Review at  https://gerrit.osmocom.org/7779

jibuf: re-sync clock out of sync timestamps

Change-Id: I33556b33d7549654442d9bdd7f31128792506652
---
M src/jibuf.c
M tests/jibuf/jibuf_test.c
M tests/jibuf/jibuf_test.ok
M tests/jibuf/jibuf_tool.c
4 files changed, 152 insertions(+), 7 deletions(-)


  git pull ssh://gerrit.osmocom.org:29418/libosmo-netif refs/changes/79/7779/1

diff --git a/src/jibuf.c b/src/jibuf.c
index aa66302..5d568e1 100644
--- a/src/jibuf.c
+++ b/src/jibuf.c
@@ -32,6 +32,7 @@
 
 /* Sampling rate (in Hz) */
 /* TODO: SAMPLE RATE can be guessed from rtp.p_type */
+#define SAMPLES_PER_PKT 160
 #define SAMPLE_RATE 8000
 
 /* TUNABLE PARAMETERS: */
@@ -146,6 +147,27 @@
        uint32_t current_tx_ts = msg_get_timestamp(msg);
 
        return samples2ms((current_tx_ts - jb->ref_tx_ts)) - (current_rx_ts - 
jb->ref_rx_ts);
+}
+
+static bool msg_is_in_sequence(struct jibuf *jb, struct msgb *msg)
+{
+       uint32_t current_tx_ts = msg_get_timestamp(msg);
+       uint16_t current_seq = msg_get_sequence(msg);
+       return (current_tx_ts - jb->ref_tx_ts) == (current_seq - 
jb->ref_tx_seq)*SAMPLES_PER_PKT;
+}
+
+/* If packet contains a mark -> start of talkspurt.
+ * A lot of packets may have been suppressed by the sender before it,
+ * so let's take it as a reference
+ * If packet timestamp is not aligned with sequence
+ * number, then we are most probaly starting a talkspurt */
+static bool msg_is_syncpoint(struct jibuf *jb, struct msgb* msg)
+{
+       bool res = msg_get_marker(msg) || !msg_is_in_sequence(jb, msg);
+       if(res)
+               LOGP(DLMIB, LOGL_DEBUG, "syncpoint: %"PRIu16": marker=%d 
in_seq=%d\n",
+                       msg_get_sequence(msg), msg_get_marker(msg), 
msg_is_in_sequence(jb, msg));
+       return res;
 }
 
 static void msg_set_as_reference(struct jibuf *jb, struct msgb *msg)
@@ -280,11 +302,8 @@
 
        clock_gettime_timeval(CLOCK_MONOTONIC, &jb->last_enqueue_time);
 
-       /* If packet contains a mark -> start of talkspurt.
-        * A lot of packets may have been suppressed by the sender before it,
-        * so let's take it as a reference
-        */
-       if (!jb->started || msg_get_marker(msg)) {
+       /* Check if it's time to sync, ie. start of talkspurt */
+       if (!jb->started || msg_is_syncpoint(jb, msg)) {
                jb->started = true;
                msg_set_as_reference(jb, msg);
                rel_delay = 0;
diff --git a/tests/jibuf/jibuf_test.c b/tests/jibuf/jibuf_test.c
index a6fe1b9..a9a7ff0 100644
--- a/tests/jibuf/jibuf_test.c
+++ b/tests/jibuf/jibuf_test.c
@@ -614,6 +614,72 @@
        osmo_jibuf_delete(jb);
 }
 
+void test_rtp_out_of_sync(unsigned int time_inc_ms, uint16_t seq_nosync_inc, 
uint32_t ts_nosync_inc, bool expect_drop)
+{
+       int min_delay = 60;
+       struct msgb *msg;
+       int rc;
+
+       printf("===test_rtp_out_of_sync(%u, %"PRIu16", %"PRIu32", %d)===\n",
+               time_inc_ms, seq_nosync_inc, ts_nosync_inc, expect_drop);
+
+       clock_override_enable(true);
+       clock_override_set(0, 0);
+       rtp_init(32, 400);
+       jb = osmo_jibuf_alloc(NULL);
+       osmo_jibuf_set_dequeue_cb(jb, dequeue_cb, NULL);
+       osmo_jibuf_set_min_delay(jb, min_delay);
+       osmo_jibuf_set_max_delay(jb, 200);
+
+       /* First rtp at t=0, should be scheduled in min_delay time */
+       clock_debug("enqueue 1st packet (seq=33, ts=560)");
+       ENQUEUE_NEXT(jb);
+       clock_override_add(0, TIME_RTP_PKT_MS*1000);
+       clock_debug("enqueue 2nd packet (seq=34, ts=720)");
+       ENQUEUE_NEXT(jb);
+
+       clock_override_add(0, time_inc_ms*1000);
+       clock_debug("2 packets dequeued");
+       osmo_select_main(0);
+
+        /* We are at t=20+time_inc_ms, next pkt would normally be dropped 
since it is
+         * pretty late, but since seq and timestamp are out of sync, which
+         * means the sender had some clock issues, the jibuf is going to take
+         * this new tuple as reference and accept it.
+       */
+       clock_debug("enqueue late pkt with possible sync change");
+       rtp_init(rtp_next_seq + seq_nosync_inc, rtp_next_ts + ts_nosync_inc);
+       msg = rtp_new(rtp_next_seq, rtp_next_ts);
+       rc = osmo_jibuf_enqueue(jb, msg);
+       if (expect_drop) {
+               OSMO_ASSERT(rc < 0);
+               msgb_free(msg);
+       } else {
+               OSMO_ASSERT(rc == 0);
+       }
+
+       clock_debug("enqueue late pkt after possible resync");
+       clock_override_add(0, TIME_RTP_PKT_MS*1000);
+       msg = rtp_next();
+       rc = osmo_jibuf_enqueue(jb, msg);
+       if (expect_drop) {
+               OSMO_ASSERT(rc < 0);
+               msgb_free(msg);
+       } else {
+               OSMO_ASSERT(rc == 0);
+       }
+
+       if (!expect_drop) {
+               clock_debug("2 packets dequeued");
+               clock_override_add(0, min_delay*1000);
+               osmo_select_main(0);
+       }
+
+       OSMO_ASSERT(osmo_jibuf_empty(jb));
+
+       osmo_jibuf_delete(jb);
+}
+
 int main(int argc, char **argv)
 {
 
@@ -639,6 +705,9 @@
        test_seq_wraparound();
        test_timestamp_wraparound();
        test_rtp_marker();
+       test_rtp_out_of_sync(80*TIME_RTP_PKT_MS, 5, 5*SAMPLES_PER_PKT, true);
+       test_rtp_out_of_sync(80*TIME_RTP_PKT_MS, 6, 5*SAMPLES_PER_PKT, false);
+       test_rtp_out_of_sync(80*TIME_RTP_PKT_MS, 5, 5*SAMPLES_PER_PKT + 3, 
false);
 
        fprintf(stdout, "OK: Test passed\n");
        return EXIT_SUCCESS;
diff --git a/tests/jibuf/jibuf_test.ok b/tests/jibuf/jibuf_test.ok
index 3103781..a152d24 100644
--- a/tests/jibuf/jibuf_test.ok
+++ b/tests/jibuf/jibuf_test.ok
@@ -365,4 +365,48 @@
 sys={0.200000}, mono={0.200000}: clock_override_add
 sys={0.200000}, mono={0.200000}: dequeue: seq=35 ts=880 INTERMEDIATE
 sys={0.200000}, mono={0.200000}: dequeue: seq=36 ts=1040 LATEST
+===test_rtp_out_of_sync(1600, 5, 800, 1)===
+sys={0.000000}, mono={0.000000}: clock_override_set
+sys={0.000000}, mono={0.000000}: enqueue 1st packet (seq=33, ts=560)
+sys={0.020000}, mono={0.020000}: clock_override_add
+sys={0.020000}, mono={0.020000}: enqueue 2nd packet (seq=34, ts=720)
+sys={1.620000}, mono={1.620000}: clock_override_add
+sys={1.620000}, mono={1.620000}: 2 packets dequeued
+sys={1.620000}, mono={1.620000}: dequeue: seq=33 ts=560 INTERMEDIATE
+sys={1.620000}, mono={1.620000}: dequeue: seq=34 ts=720 LATEST
+sys={1.620000}, mono={1.620000}: enqueue late pkt with possible sync change
+sys={1.620000}, mono={1.620000}: enqueue late pkt after possible resync
+sys={1.640000}, mono={1.640000}: clock_override_add
+===test_rtp_out_of_sync(1600, 6, 800, 0)===
+sys={0.000000}, mono={0.000000}: clock_override_set
+sys={0.000000}, mono={0.000000}: enqueue 1st packet (seq=33, ts=560)
+sys={0.020000}, mono={0.020000}: clock_override_add
+sys={0.020000}, mono={0.020000}: enqueue 2nd packet (seq=34, ts=720)
+sys={1.620000}, mono={1.620000}: clock_override_add
+sys={1.620000}, mono={1.620000}: 2 packets dequeued
+sys={1.620000}, mono={1.620000}: dequeue: seq=33 ts=560 INTERMEDIATE
+sys={1.620000}, mono={1.620000}: dequeue: seq=34 ts=720 LATEST
+sys={1.620000}, mono={1.620000}: enqueue late pkt with possible sync change
+sys={1.620000}, mono={1.620000}: enqueue late pkt after possible resync
+sys={1.640000}, mono={1.640000}: clock_override_add
+sys={1.640000}, mono={1.640000}: 2 packets dequeued
+sys={1.700000}, mono={1.700000}: clock_override_add
+sys={1.700000}, mono={1.700000}: dequeue: seq=40 ts=1520 INTERMEDIATE
+sys={1.700000}, mono={1.700000}: dequeue: seq=41 ts=1680 LATEST
+===test_rtp_out_of_sync(1600, 5, 803, 0)===
+sys={0.000000}, mono={0.000000}: clock_override_set
+sys={0.000000}, mono={0.000000}: enqueue 1st packet (seq=33, ts=560)
+sys={0.020000}, mono={0.020000}: clock_override_add
+sys={0.020000}, mono={0.020000}: enqueue 2nd packet (seq=34, ts=720)
+sys={1.620000}, mono={1.620000}: clock_override_add
+sys={1.620000}, mono={1.620000}: 2 packets dequeued
+sys={1.620000}, mono={1.620000}: dequeue: seq=33 ts=560 INTERMEDIATE
+sys={1.620000}, mono={1.620000}: dequeue: seq=34 ts=720 LATEST
+sys={1.620000}, mono={1.620000}: enqueue late pkt with possible sync change
+sys={1.620000}, mono={1.620000}: enqueue late pkt after possible resync
+sys={1.640000}, mono={1.640000}: clock_override_add
+sys={1.640000}, mono={1.640000}: 2 packets dequeued
+sys={1.700000}, mono={1.700000}: clock_override_add
+sys={1.700000}, mono={1.700000}: dequeue: seq=39 ts=1523 INTERMEDIATE
+sys={1.700000}, mono={1.700000}: dequeue: seq=40 ts=1683 LATEST
 OK: Test passed
diff --git a/tests/jibuf/jibuf_tool.c b/tests/jibuf/jibuf_tool.c
index 4a35a94..dcfd130 100644
--- a/tests/jibuf/jibuf_tool.c
+++ b/tests/jibuf/jibuf_tool.c
@@ -35,6 +35,7 @@
        int transit;
        double jitter;
        uint32_t timestamp;
+       uint16_t seq;
 };
 
 struct rtp_pkt_info {
@@ -127,6 +128,16 @@
        return ts->tv_sec * 1000 + ts->tv_usec / 1000;
 }
 
+bool pkt_is_syncpoint(struct msgb* msg, uint16_t prev_seq, uint32_t 
prev_timestamp)
+{
+       struct rtp_hdr *rtph = osmo_rtp_get_hdr(msg);
+
+       uint16_t current_seq = ntohs(rtph->sequence);
+       uint32_t current_tx_ts = ntohl(rtph->timestamp);
+       bool insync = (current_tx_ts - prev_timestamp) == (current_seq - 
prev_seq)*SAMPLES_PER_PKT;
+       return !insync || rtph->marker;
+}
+
 int32_t calc_rel_transmit_time(uint32_t tx_0, uint32_t tx_f, uint32_t rx_0, 
uint32_t rx_f, bool tx_is_samples, bool pre)
 {
        int32_t rxdiff, txdiff, res;
@@ -216,7 +227,7 @@
 
        /* If pkt->marker -> init of talkspurt, there may be missing packets 
before,
         * better to start calculating the jitter from here */
-       if (postqueue_started && !rtph->marker) {
+       if (postqueue_started && !pkt_is_syncpoint(msg, postqueue_prev.seq, 
postqueue_prev.timestamp)) {
                /* In random test mode we now the sender time, so we get real
                 * jitter results using it */
                if(opt_test_rand) {
@@ -244,6 +255,7 @@
        }
 
        postqueue_prev = pinfo->postqueue;
+       postqueue_prev.seq = htons(rtph->sequence);
 
        pkt_add_result(msg, false);
 
@@ -269,7 +281,7 @@
 
        /* If pkt->marker -> init of talkspurt, there may be missing packets 
before,
         * better to start calculating the jitter from here */
-       if (prequeue_started && !rtph->marker) {
+       if (prequeue_started && !pkt_is_syncpoint(msg, prequeue_prev.seq, 
prequeue_prev.timestamp)) {
                /* In random test mode we now the sender time, so we get real
                 * jitter results using it */
                if(opt_test_rand) {
@@ -297,6 +309,7 @@
        }
 
        prequeue_prev = pinfo->prequeue;
+       prequeue_prev.seq = htons(rtph->sequence);
 
        int n = osmo_jibuf_enqueue(jb, msg);
 

-- 
To view, visit https://gerrit.osmocom.org/7779
To unsubscribe, visit https://gerrit.osmocom.org/settings

Gerrit-MessageType: newchange
Gerrit-Change-Id: I33556b33d7549654442d9bdd7f31128792506652
Gerrit-PatchSet: 1
Gerrit-Project: libosmo-netif
Gerrit-Branch: master
Gerrit-Owner: Pau Espin Pedrol <pes...@sysmocom.de>

Reply via email to