X-Git-Url: https://git.liburcu.org/?a=blobdiff_plain;f=src%2Fcommon%2Fconsumer%2Fconsumer.h;h=6455692968d8f7a49f04c9a37fbeeecb7f5ecbf2;hb=a73d94960e33d71f20cb99ea648cd24d064aabee;hp=aa8a401a0416efe089057f2e07b6187c38bf78d5;hpb=6f9449c22eef59294cf1e1dc3610a5cbf14baec0;p=lttng-tools.git diff --git a/src/common/consumer/consumer.h b/src/common/consumer/consumer.h index aa8a401a0..645569296 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 @@ -27,6 +28,8 @@ #include #include #include +#include +#include struct lttng_consumer_local_data; @@ -64,6 +67,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 +92,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 +162,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 @@ -176,6 +186,11 @@ struct lttng_consumer_channel { /* Metadata cache is metadata channel */ struct consumer_metadata_cache *metadata_cache; + /* + * Wait queue awaiting updates to metadata stream's flushed position. + */ + struct lttng_wait_queue metadata_pushed_wait_queue; + /* For UST metadata periodical flush */ int switch_timer_enabled; timer_t switch_timer; @@ -265,6 +280,14 @@ struct stream_subbuffer { unsigned long subbuf_size; unsigned long padded_subbuf_size; uint64_t version; + /* + * Left unset when unsupported. + * + * Indicates that this is the last sub-buffer of + * a series of sub-buffer that makes-up a coherent + * (parseable) unit of metadata. + */ + LTTNG_OPTIONAL(bool) coherent; } metadata; struct { unsigned long subbuf_size; @@ -283,6 +306,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). @@ -305,8 +334,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 @@ -374,6 +403,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. @@ -393,6 +429,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; /* @@ -524,8 +565,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; @@ -582,6 +622,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 @@ -618,11 +661,13 @@ 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; }; /* @@ -873,7 +918,8 @@ void lttng_consumer_set_command_sock_path( * Returns the return code of sendmsg : the number of bytes transmitted or -1 * on error. */ -int lttng_consumer_send_error(struct lttng_consumer_local_data *ctx, int cmd); +int lttng_consumer_send_error(struct lttng_consumer_local_data *ctx, + enum lttcomm_return_code error_code); /* * Called from signal handler to ensure a clean exit. @@ -954,7 +1000,6 @@ struct lttng_consumer_local_data *lttng_consumer_create( int (*update_stream)(uint64_t sessiond_key, uint32_t state)); void lttng_consumer_destroy(struct lttng_consumer_local_data *ctx); ssize_t lttng_consumer_on_read_subbuffer_mmap( - struct lttng_consumer_local_data *ctx, struct lttng_consumer_stream *stream, const struct lttng_buffer_view *buffer, unsigned long padding); @@ -981,10 +1026,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); @@ -1031,5 +1082,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 */