opusdec: Latency improvements.

Currently the opus decoder works on the ogg page level, i.e. during
each scheduler iteration it tries to extract one ogg page from the
stream and decodes all opus packages therein. Since ogg pages can be
quite large (20K), decoding can take a significant amount of time,
especially on slow machines. If the decoding takes longer than the
buffer time of the configured writer, audible buffer underruns result.

This changes the opus decoder to decode only a single packet during
each iteration. With this patch applied, latency improves greatly,
and underruns occur on neither an old 150MHz AMD K6 nor on the
raspberry pi.

diff --git a/opusdec_filter.c b/opusdec_filter.c
index 90e65bc369571fbf9e8cde134efc07a17b76e124..c3b405cc338df6666342e863aa3a0ac053ea8500 100644 (file)
--- a/opusdec_filter.c
+++ b/opusdec_filter.c
@@ -72,6 +72,7 @@ struct opusdec_context {
        int channels;
        int preskip;
        bool have_opus_stream;
+       bool have_more;
        ogg_int32_t opus_serialno;
 };
 
@@ -204,60 +205,84 @@ static int decode_packet(struct opusdec_context *ctx, ogg_packet *op,
        return 1;
 }
 
+#define OPUSDEC_MAX_OUTPUT_SIZE (1024 * 1024)
+
 static int opusdec_post_select(__a_unused struct sched *s, struct task *t)
 {
        struct filter_node *fn = container_of(t, struct filter_node, task);
        struct opusdec_context *ctx = fn->private_data;
        struct btr_node *btrn = fn->btrn;
        int ret;
-       char *btr_buf, *data;
-       size_t nbytes;
        ogg_packet op;
 
        ret = btr_node_status(btrn, fn->min_iqs, BTR_NT_INTERNAL);
-       if (ret <= 0)
-               goto out;
-       btr_merge(btrn, fn->min_iqs);
-       nbytes = btr_next_buffer(btrn, &btr_buf);
-       nbytes = PARA_MIN(nbytes, (size_t)32768);
-       ret = 0;
-       if (nbytes == 0)
-               goto out;
-       data = ogg_sync_buffer(&ctx->oy, nbytes);
-       memcpy(data, btr_buf, nbytes);
-       btr_consume(btrn, nbytes);
-       ogg_sync_wrote(&ctx->oy, nbytes);
-       for (;;) { /* loop over all ogg pages we got */
-               ret = 0;
-               if (ogg_sync_pageout(&ctx->oy, &ctx->ogg_page) != 1)
+       if (ret < 0) {
+               if (ret != -E_BTR_EOF) /* fatal error */
+                       goto out;
+               if (!ctx->have_more) /* EOF */
                        goto out;
-               if (!ctx->stream_init) {
-                       ogg_stream_init(&ctx->os, ogg_page_serialno(&ctx->ogg_page));
+       } else if (ret == 0 && !ctx->have_more) /* nothing to do */
+               goto out;
+       if (btr_get_output_queue_size(btrn) > OPUSDEC_MAX_OUTPUT_SIZE)
+               return 0;
+       for (;;) {
+               int serial;
+               if (ctx->stream_init) {
+                       ret = ogg_stream_packetout(&ctx->os, &op);
+                       if (ret == 1)
+                               break;
+               }
+               while (ogg_sync_pageout(&ctx->oy, &ctx->ogg_page) != 1) {
+                       char *btr_buf, *data;
+                       size_t nbytes = btr_next_buffer(btrn, &btr_buf);
+                       nbytes = PARA_MIN(nbytes, (size_t)24 * 1024);
+                       //PARA_CRIT_LOG("nbytes: %d\n", nbytes);
+                       ctx->have_more = false;
+                       if (nbytes == 0)
+                               return 0;
+                       data = ogg_sync_buffer(&ctx->oy, nbytes);
+                       memcpy(data, btr_buf, nbytes);
+                       btr_consume(btrn, nbytes);
+                       ogg_sync_wrote(&ctx->oy, nbytes);
+               }
+               ctx->have_more = true;
+               serial = ogg_page_serialno(&ctx->ogg_page);
+               if (ctx->stream_init) {
+                       if (serial != ctx->os.serialno)
+                               ogg_stream_reset_serialno(&ctx->os, serial);
+               } else {
+                       ogg_stream_init(&ctx->os, serial);
                        ctx->stream_init = true;
                }
-               if (ogg_page_serialno(&ctx->ogg_page) != ctx->os.serialno)
-                       ogg_stream_reset_serialno(&ctx->os,
-                               ogg_page_serialno(&ctx->ogg_page));
                /* Add page to the bitstream */
                ogg_stream_pagein(&ctx->os, &ctx->ogg_page);
-               for (;;) { /* loop over all opus packets */
-                       ret = ogg_stream_packetout(&ctx->os, &op);
-                       if (ret != 1)
-                               break;
-                       ret = decode_packet(ctx, &op, btrn);
-                       if (ret < 0)
-                               goto out;
-                       ctx->packet_count++;
-                       if (ctx->eos)
-                               ctx->have_opus_stream = false;
-               }
        }
+       ret = decode_packet(ctx, &op, btrn);
+       if (ret < 0)
+               goto out;
+       ctx->packet_count++;
+       if (ctx->eos)
+               ctx->have_opus_stream = false;
 out:
        if (ret < 0)
                btr_remove_node(&fn->btrn);
        return ret;
 }
 
+static void opusdec_pre_select(struct sched *s, struct task *t)
+{
+       struct filter_node *fn = container_of(t, struct filter_node, task);
+       struct opusdec_context *ctx = fn->private_data;
+       int ret = btr_node_status(fn->btrn, fn->min_iqs, BTR_NT_INTERNAL);
+
+       if (ret != 0)
+               return sched_min_delay(s);
+       if (ctx->have_more)
+               return;
+       if (btr_get_output_queue_size(fn->btrn) <= OPUSDEC_MAX_OUTPUT_SIZE)
+               return sched_min_delay(s);
+}
+
 /**
  * The init function of the opusdec filter.
  *
@@ -269,7 +294,7 @@ void opusdec_filter_init(struct filter *f)
 {
        f->open = opusdec_open;
        f->close = opusdec_close;
-       f->pre_select = generic_filter_pre_select;
+       f->pre_select = opusdec_pre_select;
        f->post_select = opusdec_post_select;
        f->execute = opusdec_execute;
 }