X-Git-Url: https://git.lttng.org/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fbin%2Flttng-sessiond%2Fconsumer.h;h=3616d467cd4e76e65246ce5c2e07cbf80e75434e;hp=1337f32ac0e09e29d39cd5612cfc26e0d0d668c1;hb=5d2e1e66a968d9e555f9b8b00d0589ebfaf3de32;hpb=2f77fc4b3720dc8f75847130498c2d4aad7c03ec diff --git a/src/bin/lttng-sessiond/consumer.h b/src/bin/lttng-sessiond/consumer.h index 1337f32ac..3616d467c 100644 --- a/src/bin/lttng-sessiond/consumer.h +++ b/src/bin/lttng-sessiond/consumer.h @@ -18,8 +18,6 @@ #ifndef _CONSUMER_H #define _CONSUMER_H -#include - #include #include #include @@ -47,6 +45,10 @@ struct consumer_socket { */ 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; }; @@ -54,7 +56,24 @@ 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; @@ -67,9 +86,6 @@ 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; }; @@ -95,9 +111,6 @@ struct consumer_net { /* Data path for network streaming. */ struct lttng_uri data; - - /* Flag if network sockets were sent to the consumer. */ - unsigned int relayd_socks_sent; }; /* @@ -113,7 +126,7 @@ struct consumer_output { * 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; + uint64_t net_seq_index; /* * Subdirectory path name used for both local and network consumer. @@ -147,39 +160,62 @@ 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_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); + 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, - int channel_key, - int stream_key, - uint32_t state, - enum lttng_event_output output, - uint64_t mmap_len, + 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, - int net_index, - unsigned int metadata_flag, + uint64_t relayd_id, const char *name, - const char *pathname); -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); + 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 */