LTTNG_CONSUMER_STREAMS_SENT,
LTTNG_CONSUMER_DISCARDED_EVENTS,
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 */
/* Metadata cache is metadata channel */
struct consumer_metadata_cache *metadata_cache;
+
/* For UST metadata periodical flush */
int switch_timer_enabled;
timer_t switch_timer;
timer_t live_timer;
int live_timer_error;
+ /* For channel monitoring timer. */
+ int monitor_timer_enabled;
+ timer_t monitor_timer;
+
/* On-disk circular buffer */
uint64_t tracefile_size;
uint64_t tracefile_count;
uint64_t discarded_events;
/* Total number of missed packets due to overwriting (overwrite). */
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;
};
/*
int data_read;
int hangup_flush_done;
+ /*
+ * Whether the stream is in a "complete" state (e.g. it does not have a
+ * partially written sub-buffer.
+ *
+ * Initialized to "false" on stream creation (first packet is empty).
+ *
+ * The various transitions of the quiescent state are:
+ * - On "start" tracing: set to false, since the stream is not
+ * "complete".
+ * - On "stop" tracing: if !quiescent -> flush FINAL (update
+ * timestamp_end), and set to true; the stream has entered a
+ * complete/quiescent state.
+ * - On "destroy" or stream/application hang-up: if !quiescent ->
+ * flush FINAL, and set to true.
+ *
+ * NOTE: Update and read are protected by the stream lock.
+ */
+ bool quiescent;
+
/*
* metadata_timer_lock protects flags waiting_on_metadata and
* missed_metadata_flush.
* Lock to use the stream FDs since they are used between threads.
*
* This is nested INSIDE the consumer_data lock.
- * This is nested INSIDE the metadata cache lock.
* This is nested INSIDE the channel lock.
* This is nested INSIDE the channel timer lock.
+ * This is nested OUTSIDE the metadata cache lock.
* This is nested OUTSIDE consumer_relayd_sock_pair lock.
*/
pthread_mutex_t lock;
/* Copy of the sequence number of the last packet extracted. */
uint64_t last_sequence_number;
/*
- * FD of the index file for this stream.
+ * Index file object of the index file for this stream.
*/
- int index_fd;
+ struct lttng_index_file *index_file;
/*
* Local pipe to extract data when using splice.
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;
/*
int consumer_should_quit[2];
/* Metadata poll thread pipe. Transfer metadata stream to it */
struct lttng_pipe *consumer_metadata_pipe;
+ /*
+ * Pipe used by the channel monitoring timers to provide state samples
+ * 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;
};
/*
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;
+
/*
* Init consumer data structures.
*/
*/
void lttng_consumer_cleanup(void);
-/*
- * Flush pending writes to trace output disk file.
- */
-void lttng_consumer_sync_trace_file(struct lttng_consumer_stream *stream,
- off_t orig_offset);
-
/*
* Poll on the should_quit pipe and the command socket return -1 on error and
* should exit, 0 if data is available on the command socket
*/
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);
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);
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 */
projects / lttng-tools.git / blobdiff