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

#ifndef _STREAM_H
#define _STREAM_H

/*
 * Copyright (C) 2013 - Julien Desfossez <jdesfossez@efficios.com>
 *                      David Goulet <dgoulet@efficios.com>
 *               2015 - Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
 *
 * This program is free software; you can redistribute it and/or modify it
 * under the terms of the GNU General Public License, version 2 only, as
 * published by the Free Software Foundation.
 *
 * This program is distributed in the hope that it will be useful, but WITHOUT
 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
 * more details.
 *
 * You should have received a copy of the GNU General Public License along with
 * this program; if not, write to the Free Software Foundation, Inc., 51
 * Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA.
 */

#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 "stream-fd.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;
	/*
	 * Sequence number of the first packet of the new trace chunk to which
	 * the stream is rotating.
	 */
	uint64_t seq_num;
	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;

	/* FD on which to write the stream data. */
	struct stream_fd *stream_fd;
	/* 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;

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