roc-streaminggh-688: Dynamic latency adjustment roc-streaming#688

docs/sphinx/manuals/roc_recv.rst

@@ -28,8 +28,11 @@ Options
 --miface=MIFACE               IPv4 or IPv6 address of the network interface on which to join the multicast group
 --reuseaddr                   enable SO_REUSEADDR when binding sockets
 --target-latency=STRING       Target latency, TIME units
---io-latency=STRING           Playback target latency, TIME units
 --latency-tolerance=STRING    Maximum deviation from target latency, TIME units
+--start-latency=STRING        Target latency, TIME units
+--min-latency=STRING          Minimum allowed latency, TIME units
+--max-latency=STRING          Maximum allowed latency, TIME units
+--io-latency=STRING           Playback target latency, TIME units
 --no-play-timeout=STRING      No playback timeout, TIME units
 --choppy-play-timeout=STRING  Choppy playback timeout, TIME units
 --frame-len=TIME              Duration of the internal frames, TIME units

src/internal_modules/roc_audio/feedback_monitor.cpp

@@ -20,8 +20,9 @@ FeedbackMonitor::FeedbackMonitor(IFrameWriter& writer,
                                  ResamplerWriter* resampler,
                                  const FeedbackConfig& feedback_config,
                                  const LatencyConfig& latency_config,
-                                 const SampleSpec& sample_spec)
-    : tuner_(latency_config, sample_spec)
+                                 const SampleSpec& sample_spec,
+                                 core::CsvDumper* dumper)
+    : tuner_(latency_config, sample_spec, dumper)
     , use_packetizer_(false)
     , has_feedback_(false)
     , last_feedback_ts_(0)
