#include <common/compat/uuid.h>
#include <common/sessiond-comm/sessiond-comm.h>
#include <common/pipe.h>
-#include <common/index/lttng-index.h>
+#include <common/index/ctf-index.h>
/* Commands for consumer */
enum lttng_consumer_command {
/* pause, delete, active depending on fd state */
LTTNG_CONSUMER_UPDATE_STREAM,
/* inform the consumer to quit when all fd has hang up */
- LTTNG_CONSUMER_STOP,
+ LTTNG_CONSUMER_STOP, /* deprecated */
LTTNG_CONSUMER_ADD_RELAYD_SOCKET,
/* Inform the consumer to kill a specific relayd connection */
LTTNG_CONSUMER_DESTROY_RELAYD,
LTTNG_CONSUMER_FLUSH_CHANNEL,
LTTNG_CONSUMER_SNAPSHOT_CHANNEL,
LTTNG_CONSUMER_SNAPSHOT_METADATA,
+ LTTNG_CONSUMER_STREAMS_SENT,
};
/* State of each fd in consumer */
* FD of the index file for this stream.
*/
int index_fd;
+
+ /*
+ * Rendez-vous point between data and metadata stream in live mode.
+ */
+ pthread_cond_t metadata_rdv;
+ pthread_mutex_t metadata_rdv_lock;
+
+ /* Indicate if the stream still has some data to be read. */
+ unsigned int has_data:1;
};
/*
int consumer_splice_metadata_pipe[2];
/* Data stream poll thread pipe. To transfer data stream to the thread */
struct lttng_pipe *consumer_data_pipe;
+
+ /*
+ * Data thread use that pipe to catch wakeup from read subbuffer that
+ * detects that there is still data to be read for the stream encountered.
+ * Before doing so, the stream is flagged to indicate that there is still
+ * data to be read.
+ *
+ * Both pipes (read/write) are owned and used inside the data thread.
+ */
+ struct lttng_pipe *consumer_wakeup_pipe;
+ /* Indicate if the wakeup thread has been notified. */
+ unsigned int has_wakeup:1;
+
/* to let the signal handler wake up the fd receiver thread */
int consumer_should_quit[2];
/* Metadata poll thread pipe. Transfer metadata stream to it */
/*
* Init consumer data structures.
*/
-void lttng_consumer_init(void);
+int lttng_consumer_init(void);
/*
* Set the error socket for communication with a session daemon.
uint64_t net_seq_idx);
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);
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,
struct lttng_consumer_local_data *ctx,
struct lttng_consumer_stream *stream, unsigned long len,
unsigned long padding,
- struct lttng_packet_index *index);
+ struct ctf_packet_index *index);
ssize_t lttng_consumer_on_read_subbuffer_splice(
struct lttng_consumer_local_data *ctx,
struct lttng_consumer_stream *stream, unsigned long len,
unsigned long padding,
- struct lttng_packet_index *index);
+ struct ctf_packet_index *index);
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_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);
void *consumer_thread_data_poll(void *data);
void *consumer_thread_sessiond_poll(void *data);