X-Git-Url: http://git.efficios.com/?p=lttng-tools.git;a=blobdiff_plain;f=src%2Fcommon%2Fconsumer%2Fconsumer.h;h=568ba8c6f7fab7cc80cc754c1b671c6f842b5716;hp=59764e105c36ef086c77dec6508312e7d8b54d58;hb=dc722af3d0305807fca29ad00d694cbd5967497a;hpb=93ec662e687dc15a3601704a1e0c96c51ad228c9 diff --git a/src/common/consumer/consumer.h b/src/common/consumer/consumer.h index 59764e105..568ba8c6f 100644 --- a/src/common/consumer/consumer.h +++ b/src/common/consumer/consumer.h @@ -2,6 +2,7 @@ * Copyright (C) 2011 - Julien Desfossez * Mathieu Desnoyers * 2012 - David Goulet + * 2018 - Jérémie Galarneau * * This program is free software; you can redistribute it and/or modify * it under the terms of the GNU General Public License, version 2 only, @@ -60,6 +61,14 @@ enum lttng_consumer_command { LTTNG_CONSUMER_STREAMS_SENT, LTTNG_CONSUMER_DISCARDED_EVENTS, LTTNG_CONSUMER_LOST_PACKETS, + LTTNG_CONSUMER_CLEAR_QUIESCENT_CHANNEL, + LTTNG_CONSUMER_SET_CHANNEL_MONITOR_PIPE, + LTTNG_CONSUMER_ROTATE_CHANNEL, + LTTNG_CONSUMER_ROTATE_RENAME, + LTTNG_CONSUMER_CHECK_ROTATION_PENDING_LOCAL, + LTTNG_CONSUMER_CHECK_ROTATION_PENDING_RELAY, + LTTNG_CONSUMER_MKDIR, + LTTNG_CONSUMER_CLEAR_CHANNEL, }; /* State of each fd in consumer */ @@ -159,6 +168,7 @@ struct lttng_consumer_channel { /* Metadata cache is metadata channel */ struct consumer_metadata_cache *metadata_cache; + /* For UST metadata periodical flush */ int switch_timer_enabled; timer_t switch_timer; @@ -169,6 +179,10 @@ struct lttng_consumer_channel { timer_t live_timer; int live_timer_error; + /* For channel monitoring timer. */ + int monitor_timer_enabled; + timer_t monitor_timer; + /* On-disk circular buffer */ uint64_t tracefile_size; uint64_t tracefile_count; @@ -217,6 +231,15 @@ struct lttng_consumer_channel { uint64_t discarded_events; /* Total number of missed packets due to overwriting (overwrite). */ uint64_t lost_packets; + + bool streams_sent_to_relayd; + + /* + * The chunk id where we currently write the data. This value is sent + * to the relay when we add a stream and when a stream rotates. This + * allows to keep track of where each stream on the relay is writing. + */ + uint64_t current_chunk_id; }; /* @@ -249,6 +272,25 @@ struct lttng_consumer_stream { int data_read; int hangup_flush_done; + /* + * Whether the stream is in a "complete" state (e.g. it does not have a + * partially written sub-buffer. + * + * Initialized to "false" on stream creation (first packet is empty). + * + * The various transitions of the quiescent state are: + * - On "start" tracing: set to false, since the stream is not + * "complete". + * - On "stop" tracing: if !quiescent -> flush FINAL (update + * timestamp_end), and set to true; the stream has entered a + * complete/quiescent state. + * - On "destroy" or stream/application hang-up: if !quiescent -> + * flush FINAL, and set to true. + * + * NOTE: Update and read are protected by the stream lock. + */ + bool quiescent; + /* * metadata_timer_lock protects flags waiting_on_metadata and * missed_metadata_flush. @@ -317,9 +359,9 @@ struct lttng_consumer_stream { * Lock to use the stream FDs since they are used between threads. * * This is nested INSIDE the consumer_data lock. - * This is nested INSIDE the metadata cache lock. * This is nested INSIDE the channel lock. * This is nested INSIDE the channel timer lock. + * This is nested OUTSIDE the metadata cache lock. * This is nested OUTSIDE consumer_relayd_sock_pair lock. */ pthread_mutex_t lock; @@ -370,9 +412,15 @@ struct lttng_consumer_stream { /* Copy of the sequence number of the last packet extracted. */ uint64_t last_sequence_number; /* - * FD of the index file for this stream. + * A stream is created with a trace_archive_id matching the session's + * current trace archive id at the time of the creation of the stream. + * It is incremented when the rotate_position is reached. + */ + uint64_t trace_archive_id; + /* + * Index file object of the index file for this stream. */ - int index_fd; + struct lttng_index_file *index_file; /* * Local pipe to extract data when using splice. @@ -385,6 +433,31 @@ struct lttng_consumer_stream { pthread_cond_t metadata_rdv; pthread_mutex_t metadata_rdv_lock; + /* + * rotate_position represents the position in the ring-buffer that has to + * be flushed to disk to complete the ongoing rotation. When that position + * is reached, this tracefile can be closed and a new one is created in + * channel_read_only_attributes.path. + */ + unsigned long rotate_position; + + /* + * 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 + * fields in the stream object. These fields should be removed from + * the stream objects when we introduce refcounting. + */ + struct { + char path[LTTNG_PATH_MAX]; + uint64_t tracefile_size; + } channel_read_only_attributes; + + /* + * Flag to inform the data or metadata thread that a stream is + * ready to be rotated. + */ + bool rotate_ready; + /* Indicate if the stream still has some data to be read. */ unsigned int has_data:1; /* @@ -401,7 +474,7 @@ struct consumer_relayd_sock_pair { /* Network sequence number. */ uint64_t net_seq_idx; /* Number of stream associated with this relayd */ - unsigned int refcount; + int refcount; /* * This flag indicates whether or not we should destroy this object. The @@ -435,6 +508,7 @@ struct consumer_relayd_sock_pair { /* Session id on both sides for the sockets. */ uint64_t relayd_session_id; uint64_t sessiond_session_id; + struct lttng_consumer_local_data *ctx; }; /* @@ -519,6 +593,11 @@ struct lttng_consumer_local_data { int consumer_should_quit[2]; /* Metadata poll thread pipe. Transfer metadata stream to it */ struct lttng_pipe *consumer_metadata_pipe; + /* + * Pipe used by the channel monitoring timers to provide state samples + * to the session daemon (write-only). + */ + int channel_monitor_pipe; }; /* @@ -571,6 +650,39 @@ struct lttng_consumer_global_data { struct lttng_ht *stream_per_chan_id_ht; }; +/* + * Set to nonzero when the consumer is exiting. Updated by signal + * handler and thread exit, read by threads. + */ +extern int consumer_quit; + +/* + * Set to nonzero when the consumer is exiting. Updated by signal + * handler and thread exit, read by threads. + */ +extern int consumer_quit; + +/* Flag used to temporarily pause data consumption from testpoints. */ +extern int data_consumption_paused; + +/* Return a human-readable consumer type string that is suitable for logging. */ +static inline +const char *lttng_consumer_type_str(enum lttng_consumer_type type) +{ + switch (type) { + case LTTNG_CONSUMER_UNKNOWN: + return "unknown"; + case LTTNG_CONSUMER_KERNEL: + return "kernel"; + case LTTNG_CONSUMER32_UST: + return "32-bit user space"; + case LTTNG_CONSUMER64_UST: + return "64-bit user space"; + default: + abort(); + } +} + /* * Init consumer data structures. */ @@ -606,18 +718,20 @@ void lttng_consumer_should_exit(struct lttng_consumer_local_data *ctx); */ void lttng_consumer_cleanup(void); -/* - * Flush pending writes to trace output disk file. - */ -void lttng_consumer_sync_trace_file(struct lttng_consumer_stream *stream, - off_t orig_offset); - /* * Poll on the should_quit pipe and the command socket return -1 on error and * should exit, 0 if data is available on the command socket */ int lttng_consumer_poll_socket(struct pollfd *kconsumer_sockpoll); +/* + * Copy the fields from the channel that need to be accessed (read-only) + * directly from the stream. + */ +void consumer_stream_update_channel_attributes( + struct lttng_consumer_stream *stream, + struct lttng_consumer_channel *channel); + struct lttng_consumer_stream *consumer_allocate_stream(uint64_t channel_key, uint64_t stream_key, enum lttng_consumer_stream_state state, @@ -629,7 +743,8 @@ struct lttng_consumer_stream *consumer_allocate_stream(uint64_t channel_key, int cpu, int *alloc_ret, enum consumer_channel_type type, - unsigned int monitor); + unsigned int monitor, + uint64_t trace_archive_id); struct lttng_consumer_channel *consumer_allocate_channel(uint64_t key, uint64_t session_id, const char *pathname, @@ -681,9 +796,12 @@ ssize_t lttng_consumer_on_read_subbuffer_splice( struct lttng_consumer_stream *stream, unsigned long len, unsigned long padding, struct ctf_packet_index *index); +int lttng_consumer_sample_snapshot_positions(struct lttng_consumer_stream *stream); int lttng_consumer_take_snapshot(struct lttng_consumer_stream *stream); int lttng_consumer_get_produced_snapshot(struct lttng_consumer_stream *stream, unsigned long *pos); +int lttng_consumer_get_consumed_snapshot(struct lttng_consumer_stream *stream, + unsigned long *pos); int lttng_ustconsumer_get_wakeup_fd(struct lttng_consumer_stream *stream); int lttng_ustconsumer_close_wakeup_fd(struct lttng_consumer_stream *stream); void *consumer_thread_metadata_poll(void *data); @@ -696,7 +814,7 @@ int lttng_consumer_recv_cmd(struct lttng_consumer_local_data *ctx, ssize_t lttng_consumer_read_subbuffer(struct lttng_consumer_stream *stream, struct lttng_consumer_local_data *ctx); int lttng_consumer_on_recv_stream(struct lttng_consumer_stream *stream); -int consumer_add_relayd_socket(uint64_t net_seq_idx, int sock_type, +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); @@ -712,10 +830,30 @@ void consumer_destroy_relayd(struct consumer_relayd_sock_pair *relayd); unsigned long consumer_get_consume_start_pos(unsigned long consumed_pos, unsigned long produced_pos, uint64_t nb_packets_per_stream, uint64_t max_sb_size); -int consumer_add_data_stream(struct lttng_consumer_stream *stream); +void consumer_add_data_stream(struct lttng_consumer_stream *stream); void consumer_del_stream_for_data(struct lttng_consumer_stream *stream); -int consumer_add_metadata_stream(struct lttng_consumer_stream *stream); +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_sample_channel(struct lttng_consumer_channel *channel, + uint64_t key, const char *path, uint64_t relayd_id, + uint32_t metadata, uint64_t new_chunk_id, + struct lttng_consumer_local_data *ctx); +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, bool *rotated); +int lttng_consumer_rotate_ready_streams(struct lttng_consumer_channel *channel, + uint64_t key, struct lttng_consumer_local_data *ctx); +int lttng_consumer_rotate_rename(const char *current_path, const char *new_path, + uid_t uid, gid_t gid, uint64_t relayd_id); +int lttng_consumer_check_rotation_pending_local(uint64_t session_id, + uint64_t chunk_id); +int lttng_consumer_check_rotation_pending_relay(uint64_t session_id, + uint64_t relayd_id, uint64_t chunk_id); +void lttng_consumer_reset_stream_rotate_state(struct lttng_consumer_stream *stream); +int lttng_consumer_mkdir(const char *path, uid_t uid, gid_t gid, + uint64_t relayd_id); +void lttng_consumer_cleanup_relayd(struct consumer_relayd_sock_pair *relayd); +int lttng_consumer_clear_channel(struct lttng_consumer_channel *channel); #endif /* LIB_CONSUMER_H */