relayd protocol: reply path for close chunk and create session 2.11
[lttng-tools.git] / src / common / consumer / consumer.c
index 8483146c65ab6ecf3d1b026d1f33c62be4442c8a..716b560a96ddbf9cd397147fe3ed01eb076781ad 100644 (file)
@@ -411,7 +411,8 @@ void consumer_del_channel(struct lttng_consumer_channel *channel)
                rcu_read_unlock();
        }
 
-        call_rcu(&channel->node.head, free_channel_rcu);
+       channel->is_deleted = true;
+       call_rcu(&channel->node.head, free_channel_rcu);
 end:
        pthread_mutex_unlock(&channel->lock);
        pthread_mutex_unlock(&consumer_data.lock);
@@ -1021,6 +1022,16 @@ int lttng_consumer_channel_set_trace_chunk(
        unsigned long channel_hash;
 
        pthread_mutex_lock(&channel->lock);
+       if (channel->is_deleted) {
+               /*
+                * The channel has been logically deleted and should no longer
+                * be used. It has released its reference to its current trace
+                * chunk and should not acquire a new one.
+                *
+                * Return success as there is nothing for the caller to do.
+                */
+               goto end;
+       }
        /*
         * A stream can transition to a state where it and its channel
         * no longer belong to a trace chunk. For instance, this happens when
@@ -1397,6 +1408,7 @@ void lttng_consumer_cleanup(void)
 {
        struct lttng_ht_iter iter;
        struct lttng_consumer_channel *channel;
+       unsigned int trace_chunks_left;
 
        rcu_read_lock();
 
@@ -1421,6 +1433,27 @@ void lttng_consumer_cleanup(void)
         */
        lttng_ht_destroy(consumer_data.stream_list_ht);
 
+       /*
+        * Trace chunks in the registry may still exist if the session
+        * daemon has encountered an internal error and could not
+        * tear down its sessions and/or trace chunks properly.
+        *
+        * Release the session daemon's implicit reference to any remaining
+        * trace chunk and print an error if any trace chunk was found. Note
+        * that there are _no_ legitimate cases for trace chunks to be left,
+        * it is a leak. However, it can happen following a crash of the
+        * session daemon and not emptying the registry would cause an assertion
+        * to hit.
+        */
+       trace_chunks_left = lttng_trace_chunk_registry_put_each_chunk(
+                       consumer_data.chunk_registry);
+       if (trace_chunks_left) {
+               ERR("%u trace chunks are leaked by lttng-consumerd. "
+                               "This can be caused by an internal error of the session daemon.",
+                               trace_chunks_left);
+       }
+       /* Run all callbacks freeing each chunk. */
+       rcu_barrier();
        lttng_trace_chunk_registry_destroy(consumer_data.chunk_registry);
 }
 
