X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=src%2Fbin%2Flttng-sessiond%2Fconsumer.c;h=8f05476a7fe606935d8f3fcca23d75b2e945bb43;hb=92db7cdc97f4bb5776fa698442d8af4e5c1e3bf3;hp=39a1946380120db04419809e52a147a08b999726;hpb=4ce514c43483ba24fd935024da5b7aca681a7e52;p=lttng-tools.git

diff --git a/src/bin/lttng-sessiond/consumer.c b/src/bin/lttng-sessiond/consumer.c
index 39a194638..8f05476a7 100644
--- a/src/bin/lttng-sessiond/consumer.c
+++ b/src/bin/lttng-sessiond/consumer.c
@@ -34,6 +34,94 @@
 #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_ptr);
+	assert(msg);
+
+	/* Consumer socket is invalid. Stopping. */
+	fd = *socket->fd_ptr;
+	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 closing it and
+		 * setting the file descriptor to -1 so it is not reused.
+		 */
+
+		/* This call will PERROR on error. */
+		(void) lttcomm_close_unix_sock(fd);
+		*socket->fd_ptr = -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_ptr);
+	assert(msg);
+
+	/* Consumer socket is invalid. Stopping. */
+	fd = *socket->fd_ptr;
+	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 closing it and
+		 * setting the file descriptor to -1 so it is not reused.
+		 */
+
+		/* This call will PERROR on error. */
+		(void) lttcomm_close_unix_sock(fd);
+		*socket->fd_ptr = -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.
@@ -48,14 +136,8 @@ int consumer_recv_status_reply(struct consumer_socket *sock)
 
 	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;
 	}
 
@@ -89,14 +171,8 @@ int consumer_recv_status_channel(struct consumer_socket *sock,
 	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;
 	}
 
@@ -127,24 +203,15 @@ int consumer_send_destroy_relayd(struct consumer_socket *sock,
 	assert(consumer);
 	assert(sock);
 
-	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;
-	}
+	DBG2("Sending destroy relayd command to consumer sock %d", *sock->fd_ptr);
 
 	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. */
@@ -152,9 +219,8 @@ int consumer_send_destroy_relayd(struct consumer_socket *sock,
 
 	DBG2("Consumer send destroy relayd command done");
 
-error_send:
-	pthread_mutex_unlock(sock->lock);
 error:
+	pthread_mutex_unlock(sock->lock);
 	return ret;
 }
 
@@ -312,7 +378,7 @@ struct consumer_socket *consumer_allocate_socket(int *fd)
 		goto error;
 	}
 
-	socket->fd = fd;
+	socket->fd_ptr = fd;
 	lttng_ht_node_init_ulong(&socket->node, *fd);
 
 error:
@@ -377,8 +443,8 @@ void consumer_destroy_socket(struct consumer_socket *sock)
 	 * consumer was registered,
 	 */
 	if (sock->registered) {
-		DBG3("Consumer socket was registered. Closing fd %d", *sock->fd);
-		lttcomm_close_unix_sock(*sock->fd);
+		DBG3("Consumer socket was registered. Closing fd %d", *sock->fd_ptr);
+		lttcomm_close_unix_sock(*sock->fd_ptr);
 	}
 
 	call_rcu(&sock->node.head, destroy_socket_rcu);
