X-Git-Url: https://git.liburcu.org/?a=blobdiff_plain;f=src%2Fcommon%2Fconsumer%2Fconsumer.h;h=1c6eeb39a83aa4f4e2c7db1107cfdc0955d86823;hb=4878de5c7deb512bbdac4fdfc498907efa06fb7c;hp=1b0ee000b5bdd944ac4aba946a83d256327a5195;hpb=4bd69c5f1161cd065f487da0f4c1aa03a73c47e4;p=lttng-tools.git diff --git a/src/common/consumer/consumer.h b/src/common/consumer/consumer.h index 1b0ee000b..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 @@ -29,10 +30,6 @@ #include #include -#ifdef __cplusplus -extern "C" { -#endif - struct lttng_consumer_local_data; /* Commands for consumer */ @@ -400,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. @@ -419,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; /* @@ -551,7 +560,6 @@ struct lttng_consumer_stream { char name[LTTNG_SYMBOL_NAME_LEN]; /* Internal state of libustctl. */ struct lttng_ust_ctl_consumer_stream *ustream; - struct cds_list_head send_node; /* On-disk circular buffer */ uint64_t tracefile_size_current; uint64_t tracefile_count_current; @@ -651,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; }; @@ -802,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 @@ -834,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; }; /* @@ -1014,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); @@ -1036,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, @@ -1069,8 +1082,4 @@ enum lttcomm_return_code lttng_consumer_open_channel_packets( int consumer_metadata_wakeup_pipe(const struct lttng_consumer_channel *channel); void lttng_consumer_sigbus_handle(void *addr); -#ifdef __cplusplus -} -#endif - #endif /* LIB_CONSUMER_H */