[lttng-tools.git] / src / bin / lttng-relayd / stream.h

#ifndef _STREAM_H
#define _STREAM_H

/*
 * Copyright (C) 2013 Julien Desfossez <jdesfossez@efficios.com>
 * Copyright (C) 2013 David Goulet <dgoulet@efficios.com>
 * Copyright (C) 2015 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
 *
 * SPDX-License-Identifier: GPL-2.0-only
 *
 */

#include <limits.h>
#include <inttypes.h>
#include <pthread.h>
#include <urcu/list.h>

#include <common/hashtable/hashtable.h>
#include <common/trace-chunk.h>
#include <common/optional.h>
#include <common/buffer-view.h>

#include "session.h"
#include "tracefile-array.h"

struct lttcomm_relayd_index;

struct relay_stream_rotation {
	/*
	 * Indicates 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;
	/*
	 * Packet sequence number of the first packet of the new trace chunk to
	 * which the stream is rotating.
	 */
	uint64_t packet_seq_num;
	/*
	 * Monotonically increasing previous network sequence number of first
	 * data packet of the new trace chunk to which the stream is rotating.
	 */
	uint64_t prev_data_net_seq;
	struct lttng_trace_chunk *next_trace_chunk;
};

/*
 * Represents a stream in the relay
 */
struct relay_stream {
	uint64_t stream_handle;

	struct urcu_ref ref;
	/* Back reference to trace. Protected by refcount on trace object. */
	struct ctf_trace *trace;

	/*
	 * To protect from concurrent read/update. The viewer stream
	 * lock nests inside the stream lock. The stream lock nests
	 * inside the ctf_trace lock.
	 */
	pthread_mutex_t lock;
	/* 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;

	struct fs_handle *file;
	/* index file on which to write the index data. */
	struct lttng_index_file *index_file;

	char *path_name;
	char *channel_name;

	/* On-disk circular buffer of tracefiles. */
	uint64_t tracefile_size;
	uint64_t tracefile_size_current;
	/* Max number of trace files for this stream. */
	uint64_t tracefile_count;
	/*
	 * Index of the currently active file for this stream's on-disk
	 * ring buffer.
	 */
	uint64_t tracefile_current_index;
	/*
	 * Indicates that the on-disk buffer has wrapped around. Stream
	 * files shall be unlinked before being opened after this has occurred.
	 */
	bool tracefile_wrapped_around;

	/*
	 * 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.
	 * Therefore, the sequence tag associated with the last index
	 * received is always index_received_seqcount - 1.
	 */
	uint64_t index_received_seqcount;

	/*
	 * Packet sequence number of the last received packet index.
	 * Only populated when interacting with CTF_INDEX 1.1+.
	 */
	LTTNG_OPTIONAL(uint64_t) received_packet_seq_num;

	/*
	 * 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 close_requested;	/* Close command has been received. */

	/*
	 * Counts number of indexes in indexes_ht. Redundant info.
	 * Protected by stream lock.
	 */
	int indexes_in_flight;
	struct lttng_ht *indexes_ht;

	/*
	 * If the stream is inactive, this field is updated with the
	 * live beacon timestamp end, when it is active, this
	 * field == -1ULL.
	 */
	uint64_t beacon_ts_end;

	/* CTF stream ID, -1ULL when unset (first packet not received yet). */
	uint64_t ctf_stream_id;

	/* Indicate if the stream was initialized for a data pending command. */
	bool data_pending_check_done;

	/* Is this stream a metadata stream ? */
	bool is_metadata;
	/* Amount of metadata received (bytes). */
	uint64_t metadata_received;

