Propagate trace format all the way to the consumer

[deliverable/lttng-tools.git] / src / common / ust-consumer / ust-consumer.cpp

diff --git a/src/common/ust-consumer/ust-consumer.cpp b/src/common/ust-consumer/ust-consumer.cpp
index 766c9d30c7ff27f1a7ff796043f24aaf2a3ad90a..25da16ff0ec85252e4ab0285007aeee50730f8a9 100644 (file)
--- a/src/common/ust-consumer/ust-consumer.cpp
+++ b/src/common/ust-consumer/ust-consumer.cpp
@@ -49,46 +49,6 @@ extern int consumer_poll_timeout;
 
 LTTNG_EXPORT DEFINE_LTTNG_UST_SIGBUS_STATE();
 
-/*
- * Free channel object and all streams associated with it. This MUST be used
- * only and only if the channel has _NEVER_ been added to the global channel
- * hash table.
- */
-static void destroy_channel(struct lttng_consumer_channel *channel)
-{
-       struct lttng_consumer_stream *stream, *stmp;
-
-       LTTNG_ASSERT(channel);
-
-       DBG("UST consumer cleaning stream list");
-
-       cds_list_for_each_entry_safe(stream, stmp, &channel->streams.head,
-                       send_node) {
-
-               health_code_update();
-
-               cds_list_del_init(&stream->send_node);
-               lttng_ust_ctl_destroy_stream(stream->ustream);
-               lttng_trace_chunk_put(stream->trace_chunk);
-               free(stream);
-       }
-
-       /*
-        * If a channel is available meaning that was created before the streams
-        * were, delete it.
-        */
-       if (channel->uchan) {
-               lttng_ustconsumer_del_channel(channel);
-               lttng_ustconsumer_free_channel(channel);
-       }
-
-       if (channel->trace_chunk) {
-               lttng_trace_chunk_put(channel->trace_chunk);
-       }
-
-       free(channel);
-}
-
 /*
  * Add channel to internal consumer state.
  *
@@ -137,18 +97,9 @@ static struct lttng_consumer_stream *allocate_stream(int cpu, int key,
        LTTNG_ASSERT(channel);
        LTTNG_ASSERT(ctx);
 
-       stream = consumer_stream_create(
-                       channel,
-                       channel->key,
-                       key,
-                       channel->name,
-                       channel->relayd_id,
-                       channel->session_id,
-                       channel->trace_chunk,
-                       cpu,
-                       &alloc_ret,
-                       channel->type,
-                       channel->monitor);
+       stream = consumer_stream_create(channel, channel->key, key, channel->name,
+                       channel->relayd_id, channel->session_id, channel->trace_chunk, cpu,
+                       &alloc_ret, channel->type, channel->monitor, channel->trace_format);
        if (stream == NULL) {
                switch (alloc_ret) {
                case -ENOENT:
@@ -742,6 +693,14 @@ static int flush_channel(uint64_t chan_key)
 next:
                pthread_mutex_unlock(&stream->lock);
        }
+
+       /*
+        * Send one last buffer statistics update to the session daemon. This
+        * ensures that the session daemon gets at least one statistics update
+        * per channel even in the case of short-lived channels, such as when a
+        * short-lived app is traced in per-pid mode.
+        */
+       sample_and_send_channel_buffer_stats(channel);
 error:
        rcu_read_unlock();
        return ret;
@@ -1514,23 +1473,17 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
                };
 
                /* Create a plain object and reserve a channel key. */
-               channel = consumer_allocate_channel(
-                               msg.u.ask_channel.key,
+               channel = consumer_allocate_channel(msg.u.ask_channel.key,
                                msg.u.ask_channel.session_id,
-                               msg.u.ask_channel.chunk_id.is_set ?
-                                               &chunk_id : NULL,
-                               msg.u.ask_channel.pathname,
-                               msg.u.ask_channel.name,
+                               msg.u.ask_channel.chunk_id.is_set ? &chunk_id : NULL,
+                               msg.u.ask_channel.pathname, msg.u.ask_channel.name,
                                msg.u.ask_channel.relayd_id,
                                (enum lttng_event_output) msg.u.ask_channel.output,
-                               msg.u.ask_channel.tracefile_size,
-                               msg.u.ask_channel.tracefile_count,
-                               msg.u.ask_channel.session_id_per_pid,
-                               msg.u.ask_channel.monitor,
-                               msg.u.ask_channel.live_timer_interval,
-                               msg.u.ask_channel.is_live,
-                               msg.u.ask_channel.root_shm_path,
-                               msg.u.ask_channel.shm_path);
+                               msg.u.ask_channel.tracefile_size, msg.u.ask_channel.tracefile_count,
+                               msg.u.ask_channel.session_id_per_pid, msg.u.ask_channel.monitor,
+                               msg.u.ask_channel.live_timer_interval, msg.u.ask_channel.is_live,
+                               msg.u.ask_channel.root_shm_path, msg.u.ask_channel.shm_path,
+                               msg.u.ask_channel.trace_format);
                if (!channel) {
                        goto end_channel_error;
                }
@@ -2133,9 +2086,11 @@ end_rotate_channel_nosignal:
        case LTTNG_CONSUMER_INIT:
        {
                int ret_send_status;
+               lttng_uuid sessiond_uuid;
 
-               ret_code = lttng_consumer_init_command(ctx,
-                               msg.u.init.sessiond_uuid);
+               std::copy(std::begin(msg.u.init.sessiond_uuid), std::end(msg.u.init.sessiond_uuid),
+                               sessiond_uuid.begin());
+               ret_code = lttng_consumer_init_command(ctx, sessiond_uuid);
                health_code_update();
                ret_send_status = consumer_send_status_msg(sock, ret_code);
                if (ret_send_status < 0) {
@@ -2315,11 +2270,7 @@ end_msg_sessiond:
 
 end_channel_error:
        if (channel) {
-               /*
-                * Free channel here since no one has a reference to it. We don't
-                * free after that because a stream can store this pointer.
-                */
-               destroy_channel(channel);
+               consumer_del_channel(channel);
        }
        /* We have to send a status channel message indicating an error. */
        {
@@ -2465,8 +2416,9 @@ void lttng_ustconsumer_on_stream_hangup(struct lttng_consumer_stream *stream)
                        stream->quiescent = true;
                }
        }
-       pthread_mutex_unlock(&stream->lock);
+
        stream->hangup_flush_done = 1;
+       pthread_mutex_unlock(&stream->lock);
 }
 
 void lttng_ustconsumer_del_channel(struct lttng_consumer_channel *chan)