#include "ust-app.h"
#include "utils.h"
+/*
+ * Send a data payload using a given consumer socket of size len.
+ *
+ * The consumer socket lock MUST be acquired before calling this since this
+ * function can change the fd value.
+ *
+ * Return 0 on success else a negative value on error.
+ */
+int consumer_socket_send(struct consumer_socket *socket, void *msg, size_t len)
+{
+ int fd;
+ ssize_t size;
+
+ assert(socket);
+ assert(socket->fd);
+ assert(msg);
+
+ /* Consumer socket is invalid. Stopping. */
+ fd = *socket->fd;
+ if (fd < 0) {
+ goto error;
+ }
+
+ size = lttcomm_send_unix_sock(fd, msg, len);
+ if (size < 0) {
+ /* The above call will print a PERROR on error. */
+ DBG("Error when sending data to consumer on sock %d", fd);
+ /*
+ * At this point, the socket is not usable anymore thus flagging it
+ * invalid and closing it.
+ */
+
+ /* This call will PERROR on error. */
+ (void) lttcomm_close_unix_sock(fd);
+ *socket->fd = -1;
+ goto error;
+ }
+
+ return 0;
+
+error:
+ return -1;
+}
+
+/*
+ * Receive a data payload using a given consumer socket of size len.
+ *
+ * The consumer socket lock MUST be acquired before calling this since this
+ * function can change the fd value.
+ *
+ * Return 0 on success else a negative value on error.
+ */
+int consumer_socket_recv(struct consumer_socket *socket, void *msg, size_t len)
+{
+ int fd;
+ ssize_t size;
+
+ assert(socket);
+ assert(socket->fd);
+ assert(msg);
+
+ /* Consumer socket is invalid. Stopping. */
+ fd = *socket->fd;
+ if (fd < 0) {
+ goto error;
+ }
+
+ size = lttcomm_recv_unix_sock(fd, msg, len);
+ if (size <= 0) {
+ /* The above call will print a PERROR on error. */
+ DBG("Error when receiving data from the consumer socket %d", fd);
+ /*
+ * At this point, the socket is not usable anymore thus flagging it
+ * invalid and closing it.
+ */
+
+ /* This call will PERROR on error. */
+ (void) lttcomm_close_unix_sock(fd);
+ *socket->fd = -1;
+ goto error;
+ }
+
+ return 0;
+
+error:
+ return -1;
+}
+
/*
* Receive a reply command status message from the consumer. Consumer socket
* lock MUST be acquired before calling this function.
assert(sock);
- ret = lttcomm_recv_unix_sock(*sock->fd, &reply, sizeof(reply));
- if (ret <= 0) {
- if (ret == 0) {
- /* Orderly shutdown. Don't return 0 which means success. */
- ret = -1;
- }
- /* The above call will print a PERROR on error. */
- DBG("Fail to receive status reply on sock %d", *sock->fd);
+ ret = consumer_socket_recv(sock, &reply, sizeof(reply));
+ if (ret < 0) {
goto end;
}
assert(stream_count);
assert(key);
- ret = lttcomm_recv_unix_sock(*sock->fd, &reply, sizeof(reply));
- if (ret <= 0) {
- if (ret == 0) {
- /* Orderly shutdown. Don't return 0 which means success. */
- ret = -1;
- }
- /* The above call will print a PERROR on error. */
- DBG("Fail to receive status reply on sock %d", *sock->fd);
+ ret = consumer_socket_recv(sock, &reply, sizeof(reply));
+ if (ret < 0) {
goto end;
}
DBG2("Sending destroy relayd command to consumer sock %d", *sock->fd);
- /* Bail out if consumer is disabled */
- if (!consumer->enabled) {
- ret = LTTNG_OK;
- DBG3("Consumer is disabled");
- goto error;
- }
-
msg.cmd_type = LTTNG_CONSUMER_DESTROY_RELAYD;
msg.u.destroy_relayd.net_seq_idx = consumer->net_seq_index;
pthread_mutex_lock(sock->lock);
- ret = lttcomm_send_unix_sock(*sock->fd, &msg, sizeof(msg));
+ ret = consumer_socket_send(sock, &msg, sizeof(msg));
if (ret < 0) {
- /* Indicate that the consumer is probably closing at this point. */
- DBG("send consumer destroy relayd command");
- goto error_send;
+ goto error;
}
/* Don't check the return value. The caller will do it. */
DBG2("Consumer send destroy relayd command done");
-error_send:
- pthread_mutex_unlock(sock->lock);
error:
+ pthread_mutex_unlock(sock->lock);
return ret;
}
assert(sock);
assert(sock->fd);
- ret = lttcomm_send_unix_sock(*sock->fd, msg,
- sizeof(struct lttcomm_consumer_msg));
+ ret = consumer_socket_send(sock, msg, sizeof(struct lttcomm_consumer_msg));
if (ret < 0) {
- /* The above call will print a PERROR on error. */
- DBG("Error when sending consumer channel on sock %d", *sock->fd);
goto error;
}
assert(sock);
assert(sock->fd);
- ret = lttcomm_send_unix_sock(*sock->fd, msg,
- sizeof(struct lttcomm_consumer_msg));
+ ret = consumer_send_msg(sock, msg);
if (ret < 0) {
- /* The above call will print a PERROR on error. */
- DBG("Error when sending consumer channel on sock %d", *sock->fd);
goto error;
}
- ret = consumer_recv_status_reply(sock);
-
error:
return ret;
}
assert(sock->fd);
assert(fds);
- /* Send on socket */
- ret = lttcomm_send_unix_sock(*sock->fd, msg,
- sizeof(struct lttcomm_consumer_msg));
- if (ret < 0) {
- /* The above call will print a PERROR on error. */
- DBG("Error when sending consumer stream on sock %d", *sock->fd);
- goto error;
- }
-
- ret = consumer_recv_status_reply(sock);
+ ret = consumer_send_msg(sock, msg);
if (ret < 0) {
goto error;
}
memcpy(&msg.u.relayd_sock.sock, rsock, sizeof(msg.u.relayd_sock.sock));
DBG3("Sending relayd sock info to consumer on %d", *consumer_sock->fd);
- ret = lttcomm_send_unix_sock(*consumer_sock->fd, &msg, sizeof(msg));
- if (ret < 0) {
- /* The above call will print a PERROR on error. */
- DBG("Error when sending relayd sockets on sock %d", rsock->sock.fd);
- goto error;
- }
-
- ret = consumer_recv_status_reply(consumer_sock);
+ ret = consumer_send_msg(consumer_sock, &msg);
if (ret < 0) {
goto error;
}
assert(socket->fd);
pthread_mutex_lock(socket->lock);
-
- ret = lttcomm_send_unix_sock(*socket->fd, &msg, sizeof(msg));
+ ret = consumer_socket_send(socket, &msg, sizeof(msg));
if (ret < 0) {
- /* The above call will print a PERROR on error. */
- DBG("Error on consumer is data pending on sock %d", *socket->fd);
pthread_mutex_unlock(socket->lock);
goto error_unlock;
}
* the reply status message.
*/
- ret = lttcomm_recv_unix_sock(*socket->fd, &ret_code, sizeof(ret_code));
- if (ret <= 0) {
- if (ret == 0) {
- /* Orderly shutdown. Don't return 0 which means success. */
- ret = -1;
- }
- /* The above call will print a PERROR on error. */
- DBG("Error on recv consumer is data pending on sock %d", *socket->fd);
+ ret = consumer_socket_recv(socket, &ret_code, sizeof(ret_code));
+ if (ret < 0) {
pthread_mutex_unlock(socket->lock);
goto error_unlock;
}
-
pthread_mutex_unlock(socket->lock);
if (ret_code == 1) {
DBG3("Consumer pushing metadata on sock %d of len %zu", *socket->fd, len);
- ret = lttcomm_send_unix_sock(*socket->fd, metadata_str, len);
+ ret = consumer_socket_send(socket, metadata_str, len);
if (ret < 0) {
goto end;
}
projects / lttng-tools.git / blobdiff