X-Git-Url: https://git.lttng.org/?a=blobdiff_plain;f=src%2Fcommon%2Fust-consumer%2Fust-consumer.c;h=3731cbb26b527c4deb1621de5bf3f84051007a20;hb=13886d2d433762b7cf54f8e812f747ec2829de57;hp=0edb513f89c15263b2244983dcbf22626f95c734;hpb=8fd623e0d1c389a76e4dc3ee74a77490b678c5b8;p=lttng-tools.git diff --git a/src/common/ust-consumer/ust-consumer.c b/src/common/ust-consumer/ust-consumer.c index 0edb513f8..3731cbb26 100644 --- a/src/common/ust-consumer/ust-consumer.c +++ b/src/common/ust-consumer/ust-consumer.c @@ -189,9 +189,9 @@ static int send_stream_to_thread(struct lttng_consumer_stream *stream, /* Get the right pipe where the stream will be sent. */ if (stream->metadata_flag) { - stream_pipe = ctx->consumer_metadata_pipe[1]; + stream_pipe = lttng_pipe_get_writefd(ctx->consumer_metadata_pipe); } else { - stream_pipe = ctx->consumer_data_pipe[1]; + stream_pipe = lttng_pipe_get_writefd(ctx->consumer_data_pipe); } do { @@ -278,6 +278,12 @@ static int create_ust_streams(struct lttng_consumer_channel *channel, */ stream->wait_fd = wait_fd; + /* + * Increment channel refcount since the channel reference has now been + * assigned in the allocation process above. + */ + uatomic_inc(&stream->chan->refcount); + /* * Order is important this is why a list is used. On error, the caller * should clean this list. @@ -581,6 +587,7 @@ static int flush_channel(uint64_t chan_key) DBG("UST consumer flush channel key %" PRIu64, chan_key); + rcu_read_lock(); channel = consumer_find_channel(chan_key); if (!channel) { ERR("UST consumer flush channel %" PRIu64 " not found", chan_key); @@ -591,26 +598,25 @@ static int flush_channel(uint64_t chan_key) ht = consumer_data.stream_per_chan_id_ht; /* For each stream of the channel id, flush it. */ - rcu_read_lock(); cds_lfht_for_each_entry_duplicate(ht->ht, ht->hash_fct(&channel->key, lttng_ht_seed), ht->match_fct, &channel->key, &iter.iter, stream, node_channel_id.node) { ustctl_flush_buffer(stream->ustream, 1); } - rcu_read_unlock(); - error: + rcu_read_unlock(); return ret; } /* * Close metadata stream wakeup_fd using the given key to retrieve the channel. + * RCU read side lock MUST be acquired before calling this function. * * Return 0 on success else an LTTng error code. */ static int close_metadata(uint64_t chan_key) { - int ret; + int ret = 0; struct lttng_consumer_channel *channel; DBG("UST consumer close metadata key %" PRIu64, chan_key); @@ -622,18 +628,22 @@ static int close_metadata(uint64_t chan_key) goto error; } - 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 = LTTCOMM_CONSUMERD_ERROR_METADATA; - goto error; - } - if (channel->switch_timer_enabled == 1) { - DBG("Deleting timer on metadata channel"); - consumer_timer_switch_stop(channel); + 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); + } + 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 = LTTCOMM_CONSUMERD_ERROR_METADATA; + goto error_unlock; + } } - consumer_metadata_cache_destroy(channel); +error_unlock: + pthread_mutex_unlock(&consumer_data.lock); error: return ret; } @@ -883,6 +893,12 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, case LTTNG_UST_CHAN_PER_CPU: channel->type = CONSUMER_CHANNEL_TYPE_DATA; attr.type = LTTNG_UST_CHAN_PER_CPU; + /* + * Set refcount to 1 for owner. Below, we will + * pass ownership to the + * consumer_thread_channel_poll() thread. + */ + channel->refcount = 1; break; case LTTNG_UST_CHAN_METADATA: channel->type = CONSUMER_CHANNEL_TYPE_METADATA; @@ -898,17 +914,34 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, goto end_channel_error; } + if (msg.u.ask_channel.type == LTTNG_UST_CHAN_METADATA) { + ret = consumer_metadata_cache_allocate(channel); + if (ret < 0) { + ERR("Allocating metadata cache"); + goto end_channel_error; + } + consumer_timer_switch_start(channel, attr.switch_timer_interval); + attr.switch_timer_interval = 0; + } + /* * Add the channel to the internal state AFTER all streams were created * and successfully sent to session daemon. This way, all streams must * be ready before this channel is visible to the threads. + * If add_channel succeeds, ownership of the channel is + * passed to consumer_thread_channel_poll(). */ ret = add_channel(channel, ctx); if (ret < 0) { + if (msg.u.ask_channel.type == LTTNG_UST_CHAN_METADATA) { + if (channel->switch_timer_enabled == 1) { + consumer_timer_switch_stop(channel); + } + consumer_metadata_cache_destroy(channel); + } goto end_channel_error; } - /* * Channel and streams are now created. Inform the session daemon that * everything went well and should wait to receive the channel and @@ -923,16 +956,6 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, goto end_nosignal; } - if (msg.u.ask_channel.type == LTTNG_UST_CHAN_METADATA) { - ret = consumer_metadata_cache_allocate(channel); - if (ret < 0) { - ERR("Allocating metadata cache"); - goto end_channel_error; - } - consumer_timer_switch_start(channel, attr.switch_timer_interval); - attr.switch_timer_interval = 0; - } - break; } case LTTNG_CONSUMER_GET_CHANNEL: @@ -990,17 +1013,13 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, case LTTNG_CONSUMER_DESTROY_CHANNEL: { uint64_t key = msg.u.destroy_channel.key; - struct lttng_consumer_channel *channel; - - channel = consumer_find_channel(key); - if (!channel) { - ERR("UST consumer get channel key %" PRIu64 " not found", key); - ret_code = LTTNG_ERR_UST_CHAN_NOT_FOUND; - goto end_msg_sessiond; - } - - destroy_channel(channel); + /* + * Only called if streams have not been sent to stream + * manager thread. However, channel has been sent to + * channel manager thread. + */ + notify_thread_del_channel(ctx, key); goto end_msg_sessiond; } case LTTNG_CONSUMER_CLOSE_METADATA: @@ -1040,6 +1059,7 @@ int lttng_ustconsumer_recv_cmd(struct lttng_consumer_local_data *ctx, if (!channel) { ERR("UST consumer push metadata %" PRIu64 " not found", key); ret_code = LTTNG_ERR_UST_CHAN_NOT_FOUND; + goto end_msg_sessiond; } /* Tell session daemon we are ready to receive the metadata. */ @@ -1188,6 +1208,10 @@ void lttng_ustconsumer_del_channel(struct lttng_consumer_channel *chan) assert(chan); assert(chan->uchan); + if (chan->switch_timer_enabled == 1) { + consumer_timer_switch_stop(chan); + } + consumer_metadata_cache_destroy(chan); ustctl_destroy_channel(chan->uchan); } @@ -1196,6 +1220,9 @@ void lttng_ustconsumer_del_stream(struct lttng_consumer_stream *stream) assert(stream); assert(stream->ustream); + if (stream->chan->switch_timer_enabled == 1) { + consumer_timer_switch_stop(stream->chan); + } ustctl_destroy_stream(stream->ustream); }