ae/ECAP_Server
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

改进协程循环逻辑

Browse Source
This commit is contained in:
种太阳的热带美短
2026-06-25 15:01:29 +08:00
parent 03bfac8d0c
commit 1504db9db0
12 changed files with 409 additions and 484 deletions
Download Patch File Download Diff File Expand all files Collapse all files
config/config.json
+1 -1
View File
@@ -122,7 +122,7 @@
},
{
"key": "Port_10005",
"enable": true,
"enable": false,
"type": "Data_Feed_TCP_Client",
"output_format": {
"type": "BIN",
module/Local_Server.rar
BIN
View File
Binary file not shown.
module/Local_Server/Data_Feed/Data_Feed.cpp
+67 -9
View File
@@ -1,5 +1,10 @@
#include "Local_Server/Data_Feed/Data_Feed.h"
#include "../server/Global.h"
bool Data_Feed::registered() {
auto all_feed = Global::instance()->mode_acs.data_feed_config.map.list();
for (auto& item : all_feed) {
@@ -100,11 +105,11 @@ void Data_feed_Config::server(Global* g) {
HTTP_REQUIRE_VALUE(key, params.try_get_string("key"))
HTTP_REQUIRE_VALUE(t, map.get(key))
// 直接关闭
t->close();
// t->close();
t->from_json(&params);
if (t->enable) {
t->check_and_open();
}
// if (t->enable) {
// t->check_and_open();
// }
Global::save();
g->mode_acs.source_feed_relation_config.set_need_refresh();
});
@@ -117,10 +122,10 @@ void Data_feed_Config::server(Global* g) {
auto df = create_data_feed_from_type(t);
df->from_json(data);
HTTP_REQUIRE_VALUE(enable, data->try_get_bool("enable"))
if (enable)
{
df->check_and_open();
}
// if (enable)
// {
// df->check_and_open();
// }
HTTP_REQUIRE_TRUE(map.insert(index, df), "index")
res->setBody(warp(to_json()).to_json_string());
Global::save();
@@ -131,7 +136,7 @@ void Data_feed_Config::server(Global* g) {
CHECK_JSON_PARAM
HTTP_REQUIRE_VALUE(index, params.try_get_number<int>("index"))
HTTP_REQUIRE_VALUE(df, map.try_get(index))
df->close();
df->async_stop();
HTTP_REQUIRE_TRUE(map.remove(index), "index")
g->save();
res->setBody(warp(to_json()).to_json_string());
@@ -194,3 +199,56 @@ Psc::JSON BIN_Msg_Buffer::state_json() {
}
return VAR_JSON_7(mode_s_msg_num, mode_other_msg_num, output_packet_size, cache_size, size, pool_capacity, pool_free_count);
}
concurrencpp::result<void> Data_Feed::loop_coro(
std::shared_ptr<concurrencpp::worker_thread_executor> executor_) {
auto feed = this;
co_await concurrencpp::resume_on(executor_);
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) {
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 concurrencpp::resume_on(executor_);
continue;
}
if (!feed->running() || !feed->registered()) {
break;
}
for (auto &msg : messages) {
co_await feed->send_coro(msg);
}
co_await concurrencpp::resume_on(executor_);
}
co_return;
}
module/Local_Server/Data_Feed/Data_Feed.h
+4 -23
View File
@@ -1,6 +1,7 @@
#pragma once
#include "global.h"
#include "Local_Server/server/With_Loop_Coro.h"
enum class Output_Data_Format {
AVR,
@@ -44,16 +45,13 @@ struct Output_Format {
};
class Data_Feed {
class Data_Feed : public With_Loop_Coro {
public:
virtual ~Data_Feed() = default;
std::unique_ptr<concurrencpp::result<void>> loop_task;
BIN_Msg_Buffer msg_buffer{};
std::string key;
bool enable = false;
std::string type;
Output_Format output_format;
Frequency_Limit_Multi sbs_flm{};
concurrencpp::result<void> loop_coro(std::shared_ptr<concurrencpp::worker_thread_executor> executor) override;
virtual void handle_in_loop() {
}
@@ -86,17 +84,7 @@ public:
ret.append_list(get_custom_state_json().children);
return ret;
}
bool check_and_open() {
if (_open_) return true;
bool ret = _open();
_open_ = true;
return ret;
}
void close() {
_open_ = false;
_close();
}
std::string to_string() {
return "Data_Feed{" + VAR_STR_3(key, enable, type) + "}";
@@ -105,14 +93,7 @@ public:
bool registered();
protected:
Data_Feed() = default;
bool _open_ = false;
virtual bool _open() {
return true;
}
virtual void _close() {
}
Frequency_Limit too_many_msg_limit;
};
module/Local_Server/Data_Source/Data_Source.cpp
+21 -6
View File
@@ -590,8 +590,8 @@ void Data_Source_Config::server(Global* g) {
HTTP_REQUIRE_VALUE(key, params.try_get_string("key"))
HTTP_REQUIRE_VALUE(ds, map.get(key))
// 直接关闭
ds->enable = false;
ds->close();
// ds->enable = false;
// ds->close();
ds->from_json(&params);
g->save();
@@ -621,11 +621,8 @@ void Data_Source_Config::server(Global* g) {
CHECK_JSON_PARAM
HTTP_REQUIRE_VALUE(index, params.try_get_number<int>("index"))
HTTP_REQUIRE_VALUE(ds, map.try_get(index))
ds->enable = false;
ds->close();
ds->async_stop();
HTTP_REQUIRE_TRUE(map.remove(index), "index")
g->save();
res->setBody(warp(to_json()).to_json_string());
g->mode_acs.source_feed_relation_config.set_need_refresh();
@@ -650,3 +647,21 @@ void Data_Source_Config::server(Global* g) {
}
concurrencpp::result<void> Data_Source::loop_coro(
std::shared_ptr<concurrencpp::worker_thread_executor> executor_) {
auto source = this;
co_await concurrencpp::resume_on(executor_);
while (enable) {
if (!source->running() || !source->registered()) {
break;
}
co_await source->handle_in_loop_coro();
auto mode_data = co_await source->read_coro();
if (!enable || !source->running() || !source->registered()) {
break;
}
source->process_mode_acs_data(mode_data);
co_await concurrencpp::resume_on(executor_);
}
co_return;
}
module/Local_Server/Data_Source/Data_Source.h
+13 -28
View File
@@ -4,6 +4,7 @@
#include "Data_Source_Handler.h"
namespace Psc {
class SM_RingBuffer;
}
@@ -67,19 +68,16 @@ protected:
class Data_Source : public std::enable_shared_from_this<Data_Source>, public Data_Source_Handler {
public:
~Data_Source() override = default;
std::unique_ptr<concurrencpp::result<void>> loop_task;
std::shared_ptr<Data_Source> that();
virtual concurrencpp::result<std::string> read_coro() {co_return "";};
bool registered() const;
virtual Psc::JSON get_custom_state_json() = 0;
Psc::JSON get_state();
std::string last_char; // 用于处理奇数字节的情�?
// std::shared_ptr<Input_Format> parse_format;
std::string last_char; // 用于处理奇数字节
Data_Source();
concurrencpp::result<void> loop_coro(std::shared_ptr<concurrencpp::worker_thread_executor> executor) override;
std::shared_ptr<SSR::Mode_Msg> last_prase_msg = nullptr;
virtual concurrencpp::result<void> handle_in_loop_coro(){co_return;}
std::atomic<bool> enable{};
std::string type;
std::atomic<bool> base_station_show{};
std::atomic<bool> aircraft_show{};
@@ -144,29 +142,12 @@ public:
Get_J(keep_mode);
// parse_format->from_json(that_json->get("parse_format"));
}
bool open() {
if (!_open_) {
_open_ = _open();
return _open_;
}
return true;
}
bool is_open() const { return _open_; }
void close() {
if (_open_) {
_open_ = false;
_close();
}
}
std::string buffer;
protected:
bool _open_ = false;
virtual bool _open() = 0;
virtual void _close() = 0;
};
struct TCP_Client_Data_Source : Data_Source {
class TCP_Client_Data_Source : public Data_Source {
public:
concurrencpp::result<void> handle_in_loop_coro() override {
co_await cli.tick_coro();
co_return;
@@ -212,7 +193,8 @@ struct TCP_Client_Data_Source : Data_Source {
co_return co_await cli.read_coro();
}
};
struct Serial_Data_Source : Data_Source {
class Serial_Data_Source : public Data_Source {
public:
std::string port_name;
Baud_Rate_Type baud_rate{};
Psc::JSON get_custom_state_json() override {
@@ -283,7 +265,8 @@ enum class File_Data_Type {
enum Play_Mode { one, loop, analysis, time_batch };
struct File_Data_Source : public Data_Source {
class File_Data_Source : public Data_Source {
public:
std::string file_path;
File_Data_Type data_type = File_Data_Type::BIN_Blank_Text;
Play_Mode play_mode = Play_Mode::one;
@@ -343,7 +326,8 @@ struct File_Data_Source : public Data_Source {
std::vector<std::string> part_infos;
std::mutex mtx;
};
struct Dll_Data_Source : Data_Source {
class Dll_Data_Source : public Data_Source {
public:
Psc::JSON get_custom_state_json() override {
auto ret = VAR_JSON_1(state);
ret.append({"read_vaild_len", vs.to_string()});
@@ -410,7 +394,8 @@ struct Dll_Data_Source : Data_Source {
};
struct Shared_Memory_Data_Source : Data_Source {
class Shared_Memory_Data_Source : public Data_Source {
public:
Psc::JSON get_custom_state_json() override {
return Psc::JSON::object();
}
module/Local_Server/Data_Source/Database.h
+9 -8
View File
@@ -1,14 +1,15 @@
#pragma once
#include "BaseStation.h"
#include "../Aircraft/Aircraft.h"
#include "../Aircraft/Flight_VTO.h"
#include "../External_Database/export.h"
#include <unordered_map>
#include <memory>
#include <vector>
#include <shared_mutex>
#include <functional>
#include "BaseStation.h"
#include "Local_Server/server/With_Loop_Coro.h"
#include <functional>
#include <memory>
#include <shared_mutex>
#include <unordered_map>
#include <vector>
template <
typename Key_Type,
@@ -191,7 +192,7 @@ protected:
};
class DataBase : public SSR::Data_Source_Interface {
class DataBase : public SSR::Data_Source_Interface, public With_Loop_Coro {
public:
std::shared_ptr<SSR::Aircraft_Info> get_aircraft(const std::string& icao) override;
std::shared_ptr<SSR::Aircraft_Info> create_aircraft(const std::string& icao) override;
@@ -202,7 +203,7 @@ public:
std::string get_key() override {
return key;
}
std::string key;
std::atomic<size_t> have_pos_aircraft_num{};
Psc::JSON get_all_aircraft_json();
Psc::JSON get_aircraft_list_after(time_t timestamp);
module/Local_Server/server/With_Loop_Coro.cpp
+33
View File
@@ -0,0 +1,33 @@
#include "With_Loop_Coro.h"
With_Loop_Coro::~With_Loop_Coro() {
}
void With_Loop_Coro::async_stop() {
enable = false;
loop_task.reset();
}
void With_Loop_Coro::sync_wait() {
if (!loop_task) return;
while (loop_task->status() != concurrencpp::result_status::idle)
;
}
bool With_Loop_Coro::running() {
if (!loop_task) return false;
return loop_task->status() == concurrencpp::result_status::idle;
}
void With_Loop_Coro::sync_coro_loop_and_enable(
std::shared_ptr<concurrencpp::worker_thread_executor> executor) {
if (enable && !running()) {
_open();
loop_task = std::make_shared<concurrencpp::result<void>>(loop_coro(executor));
// 启动任务
std::cout <<"启动任务！ " << key << "!" <<std::endl;
} else if (!enable && running()) {
std::cout <<"等待停止任务！ " << key << "!" <<std::endl;
sync_wait();
_close();
std::cout <<"停止任务成功！ " << key << "!" <<std::endl;
}
}
module/Local_Server/server/With_Loop_Coro.h
+29
View File
@@ -0,0 +1,29 @@
#pragma once
#include <atomic>
#include <chrono>
#include <concurrencpp/concurrencpp.h>
#include <exception>
#include <iostream>
#include <memory>
#include <sstream>
#include <utility>
#include <vector>
class With_Loop_Coro {
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的关系
void sync_coro_loop_and_enable(std::shared_ptr<concurrencpp::worker_thread_executor> executor);
std::string key;
std::atomic<bool> enable{};
virtual bool _open() { return true; }
virtual void _close() {}
};
module/Local_Server/server/io_coro.cpp
+72 -175
View File
@@ -1,186 +1,83 @@
#include "io_coro.h"
#include "../server/Global.h"
#include "../server/Mode_Msg_Buffer.h"
#include "io_coro.h"
#include "Core/Statistics/Frequency_Limit.h"
#include "Local_Server/Data_Feed/Data_Feed.h"
#include "Local_Server/Data_Source/Data_Source.h"
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) {
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;
}
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<concurrencpp::result<void>>(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();
}
}
concurrencpp::result<void> Coro::coro_thread() {
co_await concurrencpp::resume_on(executor_);
auto g = Global::instance();
for (auto& feed : g->mode_acs.data_feed_config.map.list()) {
if (feed->enable) {
auto ret = feed->check_and_open();
if (!ret) {
std::cout << "Data feed first open failed: " << feed->key << std::endl;
}
}
}
while (running.load(std::memory_order_acquire)) {
auto sources = Global::instance()->mode_acs.data_source_config.map.list();
for (auto& source : sources) {
if (!source || !source->enable) {
continue;
}
if (task_is_running(source->loop_task)) {
continue;
}
if (!source->is_open() && !source->open()) {
continue;
}
source->loop_task = std::make_unique<concurrencpp::result<
void>>(data_source_loop_coro(source));
}
auto feeds = Global::instance()->mode_acs.data_feed_config.map.list();
for (auto& feed : feeds) {
if (!feed || !feed->enable) {
continue;
}
if (task_is_running(feed->loop_task)) {
continue;
}
auto ret = feed->check_and_open();
if (!ret) {
continue;
}
feed->loop_task = std::make_unique<concurrencpp::result<void>>(data_feed_loop_coro(feed));
}
co_await concurrencpp::resume_on(executor_);
}
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(executor_);
void Coro::start() {
if (data_feed_thread_task) {
return;
}
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<concurrencpp::result<void>>(coro_thread());
}
while (running.load(std::memory_order_acquire)) {
auto list = get_all();
// 同步协程循环 和enable的关系
for (auto &li : list) li->sync_coro_loop_and_enable(executor_);
void Coro::stop() {
running.store(false, std::memory_order_release);
if (!data_feed_thread_task) {
return;
}
try {
data_feed_thread_task->get();
}
catch (const std::exception& e) {
std::cerr << "data_feed_thread_coro exception: " << e.what() << std::endl;
Psc::fail_fast_core_dump("");
}
catch (...) {
std::cerr << "data_feed_thread_coro unknown exception" << std::endl;
Psc::fail_fast_core_dump("");
}
data_feed_thread_task.reset();
executor_.reset();
}
bool Coro::task_is_running(std::unique_ptr<concurrencpp::result<void>>& task) {
if (!task) {
return false;
}
if (task->status() == concurrencpp::result_status::idle) {
return true;
}
task->get();
task.reset();
return false;
}
concurrencpp::result<void> Coro::data_source_loop_coro(std::shared_ptr<Data_Source> source) {
co_await concurrencpp::resume_on(executor_);
while (running.load(std::memory_order_acquire)) {
if (!source->enable || !source->registered()) {
break;
}
if (!source->is_open() && !source->open()) {
co_await concurrencpp::resume_on(executor_);
continue;
}
co_await source->handle_in_loop_coro();
auto mode_data = co_await source->read_coro();
if (!running.load(std::memory_order_acquire) || !source->enable || !source->registered()) {
break;
}
source->process_mode_acs_data(mode_data);
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_return;
}
concurrencpp::result<void> Coro::data_feed_loop_coro(std::shared_ptr<Data_Feed> feed) {
co_await concurrencpp::resume_on(executor_);
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 (running.load(std::memory_order_acquire)) {
std::vector<std::string> messages;
{
if (!feed->enable || !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 concurrencpp::resume_on(executor_);
continue;
}
if (!feed->enable || !feed->registered()) {
break;
}
for (auto& msg : messages) {
co_await feed->send_coro(msg);
}
co_await concurrencpp::resume_on(executor_);
}
co_return;
}
bool Coro::has_running_loop_tasks() {
auto sources = Global::instance()->mode_acs.data_source_config.map.list();
for (auto& source : sources) {
if (source && task_is_running(source->loop_task)) {
return true;
}
}
auto feeds = Global::instance()->mode_acs.data_feed_config.map.list();
for (auto& feed : feeds) {
if (feed && task_is_running(feed->loop_task)) {
return true;
}
}
return false;
co_await concurrencpp::resume_on(executor_);
}
co_return;
}
module/Local_Server/server/io_coro.h
+11 -25
View File
@@ -1,31 +1,17 @@
#pragma once
#include <concurrencpp/concurrencpp.h>
#include <atomic>
#include <chrono>
#include <exception>
#include <iostream>
#include <memory>
#include <sstream>
#include <utility>
#include <vector>
#include "Core/Statistics/Frequency_Limit.h"
#include "Local_Server/Data_Feed/Data_Feed.h"
#include "Local_Server/Data_Source/Data_Source.h"
#include "With_Loop_Coro.h"
class Coro {
public:
void start();
void stop();
void start();
void stop();
private:
bool task_is_running(std::unique_ptr<concurrencpp::result<void>>& task);
concurrencpp::result<void> data_source_loop_coro(std::shared_ptr<Data_Source> source);
concurrencpp::result<void> data_feed_loop_coro(std::shared_ptr<Data_Feed> feed);
concurrencpp::result<void> coro_thread();
bool has_running_loop_tasks();
Frequency_Limit too_many_msg_limit;
Frequency_Limit flush_limit;
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::shared_ptr<concurrencpp::worker_thread_executor> executor_;
std::unique_ptr<concurrencpp::result<void>> data_feed_thread_task;
std::atomic<bool> running = false;
};
module/Local_Server_main.cpp
+149 -209
View File
@@ -1,237 +1,177 @@
#include <iostream>
#include <thread>
#include "Local_Server/Crawler/Crawler.h"
#include "Local_Server/Data_Source/Data_Source.h"
#include "Local_Server/State_Report/State_Report.h"
#include "Local_Server/server/Global.h"
#include <SQLiteCpp/SQLiteCpp.h>
#include <SQLiteCpp/VariadicBind.h>
#include <codecvt>
#include <csignal>
#include <future>
#include "Local_Server/server/io_coro.h"
// ./server ./config.json
#ifdef __linux__
#include <malloc.h>
#endif
struct Catch_Memory {
Catch_Memory() {
sm = (std::int64_t)get_system_memory();
cur = (std::int64_t)get_process_memory();
}
std::int64_t MB = 1024 * 1024;
bool catch_ok() {
auto old_memory = cur;
cur = (std::int64_t)get_process_memory();
auto change_mb = (cur - old_memory) / MB;
auto cur_mb = cur / MB;
// std::cout << get_current_date_string() << " 时内存" << cur_mb << "MB "
// << " 变更" << change_mb << "MB" << std::endl;
if (cur > memory_max) {
auto rate = (double)cur / (double)sm;
std::cout << get_current_date_string() + " 抓住内存暴涨:" << cur_mb
Catch_Memory() {
sm = (std::int64_t)get_system_memory();
cur = (std::int64_t)get_process_memory();
}
std::int64_t MB = 1024 * 1024;
bool catch_ok() {
auto old_memory = cur;
cur = (std::int64_t)get_process_memory();
auto change_mb = (cur - old_memory) / MB;
auto cur_mb = cur / MB;
// std::cout << get_current_date_string() << " 时内存" << cur_mb << "MB "
// << " 变更" << change_mb << "MB" << std::endl;
if (cur > memory_max) {
auto rate = (double)cur / (double)sm;
std::cout << get_current_date_string() + " 抓住内存暴涨:" << cur_mb
<< "MB" << " 变更" << change_mb << "MB" << std::endl;
std::cout << "\t内存占用" << rate << "%" << std::endl;
stop_program = SIGINT;
std::cout << "\t内存占用" << rate << "%" << std::endl;
stop_program = SIGINT;
#if WIN32
#else
system("top");
system("top");
#endif
return true;
}
return false;
}
#if WIN32
std::int64_t memory_max = 1 * 1024 * 1024 * 1024;
#else
std::int64_t memory_max = 1 * 1024 * 1024 * 1024;
#endif
std::int64_t cur;
std::int64_t sm;
};
int psc_main(int argc, char *argv[]) {
std::cout << "wyc_main" << std::endl;
if (argc >= 2) {
std::string config_path = argv[1];
if (config_path.at(0) == '@') {
config_path.replace(0, 1, get_exe_dir());
}
::default_config_path = config_path;
}
// std::cout << get_stack_trace() << std::endl;
init_signal_config({SIGINT, SIGTERM},
#ifdef WIN32
{}
#else
{SIGPIPE}
#endif
);
auto clear = []() {
std::cout << "保存配置文件 " << LOG_POS << std::endl;
Config::save();
std::cout << "shutdown" << std::endl;
auto i = Global::instance();
i->svr.stop();
i->thread_manager.stop_all_thread();
Global::instance()->dsp_config.close_dsp_device();
std::cout << "destroy开始" << std::endl;
Global::destroy();
std::cout << " Base_Logger_Manager::instance()->destroy()开始" << std::endl;
Base_Logger_Manager::destroy();
std::cout << " clear完成" << std::endl;
std::cout << " 正常退出" << std::endl;
};
auto g = Global::instance();
Psc::set_fail_fast([](const std::string& name) {
// 如果是在 Global::instance() 里面失败是不能调用这个处理函数的
auto g = Global::instance();
std::cout << "set_fail_fast 退出原因： 222" << name << std::endl;
std::cout << "shutdown" << LOG_POS << std::endl;
auto id = std::this_thread::get_id();
std::cout << "id: " << id << std::endl;
Detach_Thread *dt = g->thread_manager.get_thread(id);
if (!dt)
{
std::cout << VAR_STR_1((void*)dt) << std::endl;
std::terminate();
}
std::set<std::string> excluded_thread;
excluded_thread.insert(dt->name);
std::cout << dt->name << std::endl;
// std::cout << "保存配置文件 " << LOG_POS << std::endl;
// Config::save();
g->svr.stop();
g->thread_manager.stop_all_thread(10000, excluded_thread);
std::cout << "22222222" << std::endl;
Global::instance()->dsp_config.close_dsp_device();
std::cout << "destroy开始" << std::endl;
Global::destroy();
std::cout << " Base_Logger_Manager::instance()->destroy()开始" << std::endl;
Base_Logger_Manager::destroy();
std::cout << " clear完成" << std::endl;
std::cout << " std::exit(75) " << name << std::endl;
std::exit(75);
std::cout << "std::exit(75) 完成：" << name << std::endl;
});
Detach_Thread_Manager &manager = g->thread_manager;
std::cout << "[主线程:" << std::this_thread::get_id() << "]" << " \n"
<< std::flush;
manager.test_and_start_thread(
"web服务器线程", [g](std::atomic<bool> &running) {
auto &c = Global::instance()->device_config;
std::string ip = "0.0.0.0";
#ifdef WIN32
ip = "127.0.0.1";
#endif
Net_Config *device = c.list.get("device").value();
std::cout << "http://" + ip + ":" + std::to_string(device->port) + "\n"
<< std::flush;
std::cout << "http://" + device->ip + ":" +
std::to_string(device->port) + "\n"
<< std::flush;
if (!g->svr.listen("0.0.0.0", device->port)) {
if (running.load(std::memory_order_acquire)) {
std::cout << "http://" + ip + ":" + std::to_string(device->port) +
"启动失败 退出！\n"
<< std::flush;
std::cout << "http://" + device->ip + ":" +
std::to_string(device->port) + "启动失败 退出！\n"
<< std::flush;
std::exit(0);
}
return true;
}
});
// manager.test_and_start_thread("io_coro 协程线程", io_coro);
static int t = Global::instance()->mode_acs.read_milliseconds;
g->dsp_config.init_env();
bool enable = g->mlat.enable;
Coro coro;
coro.start();
// Catch_Memory cm;
while (stop_program == 0) {
for (auto &ds : g->mode_acs.data_source_config.map.list()) {
ds->delete_timeout_aircraft();
return false;
}
if (enable) {
g->mlat.refresh();
}
#ifdef __linux__
if (g->mode_acs.monitor_msg_live)
malloc_trim(0);
#if WIN32
std::int64_t memory_max = 1 * 1024 * 1024 * 1024;
#else
std::int64_t memory_max = 1 * 1024 * 1024 * 1024;
#endif
std::this_thread::sleep_for(std::chrono::seconds(3));
if (Global::instance()->init_ok) {
// if (cm.catch_ok()) {
// stop_program = SIGINT;
// break;
// }
std::int64_t cur;
std::int64_t sm;
};
int psc_main(int argc, char* argv[]) {
std::cout << "wyc_main" << std::endl;
if (argc >= 2) {
std::string config_path = argv[1];
if (config_path.at(0) == '@') {
config_path.replace(0, 1, get_exe_dir());
}
::default_config_path = config_path;
}
}
coro.stop();
if (stop_program == SIGINT || stop_program == SIGTERM) {
clear();
}
return 0;
// std::cout << get_stack_trace() << std::endl;
init_signal_config({SIGINT, SIGTERM},
#ifdef WIN32
{}
#else
{ SIGPIPE }
#endif
);
auto clear = []() {
std::cout << "保存配置文件 " << LOG_POS << std::endl;
Config::save();
std::cout << "shutdown" << std::endl;
auto i = Global::instance();
i->svr.stop();
i->thread_manager.stop_all_thread();
Global::instance()->dsp_config.close_dsp_device();
std::cout << "destroy开始" << std::endl;
Global::destroy();
std::cout << " Base_Logger_Manager::instance()->destroy()开始" << std::endl;
Base_Logger_Manager::destroy();
std::cout << " clear完成" << std::endl;
std::cout << " 正常退出" << std::endl;
};
auto g = Global::instance();
Psc::set_fail_fast([](const std::string& name) {
// 如果是在 Global::instance() 里面失败是不能调用这个处理函数的
auto g = Global::instance();
std::cout << "set_fail_fast 退出原因： 222" << name << std::endl;
std::cout << "shutdown" << LOG_POS << std::endl;
auto id = std::this_thread::get_id();
std::cout << "id: " << id << std::endl;
Detach_Thread* dt = g->thread_manager.get_thread(id);
if (!dt) {
std::cout << VAR_STR_1((void*)dt) << std::endl;
std::terminate();
}
std::set<std::string> excluded_thread;
excluded_thread.insert(dt->name);
std::cout << dt->name << std::endl;
// std::cout << "保存配置文件 " << LOG_POS << std::endl;
// Config::save();
g->svr.stop();
g->thread_manager.stop_all_thread(10000, excluded_thread);
std::cout << "22222222" << std::endl;
Global::instance()->dsp_config.close_dsp_device();
std::cout << "destroy开始" << std::endl;
Global::destroy();
std::cout << " Base_Logger_Manager::instance()->destroy()开始" << std::endl;
Base_Logger_Manager::destroy();
std::cout << " clear完成" << std::endl;
std::cout << " std::exit(75) " << name << std::endl;
std::exit(75);
std::cout << "std::exit(75) 完成：" << name << std::endl;
});
Detach_Thread_Manager& manager = g->thread_manager;
std::cout << "[主线程:" << std::this_thread::get_id() << "]" << " \n"
<< std::flush;
manager.test_and_start_thread(
"web服务器线程", [g](std::atomic<bool>& running) {
auto& c = Global::instance()->device_config;
std::string ip = "0.0.0.0";
#ifdef WIN32
ip = "127.0.0.1";
#endif
Net_Config* device = c.list.get("device").value();
std::cout << "http://" + ip + ":" + std::to_string(device->port) + "\n"
<< std::flush;
std::cout << "http://" + device->ip + ":" +
std::to_string(device->port) + "\n"
<< std::flush;
if (!g->svr.listen("0.0.0.0", device->port)) {
if (running.load(std::memory_order_acquire)) {
std::cout << "http://" + ip + ":" + std::to_string(device->port) +
"启动失败 退出！\n"
<< std::flush;
std::cout << "http://" + device->ip + ":" +
std::to_string(device->port) + "启动失败 退出！\n"
<< std::flush;
std::exit(0);
}
}
});
// manager.test_and_start_thread("io_coro 协程线程", io_coro);
static int t = Global::instance()->mode_acs.read_milliseconds;
g->dsp_config.init_env();
bool enable = g->mlat.enable;
Coro coro;
coro.start();
// Catch_Memory cm;
while (stop_program == 0) {
for (auto& ds : g->mode_acs.data_source_config.map.list()) {
ds->delete_timeout_aircraft();
}
if (enable) {
g->mlat.refresh();
}
#ifdef __linux__
if (g->mode_acs.monitor_msg_live)
malloc_trim(0);
#endif
std::this_thread::sleep_for(std::chrono::seconds(3));
if (Global::instance()->init_ok) {
// if (cm.catch_ok()) {
// stop_program = SIGINT;
// break;
// }
}
}
coro.stop();
if (stop_program == SIGINT || stop_program == SIGTERM) {
clear();
}
return 0;
}
int main(int argc, char* argv[]) {
return redict_main_with_gtest(argc, argv, psc_main);
}
// int main(int argc, char* argv[]) {
//
// Psc::set_console_utf8();
// std::cout << "asio 起点" << std::endl;
//
// return 0;
// }
}