执行改成多线程

config/config.json

@@ -46,16 +46,16 @@
      {
        "key": "device",
        "ip": "192.168.1.75",
+       "port": 80,
        "netmask": "255.255.255.0",
-       "gateway": "192.168.1.1",
+       "gateway": "192.168.1.1"
-       "port": 80
      },
      {
        "key": "wifi",
        "ip": "192.168.10.193",
+       "port": 80,
        "netmask": "255.255.255.0",
-       "gateway": "192.168.10.1",
+       "gateway": "192.168.10.1"
-       "port": 80
      }
     ]
   },
@@ -80,7 +80,7 @@
       "list": [
        {
          "key": "Port_10003",
-         "enable": true,
+         "enable": false,
          "type": "Data_Feed_TCP_Server",
          "output_format": {
            "type": "BIN",
@@ -117,8 +117,8 @@
            "use_mode_ac": false,
            "sbs_only_pos": true
          },
-         "url": "192.168.1.75",
+         "port": 30004,
-         "port": 30004
+         "url": "192.168.1.75"
        },
        {
          "key": "Port_10005",
@@ -155,12 +155,12 @@
       "list": [
        {
          "key": "lzy_dll",
-         "enable": true,
+         "enable": false,
+         "type": "Dll_Data_Source",
          "base_station_show": true,
          "aircraft_show": true,
          "color": "#454641",
          "aircraft_pixel_size": 28,
-         "type": "Dll_Data_Source",
          "lat": 37.433547,
          "lon": 121.408730,
          "alt": 23.000000,
@@ -177,11 +177,11 @@
        {
          "key": "aaaa",
          "enable": false,
+         "type": "File_Data_Source",
          "base_station_show": true,
          "aircraft_show": true,
          "color": "#d1a54c",
          "aircraft_pixel_size": 20,
-         "type": "File_Data_Source",
          "lat": 37.000000,
          "lon": 121.000000,
          "alt": 0.000000,

module/Local_Server/server/io_coro.cpp

@@ -1,23 +1,29 @@
 #include "io_coro.h"
 #include "../server/Global.h"
 #include "../server/Mode_Msg_Buffer.h"
-
 #include "Core/Statistics/Frequency_Limit.h"
 #include "Local_Server/Data_Feed/Data_Feed.h"
 #include "Local_Server/Data_Source/Data_Source.h"
+#include <chrono>
+#include <format>
+#include <sstream>
+#include <thread>
 
-
+static std::string thread_id_str() {
+	std::ostringstream oss;
+	oss << std::this_thread::get_id();
+	return oss.str();
+}
 
 std::vector<std::shared_ptr<With_Loop_Coro>> get_all() {
 	auto g = Global::instance();
 	std::vector<std::shared_ptr<With_Loop_Coro>> ret;
-
 	auto sources = g->mode_acs.data_source_config.map.list();
-  for (auto &source : sources) {
+	for (auto& source : sources) {
 		ret.emplace_back(source);
 	}
 	auto feeds = g->mode_acs.data_feed_config.map.list();
-  for (auto &feed : feeds) {
+	for (auto& feed : feeds) {
 		ret.emplace_back(feed);
 	}
 	return ret;
@@ -25,7 +31,7 @@ std::vector<std::shared_ptr<With_Loop_Coro>> get_all() {
 
 bool Coro::has_running_loop_tasks() {
 	auto list = get_all();
-  for (auto &li : list) {
+	for (auto& li : list) {
 		if (li->running()) {
 			return true;
 		}
@@ -36,47 +42,187 @@ bool Coro::has_running_loop_tasks() {
 void Coro::start() {
 	std::cout << "start_io_coro" << std::endl;
 	running.store(true, std::memory_order_release);
-  executor_ = runtime_.make_worker_thread_executor();
+	io = runtime_.make_executor<concurrencpp::worker_thread_executor>(
-  data_feed_thread_task =
+		[](std::string_view thread_name) {
-      std::make_unique<concurrencpp::result<void>>(coro_thread());
+			std::cout << std::format("io:{} 协程启动！\n", thread_name);
+		},
+		[](std::string_view thread_name) {
+			std::cout << std::format("io:{} 协程销毁！\n", thread_name);
+		});
+	process_data = runtime_.make_executor<concurrencpp::thread_pool_executor>(
+		"process_data",
+		4,
+		std::chrono::seconds(5),
+		[](std::string_view thread_name) {
+			std::cout << std::format("{} 协程启动！\n", thread_name);
+		},
+		[](std::string_view thread_name) {
+			std::cout << std::format("{} 协程销毁！\n", thread_name);
+		});
+	std::cout << std::format("process_data concurrency: {}\n", process_data->max_concurrency_level());
+	data_feed_thread_task = std::make_unique<concurrencpp::result<void>>(coro_thread());
 }
 
 void Coro::stop() {
-  running.store(false, std::memory_order_release);
+	if (!running.exchange(false, std::memory_order_acq_rel)) {
+		return;
+	}
+	std::cout << "stop_io_coro begin" << std::endl;
 	auto list = get_all();
-  for (auto &li : list) {
+	for (auto& li : list) {
+		std::cout << std::format("请求停止 {}:{}\n", li->type, li->key);
 		li->async_stop();
 	}
-  for (auto &li : list) {
+	for (auto& li : list) {
+		std::cout << std::format("强制关闭 {}:{}\n", li->type, li->key);
+		li->force_close();
+	}
+	if (data_feed_thread_task) {
+		std::cout << "等待 coro_thread 退出" << std::endl;
+		data_feed_thread_task->get();
+		data_feed_thread_task.reset();
+		std::cout << "coro_thread 已退出" << std::endl;
+	}
+	for (auto& li : list) {
 		li->sync_wait();
 	}
+	for (auto& li : list) {
+		li->loop_task.reset();
+	}
+	process_data.reset();
+	io.reset();
+	std::cout << "stop_io_coro end" << std::endl;
 }
 
-
+concurrencpp::result<void> Coro::sleep_for(std::chrono::milliseconds ms) {
-concurrencpp::result<void> Coro::coro_thread() {
+	co_await runtime_.timer_queue()->make_delay_object(ms, io);
-  co_await concurrencpp::resume_on(executor_);
-
-  while (running.load(std::memory_order_acquire)) {
-    auto list = get_all();
-    // 同步协程循环 和enable的关系
-    for (auto &li : list) co_await li->sync_coro_loop_and_enable(executor_);
-
-    co_await concurrencpp::resume_on(executor_);
-  }
-  auto list = get_all();
-  for (auto &li : list)  li->sync_wait();
-  // 退出时清理
-  auto start = std::chrono::steady_clock::now();
-  while (has_running_loop_tasks()) {
-    if (std::chrono::steady_clock::now() - start > std::chrono::seconds(10)) {
-      std::cerr << "data loop task stop timeout" << std::endl;
-      break;
-    }
-    co_await concurrencpp::resume_on(executor_);
-  }
 	co_return;
 }
 
+concurrencpp::result<void> Coro::coro_thread() {
+	co_await concurrencpp::resume_on(io);
+	while (running.load(std::memory_order_acquire)) {
+		auto list = get_all();
+		for (auto& li : list) {
+			co_await li->sync_coro_loop_and_enable();
+		}
+		co_await sleep_for(std::chrono::milliseconds(10));
+	}
+	std::cout << "coro_thread exit begin" << std::endl;
+	auto list = get_all();
+	for (auto& li : list) {
+		li->async_stop();
+		li->force_close();
+	}
+	std::cout << "coro_thread exit end" << std::endl;
+	co_return;
+}
 
+bool debug = false;
 
+concurrencpp::result<void> Data_Feed::loop_coro() {
+	auto feed = this;
+	auto co = Coro::instance();
+	co_await concurrencpp::resume_on(co->io);
+	auto& mode_acs = Global::instance()->mode_acs;
+	auto& cfg = mode_acs.data_feed_config;
+	auto& pool = cfg.pool_;
+	auto& report_data_feed_msg_mum = mode_acs.report_data_feed_msg_mum;
+	while (feed->enable) {
+		if (debug) {
+			std::cout << std::format("{} feed loop begin {}\n", key, thread_id_str());
+		}
+		std::vector<std::string> messages;
+		{
+			if (!feed->running() || !feed->registered()) {
+				break;
+			}
+			co_await feed->handle_in_loop_coro();
+			auto s_num = feed->msg_buffer.mode_s_msg_num.load();
+			auto other_num = feed->msg_buffer.mode_other_msg_num.load();
+			const std::vector<std::string*>& all = feed->msg_buffer.get_all();
+			Pool_Guard pg(&pool, all);
+			auto num = report_data_feed_msg_mum.load();
+			auto monitor_msg_live = mode_acs.monitor_msg_live.load();
+			if (monitor_msg_live && num != 0 && all.size() > num && too_many_msg_limit.test()) {
+				std::ostringstream oss;
+				oss << "feed_key:" << feed->key << "   ";
+				oss << "recv_s:" << s_num << "   ";
+				oss << "recv_other:" << other_num << "   ";
+				std::cout << oss.str() << std::endl;
+			}
+			messages.reserve(all.size());
+			for (const auto* str : all) {
+				if (!str || str->empty()) {
+					std::cout << "empty feed message:" << feed->key << std::endl;
+					continue;
+				}
+				messages.emplace_back(*str);
+			}
+		}
+		if (messages.empty()) {
+			co_await co->sleep_for(std::chrono::milliseconds(10));
+			continue;
+		}
+		if (!feed->running() || !feed->registered()) {
+			break;
+		}
+		for (auto& msg : messages) {
+			if (!feed->enable) {
+				break;
+			}
+			if (debug) {
+				std::cout << std::format("{} before send {}\n", key, thread_id_str());
+			}
+			co_await feed->send_coro(msg);
+			if (debug) {
+				std::cout << std::format("{} after send {}\n", key, thread_id_str());
+			}
+		}
+		co_await concurrencpp::resume_on(co->io);
+	}
+	std::cout << std::format("{} feed loop exit {}\n", key, thread_id_str());
+	co_return;
+}
 
+concurrencpp::result<void> Data_Source::loop_coro() {
+	auto source = this;
+	auto co = Coro::instance();
+	co_await concurrencpp::resume_on(co->io);
+	while (enable) {
+		if (debug) {
+			std::cout << std::format("{} source loop begin {}\n", key, thread_id_str());
+		}
+		if (!source->running() || !source->registered()) {
+			break;
+		}
+		if (debug) {
+			std::cout << std::format("{} before handle {}\n", key, thread_id_str());
+		}
+		co_await source->handle_in_loop_coro();
+		if (debug) {
+			std::cout << std::format("{} before read {}\n", key, thread_id_str());
+		}
+		auto mode_data = co_await source->read_coro();
+		if (debug) {
+			std::cout << std::format("{} after read {}\n", key, thread_id_str());
+		}
+		if (!enable || !source->running() || !source->registered()) {
+			break;
+		}
+		co_await concurrencpp::resume_on(co->process_data);
+		if (debug) {
+			std::cout << std::format("{} before process {}\n", key, thread_id_str());
+		}
+		auto num = source->process_mode_acs_data(mode_data);
+		if (debug) {
+			std::cout << std::format("{} after process {} {}\n", key, num, thread_id_str());
+		}
+		if (num == 0) {
+			co_await co->sleep_for(std::chrono::milliseconds(10));
+		}
+		co_await concurrencpp::resume_on(co->io);
+	}
+	std::cout << std::format("{} source loop exit {}\n", key, thread_id_str());
+	co_return;
+}