tunnel: Fix latency calculations

Currently module-tunnel uses only a rough estimate of the current stream
latency and reports wrong latencies in certain situations. This leads to
very inexact and unstable latency reports for the virtual sink.

This patch fixes the issue by introducing latency snapshots like they
are used in module-loopback. Because the latency reports are now correct,
the update interval for latency re-calculations can be reduced to 1s.

Part-of: <https://gitlab.freedesktop.org/pulseaudio/pulseaudio/-/merge_requests/53>

src/modules/module-tunnel.c

@@ -121,7 +121,7 @@ static const char* const valid_modargs[] = {
 
 #define DEFAULT_TIMEOUT 5
 
-#define LATENCY_INTERVAL (10*PA_USEC_PER_SEC)
+#define LATENCY_INTERVAL (1*PA_USEC_PER_SEC)
 
 #define MIN_NETWORK_LATENCY_USEC (8*PA_USEC_PER_MSEC)
 
@@ -131,6 +131,7 @@ enum {
     SINK_MESSAGE_REQUEST = PA_SINK_MESSAGE_MAX,
     SINK_MESSAGE_REMOTE_SUSPEND,
     SINK_MESSAGE_UPDATE_LATENCY,
+    SINK_MESSAGE_GET_LATENCY_SNAPSHOT,
     SINK_MESSAGE_POST
 };
 
@@ -142,7 +143,8 @@ enum {
 enum {
     SOURCE_MESSAGE_POST = PA_SOURCE_MESSAGE_MAX,
     SOURCE_MESSAGE_REMOTE_SUSPEND,
-    SOURCE_MESSAGE_UPDATE_LATENCY
+    SOURCE_MESSAGE_UPDATE_LATENCY,
+    SOURCE_MESSAGE_GET_LATENCY_SNAPSHOT
 };
 
 #define DEFAULT_FRAGSIZE_MSEC 25
@@ -212,7 +214,9 @@ struct userdata {
     uint32_t device_index;
     uint32_t channel;
 
-    int64_t counter, counter_delta;
+    int64_t counter;
+    uint64_t receive_counter;
+    uint64_t receive_snapshot;
 
     bool remote_corked:1;
     bool remote_suspended:1;
@@ -517,6 +521,13 @@ static int sink_process_msg(pa_msgobject *o, int code, void *data, int64_t offse
             return 0;
         }
 
+        case SINK_MESSAGE_GET_LATENCY_SNAPSHOT: {
+            int64_t *send_counter = data;
+
+            *send_counter = u->counter;
+            return 0;
+        }
+
         case SINK_MESSAGE_REQUEST:
 
             pa_assert(offset > 0);
@@ -559,7 +570,7 @@ static int sink_process_msg(pa_msgobject *o, int code, void *data, int64_t offse
 
             pa_pstream_send_memblock(u->pstream, u->channel, 0, PA_SEEK_RELATIVE, chunk);
 
-            u->counter_delta += (int64_t) chunk->length;
+            u->receive_counter += chunk->length;
 
             return 0;
     }
@@ -628,6 +639,13 @@ static int source_process_msg(pa_msgobject *o, int code, void *data, int64_t off
             return 0;
         }
 
+        case SOURCE_MESSAGE_GET_LATENCY_SNAPSHOT: {
+            int64_t *send_counter = data;
+
+            *send_counter = u->counter;
+            return 0;
+        }
+
         case SOURCE_MESSAGE_POST: {
             pa_memchunk c;
 
@@ -779,6 +797,9 @@ static void stream_get_latency_callback(pa_pdispatch *pd, uint32_t command, uint
     struct timeval local, remote, now;
     pa_sample_spec *ss;
     int64_t delay;
+#ifdef TUNNEL_SINK
+    uint64_t send_counter;
+#endif
 
     pa_assert(pd);
     pa_assert(u);
@@ -826,7 +847,7 @@ static void stream_get_latency_callback(pa_pdispatch *pd, uint32_t command, uint
     pa_gettimeofday(&now);
 
     /* Calculate transport usec */
-    if (pa_timeval_cmp(&local, &remote) < 0 && pa_timeval_cmp(&remote, &now)) {
+    if (pa_timeval_cmp(&local, &remote) < 0 && pa_timeval_cmp(&remote, &now) < 0) {
         /* local and remote seem to have synchronized clocks */
 #ifdef TUNNEL_SINK
         u->transport_usec = pa_timeval_diff(&remote, &local);
@@ -859,11 +880,12 @@ static void stream_get_latency_callback(pa_pdispatch *pd, uint32_t command, uint
     delay += (int64_t) u->transport_usec;
 #endif
 
-    /* Now correct by what we have have read/written since we requested the update */
+    /* Now correct by what we have have written since we requested the update. This
+     * is not necessary for the source, because if data is received between request
+     * and reply, it was already posted before we requested the source latency. */
 #ifdef TUNNEL_SINK
-    delay += (int64_t) pa_bytes_to_usec((uint64_t) u->counter_delta, ss);
-#else
-    delay -= (int64_t) pa_bytes_to_usec((uint64_t) u->counter_delta, ss);
+    pa_asyncmsgq_send(u->sink->asyncmsgq, PA_MSGOBJECT(u->sink), SINK_MESSAGE_GET_LATENCY_SNAPSHOT, &send_counter, 0, NULL);
+    delay += (int64_t) pa_bytes_to_usec(send_counter - u->receive_snapshot, ss);
 #endif
 
 #ifdef TUNNEL_SINK
@@ -901,7 +923,7 @@ static void request_latency(struct userdata *u) {
     pa_pdispatch_register_reply(u->pdispatch, tag, DEFAULT_TIMEOUT, stream_get_latency_callback, u, NULL);
 
     u->ignore_latency_before = tag;
-    u->counter_delta = 0;
+    u->receive_snapshot = u->receive_counter;
 }
 
 /* Called from main context */
@@ -1823,7 +1845,7 @@ static void pstream_memblock_callback(pa_pstream *p, uint32_t channel, int64_t o
 
     pa_asyncmsgq_send(u->source->asyncmsgq, PA_MSGOBJECT(u->source), SOURCE_MESSAGE_POST, PA_UINT_TO_PTR(seek), offset, chunk);
 
-    u->counter_delta += (int64_t) chunk->length;
+    u->receive_counter += chunk->length;
 }
 #endif
 
@@ -1979,7 +2001,9 @@ int pa__init(pa_module*m) {
     u->ignore_latency_before = 0;
     u->transport_usec = u->thread_transport_usec = 0;
     u->remote_suspended = u->remote_corked = false;
-    u->counter = u->counter_delta = 0;
+    u->counter = 0;
+    u->receive_snapshot = 0;
+    u->receive_counter = 0;
 
     u->rtpoll = pa_rtpoll_new();