[lttng-tools.git] / src / bin / lttng-relayd / index.cpp

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

#define _LGPL_SOURCE

#include <common/common.hpp>
#include <common/utils.hpp>
#include <common/compat/endian.hpp>

#include "lttng-relayd.hpp"
#include "stream.hpp"
#include "index.hpp"
#include "connection.hpp"

/*
 * Allocate a new relay index object. Pass the stream in which it is
 * contained as parameter. The sequence number will be used as the hash
 * table key.
 *
 * Called with stream mutex held.
 * Return allocated object or else NULL on error.
 */
static struct relay_index *relay_index_create(struct relay_stream *stream,
		uint64_t net_seq_num)
{
	struct relay_index *index;

	DBG2("Creating relay index for stream id %" PRIu64 " and seqnum %" PRIu64,
			stream->stream_handle, net_seq_num);

	index = zmalloc<relay_index>();
	if (!index) {
		PERROR("Relay index zmalloc");
		goto end;
	}
	if (!stream_get(stream)) {
		ERR("Cannot get stream");
		free(index);
		index = NULL;
		goto end;
	}
	index->stream = stream;

	lttng_ht_node_init_u64(&index->index_n, net_seq_num);
	pthread_mutex_init(&index->lock, NULL);
	urcu_ref_init(&index->ref);

end:
	return index;
}

/*
 * Add unique relay index to the given hash table. In case of a collision, the
 * already existing object is put in the given _index variable.
 *
 * RCU read side lock MUST be acquired.
 */
static struct relay_index *relay_index_add_unique(struct relay_stream *stream,
		struct relay_index *index)
{
	struct cds_lfht_node *node_ptr;
	struct relay_index *_index;

	ASSERT_RCU_READ_LOCKED();

	DBG2("Adding relay index with stream id %" PRIu64 " and seqnum %" PRIu64,
			stream->stream_handle, index->index_n.key);

	node_ptr = cds_lfht_add_unique(stream->indexes_ht->ht,
			stream->indexes_ht->hash_fct(&index->index_n, lttng_ht_seed),
			stream->indexes_ht->match_fct, &index->index_n,
			&index->index_n.node);
	if (node_ptr != &index->index_n.node) {
		_index = caa_container_of(node_ptr, struct relay_index,
				index_n.node);
	} else {
		_index = NULL;
	}
	return _index;
}

/*
 * Should be called with RCU read-side lock held.
 */
static bool relay_index_get(struct relay_index *index)
{
	ASSERT_RCU_READ_LOCKED();

	DBG2("index get for stream id %" PRIu64 " and seqnum %" PRIu64 " refcount %d",
			index->stream->stream_handle, index->index_n.key,
			(int) index->ref.refcount);

	return urcu_ref_get_unless_zero(&index->ref);
}

/*
 * Get a relayd index in within the given stream, or create it if not
 * present.
 *
 * Called with stream mutex held.
 * Return index object or else NULL on error.
 */
struct relay_index *relay_index_get_by_id_or_create(struct relay_stream *stream,
		 uint64_t net_seq_num)
{
	struct lttng_ht_node_u64 *node;
	struct lttng_ht_iter iter;
	struct relay_index *index = NULL;

	DBG3("Finding index for stream id %" PRIu64 " and seq_num %" PRIu64,
			stream->stream_handle, net_seq_num);

	rcu_read_lock();
	lttng_ht_lookup(stream->indexes_ht, &net_seq_num, &iter);
	node = lttng_ht_iter_get_node_u64(&iter);
	if (node) {
		index = caa_container_of(node, struct relay_index, index_n);
	} else {
		struct relay_index *oldindex;

		index = relay_index_create(stream, net_seq_num);
		if (!index) {
			ERR("Cannot create index for stream id %" PRIu64 " and seq_num %" PRIu64,
				stream->stream_handle, net_seq_num);
			goto end;
		}
		oldindex = relay_index_add_unique(stream, index);
		if (oldindex) {
			/* Added concurrently, keep old. */
			relay_index_put(index);
			index = oldindex;
			if (!relay_index_get(index)) {
				index = NULL;
			}
		} else {
			stream->indexes_in_flight++;
			index->in_hash_table = true;
		}
	}
end:
	rcu_read_unlock();
	DBG2("Index %sfound or created in HT for stream ID %" PRIu64 " and seqnum %" PRIu64,
			(index == NULL) ? "NOT " : "", stream->stream_handle, net_seq_num);
	return index;
}

