Merge branch 'feature/scheduler' into feature/library

headers/enduro2d/utils/_all.hpp

@@ -18,6 +18,7 @@
 #include "mesh.hpp"
 #include "module.hpp"
 #include "path.hpp"
+#include "scheduler.hpp"
 #include "streams.hpp"
 #include "strfmts.hpp"
 #include "strings.hpp"

headers/enduro2d/utils/_utils.hpp

@@ -17,6 +17,7 @@ namespace e2d
     class image;
     class jobber;
     class mesh;
+    class scheduler;
     class input_stream;
     class output_stream;
     class input_sequence;

headers/enduro2d/utils/jobber.hpp

@@ -7,6 +7,7 @@
 #pragma once
 
 #include "_utils.hpp"
+#include "time.hpp"
 
 namespace e2d
 {
@@ -42,6 +43,18 @@ namespace e2d
 
         void wait_all() const noexcept;
         void active_wait_all() noexcept;
+
+        template < typename T, typename TimeTag >
+        void wait_for(const unit<T, TimeTag>& time_for) const noexcept;
+
+        template < typename T, typename TimeTag >
+        void wait_until(const unit<T, TimeTag>& time_until) const noexcept;
+
+        template < typename T, typename TimeTag >
+        void active_wait_for(const unit<T, TimeTag>& time_for) noexcept;
+
+        template < typename T, typename TimeTag >
+        void active_wait_until(const unit<T, TimeTag>& time_until) noexcept;
     private:
         class task;
         using task_ptr = std::unique_ptr<task>;
@@ -123,6 +136,55 @@ namespace e2d
         return future;
     }
 
+    //
+    // wait
+    //
+
+    template < typename T, typename TimeTag >
+    void jobber::wait_for(const unit<T, TimeTag>& time_for) const noexcept {
+        if ( time_for.value > T(0) ) {
+            wait_until(
+                time::now<TimeTag, u64>() +
+                time_for.template cast_to<u64>());
+        }
+    }
+
+    template < typename T, typename TimeTag >
+    void jobber::wait_until(const unit<T, TimeTag>& time_until) const noexcept {
+        while ( active_task_count_ ) {
+            const auto time_now = time::now<TimeTag, u64>();
+            if ( time_now >= time_until.template cast_to<u64>() ) {
+                break;
+            }
+            std::this_thread::yield();
+        }
+    }
+
+    //
+    // active_wait
+    //
+
+    template < typename T, typename TimeTag >
+    void jobber::active_wait_for(const unit<T, TimeTag>& time_for) noexcept {
+        if ( time_for.value > T(0) ) {
+            active_wait_until(
+                time::now<TimeTag, u64>() +
+                time_for.template cast_to<u64>());
+        }
+    }
+
+    template < typename T, typename TimeTag >
+    void jobber::active_wait_until(const unit<T, TimeTag>& time_until) noexcept {
+        while ( active_task_count_ ) {
+            const auto time_now = time::now<TimeTag, u64>();
+            if ( time_now >= time_until.template cast_to<u64>() ) {
+                break;
+            }
+            std::unique_lock<std::mutex> lock(tasks_mutex_);
+            process_task_(std::move(lock));
+        }
+    }
+
     //
     // concrete_task<R, F, Args...>
     //

headers/enduro2d/utils/scheduler.hpp

