diff --git a/config/config.json b/config/config.json index bc8e183..49d2622 100644 --- a/config/config.json +++ b/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", - "port": 80 + "gateway": "192.168.1.1" }, { "key": "wifi", "ip": "192.168.10.193", + "port": 80, "netmask": "255.255.255.0", - "gateway": "192.168.10.1", - "port": 80 + "gateway": "192.168.10.1" } ] }, @@ -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, diff --git a/module/Local_Server/server/io_coro.cpp b/module/Local_Server/server/io_coro.cpp index 6b47e94..d8083d5 100644 --- a/module/Local_Server/server/io_coro.cpp +++ b/module/Local_Server/server/io_coro.cpp @@ -1,82 +1,228 @@ #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 +#include +#include +#include - +static std::string thread_id_str() { + std::ostringstream oss; + oss << std::this_thread::get_id(); + return oss.str(); +} std::vector> get_all() { - auto g = Global::instance(); - std::vector> ret; - - auto sources = g->mode_acs.data_source_config.map.list(); - for (auto &source : sources) { - ret.emplace_back(source); - } - auto feeds = g->mode_acs.data_feed_config.map.list(); - for (auto &feed : feeds) { - ret.emplace_back(feed); - } - return ret; + auto g = Global::instance(); + std::vector> ret; + auto sources = g->mode_acs.data_source_config.map.list(); + for (auto& source : sources) { + ret.emplace_back(source); + } + auto feeds = g->mode_acs.data_feed_config.map.list(); + for (auto& feed : feeds) { + ret.emplace_back(feed); + } + return ret; } bool Coro::has_running_loop_tasks() { - auto list = get_all(); - for (auto &li : list) { - if (li->running()) { - return true; - } - } - return false; + auto list = get_all(); + for (auto& li : list) { + if (li->running()) { + return true; + } + } + return false; } void Coro::start() { - std::cout << "start_io_coro" << std::endl; - running.store(true, std::memory_order_release); - executor_ = runtime_.make_worker_thread_executor(); - data_feed_thread_task = - std::make_unique>(coro_thread()); + std::cout << "start_io_coro" << std::endl; + running.store(true, std::memory_order_release); + io = runtime_.make_executor( + [](std::string_view thread_name) { + 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( + "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>(coro_thread()); } void Coro::stop() { - running.store(false, std::memory_order_release); - auto list = get_all(); - for (auto &li : list) { - li->async_stop(); - } - for (auto &li : list) { - li->sync_wait(); - } + 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) { + std::cout << std::format("请求停止 {}:{}\n", li->type, li->key); + li->async_stop(); + } + 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 Coro::sleep_for(std::chrono::milliseconds ms) { + co_await runtime_.timer_queue()->make_delay_object(ms, io); + co_return; +} concurrencpp::result Coro::coro_thread() { - 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; + 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 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 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& 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 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; +}