int relay_index_set_file(struct relay_index *index,
		struct lttng_index_file *index_file,
		uint64_t data_offset)
{
	int ret = 0;

	pthread_mutex_lock(&index->lock);
	if (index->index_file) {
		ret = -1;
		goto end;
	}
	lttng_index_file_get(index_file);
	index->index_file = index_file;
	index->index_data.offset = data_offset;
end:
	pthread_mutex_unlock(&index->lock);
	return ret;
}

int relay_index_set_data(struct relay_index *index,
		const struct ctf_packet_index *data)
{
	int ret = 0;

	pthread_mutex_lock(&index->lock);
	if (index->has_index_data) {
		ret = -1;
		goto end;
	}
	/* Set everything except data_offset. */
	index->index_data.packet_size = data->packet_size;
	index->index_data.content_size = data->content_size;
	index->index_data.timestamp_begin = data->timestamp_begin;
	index->index_data.timestamp_end = data->timestamp_end;
	index->index_data.events_discarded = data->events_discarded;
	index->index_data.stream_id = data->stream_id;
	index->has_index_data = true;
end:
	pthread_mutex_unlock(&index->lock);
	return ret;
}

static void index_destroy(struct relay_index *index)
{
	free(index);
}

static void index_destroy_rcu(struct rcu_head *rcu_head)
{
	struct relay_index *index =
		caa_container_of(rcu_head, struct relay_index, rcu_node);

	index_destroy(index);
}

/* Stream lock must be held by the caller. */
static void index_release(struct urcu_ref *ref)
{
	struct relay_index *index = caa_container_of(ref, struct relay_index, ref);
	struct relay_stream *stream = index->stream;
	int ret;
	struct lttng_ht_iter iter;

	if (index->index_file) {
		lttng_index_file_put(index->index_file);
		index->index_file = NULL;
	}
	if (index->in_hash_table) {
		/* Delete index from hash table. */
		iter.iter.node = &index->index_n.node;
		ret = lttng_ht_del(stream->indexes_ht, &iter);
		LTTNG_ASSERT(!ret);
		stream->indexes_in_flight--;
	}

	stream_put(index->stream);
	index->stream = NULL;

	call_rcu(&index->rcu_node, index_destroy_rcu);
}

/*
 * Called with stream mutex held.
 *
 * Stream lock must be held by the caller.
 */
void relay_index_put(struct relay_index *index)
{
	DBG2("index put for stream id %" PRIu64 " and seqnum %" PRIu64 " refcount %d",
			index->stream->stream_handle, index->index_n.key,
			(int) index->ref.refcount);
	/*
	 * Ensure existence of index->lock for index unlock.
	 */
	rcu_read_lock();
	/*
	 * Index lock ensures that concurrent test and update of stream
	 * ref is atomic.
	 */
	LTTNG_ASSERT(index->ref.refcount != 0);
	urcu_ref_put(&index->ref, index_release);
	rcu_read_unlock();
}

/*
 * Try to flush index to disk. Releases self-reference to index once
 * flush succeeds.
 *
 * Stream lock must be held by the caller.
 * Return 0 on successful flush, a negative value on error, or positive
 * value if no flush was performed.
 */
