From 0b127d48fdfd612b8738c6d53c2fcc9e48debfa2 Mon Sep 17 00:00:00 2001
From: Joel Filho <joelpcfilho@gmail.com>
Date: Sun, 31 May 2020 13:57:15 -0300
Subject: [PATCH 1/3] CMake fix: using EXCLUDE_FROM_ALL

- preventing doctest target installation without relying on cache
---
 tests/CMakeLists.txt | 4 +---
 1 file changed, 1 insertion(+), 3 deletions(-)

diff --git a/tests/CMakeLists.txt b/tests/CMakeLists.txt
index 3bb76a9..ba4d40c 100644
--- a/tests/CMakeLists.txt
+++ b/tests/CMakeLists.txt
@@ -1,6 +1,4 @@
-set(DOCTEST_NO_INSTALL ON  CACHE BOOL "Do not install doctest with TDP by default")
-set(DOCTEST_WITH_TESTS OFF CACHE BOOL "Do not install build doctest's tests and examples")
-add_subdirectory(doctest)
+add_subdirectory(doctest EXCLUDE_FROM_ALL)
 
 file(GLOB TEST_FILES CONFIGURE_DEPENDS "*.cpp")
 

From cbd0eb83baa956a4e251f9a5a6fd5ad15b8b49a7 Mon Sep 17 00:00:00 2001
From: Joel Filho <joelpcfilho@gmail.com>
Date: Sun, 31 May 2020 14:05:47 -0300
Subject: [PATCH 2/3] Remove initialization from public interface

---
 include/tdp/pipeline.impl.hpp | 11 ++++++-----
 1 file changed, 6 insertions(+), 5 deletions(-)

diff --git a/include/tdp/pipeline.impl.hpp b/include/tdp/pipeline.impl.hpp
index 1fb9db4..0791160 100644
--- a/include/tdp/pipeline.impl.hpp
+++ b/include/tdp/pipeline.impl.hpp
@@ -230,6 +230,7 @@ struct pipeline<Queue, jtc::type_list<InputArgs...>, Stages...> final
 
   ~pipeline() { stop_threads(); }
 
+ private:
   template <std::size_t I>
   void init_intermediary_threads(std::tuple<Stages...>& stages) {
     using inputs = util::result_list_t<input_list_t, Stages...>;
@@ -332,11 +333,6 @@ struct pipeline<Queue, jtc::type_list<InputArgs...>, Stages...> final
     }
   }
 
- private:
-  std::atomic_bool _stop = false;
-  tuple_t _queues;
-  std::array<std::thread, N> _threads;
-
   void stop_threads() {
     // Set the "stop token" flag
     _stop = true;
@@ -355,6 +351,11 @@ struct pipeline<Queue, jtc::type_list<InputArgs...>, Stages...> final
       if (thread.joinable())
         thread.join();
   }
+
+ private:
+  std::atomic_bool _stop = false;
+  tuple_t _queues;
+  std::array<std::thread, N> _threads;
 };
 
 //-------------------------------------------------------------------------------------------------

From f1289a8d41b157fa3e22f330e1caca5ef897af71 Mon Sep 17 00:00:00 2001
From: Joel Filho <joelpcfilho@gmail.com>
Date: Sun, 7 Jun 2020 17:15:45 -0300
Subject: [PATCH 3/3] Pipeline idling interface (1)

Initial tentative for the idle interface
---
 TODO.md                              |   2 +-
 examples/00_introduction.cpp         |   5 +
 examples/01_consumer_threads.cpp     |  14 +-
 examples/07_reference_parameters.cpp |   6 +-
 include/tdp/pipeline.impl.hpp        | 189 ++++++++++++++++++++++++---
 tests/test_consumers.cpp             |   6 +-
 tests/test_policies.cpp              |   6 +-
 tests/test_producers.cpp             |   4 +
 8 files changed, 197 insertions(+), 35 deletions(-)

