chore: executor group

Author: vinniefalco

include/mrdox/Support/Thread.hpp

@@ -14,10 +14,15 @@
 
 #include <mrdox/Platform.hpp>
 #include <mrdox/Support/Error.hpp>
-#include <functional>
+#include <atomic>
+#include <condition_variable>
+#include <deque>
 #include <iterator>
+#include <memory>
 #include <mutex>
 #include <thread>
+#include <tuple>
+#include <type_traits>
 #include <utility>
 #include <vector>
 
@@ -33,6 +38,74 @@ class TaskGroup;
 
 //------------------------------------------------
 
+class unlock_guard
+{
+    std::mutex& m_;
+
+public:
+    explicit
+    unlock_guard(std::mutex& m)
+        : m_(m)
+    {
+        m_.unlock();
+    }
+
+    ~unlock_guard()
+    {
+        m_.lock();
+    }
+};
+
+//------------------------------------------------
+
+template<class>
+class any_callable;
+
+template<class R, class... Args>
+class any_callable<R(Args...)>
+{
+    struct base
+    {
+        virtual ~base() = default;
+        virtual R invoke(Args&&...args) = 0;
+    };
+
+    std::unique_ptr<base> p_;
+
+public:
+    any_callable() = delete;
+
+    template<class Callable>
+    requires std::is_invocable_r_v<R, Callable, Args...>
+    any_callable(Callable&& f)
+    {
+        class impl : public base
+        {
+            Callable f_;
+
+        public:
+            explicit impl(Callable&& f)
+                : f_(std::forward<Callable>(f))
+            {
+            }
+
+            R invoke(Args&&... args) override
+            {
+                return f_(std::forward<Args>(args)...);
+            }
+        };
+
+        p_ = std::make_unique<impl>(std::forward<Callable>(f));
+    }
+
+    R operator()(Args&&...args) const
+    {
+        return p_->invoke(std::forward<Args>(args)...);
+    }
+};
+
+//------------------------------------------------
+
 /** A pool of threads for executing work concurrently.
 */
 class MRDOX_VISIBLE
@@ -43,6 +116,11 @@ class MRDOX_VISIBLE
     friend class TaskGroup;
 
 public:
+    template<class Agent> struct arg_ty { using type = Agent; };
+    template<class Agent> struct arg_ty<Agent&> { using type =
+        std::conditional_t< std::is_const_v<Agent>, Agent, Agent&>; };
+    template<class Agent> using arg_t = typename arg_ty<Agent>::type;
+
     /** Destructor.
     */
     MRDOX_DECL
@@ -82,8 +160,12 @@ class MRDOX_VISIBLE
         The signature of the submitted function
         object should be `void(void)`.
     */
+    template<class F>
     void
-    async(std::function<void(void)> f);
+    async(F&& f)
+    {
+        post(std::forward<F>(f));
+    }
 
     /** Invoke a function object for each element of a range.
     */
@@ -95,6 +177,9 @@ class MRDOX_VISIBLE
     MRDOX_DECL
     void
     wait();
+
+private:
+    MRDOX_DECL void post(any_callable<void(void)>);
 };
 
 //------------------------------------------------
@@ -120,14 +205,21 @@ class MRDOX_VISIBLE
         The signature of the submitted function
         object should be `void(void)`.
     */
+    template<class F>
     void
-    async(std::function<void(void)> f);
+    async(F&& f)
+    {
+        post(std::forward<F>(f));
+    }
 
     /** Block until all work has completed.
     */
     MRDOX_DECL
     void
     wait();
+
+private:
+    MRDOX_DECL void post(any_callable<void(void)>);
 };
 
 //------------------------------------------------
