X-Git-Url: https://git.lttng.org/?a=blobdiff_plain;f=src%2Fbin%2Flttng-relayd%2Findex.c;h=92d4581d124a95921694129cba29213c08a7bd90;hb=e2acc8d25bc0073554a2f3e742e070b7034ec1d0;hp=b15bbcd7702954e675a98c06e02ab31058ff7499;hpb=f8f3885cc52af9d3c951da78989d6f4a25270411;p=lttng-tools.git

diff --git a/src/bin/lttng-relayd/index.c b/src/bin/lttng-relayd/index.c
index b15bbcd77..92d4581d1 100644
--- a/src/bin/lttng-relayd/index.c
+++ b/src/bin/lttng-relayd/index.c
@@ -22,6 +22,7 @@
 
 #include <common/common.h>
 #include <common/utils.h>
+#include <common/compat/endian.h>
 
 #include "lttng-relayd.h"
 #include "stream.h"
@@ -58,7 +59,6 @@ static struct relay_index *relay_index_create(struct relay_stream *stream,
 
 	lttng_ht_node_init_u64(&index->index_n, net_seq_num);
 	pthread_mutex_init(&index->lock, NULL);
-	pthread_mutex_init(&index->reflock, NULL);
 	urcu_ref_init(&index->ref);
 
 end:
@@ -98,21 +98,11 @@ static struct relay_index *relay_index_add_unique(struct relay_stream *stream,
  */
 static bool relay_index_get(struct relay_index *index)
 {
-	bool has_ref = false;
-
 	DBG2("index get for stream id %" PRIu64 " and seqnum %" PRIu64 " refcount %d",
 			index->stream->stream_handle, index->index_n.key,
 			(int) index->ref.refcount);
 
-	/* Confirm that the index refcount has not reached 0. */
-	pthread_mutex_lock(&index->reflock);
-	if (index->ref.refcount != 0) {
-		has_ref = true;
-		urcu_ref_get(&index->ref);
-	}
-	pthread_mutex_unlock(&index->reflock);
-
-	return has_ref;
+	return urcu_ref_get_unless_zero(&index->ref);
 }
 
 /*
@@ -265,10 +255,8 @@ void relay_index_put(struct relay_index *index)
 	 * Index lock ensures that concurrent test and update of stream
 	 * ref is atomic.
 	 */
-	pthread_mutex_lock(&index->reflock);
 	assert(index->ref.refcount != 0);
 	urcu_ref_put(&index->ref, index_release);
-	pthread_mutex_unlock(&index->reflock);
 	rcu_read_unlock();
 }
 
@@ -367,3 +355,59 @@ uint64_t relay_index_find_last(struct relay_stream *stream)
 	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) {
+		DBG("Update index to fd %d", stream->index_file->fd);
+		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;
+}