int relay_index_try_flush(struct relay_index *index)
{
	int ret = 1;
	bool flushed = false;

	pthread_mutex_lock(&index->lock);
	if (index->flushed) {
		goto skip;
	}
	/* Check if we are ready to flush. */
	if (!index->has_index_data || !index->index_file) {
		goto skip;
	}

	DBG2("Writing index for stream ID %" PRIu64 " and seq num %" PRIu64,
			index->stream->stream_handle, index->index_n.key);
	flushed = true;
	index->flushed = true;
	ret = lttng_index_file_write(index->index_file, &index->index_data);
skip:
	pthread_mutex_unlock(&index->lock);

	if (flushed) {
		/* Put self-ref from index now that it has been flushed. */
		relay_index_put(index);
	}
	return ret;
}

/*
 * Close every relay index within a given stream, without flushing
 * them.
 */
void relay_index_close_all(struct relay_stream *stream)
{
	struct lttng_ht_iter iter;
	struct relay_index *index;

	rcu_read_lock();
	cds_lfht_for_each_entry(stream->indexes_ht->ht, &iter.iter,
			index, index_n.node) {
		/* Put self-ref from index. */
		relay_index_put(index);
	}
	rcu_read_unlock();
}

void relay_index_close_partial_fd(struct relay_stream *stream)
{
	struct lttng_ht_iter iter;
	struct relay_index *index;

	rcu_read_lock();
	cds_lfht_for_each_entry(stream->indexes_ht->ht, &iter.iter,
			index, index_n.node) {
		if (!index->index_file) {
			continue;
		}
		/*
		 * Partial index has its index_file: we have only
		 * received its info from the data socket.
		 * Put self-ref from index.
		 */
		relay_index_put(index);
	}
	rcu_read_unlock();
}

uint64_t relay_index_find_last(struct relay_stream *stream)
{
	struct lttng_ht_iter iter;
	struct relay_index *index;
	uint64_t net_seq_num = -1ULL;

	rcu_read_lock();
	cds_lfht_for_each_entry(stream->indexes_ht->ht, &iter.iter,
			index, index_n.node) {
		if (net_seq_num == -1ULL ||
				index->index_n.key > net_seq_num) {
			net_seq_num = index->index_n.key;
		}
	}
	rcu_read_unlock();
	return net_seq_num;
}

/*
 * Update the index file of an already existing relay_index.
 * Offsets by 'removed_data_count' the offset field of an index.
 */
static
int relay_index_switch_file(struct relay_index *index,
		struct lttng_index_file *new_index_file,
		uint64_t removed_data_count)
{
	int ret = 0;
	uint64_t offset;

	pthread_mutex_lock(&index->lock);
	if (!index->index_file) {
		ERR("No index_file");
		ret = 0;
		goto end;
	}

	lttng_index_file_put(index->index_file);
	lttng_index_file_get(new_index_file);
	index->index_file = new_index_file;
	offset = be64toh(index->index_data.offset);
	index->index_data.offset = htobe64(offset - removed_data_count);

end:
	pthread_mutex_unlock(&index->lock);
	return ret;
}

/*
 * Switch the index file of all pending indexes for a stream and update the
 * data offset by substracting the last safe position.
 * Stream lock must be held.
 */
int relay_index_switch_all_files(struct relay_stream *stream)
{
	struct lttng_ht_iter iter;
	struct relay_index *index;
	int ret = 0;

	rcu_read_lock();
	cds_lfht_for_each_entry(stream->indexes_ht->ht, &iter.iter,
			index, index_n.node) {
		ret = relay_index_switch_file(index, stream->index_file,
				stream->pos_after_last_complete_data_index);
		if (ret) {
			goto end;
		}
	}
end:
	rcu_read_unlock();
	return ret;
}

/*
 * Set index data from the control port to a given index object.
 */
