X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fbin%2Flttng-relayd%2Fstream.h;h=5e23e7339cf4fcf7495f2ae4825fcdfdd57641d1;hp=7e2b1334ec9c54e3e3ca3c5229b4e664c31c06b8;hb=c6db3843828a8fbf08444a2bc4191291a4807936;hpb=7591bab11eceedc6a0d1e02fd6f85592267a63b5 diff --git a/src/bin/lttng-relayd/stream.h b/src/bin/lttng-relayd/stream.h index 7e2b1334e..5e23e7339 100644 --- a/src/bin/lttng-relayd/stream.h +++ b/src/bin/lttng-relayd/stream.h @@ -29,6 +29,12 @@ #include "session.h" #include "stream-fd.h" +#include "tracefile-array.h" + +struct relay_stream_chunk_id { + bool is_set; + uint64_t value; +}; /* * Represents a stream in the relay @@ -36,12 +42,6 @@ struct relay_stream { uint64_t stream_handle; - /* - * reflock used to synchronize the closing of this stream. - * stream reflock nests inside viewer stream reflock. - * stream reflock nests inside index reflock. - */ - pthread_mutex_t reflock; struct urcu_ref ref; /* Back reference to trace. Protected by refcount on trace object. */ struct ctf_trace *trace; @@ -52,32 +52,63 @@ struct relay_stream { * inside the ctf_trace lock. */ pthread_mutex_t lock; - uint64_t prev_seq; /* previous data sequence number encountered. */ - uint64_t last_net_seq_num; /* seq num to encounter before closing. */ + /* previous data sequence number written to disk. */ + uint64_t prev_data_seq; + /* previous index sequence number written to disk. */ + uint64_t prev_index_seq; + /* seq num to encounter before closing. */ + uint64_t last_net_seq_num; /* FD on which to write the stream data. */ struct stream_fd *stream_fd; - /* FD on which to write the index data. */ - struct stream_fd *index_fd; + /* index file on which to write the index data. */ + struct lttng_index_file *index_file; char *path_name; + /* + * prev_path_name is only used for session rotation support. + * It is essentially used to work around the fact that index + * files are always created from the 'data' connection. + * + * Hence, it is possible to receive a ROTATE_STREAM command + * which affects the stream's path_name before the creation of + * an index file. In this situation, the index file of the + * 'previous' chunk would be created in the new destination folder. + * + * It would then be unlinked when the actual index of the new chunk + * is created. + */ + char *prev_path_name; char *channel_name; /* On-disk circular buffer of tracefiles. */ uint64_t tracefile_size; uint64_t tracefile_size_current; uint64_t tracefile_count; - uint64_t current_tracefile_id; - uint64_t current_tracefile_seq; /* Free-running counter. */ - uint64_t oldest_tracefile_seq; /* Free-running counter. */ + /* + * Position in the tracefile where we have the full index also on disk. + */ + uint64_t pos_after_last_complete_data_index; - /* To inform the viewer up to where it can go back in time. */ - uint64_t oldest_tracefile_id; + /* + * Counts the number of received indexes. The "tag" associated + * with an index is taken before incrementing this seqcount. + * Therefore, the sequence tag associated with the last index + * received is always index_received_seqcount - 1. + */ + uint64_t index_received_seqcount; - uint64_t total_index_received; + /* + * Tracefile array is an index of the stream trace files, + * indexed by position. It allows keeping track of the oldest + * available indexes when overwriting trace files in tracefile + * rotation. + */ + struct tracefile_array *tfa; - bool closed; /* Stream is closed. */ + bool closed; /* Stream is closed. */ + bool close_requested; /* Close command has been received. */ /* * Counts number of indexes in indexes_ht. Redundant info. @@ -124,18 +155,48 @@ struct relay_stream { * Node of stream within global stream hash table. */ struct lttng_ht_node_u64 node; + bool in_stream_ht; /* is stream in stream hash table. */ struct rcu_head rcu_node; /* For call_rcu teardown. */ + /* + * When we have written the data and index corresponding to this + * seq_num, rotate the tracefile (session rotation). The path_name is + * already up-to-date. + * This is set to -1ULL when no rotation is pending. + * + * Always access with stream lock held. + */ + uint64_t rotate_at_seq_num; + /* + * When rotate_at_seq_num != -1ULL, meaning that a rotation is ongoing, + * data_rotated and index_rotated respectively indicate if the stream's + * data and index have been rotated. A rotation is considered completed + * when both rotations have occurred. + */ + bool data_rotated; + bool index_rotated; + /* + * This is the id of the chunk where we are writing to if no rotation is + * pending (rotate_at_seq_num == -1ULL). If a rotation is pending, this + * is the chunk_id we will have after the rotation. It must be updated + * atomically with rotate_at_seq_num. + * + * Always access with stream lock held. + * + * This attribute is not set if the stream is created by a pre-2.11 + * consumer. + */ + struct relay_stream_chunk_id current_chunk_id; }; struct relay_stream *stream_create(struct ctf_trace *trace, uint64_t stream_handle, char *path_name, char *channel_name, uint64_t tracefile_size, - uint64_t tracefile_count); + uint64_t tracefile_count, const struct relay_stream_chunk_id *chunk_id); struct relay_stream *stream_get_by_id(uint64_t stream_id); bool stream_get(struct relay_stream *stream); void stream_put(struct relay_stream *stream); -void stream_close(struct relay_stream *stream); +void try_stream_close(struct relay_stream *stream); void stream_publish(struct relay_stream *stream); void print_relay_streams(void);