* implemented executor for split_every and split_evenly parallelism strategy

include/ecs/context/base.inl

@@ -106,7 +106,7 @@ namespace detail {
         auto& bitset = meta.bitset();
         auto& cmd    = meta.storage_meta_data();
         if (!bitset.has_component(ct))
-            throw std::invalid_argument("entity does not contain the requested compnent");
+            throw std::invalid_argument("entity does not contain the requested component");
         return _components.get(ct, handle, cmd);
     }
 

include/ecs/core/system/data_proxy.h

@@ -1,7 +1,9 @@
 #pragma once
 
 #include "./data_proxy/base.h"
+#include "./data_proxy/multi.h"
 #include "./data_proxy/single.h"
 
 #include "./data_proxy/base.inl"
+#include "./data_proxy/multi.inl"
 #include "./data_proxy/single.inl"

include/ecs/core/system/data_proxy/multi.h

@@ -0,0 +1,112 @@
+#pragma once
+
+#include <ecs/config.h>
+#include <ecs/core/system/data_proxy/base.h>
+
+namespace ecs {
+namespace core {
+namespace system {
+namespace data_proxy {
+
+    /**
+     * data proxy to execute system non-parallel
+     *
+     * @tparam T_context    context type of the ECS environment
+     * @tparam T_instance   system instance type
+     */
+    template<typename T_context, typename T_instance>
+    struct multi
+        : public base<T_context, T_instance>
+    {
+    private:
+        using context_type  = T_context;
+        using instance_type = T_instance;
+        using base_type     = base<context_type, instance_type>;
+
+    public: // private:
+        size_t  _index;
+        size_t  _begin;
+        size_t  _end;
+
+    public:
+        /**
+         * create the multi data proxy
+         *
+         * @param p_context     context of the ECS environment
+         * @param p_instance    system instance
+         * @param p_index       index of this data proxy
+         * @param p_begin       index of the first entity to process
+         * @param p_end         index of the first entity not to process
+         */
+        inline multi(context_type& p_context, instance_type& p_instance, size_t p_index, size_t p_begin, size_t p_end);
+
+        /**
+         * execute the given function for all entities handled by this data proxy
+         *
+         * @tparam T_func   function type to execute
+         *
+         * @param func      function to execute
+         */
+        template<typename T_func>
+        inline void for_entities(T_func&& func) const;
+
+        /**
+         * execute the given function for all entities not handled by this data proxy
+         *
+         * @tparam T_func   function type to execute
+         *
+         * @param func      function to execute
+         */
+        template<typename T_func>
+        void for_other_entities(T_func&& func) const;
+
+        /**
+         * execute the given function for all entities
+         *
+         * @tparam T_func   function type to execute
+         *
+         * @param func      function to execute
+         */
+        template<typename T_func>
+        void for_all_entities(T_func&& func) const;
+
+        /**
+         * get the number of entities handled by this data proxy
+         *
+         * @return number of entities handle by this data proxy
+         */
+        inline size_t entity_count() const;
+
+        /**
+         * get the number of entities not handled by this data proxy
+         *
+         * @return number of entities not handle by this data proxy
+         */
+        inline size_t other_entity_count() const;
+
+        /**
+         * get the number of all entities
+         *
+         * @return number of all entities
+         */
+        inline size_t all_entity_count() const;
+    };
+
+    /**
+     * create a multi data proxy (data proxy that does not execute in parallel)
+     *
+     * @tparam T_context    context type
+     * @tparam T_instance   instance type
+     *
+     * @param context       context
+     * @param instance      instance
+     * @param index         index of this data proxy
+     * @param begin         index of the first entity to process
+     * @param end           index of the first entity not to process
+     *
+     * @return              multi data proxy
+     */
+    template<typename T_context, typename T_instance>
+    constexpr decltype(auto) make_multi(T_context&& context, T_instance&& instance, size_t index, size_t begin, size_t end);
+
+} } } }

include/ecs/core/system/data_proxy/multi.inl

