|
|
@@ -1,7 +1,9 @@
|
|
|
#include "tristate_obj.h"
|
|
|
|
|
|
+#include "utility/debug_utility.hpp"
|
|
|
+
|
|
|
#include "third_party/static_block.hpp"
|
|
|
-#include "utility/coro_signal.hpp"
|
|
|
+#include "utility/coro_signal2.hpp"
|
|
|
#include "utility/name_translator.hpp"
|
|
|
|
|
|
#include <spdlog/spdlog.h>
|
|
|
@@ -34,12 +36,12 @@ namespace sophiar {
|
|
|
tristate_obj *q_this = nullptr;
|
|
|
|
|
|
state_type state = state_type::INITIAL;
|
|
|
- coro_signal init_cancel_signal;
|
|
|
- coro_signal start_cancel_signal;
|
|
|
- coro_signal init_finished_signal;
|
|
|
- coro_signal start_finished_signal;
|
|
|
- coro_signal stop_finished_signal;
|
|
|
- coro_signal reset_finished_signal;
|
|
|
+ coro_signal2 init_cancel_signal;
|
|
|
+ coro_signal2 start_cancel_signal;
|
|
|
+ coro_signal2 init_finished_signal;
|
|
|
+ coro_signal2 start_finished_signal;
|
|
|
+ coro_signal2 stop_finished_signal;
|
|
|
+ coro_signal2 reset_finished_signal;
|
|
|
|
|
|
impl()
|
|
|
: init_cancel_signal(get_context()),
|
|
|
@@ -58,12 +60,15 @@ namespace sophiar {
|
|
|
|
|
|
awaitable<bool> init(const nlohmann::json &config) {
|
|
|
if (state == state_type::INITIALIZING) {
|
|
|
- auto result = co_await (init_finished_signal.coro_wait() ||
|
|
|
- reset_finished_signal.coro_wait());
|
|
|
+ auto init_finished_watcher = init_finished_signal.new_watcher(get_context());
|
|
|
+ auto reset_finished_watcher = reset_finished_signal.new_watcher(get_context());
|
|
|
+ auto result = co_await (init_finished_watcher.coro_wait() ||
|
|
|
+ reset_finished_watcher.coro_wait());
|
|
|
co_return result.index() == 0;
|
|
|
}
|
|
|
if (state == state_type::RESETTING) {
|
|
|
- co_await reset_finished_signal.coro_wait();
|
|
|
+ auto reset_finished_watcher = reset_finished_signal.new_watcher(get_context());
|
|
|
+ co_await reset_finished_watcher.coro_wait();
|
|
|
co_return false;
|
|
|
}
|
|
|
if (state != state_type::INITIAL) co_return true; // >= PENDING
|
|
|
@@ -71,7 +76,8 @@ namespace sophiar {
|
|
|
log_state_change(state_type::INITIAL, state_type::INITIALIZING);
|
|
|
SPDLOG_TRACE("Initializing object [name = {}] with config {}.",
|
|
|
get_manager().get_object_name(q_this), config.dump());
|
|
|
- auto result = co_await (q_this->on_init(config) || init_cancel_signal.coro_wait());
|
|
|
+ auto init_cancel_watcher = init_cancel_signal.new_watcher(get_context());
|
|
|
+ auto result = co_await (q_this->on_init(config) || init_cancel_watcher.coro_wait());
|
|
|
if (result.index() == 0 && std::get<0>(result) == true) { // succeeded
|
|
|
state = state_type::PENDING;
|
|
|
log_state_change(state_type::INITIALIZING, state_type::PENDING);
|
|
|
@@ -90,12 +96,15 @@ namespace sophiar {
|
|
|
|
|
|
awaitable<bool> start(const nlohmann::json &config) {
|
|
|
if (state == state_type::STARTING) {
|
|
|
- auto result = co_await (start_finished_signal.coro_wait() ||
|
|
|
- stop_finished_signal.coro_wait());
|
|
|
+ auto start_finished_watcher = start_finished_signal.new_watcher(get_context());
|
|
|
+ auto stop_finished_watcher = stop_finished_signal.new_watcher(get_context());
|
|
|
+ auto result = co_await (start_finished_watcher.coro_wait() ||
|
|
|
+ stop_finished_watcher.coro_wait());
|
|
|
co_return result.index() == 0;
|
|
|
}
|
|
|
if (state == state_type::STOPPING) {
|
|
|
- co_await stop_finished_signal.coro_wait();
|
|
|
+ auto stop_finished_watcher = stop_finished_signal.new_watcher(get_context());
|
|
|
+ co_await stop_finished_watcher.coro_wait();
|
|
|
co_return false;
|
|
|
}
|
|
|
if (state == state_type::RUNNING) co_return true;
|
|
|
@@ -104,7 +113,8 @@ namespace sophiar {
|
|
|
log_state_change(state_type::PENDING, state_type::STARTING);
|
|
|
SPDLOG_TRACE("Starting object [name = {}] with config {}.",
|
|
|
get_manager().get_object_name(q_this), config.dump());
|
|
|
- auto result = co_await (q_this->on_start(config) || start_cancel_signal.coro_wait());
|
|
|
+ auto start_cancel_watcher = start_cancel_signal.new_watcher(get_context());
|
|
|
+ auto result = co_await (q_this->on_start(config) || start_cancel_watcher.coro_wait());
|
|
|
if (result.index() == 0 && std::get<0>(result) == true) { // succeeded
|
|
|
state = state_type::RUNNING;
|
|
|
log_state_change(state_type::STARTING, state_type::RUNNING);
|
|
|
@@ -132,10 +142,12 @@ namespace sophiar {
|
|
|
stop_finished_signal.try_notify_all();
|
|
|
get_manager().notify_object_stop(q_this);
|
|
|
} else if (state == state_type::STOPPING) {
|
|
|
- co_await stop_finished_signal.coro_wait();
|
|
|
+ auto stop_finished_watcher = stop_finished_signal.new_watcher(get_context());
|
|
|
+ co_await stop_finished_watcher.coro_wait();
|
|
|
} else if (state == state_type::STARTING) {
|
|
|
start_cancel_signal.try_notify_all();
|
|
|
- co_await stop_finished_signal.coro_wait();
|
|
|
+ auto stop_finished_watcher = stop_finished_signal.new_watcher(get_context());
|
|
|
+ co_await stop_finished_watcher.coro_wait();
|
|
|
}
|
|
|
co_return;
|
|
|
}
|
|
|
@@ -158,10 +170,12 @@ namespace sophiar {
|
|
|
SPDLOG_DEBUG("Reset object [name = {}].", get_manager().get_object_name(q_this));
|
|
|
reset_finished_signal.try_notify_all();
|
|
|
} else if (state == state_type::RESETTING) {
|
|
|
- co_await reset_finished_signal.coro_wait();
|
|
|
+ auto reset_finished_watcher = reset_finished_signal.new_watcher(get_context());
|
|
|
+ co_await reset_finished_watcher.coro_wait();
|
|
|
} else if (state == state_type::INITIALIZING) {
|
|
|
init_cancel_signal.try_notify_all();
|
|
|
- co_await reset_finished_signal.coro_wait();
|
|
|
+ auto reset_finished_watcher = reset_finished_signal.new_watcher(get_context());
|
|
|
+ co_await reset_finished_watcher.coro_wait();
|
|
|
}
|
|
|
co_return;
|
|
|
}
|