X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fbin%2Flttng-sessiond%2Fconsumer.h;h=11c98773ee363de996ff03d6d158695c4f6e67c7;hp=ab36a685c448de22e4645752d7e4beb69b567aba;hb=840cb59cfc335e2ca44841f6fd57972ac3a623be;hpb=37278a1e7efe00011260569fa90909601e8c5184 diff --git a/src/bin/lttng-sessiond/consumer.h b/src/bin/lttng-sessiond/consumer.h index ab36a685c..11c98773e 100644 --- a/src/bin/lttng-sessiond/consumer.h +++ b/src/bin/lttng-sessiond/consumer.h @@ -18,9 +18,8 @@ #ifndef _CONSUMER_H #define _CONSUMER_H -#include - #include +#include #include #include "health.h" @@ -30,11 +29,47 @@ enum consumer_dst_type { 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; + + 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; @@ -47,8 +82,8 @@ struct consumer_data { char err_unix_sock_path[PATH_MAX]; char cmd_unix_sock_path[PATH_MAX]; - /* Health check of the thread */ - struct health_state health; + /* communication lock */ + pthread_mutex_t lock; }; /* @@ -72,45 +107,76 @@ struct consumer_net { /* Data path for network streaming. */ struct lttng_uri data; + + /* Flag if network sockets were sent to the consumer. */ + unsigned int control_sock_sent; + unsigned int data_sock_sent; }; /* * Consumer output object describing where and how to send data. */ struct consumer_output { - /* Consumer socket file descriptor */ - int sock; /* 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's basically the relayd socket file descriptor value so the - * consumer can identify which streams goes with which socket. + * 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. */ int 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(int sock, int *fds, size_t nb_fd); -int consumer_send_stream(int sock, struct consumer_output *dst, - struct lttcomm_consumer_msg *msg, int *fds, size_t nb_fd); -int consumer_send_channel(int sock, struct lttcomm_consumer_msg *msg); -int consumer_send_relayd_socket(int consumer_sock, +int consumer_send_fds(struct consumer_socket *sock, int *fds, size_t nb_fd); +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); + enum lttng_stream_type type, unsigned int session_id); +int consumer_send_destroy_relayd(struct consumer_socket *sock, + struct consumer_output *consumer); +int consumer_recv_status_reply(struct consumer_socket *sock); +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_stream_comm_msg(struct lttcomm_consumer_msg *msg, enum lttng_consumer_command cmd, @@ -124,12 +190,16 @@ void consumer_init_stream_comm_msg(struct lttcomm_consumer_msg *msg, int net_index, unsigned int metadata_flag, const char *name, - const char *pathname); + const char *pathname, + unsigned int session_id); void consumer_init_channel_comm_msg(struct lttcomm_consumer_msg *msg, enum lttng_consumer_command cmd, int channel_key, uint64_t max_sb_size, uint64_t mmap_len, - const char *name); + const char *name, + unsigned int nb_init_streams); +int consumer_is_data_pending(unsigned int id, + struct consumer_output *consumer); #endif /* _CONSUMER_H */