@@ -0,0 +1,96 @@
+#pragma once
+
+#include <ecs/core/system/data_proxy/multi.h>
+
+namespace ecs {
+namespace core {
+namespace system {
+namespace data_proxy {
+
+    /* multi */
+
+    template<typename T_context, typename T_instance>
+    inline multi<T_context, T_instance>
+        ::multi(context_type& p_context, instance_type& p_instance, size_t p_index, size_t p_begin, size_t p_end)
+            : base_type (p_context, p_instance)
+            , _index    (p_index)
+            , _begin    (p_begin)
+            , _end      (p_end)
+        { }
+
+    template<typename T_context, typename T_instance>
+    template<typename T_func>
+    inline void multi<T_context, T_instance>
+        ::for_entities(T_func&& func) const
+    {
+        for (size_t i = _begin; i < _end; ++i)
+        {
+            std::forward<T_func>(func)(this->instance.subscribed()[i]);
+        }
+    }
+
+    template<typename T_context, typename T_instance>
+    template<typename T_func>
+    inline void multi<T_context, T_instance>
+        ::for_other_entities(T_func&& func) const
+    {
+        auto& subscribed  = this->instance.subscribed();
+        auto  total_count = this->instance.subscribed_count();
+        for (size_t i = 0; i < _begin; ++i)
+        {
+            std::forward<T_func>(func)(subscribed()[i]);
+        }
+        for (size_t i = _end; i < total_count; ++i)
+        {
+            std::forward<T_func>(func)(subscribed()[i]);
+        }
+    }
+
+    template<typename T_context, typename T_instance>
+    template<typename T_func>
+    inline void multi<T_context, T_instance>
+        ::for_all_entities(T_func&& func) const
+    {
+        for (auto& e : this->instance.subscribed())
+        {
+            std::forward<T_func>(func)(e);
+        }
+    }
+
+    template<typename T_context, typename T_instance>
+    inline size_t multi<T_context, T_instance>
+        ::entity_count() const
+    {
+        return _end - _begin;
+    }
+
+    template<typename T_context, typename T_instance>
+    inline size_t multi<T_context, T_instance>
+        ::other_entity_count() const
+    {
+        return all_entity_count() - entity_count();
+    }
+
+    template<typename T_context, typename T_instance>
+    inline size_t multi<T_context, T_instance>
+        ::all_entity_count() const
+    {
+        return this->instance.subscribed_count();
+    }
+
+    /* make_multi */
+
+    template<typename T_context, typename T_instance>
+    constexpr decltype(auto) make_multi(T_context&& context, T_instance&& instance, size_t index, size_t begin, size_t end)
+    {
+        using context_type  = mp::decay_t<T_context>;
+        using instance_type = mp::decay_t<T_instance>;
+        return multi<context_type, instance_type>(
+            std::forward<T_context>(context),
+            std::forward<T_instance>(instance),
+            index,
+            begin,
+            end);
+    }
+
+} } } }

include/ecs/core/system/data_proxy/single.h