	/*
	 * Member of the stream list in struct ctf_trace.
	 * Updates are protected by the stream_list_lock.
	 * Traversals are protected by RCU.
	 */
	struct cds_list_head stream_node;
	/*
	 * Temporary list belonging to the connection until all streams
	 * are received for that connection.
	 * Member of the stream recv list in the connection.
	 * Updates are protected by the stream_recv_list_lock.
	 * Traversals are protected by RCU.
	 */
	bool in_recv_list;
	struct cds_list_head recv_node;
	/* Protected by session lock. */
	bool published;
	/* Notified viewer that no new metadata is available. */
	bool no_new_metadata_notified;
	/*
	 * 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. */
	/*
	 * The trace chunk to which the file currently being produced (if any)
	 * belongs.
	 */
	struct lttng_trace_chunk *trace_chunk;
	LTTNG_OPTIONAL(struct relay_stream_rotation) ongoing_rotation;
};

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);

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);
int stream_rotate_output_files(struct relay_session *session,
		struct relay_stream *stream);
int stream_set_pending_rotation(struct relay_stream *stream,
		struct lttng_trace_chunk *next_trace_chunk,
		uint64_t rotation_sequence_number);
void try_stream_close(struct relay_stream *stream);
void stream_publish(struct relay_stream *stream);
int stream_init_packet(struct relay_stream *stream, size_t packet_size,
		bool *file_rotated);
int stream_write(struct relay_stream *stream,
		const struct lttng_buffer_view *packet, size_t padding_len);
/* Called after the reception of a complete data packet. */
int stream_update_index(struct relay_stream *stream, uint64_t net_seq_num,
		bool rotate_index, bool *flushed, uint64_t total_size);
int stream_complete_packet(struct relay_stream *stream,
		size_t packet_total_size, uint64_t sequence_number,
		bool index_flushed);
/* Index info is in host endianness. */
int stream_add_index(struct relay_stream *stream,
		const struct lttcomm_relayd_index *index_info);
int stream_reset_file(struct relay_stream *stream);

void print_relay_streams(void);

#endif /* _STREAM_H */
Commit	Line	Data
7591bab1 MD	1	#ifndef _STREAM_H
	2	#define _STREAM_H
	3
2a174661	4	/*
ab5be9fa MJ	5	* Copyright (C) 2013 Julien Desfossez <jdesfossez@efficios.com>
	6	* Copyright (C) 2013 David Goulet <dgoulet@efficios.com>
	7	* Copyright (C) 2015 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
2a174661	8	*
ab5be9fa	9	* SPDX-License-Identifier: GPL-2.0-only
2a174661	10	*
2a174661 DG	11	*/
2a174661 DG	12
2a174661 DG	13	#include <limits.h>
	14	#include <inttypes.h>
	15	#include <pthread.h>
	16	#include <urcu/list.h>
	17
	18	#include <common/hashtable/hashtable.h>
348a81dc	19	#include <common/trace-chunk.h>
c35f9726 JG	20	#include <common/optional.h>
c35f9726 JG	21	#include <common/buffer-view.h>
2a174661 DG	22
2a174661 DG	23	#include "session.h"
a44ca2ca	24	#include "tracefile-array.h"
2a174661	25
c35f9726 JG	26	struct lttcomm_relayd_index;
	27
	28	struct relay_stream_rotation {
	29	/*
	30	* Indicates if the stream's data and index have been rotated. A
	31	* rotation is considered completed when both rotations have occurred.
	32	*/
	33	bool data_rotated;
	34	bool index_rotated;
	35	/*
0f83d1cc MD	36	* Packet sequence number of the first packet of the new trace chunk to
0f83d1cc MD	37	* which the stream is rotating.
c35f9726	38	*/
0f83d1cc MD	39	uint64_t packet_seq_num;
	40	/*
	41	* Monotonically increasing previous network sequence number of first
	42	* data packet of the new trace chunk to which the stream is rotating.
	43	*/
	44	uint64_t prev_data_net_seq;
c35f9726 JG	45	struct lttng_trace_chunk *next_trace_chunk;
	46	};
	47