@@ -0,0 +1,196 @@
+/*******************************************************************************
+ * This file is part of the "Enduro2D"
+ * For conditions of distribution and use, see copyright notice in LICENSE.md
+ * Copyright (C) 2018 Matvey Cherevko
+ ******************************************************************************/
+
+#pragma once
+
+#include "_utils.hpp"
+#include "time.hpp"
+
+namespace e2d
+{
+    class scheduler final : private noncopyable {
+    public:
+        enum class priority {
+            lowest,
+            below_normal,
+            normal,
+            above_normal,
+            highest
+        };
+    public:
+        scheduler();
+        ~scheduler() noexcept;
+
+        template < typename F, typename... Args >
+        using schedule_invoke_result_t = stdex::invoke_result_t<
+            std::decay_t<F>,
+            std::decay_t<Args>...>;
+
+        template < typename F, typename... Args
+                 , typename R = schedule_invoke_result_t<F, Args...> >
+        std::future<R> schedule(F&& f, Args&&... args);
+
+        template < typename F, typename... Args
+                 , typename R = schedule_invoke_result_t<F, Args...> >
+        std::future<R> schedule(priority priority, F&& f, Args&&... args);
+
+        void process_all_tasks() noexcept;
+
+        template < typename T, typename TimeTag >
+        void process_tasks_for(const unit<T, TimeTag>& time_for) noexcept;
+
+        template < typename T, typename TimeTag >
+        void process_tasks_until(const unit<T, TimeTag>& time_until) noexcept;
+    private:
+        class task;
+        using task_ptr = std::unique_ptr<task>;
+        template < typename R, typename F, typename... Args >
+        class concrete_task;
+    private:
+        void push_task_(priority priority, task_ptr task);
+        task_ptr pop_task_() noexcept;
+        void process_task_(std::unique_lock<std::mutex> lock) noexcept;
+    private:
+        std::mutex tasks_mutex_;
+        std::atomic<size_t> active_task_count_{0};
+        vector<std::pair<priority, task_ptr>> tasks_;
+    };
+
+    class scheduler::task : private noncopyable {
+    public:
+        virtual ~task() noexcept = default;
+        virtual void run() noexcept = 0;
+    };
+
+    template < typename R, typename F, typename... Args >
+    class scheduler::concrete_task : public task {
+        F f_;
+        std::tuple<Args...> args_;
+        std::promise<R> promise_;
+    public:
+        template < typename U >
+        concrete_task(U&& u, std::tuple<Args...>&& args);
+        void run() noexcept final;
+        std::future<R> future() noexcept;
+    };
+
+    template < typename F, typename... Args >
+    class scheduler::concrete_task<void, F, Args...> : public task {
+        F f_;
+        std::tuple<Args...> args_;
+        std::promise<void> promise_;
+    public:
+        template < typename U >
+        concrete_task(U&& u, std::tuple<Args...>&& args);
+        void run() noexcept final;
+        std::future<void> future() noexcept;
+    };
+}
+
+namespace e2d
+{
+    //
+    // schedule
+    //
+
+    template < typename F, typename... Args, typename R >
+    std::future<R> scheduler::schedule(F&& f, Args&&... args) {
+        return schedule(
+            priority::normal,
+            std::forward<F>(f),
+            std::forward<Args>(args)...);
+    }
+
+    template < typename F, typename... Args, typename R >
+    std::future<R> scheduler::schedule(priority priority, F&& f, Args&&... args) {
+        using task_t = concrete_task<
+            R,
+            std::decay_t<F>,
+            std::decay_t<Args>...>;
+        std::unique_ptr<task_t> task = std::make_unique<task_t>(
+            std::forward<F>(f),
+            std::make_tuple(std::forward<Args>(args)...));
+        std::future<R> future = task->future();
+        std::lock_guard<std::mutex> guard(tasks_mutex_);
+        push_task_(priority, std::move(task));
+        return future;
+    }
+
+    //
+    // process_all_tasks
+    //
+
+    template < typename T, typename TimeTag >
+    void scheduler::process_tasks_for(const unit<T, TimeTag>& time_for) noexcept {
+        if ( time_for.value > T(0) ) {
+            process_tasks_until(
+                time::now<TimeTag, u64>() +
+                time_for.template cast_to<u64>());
+        }
+    }
+
+    template < typename T, typename TimeTag >
+    void scheduler::process_tasks_until(const unit<T, TimeTag>& time_until) noexcept {
+        while ( active_task_count_ ) {
+            const auto time_now = time::now<TimeTag, u64>();
+            if ( time_now >= time_until.template cast_to<u64>() ) {
+                break;
+            }
+            std::unique_lock<std::mutex> lock(tasks_mutex_);
+            process_task_(std::move(lock));
+        }
+    }
+
+    //
+    // concrete_task<R, F, Args...>
+    //
+
+    template < typename R, typename F, typename... Args >
+    template < typename U >
+    scheduler::concrete_task<R, F, Args...>::concrete_task(U&& u, std::tuple<Args...>&& args)
+    : f_(std::forward<U>(u))
+    , args_(std::move(args)) {}
+
+    template < typename R, typename F, typename... Args >
+    void scheduler::concrete_task<R, F, Args...>::run() noexcept {
+        try {
+            R value = stdex::apply(std::move(f_), std::move(args_));
+            promise_.set_value(std::move(value));
+        } catch (...) {
+            promise_.set_exception(std::current_exception());
+        }
+    }
+
+    template < typename R, typename F, typename... Args >
+    std::future<R> scheduler::concrete_task<R, F, Args...>::future() noexcept {
+        return promise_.get_future();
+    }
+
+    //
+    // concrete_task<void, F, Args...>
+    //
+
+    template < typename F, typename... Args >
+    template < typename U >
+    scheduler::concrete_task<void, F, Args...>::concrete_task(U&& u, std::tuple<Args...>&& args)
+    : f_(std::forward<U>(u))
+    , args_(std::move(args)) {}
+
+    template < typename F, typename... Args >
+    void scheduler::concrete_task<void, F, Args...>::run() noexcept {
+        try {
+            stdex::apply(std::move(f_), std::move(args_));
+            promise_.set_value();
+        } catch (...) {
+            promise_.set_exception(std::current_exception());
+        }
+    }
+
+    template < typename F, typename... Args >
+    std::future<void> scheduler::concrete_task<void, F, Args...>::future() noexcept {
+        return promise_.get_future();
+    }
+}