diff --git a/TODO.md b/TODO.md
index 65b2ebe..71a673f 100644
--- a/TODO.md
+++ b/TODO.md
@@ -12,7 +12,7 @@
 - [x] Non-locking `get()` interface
   - [x] Rename old interface to make it clearer it blocks
 - [x] Apply execution policies
-- [ ] `idle()` interface
+- [x] `idle()` interface
   - [x] Rename `running()` to remove ambiguity
 - [x] Use `std::invoke` wherever applicable
 - [x] Prohibit reference outputs (mutable lvalue reference parameters already invalid)
diff --git a/examples/00_introduction.cpp b/examples/00_introduction.cpp
index 4875b48..274735b 100644
--- a/examples/00_introduction.cpp
+++ b/examples/00_introduction.cpp
@@ -50,6 +50,11 @@ int main() {
     std::cout << "Pipeline is empty, can't get another result!\n";
   }
 
+  // We can verify all threads on the pipeline are waiting for input with idle()
+  // Note: idle() may return false if any pipeline stage hasn't registered its idle state yet.
+  // To guarantee the pipeline is idle before exitting, see the next example.
+  std::cout << "pipeline.idle() = " << std::boolalpha << pipeline.idle() << std::endl;
+
   // At the end, we don't need to do anything.
   // The pipeline stops all threads when its destructor is called.
 }
\ No newline at end of file
diff --git a/examples/01_consumer_threads.cpp b/examples/01_consumer_threads.cpp
index 9dffd62..23e3eab 100644
--- a/examples/01_consumer_threads.cpp
+++ b/examples/01_consumer_threads.cpp
@@ -35,13 +35,11 @@ int main() {
   pipeline.input(2, 3.5);
   pipeline.input(7, 0.75);
 
-  // We can wait a little, do anything else, and the production will be done.
-  // Sometimes we want to guarantee the pipeline has finished running.
-  // For that, we'll use the idle() member function.
-  // TODO
-  using namespace std::chrono_literals;
-  std::this_thread::sleep_for(100ms);
-
-  // The capture we used from the lambda!
+  // We can wait a little, do anything else, and the processing will be done.
+  // Sometimes we want to guarantee the pipeline has finished running before exitting.
+  // For that, we'll use the wait_until_idle() member function.
+  pipeline.wait_until_idle();
+
+  // The capture we used from the lambda shows us we have processed everything!
   std::cout << "print() has been called " << count << " times.\n";
 }
\ No newline at end of file
diff --git a/examples/07_reference_parameters.cpp b/examples/07_reference_parameters.cpp
index a41dcf4..b9bbb6f 100644
--- a/examples/07_reference_parameters.cpp
+++ b/examples/07_reference_parameters.cpp
@@ -33,7 +33,7 @@ int main() {
   pipeline.input("!dlroW olleH");
   pipeline.input("TACOCAT");
 
-  // We insert a delay before destruction, so the pipeline finishes execution.
-  using namespace std::chrono_literals;
-  std::this_thread::sleep_for(100ms);
+  // We wait until execution is complete with wait_until_idle()
+  // This way, we free the processor from scheduling this thread until then, and don't need to check pipeline.idle()
+  pipeline.wait_until_idle();
 }
\ No newline at end of file
diff --git a/include/tdp/pipeline.impl.hpp b/include/tdp/pipeline.impl.hpp
index 0791160..199cfa8 100644
--- a/include/tdp/pipeline.impl.hpp
+++ b/include/tdp/pipeline.impl.hpp
@@ -10,7 +10,9 @@
 
 #include <array>
 #include <atomic>
+#include <bitset>
 #include <functional>
+#include <limits>
 #include <thread>
 #include <tuple>
 #include <type_traits>
