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-07-14 10:35:04 +08:00
parent 39197a697c
commit 9ccb8aaa3a
7 changed files with 269 additions and 153 deletions
Download Patch File Download Diff File Expand all files Collapse all files
module/Local_Server/server/With_Loop_Coro.cpp
+124 -89
View File
@@ -1,105 +1,140 @@
#include "With_Loop_Coro.h"
#include "io_coro.h"
#include <string_view>
With_Loop_Coro::~With_Loop_Coro() = default;
void With_Loop_Coro::async_stop() {
std::string name = type + ":" + key;
set_state(State::Force_Quit, std::format("任务正常收到请求,等待退出 {}!\n", name).c_str());
loop_running = false;
std::string name = type + ":" + key;
set_state(State::Force_Quit, std::format("任务正常收到请求,等待退出 {}!\n", name).c_str());
loop_running = false;
}
void With_Loop_Coro::sync_wait() {
std::string name = type + ":" + key;
if (!loop_task) {
std::cout << std::format("{}退出成功！ loop_task 未启动 \n", name);
return;
}
while (loop_task->wait_for(std::chrono::milliseconds(0)) != std::future_status::ready) {
std::cout << std::format("{}等待退出 当前状态为{} \n", name, Psc::to_string(this->state));
std::this_thread::sleep_for(std::chrono::milliseconds(1000));
}
std::cout << std::format("{}退出成功！ \n", name);
std::string name = type + ":" + key;
if (!loop_task) {
std::cout << std::format("{}退出成功！ loop_task 未启动 \n", name);
return;
}
while (loop_task->wait_for(std::chrono::milliseconds(0)) != std::future_status::ready) {
std::cout << std::format("{}等待退出 当前状态为{} \n", name, Psc::to_string(this->state));
std::this_thread::sleep_for(std::chrono::milliseconds(1000));
}
std::cout << std::format("{}退出成功！ \n", name);
}
bool With_Loop_Coro::running() const {
return loop_running;
return loop_running.load();
}
void With_Loop_Coro::set_state(State state, std::string_view action) {
if (!action.empty()) {
std::string name = type + "_" + key;
std::cout << std::format("协程状态机 {} {} {} ==> {} \n", name, action, Psc::to_string(this->state),
Psc::to_string(state));
}
this->state = state;
if (!action.empty()) {
std::string name = type + "_" + key;
std::cout << std::format("协程状态机 {} {} {} ==> {} \n", name, action, Psc::to_string(this->state),
Psc::to_string(state));
}
this->state = state;
}
With_Loop_Coro::With_Loop_Coro() : stop(0.1) {
}
With_Loop_Coro::With_Loop_Coro() : stop(0.1) {}
asio::awaitable<void> With_Loop_Coro::run_loop_coro() {
loop_running.store(true, std::memory_order_release);
try {
co_await loop_coro();
loop_running.store(false, std::memory_order_release);
co_return;
} catch (...) {
loop_running.store(false, std::memory_order_release);
throw;
}
loop_running.store(true, std::memory_order_release);
try {
co_await loop_coro();
loop_running.store(false, std::memory_order_release);
co_return;
}
catch (const std::exception& e) {
//loop_running.store(false, std::memory_order_release);
auto str = std::format("{} loop_coro 异常退出: {}\n", key, e.what());
std::cerr << str;
set_state(State::Loop_Exception, str);
throw;
}
catch (...) {
//loop_running.store(false, std::memory_order_release);
std::cerr << "loop_coro 未知异常退出\n";
throw;
}
}
asio::awaitable<void> With_Loop_Coro::tick() {
std::string name = type + ":" + key;
if (state == State::Force_Quit) {
static Frequency_Limit_Multi flm;
if (flm.test(name)) {
std::cout << std::format("{} 正在强制退出!\n", name);
}
}
if (state == State::Start) {
if (enable) {
set_state(State::Before_Request_Start_Loop);
}
} else if (state == State::Before_Request_Start_Loop) {
co_await this->_open();
loop_running = enable;
loop_task = Coro::instance()->spawn(run_loop_coro());
set_state(State::Waiting_Loop_Start, "开始启动任务");
} else if (state == State::Waiting_Loop_Start) {
if (running()) {
set_state(State::Loop_Running, "启动任务成功！");
}
} else if (state == State::Loop_Running) {
if (enable && !running()) {
std::cout << std::format("任务未知原因已经退出 {}!\n", name);
try {
loop_task->get();
} catch (const std::exception &e) {
std::cout << std::format("任务异常退出 {}! {}\n", name, e.what());
loop_running = false;
} catch (...) {
std::cout << std::format("任务未知异常退出 {}!\n", name);
loop_running = false;
}
loop_task.reset();
co_await _close();
co_return;
}
if (!enable) {
set_state(State::Before_Request_Stop_Loop, std::format("{} 检测到 enable变化 异步退出开始！", name));
}
} else if (state == State::Before_Request_Stop_Loop) {
loop_running = false;
set_state(State::Waiting_Stop_Loop, "退出变量已设置 等待退出");
} else if (state == State::Waiting_Stop_Loop) {
if (!running()) {
loop_task.reset();
co_await this->_close();
set_state(State::Start, "任务正常退出");
}
}
co_return;
std::string name = type + ":" + key;
if (state == State::Force_Quit) {
static Frequency_Limit_Multi flm;
if (flm.test(name)) {
std::cout << std::format("{} 正在强制退出!\n", name);
}
}
if (state == State::Start) {
if (enable) {
set_state(State::Before_Request_Start_Loop);
}
}
else if (state == State::Before_Request_Start_Loop) {
co_await this->_open();
loop_running = enable;
auto future = asio::co_spawn(Coro::instance()->io, run_loop_coro(), asio::use_future);
loop_task = std::make_shared<std::future<void>>(std::move(future));
//loop_task = Coro::instance()->spawn(run_loop_coro());
// loop_task = asio::co_spawn(Coro::instance()->io, run_loop_coro(), [](std::exception_ptr exception) {
// if (exception) {
// std::cout << " 发生了异常！" << std::endl;
// std::rethrow_exception(exception);
// }
// });
set_state(State::Waiting_Loop_Start, "开始启动任务");
}
else if (state == State::Waiting_Loop_Start) {
if (running()) {
set_state(State::Loop_Running, "启动任务成功！");
}
}
else if (state == State::Loop_Running) {
if (!enable) {
set_state(State::Before_Request_Stop_Loop, std::format("{} 检测到 enable变化 异步退出开始！", name));
}
}
else if (state == State::Loop_Exception) {
bool catch_exception = false;
if (!catch_exception) {
loop_task->get();
}
else {
std::exception_ptr task_exception;
try {
// 走到这里直接崩溃
loop_task->get();
}
catch (...) {
task_exception = std::current_exception();
}
loop_task.reset();
co_await _close();
if (task_exception) {
loop_running = false;
try {
std::rethrow_exception(task_exception);
}
catch (const std::exception& e) {
std::cout << std::format("任务异常退出 {}! {}\n", name, e.what()) << std::flush;
}
catch (...) {
std::cout << std::format("任务未知异常退出 {}!\n", name) << std::flush;
}
set_state(State::Start, "任务异常退出");
co_return;
}
else {
std::cout << "没有异常 状态机 紊乱" << std::endl;
std::terminate();
}
co_return;
}
}
else if (state == State::Before_Request_Stop_Loop) {
loop_running = false;
set_state(State::Waiting_Stop_Loop, "退出变量已设置 等待退出");
}
else if (state == State::Waiting_Stop_Loop) {
if (!running()) {
loop_task.reset();
co_await this->_close();
set_state(State::Start, "任务正常退出");
}
}
co_return;
}
module/Local_Server/server/With_Loop_Coro.h
+28 -27
View File
@@ -16,35 +16,36 @@
#include "Core/Statistics/Frequency_Limit.h"
class With_Loop_Coro_Data {
public:
std::string type;
std::string key;
Psc::Copyable_Atomic<bool> enable{};
PSC_USE_JSON
std::string type;
std::string key;
Psc::Copyable_Atomic<bool> enable{};
PSC_USE_JSON
};
class With_Loop_Coro : public With_Loop_Coro_Data {
public:
virtual ~With_Loop_Coro();
std::shared_ptr<std::future<void>> loop_task;
virtual asio::awaitable<void> loop_coro() = 0;
void async_stop();
void sync_wait();
bool running() const;
asio::awaitable<void> tick();
virtual asio::awaitable<void> _open() = 0;
virtual asio::awaitable<void> _close() = 0;
enum class State {
Force_Quit,
Start,
Before_Request_Start_Loop,
Waiting_Loop_Start,
Loop_Running,
Before_Request_Stop_Loop,
Waiting_Stop_Loop
} state = State::Start;
void set_state(State state, std::string_view action = "");
With_Loop_Coro();
virtual ~With_Loop_Coro();
std::shared_ptr<std::future<void>> loop_task;
virtual asio::awaitable<void> loop_coro() = 0;
void async_stop();
void sync_wait();
bool running() const;
asio::awaitable<void> tick();
virtual asio::awaitable<void> _open() = 0;
virtual asio::awaitable<void> _close() = 0;
enum class State {
Force_Quit,
Start,
Before_Request_Start_Loop,
Waiting_Loop_Start,
Loop_Running,
Loop_Exception,
Before_Request_Stop_Loop,
Waiting_Stop_Loop
} state = State::Start;
void set_state(State state, std::string_view action = "");
With_Loop_Coro();
protected:
Psc::Copyable_Atomic<bool> loop_running{};
asio::awaitable<void> run_loop_coro();
Frequency_Limit_Multi stop;
Psc::Copyable_Atomic<bool> loop_running{};
asio::awaitable<void> run_loop_coro();
Frequency_Limit_Multi stop;
};
module/Local_Server/server/io_coro.cpp
+1
View File
@@ -82,6 +82,7 @@ void Coro::start() {
auto promise = std::make_shared<std::promise<void> >();
loop_task = std::make_unique<std::future<void> >(promise->get_future());
asio::co_spawn(io, coro_thread(), [promise](std::exception_ptr ep) {
std::cout << "发生了异常" << std::endl;
if (ep) {
promise->set_exception(ep);
std::rethrow_exception(ep);
module/Local_Server/server/io_coro.h
+9
View File
@@ -31,6 +31,15 @@ public:
return std::make_shared<std::future<void> >(std::move(future));
}
// template<typename Awaitable>
// void spawn(Awaitable &&awaitable) {
// asio::co_spawn(io, std::forward<Awaitable>(awaitable), [](std::exception_ptr exception) {
// if (exception) {
// std::rethrow_exception(exception);
// }
// });
// }
private:
asio::awaitable<void> coro_thread();
module/dll_source/main.cpp
+8 -37
View File
@@ -18,12 +18,9 @@
#include "Core/Shared_Memory/Shared_Memory.h"
#include "Core/system/export.h"
#include "Core/spdlog/export.h"
#include "Core/Base/SM_RingBuffer.h"
#include "Dll_Global.h"
#include <string_view>
std::vector<std::string> ddd = {
"1A 33 1A 1A F1 DA 08 2A 73 60 8D 78 12 0A EA 28 88 64 25 3C 08 33 22 41",
"1A 33 1A 1A F1 D9 13 EA E7 11 8D 78 1D 40 58 BF 41 E2 45 49 09 7B C8 24",
@@ -41,8 +38,6 @@ std::vector<std::string> ddd = {
"1A 33 1A 1A F1 DF AF 71 5F 6C A0 00 0E 3E A8 7A 23 26 E1 0C 22 4E D9 88",
"1A 33 1A 1A F1 E2 0A DD 76 2A A0 00 11 B1 B5 E8 00 30 AA 00 00 D5 D0 A3",
};
std::string get_absb() {
static size_t index = 0; // 静态变量用于保存当前读取位置
index++;
@@ -51,89 +46,74 @@ std::string get_absb() {
}
return ddd[index];
}
class TextLineReader {
public:
explicit TextLineReader(std::string_view file_path) {
load(file_path);
}
// 每调用一次，返回一行
// 返回 true 表示成功取到一行
// 返回 false 表示已经没有更多行
bool get(std::string &ret_line) {
bool get(std::string& ret_line) {
if (index_ >= lines_.size()) {
return false;
}
ret_line = lines_[index_];
++index_;
return true;
}
// 重新从第一行开始读
void reset() {
index_ = 0;
}
// 总行数
std::size_t size() const {
return lines_.size();
}
// 是否已经读完
bool empty() const {
return lines_.empty();
}
private:
void load(std::string_view file_path) {
std::ifstream file{std::string(file_path)};
// std::cout << "抛出异常" << std::endl;
// throw std::runtime_error("222");
if (!file.is_open()) {
auto err = "Failed to open file: " + std::string(file_path);
std::cout << err << std::endl;
std::exit(895);
throw std::runtime_error(err);
}
std::string line;
while (std::getline(file, line)) {
lines_.push_back(line);
}
index_ = 0;
}
private:
std::vector<std::string> lines_;
std::size_t index_ = 0;
};
std::size_t adsb_read(char *buf, std::size_t len) {
std::size_t adsb_read(char* buf, std::size_t len) {
static TextLineReader reader(get_exe_dir() + "/ADS-B_195_0205.txt");
std::string line;
reader.get(line);
auto l = line.size();
std::memcpy(buf, line.c_str(), l);
/*static Frequency_Limit fl;
if (fl.test()) {
std::cout << "adsb_read" << VAR_STR_2(l, len) << std::endl;
}*/
return l;
}
int adsb_set_iq_cbk(iq_cbk_t recv_cbk, void *pusr) {
int adsb_set_iq_cbk(iq_cbk_t recv_cbk, void* pusr) {
auto dg = Dll_Global::instance();
dg->recv_cbk = recv_cbk;
dg->puser = pusr;
dg->init_when_dll();
return 0;
}
int adsb_set_less_30Mhz_ddc_param(uint64_t cf_hz, uint64_t sf_hz, uint64_t bw_hz, int8_t sw) {
auto dg = Dll_Global::instance();
dg->center_freq = cf_hz;
@@ -141,41 +121,33 @@ int adsb_set_less_30Mhz_ddc_param(uint64_t cf_hz, uint64_t sf_hz, uint64_t bw_hz
dg->band_width = bw_hz;
return 0;
}
int adsb_set_less_30Mhz_fft_param(uint64_t fftWin, uint64_t fftLen, int fftPeriod_ms, int8_t sw) {
auto dg = Dll_Global::instance();
dg->fft_point_number = fftLen;
dg->fft_period_ms = fftPeriod_ms;
dg->fft_win_type = (UHD_FFTWin) fftWin;
dg->fft_win_type = (UHD_FFTWin)fftWin;
return 0;
}
int main() {
init_signal_config();
auto dg = Dll_Global::instance();
dg->init_when_exe();
bool debug = false;
auto mb = get_system_memory() / (1024 * 1024);
auto gb = mb / (1024);
std::cout << "当前系统的物理内存大小" << mb << " MB" << std::endl;
std::cout << "当前系统的物理内存大小" << gb << " GB" << std::endl;
dg->dtm.test_and_start_thread("dsp共享内存数据推送", [dg](std::atomic<bool> &running) {
dg->dtm.test_and_start_thread("dsp共享内存数据推送", [dg](std::atomic<bool>& running) {
bool debug = true;
while (running) {
auto list = dg->iq_memory_buffer.get_all();
// 释放内存池
Pool_Guard pg(&dg->iq_memory_buffer.pool, list);
if (debug && list.size() > 1000) {
static Frequency_Limit fl(1.0 / 60.0);
std::ostringstream oss;
oss << get_current_millisecond_timestamp() << " 警告： dsp共享内存数据推送数据过多 当前数据队列长度:" << list.size() <<
std::endl;
std::endl;
if (fl.test()) {
std::cout << oss.str() << std::endl;
}
@@ -189,7 +161,6 @@ int main() {
if (fl.test()) {
std::cout << dg->rb->state_str() << " dsp数据推送" << frame_number << "条" << std::endl;
}
for (int i = 0; i < frame_number; i++) {
auto buf = list[i];
dg->handle_2112(buf->data());