#include "session.h"
#include "stream.h"
#include "connection.h"
+#include "tracefile-array.h"
/* command line options */
char *opt_output_path;
* Only flag a stream inactive when it has already
* received data and no indexes are in flight.
*/
- if (stream->total_index_received > 0
+ if (stream->index_received_seqcount > 0
&& stream->indexes_in_flight == 0) {
stream->beacon_ts_end =
be64toh(index_info.timestamp_end);
}
ret = relay_index_try_flush(index);
if (ret == 0) {
- stream->total_index_received++;
+ tracefile_array_commit_seq(stream->tfa);
+ stream->index_received_seqcount++;
} else if (ret > 0) {
/* no flush. */
ret = 0;
fd = index_create_file(stream->path_name, stream->channel_name,
-1, -1, stream->tracefile_size,
- stream->current_tracefile_id);
+ tracefile_array_get_file_index_head(stream->tfa));
if (fd < 0) {
ret = -1;
/* Put self-ref for this index due to error. */
ret = relay_index_try_flush(index);
if (ret == 0) {
- stream->total_index_received++;
+ tracefile_array_commit_seq(stream->tfa);
+ stream->index_received_seqcount++;
} else if (ret > 0) {
/* No flush. */
ret = 0;
uint64_t net_seq_num;
uint32_t data_size;
struct relay_session *session;
+ bool new_stream = false;
ret = conn->sock->ops->recvmsg(conn->sock, &data_hdr,
sizeof(struct lttcomm_relayd_data_hdr), 0);
if (stream->tracefile_size > 0 &&
(stream->tracefile_size_current + data_size) >
stream->tracefile_size) {
- uint64_t new_id;
+ uint64_t old_id, new_id;
+
+ old_id = tracefile_array_get_file_index_head(stream->tfa);
+ tracefile_array_file_rotate(stream->tfa);
+
+ /* new_id is updated by utils_rotate_stream_file. */
+ new_id = old_id;
- new_id = (stream->current_tracefile_id + 1) %
- stream->tracefile_count;
- /*
- * Move viewer oldest available data position forward if
- * we are overwriting a tracefile.
- */
- if (new_id == stream->oldest_tracefile_id) {
- stream->oldest_tracefile_id =
- (stream->oldest_tracefile_id + 1) %
- stream->tracefile_count;
- }
ret = utils_rotate_stream_file(stream->path_name,
stream->channel_name, stream->tracefile_size,
stream->tracefile_count, -1,
-1, stream->stream_fd->fd,
- &stream->current_tracefile_id,
- &stream->stream_fd->fd);
+ &new_id, &stream->stream_fd->fd);
if (ret < 0) {
ERR("Rotating stream output file");
goto end_stream_unlock;
}
- stream->current_tracefile_seq++;
- if (stream->current_tracefile_seq
- - stream->oldest_tracefile_seq >=
- stream->tracefile_count) {
- stream->oldest_tracefile_seq++;
- }
/*
* Reset current size because we just performed a stream
* rotation.
}
stream->tracefile_size_current +=
data_size + be32toh(data_hdr.padding_size);
+ if (stream->prev_seq == -1ULL) {
+ new_stream = true;
+ }
+
stream->prev_seq = net_seq_num;
end_stream_unlock:
pthread_mutex_unlock(&stream->lock);
+ if (new_stream) {
+ pthread_mutex_lock(&session->lock);
+ uatomic_set(&session->new_streams, 1);
+ pthread_mutex_unlock(&session->lock);
+ }
end_stream_put:
stream_put(stream);
end: