scheduler: allow attaching tasks from already running processes (close #845)

This commit is contained in:
Michele Caini
2022-02-18 11:34:42 +01:00
parent bcd16ea5ea
commit 790aa03834
3 changed files with 58 additions and 42 deletions
+2 -1
View File
@@ -4,7 +4,8 @@
* add examples (and credits) from @alanjfs :)
WIP:
* dense map/set: support uses-allocator construction (single arg, multi-arg, key construction)
* uses-allocator construction: dense set (review), dense map, compressed pair, any (with allocator support), cache, dispatcher, poly, ...
* process scheduler: reviews, use free lists internally
* runtime events (emitter)
* iterator based try_emplace vs try_insert for perf reasons
* registry: remove reference to basic_sparse_set<E>
+18 -26
View File
@@ -42,8 +42,8 @@ template<typename Delta>
class scheduler {
struct process_handler {
using instance_type = std::unique_ptr<void, void (*)(void *)>;
using update_fn_type = bool(process_handler &, Delta, void *);
using abort_fn_type = void(process_handler &, bool);
using update_fn_type = bool(scheduler &, std::size_t, Delta, void *);
using abort_fn_type = void(scheduler &, std::size_t, bool);
using next_type = std::unique_ptr<process_handler>;
instance_type instance;
@@ -75,17 +75,17 @@ class scheduler {
};
template<typename Proc>
[[nodiscard]] static bool update(process_handler &handler, const Delta delta, void *data) {
auto *process = static_cast<Proc *>(handler.instance.get());
[[nodiscard]] static bool update(scheduler &owner, std::size_t pos, const Delta delta, void *data) {
auto *process = static_cast<Proc *>(owner.handlers[pos].instance.get());
process->tick(delta, data);
if(process->rejected()) {
return true;
} else if(process->finished()) {
if(handler.next) {
if(auto &&handler = owner.handlers[pos]; handler.next) {
handler = std::move(*handler.next);
// forces the process to exit the uninitialized state
return handler.update(handler, {}, nullptr);
return handler.update(owner, pos, {}, nullptr);
}
return true;
@@ -95,8 +95,8 @@ class scheduler {
}
template<typename Proc>
static void abort(process_handler &handler, const bool immediately) {
static_cast<Proc *>(handler.instance.get())->abort(immediately);
static void abort(scheduler &owner, std::size_t pos, const bool immediately) {
static_cast<Proc *>(owner.handlers[pos].instance.get())->abort(immediately);
}
template<typename Proc>
@@ -172,10 +172,10 @@ public:
auto attach(Args &&...args) {
static_assert(std::is_base_of_v<process<Proc, Delta>, Proc>, "Invalid process type");
auto proc = typename process_handler::instance_type{new Proc{std::forward<Args>(args)...}, &scheduler::deleter<Proc>};
process_handler handler{std::move(proc), &scheduler::update<Proc>, &scheduler::abort<Proc>, nullptr};
auto &&ref = handlers.emplace_back(process_handler{std::move(proc), &scheduler::update<Proc>, &scheduler::abort<Proc>, nullptr});
// forces the process to exit the uninitialized state
handler.update(handler, {}, nullptr);
return continuation{&handlers.emplace_back(std::move(handler))};
ref.update(*this, handlers.size() - 1u, {}, nullptr);
return continuation{&handlers.back()};
}
/**
@@ -247,17 +247,14 @@ public:
* @param data Optional data.
*/
void update(const Delta delta, void *data = nullptr) {
auto sz = handlers.size();
for(auto pos = handlers.size(); pos; --pos) {
auto &handler = handlers[pos - 1];
const auto curr = pos - 1u;
if(const auto dead = handler.update(handler, delta, data); dead) {
std::swap(handler, handlers[--sz]);
if(const auto dead = handlers[curr].update(*this, curr, delta, data); dead) {
std::swap(handlers[curr], handlers.back());
handlers.pop_back();
}
}
handlers.erase(handlers.begin() + sz, handlers.end());
}
/**
@@ -271,15 +268,10 @@ public:
* @param immediately Requests an immediate operation.
*/
void abort(const bool immediately = false) {
decltype(handlers) exec;
exec.swap(handlers);
for(auto &&handler: exec) {
handler.abort(handler, immediately);
for(auto pos = handlers.size(); pos; --pos) {
const auto curr = pos - 1u;
handlers[curr].abort(*this, curr, immediately);
}
std::move(handlers.begin(), handlers.end(), std::back_inserter(exec));
handlers.swap(exec);
}
private:
+38 -15
View File
@@ -22,29 +22,30 @@ struct foo_process: entt::process<foo_process, int> {
struct succeeded_process: entt::process<succeeded_process, int> {
void update(delta_type, void *) {
ASSERT_FALSE(updated);
updated = true;
++invoked;
succeed();
}
static unsigned int invoked;
bool updated = false;
static inline unsigned int invoked;
};
unsigned int succeeded_process::invoked = 0;
struct failed_process: entt::process<failed_process, int> {
void update(delta_type, void *) {
ASSERT_FALSE(updated);
updated = true;
++invoked;
fail();
}
bool updated = false;
static inline unsigned int invoked;
};
TEST(Scheduler, Functionalities) {
struct Scheduler: ::testing::Test {
void SetUp() override {
succeeded_process::invoked = 0u;
failed_process::invoked = 0u;
}
};
TEST_F(Scheduler, Functionalities) {
entt::scheduler<int> scheduler{};
bool updated = false;
@@ -75,7 +76,7 @@ TEST(Scheduler, Functionalities) {
ASSERT_TRUE(scheduler.empty());
}
TEST(Scheduler, Then) {
TEST_F(Scheduler, Then) {
entt::scheduler<int> scheduler;
// failing process with successor
@@ -93,15 +94,18 @@ TEST(Scheduler, Then) {
scheduler.attach<succeeded_process>()
.then<succeeded_process>();
for(auto i = 0; i < 8; ++i) {
ASSERT_EQ(succeeded_process::invoked, 0u);
ASSERT_EQ(failed_process::invoked, 0u);
while(!scheduler.empty()) {
scheduler.update(0);
}
ASSERT_EQ(succeeded_process::invoked, 6u);
ASSERT_TRUE(scheduler.empty());
ASSERT_EQ(failed_process::invoked, 2u);
}
TEST(Scheduler, Functor) {
TEST_F(Scheduler, Functor) {
entt::scheduler<int> scheduler;
bool first_functor = false;
@@ -121,7 +125,7 @@ TEST(Scheduler, Functor) {
scheduler.attach(std::move(attach)).then(std::move(then)).then([](auto...) { FAIL(); });
for(auto i = 0; i < 8; ++i) {
while(!scheduler.empty()) {
scheduler.update(0);
}
@@ -129,3 +133,22 @@ TEST(Scheduler, Functor) {
ASSERT_TRUE(second_functor);
ASSERT_TRUE(scheduler.empty());
}
TEST_F(Scheduler, SpawningProcess) {
entt::scheduler<int> scheduler;
scheduler.attach([&scheduler](auto, void *, auto resolve, auto) {
scheduler.attach<succeeded_process>().then<failed_process>();
resolve();
});
ASSERT_EQ(succeeded_process::invoked, 0u);
ASSERT_EQ(failed_process::invoked, 0u);
while(!scheduler.empty()) {
scheduler.update(0);
}
ASSERT_EQ(succeeded_process::invoked, 1u);
ASSERT_EQ(failed_process::invoked, 1u);
}