stream: add more timing info

Keep track of queued data in the stream
Pass delay field around to make it possible to know about the raw
software read/write pointer and the hardware one.
Start stream position at 0

spa/include/spa/node/io.h

@@ -74,9 +74,12 @@ struct spa_io_control_range {
 
 /** A time source */
 struct spa_io_clock {
-	uint64_t nsec;                  /**< time in nanoseconds */
-        struct spa_fraction rate;       /**< rate */
-        uint32_t position;              /**< current position expressed in rate */
+	uint64_t nsec;			/**< time in nanoseconds */
+	struct spa_fraction rate;	/**< rate for position/delay */
+	uint64_t position;		/**< current position */
+	uint64_t delay;			/**< delay between position and hardware,
+					     add to position for capture,
+					     subtract for playback */
 };
 
 struct spa_type_io {

spa/plugins/alsa/alsa-utils.c

@@ -36,6 +36,7 @@ static int spa_alsa_open(struct state *state)
 
 	state->timerfd = timerfd_create(CLOCK_MONOTONIC, TFD_CLOEXEC | TFD_NONBLOCK);
 	state->opened = true;
+	state->sample_count = 0;
 
 	return 0;
 }
@@ -565,7 +566,8 @@ static void alsa_on_playback_timeout_event(struct spa_source *source)
 	if (state->clock) {
 		state->clock->nsec = SPA_TIMESPEC_TO_TIME(&state->now);
 		state->clock->rate = SPA_FRACTION(state->rate, 1);
-		state->clock->position = state->sample_count - state->filled;
+		state->clock->position = state->sample_count;
+		state->clock->delay = state->filled;
 	}
 
 	spa_log_trace(state->log, "timeout %ld %d %ld %ld %ld", state->filled, state->threshold,
@@ -636,7 +638,8 @@ static void alsa_on_capture_timeout_event(struct spa_source *source)
 	if (state->clock) {
 		state->clock->nsec = SPA_TIMESPEC_TO_TIME(&state->now);
 		state->clock->rate = SPA_FRACTION(state->rate, 1);
-		state->clock->position = state->sample_count + avail;
+		state->clock->position = state->sample_count;
+		state->clock->delay = avail;
 	}
 
 	spa_log_trace(state->log, "timeout %ld %d %ld %ld %ld %ld %ld", avail, state->threshold,

src/modules/module-client-node/client-node.c

@@ -163,6 +163,7 @@ struct impl {
 	int other_fds[2];
 
 	struct pw_client_node_position *position;
+	uint64_t start;
 };
 
 /** \endcond */
@@ -818,6 +819,7 @@ impl_node_port_use_buffers(struct spa_node *node,
 			}
 		}
 	}
+	impl->start = -1;
 
 	pw_client_node_resource_port_use_buffers(this->resource,
 						 this->seq,
@@ -912,7 +914,10 @@ static int impl_node_process(struct spa_node *node)
 	rq = SPA_MEMBER(impl->position, sizeof(struct pw_client_node_position),
 			struct pw_driver_quantum);
 
+	if (impl->start == -1)
+		impl->start = q->position;
 	*rq = *q;
+	rq->position -= impl->start;
 
 	if (write(this->writefd, &cmd, 8) != 8)
 		spa_log_warn(this->log, "node %p: error %s", this, strerror(errno));

src/modules/module-client-node/client-stream.c

@@ -771,10 +771,10 @@ static int impl_node_process(struct spa_node *node)
 {
 	struct node *this = SPA_CONTAINER_OF(node, struct node, node);
 	struct impl *impl = this->impl;
+	struct pw_driver_quantum *q = impl->this.node->driver_node->rt.quantum;
 	int status, trigger;
 
-	impl->ctrl.min_size = impl->ctrl.max_size =
-		impl->this.node->driver_node->rt.quantum->size * sizeof(float);
+	impl->ctrl.min_size = impl->ctrl.max_size = q->size * sizeof(float);
 
 	spa_log_trace(this->log, "%p: process %d", this, impl->ctrl.max_size);
 

src/pipewire/node.c

@@ -626,23 +626,24 @@ static void node_process(void *data, int status)
 	if (node->driver) {
 		if (node->rt.driver->state->pending == 0 || !node->remote) {
 			struct timespec ts;
-			struct pw_driver_quantum *q;
-			q = node->rt.quantum;
+			struct pw_driver_quantum *q = node->rt.quantum;
 
 			if (node->rt.clock) {
 				q->nsec = node->rt.clock->nsec;
 				q->rate = node->rt.clock->rate;
 				q->position = node->rt.clock->position;
+				q->delay = node->rt.clock->delay;
 			}
 			else {
 				clock_gettime(CLOCK_MONOTONIC, &ts);
 				q->nsec = SPA_TIMESPEC_TO_TIME(&ts);
 				q->position = impl->next_position;
+				q->delay = 0;
 			}
 			impl->next_position += q->size;
 
-			pw_log_trace("node %p: run %"PRIu64" %d %d", node,
-					q->nsec, q->position, q->size);
+			pw_log_trace("node %p: run %"PRIu64" %"PRIu64" %"PRIi64" %d", node,
+					q->nsec, q->position, q->delay, q->size);
 
 			spa_graph_run(node->rt.driver);
 		}

src/pipewire/private.h

@@ -227,7 +227,8 @@ struct pw_module {
 struct pw_driver_quantum {
 	uint64_t nsec;			/**< time in nanoseconds */
 	struct spa_fraction rate;	/**< rate */
-	uint32_t position;		/**< current position expressed in rate */
+	uint64_t position;		/**< current position expressed in rate */
+	uint64_t delay;			/**< delay to hardware */
 	uint32_t size;			/**< size of one period expressed in rate */
 };
 

src/pipewire/stream.c

@@ -71,6 +71,8 @@ struct buffer {
 struct queue {
 	uint32_t ids[MAX_BUFFERS];
 	struct spa_ringbuffer ring;
+	uint64_t incount;
+	uint64_t outcount;
 };
 
 struct data {
@@ -143,6 +145,9 @@ struct stream {
 
 	bool free_data;
 	struct data data;
+
+	uint32_t seq1, seq2;
+	struct pw_time time;
 };
 
 static struct param *add_param(struct pw_stream *stream,
@@ -184,6 +189,7 @@ static inline int push_queue(struct stream *stream, struct queue *queue, struct
 		return -EINVAL;
 
 	SPA_FLAG_SET(buffer->flags, BUFFER_FLAG_QUEUED);
+	queue->incount += buffer->this.size;
 
 	spa_ringbuffer_get_write_index(&queue->ring, &index);
 	queue->ids[index & MASK_BUFFERS] = buffer->id;
@@ -205,6 +211,7 @@ static inline struct buffer *pop_queue(struct stream *stream, struct queue *queu
 	spa_ringbuffer_read_update(&queue->ring, index + 1);
 
 	buffer = &stream->buffers[id];
+	queue->outcount += buffer->this.size;
 	SPA_FLAG_UNSET(buffer->flags, BUFFER_FLAG_QUEUED);
 
 	return buffer;
@@ -632,6 +639,18 @@ static int impl_port_reuse_buffer(struct spa_node *node, uint32_t port_id, uint3
 	return 0;
 }
 
+static inline void copy_quantum(struct stream *impl, int64_t queued)
+{
+	struct pw_driver_quantum *q = impl->node->rt.quantum;
+	impl->seq1++;
+	impl->time.now = q->nsec;
+	impl->time.rate = q->rate;
+	impl->time.ticks = q->position;
+	impl->time.delay = q->delay;
+	impl->time.queued = queued;
+	impl->seq2 = impl->seq1;
+}
+
 static int impl_node_process_input(struct spa_node *node)
 {
 	struct stream *impl = SPA_CONTAINER_OF(node, struct stream, impl_node);
@@ -639,7 +658,8 @@ static int impl_node_process_input(struct spa_node *node)
 	struct spa_io_buffers *io = impl->io;
 	struct buffer *b;
 
-	pw_log_trace("stream %p: process input %d %d", stream, io->status, io->buffer_id);
+	pw_log_trace("stream %p: process in %d %d %"PRIu64" %"PRIi64, stream,
+			io->status, io->buffer_id, impl->time.ticks, impl->time.delay);
 
 	if (io->status != SPA_STATUS_HAVE_BUFFER)
 		goto done;
@@ -652,6 +672,8 @@ static int impl_node_process_input(struct spa_node *node)
 		call_process(impl);
 
       done:
+	copy_quantum(impl, impl->dequeued.incount);
+
 	/* pop buffer to recycle */
 	if ((b = pop_queue(impl, &impl->queued))) {
 		pw_log_trace("stream %p: recycle buffer %d", stream, b->id);
@@ -673,7 +695,8 @@ static int impl_node_process_output(struct spa_node *node)
 	uint32_t index;
 
      again:
-	pw_log_trace("stream %p: process out %d %d", stream, io->status, io->buffer_id);
+	pw_log_trace("stream %p: process out %d %d %"PRIu64" %"PRIi64, stream,
+			io->status, io->buffer_id, impl->time.ticks, impl->time.delay);
 
 	res = 0;
 	if (io->status != SPA_STATUS_HAVE_BUFFER) {
@@ -694,12 +717,15 @@ static int impl_node_process_output(struct spa_node *node)
 			pw_log_trace("stream %p: no more buffers %p", stream, io);
 		}
 	}
+
 	if (!SPA_FLAG_CHECK(impl->flags, PW_STREAM_FLAG_DRIVER)) {
 		call_process(impl);
 		if (spa_ringbuffer_get_read_index(&impl->queued.ring, &index) >= MIN_QUEUED &&
 		    io->status == SPA_STATUS_NEED_BUFFER)
 			goto again;
 	}
+	copy_quantum(impl, impl->queued.outcount);
+
 	res = io->status;
 	pw_log_trace("stream %p: res %d", stream, res);
 
@@ -1104,13 +1130,17 @@ int pw_stream_set_active(struct pw_stream *stream, bool active)
 int pw_stream_get_time(struct pw_stream *stream, struct pw_time *time)
 {
 	struct stream *impl = SPA_CONTAINER_OF(stream, struct stream, this);
-	struct pw_driver_quantum *q;
+	uint32_t seq;
 
-	q = impl->node->rt.quantum;
+	do {
+		seq = impl->seq2;
+		*time = impl->time;
+	} while (impl->seq1 != seq);
 
-	time->now = q->nsec;
-	time->ticks = q->position;
-	time->rate = q->rate;
+	if (impl->direction == SPA_DIRECTION_INPUT)
+		time->queued = time->queued - impl->dequeued.outcount;
+	else
+		time->queued = impl->queued.incount - time->queued;
 
 	return 0;
 }

src/pipewire/stream.h

@@ -166,6 +166,7 @@ enum pw_stream_state {
 struct pw_buffer {
 	struct spa_buffer *buffer;	/* the spa buffer */
 	void *user_data;		/* user data attached to the buffer */
+	uint64_t size;			/* size to keep track of queued size */
 };
 
 /** Events for a stream */
@@ -308,8 +309,15 @@ int pw_stream_set_active(struct pw_stream *stream, bool active);
 /** A time structure \memberof pw_stream */
 struct pw_time {
 	int64_t now;			/**< the monotonic time */
-	int64_t ticks;			/**< the ticks at \a now */
-	struct spa_fraction rate;	/**< the rate of \a ticks */
+	struct spa_fraction rate;	/**< the rate of \a ticks and delay */
+	uint64_t ticks;			/**< the ticks at \a now. This is the current time that
+					     the remote end is reading/writing. */
+	uint64_t delay;			/**< delay to device, add to ticks for INPUT streams and
+					     subtract from ticks for OUTPUT streams to get the
+					     time of the device. */
+	uint64_t queued;		/**< data queued in the stream, this is the sum
+					     of the size fields in the pw_buffer that are
+					     currently queued */
 };
 /** Query the time on the stream \memberof pw_stream */
 int pw_stream_get_time(struct pw_stream *stream, struct pw_time *time);