osmux: Improve code handling lost packets

With previous implementation, only up to a maximum of 6-7 rtp lost packets
can be replayed in best-case scenario. In other scenarios, big gaps can
easily occur, for instance if a lot of packets are lost when the current
batch is still empty.

If we forget to replay some lost RTP packets, then those are not taken into
account in the Osmux stream, which means sequence numbers and timestamps will
become different when Osmux re-generates the frames into RTP packets on the
other side, and some receivers may not play nice with this jumps in time,
eg. audio delays.

This new implementation tries to maintain status on the whole circuit
level rather than only on current batch. To simplify things, it clones
payload from latest packet to enqueue rather than last previosuly packet
enqueued.

Change-Id: I3b77d372cedadfd5588384bea0e5d3fd475195ce

1 files changed, 36 insertions, 21 deletions

diff --git a/src/osmux.c b/src/osmux.c
index b3c43e2..c3b359b 100644
--- a/src/osmux.c
+++ b/src/osmux.c
@@ -49,6 +49,11 @@
 
 static void *osmux_ctx;
 
+static bool seq_before(uint32_t seq1, uint32_t seq2)
+{
+        return (int32_t)(seq1-seq2) < 0;
+}
+
 static uint32_t osmux_get_payload_len(struct osmux_hdr *osmuxh)
 {
 	return osmo_amr_bytes(osmuxh->amr_ft) * (osmuxh->ctr+1);
@@ -207,6 +212,9 @@ struct osmux_circuit {
 	struct llist_head	msg_list;
 	int			nmsgs;
 	int			dummy;
+	uint16_t		last_snd_sequence;
+	uint32_t		last_snd_timestamp;
+	bool			first_queued;
 };
 
 static int osmux_batch_enqueue(struct msgb *msg, struct osmux_circuit *circuit,
@@ -458,25 +466,15 @@ static int osmux_rtp_amr_payload_len(struct msgb *msg, struct rtp_hdr *rtph)
 	return amr_payload_len;
 }
 
-static void osmux_replay_lost_packets(struct osmux_circuit *circuit,
-				      struct rtp_hdr *cur_rtph, int batch_factor)
+static int
+osmux_replay_lost_packets(struct osmux_circuit *circuit,  struct msgb *msg,
+			  struct rtp_hdr *cur_rtph, int batch_factor)
 {
 	int16_t diff;
-	struct msgb *last;
 	struct rtp_hdr *rtph;
 	int i;
 
-	/* Have we see any RTP packet in this batch before? */
-	if (llist_empty(&circuit->msg_list))
-		return;
-
-	/* Get last RTP packet seen in this batch */
-	last = llist_entry(circuit->msg_list.prev, struct msgb, list);
-	rtph = osmo_rtp_get_hdr(last);
-	if (rtph == NULL)
-		return;
-
-	diff = ntohs(cur_rtph->sequence) - ntohs(rtph->sequence);
+	diff = ntohs(cur_rtph->sequence) - ntohs(circuit->last_snd_sequence);
 
 	/* Lifesaver: make sure bugs don't spawn lots of clones */
 	if (diff > 16)
@@ -489,29 +487,33 @@ static void osmux_replay_lost_packets(struct osmux_circuit *circuit,
 		struct msgb *clone;
 
 		/* Clone last RTP packet seen */
-		clone = msgb_alloc(last->data_len, "RTP clone");
+		clone = msgb_alloc(msg->data_len, "RTP clone");
 		if (!clone)
 			continue;
 
-		memcpy(clone->data, last->data, last->len);
-		msgb_put(clone, last->len);
+		memcpy(clone->data, msg->data, msg->len);
+		msgb_put(clone, msg->len);
 
 		/* The original RTP message has been already sanity check. */
 		rtph = osmo_rtp_get_hdr(clone);
 
 		/* Adjust sequence number and timestamp */
-		rtph->sequence = htons(ntohs(rtph->sequence) + i);
-		rtph->timestamp = htonl(ntohl(rtph->timestamp) +
+		rtph->sequence = htons(ntohs(circuit->last_snd_sequence) + i);
+		rtph->timestamp = htonl(ntohl(circuit->last_snd_timestamp) +
 					DELTA_RTP_TIMESTAMP);
 
 		/* No more room in this batch, skip padding with more clones */
 		if (osmux_batch_enqueue(clone, circuit, batch_factor) < 0) {
 			msgb_free(clone);
-			break;
+			return 1;
 		}
 
 		LOGP(DLMIB, LOGL_ERROR, "adding cloned RTP\n");
+		circuit->last_snd_sequence = rtph->sequence;
+		circuit->last_snd_timestamp = rtph->timestamp;
+
 	}
+	return 0;
 }
 
 static struct osmux_circuit *
@@ -607,6 +609,7 @@ osmux_batch_add(struct osmux_batch *batch, uint32_t batch_factor, struct msgb *m
 	/* Extra validation: check if this message already exists, should not
 	 * happen but make sure we don't propagate duplicated messages.
 	 */
+	 /* FIXME: overwrite old packet with new one as the old one was probably lost & cloned */
 	llist_for_each_entry(cur, &circuit->msg_list, list) {
 		struct rtp_hdr *rtph2 = osmo_rtp_get_hdr(cur);
 		if (rtph2 == NULL)
@@ -620,8 +623,16 @@ osmux_batch_add(struct osmux_batch *batch, uint32_t batch_factor, struct msgb *m
 			return -1;
 		}
 	}
+
+	/* Avoid going further if the packet is old, drop the packet */
+	if (circuit->first_queued &&
+	    seq_before(ntohs(rtph->sequence), ntohs(circuit->last_snd_sequence)))
+		return -1;
+
 	/* Handle RTP packet loss scenario */
-	osmux_replay_lost_packets(circuit, rtph, batch_factor);
+	if (circuit->first_queued &&
+	    osmux_replay_lost_packets(circuit, msg, rtph, batch_factor) > 0)
+		return 1;
 
 	/* This batch is full, force batch delivery */
 	if (osmux_batch_enqueue(msg, circuit, batch_factor) < 0)
@@ -644,6 +655,10 @@ osmux_batch_add(struct osmux_batch *batch, uint32_t batch_factor, struct msgb *m
 	}
 	batch->nmsgs++;
 
+	/* Update circuit state */
+	circuit->last_snd_sequence = rtph->sequence;
+	circuit->last_snd_timestamp = rtph->timestamp;
+	circuit->first_queued = true;
 	return 0;
 }