X-Git-Url: https://git.liburcu.org/?a=blobdiff_plain;f=src%2Fcommon%2Fust-consumer%2Fust-consumer.cpp;h=aeafae642b554e1aafc825705b8be25bf1437ebf;hb=f46376a14da2eb796690cb4e718e8b213839d6ea;hp=a2bf099b236f94772c9fd1d9fbb0fa3831639848;hpb=edb555b5791df39308cb5d9b1fc9883aab21951e;p=lttng-tools.git diff --git a/src/common/ust-consumer/ust-consumer.cpp b/src/common/ust-consumer/ust-consumer.cpp index a2bf099b2..aeafae642 100644 --- a/src/common/ust-consumer/ust-consumer.cpp +++ b/src/common/ust-consumer/ust-consumer.cpp @@ -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 * @@ -67,7 +67,7 @@ static void destroy_channel(struct lttng_consumer_channel *channel) health_code_update(); - cds_list_del(&stream->send_node); + cds_list_del_init(&stream->send_node); lttng_ust_ctl_destroy_stream(stream->ustream); lttng_trace_chunk_put(stream->trace_chunk); free(stream); @@ -203,7 +203,7 @@ static int send_stream_to_thread(struct lttng_consumer_stream *stream, * global. */ stream->globally_visible = 1; - cds_list_del(&stream->send_node); + cds_list_del_init(&stream->send_node); ret = lttng_pipe_write(stream_pipe, &stream, sizeof(stream)); if (ret < 0) { @@ -876,6 +876,8 @@ static int setup_metadata(struct lttng_consumer_local_data *ctx, uint64_t key) int ret; struct lttng_consumer_channel *metadata; + ASSERT_RCU_READ_LOCKED(); + DBG("UST consumer setup metadata key %" PRIu64, key); metadata = consumer_find_channel(key); @@ -948,7 +950,6 @@ error: * will make sure to clean that list. */ consumer_stream_destroy(metadata->metadata_stream, NULL); - cds_list_del(&metadata->metadata_stream->send_node); metadata->metadata_stream = NULL; send_streams_error: error_no_stream: @@ -971,6 +972,7 @@ static int snapshot_metadata(struct lttng_consumer_channel *metadata_channel, LTTNG_ASSERT(path); LTTNG_ASSERT(ctx); + ASSERT_RCU_READ_LOCKED(); DBG("UST consumer snapshot metadata with key %" PRIu64 " at path %s", key, path); @@ -1004,7 +1006,7 @@ static int snapshot_metadata(struct lttng_consumer_channel *metadata_channel, metadata_stream = metadata_channel->metadata_stream; LTTNG_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); @@ -1012,14 +1014,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; @@ -1027,12 +1027,12 @@ 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); metadata_channel->metadata_stream = NULL; error: @@ -1087,6 +1087,7 @@ static int snapshot_channel(struct lttng_consumer_channel *channel, LTTNG_ASSERT(path); LTTNG_ASSERT(ctx); + ASSERT_RCU_READ_LOCKED(); rcu_read_lock(); @@ -1323,10 +1324,21 @@ int lttng_ustconsumer_recv_metadata(int sock, uint64_t key, uint64_t offset, * metadata position to ensure the metadata poll thread consumes * the whole cache. */ - pthread_mutex_lock(&channel->metadata_stream->lock); - metadata_stream_reset_cache_consumed_position( - channel->metadata_stream); - pthread_mutex_unlock(&channel->metadata_stream->lock); + + /* + * 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); + } else { + /* Validate we are in snapshot mode. */ + LTTNG_ASSERT(!channel->monitor); + } /* Fall-through. */ case CONSUMER_METADATA_CACHE_WRITE_STATUS_APPENDED_CONTENT: /* @@ -1417,11 +1429,18 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, switch (msg.cmd_type) { case LTTNG_CONSUMER_ADD_RELAYD_SOCKET: { + uint32_t major = msg.u.relayd_sock.major; + uint32_t minor = msg.u.relayd_sock.minor; + enum lttcomm_sock_proto protocol = + (enum lttcomm_sock_proto) msg.u.relayd_sock + .relayd_socket_protocol; + /* Session daemon status message are handled in the following call. */ consumer_add_relayd_socket(msg.u.relayd_sock.net_index, - msg.u.relayd_sock.type, ctx, sock, consumer_sockpoll, - &msg.u.relayd_sock.sock, msg.u.relayd_sock.session_id, - msg.u.relayd_sock.relayd_session_id); + msg.u.relayd_sock.type, ctx, sock, + consumer_sockpoll, msg.u.relayd_sock.session_id, + msg.u.relayd_sock.relayd_session_id, major, + minor, protocol); goto end_nosignal; } case LTTNG_CONSUMER_DESTROY_RELAYD: @@ -2046,8 +2065,7 @@ error_push_metadata_fatal: */ rotate_channel = lttng_consumer_rotate_channel( found_channel, key, - msg.u.rotate_channel.relayd_id, - msg.u.rotate_channel.metadata, ctx); + msg.u.rotate_channel.relayd_id); if (rotate_channel < 0) { ERR("Rotate channel failed"); ret_code = LTTCOMM_CONSUMERD_ROTATION_FAIL; @@ -2074,8 +2092,7 @@ error_push_metadata_fatal: ret_rotate_read_streams = lttng_consumer_rotate_ready_streams( - found_channel, key, - ctx); + found_channel, key); if (ret_rotate_read_streams < 0) { ERR("Rotate channel failed"); } @@ -2639,6 +2656,7 @@ enum sync_metadata_status lttng_ustconsumer_sync_metadata( LTTNG_ASSERT(ctx); LTTNG_ASSERT(metadata_stream); + ASSERT_RCU_READ_LOCKED(); metadata_channel = metadata_stream->chan; pthread_mutex_unlock(&metadata_stream->lock); @@ -3049,7 +3067,7 @@ end: } static int put_next_subbuffer(struct lttng_consumer_stream *stream, - struct stream_subbuffer *subbuffer) + struct stream_subbuffer *subbuffer __attribute__((unused))) { const int ret = lttng_ust_ctl_put_next_subbuf(stream->ustream); @@ -3058,7 +3076,7 @@ static int put_next_subbuffer(struct lttng_consumer_stream *stream, } static int signal_metadata(struct lttng_consumer_stream *stream, - struct lttng_consumer_local_data *ctx) + struct lttng_consumer_local_data *ctx __attribute__((unused))) { ASSERT_LOCKED(stream->metadata_rdv_lock); return pthread_cond_broadcast(&stream->metadata_rdv) ? -errno : 0;