[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 "session.h"
#include "stream-fd.h"
#include "tracefile-array.h"

/*
 * Represents a stream in the relay
 */
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;

	/*
	 * 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;
	uint64_t prev_seq;		/* previous data sequence number encountered. */
	uint64_t prev_index_seq;	/* previous index sequence number encountered. */
	uint64_t last_net_seq_num;	/* seq num to encounter before closing. */

	/* 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;
	uint64_t tracefile_count;
	uint64_t tracefile_count_current;

	/*
	 * 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;


	/*
	 * Index sequence number for the position at which clear needs to be
	 * performed.
	 */
	uint64_t clear_position_index_seqcount;
	uint64_t clear_position_data_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 ? */
	int32_t 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;
	bool published;	/* Protected by session lock. */
	/*
	 * 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. */
};

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);
void try_stream_close(struct relay_stream *stream);
void stream_publish(struct relay_stream *stream);
void print_relay_streams(void);
int stream_clear(struct relay_stream *stream);
int try_stream_clear_index_data(struct relay_stream *stream);

#endif /* _STREAM_H */
Commit	Line	Data
7591bab1 MD	1	#ifndef _STREAM_H
	2	#define _STREAM_H
	3
2a174661 DG	4	/*
	5	* Copyright (C) 2013 - Julien Desfossez <jdesfossez@efficios.com>
	6	* David Goulet <dgoulet@efficios.com>
7591bab1	7	* 2015 - Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
2a174661 DG	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
2a174661 DG	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
	30	#include "session.h"
7591bab1	31	#include "stream-fd.h"
a44ca2ca	32	#include "tracefile-array.h"
2a174661 DG	33
	34	/*
	35	* Represents a stream in the relay
	36	*/
	37	struct relay_stream {
	38	uint64_t stream_handle;
7591bab1	39
2a174661	40	/*
7591bab1 MD	41	* reflock used to synchronize the closing of this stream.
	42	* stream reflock nests inside viewer stream reflock.
	43	* stream reflock nests inside index reflock.
2a174661	44	*/
7591bab1 MD	45	pthread_mutex_t reflock;
	46	struct urcu_ref ref;
	47	/* Back reference to trace. Protected by refcount on trace object. */
	48	struct ctf_trace *trace;
2a174661	49
7591bab1 MD	50	/*
	51	* To protect from concurrent read/update. The viewer stream
	52	* lock nests inside the stream lock. The stream lock nests
	53	* inside the ctf_trace lock.
	54	*/
	55	pthread_mutex_t lock;
	56	uint64_t prev_seq; /* previous data sequence number encountered. */
82b4cfcc	57	uint64_t prev_index_seq; /* previous index sequence number encountered. */
7591bab1 MD	58	uint64_t last_net_seq_num; /* seq num to encounter before closing. */
	59
	60	/* FD on which to write the stream data. */
	61	struct stream_fd *stream_fd;
f8f3885c MD	62	/* index file on which to write the index data. */
f8f3885c MD	63	struct lttng_index_file *index_file;
2a174661 DG	64
	65	char *path_name;
	66	char *channel_name;
7591bab1 MD	67
7591bab1 MD	68	/* On-disk circular buffer of tracefiles. */
2a174661 DG	69	uint64_t tracefile_size;
	70	uint64_t tracefile_size_current;
	71	uint64_t tracefile_count;
82b4cfcc	72	uint64_t tracefile_count_current;
7591bab1	73
a44ca2ca MD	74	/*
	75	* Counts the number of received indexes. The "tag" associated
	76	* with an index is taken before incrementing this seqcount.
	77	* Therefore, the sequence tag associated with the last index
	78	* received is always index_received_seqcount - 1.
	79	*/
	80	uint64_t index_received_seqcount;
2a174661	81
82b4cfcc MD	82
	83	/*
	84	* Index sequence number for the position at which clear needs to be
	85	* performed.
	86	*/
	87	uint64_t clear_position_index_seqcount;
	88	uint64_t clear_position_data_seqcount;
	89
a44ca2ca MD	90	/*
	91	* Tracefile array is an index of the stream trace files,
	92	* indexed by position. It allows keeping track of the oldest
	93	* available indexes when overwriting trace files in tracefile
	94	* rotation.
	95	*/
	96	struct tracefile_array *tfa;
2a174661	97
bda7c7b9 JG	98	bool closed; /* Stream is closed. */
bda7c7b9 JG	99	bool close_requested; /* Close command has been received. */
2a174661 DG	100
2a174661 DG	101	/*
7591bab1 MD	102	* Counts number of indexes in indexes_ht. Redundant info.
7591bab1 MD	103	* Protected by stream lock.
6e7241fe JD	104	*/
6e7241fe JD	105	int indexes_in_flight;
7591bab1 MD	106	struct lttng_ht *indexes_ht;
7591bab1 MD	107
528f2ffa	108	/*
7591bab1 MD	109	* If the stream is inactive, this field is updated with the
	110	* live beacon timestamp end, when it is active, this
	111	* field == -1ULL.
528f2ffa	112	*/
7591bab1 MD	113	uint64_t beacon_ts_end;
	114
	115	/* CTF stream ID, -1ULL when unset (first packet not received yet). */
528f2ffa	116	uint64_t ctf_stream_id;
2a174661	117
7591bab1 MD	118	/* Indicate if the stream was initialized for a data pending command. */
	119	bool data_pending_check_done;
	120
	121	/* Is this stream a metadata stream ? */
	122	int32_t is_metadata;
	123	/* Amount of metadata received (bytes). */
	124	uint64_t metadata_received;
	125
2a174661	126	/*
7591bab1 MD	127	* Member of the stream list in struct ctf_trace.
	128	* Updates are protected by the stream_list_lock.
	129	* Traversals are protected by RCU.
2a174661	130	*/
7591bab1	131	struct cds_list_head stream_node;
2a174661	132	/*
7591bab1 MD	133	* Temporary list belonging to the connection until all streams
	134	* are received for that connection.
	135	* Member of the stream recv list in the connection.
	136	* Updates are protected by the stream_recv_list_lock.
	137	* Traversals are protected by RCU.
2a174661	138	*/
7591bab1 MD	139	bool in_recv_list;
	140	struct cds_list_head recv_node;
	141	bool published; /* Protected by session lock. */
2a174661	142	/*
7591bab1	143	* Node of stream within global stream hash table.
2a174661	144	*/
7591bab1	145	struct lttng_ht_node_u64 node;
77f7bd85	146	bool in_stream_ht; /* is stream in stream hash table. */
7591bab1	147	struct rcu_head rcu_node; /* For call_rcu teardown. */
2a174661 DG	148	};
2a174661 DG	149
7591bab1 MD	150	struct relay_stream stream_create(struct ctf_trace trace,
	151	uint64_t stream_handle, char *path_name,
	152	char *channel_name, uint64_t tracefile_size,
	153	uint64_t tracefile_count);
	154
	155	struct relay_stream *stream_get_by_id(uint64_t stream_id);
	156	bool stream_get(struct relay_stream *stream);
	157	void stream_put(struct relay_stream *stream);
bda7c7b9	158	void try_stream_close(struct relay_stream *stream);
7591bab1 MD	159	void stream_publish(struct relay_stream *stream);
7591bab1 MD	160	void print_relay_streams(void);
82b4cfcc MD	161	int stream_clear(struct relay_stream *stream);
82b4cfcc MD	162	int try_stream_clear_index_data(struct relay_stream *stream);
2a174661 DG	163
2a174661 DG	164	#endif /* _STREAM_H */