X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fconsumer.h;h=29836e5b09cf6e25220e20baac2d3fd48b5cfa08;hp=53b6151822ba36102b6b6df5a53878ac35b714a7;hb=d8ef542d25837bdfb960e5df2a91c5d18f5ef401;hpb=63281d5d013d47087217291ba7724d8302fd63e4 diff --git a/src/common/consumer.h b/src/common/consumer.h index 53b615182..29836e5b0 100644 --- a/src/common/consumer.h +++ b/src/common/consumer.h @@ -23,28 +23,15 @@ #include #include #include +#include #include #include #include +#include #include -/* - * When the receiving thread dies, we need to have a way to make the polling - * thread exit eventually. If all FDs hang up (normal case when the - * lttng-sessiond stops), we can exit cleanly, but if there is a problem and - * for whatever reason some FDs remain open, the consumer should still exit - * eventually. - * - * If the timeout is reached, it means that during this period no events - * occurred on the FDs so we need to force an exit. This case should not happen - * but it is a safety to ensure we won't block the consumer indefinitely. - * - * The value of 2 seconds is an arbitrary choice. - */ -#define LTTNG_CONSUMER_POLL_TIMEOUT 2000 - /* Commands for consumer */ enum lttng_consumer_command { LTTNG_CONSUMER_ADD_CHANNEL, @@ -57,7 +44,14 @@ enum lttng_consumer_command { /* Inform the consumer to kill a specific relayd connection */ LTTNG_CONSUMER_DESTROY_RELAYD, /* Return to the sessiond if there is data pending for a session */ - LTTNG_CONSUMER_DATA_AVAILABLE, + LTTNG_CONSUMER_DATA_PENDING, + /* Consumer creates a channel and returns it to sessiond. */ + LTTNG_CONSUMER_ASK_CHANNEL_CREATION, + LTTNG_CONSUMER_GET_CHANNEL, + LTTNG_CONSUMER_DESTROY_CHANNEL, + LTTNG_CONSUMER_PUSH_METADATA, + LTTNG_CONSUMER_CLOSE_METADATA, + LTTNG_CONSUMER_SETUP_METADATA, }; /* State of each fd in consumer */ @@ -74,82 +68,166 @@ enum lttng_consumer_type { LTTNG_CONSUMER32_UST, }; +enum consumer_endpoint_status { + CONSUMER_ENDPOINT_ACTIVE, + CONSUMER_ENDPOINT_INACTIVE, +}; + +enum consumer_channel_output { + CONSUMER_CHANNEL_MMAP = 0, + CONSUMER_CHANNEL_SPLICE = 1, +}; + +enum consumer_channel_type { + CONSUMER_CHANNEL_TYPE_METADATA = 0, + CONSUMER_CHANNEL_TYPE_DATA = 1, +}; + +struct stream_list { + struct cds_list_head head; + unsigned int count; +}; + struct lttng_consumer_channel { - struct lttng_ht_node_ulong node; - int key; - uint64_t max_sb_size; /* the subbuffer size for this channel */ - int refcount; /* Number of streams referencing this channel */ + /* HT node used for consumer_data.channel_ht */ + struct lttng_ht_node_u64 node; + /* Indexed key. Incremented value in the consumer. */ + uint64_t key; + /* Number of streams referencing this channel */ + int refcount; + /* Tracing session id on the session daemon side. */ + uint64_t session_id; + /* Channel trace file path name. */ + char pathname[PATH_MAX]; + /* Channel name. */ + char name[LTTNG_SYMBOL_NAME_LEN]; + /* UID and GID of the channel. */ + uid_t uid; + gid_t gid; + /* Relayd id of the channel. -1 if it does not apply. */ + int64_t relayd_id; /* - * The number of streams to receive initially. Used to guarantee that we do - * not destroy a channel before receiving all its associated streams. + * Number of streams NOT initialized yet. This is used in order to not + * delete this channel if streams are getting initialized. */ - unsigned int nb_init_streams; + unsigned int nb_init_stream_left; + /* Output type (mmap or splice). */ + enum consumer_channel_output output; + /* Channel type for stream */ + enum consumer_channel_type type; /* For UST */ - int shm_fd; + struct ustctl_consumer_channel *uchan; + unsigned char uuid[UUID_STR_LEN]; + /* + * Temporary stream list used to store the streams once created and waiting + * to be sent to the session daemon by receiving the + * LTTNG_CONSUMER_GET_CHANNEL. + */ + struct stream_list streams; + /* + * 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 + * regular channel, this is always set to NULL. + */ + struct lttng_consumer_stream *metadata_stream; + /* + * Metadata written so far. Helps keeping track of + * contiguousness and order. + */ + uint64_t contig_metadata_written; + + /* for UST */ int wait_fd; - void *mmap_base; - size_t mmap_len; - struct lttng_ust_shm_handle *handle; - int wait_fd_is_copy; - int cpucount; + /* Node within channel thread ht */ + struct lttng_ht_node_u64 wait_fd_node; }; -/* Forward declaration for UST. */ -struct lttng_ust_lib_ring_buffer; - /* * Internal representation of the streams, sessiond_key is used to identify * uniquely a stream. */ struct lttng_consumer_stream { /* HT node used by the data_ht and metadata_ht */ - struct lttng_ht_node_ulong node; + struct lttng_ht_node_u64 node; + /* stream indexed per channel key node */ + struct lttng_ht_node_u64 node_channel_id; /* HT node used in consumer_data.stream_list_ht */ - struct lttng_ht_node_ulong node_session_id; - struct lttng_consumer_channel *chan; /* associated channel */ + struct lttng_ht_node_u64 node_session_id; + /* Pointer to associated channel. */ + struct lttng_consumer_channel *chan; + + /* Key by which the stream is indexed for 'node'. */ + uint64_t key; /* - * key is the key used by the session daemon to refer to the - * object in the consumer daemon. + * File descriptor of the data output file. This can be either a file or a + * socket fd for relayd streaming. */ - int key; - int shm_fd; - int wait_fd; int out_fd; /* output file to write the data */ - off_t out_fd_offset; /* write position in the output file descriptor */ - char path_name[PATH_MAX]; /* tracefile name */ + /* Write position in the output file descriptor */ + off_t out_fd_offset; enum lttng_consumer_stream_state state; - size_t shm_len; - void *mmap_base; - size_t mmap_len; - enum lttng_event_output output; /* splice or mmap */ int shm_fd_is_copy; - int wait_fd_is_copy; - /* For UST */ - struct lttng_ust_lib_ring_buffer *buf; - int cpu; int data_read; int hangup_flush_done; + enum lttng_event_output output; + /* Maximum subbuffer size. */ + unsigned long max_sb_size; + + /* + * Still used by the kernel for MMAP output. For UST, the ustctl getter is + * used for the mmap base and offset. + */ + void *mmap_base; + unsigned long mmap_len; + + /* For UST */ + + int wait_fd; /* 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. */ - int net_seq_idx; + uint64_t net_seq_idx; /* Identify if the stream is the metadata */ 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 */ + /* + * When sending a stream packet to a relayd, this number is used to track + * the packet sent by the consumer and seen by the relayd. When sending the + * data header to the relayd, this number is sent and if the transmission + * was successful, it is incremented. + * + * Even if the full data is not fully transmitted it won't matter since + * only two possible error can happen after that where either the relayd + * died or a read error is detected on the stream making this value useless + * after that. + * + * This value SHOULD be read/updated atomically or with the lock acquired. + */ uint64_t next_net_seq_num; /* - * Lock to use the stream FDs since they are used between threads. Using - * this lock with network streaming, when using the control mutex of a - * consumer_relayd_sock_pair, make sure to acquire this lock BEFORE locking - * it and releasing it AFTER the control mutex unlock. + * Lock to use the stream FDs since they are used between threads. + * + * This is nested INSIDE the consumer_data lock. + * This is nested OUTSIDE consumer_relayd_sock_pair lock. */ pthread_mutex_t lock; /* Tracing session id */ uint64_t session_id; + /* + * Indicates if the stream end point is still active or not (network + * streaming or local file system). The thread "owning" the stream is + * handling this status and can be notified of a state change through the + * consumer data appropriate pipe. + */ + enum consumer_endpoint_status endpoint_status; + /* Stream name. Format is: _ */ + char name[LTTNG_SYMBOL_NAME_LEN]; + /* Internal state of libustctl. */ + struct ustctl_consumer_stream *ustream; + struct cds_list_head send_node; }; /* @@ -157,7 +235,7 @@ struct lttng_consumer_stream { */ struct consumer_relayd_sock_pair { /* Network sequence number. */ - int net_seq_idx; + int64_t net_seq_idx; /* Number of stream associated with this relayd */ unsigned int refcount; @@ -173,6 +251,9 @@ struct consumer_relayd_sock_pair { * 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. + * + * This is nested INSIDE the consumer_data lock. + * This is nested INSIDE the stream lock. */ pthread_mutex_t ctrl_sock_mutex; @@ -185,7 +266,11 @@ struct consumer_relayd_sock_pair { * this socket is for now only used in a single thread. */ struct lttcomm_sock data_sock; - struct lttng_ht_node_ulong node; + struct lttng_ht_node_u64 node; + + /* Session id on both sides for the sockets. */ + uint64_t relayd_session_id; + uint64_t sessiond_session_id; }; /* @@ -242,6 +327,7 @@ struct lttng_consumer_local_data { char *consumer_command_sock_path; /* communication with splice */ int consumer_thread_pipe[2]; + int consumer_channel_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]; @@ -260,8 +346,8 @@ struct lttng_consumer_global_data { * and number of element in the hash table. It's also a protection for * concurrent read/write between threads. * - * XXX: We need to see if this lock is still needed with the lockless RCU - * hash tables. + * This is nested OUTSIDE the stream lock. + * This is nested OUTSIDE the consumer_relayd_sock_pair lock. */ pthread_mutex_t lock; @@ -294,27 +380,28 @@ struct lttng_consumer_global_data { * This HT uses the "node_session_id" of the consumer stream. */ struct lttng_ht *stream_list_ht; -}; -/* Defined in consumer.c and coupled with explanations */ -extern struct lttng_ht *metadata_ht; -extern struct lttng_ht *data_ht; + /* + * This HT uses the "node_channel_id" of the consumer stream. + */ + struct lttng_ht *stream_per_chan_id_ht; +}; /* * Init consumer data structures. */ -extern void lttng_consumer_init(void); +void lttng_consumer_init(void); /* * Set the error socket for communication with a session daemon. */ -extern void lttng_consumer_set_error_sock( - struct lttng_consumer_local_data *ctx, int sock); +void lttng_consumer_set_error_sock(struct lttng_consumer_local_data *ctx, + int sock); /* * Set the command socket path for communication with a session daemon. */ -extern void lttng_consumer_set_command_sock_path( +void lttng_consumer_set_command_sock_path( struct lttng_consumer_local_data *ctx, char *sock); /* @@ -323,92 +410,90 @@ extern void lttng_consumer_set_command_sock_path( * Returns the return code of sendmsg : the number of bytes transmitted or -1 * on error. */ -extern int lttng_consumer_send_error( - struct lttng_consumer_local_data *ctx, int cmd); +int lttng_consumer_send_error(struct lttng_consumer_local_data *ctx, int cmd); /* * Called from signal handler to ensure a clean exit. */ -extern void lttng_consumer_should_exit( - struct lttng_consumer_local_data *ctx); +void lttng_consumer_should_exit(struct lttng_consumer_local_data *ctx); /* * Cleanup the daemon's socket on exit. */ -extern void lttng_consumer_cleanup(void); +void lttng_consumer_cleanup(void); /* * Flush pending writes to trace output disk file. */ -extern void lttng_consumer_sync_trace_file( - struct lttng_consumer_stream *stream, off_t orig_offset); +void lttng_consumer_sync_trace_file(struct lttng_consumer_stream *stream, + off_t orig_offset); /* * Poll on the should_quit pipe and the command socket return -1 on error and * should exit, 0 if data is available on the command socket */ -extern int lttng_consumer_poll_socket(struct pollfd *kconsumer_sockpoll); +int lttng_consumer_poll_socket(struct pollfd *kconsumer_sockpoll); -extern struct lttng_consumer_stream *consumer_allocate_stream( - int channel_key, int stream_key, - int shm_fd, int wait_fd, +struct lttng_consumer_stream *consumer_allocate_stream(uint64_t channel_key, + uint64_t stream_key, enum lttng_consumer_stream_state state, - uint64_t mmap_len, - enum lttng_event_output output, - const char *path_name, + const char *channel_name, uid_t uid, gid_t gid, - int net_index, - int metadata_flag, + int relayd_id, uint64_t session_id, - int *alloc_ret); -extern void consumer_del_stream(struct lttng_consumer_stream *stream, + int cpu, + int *alloc_ret, + enum consumer_channel_type type); +struct lttng_consumer_channel *consumer_allocate_channel(uint64_t key, + uint64_t session_id, + const char *pathname, + const char *name, + uid_t uid, + gid_t gid, + int relayd_id, + enum lttng_event_output output); +void consumer_del_stream(struct lttng_consumer_stream *stream, struct lttng_ht *ht); -extern void consumer_del_metadata_stream(struct lttng_consumer_stream *stream, +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, - unsigned int nb_init_streams); -int consumer_add_channel(struct lttng_consumer_channel *channel); +int consumer_add_channel(struct lttng_consumer_channel *channel, + struct lttng_consumer_local_data *ctx); +void consumer_del_channel(struct lttng_consumer_channel *channel); /* lttng-relayd consumer command */ struct consumer_relayd_sock_pair *consumer_allocate_relayd_sock_pair( int net_seq_idx); -struct consumer_relayd_sock_pair *consumer_find_relayd(int key); +struct consumer_relayd_sock_pair *consumer_find_relayd(uint64_t key); +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); void consumer_steal_stream_key(int key, struct lttng_ht *ht); -extern struct lttng_consumer_local_data *lttng_consumer_create( +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), 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)); -extern void lttng_consumer_destroy(struct lttng_consumer_local_data *ctx); -extern ssize_t lttng_consumer_on_read_subbuffer_mmap( +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); -extern ssize_t lttng_consumer_on_read_subbuffer_splice( +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); -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, +int lttng_consumer_take_snapshot(struct lttng_consumer_stream *stream); +int lttng_consumer_get_produced_snapshot(struct lttng_consumer_stream *stream, unsigned long *pos); -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, +void *consumer_thread_metadata_poll(void *data); +void *consumer_thread_data_poll(void *data); +void *consumer_thread_sessiond_poll(void *data); +void *consumer_thread_channel_poll(void *data); +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, @@ -416,9 +501,13 @@ ssize_t lttng_consumer_read_subbuffer(struct lttng_consumer_stream *stream, 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); + struct pollfd *consumer_sockpoll, struct lttcomm_sock *relayd_sock, + unsigned int sessiond_id); void consumer_flag_relayd_for_destroy( struct consumer_relayd_sock_pair *relayd); -int consumer_data_available(uint64_t id); +int consumer_data_pending(uint64_t id); +int consumer_send_status_msg(int sock, int ret_code); +int consumer_send_status_channel(int sock, + struct lttng_consumer_channel *channel); #endif /* LIB_CONSUMER_H */