Implemented variable callback.

app/data/tka_ar_navigation.json

@@ -421,19 +421,19 @@
       "start_config": {
         "fiducial_points": [
           [
-            -39.9547,
-            -25.8887,
-            204.005
+            47.351,
+            46.6799,
+            -70.5804
           ],
           [
-            -51.1875,
-            12.4128,
-            203.383
+            -45.3668,
+            23.3452,
+            1.94304
           ],
           [
-            -71.8993,
-            -18.712,
-            204.388
+            -55.3382,
+            -4.55638,
+            54.7446
           ]
         ],
         "point_var_name": "picked_point_in_tibia_ref",

src/core/sophiar_pool.cpp

@@ -1,32 +1,78 @@
 #include "sophiar_pool.h"
 #include "core/basic_obj_types.hpp"
+#include "core/global_defs.h"
 #include "utility/config_utility.hpp"
 #include "utility/named_vector.hpp"
 #include "utility/string_map.hpp"
 #include "utility/versatile_buffer2.hpp"
 
+#include <boost/asio/co_spawn.hpp>
+#include <boost/asio/post.hpp>
+
+#include <list>
 #include <vector>
 #include <unordered_map>
 
+using boost::asio::co_spawn;
+using boost::asio::post;
+
 namespace sophiar {
 
     struct sophiar_pool::impl {
 
+        struct variable_info_impl;
+        struct callback_info;
+
+        using callback_queue_type = std::list<callback_info *>;
+        using callback_queue_iter_type = callback_queue_type::iterator;
+
+        struct attach_info {
+            variable_info_impl *var_info = nullptr;
+            callback_info *callback = nullptr;
+            void *callback_list_iter = nullptr; // callback_list_iter_type
+            void *attach_iter = nullptr; // attach_iter_type
+        };
+
+        using attach_list_type = std::list<attach_info>;
+        using attach_iter_type = attach_list_type::iterator;
+        static_assert(sizeof(attach_iter_type) == sizeof(void *));
+        static_assert(sizeof(attach_iter_type) == sizeof(attach_token_type));
+
+        struct callback_info {
+            enum status_type {
+                IDLE, PENDING, RUNNING
+            } status = IDLE;
+
+            using callback_store_type = std::variant<function_callback_type, coroutine_callback_type>;
+            callback_store_type func_store;
+            attach_list_type attach_list;
+            callback_queue_iter_type queue_iter;
+        };
+
+        using callback_pool_type = std::list<callback_info>;
+        using callback_iter_type = callback_pool_type::iterator;
+        static_assert(sizeof(callback_iter_type) == sizeof(callback_token_type));
+
+        using callback_list_type = std::list<callback_info *>;
+        using callback_list_iter_type = callback_list_type::iterator;
+        static_assert(sizeof(callback_list_iter_type) == sizeof(void *));
+
         struct variable_type_info {
             std::type_index type = typeid(void);
-            variable_type_index_type type_index;
+            variable_type_index_type type_index = -1;
             std::string type_name;
 
             // function list
             using creator_func_type = void *(*)(const nlohmann::json &);
-            creator_func_type creator_func;
+            creator_func_type creator_func = nullptr;
         };
 
         struct variable_info_impl {
-            void *placeholder;
-            coro_signal2 *update_signal;
-            const variable_type_info *type_info;
-            timestamp_type last_update_ts;
+            void *placeholder = nullptr;
+            coro_signal2 *update_signal = nullptr;
+            const variable_type_info *type_info = nullptr;
+            timestamp_type last_update_ts = 0;
+            callback_list_type callback_list;
         };
 
         string_map<variable_type_index_type> variable_type_name_index_map; // typename -> index
@@ -34,6 +80,9 @@ namespace sophiar {
         std::vector<variable_type_info> variable_type_info_pool;
 
         named_vector<variable_index_type, variable_info_impl> variable_pool;
+        callback_pool_type callback_pool;
+        callback_queue_type callback_queue;
+        bool is_callback_handler_active = false;
 
         template<typename SmallObjType>
         static void *create_variable_pointer(const nlohmann::json &config) {
@@ -108,6 +157,76 @@ namespace sophiar {
             info.last_update_ts = 0;
         }
 
+        callback_token_type register_callback(callback_info::callback_store_type &&callback) {
+            auto &item = callback_pool.emplace_front();
+            item.func_store = std::move(callback);
+            return std::bit_cast<callback_token_type>(callback_pool.begin());
+        }
+
+        void unregister_callback(callback_iter_type callback) {
+            while (!callback->attach_list.empty()) {
+                detach_callback(callback->attach_list.begin());
+            }
+            if (callback->status == callback_info::PENDING) {
+                callback_queue.erase(callback->queue_iter);
+            }
+            callback_pool.erase(callback);
+        }
+
+        attach_token_type attach_callback(variable_index_type var_index, callback_iter_type callback) {
+            auto &var_info = variable_pool[var_index];
+            var_info.callback_list.push_front(&(*callback));
+
+            auto &attach_item = callback->attach_list.emplace_front();
+            attach_item.var_info = &var_info;
+            attach_item.callback = &(*callback);
+            attach_item.callback_list_iter = std::bit_cast<void *>(var_info.callback_list.begin());
+            attach_item.attach_iter = std::bit_cast<void *>(callback->attach_list.begin());
+            return std::bit_cast<attach_token_type>(callback->attach_list.begin());
+        }
+
+        static void detach_callback(attach_iter_type iter) {
+            auto callback_iter = std::bit_cast<callback_list_iter_type>(iter->callback_list_iter);
+            auto attach_iter = std::bit_cast<attach_iter_type>(iter->attach_iter);
+            iter->var_info->callback_list.erase(callback_iter);
+            iter->callback->attach_list.erase(attach_iter);
+        }
+
+        void handle_callback_func() {
+            while (!callback_queue.empty()) {
+                auto callback = callback_queue.back();
+                callback_queue.pop_back();
+                callback->status = callback_info::RUNNING;
+                callback->queue_iter = {};
+                if (callback->func_store.index() == 0) {
+                    std::get<0>(callback->func_store)();
+                    callback->status = callback_info::IDLE;
+                } else {
+                    assert(callback->func_store.index() == 1);
+                    co_spawn(*global_context, std::get<1>(callback->func_store)(), [=](std::exception_ptr eptr) {
+                        assert(eptr == nullptr);
+                        callback->status = callback_info::IDLE;
+                    });
+                }
+            }
+            is_callback_handler_active = false;
+        }
+
+        void trigger_callback(variable_info_impl *var_info) {
+            if (var_info->callback_list.empty()) return;
+            bool has_callback = false;
+            for (auto &callback: var_info->callback_list) {
+                if (callback->status != callback_info::IDLE) continue;
+                callback_queue.push_front(callback);
+                callback->status = callback_info::PENDING;
+                callback->queue_iter = callback_queue.begin();
+                has_callback = true;
+            }
+            if (!has_callback || is_callback_handler_active) return;
+            post(*global_context, [this]() { handle_callback_func(); });
+            is_callback_handler_active = true;
+        }
+
         void load_config(const nlohmann::json &config) {
 
 #ifdef SOPHIAR_TEST
@@ -186,12 +305,34 @@ namespace sophiar {
         assert(ts > info.last_update_ts);
         info.last_update_ts = ts;
         info.update_signal->try_notify_all(ts);
+        pimpl->trigger_callback(&info);
     }
 
     void sophiar_pool::load_config(const nlohmann::json &config) {
         pimpl->load_config(config);
     }
 
+    sophiar_pool::callback_token_type sophiar_pool::register_callback(function_callback_type &&callback) {
+        return pimpl->register_callback(std::move(callback));
+    }
+
+    sophiar_pool::callback_token_type sophiar_pool::register_coro_callback(coroutine_callback_type &&callback) {
+        return pimpl->register_callback(std::move(callback));
+    }
+
+    void sophiar_pool::unregister_callback(callback_token_type token) {
+        return pimpl->unregister_callback(std::bit_cast<impl::callback_iter_type>(token));
+    }
+
+    sophiar_pool::attach_token_type sophiar_pool::attach_callback(variable_index_type var_index,
+                                                                  callback_token_type token) {
+        return pimpl->attach_callback(var_index, std::bit_cast<impl::callback_iter_type>(token));
+    }
+
+    void sophiar_pool::detach_callback(attach_token_type token) {
+        return pimpl->detach_callback(std::bit_cast<impl::attach_iter_type>(token));
+    }
+
     sophiar_pool::~sophiar_pool() = default;
 
 #ifdef SOPHIAR_TEST

src/core/sophiar_pool.h

@@ -5,9 +5,12 @@
 #include "utility/coro_signal2.hpp"
 #include "utility/versatile_buffer2.hpp"
 
+#include <boost/asio/awaitable.hpp>
+
 #include <nlohmann/json.hpp>
 
 #include <cstdint>
+#include <functional>
 #include <string_view>
 #include <typeindex>
 #include <typeinfo>
@@ -58,6 +61,24 @@ namespace sophiar {
 
         timestamp_type query_variable_update_ts(variable_index_type var_index);
 
+        // callback support
+
+        using callback_token_type = void *;
+        using attach_token_type = void *;
+        using function_callback_type = std::function<void()>;
+        using coroutine_callback_type = std::function<boost::asio::awaitable<void>()>;
+
+        callback_token_type register_callback(function_callback_type &&callback);
+
+        callback_token_type register_coro_callback(coroutine_callback_type &&callback);
+
+        void unregister_callback(callback_token_type token);
+
+        attach_token_type attach_callback(variable_index_type var_index,
+                                          callback_token_type token);
+
+        void detach_callback(attach_token_type token);
+
 #ifdef SOPHIAR_TEST
 
     public:

tests/core/sophiar_pool.cpp

@@ -3,7 +3,11 @@
 #include "core/basic_obj_types.hpp"
 #include "core/global_defs.h"
 #include "core/sophiar_pool.h"
+#include "utility/debug_utility.hpp"
 
+#include <boost/asio/awaitable.hpp>
+#include <boost/asio/co_spawn.hpp>
+#include <boost/asio/detached.hpp>
 #include <boost/test/unit_test.hpp>
 
 #include <spdlog/spdlog.h>
@@ -12,6 +16,100 @@
 
 using namespace sophiar;
 
+using boost::asio::awaitable;
+using boost::asio::co_spawn;
+using boost::asio::detached;
+using namespace std::chrono_literals;
+
+awaitable<void> test_callback() {
+    int test_flag = 0;
+    auto callback_token = global_sophiar_pool->register_callback([&]() { ++test_flag; });
+    auto var_index = REQUIRE_VARIABLE(u64int_obj, "var_num");
+
+    auto attach_token = global_sophiar_pool->attach_callback(var_index, callback_token);
+    UPDATE_VARIABLE_VAL(u64int_obj, var_index, 1);
+    co_await coro_sleep(50ms);
+    BOOST_TEST(test_flag == 1);
+
+    global_sophiar_pool->detach_callback(attach_token);
+    UPDATE_VARIABLE_VAL(u64int_obj, var_index, 1);
+    co_await coro_sleep(50ms);
+    BOOST_TEST(test_flag == 1);
+
+    global_sophiar_pool->attach_callback(var_index, callback_token);
+    UPDATE_VARIABLE_VAL(u64int_obj, var_index, 2);
+    co_await coro_sleep(50ms);
+    BOOST_TEST(test_flag == 2);
+
+    global_sophiar_pool->unregister_callback(callback_token);
+    UPDATE_VARIABLE_VAL(u64int_obj, var_index, 2);
+    co_await coro_sleep(50ms);
+    BOOST_TEST(test_flag == 2);
+
+    int test_flag_2 = 0;
+    auto var_index_2 = REQUIRE_VARIABLE(double_obj, "var_float");
+    auto callback_token_1 = global_sophiar_pool->register_callback([&]() {
+        SPDLOG_DEBUG("callback 1");
+        ++test_flag;
+        UPDATE_VARIABLE_VAL(double_obj, var_index_2, 1.0);
+    });
+    auto callback_token_2 = global_sophiar_pool->register_callback([&]() {
+        SPDLOG_DEBUG("callback 2");
+        ++test_flag_2;
+    });
+    global_sophiar_pool->attach_callback(var_index, callback_token_1);
+    global_sophiar_pool->attach_callback(var_index_2, callback_token_2);
+    UPDATE_VARIABLE_VAL(u64int_obj, var_index, 3);
+    co_await coro_sleep(50ms);
+    BOOST_TEST(test_flag == 3);
+    BOOST_TEST(test_flag_2 == 1);
+
+    UPDATE_VARIABLE_VAL(u64int_obj, var_index, 4);
+    co_await coro_sleep(50ms);
+    BOOST_TEST(test_flag == 4);
+    BOOST_TEST(test_flag_2 == 2);
+
+    auto coro_token_1 = global_sophiar_pool->register_coro_callback([&]() -> awaitable<void> {
+        co_await coro_sleep(100ms);
+        SPDLOG_DEBUG("coro callback 1");
+        ++test_flag;
+        UPDATE_VARIABLE_VAL(double_obj, var_index_2, 1.0);
+        co_return;
+    });
+    auto coro_token_2 = global_sophiar_pool->register_coro_callback([&]() -> awaitable<void> {
+        co_await coro_sleep(100ms);
+        SPDLOG_DEBUG("coro callback 2");
+        ++test_flag_2;
+        co_return;
+    });
+    global_sophiar_pool->attach_callback(var_index, coro_token_1);
+    global_sophiar_pool->attach_callback(var_index_2, coro_token_2);
+    UPDATE_VARIABLE_VAL(u64int_obj, var_index, 5);
+    co_await coro_sleep(50ms);
+    BOOST_TEST(test_flag == 5);
+    BOOST_TEST(test_flag_2 == 3);
+
+    co_await coro_sleep(200ms);
+    BOOST_TEST(test_flag == 6);
+    BOOST_TEST(test_flag_2 == 5);
+
+    global_sophiar_pool->unregister_callback(callback_token_1);
+    global_sophiar_pool->unregister_callback(callback_token_2);
+    global_sophiar_pool->unregister_callback(coro_token_1);
+    global_sophiar_pool->unregister_callback(coro_token_2);
+
+    test_flag = 0;
+    callback_token = global_sophiar_pool->register_callback([&]() { ++test_flag; });
+    global_sophiar_pool->attach_callback(var_index, callback_token);
+    global_sophiar_pool->attach_callback(var_index_2, callback_token);
+    UPDATE_VARIABLE_VAL(u64int_obj, var_index, 1);
+    UPDATE_VARIABLE_VAL(double_obj, var_index_2, 1.0);
+    co_await coro_sleep(50ms);
+    BOOST_TEST(test_flag == 1);
+
+    co_return;
+}
+
 BOOST_AUTO_TEST_CASE(test_sophiar_pool) {
 
     spdlog::set_level(spdlog::level::trace);
@@ -64,5 +162,6 @@ BOOST_AUTO_TEST_CASE(test_sophiar_pool) {
         BOOST_TEST(var_ptr->value[0] == 6);
     }
 
+    co_spawn(*global_context, test_callback(), detached);
     global_context->run();
 }