int relay_index_set_control_data(struct relay_index *index,
		const struct lttcomm_relayd_index *data,
		unsigned int minor_version)
{
	/* The index on disk is encoded in big endian. */
	ctf_packet_index index_data {};

	index_data.packet_size = htobe64(data->packet_size);
	index_data.content_size = htobe64(data->content_size);
	index_data.timestamp_begin = htobe64(data->timestamp_begin);
	index_data.timestamp_end = htobe64(data->timestamp_end);
	index_data.events_discarded = htobe64(data->events_discarded);
	index_data.stream_id = htobe64(data->stream_id);

	if (minor_version >= 8) {
		index->index_data.stream_instance_id = htobe64(data->stream_instance_id);
		index->index_data.packet_seq_num = htobe64(data->packet_seq_num);
	} else {
		uint64_t unset_value = -1ULL;

		index->index_data.stream_instance_id = htobe64(unset_value);
		index->index_data.packet_seq_num = htobe64(unset_value);
	}

	return relay_index_set_data(index, &index_data);
}
Commit	Line	Data
1c20f0e2	1	/*
ab5be9fa MJ	2	* Copyright (C) 2013 Julien Desfossez <jdesfossez@efficios.com>
	3	* Copyright (C) 2013 David Goulet <dgoulet@efficios.com>
	4	* Copyright (C) 2015 Mathieu Desnoyers <mathieu.desnoyers@efficios.com>
1c20f0e2	5	*
ab5be9fa	6	* SPDX-License-Identifier: GPL-2.0-only
1c20f0e2	7	*
1c20f0e2 JD	8	*/
1c20f0e2 JD	9
6c1c0768	10	#define _LGPL_SOURCE
1c20f0e2	11
c9e313bc SM	12	#include <common/common.hpp>
	13	#include <common/utils.hpp>
	14	#include <common/compat/endian.hpp>
1c20f0e2	15
c9e313bc SM	16	#include "lttng-relayd.hpp"
	17	#include "stream.hpp"
	18	#include "index.hpp"
	19	#include "connection.hpp"
