X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fconsumer%2Fconsumer.c;h=d45771e7eb19fb45123041736853143704db3aff;hp=f13e90a6881ac4ab87312c61a7e632004d97e81d;hb=7a86c13d20ba17a39767f808e6f0ea363430eeda;hpb=23d565989350c270c68e9a6c8edfbe2dd6a6895d

diff --git a/src/common/consumer/consumer.c b/src/common/consumer/consumer.c
index f13e90a68..d45771e7e 100644
--- a/src/common/consumer/consumer.c
+++ b/src/common/consumer/consumer.c
@@ -7,6 +7,7 @@
  *
  */
 
+#include "common/index/ctf-index.h"
 #define _LGPL_SOURCE
 #include <assert.h>
 #include <poll.h>
@@ -568,98 +569,6 @@ void consumer_stream_update_channel_attributes(
 			channel->tracefile_size;
 }
 
-struct lttng_consumer_stream *consumer_allocate_stream(
-		struct lttng_consumer_channel *channel,
-		uint64_t channel_key,
-		uint64_t stream_key,
-		const char *channel_name,
-		uint64_t relayd_id,
-		uint64_t session_id,
-		struct lttng_trace_chunk *trace_chunk,
-		int cpu,
-		int *alloc_ret,
-		enum consumer_channel_type type,
-		unsigned int monitor)
-{
-	int ret;
-	struct lttng_consumer_stream *stream;
-
-	stream = zmalloc(sizeof(*stream));
-	if (stream == NULL) {
-		PERROR("malloc struct lttng_consumer_stream");
-		ret = -ENOMEM;
-		goto end;
-	}
-
-	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;
-	stream->out_fd = -1;
-	stream->out_fd_offset = 0;
-	stream->output_written = 0;
-	stream->net_seq_idx = relayd_id;
-	stream->session_id = session_id;
-	stream->monitor = monitor;
-	stream->endpoint_status = CONSUMER_ENDPOINT_ACTIVE;
-	stream->index_file = NULL;
-	stream->last_sequence_number = -1ULL;
-	stream->rotate_position = -1ULL;
-	pthread_mutex_init(&stream->lock, NULL);
-	pthread_mutex_init(&stream->metadata_timer_lock, NULL);
-
-	/* If channel is the metadata, flag this stream as metadata. */
-	if (type == CONSUMER_CHANNEL_TYPE_METADATA) {
-		stream->metadata_flag = 1;
-		/* Metadata is flat out. */
-		strncpy(stream->name, DEFAULT_METADATA_NAME, sizeof(stream->name));
-		/* Live rendez-vous point. */
-		pthread_cond_init(&stream->metadata_rdv, NULL);
-		pthread_mutex_init(&stream->metadata_rdv_lock, NULL);
-	} else {
-		/* Format stream name to <channel_name>_<cpu_number> */
-		ret = snprintf(stream->name, sizeof(stream->name), "%s_%d",
-				channel_name, cpu);
-		if (ret < 0) {
-			PERROR("snprintf stream name");
-			goto error;
-		}
-	}
-
-	/* Key is always the wait_fd for streams. */
-	lttng_ht_node_init_u64(&stream->node, stream->key);
-
-	/* Init node per channel id key */
-	lttng_ht_node_init_u64(&stream->node_channel_id, channel_key);
-
-	/* Init session id node with the stream session id */
-	lttng_ht_node_init_u64(&stream->node_session_id, stream->session_id);
-
-	DBG3("Allocated stream %s (key %" PRIu64 ", chan_key %" PRIu64
-			" relayd_id %" PRIu64 ", session_id %" PRIu64,
-			stream->name, stream->key, channel_key,
-			stream->net_seq_idx, stream->session_id);
-
-	rcu_read_unlock();
-	return stream;
-
-error:
-	rcu_read_unlock();
-	lttng_trace_chunk_put(stream->trace_chunk);
-	free(stream);
-end:
-	if (alloc_ret) {
-		*alloc_ret = ret;
-	}
-	return NULL;
-}
-
 /*
  * Add a stream to the global list protected by a mutex.
  */
@@ -1467,7 +1376,7 @@ void lttng_consumer_sync_trace_file(struct lttng_consumer_stream *stream,
 struct lttng_consumer_local_data *lttng_consumer_create(
 		enum lttng_consumer_type type,
 		ssize_t (*buffer_ready)(struct lttng_consumer_stream *stream,
-			struct lttng_consumer_local_data *ctx),
+			struct lttng_consumer_local_data *ctx, bool locked_by_caller),
 		int (*recv_channel)(struct lttng_consumer_channel *channel),
 		int (*recv_stream)(struct lttng_consumer_stream *stream),
 		int (*update_stream)(uint64_t stream_key, uint32_t state))
