X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fconsumer%2Fconsumer.c;h=84fdbcaaa6902e076476489c3ab4e6b1aa7a1c9a;hp=b2c5eb3ddbd1484e5aa2f44f64bc48f17a8dceda;hb=1cbd136b2479ef142bfb339b13d3d25aa772dda5;hpb=503fefca8a1b82cfafccaa096e900c41bf4570f6

diff --git a/src/common/consumer/consumer.c b/src/common/consumer/consumer.c
index b2c5eb3dd..84fdbcaaa 100644
--- a/src/common/consumer/consumer.c
+++ b/src/common/consumer/consumer.c
@@ -9,7 +9,6 @@
 
 #include "common/index/ctf-index.h"
 #define _LGPL_SOURCE
-#include <assert.h>
 #include <poll.h>
 #include <pthread.h>
 #include <stdlib.h>
@@ -45,10 +44,10 @@
 #include <common/string-utils/format.h>
 #include <common/dynamic-array.h>
 
-struct lttng_consumer_global_data consumer_data = {
-	.stream_count = 0,
-	.need_update = 1,
-	.type = LTTNG_CONSUMER_UNKNOWN,
+struct lttng_consumer_global_data the_consumer_data = {
+		.stream_count = 0,
+		.need_update = 1,
+		.type = LTTNG_CONSUMER_UNKNOWN,
 };
 
 enum consumer_channel_action {
@@ -84,7 +83,7 @@ static struct lttng_ht *data_ht;
 
 static const char *get_consumer_domain(void)
 {
-	switch (consumer_data.type) {
+	switch (the_consumer_data.type) {
 	case LTTNG_CONSUMER_KERNEL:
 		return DEFAULT_KERNEL_TRACE_DIR;
 	case LTTNG_CONSUMER64_UST:
@@ -105,7 +104,7 @@ static void notify_thread_lttng_pipe(struct lttng_pipe *pipe)
 {
 	struct lttng_consumer_stream *null_stream = NULL;
 
-	assert(pipe);
+	LTTNG_ASSERT(pipe);
 
 	(void) lttng_pipe_write(pipe, &null_stream, sizeof(null_stream));
 }
@@ -173,7 +172,7 @@ static void clean_channel_stream_list(struct lttng_consumer_channel *channel)
 {
 	struct lttng_consumer_stream *stream, *stmp;
 
-	assert(channel);
+	LTTNG_ASSERT(channel);
 
 	/* Delete streams that might have been left in the stream list. */
 	cds_list_for_each_entry_safe(stream, stmp, &channel->streams.head,
@@ -201,7 +200,7 @@ static struct lttng_consumer_stream *find_stream(uint64_t key,
 	struct lttng_ht_node_u64 *node;
 	struct lttng_consumer_stream *stream = NULL;
 
-	assert(ht);
+	LTTNG_ASSERT(ht);
 
 	/* -1ULL keys are lookup failures */
 	if (key == (uint64_t) -1ULL) {
@@ -256,7 +255,7 @@ struct lttng_consumer_channel *consumer_find_channel(uint64_t key)
 		return NULL;
 	}
 
-	lttng_ht_lookup(consumer_data.channel_ht, &key, &iter);
+	lttng_ht_lookup(the_consumer_data.channel_ht, &key, &iter);
 	node = lttng_ht_iter_get_node_u64(&iter);
 	if (node != NULL) {
 		channel = caa_container_of(node, struct lttng_consumer_channel, node);
@@ -298,7 +297,7 @@ static void free_channel_rcu(struct rcu_head *head)
 	struct lttng_consumer_channel *channel =
 		caa_container_of(node, struct lttng_consumer_channel, node);
 
-	switch (consumer_data.type) {
+	switch (the_consumer_data.type) {
 	case LTTNG_CONSUMER_KERNEL:
 		break;
 	case LTTNG_CONSUMER32_UST:
@@ -352,7 +351,7 @@ void consumer_destroy_relayd(struct consumer_relayd_sock_pair *relayd)
 	DBG("Consumer destroy and close relayd socket pair");
 
 	iter.iter.node = &relayd->node.node;
-	ret = lttng_ht_del(consumer_data.relayd_ht, &iter);
+	ret = lttng_ht_del(the_consumer_data.relayd_ht, &iter);
 	if (ret != 0) {
 		/* We assume the relayd is being or is destroyed */
 		return;
@@ -372,7 +371,7 @@ void consumer_del_channel(struct lttng_consumer_channel *channel)
 
 	DBG("Consumer delete channel key %" PRIu64, channel->key);
 
-	pthread_mutex_lock(&consumer_data.lock);
+	pthread_mutex_lock(&the_consumer_data.lock);
 	pthread_mutex_lock(&channel->lock);
 
 	/* Destroy streams that might have been left in the stream list. */
@@ -385,7 +384,7 @@ void consumer_del_channel(struct lttng_consumer_channel *channel)
 		consumer_timer_monitor_stop(channel);
 	}
 
-	switch (consumer_data.type) {
+	switch (the_consumer_data.type) {
 	case LTTNG_CONSUMER_KERNEL:
 		break;
 	case LTTNG_CONSUMER32_UST:
@@ -394,7 +393,7 @@ void consumer_del_channel(struct lttng_consumer_channel *channel)
 		break;
 	default:
 		ERR("Unknown consumer_data type");
-		assert(0);
+		abort();
 		goto end;
 	}
 
@@ -406,13 +405,13 @@ void consumer_del_channel(struct lttng_consumer_channel *channel)
 
 		rcu_read_lock();
 		iter.iter.node = &channel->node.node;
-		ret = lttng_ht_del(consumer_data.channel_ht, &iter);
-		assert(!ret);
+		ret = lttng_ht_del(the_consumer_data.channel_ht, &iter);
+		LTTNG_ASSERT(!ret);
 
 		iter.iter.node = &channel->channels_by_session_id_ht_node.node;
-		ret = lttng_ht_del(consumer_data.channels_by_session_id_ht,
+		ret = lttng_ht_del(the_consumer_data.channels_by_session_id_ht,
 				&iter);
-		assert(!ret);
+		LTTNG_ASSERT(!ret);
 		rcu_read_unlock();
 	}
 
@@ -420,7 +419,7 @@ void consumer_del_channel(struct lttng_consumer_channel *channel)
 	call_rcu(&channel->node.head, free_channel_rcu);
 end:
 	pthread_mutex_unlock(&channel->lock);
-	pthread_mutex_unlock(&consumer_data.lock);
+	pthread_mutex_unlock(&the_consumer_data.lock);
 }
 
 /*
@@ -434,14 +433,14 @@ static void cleanup_relayd_ht(void)
 
 	rcu_read_lock();
 
-	cds_lfht_for_each_entry(consumer_data.relayd_ht->ht, &iter.iter, relayd,
-			node.node) {
+	cds_lfht_for_each_entry(the_consumer_data.relayd_ht->ht, &iter.iter,
+			relayd, node.node) {
 		consumer_destroy_relayd(relayd);
 	}
 
 	rcu_read_unlock();
 
-	lttng_ht_destroy(consumer_data.relayd_ht);
+	lttng_ht_destroy(the_consumer_data.relayd_ht);
 }
 
 /*
@@ -491,7 +490,7 @@ void lttng_consumer_cleanup_relayd(struct consumer_relayd_sock_pair *relayd)
 {
 	uint64_t netidx;
 
-	assert(relayd);
+	LTTNG_ASSERT(relayd);
 
 	DBG("Cleaning up relayd object ID %"PRIu64, relayd->net_seq_idx);
 
@@ -525,7 +524,7 @@ void lttng_consumer_cleanup_relayd(struct consumer_relayd_sock_pair *relayd)
  */
 void consumer_flag_relayd_for_destroy(struct consumer_relayd_sock_pair *relayd)
 {
-	assert(relayd);
+	LTTNG_ASSERT(relayd);
 
 	/* Set destroy flag for this object */
 	uatomic_set(&relayd->destroy_flag, 1);
@@ -576,12 +575,12 @@ void consumer_add_data_stream(struct lttng_consumer_stream *stream)
 {
 	struct lttng_ht *ht = data_ht;
 
-	assert(stream);
-	assert(ht);
+	LTTNG_ASSERT(stream);
+	LTTNG_ASSERT(ht);
 
 	DBG3("Adding consumer stream %" PRIu64, stream->key);
 
-	pthread_mutex_lock(&consumer_data.lock);
+	pthread_mutex_lock(&the_consumer_data.lock);
 	pthread_mutex_lock(&stream->chan->lock);
 	pthread_mutex_lock(&stream->chan->timer_lock);
 	pthread_mutex_lock(&stream->lock);
@@ -592,7 +591,7 @@ void consumer_add_data_stream(struct lttng_consumer_stream *stream)
 
 	lttng_ht_add_unique_u64(ht, &stream->node);
 
-	lttng_ht_add_u64(consumer_data.stream_per_chan_id_ht,
+	lttng_ht_add_u64(the_consumer_data.stream_per_chan_id_ht,
 			&stream->node_channel_id);
 
 	/*
@@ -600,7 +599,8 @@ void consumer_add_data_stream(struct lttng_consumer_stream *stream)
 	 * the key since the HT does not use it and we allow to add redundant keys
 	 * into this table.
 	 */
-	lttng_ht_add_u64(consumer_data.stream_list_ht, &stream->node_session_id);
+	lttng_ht_add_u64(the_consumer_data.stream_list_ht,
+			&stream->node_session_id);
 
 	/*
 	 * When nb_init_stream_left reaches 0, we don't need to trigger any action
@@ -616,14 +616,14 @@ void consumer_add_data_stream(struct lttng_consumer_stream *stream)
 	}
 
 	/* Update consumer data once the node is inserted. */
-	consumer_data.stream_count++;
-	consumer_data.need_update = 1;
+	the_consumer_data.stream_count++;
+	the_consumer_data.need_update = 1;
 
 	rcu_read_unlock();
 	pthread_mutex_unlock(&stream->lock);
 	pthread_mutex_unlock(&stream->chan->timer_lock);
 	pthread_mutex_unlock(&stream->chan->lock);
-	pthread_mutex_unlock(&consumer_data.lock);
+	pthread_mutex_unlock(&the_consumer_data.lock);
 }
 
 /*
@@ -636,15 +636,15 @@ static int add_relayd(struct consumer_relayd_sock_pair *relayd)
 	struct lttng_ht_node_u64 *node;
 	struct lttng_ht_iter iter;
 
-	assert(relayd);
+	LTTNG_ASSERT(relayd);
 
-	lttng_ht_lookup(consumer_data.relayd_ht,
-			&relayd->net_seq_idx, &iter);
+	lttng_ht_lookup(the_consumer_data.relayd_ht, &relayd->net_seq_idx,
+			&iter);
 	node = lttng_ht_iter_get_node_u64(&iter);
 	if (node != NULL) {
 		goto end;
 	}
-	lttng_ht_add_unique_u64(consumer_data.relayd_ht, &relayd->node);
+	lttng_ht_add_unique_u64(the_consumer_data.relayd_ht, &relayd->node);
 
 end:
 	return ret;
@@ -699,8 +699,7 @@ struct consumer_relayd_sock_pair *consumer_find_relayd(uint64_t key)
 		goto error;
 	}
 
-	lttng_ht_lookup(consumer_data.relayd_ht, &key,
-			&iter);
+	lttng_ht_lookup(the_consumer_data.relayd_ht, &key, &iter);
 	node = lttng_ht_iter_get_node_u64(&iter);
 	if (node != NULL) {
 		relayd = caa_container_of(node, struct consumer_relayd_sock_pair, node);
@@ -721,9 +720,9 @@ int consumer_send_relayd_stream(struct lttng_consumer_stream *stream,
 	int ret = 0;
 	struct consumer_relayd_sock_pair *relayd;
 
-	assert(stream);
-	assert(stream->net_seq_idx != -1ULL);
-	assert(path);
+	LTTNG_ASSERT(stream);
+	LTTNG_ASSERT(stream->net_seq_idx != -1ULL);
+	LTTNG_ASSERT(path);
 
 	/* The stream is not metadata. Get relayd reference if exists. */
 	rcu_read_lock();
@@ -770,7 +769,7 @@ int consumer_send_relayd_streams_sent(uint64_t net_seq_idx)
 	int ret = 0;
 	struct consumer_relayd_sock_pair *relayd;
 
-	assert(net_seq_idx != -1ULL);
+	LTTNG_ASSERT(net_seq_idx != -1ULL);
 
 	/* The stream is not metadata. Get relayd reference if exists. */
 	rcu_read_lock();
@@ -830,8 +829,8 @@ static int write_relayd_stream_header(struct lttng_consumer_stream *stream,
 	struct lttcomm_relayd_data_hdr data_hdr;
 
 	/* Safety net */
-	assert(stream);
-	assert(relayd);
+	LTTNG_ASSERT(stream);
+	LTTNG_ASSERT(relayd);
 
 	/* Reset data header */
 	memset(&data_hdr, 0, sizeof(data_hdr));
@@ -877,6 +876,43 @@ error:
 	return outfd;
 }
 
+/*
+ * Write a character on the metadata poll pipe to wake the metadata thread.
+ * Returns 0 on success, -1 on error.
+ */
+int consumer_metadata_wakeup_pipe(const struct lttng_consumer_channel *channel)
+{
+	int ret = 0;
+
+	DBG("Waking up metadata poll thread (writing to pipe): channel name = '%s'",
+			channel->name);
+	if (channel->monitor && channel->metadata_stream) {
+		const char dummy = 'c';
+		const ssize_t write_ret = lttng_write(
+				channel->metadata_stream->ust_metadata_poll_pipe[1],
+				&dummy, 1);
+
+		if (write_ret < 1) {
+			if (errno == EWOULDBLOCK) {
+				/*
+				 * This is fine, the metadata poll thread
+				 * is having a hard time keeping-up, but
+				 * it will eventually wake-up and consume
+				 * the available data.
+				 */
+				ret = 0;
+			} else {
+				PERROR("Failed to write to UST metadata pipe while attempting to wake-up the metadata poll thread");
+				ret = -1;
+				goto end;
+			}
+		}
+	}
+
+end:
+	return ret;
+}
+
 /*
  * Trigger a dump of the metadata content. Following/during the succesful
  * completion of this call, the metadata poll thread will start receiving
@@ -891,10 +927,10 @@ int consumer_metadata_stream_dump(struct lttng_consumer_stream *stream)
 
 	ASSERT_LOCKED(stream->chan->lock);
 	ASSERT_LOCKED(stream->lock);
-	assert(stream->metadata_flag);
-	assert(stream->chan->trace_chunk);
+	LTTNG_ASSERT(stream->metadata_flag);
+	LTTNG_ASSERT(stream->chan->trace_chunk);
 
-	switch (consumer_data.type) {
+	switch (the_consumer_data.type) {
 	case LTTNG_CONSUMER_KERNEL:
 		/*
 		 * Reset the position of what has been read from the
@@ -947,7 +983,7 @@ int lttng_consumer_channel_set_trace_chunk(
 		const bool acquired_reference = lttng_trace_chunk_get(
 				new_trace_chunk);
 
-		assert(acquired_reference);
+		LTTNG_ASSERT(acquired_reference);
 	}
 
 	lttng_trace_chunk_put(channel->trace_chunk);
@@ -984,7 +1020,7 @@ struct lttng_consumer_channel *consumer_allocate_channel(uint64_t key,
 
 	if (chunk_id) {
 		trace_chunk = lttng_trace_chunk_registry_find_chunk(
-				consumer_data.chunk_registry, session_id,
+				the_consumer_data.chunk_registry, session_id,
 				*chunk_id);
 		if (!trace_chunk) {
 			ERR("Failed to find trace chunk reference during creation of channel");
@@ -1019,7 +1055,7 @@ struct lttng_consumer_channel *consumer_allocate_channel(uint64_t key,
 		channel->output = CONSUMER_CHANNEL_MMAP;
 		break;
 	default:
-		assert(0);
+		abort();
 		free(channel);
 		channel = NULL;
 		goto end;
@@ -1087,7 +1123,7 @@ error:
 int consumer_add_channel(struct lttng_consumer_channel *channel,
 		struct lttng_consumer_local_data *ctx)
 {
-	pthread_mutex_lock(&consumer_data.lock);
+	pthread_mutex_lock(&the_consumer_data.lock);
 	pthread_mutex_lock(&channel->lock);
 	pthread_mutex_lock(&channel->timer_lock);
 
@@ -1099,15 +1135,15 @@ int consumer_add_channel(struct lttng_consumer_channel *channel,
 	steal_channel_key(channel->key);
 
 	rcu_read_lock();
-	lttng_ht_add_unique_u64(consumer_data.channel_ht, &channel->node);
-	lttng_ht_add_u64(consumer_data.channels_by_session_id_ht,
+	lttng_ht_add_unique_u64(the_consumer_data.channel_ht, &channel->node);
+	lttng_ht_add_u64(the_consumer_data.channels_by_session_id_ht,
 			&channel->channels_by_session_id_ht_node);
 	rcu_read_unlock();
 	channel->is_published = true;
 
 	pthread_mutex_unlock(&channel->timer_lock);
 	pthread_mutex_unlock(&channel->lock);
-	pthread_mutex_unlock(&consumer_data.lock);
+	pthread_mutex_unlock(&the_consumer_data.lock);
 
 	if (channel->wait_fd != -1 && channel->type == CONSUMER_CHANNEL_TYPE_DATA) {
 		notify_channel_pipe(ctx, channel, -1, CONSUMER_CHANNEL_ADD);
@@ -1131,10 +1167,10 @@ static int update_poll_array(struct lttng_consumer_local_data *ctx,
 	struct lttng_ht_iter iter;
 	struct lttng_consumer_stream *stream;
 
-	assert(ctx);
-	assert(ht);
-	assert(pollfd);
-	assert(local_stream);
+	LTTNG_ASSERT(ctx);
+	LTTNG_ASSERT(ht);
+	LTTNG_ASSERT(pollfd);
+	LTTNG_ASSERT(local_stream);
 
 	DBG("Updating poll fd array");
 	*nb_inactive_fd = 0;
@@ -1251,26 +1287,26 @@ void lttng_consumer_cleanup(void)
 
 	rcu_read_lock();
 
-	cds_lfht_for_each_entry(consumer_data.channel_ht->ht, &iter.iter, channel,
-			node.node) {
+	cds_lfht_for_each_entry(the_consumer_data.channel_ht->ht, &iter.iter,
+			channel, node.node) {
 		consumer_del_channel(channel);
 	}
 
 	rcu_read_unlock();
 
-	lttng_ht_destroy(consumer_data.channel_ht);
-	lttng_ht_destroy(consumer_data.channels_by_session_id_ht);
+	lttng_ht_destroy(the_consumer_data.channel_ht);
+	lttng_ht_destroy(the_consumer_data.channels_by_session_id_ht);
 
 	cleanup_relayd_ht();
 
-	lttng_ht_destroy(consumer_data.stream_per_chan_id_ht);
+	lttng_ht_destroy(the_consumer_data.stream_per_chan_id_ht);
 
 	/*
 	 * This HT contains streams that are freed by either the metadata thread or
 	 * the data thread so we do *nothing* on the hash table and simply destroy
 	 * it.
 	 */
-	lttng_ht_destroy(consumer_data.stream_list_ht);
+	lttng_ht_destroy(the_consumer_data.stream_list_ht);
 
 	/*
 	 * Trace chunks in the registry may still exist if the session
@@ -1285,7 +1321,7 @@ void lttng_consumer_cleanup(void)
 	 * to hit.
 	 */
 	trace_chunks_left = lttng_trace_chunk_registry_put_each_chunk(
-			consumer_data.chunk_registry);
+			the_consumer_data.chunk_registry);
 	if (trace_chunks_left) {
 		ERR("%u trace chunks are leaked by lttng-consumerd. "
 				"This can be caused by an internal error of the session daemon.",
@@ -1293,7 +1329,7 @@ void lttng_consumer_cleanup(void)
 	}
 	/* Run all callbacks freeing each chunk. */
 	rcu_barrier();
-	lttng_trace_chunk_registry_destroy(consumer_data.chunk_registry);
+	lttng_trace_chunk_registry_destroy(the_consumer_data.chunk_registry);
 }
 
 /*
@@ -1384,9 +1420,9 @@ struct lttng_consumer_local_data *lttng_consumer_create(
 	int ret;
 	struct lttng_consumer_local_data *ctx;
 
-	assert(consumer_data.type == LTTNG_CONSUMER_UNKNOWN ||
-		consumer_data.type == type);
-	consumer_data.type = type;
+	LTTNG_ASSERT(the_consumer_data.type == LTTNG_CONSUMER_UNKNOWN ||
+			the_consumer_data.type == type);
+	the_consumer_data.type = type;
 
 	ctx = zmalloc(sizeof(struct lttng_consumer_local_data));
 	if (ctx == NULL) {
@@ -1598,7 +1634,7 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap(
 
 	/* RCU lock for the relayd pointer */
 	rcu_read_lock();
-	assert(stream->net_seq_idx != (uint64_t) -1ULL ||
+	LTTNG_ASSERT(stream->net_seq_idx != (uint64_t) -1ULL ||
 			stream->trace_chunk);
 
 	/* Flag that the current stream if set for network streaming. */
@@ -1765,7 +1801,7 @@ ssize_t lttng_consumer_on_read_subbuffer_splice(
 	int *splice_pipe;
 	unsigned int relayd_hang_up = 0;
 
-	switch (consumer_data.type) {
+	switch (the_consumer_data.type) {
 	case LTTNG_CONSUMER_KERNEL:
 		break;
 	case LTTNG_CONSUMER32_UST:
@@ -1774,7 +1810,7 @@ ssize_t lttng_consumer_on_read_subbuffer_splice(
 		return -ENOSYS;
 	default:
 		ERR("Unknown consumer_data type");
-		assert(0);
+		abort();
 	}
 
 	/* RCU lock for the relayd pointer */
@@ -1969,7 +2005,7 @@ end:
  */
 int lttng_consumer_sample_snapshot_positions(struct lttng_consumer_stream *stream)
 {
-	switch (consumer_data.type) {
+	switch (the_consumer_data.type) {
 	case LTTNG_CONSUMER_KERNEL:
 		return lttng_kconsumer_sample_snapshot_positions(stream);
 	case LTTNG_CONSUMER32_UST:
@@ -1977,7 +2013,7 @@ int lttng_consumer_sample_snapshot_positions(struct lttng_consumer_stream *strea
 		return lttng_ustconsumer_sample_snapshot_positions(stream);
 	default:
 		ERR("Unknown consumer_data type");
-		assert(0);
+		abort();
 		return -ENOSYS;
 	}
 }
@@ -1988,7 +2024,7 @@ int lttng_consumer_sample_snapshot_positions(struct lttng_consumer_stream *strea
  */
 int lttng_consumer_take_snapshot(struct lttng_consumer_stream *stream)
 {
-	switch (consumer_data.type) {
+	switch (the_consumer_data.type) {
 	case LTTNG_CONSUMER_KERNEL:
 		return lttng_kconsumer_take_snapshot(stream);
 	case LTTNG_CONSUMER32_UST:
@@ -1996,7 +2032,7 @@ int lttng_consumer_take_snapshot(struct lttng_consumer_stream *stream)
 		return lttng_ustconsumer_take_snapshot(stream);
 	default:
 		ERR("Unknown consumer_data type");
-		assert(0);
+		abort();
 		return -ENOSYS;
 	}
 }
@@ -2009,7 +2045,7 @@ int lttng_consumer_take_snapshot(struct lttng_consumer_stream *stream)
 int lttng_consumer_get_produced_snapshot(struct lttng_consumer_stream *stream,
 		unsigned long *pos)
 {
-	switch (consumer_data.type) {
+	switch (the_consumer_data.type) {
 	case LTTNG_CONSUMER_KERNEL:
 		return lttng_kconsumer_get_produced_snapshot(stream, pos);
 	case LTTNG_CONSUMER32_UST:
@@ -2017,7 +2053,7 @@ int lttng_consumer_get_produced_snapshot(struct lttng_consumer_stream *stream,
 		return lttng_ustconsumer_get_produced_snapshot(stream, pos);
 	default:
 		ERR("Unknown consumer_data type");
-		assert(0);
+		abort();
 		return -ENOSYS;
 	}
 }
@@ -2030,7 +2066,7 @@ int lttng_consumer_get_produced_snapshot(struct lttng_consumer_stream *stream,
 int lttng_consumer_get_consumed_snapshot(struct lttng_consumer_stream *stream,
 		unsigned long *pos)
 {
-	switch (consumer_data.type) {
+	switch (the_consumer_data.type) {
 	case LTTNG_CONSUMER_KERNEL:
 		return lttng_kconsumer_get_consumed_snapshot(stream, pos);
 	case LTTNG_CONSUMER32_UST:
@@ -2038,7 +2074,7 @@ int lttng_consumer_get_consumed_snapshot(struct lttng_consumer_stream *stream,
 		return lttng_ustconsumer_get_consumed_snapshot(stream, pos);
 	default:
 		ERR("Unknown consumer_data type");
-		assert(0);
+		abort();
 		return -ENOSYS;
 	}
 }
@@ -2046,7 +2082,7 @@ int lttng_consumer_get_consumed_snapshot(struct lttng_consumer_stream *stream,
 int lttng_consumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 		int sock, struct pollfd *consumer_sockpoll)
 {
-	switch (consumer_data.type) {
+	switch (the_consumer_data.type) {
 	case LTTNG_CONSUMER_KERNEL:
 		return lttng_kconsumer_recv_cmd(ctx, sock, consumer_sockpoll);
 	case LTTNG_CONSUMER32_UST:
@@ -2054,7 +2090,7 @@ int lttng_consumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 		return lttng_ustconsumer_recv_cmd(ctx, sock, consumer_sockpoll);
 	default:
 		ERR("Unknown consumer_data type");
-		assert(0);
+		abort();
 		return -ENOSYS;
 	}
 }
@@ -2062,7 +2098,7 @@ int lttng_consumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 static
 void lttng_consumer_close_all_metadata(void)
 {
-	switch (consumer_data.type) {
+	switch (the_consumer_data.type) {
 	case LTTNG_CONSUMER_KERNEL:
 		/*
 		 * The Kernel consumer has a different metadata scheme so we don't
@@ -2082,7 +2118,7 @@ void lttng_consumer_close_all_metadata(void)
 		break;
 	default:
 		ERR("Unknown consumer_data type");
-		assert(0);
+		abort();
 	}
 }
 
@@ -2095,16 +2131,16 @@ void consumer_del_metadata_stream(struct lttng_consumer_stream *stream,
 	struct lttng_consumer_channel *channel = NULL;
 	bool free_channel = false;
 
-	assert(stream);
+	LTTNG_ASSERT(stream);
 	/*
 	 * This call should NEVER receive regular stream. It must always be
 	 * metadata stream and this is crucial for data structure synchronization.
 	 */
-	assert(stream->metadata_flag);
+	LTTNG_ASSERT(stream->metadata_flag);
 
 	DBG3("Consumer delete metadata stream %d", stream->wait_fd);
 
-	pthread_mutex_lock(&consumer_data.lock);
+	pthread_mutex_lock(&the_consumer_data.lock);
 	/*
 	 * Note that this assumes that a stream's channel is never changed and
 	 * that the stream's lock doesn't need to be taken to sample its
@@ -2146,7 +2182,7 @@ void consumer_del_metadata_stream(struct lttng_consumer_stream *stream,
 	}
 	pthread_mutex_unlock(&stream->lock);
 	pthread_mutex_unlock(&channel->lock);
-	pthread_mutex_unlock(&consumer_data.lock);
+	pthread_mutex_unlock(&the_consumer_data.lock);
 
 	if (free_channel) {
 		consumer_del_channel(channel);
@@ -2167,12 +2203,12 @@ void consumer_add_metadata_stream(struct lttng_consumer_stream *stream)
 	struct lttng_ht_iter iter;
 	struct lttng_ht_node_u64 *node;
 
-	assert(stream);
-	assert(ht);
+	LTTNG_ASSERT(stream);
+	LTTNG_ASSERT(ht);
 
 	DBG3("Adding metadata stream %" PRIu64 " to hash table", stream->key);
 
-	pthread_mutex_lock(&consumer_data.lock);
+	pthread_mutex_lock(&the_consumer_data.lock);
 	pthread_mutex_lock(&stream->chan->lock);
 	pthread_mutex_lock(&stream->chan->timer_lock);
 	pthread_mutex_lock(&stream->lock);
@@ -2190,7 +2226,7 @@ void consumer_add_metadata_stream(struct lttng_consumer_stream *stream)
 	 */
 	lttng_ht_lookup(ht, &stream->key, &iter);
 	node = lttng_ht_iter_get_node_u64(&iter);
-	assert(!node);
+	LTTNG_ASSERT(!node);
 
 	/*
 	 * When nb_init_stream_left reaches 0, we don't need to trigger any action
@@ -2207,22 +2243,23 @@ void consumer_add_metadata_stream(struct lttng_consumer_stream *stream)
 
 	lttng_ht_add_unique_u64(ht, &stream->node);
 
-	lttng_ht_add_u64(consumer_data.stream_per_chan_id_ht,
-		&stream->node_channel_id);
+	lttng_ht_add_u64(the_consumer_data.stream_per_chan_id_ht,
+			&stream->node_channel_id);
 
 	/*
 	 * Add stream to the stream_list_ht of the consumer data. No need to steal
 	 * the key since the HT does not use it and we allow to add redundant keys
 	 * into this table.
 	 */
-	lttng_ht_add_u64(consumer_data.stream_list_ht, &stream->node_session_id);
+	lttng_ht_add_u64(the_consumer_data.stream_list_ht,
+			&stream->node_session_id);
 
 	rcu_read_unlock();
 
 	pthread_mutex_unlock(&stream->lock);
 	pthread_mutex_unlock(&stream->chan->lock);
 	pthread_mutex_unlock(&stream->chan->timer_lock);
-	pthread_mutex_unlock(&consumer_data.lock);
+	pthread_mutex_unlock(&the_consumer_data.lock);
 }
 
 /*
@@ -2258,7 +2295,7 @@ static void validate_endpoint_status_metadata_stream(
 
 	DBG("Consumer delete flagged metadata stream");
 
-	assert(pollset);
+	LTTNG_ASSERT(pollset);
 
 	rcu_read_lock();
 	cds_lfht_for_each_entry(metadata_ht->ht, &iter.iter, stream, node.node) {
@@ -2410,7 +2447,7 @@ restart:
 				lttng_ht_lookup(metadata_ht, &tmp_id, &iter);
 			}
 			node = lttng_ht_iter_get_node_u64(&iter);
-			assert(node);
+			LTTNG_ASSERT(node);
 
 			stream = caa_container_of(node, struct lttng_consumer_stream,
 					node);
@@ -2418,7 +2455,7 @@ restart:
 			if (revents & (LPOLLIN | LPOLLPRI)) {
 				/* Get the data out of the metadata file descriptor */
 				DBG("Metadata available on fd %d", pollfd);
-				assert(stream->wait_fd == pollfd);
+				LTTNG_ASSERT(stream->wait_fd == pollfd);
 
 				do {
 					health_code_update();
@@ -2440,9 +2477,10 @@ restart:
 				}
 			} else if (revents & (LPOLLERR | LPOLLHUP)) {
 				DBG("Metadata fd %d is hup|err.", pollfd);
-				if (!stream->hangup_flush_done
-						&& (consumer_data.type == LTTNG_CONSUMER32_UST
-							|| consumer_data.type == LTTNG_CONSUMER64_UST)) {
+				if (!stream->hangup_flush_done &&
+						(the_consumer_data.type == LTTNG_CONSUMER32_UST ||
+								the_consumer_data.type ==
+										LTTNG_CONSUMER64_UST)) {
 					DBG("Attempting to flush and consume the UST buffers");
 					lttng_ustconsumer_on_stream_hangup(stream);
 
@@ -2538,8 +2576,8 @@ void *consumer_thread_data_poll(void *data)
 		 * the fds set has been updated, we need to update our
 		 * local array as well
 		 */
-		pthread_mutex_lock(&consumer_data.lock);
-		if (consumer_data.need_update) {
+		pthread_mutex_lock(&the_consumer_data.lock);
+		if (the_consumer_data.need_update) {
 			free(pollfd);
 			pollfd = NULL;
 
@@ -2547,18 +2585,21 @@ void *consumer_thread_data_poll(void *data)
 			local_stream = NULL;
 
 			/* Allocate for all fds */
-			pollfd = zmalloc((consumer_data.stream_count + nb_pipes_fd) * sizeof(struct pollfd));
+			pollfd = zmalloc((the_consumer_data.stream_count +
+							 nb_pipes_fd) *
+					sizeof(struct pollfd));
 			if (pollfd == NULL) {
 				PERROR("pollfd malloc");
-				pthread_mutex_unlock(&consumer_data.lock);
+				pthread_mutex_unlock(&the_consumer_data.lock);
 				goto end;
 			}
 
-			local_stream = zmalloc((consumer_data.stream_count + nb_pipes_fd) *
+			local_stream = zmalloc((the_consumer_data.stream_count +
+							       nb_pipes_fd) *
 					sizeof(struct lttng_consumer_stream *));
 			if (local_stream == NULL) {
 				PERROR("local_stream malloc");
-				pthread_mutex_unlock(&consumer_data.lock);
+				pthread_mutex_unlock(&the_consumer_data.lock);
 				goto end;
 			}
 			ret = update_poll_array(ctx, &pollfd, local_stream,
@@ -2566,13 +2607,13 @@ void *consumer_thread_data_poll(void *data)
 			if (ret < 0) {
 				ERR("Error in allocating pollfd or local_outfds");
 				lttng_consumer_send_error(ctx, LTTCOMM_CONSUMERD_POLL_ERROR);
-				pthread_mutex_unlock(&consumer_data.lock);
+				pthread_mutex_unlock(&the_consumer_data.lock);
 				goto end;
 			}
 			nb_fd = ret;
-			consumer_data.need_update = 0;
+			the_consumer_data.need_update = 0;
 		}
-		pthread_mutex_unlock(&consumer_data.lock);
+		pthread_mutex_unlock(&the_consumer_data.lock);
 
 		/* No FDs and consumer_quit, consumer_cleanup the thread */
 		if (nb_fd == 0 && nb_inactive_fd == 0 &&
@@ -2718,8 +2759,8 @@ void *consumer_thread_data_poll(void *data)
 			}
 			if (!local_stream[i]->hangup_flush_done
 					&& (pollfd[i].revents & (POLLHUP | POLLERR | POLLNVAL))
-					&& (consumer_data.type == LTTNG_CONSUMER32_UST
-						|| consumer_data.type == LTTNG_CONSUMER64_UST)) {
+					&& (the_consumer_data.type == LTTNG_CONSUMER32_UST
+						|| the_consumer_data.type == LTTNG_CONSUMER64_UST)) {
 				DBG("fd %d is hup|err|nval. Attempting flush and read.",
 						pollfd[i].fd);
 				lttng_ustconsumer_on_stream_hangup(local_stream[i]);
@@ -2798,7 +2839,7 @@ void consumer_close_channel_streams(struct lttng_consumer_channel *channel)
 	struct lttng_consumer_stream *stream;
 	struct lttng_ht_iter iter;
 
-	ht = consumer_data.stream_per_chan_id_ht;
+	ht = the_consumer_data.stream_per_chan_id_ht;
 
 	rcu_read_lock();
 	cds_lfht_for_each_entry_duplicate(ht->ht,
@@ -2812,7 +2853,7 @@ void consumer_close_channel_streams(struct lttng_consumer_channel *channel)
 		if (cds_lfht_is_node_deleted(&stream->node.node)) {
 			goto next;
 		}
-		switch (consumer_data.type) {
+		switch (the_consumer_data.type) {
 		case LTTNG_CONSUMER_KERNEL:
 			break;
 		case LTTNG_CONSUMER32_UST:
@@ -2831,7 +2872,7 @@ void consumer_close_channel_streams(struct lttng_consumer_channel *channel)
 			break;
 		default:
 			ERR("Unknown consumer_data type");
-			assert(0);
+			abort();
 		}
 	next:
 		pthread_mutex_unlock(&stream->lock);
@@ -2852,7 +2893,7 @@ static void destroy_channel_ht(struct lttng_ht *ht)
 	rcu_read_lock();
 	cds_lfht_for_each_entry(ht->ht, &iter.iter, channel, wait_fd_node.node) {
 		ret = lttng_ht_del(ht, &iter);
-		assert(ret != 0);
+		LTTNG_ASSERT(ret != 0);
 	}
 	rcu_read_unlock();
 
@@ -2990,9 +3031,9 @@ restart:
 						lttng_poll_del(&events, chan->wait_fd);
 						iter.iter.node = &chan->wait_fd_node.node;
 						ret = lttng_ht_del(channel_ht, &iter);
-						assert(ret == 0);
+						LTTNG_ASSERT(ret == 0);
 
-						switch (consumer_data.type) {
+						switch (the_consumer_data.type) {
 						case LTTNG_CONSUMER_KERNEL:
 							break;
 						case LTTNG_CONSUMER32_UST:
@@ -3003,7 +3044,7 @@ restart:
 							break;
 						default:
 							ERR("Unknown consumer_data type");
-							assert(0);
+							abort();
 						}
 
 						/*
@@ -3051,7 +3092,7 @@ restart:
 				lttng_ht_lookup(channel_ht, &tmp_id, &iter);
 			}
 			node = lttng_ht_iter_get_node_u64(&iter);
-			assert(node);
+			LTTNG_ASSERT(node);
 
 			chan = caa_container_of(node, struct lttng_consumer_channel,
 					wait_fd_node);
@@ -3062,7 +3103,7 @@ restart:
 
 				lttng_poll_del(&events, chan->wait_fd);
 				ret = lttng_ht_del(channel_ht, &iter);
-				assert(ret == 0);
+				LTTNG_ASSERT(ret == 0);
 
 				/*
 				 * This will close the wait fd for each stream associated to
@@ -3110,8 +3151,8 @@ static int set_metadata_socket(struct lttng_consumer_local_data *ctx,
 {
 	int ret;
 
-	assert(ctx);
-	assert(sockpoll);
+	LTTNG_ASSERT(ctx);
+	LTTNG_ASSERT(sockpoll);
 
 	ret = lttng_consumer_poll_socket(sockpoll);
 	if (ret) {
@@ -3253,7 +3294,7 @@ void *consumer_thread_sessiond_poll(void *data)
 			err = 0;	/* All is OK */
 			goto end;
 		}
-		DBG("received command on sock");
+		DBG("Received command on sock");
 	}
 	/* All is OK */
 	err = 0;
@@ -3340,6 +3381,7 @@ ssize_t lttng_consumer_read_subbuffer(struct lttng_consumer_stream *stream,
 	ssize_t ret, written_bytes = 0;
 	int rotation_ret;
 	struct stream_subbuffer subbuffer = {};
+	enum get_next_subbuffer_status get_next_status;
 
 	if (!locked_by_caller) {
 		stream->read_subbuffer_ops.lock(stream);
@@ -3365,14 +3407,20 @@ ssize_t lttng_consumer_read_subbuffer(struct lttng_consumer_stream *stream,
 		}
 	}
 
-	ret = stream->read_subbuffer_ops.get_next_subbuffer(stream, &subbuffer);
-	if (ret) {
-		if (ret == -ENODATA) {
-			/* Not an error. */
-			ret = 0;
-			goto sleep_stream;
-		}
+	get_next_status = stream->read_subbuffer_ops.get_next_subbuffer(
+			stream, &subbuffer);
+	switch (get_next_status) {
+	case GET_NEXT_SUBBUFFER_STATUS_OK:
+		break;
+	case GET_NEXT_SUBBUFFER_STATUS_NO_DATA:
+		/* Not an error. */
+		ret = 0;
+		goto sleep_stream;
+	case GET_NEXT_SUBBUFFER_STATUS_ERROR:
+		ret = -1;
 		goto end;
+	default:
+		abort();
 	}
 
 	ret = stream->read_subbuffer_ops.pre_consume_subbuffer(
@@ -3383,24 +3431,10 @@ ssize_t lttng_consumer_read_subbuffer(struct lttng_consumer_stream *stream,
 
 	written_bytes = stream->read_subbuffer_ops.consume_subbuffer(
 			ctx, stream, &subbuffer);
-	/*
-	 * Should write subbuf_size amount of data when network streaming or
-	 * the full padded size when we are not streaming.
-	 */
-	if ((written_bytes != subbuffer.info.data.subbuf_size &&
-			    stream->net_seq_idx != (uint64_t) -1ULL) ||
-			(written_bytes != subbuffer.info.data.padded_subbuf_size &&
-					stream->net_seq_idx ==
-							(uint64_t) -1ULL)) {
-		/*
-		 * Display the error but continue processing to try to
-		 * release the subbuffer. This is a DBG statement
-		 * since this can happen without being a critical
-		 * error.
-		 */
-		DBG("Failed to write to tracefile (written_bytes: %zd != padded subbuffer size: %lu, subbuffer size: %lu)",
-				written_bytes, subbuffer.info.data.padded_subbuf_size,
-				subbuffer.info.data.subbuf_size);
+	if (written_bytes <= 0) {
+		ERR("Error consuming subbuffer: (%zd)", written_bytes);
+		ret = (int) written_bytes;
+		goto error_put_subbuf;
 	}
 
 	ret = stream->read_subbuffer_ops.put_next_subbuffer(stream, &subbuffer);
@@ -3454,7 +3488,7 @@ error_put_subbuf:
 
 int lttng_consumer_on_recv_stream(struct lttng_consumer_stream *stream)
 {
-	switch (consumer_data.type) {
+	switch (the_consumer_data.type) {
 	case LTTNG_CONSUMER_KERNEL:
 		return lttng_kconsumer_on_recv_stream(stream);
 	case LTTNG_CONSUMER32_UST:
@@ -3462,7 +3496,7 @@ int lttng_consumer_on_recv_stream(struct lttng_consumer_stream *stream)
 		return lttng_ustconsumer_on_recv_stream(stream);
 	default:
 		ERR("Unknown consumer_data type");
-		assert(0);
+		abort();
 		return -ENOSYS;
 	}
 }
@@ -3472,29 +3506,30 @@ int lttng_consumer_on_recv_stream(struct lttng_consumer_stream *stream)
  */
 int lttng_consumer_init(void)
 {
-	consumer_data.channel_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
-	if (!consumer_data.channel_ht) {
+	the_consumer_data.channel_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
+	if (!the_consumer_data.channel_ht) {
 		goto error;
 	}
 
-	consumer_data.channels_by_session_id_ht =
+	the_consumer_data.channels_by_session_id_ht =
 			lttng_ht_new(0, LTTNG_HT_TYPE_U64);
-	if (!consumer_data.channels_by_session_id_ht) {
+	if (!the_consumer_data.channels_by_session_id_ht) {
 		goto error;
 	}
 
-	consumer_data.relayd_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
-	if (!consumer_data.relayd_ht) {
+	the_consumer_data.relayd_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
+	if (!the_consumer_data.relayd_ht) {
 		goto error;
 	}
 
-	consumer_data.stream_list_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
-	if (!consumer_data.stream_list_ht) {
+	the_consumer_data.stream_list_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
+	if (!the_consumer_data.stream_list_ht) {
 		goto error;
 	}
 
-	consumer_data.stream_per_chan_id_ht = lttng_ht_new(0, LTTNG_HT_TYPE_U64);
-	if (!consumer_data.stream_per_chan_id_ht) {
+	the_consumer_data.stream_per_chan_id_ht =
+			lttng_ht_new(0, LTTNG_HT_TYPE_U64);
+	if (!the_consumer_data.stream_per_chan_id_ht) {
 		goto error;
 	}
 
@@ -3508,8 +3543,8 @@ int lttng_consumer_init(void)
 		goto error;
 	}
 
-	consumer_data.chunk_registry = lttng_trace_chunk_registry_create();
-	if (!consumer_data.chunk_registry) {
+	the_consumer_data.chunk_registry = lttng_trace_chunk_registry_create();
+	if (!the_consumer_data.chunk_registry) {
 		goto error;
 	}
 
@@ -3535,15 +3570,15 @@ error:
 	enum lttcomm_return_code ret_code = LTTCOMM_CONSUMERD_SUCCESS;
 	struct consumer_relayd_sock_pair *relayd = NULL;
 
-	assert(ctx);
-	assert(relayd_sock);
+	LTTNG_ASSERT(ctx);
+	LTTNG_ASSERT(relayd_sock);
 
 	DBG("Consumer adding relayd socket (idx: %" PRIu64 ")", net_seq_idx);
 
 	/* Get relayd reference if exists. */
 	relayd = consumer_find_relayd(net_seq_idx);
 	if (relayd == NULL) {
-		assert(sock_type == LTTNG_STREAM_CONTROL);
+		LTTNG_ASSERT(sock_type == LTTNG_STREAM_CONTROL);
 		/* Not found. Allocate one. */
 		relayd = consumer_allocate_relayd_sock_pair(net_seq_idx);
 		if (relayd == NULL) {
@@ -3563,7 +3598,7 @@ error:
 		/*
 		 * relayd key should never be found for control socket.
 		 */
-		assert(sock_type != LTTNG_STREAM_CONTROL);
+		LTTNG_ASSERT(sock_type != LTTNG_STREAM_CONTROL);
 	}
 
 	/* First send a status message before receiving the fds. */
@@ -3717,8 +3752,8 @@ static struct consumer_relayd_sock_pair *find_relayd_by_session_id(uint64_t id)
 	struct consumer_relayd_sock_pair *relayd = NULL;
 
 	/* Iterate over all relayd since they are indexed by net_seq_idx. */
-	cds_lfht_for_each_entry(consumer_data.relayd_ht->ht, &iter.iter, relayd,
-			node.node) {
+	cds_lfht_for_each_entry(the_consumer_data.relayd_ht->ht, &iter.iter,
+			relayd, node.node) {
 		/*
 		 * Check by sessiond id which is unique here where the relayd session
 		 * id might not be when having multiple relayd.
@@ -3753,9 +3788,9 @@ int consumer_data_pending(uint64_t id)
 	DBG("Consumer data pending command on session id %" PRIu64, id);
 
 	rcu_read_lock();
-	pthread_mutex_lock(&consumer_data.lock);
+	pthread_mutex_lock(&the_consumer_data.lock);
 
-	switch (consumer_data.type) {
+	switch (the_consumer_data.type) {
 	case LTTNG_CONSUMER_KERNEL:
 		data_pending = lttng_kconsumer_data_pending;
 		break;
@@ -3765,11 +3800,11 @@ int consumer_data_pending(uint64_t id)
 		break;
 	default:
 		ERR("Unknown consumer data type");
-		assert(0);
+		abort();
 	}
 
 	/* Ease our life a bit */
-	ht = consumer_data.stream_list_ht;
+	ht = the_consumer_data.stream_list_ht;
 
 	cds_lfht_for_each_entry_duplicate(ht->ht,
 			ht->hash_fct(&id, lttng_ht_seed),
@@ -3856,13 +3891,13 @@ int consumer_data_pending(uint64_t id)
 
 data_not_pending:
 	/* Data is available to be read by a viewer. */
-	pthread_mutex_unlock(&consumer_data.lock);
+	pthread_mutex_unlock(&the_consumer_data.lock);
 	rcu_read_unlock();
 	return 0;
 
 data_pending:
 	/* Data is still being extracted from buffers. */
-	pthread_mutex_unlock(&consumer_data.lock);
+	pthread_mutex_unlock(&the_consumer_data.lock);
 	rcu_read_unlock();
 	return 1;
 }
@@ -3892,7 +3927,7 @@ int consumer_send_status_channel(int sock,
 {
 	struct lttcomm_consumer_status_channel msg;
 
-	assert(sock >= 0);
+	LTTNG_ASSERT(sock >= 0);
 
 	memset(&msg, 0, sizeof(msg));
 	if (!channel) {
@@ -3915,7 +3950,7 @@ unsigned long consumer_get_consume_start_pos(unsigned long consumed_pos,
 	if (!nb_packets_per_stream) {
 		return consumed_pos;	/* Grab everything */
 	}
-	start_pos = produced_pos - offset_align_floor(produced_pos, max_sb_size);
+	start_pos = produced_pos - lttng_offset_align_floor(produced_pos, max_sb_size);
 	start_pos -= max_sb_size * nb_packets_per_stream;
 	if ((long) (start_pos - consumed_pos) < 0) {
 		return consumed_pos;	/* Grab everything */
@@ -3923,6 +3958,36 @@ unsigned long consumer_get_consume_start_pos(unsigned long consumed_pos,
 	return start_pos;
 }
 
+/* Stream lock must be held by the caller. */
+static int sample_stream_positions(struct lttng_consumer_stream *stream,
+			    unsigned long *produced, unsigned long *consumed)
+{
+	int ret;
+
+	ASSERT_LOCKED(stream->lock);
+
+	ret = lttng_consumer_sample_snapshot_positions(stream);
+	if (ret < 0) {
+		ERR("Failed to sample snapshot positions");
+		goto end;
+	}
+
+	ret = lttng_consumer_get_produced_snapshot(stream, produced);
+	if (ret < 0) {
+		ERR("Failed to sample produced position");
+		goto end;
+	}
+
+	ret = lttng_consumer_get_consumed_snapshot(stream, consumed);
+	if (ret < 0) {
+		ERR("Failed to sample consumed position");
+		goto end;
+	}
+
+end:
+	return ret;
+}
+
 /*
  * Sample the rotate position for all the streams of a channel. If a stream
  * is already at the rotate position (produced == consumed), we flag it as
@@ -3939,23 +4004,27 @@ int lttng_consumer_rotate_channel(struct lttng_consumer_channel *channel,
 	int ret;
 	struct lttng_consumer_stream *stream;
 	struct lttng_ht_iter iter;
-	struct lttng_ht *ht = consumer_data.stream_per_chan_id_ht;
+	struct lttng_ht *ht = the_consumer_data.stream_per_chan_id_ht;
 	struct lttng_dynamic_array stream_rotation_positions;
 	uint64_t next_chunk_id, stream_count = 0;
 	enum lttng_trace_chunk_status chunk_status;
 	const bool is_local_trace = relayd_id == -1ULL;
 	struct consumer_relayd_sock_pair *relayd = NULL;
 	bool rotating_to_new_chunk = true;
+	/* Array of `struct lttng_consumer_stream *` */
+	struct lttng_dynamic_pointer_array streams_packet_to_open;
+	size_t stream_idx;
 
 	DBG("Consumer sample rotate position for channel %" PRIu64, key);
 
 	lttng_dynamic_array_init(&stream_rotation_positions,
 			sizeof(struct relayd_stream_rotation_position), NULL);
+	lttng_dynamic_pointer_array_init(&streams_packet_to_open, NULL);
 
 	rcu_read_lock();
 
 	pthread_mutex_lock(&channel->lock);
-	assert(channel->trace_chunk);
+	LTTNG_ASSERT(channel->trace_chunk);
 	chunk_status = lttng_trace_chunk_get_id(channel->trace_chunk,
 			&next_chunk_id);
 	if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
@@ -3981,19 +4050,96 @@ int lttng_consumer_rotate_channel(struct lttng_consumer_channel *channel,
 		}
 
 		/*
-		 * Do not flush an empty packet when rotating from a NULL trace
+		 * Do not flush a packet when rotating from a NULL trace
 		 * chunk. The stream has no means to output data, and the prior
-		 * rotation which rotated to NULL performed that side-effect already.
+		 * rotation which rotated to NULL performed that side-effect
+		 * already. No new data can be produced when a stream has no
+		 * associated trace chunk (e.g. a stop followed by a rotate).
 		 */
 		if (stream->trace_chunk) {
+			bool flush_active;
+
+			if (stream->metadata_flag) {
+				/*
+				 * Don't produce an empty metadata packet,
+				 * simply close the current one.
+				 *
+				 * Metadata is regenerated on every trace chunk
+				 * switch; there is no concern that no data was
+				 * produced.
+				 */
+				flush_active = true;
+			} else {
+				/*
+				 * Only flush an empty packet if the "packet
+				 * open" could not be performed on transition
+				 * to a new trace chunk and no packets were
+				 * consumed within the chunk's lifetime.
+				 */
+				if (stream->opened_packet_in_current_trace_chunk) {
+					flush_active = true;
+				} else {
+					/*
+					 * Stream could have been full at the
+					 * time of rotation, but then have had
+					 * no activity at all.
+					 *
+					 * It is important to flush a packet
+					 * to prevent 0-length files from being
+					 * produced as most viewers choke on
+					 * them.
+					 *
+					 * Unfortunately viewers will not be
+					 * able to know that tracing was active
+					 * for this stream during this trace
+					 * chunk's lifetime.
+					 */
+					ret = sample_stream_positions(stream, &produced_pos, &consumed_pos);
+					if (ret) {
+						goto end_unlock_stream;
+					}
+
+					/*
+					 * Don't flush an empty packet if data
+					 * was produced; it will be consumed
+					 * before the rotation completes.
+					 */
+					flush_active = produced_pos != consumed_pos;
+					if (!flush_active) {
+						const char *trace_chunk_name;
+						uint64_t trace_chunk_id;
+
+						chunk_status = lttng_trace_chunk_get_name(
+								stream->trace_chunk,
+								&trace_chunk_name,
+								NULL);
+						if (chunk_status == LTTNG_TRACE_CHUNK_STATUS_NONE) {
+							trace_chunk_name = "none";
+						}
+
+						/*
+						 * Consumer trace chunks are
+						 * never anonymous.
+						 */
+						chunk_status = lttng_trace_chunk_get_id(
+								stream->trace_chunk,
+								&trace_chunk_id);
+						LTTNG_ASSERT(chunk_status ==
+								LTTNG_TRACE_CHUNK_STATUS_OK);
+
+						DBG("Unable to open packet for stream during trace chunk's lifetime. "
+								"Flushing an empty packet to prevent an empty file from being created: "
+								"stream id = %" PRIu64 ", trace chunk name = `%s`, trace chunk id = %" PRIu64,
+								stream->key, trace_chunk_name, trace_chunk_id);
+					}
+				}
+			}
+
 			/*
-			 * For metadata stream, do an active flush, which does not
-			 * produce empty packets. For data streams, empty-flush;
-			 * ensures we have at least one packet in each stream per trace
-			 * chunk, even if no data was produced.
+			 * Close the current packet before sampling the
+			 * ring buffer positions.
 			 */
-			ret = consumer_stream_flush_buffer(
-					stream, stream->metadata_flag ? 1 : 0);
+			ret = consumer_stream_flush_buffer(stream, flush_active);
 			if (ret < 0) {
 				ERR("Failed to flush stream %" PRIu64 " during channel rotation",
 						stream->key);
@@ -4025,7 +4171,7 @@ int lttng_consumer_rotate_channel(struct lttng_consumer_channel *channel,
 		 * Align produced position on the start-of-packet boundary of the first
 		 * packet going into the next trace chunk.
 		 */
-		produced_pos = ALIGN_FLOOR(produced_pos, stream->max_sb_size);
+		produced_pos = lttng_align_floor(produced_pos, stream->max_sb_size);
 		if (consumed_pos == produced_pos) {
 			DBG("Set rotate ready for stream %" PRIu64 " produced = %lu consumed = %lu",
 					stream->key, produced_pos, consumed_pos);
@@ -4128,67 +4274,88 @@ int lttng_consumer_rotate_channel(struct lttng_consumer_channel *channel,
 			 * packets in this scenario and allows the tracer to
 			 * "stamp" the beginning of the new trace chunk at the
 			 * earliest possible point.
+			 *
+			 * The packet open is performed after the channel
+			 * rotation to ensure that no attempt to open a packet
+			 * is performed in a stream that has no active trace
+			 * chunk.
 			 */
-			const enum consumer_stream_open_packet_status status =
-				consumer_stream_open_packet(stream);
-
-			switch (status) {
-			case CONSUMER_STREAM_OPEN_PACKET_STATUS_OPENED:
-				DBG("Opened a packet after a rotation: stream id = %" PRIu64
-						", channel name = %s, session id = %" PRIu64,
-						stream->key, stream->chan->name,
-						stream->chan->session_id);
-				break;
-			case CONSUMER_STREAM_OPEN_PACKET_STATUS_NO_SPACE:
-				/*
-				 * Can't open a packet as there is no space left
-				 * in the buffer. A new packet will be opened
-				 * once one has been consumed.
-				 */
-				DBG("No space left to open a packet after a rotation: stream id = %" PRIu64
-						", channel name = %s, session id = %" PRIu64,
-						stream->key, stream->chan->name,
-						stream->chan->session_id);
-				break;
-			case CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR:
-				/* Logged by callee. */
+			ret = lttng_dynamic_pointer_array_add_pointer(
+					&streams_packet_to_open, stream);
+			if (ret) {
+				PERROR("Failed to add a stream pointer to array of streams in which to open a packet");
 				ret = -1;
 				goto end_unlock_stream;
-			default:
-				abort();
 			}
 		}
 
 		pthread_mutex_unlock(&stream->lock);
 	}
 	stream = NULL;
-	pthread_mutex_unlock(&channel->lock);
 
-	if (is_local_trace) {
-		ret = 0;
-		goto end;
-	}
+	if (!is_local_trace) {
+		relayd = consumer_find_relayd(relayd_id);
+		if (!relayd) {
+			ERR("Failed to find relayd %" PRIu64, relayd_id);
+			ret = -1;
+			goto end_unlock_channel;
+		}
 
-	relayd = consumer_find_relayd(relayd_id);
-	if (!relayd) {
-		ERR("Failed to find relayd %" PRIu64, relayd_id);
-		ret = -1;
-		goto end;
+		pthread_mutex_lock(&relayd->ctrl_sock_mutex);
+		ret = relayd_rotate_streams(&relayd->control_sock, stream_count,
+				rotating_to_new_chunk ? &next_chunk_id : NULL,
+				(const struct relayd_stream_rotation_position *)
+						stream_rotation_positions.buffer
+								.data);
+		pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
+		if (ret < 0) {
+			ERR("Relayd rotate stream failed. Cleaning up relayd %" PRIu64,
+					relayd->net_seq_idx);
+			lttng_consumer_cleanup_relayd(relayd);
+			goto end_unlock_channel;
+		}
 	}
 
-	pthread_mutex_lock(&relayd->ctrl_sock_mutex);
-	ret = relayd_rotate_streams(&relayd->control_sock, stream_count,
-			rotating_to_new_chunk ? &next_chunk_id : NULL,
-			(const struct relayd_stream_rotation_position *)
-					stream_rotation_positions.buffer.data);
-	pthread_mutex_unlock(&relayd->ctrl_sock_mutex);
-	if (ret < 0) {
-		ERR("Relayd rotate stream failed. Cleaning up relayd %" PRIu64,
-				relayd->net_seq_idx);
-		lttng_consumer_cleanup_relayd(relayd);
-		goto end;
+	for (stream_idx = 0;
+			stream_idx < lttng_dynamic_pointer_array_get_count(
+				&streams_packet_to_open);
+			stream_idx++) {
+		enum consumer_stream_open_packet_status status;
+
+		stream = lttng_dynamic_pointer_array_get_pointer(
+				&streams_packet_to_open, stream_idx);
+
+		pthread_mutex_lock(&stream->lock);
+		status = consumer_stream_open_packet(stream);
+		pthread_mutex_unlock(&stream->lock);
+		switch (status) {
+		case CONSUMER_STREAM_OPEN_PACKET_STATUS_OPENED:
+			DBG("Opened a packet after a rotation: stream id = %" PRIu64
+			    ", channel name = %s, session id = %" PRIu64,
+					stream->key, stream->chan->name,
+					stream->chan->session_id);
+			break;
+		case CONSUMER_STREAM_OPEN_PACKET_STATUS_NO_SPACE:
+			/*
+			 * Can't open a packet as there is no space left
+			 * in the buffer. A new packet will be opened
+			 * once one has been consumed.
+			 */
+			DBG("No space left to open a packet after a rotation: stream id = %" PRIu64
+			    ", channel name = %s, session id = %" PRIu64,
+					stream->key, stream->chan->name,
+					stream->chan->session_id);
+			break;
+		case CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR:
+			/* Logged by callee. */
+			ret = -1;
+			goto end_unlock_channel;
+		default:
+			abort();
+		}
 	}
 
+	pthread_mutex_unlock(&channel->lock);
 	ret = 0;
 	goto end;
 
@@ -4199,6 +4366,7 @@ end_unlock_channel:
 end:
 	rcu_read_unlock();
 	lttng_dynamic_array_reset(&stream_rotation_positions);
+	lttng_dynamic_pointer_array_reset(&streams_packet_to_open);
 	return ret;
 }
 
@@ -4220,7 +4388,7 @@ int consumer_clear_buffer(struct lttng_consumer_stream *stream)
 		goto end;
 	}
 
-	switch (consumer_data.type) {
+	switch (the_consumer_data.type) {
 	case LTTNG_CONSUMER_KERNEL:
 		ret = kernctl_buffer_clear(stream->wait_fd);
 		if (ret < 0) {
@@ -4230,7 +4398,11 @@ int consumer_clear_buffer(struct lttng_consumer_stream *stream)
 		break;
 	case LTTNG_CONSUMER32_UST:
 	case LTTNG_CONSUMER64_UST:
-		lttng_ustconsumer_clear_buffer(stream);
+		ret = lttng_ustconsumer_clear_buffer(stream);
+		if (ret < 0) {
+			ERR("Failed to clear ust stream (ret = %d)", ret);
+			goto end;
+		}
 		break;
 	default:
 		ERR("Unknown consumer_data type");
@@ -4500,7 +4672,7 @@ int lttng_consumer_rotate_ready_streams(struct lttng_consumer_channel *channel,
 	int ret;
 	struct lttng_consumer_stream *stream;
 	struct lttng_ht_iter iter;
-	struct lttng_ht *ht = consumer_data.stream_per_chan_id_ht;
+	struct lttng_ht *ht = the_consumer_data.stream_per_chan_id_ht;
 
 	rcu_read_lock();
 
@@ -4586,10 +4758,10 @@ enum lttcomm_return_code lttng_consumer_create_trace_chunk(
 		} else {
 			relayd_id_str = "(formatting error)";
 		}
-        }
+	}
 
-        /* Local protocol error. */
-	assert(chunk_creation_timestamp);
+	/* Local protocol error. */
+	LTTNG_ASSERT(chunk_creation_timestamp);
 	ret = time_to_iso8601_str(chunk_creation_timestamp,
 			creation_timestamp_buffer,
 			sizeof(creation_timestamp_buffer));
@@ -4658,7 +4830,7 @@ enum lttcomm_return_code lttng_consumer_create_trace_chunk(
 	}
 
 	published_chunk = lttng_trace_chunk_registry_publish_chunk(
-			consumer_data.chunk_registry, session_id,
+			the_consumer_data.chunk_registry, session_id,
 			created_chunk);
 	lttng_trace_chunk_put(created_chunk);
 	created_chunk = NULL;
@@ -4669,10 +4841,11 @@ enum lttcomm_return_code lttng_consumer_create_trace_chunk(
 	}
 
 	rcu_read_lock();
-	cds_lfht_for_each_entry_duplicate(consumer_data.channels_by_session_id_ht->ht,
-			consumer_data.channels_by_session_id_ht->hash_fct(
+	cds_lfht_for_each_entry_duplicate(
+			the_consumer_data.channels_by_session_id_ht->ht,
+			the_consumer_data.channels_by_session_id_ht->hash_fct(
 					&session_id, lttng_ht_seed),
-			consumer_data.channels_by_session_id_ht->match_fct,
+			the_consumer_data.channels_by_session_id_ht->match_fct,
 			&session_id, &iter.iter, channel,
 			channels_by_session_id_ht_node.node) {
 		ret = lttng_consumer_channel_set_trace_chunk(channel,
@@ -4785,7 +4958,7 @@ enum lttcomm_return_code lttng_consumer_close_trace_chunk(
 			close_command_name);
 
 	chunk = lttng_trace_chunk_registry_find_chunk(
-			consumer_data.chunk_registry, session_id, chunk_id);
+			the_consumer_data.chunk_registry, session_id, chunk_id);
 	if (!chunk) {
 		ERR("Failed to find chunk: session_id = %" PRIu64
 				", chunk_id = %" PRIu64,
@@ -4816,7 +4989,7 @@ enum lttcomm_return_code lttng_consumer_close_trace_chunk(
 	 * current chunk found in the session's channels.
 	 */
 	rcu_read_lock();
-	cds_lfht_for_each_entry(consumer_data.channel_ht->ht, &iter.iter,
+	cds_lfht_for_each_entry(the_consumer_data.channel_ht->ht, &iter.iter,
 			channel, node.node) {
 		int ret;
 
@@ -4887,8 +5060,6 @@ enum lttcomm_return_code lttng_consumer_trace_chunk_exists(
 	bool chunk_exists_local, chunk_exists_remote;
 
 	if (relayd_id) {
-		int ret;
-
 		/* Only used for logging purposes. */
 		ret = snprintf(relayd_id_buffer, sizeof(relayd_id_buffer),
 				"%" PRIu64, *relayd_id);
@@ -4897,14 +5068,14 @@ enum lttcomm_return_code lttng_consumer_trace_chunk_exists(
 		} else {
 			relayd_id_str = "(formatting error)";
 		}
-        }
+	}
 
 	DBG("Consumer trace chunk exists command: relayd_id = %s"
 			", chunk_id = %" PRIu64, relayd_id_str,
 			chunk_id);
 	ret = lttng_trace_chunk_registry_chunk_exists(
-			consumer_data.chunk_registry, session_id,
-			chunk_id, &chunk_exists_local);
+			the_consumer_data.chunk_registry, session_id, chunk_id,
+			&chunk_exists_local);
 	if (ret) {
 		/* Internal error. */
 		ERR("Failed to query the existence of a trace chunk");
@@ -4959,7 +5130,7 @@ int consumer_clear_monitored_channel(struct lttng_consumer_channel *channel)
 	struct lttng_ht_iter iter;
 	int ret;
 
-	ht = consumer_data.stream_per_chan_id_ht;
+	ht = the_consumer_data.stream_per_chan_id_ht;
 
 	rcu_read_lock();
 	cds_lfht_for_each_entry_duplicate(ht->ht,
@@ -5081,3 +5252,8 @@ error_unlock:
 	pthread_mutex_unlock(&stream->lock);
 	goto end_rcu_unlock;
 }
+
+void lttng_consumer_sigbus_handle(void *addr)
+{
+	lttng_ustconsumer_sigbus_handle(addr);
+}