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 */
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;
};
/*
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;
/*
* 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;
};
/*
*/
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,
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);
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 */