1c20f0e2 JD	20
1c20f0e2 JD	21	/*
7591bab1 MD	22	* Allocate a new relay index object. Pass the stream in which it is
	23	* contained as parameter. The sequence number will be used as the hash
	24	* table key.
	25	*
	26	* Called with stream mutex held.
	27	* Return allocated object or else NULL on error.
1c20f0e2	28	*/
7591bab1 MD	29	static struct relay_index relay_index_create(struct relay_stream stream,
7591bab1 MD	30	uint64_t net_seq_num)
1c20f0e2	31	{
7591bab1	32	struct relay_index *index;
1c20f0e2	33
7591bab1 MD	34	DBG2("Creating relay index for stream id %" PRIu64 " and seqnum %" PRIu64,
7591bab1 MD	35	stream->stream_handle, net_seq_num);
1c20f0e2	36
64803277	37	index = zmalloc<relay_index>();
7591bab1 MD	38	if (!index) {
	39	PERROR("Relay index zmalloc");
	40	goto end;
	41	}
	42	if (!stream_get(stream)) {
	43	ERR("Cannot get stream");
	44	free(index);
	45	index = NULL;
	46	goto end;
1c20f0e2	47	}
7591bab1	48	index->stream = stream;
1c20f0e2	49
7591bab1 MD	50	lttng_ht_node_init_u64(&index->index_n, net_seq_num);
7591bab1 MD	51	pthread_mutex_init(&index->lock, NULL);
7591bab1 MD	52	urcu_ref_init(&index->ref);
	53
	54	end:
	55	return index;
1c20f0e2 JD	56	}
	57
	58	/*
7591bab1 MD	59	* Add unique relay index to the given hash table. In case of a collision, the
7591bab1 MD	60	* already existing object is put in the given _index variable.
1c20f0e2	61	*
7591bab1	62	* RCU read side lock MUST be acquired.
1c20f0e2	63	*/
7591bab1 MD	64	static struct relay_index relay_index_add_unique(struct relay_stream stream,
7591bab1 MD	65	struct relay_index *index)
1c20f0e2	66	{
7591bab1 MD	67	struct cds_lfht_node *node_ptr;
7591bab1 MD	68	struct relay_index *_index;
1c20f0e2	69
48b7cdc2 FD	70	ASSERT_RCU_READ_LOCKED();
48b7cdc2 FD	71
7591bab1 MD	72	DBG2("Adding relay index with stream id %" PRIu64 " and seqnum %" PRIu64,
7591bab1 MD	73	stream->stream_handle, index->index_n.key);
1c20f0e2	74
7591bab1 MD	75	node_ptr = cds_lfht_add_unique(stream->indexes_ht->ht,
	76	stream->indexes_ht->hash_fct(&index->index_n, lttng_ht_seed),
	77	stream->indexes_ht->match_fct, &index->index_n,
	78	&index->index_n.node);
	79	if (node_ptr != &index->index_n.node) {
	80	_index = caa_container_of(node_ptr, struct relay_index,
	81	index_n.node);
	82	} else {
	83	_index = NULL;
1c20f0e2	84	}
7591bab1 MD	85	return _index;
	86	}
	87
	88	/*
	89	* Should be called with RCU read-side lock held.
	90	*/
	91	static bool relay_index_get(struct relay_index *index)
	92	{
48b7cdc2 FD	93	ASSERT_RCU_READ_LOCKED();
48b7cdc2 FD	94
7591bab1 MD	95	DBG2("index get for stream id %" PRIu64 " and seqnum %" PRIu64 " refcount %d",
	96	index->stream->stream_handle, index->index_n.key,
	97	(int) index->ref.refcount);
1c20f0e2	98
ce4d4083	99	return urcu_ref_get_unless_zero(&index->ref);
1c20f0e2 JD	100	}
	101
	102	/*
7591bab1 MD	103	* Get a relayd index in within the given stream, or create it if not
7591bab1 MD	104	* present.
1c20f0e2	105	*
7591bab1	106	* Called with stream mutex held.
1c20f0e2 JD	107	* Return index object or else NULL on error.
1c20f0e2 JD	108	*/
7591bab1 MD	109	struct relay_index relay_index_get_by_id_or_create(struct relay_stream stream,
7591bab1 MD	110	uint64_t net_seq_num)
1c20f0e2	111	{
7591bab1	112	struct lttng_ht_node_u64 *node;
1c20f0e2	113	struct lttng_ht_iter iter;
1c20f0e2 JD	114	struct relay_index *index = NULL;
1c20f0e2 JD	115
1c20f0e2	116	DBG3("Finding index for stream id %" PRIu64 " and seq_num %" PRIu64,
7591bab1	117	stream->stream_handle, net_seq_num);
1c20f0e2	118
7591bab1 MD	119	rcu_read_lock();
	120	lttng_ht_lookup(stream->indexes_ht, &net_seq_num, &iter);
	121	node = lttng_ht_iter_get_node_u64(&iter);
	122	if (node) {
	123	index = caa_container_of(node, struct relay_index, index_n);
	124	} else {
	125	struct relay_index *oldindex;
	126
	127	index = relay_index_create(stream, net_seq_num);
	128	if (!index) {
	129	ERR("Cannot create index for stream id %" PRIu64 " and seq_num %" PRIu64,
811a4037	130	stream->stream_handle, net_seq_num);
7591bab1 MD	131	goto end;
	132	}
	133	oldindex = relay_index_add_unique(stream, index);
	134	if (oldindex) {
	135	/* Added concurrently, keep old. */
	136	relay_index_put(index);
	137	index = oldindex;
	138	if (!relay_index_get(index)) {
	139	index = NULL;
	140	}
	141	} else {
	142	stream->indexes_in_flight++;
	143	index->in_hash_table = true;
	144	}
1c20f0e2	145	}
1c20f0e2	146	end:
7591bab1 MD	147	rcu_read_unlock();
7591bab1 MD	148	DBG2("Index %sfound or created in HT for stream ID %" PRIu64 " and seqnum %" PRIu64,
719155c2	149	(index == NULL) ? "NOT " : "", stream->stream_handle, net_seq_num);
1c20f0e2 JD	150	return index;
	151	}
	152
