Implement the RELAYD_ROTATE_STREAM relay daemon command

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

diff --git a/src/bin/lttng-relayd/index.c b/src/bin/lttng-relayd/index.c
index 1b14e3e078e1d96fb0d88f93301fac8e0c3e2d13..92d4581d124a95921694129cba29213c08a7bd90 100644 (file)
--- 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"
@@ -354,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;
+}