latency: handle negative latency correctly

In our current world, it is possible to have a negative delay. This
means that the stream should be delayed to sync with other streams.

The pulse-server sets negative delay and the Latency message can hold
those negative values so make sure we handle them in the helper
functions as well.

Do the delay calculations in pw_stream and JACK with signed values to
correctly handle negative values. Clamp JACK latency range to 0 because
negative latency is not supported in JACK.

We should also probably make sure we never end up with negative
latency, mostly in ALSA when we set a Latency offset, but that is
another detail.

pipewire-jack/src/pipewire-jack.c

@@ -6583,6 +6583,7 @@ void jack_port_get_latency_range (jack_port_t *port, jack_latency_callback_mode_
 	jack_nframes_t nframes, rate;
 	int direction;
 	struct spa_latency_info *info;
+	int64_t min, max;
 
 	return_if_fail(o != NULL);
 	c = o->client;
@@ -6601,10 +6602,15 @@ void jack_port_get_latency_range (jack_port_t *port, jack_latency_callback_mode_
 	rate = jack_get_sample_rate((jack_client_t*)c);
 	info = &o->port.latency[direction];
 
-	range->min = (jack_nframes_t)((info->min_quantum * nframes) +
-		info->min_rate + (info->min_ns * rate) / SPA_NSEC_PER_SEC);
-	range->max = (jack_nframes_t)((info->max_quantum * nframes) +
-		info->max_rate + (info->max_ns * rate) / SPA_NSEC_PER_SEC);
+	min = (int64_t)(info->min_quantum * nframes) +
+		info->min_rate +
+		(info->min_ns * (int64_t)rate) / (int64_t)SPA_NSEC_PER_SEC;
+	max = (int64_t)(info->max_quantum * nframes) +
+		info->max_rate +
+		(info->max_ns * (int64_t)rate) / (int64_t)SPA_NSEC_PER_SEC;
+
+	range->min = SPA_MAX(min, 0);
+	range->max = SPA_MAX(max, 0);
 
 	pw_log_debug("%p: %s get %d latency range %d %d", c, o->port.name,
 			mode, range->min, range->max);
@@ -6649,13 +6655,13 @@ void jack_port_set_latency_range (jack_port_t *port, jack_latency_callback_mode_
 		nframes = 1;
 
 	latency.min_rate = range->min;
-	if (latency.min_rate >= nframes) {
+	if (latency.min_rate >= (int32_t)nframes) {
 		latency.min_quantum = latency.min_rate / nframes;
 		latency.min_rate %= nframes;
 	}
 
 	latency.max_rate = range->max;
-	if (latency.max_rate >= nframes) {
+	if (latency.max_rate >= (int32_t)nframes) {
 		latency.max_quantum = latency.max_rate / nframes;
 		latency.max_rate %= nframes;
 	}

spa/include/spa/param/latency-utils.h

@@ -38,21 +38,27 @@ spa_latency_info_combine_start(struct spa_latency_info *info, enum spa_direction
 {
 	*info = SPA_LATENCY_INFO(direction,
 			.min_quantum = FLT_MAX,
-			.max_quantum = 0.0f,
-			.min_rate = UINT32_MAX,
-			.max_rate = 0,
-			.min_ns = UINT64_MAX,
-			.max_ns = 0);
+			.max_quantum = FLT_MIN,
+			.min_rate = INT32_MAX,
+			.max_rate = INT32_MIN,
+			.min_ns = INT64_MAX,
+			.max_ns = INT64_MIN);
 }
 static inline void
 spa_latency_info_combine_finish(struct spa_latency_info *info)
 {
 	if (info->min_quantum == FLT_MAX)
 		info->min_quantum = 0;
-	if (info->min_rate == UINT32_MAX)
+	if (info->max_quantum == FLT_MIN)
+		info->max_quantum = 0;
+	if (info->min_rate == INT32_MAX)
 		info->min_rate = 0;
-	if (info->min_ns == UINT64_MAX)
+	if (info->max_rate == INT32_MIN)
+		info->max_rate = 0;
+	if (info->min_ns == INT64_MAX)
 		info->min_ns = 0;
+	if (info->max_ns == INT64_MIN)
+		info->max_ns = 0;
 }
 
 static inline int

spa/include/spa/param/latency.h

@@ -50,10 +50,10 @@ struct spa_latency_info {
 	enum spa_direction direction;
 	float min_quantum;
 	float max_quantum;
-	uint32_t min_rate;
-	uint32_t max_rate;
-	uint64_t min_ns;
-	uint64_t max_ns;
+	int32_t min_rate;
+	int32_t max_rate;
+	int64_t min_ns;
+	int64_t max_ns;
 };
 
 #define SPA_LATENCY_INFO(dir,...) ((struct spa_latency_info) { .direction = (dir), ## __VA_ARGS__ })
@@ -74,8 +74,8 @@ enum spa_param_process_latency {
 /** Helper structure for managing process latency objects */
 struct spa_process_latency_info {
 	float quantum;
-	uint32_t rate;
-	uint64_t ns;
+	int32_t rate;
+	int64_t ns;
 };
 
 #define SPA_PROCESS_LATENCY_INFO_INIT(...)	((struct spa_process_latency_info) { __VA_ARGS__ })

src/pipewire/stream.c

@@ -2371,7 +2371,8 @@ int pw_stream_get_time_n(struct pw_stream *stream, struct pw_time *time, size_t
 
 	time->delay += (int64_t)(((impl->latency.min_quantum + impl->latency.max_quantum) / 2.0f) * quantum);
 	time->delay += (impl->latency.min_rate + impl->latency.max_rate) / 2;
-	time->delay += ((impl->latency.min_ns + impl->latency.max_ns) / 2) * time->rate.denom / SPA_NSEC_PER_SEC;
+	time->delay += ((impl->latency.min_ns + impl->latency.max_ns) / 2) *
+		(int64_t)time->rate.denom / (int64_t)SPA_NSEC_PER_SEC;
 
 	avail_buffers = spa_ringbuffer_get_read_index(&impl->dequeued.ring, &index);
 	avail_buffers = SPA_CLAMP(avail_buffers, 0, (int32_t)impl->n_buffers);

src/pipewire/stream.h

@@ -303,6 +303,13 @@ struct pw_stream_control {
  * caused by the hardware. The delay is usually quite stable and should only change when
  * the topology, quantum or samplerate of the graph changes.
  *
+ * The delay requires the application to send the stream early relative to other synchronized
+ * streams in order to arrive at the edge of the graph in time. This is usually done by
+ * delaying the other streams with the given delay.
+ *
+ * Note that the delay can be negative. A negative delay means that this stream should be
+ * delayed with the (positive) delay relative to other streams.
+ *
  * pw_time.queued and pw_time.buffered is expressed in the time domain of the stream,
  * or the format that is used for the buffers of this stream.
  *