X-Git-Url: https://git.lttng.org/?a=blobdiff_plain;f=src%2Fcommon%2Fconsumer%2Fconsumer-stream.c;h=be19c1bfd17a3bcf3526096f51ed389c387bd849;hb=1cbd136b2479ef142bfb339b13d3d25aa772dda5;hp=398d71ae05f92f6d9c0437b4fb203532870eb5d6;hpb=f96af312b7d09d834f9d13408654ada93320ae2a;p=lttng-tools.git

diff --git a/src/common/consumer/consumer-stream.c b/src/common/consumer/consumer-stream.c
index 398d71ae0..be19c1bfd 100644
--- a/src/common/consumer/consumer-stream.c
+++ b/src/common/consumer/consumer-stream.c
@@ -8,7 +8,6 @@
  */
 
 #define _LGPL_SOURCE
-#include <assert.h>
 #include <inttypes.h>
 #include <sys/mman.h>
 #include <unistd.h>
@@ -22,6 +21,7 @@
 #include <common/consumer/consumer.h>
 #include <common/consumer/consumer-timer.h>
 #include <common/consumer/metadata-bucket.h>
+#include <common/kernel-ctl/kernel-ctl.h>
 
 #include "consumer-stream.h"
 
@@ -151,9 +151,40 @@ static ssize_t consumer_stream_consume_mmap(
 	const unsigned long padding_size =
 			subbuffer->info.data.padded_subbuf_size -
 			subbuffer->info.data.subbuf_size;
-
-	return lttng_consumer_on_read_subbuffer_mmap(
+	const ssize_t written_bytes = lttng_consumer_on_read_subbuffer_mmap(
 			stream, &subbuffer->buffer.buffer, padding_size);
+
+	if (stream->net_seq_idx == -1ULL) {
+		/*
+		 * When writing on disk, check that only the subbuffer (no
+		 * padding) was written to disk.
+		 */
+		if (written_bytes != subbuffer->info.data.padded_subbuf_size) {
+			DBG("Failed to write the entire padded subbuffer on disk (written_bytes: %zd, padded subbuffer size %lu)",
+					written_bytes,
+					subbuffer->info.data.padded_subbuf_size);
+		}
+	} else {
+		/*
+		 * When streaming over the network, check that the entire
+		 * subbuffer including padding was successfully written.
+		 */
+		if (written_bytes != subbuffer->info.data.subbuf_size) {
+			DBG("Failed to write only the subbuffer over the network (written_bytes: %zd, subbuffer size %lu)",
+					written_bytes,
+					subbuffer->info.data.subbuf_size);
+		}
+	}
+
+	/*
+	 * If `lttng_consumer_on_read_subbuffer_mmap()` returned an error, pass
+	 * it along to the caller, else return zero.
+	 */
+	if (written_bytes < 0) {
+		ERR("Error reading mmap subbuffer: %zd", written_bytes);
+	}
+
+	return written_bytes;
 }
 
 static ssize_t consumer_stream_consume_splice(
@@ -161,8 +192,24 @@ static ssize_t consumer_stream_consume_splice(
 		struct lttng_consumer_stream *stream,
 		const struct stream_subbuffer *subbuffer)
 {
-	return lttng_consumer_on_read_subbuffer_splice(ctx, stream,
-			subbuffer->info.data.padded_subbuf_size, 0);
+	const ssize_t written_bytes = lttng_consumer_on_read_subbuffer_splice(
+			ctx, stream, subbuffer->info.data.padded_subbuf_size, 0);
+
+	if (written_bytes != subbuffer->info.data.padded_subbuf_size) {
+		DBG("Failed to write the entire padded subbuffer (written_bytes: %zd, padded subbuffer size %lu)",
+				written_bytes,
+				subbuffer->info.data.padded_subbuf_size);
+	}
+
+	/*
+	 * If `lttng_consumer_on_read_subbuffer_splice()` returned an error,
+	 * pass it along to the caller, else return zero.
+	 */
+	if (written_bytes < 0) {
+		ERR("Error reading splice subbuffer: %zd", written_bytes);
+	}
+
+	return written_bytes;
 }
 
 static int consumer_stream_send_index(
@@ -198,9 +245,9 @@ static int do_sync_metadata(struct lttng_consumer_stream *metadata,
 	int ret;
 	enum sync_metadata_status status;
 
-	assert(metadata);
-	assert(metadata->metadata_flag);
-	assert(ctx);
+	LTTNG_ASSERT(metadata);
+	LTTNG_ASSERT(metadata->metadata_flag);
+	LTTNG_ASSERT(ctx);
 
 	/*
 	 * In UST, since we have to write the metadata from the cache packet
@@ -318,10 +365,10 @@ int consumer_stream_sync_metadata(struct lttng_consumer_local_data *ctx,
 	struct lttng_ht_iter iter;
 	struct lttng_ht *ht;
 
-	assert(ctx);
+	LTTNG_ASSERT(ctx);
 
 	/* Ease our life a bit. */
-	ht = consumer_data.stream_list_ht;
+	ht = the_consumer_data.stream_list_ht;
 
 	rcu_read_lock();
 
@@ -360,7 +407,7 @@ static int consumer_stream_sync_metadata_index(
 
 	/* Block until all the metadata is sent. */
 	pthread_mutex_lock(&stream->metadata_timer_lock);
-	assert(!stream->missed_metadata_flush);
+	LTTNG_ASSERT(!stream->missed_metadata_flush);
 	stream->waiting_on_metadata = true;
 	pthread_mutex_unlock(&stream->metadata_timer_lock);
 
@@ -409,6 +456,155 @@ end:
 	return 0;
 }
 
+static
+bool stream_is_rotating_to_null_chunk(
+		const struct lttng_consumer_stream *stream)
+{
+	bool rotating_to_null_chunk = false;
+
+	if (stream->rotate_position == -1ULL) {
+		/* No rotation ongoing. */
+		goto end;
+	}
+
+	if (stream->trace_chunk == stream->chan->trace_chunk ||
+			!stream->chan->trace_chunk) {
+		rotating_to_null_chunk = true;
+	}
+end:
+	return rotating_to_null_chunk;
+}
+
+enum consumer_stream_open_packet_status consumer_stream_open_packet(
+		struct lttng_consumer_stream *stream)
+{
+	int ret;
+	enum consumer_stream_open_packet_status status;
+	unsigned long produced_pos_before, produced_pos_after;
+
+	ret = lttng_consumer_sample_snapshot_positions(stream);
+	if (ret < 0) {
+		ERR("Failed to snapshot positions before post-rotation empty packet flush: stream id = %" PRIu64
+				", channel name = %s, session id = %" PRIu64,
+				stream->key, stream->chan->name,
+				stream->chan->session_id);
+		status = CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR;
+		goto end;
+	}
+
+	ret = lttng_consumer_get_produced_snapshot(
+			stream, &produced_pos_before);
+	if (ret < 0) {
+		ERR("Failed to read produced position before post-rotation empty packet flush: stream id = %" PRIu64
+				", channel name = %s, session id = %" PRIu64,
+				stream->key, stream->chan->name,
+				stream->chan->session_id);
+		status = CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR;
+		goto end;
+	}
+
+	ret = consumer_stream_flush_buffer(stream, 0);
+	if (ret) {
+		ERR("Failed to flush an empty packet at rotation point: stream id = %" PRIu64
+				", channel name = %s, session id = %" PRIu64,
+				stream->key, stream->chan->name,
+				stream->chan->session_id);
+		status = CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR;
+		goto end;
+	}
+
+	ret = lttng_consumer_sample_snapshot_positions(stream);
+	if (ret < 0) {
+		ERR("Failed to snapshot positions after post-rotation empty packet flush: stream id = %" PRIu64
+				", channel name = %s, session id = %" PRIu64,
+				stream->key, stream->chan->name,
+				stream->chan->session_id);
+		status = CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR;
+		goto end;
+	}
+
+	ret = lttng_consumer_get_produced_snapshot(stream, &produced_pos_after);
+	if (ret < 0) {
+		ERR("Failed to read produced position after post-rotation empty packet flush: stream id = %" PRIu64
+				", channel name = %s, session id = %" PRIu64,
+				stream->key, stream->chan->name,
+				stream->chan->session_id);
+		status = CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR;
+		goto end;
+	}
+
+	/*
+	 * Determine if the flush had an effect by comparing the produced
+	 * positons before and after the flush.
+	 */
+	status = produced_pos_before != produced_pos_after ?
+			CONSUMER_STREAM_OPEN_PACKET_STATUS_OPENED :
+			CONSUMER_STREAM_OPEN_PACKET_STATUS_NO_SPACE;
+	if (status == CONSUMER_STREAM_OPEN_PACKET_STATUS_OPENED) {
+		stream->opened_packet_in_current_trace_chunk = true;
+	}
+
+end:
+	return status;
+}
+
+/*
+ * An attempt to open a new packet is performed after a rotation completes to
+ * get a begin timestamp as close as possible to the rotation point.
+ *
+ * However, that initial attempt at opening a packet can fail due to a full
+ * ring-buffer. In that case, a second attempt is performed after consuming
+ * a packet since that will have freed enough space in the ring-buffer.
+ */
+static
+int post_consume_open_new_packet(struct lttng_consumer_stream *stream,
+		const struct stream_subbuffer *subbuffer,
+		struct lttng_consumer_local_data *ctx)
+{
+	int ret = 0;
+
+	if (!stream->opened_packet_in_current_trace_chunk &&
+			stream->trace_chunk &&
+			!stream_is_rotating_to_null_chunk(stream)) {
+		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 consuming a packet rotation: stream id = %" PRIu64
+					", channel name = %s, session id = %" PRIu64,
+					stream->key, stream->chan->name,
+					stream->chan->session_id);
+			stream->opened_packet_in_current_trace_chunk = true;
+			break;
+		case CONSUMER_STREAM_OPEN_PACKET_STATUS_NO_SPACE:
+			/*
+			 * Can't open a packet as there is no space left.
+			 * This means that new events were produced, resulting
+			 * in a packet being opened, which is what we want
+			 * anyhow.
+			 */
+			DBG("No space left to open a packet after consuming a packet: stream id = %" PRIu64
+					", channel name = %s, session id = %" PRIu64,
+					stream->key, stream->chan->name,
+					stream->chan->session_id);
+			stream->opened_packet_in_current_trace_chunk = true;
+			break;
+		case CONSUMER_STREAM_OPEN_PACKET_STATUS_ERROR:
+			/* Logged by callee. */
+			ret = -1;
+			goto end;
+		default:
+			abort();
+		}
+
+		stream->opened_packet_in_current_trace_chunk = true;
+	}
+
+end:
+	return ret;
+}
+
 struct lttng_consumer_stream *consumer_stream_create(
 		struct lttng_consumer_channel *channel,
 		uint64_t channel_key,
@@ -432,13 +628,14 @@ struct lttng_consumer_stream *consumer_stream_create(
 		goto end;
 	}
 
+	rcu_read_lock();
+
 	if (trace_chunk && !lttng_trace_chunk_get(trace_chunk)) {
 		ERR("Failed to acquire trace chunk reference during the creation of a stream");
 		ret = -1;
 		goto error;
 	}
 
-	rcu_read_lock();
 	stream->chan = channel;
 	stream->key = stream_key;
 	stream->trace_chunk = trace_chunk;
@@ -506,6 +703,9 @@ struct lttng_consumer_stream *consumer_stream_create(
 
 	rcu_read_unlock();
 
+	lttng_dynamic_array_init(&stream->read_subbuffer_ops.post_consume_cbs,
+			sizeof(post_consume_cb), NULL);
+
 	if (type == CONSUMER_CHANNEL_TYPE_METADATA) {
 		stream->read_subbuffer_ops.lock =
 				consumer_stream_metadata_lock_all;
@@ -514,18 +714,31 @@ struct lttng_consumer_stream *consumer_stream_create(
 		stream->read_subbuffer_ops.pre_consume_subbuffer =
 				metadata_stream_check_version;
 	} else {
+		const post_consume_cb post_consume_index_op = channel->is_live ?
+				consumer_stream_sync_metadata_index :
+				consumer_stream_send_index;
+
+		ret = lttng_dynamic_array_add_element(
+				&stream->read_subbuffer_ops.post_consume_cbs,
+				&post_consume_index_op);
+		if (ret) {
+			PERROR("Failed to add `send index` callback to stream's post consumption callbacks");
+			goto error;
+		}
+
+		ret = lttng_dynamic_array_add_element(
+				&stream->read_subbuffer_ops.post_consume_cbs,
+				&(post_consume_cb) { post_consume_open_new_packet });
+		if (ret) {
+			PERROR("Failed to add `open new packet` callback to stream's post consumption callbacks");
+			goto error;
+		}
+
 		stream->read_subbuffer_ops.lock = consumer_stream_data_lock_all;
 		stream->read_subbuffer_ops.unlock =
 				consumer_stream_data_unlock_all;
 		stream->read_subbuffer_ops.pre_consume_subbuffer =
 				consumer_stream_update_stats;
-		if (channel->is_live) {
-			stream->read_subbuffer_ops.post_consume =
-					consumer_stream_sync_metadata_index;
-		} else {
-			stream->read_subbuffer_ops.post_consume =
-					consumer_stream_send_index;
-		}
 	}
 
 	if (channel->output == CONSUMER_CHANNEL_MMAP) {
@@ -541,6 +754,7 @@ struct lttng_consumer_stream *consumer_stream_create(
 error:
 	rcu_read_unlock();
 	lttng_trace_chunk_put(stream->trace_chunk);
+	lttng_dynamic_array_reset(&stream->read_subbuffer_ops.post_consume_cbs);
 	free(stream);
 end:
 	if (alloc_ret) {
@@ -561,12 +775,12 @@ void consumer_stream_relayd_close(struct lttng_consumer_stream *stream,
 {
 	int ret;
 
-	assert(stream);
-	assert(relayd);
+	LTTNG_ASSERT(stream);
+	LTTNG_ASSERT(relayd);
 
 	if (stream->sent_to_relayd) {
 		uatomic_dec(&relayd->refcount);
-		assert(uatomic_read(&relayd->refcount) >= 0);
+		LTTNG_ASSERT(uatomic_read(&relayd->refcount) >= 0);
 	}
 
 	/* Closing streams requires to lock the control socket. */
@@ -601,9 +815,9 @@ void consumer_stream_close(struct lttng_consumer_stream *stream)
 	int ret;
 	struct consumer_relayd_sock_pair *relayd;
 
-	assert(stream);
+	LTTNG_ASSERT(stream);
 
-	switch (consumer_data.type) {
+	switch (the_consumer_data.type) {
 	case LTTNG_CONSUMER_KERNEL:
 		if (stream->mmap_base != NULL) {
 			ret = munmap(stream->mmap_base, stream->mmap_len);
@@ -650,7 +864,7 @@ void consumer_stream_close(struct lttng_consumer_stream *stream)
 	}
 	default:
 		ERR("Unknown consumer_data type");
-		assert(0);
+		abort();
 	}
 
 	/* Close output fd. Could be a socket or local file at this point. */
@@ -691,16 +905,16 @@ void consumer_stream_delete(struct lttng_consumer_stream *stream,
 	int ret;
 	struct lttng_ht_iter iter;
 
-	assert(stream);
+	LTTNG_ASSERT(stream);
 	/* Should NEVER be called not in monitor mode. */
-	assert(stream->chan->monitor);
+	LTTNG_ASSERT(stream->chan->monitor);
 
 	rcu_read_lock();
 
 	if (ht) {
 		iter.iter.node = &stream->node.node;
 		ret = lttng_ht_del(ht, &iter);
-		assert(!ret);
+		LTTNG_ASSERT(!ret);
 	}
 
 	/* Delete from stream per channel ID hash table. */
@@ -711,19 +925,19 @@ void consumer_stream_delete(struct lttng_consumer_stream *stream,
 	 * that did not add the stream to a (all) hash table. Same goes for the
 	 * next call ht del call.
 	 */
-	(void) lttng_ht_del(consumer_data.stream_per_chan_id_ht, &iter);
+	(void) lttng_ht_del(the_consumer_data.stream_per_chan_id_ht, &iter);
 
 	/* Delete from the global stream list. */
 	iter.iter.node = &stream->node_session_id.node;
 	/* See the previous ht del on why we ignore the returned value. */
-	(void) lttng_ht_del(consumer_data.stream_list_ht, &iter);
+	(void) lttng_ht_del(the_consumer_data.stream_list_ht, &iter);
 
 	rcu_read_unlock();
 
 	if (!stream->metadata_flag) {
 		/* Decrement the stream count of the global consumer data. */
-		assert(consumer_data.stream_count > 0);
-		consumer_data.stream_count--;
+		LTTNG_ASSERT(the_consumer_data.stream_count > 0);
+		the_consumer_data.stream_count--;
 	}
 }
 
@@ -732,7 +946,7 @@ void consumer_stream_delete(struct lttng_consumer_stream *stream,
  */
 void consumer_stream_free(struct lttng_consumer_stream *stream)
 {
-	assert(stream);
+	LTTNG_ASSERT(stream);
 
 	metadata_bucket_destroy(stream->metadata_bucket);
 	call_rcu(&stream->node.head, free_stream_rcu);
@@ -743,9 +957,9 @@ void consumer_stream_free(struct lttng_consumer_stream *stream)
  */
 void consumer_stream_destroy_buffers(struct lttng_consumer_stream *stream)
 {
-	assert(stream);
+	LTTNG_ASSERT(stream);
 
-	switch (consumer_data.type) {
+	switch (the_consumer_data.type) {
 	case LTTNG_CONSUMER_KERNEL:
 		break;
 	case LTTNG_CONSUMER32_UST:
@@ -754,7 +968,7 @@ void consumer_stream_destroy_buffers(struct lttng_consumer_stream *stream)
 		break;
 	default:
 		ERR("Unknown consumer_data type");
-		assert(0);
+		abort();
 	}
 }
 
@@ -763,7 +977,7 @@ void consumer_stream_destroy_buffers(struct lttng_consumer_stream *stream)
  */
 static void destroy_close_stream(struct lttng_consumer_stream *stream)
 {
-	assert(stream);
+	LTTNG_ASSERT(stream);
 
 	DBG("Consumer stream destroy monitored key: %" PRIu64, stream->key);
 
@@ -782,8 +996,8 @@ static struct lttng_consumer_channel *unref_channel(
 {
 	struct lttng_consumer_channel *free_chan = NULL;
 
-	assert(stream);
-	assert(stream->chan);
+	LTTNG_ASSERT(stream);
+	LTTNG_ASSERT(stream->chan);
 
 	/* Update refcount of channel and see if we need to destroy it. */
 	if (!uatomic_sub_return(&stream->chan->refcount, 1)
@@ -805,7 +1019,7 @@ static struct lttng_consumer_channel *unref_channel(
 void consumer_stream_destroy(struct lttng_consumer_stream *stream,
 		struct lttng_ht *ht)
 {
-	assert(stream);
+	LTTNG_ASSERT(stream);
 
 	/* Stream is in monitor mode. */
 	if (stream->monitor) {
@@ -817,7 +1031,7 @@ void consumer_stream_destroy(struct lttng_consumer_stream *stream,
 		 * stream thus being globally visible.
 		 */
 		if (stream->globally_visible) {
-			pthread_mutex_lock(&consumer_data.lock);
+			pthread_mutex_lock(&the_consumer_data.lock);
 			pthread_mutex_lock(&stream->chan->lock);
 			pthread_mutex_lock(&stream->lock);
 			/* Remove every reference of the stream in the consumer. */
@@ -829,11 +1043,11 @@ void consumer_stream_destroy(struct lttng_consumer_stream *stream,
 			free_chan = unref_channel(stream);
 
 			/* Indicates that the consumer data state MUST be updated after this. */
-			consumer_data.need_update = 1;
+			the_consumer_data.need_update = 1;
 
 			pthread_mutex_unlock(&stream->lock);
 			pthread_mutex_unlock(&stream->chan->lock);
-			pthread_mutex_unlock(&consumer_data.lock);
+			pthread_mutex_unlock(&the_consumer_data.lock);
 		} else {
 			/*
 			 * If the stream is not visible globally, this needs to be done
@@ -852,6 +1066,7 @@ void consumer_stream_destroy(struct lttng_consumer_stream *stream,
 	/* Free stream within a RCU call. */
 	lttng_trace_chunk_put(stream->trace_chunk);
 	stream->trace_chunk = NULL;
+	lttng_dynamic_array_reset(&stream->read_subbuffer_ops.post_consume_cbs);
 	consumer_stream_free(stream);
 }
 
@@ -865,8 +1080,8 @@ int consumer_stream_write_index(struct lttng_consumer_stream *stream,
 {
 	int ret;
 
-	assert(stream);
-	assert(element);
+	LTTNG_ASSERT(stream);
+	LTTNG_ASSERT(element);
 
 	rcu_read_lock();
 	if (stream->net_seq_idx != (uint64_t) -1ULL) {
@@ -917,7 +1132,7 @@ int consumer_stream_create_output_files(struct lttng_consumer_stream *stream,
 	char stream_path[LTTNG_PATH_MAX];
 
 	ASSERT_LOCKED(stream->lock);
-	assert(stream->trace_chunk);
+	LTTNG_ASSERT(stream->trace_chunk);
 
 	ret = utils_stream_file_path(stream->chan->pathname, stream->name,
 			stream->chan->tracefile_size,
@@ -935,16 +1150,16 @@ int consumer_stream_create_output_files(struct lttng_consumer_stream *stream,
 			goto end;
 		}
 		stream->out_fd = -1;
-        }
+	}
 
 	DBG("Opening stream output file \"%s\"", stream_path);
 	chunk_status = lttng_trace_chunk_open_file(stream->trace_chunk, stream_path,
 			flags, mode, &stream->out_fd, false);
-        if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
+	if (chunk_status != LTTNG_TRACE_CHUNK_STATUS_OK) {
 		ERR("Failed to open stream file \"%s\"", stream->name);
 		ret = -1;
 		goto end;
-        }
+	}
 
 	if (!stream->metadata_flag && (create_index || stream->index_file)) {
 		if (stream->index_file) {
@@ -997,7 +1212,7 @@ bool consumer_stream_is_deleted(struct lttng_consumer_stream *stream)
 	 * This function does not take a const stream since
 	 * cds_lfht_is_node_deleted was not const before liburcu 0.12.
 	 */
-	assert(stream);
+	LTTNG_ASSERT(stream);
 	return cds_lfht_is_node_deleted(&stream->node.node);
 }
 
@@ -1041,9 +1256,9 @@ int consumer_stream_enable_metadata_bucketization(
 {
 	int ret = 0;
 
-	assert(stream->metadata_flag);
-	assert(!stream->metadata_bucket);
-	assert(stream->chan->output == CONSUMER_CHANNEL_MMAP);
+	LTTNG_ASSERT(stream->metadata_flag);
+	LTTNG_ASSERT(!stream->metadata_bucket);
+	LTTNG_ASSERT(stream->chan->output == CONSUMER_CHANNEL_MMAP);
 
 	stream->metadata_bucket = metadata_bucket_create(
 			metadata_bucket_flush, stream);
@@ -1060,7 +1275,7 @@ end:
 void consumer_stream_metadata_set_version(
 		struct lttng_consumer_stream *stream, uint64_t new_version)
 {
-	assert(new_version > stream->metadata_version);
+	LTTNG_ASSERT(new_version > stream->metadata_version);
 	stream->metadata_version = new_version;
 	stream->reset_metadata_flag = 1;
 
@@ -1068,3 +1283,47 @@ void consumer_stream_metadata_set_version(
 		metadata_bucket_reset(stream->metadata_bucket);
 	}
 }
+
+int consumer_stream_flush_buffer(struct lttng_consumer_stream *stream,
+		bool producer_active)
+{
+	int ret = 0;
+
+	switch (the_consumer_data.type) {
+	case LTTNG_CONSUMER_KERNEL:
+		if (producer_active) {
+			ret = kernctl_buffer_flush(stream->wait_fd);
+			if (ret < 0) {
+				ERR("Failed to flush kernel stream");
+				goto end;
+			}
+		} else {
+			ret = kernctl_buffer_flush_empty(stream->wait_fd);
+			if (ret < 0) {
+				/*
+				 * Doing a buffer flush which does not take into
+				 * account empty packets. This is not perfect,
+				 * but required as a fall-back when
+				 * "flush_empty" is not implemented by
+				 * lttng-modules.
+				 */
+				ret = kernctl_buffer_flush(stream->wait_fd);
+				if (ret < 0) {
+					ERR("Failed to flush kernel stream");
+					goto end;
+				}
+			}
+		}
+		break;
+	case LTTNG_CONSUMER32_UST:
+	case LTTNG_CONSUMER64_UST:
+		ret = lttng_ustconsumer_flush_buffer(stream, (int) producer_active);
+		break;
+	default:
+		ERR("Unknown consumer_data type");
+		abort();
+	}
+
+end:
+	return ret;
+}