X-Git-Url: http://git.lttng.org/?a=blobdiff_plain;f=src%2Fcommon%2Fconsumer%2Fconsumer.c;h=3c20c57bda9d68ac1967de0cf433901ab8cce7e6;hb=09ede842515f606976233d5cc538fc47a60abc09;hp=c7b77010c76bf4a157941f1d35ade3e00286e4a9;hpb=92b7a7f81a94f7616c155aed8a213edd68937a57;p=lttng-tools.git

diff --git a/src/common/consumer/consumer.c b/src/common/consumer/consumer.c
index c7b77010c..3c20c57bd 100644
--- a/src/common/consumer/consumer.c
+++ b/src/common/consumer/consumer.c
@@ -393,6 +393,10 @@ void consumer_del_channel(struct lttng_consumer_channel *channel)
 	iter.iter.node = &channel->node.node;
 	ret = lttng_ht_del(consumer_data.channel_ht, &iter);
 	assert(!ret);
+
+	iter.iter.node = &channel->channels_by_session_id_ht_node.node;
+	ret = lttng_ht_del(consumer_data.channels_by_session_id_ht, &iter);
+	assert(!ret);
 	rcu_read_unlock();
 
 	call_rcu(&channel->node.head, free_channel_rcu);
@@ -1035,6 +1039,8 @@ struct lttng_consumer_channel *consumer_allocate_channel(uint64_t key,
 	}
 
 	lttng_ht_node_init_u64(&channel->node, channel->key);
+	lttng_ht_node_init_u64(&channel->channels_by_session_id_ht_node,
+			channel->session_id);
 
 	channel->wait_fd = -1;
 
@@ -1067,6 +1073,8 @@ int consumer_add_channel(struct lttng_consumer_channel *channel,
 
 	rcu_read_lock();
 	lttng_ht_add_unique_u64(consumer_data.channel_ht, &channel->node);
+	lttng_ht_add_u64(consumer_data.channels_by_session_id_ht,
+			&channel->channels_by_session_id_ht_node);
 	rcu_read_unlock();
 
 	pthread_mutex_unlock(&channel->timer_lock);
@@ -1225,6 +1233,7 @@ void lttng_consumer_cleanup(void)
 	rcu_read_unlock();
 
 	lttng_ht_destroy(consumer_data.channel_ht);
+	lttng_ht_destroy(consumer_data.channels_by_session_id_ht);
 
 	cleanup_relayd_ht();
 
@@ -1236,6 +1245,8 @@ void lttng_consumer_cleanup(void)
 	 * it.
 	 */
 	lttng_ht_destroy(consumer_data.stream_list_ht);
+
+	lttng_trace_chunk_registry_destroy(consumer_data.chunk_registry);
 }
 
 /*
@@ -2396,11 +2407,6 @@ restart:
 			revents = LTTNG_POLL_GETEV(&events, i);
 			pollfd = LTTNG_POLL_GETFD(&events, i);
 
-			if (!revents) {
-				/* No activity for this FD (poll implementation). */
-				continue;
-			}
-
 			if (pollfd == lttng_pipe_get_readfd(ctx->consumer_metadata_pipe)) {
 				if (revents & LPOLLIN) {
 					ssize_t pipe_len;
@@ -2990,11 +2996,6 @@ restart:
 			revents = LTTNG_POLL_GETEV(&events, i);
 			pollfd = LTTNG_POLL_GETFD(&events, i);
 
-			if (!revents) {
-				/* No activity for this FD (poll implementation). */
-				continue;
-			}
-
 			if (pollfd == ctx->consumer_channel_pipe[0]) {
 				if (revents & LPOLLIN) {
 					enum consumer_channel_action action;
@@ -3433,6 +3434,12 @@ int lttng_consumer_init(void)
 		goto error;
 	}
 
+	consumer_data.channels_by_session_id_ht =
+			lttng_ht_new(0, LTTNG_HT_TYPE_U64);
+	if (!consumer_data.channels_by_session_id_ht) {
+		goto error;
+	}
+
 	consumer_data.relayd_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
 	if (!consumer_data.relayd_ht) {
 		goto error;
@@ -3458,6 +3465,11 @@ int lttng_consumer_init(void)
 		goto error;
 	}
 
+	consumer_data.chunk_registry = lttng_trace_chunk_registry_create();
+	if (!consumer_data.chunk_registry) {
+		goto error;
+	}
+
 	return 0;
 
 error:
@@ -3568,7 +3580,6 @@ error:
 
 		/* Assign new file descriptor */
 		relayd->control_sock.sock.fd = fd;
-		fd = -1;	/* For error path */
 		/* Assign version values. */
 		relayd->control_sock.major = relayd_sock->major;
 		relayd->control_sock.minor = relayd_sock->minor;
@@ -3596,7 +3607,6 @@ error:
 
 		/* Assign new file descriptor */
 		relayd->data_sock.sock.fd = fd;
-		fd = -1;	/* for eventual error paths */
 		/* Assign version values. */
 		relayd->data_sock.major = relayd_sock->major;
 		relayd->data_sock.minor = relayd_sock->minor;
@@ -3610,6 +3620,11 @@ error:
 	DBG("Consumer %s socket created successfully with net idx %" PRIu64 " (fd: %d)",
 			sock_type == LTTNG_STREAM_CONTROL ? "control" : "data",
 			relayd->net_seq_idx, fd);
+	/*
+	 * We gave the ownership of the fd to the relayd structure. Set the
+	 * fd to -1 so we don't call close() on it in the error path below.
+	 */
+	fd = -1;
 
 	/* We successfully added the socket. Send status back. */
 	ret = consumer_send_status_msg(sock, ret_code);
@@ -4470,3 +4485,24 @@ int lttng_consumer_mkdir(const char *path, uid_t uid, gid_t gid,
 		return mkdir_local(path, uid, gid);
 	}
 }
+
+enum lttcomm_return_code lttng_consumer_init_command(
+		struct lttng_consumer_local_data *ctx,
+		const lttng_uuid sessiond_uuid)
+{
+	enum lttcomm_return_code ret;
+	char uuid_str[UUID_STR_LEN];
+
+	if (ctx->sessiond_uuid.is_set) {
+		ret = LTTCOMM_CONSUMERD_ALREADY_SET;
+		goto end;
+	}
+
+	ctx->sessiond_uuid.is_set = true;
+	memcpy(ctx->sessiond_uuid.value, sessiond_uuid, sizeof(lttng_uuid));
+	ret = LTTCOMM_CONSUMERD_SUCCESS;
+	lttng_uuid_to_str(sessiond_uuid, uuid_str);
+	DBG("Received session daemon UUID: %s", uuid_str);
+end:
+	return ret;
+}