X-Git-Url: https://git.liburcu.org/?a=blobdiff_plain;f=src%2Fcommon%2Fconsumer.c;h=892c841ba0f8958ae46cff1f4d577f0426df1415;hb=41ba603553a90bde7c47eb10098224bde504d571;hp=191367cd0cd55c55be523cdf139182d4dbe8feca;hpb=1c20f0e29cbf8627bfb1ff444572d52d6655c4e2;p=lttng-tools.git

diff --git a/src/common/consumer.c b/src/common/consumer.c
index 191367cd0..892c841ba 100644
--- a/src/common/consumer.c
+++ b/src/common/consumer.c
@@ -40,6 +40,7 @@
 #include <common/kernel-consumer/kernel-consumer.h>
 #include <common/relayd/relayd.h>
 #include <common/ust-consumer/ust-consumer.h>
+#include <common/consumer-timer.h>
 
 #include "consumer.h"
 #include "consumer-stream.h"
@@ -305,6 +306,10 @@ void consumer_del_channel(struct lttng_consumer_channel *channel)
 		consumer_stream_destroy(stream, NULL);
 	}
 
+	if (channel->live_timer_enabled == 1) {
+		consumer_timer_live_stop(channel);
+	}
+
 	switch (consumer_data.type) {
 	case LTTNG_CONSUMER_KERNEL:
 		break;
@@ -515,6 +520,9 @@ struct lttng_consumer_stream *consumer_allocate_stream(uint64_t channel_key,
 		stream->metadata_flag = 1;
 		/* Metadata is flat out. */
 		strncpy(stream->name, DEFAULT_METADATA_NAME, sizeof(stream->name));
+		/* Live rendez-vous point. */
+		pthread_cond_init(&stream->metadata_rdv, NULL);
+		pthread_mutex_init(&stream->metadata_rdv_lock, NULL);
 	} else {
 		/* Format stream name to <channel_name>_<cpu_number> */
 		ret = snprintf(stream->name, sizeof(stream->name), "%s_%d",
@@ -3000,6 +3008,7 @@ void *consumer_thread_sessiond_poll(void *data)
 			 * ERR() here.
 			 */
 			DBG("Communication interrupted on command socket");
+			err = 0;
 			goto end;
 		}
 		if (consumer_quit) {
@@ -3057,6 +3066,9 @@ ssize_t lttng_consumer_read_subbuffer(struct lttng_consumer_stream *stream,
 	ssize_t ret;
 
 	pthread_mutex_lock(&stream->lock);
+	if (stream->metadata_flag) {
+		pthread_mutex_lock(&stream->metadata_rdv_lock);
+	}
 
 	switch (consumer_data.type) {
 	case LTTNG_CONSUMER_KERNEL:
@@ -3073,6 +3085,10 @@ ssize_t lttng_consumer_read_subbuffer(struct lttng_consumer_stream *stream,
 		break;
 	}
 
+	if (stream->metadata_flag) {
+		pthread_cond_broadcast(&stream->metadata_rdv);
+		pthread_mutex_unlock(&stream->metadata_rdv_lock);
+	}
 	pthread_mutex_unlock(&stream->lock);
 	return ret;
 }
@@ -3112,7 +3128,8 @@ void lttng_consumer_init(void)
 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, uint64_t sessiond_id)
+		struct lttcomm_relayd_sock *relayd_sock, uint64_t sessiond_id,
+		uint64_t relayd_session_id)
 {
 	int fd = -1, ret = -1, relayd_created = 0;
 	enum lttng_error_code ret_code = LTTNG_OK;
@@ -3212,29 +3229,7 @@ int consumer_add_relayd_socket(uint64_t net_seq_idx, int sock_type,
 		relayd->control_sock.major = relayd_sock->major;
 		relayd->control_sock.minor = relayd_sock->minor;
 
-		/*
-		 * Create a session on the relayd and store the returned id. Lock the
-		 * control socket mutex if the relayd was NOT created before.
-		 */
-		if (!relayd_created) {
-			pthread_mutex_lock(&relayd->ctrl_sock_mutex);
-		}
-		ret = relayd_create_session(&relayd->control_sock,
-				&relayd->relayd_session_id);
-		if (!relayd_created) {
-			pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
-		}
-		if (ret < 0) {
-			/*
-			 * Close all sockets of a relayd object. It will be freed if it was
-			 * created at the error code path or else it will be garbage
-			 * collect.
-			 */
-			(void) relayd_close(&relayd->control_sock);
-			(void) relayd_close(&relayd->data_sock);
-			ret_code = LTTCOMM_CONSUMERD_RELAYD_FAIL;
-			goto error;
-		}
+		relayd->relayd_session_id = relayd_session_id;
 
 		break;
 	case LTTNG_STREAM_DATA: