aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorPau Espin Pedrol <pespin@sysmocom.de>2017-04-23 19:22:15 +0200
committerPau Espin Pedrol <pespin@sysmocom.de>2018-02-28 11:31:06 +0100
commit818c785dc5db6dd75f8090bcfa6c2e3b6cde0946 (patch)
tree90590738551c3c3ba1bfe2a6866e59d371456e67
parent332214af6f77c4e91cd58abfc83ec2a46ea566af (diff)
jibuf: re-sync clock out of sync timestamps
-rw-r--r--src/jibuf.c29
-rw-r--r--tests/jibuf/jibuf_test.c69
-rw-r--r--tests/jibuf/jibuf_test.ok44
-rw-r--r--tests/jibuf/jibuf_tool.c17
4 files changed, 152 insertions, 7 deletions
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: */
@@ -148,6 +149,27 @@ static int calc_pkt_rel_delay(struct jibuf *jb, struct msgb *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)
{
jb->ref_rx_ts = timeval2ms(&jb->last_enqueue_time);
@@ -280,11 +302,8 @@ int osmo_jibuf_enqueue(struct jibuf *jb, struct msgb *msg)
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 @@ static void test_rtp_marker(void)
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 @@ int main(int argc, char **argv)
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.140000}, mono={0.140000}: 2 packets dequeued
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 @@ struct checkpoint {
int transit;
double jitter;
uint32_t timestamp;
+ uint16_t seq;
};
struct rtp_pkt_info {
@@ -127,6 +128,16 @@ static uint32_t timeval2ms(const struct timeval *ts)
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 @@ void dequeue_cb(struct msgb *msg, void *data)
/* 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 @@ void dequeue_cb(struct msgb *msg, void *data)
}
postqueue_prev = pinfo->postqueue;
+ postqueue_prev.seq = htons(rtph->sequence);
pkt_add_result(msg, false);
@@ -269,7 +281,7 @@ void pkt_arrived_cb(void *data)
/* 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 @@ void pkt_arrived_cb(void *data)
}
prequeue_prev = pinfo->prequeue;
+ prequeue_prev.seq = htons(rtph->sequence);
int n = osmo_jibuf_enqueue(jb, msg);