Fix: channel management thread should hold a refcount

In order to handle teardown caused by relayd being unresponsive, we need
to ensure that we correctly handle that the streams can be cleaned up
before the channel wakeup pipe gets closed. We achieve this by making
the channel management thread hold a refcount on the channel object, so
it only gets destroyed when all streams, _and_ the channel management
thread, have released their reference.

Signed-off-by: Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
Signed-off-by: David Goulet <dgoulet@efficios.com>

diff --git a/src/common/consumer.c b/src/common/consumer.c
index cc6c978686ffaf18697c7a54a84ff581f7bf579b..34385579741f6f0273aee213576bd0a188a95b0e 100644 (file)
--- a/src/common/consumer.c
+++ b/src/common/consumer.c
@@ -516,8 +516,7 @@ void consumer_del_stream(struct lttng_consumer_stream *stream,
        }
        rcu_read_unlock();
 
-       uatomic_dec(&stream->chan->refcount);
-       if (!uatomic_read(&stream->chan->refcount)
+       if (!uatomic_sub_return(&stream->chan->refcount, 1)
                        && !uatomic_read(&stream->chan->nb_init_stream_left)) {
                free_chan = stream->chan;
        }
@@ -659,6 +658,8 @@ static int add_stream(struct lttng_consumer_stream *stream,
         * stream.
         */
        if (uatomic_read(&stream->chan->nb_init_stream_left) > 0) {
+               /* Increment refcount before decrementing nb_init_stream_left */
+               cmm_smp_wmb();
                uatomic_dec(&stream->chan->nb_init_stream_left);
        }
 
@@ -1937,8 +1938,7 @@ void consumer_del_metadata_stream(struct lttng_consumer_stream *stream,
        rcu_read_unlock();
 
        /* Atomically decrement channel refcount since other threads can use it. */
-       uatomic_dec(&stream->chan->refcount);
-       if (!uatomic_read(&stream->chan->refcount)
+       if (!uatomic_sub_return(&stream->chan->refcount, 1)
                        && !uatomic_read(&stream->chan->nb_init_stream_left)) {
                /* Go for channel deletion! */
                free_chan = stream->chan;
@@ -2008,6 +2008,8 @@ static int add_metadata_stream(struct lttng_consumer_stream *stream,
         * stream.
         */
        if (uatomic_read(&stream->chan->nb_init_stream_left) > 0) {
+               /* Increment refcount before decrementing nb_init_stream_left */
+               cmm_smp_wmb();
                uatomic_dec(&stream->chan->nb_init_stream_left);
        }
 
@@ -2557,6 +2559,13 @@ void consumer_close_channel_streams(struct lttng_consumer_channel *channel)
                        ht->hash_fct(&channel->key, lttng_ht_seed),
                        ht->match_fct, &channel->key,
                        &iter.iter, stream, node_channel_id.node) {
+               /*
+                * Protect against teardown with mutex.
+                */
+               pthread_mutex_lock(&stream->lock);
+               if (cds_lfht_is_node_deleted(&stream->node.node)) {
+                       goto next;
+               }
                switch (consumer_data.type) {
                case LTTNG_CONSUMER_KERNEL:
                        break;
@@ -2573,6 +2582,8 @@ void consumer_close_channel_streams(struct lttng_consumer_channel *channel)
                        ERR("Unknown consumer_data type");
                        assert(0);
                }
+       next:
+               pthread_mutex_unlock(&stream->lock);
        }
        rcu_read_unlock();
 }
@@ -2737,6 +2748,12 @@ restart:
                                ret = lttng_ht_del(channel_ht, &iter);
                                assert(ret == 0);
                                consumer_close_channel_streams(chan);
+
+                               /* Release our own refcount */
+                               if (!uatomic_sub_return(&chan->refcount, 1)
+                                               && !uatomic_read(&chan->nb_init_stream_left)) {
+                                       consumer_del_channel(chan);
+                               }
                        }
 
                        /* Release RCU lock for the channel looked up */

diff --git a/src/common/ust-consumer/ust-consumer.c b/src/common/ust-consumer/ust-consumer.c
index 89b946896fc2858844be1f63c77bbd55c9e7642a..2bbc3f5e4fab1ba7b89fd0820ddf93080229b15d 100644 (file)
--- a/src/common/ust-consumer/ust-consumer.c
+++ b/src/common/ust-consumer/ust-consumer.c
@@ -859,6 +859,13 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
                        goto end_channel_error;
                }
 
+               /*
+                * Set refcount to 1 for owner. Below, we will pass
+                * ownership to the consumer_thread_channel_poll()
+                * thread.
+                */
+               channel->refcount = 1;
+
                /* Build channel attributes from received message. */
                attr.subbuf_size = msg.u.ask_channel.subbuf_size;
                attr.num_subbuf = msg.u.ask_channel.num_subbuf;