@@ -104,10 +105,10 @@ void FeedbackMonitor::process_feedback(packet::stream_source_t source_id,
     latency_metrics_ = latency_metrics;
     link_metrics_ = link_metrics;
 
-    if (link_metrics_.total_packets == 0 || use_packetizer_) {
+    if (link_metrics_.expected_packets == 0 || use_packetizer_) {
         // If packet counter is not reported from receiver, fallback to
         // counter from sender.
-        link_metrics_.total_packets = packetizer_.metrics().encoded_packet_count;
+        link_metrics_.expected_packets = packetizer_.metrics().encoded_packet_count;
         use_packetizer_ = true;
     }
 

src/internal_modules/roc_audio/feedback_monitor.h

@@ -68,7 +68,8 @@ class FeedbackMonitor : public IFrameWriter, public core::NonCopyable<> {
                     ResamplerWriter* resampler,
                     const FeedbackConfig& feedback_config,
                     const LatencyConfig& latency_config,
-                    const SampleSpec& sample_spec);
+                    const SampleSpec& sample_spec,
+                    core::CsvDumper* dumper);
 
     //! Check if the object was successfully constructed.
     status::StatusCode init_status() const;

src/internal_modules/roc_audio/freq_estimator.cpp

@@ -9,6 +9,7 @@
 #include "roc_audio/freq_estimator.h"
 #include "roc_core/log.h"
 #include "roc_core/panic.h"
+#include "roc_core/time.h"
 
 namespace roc {
 namespace audio {
@@ -25,15 +26,19 @@ FreqEstimatorConfig make_config(FreqEstimatorProfile profile) {
         config.I = 1e-10;
         config.decimation_factor1 = fe_decim_factor_max;
         config.decimation_factor2 = 0;
+        config.stable_criteria = 0.1;
         break;
 
     case FreqEstimatorProfile_Gradual:
         config.P = 1e-6;
         config.I = 5e-9;
         config.decimation_factor1 = fe_decim_factor_max;
         config.decimation_factor2 = fe_decim_factor_max;
+        config.stable_criteria = 0.05;
         break;
     }
+    config.stability_duration_criteria = 15 * core::Second;
+    config.control_action_saturation_cap = 1e-2;
 
     return config;
 }
@@ -62,14 +67,18 @@ double dot_prod(const double* coeff,
 } // namespace
 
 FreqEstimator::FreqEstimator(FreqEstimatorProfile profile,
-                             packet::stream_timestamp_t target_latency)
+                             packet::stream_timestamp_t target_latency,
+                             core::CsvDumper* dumper)
     : config_(make_config(profile))
     , target_(target_latency)
     , dec1_ind_(0)
     , dec2_ind_(0)
     , samples_counter_(0)
     , accum_(0)
-    , coeff_(1) {
+    , coeff_(1)
+    , stable_(false)
+    , last_unstable_time_(core::timestamp(core::ClockMonotonic))
+    , dumper_(dumper) {
     roc_log(LogDebug, "freq estimator: initializing: P=%e I=%e dc1=%lu dc2=%lu",
             config_.P, config_.I, (unsigned long)config_.decimation_factor1,
             (unsigned long)config_.decimation_factor2);
@@ -101,10 +110,13 @@ float FreqEstimator::freq_coeff() const {
     return (float)coeff_;
 }
 
-void FreqEstimator::update(packet::stream_timestamp_t current) {
+void FreqEstimator::update_current_latency(packet::stream_timestamp_t current_latency) {
     double filtered;
 
-    if (run_decimators_(current, filtered)) {
+    if (run_decimators_(current_latency, filtered)) {
+        if (dumper_) {
+            dump_(filtered);
+        }
         coeff_ = run_controller_(filtered);
     }
 }
@@ -149,8 +161,76 @@ bool FreqEstimator::run_decimators_(packet::stream_timestamp_t current,
 double FreqEstimator::run_controller_(double current) {
     const double error = (current - target_);
 
-    accum_ = accum_ + error;
-    return 1 + config_.P * error + config_.I * accum_;
+    roc_log(LogTrace,
+            "freq estimator:"
+            " current latency error: %.0f",
+            error);
+
+    const core::nanoseconds_t now = core::timestamp(core::ClockMonotonic);
+
+    if (std::abs(error) > target_ * config_.stable_criteria && stable_) {
+        stable_ = false;
+        accum_ = 0;
+        last_unstable_time_ = now;
+        roc_log(LogDebug,
+                "freq estimator: "
+                " unstable, %0.f > %.0f / %0.f",
+                config_.stable_criteria, error, target_);
+    } else if (std::abs(error) < target_ * config_.stable_criteria && !stable_
+               && now - last_unstable_time_ > config_.stability_duration_criteria) {
+        stable_ = true;
+        roc_log(LogDebug,
+                "freq estimator: "
+                " stabilized");
+    }
+
+    double res = 0.;
+    // In stable state we are not using P term in order to avoid permanent variation
+    // of resampler control input.
+    if (stable_) {
+        accum_ = accum_ + error;
+        res += config_.I * accum_;
+    } else {
+        res += config_.P * error;
+    }
+    if (std::abs(res) > config_.control_action_saturation_cap) {
+        res = res / std::abs(res) * config_.control_action_saturation_cap;
+    }
+    res += 1.;
+
+    return res;
+}
+
+void FreqEstimator::dump_(double filtered) {
+    core::CsvEntry e;
+    e.type = 'f';
+    e.n_fields = 5;
+    e.fields[0] = core::timestamp(core::ClockUnix);
+    e.fields[1] = filtered;
+    e.fields[2] = target_;
+    e.fields[3] = (filtered - target_) * config_.P;
+    e.fields[4] = accum_ * config_.I;
+    dumper_->write(e);
+}
+
+void FreqEstimator::update_target_latency(packet::stream_timestamp_t target_latency) {
+    target_ = (double)target_latency;
+}
+
+bool FreqEstimator::is_stable() const {
+    return stable_;
+}
+
+static const char* fe_profile_to_str(FreqEstimatorProfile profile) {
+    switch (profile) {
+    case FreqEstimatorProfile_Responsive:
+        return "responsive";
+
+    case FreqEstimatorProfile_Gradual:
+        return "gradual";
+    }
+
+    return "<invalid>";
 }
 
 } // namespace audio

src/internal_modules/roc_audio/freq_estimator.h

@@ -14,6 +14,7 @@
 
 #include "roc_audio/freq_estimator_decim.h"
 #include "roc_audio/sample.h"
+#include "roc_core/csv_dumper.h"
 #include "roc_core/noncopyable.h"
 #include "roc_packet/units.h"
 
@@ -44,11 +45,26 @@ struct FreqEstimatorConfig {
     //! to fe_decim_factor_max. Could be zero to disable the second decimation stage.
     size_t decimation_factor2;
 
+    //! Within this range we consider the FreqEstimator is stable.
+    //! stable_criteria > error / target;
+    double stable_criteria;
+
+    //! How much time current latency readings must be within stable_criteria range
+    //! to let FreqEstimator switch into stable state.
+    core::nanoseconds_t stability_duration_criteria;
+
+    //! FreqEstimator limits its output control action value with this value so as to
+    //! keep sensible pace of latency adjustment if there is a long way to go.
+    double control_action_saturation_cap;
+
     FreqEstimatorConfig()
         : P(0)
         , I(0)
         , decimation_factor1(0)
-        , decimation_factor2(0) {
+        , decimation_factor2(0)
+        , stable_criteria(0.1)
+        , stability_duration_criteria(0)
+        , control_action_saturation_cap(0) {
     }
 };
 
@@ -67,20 +83,32 @@ class FreqEstimator : public core::NonCopyable<> {
     //!  - @p profile defines configuration preset.
     //!  - @p target_latency defines latency we want to archive.
     FreqEstimator(FreqEstimatorProfile profile,
-                  packet::stream_timestamp_t target_latency);
+                  packet::stream_timestamp_t target_latency,
+                  roc::core::CsvDumper* dumper);
 
     //! Get current frequecy coefficient.
     float freq_coeff() const;
 
     //! Compute new value of frequency coefficient.
-    void update(packet::stream_timestamp_t current_latency);
+    void update_current_latency(packet::stream_timestamp_t current_latency);
+
+    //! Update target latency.
+    void update_target_latency(packet::stream_timestamp_t target_latency);
+
+    //! Is FreqEstimator in stable state.
+    //! @remarks
+    //!  If current_latency is in kept within certain limits around target_latency
+    //!  FreqEstimator is in 'stable' state, otherwise it is 'not-stable' state.
+    //!  The state affects internal regulator strategy and it effectiveness.
+    bool is_stable() const;
 
 private:
     bool run_decimators_(packet::stream_timestamp_t current, double& filtered);
     double run_controller_(double current);
+    void dump_(double filtered);
 
     const FreqEstimatorConfig config_;
-    const double target_; // Target latency.
+    double target_; // Target latency.
 
     double dec1_casc_buff_[fe_decim_len];
     size_t dec1_ind_;
@@ -92,6 +120,12 @@ class FreqEstimator : public core::NonCopyable<> {
     double accum_;           // Integrator value.
 
     double coeff_; // Current frequency coefficient value.
+
+    bool stable_; // True if FreqEstimator has stabilized.
+    // Last time when FreqEstimator was out of range.
+    core::nanoseconds_t last_unstable_time_;
+
+    core::CsvDumper* dumper_;
 };
 
 } // namespace audio

src/internal_modules/roc_audio/latency_monitor.cpp

@@ -10,9 +10,7 @@
 #include "roc_audio/freq_estimator.h"
 #include "roc_core/log.h"
 #include "roc_core/panic.h"
-#include "roc_core/stddefs.h"
 #include "roc_core/time.h"
-#include "roc_rtp/link_meter.h"
 
 namespace roc {
 namespace audio {
@@ -25,8 +23,9 @@ LatencyMonitor::LatencyMonitor(IFrameReader& frame_reader,
                                ResamplerReader* resampler,
                                const LatencyConfig& config,
                                const SampleSpec& packet_sample_spec,
-                               const SampleSpec& frame_sample_spec)
-    : tuner_(config, frame_sample_spec)
+                               const SampleSpec& frame_sample_spec,
+                               core::CsvDumper* dumper)
+    : tuner_(config, frame_sample_spec, dumper)
     , frame_reader_(frame_reader)
     , incoming_queue_(incoming_queue)
     , depacketizer_(depacketizer)

src/internal_modules/roc_audio/latency_monitor.h

@@ -19,6 +19,7 @@
 #include "roc_audio/resampler_reader.h"
 #include "roc_audio/sample_spec.h"
 #include "roc_core/attributes.h"
+#include "roc_core/csv_dumper.h"
 #include "roc_core/noncopyable.h"
 #include "roc_core/optional.h"
 #include "roc_core/time.h"
@@ -68,7 +69,8 @@ class LatencyMonitor : public IFrameReader, public core::NonCopyable<> {
                    ResamplerReader* resampler,
                    const LatencyConfig& config,
                    const SampleSpec& packet_sample_spec,
-                   const SampleSpec& frame_sample_spec);
+                   const SampleSpec& frame_sample_spec,
+                   core::CsvDumper* dumper);
 
     //! Check if the object was successfully constructed.
     status::StatusCode init_status() const;