@@ -153,48 +245,142 @@ forEach(
 
 //------------------------------------------------
 
-#if 0
 /** A set of execution agents for performing concurrent work.
 */
-template<class... Args>
-class ExecutionGroup
+template<class Agent>
+class ExecutorGroup
 {
-    std::vector<Agent*> agents_;
-    std::vector<std::function<void(Agent&)>> work_;
-    ThreadPool threadPool_;
-    TaskGroup taskGroup_;
+    struct Impl
+    {
+        std::mutex mutex_;
+        std::condition_variable cv_;
+    };
+
+    ThreadPool& threadPool_;
+    std::unique_ptr<Impl> impl_;
+    std::vector<std::unique_ptr<Agent>> agents_;
+    std::deque<any_callable<void(Agent&)>> work_;
+    std::size_t busy_ = 0;
 
 public:
-    template<class Agents>
+    template<class T>
+    using arg_t = ThreadPool::arg_t<T>;
+
+    ExecutorGroup(ExecutorGroup const&) = delete;
+    ExecutorGroup& operator=(ExecutorGroup&&) = delete;
+    ExecutorGroup& operator=(ExecutorGroup const&) = delete;
+    ExecutorGroup(ExecutorGroup&&) = default;
+
     explicit
-    Executors(
-        Agents&& agents,
-        ThreadPool &threadPool)
+    ExecutorGroup(
+        ThreadPool& threadPool) noexcept
         : threadPool_(threadPool)
+        , impl_(std::make_unique<Impl>())
+    {
+    }
+
+    /** Construct a new agent in the group.
+
+        The behavior is undefined if there is
+        any outstanding work or busy threads.
+    */
+    template<class... Args>
+    void
+    emplace(Args&&... args)
     {
-        agents_.reserve(std::distance(
-            std::begin(agents), std::end(agents)));
-        for(auto& agent : agents)
-            agents_.push_back(&agent);
+        agents_.emplace_back(std::make_unique<Agent>(
+            std::forward<Args>(args)...));
     }
 
     /** Submit work to be executed.
 
-        The signature of the submitted function
-        object should be `void(Agent&)`.
+        The function object must have this
+        equivalent signature:
+        @code
+        void( Agent&, Args... );
+        @endcode
     */
+    template<class F, class... Args>
     void
-    async(std::function<void(void)> f);
+    async(F&& f, Args&&... args)
+    {
+        static_assert(std::is_invocable_v<F, Agent&, arg_t<Args>...>);
+        std::unique_lock<std::mutex> lock(impl_->mutex_);
+        work_.emplace_back(
+            [
+                f = std::forward<F>(f),
+                args = std::tuple<arg_t<Args>...>(args...)
+            ](Agent& agent)
+            {
+                std::apply(f, std::tuple_cat(
+                    std::tuple<Agent&>(agent),
+                    std::move(args)));
+            });
+        if(agents_.empty())
+            return;
+        run(std::move(lock));
+    }
 
     /** Block until all work has completed.
     */
     void
     wait()
     {
-        taskGroup_.wait();
+        std::unique_lock<std::mutex> lock(impl_->mutex_);
+        impl_->cv_.wait(lock,
+            [&]
+            {
+                return work_.empty() && busy_ == 0;
+            });
+    }
+
+private:
+    class scoped_agent
+    {
+        ExecutorGroup& group_;
+        std::unique_ptr<Agent> agent_;
+
+    public:
+        scoped_agent(
+            ExecutorGroup& group,
+            std::unique_ptr<Agent> agent) noexcept
+            : group_(group)
+            , agent_(std::move(agent))
+        {
+        }
+
+        ~scoped_agent()
+        {
+            --group_.busy_;
+            group_.agents_.emplace_back(std::move(agent_));
+            group_.impl_->cv_.notify_all();
+        }
+    };
+
+    void
+    run(std::unique_lock<std::mutex> lock)
+    {
+        std::unique_ptr<Agent> agent(std::move(agents_.back()));
+        agents_.pop_back();
+        ++busy_;
+
+        threadPool_.async(
+            [this, agent = std::move(agent)]() mutable
+            {
+                scoped_agent scope(*this, std::move(agent));
+                std::unique_lock<std::mutex> lock(impl_->mutex_);
+                for(;;)
+                {
+                    if(work_.empty())
+                        break;
+                    any_callable<void(Agent&)> work(std::move(work_.front()));
+                    work_.pop_front();
+                    unlock_guard unlock(impl_->mutex_);
+                    work(*agent);
+                }
+            });
     }
 };
-#endif
 
 } // mrdox
 } // clang

source/Support/Thread.cpp

@@ -57,17 +57,23 @@ getThreadCount() const noexcept
 
 void
 ThreadPool::
-async(
-    std::function<void(void)> f)
+wait()
 {
-    impl_->async(std::move(f));
+    impl_->wait();
 }
 
 void
 ThreadPool::
-wait()
+post(
+    any_callable<void(void)> f)
 {
-    impl_->wait();
+    auto sp = std::make_shared<
+        any_callable<void(void)>>(std::move(f));
+    impl_->async(
+        [sp]
+        {
+            (*sp)();
+        });
 }
 
 //------------------------------------------------
@@ -90,17 +96,23 @@ TaskGroup(
 
 void
 TaskGroup::
-async(
-    std::function<void(void)> f)
+wait()
 {
-    impl_->async(std::move(f));
+    impl_->wait();
 }
 
 void
 TaskGroup::
-wait()
+post(
+    any_callable<void(void)> f)
 {
-    impl_->wait();
+    auto sp = std::make_shared<
+        any_callable<void(void)>>(std::move(f));
+    impl_->async(
+        [sp]
+        {
+            (*sp)();
+        });
 }
 
 } // mrdox