X-Git-Url: https://git.lttng.org/?a=blobdiff_plain;f=src%2Fcommon%2Fconsumer%2Fconsumer.h;h=9e77ff999601f1ed3efddd67e5dcf378eb5c2760;hb=d73bf3d793ee0b0c5b56cb47cb50c27d1789d3bd;hp=527331298c0440ece041d2a73a5296352aa8304b;hpb=b35308203f7844f48542978fd00aab21f2057c17;p=lttng-tools.git diff --git a/src/common/consumer/consumer.h b/src/common/consumer/consumer.h index 527331298..9e77ff999 100644 --- 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_RENAME, + LTTNG_CONSUMER_MKDIR, }; /* State of each fd in consumer */ @@ -161,6 +164,7 @@ struct lttng_consumer_channel { /* Metadata cache is metadata channel */ struct consumer_metadata_cache *metadata_cache; + /* For UST metadata periodical flush */ int switch_timer_enabled; timer_t switch_timer; @@ -223,6 +227,22 @@ struct lttng_consumer_channel { 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; + + /* + * The chunk id where we currently write the data. This value is sent + * to the relay when we add a stream and when a stream rotates. This + * allows to keep track of where each stream on the relay is writing. + */ + uint64_t current_chunk_id; }; /* @@ -410,6 +430,17 @@ struct lttng_consumer_stream { pthread_cond_t metadata_rdv; pthread_mutex_t metadata_rdv_lock; + /* + * 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. + */ + struct { + char path[LTTNG_PATH_MAX]; + uint64_t tracefile_size; + } channel_read_only_attributes; + /* Indicate if the stream still has some data to be read. */ unsigned int has_data:1; /* @@ -549,6 +580,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; }; /* @@ -601,6 +637,21 @@ 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; + /* * Init consumer data structures. */ @@ -642,6 +693,14 @@ 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 (read-only) + * directly from the stream. + */ +void consumer_stream_update_channel_attributes( + 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, @@ -705,9 +764,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); @@ -720,7 +782,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); @@ -736,10 +798,17 @@ void consumer_destroy_relayd(struct consumer_relayd_sock_pair *relayd); unsigned long consumer_get_consume_start_pos(unsigned long consumed_pos, unsigned long produced_pos, uint64_t nb_packets_per_stream, uint64_t max_sb_size); -int consumer_add_data_stream(struct lttng_consumer_stream *stream); +void consumer_add_data_stream(struct lttng_consumer_stream *stream); void consumer_del_stream_for_data(struct lttng_consumer_stream *stream); -int consumer_add_metadata_stream(struct lttng_consumer_stream *stream); +void 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_stream(struct lttng_consumer_local_data *ctx, + struct lttng_consumer_stream *stream); +int lttng_consumer_rotate_rename(const char *current_path, const char *new_path, + uid_t uid, gid_t gid, uint64_t relayd_id); +void lttng_consumer_reset_stream_rotate_state(struct lttng_consumer_stream *stream); +int lttng_consumer_mkdir(const char *path, uid_t uid, gid_t gid, + uint64_t relayd_id); #endif /* LIB_CONSUMER_H */