sources/enduro2d/utils/scheduler.cpp

@@ -0,0 +1,51 @@
+/*******************************************************************************
+ * This file is part of the "Enduro2D"
+ * For conditions of distribution and use, see copyright notice in LICENSE.md
+ * Copyright (C) 2018 Matvey Cherevko
+ ******************************************************************************/
+
+#include <enduro2d/utils/scheduler.hpp>
+
+namespace
+{
+    using namespace e2d;
+}
+
+namespace e2d
+{
+    scheduler::scheduler() = default;
+    scheduler::~scheduler() noexcept = default;
+
+    void scheduler::process_all_tasks() noexcept {
+        while ( active_task_count_ ) {
+            std::unique_lock<std::mutex> lock(tasks_mutex_);
+            process_task_(std::move(lock));
+        }
+    }
+
+    void scheduler::push_task_(priority priority, task_ptr task) {
+        tasks_.emplace_back(priority, std::move(task));
+        std::push_heap(tasks_.begin(), tasks_.end());
+        ++active_task_count_;
+    }
+
+    scheduler::task_ptr scheduler::pop_task_() noexcept {
+        if ( !tasks_.empty() ) {
+            std::pop_heap(tasks_.begin(), tasks_.end());
+            task_ptr task = std::move(tasks_.back().second);
+            tasks_.pop_back();
+            return task;
+        }
+        return nullptr;
+    }
+
+    void scheduler::process_task_(std::unique_lock<std::mutex> lock) noexcept {
+        E2D_ASSERT(lock.owns_lock());
+        task_ptr task = pop_task_();
+        if ( task ) {
+            lock.unlock();
+            task->run();
+            --active_task_count_;
+        }
+    }
+}

untests/sources/untests_utils/jobber.cpp

@@ -69,6 +69,35 @@ TEST_CASE("jobber") {
         j.resume();
         j.wait_all();
     }
