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.
*
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);
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);
struct relay_stream *stream;
ssize_t send_ret;
int ret;
+ uint64_t stream_seq;
DBG("Data pending command received");
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 {
}
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);
tracefile_array_commit_seq(stream->tfa);
stream->index_received_seqcount++;
stream->pos_after_last_complete_data_index += index->total_size;
+ stream->prev_index_seq = index_info.net_seq_num;
} else if (ret > 0) {
/* no flush. */
ret = 0;
}
- 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) {
if (index_flushed) {
stream->pos_after_last_complete_data_index =
stream->tracefile_size_current;
+ stream->prev_index_seq = state->header.net_seq_num;
}
stream->prev_seq = state->header.net_seq_num;