diff --git a/daemon/codec.c b/daemon/codec.c index e59c1c204..733788ff3 100644 --- a/daemon/codec.c +++ b/daemon/codec.c @@ -2298,7 +2298,7 @@ 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, (GDestroyNotify) __transcode_packet_free); + packet_sequencer_init(seq, (void (*)(seq_packet_t *)) __transcode_packet_free); g_hash_table_insert(ssrc_in->sequencers, mp->media_out, seq); ssrc_in->media_cache = mp->media_out; ssrc_in->sequencer_cache = seq; @@ -2307,12 +2307,12 @@ static int __handler_func_sequencer(struct media_packet *mp, struct transcode_pa if(stats_ext_seq) { seq->roc = stats_ext_seq>>16; seq->ext_seq = stats_ext_seq-1; - seq->seq = stats_ext_seq & 0xffff; - ilog(LOG_DEBUG, "transcode: restoring sequencer, roc: %d ext_seq: %u seq: %u", seq->roc, seq->ext_seq, seq->seq); + seq->a_seq = stats_ext_seq & 0xffff; + ilog(LOG_DEBUG, "transcode: restoring sequencer, roc: %d ext_seq: %u seq: %u", seq->roc, seq->ext_seq, seq->a_seq); } } - uint16_t seq_ori = (seq->seq < 0) ? 0 : seq->seq; + uint16_t seq_ori = (seq->a_seq == -1u) ? 0 : seq->a_seq; int seq_ret = packet_sequencer_insert(seq, &packet->p); if (seq_ret < 0) { // dupe diff --git a/daemon/dtmf.c b/daemon/dtmf.c index 84aa852a7..b8fb876c1 100644 --- a/daemon/dtmf.c +++ b/daemon/dtmf.c @@ -747,7 +747,7 @@ static const char *dtmf_inject_pcm(struct call_media *media, struct call_media * struct rtp_header rtp = { .m_pt = 0xff, .timestamp = 0, - .seq_num = htons(seq->seq), + .seq_num = htons(seq->a_seq), .ssrc = htonl(ssrc_in->h.ssrc), }; struct media_packet packet = { @@ -773,7 +773,7 @@ static const char *dtmf_inject_pcm(struct call_media *media, struct call_media * // insert pause tep.event = 0xff; tep.duration = htons(pause_samples); - rtp.seq_num = htons(seq->seq); + rtp.seq_num = htons(seq->a_seq); ch->dtmf_injector->handler_func(ch->dtmf_injector, &packet); diff --git a/daemon/t38.c b/daemon/t38.c index 412875ce3..ac3f01c9b 100644 --- a/daemon/t38.c +++ b/daemon/t38.c @@ -450,8 +450,8 @@ 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, (GDestroyNotify) __udptl_packet_free); - tg->sequencer.seq = 0; + packet_sequencer_init(&tg->sequencer, (void (*)(seq_packet_t *)) __udptl_packet_free); + tg->sequencer.a_seq = 0; // done - add references to media structs t38_media->t38_gateway = tg; diff --git a/lib/codeclib.c b/lib/codeclib.c index 199ca70b3..c7f06fd66 100644 --- a/lib/codeclib.c +++ b/lib/codeclib.c @@ -12,7 +12,6 @@ -#define PACKET_SEQ_DUPE_THRES 100 #define PACKET_TS_RESET_THRES 5000 // milliseconds @@ -473,101 +472,73 @@ void codeclib_init(int print) { -static int ptr_cmp(const void *a, const void *b, void *dummy) { - if (a < b) - return -1; - if (a > b) - return 1; - return 0; +void packet_sequencer_init(packet_sequencer_t *ps, void (*ffunc)(seq_packet_t *)) { + ps->free_func = ffunc; + ps->a_seq = -1u; + ps->a_nxt = -1u; } - -void __packet_sequencer_init(packet_sequencer_t *ps, GDestroyNotify ffunc) { - ps->packets = g_tree_new_full(ptr_cmp, NULL, NULL, ffunc); - ps->seq = -1; +static void sequencer_packets_clear(packet_sequencer_t *ps) { + for (unsigned int i = 0; i < G_N_ELEMENTS(ps->packets); i++) { + if (ps->packets[i]) + ps->free_func(ps->packets[i]); + ps->packets[i] = NULL; + } + ps->n_pks = 0; + ps->a_nxt = -1u; } void packet_sequencer_destroy(packet_sequencer_t *ps) { - if (ps->packets) - g_tree_destroy(ps->packets); - ps->packets = NULL; -} -struct tree_searcher { - int find_seq, - found_seq; -}; -static int packet_tree_search(const void *testseq_p, const void *ts_p) { - struct tree_searcher *ts = (void *) ts_p; - int testseq = GPOINTER_TO_INT(testseq_p); - // called as a binary search test function. we're looking for the lowest - // seq number that is higher than find_seq. if our test number is too low, - // we proceed with higher numbers. if it's too high, we proceed to the lower - // numbers, but remember the lowest we've seen along that path. - if (G_UNLIKELY(testseq == ts->find_seq)) { - // we've struck gold - ts->found_seq = testseq; - return 0; - } - if (testseq < ts->find_seq) - return 1; - // testseq > ts->find_seq - if (ts->found_seq == -1 || testseq < ts->found_seq) - ts->found_seq = testseq; - return -1; + sequencer_packets_clear(ps); } // caller must take care of locking -static void *__packet_sequencer_next_packet(packet_sequencer_t *ps, int num_wait) { +static void *__packet_sequencer_next_packet(packet_sequencer_t *ps, unsigned int num_wait) { // see if we have a packet with the correct seq nr in the queue - seq_packet_t *packet = g_tree_lookup(ps->packets, GINT_TO_POINTER(ps->seq)); + unsigned int ix; + seq_packet_t *packet = ps->packets[ps->a_idx]; if (G_LIKELY(packet != NULL)) { - cdbg("returning in-sequence packet (seq %i)", ps->seq); + cdbg("returning in-sequence packet (seq %i)", ps->a_seq); + ix = ps->a_idx; goto out; } // why not? do we have anything? (we should) - int nnodes = g_tree_nnodes(ps->packets); - if (G_UNLIKELY(nnodes == 0)) { + if (G_UNLIKELY(ps->n_pks == 0)) { cdbg("packet queue empty"); return NULL; } - if (G_LIKELY(nnodes < num_wait)) { - cdbg("only %i packets in queue - waiting for more", nnodes); + if (G_LIKELY(ps->n_pks < num_wait)) { + cdbg("only %i packets in queue - waiting for more", ps->n_pks); return NULL; // need to wait for more } // packet was probably lost. search for the next highest seq - struct tree_searcher ts = { .find_seq = ps->seq + 1, .found_seq = -1 }; - packet = g_tree_search(ps->packets, packet_tree_search, &ts); - if (packet) { - // bullseye + if (ps->a_nxt != -1u) { + ix = ps->a_nxt; + packet = ps->packets[ix]; cdbg("lost packet - returning packet with next seq %i", packet->seq); goto out; } - if (G_UNLIKELY(ts.found_seq == -1)) { - // didn't find anything. seq must have wrapped around. retry - // starting from zero - ts.find_seq = 0; - packet = g_tree_search(ps->packets, packet_tree_search, &ts); - if (packet) { - cdbg("lost packet - returning packet with next seq %i (after wrap)", packet->seq); + + for (unsigned int i = 0; i < G_N_ELEMENTS(ps->packets); i++) { + ix = (ps->a_idx + i) % G_N_ELEMENTS(ps->packets); + if (ps->packets[ix]) { + packet = ps->packets[ix]; + cdbg("lost packet - returning packet with next seq %i", packet->seq); goto out; } - if (G_UNLIKELY(ts.found_seq == -1)) - abort(); } - // pull out the packet we found - packet = g_tree_lookup(ps->packets, GINT_TO_POINTER(ts.found_seq)); - if (G_UNLIKELY(packet == NULL)) - abort(); + abort(); - cdbg("lost multiple packets - returning packet with next highest seq %i", packet->seq); - -out: - ; - uint16_t l = packet->seq - ps->seq; +out:; + uint16_t l = packet->seq - ps->a_seq; ps->lost_count += l; - g_tree_steal(ps->packets, GINT_TO_POINTER(packet->seq)); - ps->seq = (packet->seq + 1) & 0xffff; + ps->packets[ix] = NULL; + ps->n_pks--; + ps->a_seq = (packet->seq + 1) & 0xffff; + ps->a_idx = (ix + 1) % G_N_ELEMENTS(ps->packets); + if (ps->a_nxt == ix) + ps->a_nxt = -1u; unsigned int ext_seq = ps->roc << 16 | packet->seq; while (ext_seq < ps->ext_seq) { @@ -585,47 +556,63 @@ void *packet_sequencer_force_next_packet(packet_sequencer_t *ps) { return __packet_sequencer_next_packet(ps, 0); } -int packet_sequencer_next_ok(packet_sequencer_t *ps) { - if (g_tree_lookup(ps->packets, GINT_TO_POINTER(ps->seq))) - return 1; - return 0; +bool packet_sequencer_next_ok(packet_sequencer_t *ps) { + if (ps->packets[ps->a_idx]) + return true; + return false; } int packet_sequencer_insert(packet_sequencer_t *ps, seq_packet_t *p) { int ret = 0; // check seq for dupes - if (G_UNLIKELY(ps->seq == -1)) { + if (G_UNLIKELY(ps->a_seq == -1u)) { // first packet we see - ps->seq = p->seq; + ps->a_seq = p->seq; goto seq_ok; } - int diff = p->seq - ps->seq; - // early packet: p->seq = 200, ps->seq = 150, diff = 50 - if (G_LIKELY(diff >= 0 && diff < PACKET_SEQ_DUPE_THRES)) + int diff = p->seq - ps->a_seq; + // early packet: p->seq = 200, ps->a_seq = 150, diff = 50 + // or exact match: p->seq = 200, ps->a_seq = 200, diff = 0 + if (G_LIKELY(diff >= 0 && diff < G_N_ELEMENTS(ps->packets))) goto seq_ok; - // early packet with wrap-around: p->seq = 20, ps->seq = 65530, diff = -65510 - if (diff < (-0xffff + PACKET_SEQ_DUPE_THRES)) + // early packet with wrap-around: p->seq = 20, ps->a_seq = 65530, diff = -65510 + 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 > -PACKET_SEQ_DUPE_THRES) + if (diff < 0 && diff > -(signed) G_N_ELEMENTS(ps->packets)) return -1; // recent duplicate after wrap-around: p->seq = 65530, ps->seq = 30, diff = 65500 - if (diff > (0xffff - PACKET_SEQ_DUPE_THRES)) + if (diff > (0xffff - (signed) G_N_ELEMENTS(ps->packets))) return -1; // everything else we consider a seq reset - ilog(LOG_DEBUG, "Seq reset detected: expected seq %i, received seq %i", ps->seq, p->seq); - ps->seq = p->seq; + ilog(LOG_DEBUG, "Seq reset detected: expected seq %i, received seq %i", ps->a_seq, p->seq); + ps->a_seq = p->seq; ret = 1; + sequencer_packets_clear(ps); // seq ok - fall through - g_tree_clear(ps->packets); -seq_ok: - if (g_tree_lookup(ps->packets, GINT_TO_POINTER(p->seq))) +seq_ok:; + // slot of this packet + unsigned int idx = (ps->a_idx + p->seq - ps->a_seq) % G_N_ELEMENTS(ps->packets); + // packet already present? + if (ps->packets[idx]) return -1; - ret = g_tree_nnodes(ps->packets) == 0 ? ret : 2; // indicates an out-of-order packet - g_tree_insert(ps->packets, GINT_TO_POINTER(p->seq), p); + + if (ps->a_idx != idx) + ret = 2; // indicates an out-of-order packet + + // if not set + if (ps->a_nxt == -1u) + ps->a_nxt = idx; + // or if new idx is less than a_nxt + else if ((ps->a_nxt - ps->a_idx) % G_N_ELEMENTS(ps->packets) + > (idx - ps->a_idx) % G_N_ELEMENTS(ps->packets)) + ps->a_nxt = idx; + + ps->packets[idx] = p; + ps->n_pks++; return ret; } diff --git a/lib/codeclib.h b/lib/codeclib.h index c4f522df3..da4e4858d 100644 --- a/lib/codeclib.h +++ b/lib/codeclib.h @@ -376,9 +376,13 @@ struct seq_packet_s { int seq; }; struct packet_sequencer_s { - GTree *packets; + seq_packet_t *packets[128]; // should be 2^n + unsigned int a_idx; // start of queue, 0..127 + unsigned int a_seq; // seq of head of queue + unsigned int n_pks; // number of packets + unsigned int a_nxt; // index of next closest + void (*free_func)(seq_packet_t *); unsigned int lost_count; - int seq; // next expected unsigned int ext_seq; // last received int roc; // rollover counter XXX duplicate with SRTP encryption context }; @@ -437,11 +441,10 @@ int encoder_input_fifo(encoder_t *enc, AVFrame *frame, int (*callback)(encoder_t *, void *u1, void *u2), void *u1, void *u2); -void __packet_sequencer_init(packet_sequencer_t *ps, GDestroyNotify); -INLINE void packet_sequencer_init(packet_sequencer_t *ps, GDestroyNotify); +void packet_sequencer_init(packet_sequencer_t *ps, void (*)(seq_packet_t *)); void packet_sequencer_destroy(packet_sequencer_t *ps); void *packet_sequencer_next_packet(packet_sequencer_t *ps); -int packet_sequencer_next_ok(packet_sequencer_t *ps); +bool packet_sequencer_next_ok(packet_sequencer_t *ps); void *packet_sequencer_force_next_packet(packet_sequencer_t *ps); int packet_sequencer_insert(packet_sequencer_t *ps, seq_packet_t *); @@ -524,11 +527,6 @@ AVPacket *codec_cc_input_data(codec_cc_t *c, const str *data, unsigned long ts, // `ps` must be zero allocated -INLINE void packet_sequencer_init(packet_sequencer_t *ps, GDestroyNotify n) { - if (ps->packets) - return; - __packet_sequencer_init(ps, n); -} INLINE int format_eq(const format_t *a, const format_t *b) { if (G_UNLIKELY(a->clockrate != b->clockrate)) return 0; diff --git a/recording-daemon/packet.c b/recording-daemon/packet.c index a26164663..b7045026d 100644 --- a/recording-daemon/packet.c +++ b/recording-daemon/packet.c @@ -22,8 +22,7 @@ #include "mix.h" -static void packet_free(void *p) { - packet_t *packet = p; +static void packet_free(packet_t *packet) { if (!packet) return; free(packet->buffer); @@ -65,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, packet_free); + packet_sequencer_init(&ret->sequencer, (void (*)(seq_packet_t *)) packet_free); g_hash_table_insert(mf->ssrc_hash, GUINT_TO_POINTER(ssrc), ret); @@ -154,7 +153,7 @@ static void ssrc_run(ssrc_t *ssrc) { packet_decode(ssrc, packet); packet_free(packet); - dbg("packets left in queue: %i", g_tree_nnodes(ssrc->sequencer.packets)); + dbg("packets left in queue: %i", ssrc->sequencer.n_pks); } pthread_mutex_unlock(&ssrc->lock); @@ -202,7 +201,7 @@ void packet_process(stream_t *stream, unsigned char *buf, unsigned len) { if (!ssrc) // stream shutdown goto out; if (packet_sequencer_insert(&ssrc->sequencer, &packet->p) < 0) { - dbg("skipping dupe packet (new seq %i prev seq %i)", packet->p.seq, ssrc->sequencer.seq); + dbg("skipping dupe packet (new seq %i prev seq %i)", packet->p.seq, ssrc->sequencer.a_seq); goto skip; } diff --git a/t/test-sequencer.c b/t/test-sequencer.c index 119c5dce3..ca9d2eea3 100644 --- a/t/test-sequencer.c +++ b/t/test-sequencer.c @@ -3,7 +3,7 @@ static unsigned int num_freed; -static void ffunc(void *a) { +static void ffunc(seq_packet_t *a) { num_freed++; } @@ -73,7 +73,7 @@ int main(void) { pks[1].seq = 104; i = packet_sequencer_insert(&ps, &pks[1]); - assert(i == 0); + assert(i == 2); assert(num_freed == 0); i = packet_sequencer_next_ok(&ps); @@ -83,7 +83,7 @@ int main(void) { assert(p == NULL); i = packet_sequencer_insert(&ps, &pks[0]); - assert(i == 2); + assert(i == 0); assert(num_freed == 0); i = packet_sequencer_next_ok(&ps); @@ -122,7 +122,7 @@ int main(void) { pks[1].seq = 107; i = packet_sequencer_insert(&ps, &pks[1]); - assert(i == 0); + assert(i == 2); assert(num_freed == 0); i = packet_sequencer_next_ok(&ps); @@ -156,7 +156,7 @@ int main(void) { pks[1].seq = 111; i = packet_sequencer_insert(&ps, &pks[1]); - assert(i == 0); + assert(i == 2); assert(num_freed == 0); i = packet_sequencer_next_ok(&ps);