+    {
+        jobber j(1);
+        i32 counter = 0;
+        j.pause();
+        for ( std::size_t i = 0; i < 10; ++i ) {
+            j.async([&counter](){
+                ++counter;
+                std::this_thread::sleep_for(std::chrono::milliseconds(15));
+            });
+        }
+
+        const auto b = time::now_ms();
+
+        j.wait_for(make_milliseconds(15));
+        REQUIRE(time::now_ms() - b > make_milliseconds<i64>(10));
+        REQUIRE(counter == 0);
+
+        j.wait_until(time::now_ms() + make_milliseconds<i64>(15));
+        REQUIRE(time::now_ms() - b > make_milliseconds<i64>(20));
+        REQUIRE(counter == 0);
+
+        j.active_wait_for(make_milliseconds(60));
+        REQUIRE(time::now_ms() - b > make_milliseconds<i64>(70));
+        REQUIRE(counter > 2);
+        REQUIRE(counter < 10);
+
+        j.active_wait_until(time::now_s() + make_seconds<i64>(1));
+        REQUIRE(counter == 10);
+    }
     {
         jobber j(2);
         jobber g(2);

untests/sources/untests_utils/scheduler.cpp

@@ -0,0 +1,80 @@
+/*******************************************************************************
+ * This file is part of the "Enduro2D"
+ * For conditions of distribution and use, see copyright notice in LICENSE.md
+ * Copyright (C) 2018 Matvey Cherevko
+ ******************************************************************************/
+
+#include "_utils.hpp"
+using namespace e2d;
+
+TEST_CASE("scheduler") {
+    {
+        scheduler s;
+        i32 counter = 0;
+        s.schedule([&counter](){ ++counter; });
+        REQUIRE(counter == 0);
+        s.process_all_tasks();
+        REQUIRE(counter == 1);
+        s.schedule([&counter](){ ++counter; });
+        REQUIRE(counter == 1);
+        s.process_all_tasks();
+        REQUIRE(counter == 2);
+    }
+    {
+        scheduler s;
+        i32 counter = 0;
+        for ( std::size_t i = 0; i < 10; ++i ) {
+            s.schedule([&counter](){
+                ++counter;
+                std::this_thread::sleep_for(std::chrono::milliseconds(15));
+            });
+        }
+        s.process_tasks_for(make_milliseconds(-1));
+        s.process_tasks_for(make_milliseconds(0));
+        REQUIRE(counter == 0);
+        s.process_tasks_for(make_milliseconds(60));
+        REQUIRE(counter > 2);
+        REQUIRE(counter < 10);
+        s.process_tasks_for(make_seconds(1));
+        REQUIRE(counter == 10);
+    }
+    {
+        scheduler s;
+        i32 counter = 0;
+        for ( std::size_t i = 0; i < 10; ++i ) {
+            s.schedule([&counter](){
+                ++counter;
+                std::this_thread::sleep_for(std::chrono::milliseconds(15));
+            });
+        }
+        s.process_tasks_until(time::now_ms() - make_milliseconds<i64>(1));
+        s.process_tasks_until(time::now_ms());
+        REQUIRE(counter == 0);
+        s.process_tasks_until(time::now_ms() + make_milliseconds<i64>(60));
+        REQUIRE(counter > 2);
+        REQUIRE(counter < 10);
+        s.process_tasks_until(time::now_s() + make_seconds<i64>(1));
+        REQUIRE(counter == 10);
+    }
+    {
+        scheduler s;
+        str accumulator;
+        s.schedule(scheduler::priority::lowest, [](str& acc){
+            acc.append("o");
+        }, std::ref(accumulator));
+        s.schedule(scheduler::priority::below_normal, [](str& acc){
+            acc.append("l");
+        }, std::ref(accumulator));
+        s.schedule(scheduler::priority::highest, [](str& acc){
+            acc.append("h");
+        }, std::ref(accumulator));
+        s.schedule(scheduler::priority::above_normal, [](str& acc){
+            acc.append("e");
+        }, std::ref(accumulator));
+        s.schedule(scheduler::priority::normal, [](str& acc){
+            acc.append("l");
+        }, std::ref(accumulator));
+        s.process_all_tasks();
+        REQUIRE(accumulator == "hello");
+    }
+}