X-Git-Url: https://git.lttng.org/?a=blobdiff_plain;f=src%2Fcommon%2Fconsumer%2Fconsumer.cpp;h=1da243601cd7ff3c72fc8febb5d6c11f2f150c4b;hb=HEAD;hp=01845871b91a2a2423e4375e983de1cac6243b7f;hpb=671e39d79a1ad9c3f13c4784a26710a5b1f14237;p=lttng-tools.git

diff --git a/src/common/consumer/consumer.cpp b/src/common/consumer/consumer.cpp
index 01845871b..1da243601 100644
--- a/src/common/consumer/consumer.cpp
+++ b/src/common/consumer/consumer.cpp
@@ -45,6 +45,7 @@
 #include <sys/mman.h>
 #include <sys/socket.h>
 #include <sys/types.h>
+#include <type_traits>
 #include <unistd.h>
 
 lttng_consumer_global_data the_consumer_data;
@@ -177,13 +178,6 @@ static void clean_channel_stream_list(struct lttng_consumer_channel *channel)
 
 	/* Delete streams that might have been left in the stream list. */
 	cds_list_for_each_entry_safe (stream, stmp, &channel->streams.head, send_node) {
-		/*
-		 * Once a stream is added to this list, the buffers were created so we
-		 * have a guarantee that this call will succeed. Setting the monitor
-		 * mode to 0 so we don't lock nor try to delete the stream from the
-		 * global hash table.
-		 */
-		stream->monitor = 0;
 		consumer_stream_destroy(stream, nullptr);
 	}
 }
@@ -303,7 +297,8 @@ static void free_channel_rcu(struct rcu_head *head)
 		ERR("Unknown consumer_data type");
 		abort();
 	}
-	free(channel);
+
+	delete channel;
 }
 
 /*
@@ -462,6 +457,8 @@ static void update_endpoint_status_by_netidx(uint64_t net_seq_idx,
 	cds_lfht_for_each_entry (metadata_ht->ht, &iter.iter, stream, node.node) {
 		if (stream->net_seq_idx == net_seq_idx) {
 			uatomic_set(&stream->endpoint_status, status);
+			stream->chan->metadata_pushed_wait_queue.wake_all();
+
 			DBG("Delete flag set to metadata stream %d", stream->wait_fd);
 		}
 	}
@@ -1017,9 +1014,11 @@ struct lttng_consumer_channel *consumer_allocate_channel(uint64_t key,
 		}
 	}
 
-	channel = zmalloc<lttng_consumer_channel>();
-	if (channel == nullptr) {
-		PERROR("malloc struct lttng_consumer_channel");
+	try {
+		channel = new lttng_consumer_channel;
+	} catch (const std::bad_alloc& e) {
+		ERR("Failed to allocate lttng_consumer_channel: %s", e.what());
+		channel = nullptr;
 		goto end;
 	}
 
@@ -1033,8 +1032,8 @@ struct lttng_consumer_channel *consumer_allocate_channel(uint64_t key,
 	channel->monitor = monitor;
 	channel->live_timer_interval = live_timer_interval;
 	channel->is_live = is_in_live_session;
-	pthread_mutex_init(&channel->lock, nullptr);
-	pthread_mutex_init(&channel->timer_lock, nullptr);
+	pthread_mutex_init(&channel->lock, NULL);
+	pthread_mutex_init(&channel->timer_lock, NULL);
 
 	switch (output) {
 	case LTTNG_EVENT_SPLICE:
@@ -1045,7 +1044,7 @@ struct lttng_consumer_channel *consumer_allocate_channel(uint64_t key,
 		break;
 	default:
 		abort();
-		free(channel);
+		delete channel;
 		channel = nullptr;
 		goto end;
 	}
@@ -1250,11 +1249,17 @@ void lttng_consumer_set_command_sock_path(struct lttng_consumer_local_data *ctx,
  * Send return code to the session daemon.
  * If the socket is not defined, we return 0, it is not a fatal error
  */
-int lttng_consumer_send_error(struct lttng_consumer_local_data *ctx, int cmd)
+int lttng_consumer_send_error(struct lttng_consumer_local_data *ctx,
+			      enum lttcomm_return_code error_code)
 {
 	if (ctx->consumer_error_socket > 0) {
+		const std::int32_t comm_code = std::int32_t(error_code);
+
+		static_assert(
+			sizeof(comm_code) >= sizeof(std::underlying_type<lttcomm_return_code>),
+			"Fixed-size communication type too small to accomodate lttcomm_return_code");
 		return lttcomm_send_unix_sock(
-			ctx->consumer_error_socket, &cmd, sizeof(enum lttcomm_sessiond_command));
+			ctx->consumer_error_socket, &comm_code, sizeof(comm_code));
 	}
 
 	return 0;
@@ -2130,6 +2135,7 @@ void consumer_del_metadata_stream(struct lttng_consumer_stream *stream, struct l
 	 * pointer value.
 	 */
 	channel->metadata_stream = nullptr;
+	channel->metadata_pushed_wait_queue.wake_all();
 
 	if (channel->metadata_cache) {
 		pthread_mutex_unlock(&channel->metadata_cache->lock);