Fix: get consumer lock before closing/pushing metadata

This is to ensure the validity of the metadata stream in the channel
before closing or pushing the metadata on it.

Fixes #536

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

diff --git a/src/common/consumer-metadata-cache.h b/src/common/consumer-metadata-cache.h
index 164f9eaae0689d68262e007278bbfc82a3a7d6ac..b1a4dc2b777fad31b91a8ecf040b439f180776c3 100644 (file)
--- a/src/common/consumer-metadata-cache.h
+++ b/src/common/consumer-metadata-cache.h
@@ -43,6 +43,8 @@ struct consumer_metadata_cache {
        /*
         * Lock to update the metadata cache and push into the ring_buffer
         * (ustctl_write_metadata_to_channel).
        /*
         * Lock to update the metadata cache and push into the ring_buffer
         * (ustctl_write_metadata_to_channel).

+        *
+        * This is nested INSIDE the consumer_data lock.

         */
        pthread_mutex_t lock;
 };
         */
        pthread_mutex_t lock;
 };

diff --git a/src/common/consumer.c b/src/common/consumer.c
index b5fe9831640217455c40c11b9e85e36b470ec05c..78b3f0799737139fc9067a601a39452e384b3e0e 100644 (file)
--- a/src/common/consumer.c
+++ b/src/common/consumer.c
@@ -1960,6 +1960,13 @@ void consumer_del_metadata_stream(struct lttng_consumer_stream *stream,
        }
 
 end:
        }
 
 end:

+       /*
+        * Nullify the stream reference so it is not used after deletion. The
+        * consumer data lock MUST be acquired before being able to check for a
+        * NULL pointer value.
+        */
+       stream->chan->metadata_stream = NULL;
+

        pthread_mutex_unlock(&stream->lock);
        pthread_mutex_unlock(&consumer_data.lock);
 
        pthread_mutex_unlock(&stream->lock);
        pthread_mutex_unlock(&consumer_data.lock);
 

diff --git a/src/common/consumer.h b/src/common/consumer.h
index 3726fd1e6a2b19b3ca35623243524fd8e608fda9..bf87609f3f1aea428128895cdab7ba20cf1f8568 100644 (file)
--- a/src/common/consumer.h
+++ b/src/common/consumer.h
@@ -222,6 +222,7 @@ struct lttng_consumer_stream {
         * Lock to use the stream FDs since they are used between threads.
         *
         * This is nested INSIDE the consumer_data lock.
         * Lock to use the stream FDs since they are used between threads.
         *
         * This is nested INSIDE the consumer_data lock.

+        * This is nested INSIDE the metadata cache lock.

         * This is nested OUTSIDE consumer_relayd_sock_pair lock.
         */
        pthread_mutex_t lock;
         * This is nested OUTSIDE consumer_relayd_sock_pair lock.
         */
        pthread_mutex_t lock;

diff --git a/src/common/ust-consumer/ust-consumer.c b/src/common/ust-consumer/ust-consumer.c
index e0280f1489e2e8defa25e1311023f4f518e4d666..65220729766d4ad56c6a5c861e9b8d8f6af84927 100644 (file)
--- a/src/common/ust-consumer/ust-consumer.c
+++ b/src/common/ust-consumer/ust-consumer.c
@@ -558,6 +558,11 @@ int lttng_ustconsumer_push_metadata(struct lttng_consumer_channel *metadata,
 
        DBG("UST consumer writing metadata to channel %s", metadata->name);
 
 
        DBG("UST consumer writing metadata to channel %s", metadata->name);
 

+       if (!metadata->metadata_stream) {
+               ret = 0;
+               goto error;
+       }
+

        assert(target_offset <= metadata->metadata_cache->max_offset);
        ret = ustctl_write_metadata_to_channel(metadata->uchan,
                        metadata_str + target_offset, len);
        assert(target_offset <= metadata->metadata_cache->max_offset);
        ret = ustctl_write_metadata_to_channel(metadata->uchan,
                        metadata_str + target_offset, len);


@@ -629,11 +634,17 @@ static int close_metadata(uint64_t chan_key)
        }
 
        pthread_mutex_lock(&consumer_data.lock);
        }
 
        pthread_mutex_lock(&consumer_data.lock);

-       if (!cds_lfht_is_node_deleted(&channel->node.node)) {
-               if (channel->switch_timer_enabled == 1) {
-                       DBG("Deleting timer on metadata channel");
-                       consumer_timer_switch_stop(channel);
-               }
+
+       if (cds_lfht_is_node_deleted(&channel->node.node)) {
+               goto error_unlock;
+       }
+
+       if (channel->switch_timer_enabled == 1) {
+               DBG("Deleting timer on metadata channel");
+               consumer_timer_switch_stop(channel);
+       }
+
+       if (channel->metadata_stream) {

                ret = ustctl_stream_close_wakeup_fd(channel->metadata_stream->ustream);
                if (ret < 0) {
                        ERR("UST consumer unable to close fd of metadata (ret: %d)", ret);
                ret = ustctl_stream_close_wakeup_fd(channel->metadata_stream->ustream);
                if (ret < 0) {
                        ERR("UST consumer unable to close fd of metadata (ret: %d)", ret);


@@ -728,6 +739,17 @@ int lttng_ustconsumer_recv_metadata(int sock, uint64_t key, uint64_t offset,
                goto end_free;
        }
 
                goto end_free;
        }
 

+       /*
+        * XXX: The consumer data lock is acquired before calling metadata cache
+        * write which calls push metadata that MUST be protected by the consumer
+        * lock in order to be able to check the validity of the metadata stream of
+        * the channel.
+        *
+        * Note that this will be subject to change to better fine grained locking
+        * and ultimately try to get rid of this global consumer data lock.
+        */
+       pthread_mutex_lock(&consumer_data.lock);
+

        pthread_mutex_lock(&channel->metadata_cache->lock);
        ret = consumer_metadata_cache_write(channel, offset, len, metadata_str);
        if (ret < 0) {
        pthread_mutex_lock(&channel->metadata_cache->lock);
        ret = consumer_metadata_cache_write(channel, offset, len, metadata_str);
        if (ret < 0) {


@@ -735,6 +757,7 @@ int lttng_ustconsumer_recv_metadata(int sock, uint64_t key, uint64_t offset,
                ret_code = LTTCOMM_CONSUMERD_ERROR_METADATA;
        }
        pthread_mutex_unlock(&channel->metadata_cache->lock);
                ret_code = LTTCOMM_CONSUMERD_ERROR_METADATA;
        }
        pthread_mutex_unlock(&channel->metadata_cache->lock);

+       pthread_mutex_unlock(&consumer_data.lock);

 
        while (consumer_metadata_cache_flushed(channel, offset + len)) {
                DBG("Waiting for metadata to be flushed");
 
        while (consumer_metadata_cache_flushed(channel, offset + len)) {
                DBG("Waiting for metadata to be flushed");