X-Git-Url: http://git.lttng.org./?a=blobdiff_plain;f=src%2Fcommon%2Fust-consumer%2Fust-consumer.c;h=10b1481c0e6ac7b1ce5fdad560f1d5bb4443709f;hb=a4eb26f0f09cd6d031166329a7d5d7b5d40408a6;hp=88520dbe260fbfd14a27d1431b3bc0df616e86e7;hpb=7f1b22f2ae8abb4e0c327d5c0e561845634022ff;p=lttng-tools.git diff --git a/src/common/ust-consumer/ust-consumer.c b/src/common/ust-consumer/ust-consumer.c index 88520dbe2..10b1481c0 100644 --- a/src/common/ust-consumer/ust-consumer.c +++ b/src/common/ust-consumer/ust-consumer.c @@ -1,5 +1,5 @@ /* - * Copyright (C) 2011 Julien Desfossez + * Copyright (C) 2011 EfficiOS Inc. * Copyright (C) 2011 Mathieu Desnoyers * Copyright (C) 2017 Jérémie Galarneau * @@ -1031,7 +1031,7 @@ static int snapshot_metadata(struct lttng_consumer_channel *metadata_channel, metadata_stream = metadata_channel->metadata_stream; assert(metadata_stream); - pthread_mutex_lock(&metadata_stream->lock); + metadata_stream->read_subbuffer_ops.lock(metadata_stream); if (relayd_id != (uint64_t) -1ULL) { metadata_stream->net_seq_idx = relayd_id; ret = consumer_send_relayd_stream(metadata_stream, path); @@ -1039,14 +1039,12 @@ static int snapshot_metadata(struct lttng_consumer_channel *metadata_channel, ret = consumer_stream_create_output_files(metadata_stream, false); } - pthread_mutex_unlock(&metadata_stream->lock); if (ret < 0) { goto error_stream; } do { health_code_update(); - ret = lttng_consumer_read_subbuffer(metadata_stream, ctx, true); if (ret < 0) { goto error_stream; @@ -1054,9 +1052,10 @@ static int snapshot_metadata(struct lttng_consumer_channel *metadata_channel, } while (ret > 0); error_stream: + metadata_stream->read_subbuffer_ops.unlock(metadata_stream); /* - * Clean up the stream completly because the next snapshot will use a new - * metadata stream. + * Clean up the stream completely because the next snapshot will use a + * new metadata stream. */ consumer_stream_destroy(metadata_stream, NULL); cds_list_del(&metadata_stream->send_node); @@ -1278,6 +1277,17 @@ error_unlock: return ret; } +static +void metadata_stream_reset_cache_consumed_position( + struct lttng_consumer_stream *stream) +{ + ASSERT_LOCKED(stream->lock); + + DBG("Reset metadata cache of session %" PRIu64, + stream->chan->session_id); + stream->ust_metadata_pushed = 0; +} + /* * Receive the metadata updates from the sessiond. Supports receiving * overlapping metadata, but is needs to always belong to a contiguous @@ -1292,6 +1302,7 @@ int lttng_ustconsumer_recv_metadata(int sock, uint64_t key, uint64_t offset, { int ret, ret_code = LTTCOMM_CONSUMERD_SUCCESS; char *metadata_str; + enum consumer_metadata_cache_write_status cache_write_status; DBG("UST consumer push metadata key %" PRIu64 " of len %" PRIu64, key, len); @@ -1315,10 +1326,48 @@ int lttng_ustconsumer_recv_metadata(int sock, uint64_t key, uint64_t offset, health_code_update(); pthread_mutex_lock(&channel->metadata_cache->lock); - ret = consumer_metadata_cache_write(channel, offset, len, version, - metadata_str); + cache_write_status = consumer_metadata_cache_write( + channel, offset, len, version, metadata_str); pthread_mutex_unlock(&channel->metadata_cache->lock); - if (ret < 0) { + switch (cache_write_status) { + case CONSUMER_METADATA_CACHE_WRITE_STATUS_NO_CHANGE: + /* + * The write entirely overlapped with existing contents of the + * same metadata version (same content); there is nothing to do. + */ + break; + case CONSUMER_METADATA_CACHE_WRITE_STATUS_INVALIDATED: + /* + * The metadata cache was invalidated (previously pushed + * content has been overwritten). Reset the stream's consumed + * metadata position to ensure the metadata poll thread consumes + * the whole cache. + */ + + /* + * channel::metadata_stream can be null when the metadata + * channel is under a snapshot session type. No need to update + * the stream position in that scenario. + */ + if (channel->metadata_stream != NULL) { + pthread_mutex_lock(&channel->metadata_stream->lock); + metadata_stream_reset_cache_consumed_position( + channel->metadata_stream); + pthread_mutex_unlock(&channel->metadata_stream->lock); + } + /* Fall-through. */ + case CONSUMER_METADATA_CACHE_WRITE_STATUS_APPENDED_CONTENT: + /* + * In both cases, the metadata poll thread has new data to + * consume. + */ + ret = consumer_metadata_wakeup_pipe(channel); + if (ret) { + ret_code = LTTCOMM_CONSUMERD_ERROR_METADATA; + goto end_free; + } + break; + case CONSUMER_METADATA_CACHE_WRITE_STATUS_ERROR: /* Unable to handle metadata. Notify session daemon. */ ret_code = LTTCOMM_CONSUMERD_ERROR_METADATA; /* @@ -1327,6 +1376,8 @@ int lttng_ustconsumer_recv_metadata(int sock, uint64_t key, uint64_t offset, * waiting for the metadata cache to be flushed. */ goto end_free; + default: + abort(); } if (!wait) { @@ -2142,7 +2193,7 @@ end_rotate_channel_nosignal: const uint64_t relayd_id = msg.u.close_trace_chunk.relayd_id.value; struct lttcomm_consumer_close_trace_chunk_reply reply; - char closed_trace_chunk_path[LTTNG_PATH_MAX]; + char closed_trace_chunk_path[LTTNG_PATH_MAX] = {}; int ret; ret_code = lttng_consumer_close_trace_chunk( @@ -2455,15 +2506,6 @@ int lttng_ustconsumer_close_wakeup_fd(struct lttng_consumer_stream *stream) return ustctl_stream_close_wakeup_fd(stream->ustream); } -static -void metadata_stream_reset_cache_consumed_position( - struct lttng_consumer_stream *stream) -{ - DBG("Reset metadata cache of session %" PRIu64, - stream->chan->session_id); - stream->ust_metadata_pushed = 0; -} - /* * Write up to one packet from the metadata cache to the channel. * @@ -3041,6 +3083,7 @@ int lttng_ustconsumer_data_pending(struct lttng_consumer_stream *stream) assert(stream); assert(stream->ustream); + ASSERT_LOCKED(stream->lock); DBG("UST consumer checking data pending");