@@ -2253,7 +2286,8 @@ void lttng_consumer_close_all_metadata(void)
 void consumer_del_metadata_stream(struct lttng_consumer_stream *stream,
                struct lttng_ht *ht)
 {
-       struct lttng_consumer_channel *free_chan = NULL;
+       struct lttng_consumer_channel *channel = NULL;
+       bool free_channel = false;
 
        assert(stream);
        /*
@@ -2265,11 +2299,17 @@ void consumer_del_metadata_stream(struct lttng_consumer_stream *stream,
        DBG3("Consumer delete metadata stream %d", stream->wait_fd);
 
        pthread_mutex_lock(&consumer_data.lock);
-       pthread_mutex_lock(&stream->chan->lock);
+       /*
+        * Note that this assumes that a stream's channel is never changed and
+        * that the stream's lock doesn't need to be taken to sample its
+        * channel.
+        */
+       channel = stream->chan;
+       pthread_mutex_lock(&channel->lock);
        pthread_mutex_lock(&stream->lock);
-       if (stream->chan->metadata_cache) {
+       if (channel->metadata_cache) {
                /* Only applicable to userspace consumers. */
-               pthread_mutex_lock(&stream->chan->metadata_cache->lock);
+               pthread_mutex_lock(&channel->metadata_cache->lock);
        }
 
        /* Remove any reference to that stream. */
@@ -2281,28 +2321,29 @@ void consumer_del_metadata_stream(struct lttng_consumer_stream *stream,
        consumer_stream_destroy_buffers(stream);
 
        /* Atomically decrement channel refcount since other threads can use it. */
-       if (!uatomic_sub_return(&stream->chan->refcount, 1)
-                       && !uatomic_read(&stream->chan->nb_init_stream_left)) {
+       if (!uatomic_sub_return(&channel->refcount, 1)
+                       && !uatomic_read(&channel->nb_init_stream_left)) {
                /* Go for channel deletion! */
-               free_chan = stream->chan;
+               free_channel = true;
        }
+       stream->chan = NULL;
 
        /*
         * Nullify the stream reference so it is not used after deletion. The
         * channel lock MUST be acquired before being able to check for a NULL
         * pointer value.
         */
-       stream->chan->metadata_stream = NULL;
+       channel->metadata_stream = NULL;
 
-       if (stream->chan->metadata_cache) {
-               pthread_mutex_unlock(&stream->chan->metadata_cache->lock);
+       if (channel->metadata_cache) {
+               pthread_mutex_unlock(&channel->metadata_cache->lock);
        }
        pthread_mutex_unlock(&stream->lock);
-       pthread_mutex_unlock(&stream->chan->lock);
+       pthread_mutex_unlock(&channel->lock);
        pthread_mutex_unlock(&consumer_data.lock);
 
-       if (free_chan) {
-               consumer_del_channel(free_chan);
+       if (free_channel) {
+               consumer_del_channel(channel);
        }
 
        lttng_trace_chunk_put(stream->trace_chunk);
@@ -4539,11 +4580,13 @@ enum lttcomm_return_code lttng_consumer_create_trace_chunk(
                         * channels.
                         */
                        enum lttcomm_return_code close_ret;
+                       char path[LTTNG_PATH_MAX];
 
                        DBG("Failed to set new trace chunk on existing channels, rolling back");
                        close_ret = lttng_consumer_close_trace_chunk(relayd_id,
                                        session_id, chunk_id,
-                                       chunk_creation_timestamp, NULL);
+                                       chunk_creation_timestamp, NULL,
+                                       path);
                        if (close_ret != LTTCOMM_CONSUMERD_SUCCESS) {
                                ERR("Failed to roll-back the creation of new chunk: session_id = %" PRIu64 ", chunk_id = %" PRIu64,
                                                session_id, chunk_id);
@@ -4569,12 +4612,13 @@ enum lttcomm_return_code lttng_consumer_create_trace_chunk(
 
                if (!relayd || ret) {
                        enum lttcomm_return_code close_ret;
+                       char path[LTTNG_PATH_MAX];
 
                        close_ret = lttng_consumer_close_trace_chunk(relayd_id,
                                        session_id,
                                        chunk_id,
                                        chunk_creation_timestamp,
-                                       NULL);
+                                       NULL, path);
                        if (close_ret != LTTCOMM_CONSUMERD_SUCCESS) {
                                ERR("Failed to roll-back the creation of new chunk: session_id = %" PRIu64 ", chunk_id = %" PRIu64,
                                                session_id,
@@ -4596,7 +4640,8 @@ end:
 enum lttcomm_return_code lttng_consumer_close_trace_chunk(
                const uint64_t *relayd_id, uint64_t session_id,
                uint64_t chunk_id, time_t chunk_close_timestamp,
-               const enum lttng_trace_chunk_command_type *close_command)
+               const enum lttng_trace_chunk_command_type *close_command,
+               char *path)
 {
        enum lttcomm_return_code ret_code = LTTCOMM_CONSUMERD_SUCCESS;
        struct lttng_trace_chunk *chunk;
@@ -4694,7 +4739,8 @@ enum lttcomm_return_code lttng_consumer_close_trace_chunk(
                if (relayd) {
                        pthread_mutex_lock(&relayd->ctrl_sock_mutex);
                        ret = relayd_close_trace_chunk(
-                                       &relayd->control_sock, chunk);
+                                       &relayd->control_sock, chunk,
+                                       path);
                        pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
                } else {
                        ERR("Failed to find relay daemon socket: relayd_id = %" PRIu64,
This page took 0.026264 seconds and 4 git commands to generate.