2a174661 DG	48	/*
	49	* Represents a stream in the relay
	50	*/
	51	struct relay_stream {
	52	uint64_t stream_handle;
7591bab1	53
7591bab1 MD	54	struct urcu_ref ref;
	55	/* Back reference to trace. Protected by refcount on trace object. */
	56	struct ctf_trace *trace;
2a174661	57
7591bab1 MD	58	/*
	59	* To protect from concurrent read/update. The viewer stream
	60	* lock nests inside the stream lock. The stream lock nests
	61	* inside the ctf_trace lock.
	62	*/
	63	pthread_mutex_t lock;
a8f9f353 JG	64	/* previous data sequence number written to disk. */
a8f9f353 JG	65	uint64_t prev_data_seq;
7a45c7e6 JG	66	/* previous index sequence number written to disk. */
7a45c7e6 JG	67	uint64_t prev_index_seq;
a8f9f353 JG	68	/* seq num to encounter before closing. */
a8f9f353 JG	69	uint64_t last_net_seq_num;
7591bab1	70
8bb66c3c	71	struct fs_handle *file;
f8f3885c MD	72	/* index file on which to write the index data. */
f8f3885c MD	73	struct lttng_index_file *index_file;
2a174661 DG	74
	75	char *path_name;
	76	char *channel_name;
7591bab1 MD	77
7591bab1 MD	78	/* On-disk circular buffer of tracefiles. */
2a174661 DG	79	uint64_t tracefile_size;
2a174661 DG	80	uint64_t tracefile_size_current;
c35f9726	81	/* Max number of trace files for this stream. */
2a174661	82	uint64_t tracefile_count;
c35f9726 JG	83	/*
	84	* Index of the currently active file for this stream's on-disk
	85	* ring buffer.
	86	*/
	87	uint64_t tracefile_current_index;
	88	/*
	89	* Indicates that the on-disk buffer has wrapped around. Stream
	90	* files shall be unlinked before being opened after this has occurred.
	91	*/
	92	bool tracefile_wrapped_around;
7591bab1	93
d3ecc550 JD	94	/*
	95	* Position in the tracefile where we have the full index also on disk.
	96	*/
	97	uint64_t pos_after_last_complete_data_index;
	98
a44ca2ca MD	99	/*
	100	* Counts the number of received indexes. The "tag" associated
	101	* with an index is taken before incrementing this seqcount.
	102	* Therefore, the sequence tag associated with the last index
	103	* received is always index_received_seqcount - 1.
	104	*/
	105	uint64_t index_received_seqcount;
2a174661	106
0f83d1cc MD	107	/*
	108	* Packet sequence number of the last received packet index.
	109	* Only populated when interacting with CTF_INDEX 1.1+.
	110	*/
	111	LTTNG_OPTIONAL(uint64_t) received_packet_seq_num;
	112
a44ca2ca MD	113	/*
	114	* Tracefile array is an index of the stream trace files,
	115	* indexed by position. It allows keeping track of the oldest
	116	* available indexes when overwriting trace files in tracefile
	117	* rotation.
	118	*/
	119	struct tracefile_array *tfa;
2a174661	120
bda7c7b9 JG	121	bool closed; /* Stream is closed. */
bda7c7b9 JG	122	bool close_requested; /* Close command has been received. */
2a174661 DG	123
2a174661 DG	124	/*
7591bab1 MD	125	* Counts number of indexes in indexes_ht. Redundant info.
7591bab1 MD	126	* Protected by stream lock.
6e7241fe JD	127	*/
6e7241fe JD	128	int indexes_in_flight;
7591bab1 MD	129	struct lttng_ht *indexes_ht;
7591bab1 MD	130
528f2ffa	131	/*
7591bab1 MD	132	* If the stream is inactive, this field is updated with the
	133	* live beacon timestamp end, when it is active, this
	134	* field == -1ULL.
528f2ffa	135	*/
7591bab1 MD	136	uint64_t beacon_ts_end;
	137
	138	/* CTF stream ID, -1ULL when unset (first packet not received yet). */
528f2ffa	139	uint64_t ctf_stream_id;
2a174661	140
7591bab1 MD	141	/* Indicate if the stream was initialized for a data pending command. */
	142	bool data_pending_check_done;
	143
	144	/* Is this stream a metadata stream ? */
819cb727	145	bool is_metadata;
7591bab1 MD	146	/* Amount of metadata received (bytes). */
	147	uint64_t metadata_received;
	148
2a174661	149	/*
7591bab1 MD	150	* Member of the stream list in struct ctf_trace.
	151	* Updates are protected by the stream_list_lock.
	152	* Traversals are protected by RCU.
2a174661	153	*/
7591bab1	154	struct cds_list_head stream_node;
2a174661	155	/*
7591bab1 MD	156	* Temporary list belonging to the connection until all streams
	157	* are received for that connection.
	158	* Member of the stream recv list in the connection.
	159	* Updates are protected by the stream_recv_list_lock.
	160	* Traversals are protected by RCU.
2a174661	161	*/
7591bab1 MD	162	bool in_recv_list;
7591bab1 MD	163	struct cds_list_head recv_node;
c35f9726 JG	164	/* Protected by session lock. */
c35f9726 JG	165	bool published;
94f73d08 MD	166	/* Notified viewer that no new metadata is available. */
94f73d08 MD	167	bool no_new_metadata_notified;
2a174661	168	/*
7591bab1	169	* Node of stream within global stream hash table.
2a174661	170	*/
7591bab1	171	struct lttng_ht_node_u64 node;
77f7bd85	172	bool in_stream_ht; /* is stream in stream hash table. */
7591bab1	173	struct rcu_head rcu_node; /* For call_rcu teardown. */
d3ecc550	174	/*
c35f9726 JG	175	* The trace chunk to which the file currently being produced (if any)
c35f9726 JG	176	* belongs.
d3ecc550	177	*/
348a81dc	178	struct lttng_trace_chunk *trace_chunk;
c35f9726	179	LTTNG_OPTIONAL(struct relay_stream_rotation) ongoing_rotation;
2a174661 DG	180	};
2a174661 DG	181
7591bab1 MD	182	struct relay_stream stream_create(struct ctf_trace trace,
	183	uint64_t stream_handle, char *path_name,
	184	char *channel_name, uint64_t tracefile_size,
348a81dc	185	uint64_t tracefile_count);
7591bab1 MD	186
	187	struct relay_stream *stream_get_by_id(uint64_t stream_id);
	188	bool stream_get(struct relay_stream *stream);
	189	void stream_put(struct relay_stream *stream);
