X-Git-Url: https://git.lttng.org/?a=blobdiff_plain;f=src%2Fbin%2Flttng-relayd%2Fstream.h;h=47ae2e8f9927fdcd672a595bf3c20902fd375846;hb=40bbd087934edf0be7c21165786aad3fabb7e7b5;hp=ca6be8133dc9cd963f5fcdbdae379b813119ef84;hpb=a44ca2ca85e4b64729f7b88b1919fd6737dfff8a;p=lttng-tools.git diff --git a/src/bin/lttng-relayd/stream.h b/src/bin/lttng-relayd/stream.h index ca6be8133..47ae2e8f9 100644 --- a/src/bin/lttng-relayd/stream.h +++ b/src/bin/lttng-relayd/stream.h @@ -37,12 +37,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; @@ -58,8 +52,8 @@ struct relay_stream { /* 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; char *channel_name; @@ -69,6 +63,11 @@ struct relay_stream { uint64_t tracefile_size_current; uint64_t tracefile_count; + /* + * Position in the tracefile where we have the full index also on disk. + */ + uint64_t pos_after_last_complete_data_index; + /* * Counts the number of received indexes. The "tag" associated * with an index is taken before incrementing this seqcount. @@ -85,7 +84,8 @@ struct relay_stream { */ 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. @@ -132,7 +132,26 @@ 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; + /* + * 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. + */ + uint64_t chunk_id; }; struct relay_stream *stream_create(struct ctf_trace *trace, @@ -143,7 +162,7 @@ struct relay_stream *stream_create(struct ctf_trace *trace, 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);