X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fconsumer.h;h=5cbb47dcfa5945534e48551f2fa9393486ddce2f;hp=6ac781605c7284ecf7319e2b10fa0c3920b441b3;hb=f02e1e8a5820da2eda835add020f92ca8d32c973;hpb=d14d33bf091e72b23b1f90ea18a0a01bed098b76 diff --git a/src/common/consumer.h b/src/common/consumer.h index 6ac781605..5cbb47dcf 100644 --- a/src/common/consumer.h +++ b/src/common/consumer.h @@ -1,6 +1,7 @@ /* * 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, version 2 only, @@ -16,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 @@ -27,6 +28,7 @@ #include #include +#include /* * When the receiving thread dies, we need to have a way to make the polling @@ -51,6 +53,9 @@ 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, }; /* State of each fd in consumer */ @@ -92,6 +97,7 @@ struct lttng_ust_lib_ring_buffer; */ struct lttng_consumer_stream { 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 @@ -118,6 +124,50 @@ 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; +}; + +/* + * 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; }; /* @@ -184,7 +234,6 @@ struct lttng_consumer_local_data { * 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 @@ -210,6 +259,12 @@ 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; }; /* @@ -263,7 +318,8 @@ 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_consumer_stream **local_consumer_streams, + struct lttng_ht *metadata_ht); extern struct lttng_consumer_stream *consumer_allocate_stream( int channel_key, int stream_key, @@ -273,7 +329,9 @@ extern struct lttng_consumer_stream *consumer_allocate_stream( enum lttng_event_output output, const char *path_name, uid_t uid, - gid_t gid); + 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, @@ -286,6 +344,15 @@ extern struct lttng_consumer_channel *consumer_allocate_channel( uint64_t max_sb_size); 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); + extern struct lttng_consumer_local_data *lttng_consumer_create( enum lttng_consumer_type type, ssize_t (*buffer_ready)(struct lttng_consumer_stream *stream, @@ -315,4 +382,4 @@ 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); -#endif /* _LTTNG_CONSUMER_H */ +#endif /* LIB_CONSUMER_H */