Fix teardown after silence detector trigger

8 files changed, 60 insertions, 36 deletions

diff --git a/src/AlsaInput.cpp b/src/AlsaInput.cpp
index 0d3b40e..442304c 100644
--- a/src/AlsaInput.cpp
+++ b/src/AlsaInput.cpp
@@ -31,6 +31,9 @@ using namespace std;
 
 AlsaInput::~AlsaInput()
 {
+    // Ensures push() doesn't get blocked
+    m_queue.clear();
+
     if (m_alsa_handle) {
         snd_pcm_close(m_alsa_handle);
         m_alsa_handle = nullptr;
@@ -123,6 +126,18 @@ ssize_t AlsaInput::m_read(uint8_t* buf, snd_pcm_uframes_t length)
     return err;
 }
 
+AlsaInputThreaded::~AlsaInputThreaded()
+{
+    m_running = false;
+
+    // Ensures push() doesn't get blocked
+    m_queue.clear();
+
+    if (m_thread.joinable()) {
+        m_thread.join();
+    }
+}
+
 void AlsaInputThreaded::prepare()
 {
     if (m_fault) {
diff --git a/src/AlsaInput.h b/src/AlsaInput.h
index 8055140..f697a3c 100644
--- a/src/AlsaInput.h
+++ b/src/AlsaInput.h
@@ -111,13 +111,7 @@ class AlsaInputThreaded : public AlsaInput
             m_fault(false),
             m_running(false) { }
 
-        virtual ~AlsaInputThreaded()
-        {
-            if (m_running) {
-                m_running = false;
-                m_thread.join();
-            }
-        }
+        virtual ~AlsaInputThreaded();
 
         /*! Start the ALSA thread that fills the queue */
         virtual void prepare(void) override;
diff --git a/src/FileInput.cpp b/src/FileInput.cpp
index 51b0456..517ea7d 100644
--- a/src/FileInput.cpp
+++ b/src/FileInput.cpp
@@ -27,6 +27,9 @@ using namespace std;
 
 FileInput::~FileInput()
 {
+    // Ensures push() doesn't get blocked
+    m_queue.clear();
+
     if (m_raw_input and m_in_fh) {
         fclose(m_in_fh);
     }
diff --git a/src/GSTInput.cpp b/src/GSTInput.cpp
index aa46003..a2401cb 100644
--- a/src/GSTInput.cpp
+++ b/src/GSTInput.cpp
@@ -271,6 +271,10 @@ void GSTInput::process()
 GSTInput::~GSTInput()
 {
     m_running = false;
+
+    // Ensures push() doesn't get blocked
+    m_gst_data.samplequeue.clear();
+
     if (m_thread.joinable()) {
         m_thread.join();
     }
diff --git a/src/JackInput.cpp b/src/JackInput.cpp
index bc57588..4d9a530 100644
--- a/src/JackInput.cpp
+++ b/src/JackInput.cpp
@@ -34,6 +34,9 @@ using namespace std;
 
 JackInput::~JackInput()
 {
+    // Ensures push() doesn't get blocked
+    m_queue.clear();
+
     if (m_client) {
         jack_client_close(m_client);
     }
diff --git a/src/SampleQueue.h b/src/SampleQueue.h
index 95706ee..f8bc799 100644
--- a/src/SampleQueue.h
+++ b/src/SampleQueue.h
@@ -1,5 +1,5 @@
 /*
- * Copyright (C) 2018 Matthias P. Braendli
+ * Copyright (C) 2020 Matthias P. Braendli
  *
  * Licensed under the Apache License, Version 2.0 (the "License");
  * you may not use this file except in compliance with the License.
@@ -63,15 +63,15 @@ template<typename T>
 class SampleQueue
 {
 public:
-    SampleQueue(unsigned int bytes_per_sample,
-            unsigned int channels,
-            size_t max_size,
-            bool drift_compensation) :
-        m_channels(channels),
-        m_bytes_per_sample(bytes_per_sample),
-        m_max_size(max_size),
-        m_push_block(not drift_compensation),
-        m_overruns(0) {}
+    SampleQueue(unsigned int bytes_per_sample) :
+        m_bytes_per_sample(bytes_per_sample) {}
+
+    void configure(size_t max_size, bool push_block, unsigned int channels)
+    {
+        m_max_size = max_size;
+        m_push_block = push_block;
+        m_channels = channels;
+    }
 
 
     /*! Push a bunch of samples into the buffer
@@ -275,9 +275,16 @@ public:
         return ret;
     }
 
-    void set_max_size(size_t max_size)
+    void clear()
     {
-        m_max_size = max_size;
+        {
+            std::lock_guard<std::mutex> lock(m_mutex);
+            m_queue.clear();
+#if DEBUG_SAMPLE_QUEUE
+            fprintf(stdout, "clear\n");
+#endif
+        }
+        m_pop_notification.notify_all();
     }
 
 private:
@@ -286,15 +293,15 @@ private:
     std::condition_variable m_push_notification;
     std::condition_variable m_pop_notification;
 
-    unsigned int m_channels;
     unsigned int m_bytes_per_sample;
-    size_t m_max_size;
-    bool m_push_block;
+    unsigned int m_channels = 2;
+    size_t m_max_size = 1;
+    bool m_push_block = true;
 
     /*! Counter to keep track of number of overruns between calls
      * to pop()
      */
-    size_t m_overruns;
+    size_t m_overruns = 0;
 };
 
 #endif
diff --git a/src/VLCInput.cpp b/src/VLCInput.cpp
index 7b10d81..0547696 100644
--- a/src/VLCInput.cpp
+++ b/src/VLCInput.cpp
@@ -125,6 +125,9 @@ VLCInput::~VLCInput()
 {
     m_running = false;
 
+    // Ensures push() doesn't get blocked
+    m_samplequeue.clear();
+
     if (m_thread.joinable()) {
         m_thread.join();
     }
diff --git a/src/odr-audioenc.cpp b/src/odr-audioenc.cpp
index 9a0b0c4..9264ac7 100644
--- a/src/odr-audioenc.cpp
+++ b/src/odr-audioenc.cpp
@@ -491,7 +491,7 @@ public:
     unique_ptr<AACDecoder> decoder;
     unique_ptr<StatsPublisher> stats_publisher;
 
-    AudioEnc() : queue(BYTES_PER_SAMPLE, channels, 0, drift_compensation) { }
+    AudioEnc() : queue(BYTES_PER_SAMPLE) { }
     AudioEnc(const AudioEnc&) = delete;
     AudioEnc& operator=(const AudioEnc&) = delete;
     ~AudioEnc();
@@ -764,7 +764,7 @@ int AudioEnc::run()
     /*! The SampleQueue \c queue is given to the inputs, so that they
      * can fill it.
      */
-    queue.set_max_size(max_size);
+    queue.configure(max_size, not drift_compensation, channels);
 
     /* symsize=8, gfpoly=0x11d, fcr=0, prim=1, nroots=10, pad=135 */
     rs_handler = init_rs_char(8, 0x11d, 0, 1, 10, 135);
@@ -1319,20 +1319,17 @@ shared_ptr<InputInterface> AudioEnc::initialise_input()
     shared_ptr<InputInterface> input;
 
     if (not infile.empty()) {
-        input = make_shared<FileInput>(infile, raw_input, sample_rate,
-                continue_after_eof, queue);
+        input = make_shared<FileInput>(infile, raw_input, sample_rate, continue_after_eof, queue);
     }
 #if HAVE_JACK
     else if (not jack_name.empty()) {
-        input = make_shared<JackInput>(jack_name, channels, sample_rate,
-                queue);
+        input = make_shared<JackInput>(jack_name, channels, sample_rate, queue);
     }
 #endif
 #if HAVE_VLC
     else if (not vlc_uri.empty()) {
-        input = make_shared<VLCInput>(vlc_uri, sample_rate, channels,
-                verbosity, vlc_gain, vlc_cache, vlc_additional_opts,
-                queue);
+        input = make_shared<VLCInput>(vlc_uri, sample_rate, channels, verbosity,
+                vlc_gain, vlc_cache, vlc_additional_opts, queue);
     }
 #endif
 #if HAVE_GST
@@ -1342,12 +1339,10 @@ shared_ptr<InputInterface> AudioEnc::initialise_input()
 #endif
 #if HAVE_ALSA
     else if (drift_compensation) {
-        input = make_shared<AlsaInputThreaded>(alsa_device, channels,
-                sample_rate, queue);
+        input = make_shared<AlsaInputThreaded>(alsa_device, channels, sample_rate, queue);
     }
     else {
-        input = make_shared<AlsaInputDirect>(alsa_device, channels,
-                sample_rate, queue);
+        input = make_shared<AlsaInputDirect>(alsa_device, channels, sample_rate, queue);
     }
 #endif