Merge remote branch 'mkbosmans/rate-adjustment'

src/modules/module-combine.c

@@ -217,23 +217,29 @@ static void adjust_rates(struct userdata *u) {
     base_rate = u->sink->sample_spec.rate;
 
     PA_IDXSET_FOREACH(o, u->outputs, idx) {
-        uint32_t r = base_rate;
+        uint32_t new_rate = base_rate;
+        uint32_t current_rate = o->sink_input->sample_spec.rate;
 
         if (!o->sink_input || !PA_SINK_IS_OPENED(pa_sink_get_state(o->sink)))
             continue;
 
-        if (o->total_latency < target_latency)
-            r -= (uint32_t) ((((double) (target_latency - o->total_latency))/(double)u->adjust_time)*(double)r);
-        else if (o->total_latency > target_latency)
-            r += (uint32_t) ((((double) (o->total_latency - target_latency))/(double)u->adjust_time)*(double)r);
+        if (o->total_latency != target_latency)
+            new_rate += (uint32_t) (((double) o->total_latency - (double) target_latency) / (double) u->adjust_time * (double) new_rate);
 
-        if (r < (uint32_t) (base_rate*0.9) || r > (uint32_t) (base_rate*1.1)) {
-            pa_log_warn("[%s] sample rates too different, not adjusting (%u vs. %u).", o->sink_input->sink->name, base_rate, r);
-            pa_sink_input_set_rate(o->sink_input, base_rate);
+        if (new_rate < (uint32_t) (base_rate*0.8) || new_rate > (uint32_t) (base_rate*1.25)) {
+            pa_log_warn("[%s] sample rates too different, not adjusting (%u vs. %u).", o->sink_input->sink->name, base_rate, new_rate);
+            new_rate = base_rate;
         } else {
-            pa_log_info("[%s] new rate is %u Hz; ratio is %0.3f; latency is %0.0f usec.", o->sink_input->sink->name, r, (double) r / base_rate, (float) o->total_latency);
-            pa_sink_input_set_rate(o->sink_input, r);
+            if (base_rate < new_rate + 20 && new_rate < base_rate + 20)
+              new_rate = base_rate;
+            /* Do the adjustment in small steps; 2‰ can be considered inaudible */
+            if (new_rate < (uint32_t) (current_rate*0.998) || new_rate > (uint32_t) (current_rate*1.002)) {
+                pa_log_info("[%s] new rate of %u Hz not within 2‰ of %u Hz, forcing smaller adjustment", o->sink_input->sink->name, new_rate, current_rate);
+                new_rate = PA_CLAMP(new_rate, (uint32_t) (current_rate*0.998), (uint32_t) (current_rate*1.002));
+            }
+            pa_log_info("[%s] new rate is %u Hz; ratio is %0.3f; latency is %0.2f msec.", o->sink_input->sink->name, new_rate, (double) new_rate / base_rate, (double) o->total_latency / PA_USEC_PER_MSEC);
         }
+        pa_sink_input_set_rate(o->sink_input, new_rate);
     }
 
     pa_asyncmsgq_send(u->sink->asyncmsgq, PA_MSGOBJECT(u->sink), SINK_MESSAGE_UPDATE_LATENCY, NULL, (int64_t) avg_total_latency, NULL);

src/modules/module-loopback.c

@@ -108,6 +108,7 @@ struct userdata {
 static const char* const valid_modargs[] = {
     "source",
     "sink",
+    "adjust_time",
     "latency_msec",
     "format",
     "rate",
@@ -178,13 +179,13 @@ static void adjust_rates(struct userdata *u) {
 
     buffer_latency = pa_bytes_to_usec(buffer, &u->sink_input->sample_spec);
 
-    pa_log_info("Loopback overall latency is %0.2f ms + %0.2f ms + %0.2f ms = %0.2f ms",
+    pa_log_debug("Loopback overall latency is %0.2f ms + %0.2f ms + %0.2f ms = %0.2f ms",
                 (double) u->latency_snapshot.sink_latency / PA_USEC_PER_MSEC,
                 (double) buffer_latency / PA_USEC_PER_MSEC,
                 (double) u->latency_snapshot.source_latency / PA_USEC_PER_MSEC,
                 ((double) u->latency_snapshot.sink_latency + buffer_latency + u->latency_snapshot.source_latency) / PA_USEC_PER_MSEC);
 
-    pa_log_info("Should buffer %zu bytes, buffered at minimum %zu bytes",
+    pa_log_debug("Should buffer %zu bytes, buffered at minimum %zu bytes",
                 u->latency_snapshot.max_request*2,
                 u->latency_snapshot.min_memblockq_length);
 
@@ -197,9 +198,21 @@ static void adjust_rates(struct userdata *u) {
     else
         new_rate = base_rate + (((u->latency_snapshot.min_memblockq_length - u->latency_snapshot.max_request*2) / fs) *PA_USEC_PER_SEC)/u->adjust_time;
 
-    pa_log_info("Old rate %lu Hz, new rate %lu Hz", (unsigned long) old_rate, (unsigned long) new_rate);
+    if (new_rate < (uint32_t) (base_rate*0.8) || new_rate > (uint32_t) (base_rate*1.25)) {
+        pa_log_warn("Sample rates too different, not adjusting (%u vs. %u).", base_rate, new_rate);
+        new_rate = base_rate;
+    } else {
+        if (base_rate < new_rate + 20 && new_rate < base_rate + 20)
+          new_rate = base_rate;
+        /* Do the adjustment in small steps; 2‰ can be considered inaudible */
+        if (new_rate < (uint32_t) (old_rate*0.998) || new_rate > (uint32_t) (old_rate*1.002)) {
+            pa_log_info("New rate of %u Hz not within 2‰ of %u Hz, forcing smaller adjustment", new_rate, old_rate);
+            new_rate = PA_CLAMP(new_rate, (uint32_t) (old_rate*0.998), (uint32_t) (old_rate*1.002));
+        }
+    }
 
     pa_sink_input_set_rate(u->sink_input, new_rate);
+    pa_log_debug("[%s] Updated sampling rate to %lu Hz.", u->sink_input->sink->name, (unsigned long) new_rate);
 
     pa_core_rttime_restart(u->core, u->time_event, pa_rtclock_now() + u->adjust_time);
 }

src/modules/rtp/module-rtp-recv.c

@@ -52,7 +52,6 @@
 #include <pulsecore/macro.h>
 #include <pulsecore/atomic.h>
 #include <pulsecore/atomic.h>
-#include <pulsecore/time-smoother.h>
 #include <pulsecore/socket-util.h>
 #include <pulsecore/once.h>
 
@@ -104,11 +103,13 @@ struct session {
 
     pa_atomic_t timestamp;
 
-    pa_smoother *smoother;
     pa_usec_t intended_latency;
     pa_usec_t sink_latency;
 
     pa_usec_t last_rate_update;
+    pa_usec_t last_latency;
+    double estimated_rate;
+    double avg_estimated_rate;
 };
 
 struct userdata {
@@ -194,10 +195,9 @@ static void sink_input_suspend_within_thread(pa_sink_input* i, pa_bool_t b) {
     pa_sink_input_assert_ref(i);
     pa_assert_se(s = i->userdata);
 
-    if (b) {
-        pa_smoother_pause(s->smoother, pa_rtclock_now());
+    if (b)
         pa_memblockq_flush_read(s->memblockq);
-    } else
+    else
         s->first_packet = FALSE;
 }
 
@@ -266,11 +266,6 @@ static int rtpoll_work_cb(pa_rtpoll_item *i) {
     } else
         pa_rtclock_from_wallclock(&now);
 
-    pa_smoother_put(s->smoother, pa_timeval_load(&now), pa_bytes_to_usec((uint64_t) pa_memblockq_get_write_index(s->memblockq), &s->sink_input->sample_spec));
-
-    /* Tell the smoother that we are rolling now, in case it is still paused */
-    pa_smoother_resume(s->smoother, pa_timeval_load(&now), TRUE);
-
     if (pa_memblockq_push(s->memblockq, &chunk) < 0) {
         pa_log_warn("Queue overrun");
         pa_memblockq_seek(s->memblockq, (int64_t) chunk.length, PA_SEEK_RELATIVE, TRUE);
@@ -286,12 +281,15 @@ static int rtpoll_work_cb(pa_rtpoll_item *i) {
     pa_atomic_store(&s->timestamp, (int) now.tv_sec);
 
     if (s->last_rate_update + RATE_UPDATE_INTERVAL < pa_timeval_load(&now)) {
-        pa_usec_t wi, ri, render_delay, sink_delay = 0, latency, fix;
-        unsigned fix_samples;
+        pa_usec_t wi, ri, render_delay, sink_delay = 0, latency;
+        uint32_t base_rate = s->sink_input->sink->sample_spec.rate;
+        uint32_t current_rate = s->sink_input->sample_spec.rate;
+        uint32_t new_rate;
+        double estimated_rate, alpha = 0.02;
 
         pa_log_debug("Updating sample rate");
 
-        wi = pa_smoother_get(s->smoother, pa_timeval_load(&now));
+        wi = pa_bytes_to_usec((uint64_t) pa_memblockq_get_write_index(s->memblockq), &s->sink_input->sample_spec);
         ri = pa_bytes_to_usec((uint64_t) pa_memblockq_get_read_index(s->memblockq), &s->sink_input->sample_spec);
 
         pa_log_debug("wi=%lu ri=%lu", (unsigned long) wi, (unsigned long) ri);
@@ -309,30 +307,61 @@ static int rtpoll_work_cb(pa_rtpoll_item *i) {
         else
             latency = wi - ri;
 
-        pa_log_debug("Write index deviates by %0.2f ms, expected %0.2f ms", (double) latency/PA_USEC_PER_MSEC, (double)  s->intended_latency/PA_USEC_PER_MSEC);
+        pa_log_debug("Write index deviates by %0.2f ms, expected %0.2f ms", (double) latency/PA_USEC_PER_MSEC, (double) s->intended_latency/PA_USEC_PER_MSEC);
 
-        /* Calculate deviation */
-        if (latency < s->intended_latency)
-            fix = s->intended_latency - latency;
-        else
-            fix = latency - s->intended_latency;
-
-        /* How many samples is this per second? */
-        fix_samples = (unsigned) (fix * (pa_usec_t) s->sink_input->thread_info.sample_spec.rate / (pa_usec_t) RATE_UPDATE_INTERVAL);
-
-        /* Check if deviation is in bounds */
-        if (fix_samples > s->sink_input->sample_spec.rate*.50)
-            pa_log_debug("Hmmm, rate fix is too large (%lu Hz), not applying.", (unsigned long) fix_samples);
-        else {
-            /* Fix up rate */
-            if (latency < s->intended_latency)
-                s->sink_input->sample_spec.rate -= fix_samples;
-            else
-                s->sink_input->sample_spec.rate += fix_samples;
-
-            if (s->sink_input->sample_spec.rate > PA_RATE_MAX)
-                s->sink_input->sample_spec.rate = PA_RATE_MAX;
+        /* The buffer is filling with some unknown rate R̂ samples/second. If the rate of reading in
+         * the last T seconds was Rⁿ, then the increase in buffer latency ΔLⁿ = Lⁿ - Lⁿ⁻ⁱ in that
+         * same period is ΔLⁿ = (TR̂ - TRⁿ) / R̂, giving the estimated target rate
+         *                                           T
+         *                                 R̂ = ─────────────── Rⁿ .                             (1)
+         *                                     T - (Lⁿ - Lⁿ⁻ⁱ)
+         *
+         * Setting the sample rate to R̂ results in the latency being constant (if the estimate of R̂
+         * is correct).  But there is also the requirement to keep the buffer at a predefined target
+         * latency L̂.  So instead of setting Rⁿ⁺ⁱ to R̂ immediately, the strategy will be to reduce R
+         * from Rⁿ⁺ⁱ to R̂ in a steps of T seconds, where Rⁿ⁺ⁱ is chosen such that in the total time
+         * aT the latency is reduced from Lⁿ to L̂.  This strategy translates to the requirements
+         *            ₐ      R̂ - Rⁿ⁺ʲ                            a-j+1         j-1
+         *            Σ  T ────────── = L̂ - Lⁿ    with    Rⁿ⁺ʲ = ───── Rⁿ⁺ⁱ + ───── R̂ .
+         *           ʲ⁼ⁱ        R̂                                  a            a
+         * Solving for Rⁿ⁺ⁱ gives
+         *                                     T - ²∕ₐ₊₁(L̂ - Lⁿ)
+         *                              Rⁿ⁺ⁱ = ───────────────── R̂ .                            (2)
+         *                                            T
+         * In the code below a = 7 is used.
+         *
+         * Equation (1) is not directly used in (2), but instead an exponentially weighted average
+         * of the estimated rate R̂ is used.  This average R̅ is defined as
+         *                                R̅ⁿ = α R̂ⁿ + (1-α) R̅ⁿ⁻ⁱ .
+         * Because it is difficult to find a fixed value for the coefficient α such that the
+         * averaging is without significant lag but oscillations are filtered out, a heuristic is
+         * used.  When the successive estimates R̂ⁿ do not change much then α→1, but when there is a
+         * sudden spike in the estimated rate α→0, such that the deviation is given little weight.
+         */
+        estimated_rate = (double) current_rate * (double) RATE_UPDATE_INTERVAL / (double) (RATE_UPDATE_INTERVAL + s->last_latency - latency);
+        if (fabs(s->estimated_rate - s->avg_estimated_rate) > 1) {
+          double ratio = (estimated_rate + s->estimated_rate - 2*s->avg_estimated_rate) / (s->estimated_rate - s->avg_estimated_rate);
+          alpha = PA_CLAMP(2 * (ratio + fabs(ratio)) / (4 + ratio*ratio), 0.02, 0.8);
         }
+        s->avg_estimated_rate = alpha * estimated_rate + (1-alpha) * s->avg_estimated_rate;
+        s->estimated_rate = estimated_rate;
+        pa_log_debug("Estimated target rate: %.0f Hz, using average of %.0f Hz  (α=%.3f)", estimated_rate, s->avg_estimated_rate, alpha);
+        new_rate = (uint32_t) ((double) (RATE_UPDATE_INTERVAL + latency/4 - s->intended_latency/4) / (double) RATE_UPDATE_INTERVAL * s->avg_estimated_rate);
+        s->last_latency = latency;
+
+        if (new_rate < (uint32_t) (base_rate*0.8) || new_rate > (uint32_t) (base_rate*1.25)) {
+            pa_log_warn("Sample rates too different, not adjusting (%u vs. %u).", base_rate, new_rate);
+            new_rate = base_rate;
+        } else {
+            if (base_rate < new_rate + 20 && new_rate < base_rate + 20)
+              new_rate = base_rate;
+            /* Do the adjustment in small steps; 2‰ can be considered inaudible */
+            if (new_rate < (uint32_t) (current_rate*0.998) || new_rate > (uint32_t) (current_rate*1.002)) {
+                pa_log_info("New rate of %u Hz not within 2‰ of %u Hz, forcing smaller adjustment", new_rate, current_rate);
+                new_rate = PA_CLAMP(new_rate, (uint32_t) (current_rate*0.998), (uint32_t) (current_rate*1.002));
+            }
+        }
+        s->sink_input->sample_spec.rate = new_rate;
 
         pa_assert(pa_sample_spec_valid(&s->sink_input->sample_spec));
 
@@ -346,7 +375,9 @@ static int rtpoll_work_cb(pa_rtpoll_item *i) {
     if (pa_memblockq_is_readable(s->memblockq) &&
         s->sink_input->thread_info.underrun_for > 0) {
         pa_log_debug("Requesting rewind due to end of underrun");
-        pa_sink_input_request_rewind(s->sink_input, 0, FALSE, TRUE, FALSE);
+        pa_sink_input_request_rewind(s->sink_input,
+                                     (size_t) (s->sink_input->thread_info.underrun_for == (uint64_t) -1 ? 0 : s->sink_input->thread_info.underrun_for),
+                                     FALSE, TRUE, FALSE);
     }
 
     return 1;
@@ -471,15 +502,10 @@ static struct session *session_new(struct userdata *u, const pa_sdp_info *sdp_in
     s->sdp_info = *sdp_info;
     s->rtpoll_item = NULL;
     s->intended_latency = LATENCY_USEC;
-    s->smoother = pa_smoother_new(
-            PA_USEC_PER_SEC*5,
-            PA_USEC_PER_SEC*2,
-            TRUE,
-            TRUE,
-            10,
-            pa_timeval_load(&now),
-            TRUE);
     s->last_rate_update = pa_timeval_load(&now);
+    s->last_latency = LATENCY_USEC;
+    s->estimated_rate = (double) sink->sample_spec.rate;
+    s->avg_estimated_rate = (double) sink->sample_spec.rate;
     pa_atomic_store(&s->timestamp, (int) now.tv_sec);
 
     if ((fd = mcast_socket((const struct sockaddr*) &sdp_info->sa, sdp_info->salen)) < 0)
@@ -579,8 +605,6 @@ static void session_free(struct session *s) {
     pa_sdp_info_destroy(&s->sdp_info);
     pa_rtp_context_destroy(&s->rtp_context);
 
-    pa_smoother_free(s->smoother);
-
     pa_xfree(s);
 }