Some experimental fixes

Author: Stringy

collector/lib/Pipeline.h

@@ -39,22 +39,30 @@ class Queue {
 
   void push(const T& elem) {
     auto lock = write_lock();
-    return inner_.push(elem);
+    auto e = elem;
+    inner_.push(std::move(e));
+    state_changed_.notify_one();
   }
 
   void push(T&& elem) {
     auto lock = write_lock();
-    return inner_.push(elem);
+    inner_.push(elem);
+    state_changed_.notify_one();
   }
 
   template <class... Args>
   decltype(auto) emplace(Args&&... args) {
     auto lock = write_lock();
-    return inner_.emplace(std::forward<Args>(args)...);
+    decltype(auto) out = inner_.emplace(std::forward<Args>(args)...);
+    state_changed_.notify_one();
+    return out;
   }
 
   T pop() {
     auto lock = write_lock();
+    if (empty()) {
+      state_changed_.wait(lock, [this]() { return empty(); });
+    }
     T data = inner_.front();
     inner_.pop();
     return data;
@@ -72,14 +80,19 @@ class Queue {
   std::queue<T> inner_;
 
   mutable std::shared_mutex mx_;
+  mutable std::condition_variable_any state_changed_;
 };
 
 template <class T>
 class Producer {
  public:
   Producer(std::shared_ptr<Queue<T>>& output) : output_(output) {}
 
-  ~Producer() { Stop(); }
+  ~Producer() {
+    if (thread_.running()) {
+      Stop();
+    }
+  }
 
   virtual T next() = 0;
 
@@ -108,7 +121,11 @@ class Consumer {
  public:
   Consumer(std::shared_ptr<Queue<T>>& input) : input_(input) {}
 
-  ~Consumer() { Stop(); }
+  ~Consumer() {
+    if (thread_.running()) {
+      Stop();
+    }
+  }
 
   virtual void handle(const T& event) = 0;
 

collector/lib/StoppableThread.cpp

@@ -1,6 +1,8 @@
 #include "StoppableThread.h"
 
+#include <chrono>
 #include <iostream>
+#include <thread>
 #include <unistd.h>
 
 #include "Utility.h"
@@ -44,6 +46,10 @@ void StoppableThread::Stop() {
     }
     break;
   }
+  if (!thread_->joinable()) {
+    CLOG(WARNING) << "thread not yet joinable...";
+    std::this_thread::sleep_for(std::chrono::milliseconds(500));
+  }
   thread_->join();
   thread_.reset();
   int rv = close(stop_pipe_[0]);

collector/test/PipelineTests.cpp

@@ -17,7 +17,7 @@ class IntProducer : public Producer<int> {
   int next() override {
     n_++;
     if (n_ > limit_) {
-      Stop();
+      return limit_;
     }
     return n_;
   }
@@ -54,7 +54,7 @@ class EvenIntFilter : public Filter<int> {
 };
 
 TEST(PipelineTests, TestBasic) {
-  std::shared_ptr<Queue<int>> queue;
+  std::shared_ptr<Queue<int>> queue = std::make_shared<Queue<int>>();
 
   std::vector<int> output;