X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fbin%2Flttng-sessiond%2Fconsumer.h;h=3616d467cd4e76e65246ce5c2e07cbf80e75434e;hp=21d762a1582028a8ac206104943f8e57a16d9cfa;hb=5d2e1e66a968d9e555f9b8b00d0589ebfaf3de32;hpb=f1e1679499bf3e76127c8812c72a6e9ba69e02e5 diff --git a/src/bin/lttng-sessiond/consumer.h b/src/bin/lttng-sessiond/consumer.h index 21d762a15..3616d467c 100644 --- a/src/bin/lttng-sessiond/consumer.h +++ b/src/bin/lttng-sessiond/consumer.h @@ -18,15 +18,62 @@ #ifndef _CONSUMER_H #define _CONSUMER_H -#include - #include +#include +#include + +#include "health.h" + +enum consumer_dst_type { + CONSUMER_DST_LOCAL, + CONSUMER_DST_NET, +}; + +struct consumer_socket { + /* File descriptor */ + int fd; + /* + * To use this socket (send/recv), this lock MUST be acquired. + */ + pthread_mutex_t *lock; + + /* + * Indicates if the socket was registered by a third part + * (REGISTER_CONSUMER) or is the spawn consumer of the session daemon. + * During the destroy phase of a consumer output, we close the socket if + * this flag is set to 1 since we don't need the fd anymore. + */ + unsigned int registered; + + /* Flag if network sockets were sent to the consumer. */ + unsigned int control_sock_sent; + unsigned int data_sock_sent; + + struct lttng_ht_node_ulong node; +}; struct consumer_data { enum lttng_consumer_type type; pthread_t thread; /* Worker thread interacting with the consumer */ - sem_t sem; + + /* Conditions used by the consumer thread to indicate readiness. */ + pthread_cond_t cond; + pthread_condattr_t condattr; + pthread_mutex_t cond_mutex; + + /* + * This is a flag condition indicating that the consumer thread is ready + * and connected to the lttng-consumerd daemon. This flag MUST only be + * updated by locking the condition mutex above or before spawning a + * consumer thread. + * + * A value of 0 means that the thread is NOT ready. A value of 1 means that + * the thread consumer did connect successfully to the lttng-consumerd + * daemon. A negative value indicates that there is been an error and the + * thread has likely quit. + */ + int consumer_thread_is_ready; /* Mutex to control consumerd pid assignation */ pthread_mutex_t pid_mutex; @@ -38,6 +85,137 @@ struct consumer_data { /* consumer error and command Unix socket path */ char err_unix_sock_path[PATH_MAX]; char cmd_unix_sock_path[PATH_MAX]; + + /* communication lock */ + pthread_mutex_t lock; +}; + +/* + * Network URIs + */ +struct consumer_net { + /* + * Indicate if URI type is set. Those flags should only be set when the + * created URI is done AND valid. + */ + int control_isset; + int data_isset; + + /* + * The following two URIs MUST have the same destination address for + * network streaming to work. Network hop are not yet supported. + */ + + /* Control path for network streaming. */ + struct lttng_uri control; + + /* Data path for network streaming. */ + struct lttng_uri data; }; +/* + * Consumer output object describing where and how to send data. + */ +struct consumer_output { + /* If the consumer is enabled meaning that should be used */ + unsigned int enabled; + enum consumer_dst_type type; + + /* + * The net_seq_index is the index of the network stream on the consumer + * side. It tells the consumer which streams goes to which relayd with this + * index. The relayd sockets are index with it on the consumer side. + */ + uint64_t net_seq_index; + + /* + * Subdirectory path name used for both local and network consumer. + */ + char subdir[PATH_MAX]; + + /* + * Hashtable of consumer_socket index by the file descriptor value. For + * multiarch consumer support, we can have more than one consumer (ex: 32 + * and 64 bit). + */ + struct lttng_ht *socks; + + union { + char trace_path[PATH_MAX]; + struct consumer_net net; + } dst; +}; + +struct consumer_socket *consumer_find_socket(int key, + struct consumer_output *consumer); +struct consumer_socket *consumer_allocate_socket(int fd); +void consumer_add_socket(struct consumer_socket *sock, + struct consumer_output *consumer); +void consumer_del_socket(struct consumer_socket *sock, + struct consumer_output *consumer); +void consumer_destroy_socket(struct consumer_socket *sock); + +struct consumer_output *consumer_create_output(enum consumer_dst_type type); +struct consumer_output *consumer_copy_output(struct consumer_output *obj); +void consumer_destroy_output(struct consumer_output *obj); +int consumer_set_network_uri(struct consumer_output *obj, + struct lttng_uri *uri); +int consumer_send_fds(struct consumer_socket *sock, int *fds, size_t nb_fd); +int consumer_send_msg(struct consumer_socket *sock, + struct lttcomm_consumer_msg *msg); +int consumer_send_stream(struct consumer_socket *sock, + struct consumer_output *dst, struct lttcomm_consumer_msg *msg, + int *fds, size_t nb_fd); +int consumer_send_channel(struct consumer_socket *sock, + struct lttcomm_consumer_msg *msg); +int consumer_send_relayd_socket(struct consumer_socket *consumer_sock, + struct lttcomm_sock *sock, struct consumer_output *consumer, + enum lttng_stream_type type, uint64_t session_id); +int consumer_send_destroy_relayd(struct consumer_socket *sock, + struct consumer_output *consumer); +int consumer_recv_status_reply(struct consumer_socket *sock); +int consumer_recv_status_channel(struct consumer_socket *sock, + uint64_t *key, unsigned int *stream_count); +void consumer_output_send_destroy_relayd(struct consumer_output *consumer); +int consumer_create_socket(struct consumer_data *data, + struct consumer_output *output); +int consumer_set_subdir(struct consumer_output *consumer, + const char *session_name); + +void consumer_init_ask_channel_comm_msg(struct lttcomm_consumer_msg *msg, + uint64_t subbuf_size, + uint64_t num_subbuf, + int overwrite, + unsigned int switch_timer_interval, + unsigned int read_timer_interval, + int output, + int type, + uint64_t session_id, + const char *pathname, + const char *name, + uid_t uid, + gid_t gid, + uint64_t relayd_id, + uint64_t key, + unsigned char *uuid); +void consumer_init_stream_comm_msg(struct lttcomm_consumer_msg *msg, + enum lttng_consumer_command cmd, + uint64_t channel_key, + uint64_t stream_key, + int cpu); +void consumer_init_channel_comm_msg(struct lttcomm_consumer_msg *msg, + enum lttng_consumer_command cmd, + uint64_t channel_key, + uint64_t session_id, + const char *pathname, + uid_t uid, + gid_t gid, + uint64_t relayd_id, + const char *name, + unsigned int nb_init_streams, + enum lttng_event_output output, + int type); +int consumer_is_data_pending(uint64_t session_id, + struct consumer_output *consumer); + #endif /* _CONSUMER_H */