From: Mathieu Desnoyers Date: Thu, 12 Dec 2019 16:31:25 +0000 (-0500) Subject: consumer: implement clear channel X-Git-Url: http://git.efficios.com/?p=lttng-tools.git;a=commitdiff_plain;h=5f3aff8bbb7dfaa4aa9eb9234b6f2393c40b69bf consumer: implement clear channel Signed-off-by: Mathieu Desnoyers Change-Id: I4eefc1d23b1b781f055790933572c4796375da26 Signed-off-by: Jérémie Galarneau --- diff --git a/src/common/consumer/consumer.c b/src/common/consumer/consumer.c index 2963c801d..a1c669894 100644 --- a/src/common/consumer/consumer.c +++ b/src/common/consumer/consumer.c @@ -4173,6 +4173,115 @@ end: return ret; } +static +int consumer_clear_buffer(struct lttng_consumer_stream *stream) +{ + int ret = 0; + unsigned long consumed_pos_before, consumed_pos_after; + + ret = lttng_consumer_sample_snapshot_positions(stream); + if (ret < 0) { + ERR("Taking snapshot positions"); + goto end; + } + + ret = lttng_consumer_get_consumed_snapshot(stream, &consumed_pos_before); + if (ret < 0) { + ERR("Consumed snapshot position"); + goto end; + } + + switch (consumer_data.type) { + case LTTNG_CONSUMER_KERNEL: + ret = kernctl_buffer_clear(stream->wait_fd); + if (ret < 0) { + ERR("Failed to flush kernel stream"); + goto end; + } + break; + case LTTNG_CONSUMER32_UST: + case LTTNG_CONSUMER64_UST: + lttng_ustconsumer_clear_buffer(stream); + break; + default: + ERR("Unknown consumer_data type"); + abort(); + } + + ret = lttng_consumer_sample_snapshot_positions(stream); + if (ret < 0) { + ERR("Taking snapshot positions"); + goto end; + } + ret = lttng_consumer_get_consumed_snapshot(stream, &consumed_pos_after); + if (ret < 0) { + ERR("Consumed snapshot position"); + goto end; + } + DBG("clear: before: %lu after: %lu", consumed_pos_before, consumed_pos_after); +end: + return ret; +} + +static +int consumer_clear_stream(struct lttng_consumer_stream *stream) +{ + int ret; + + ret = consumer_flush_buffer(stream, 1); + if (ret < 0) { + ERR("Failed to flush stream %" PRIu64 " during channel clear", + stream->key); + ret = LTTCOMM_CONSUMERD_FATAL; + goto error; + } + + ret = consumer_clear_buffer(stream); + if (ret < 0) { + ERR("Failed to clear stream %" PRIu64 " during channel clear", + stream->key); + ret = LTTCOMM_CONSUMERD_FATAL; + goto error; + } + + ret = LTTCOMM_CONSUMERD_SUCCESS; +error: + return ret; +} + +static +int consumer_clear_unmonitored_channel(struct lttng_consumer_channel *channel) +{ + int ret; + struct lttng_consumer_stream *stream; + + rcu_read_lock(); + pthread_mutex_lock(&channel->lock); + cds_list_for_each_entry(stream, &channel->streams.head, send_node) { + health_code_update(); + pthread_mutex_lock(&stream->lock); + ret = consumer_clear_stream(stream); + if (ret) { + goto error_unlock; + } + pthread_mutex_unlock(&stream->lock); + } + pthread_mutex_unlock(&channel->lock); + rcu_read_unlock(); + return 0; + +error_unlock: + pthread_mutex_unlock(&stream->lock); + pthread_mutex_unlock(&channel->lock); + rcu_read_unlock(); + if (ret) { + goto error; + } + ret = LTTCOMM_CONSUMERD_SUCCESS; +error: + return ret; +} + /* * Check if a stream is ready to be rotated after extracting it. * @@ -4804,3 +4913,67 @@ end_rcu_unlock: end: return ret_code; } + +static +int consumer_clear_monitored_channel(struct lttng_consumer_channel *channel) +{ + struct lttng_ht *ht; + struct lttng_consumer_stream *stream; + struct lttng_ht_iter iter; + int ret; + + ht = consumer_data.stream_per_chan_id_ht; + + 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) { + /* + * Protect against teardown with mutex. + */ + pthread_mutex_lock(&stream->lock); + if (cds_lfht_is_node_deleted(&stream->node.node)) { + goto next; + } + ret = consumer_clear_stream(stream); + if (ret) { + goto error_unlock; + } + next: + pthread_mutex_unlock(&stream->lock); + } + rcu_read_unlock(); + return LTTCOMM_CONSUMERD_SUCCESS; + +error_unlock: + pthread_mutex_unlock(&stream->lock); + rcu_read_unlock(); + return ret; +} + +int lttng_consumer_clear_channel(struct lttng_consumer_channel *channel) +{ + int ret; + + DBG("Consumer clear channel %" PRIu64, channel->key); + + if (channel->type == CONSUMER_CHANNEL_TYPE_METADATA) { + /* + * Nothing to do for the metadata channel/stream. + * Snapshot mechanism already take care of the metadata + * handling/generation, and monitored channels only need to + * have their data stream cleared.. + */ + ret = LTTCOMM_CONSUMERD_SUCCESS; + goto end; + } + + if (!channel->monitor) { + ret = consumer_clear_unmonitored_channel(channel); + } else { + ret = consumer_clear_monitored_channel(channel); + } +end: + return ret; +} diff --git a/src/common/consumer/consumer.h b/src/common/consumer/consumer.h index 17c3ee581..9dfa91353 100644 --- a/src/common/consumer/consumer.h +++ b/src/common/consumer/consumer.h @@ -70,6 +70,7 @@ enum lttng_consumer_command { LTTNG_CONSUMER_CREATE_TRACE_CHUNK, LTTNG_CONSUMER_CLOSE_TRACE_CHUNK, LTTNG_CONSUMER_TRACE_CHUNK_EXISTS, + LTTNG_CONSUMER_CLEAR_CHANNEL, }; enum lttng_consumer_type { @@ -875,5 +876,6 @@ void lttng_consumer_cleanup_relayd(struct consumer_relayd_sock_pair *relayd); enum lttcomm_return_code lttng_consumer_init_command( struct lttng_consumer_local_data *ctx, const lttng_uuid sessiond_uuid); +int lttng_consumer_clear_channel(struct lttng_consumer_channel *channel); #endif /* LIB_CONSUMER_H */ diff --git a/src/common/kernel-consumer/kernel-consumer.c b/src/common/kernel-consumer/kernel-consumer.c index 2cd970444..c5c87cc9e 100644 --- a/src/common/kernel-consumer/kernel-consumer.c +++ b/src/common/kernel-consumer/kernel-consumer.c @@ -1159,6 +1159,32 @@ end_destroy_channel: error_rotate_channel: goto end_nosignal; } + case LTTNG_CONSUMER_CLEAR_CHANNEL: + { + struct lttng_consumer_channel *channel; + uint64_t key = msg.u.clear_channel.key; + + channel = consumer_find_channel(key); + if (!channel) { + DBG("Channel %" PRIu64 " not found", key); + ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND; + } else { + ret = lttng_consumer_clear_channel(channel); + if (ret) { + ERR("Clear channel failed"); + ret_code = ret; + } + + health_code_update(); + } + ret = consumer_send_status_msg(sock, ret_code); + if (ret < 0) { + /* Somehow, the session daemon is not responding anymore. */ + goto end_nosignal; + } + + break; + } case LTTNG_CONSUMER_INIT: { ret_code = lttng_consumer_init_command(ctx, diff --git a/src/common/ust-consumer/ust-consumer.c b/src/common/ust-consumer/ust-consumer.c index b053df521..209869e76 100644 --- a/src/common/ust-consumer/ust-consumer.c +++ b/src/common/ust-consumer/ust-consumer.c @@ -2013,6 +2013,31 @@ error_push_metadata_fatal: end_rotate_channel_nosignal: goto end_nosignal; } + case LTTNG_CONSUMER_CLEAR_CHANNEL: + { + struct lttng_consumer_channel *channel; + uint64_t key = msg.u.clear_channel.key; + + channel = consumer_find_channel(key); + if (!channel) { + DBG("Channel %" PRIu64 " not found", key); + ret_code = LTTCOMM_CONSUMERD_CHAN_NOT_FOUND; + } else { + ret = lttng_consumer_clear_channel(channel); + if (ret) { + ERR("Clear channel failed key %" PRIu64, key); + ret_code = ret; + } + + health_code_update(); + } + ret = consumer_send_status_msg(sock, ret_code); + if (ret < 0) { + /* Somehow, the session daemon is not responding anymore. */ + goto end_nosignal; + } + break; + } case LTTNG_CONSUMER_INIT: { ret_code = lttng_consumer_init_command(ctx,