Fix: take index seq number into account for rotation pending check

[lttng-tools.git] / src / bin / lttng-relayd / main.c

diff --git a/src/bin/lttng-relayd/main.c b/src/bin/lttng-relayd/main.c
index 92f9a5600ac28b7280dcea9f1041b7725b13b748..f111b931c7792c3b3f627977bb7ed62e6999c1a4 100644 (file)
--- a/src/bin/lttng-relayd/main.c
+++ b/src/bin/lttng-relayd/main.c
@@ -1083,6 +1083,11 @@ static int set_index_control_data(struct relay_index *index,
        return relay_index_set_data(index, &index_data);
 }
 
+static bool session_streams_have_index(const struct relay_session *session)
+{
+       return session->minor >= 4 && !session->snapshot;
+}
+
 /*
  * Handle the RELAYD_CREATE_SESSION command.
  *
@@ -1741,18 +1746,27 @@ static
 int try_rotate_stream(struct relay_stream *stream)
 {
        int ret = 0;
+       uint64_t trace_seq;
 
        /* No rotation expected. */
        if (stream->rotate_at_seq_num == -1ULL) {
                goto end;
        }
 
-       if (stream->prev_seq < stream->rotate_at_seq_num ||
-                       stream->prev_seq == -1ULL) {
-               DBG("Stream %" PRIu64 " no yet ready for rotation",
-                               stream->stream_handle);
+       trace_seq = min(stream->prev_seq, stream->prev_index_seq);
+       if (stream->prev_seq == -1ULL || stream->prev_index_seq == -1ULL ||
+                       trace_seq < stream->rotate_at_seq_num) {
+               DBG("Stream %" PRIu64 " not yet ready for rotation (rotate_at_seq_num = %" PRIu64 ", prev_seq = %" PRIu64 ", prev_index_seq = %" PRIu64 ")",
+                               stream->stream_handle,
+                               stream->rotate_at_seq_num,
+                               stream->prev_seq,
+                               stream->prev_index_seq);
                goto end;
        } else if (stream->prev_seq > stream->rotate_at_seq_num) {
+               /*
+                * prev_seq is checked here since indexes and rotation
+                * commands are serialized with respect to each other.
+                */
                DBG("Rotation after too much data has been written in tracefile "
                                "for stream %" PRIu64 ", need to truncate before "
                                "rotating", stream->stream_handle);
@@ -1762,7 +1776,20 @@ int try_rotate_stream(struct relay_stream *stream)
                        goto end;
                }
        } else {
-               /* stream->prev_seq == stream->rotate_at_seq_num */
+               if (trace_seq != stream->rotate_at_seq_num) {
+                       /*
+                        * Unexpected, protocol error/bug.
+                        * It could mean that we received a rotation position
+                        * that is in the past.
+                        */
+                       ERR("Stream %" PRIu64 " is in an inconsistent state (rotate_at_seq_num = %" PRIu64 ", prev_seq = %" PRIu64 ", prev_index_seq = %" PRIu64 ")",
+                               stream->stream_handle,
+                               stream->rotate_at_seq_num,
+                               stream->prev_seq,
+                               stream->prev_index_seq);
+                       ret = -1;
+                       goto end;
+               }
                DBG("Stream %" PRIu64 " ready for rotation",
                                stream->stream_handle);
                ret = do_rotate_stream(stream);
@@ -1931,6 +1958,7 @@ static int relay_data_pending(const struct lttcomm_relayd_hdr *recv_hdr,
        struct relay_stream *stream;
        ssize_t send_ret;
        int ret;
+       uint64_t stream_seq;
 
        DBG("Data pending command received");
 
@@ -1958,12 +1986,23 @@ static int relay_data_pending(const struct lttcomm_relayd_hdr *recv_hdr,
 
        pthread_mutex_lock(&stream->lock);
 
-       DBG("Data pending for stream id %" PRIu64 " prev_seq %" PRIu64
-                       " and last_seq %" PRIu64, msg.stream_id,
-                       stream->prev_seq, msg.last_net_seq_num);
+       if (session_streams_have_index(session)) {
+               /*
+                * Ensure that both the index and stream data have been
+                * flushed up to the requested point.
+                */
+               stream_seq = min(stream->prev_seq, stream->prev_index_seq);
+       } else {
+               stream_seq = stream->prev_seq;
+       }
+       DBG("Data pending for stream id %" PRIu64 ": prev_seq %" PRIu64
+                       ", prev_index_seq %" PRIu64
+                       ", and last_seq %" PRIu64, msg.stream_id,
+                       stream->prev_seq, stream->prev_index_seq,
+                       msg.last_net_seq_num);
 
        /* Avoid wrapping issue */
-       if (((int64_t) (stream->prev_seq - msg.last_net_seq_num)) >= 0) {
+       if (((int64_t) (stream_seq - msg.last_net_seq_num)) >= 0) {
                /* Data has in fact been written and is NOT pending */
                ret = 0;
        } else {
@@ -2183,7 +2222,18 @@ static int relay_end_data_pending(const struct lttcomm_relayd_hdr *recv_hdr,
                }
                pthread_mutex_lock(&stream->lock);
                if (!stream->data_pending_check_done) {
-                       if (!stream->closed || !(((int64_t) (stream->prev_seq - stream->last_net_seq_num)) >= 0)) {
+                       uint64_t stream_seq;
+
+                       if (session_streams_have_index(conn->session)) {
+                               /*
+                                * Ensure that both the index and stream data have been
+                                * flushed up to the requested point.
+                                */
+                               stream_seq = min(stream->prev_seq, stream->prev_index_seq);
+                       } else {
+                               stream_seq = stream->prev_seq;
+                       }
+                       if (!stream->closed || !(((int64_t) (stream_seq - stream->last_net_seq_num)) >= 0)) {
                                is_data_inflight = 1;
                                DBG("Data is still in flight for stream %" PRIu64,
                                                stream->stream_handle);
@@ -3462,7 +3512,7 @@ static enum relay_connection_status relay_process_data_receive_payload(
        }
 
 
-       if (session->minor >= 4 && !session->snapshot) {
+       if (session_streams_have_index(session)) {
                ret = handle_index_data(stream, state->header.net_seq_num,
                                state->rotate_index, &index_flushed, state->header.data_size + state->header.padding_size);
                if (ret < 0) {