MT#65420 marker handling in sequencer

Change-Id: Ib1d66943cf1b0de9b9a6ac83d36b841b0c11245a

daemon/codec.c

@@ -2302,7 +2302,8 @@ static int __handler_func_sequencer(struct media_packet *mp, struct transcode_pa
 		seq = g_hash_table_lookup(ssrc_in->sequencers, mp->media_out);
 	if (!seq) {
 		seq = g_new0(__typeof(*seq), 1);
-		packet_sequencer_init(seq, (void (*)(seq_packet_t *)) __transcode_packet_free);
+		packet_sequencer_init(seq, (void (*)(seq_packet_t *)) __transcode_packet_free,
+				h->source_pt.codec_def->media_type == MT_VIDEO ? true : false);
 		g_hash_table_insert(ssrc_in->sequencers, mp->media_out, seq);
 		ssrc_in->media_cache = mp->media_out;
 		ssrc_in->sequencer_cache = seq;

daemon/t38.c

@@ -450,7 +450,7 @@ int t38_gateway_pair(struct call_media *t38_media, struct call_media *pcm_media,
 	my_span_set_log(ls, spandsp_logging_func);
 	span_log_set_level(ls, span_log_level_map(get_log_level(spandsp)));
 
-	packet_sequencer_init(&tg->sequencer, (void (*)(seq_packet_t *)) __udptl_packet_free);
+	packet_sequencer_init(&tg->sequencer, (void (*)(seq_packet_t *)) __udptl_packet_free, false);
 	tg->sequencer.a_seq = 0;
 
 	// done - add references to media structs

lib/codeclib.c

@@ -515,10 +515,11 @@ void codeclib_init(int print) {
 
 
 
-void packet_sequencer_init(packet_sequencer_t *ps, void (*ffunc)(seq_packet_t *)) {
+void packet_sequencer_init(packet_sequencer_t *ps, void (*ffunc)(seq_packet_t *), bool marker_delay) {
 	ps->free_func = ffunc;
 	ps->a_seq = -1u;
 	ps->a_nxt = -1u;
+	ps->marker_delay = marker_delay;
 }
 static void sequencer_packets_clear(packet_sequencer_t *ps) {
 	for (unsigned int i = 0; i < G_N_ELEMENTS(ps->packets); i++) {
@@ -534,6 +535,11 @@ void packet_sequencer_destroy(packet_sequencer_t *ps) {
 }
 // caller must take care of locking
 static void *__packet_sequencer_next_packet(packet_sequencer_t *ps, unsigned int num_wait) {
+	if (G_UNLIKELY(ps->marker_delay)) {
+		cdbg("waiting for marker");
+		return NULL;
+	}
+
 	// see if we have a packet with the correct seq nr in the queue
 	unsigned int ix;
 	seq_packet_t *packet = ps->packets[ps->a_idx];
@@ -600,6 +606,8 @@ void *packet_sequencer_force_next_packet(packet_sequencer_t *ps) {
 }
 
 bool packet_sequencer_next_ok(packet_sequencer_t *ps) {
+	if (ps->marker_delay)
+		return false;
 	if (ps->packets[ps->a_idx])
 		return true;
 	return false;
@@ -624,8 +632,15 @@ int packet_sequencer_insert(packet_sequencer_t *ps, seq_packet_t *p) {
 	if (diff < (-0xffff + (signed) G_N_ELEMENTS(ps->packets)))
 		goto seq_ok;
 	// recent duplicate: p->seq = 1000, ps->seq = 1080, diff = -80
-	if (diff < 0 && diff > -(signed) G_N_ELEMENTS(ps->packets))
-		return -1;
+	if (diff < 0 && diff > -(signed) G_N_ELEMENTS(ps->packets)) {
+		if (!ps->marker_delay)
+			return -1;
+		// special case: we're waiting for the marker bit and there was
+		// an initial out-of-order packet
+		ps->a_seq += diff;
+		ps->a_idx = (ps->a_idx + diff) % G_N_ELEMENTS(ps->packets);
+		goto seq_ok;
+	}
 	// recent duplicate after wrap-around: p->seq = 65530, ps->seq = 30, diff = 65500
 	if (diff > (0xffff - (signed) G_N_ELEMENTS(ps->packets)))
 		return -1;
@@ -636,7 +651,10 @@ int packet_sequencer_insert(packet_sequencer_t *ps, seq_packet_t *p) {
 	ret = 1;
 	sequencer_packets_clear(ps);
 	// seq ok - fall through
-seq_ok:;
+seq_ok:
+	if (p->marker)
+		ps->marker_delay = false;
+
 	// slot of this packet
 	unsigned int idx = (ps->a_idx + p->seq - ps->a_seq) % G_N_ELEMENTS(ps->packets);
 	// packet already present?

lib/codeclib.h

@@ -434,6 +434,7 @@ struct packet_sequencer_s {
 	unsigned int lost_count;
 	unsigned int ext_seq; // last received
 	int roc; // rollover counter XXX duplicate with SRTP encryption context
+	bool marker_delay; // initially set for video-type RTP
 };
 
 
@@ -494,7 +495,7 @@ INLINE int encoder_input_data(encoder_t *enc, AVFrame *frame,
 }
 
 
-void packet_sequencer_init(packet_sequencer_t *ps, void (*)(seq_packet_t *));
+void packet_sequencer_init(packet_sequencer_t *ps, void (*)(seq_packet_t *), bool marker_delay);
 void packet_sequencer_destroy(packet_sequencer_t *ps);
 void *packet_sequencer_next_packet(packet_sequencer_t *ps);
 bool packet_sequencer_next_ok(packet_sequencer_t *ps);

recording-daemon/packet.c

@@ -64,7 +64,7 @@ static ssrc_t *ssrc_get(stream_t *stream, unsigned long ssrc) {
 	ret->metafile = mf;
 	ret->stream = stream;
 	ret->ssrc = ssrc;
-	packet_sequencer_init(&ret->sequencer, (void (*)(seq_packet_t *)) packet_free);
+	packet_sequencer_init(&ret->sequencer, (void (*)(seq_packet_t *)) packet_free, false);
 
 	g_hash_table_insert(mf->ssrc_hash, GUINT_TO_POINTER(ssrc), ret);
 

t/test-sequencer.c

@@ -7,9 +7,9 @@ static void ffunc(seq_packet_t *a) {
 	num_freed++;
 }
 
-int main(void) {
+void test1(void) {
 	packet_sequencer_t ps = {0};
-	packet_sequencer_init(&ps, ffunc);
+	packet_sequencer_init(&ps, ffunc, false);
 
 	void *p;
 	int i;
@@ -25,7 +25,7 @@ int main(void) {
 
 	assert(num_freed == 0);
 
-	seq_packet_t pks[256];
+	seq_packet_t pks[10] = {0};
 
 	pks[0].seq = 100;
 	i = packet_sequencer_insert(&ps, &pks[0]);
@@ -284,6 +284,227 @@ int main(void) {
 
 
 	packet_sequencer_destroy(&ps);
+}
+
+void test2(void) {
+	packet_sequencer_t ps = {0};
+	packet_sequencer_init(&ps, ffunc, false);
+
+	void *p;
+	int i;
+
+	seq_packet_t pks[3] = {0};
+
+	pks[0].seq = 100;
+	i = packet_sequencer_insert(&ps, &pks[0]);
+	assert(i == 0);
+	assert(num_freed == 0);
+
+	i = packet_sequencer_next_ok(&ps);
+	assert(i);
+
+	p = packet_sequencer_next_packet(&ps);
+	assert(p == &pks[0]);
+	assert(num_freed == 0);
+
+	pks[1].seq = 99;
+	i = packet_sequencer_insert(&ps, &pks[1]);
+	assert(i == -1);
+	assert(num_freed == 0);
+
+	i = packet_sequencer_next_ok(&ps);
+	assert(!i);
+
+	p = packet_sequencer_next_packet(&ps);
+	assert(!p);
+	assert(num_freed == 0);
+
+	pks[2].seq = 101;
+	i = packet_sequencer_insert(&ps, &pks[2]);
+	assert(i == 0);
+	assert(num_freed == 0);
+
+	i = packet_sequencer_next_ok(&ps);
+	assert(i);
+
+	p = packet_sequencer_next_packet(&ps);
+	assert(p == &pks[2]);
+	assert(num_freed == 0);
+
+	packet_sequencer_destroy(&ps);
+}
+
+void test3(void) {
+	packet_sequencer_t ps = {0};
+	packet_sequencer_init(&ps, ffunc, false);
+
+	void *p;
+	int i;
+
+	seq_packet_t pks[3] = {0};
+
+	pks[0].seq = 100;
+	i = packet_sequencer_insert(&ps, &pks[0]);
+	assert(i == 0);
+	assert(num_freed == 0);
+
+	i = packet_sequencer_next_ok(&ps);
+	assert(i);
+
+	pks[1].seq = 99;
+	i = packet_sequencer_insert(&ps, &pks[1]);
+	assert(i == -1);
+	assert(num_freed == 0);
+
+	p = packet_sequencer_next_packet(&ps);
+	assert(p == &pks[0]);
+	assert(num_freed == 0);
+
+	i = packet_sequencer_next_ok(&ps);
+	assert(!i);
+
+	p = packet_sequencer_next_packet(&ps);
+	assert(!p);
+	assert(num_freed == 0);
+
+	pks[2].seq = 101;
+	i = packet_sequencer_insert(&ps, &pks[2]);
+	assert(i == 0);
+	assert(num_freed == 0);
+
+	i = packet_sequencer_next_ok(&ps);
+	assert(i);
+
+	p = packet_sequencer_next_packet(&ps);
+	assert(p == &pks[2]);
+	assert(num_freed == 0);
+
+	packet_sequencer_destroy(&ps);
+}
+
+void test4(void) {
+	packet_sequencer_t ps = {0};
+	packet_sequencer_init(&ps, ffunc, true);
+
+	void *p;
+	int i;
+
+	seq_packet_t pks[3] = {0};
+
+	pks[0].seq = 100;
+	i = packet_sequencer_insert(&ps, &pks[0]);
+	assert(i == 0);
+	assert(num_freed == 0);
+
+	i = packet_sequencer_next_ok(&ps);
+	assert(!i);
+
+	p = packet_sequencer_next_packet(&ps);
+	assert(!p);
+	assert(num_freed == 0);
+
+	pks[1].seq = 99;
+	i = packet_sequencer_insert(&ps, &pks[1]);
+	assert(i == 0);
+	assert(num_freed == 0);
+
+	i = packet_sequencer_next_ok(&ps);
+	assert(!i);
+
+	p = packet_sequencer_next_packet(&ps);
+	assert(!p);
+	assert(num_freed == 0);
+
+	pks[2].seq = 101;
+	pks[2].marker = true;
+	i = packet_sequencer_insert(&ps, &pks[2]);
+	assert(i == 2);
+	assert(num_freed == 0);
+
+	i = packet_sequencer_next_ok(&ps);
+	assert(i);
+
+	p = packet_sequencer_next_packet(&ps);
+	assert(p == &pks[1]);
+	assert(num_freed == 0);
+
+	i = packet_sequencer_next_ok(&ps);
+	assert(i);
+
+	p = packet_sequencer_next_packet(&ps);
+	assert(p == &pks[0]);
+	assert(num_freed == 0);
+
+	i = packet_sequencer_next_ok(&ps);
+	assert(i);
+
+	p = packet_sequencer_next_packet(&ps);
+	assert(p == &pks[2]);
+	assert(num_freed == 0);
+
+	packet_sequencer_destroy(&ps);
+}
+
+void test5(void) {
+	packet_sequencer_t ps = {0};
+	packet_sequencer_init(&ps, ffunc, true);
+
+	void *p;
+	int i;
+
+	seq_packet_t pks[3] = {0};
+
+	pks[0].seq = 100;
+	i = packet_sequencer_insert(&ps, &pks[0]);
+	assert(i == 0);
+	assert(num_freed == 0);
+
+	i = packet_sequencer_next_ok(&ps);
+	assert(!i);
+
+	pks[1].seq = 99;
+	i = packet_sequencer_insert(&ps, &pks[1]);
+	assert(i == 0);
+	assert(num_freed == 0);
+
+	p = packet_sequencer_next_packet(&ps);
+	assert(!p);
+	assert(num_freed == 0);
+
+	i = packet_sequencer_next_ok(&ps);
+	assert(!i);
+
+	p = packet_sequencer_next_packet(&ps);
+	assert(!p);
+	assert(num_freed == 0);
+
+	pks[2].seq = 101;
+	pks[2].marker = true;
+	i = packet_sequencer_insert(&ps, &pks[2]);
+	assert(i == 2);
+	assert(num_freed == 0);
+
+	i = packet_sequencer_next_ok(&ps);
+	assert(i);
+
+	p = packet_sequencer_next_packet(&ps);
+	assert(p == &pks[1]);
+	assert(num_freed == 0);
+
+	p = packet_sequencer_next_packet(&ps);
+	assert(p == &pks[0]);
+	assert(num_freed == 0);
+
+	p = packet_sequencer_next_packet(&ps);
+	assert(p == &pks[2]);
+	assert(num_freed == 0);
+
+	packet_sequencer_destroy(&ps);
+}
+
+
+int main(void) {
+	test1();
 
 	return 0;
 }