@@ -1674,11 +1583,9 @@ end:
  * Returns the number of bytes written
  */
 ssize_t lttng_consumer_on_read_subbuffer_mmap(
-		struct lttng_consumer_local_data *ctx,
 		struct lttng_consumer_stream *stream,
 		const struct lttng_buffer_view *buffer,
-		unsigned long padding,
-		struct ctf_packet_index *index)
+		unsigned long padding)
 {
 	ssize_t ret = 0;
 	off_t orig_offset = stream->out_fd_offset;
@@ -1770,10 +1677,6 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap(
 			orig_offset = 0;
 		}
 		stream->tracefile_size_current += buffer->size;
-		if (index) {
-			index->offset = htobe64(stream->out_fd_offset);
-		}
-
 		write_len = buffer->size;
 	}
 
@@ -1782,7 +1685,7 @@ ssize_t lttng_consumer_on_read_subbuffer_mmap(
 	 * receive a ret value that is bigger than len.
 	 */
 	ret = lttng_write(outfd, buffer->data, write_len);
-	DBG("Consumer mmap write() ret %zd (len %lu)", ret, write_len);
+	DBG("Consumer mmap write() ret %zd (len %zu)", ret, write_len);
 	if (ret < 0 || ((size_t) ret != write_len)) {
 		/*
 		 * Report error to caller if nothing was written else at least send the
@@ -1850,8 +1753,7 @@ end:
 ssize_t lttng_consumer_on_read_subbuffer_splice(
 		struct lttng_consumer_local_data *ctx,
 		struct lttng_consumer_stream *stream, unsigned long len,
-		unsigned long padding,
-		struct ctf_packet_index *index)
+		unsigned long padding)
 {
 	ssize_t ret = 0, written = 0, ret_splice = 0;
 	loff_t offset = 0;
@@ -1955,7 +1857,6 @@ ssize_t lttng_consumer_on_read_subbuffer_splice(
 			orig_offset = 0;
 		}
 		stream->tracefile_size_current += len;
-		index->offset = htobe64(stream->out_fd_offset);
 	}
 
 	while (len > 0) {
@@ -2522,7 +2423,7 @@ restart:
 				do {
 					health_code_update();
 
-					len = ctx->on_buffer_ready(stream, ctx);
+					len = ctx->on_buffer_ready(stream, ctx, false);
 					/*
 					 * We don't check the return value here since if we get
 					 * a negative len, it means an error occurred thus we
@@ -2549,7 +2450,7 @@ restart:
 					do {
 						health_code_update();
 
-						len = ctx->on_buffer_ready(stream, ctx);
+						len = ctx->on_buffer_ready(stream, ctx, false);
 						/*
 						 * We don't check the return value here since if we get
 						 * a negative len, it means an error occurred thus we
@@ -2765,7 +2666,7 @@ void *consumer_thread_data_poll(void *data)
 			if (pollfd[i].revents & POLLPRI) {
 				DBG("Urgent read on fd %d", pollfd[i].fd);
 				high_prio = 1;
-				len = ctx->on_buffer_ready(local_stream[i], ctx);
+				len = ctx->on_buffer_ready(local_stream[i], ctx, false);
 				/* it's ok to have an unavailable sub-buffer */
 				if (len < 0 && len != -EAGAIN && len != -ENODATA) {
 					/* Clean the stream and free it. */
@@ -2796,7 +2697,7 @@ void *consumer_thread_data_poll(void *data)
 					local_stream[i]->hangup_flush_done ||
 					local_stream[i]->has_data) {
 				DBG("Normal read on fd %d", pollfd[i].fd);
-				len = ctx->on_buffer_ready(local_stream[i], ctx);
+				len = ctx->on_buffer_ready(local_stream[i], ctx, false);
 				/* it's ok to have an unavailable sub-buffer */
 				if (len < 0 && len != -EAGAIN && len != -ENODATA) {
 					/* Clean the stream and free it. */
@@ -3409,16 +3310,46 @@ error_testpoint:
 	return NULL;
 }
 
-ssize_t lttng_consumer_read_subbuffer(struct lttng_consumer_stream *stream,
+static int post_consume(struct lttng_consumer_stream *stream,
+		const struct stream_subbuffer *subbuffer,
 		struct lttng_consumer_local_data *ctx)
 {
-	ssize_t ret;
+	size_t i;
+	int ret = 0;
+	const size_t count = lttng_dynamic_array_get_count(
+			&stream->read_subbuffer_ops.post_consume_cbs);
+
+	for (i = 0; i < count; i++) {
+		const post_consume_cb op = *(post_consume_cb *) lttng_dynamic_array_get_element(
+				&stream->read_subbuffer_ops.post_consume_cbs,
+				i);
+
+		ret = op(stream, subbuffer, ctx);
+		if (ret) {
+			goto end;
+		}
+	}
+end:
+	return ret;
+}
+
+ssize_t lttng_consumer_read_subbuffer(struct lttng_consumer_stream *stream,
+		struct lttng_consumer_local_data *ctx,
+		bool locked_by_caller)
+{
+	ssize_t ret, written_bytes = 0;
 	int rotation_ret;
+	struct stream_subbuffer subbuffer = {};
 
-	pthread_mutex_lock(&stream->chan->lock);
-	pthread_mutex_lock(&stream->lock);
-	if (stream->metadata_flag) {
-		pthread_mutex_lock(&stream->metadata_rdv_lock);
+	if (!locked_by_caller) {
+		stream->read_subbuffer_ops.lock(stream);
+	}
+
+	if (stream->read_subbuffer_ops.on_wake_up) {
+		ret = stream->read_subbuffer_ops.on_wake_up(stream);
+		if (ret) {
+			goto end;
+		}
 	}
 
 	/*
@@ -3434,22 +3365,51 @@ ssize_t lttng_consumer_read_subbuffer(struct lttng_consumer_stream *stream,
 		}
 	}
 
-	switch (consumer_data.type) {
-	case LTTNG_CONSUMER_KERNEL:
-		ret = lttng_kconsumer_read_subbuffer(stream, ctx);
-		break;
-	case LTTNG_CONSUMER32_UST:
-	case LTTNG_CONSUMER64_UST:
-		ret = lttng_ustconsumer_read_subbuffer(stream, ctx);
-		break;
-	default:
-		ERR("Unknown consumer_data type");
-		assert(0);
-		ret = -ENOSYS;
-		break;
+	ret = stream->read_subbuffer_ops.get_next_subbuffer(stream, &subbuffer);
+	if (ret) {
+		if (ret == -ENODATA) {
+			/* Not an error. */
+			ret = 0;
+			goto sleep_stream;
+		}
+		goto end;
 	}
 
-	if (ret < 0) {
+	ret = stream->read_subbuffer_ops.pre_consume_subbuffer(
+			stream, &subbuffer);
+	if (ret) {
+		goto error_put_subbuf;
+	}
+
+	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);
+	}
+
+	ret = stream->read_subbuffer_ops.put_next_subbuffer(stream, &subbuffer);
+	if (ret) {
+		goto end;
+	}
+
+	ret = post_consume(stream, &subbuffer, ctx);
+	if (ret) {
 		goto end;
 	}
 
@@ -3468,20 +3428,28 @@ ssize_t lttng_consumer_read_subbuffer(struct lttng_consumer_stream *stream,
 			ERR("Stream rotation error after consuming data");
 			goto end;
 		}
+
 	} else if (rotation_ret < 0) {
 		ret = rotation_ret;
 		ERR("Failed to check if stream was ready to rotate after consuming data");
 		goto end;
 	}
 
+sleep_stream:
+	if (stream->read_subbuffer_ops.on_sleep) {
+		stream->read_subbuffer_ops.on_sleep(stream, ctx);
+	}
+
+	ret = written_bytes;
 end:
-	if (stream->metadata_flag) {
-		pthread_cond_broadcast(&stream->metadata_rdv);
-		pthread_mutex_unlock(&stream->metadata_rdv_lock);
+	if (!locked_by_caller) {
+		stream->read_subbuffer_ops.unlock(stream);
 	}
-	pthread_mutex_unlock(&stream->lock);
-	pthread_mutex_unlock(&stream->chan->lock);
+
 	return ret;
+error_put_subbuf:
+	(void) stream->read_subbuffer_ops.put_next_subbuffer(stream, &subbuffer);
+	goto end;
 }
 
 int lttng_consumer_on_recv_stream(struct lttng_consumer_stream *stream)
@@ -3955,50 +3923,6 @@ unsigned long consumer_get_consume_start_pos(unsigned long consumed_pos,
 	return start_pos;
 }
 
-static
-int consumer_flush_buffer(struct lttng_consumer_stream *stream, int producer_active)
-{
-	int ret = 0;
-
-	switch (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:
-		lttng_ustconsumer_flush_buffer(stream, producer_active);
-		break;
-	default:
-		ERR("Unknown consumer_data type");
-		abort();
-	}
-
-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
@@ -4022,11 +3946,15 @@ int lttng_consumer_rotate_channel(struct lttng_consumer_channel *channel,
 	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();
 
@@ -4068,7 +3996,8 @@ int lttng_consumer_rotate_channel(struct lttng_consumer_channel *channel,
 			 * ensures we have at least one packet in each stream per trace
 			 * chunk, even if no data was produced.
 			 */
-			ret = consumer_flush_buffer(stream, stream->metadata_flag ? 1 : 0);
+			ret = consumer_stream_flush_buffer(
+					stream, stream->metadata_flag ? 1 : 0);
 			if (ret < 0) {
 				ERR("Failed to flush stream %" PRIu64 " during channel rotation",
 						stream->key);
@@ -4156,36 +4085,135 @@ int lttng_consumer_rotate_channel(struct lttng_consumer_channel *channel,
 			}
 			stream_count++;
 		}
+
+		stream->opened_packet_in_current_trace_chunk = false;
+
+		if (rotating_to_new_chunk && !stream->metadata_flag) {
+			/*
+			 * Attempt to flush an empty packet as close to the
+			 * rotation point as possible. In the event where a
+			 * stream remains inactive after the rotation point,
+			 * this ensures that the new trace chunk has a
+			 * beginning timestamp set at the begining of the
+			 * trace chunk instead of only creating an empty
+			 * packet when the trace chunk is stopped.
+			 *
+			 * This indicates to the viewers that the stream
+			 * was being recorded, but more importantly it
+			 * allows viewers to determine a useable trace
+			 * intersection.
+			 *
+			 * This presents a problem in the case where the
+			 * ring-buffer is completely full.
+			 *
+			 * Consider the following scenario:
+			 *   - The consumption of data is slow (slow network,
+			 *     for instance),
+			 *   - The ring buffer is full,
+			 *   - A rotation is initiated,
+			 *     - The flush below does nothing (no space left to
+			 *       open a new packet),
+			 *   - The other streams rotate very soon, and new
+			 *     data is produced in the new chunk,
+			 *   - This stream completes its rotation long after the
+			 *     rotation was initiated
+			 *   - The session is stopped before any event can be
+			 *     produced	in this stream's buffers.
+			 *
+			 * The resulting trace chunk will have a single packet
+			 * temporaly at the end of the trace chunk for this
+			 * stream making the stream intersection more narrow
+			 * than it should be.
+			 *
+			 * To work-around this, an empty flush is performed
+			 * after the first consumption of a packet during a
+			 * rotation if open_packet fails. The idea is that
+			 * consuming a packet frees enough space to switch
+			 * 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.
+			 */
+			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;
+			}
+		}
+
 		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;
 
@@ -4196,6 +4224,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;
 }
 
@@ -4254,7 +4283,7 @@ int consumer_clear_stream(struct lttng_consumer_stream *stream)
 {
 	int ret;
 
-	ret = consumer_flush_buffer(stream, 1);
+	ret = consumer_stream_flush_buffer(stream, 1);
 	if (ret < 0) {
 		ERR("Failed to flush stream %" PRIu64 " during channel clear",
 				stream->key);
@@ -5011,3 +5040,70 @@ int lttng_consumer_clear_channel(struct lttng_consumer_channel *channel)
 end:
 	return ret;
 }
+
+enum lttcomm_return_code lttng_consumer_open_channel_packets(
+		struct lttng_consumer_channel *channel)
+{
+	struct lttng_consumer_stream *stream;
+	enum lttcomm_return_code ret = LTTCOMM_CONSUMERD_SUCCESS;
+
+	if (channel->metadata_stream) {
+		ERR("Open channel packets command attempted on a metadata channel");
+		ret = LTTCOMM_CONSUMERD_INVALID_PARAMETERS;
+		goto end;
+	}
+
+	rcu_read_lock();
+	cds_list_for_each_entry(stream, &channel->streams.head, send_node) {
+		enum consumer_stream_open_packet_status status;
+
+		pthread_mutex_lock(&stream->lock);
+		if (cds_lfht_is_node_deleted(&stream->node.node)) {
+			goto next;
+		}
+
+		status = consumer_stream_open_packet(stream);
+		switch (status) {
+		case CONSUMER_STREAM_OPEN_PACKET_STATUS_OPENED:
+			DBG("Opened a packet in \"open channel packets\" command: 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:
+			DBG("No space left to open a packet in \"open channel packets\" command: 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:
+			/*
+			 * Only unexpected internal errors can lead to this
+			 * failing. Report an unknown error.
+			 */
+			ERR("Failed to flush empty buffer in \"open channel packets\" command: stream id = %" PRIu64
+					", channel id = %" PRIu64
+					", channel name = %s"
+					", session id = %" PRIu64,
+					stream->key, channel->key,
+					channel->name, channel->session_id);
+			ret = LTTCOMM_CONSUMERD_UNKNOWN_ERROR;
+			goto error_unlock;
+		default:
+			abort();
+		}
+
+	next:
+		pthread_mutex_unlock(&stream->lock);
+	}
+
+end_rcu_unlock:
+	rcu_read_unlock();
+end:
+	return ret;
+
+error_unlock:
+	pthread_mutex_unlock(&stream->lock);
+	goto end_rcu_unlock;
+}