@@ -23,6 +25,16 @@
 
 namespace tdp::detail {
 
+//-------------------------------------------------------------------------------------------------
+// Idle Callback: Each thread notifies its idling state
+//-------------------------------------------------------------------------------------------------
+
+struct idle_callback {
+  virtual void set(std::size_t index) noexcept = 0;
+  virtual void reset(std::size_t index) noexcept = 0;
+  virtual ~idle_callback() = default;
+};
+
 //-------------------------------------------------------------------------------------------------
 // Processing threads
 //-------------------------------------------------------------------------------------------------
@@ -38,19 +50,31 @@ struct thread_worker<Queue, jtc::type_list<InputArgs...>, Callable,  //
   using input_t = std::tuple<InputArgs...>;
   using output_t = std::invoke_result_t<Callable, InputArgs...>;
 
+  const std::size_t _id;
+  idle_callback& _idle;
   Callable _f;
   Queue<input_t>& _input_queue;
   Queue<output_t>& _output_queue;
   const std::atomic_bool& _stop;
 
   void operator()() noexcept {
+    _idle.reset(_id);
     while (!_stop) {
-      auto val = _input_queue.pop_unless([&] { return _stop.load(); });
-      if (!val)
-        break;
-      auto&& res = std::apply(_f, std::move(*val));
-      _output_queue.push(std::move(res));
+      if (_input_queue.empty()) {
+        _idle.set(_id);
+        auto val = _input_queue.pop_unless([&] { return _stop.load(); });
+        if (!val)
+          break;
+        _idle.reset(_id);
+        auto&& res = std::apply(_f, std::move(*val));
+        _output_queue.push(std::move(res));
+      } else {
+        auto&& val = _input_queue.pop();
+        auto&& res = std::apply(_f, std::move(val));
+        _output_queue.push(std::move(res));
+      }
     }
+    _idle.set(_id);
     _output_queue.wake();
   }
 };
@@ -60,16 +84,27 @@ template <template <typename...> class Queue, typename Callable>
 struct thread_worker<Queue, jtc::type_list<>, Callable> {
   using output_t = std::invoke_result_t<Callable>;
 
+  const std::size_t _id;
+  idle_callback& _idle;
   Callable _f;
   Queue<output_t>& _output_queue;
   const std::atomic_bool& _pause;
   const std::atomic_bool& _stop;
 
   void operator()() noexcept {
+    _idle.reset(_id);
     while (!_stop) {
-      if (!_pause)
+      if (!_pause) {
         _output_queue.push(std::invoke(_f));
+      } else {
+        _idle.set(_id);
+        while (_pause && !_stop)
+          ;
+        if (!_stop)
+          _idle.reset(_id);
+      }
     }
+    _idle.set(_id);
     _output_queue.wake();
   }
 };
@@ -79,17 +114,28 @@ template <template <typename...> class Queue, typename Input, typename Callable>
 struct thread_worker<Queue, Input, Callable,                           //
     std::enable_if_t<!util::is_instance_of_v<Input, jtc::type_list>>,  //
     std::enable_if_t<std::is_same_v<std::invoke_result_t<Callable, Input>, void>>> {
+  const std::size_t _id;
+  idle_callback& _idle;
   Callable _f;
   Queue<Input>& _input_queue;
   const std::atomic_bool& _stop;
 
   void operator()() noexcept {
     while (!_stop) {
-      auto val = _input_queue.pop_unless([&] { return _stop.load(); });
-      if (!val)
-        break;
-      std::invoke(_f, std::move(*val));
+      _idle.reset(_id);
+      if (_input_queue.empty()) {
+        _idle.set(_id);
+        auto val = _input_queue.pop_unless([&] { return _stop.load(); });
+        if (!val)
+          break;
+        _idle.reset(_id);
+        std::invoke(_f, std::move(*val));
+      } else {
+        auto&& val = _input_queue.pop();
+        std::invoke(_f, std::move(val));
+      }
     }
+    _idle.set(_id);
   }
 };
 
