X-Git-Url: https://git.lttng.org/?a=blobdiff_plain;f=src%2Fcommon%2Fconsumer%2Fconsumer.h;h=737f2a42c449099035bbf8c671090c4fb1234310;hb=947bd0978b3d17b34eb1158cc6439eb1d4e5b6c3;hp=4770671c003a35f40d67ce4d85f199587323e93f;hpb=f5ba75b4f0c0b44092c76bc931b25b24a2e62718;p=lttng-tools.git diff --git a/src/common/consumer/consumer.h b/src/common/consumer/consumer.h index 4770671c0..737f2a42c 100644 --- a/src/common/consumer/consumer.h +++ b/src/common/consumer/consumer.h @@ -27,6 +27,7 @@ #include #include #include +#include struct lttng_consumer_local_data; @@ -64,6 +65,7 @@ enum lttng_consumer_command { LTTNG_CONSUMER_CLOSE_TRACE_CHUNK, LTTNG_CONSUMER_TRACE_CHUNK_EXISTS, LTTNG_CONSUMER_CLEAR_CHANNEL, + LTTNG_CONSUMER_OPEN_CHANNEL_PACKETS, }; enum lttng_consumer_type { @@ -88,7 +90,13 @@ enum consumer_channel_type { CONSUMER_CHANNEL_TYPE_DATA = 1, }; -extern struct lttng_consumer_global_data consumer_data; +enum sync_metadata_status { + SYNC_METADATA_STATUS_NEW_DATA, + SYNC_METADATA_STATUS_NO_DATA, + SYNC_METADATA_STATUS_ERROR, +}; + +extern struct lttng_consumer_global_data the_consumer_data; struct stream_list { struct cds_list_head head; @@ -152,7 +160,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 @@ -291,6 +299,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). @@ -313,8 +327,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 @@ -382,6 +396,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. @@ -532,7 +553,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 lttng_ust_ctl_consumer_stream *ustream; struct cds_list_head send_node; /* On-disk circular buffer */ uint64_t tracefile_size_current; @@ -590,6 +611,9 @@ struct lttng_consumer_stream { */ uint64_t rotate_position; + /* Whether or not a packet was opened during the current trace chunk. */ + bool opened_packet_in_current_trace_chunk; + /* * Read-only copies of channel values. We cannot safely access the * channel from a stream, so we need to have a local copy of these @@ -626,10 +650,11 @@ struct lttng_consumer_stream { reset_metadata_cb reset_metadata; consume_subbuffer_cb consume_subbuffer; put_next_subbuffer_cb put_next_subbuffer; - post_consume_cb post_consume; + struct lttng_dynamic_array post_consume_cbs; 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; }; @@ -781,30 +806,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 @@ -813,17 +838,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; }; /* @@ -838,8 +863,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 @@ -1039,5 +1068,9 @@ enum lttcomm_return_code lttng_consumer_init_command( struct lttng_consumer_local_data *ctx, const lttng_uuid sessiond_uuid); 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 */