sink-input: Add history memblockq

A new memblockq is added to the sink input code to keep some history of the
input data. The queue is kept in sync with the render memblockq. The old input
data will be used to prepare the resampler after a rewind.

pa_resampler_request() and pa_resampler_result() have been changed to round
as good as possible to avoid loosing or duplicating samples during rewinds.

Part-of: <https://gitlab.freedesktop.org/pulseaudio/pulseaudio/-/merge_requests/120>

4 files changed, 134 insertions, 34 deletions

diff --git a/src/pulsecore/resampler.c b/src/pulsecore/resampler.c
index 58463f16b..9c39a0241 100644
--- a/src/pulsecore/resampler.c
+++ b/src/pulsecore/resampler.c
@@ -22,6 +22,7 @@
 #endif
 
 #include <string.h>
+#include <math.h>
 
 #include <pulse/xmalloc.h>
 #include <pulsecore/log.h>
@@ -500,34 +501,73 @@ void pa_resampler_set_output_rate(pa_resampler *r, uint32_t rate) {
         pa_lfe_filter_update_rate(r->lfe_filter, rate);
 }
 
+/* pa_resampler_request() and pa_resampler_result() should be as exact as
+ * possible to ensure that no samples are lost or duplicated during rewinds.
+ * Ignore the leftover buffer, the value appears to be wrong for ffmpeg
+ * and 0 in all other cases. If the resampler is NULL it means that no
+ * resampling is necessary and the input length equals the output length.
+ * FIXME: These functions are not exact for the soxr resamplers because
+ * soxr uses a different algorithm. */
 size_t pa_resampler_request(pa_resampler *r, size_t out_length) {
-    pa_assert(r);
+    size_t in_length;
+
+    if (!r || out_length == 0)
+        return out_length;
+
+    /* Convert to output frames */
+    out_length = out_length / r->o_fz;
+
+    /* Convert to input frames. The equation matches exactly the
+     * behavior of the used resamplers and will calculate the
+     * minimum number of input frames that are needed to produce
+     * the given number of output frames. */
+    in_length = (out_length - 1) * r->i_ss.rate / r->o_ss.rate + 1;
 
-    /* Let's round up here to make it more likely that the caller will get at
-     * least out_length amount of data from pa_resampler_run().
-     *
-     * We don't take the leftover into account here. If we did, then it might
-     * be in theory possible that this function would return 0 and
-     * pa_resampler_run() would also return 0. That could lead to infinite
-     * loops. When the leftover is ignored here, such loops would eventually
-     * terminate, because the leftover would grow each round, finally
-     * surpassing the minimum input threshold of the resampler. */
-    return ((((uint64_t) ((out_length + r->o_fz-1) / r->o_fz) * r->i_ss.rate) + r->o_ss.rate-1) / r->o_ss.rate) * r->i_fz;
+    /* Convert to input length */
+    return in_length * r->i_fz;
 }
 
 size_t pa_resampler_result(pa_resampler *r, size_t in_length) {
-    size_t frames;
+    size_t out_length;
 
-    pa_assert(r);
+    if (!r)
+        return in_length;
 
-    /* Let's round up here to ensure that the caller will always allocate big
-     * enough output buffer. */
+    /* Convert to intput frames */
+    in_length = in_length / r->i_fz;
 
-    frames = (in_length + r->i_fz - 1) / r->i_fz;
-    if (*r->have_leftover)
-        frames += r->leftover_buf->length / r->w_fz;
+     /* soxr processes samples in blocks, depending on the ratio.
+      * Therefore samples  that do not fit into a block must be
+      * ignored. */
+    if (r->method == PA_RESAMPLER_SOXR_MQ || r->method == PA_RESAMPLER_SOXR_HQ || r->method == PA_RESAMPLER_SOXR_VHQ) {
+        double ratio;
+        size_t block_size;
+        int k;
+
+        ratio = (double)r->i_ss.rate / (double)r->o_ss.rate;
+
+        for (k = 0; k < 7; k++) {
+            if (ratio < pow(2, k + 1))
+                break;
+        }
+        block_size = pow(2, k);
+        in_length = in_length - in_length % block_size;
+    }
+
+    /* Convert to output frames. This matches exactly the algorithm
+     * used by the resamplers except for the soxr resamplers. */
+
+     out_length = in_length * r->o_ss.rate / r->i_ss.rate;
+     if ((double)in_length * (double)r->o_ss.rate / (double)r->i_ss.rate - out_length > 0)
+         out_length++;
+     /* The libsamplerate resamplers return one sample more if the result is integral and the ratio is not integral. */
+     else if (r->method >= PA_RESAMPLER_SRC_SINC_BEST_QUALITY && r->method <= PA_RESAMPLER_SRC_SINC_FASTEST && r->i_ss.rate > r->o_ss.rate && r->i_ss.rate % r->o_ss.rate > 0 && (double)in_length * (double)r->o_ss.rate / (double)r->i_ss.rate - out_length <= 0)
+         out_length++;
+     else if (r->method == PA_RESAMPLER_SRC_ZERO_ORDER_HOLD && r->i_ss.rate > r->o_ss.rate && (double)in_length * (double)r->o_ss.rate / (double)r->i_ss.rate - out_length <= 0)
+         out_length++;
 
-    return (((uint64_t) frames * r->o_ss.rate + r->i_ss.rate - 1) / r->i_ss.rate) * r->o_fz;
+    /* Convert to output length */
+    return out_length * r->o_fz;
 }
 
 size_t pa_resampler_max_block_size(pa_resampler *r) {
diff --git a/src/pulsecore/resampler.h b/src/pulsecore/resampler.h
index 5a264b342..783f69999 100644
--- a/src/pulsecore/resampler.h
+++ b/src/pulsecore/resampler.h
@@ -73,6 +73,8 @@ typedef enum pa_resample_flags {
     PA_RESAMPLER_CONSUME_LFE   = 0x0040U,
 } pa_resample_flags_t;
 
+#define PA_RESAMPLER_MAX_HISTORY 64
+
 struct pa_resampler {
     pa_resample_method_t method;
     pa_resample_flags_t flags;
diff --git a/src/pulsecore/sink-input.c b/src/pulsecore/sink-input.c
index 5e7657c45..31dae6a78 100644
--- a/src/pulsecore/sink-input.c
+++ b/src/pulsecore/sink-input.c
@@ -301,6 +301,7 @@ int pa_sink_input_new(
     int r;
     char *pt;
     char *memblockq_name;
+    pa_memchunk silence;
 
     pa_assert(_i);
     pa_assert(core);
@@ -582,6 +583,21 @@ int pa_sink_input_new(
             &i->sink->silence);
     pa_xfree(memblockq_name);
 
+    memblockq_name = pa_sprintf_malloc("sink input history memblockq [%u]", i->index);
+    pa_sink_input_get_silence(i, &silence);
+    i->thread_info.history_memblockq = pa_memblockq_new(
+            memblockq_name,
+            0,
+            MEMBLOCKQ_MAXLENGTH,
+            0,
+            &i->sample_spec,
+            0,
+            1,
+            0,
+            &silence);
+    pa_xfree(memblockq_name);
+    pa_memblock_unref(silence.memblock);
+
     pt = pa_proplist_to_string_sep(i->proplist, "\n    ");
     pa_log_info("Created input %u \"%s\" on %s with sample spec %s and channel map %s\n    %s",
                 i->index,
@@ -765,6 +781,9 @@ static void sink_input_free(pa_object *o) {
     if (i->thread_info.render_memblockq)
         pa_memblockq_free(i->thread_info.render_memblockq);
 
+    if (i->thread_info.history_memblockq)
+        pa_memblockq_free(i->thread_info.history_memblockq);
+
     if (i->thread_info.resampler)
         pa_resampler_free(i->thread_info.resampler);
 
@@ -934,6 +953,7 @@ void pa_sink_input_peek(pa_sink_input *i, size_t slength /* in sink bytes */, pa
              * data, so let's just hand out silence */
 
             pa_memblockq_seek(i->thread_info.render_memblockq, (int64_t) slength, PA_SEEK_RELATIVE, true);
+            pa_memblockq_seek(i->thread_info.history_memblockq, (int64_t) ilength_full, PA_SEEK_RELATIVE, true);
             i->thread_info.playing_for = 0;
             if (i->thread_info.underrun_for != (uint64_t) -1) {
                 i->thread_info.underrun_for += ilength_full;
@@ -981,6 +1001,9 @@ void pa_sink_input_peek(pa_sink_input *i, size_t slength /* in sink bytes */, pa
                     pa_volume_memchunk(&wchunk, &i->thread_info.sample_spec, &i->thread_info.soft_volume);
             }
 
+            /* Push chunk into history queue to retain some resampler input history. */
+            pa_memblockq_push(i->thread_info.history_memblockq, &wchunk);
+
             if (!i->thread_info.resampler) {
 
                 if (nvfs) {
@@ -1045,6 +1068,7 @@ void pa_sink_input_peek(pa_sink_input *i, size_t slength /* in sink bytes */, pa
 
 /* Called from thread context */
 void pa_sink_input_drop(pa_sink_input *i, size_t nbytes /* in sink sample spec */) {
+    int64_t rbq, hbq;
 
     pa_sink_input_assert_ref(i);
     pa_sink_input_assert_io_context(i);
@@ -1057,6 +1081,22 @@ void pa_sink_input_drop(pa_sink_input *i, size_t nbytes /* in sink sample spec *
 #endif
 
     pa_memblockq_drop(i->thread_info.render_memblockq, nbytes);
+
+    /* Keep memblockq's in sync. Using pa_resampler_request()
+     * on nbytes will not work here because of rounding. */
+    rbq = pa_memblockq_get_write_index(i->thread_info.render_memblockq);
+    rbq -= pa_memblockq_get_read_index(i->thread_info.render_memblockq);
+    hbq = pa_memblockq_get_write_index(i->thread_info.history_memblockq);
+    hbq -= pa_memblockq_get_read_index(i->thread_info.history_memblockq);
+    if (rbq >= 0)
+        rbq = pa_resampler_request(i->thread_info.resampler, rbq);
+    else
+        rbq = - (int64_t) pa_resampler_request(i->thread_info.resampler, - rbq);
+
+    if (hbq > rbq)
+        pa_memblockq_drop(i->thread_info.history_memblockq, hbq - rbq);
+    else if (rbq > hbq)
+        pa_memblockq_rewind(i->thread_info.history_memblockq, rbq - hbq);
 }
 
 /* Called from thread context */
@@ -1070,6 +1110,7 @@ bool pa_sink_input_process_underrun(pa_sink_input *i) {
     if (i->process_underrun && i->process_underrun(i)) {
         /* All valid data has been played back, so we can empty this queue. */
         pa_memblockq_silence(i->thread_info.render_memblockq);
+        pa_memblockq_silence(i->thread_info.history_memblockq);
         return true;
     }
     return false;
@@ -1079,6 +1120,7 @@ bool pa_sink_input_process_underrun(pa_sink_input *i) {
 void pa_sink_input_process_rewind(pa_sink_input *i, size_t nbytes /* in sink sample spec */) {
     size_t lbq;
     bool called = false;
+    size_t sink_input_nbytes;
 
     pa_sink_input_assert_ref(i);
     pa_sink_input_assert_io_context(i);
@@ -1090,10 +1132,12 @@ void pa_sink_input_process_rewind(pa_sink_input *i, size_t nbytes /* in sink sam
 #endif
 
     lbq = pa_memblockq_get_length(i->thread_info.render_memblockq);
+    sink_input_nbytes = pa_resampler_request(i->thread_info.resampler, nbytes);
 
     if (nbytes > 0 && !i->thread_info.dont_rewind_render) {
         pa_log_debug("Have to rewind %lu bytes on render memblockq.", (unsigned long) nbytes);
         pa_memblockq_rewind(i->thread_info.render_memblockq, nbytes);
+        pa_memblockq_rewind(i->thread_info.history_memblockq, sink_input_nbytes);
     }
 
     if (i->thread_info.rewrite_nbytes == (size_t) -1) {
@@ -1102,6 +1146,7 @@ void pa_sink_input_process_rewind(pa_sink_input *i, size_t nbytes /* in sink sam
          * data from implementor the next time peek() is called */
 
         pa_memblockq_flush_write(i->thread_info.render_memblockq, true);
+        pa_memblockq_flush_write(i->thread_info.history_memblockq, true);
 
     } else if (i->thread_info.rewrite_nbytes > 0) {
         size_t max_rewrite, amount;
@@ -1112,8 +1157,7 @@ void pa_sink_input_process_rewind(pa_sink_input *i, size_t nbytes /* in sink sam
             max_rewrite += lbq;
 
         /* Transform into local domain */
-        if (i->thread_info.resampler)
-            max_rewrite = pa_resampler_request(i->thread_info.resampler, max_rewrite);
+        max_rewrite = pa_resampler_request(i->thread_info.resampler, max_rewrite);
 
         /* Calculate how much of the rewinded data should actually be rewritten */
         amount = PA_MIN(i->thread_info.rewrite_nbytes, max_rewrite);
@@ -1126,20 +1170,25 @@ void pa_sink_input_process_rewind(pa_sink_input *i, size_t nbytes /* in sink sam
                 i->process_rewind(i, amount);
             called = true;
 
+            if (amount > 0)
+                /* Update the history write pointer */
+                pa_memblockq_seek(i->thread_info.history_memblockq, - ((int64_t) amount), PA_SEEK_RELATIVE, true);
+
             /* Convert back to sink domain */
-            if (i->thread_info.resampler)
-                amount = pa_resampler_result(i->thread_info.resampler, amount);
+            amount = pa_resampler_result(i->thread_info.resampler, amount);
 
             if (amount > 0)
                 /* Ok, now update the write pointer */
                 pa_memblockq_seek(i->thread_info.render_memblockq, - ((int64_t) amount), PA_SEEK_RELATIVE, true);
 
-            if (i->thread_info.rewrite_flush)
-                pa_memblockq_silence(i->thread_info.render_memblockq);
-
             /* And rewind the resampler */
             if (i->thread_info.resampler)
                 pa_resampler_rewind(i->thread_info.resampler, amount);
+
+            if (i->thread_info.rewrite_flush) {
+                pa_memblockq_silence(i->thread_info.render_memblockq);
+                pa_memblockq_silence(i->thread_info.history_memblockq);
+            }
         }
     }
 
@@ -1157,7 +1206,7 @@ size_t pa_sink_input_get_max_rewind(pa_sink_input *i) {
     pa_sink_input_assert_ref(i);
     pa_sink_input_assert_io_context(i);
 
-    return i->thread_info.resampler ? pa_resampler_request(i->thread_info.resampler, i->sink->thread_info.max_rewind) : i->sink->thread_info.max_rewind;
+    return pa_resampler_request(i->thread_info.resampler, i->sink->thread_info.max_rewind);
 }
 
 /* Called from thread context */
@@ -1168,11 +1217,13 @@ size_t pa_sink_input_get_max_request(pa_sink_input *i) {
     /* We're not verifying the status here, to allow this to be called
      * in the state change handler between _INIT and _RUNNING */
 
-    return i->thread_info.resampler ? pa_resampler_request(i->thread_info.resampler, i->sink->thread_info.max_request) : i->sink->thread_info.max_request;
+    return pa_resampler_request(i->thread_info.resampler, i->sink->thread_info.max_request);
 }
 
 /* Called from thread context */
 void pa_sink_input_update_max_rewind(pa_sink_input *i, size_t nbytes  /* in the sink's sample spec */) {
+    size_t max_rewind;
+
     pa_sink_input_assert_ref(i);
     pa_sink_input_assert_io_context(i);
     pa_assert(PA_SINK_INPUT_IS_LINKED(i->thread_info.state));
@@ -1180,8 +1231,12 @@ void pa_sink_input_update_max_rewind(pa_sink_input *i, size_t nbytes  /* in the
 
     pa_memblockq_set_maxrewind(i->thread_info.render_memblockq, nbytes);
 
+    max_rewind = pa_resampler_request(i->thread_info.resampler, nbytes);
+
+    pa_memblockq_set_maxrewind(i->thread_info.history_memblockq, max_rewind + PA_RESAMPLER_MAX_HISTORY * pa_frame_size(&i->sample_spec));
+
     if (i->update_max_rewind)
-        i->update_max_rewind(i, i->thread_info.resampler ? pa_resampler_request(i->thread_info.resampler, nbytes) : nbytes);
+        i->update_max_rewind(i, max_rewind);
 }
 
 /* Called from thread context */
@@ -1192,7 +1247,7 @@ void pa_sink_input_update_max_request(pa_sink_input *i, size_t nbytes  /* in the
     pa_assert(pa_frame_aligned(nbytes, &i->sink->sample_spec));
 
     if (i->update_max_request)
-        i->update_max_request(i, i->thread_info.resampler ? pa_resampler_request(i->thread_info.resampler, nbytes) : nbytes);
+        i->update_max_request(i, pa_resampler_request(i->thread_info.resampler, nbytes));
 }
 
 /* Called from thread context */
@@ -2220,8 +2275,7 @@ void pa_sink_input_request_rewind(
         nbytes = i->sink->thread_info.max_rewind + lbq;
 
         /* Transform from sink domain */
-        if (i->thread_info.resampler)
-            nbytes = pa_resampler_request(i->thread_info.resampler, nbytes);
+        nbytes = pa_resampler_request(i->thread_info.resampler, nbytes);
     }
 
     /* Remember how much we actually want to rewrite */
@@ -2247,8 +2301,7 @@ void pa_sink_input_request_rewind(
     if (nbytes != (size_t) -1) {
 
         /* Transform to sink domain */
-        if (i->thread_info.resampler)
-            nbytes = pa_resampler_result(i->thread_info.resampler, nbytes);
+        nbytes = pa_resampler_result(i->thread_info.resampler, nbytes);
 
         if (nbytes > lbq)
             pa_sink_request_rewind(i->sink, nbytes - lbq);
@@ -2354,6 +2407,7 @@ int pa_sink_input_update_resampler(pa_sink_input *i) {
     i->thread_info.resampler = new_resampler;
 
     pa_memblockq_free(i->thread_info.render_memblockq);
+    pa_memblockq_flush_write(i->thread_info.history_memblockq, true);
 
     memblockq_name = pa_sprintf_malloc("sink input render_memblockq [%u]", i->index);
     i->thread_info.render_memblockq = pa_memblockq_new(
diff --git a/src/pulsecore/sink-input.h b/src/pulsecore/sink-input.h
index d3de6e3f8..fbb304bfe 100644
--- a/src/pulsecore/sink-input.h
+++ b/src/pulsecore/sink-input.h
@@ -252,6 +252,10 @@ struct pa_sink_input {
         /* We maintain a history of resampled audio data here. */
         pa_memblockq *render_memblockq;
 
+        /* This queue keeps the history before resampling and is used
+         * when rewinding the resampler. */
+        pa_memblockq *history_memblockq;
+
         pa_sink_input *sync_prev, *sync_next;
 
         /* The requested latency for the sink */