@@ -100,17 +146,28 @@ struct thread_worker<Queue, jtc::type_list<InputArgs...>, Callable,  //
     std::enable_if_t<std::is_same_v<std::invoke_result_t<Callable, InputArgs...>, void>>> {
   using input_t = std::tuple<InputArgs...>;
 
+  const std::size_t _id;
+  idle_callback& _idle;
   Callable _f;
   Queue<input_t>& _input_queue;
   const std::atomic_bool& _stop;
 
   void operator()() noexcept {
+    _idle.reset(_id);
     while (!_stop) {
-      auto val = _input_queue.pop_unless([&] { return _stop.load(); });
-      if (!val)
-        break;
-      std::apply(_f, std::move(*val));
+      if (_input_queue.empty()) {
+        _idle.set(_id);
+        auto val = _input_queue.pop_unless([&] { return _stop.load(); });
+        if (!val)
+          break;
+        _idle.reset(_id);
+        std::apply(_f, std::move(*val));
+      } else {
+        auto&& val = _input_queue.pop();
+        std::apply(_f, std::move(val));
+      }
     }
+    _idle.set(_id);
   }
 };
 
@@ -121,19 +178,31 @@ struct thread_worker<Queue, Input, Callable,                           //
     std::enable_if_t<!std::is_same_v<std::invoke_result_t<Callable, Input>, void>>> {
   using output_t = std::invoke_result_t<Callable, Input>;
 
+  const std::size_t _id;
+  idle_callback& _idle;
   Callable _f;
   Queue<Input>& _input_queue;
   Queue<output_t>& _output_queue;
   const std::atomic_bool& _stop;
 
   void operator()() noexcept {
+    _idle.reset(_id);
     while (!_stop) {
-      auto val = _input_queue.pop_unless([&] { return _stop.load(); });
-      if (!val)
-        break;
-      auto&& res = std::invoke(_f, std::move(*val));
-      _output_queue.push(std::move(res));
+      if (_input_queue.empty()) {
+        _idle.set(_id);
+        auto val = _input_queue.pop_unless([&] { return _stop.load(); });
+        if (!val)
+          break;
+        _idle.reset(_id);
+        auto&& res = std::invoke(_f, std::move(*val));
+        _output_queue.push(std::move(res));
+      } else {
+        auto&& val = _input_queue.pop();
+        auto&& res = std::invoke(_f, std::move(val));
+        _output_queue.push(std::move(res));
+      }
     }
+    _idle.set(_id);
     _output_queue.wake();
   }
 };