@@ -34,7 +34,7 @@ namespace data_proxy {
          * @param func      function to execute
          */
         template<typename T_func>
-        void for_entities(T_func&& func) const;
+        inline void for_entities(T_func&& func) const;
 
         /**
          * execute the given function for all entities not handled by this data proxy
@@ -44,7 +44,7 @@ namespace data_proxy {
          * @param func      function to execute
          */
         template<typename T_func>
-        void for_other_entities(T_func&& func) const;
+        inline void for_other_entities(T_func&& func) const;
 
         /**
          * execute the given function for all entities
@@ -54,7 +54,7 @@ namespace data_proxy {
          * @param func      function to execute
          */
         template<typename T_func>
-        void for_all_entities(T_func&& func) const;
+        inline void for_all_entities(T_func&& func) const;
 
         /**
          * get the number of entities handled by this data proxy

include/ecs/core/system/data_proxy/single.inl

@@ -11,24 +11,24 @@ namespace data_proxy {
 
     template<typename T_context, typename T_instance>
     template<typename T_func>
-    void single<T_context, T_instance>
+    inline void single<T_context, T_instance>
         ::for_entities(T_func&& func) const
     {
         for (auto& e : this->instance.subscribed())
         {
-            func(e);
+            std::forward<T_func>(func)(e);
         }
     }
 
     template<typename T_context, typename T_instance>
     template<typename T_func>
-    void single<T_context, T_instance>
+    inline void single<T_context, T_instance>
         ::for_other_entities(T_func&& func) const
         { }
 
     template<typename T_context, typename T_instance>
     template<typename T_func>
-    void single<T_context, T_instance>
+    inline void single<T_context, T_instance>
         ::for_all_entities(T_func&& func) const
     {
         for_entities(std::forward<T_func>(func));

include/ecs/core/system/instance.h

@@ -137,29 +137,6 @@ namespace system {
             clear_and_prepare(n);
         }
 
-        template<typename T_context, typename T_func>
-        inline void prepare_and_wait_subtasks(T_context& context, size_t n, T_func& func)
-        {
-            assert(n > 0);
-            clear_and_prepare(n);
-            utils::counter_blocker b(n-1);
-
-            auto run_in_seperate_thread = [this, &context, &b](auto& f) {
-                return [this, &b, &context, &f](auto&&... xs) {
-                    context.post_in_thread_pool([&f, &b, xs...]() {
-                        ecs_make_scope_guard([&b](){
-                            b.decrement();
-                        });
-                        f(xs...);
-                    });
-                };
-            };
-
-            b.execute_and_wait([&func, &run_in_seperate_thread]{
-                func(run_in_seperate_thread);
-            });
-        }
-
         template<typename T_context, typename T_func>
         inline void execute(T_context& context, T_func&& func)
             { _executor(context, *this, std::forward<T_func>(func)); }

include/ecs/core/system/parallelism/composer/fixed_threshold.h

@@ -20,7 +20,7 @@ namespace parallelism {
         template<typename T_context, typename T_instance, typename T_func>
         inline void operator()(T_context& context, T_instance& instance, T_func&& func) const
         {
-            bool threshold_reached = false; // TODO
+            bool threshold_reached = instance.subscribed_count() >= parameters_type::get_threshold();
             if (!threshold_reached) _strategy_lower  (instance, context, std::forward<T_func>(func));
             else                    _strategy_greater(instance, context, std::forward<T_func>(func));
         }

include/ecs/core/system/parallelism/strategy/none.h

@@ -27,7 +27,7 @@ namespace parallelism {
             {
                 instance.prepare_states(1);
                 auto data = data_proxy::make_single(context, instance);
-                func(data);
+                std::forward<T_func>(func)(data);
             }
         };
 
@@ -58,7 +58,7 @@ namespace parallelism {
             #endif
 
             executor_proxy<T_context, T_instance> ep { context, instance };
-            func(instance, ep);
+            std::forward<T_func>(func)(instance, ep);
         }
     };
 

include/ecs/core/system/parallelism/strategy/split_base.h

@@ -0,0 +1,66 @@
+#pragma once
+
+#include <ecs/config.h>
+#include <ecs/core/system/data_proxy.h>
+
+namespace ecs {
+namespace core {
+namespace system {
+namespace parallelism {
+
+    struct split_base
+    {
+    public:
+        template<typename T_context, typename T_instance>
+        struct executor_proxy
+        {
+        public:
+            using context_type  = T_context;
+            using instance_type = T_instance;
+
+            context_type&   context;
+            instance_type&  instance;
+            size_t          split_count;
+            size_t          per_split;
+
+        public:
+            template<typename T_func>
+            inline void for_subtasks(T_func&& func)
+            {
+                auto total_count = instance.subscribed_count();
+
+                instance.prepare_states(split_count);
+                utils::counter_blocker counter(split_count - 1);
+
+                /* execute the splitted subtask */
+                auto execute = [this, &func](size_t index, size_t from, size_t to)
+                {
+                    auto data = data_proxy::make_multi(context, instance, index, from, to);
+                    func(data);
+                };
+
+                /* create subtasks */
+                for (size_t i = 0; i < split_count - 1; ++i)
+                {
+                    context.post_in_thread_pool([this, &counter, &execute, i](auto){
+                        ecs_make_scope_guard([&counter](){
+                            counter.decrement();
+                        });
+                        auto beg =  i      * per_split;
+                        auto end = (i + 1) * per_split;
+                        execute(i, beg, end);
+                    });
+                }
+
+                /* execute and wait */
+                counter.execute_and_wait([this, &execute, total_count]{
+                    auto index  = split_count - 1;
+                    auto beg    = index * per_split;
+                    auto end    = total_count;
+                    execute(index, beg, end);
+                });
+            }
+        };
+    };
+
+} } } }

include/ecs/core/system/parallelism/strategy/split_evenly.h

@@ -1,6 +1,7 @@
 #pragma once
 
 #include <ecs/config.h>
