X-Git-Url: https://git.lttng.org/?a=blobdiff_plain;f=src%2Fcommon%2Fust-consumer%2Fust-consumer.cpp;h=3f40d03ed3d846cb8fa95788c94bf555be5fb28f;hb=HEAD;hp=d0b39a887b467f1098397ba53213a69bd670067e;hpb=5c7248cd5bce45bf64d563fb4e130a63bf345f11;p=lttng-tools.git diff --git a/src/common/ust-consumer/ust-consumer.cpp b/src/common/ust-consumer/ust-consumer.cpp index d0b39a887..3f40d03ed 100644 --- a/src/common/ust-consumer/ust-consumer.cpp +++ b/src/common/ust-consumer/ust-consumer.cpp @@ -12,7 +12,6 @@ #include #include -#include #include #include #include @@ -22,12 +21,14 @@ #include #include #include +#include #include #include #include #include +#include #include #include #include @@ -666,7 +667,7 @@ static int flush_channel(uint64_t chan_key) DBG("UST consumer flush channel key %" PRIu64, chan_key); - rcu_read_lock(); + lttng::urcu::read_lock_guard read_lock; channel = consumer_find_channel(chan_key); if (!channel) { ERR("UST consumer flush channel %" PRIu64 " not found", chan_key); @@ -721,7 +722,6 @@ static int flush_channel(uint64_t chan_key) */ sample_and_send_channel_buffer_stats(channel); error: - rcu_read_unlock(); return ret; } @@ -741,7 +741,7 @@ static int clear_quiescent_channel(uint64_t chan_key) DBG("UST consumer clear quiescent channel key %" PRIu64, chan_key); - rcu_read_lock(); + lttng::urcu::read_lock_guard read_lock; channel = consumer_find_channel(chan_key); if (!channel) { ERR("UST consumer clear quiescent channel %" PRIu64 " not found", chan_key); @@ -767,7 +767,6 @@ static int clear_quiescent_channel(uint64_t chan_key) pthread_mutex_unlock(&stream->lock); } error: - rcu_read_unlock(); return ret; } @@ -931,6 +930,8 @@ error: */ consumer_stream_destroy(metadata->metadata_stream, nullptr); metadata->metadata_stream = nullptr; + metadata->metadata_pushed_wait_queue.wake_all(); + send_streams_error: error_no_stream: end: @@ -958,7 +959,7 @@ static int snapshot_metadata(struct lttng_consumer_channel *metadata_channel, DBG("UST consumer snapshot metadata with key %" PRIu64 " at path %s", key, path); - rcu_read_lock(); + lttng::urcu::read_lock_guard read_lock; LTTNG_ASSERT(!metadata_channel->monitor); @@ -968,7 +969,7 @@ static int snapshot_metadata(struct lttng_consumer_channel *metadata_channel, * Ask the sessiond if we have new metadata waiting and update the * consumer metadata cache. */ - ret = lttng_ustconsumer_request_metadata(ctx, metadata_channel, 0, 1); + ret = lttng_ustconsumer_request_metadata(ctx, metadata_channel, false, 1); if (ret < 0) { goto error; } @@ -1014,9 +1015,9 @@ error_stream: */ consumer_stream_destroy(metadata_stream, nullptr); metadata_channel->metadata_stream = nullptr; + metadata_channel->metadata_pushed_wait_queue.wake_all(); error: - rcu_read_unlock(); return ret; } @@ -1067,7 +1068,7 @@ static int snapshot_channel(struct lttng_consumer_channel *channel, LTTNG_ASSERT(ctx); ASSERT_RCU_READ_LOCKED(); - rcu_read_lock(); + lttng::urcu::read_lock_guard read_lock; if (relayd_id != (uint64_t) -1ULL) { use_relayd = 1; @@ -1218,7 +1219,6 @@ static int snapshot_channel(struct lttng_consumer_channel *channel, pthread_mutex_unlock(&stream->lock); } - rcu_read_unlock(); return 0; error_put_subbuf: @@ -1229,7 +1229,6 @@ error_close_stream: consumer_stream_close_output(stream); error_unlock: pthread_mutex_unlock(&stream->lock); - rcu_read_unlock(); return ret; } @@ -1255,7 +1254,7 @@ int lttng_ustconsumer_recv_metadata(int sock, uint64_t len, uint64_t version, struct lttng_consumer_channel *channel, - int timer, + bool invoked_by_timer, int wait) { int ret, ret_code = LTTCOMM_CONSUMERD_SUCCESS; @@ -1343,13 +1342,8 @@ int lttng_ustconsumer_recv_metadata(int sock, if (!wait) { goto end_free; } - while (consumer_metadata_cache_flushed(channel, offset + len, timer)) { - DBG("Waiting for metadata to be flushed"); - health_code_update(); - - usleep(DEFAULT_METADATA_AVAILABILITY_WAIT_TIME); - } + consumer_wait_metadata_cache_flushed(channel, offset + len, invoked_by_timer); end_free: free(metadata_str); @@ -1401,7 +1395,7 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, health_code_update(); /* relayd needs RCU read-side lock */ - rcu_read_lock(); + lttng::urcu::read_lock_guard read_lock; switch (msg.cmd_type) { case LTTNG_CONSUMER_ADD_RELAYD_SOCKET: @@ -1456,7 +1450,6 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, } case LTTNG_CONSUMER_UPDATE_STREAM: { - rcu_read_unlock(); return -ENOSYS; } case LTTNG_CONSUMER_DATA_PENDING: @@ -1795,7 +1788,7 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, health_code_update(); ret = lttng_ustconsumer_recv_metadata( - sock, key, offset, len, version, found_channel, 0, 1); + sock, key, offset, len, version, found_channel, false, 1); if (ret < 0) { /* error receiving from sessiond */ goto error_push_metadata_fatal; @@ -1877,7 +1870,6 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, uint64_t key = msg.u.discarded_events.channel_key; DBG("UST consumer discarded events command for session id %" PRIu64, id); - rcu_read_lock(); pthread_mutex_lock(&the_consumer_data.lock); ht = the_consumer_data.stream_list_ht; @@ -1904,7 +1896,6 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, } } pthread_mutex_unlock(&the_consumer_data.lock); - rcu_read_unlock(); DBG("UST consumer discarded events command for session id %" PRIu64 ", channel key %" PRIu64, @@ -1933,7 +1924,6 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, uint64_t key = msg.u.lost_packets.channel_key; DBG("UST consumer lost packets command for session id %" PRIu64, id); - rcu_read_lock(); pthread_mutex_lock(&the_consumer_data.lock); ht = the_consumer_data.stream_list_ht; @@ -1959,7 +1949,6 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, } } pthread_mutex_unlock(&the_consumer_data.lock); - rcu_read_unlock(); DBG("UST consumer lost packets command for session id %" PRIu64 ", channel key %" PRIu64, @@ -2296,7 +2285,6 @@ error_fatal: goto end; end: - rcu_read_unlock(); health_code_update(); return ret_func; } @@ -2560,7 +2548,9 @@ static int commit_one_metadata_packet(struct lttng_consumer_stream *stream) ret = write_len; goto end; } + stream->ust_metadata_pushed += write_len; + stream->chan->metadata_pushed_wait_queue.wake_all(); LTTNG_ASSERT(stream->chan->metadata_cache->contents.size >= stream->ust_metadata_pushed); ret = write_len; @@ -2609,7 +2599,7 @@ lttng_ustconsumer_sync_metadata(struct lttng_consumer_local_data *ctx, * Request metadata from the sessiond, but don't wait for the flush * because we locked the metadata thread. */ - ret = lttng_ustconsumer_request_metadata(ctx, metadata_channel, 0, 0); + ret = lttng_ustconsumer_request_metadata(ctx, metadata_channel, false, 0); pthread_mutex_lock(&metadata_stream->lock); if (ret < 0) { status = SYNC_METADATA_STATUS_ERROR; @@ -3213,15 +3203,17 @@ void lttng_ustconsumer_close_all_metadata(struct lttng_ht *metadata_ht) DBG("UST consumer closing all metadata streams"); - rcu_read_lock(); - cds_lfht_for_each_entry (metadata_ht->ht, &iter.iter, stream, node.node) { - health_code_update(); + { + lttng::urcu::read_lock_guard read_lock; - pthread_mutex_lock(&stream->chan->lock); - lttng_ustconsumer_close_metadata(stream->chan); - pthread_mutex_unlock(&stream->chan->lock); + cds_lfht_for_each_entry (metadata_ht->ht, &iter.iter, stream, node.node) { + health_code_update(); + + pthread_mutex_lock(&stream->chan->lock); + lttng_ustconsumer_close_metadata(stream->chan); + pthread_mutex_unlock(&stream->chan->lock); + } } - rcu_read_unlock(); } void lttng_ustconsumer_close_stream_wakeup(struct lttng_consumer_stream *stream) @@ -3246,7 +3238,7 @@ void lttng_ustconsumer_close_stream_wakeup(struct lttng_consumer_stream *stream) */ int lttng_ustconsumer_request_metadata(struct lttng_consumer_local_data *ctx, struct lttng_consumer_channel *channel, - int timer, + bool invoked_by_timer, int wait) { struct lttcomm_metadata_request_msg request; @@ -3351,8 +3343,14 @@ int lttng_ustconsumer_request_metadata(struct lttng_consumer_local_data *ctx, health_code_update(); - ret = lttng_ustconsumer_recv_metadata( - ctx->consumer_metadata_socket, key, offset, len, version, channel, timer, wait); + ret = lttng_ustconsumer_recv_metadata(ctx->consumer_metadata_socket, + key, + offset, + len, + version, + channel, + invoked_by_timer, + wait); if (ret >= 0) { /* * Only send the status msg if the sessiond is alive meaning a positive