X-Git-Url: https://git.lttng.org/?a=blobdiff_plain;f=src%2Fcommon%2Fconsumer.h;h=fe1568382f3971728fb2a15fbc4b7bec348541d5;hb=3cc2f24a5cdabfbcb1022c0798f6b4845f72b498;hp=da3c4ce5a69b75c8eafc5bb9eeefa4cba2778026;hpb=7735ef9e674217413a63bd4a09a93ac0958fe58a;p=lttng-tools.git diff --git a/src/common/consumer.h b/src/common/consumer.h index da3c4ce5a..fe1568382 100644 --- a/src/common/consumer.h +++ b/src/common/consumer.h @@ -77,13 +77,18 @@ struct lttng_consumer_channel { int key; uint64_t max_sb_size; /* the subbuffer size for this channel */ int refcount; /* Number of streams referencing this channel */ + /* + * The number of streams to receive initially. Used to guarantee that we do + * not destroy a channel before receiving all its associated streams. + */ + unsigned int nb_init_streams; + /* For UST */ int shm_fd; int wait_fd; void *mmap_base; size_t mmap_len; struct lttng_ust_shm_handle *handle; - int nr_streams; int wait_fd_is_copy; int cpucount; }; @@ -96,8 +101,8 @@ struct lttng_ust_lib_ring_buffer; * uniquely a stream. */ struct lttng_consumer_stream { + /* Hash table node for both metadata and data type */ struct lttng_ht_node_ulong node; - struct lttng_ht_node_ulong waitfd_node; struct lttng_consumer_channel *chan; /* associated channel */ /* * key is the key used by the session daemon to refer to the @@ -224,10 +229,13 @@ struct lttng_consumer_local_data { char *consumer_command_sock_path; /* communication with splice */ int consumer_thread_pipe[2]; - /* pipe to wake the poll thread when necessary */ - int consumer_poll_pipe[2]; + int consumer_splice_metadata_pipe[2]; + /* Data stream poll thread pipe. To transfer data stream to the thread */ + int consumer_data_pipe[2]; /* to let the signal handler wake up the fd receiver thread */ int consumer_should_quit[2]; + /* Metadata poll thread pipe. Transfer metadata stream to it */ + int consumer_metadata_pipe[2]; }; /* @@ -245,13 +253,12 @@ struct lttng_consumer_global_data { pthread_mutex_t lock; /* - * Number of streams in the hash table. Protected by consumer_data.lock. + * Number of streams in the data stream hash table declared outside. + * Protected by consumer_data.lock. */ int stream_count; - /* - * Hash tables of streams and channels. Protected by consumer_data.lock. - */ - struct lttng_ht *stream_ht; + + /* Channel hash table protected by consumer_data.lock. */ struct lttng_ht *channel_ht; /* * Flag specifying if the local array of FDs needs update in the @@ -267,6 +274,10 @@ struct lttng_consumer_global_data { struct lttng_ht *relayd_ht; }; +/* Defined in consumer.c and coupled with explanations */ +extern struct lttng_ht *metadata_ht; +extern struct lttng_ht *data_ht; + /* * Init consumer data structures. */ @@ -316,11 +327,6 @@ extern void lttng_consumer_sync_trace_file( */ extern int lttng_consumer_poll_socket(struct pollfd *kconsumer_sockpoll); -extern int consumer_update_poll_array( - struct lttng_consumer_local_data *ctx, struct pollfd **pollfd, - struct lttng_consumer_stream **local_consumer_streams, - struct lttng_ht *metadata_ht); - extern struct lttng_consumer_stream *consumer_allocate_stream( int channel_key, int stream_key, int shm_fd, int wait_fd, @@ -331,27 +337,28 @@ extern struct lttng_consumer_stream *consumer_allocate_stream( uid_t uid, gid_t gid, int net_index, - int metadata_flag); -extern int consumer_add_stream(struct lttng_consumer_stream *stream); -extern void consumer_del_stream(struct lttng_consumer_stream *stream); -extern void consumer_change_stream_state(int stream_key, - enum lttng_consumer_stream_state state); + int metadata_flag, + int *alloc_ret); +extern void consumer_del_stream(struct lttng_consumer_stream *stream, + struct lttng_ht *ht); +extern void consumer_del_metadata_stream(struct lttng_consumer_stream *stream, + struct lttng_ht *ht); extern void consumer_del_channel(struct lttng_consumer_channel *channel); extern struct lttng_consumer_channel *consumer_allocate_channel( int channel_key, int shm_fd, int wait_fd, uint64_t mmap_len, - uint64_t max_sb_size); + uint64_t max_sb_size, + unsigned int nb_init_streams); int consumer_add_channel(struct lttng_consumer_channel *channel); /* lttng-relayd consumer command */ -int consumer_add_relayd(struct consumer_relayd_sock_pair *relayd); struct consumer_relayd_sock_pair *consumer_allocate_relayd_sock_pair( int net_seq_idx); struct consumer_relayd_sock_pair *consumer_find_relayd(int key); int consumer_handle_stream_before_relayd(struct lttng_consumer_stream *stream, size_t data_size); -void consumer_destroy_relayd(struct consumer_relayd_sock_pair *relayd); +void consumer_steal_stream_key(int key, struct lttng_ht *ht); extern struct lttng_consumer_local_data *lttng_consumer_create( enum lttng_consumer_type type, @@ -363,18 +370,21 @@ extern struct lttng_consumer_local_data *lttng_consumer_create( extern void lttng_consumer_destroy(struct lttng_consumer_local_data *ctx); extern ssize_t lttng_consumer_on_read_subbuffer_mmap( struct lttng_consumer_local_data *ctx, - struct lttng_consumer_stream *stream, unsigned long len); + struct lttng_consumer_stream *stream, unsigned long len, + unsigned long padding); extern ssize_t lttng_consumer_on_read_subbuffer_splice( struct lttng_consumer_local_data *ctx, - struct lttng_consumer_stream *stream, unsigned long len); + struct lttng_consumer_stream *stream, unsigned long len, + unsigned long padding); extern int lttng_consumer_take_snapshot(struct lttng_consumer_local_data *ctx, struct lttng_consumer_stream *stream); extern int lttng_consumer_get_produced_snapshot( struct lttng_consumer_local_data *ctx, struct lttng_consumer_stream *stream, unsigned long *pos); -extern void *lttng_consumer_thread_poll_fds(void *data); -extern void *lttng_consumer_thread_receive_fds(void *data); +extern void *consumer_thread_metadata_poll(void *data); +extern void *consumer_thread_data_poll(void *data); +extern void *consumer_thread_sessiond_poll(void *data); extern int lttng_consumer_recv_cmd(struct lttng_consumer_local_data *ctx, int sock, struct pollfd *consumer_sockpoll); @@ -384,5 +394,7 @@ int lttng_consumer_on_recv_stream(struct lttng_consumer_stream *stream); int consumer_add_relayd_socket(int net_seq_idx, int sock_type, struct lttng_consumer_local_data *ctx, int sock, struct pollfd *consumer_sockpoll, struct lttcomm_sock *relayd_sock); +void consumer_flag_relayd_for_destroy( + struct consumer_relayd_sock_pair *relayd); #endif /* LIB_CONSUMER_H */