+#include <ecs/core/system/parallelism/strategy/split_base.h>
 
 namespace ecs {
 namespace core {
@@ -9,14 +10,19 @@ namespace parallelism {
 
     template<typename T_parameters>
     struct split_evenly
+        : private split_base
     {
+    public:
         using parameters_type = T_parameters;
 
         template<typename T_context, typename T_instance, typename T_func>
         inline void operator()(T_context& context, T_instance& instance, T_func&& func) const
         {
-            // TODO
-            // auto split_count = parameters_type::get_split_count();
+            auto split_count = parameters_type::get_split_count();
+            auto total_count = instance.subscribed_count();
+            auto per_split   = total_count / split_count;
+            executor_proxy<T_context, T_instance> ep { context, instance, split_count, per_split };
+            func(instance, ep);
         }
     };
 

include/ecs/core/system/parallelism/strategy/split_every.h

@@ -1,6 +1,7 @@
 #pragma once
 
 #include <ecs/config.h>
+#include <ecs/core/system/parallelism/strategy/split_base.h>
 
 namespace ecs {
 namespace core {
@@ -9,14 +10,19 @@ namespace parallelism {
 
     template<typename T_parameters>
     struct split_every
+        : private split_base
     {
         using parameters_type = T_parameters;
 
         template<typename T_context, typename T_instance, typename T_func>
         inline void operator()(T_context& context, T_instance& instance, T_func&& func) const
         {
-            // TODO
-            // auto per_split = parameters_type::get_per_split_count();
+            auto per_split   = parameters_type::get_per_split_count();
+            auto total_count = instance.subscribed_count();
+            auto split_count = (total_count + per_split - 1) / per_split;
+
+            executor_proxy<T_context, T_instance> ep { context, instance, split_count, per_split };
+            func(instance, ep);
         }
     };
 

include/ecs/core/utils/counter_blocker.inl

@@ -1,5 +1,6 @@
 #pragma once
 
+#include <iostream> // TODO debug!
 #include <ecs/core/utils/counter_blocker.h>
 
 namespace ecs {

include/ecs/core/utils/thread_pool/worker.inl

@@ -25,7 +25,7 @@ namespace utils {
                 _own_tasks.pop();
                 return true;
             }
-            if (_all_tasks.wait_dequeue_timed(t, 0))
+            if (_all_tasks.wait_dequeue_timed(t, 0)) // TODO: this will sometimes return false, even if the queue is not yet empty
             {
                 return true;
             }

include/ecs/signature/system/parallelism/strategy/split_every.h

@@ -15,7 +15,7 @@ namespace parallelism {
         {
             struct parameters
             {
-                static constexpr size_t get_split_count() noexcept
+                static constexpr size_t get_per_split_count() noexcept
                     { return T_split::value; }
             };
 

test/dummy.cpp

@@ -64,6 +64,8 @@ namespace test
 
 using namespace test;
 
+namespace hana = ::boost::hana;
+
 namespace c = component;
 namespace ct = c::tag;
 namespace cs = ecs::signature::component;
@@ -95,11 +97,13 @@ constexpr decltype(auto) cs_list = csl::make(
 
 constexpr decltype(auto) ss_accelerate =
     ss::make(st::accelerate)
+        // .parallelism(ss::parallelism::split_evenly(hana::size_c<3>))
         .read(ct::acceleration)
         .write(ct::velocity);
 
 constexpr decltype(auto) ss_move =
     ss::make(st::move)
+        .parallelism(ss::parallelism::split_every(hana::size_c<3>))
         .read(ct::velocity)
         .write(ct::position);
 
@@ -126,6 +130,31 @@ double frand(double min, double max)
     return min + f * (max - min);
 }
 
+inline void log_system_execution(const std::string& system)
+{
+    std::ostringstream os;
+    os  << system
+        << " (thread=" << std::this_thread::get_id()
+        << ")"
+        << std::endl;
+    std::cout << os.str();
+}
+
+template<typename T_data_proxy>
+inline void log_system_execution(const std::string& system, const T_data_proxy& data)
+{
+    std::ostringstream os;
+    os  << system
+        << " (thread=" << std::this_thread::get_id()
+        << ", index=" << data._index
+        << ", begin=" << data._begin
+        << ", end=" << data._end
+        << ", count=" << data.entity_count()
+        << ")"
+        << std::endl;
+    std::cout << os.str();
+}
+
 TEST(DummyTest, fuu)
 {
     srand(static_cast<unsigned int>(time(nullptr)));
@@ -157,7 +186,7 @@ TEST(DummyTest, fuu)
             proxy.execute_systems()(
                 sea::tags(st::accelerate)
                     .for_subtasks([](auto& s, auto& data){
-                        std::cout << "accelerate " << std::this_thread::get_id() << std::endl;
+                        log_system_execution("accelerate");
                         data.for_entities([&data](auto& handle) {
                             auto& velocity     = data.get_component(ct::velocity, handle);
                             auto& acceleration = data.get_component(ct::acceleration, handle);
@@ -168,7 +197,7 @@ TEST(DummyTest, fuu)
                     }),
                 sea::tags(st::move)
                     .for_subtasks([](auto& s, auto& data){
-                        std::cout << "move " << std::this_thread::get_id() << std::endl;
+                        log_system_execution("move", data);
                         data.for_entities([&data](auto& handle) {
                             auto& position = data.get_component(ct::position, handle);
                             auto& velocity = data.get_component(ct::velocity, handle);
@@ -179,7 +208,7 @@ TEST(DummyTest, fuu)
                     }),
                 sea::tags(st::render)
                     .for_subtasks([](auto& s, auto& data){
-                        std::cout << "render " << std::this_thread::get_id() << std::endl;
+                        log_system_execution("render");
                         data.for_entities([&data](auto& handle) {
                             auto& position     = data.get_component(ct::position, handle);
                             auto& velocity     = data.get_component(ct::velocity, handle);