Move mkdir/rename code to the consumer

[lttng-tools.git] / src / common / consumer / consumer.h

diff --git a/src/common/consumer/consumer.h b/src/common/consumer/consumer.h
index 7375373c76113ab791a5e06027e8f0599a1494f6..3aa63eed1fecc21dc415a8dbf1af68c09cf40524 100644 (file)
--- a/src/common/consumer/consumer.h
+++ b/src/common/consumer/consumer.h
@@ -62,6 +62,9 @@ enum lttng_consumer_command {
        LTTNG_CONSUMER_LOST_PACKETS,
        LTTNG_CONSUMER_CLEAR_QUIESCENT_CHANNEL,
        LTTNG_CONSUMER_SET_CHANNEL_MONITOR_PIPE,
+       LTTNG_CONSUMER_SET_CHANNEL_ROTATE_PIPE,
+       LTTNG_CONSUMER_ROTATE_CHANNEL,
+       LTTNG_CONSUMER_ROTATE_RENAME,
 };
 
 /* State of each fd in consumer */
@@ -226,6 +229,13 @@ struct lttng_consumer_channel {
        uint64_t lost_packets;
 
        bool streams_sent_to_relayd;
+
+       /*
+        * Account how many streams are waiting for their rotation to be
+        * complete. When this number reaches 0, we inform the session
+        * daemon that this channel has finished its rotation.
+        */
+       uint64_t nr_stream_rotate_pending;
 };
 
 /*
@@ -413,6 +423,37 @@ struct lttng_consumer_stream {
        pthread_cond_t metadata_rdv;
        pthread_mutex_t metadata_rdv_lock;
 
+       /*
+        * If rotate_position != 0, when we reach this position in the
+        * ring-buffer, close this tracefile and create a new one in
+        * chan->pathname.
+        */
+       uint64_t rotate_position;
+
+       /*
+        * Read-only copies of channel values. We cannot safely access the
+        * channel from a stream, so we need to have a local copy of these
+        * fields in the stream object. These fields should be removed from
+        * the stream objects when we introduce refcounting.
+        */
+       char channel_ro_pathname[PATH_MAX];
+       uint64_t channel_ro_tracefile_size;
+
+       /*
+        * If rotate_ready is set to 1, rotate the stream the next time data
+        * need to be extracted, regardless of the rotate_position. This is
+        * used if all the metadata has been consumed when we rotate. In this
+        * case, the snapshot of the positions returns -EAGAIN and we cannot
+        * use the produced/consumed positions as reference.
+        */
+       unsigned int rotate_ready:1;
+       /*
+        * Flag set to 1 if the stream just got rotated. This is used to
+        * perform actions on the channel after a rotation without needing
+        * to nest the channel lock inside the stream lock.
+        */
+       unsigned int rotated:1;
+
        /* Indicate if the stream still has some data to be read. */
        unsigned int has_data:1;
        /*
@@ -552,6 +593,11 @@ struct lttng_consumer_local_data {
         * to the session daemon (write-only).
         */
        int channel_monitor_pipe;
+       /*
+        * Pipe used to inform the session daemon that a stream has finished
+        * its rotation (write-only).
+        */
+       int channel_rotate_pipe;
 };
 
 /*
@@ -604,6 +650,18 @@ struct lttng_consumer_global_data {
        struct lttng_ht *stream_per_chan_id_ht;
 };
 
+/*
+ * Set to nonzero when the consumer is exiting. Updated by signal
+ * handler and thread exit, read by threads.
+ */
+extern int consumer_quit;
+
+/*
+ * Set to nonzero when the consumer is exiting. Updated by signal
+ * handler and thread exit, read by threads.
+ */
+extern int consumer_quit;
+
 /* Flag used to temporarily pause data consumption from testpoints. */
 extern int data_consumption_paused;
 
@@ -648,6 +706,13 @@ void lttng_consumer_cleanup(void);
  */
 int lttng_consumer_poll_socket(struct pollfd *kconsumer_sockpoll);
 
+/*
+ * Copy the fields from the channel that need to be accessed in read-only
+ * directly from the stream.
+ */
+void consumer_stream_copy_ro_channel_values(struct lttng_consumer_stream *stream,
+               struct lttng_consumer_channel *channel);
+
 struct lttng_consumer_stream *consumer_allocate_stream(uint64_t channel_key,
                uint64_t stream_key,
                enum lttng_consumer_stream_state state,
@@ -711,9 +776,12 @@ ssize_t lttng_consumer_on_read_subbuffer_splice(
                struct lttng_consumer_stream *stream, unsigned long len,
                unsigned long padding,
                struct ctf_packet_index *index);
+int lttng_consumer_sample_snapshot_positions(struct lttng_consumer_stream *stream);
 int lttng_consumer_take_snapshot(struct lttng_consumer_stream *stream);
 int lttng_consumer_get_produced_snapshot(struct lttng_consumer_stream *stream,
                unsigned long *pos);
+int lttng_consumer_get_consumed_snapshot(struct lttng_consumer_stream *stream,
+               unsigned long *pos);
 int lttng_ustconsumer_get_wakeup_fd(struct lttng_consumer_stream *stream);
 int lttng_ustconsumer_close_wakeup_fd(struct lttng_consumer_stream *stream);
 void *consumer_thread_metadata_poll(void *data);
@@ -726,7 +794,7 @@ int lttng_consumer_recv_cmd(struct lttng_consumer_local_data *ctx,
 ssize_t lttng_consumer_read_subbuffer(struct lttng_consumer_stream *stream,
                struct lttng_consumer_local_data *ctx);
 int lttng_consumer_on_recv_stream(struct lttng_consumer_stream *stream);
-int consumer_add_relayd_socket(uint64_t net_seq_idx, int sock_type,
+void consumer_add_relayd_socket(uint64_t net_seq_idx, int sock_type,
                struct lttng_consumer_local_data *ctx, int sock,
                struct pollfd *consumer_sockpoll, struct lttcomm_relayd_sock *relayd_sock,
                uint64_t sessiond_id, uint64_t relayd_session_id);
@@ -747,5 +815,14 @@ void consumer_del_stream_for_data(struct lttng_consumer_stream *stream);
 int consumer_add_metadata_stream(struct lttng_consumer_stream *stream);
 void consumer_del_stream_for_metadata(struct lttng_consumer_stream *stream);
 int consumer_create_index_file(struct lttng_consumer_stream *stream);
+int lttng_consumer_rotate_channel(uint64_t key, char *path,
+               uint64_t relayd_id, uint32_t metadata,
+               struct lttng_consumer_local_data *ctx);
+int lttng_consumer_rotate_stream(struct lttng_consumer_local_data *ctx,
+               struct lttng_consumer_stream *stream);
+int lttng_consumer_rotate_ready_streams(uint64_t key,
+               struct lttng_consumer_local_data *ctx);
+int lttng_consumer_rotate_rename(char *current_path, char *new_path,
+               uint32_t create, uid_t uid, gid_t gid);
 
 #endif /* LIB_CONSUMER_H */