X-Git-Url: https://git.lttng.org/?a=blobdiff_plain;f=src%2Fcommon%2Fconsumer.c;h=1f96fa9d460c5a97eac61a9c79acadbf239a7e91;hb=618a6a28c0956fc6829c165a39ffec97f239096c;hp=0518f44ca902e8b55054d93a28a17314b89a740f;hpb=95a1109b9ea25871941d9fff87cd8c2b2cba61d3;p=lttng-tools.git

diff --git a/src/common/consumer.c b/src/common/consumer.c
index 0518f44ca..1f96fa9d4 100644
--- a/src/common/consumer.c
+++ b/src/common/consumer.c
@@ -165,20 +165,20 @@ static struct lttng_consumer_stream *find_stream(uint64_t key,
 	return stream;
 }
 
-static void steal_stream_key(int key, struct lttng_ht *ht)
+static void steal_stream_key(uint64_t key, struct lttng_ht *ht)
 {
 	struct lttng_consumer_stream *stream;
 
 	rcu_read_lock();
 	stream = find_stream(key, ht);
 	if (stream) {
-		stream->key = -1ULL;
+		stream->key = (uint64_t) -1ULL;
 		/*
 		 * We don't want the lookup to match, but we still need
 		 * to iterate on this stream when iterating over the hash table. Just
 		 * change the node key.
 		 */
-		stream->node.key = -1ULL;
+		stream->node.key = (uint64_t) -1ULL;
 	}
 	rcu_read_unlock();
 }
@@ -370,13 +370,13 @@ static void cleanup_relayd_ht(void)
  * It's atomically set without having the stream mutex locked which is fine
  * because we handle the write/read race with a pipe wakeup for each thread.
  */
