X-Git-Url: https://git.lttng.org/?a=blobdiff_plain;f=src%2Fcommon%2Fconsumer.h;h=d0cd8fd869e2b647581e02e163f75725365124f8;hb=4dbc372b53ef1ac713497164e7a8b92100db7ae2;hp=8bcf27d2c99166393cfa035ed644b4cd267e60df;hpb=00e2e675d54dc726a7c8f8887c889cc8ef022003;p=lttng-tools.git diff --git a/src/common/consumer.h b/src/common/consumer.h index 8bcf27d2c..d0cd8fd86 100644 --- a/src/common/consumer.h +++ b/src/common/consumer.h @@ -17,8 +17,8 @@ * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301 USA. */ -#ifndef _LTTNG_CONSUMER_H -#define _LTTNG_CONSUMER_H +#ifndef LIB_CONSUMER_H +#define LIB_CONSUMER_H #include #include @@ -54,6 +54,8 @@ enum lttng_consumer_command { /* inform the consumer to quit when all fd has hang up */ LTTNG_CONSUMER_STOP, LTTNG_CONSUMER_ADD_RELAYD_SOCKET, + /* Inform the consumer to kill a specific relayd connection */ + LTTNG_CONSUMER_DESTROY_RELAYD, }; /* State of each fd in consumer */ @@ -75,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; }; @@ -128,6 +135,8 @@ struct lttng_consumer_stream { unsigned int metadata_flag; /* Used when the stream is set for network streaming */ uint64_t relayd_stream_id; + /* Next sequence number to use for trace packet */ + uint64_t next_net_seq_num; }; /* @@ -138,9 +147,30 @@ struct consumer_relayd_sock_pair { int net_seq_idx; /* Number of stream associated with this relayd */ unsigned int refcount; + + /* + * This flag indicates whether or not we should destroy this object. The + * destruction should ONLY occurs when this flag is set and the refcount is + * set to zero. + */ + unsigned int destroy_flag; + + /* + * Mutex protecting the control socket to avoid out of order packets + * between threads sending data to the relayd. Since metadata data is sent + * over that socket, at least two sendmsg() are needed (header + data) + * creating a race for packets to overlap between threads using it. + */ pthread_mutex_t ctrl_sock_mutex; - /* Sockets information */ + + /* Control socket. Command and metadata are passed over it */ struct lttcomm_sock control_sock; + + /* + * We don't need a mutex at this point since we only splice or write single + * large chunk of data with a header appended at the begining. Moreover, + * this socket is for now only used in a single thread. + */ struct lttcomm_sock data_sock; struct lttng_ht_node_ulong node; }; @@ -199,10 +229,13 @@ struct lttng_consumer_local_data { char *consumer_command_sock_path; /* communication with splice */ int consumer_thread_pipe[2]; + int consumer_splice_metadata_pipe[2]; /* pipe to wake the poll thread when necessary */ int consumer_poll_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]; }; /* @@ -293,8 +326,7 @@ 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); + struct lttng_consumer_stream **local_consumer_streams); extern struct lttng_consumer_stream *consumer_allocate_stream( int channel_key, int stream_key, @@ -306,9 +338,13 @@ extern struct lttng_consumer_stream *consumer_allocate_stream( uid_t uid, gid_t gid, int net_index, - int metadata_flag); + int metadata_flag, + int *alloc_ret); extern int consumer_add_stream(struct lttng_consumer_stream *stream); -extern void consumer_del_stream(struct lttng_consumer_stream *stream); +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_change_stream_state(int stream_key, enum lttng_consumer_stream_state state); extern void consumer_del_channel(struct lttng_consumer_channel *channel); @@ -316,11 +352,11 @@ 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); @@ -337,10 +373,12 @@ 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( @@ -355,5 +393,10 @@ extern 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(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 /* _LTTNG_CONSUMER_H */ +#endif /* LIB_CONSUMER_H */