@@ -190,6 +259,55 @@ struct pipeline_output {
 template <template <typename...> class Queue>
 struct pipeline_output<Queue, void> {};
 
+//-------------------------------------------------------------------------------------------------
+// A mechanism for suspending threads waiting for a pipeline's idle state
+//-------------------------------------------------------------------------------------------------
+
+template <std::size_t N>
+struct pipeline_idler final : idle_callback {
+  std::atomic<std::bitset<N>>& _flags;
+  mutable std::condition_variable _condition{};
+  mutable std::mutex _mutex{};
+
+  pipeline_idler(std::atomic<std::bitset<N>>& flags) noexcept : _flags(flags) {}
+
+  void set(std::size_t index) noexcept override {
+    auto old = _flags.load();
+    auto next = old;
+    do {
+      next = old;
+      next.set(index);
+    } while (!_flags.compare_exchange_weak(old, next));
+
+    if (_flags.load().all()) {
+      notify();
+    }
+  }
+
+  void reset(std::size_t index) noexcept override {
+    auto old = _flags.load();
+    auto next = old;
+    do {
+      next = old;
+      next.reset(index);
+    } while (!_flags.compare_exchange_weak(old, next));
+  }
+
+  void notify() const noexcept {
+    { std::unique_lock lock{_mutex}; }
+    _condition.notify_all();
+  }
+
+  template <typename F>
+  void wait(F&& pred) const noexcept {
+    {
+      std::unique_lock lock{_mutex};
+      _condition.wait(lock, std::forward<F>(pred));
+    }
+    _condition.notify_one();
+  }
+};
+
 //-------------------------------------------------------------------------------------------------
 // Pipeline system
 //-------------------------------------------------------------------------------------------------
@@ -230,6 +348,21 @@ struct pipeline<Queue, jtc::type_list<InputArgs...>, Stages...> final
 
   ~pipeline() { stop_threads(); }
 
+  [[nodiscard]] bool idle() const noexcept {
+    bool empty_queues = true;
+    util::tuple_foreach([&](const auto& queue) { empty_queues &= queue.empty(); }, _queues);
+    if constexpr (sizeof...(InputArgs) == 0) {
+      empty_queues &= !pipeline_input_t::producing();
+    } else {
+      empty_queues &= pipeline_input_t::input_is_empty();
+    }
+    return empty_queues && _idle_threads.load().all();
+  }
+
+  void wait_until_idle() const noexcept {
+    _idler.wait([this] { return idle(); });
+  }
+
  private:
   template <std::size_t I>
   void init_intermediary_threads(std::tuple<Stages...>& stages) {
@@ -239,6 +372,8 @@ struct pipeline<Queue, jtc::type_list<InputArgs...>, Stages...> final
     using callable_t = jtc::list_get_t<callables, I>;
 
     _threads[I] = std::thread(thread_worker<Queue, input_t, callable_t>{
+        I,
+        _idler,
         std::move(std::get<I>(stages)),
         std::get<I - 1>(_queues),
         std::get<I>(_queues),
@@ -261,6 +396,8 @@ struct pipeline<Queue, jtc::type_list<InputArgs...>, Stages...> final
     if constexpr (std::is_same_v<ret_t, void>) {
       // Consumer
       _threads[N - 1] = std::thread(thread_worker<Queue, input_t, callable_t>{
+          N - 1,
+          _idler,
           std::forward<T>(last),
           std::get<N - 2>(_queues),
           _stop,
@@ -268,6 +405,8 @@ struct pipeline<Queue, jtc::type_list<InputArgs...>, Stages...> final
     } else {
       // User output
       _threads[N - 1] = std::thread(thread_worker<Queue, input_t, callable_t>{
+          N - 1,
+          _idler,
           std::forward<T>(last),
           std::get<N - 2>(_queues),
           pipeline_output_t::_output_queue,
@@ -287,6 +426,8 @@ struct pipeline<Queue, jtc::type_list<InputArgs...>, Stages...> final
       if constexpr (N == 1) {
         // Producing directly to output
         _threads[0] = std::thread(thread_worker<Queue, input_t, callable_t>{
+            0,
+            _idler,
             std::forward<T>(first),
             pipeline_output_t::_output_queue,
             pipeline_input_t::_paused,
@@ -295,6 +436,8 @@ struct pipeline<Queue, jtc::type_list<InputArgs...>, Stages...> final
       } else {
         // Producing to another thread
         _threads[0] = std::thread(thread_worker<Queue, input_t, callable_t>{
+            0,
+            _idler,
             std::forward<T>(first),
             std::get<0>(_queues),
             pipeline_input_t::_paused,
@@ -308,6 +451,8 @@ struct pipeline<Queue, jtc::type_list<InputArgs...>, Stages...> final
         if constexpr (std::is_same_v<ret_t, void>) {
           // Consumer-only pipeline
           _threads[0] = std::thread(thread_worker<Queue, input_t, callable_t>{
+              0,
+              _idler,
               std::forward<T>(first),
               pipeline_input_t::_input_queue,
               _stop,
@@ -315,6 +460,8 @@ struct pipeline<Queue, jtc::type_list<InputArgs...>, Stages...> final
         } else {
           // Feeding directly to output
           _threads[0] = std::thread(thread_worker<Queue, input_t, callable_t>{
+              0,
+              _idler,
               std::forward<T>(first),
               pipeline_input_t::_input_queue,
               pipeline_output_t::_output_queue,
@@ -324,6 +471,8 @@ struct pipeline<Queue, jtc::type_list<InputArgs...>, Stages...> final
       } else {
         // Feeding to a second thread
         _threads[0] = std::thread(thread_worker<Queue, input_t, callable_t>{
+            0,
+            _idler,
             std::forward<T>(first),
             pipeline_input_t::_input_queue,
             std::get<0>(_queues),
@@ -356,6 +505,8 @@ struct pipeline<Queue, jtc::type_list<InputArgs...>, Stages...> final
   std::atomic_bool _stop = false;
   tuple_t _queues;
   std::array<std::thread, N> _threads;
+  std::atomic<std::bitset<N>> _idle_threads{std::numeric_limits<unsigned long long>::max()};
+  pipeline_idler<N> _idler{_idle_threads};
 };
 
 //-------------------------------------------------------------------------------------------------
diff --git a/tests/test_consumers.cpp b/tests/test_consumers.cpp
index 0828b9b..7cdbfef 100644
--- a/tests/test_consumers.cpp
+++ b/tests/test_consumers.cpp
@@ -27,7 +27,8 @@ TEST_CASE("Consumers") {
       pipeline.input(i);
     }
 
-    std::this_thread::sleep_for(100ms);
+    pipeline.wait_until_idle();
+    REQUIRE(pipeline.idle());
 
     REQUIRE_EQ(consumed.size(), input_count);
 
@@ -53,7 +54,8 @@ TEST_CASE("Single-thread Consumers") {
       pipeline->input(i, i);
     }
 
-    std::this_thread::sleep_for(100ms);
+    pipeline->wait_until_idle();
+    REQUIRE(pipeline->idle());
 
     REQUIRE_EQ(consumed.size(), input_count);
 
diff --git a/tests/test_policies.cpp b/tests/test_policies.cpp
index 36cdc3d..981582c 100644
--- a/tests/test_policies.cpp
+++ b/tests/test_policies.cpp
@@ -40,7 +40,8 @@ TEST_CASE("Blocking Triple Buffer policy") {
     pipeline.input(i, i + 2);
   }
 
-  std::this_thread::sleep_for(100ms);
+  pipeline.wait_until_idle();
+  REQUIRE(pipeline.idle());
 
   int processed = 0;
 
@@ -66,7 +67,8 @@ TEST_CASE("Lock-free Triple Buffer policy") {
     pipeline.input(i, i + 2);
   }
 
-  std::this_thread::sleep_for(100ms);
+  pipeline.wait_until_idle();
+  REQUIRE(pipeline.idle());
 
   int processed = 0;
 
diff --git a/tests/test_producers.cpp b/tests/test_producers.cpp
index 4386563..3e7a9cf 100644
--- a/tests/test_producers.cpp
+++ b/tests/test_producers.cpp
@@ -21,6 +21,7 @@ TEST_CASE("Producers") {
 
   SUBCASE("pause() actually pauses production, and producing() returns false") {
     std::this_thread::sleep_for(50ms);
+    REQUIRE_FALSE(pipeline.idle());
     pipeline.pause();
     REQUIRE_FALSE(pipeline.producing());
     REQUIRE_NE(produced, 0);
@@ -37,6 +38,7 @@ TEST_CASE("Producers") {
       std::this_thread::sleep_for(10ms);
 
       REQUIRE_NE(old_produced, produced);
+      REQUIRE_FALSE(pipeline.idle());
     }
   }
 
@@ -44,6 +46,7 @@ TEST_CASE("Producers") {
     std::this_thread::sleep_for(10ms);
     while (produced < 10)
       /* Wait until there's at least 10 produced items in the list */;
+    REQUIRE_FALSE(pipeline.idle());
     pipeline.pause();
     std::this_thread::sleep_for(10ms);
 
@@ -61,5 +64,6 @@ TEST_CASE("Producers") {
     }
 
     REQUIRE_EQ(produced, consumed);
+    REQUIRE(pipeline.idle());
   }
 }
\ No newline at end of file