f8f3885c MD	153	int relay_index_set_file(struct relay_index *index,
f8f3885c MD	154	struct lttng_index_file *index_file,
7591bab1	155	uint64_t data_offset)
1c20f0e2	156	{
7591bab1	157	int ret = 0;
1c20f0e2	158
7591bab1	159	pthread_mutex_lock(&index->lock);
f8f3885c	160	if (index->index_file) {
7591bab1 MD	161	ret = -1;
	162	goto end;
	163	}
f8f3885c MD	164	lttng_index_file_get(index_file);
f8f3885c MD	165	index->index_file = index_file;
7591bab1 MD	166	index->index_data.offset = data_offset;
	167	end:
	168	pthread_mutex_unlock(&index->lock);
	169	return ret;
	170	}
1c20f0e2	171
7591bab1	172	int relay_index_set_data(struct relay_index *index,
211d28db	173	const struct ctf_packet_index *data)
7591bab1 MD	174	{
7591bab1 MD	175	int ret = 0;
1c20f0e2	176
7591bab1 MD	177	pthread_mutex_lock(&index->lock);
	178	if (index->has_index_data) {
	179	ret = -1;
	180	goto end;
1c20f0e2	181	}
7591bab1 MD	182	/* Set everything except data_offset. */
	183	index->index_data.packet_size = data->packet_size;
	184	index->index_data.content_size = data->content_size;
	185	index->index_data.timestamp_begin = data->timestamp_begin;
	186	index->index_data.timestamp_end = data->timestamp_end;
	187	index->index_data.events_discarded = data->events_discarded;
	188	index->index_data.stream_id = data->stream_id;
	189	index->has_index_data = true;
	190	end:
	191	pthread_mutex_unlock(&index->lock);
	192	return ret;
1c20f0e2 JD	193	}
1c20f0e2 JD	194
7591bab1	195	static void index_destroy(struct relay_index *index)
1c20f0e2	196	{
7591bab1 MD	197	free(index);
7591bab1 MD	198	}
1c20f0e2	199
7591bab1 MD	200	static void index_destroy_rcu(struct rcu_head *rcu_head)
	201	{
	202	struct relay_index *index =
	203	caa_container_of(rcu_head, struct relay_index, rcu_node);
1c20f0e2	204
7591bab1	205	index_destroy(index);
1c20f0e2 JD	206	}
1c20f0e2 JD	207
7591bab1 MD	208	/* Stream lock must be held by the caller. */
7591bab1 MD	209	static void index_release(struct urcu_ref *ref)
1c20f0e2	210	{
7591bab1 MD	211	struct relay_index *index = caa_container_of(ref, struct relay_index, ref);
	212	struct relay_stream *stream = index->stream;
	213	int ret;
	214	struct lttng_ht_iter iter;
	215
f8f3885c MD	216	if (index->index_file) {
	217	lttng_index_file_put(index->index_file);
	218	index->index_file = NULL;
7591bab1 MD	219	}
	220	if (index->in_hash_table) {
	221	/* Delete index from hash table. */
	222	iter.iter.node = &index->index_n.node;
	223	ret = lttng_ht_del(stream->indexes_ht, &iter);
a0377dfe	224	LTTNG_ASSERT(!ret);
7591bab1 MD	225	stream->indexes_in_flight--;
	226	}
	227
	228	stream_put(index->stream);
	229	index->stream = NULL;
	230
	231	call_rcu(&index->rcu_node, index_destroy_rcu);
1c20f0e2 JD	232	}
	233
	234	/*
7591bab1 MD	235	* Called with stream mutex held.
	236	*
	237	* Stream lock must be held by the caller.
1c20f0e2	238	*/
7591bab1	239	void relay_index_put(struct relay_index *index)
1c20f0e2	240	{
7591bab1 MD	241	DBG2("index put for stream id %" PRIu64 " and seqnum %" PRIu64 " refcount %d",
	242	index->stream->stream_handle, index->index_n.key,
	243	(int) index->ref.refcount);
	244	/*
0f1b1d25	245	* Ensure existence of index->lock for index unlock.
7591bab1 MD	246	*/
	247	rcu_read_lock();
	248	/*
	249	* Index lock ensures that concurrent test and update of stream
	250	* ref is atomic.
	251	*/
a0377dfe	252	LTTNG_ASSERT(index->ref.refcount != 0);
7591bab1	253	urcu_ref_put(&index->ref, index_release);
7591bab1	254	rcu_read_unlock();
1c20f0e2 JD	255	}
	256
	257	/*
7591bab1 MD	258	* Try to flush index to disk. Releases self-reference to index once
7591bab1 MD	259	* flush succeeds.
1c20f0e2	260	*
7591bab1 MD	261	* Stream lock must be held by the caller.
	262	* Return 0 on successful flush, a negative value on error, or positive
	263	* value if no flush was performed.
1c20f0e2	264	*/
7591bab1	265	int relay_index_try_flush(struct relay_index *index)
1c20f0e2	266	{
7591bab1 MD	267	int ret = 1;
7591bab1 MD	268	bool flushed = false;
1c20f0e2	269
7591bab1 MD	270	pthread_mutex_lock(&index->lock);
	271	if (index->flushed) {
	272	goto skip;
	273	}
	274	/* Check if we are ready to flush. */
f8f3885c	275	if (!index->has_index_data \|\| !index->index_file) {
7591bab1 MD	276	goto skip;
7591bab1 MD	277	}
8bb66c3c JG	278
	279	DBG2("Writing index for stream ID %" PRIu64 " and seq num %" PRIu64,
	280	index->stream->stream_handle, index->index_n.key);
7591bab1 MD	281	flushed = true;
7591bab1 MD	282	index->flushed = true;
f8f3885c	283	ret = lttng_index_file_write(index->index_file, &index->index_data);
7591bab1 MD	284	skip:
7591bab1 MD	285	pthread_mutex_unlock(&index->lock);
1c20f0e2	286
7591bab1 MD	287	if (flushed) {
	288	/* Put self-ref from index now that it has been flushed. */
	289	relay_index_put(index);
	290	}
	291	return ret;
1c20f0e2 JD	292	}
	293
	294	/*
7591bab1 MD	295	* Close every relay index within a given stream, without flushing
7591bab1 MD	296	* them.
1c20f0e2	297	*/
7591bab1	298	void relay_index_close_all(struct relay_stream *stream)
1c20f0e2 JD	299	{
	300	struct lttng_ht_iter iter;
	301	struct relay_index *index;
	302
1c20f0e2	303	rcu_read_lock();
7591bab1 MD	304	cds_lfht_for_each_entry(stream->indexes_ht->ht, &iter.iter,
	305	index, index_n.node) {
	306	/* Put self-ref from index. */
	307	relay_index_put(index);
1c20f0e2 JD	308	}
	309	rcu_read_unlock();
	310	}
