Add input.backpressure_input configuration optionnext

4 files changed, 38 insertions, 7 deletions

diff --git a/doc/example.ini b/doc/example.ini
index 0d0f8e3..a1129de 100644
--- a/doc/example.ini
+++ b/doc/example.ini
@@ -76,6 +76,12 @@ loop=0
 ;transport=tcp
 ;source=localhost:9200
 
+
+; 1 means that the output device will throttle the input rate.
+; 0 means that frames will be dropped if the input is too fast.
+; Default depends on the value of transport: 1 for file, 0 for network inputs
+;backpressure_input=1
+
 [modulator]
 ;   Mode 'fix' uses a fixed factor and is really not recommended. It is more
 ; useful on an academic perspective for people trying to understand the DAB
diff --git a/src/ConfigParser.cpp b/src/ConfigParser.cpp
index 408ded9..085ed09 100644
--- a/src/ConfigParser.cpp
+++ b/src/ConfigParser.cpp
@@ -136,6 +136,27 @@ static void parse_configfile(
 
     mod_settings.inputTransport = pt.Get("input.transport", "file");
 
+    bool defaultBackpressureInput = false;
+    if (mod_settings.inputTransport == "edi" or
+        mod_settings.inputTransport == "tcp") {
+        defaultBackpressureInput = false;
+    }
+    else if (mod_settings.inputTransport == "file") {
+        defaultBackpressureInput = true;
+    }
+    else {
+        throw std::runtime_error("invalid input transport " +
+                mod_settings.inputTransport + " selected!");
+    }
+
+    // With unsynchronised sources (e.g. file input), the mod should
+    // throttle the input. For network inputs however, it's better to
+    // allow queue overflows, otherwise during startup, a slow output
+    // device will also block the entire mod and input.
+    mod_settings.sdr_device_config.blockingQueue =
+        pt.GetBoolean("input.backpressure_input", defaultBackpressureInput);
+
+
     mod_settings.edi_max_delay_ms = pt.GetReal("input.edi_max_delay", 0.0);
 
     mod_settings.inputName = pt.Get("input.source", "/dev/stdin");
@@ -241,7 +262,7 @@ static void parse_configfile(
     }
 #if defined(HAVE_OUTPUT_UHD)
     else if (output_selected == "uhd") {
-        Output::SDRDeviceConfig sdr_device_config;
+        auto& sdr_device_config = mod_settings.sdr_device_config;
 
         string device = pt.Get("uhdoutput.device", "");
         const auto usrpType = pt.Get("uhdoutput.type", "");
@@ -310,7 +331,6 @@ static void parse_configfile(
 
         sdr_device_config.dpdFeedbackServerPort = pt.GetInteger("uhdoutput.dpd_port", 0);
 
-        mod_settings.sdr_device_config = sdr_device_config;
         mod_settings.useUHDOutput = true;
     }
 #endif // defined(HAVE_OUTPUT_UHD)
diff --git a/src/output/SDR.cpp b/src/output/SDR.cpp
index 22398c7..7d5a609 100644
--- a/src/output/SDR.cpp
+++ b/src/output/SDR.cpp
@@ -188,10 +188,15 @@ meta_vec_t SDR::process_metadata(const meta_vec_t& metadataIn)
 
 
             const auto max_size = m_config.enableSync ? FRAMES_MAX_SIZE_SYNC : FRAMES_MAX_SIZE_UNSYNC;
-            auto r = m_queue.push_overflow(std::move(frame), max_size);
-            etiLog.log(trace, "SDR,push %d %zu", r.overflowed, r.new_size);
-
-            num_queue_overflows += r.overflowed ? 1 : 0;
+            if (m_config.blockingQueue) {
+                const auto new_size = m_queue.push_wait_if_full(std::move(frame), max_size);
+                etiLog.log(trace, "SDR,push 0 %zu", new_size);
+            }
+            else {
+                const auto r = m_queue.push_overflow(std::move(frame), max_size);
+                etiLog.log(trace, "SDR,push %d %zu", r.overflowed, r.new_size);
+                num_queue_overflows += r.overflowed ? 1 : 0;
+            }
         }
     }
     else {
diff --git a/src/output/SDRDevice.h b/src/output/SDRDevice.h
index ec9373d..c466285 100644
--- a/src/output/SDRDevice.h
+++ b/src/output/SDRDevice.h
@@ -58,7 +58,7 @@ struct SDRDeviceConfig {
     std::string rx_antenna;
 
     bool fixedPoint = false;
-
+    bool blockingQueue = false;
     long masterClockRate = 32768000;
     unsigned sampleRate = 2048000;
     double frequency = 0.0;