执行改成多线程

module/Local_Server/server/With_Loop_Coro.cpp

@@ -1,32 +1,46 @@
 #include "With_Loop_Coro.h"
-With_Loop_Coro::~With_Loop_Coro() {}
+With_Loop_Coro::~With_Loop_Coro() = default;
+
 void With_Loop_Coro::async_stop() {
-    enable = false;
-    loop_task.reset();
+	enable = false;
 }
-void With_Loop_Coro::sync_wait() {
-    if (!loop_task)
-        return;
-    while (loop_task->status() != concurrencpp::result_status::idle) {}
+
+void With_Loop_Coro::sync_wait() const {
+	if (!loop_task) {
+		return;
+	}
+	auto last = std::chrono::steady_clock::now();
+	while (loop_task->status() == concurrencpp::result_status::idle) {
+		auto now = std::chrono::steady_clock::now();
+		if (now - last > std::chrono::seconds(1)) {
+			std::cout << std::format("等待任务退出 {}:{}\n", type, key);
+			last = now;
+		}
+		std::this_thread::sleep_for(std::chrono::milliseconds(10));
+	}
 }
-bool With_Loop_Coro::running() {
-    if (!loop_task)
-        return false;
-    return loop_task->status() == concurrencpp::result_status::idle;
+
+bool With_Loop_Coro::running() const {
+	if (!loop_task) {
+		return false;
+	}
+	return loop_task->status() == concurrencpp::result_status::idle;
 }
-concurrencpp::result<void> With_Loop_Coro::sync_coro_loop_and_enable(
-    std::shared_ptr<concurrencpp::worker_thread_executor> executor) {
-    std::string name = type + ":" + key;
-    if (enable && !running()) {
-        co_await _open();
-        loop_task = std::make_shared<concurrencpp::result<void>>(loop_coro(executor));
-        std::cout << std::format("启动任务 {}!\n", name);
-    }
-    else if (!enable && running()) {
-        std::cout << std::format("等待停止任务 {}!\n", name);
-        sync_wait();
-        co_await _close();
-        std::cout << std::format("停止任务成功 {}!\n", name);
-    }
-    co_return;
+
+concurrencpp::result<void> With_Loop_Coro::sync_coro_loop_and_enable() {
+	std::string name = type + ":" + key;
+	if (loop_task && !running()) {
+		loop_task.reset();
+		co_await _close();
+	}
+	if (enable && !loop_task) {
+		co_await _open();
+		loop_task = std::make_shared<concurrencpp::result<void>>(loop_coro());
+		std::cout << std::format("启动任务 {}!\n", name);
+	}
+	else if (!enable && running()) {
+		std::cout << std::format("请求停止任务 {}!\n", name);
+		force_close();
+	}
+	co_return;
 }

module/Local_Server/server/With_Loop_Coro.h

@@ -3,37 +3,37 @@
 #include <chrono>
 #include <concurrencpp/concurrencpp.h>
 #include <exception>
+#include <format>
 #include <iostream>
 #include <memory>
 #include <sstream>
+#include <thread>
 #include <utility>
 #include <vector>
-
 #include "Core/Base/JSON.h"
-
 class With_Loop_Coro_Data {
 public:
-    std::string key;
-    Psc::Copyable_Atomic<bool> enable{};
-    std::string type;
-    PSC_USE_JSON
+	std::string key;
+	Psc::Copyable_Atomic<bool> enable{};
+	std::string type;
+	PSC_USE_JSON
 };
-
-
 class With_Loop_Coro : public With_Loop_Coro_Data {
 public:
-  virtual ~With_Loop_Coro();
-  std::shared_ptr<concurrencpp::result<void>> loop_task;
-  virtual concurrencpp::result<void>
-  loop_coro(std::shared_ptr<concurrencpp::worker_thread_executor> executor) = 0;
-  void async_stop();
-  void sync_wait();
-  bool running();
-  // 同步协程循环 和enable的关系
-  concurrencpp::result<void> sync_coro_loop_and_enable(
-      std::shared_ptr<concurrencpp::worker_thread_executor> executor);
+	virtual ~With_Loop_Coro();
+	std::shared_ptr<concurrencpp::result<void>> loop_task;
+	virtual concurrencpp::result<void> loop_coro() = 0;
+	void async_stop();
+	void sync_wait() const;
+	bool running() const;
+	virtual void force_close() {}
+	concurrencpp::result<void> sync_coro_loop_and_enable();
 
-  // 这个关闭打开指的是内部的socket serial 等的关闭打开
-  virtual concurrencpp::result<void> _open() { co_return; }
-  virtual concurrencpp::result<void> _close() { co_return; }
+	virtual concurrencpp::result<void> _open() {
+		co_return;
+	}
+
+	virtual concurrencpp::result<void> _close() {
+		co_return;
+	}
 };

module/Local_Server/server/io_coro.h

@@ -1,17 +1,21 @@
 #pragma once
-
-#include "With_Loop_Coro.h"
-
-class Coro {
+#include <atomic>
+#include <chrono>
+#include <concurrencpp/concurrencpp.h>
+#include <iostream>
+#include <memory>
+#include "psc_global_include/Singleton.hpp"
+class Coro : public Psc::Singleton<Coro> {
 public:
-  void start();
-  void stop();
-
+	void start();
+	void stop();
+	concurrencpp::result<void> sleep_for(std::chrono::milliseconds ms);
+	std::shared_ptr<concurrencpp::thread_pool_executor> process_data;
+	std::shared_ptr<concurrencpp::worker_thread_executor> io;
 private:
-  concurrencpp::result<void> coro_thread();
-  bool has_running_loop_tasks();
-  concurrencpp::runtime runtime_;
-  std::shared_ptr<concurrencpp::worker_thread_executor> executor_;
-  std::unique_ptr<concurrencpp::result<void>> data_feed_thread_task;
-  std::atomic<bool> running = false;
+	concurrencpp::result<void> coro_thread();
+	bool has_running_loop_tasks();
+	concurrencpp::runtime runtime_;
+	std::unique_ptr<concurrencpp::result<void>> data_feed_thread_task;
+	std::atomic<bool> running = false;
 };