3d07a857 MD	311
	312	void relay_index_close_partial_fd(struct relay_stream *stream)
	313	{
	314	struct lttng_ht_iter iter;
	315	struct relay_index *index;
	316
	317	rcu_read_lock();
	318	cds_lfht_for_each_entry(stream->indexes_ht->ht, &iter.iter,
	319	index, index_n.node) {
f8f3885c	320	if (!index->index_file) {
3d07a857 MD	321	continue;
	322	}
	323	/*
f8f3885c	324	* Partial index has its index_file: we have only
3d07a857 MD	325	* received its info from the data socket.
	326	* Put self-ref from index.
	327	*/
	328	relay_index_put(index);
	329	}
	330	rcu_read_unlock();
	331	}
	332
	333	uint64_t relay_index_find_last(struct relay_stream *stream)
	334	{
	335	struct lttng_ht_iter iter;
	336	struct relay_index *index;
	337	uint64_t net_seq_num = -1ULL;
	338
	339	rcu_read_lock();
	340	cds_lfht_for_each_entry(stream->indexes_ht->ht, &iter.iter,
	341	index, index_n.node) {
	342	if (net_seq_num == -1ULL \|\|
	343	index->index_n.key > net_seq_num) {
	344	net_seq_num = index->index_n.key;
	345	}
	346	}
	347	rcu_read_unlock();
	348	return net_seq_num;
	349	}
