X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fconsumer%2Fconsumer.c;h=a1c669894bec5e9472e072c9410817a58399ec41;hp=2963c801d5407f14c3993552aa50cee75bcb1d51;hb=5f3aff8bbb7dfaa4aa9eb9234b6f2393c40b69bf;hpb=0e2d816a62eccd0d159ec254e87ff80a7e6cf1bb

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;
+}