-static void update_endpoint_status_by_netidx(int net_seq_idx,
+static void update_endpoint_status_by_netidx(uint64_t net_seq_idx,
 		enum consumer_endpoint_status status)
 {
 	struct lttng_ht_iter iter;
 	struct lttng_consumer_stream *stream;
 
-	DBG("Consumer set delete flag on stream by idx %d", net_seq_idx);
+	DBG("Consumer set delete flag on stream by idx %" PRIu64, net_seq_idx);
 
 	rcu_read_lock();
 
@@ -409,7 +409,7 @@ static void update_endpoint_status_by_netidx(int net_seq_idx,
 static void cleanup_relayd(struct consumer_relayd_sock_pair *relayd,
 		struct lttng_consumer_local_data *ctx)
 {
-	int netidx;
+	uint64_t netidx;
 
 	assert(relayd);
 
@@ -637,12 +637,12 @@ end:
  * Allocate and return a consumer relayd socket.
  */
 struct consumer_relayd_sock_pair *consumer_allocate_relayd_sock_pair(
-		int net_seq_idx)
+		uint64_t net_seq_idx)
 {
 	struct consumer_relayd_sock_pair *obj = NULL;
 
-	/* Negative net sequence index is a failure */
-	if (net_seq_idx < 0) {
+	/* net sequence index of -1 is a failure */
+	if (net_seq_idx == (uint64_t) -1ULL) {
 		goto error;
 	}
 
@@ -1259,7 +1259,7 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap(
 	rcu_read_lock();
 
 	/* Flag that the current stream if set for network streaming. */
-	if (stream->net_seq_idx != -1) {
+	if (stream->net_seq_idx != (uint64_t) -1ULL) {
 		relayd = consumer_find_relayd(stream->net_seq_idx);
 		if (relayd == NULL) {
 			goto end;
@@ -1458,7 +1458,7 @@ ssize_t lttng_consumer_on_read_subbuffer_splice(
 	rcu_read_lock();
 
 	/* Flag that the current stream if set for network streaming. */
-	if (stream->net_seq_idx != -1) {
+	if (stream->net_seq_idx != (uint64_t) -1ULL) {
 		relayd = consumer_find_relayd(stream->net_seq_idx);
 		if (relayd == NULL) {
 			goto end;
@@ -2996,7 +2996,7 @@ void lttng_consumer_init(void)
  * This will create a relayd socket pair and add it to the relayd hash table.
  * The caller MUST acquire a RCU read side lock before calling it.
  */
-int consumer_add_relayd_socket(int net_seq_idx, int sock_type,
+int consumer_add_relayd_socket(uint64_t net_seq_idx, int sock_type,
 		struct lttng_consumer_local_data *ctx, int sock,
 		struct pollfd *consumer_sockpoll,
 		struct lttcomm_relayd_sock *relayd_sock, unsigned int sessiond_id)
@@ -3008,16 +3008,18 @@ int consumer_add_relayd_socket(int net_seq_idx, int sock_type,
 	assert(ctx);
 	assert(relayd_sock);
 
-	DBG("Consumer adding relayd socket (idx: %d)", net_seq_idx);
+	DBG("Consumer adding relayd socket (idx: %" PRIu64 ")", net_seq_idx);
 
 	/* Get relayd reference if exists. */
 	relayd = consumer_find_relayd(net_seq_idx);
 	if (relayd == NULL) {
+		assert(sock_type == LTTNG_STREAM_CONTROL);
 		/* Not found. Allocate one. */
 		relayd = consumer_allocate_relayd_sock_pair(net_seq_idx);
 		if (relayd == NULL) {
-			ret_code = LTTCOMM_CONSUMERD_ENOMEM;
 			ret = -ENOMEM;
+			ret_code = LTTCOMM_CONSUMERD_ENOMEM;
+			goto error;
 		} else {
 			relayd->sessiond_session_id = (uint64_t) sessiond_id;
 			relayd_created = 1;
@@ -3028,26 +3030,31 @@ int consumer_add_relayd_socket(int net_seq_idx, int sock_type,
 		 * we can notify the session daemon and continue our work without
 		 * killing everything.
 		 */
+	} else {
+		/*
+		 * relayd key should never be found for control socket.
+		 */
+		assert(sock_type != LTTNG_STREAM_CONTROL);
 	}
 
 	/* First send a status message before receiving the fds. */
-	ret = consumer_send_status_msg(sock, ret_code);
-	if (ret < 0 || ret_code != LTTNG_OK) {
+	ret = consumer_send_status_msg(sock, LTTNG_OK);
+	if (ret < 0) {
 		/* Somehow, the session daemon is not responding anymore. */
-		goto error;
+		lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_FATAL);
+		goto error_nosignal;
 	}
 
 	/* Poll on consumer socket. */
 	if (lttng_consumer_poll_socket(consumer_sockpoll) < 0) {
 		lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_POLL_ERROR);
 		ret = -EINTR;
-		goto error;
+		goto error_nosignal;
 	}
 
 	/* Get relayd socket from session daemon */
 	ret = lttcomm_recv_fds_unix_sock(sock, &fd, 1);
 	if (ret != sizeof(fd)) {
-		ret_code = LTTCOMM_CONSUMERD_ERROR_RECV_FD;
 		ret = -1;
 		fd = -1;	/* Just in case it gets set with an invalid value. */
 
@@ -3061,18 +3068,7 @@ int consumer_add_relayd_socket(int net_seq_idx, int sock_type,
 		 * issue when reaching the fd limit.
 		 */
 		lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_ERROR_RECV_FD);
-
-		/*
-		 * This code path MUST continue to the consumer send status message so
-		 * we can send the error to the thread expecting a reply. The above
-		 * call will make everything stop.
-		 */
-	}
-
-	/* We have the fds without error. Send status back. */
-	ret = consumer_send_status_msg(sock, ret_code);
-	if (ret < 0 || ret_code != LTTNG_OK) {
-		/* Somehow, the session daemon is not responding anymore. */
+		ret_code = LTTCOMM_CONSUMERD_ERROR_RECV_FD;
 		goto error;
 	}
 
@@ -3082,16 +3078,19 @@ int consumer_add_relayd_socket(int net_seq_idx, int sock_type,
 		/* Copy received lttcomm socket */
 		lttcomm_copy_sock(&relayd->control_sock.sock, &relayd_sock->sock);
 		ret = lttcomm_create_sock(&relayd->control_sock.sock);
-		/* Immediately try to close the created socket if valid. */
-		if (relayd->control_sock.sock.fd >= 0) {
-			if (close(relayd->control_sock.sock.fd)) {
-				PERROR("close relayd control socket");
-			}
-		}
 		/* Handle create_sock error. */
 		if (ret < 0) {
+			ret_code = LTTCOMM_CONSUMERD_ENOMEM;
 			goto error;
 		}
+		/*
+		 * Close the socket created internally by
+		 * lttcomm_create_sock, so we can replace it by the one
+		 * received from sessiond.
+		 */
+		if (close(relayd->control_sock.sock.fd)) {
+			PERROR("close");
+		}
 
 		/* Assign new file descriptor */
 		relayd->control_sock.sock.fd = fd;
@@ -3120,6 +3119,7 @@ int consumer_add_relayd_socket(int net_seq_idx, int sock_type,
 			 */
 			(void) relayd_close(&relayd->control_sock);
 			(void) relayd_close(&relayd->data_sock);
+			ret_code = LTTCOMM_CONSUMERD_RELAYD_FAIL;
 			goto error;
 		}
 
@@ -3128,16 +3128,19 @@ int consumer_add_relayd_socket(int net_seq_idx, int sock_type,
 		/* Copy received lttcomm socket */
 		lttcomm_copy_sock(&relayd->data_sock.sock, &relayd_sock->sock);
 		ret = lttcomm_create_sock(&relayd->data_sock.sock);
-		/* Immediately try to close the created socket if valid. */
-		if (relayd->data_sock.sock.fd >= 0) {
-			if (close(relayd->data_sock.sock.fd)) {
-				PERROR("close relayd data socket");
-			}
-		}
 		/* Handle create_sock error. */
 		if (ret < 0) {
+			ret_code = LTTCOMM_CONSUMERD_ENOMEM;
 			goto error;
 		}
+		/*
+		 * Close the socket created internally by
+		 * lttcomm_create_sock, so we can replace it by the one
+		 * received from sessiond.
+		 */
+		if (close(relayd->data_sock.sock.fd)) {
+			PERROR("close");
+		}
 
 		/* Assign new file descriptor */
 		relayd->data_sock.sock.fd = fd;
@@ -3149,6 +3152,7 @@ int consumer_add_relayd_socket(int net_seq_idx, int sock_type,
 	default:
 		ERR("Unknown relayd socket type (%d)", sock_type);
 		ret = -1;
+		ret_code = LTTCOMM_CONSUMERD_FATAL;
 		goto error;
 	}
 
@@ -3156,6 +3160,14 @@ int consumer_add_relayd_socket(int net_seq_idx, int sock_type,
 			sock_type == LTTNG_STREAM_CONTROL ? "control" : "data",
 			relayd->net_seq_idx, fd);
 
+	/* We successfully added the socket. Send status back. */
+	ret = consumer_send_status_msg(sock, ret_code);
+	if (ret < 0) {
+		/* Somehow, the session daemon is not responding anymore. */
+		lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_FATAL);
+		goto error_nosignal;
+	}
+
 	/*
 	 * Add relayd socket pair to consumer data hashtable. If object already
 	 * exists or on error, the function gracefully returns.
@@ -3166,6 +3178,11 @@ int consumer_add_relayd_socket(int net_seq_idx, int sock_type,
 	return 0;
 
 error:
+	if (consumer_send_status_msg(sock, ret_code) < 0) {
+		lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_FATAL);
+	}
+
+error_nosignal:
 	/* Close received socket if valid. */
 	if (fd >= 0) {
 		if (close(fd)) {