Avoid race condition on teardown of pipelined plugins

8 files changed, 28 insertions, 20 deletions

diff --git a/src/FIRFilter.cpp b/src/FIRFilter.cpp
index bc2314a..96ad1b9 100644
--- a/src/FIRFilter.cpp
+++ b/src/FIRFilter.cpp
@@ -87,6 +87,11 @@ FIRFilter::FIRFilter(const std::string& taps_file) :
     start_pipeline_thread();
 }
 
+FIRFilter::~FIRFilter()
+{
+    stop_pipeline_thread();
+}
+
 void FIRFilter::load_filter_taps(const std::string &tapsFile)
 {
     std::vector<float> filter_taps;
diff --git a/src/FIRFilter.h b/src/FIRFilter.h
index 6ace338..d04c456 100644
--- a/src/FIRFilter.h
+++ b/src/FIRFilter.h
@@ -53,6 +53,9 @@ class FIRFilter : public PipelinedModCodec, public RemoteControllable
 {
 public:
     FIRFilter(const std::string& taps_file);
+    FIRFilter(const FIRFilter& other) = delete;
+    FIRFilter& operator=(const FIRFilter& other) = delete;
+    virtual ~FIRFilter();
 
     const char* name() override { return "FIRFilter"; }
 
diff --git a/src/GainControl.cpp b/src/GainControl.cpp
index dbb9464..5657fc2 100644
--- a/src/GainControl.cpp
+++ b/src/GainControl.cpp
@@ -73,6 +73,10 @@ GainControl::GainControl(size_t framesize,
     start_pipeline_thread();
 }
 
+GainControl::~GainControl()
+{
+    stop_pipeline_thread();
+}
 
 int GainControl::internal_process(Buffer* const dataIn, Buffer* dataOut)
 {
diff --git a/src/GainControl.h b/src/GainControl.h
index 92c905b..b4579cd 100644
--- a/src/GainControl.h
+++ b/src/GainControl.h
@@ -56,7 +56,7 @@ class GainControl : public PipelinedModCodec, public RemoteControllable
                     float normalise,
                     float varVariance);
 
-        virtual ~GainControl() = default;
+        virtual ~GainControl();
         GainControl(const GainControl&) = delete;
         GainControl& operator=(const GainControl&) = delete;
 
diff --git a/src/MemlessPoly.cpp b/src/MemlessPoly.cpp
index ae097c9..5d1c02d 100644
--- a/src/MemlessPoly.cpp
+++ b/src/MemlessPoly.cpp
@@ -99,6 +99,11 @@ MemlessPoly::MemlessPoly(const std::string& coefs_file, unsigned int num_threads
     start_pipeline_thread();
 }
 
+MemlessPoly::~MemlessPoly()
+{
+    stop_pipeline_thread();
+}
+
 void MemlessPoly::load_coefficients(const std::string &coefFile)
 {
     std::ifstream coef_fstream(coefFile.c_str());
diff --git a/src/MemlessPoly.h b/src/MemlessPoly.h
index 612934f..4c67d46 100644
--- a/src/MemlessPoly.h
+++ b/src/MemlessPoly.h
@@ -59,6 +59,9 @@ class MemlessPoly : public PipelinedModCodec, public RemoteControllable
 {
 public:
     MemlessPoly(const std::string& coefs_file, unsigned int num_threads);
+    MemlessPoly(const MemlessPoly& other) = delete;
+    MemlessPoly& operator=(const MemlessPoly& other) = delete;
+    virtual ~MemlessPoly();
 
     virtual const char* name() { return "MemlessPoly"; }
 
diff --git a/src/ModPlugin.cpp b/src/ModPlugin.cpp
index f86bfb2..d567a90 100644
--- a/src/ModPlugin.cpp
+++ b/src/ModPlugin.cpp
@@ -72,17 +72,7 @@ int ModOutput::process(
     return process(dataIn[0]);
 }
 
-PipelinedModCodec::PipelinedModCodec() :
-    ModCodec(),
-    m_input_queue(),
-    m_output_queue(),
-    m_metadata_fifo(),
-    m_running(false),
-    m_thread()
-{
-}
-
-PipelinedModCodec::~PipelinedModCodec()
+void PipelinedModCodec::stop_pipeline_thread()
 {
     m_input_queue.push({});
     if (m_thread.joinable()) {
diff --git a/src/ModPlugin.h b/src/ModPlugin.h
index 5635fca..e9cfa21 100644
--- a/src/ModPlugin.h
+++ b/src/ModPlugin.h
@@ -95,13 +95,6 @@ public:
 class PipelinedModCodec : public ModCodec, public ModMetadata
 {
 public:
-    PipelinedModCodec();
-    PipelinedModCodec(const PipelinedModCodec&) = delete;
-    PipelinedModCodec& operator=(const PipelinedModCodec&) = delete;
-    PipelinedModCodec(PipelinedModCodec&&) = delete;
-    PipelinedModCodec& operator=(PipelinedModCodec&&) = delete;
-    virtual ~PipelinedModCodec();
-
     virtual int process(Buffer* const dataIn, Buffer* dataOut) final;
     virtual const char* name() = 0;
 
@@ -111,6 +104,11 @@ protected:
     // Once the instance implementing PipelinedModCodec has been constructed,
     // it must call start_pipeline_thread()
     void start_pipeline_thread(void);
+    // To avoid race conditions on teardown, plugins must call
+    // stop_pipeline_thread in their destructor.
+    void stop_pipeline_thread(void);
+
+    // The real processing must be implemented in internal_process
     virtual int internal_process(Buffer* const dataIn, Buffer* dataOut) = 0;
 
 private:
@@ -121,7 +119,7 @@ private:
 
     std::deque<meta_vec_t> m_metadata_fifo;
 
-    std::atomic<bool> m_running;
+    std::atomic<bool> m_running = ATOMIC_VAR_INIT(false);
     std::thread m_thread;
     void process_thread(void);
 };