@@ -509,13 +575,13 @@ int consumer_copy_sockets(struct consumer_output *dst,
 	rcu_read_lock();
 	cds_lfht_for_each_entry(src->socks->ht, &iter.iter, socket, node.node) {
 		/* Ignore socket that are already there. */
-		copy_sock = consumer_find_socket(*socket->fd, dst);
+		copy_sock = consumer_find_socket(*socket->fd_ptr, dst);
 		if (copy_sock) {
 			continue;
 		}
 
 		/* Create new socket object. */
-		copy_sock = consumer_allocate_socket(socket->fd);
+		copy_sock = consumer_allocate_socket(socket->fd_ptr);
 		if (copy_sock == NULL) {
 			rcu_read_unlock();
 			ret = -ENOMEM;
@@ -648,13 +714,12 @@ int consumer_send_fds(struct consumer_socket *sock, int *fds, size_t nb_fd)
 
 	assert(fds);
 	assert(sock);
-	assert(sock->fd);
 	assert(nb_fd > 0);
 
-	ret = lttcomm_send_fds_unix_sock(*sock->fd, fds, nb_fd);
+	ret = lttcomm_send_fds_unix_sock(*sock->fd_ptr, fds, nb_fd);
 	if (ret < 0) {
 		/* The above call will print a PERROR on error. */
-		DBG("Error when sending consumer fds on sock %d", *sock->fd);
+		DBG("Error when sending consumer fds on sock %d", *sock->fd_ptr);
 		goto error;
 	}
 
@@ -674,13 +739,9 @@ int consumer_send_msg(struct consumer_socket *sock,
 
 	assert(msg);
 	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;
 	}
 
@@ -700,18 +761,12 @@ int consumer_send_channel(struct consumer_socket *sock,
 
 	assert(msg);
 	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;
 }
@@ -857,19 +912,9 @@ int consumer_send_stream(struct consumer_socket *sock,
 	assert(msg);
 	assert(dst);
 	assert(sock);
-	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;
 	}
@@ -899,7 +944,6 @@ int consumer_send_relayd_socket(struct consumer_socket *consumer_sock,
 	assert(rsock);
 	assert(consumer);
 	assert(consumer_sock);
-	assert(consumer_sock->fd);
 
 	/* Bail out if consumer is disabled */
 	if (!consumer->enabled) {
@@ -918,15 +962,8 @@ int consumer_send_relayd_socket(struct consumer_socket *consumer_sock,
 	msg.u.relayd_sock.session_id = session_id;
 	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);
+	DBG3("Sending relayd sock info to consumer on %d", *consumer_sock->fd_ptr);
+	ret = consumer_send_msg(consumer_sock, &msg);
 	if (ret < 0) {
 		goto error;
 	}
@@ -1019,15 +1056,9 @@ int consumer_is_data_pending(uint64_t session_id,
 	rcu_read_lock();
 	cds_lfht_for_each_entry(consumer->socks->ht, &iter.iter, socket,
 			node.node) {
-		/* Code flow 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;
 		}
@@ -1037,18 +1068,11 @@ int consumer_is_data_pending(uint64_t session_id,
 		 * 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) {
@@ -1077,7 +1101,6 @@ int consumer_flush_channel(struct consumer_socket *socket, uint64_t key)
 	struct lttcomm_consumer_msg msg;
 
 	assert(socket);
-	assert(socket->fd);
 
 	DBG2("Consumer flush channel key %" PRIu64, key);
 
@@ -1110,7 +1133,6 @@ int consumer_close_metadata(struct consumer_socket *socket,
 	struct lttcomm_consumer_msg msg;
 
 	assert(socket);
-	assert(socket->fd);
 
 	DBG2("Consumer close metadata channel key %" PRIu64, metadata_key);
 
@@ -1143,7 +1165,6 @@ int consumer_setup_metadata(struct consumer_socket *socket,
 	struct lttcomm_consumer_msg msg;
 
 	assert(socket);
-	assert(socket->fd);
 
 	DBG2("Consumer setup metadata channel key %" PRIu64, metadata_key);
 
@@ -1177,9 +1198,8 @@ int consumer_push_metadata(struct consumer_socket *socket,
 	struct lttcomm_consumer_msg msg;
 
 	assert(socket);
-	assert(socket->fd);
 
-	DBG2("Consumer push metadata to consumer socket %d", *socket->fd);
+	DBG2("Consumer push metadata to consumer socket %d", *socket->fd_ptr);
 
 	msg.cmd_type = LTTNG_CONSUMER_PUSH_METADATA;
 	msg.u.push_metadata.key = metadata_key;
@@ -1192,9 +1212,10 @@ int consumer_push_metadata(struct consumer_socket *socket,
 		goto end;
 	}
 
-	DBG3("Consumer pushing metadata on sock %d of len %zu", *socket->fd, len);
+	DBG3("Consumer pushing metadata on sock %d of len %zu", *socket->fd_ptr,
+			len);
 
-	ret = lttcomm_send_unix_sock(*socket->fd, metadata_str, len);
+	ret = consumer_socket_send(socket, metadata_str, len);
 	if (ret < 0) {
 		goto end;
 	}
@@ -1223,7 +1244,6 @@ int consumer_snapshot_channel(struct consumer_socket *socket, uint64_t key,
 	struct lttcomm_consumer_msg msg;
 
 	assert(socket);
-	assert(socket->fd);
 	assert(output);
 	assert(output->consumer);