X-Git-Url: http://git.efficios.com/?a=blobdiff_plain;f=src%2Fcommon%2Fconsumer%2Fconsumer.h;h=ab59e6e757ffba375e7ebb71cd0ab7b1ef91d9b4;hb=6d40f8fa6e8256c448ab3b3d279069a8749c716d;hp=3af7db9136f3b011eef851397a2747ac86878ce2;hpb=fb83fe64f250bec7416f18891a8264450c61ead3;p=deliverable%2Flttng-tools.git diff --git a/src/common/consumer/consumer.h b/src/common/consumer/consumer.h index 3af7db913..ab59e6e75 100644 --- a/src/common/consumer/consumer.h +++ b/src/common/consumer/consumer.h @@ -60,6 +60,7 @@ enum lttng_consumer_command { LTTNG_CONSUMER_STREAMS_SENT, LTTNG_CONSUMER_DISCARDED_EVENTS, LTTNG_CONSUMER_LOST_PACKETS, + LTTNG_CONSUMER_CLEAR_QUIESCENT_CHANNEL, }; /* State of each fd in consumer */ @@ -217,6 +218,8 @@ 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; }; /* @@ -249,6 +252,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. @@ -280,8 +302,11 @@ struct lttng_consumer_stream { /* UID/GID of the user owning the session to which stream belongs */ uid_t uid; gid_t gid; - /* Network sequence number. Indicating on which relayd socket it goes. */ - uint64_t net_seq_idx; + /* + * Relayd id, indicating on which relayd socket it goes. Set to -1ULL if + * not the stream is not associated to a relay daemon. + */ + uint64_t relayd_id; /* * Indicate if this stream was successfully sent to a relayd. This is set * after the refcount of the relayd is incremented and is checked when the @@ -292,6 +317,11 @@ struct lttng_consumer_stream { /* Identify if the stream is the metadata */ unsigned int metadata_flag; + /* + * Last known metadata version, reset the metadata file in case + * of change. + */ + uint64_t metadata_version; /* Used when the stream is set for network streaming */ uint64_t relayd_stream_id; /* @@ -312,9 +342,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; @@ -365,9 +395,9 @@ 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. + * 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. @@ -382,6 +412,11 @@ struct lttng_consumer_stream { /* Indicate if the stream still has some data to be read. */ unsigned int has_data:1; + /* + * Inform the consumer or relay to reset the metadata + * file before writing in it (regeneration). + */ + unsigned int reset_metadata_flag:1; }; /* @@ -389,9 +424,9 @@ struct lttng_consumer_stream { */ struct consumer_relayd_sock_pair { /* Network sequence number. */ - uint64_t net_seq_idx; + uint64_t id; /* Number of stream associated with this relayd */ - unsigned int refcount; + int refcount; /* * This flag indicates whether or not we should destroy this object. The @@ -425,6 +460,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; }; /* @@ -561,6 +597,12 @@ 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 volatile int consumer_quit; + /* * Init consumer data structures. */ @@ -596,12 +638,6 @@ 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 @@ -646,7 +682,7 @@ void consumer_del_channel(struct lttng_consumer_channel *channel); /* lttng-relayd consumer command */ struct consumer_relayd_sock_pair *consumer_find_relayd(uint64_t key); int consumer_send_relayd_stream(struct lttng_consumer_stream *stream, char *path); -int consumer_send_relayd_streams_sent(uint64_t net_seq_idx); +int consumer_send_relayd_streams_sent(uint64_t relayd_id); void close_relayd_stream(struct lttng_consumer_stream *stream); struct lttng_consumer_channel *consumer_find_channel(uint64_t key); int consumer_handle_stream_before_relayd(struct lttng_consumer_stream *stream, @@ -686,7 +722,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 relayd_id, 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); @@ -707,5 +743,6 @@ void consumer_del_stream_for_data(struct lttng_consumer_stream *stream); int 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); +void lttng_consumer_cleanup_relayd(struct consumer_relayd_sock_pair *relayd); #endif /* LIB_CONSUMER_H */