+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;
+}
+
+static enum open_packet_status open_packet(struct lttng_consumer_stream *stream)
+{
+ int ret;
+ enum 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 = 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 = OPEN_PACKET_STATUS_ERROR;
+ goto end;
+ }
+
+ ret = consumer_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 = 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 = 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 = 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 ?
+ OPEN_PACKET_STATUS_OPENED :
+ OPEN_PACKET_STATUS_NO_SPACE;
+end:
+ return status;
+}
+
+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;
+}
+