stream: add resample size to pw_time

Now that the resampler input size is set in the io_rate field when we
start we can add it to the pw_time struct as well.

This way we can know the required resampler input without having
to dequeue a buffer. This can be handy when the stream is a driver
and needs to know how much data to accumulate before starting the graph.

See #3750

src/pipewire/stream.c

@@ -100,6 +100,7 @@ struct stream {
 	struct spa_io_buffers *io;
 	struct spa_io_rate_match *rate_match;
 	uint32_t rate_queued;
+	uint64_t rate_requested;
 	struct {
 		struct spa_io_position *position;
 	} rt;
@@ -420,7 +421,7 @@ static struct buffer *get_buffer(struct pw_stream *stream, uint32_t id)
 
 static inline uint32_t update_requested(struct stream *impl)
 {
-	uint32_t index, id, res = 0;
+	uint32_t index, id;
 	struct buffer *buffer;
 	struct spa_io_rate_match *r = impl->rate_match;
 
@@ -431,15 +432,11 @@ static inline uint32_t update_requested(struct stream *impl)
 
 	id = impl->dequeued.ids[index & MASK_BUFFERS];
 	buffer = &impl->buffers[id];
-	if (r) {
-		buffer->this.requested = r->size;
-		res = r->size > 0 ? 1 : 0;
-	} else {
-		buffer->this.requested = impl->quantum;
-		res = 1;
-	}
+	buffer->this.requested = r ? r->size : impl->quantum;
+
 	pw_log_trace_fp("%p: update buffer:%u req:%"PRIu64, impl, id, buffer->this.requested);
-	return res;
+
+	return buffer->this.requested > 0 ? 1 : 0;
 }
 
 static int
@@ -643,8 +640,10 @@ static inline void copy_position(struct stream *impl, int64_t queued)
 		impl->time.queued = queued;
 		impl->quantum = p->clock.duration;
 	}
-	if (SPA_LIKELY(impl->rate_match != NULL))
+	if (SPA_LIKELY(impl->rate_match != NULL)) {
 		impl->rate_queued = impl->rate_match->delay;
+		impl->rate_requested = impl->rate_match->size;
+	}
 	SPA_SEQ_WRITE(impl->seq);
 }
 
@@ -677,11 +676,11 @@ static int impl_send_command(void *object, const struct spa_command *command)
 				if (impl->io != NULL)
 					impl->io->status = SPA_STATUS_NEED_DATA;
 			}
-			else if (!impl->process_rt && !impl->driving) {
+			else {
 				copy_position(impl, impl->queued.incount);
+				if (!impl->process_rt && !impl->driving)
 					call_process(impl);
 			}
-
 			stream_set_state(stream, PW_STREAM_STATE_STREAMING, 0, NULL);
 		}
 		break;
@@ -2355,13 +2354,14 @@ int pw_stream_get_time_n(struct pw_stream *stream, struct pw_time *time, size_t
 {
 	struct stream *impl = SPA_CONTAINER_OF(stream, struct stream, this);
 	uintptr_t seq1, seq2;
-	uint32_t buffered, quantum, index;
+	uint32_t buffered, quantum, index, requested;
 	int32_t avail_buffers;
 
 	do {
 		seq1 = SPA_SEQ_READ(impl->seq);
 		memcpy(time, &impl->time, SPA_MIN(size, sizeof(struct pw_time)));
 		buffered = impl->rate_queued;
+		requested = impl->rate_requested;
 		quantum = impl->quantum;
 		seq2 = SPA_SEQ_READ(impl->seq);
 	} while (!SPA_SEQ_READ_SUCCESS(seq1, seq2));
@@ -2382,8 +2382,10 @@ int pw_stream_get_time_n(struct pw_stream *stream, struct pw_time *time, size_t
 		time->buffered = buffered;
 	if (size >= offsetof(struct pw_time, avail_buffers))
 		time->queued_buffers = impl->n_buffers - avail_buffers;
-	if (size >= sizeof(struct pw_time))
+	if (size >= offsetof(struct pw_time, requested))
 		time->avail_buffers = avail_buffers;
+	if (size >= sizeof(struct pw_time))
+		time->requested = requested;
 
 	pw_log_trace_fp("%p: %"PRIi64" %"PRIi64" %"PRIu64" %d/%d %"PRIu64" %"
 			PRIu64" %"PRIu64" %"PRIu64" %"PRIu64" %d/%d", stream,

src/pipewire/stream.h

@@ -330,8 +330,11 @@ struct pw_time {
 	uint64_t buffered;		/**< for audio/raw streams, this contains the extra
 					  *  number of frames buffered in the resampler.
 					  *  Since 0.3.50. */
-	uint32_t queued_buffers;	/**< The number of buffers that are queued. Since 0.3.50 */
-	uint32_t avail_buffers;		/**< The number of buffers that can be dequeued. Since 0.3.50 */
+	uint32_t queued_buffers;	/**< the number of buffers that are queued. Since 0.3.50 */
+	uint32_t avail_buffers;		/**< the number of buffers that can be dequeued. Since 0.3.50 */
+	uint64_t requested;		/**< for audio/raw playback streams, this contains the number of
+					  *  samples requested by the resampler to fill the current
+					  *  quantum. Since 1.1.0 */
 };
 
 #include <pipewire/port.h>

src/tools/pw-cat.c

@@ -932,11 +932,11 @@ static void do_print_delay(void *userdata, uint64_t expirations)
 	pw_stream_get_time_n(data->stream, &time, sizeof(time));
 	printf("stream time: now:%"PRIi64" rate:%u/%u ticks:%"PRIu64
 			" delay:%"PRIi64" queued:%"PRIu64
-			" buffered:%"PRIi64" buffers:%u avail:%u\n",
+			" buffered:%"PRIi64" buffers:%u avail:%u req:%"PRIu64"\n",
 		time.now,
 		time.rate.num, time.rate.denom,
 		time.ticks, time.delay, time.queued, time.buffered,
-		time.queued_buffers, time.avail_buffers);
+		time.queued_buffers, time.avail_buffers, time.requested);
 }
 
 enum {