consumer: implement clear channel

Signed-off-by: Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
Change-Id: I4eefc1d23b1b781f055790933572c4796375da26
Signed-off-by: Jérémie Galarneau <jeremie.galarneau@efficios.com>

diff --git a/src/common/consumer/consumer.c b/src/common/consumer/consumer.c
index 2963c801d5407f14c3993552aa50cee75bcb1d51..a1c669894bec5e9472e072c9410817a58399ec41 100644 (file)
--- a/src/common/consumer/consumer.c
+++ b/src/common/consumer/consumer.c
@@ -4173,6 +4173,115 @@ end:
        return ret;
 }
 
        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.
  *
 /*
  * Check if a stream is ready to be rotated after extracting it.
  *


@@ -4804,3 +4913,67 @@ end_rcu_unlock:
 end:
        return ret_code;
 }
 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 17c3ee581bf2ff079439d375337031df2b557195..9dfa91353bd69108c573b5439d2b5d3dcdb31836 100644 (file)
--- 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_CREATE_TRACE_CHUNK,
        LTTNG_CONSUMER_CLOSE_TRACE_CHUNK,
        LTTNG_CONSUMER_TRACE_CHUNK_EXISTS,

+       LTTNG_CONSUMER_CLEAR_CHANNEL,

 };
 
 enum lttng_consumer_type {
 };
 
 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);
 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 */
 
 #endif /* LIB_CONSUMER_H */

diff --git a/src/common/kernel-consumer/kernel-consumer.c b/src/common/kernel-consumer/kernel-consumer.c
index 2cd9704442fe468829d4d0e79d9e1cba394edddd..c5c87cc9e12ff191337c2e40283ca7a860193a26 100644 (file)
--- 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;
        }
 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,
        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 b053df521d0f3a9b46cb7b17d6ed92257fb9bd41..209869e762e4498cf40c706c3335afbb88ab4301 100644 (file)
--- 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;
        }
 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,
        case LTTNG_CONSUMER_INIT:
        {
                ret_code = lttng_consumer_init_command(ctx,