X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fconsumer.h;h=df002f81a606c62f75db59bf23d752ec29743e62;hp=83b86a39e2cbe7fe8ff8037123aeb6c2ee9329ba;hb=ca22feea083301934d1c8511851c86fb008c0697;hpb=f5204c9440be52e1b17bbb2c82450277ec02a583 diff --git a/src/common/consumer.h b/src/common/consumer.h index 83b86a39e..df002f81a 100644 --- a/src/common/consumer.h +++ b/src/common/consumer.h @@ -1,24 +1,24 @@ /* * Copyright (C) 2011 - Julien Desfossez - * Copyright (C) 2011 - Mathieu Desnoyers + * Mathieu Desnoyers + * 2012 - David Goulet * - * This program is free software; you can redistribute it and/or - * modify it under the terms of the GNU General Public License - * as published by the Free Software Foundation; only version 2 - * of the License. + * This program is free software; you can redistribute it and/or modify + * it under the terms of the GNU General Public License, version 2 only, + * as published by the Free Software Foundation. * - * This program is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - * GNU General Public License for more details. + * This program is distributed in the hope that it will be useful, but WITHOUT + * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or + * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for + * more details. * - * You should have received a copy of the GNU General Public License - * along with this program; if not, write to the Free Software - * Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA. + * You should have received a copy of the GNU General Public License along + * with this program; if not, write to the Free Software Foundation, Inc., + * 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 @@ -26,7 +26,9 @@ #include -#include "src/common/hashtable/hashtable.h" +#include +#include +#include /* * When the receiving thread dies, we need to have a way to make the polling @@ -51,6 +53,11 @@ enum lttng_consumer_command { LTTNG_CONSUMER_UPDATE_STREAM, /* 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, + /* Return to the sessiond if there is data pending for a session */ + LTTNG_CONSUMER_DATA_AVAILABLE, }; /* State of each fd in consumer */ @@ -72,13 +79,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; }; @@ -91,7 +103,10 @@ struct lttng_ust_lib_ring_buffer; * uniquely a stream. */ struct lttng_consumer_stream { + /* HT node used by the data_ht and metadata_ht */ struct lttng_ht_node_ulong node; + /* HT node used in consumer_data.stream_list_ht */ + struct lttng_ht_node_ulong node_session_id; struct lttng_consumer_channel *chan; /* associated channel */ /* * key is the key used by the session daemon to refer to the @@ -118,6 +133,54 @@ 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. */ + int 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 */ + uint64_t next_net_seq_num; + /* Lock to use the stream FDs since they are used between threads. */ + pthread_mutex_t lock; + /* Tracing session id */ + uint64_t session_id; +}; + +/* + * Internal representation of a relayd socket pair. + */ +struct consumer_relayd_sock_pair { + /* Network sequence number. */ + 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; + + /* 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; }; /* @@ -174,17 +237,19 @@ 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]; }; /* * Library-level data. One instance per process. */ struct lttng_consumer_global_data { - /* * At this time, this lock is used to ensure coherence between the count * and number of element in the hash table. It's also a protection for @@ -196,13 +261,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 @@ -210,8 +274,27 @@ struct lttng_consumer_global_data { */ unsigned int need_update; enum lttng_consumer_type type; + + /* + * Relayd socket(s) hashtable indexed by network sequence number. Each + * stream has an index which associate the right relayd socket to use. + */ + struct lttng_ht *relayd_ht; + + /* + * This hash table contains all streams (metadata and data) indexed by + * session id. In other words, the ht is indexed by session id and each + * bucket contains the list of associated streams. + * + * 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; + /* * Init consumer data structures. */ @@ -261,10 +344,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); - extern struct lttng_consumer_stream *consumer_allocate_stream( int channel_key, int stream_key, int shm_fd, int wait_fd, @@ -273,19 +352,32 @@ extern struct lttng_consumer_stream *consumer_allocate_stream( enum lttng_event_output output, const char *path_name, uid_t uid, - gid_t gid); -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); + gid_t gid, + int net_index, + int metadata_flag, + uint64_t session_id, + 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 */ +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_steal_stream_key(int key, struct lttng_ht *ht); + extern struct lttng_consumer_local_data *lttng_consumer_create( enum lttng_consumer_type type, ssize_t (*buffer_ready)(struct lttng_consumer_stream *stream, @@ -296,23 +388,32 @@ 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); 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); +int consumer_data_available(uint64_t id); -#endif /* _LTTNG_CONSUMER_H */ +#endif /* LIB_CONSUMER_H */