X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=src%2Fcommon%2Fconsumer%2Fconsumer.h;h=14b52bb3ee925ecec208c6e3894159a753265b66;hb=0934dd7e7c31433d5976119c05779e3800573e04;hp=acefacb218e3b8316011dff9f7c289bcda9428bd;hpb=e0547b835d76e0a3a2c3082d7c10e6902a1f3e04;p=lttng-tools.git diff --git a/src/common/consumer/consumer.h b/src/common/consumer/consumer.h index acefacb21..14b52bb3e 100644 --- a/src/common/consumer/consumer.h +++ b/src/common/consumer/consumer.h @@ -33,6 +33,10 @@ #include #include #include +#include +#include + +struct lttng_consumer_local_data; /* Commands for consumer */ enum lttng_consumer_command { @@ -61,6 +65,8 @@ enum lttng_consumer_command { LTTNG_CONSUMER_DISCARDED_EVENTS, LTTNG_CONSUMER_LOST_PACKETS, LTTNG_CONSUMER_CLEAR_QUIESCENT_CHANNEL, + LTTNG_CONSUMER_CHANNEL_STOP_LIVE_TIMER, + LTTNG_CONSUMER_CHANNEL_START_LIVE_TIMER, }; /* State of each fd in consumer */ @@ -169,6 +175,8 @@ struct lttng_consumer_channel { int live_timer_enabled; timer_t live_timer; int live_timer_error; + /* Channel is part of a live session ? */ + bool is_live; /* On-disk circular buffer */ uint64_t tracefile_size; @@ -218,8 +226,154 @@ 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; +}; + +struct stream_subbuffer { + union { + /* + * CONSUMER_CHANNEL_SPLICE + * No ownership assumed. + */ + int fd; + /* CONSUMER_CHANNEL_MMAP */ + struct lttng_buffer_view buffer; + } buffer; + union { + /* + * Common members are fine to access through either + * union entries (as per C11, Common Initial Sequence). + */ + struct { + unsigned long subbuf_size; + unsigned long padded_subbuf_size; + uint64_t version; + /* + * Left unset when unsupported. + * + * Indicates that this is the last sub-buffer of + * a series of sub-buffer that makes-up a coherent + * (parseable) unit of metadata. + */ + LTTNG_OPTIONAL(bool) coherent; + } metadata; + struct { + unsigned long subbuf_size; + unsigned long padded_subbuf_size; + uint64_t packet_size; + uint64_t content_size; + uint64_t timestamp_begin; + uint64_t timestamp_end; + uint64_t events_discarded; + /* Left unset when unsupported. */ + LTTNG_OPTIONAL(uint64_t) sequence_number; + uint64_t stream_id; + /* Left unset when unsupported. */ + LTTNG_OPTIONAL(uint64_t) stream_instance_id; + } data; + } info; }; +/* + * Perform any operation required to acknowledge + * the wake-up of a consumer stream (e.g. consume a byte on a wake-up pipe). + * + * Stream and channel locks are acquired during this call. + */ +typedef int (*on_wake_up_cb)(struct lttng_consumer_stream *); + +/* + * Perform any operation required before a consumer stream is put + * to sleep before awaiting a data availability notification. + * + * Stream and channel locks are acquired during this call. + */ +typedef int (*on_sleep_cb)(struct lttng_consumer_stream *, + struct lttng_consumer_local_data *); + +/* + * Acquire the subbuffer at the current 'consumed' position. + * + * Stream and channel locks are acquired during this call. + */ +typedef int (*get_next_subbuffer_cb)(struct lttng_consumer_stream *, + struct stream_subbuffer *); + +/* + * Populate the stream_subbuffer's info member. The info to populate + * depends on the type (metadata/data) of the stream. + * + * Stream and channel locks are acquired during this call. + */ +typedef int (*extract_subbuffer_info_cb)( + struct lttng_consumer_stream *, struct stream_subbuffer *); + +/* + * Invoked after a subbuffer's info has been filled. + * + * Stream and channel locks are acquired during this call. + */ +typedef int (*pre_consume_subbuffer_cb)(struct lttng_consumer_stream *, + const struct stream_subbuffer *); + +/* + * Consume subbuffer contents. + * + * Stream and channel locks are acquired during this call. + */ +typedef ssize_t (*consume_subbuffer_cb)(struct lttng_consumer_local_data *, + struct lttng_consumer_stream *, + const struct stream_subbuffer *); + +/* + * Release the current subbuffer and advance the 'consumed' position by + * one subbuffer. + * + * Stream and channel locks are acquired during this call. + */ +typedef int (*put_next_subbuffer_cb)(struct lttng_consumer_stream *, + struct stream_subbuffer *); + +/* + * Invoked after consuming a subbuffer. + * + * Stream and channel locks are acquired during this call. + */ +typedef int (*post_consume_cb)(struct lttng_consumer_stream *, + const struct stream_subbuffer *, + struct lttng_consumer_local_data *); + +/* + * Send a live beacon if no data is available. + * + * Stream and channel locks are acquired during this call. + */ +typedef int (*send_live_beacon_cb)(struct lttng_consumer_stream *); + +/* + * Lock the stream and channel locks and any other stream-type specific + * lock that need to be acquired during the processing of an + * availability notification. + */ +typedef void (*lock_cb)(struct lttng_consumer_stream *); + +/* + * Unlock the stream and channel locks and any other stream-type specific + * lock before sleeping until the next availability notification. + * + * Stream and channel locks are acquired during this call. + */ +typedef void (*unlock_cb)(struct lttng_consumer_stream *); + +/* + * Invoked when a subbuffer's metadata version does not match the last + * known metadata version. + * + * Stream and channel locks are acquired during this call. + */ +typedef void (*reset_metadata_cb)(struct lttng_consumer_stream *); + /* * Internal representation of the streams, sessiond_key is used to identify * uniquely a stream. @@ -300,8 +454,11 @@ struct lttng_consumer_stream { /* UID/GID of the user owning the session to which stream belongs */ uid_t uid; gid_t gid; - /* Network sequence number. Indicating on which relayd socket it goes. */ - uint64_t net_seq_idx; + /* + * Relayd id, indicating on which relayd socket it goes. Set to -1ULL if + * not the stream is not associated to a relay daemon. + */ + uint64_t relayd_id; /* * Indicate if this stream was successfully sent to a relayd. This is set * after the refcount of the relayd is incremented and is checked when the @@ -337,9 +494,9 @@ struct lttng_consumer_stream { * 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; @@ -412,6 +569,25 @@ struct lttng_consumer_stream { * file before writing in it (regeneration). */ unsigned int reset_metadata_flag:1; + struct { + /* + * Invoked in the order of declaration. + * See callback type definitions. + */ + lock_cb lock; + on_wake_up_cb on_wake_up; + get_next_subbuffer_cb get_next_subbuffer; + extract_subbuffer_info_cb extract_subbuffer_info; + pre_consume_subbuffer_cb pre_consume_subbuffer; + reset_metadata_cb reset_metadata; + consume_subbuffer_cb consume_subbuffer; + put_next_subbuffer_cb put_next_subbuffer; + post_consume_cb post_consume; + send_live_beacon_cb send_live_beacon; + on_sleep_cb on_sleep; + unlock_cb unlock; + } read_subbuffer_ops; + struct metadata_bucket *metadata_bucket; }; /* @@ -419,9 +595,9 @@ struct lttng_consumer_stream { */ struct consumer_relayd_sock_pair { /* Network sequence number. */ - uint64_t net_seq_idx; + uint64_t id; /* Number of stream associated with this relayd */ - unsigned int refcount; + int refcount; /* * This flag indicates whether or not we should destroy this object. The @@ -455,6 +631,7 @@ struct consumer_relayd_sock_pair { /* Session id on both sides for the sockets. */ uint64_t relayd_session_id; uint64_t sessiond_session_id; + struct lttng_consumer_local_data *ctx; }; /* @@ -467,7 +644,8 @@ struct lttng_consumer_local_data { * Returns the number of bytes read, or negative error value. */ ssize_t (*on_buffer_ready)(struct lttng_consumer_stream *stream, - struct lttng_consumer_local_data *ctx); + struct lttng_consumer_local_data *ctx, + bool locked_by_caller); /* * function to call when we receive a new channel, it receives a * newly allocated channel, depending on the return code of this @@ -591,6 +769,12 @@ 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 volatile int consumer_quit; + /* * Init consumer data structures. */ @@ -632,7 +816,9 @@ void lttng_consumer_cleanup(void); */ int lttng_consumer_poll_socket(struct pollfd *kconsumer_sockpoll); -struct lttng_consumer_stream *consumer_allocate_stream(uint64_t channel_key, +struct lttng_consumer_stream *consumer_allocate_stream( + struct lttng_consumer_channel *channel, + uint64_t channel_key, uint64_t stream_key, enum lttng_consumer_stream_state state, const char *channel_name, @@ -657,6 +843,7 @@ struct lttng_consumer_channel *consumer_allocate_channel(uint64_t key, uint64_t session_id_per_pid, unsigned int monitor, unsigned int live_timer_interval, + bool is_in_live_session, const char *root_shm_path, const char *shm_path); void consumer_del_stream(struct lttng_consumer_stream *stream, @@ -670,7 +857,7 @@ void consumer_del_channel(struct lttng_consumer_channel *channel); /* lttng-relayd consumer command */ struct consumer_relayd_sock_pair *consumer_find_relayd(uint64_t key); int consumer_send_relayd_stream(struct lttng_consumer_stream *stream, char *path); -int consumer_send_relayd_streams_sent(uint64_t net_seq_idx); +int consumer_send_relayd_streams_sent(uint64_t relayd_id); void close_relayd_stream(struct lttng_consumer_stream *stream); struct lttng_consumer_channel *consumer_find_channel(uint64_t key); int consumer_handle_stream_before_relayd(struct lttng_consumer_stream *stream, @@ -680,21 +867,20 @@ void consumer_steal_stream_key(int key, struct lttng_ht *ht); struct lttng_consumer_local_data *lttng_consumer_create( enum lttng_consumer_type type, ssize_t (*buffer_ready)(struct lttng_consumer_stream *stream, - struct lttng_consumer_local_data *ctx), + struct lttng_consumer_local_data *ctx, + bool locked_by_caller), int (*recv_channel)(struct lttng_consumer_channel *channel), int (*recv_stream)(struct lttng_consumer_stream *stream), int (*update_stream)(uint64_t sessiond_key, uint32_t state)); void lttng_consumer_destroy(struct lttng_consumer_local_data *ctx); ssize_t lttng_consumer_on_read_subbuffer_mmap( - struct lttng_consumer_local_data *ctx, - struct lttng_consumer_stream *stream, unsigned long len, - unsigned long padding, - struct ctf_packet_index *index); + struct lttng_consumer_stream *stream, + const struct lttng_buffer_view *buffer, + unsigned long padding); ssize_t lttng_consumer_on_read_subbuffer_splice( struct lttng_consumer_local_data *ctx, struct lttng_consumer_stream *stream, unsigned long len, - unsigned long padding, - struct ctf_packet_index *index); + unsigned long padding); int lttng_consumer_take_snapshot(struct lttng_consumer_stream *stream); int lttng_consumer_get_produced_snapshot(struct lttng_consumer_stream *stream, unsigned long *pos); @@ -708,9 +894,10 @@ int lttng_consumer_recv_cmd(struct lttng_consumer_local_data *ctx, int sock, struct pollfd *consumer_sockpoll); ssize_t lttng_consumer_read_subbuffer(struct lttng_consumer_stream *stream, - struct lttng_consumer_local_data *ctx); + struct lttng_consumer_local_data *ctx, + bool locked_by_caller); 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 relayd_id, 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); @@ -731,5 +918,7 @@ 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); +void lttng_consumer_cleanup_relayd(struct consumer_relayd_sock_pair *relayd); +int consumer_send_relayd_channel_bulk(struct lttng_consumer_channel *channel); #endif /* LIB_CONSUMER_H */