Fix: mark consumer channels as logically deleted during deletion

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

diff --git a/src/common/consumer/consumer.c b/src/common/consumer/consumer.c
index b2f4c2686d2f7f81b5bc21c4249a681e92a5bad3..82fe0168f2c85b11cac63390ddd427a763dc4149 100644 (file)
--- a/src/common/consumer/consumer.c
+++ b/src/common/consumer/consumer.c
@@ -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
@@ -1727,9 +1738,8 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap(
 
        /* RCU lock for the relayd pointer */
        rcu_read_lock();
-
        assert(stream->net_seq_idx != (uint64_t) -1ULL ||
-                       stream->chan->trace_chunk);
+                       stream->trace_chunk);
 
        /* Flag that the current stream if set for network streaming. */
        if (stream->net_seq_idx != (uint64_t) -1ULL) {
@@ -4073,6 +4083,10 @@ int lttng_consumer_rotate_channel(struct lttng_consumer_channel *channel,
                        stream->rotate_ready = true;
                }
 
+               /*
+                * Active flush; has no effect if the production position
+                * is at a packet boundary.
+                */
                ret = consumer_flush_buffer(stream, 1);
                if (ret < 0) {
                        ERR("Failed to flush stream %" PRIu64 " during channel rotation",
@@ -4081,10 +4095,34 @@ int lttng_consumer_rotate_channel(struct lttng_consumer_channel *channel,
                }
 
                if (!is_local_trace) {
+                       /*
+                        * The relay daemon control protocol expects a rotation
+                        * position as "the sequence number of the first packet
+                        * _after_ the current trace chunk.
+                        *
+                        * At the moment when the positions of the buffers are
+                        * sampled, the production position does not necessarily
+                        * sit at a packet boundary. The 'active' flush
+                        * operation above will push the production position to
+                        * the next packet boundary _if_ it is not already
+                        * sitting at such a boundary.
+                        *
+                        * Assuming a current production position that is not
+                        * on the bound of a packet, the 'target' sequence
+                        * number is
+                        *   (consumed_pos / subbuffer_size) + 1
+                        * Note the '+ 1' to ensure the current packet is
+                        * part of the current trace chunk.
+                        *
+                        * However, if the production position is already at
+                        * a packet boundary, the '+ 1' is not necessary as the
+                        * last packet of the current chunk is already
+                        * 'complete'.
+                        */
                        const struct relayd_stream_rotation_position position = {
                                .stream_id = stream->relayd_stream_id,
-                               .rotate_at_seq_num = (stream->rotate_position /
-                                               stream->max_sb_size) + 1,
+                               .rotate_at_seq_num = (stream->rotate_position / stream->max_sb_size) +
+                                       !!(stream->rotate_position % stream->max_sb_size),
                        };
 
                        ret = lttng_dynamic_array_add_element(