Fix: consumerd: leak of tracing buffers on relayd connectivity issue

[lttng-tools.git] / src / common / consumer / consumer.c

diff --git a/src/common/consumer/consumer.c b/src/common/consumer/consumer.c
index cbd3f67a900d211fc6173de56464a4faa47c754f..ff88c7f1ef31f0ef33ea5ab8d26e8c09b81813ca 100644 (file)
--- a/src/common/consumer/consumer.c
+++ b/src/common/consumer/consumer.c
@@ -179,13 +179,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, NULL);
        }
 }
@@ -465,6 +458,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);
+                       lttng_wait_queue_wake_all(&stream->chan->metadata_pushed_wait_queue);
+
                        DBG("Delete flag set to metadata stream %d", stream->wait_fd);
                }
        }
@@ -1047,6 +1042,7 @@ struct lttng_consumer_channel *consumer_allocate_channel(uint64_t key,
        channel->is_live = is_in_live_session;
        pthread_mutex_init(&channel->lock, NULL);
        pthread_mutex_init(&channel->timer_lock, NULL);
+       lttng_wait_queue_init(&channel->metadata_pushed_wait_queue);
 
        switch (output) {
        case LTTNG_EVENT_SPLICE:
@@ -1266,11 +1262,14 @@ void lttng_consumer_set_command_sock_path(
  * 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) {
-               return lttcomm_send_unix_sock(ctx->consumer_error_socket, &cmd,
-                               sizeof(enum lttcomm_sessiond_command));
+               const int32_t comm_code = (int32_t) error_code;
+
+               return lttcomm_send_unix_sock(
+                       ctx->consumer_error_socket, &comm_code, sizeof(comm_code));
        }
 
        return 0;
@@ -2177,6 +2176,7 @@ void consumer_del_metadata_stream(struct lttng_consumer_stream *stream,
         * pointer value.
         */
        channel->metadata_stream = NULL;
+       lttng_wait_queue_wake_all(&channel->metadata_pushed_wait_queue);
 
        if (channel->metadata_cache) {
                pthread_mutex_unlock(&channel->metadata_cache->lock);