重新实现了manager

CMakeLists.txt

@@ -28,7 +28,7 @@ file(GLOB_RECURSE SRC_FILES ./src/*.cpp)
 add_executable(${PROJECT_NAME} ${SRC_FILES})
 
 IF (WIN32)
-    list(APPEND EXTRA_LIBS ws2_32 winmm)
+    list(APPEND EXTRA_LIBS ws2_32 wsock32 winmm)
 ENDIF ()
 
 target_link_libraries(${PROJECT_NAME} ${EXTRA_LIBS})

doc/manager_format.txt

@@ -0,0 +1,56 @@
+Request Format:
+Length [2]
+Command [1]
+Command content [n]
+
+Reply Format:
+Length [2]
+Reply content [n]
+
+Command List:
+0x00 Query manager status
+0x01 Query obj status
+0x02 Query global obj value
+0x0A Query obj index
+0x0B Query global obj index
+0x10 Init obj
+0x11 Start obj
+0x12 Stop obj
+0x13 Reset obj
+
+Command Content List:
+0x00
+None [0]
+0x01
+Query number [2]
+Obj index list [2 * n]
+0x02
+Query number [2]
+Global obj index list [2 * n]
+0x0A
+Obj name length [2]
+Obj name str [n]
+0x0B
+Global obj name length [2]
+Global obj name str [n]
+0x10 - 0x13
+Obj index [2]
+
+Reply Content List:
+0x00
+Manager status [1] 0x00 Initial, 0x01 Normal, 0x02 Busy
+0x01
+Obj status list [1 * n] 0x00 Inital, 0x01 Initializing, 0x02 Resetting,
+                        0x03 Pending, 0x04 Starting, 0x05 Stopping, 0x06 Running
+                        0xFF 
+0x02
+Repeat of the follow [*n]
+  Global obj type [2]
+  Value length [1]
+  Value content [n]
+0x0A
+Obj index [2] (-1 means not found)
+0x0B
+Global obj index [2] (-1 means not found)
+0x10 - 0x13
+Finished [1] (Always 0xFF)

src/core/basic_obj_types.hpp

@@ -18,11 +18,72 @@ namespace sophiar {
 
         explicit small_obj_wrapper(T &&other)
                 : value(std::move(other)) {}
+
+        template<typename ReaderType>
+        void fill_from(ReaderType &reader) {
+            reader >> value;
+        }
+
+        template<typename WriterType>
+        void write_to(WriterType &writer) {
+            writer << value;
+        }
+
+        template<typename WriterType>
+        static void raw_pointer_write_to(WriterType &writer, void *raw_ptr) {
+            using this_type = small_obj_wrapper<T>;
+            auto &real_ptr = *static_cast<typename this_type::pointer *>(raw_ptr);
+            real_ptr->write_to(writer);
+        }
+
+        static constexpr size_t binary_length() {
+            return sizeof(T);
+        }
+
     };
 
-    using bool_obj = small_obj_wrapper<bool>;
-    using double_obj = small_obj_wrapper<double>;
-    using transform_obj = small_obj_wrapper<Eigen::Isometry3d>;
+    using bool_obj = small_obj_wrapper<bool>;       // 0
+    using u8int_obj = small_obj_wrapper<uint8_t>;   // 1
+    using u16int_obj = small_obj_wrapper<uint16_t>; // 2
+    using u32int_obj = small_obj_wrapper<uint32_t>; // 3
+    using u64int_obj = small_obj_wrapper<uint64_t>; // 4
+    using i8int_obj = small_obj_wrapper<int8_t>;    // 5
+    using i16int_obj = small_obj_wrapper<int16_t>;  // 6
+    using i32int_obj = small_obj_wrapper<int32_t>;  // 7
+    using i64int_obj = small_obj_wrapper<int64_t>;  // 8
+    using float_obj = small_obj_wrapper<float>;     // 9
+    using double_obj = small_obj_wrapper<double>;   // 10
+    using scalarxyz_obj = small_obj_wrapper<Eigen::Vector3d>;   // 11
+    using transform_obj = small_obj_wrapper<Eigen::Isometry3d>; // 12
+
+    // specialization
+
+    template<>
+    template<typename ReaderType>
+    inline void transform_obj::fill_from(ReaderType &reader) {
+        double tx, ty, tz, qw, qx, qy, qz;
+        reader >> tx >> ty >> tz >> qw >> qx >> qy >> qz;
+        value = Eigen::Quaterniond(qw, qx, qy, qz) * Eigen::Translation3d(tx, ty, tz);
+    }
+
+    template<>
+    template<typename WriterType>
+    inline void transform_obj::write_to(WriterType &writer) {
+        auto trans_part = value.translation();
+        writer << trans_part.x() << trans_part.y() << trans_part.z();
+        auto quat_part = Eigen::Quaterniond(value.rotation());
+        writer << quat_part.w() << quat_part.x() << quat_part.y() << quat_part.z();
+    }
+
+    template<>
+    constexpr size_t transform_obj::binary_length() {
+        return 7 * sizeof(double);
+    }
+
+    template<>
+    constexpr size_t scalarxyz_obj::binary_length() {
+        return 3 * sizeof(double);
+    }
 
 }
 

src/core/datanode_base.cpp

@@ -104,7 +104,7 @@ namespace sophiar {
         void create_trigger_slot() {
             auto trigger_slot = new trigger_slot_impl_type;
             trigger_slot->p_this = this;
-            get_manager().register_slot<>(q_this, "trigger", *trigger_slot);
+//            get_manager().register_slot<>(q_this, "trigger", *trigger_slot);
         }
 
         uint8_t create_input_slot(const std::string &name) {
@@ -113,14 +113,14 @@ namespace sophiar {
             auto update_slot = new update_slot_impl_type;
             update_slot->p_this = this;
             update_slot->channel_index = slot_index;
-            get_manager().register_slot<data_pointer_type>(q_this, name, *update_slot);
+//            get_manager().register_slot<data_pointer_type>(q_this, name, *update_slot);
             return slot_index;
         }
 
         uint8_t create_output_signal(const std::string &name) {
             assert(output_total + 1 <= MAX_CHANNEL_CNT);
             auto signal_index = output_total++;
-            get_manager().register_signal(q_this, name, output_signal[signal_index]);
+//            get_manager().register_signal(q_this, name, output_signal[signal_index]);
             return signal_index;
         }
 

src/core/sophiar_manager.h

@@ -34,9 +34,7 @@ namespace sophiar {
 
         using obj_factory_func_type = sophiar_obj *(*)();
 
-        void build_from_config(const nlohmann::json &config);
-
-        boost::asio::awaitable<bool> switch_mode(const std::string &mode_name);
+        bool load_config_and_start(const nlohmann::json &config);
 
         std::string get_object_name(sophiar_obj *obj) const;
 
@@ -79,26 +77,6 @@ namespace sophiar {
 
         signal_watcher request_global_obj_update_watcher(global_obj_index_type obj_index);
 
-        template<typename ...Args>
-        [[deprecated("Use obj pool system instead.")]]
-        void register_signal(sophiar_obj *obj,
-                             const std::string &signal_name,
-                             tiny_signal<Args...> &signal) {
-            tiny_signal_base *base_ptr = &signal;
-            register_signal_impl(obj, signal_name, base_ptr);
-        }
-
-        // 将套用 slot_demuxer
-        template<typename ...Args>
-        [[deprecated("Use obj pool system instead.")]]
-        void register_slot(sophiar_obj *obj,
-                           const std::string &slot_name,
-                           typename tiny_signal<Args...>::slot_type &slot) {
-            tiny_slot_base *slot_base = &slot;
-            slot_demuxer_base *demuxer_base = new slot_demuxer<Args...>(slot);
-            register_slot_impl(obj, slot_name, slot_base, demuxer_base);
-        }
-
         sophiar_manager();
 
         ~sophiar_manager();
@@ -109,21 +87,8 @@ namespace sophiar {
 
         sophiar_obj *get_object(const std::string &obj_name) const;
 
-        template<typename ...Args>
-        typename tiny_signal<Args...>::slot_type &get_slot(const std::string &obj_name,
-                                                           const std::string &slot_name) {
-            using ret_type = typename tiny_signal<Args...>::slot_type;
-            auto slot_base = get_slot_impl(obj_name, slot_name);
-            auto slot_ptr = dynamic_cast<ret_type *>(slot_base);
-            assert(slot_ptr != nullptr);
-            return *slot_ptr;
-        }
-
     private:
 
-        tiny_slot_base *get_slot_impl(const std::string &obj_name,
-                                      const std::string &slot_name);
-
 #endif // SOPHIAR_TEST
 
     private:
@@ -133,17 +98,6 @@ namespace sophiar {
 
         void register_object_type_impl(const std::string &type_name, obj_factory_func_type func);
 
-        // 记录 signal 的归属信息
-        void register_signal_impl(sophiar_obj *obj,
-                                  const std::string &signal_name,
-                                  tiny_signal_base *signal_base);
-
-        // 记录 slot 的归属信息
-        void register_slot_impl(sophiar_obj *obj,
-                                const std::string &slot_name,
-                                tiny_slot_base *slot_base,
-                                slot_demuxer_base *demuxer_base);
-
         global_obj_index_type register_global_obj_impl(const std::string &obj_name,
                                                        std::type_index obj_type,
                                                        void *placeholder); // small_obj<Derive>::pointer

src/core/transform_tree.cpp

@@ -135,7 +135,7 @@ namespace sophiar {
 
                 auto trans_update_slot = new trans_update_slot_impl_type;
                 trans_update_slot->node = &tree_node;
-                get_manager().register_slot<trans_obj::pointer>(q_this, name, *trans_update_slot);
+//                get_manager().register_slot<trans_obj::pointer>(q_this, name, *trans_update_slot);
             }
         }
 
@@ -156,7 +156,7 @@ namespace sophiar {
                     assert(watch_config["name"].is_string());
                     watch_name = watch_config["name"].get<std::string>();
                 }
-                get_manager().register_signal(q_this, watch_name, *watch_signal);
+//                get_manager().register_signal(q_this, watch_name, *watch_signal);
             }
         }
 

src/core/tristate_obj.h

@@ -18,13 +18,13 @@ namespace sophiar {
     public:
 
         enum class state_type {
-            INITIAL,
-            INITIALIZING,
-            RESETTING,
-            PENDING,
-            STARTING,
-            STOPPING,
-            RUNNING,
+            INITIAL = 0x00,
+            INITIALIZING = 0x01,
+            RESETTING = 0x02,
+            PENDING = 0x03,
+            STARTING = 0x04,
+            STOPPING = 0x05,
+            RUNNING = 0x06,
         };
 
         tristate_obj();

src/robot/ur/ur_interface.cpp

@@ -558,8 +558,8 @@ namespace sophiar {
         }
 
         void register_signal_slot() {
-            get_manager().register_signal(q_this, "status", ur_status_signal);
-            get_manager().register_slot<ur_command::pointer>(q_this, "command", ur_command_handler);
+//            get_manager().register_signal(q_this, "status", ur_status_signal);
+//            get_manager().register_slot<ur_command::pointer>(q_this, "command", ur_command_handler);
         }
 
         void load_init_config(const nlohmann::json &config) {

src/tracker/ndi/ndi_interface.cpp

@@ -447,32 +447,21 @@ namespace sophiar {
 
         void start_receive_reply() {
             assert(receive_reply_worker == nullptr);
-
-            // handler error
-            auto noexcept_wrapper = [this]() -> awaitable<bool> {
-                try {
-                    auto ok = co_await read_and_dispatch_reply();
-                    co_return ok;
-                } catch (std::exception &e) {
-                    // TODO show log
-                    co_return false;
-                }
-                assert(false);
-                co_return false;
+            auto worker_func = [this]() -> awaitable<bool> {
+                auto ok = co_await read_and_dispatch_reply();
+                co_return ok;
             };
-            receive_reply_worker = make_infinite_coro_worker(get_context(),
-                                                             std::move(noexcept_wrapper));
-            receive_reply_worker->run();
-
-            // start another coroutine that will reset the ndi_interface
-            // if the receiving reply coroutine stopped to work
-            auto watchdog_coro = [this, worker = receive_reply_worker]() -> awaitable<void> {
+            auto noexcept_worker_func = make_noexcept_func(std::move(worker_func), [](std::exception &e) {
+                // TODO show error
+            });
+            auto exit_func = [this]() {
                 assert(q_this != nullptr);
-                co_await worker->coro_wait_stop();
-                co_await q_this->reset();
-                co_return;
+                co_spawn(get_context(), q_this->reset(), detached);
             };
-            co_spawn(get_context(), std::move(watchdog_coro), detached);
+            receive_reply_worker = make_infinite_coro_worker(get_context(),
+                                                             std::move(noexcept_worker_func),
+                                                             std::move(exit_func));
+            receive_reply_worker->run();
         }
 
         awaitable<void> send_command(const char *cmd, size_t cmd_length) {

src/utility/coro_worker.hpp

@@ -19,7 +19,9 @@ namespace sophiar {
     class coro_worker {
     public:
 
-        using pointer = std::shared_ptr<coro_worker>;
+        using pointer = std::unique_ptr<coro_worker>;
+
+        static constexpr auto empty_func = []() {};
 
         virtual ~coro_worker() {
             assert(!is_running);
@@ -51,7 +53,7 @@ namespace sophiar {
 
     };
 
-    template<typename Executor, typename FuncType>
+    template<typename Executor, typename FuncType, typename ExitFuncType>
     class coro_worker_impl : public coro_worker {
     public:
 
@@ -59,10 +61,14 @@ namespace sophiar {
                 decltype(std::declval<FuncType>()()),
                 boost::asio::awaitable<bool>>);
 
-        coro_worker_impl(Executor &_executor, FuncType &&_func)
+        static_assert(std::is_void<
+                decltype(std::declval<ExitFuncType>()())>());
+
+        coro_worker_impl(Executor &_executor, FuncType &&_func, ExitFuncType &&_exit_func)
                 : coro_worker(_executor),
                   executor(_executor),
-                  func(std::forward<FuncType>(_func)) {}
+                  func(std::forward<FuncType>(_func)),
+                  exit_func(std::forward<ExitFuncType>(_exit_func)) {}
 
         ~coro_worker_impl() override = default;
 
@@ -82,8 +88,10 @@ namespace sophiar {
 
     private:
 
-        using store_type = std::remove_cvref_t<FuncType>;
-        store_type func;
+        using func_store_type = std::remove_cvref_t<FuncType>;
+        using exit_func_store_type = std::remove_cvref_t<ExitFuncType>;
+        func_store_type func;
+        exit_func_store_type exit_func;
 
         Executor &executor;
 
@@ -92,6 +100,7 @@ namespace sophiar {
             auto closer = sg::make_scope_guard([this]() {
                 is_running = false;
                 stop_finished_signal.try_notify_all();
+                exit_func();
             });
             for (;;) {
                 using namespace boost::asio::experimental::awaitable_operators;
@@ -104,14 +113,19 @@ namespace sophiar {
 
     };
 
-    template<typename Executor, typename FuncType>
-    auto make_infinite_coro_worker(Executor &executor, FuncType &&func) {
-        return coro_worker::pointer(new coro_worker_impl<Executor, FuncType>(
-                executor, std::forward<FuncType>(func)));
+    template<typename Executor, typename FuncType,
+            typename ExitFuncType = decltype(coro_worker::empty_func) const &>
+    auto make_infinite_coro_worker(Executor &executor, FuncType &&func,
+                                   ExitFuncType &&exit_func = coro_worker::empty_func) {
+        return coro_worker::pointer(new coro_worker_impl<Executor, FuncType, ExitFuncType>(
+                executor, std::forward<FuncType>(func), std::forward<ExitFuncType>(exit_func))
+        );
     }
 
-    template<typename Executor, typename FuncType>
-    auto make_interval_coro_worker(Executor &executor, std::chrono::milliseconds interval, FuncType &&func) {
+    template<typename Executor, typename FuncType,
+            typename ExitFuncType = decltype(coro_worker::empty_func) const &>
+    auto make_interval_coro_worker(Executor &executor, std::chrono::milliseconds interval,
+                                   FuncType &&func, ExitFuncType &&exit_func = coro_worker::empty_func) {
         auto worker_func = [
                 interval,
                 func = std::forward<FuncType>(func),
@@ -122,7 +136,37 @@ namespace sophiar {
             co_await timer.async_wait(boost::asio::use_awaitable);
             co_return ret;
         };
-        return make_infinite_coro_worker(executor, std::move(worker_func));
+        return make_infinite_coro_worker(executor, std::move(worker_func),
+                                         std::forward<ExitFuncType>(exit_func));
+    }
+
+    static constexpr auto empty_exception_handler = [](std::exception &) {};
+
+    template<typename FuncType,
+            typename ErrorHandlerType = decltype(empty_exception_handler) const &>
+    auto make_noexcept_func(FuncType &&func,
+                            ErrorHandlerType &&error_handler = empty_exception_handler) {
+        static_assert(std::is_convertible_v<
+                decltype(std::declval<FuncType>()()),
+                boost::asio::awaitable<bool>>);
+        static_assert(std::is_void<
+                decltype(std::declval<ErrorHandlerType>()(
+                        std::declval<std::exception &>()))>());
+        auto noexcept_func = [
+                real_func = std::forward<FuncType>(func),
+                error_handler = std::forward<ErrorHandlerType>(error_handler)]() mutable noexcept
+                -> boost::asio::awaitable<bool> {
+            try {
+                auto ok = co_await real_func();
+                co_return ok;
+            } catch (std::exception &e) {
+                error_handler(e);
+                co_return false;
+            }
+            assert(false);
+            co_return false;
+        };
+        return std::move(noexcept_func);
     }
 
 }

src/utility/versatile_buffer2.hpp

@@ -51,6 +51,15 @@ namespace sophiar {
 
         size_t max_length() const { return length; }
 
+        void ensure_length(size_t ext_length) {
+            if (ext_length <= length) return;
+            auto ext_buf = new char[ext_length];
+            memcpy(ext_buf, buf, length);
+            delete[] buf;
+            buf = ext_buf;
+            length = ext_length;
+        }
+
     private:
         char *buf;
         size_t length;
@@ -61,6 +70,15 @@ namespace sophiar {
         extern_memory(char *extern_buf, size_t buf_length)
                 : buf(extern_buf), length(buf_length) {}
 
+        extern_memory(extern_memory &&other) = default; // to suppress some warning
+
+        explicit extern_memory(dynamic_memory &other)
+                : buf(other.data()), length(other.max_length()) {}
+
+        template<size_t length>
+        explicit extern_memory(static_memory<length> &other)
+                :buf(other.data()), length(other.max_length()) {}
+
         char *data() { return buf; }
 
         const char *data() const { return buf; }
@@ -178,6 +196,10 @@ namespace sophiar {
             return buffer;
         }
 
+        size_t get_cur_pos() const {
+            return cur_pos;
+        }
+
         const char *cur_data() const {
             return buffer.data() + cur_pos;
         }
@@ -263,6 +285,10 @@ namespace sophiar {
             cur_pos += offset;
         }
 
+        size_t get_cur_pos() const {
+            return cur_pos;
+        }
+
         char *cur_data() {
             return buffer.data() + cur_pos;
         }

tests/core/datanode_base.cpp

@@ -260,7 +260,7 @@ BOOST_AUTO_TEST_CASE(test_datanode_base) {
     std::ifstream config_file("data/datanode_base_config.json");
     BOOST_TEST(config_file.is_open());
 
-    global_sophiar_manager.build_from_config(nlohmann::json::parse(config_file));
+    global_sophiar_manager.load_config_and_start(nlohmann::json::parse(config_file));
 
     co_spawn(global_context, test_datanode_base_1(), detached);
 //    co_spawn(high_freq_context, test_datanode_base_2(), detached);

tests/core/sophiar_manager.cpp

@@ -2,7 +2,9 @@
 
 #include "core/sophiar_manager.h"
 #include "core/tristate_obj.h"
+#include "core/basic_obj_types.hpp"
 #include "utility/debug_utility.hpp"
+#include "utility/global_obj_helper.hpp"
 
 #include <boost/asio/co_spawn.hpp>
 #include <boost/asio/detached.hpp>
@@ -26,16 +28,11 @@ using namespace std::chrono_literals;
 struct source_node_type : public tristate_obj {
     DEFAULT_NEW_INSTANCE(source_node_type)
 
-    void load_construct_config(const nlohmann::json &config) override {
-        BOOST_TEST(config.empty());
-        get_manager().register_signal(this, "output", output_signal);
-    };
+    global_obj_auto_sync_delegate<double_obj> out_obj;
 
     boost::asio::awaitable<bool> on_init(const nlohmann::json &config) override {
-        assert(config.contains("init_value"));
-        assert(config["init_value"].is_number());
-        init_value = config["init_value"].get<int64_t>();
-        co_return true;
+        init_value = config["output_obj_name"].get<int64_t>();
+        co_return true; // TODO
     }
 
     boost::asio::awaitable<bool> on_start(const nlohmann::json &config) override {
@@ -138,7 +135,7 @@ BOOST_AUTO_TEST_CASE(test_sophiar_manager) {
     std::ifstream config_file("data/sophiar_manager_config.json");
     BOOST_TEST(config_file.is_open());
 
-    global_sophiar_manager.build_from_config(nlohmann::json::parse(config_file));
+    global_sophiar_manager.load_config_and_start(nlohmann::json::parse(config_file));
 
     co_spawn(global_context, []() -> awaitable<void> {
         co_await global_sophiar_manager.switch_mode("mode_a");

tests/data/sophiar_manager_config.json

@@ -1,131 +1,37 @@
 {
-  "mode_list": [
-    {
-      "name": "mode_a"
-    },
-    {
-      "name": "mode_b",
-      "degrade_list": [
-        "mode_a"
-      ]
-    }
-  ],
+  "listen_port": 5277,
   "object_list": [
     {
       "type": "source_node_type",
       "name": "source",
-      "enabled_modes": "all",
-      "construct_config": {},
-      "init_configs": [
-        {
-          "modes": [
-            "mode_a"
-          ],
-          "config": {
-            "init_value": 2
-          }
-        },
-        {
-          "modes": [
-            "mode_b"
-          ],
-          "config": {
-            "init_value": 3
-          }
-        }
-      ],
-      "start_configs": [
-        {
-          "modes": [
-            "mode_a"
-          ],
-          "config": {
-            "start_value": 3
-          }
-        },
-        {
-          "modes": [
-            "mode_b"
-          ],
-          "config": {
-            "start_value": 4
-          }
-        }
-      ]
+      "init_config": {
+        "output_obj_name": "out_obj"
+      },
+      "start_configs": {
+        "start_value": 234
+      }
     },
     {
       "type": "proxy_node_type",
       "name": "proxy",
-      "enabled_modes": [
-        "mode_b"
-      ],
-      "construct_config": {},
-      "init_configs": [
-        {
-          "modes": "all",
-          "config": {
-            "init_value": 5
-          }
-        }
-      ],
-      "start_configs": [
-        {
-          "modes": "all",
-          "config": {}
-        }
+      "init_configs": {
+        "input_obj_name": "out_obj",
+        "output_obj_name": "proxy_obj"
+      },
+      "start_configs": {
+        "start_value": 10000
+      },
+      "dependencies": [
+        "source"
       ]
     },
     {
       "type": "target_node_type",
       "name": "target",
-      "enabled_modes": "all",
-      "construct_config": {},
-      "init_configs": [
-        {
-          "modes": "all",
-          "config": {}
-        }
-      ],
-      "start_configs": [
-        {
-          "modes": "all",
-          "config": {}
-        }
-      ]
-    }
-  ],
-  "connection_list": [
-    {
-      "modes": [
-        "mode_a"
-      ],
-      "connections": [
-        {
-          "signal_object": "source",
-          "signal_name": "output",
-          "slot_object": "target",
-          "slot_name": "input"
-        }
-      ]
-    },
-    {
-      "modes": [
-        "mode_b"
-      ],
-      "connections": [
-        {
-          "signal_object": "source",
-          "signal_name": "output",
-          "slot_object": "proxy",
-          "slot_name": "input"
-        },
-        {
-          "signal_object": "proxy",
-          "signal_name": "output",
-          "slot_object": "target",
-          "slot_name": "input"
-        }
-      ]
+      "init_configs": {
+        "listen_obj_name": "out_obj"
+      },
+      "start_configs": {}
     }
   ]
 }

tests/utility/coro_worker.cpp

@@ -95,11 +95,31 @@ awaitable<void> test_coro_worker_5() {
     co_return;
 }
 
+awaitable<void> test_coro_worker_6() {
+    int cnt = 0, exit_cnt = 0;
+    auto worker = make_interval_coro_worker(global_context, 10ms, [&]() -> awaitable<bool> {
+        co_await coro_sleep(50ms);
+        ++cnt;
+        co_return true;
+    }, [&]() {
+        ++exit_cnt;
+    });
+    worker->run();
+    co_await coro_sleep(210ms);
+    BOOST_TEST(exit_cnt == 0);
+    worker->cancel();
+    co_await worker->coro_wait_stop();
+    BOOST_TEST(cnt == 4);
+    BOOST_TEST(exit_cnt == 1);
+    co_return;
+}
+
 BOOST_AUTO_TEST_CASE(test_coro_worker) {
     co_spawn(global_context, test_coro_worker_1(), detached);
     co_spawn(global_context, test_coro_worker_2(), detached);
     co_spawn(global_context, test_coro_worker_3(), detached);
     co_spawn(global_context, test_coro_worker_4(), detached);
     co_spawn(global_context, test_coro_worker_5(), detached);
+    co_spawn(global_context, test_coro_worker_6(), detached);
     global_context.run();
 }