add opt-in, simple parallelisation to model base class

maartenscholl · maartenscholl · commit de701bfe98d9 · 2020-10-20T00:50:11.000+01:00
diff --git a/esl/agent.hpp b/esl/agent.hpp
@@ -75,7 +75,7 @@ namespace esl {
         ///
         /// \param rhs
         /// \return
-        [[nodiscard]] constexpr bool operator!=(const agent &rhs) const
+        [[nodiscard]] constexpr bool operator != (const agent &rhs) const
         {
             return !(*this == rhs);
         }
diff --git a/esl/computation/environment.cpp b/esl/computation/environment.cpp
@@ -126,21 +126,27 @@ namespace esl::computation {
     size_t environment::send_messages(simulation::model &simulation)
     {
         size_t messages_ = 0;
-        for(auto &[i, a] : simulation.agents.local_agents_) {
-            (void) i;
-            for(const auto &m: a->outbox) {
-                auto iterator_ =
-                    simulation.agents.local_agents_.find(m->recipient);
-                if(simulation.agents.local_agents_.end() == iterator_) {
-                    // not in distributed mode, and no local agent matching
-                    throw std::logic_error("agent not found "
-                                          + m->recipient.representation());
+        if(simulation.threads <= 1) {
+            for(auto &[i, a] : simulation.agents.local_agents_) {
+                (void)i;
+                for(const auto &m : a->outbox) {
+                    auto iterator_ =
+                        simulation.agents.local_agents_.find(m->recipient);
+                    if(simulation.agents.local_agents_.end() == iterator_) {
+                        // not in distributed mode, and no local agent matching
+                        throw std::logic_error("agent not found "
+                                               + m->recipient.representation());
+                    }
+                    iterator_->second->inbox.insert({m->received, m});
+                    ++messages_;
                 }
-                iterator_->second->inbox.insert({m->received, m});
-                ++messages_;
+                a->outbox.clear();
+                a->outbox.shrink_to_fit();
             }
-            a->outbox.clear();
-            a->outbox.shrink_to_fit();
+        }else{
+
+
+
         }
         return messages_;
     }
diff --git a/esl/simulation/entity.hpp b/esl/simulation/entity.hpp
@@ -86,7 +86,8 @@ namespace esl {
         ///
         ///
         ///
-        entity() : entity(identity<entity_type_>())
+        entity()
+        : entity(identity<entity_type_>())
         {
 
         }
@@ -123,7 +124,6 @@ namespace esl {
         }
 
     public:
-
         constexpr bool operator == (const entity_type_ &operand) const
         {
             return this->identifier == operand.identifier;
diff --git a/esl/simulation/model.cpp b/esl/simulation/model.cpp
@@ -26,10 +26,12 @@
 
 #include <chrono>
 using std::chrono::high_resolution_clock;
+#include <numeric>
 
 #include <esl/agent.hpp>
 #include <esl/computation/environment.hpp>
 #include <esl/data/log.hpp>
+#include <esl/quantity.hpp>
 
 
 namespace esl::simulation {
@@ -44,6 +46,7 @@ namespace esl::simulation {
         , sample(parameters.get<std::uint64_t>("sample"))
         , agents(e)
         , verbosity(parameters.get<std::uint64_t>("verbosity"))
+        , threads( std::max(1u, parameters.get<unsigned int>("threads")))
     {
 
     }
@@ -65,35 +68,41 @@ namespace esl::simulation {
         environment_.before_step();
 
         // read the sample index from the parameter collection
-        auto first_event_   = step.upper;
+
+        std::mutex mutex_first_event_;
+        time_point first_event_   = step.upper;
         unsigned int round_ = 0;
         do {
 
             if (verbosity > 0 && 0 == (rounds_ % verbosity)){
                 LOG(notice) << "time " << step << " round " << round_  << std::endl;
             }
             first_event_   = step.upper;
-            for(auto &[i, a] : agents.local_agents_) {
-
-                //double agent_cb_end_;
-                //timings_.emplace(i, 0.);
-                //timings_cb_.emplace(i, 0.);
-                //timings_act_.emplace(i, 0.);
-                //auto agent_start_ = high_resolution_clock::now();
-                //auto agent_act_ = high_resolution_clock::now();
+
+
+            auto job_ = [&](std::shared_ptr<agent> a){
+                // double agent_cb_end_;
+                // timings_.emplace(i, 0.);
+                // timings_cb_.emplace(i, 0.);
+                // timings_act_.emplace(i, 0.);
+                // auto agent_start_ = high_resolution_clock::now();
+                // auto agent_act_ = high_resolution_clock::now();
                 // The seed is deterministic in the following variables:
                 std::seed_seq seed_ {
-                    std::uint64_t(std::hash<identity<agent>>()(i)),
-                    std::uint64_t(step.lower),
-                    std::uint64_t(round_),
-                    sample
-                };
-
-                //try {
-                    first_event_ = std::min(first_event_, a->process_messages(step, seed_));
-                    //agent_cb_end_ = double((high_resolution_clock::now() - agent_start_).count());
-                    //agent_act_ = high_resolution_clock::now();
+                    std::uint64_t(std::hash<identity<agent>>()(a->identifier)),
+                    std::uint64_t(step.lower), std::uint64_t(round_),
+                    sample};
+
+                // try {
+
+                {
+                    std::unique_lock lock_(mutex_first_event_);
+                    first_event_ = std::min(first_event_,
+                                            a->process_messages(step, seed_));
+                    // agent_cb_end_ = double((high_resolution_clock::now() - agent_start_).count()); agent_act_ = high_resolution_clock::now();
                     first_event_ = std::min(first_event_, a->act(step, seed_));
+                }
+
                 //} catch(const std::runtime_error &e) {
                 //    LOG(errorlog) << e.what() << std::endl;
                 //    throw e;
@@ -103,17 +112,45 @@ namespace esl::simulation {
                 //} catch(...) {
                 //    throw;
                 //}
-                a->inbox.clear(); 
-                //auto agent_end_ = high_resolution_clock::now() - agent_start_;
-
-
-                //timings_cb_[a->identifier] += agent_cb_end_;
-                //timings_act_[a->identifier] += double( (high_resolution_clock::now() - agent_act_).count());
-                //timings_[a->identifier] += (double(agent_end_.count()) );
-                //LOG(notice) << "a" << a->identifier << " tcb " << step << " " << std::setprecision(8) << timings_cb_[a->identifier]/ 1e+9/step.lower <<  " s" << std::endl;
-                //LOG(notice) << "a" << a->identifier << " tac " << step << " " << std::setprecision(8) << timings_act_[a->identifier]/ 1e+9/step.lower <<  " s" << std::endl;
+                a->inbox.clear();
+                // auto agent_end_ = high_resolution_clock::now() - agent_start_;
+
+
+                // timings_cb_[a->identifier] += agent_cb_end_;
+                // timings_act_[a->identifier] += double( (high_resolution_clock::now() - agent_act_).count());
+                // timings_[a->identifier] += (double(agent_end_.count()) );
+                // LOG(notice) << "a" << a->identifier << " tcb " << step << " " << std::setprecision(8) << timings_cb_[a->identifier]/ 1e+9/step.lower <<  " s" << std::endl; LOG(notice) << "a" << a->identifier << " tac " << step << " " << std::setprecision(8) << timings_act_[a->identifier]/ 1e+9/step.lower <<  " s" << std::endl;
+
+            };
+
+            // important: if using a single thread, run everything in main
+            if(threads <= 1) {
+                for(auto &[i, a] : agents.local_agents_) {
+                    job_(a);
+                }
+            }else{
+                std::vector<std::thread> threads_;
+                auto iterator_ = agents.local_agents_.begin();
+                for(const auto& tasks_: quantity(agents.local_agents_.size()) / threads){
+                    std::vector<std::shared_ptr<agent>> task_split_;
+                    for(auto i = quantity(0); i < tasks_; ++i){
+                        task_split_.push_back(iterator_->second);
+                        std::advance(iterator_, 1);
+                    }
+
+                    threads_.emplace_back([&](std::vector<std::shared_ptr<agent>> ts){
+                            for(const auto& a: ts){
+                                job_(a);
+                            }
+                        }, task_split_);
+                }
+
+                for(auto &t: threads_){
+                    t.join();
+                }
 
             }
+
             environment_.send_messages(*this);
             ++round_;
             ++rounds_;
diff --git a/esl/simulation/parameter/parametrization.hpp b/esl/simulation/parameter/parametrization.hpp
@@ -85,15 +85,18 @@ namespace esl::simulation::parameter {
                        , time_point start           = time_point()
                        , time_point end             = time_point() + 1
 #ifdef ESL_RELEASE
-                       , std::uint64_t verbosity    = 1==ESL_RELEASE ? 1000 : 100 )
+                       , std::uint64_t verbosity    = 0
 #else
-                       , std::uint64_t verbosity    = 100 )
+                       , std::uint64_t verbosity    = 1
 #endif
+                       , unsigned int threads       = 1)
+
         {
-            values["sample"] = std::make_shared<constant<std::uint64_t>>(sample);
-            values["start"]  = std::make_shared<constant<time_point>>(start);
-            values["end"]    = std::make_shared<constant<time_point>>(end);
+            values["sample"]    = std::make_shared<constant<std::uint64_t>>(sample);
+            values["start"]     = std::make_shared<constant<time_point>>(start);
+            values["end"]       = std::make_shared<constant<time_point>>(end);
             values["verbosity"] = std::make_shared<constant<std::uint64_t>>(verbosity);
+            values["threads"]   = std::make_shared<constant<unsigned int>>(verbosity);
         }
 
 
diff --git a/test/test_environment.cpp b/test/test_environment.cpp
@@ -34,23 +34,28 @@
 #include <esl/agent.hpp>
 
 
+using namespace esl;
+using namespace esl::simulation;
+
 struct test_agent
-: public esl::agent
+: public agent
 {
-    using esl::agent::agent;
+    using agent::agent;
+
+    unsigned int delay = 0;
 
-    esl::simulation::time_point act(esl::simulation::time_interval step,
+    time_point act(time_interval step,
                                     std::seed_seq &seed) override
     {
-        return agent::act(step, seed);
+        return step.lower + delay;
     }
 };
 
 
 struct test_model
-    : public esl::simulation::model
+    : public model
 {
-    using esl::simulation::model::model;
+    using model::model;
 
     int initialize_called = 0;
 
@@ -74,9 +79,9 @@ BOOST_AUTO_TEST_SUITE(ESL)
 
     BOOST_AUTO_TEST_CASE(environment_constructor)
     {
-        esl::computation::environment e;
+        computation::environment e;
 
-        test_model tm(e, esl::simulation::parameter::parametrization(0, 1, 100));
+        test_model tm(e, parameter::parametrization(0, 1, 100));
 
         BOOST_CHECK_EQUAL(tm.start, 1);
         BOOST_CHECK_EQUAL(tm.time, 1);
@@ -85,14 +90,68 @@ BOOST_AUTO_TEST_SUITE(ESL)
 
     BOOST_AUTO_TEST_CASE(environment_basic_time_stepping)
     {
-        esl::computation::environment e;
+        computation::environment e;
+        test_model tm(e, parameter::parametrization(0, 0, 100));
+
+        auto next_ = tm.step( {0, 3});
+
+        std::cout << next_ << std::endl;
+
+        BOOST_CHECK_EQUAL(next_, 3);
+    }
+
+
+    BOOST_AUTO_TEST_CASE(environment_time_step_with_agents)
+    {
+        computation::environment e;
+        test_model tm(e, parameter::parametrization(0, 0, 100));
 
-        test_model tm(e, esl::simulation::parameter::parametrization(0, 0, 100));
+        auto a1 = tm.create<test_agent>();
+        a1->delay = 5;
+
+        auto next_ = tm.step( {0, 3});
+        std::cout << next_ << std::endl;
+        BOOST_CHECK_EQUAL(next_, 3);
+
+        next_ = tm.step( {3, 5});
+
+        std::cout << next_ << std::endl;
+
+        BOOST_CHECK_EQUAL(next_, 5);
+
+    }
+
+
+
+    ///
+    /// \brief
+    ///
+    BOOST_AUTO_TEST_CASE(environment_run_agents_parallel)
+    {
+        computation::environment e;
+        unsigned int threads = 8;
+
+        test_model tm(e, parameter::parametrization(0, 0, 100, 0, threads));
+
+        // We create one agent with delay 5, and other agents have more.
+        // If the parallelisation fails, we might miss that particular agent
+        // and end with a higher time point although this is nondeterministic
+        // and we are thus not guaranteed to catch it.
+        auto test_agents = 100'000;
+        for(auto i = 0; i < test_agents; ++i){
+            auto a1 = tm.create<test_agent>();
+            a1->delay = 4 + test_agents - i;
+        }
 
         auto next_ = tm.step( {0, 3});
 
         BOOST_CHECK_EQUAL(next_, 3);
+
+        // agents will delay again
+        next_ = tm.step( {3, 1234});
+        BOOST_CHECK_EQUAL(next_, 8);
     }
 
 
+
 BOOST_AUTO_TEST_SUITE_END()  // ESL

Original file line number	Diff line number	Diff line change
`@@ -75,7 +75,7 @@ namespace esl {`
`75`	`75`	`///`
`76`	`76`	`/// \param rhs`
`77`	`77`	`/// \return`
`78`		`- [[nodiscard]] constexpr bool operator!=(const agent &rhs) const`
	`78`	`+ [[nodiscard]] constexpr bool operator != (const agent &rhs) const`
`79`	`79`	`{`
`80`	`80`	`return !(*this == rhs);`
`81`	`81`	`}`
Original file line number	Diff line number	Diff line change
`@@ -86,7 +86,8 @@ namespace esl {`
`86`	`86`	`///`
`87`	`87`	`///`
`88`	`88`	`///`
`89`		`- entity() : entity(identity<entity_type_>())`
	`89`	`+ entity()`
	`90`	`+ : entity(identity<entity_type_>())`
`90`	`91`	`{`
`91`	`92`
`92`	`93`	`}`
`@@ -123,7 +124,6 @@ namespace esl {`
`123`	`124`	`}`
`124`	`125`
`125`	`126`	`public:`
`126`		`-`
`127`	`127`	`constexpr bool operator == (const entity_type_ &operand) const`
`128`	`128`	`{`
`129`	`129`	`return this->identifier == operand.identifier;`