Fix deadlock: don't take channel lock in timer

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

diff --git a/src/common/ust-consumer/ust-consumer.c b/src/common/ust-consumer/ust-consumer.c
index 739cebaefe04934573876b0ced133de8b258e338..290eb3390e80aff424d954f8d9b7a00746f62fc0 100644 (file)
--- a/src/common/ust-consumer/ust-consumer.c
+++ b/src/common/ust-consumer/ust-consumer.c
@@ -375,7 +375,7 @@ static int send_sessiond_stream(int sock, struct lttng_consumer_stream *stream)
        assert(stream);
        assert(sock >= 0);
 
-       DBG2("UST consumer sending stream %" PRIu64 " to sessiond", stream->key);
+       DBG("UST consumer sending stream %" PRIu64 " to sessiond", stream->key);
 
        /* Send stream to session daemon. */
        ret = ustctl_send_stream_to_sessiond(sock, stream->ustream);
@@ -746,7 +746,8 @@ error_find:
  * Receive the metadata updates from the sessiond.
  */
 int lttng_ustconsumer_recv_metadata(int sock, uint64_t key, uint64_t offset,
-               uint64_t len, struct lttng_consumer_channel *channel)
+               uint64_t len, struct lttng_consumer_channel *channel,
+               int timer)
 {
        int ret, ret_code = LTTNG_OK;
        char *metadata_str;
@@ -768,17 +769,10 @@ int lttng_ustconsumer_recv_metadata(int sock, uint64_t key, uint64_t offset,
                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->lock);
+       if (!timer) {
+               pthread_mutex_lock(&channel->lock);
+       }
+       pthread_mutex_lock(&channel->timer_lock);
        pthread_mutex_lock(&channel->metadata_cache->lock);
        ret = consumer_metadata_cache_write(channel, offset, len, metadata_str);
        if (ret < 0) {
@@ -790,15 +784,19 @@ int lttng_ustconsumer_recv_metadata(int sock, uint64_t key, uint64_t offset,
                 * waiting for the metadata cache to be flushed.
                 */
                pthread_mutex_unlock(&channel->metadata_cache->lock);
-               pthread_mutex_unlock(&channel->lock);
-               pthread_mutex_unlock(&consumer_data.lock);
+               pthread_mutex_unlock(&channel->timer_lock);
+               if (!timer) {
+                       pthread_mutex_unlock(&channel->lock);
+               }
                goto end_free;
        }
        pthread_mutex_unlock(&channel->metadata_cache->lock);
-       pthread_mutex_unlock(&channel->lock);
-       pthread_mutex_unlock(&consumer_data.lock);
+       pthread_mutex_unlock(&channel->timer_lock);
+       if (!timer) {
+               pthread_mutex_unlock(&channel->lock);
+       }
 
-       while (consumer_metadata_cache_flushed(channel, offset + len)) {
+       while (consumer_metadata_cache_flushed(channel, offset + len, timer)) {
                DBG("Waiting for metadata to be flushed");
                usleep(DEFAULT_METADATA_AVAILABILITY_WAIT_TIME);
        }
@@ -1134,7 +1132,7 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx,
                }
 
                ret = lttng_ustconsumer_recv_metadata(sock, key, offset,
-                               len, channel);
+                               len, channel, 0);
                if (ret < 0) {
                        /* error receiving from sessiond */
                        goto error_fatal;
@@ -1301,7 +1299,7 @@ int lttng_ustconsumer_read_subbuffer(struct lttng_consumer_stream *stream,
        assert(stream->ustream);
        assert(ctx);
 
-       DBG2("In UST read_subbuffer (wait_fd: %d, name: %s)", stream->wait_fd,
+       DBG("In UST read_subbuffer (wait_fd: %d, name: %s)", stream->wait_fd,
                        stream->name);
 
        /* Ease our life for what's next. */
@@ -1417,6 +1415,11 @@ int lttng_ustconsumer_data_pending(struct lttng_consumer_stream *stream)
 
        DBG("UST consumer checking data pending");
 
+       if (stream->endpoint_status != CONSUMER_ENDPOINT_ACTIVE) {
+               ret = 0;
+               goto end;
+       }
+
        ret = ustctl_get_next_subbuf(stream->ustream);
        if (ret == 0) {
                /* There is still data so let's put back this subbuffer. */
@@ -1481,8 +1484,14 @@ void lttng_ustconsumer_close_stream_wakeup(struct lttng_consumer_stream *stream)
        }
 }
 
+/*
+ * Please refer to consumer-timer.c before adding any lock within this
+ * function or any of its callees. Timers have a very strict locking
+ * semantic with respect to teardown. Failure to respect this semantic
+ * introduces deadlocks.
+ */
 int lttng_ustconsumer_request_metadata(struct lttng_consumer_local_data *ctx,
-               struct lttng_consumer_channel *channel)
+               struct lttng_consumer_channel *channel, int timer)
 {
        struct lttcomm_metadata_request_msg request;
        struct lttcomm_consumer_msg msg;
@@ -1569,7 +1578,7 @@ int lttng_ustconsumer_request_metadata(struct lttng_consumer_local_data *ctx,
        }
 
        ret_code = lttng_ustconsumer_recv_metadata(ctx->consumer_metadata_socket,
-                       key, offset, len, channel);
+                       key, offset, len, channel, timer);
        if (ret_code >= 0) {
                /*
                 * Only send the status msg if the sessiond is alive meaning a positive