X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fconsumer.h;h=2003cbe43003f304c7bf60387416b9a5c8a1da76;hp=fa503ea111836c18335eb7863123275daa2a6edc;hb=567eb353c7f88e2fdaa106eb7e0a38dbb8717792;hpb=da009f2c0baef56316db8087abd1867f99969374 diff --git a/src/common/consumer.h b/src/common/consumer.h index fa503ea11..2003cbe43 100644 --- a/src/common/consumer.h +++ b/src/common/consumer.h @@ -115,7 +115,7 @@ struct lttng_consumer_channel { char pathname[PATH_MAX]; /* Channel name. */ char name[LTTNG_SYMBOL_NAME_LEN]; - /* UID and GID of the channel. */ + /* UID and GID of the session owning this channel. */ uid_t uid; gid_t gid; /* Relayd id of the channel. -1ULL if it does not apply. */ @@ -131,6 +131,7 @@ struct lttng_consumer_channel { enum consumer_channel_type type; /* For UST */ + uid_t ust_app_uid; /* Application UID. */ struct ustctl_consumer_channel *uchan; unsigned char uuid[UUID_STR_LEN]; /* @@ -140,12 +141,6 @@ struct lttng_consumer_channel { */ struct stream_list streams; - /* - * List of streams in no monitor mode for this channel. Used ONLY for - * snapshots recording. - */ - struct stream_list stream_no_monitor_list; - /* * Set if the channel is metadata. We keep a reference to the stream * because we have to flush data once pushed by the session daemon. For a @@ -174,6 +169,33 @@ struct lttng_consumer_channel { * monitor list of the channel. */ unsigned int monitor; + + /* + * Channel lock. + * + * This lock protects against concurrent update of channel. + * + * This is nested INSIDE the consumer data lock. + * This is nested OUTSIDE the channel timer lock. + * This is nested OUTSIDE the metadata cache lock. + * This is nested OUTSIDE stream lock. + * This is nested OUTSIDE consumer_relayd_sock_pair lock. + */ + pthread_mutex_t lock; + + /* + * Channel teardown lock. + * + * This lock protect against teardown of channel. It is _never_ + * taken by the timer handler. + * + * This is nested INSIDE the consumer data lock. + * This is nested INSIDE the channel lock. + * This is nested OUTSIDE the metadata cache lock. + * This is nested OUTSIDE stream lock. + * This is nested OUTSIDE consumer_relayd_sock_pair lock. + */ + pthread_mutex_t timer_lock; }; /* @@ -222,6 +244,14 @@ struct lttng_consumer_stream { gid_t gid; /* Network sequence number. Indicating on which relayd socket it goes. */ uint64_t net_seq_idx; + /* + * 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 + * stream is closed before decrementing the refcount in order to avoid an + * unbalanced state. + */ + unsigned int sent_to_relayd; + /* Identify if the stream is the metadata */ unsigned int metadata_flag; /* Used when the stream is set for network streaming */ @@ -245,6 +275,8 @@ struct lttng_consumer_stream { * * 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 consumer_relayd_sock_pair lock. */ pthread_mutex_t lock; @@ -265,9 +297,28 @@ struct lttng_consumer_stream { /* On-disk circular buffer */ uint64_t tracefile_size_current; uint64_t tracefile_count_current; - - /* Node for the no monitor stream list in a channel. */ - struct cds_list_head no_monitor_node; + /* + * Monitor or not the streams of this channel meaning this indicates if the + * streams should be sent to the data/metadata thread or added to the no + * monitor list of the channel. + */ + unsigned int monitor; + /* + * Indicate if the stream is globally visible meaning that it has been + * added to the multiple hash tables. If *not* set, NO lock should be + * acquired in the destroy path. + */ + unsigned int globally_visible; + /* + * Pipe to wake up the metadata poll thread when the UST metadata + * cache is updated. + */ + int ust_metadata_poll_pipe[2]; + /* + * How much metadata was read from the metadata cache and sent + * to the channel. + */ + uint64_t ust_metadata_pushed; }; /* @@ -360,12 +411,18 @@ struct lttng_consumer_local_data { * == 0 (success, FD is left to library) * < 0 (error) */ - int (*on_update_stream)(int sessiond_key, uint32_t state); + int (*on_update_stream)(uint64_t sessiond_key, uint32_t state); enum lttng_consumer_type type; /* socket to communicate errors with sessiond */ int consumer_error_socket; - /* socket to ask metadata to sessiond */ + /* socket to ask metadata to sessiond. */ int consumer_metadata_socket; + /* + * Protect consumer_metadata_socket. + * + * This is nested OUTSIDE the metadata cache lock. + */ + pthread_mutex_t metadata_socket_lock; /* socket to exchange commands with sessiond */ char *consumer_command_sock_path; /* communication with splice */ @@ -487,7 +544,8 @@ struct lttng_consumer_stream *consumer_allocate_stream(uint64_t channel_key, uint64_t session_id, int cpu, int *alloc_ret, - enum consumer_channel_type type); + enum consumer_channel_type type, + unsigned int monitor); struct lttng_consumer_channel *consumer_allocate_channel(uint64_t key, uint64_t session_id, const char *pathname, @@ -512,6 +570,8 @@ void consumer_del_channel(struct lttng_consumer_channel *channel); struct consumer_relayd_sock_pair *consumer_allocate_relayd_sock_pair( uint64_t net_seq_idx); struct consumer_relayd_sock_pair *consumer_find_relayd(uint64_t key); +int consumer_send_relayd_stream(struct lttng_consumer_stream *stream, char *path); +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, size_t data_size); @@ -523,7 +583,7 @@ struct lttng_consumer_local_data *lttng_consumer_create( struct lttng_consumer_local_data *ctx), int (*recv_channel)(struct lttng_consumer_channel *channel), int (*recv_stream)(struct lttng_consumer_stream *stream), - int (*update_stream)(int sessiond_key, uint32_t state)); + 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, @@ -549,7 +609,7 @@ int lttng_consumer_on_recv_stream(struct lttng_consumer_stream *stream); int 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, - unsigned int sessiond_id); + uint64_t sessiond_id); void consumer_flag_relayd_for_destroy( struct consumer_relayd_sock_pair *relayd); int consumer_data_pending(uint64_t id); @@ -559,5 +619,11 @@ int consumer_send_status_channel(int sock, void notify_thread_del_channel(struct lttng_consumer_local_data *ctx, uint64_t key); void consumer_destroy_relayd(struct consumer_relayd_sock_pair *relayd); +unsigned long consumer_get_consumed_maxsize(unsigned long consumed_pos, + unsigned long produced_pos, uint64_t max_stream_size); +int 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_del_stream_for_metadata(struct lttng_consumer_stream *stream); #endif /* LIB_CONSUMER_H */