X-Git-Url: https://git.lttng.org/?a=blobdiff_plain;f=src%2Fcommon%2Fconsumer%2Fconsumer.h;h=1c6eeb39a83aa4f4e2c7db1107cfdc0955d86823;hb=4878de5c7deb512bbdac4fdfc498907efa06fb7c;hp=7b381b2c75db6c7300cc6581206e0e12b1f5c0fa;hpb=b1316da1ffbd276fc8271e7a9438e683ad352781;p=lttng-tools.git diff --git a/src/common/consumer/consumer.h b/src/common/consumer/consumer.h index 7b381b2c7..1c6eeb39a 100644 --- a/src/common/consumer/consumer.h +++ b/src/common/consumer/consumer.h @@ -1,5 +1,5 @@ /* - * Copyright (C) 2011 Julien Desfossez + * Copyright (C) 2011 EfficiOS Inc. * Copyright (C) 2011 Mathieu Desnoyers * Copyright (C) 2012 David Goulet * Copyright (C) 2018 Jérémie Galarneau @@ -13,6 +13,7 @@ #include #include +#include #include #include @@ -96,7 +97,7 @@ enum sync_metadata_status { SYNC_METADATA_STATUS_ERROR, }; -extern struct lttng_consumer_global_data consumer_data; +extern struct lttng_consumer_global_data the_consumer_data; struct stream_list { struct cds_list_head head; @@ -160,7 +161,7 @@ struct lttng_consumer_channel { /* For UST */ uid_t ust_app_uid; /* Application UID. */ - struct ustctl_consumer_channel *uchan; + struct lttng_ust_ctl_consumer_channel *uchan; unsigned char uuid[LTTNG_UUID_STR_LEN]; /* * Temporary stream list used to store the streams once created and waiting @@ -299,6 +300,12 @@ struct stream_subbuffer { } info; }; +enum get_next_subbuffer_status { + GET_NEXT_SUBBUFFER_STATUS_OK, + GET_NEXT_SUBBUFFER_STATUS_NO_DATA, + GET_NEXT_SUBBUFFER_STATUS_ERROR, +}; + /* * Perform any operation required to acknowledge * the wake-up of a consumer stream (e.g. consume a byte on a wake-up pipe). @@ -321,8 +328,8 @@ typedef int (*on_sleep_cb)(struct lttng_consumer_stream *, * * Stream and channel locks are acquired during this call. */ -typedef int (*get_next_subbuffer_cb)(struct lttng_consumer_stream *, - struct stream_subbuffer *); +typedef enum get_next_subbuffer_status (*get_next_subbuffer_cb)( + struct lttng_consumer_stream *, struct stream_subbuffer *); /* * Populate the stream_subbuffer's info member. The info to populate @@ -390,6 +397,13 @@ typedef void (*lock_cb)(struct lttng_consumer_stream *); */ typedef void (*unlock_cb)(struct lttng_consumer_stream *); +/* + * Assert that the stream and channel lock and any other stream type specific + * lock that need to be acquired during the processing of a read_subbuffer + * operation is acquired. + */ +typedef void (*assert_locked_cb)(struct lttng_consumer_stream *); + /* * Invoked when a subbuffer's metadata version does not match the last * known metadata version. @@ -409,6 +423,11 @@ struct lttng_consumer_stream { struct lttng_ht_node_u64 node_channel_id; /* HT node used in consumer_data.stream_list_ht */ struct lttng_ht_node_u64 node_session_id; + /* + * List used by channels to reference streams that are not yet globally + * visible. + */ + struct cds_list_head send_node; /* Pointer to associated channel. */ struct lttng_consumer_channel *chan; /* @@ -540,8 +559,7 @@ struct lttng_consumer_stream { /* Stream name. Format is: _ */ char name[LTTNG_SYMBOL_NAME_LEN]; /* Internal state of libustctl. */ - struct ustctl_consumer_stream *ustream; - struct cds_list_head send_node; + struct lttng_ust_ctl_consumer_stream *ustream; /* On-disk circular buffer */ uint64_t tracefile_size_current; uint64_t tracefile_count_current; @@ -641,6 +659,7 @@ struct lttng_consumer_stream { send_live_beacon_cb send_live_beacon; on_sleep_cb on_sleep; unlock_cb unlock; + assert_locked_cb assert_locked; } read_subbuffer_ops; struct metadata_bucket *metadata_bucket; }; @@ -792,30 +811,30 @@ struct lttng_consumer_global_data { * This is nested OUTSIDE the stream lock. * This is nested OUTSIDE the consumer_relayd_sock_pair lock. */ - pthread_mutex_t lock; + pthread_mutex_t lock {}; /* * Number of streams in the data stream hash table declared outside. * Protected by consumer_data.lock. */ - int stream_count; + int stream_count = 0; /* Channel hash table protected by consumer_data.lock. */ - struct lttng_ht *channel_ht; + struct lttng_ht *channel_ht = nullptr; /* Channel hash table indexed by session id. */ - struct lttng_ht *channels_by_session_id_ht; + struct lttng_ht *channels_by_session_id_ht = nullptr; /* * Flag specifying if the local array of FDs needs update in the * poll function. Protected by consumer_data.lock. */ - unsigned int need_update; - enum lttng_consumer_type type; + unsigned int need_update = 1; + enum lttng_consumer_type type = LTTNG_CONSUMER_UNKNOWN; /* * 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; + struct lttng_ht *relayd_ht = nullptr; /* * This hash table contains all streams (metadata and data) indexed by @@ -824,17 +843,17 @@ struct lttng_consumer_global_data { * * This HT uses the "node_session_id" of the consumer stream. */ - struct lttng_ht *stream_list_ht; + struct lttng_ht *stream_list_ht = nullptr; /* * This HT uses the "node_channel_id" of the consumer stream. */ - struct lttng_ht *stream_per_chan_id_ht; + struct lttng_ht *stream_per_chan_id_ht = nullptr; /* * Trace chunk registry indexed by (session_id, chunk_id). */ - struct lttng_trace_chunk_registry *chunk_registry; + struct lttng_trace_chunk_registry *chunk_registry = nullptr; }; /* @@ -849,8 +868,12 @@ extern int consumer_quit; */ extern int consumer_quit; -/* Flag used to temporarily pause data consumption from testpoints. */ -extern int data_consumption_paused; +/* + * Flag used to temporarily pause data consumption from testpoints. + * + * This variable is dlsym-ed from a test, so needs to be exported. + */ +LTTNG_EXPORT extern int data_consumption_paused; /* Return a human-readable consumer type string that is suitable for logging. */ static inline @@ -1000,10 +1023,16 @@ ssize_t lttng_consumer_read_subbuffer(struct lttng_consumer_stream *stream, struct lttng_consumer_local_data *ctx, bool locked_by_caller); int lttng_consumer_on_recv_stream(struct lttng_consumer_stream *stream); -void consumer_add_relayd_socket(uint64_t net_seq_idx, int sock_type, - struct lttng_consumer_local_data *ctx, int sock, - struct pollfd *consumer_sockpoll, struct lttcomm_relayd_sock *relayd_sock, - uint64_t sessiond_id, uint64_t relayd_session_id); +void consumer_add_relayd_socket(uint64_t net_seq_idx, + int sock_type, + struct lttng_consumer_local_data *ctx, + int sock, + struct pollfd *consumer_sockpoll, + uint64_t sessiond_id, + uint64_t relayd_session_id, + uint32_t relayd_version_major, + uint32_t relayd_version_minor, + enum lttcomm_sock_proto relayd_socket_protocol); void consumer_flag_relayd_for_destroy( struct consumer_relayd_sock_pair *relayd); int consumer_data_pending(uint64_t id); @@ -1022,13 +1051,11 @@ void consumer_add_metadata_stream(struct lttng_consumer_stream *stream); void consumer_del_stream_for_metadata(struct lttng_consumer_stream *stream); int consumer_create_index_file(struct lttng_consumer_stream *stream); int lttng_consumer_rotate_channel(struct lttng_consumer_channel *channel, - uint64_t key, uint64_t relayd_id, uint32_t metadata, - struct lttng_consumer_local_data *ctx); + uint64_t key, uint64_t relayd_id); int lttng_consumer_stream_is_rotate_ready(struct lttng_consumer_stream *stream); -int lttng_consumer_rotate_stream(struct lttng_consumer_local_data *ctx, - struct lttng_consumer_stream *stream); +int lttng_consumer_rotate_stream(struct lttng_consumer_stream *stream); int lttng_consumer_rotate_ready_streams(struct lttng_consumer_channel *channel, - uint64_t key, struct lttng_consumer_local_data *ctx); + uint64_t key); void lttng_consumer_reset_stream_rotate_state(struct lttng_consumer_stream *stream); enum lttcomm_return_code lttng_consumer_create_trace_chunk( const uint64_t *relayd_id, uint64_t session_id, @@ -1053,5 +1080,6 @@ int lttng_consumer_clear_channel(struct lttng_consumer_channel *channel); enum lttcomm_return_code lttng_consumer_open_channel_packets( struct lttng_consumer_channel *channel); int consumer_metadata_wakeup_pipe(const struct lttng_consumer_channel *channel); +void lttng_consumer_sigbus_handle(void *addr); #endif /* LIB_CONSUMER_H */