c35f9726 JG	190	int stream_rotate_output_files(struct relay_session *session,
	191	struct relay_stream *stream);
	192	int stream_set_pending_rotation(struct relay_stream *stream,
	193	struct lttng_trace_chunk *next_trace_chunk,
	194	uint64_t rotation_sequence_number);
bda7c7b9	195	void try_stream_close(struct relay_stream *stream);
7591bab1	196	void stream_publish(struct relay_stream *stream);
c35f9726 JG	197	int stream_init_packet(struct relay_stream *stream, size_t packet_size,
	198	bool *file_rotated);
	199	int stream_write(struct relay_stream *stream,
	200	const struct lttng_buffer_view *packet, size_t padding_len);
	201	/* Called after the reception of a complete data packet. */
	202	int stream_update_index(struct relay_stream *stream, uint64_t net_seq_num,
	203	bool rotate_index, bool *flushed, uint64_t total_size);
	204	int stream_complete_packet(struct relay_stream *stream,
	205	size_t packet_total_size, uint64_t sequence_number,
	206	bool index_flushed);
	207	/* Index info is in host endianness. */
	208	int stream_add_index(struct relay_stream *stream,
	209	const struct lttcomm_relayd_index *index_info);
	210	int stream_reset_file(struct relay_stream *stream);
	211
7591bab1	212	void print_relay_streams(void);
2a174661 DG	213
2a174661 DG	214	#endif /* _STREAM_H */