d3ecc550 JD	350
	351	/*
	352	* Update the index file of an already existing relay_index.
	353	* Offsets by 'removed_data_count' the offset field of an index.
	354	*/
	355	static
	356	int relay_index_switch_file(struct relay_index *index,
	357	struct lttng_index_file *new_index_file,
	358	uint64_t removed_data_count)
	359	{
	360	int ret = 0;
	361	uint64_t offset;
	362
	363	pthread_mutex_lock(&index->lock);
	364	if (!index->index_file) {
	365	ERR("No index_file");
	366	ret = 0;
	367	goto end;
	368	}
	369
	370	lttng_index_file_put(index->index_file);
	371	lttng_index_file_get(new_index_file);
	372	index->index_file = new_index_file;
	373	offset = be64toh(index->index_data.offset);
	374	index->index_data.offset = htobe64(offset - removed_data_count);
	375
	376	end:
	377	pthread_mutex_unlock(&index->lock);
	378	return ret;
	379	}
	380
	381	/*
	382	* Switch the index file of all pending indexes for a stream and update the
	383	* data offset by substracting the last safe position.
	384	* Stream lock must be held.
	385	*/
	386	int relay_index_switch_all_files(struct relay_stream *stream)
	387	{
	388	struct lttng_ht_iter iter;
	389	struct relay_index *index;
	390	int ret = 0;
	391
	392	rcu_read_lock();
	393	cds_lfht_for_each_entry(stream->indexes_ht->ht, &iter.iter,
	394	index, index_n.node) {
d3ecc550 JD	395	ret = relay_index_switch_file(index, stream->index_file,
	396	stream->pos_after_last_complete_data_index);
	397	if (ret) {
	398	goto end;
	399	}
	400	}
	401	end:
	402	rcu_read_unlock();
	403	return ret;
	404	}
c35f9726 JG	405
	406	/*
	407	* Set index data from the control port to a given index object.
	408	*/
	409	int relay_index_set_control_data(struct relay_index *index,
	410	const struct lttcomm_relayd_index *data,
	411	unsigned int minor_version)
	412	{
	413	/* The index on disk is encoded in big endian. */
ac497a37 SM	414	ctf_packet_index index_data {};
	415
	416	index_data.packet_size = htobe64(data->packet_size);
	417	index_data.content_size = htobe64(data->content_size);
	418	index_data.timestamp_begin = htobe64(data->timestamp_begin);
	419	index_data.timestamp_end = htobe64(data->timestamp_end);
	420	index_data.events_discarded = htobe64(data->events_discarded);
	421	index_data.stream_id = htobe64(data->stream_id);
c35f9726 JG	422
	423	if (minor_version >= 8) {
	424	index->index_data.stream_instance_id = htobe64(data->stream_instance_id);
	425	index->index_data.packet_seq_num = htobe64(data->packet_seq_num);
0f83d1cc MD	426	} else {
	427	uint64_t unset_value = -1ULL;
	428
	429	index->index_data.stream_instance_id = htobe64(unset_value);
	430	index->index_data.packet_seq_num = htobe64(unset_value);
c35f9726 JG	431	}
	432
	433	return relay_index_set_data(index, &index_data);
	434	}