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);
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);
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;
}
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;
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
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:
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:
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. */
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);
}
assert(stream);
assert(stream->ustream);
+ if (stream->chan->switch_timer_enabled == 1) {
+ consumer_timer_switch_stop(stream->chan);
+ }
ustctl_destroy_stream(stream->ustream);
}
projects / lttng-tools.git / blobdiff