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

预处理移动到 process_mode_acs_data 线程

Browse Source
This commit is contained in:
种太阳的热带美短
2026-06-23 14:43:14 +08:00
parent afd4bf8e69
commit 3f89fe2aab
9 changed files with 342 additions and 344 deletions
Download Patch File Download Diff File Expand all files Collapse all files
module/Local_Server/Data_Source/Data_Source.cpp
+18 -194
View File
@@ -110,10 +110,7 @@ void File_Data_Source::before_handle_msg(std::shared_ptr<SSR::Mode_Msg>& msg) {
ucoro::awaitable<void> Data_Source::source_step_coro() {
co_await handle_in_loop_coro();
auto mode_data = co_await read_coro();
if (mode_data.empty()) {
std::this_thread::sleep_for(std::chrono::milliseconds(10));
co_return;
}
struct Process_Result {
std::exception_ptr exception;
};
@@ -146,41 +143,6 @@ void Data_Source::test_and_stop_thread() const {
Global::instance()->thread_manager.test_and_stop_thread(thread_key());
}
Psc::JSON Data_Source::get_all_connect_feed_status() {
Psc::JSON ret = Psc::JSON::object();
std::vector<Cached_Source_Info> list;
{
std::lock_guard g(cdf_mtx);
list = cached_data_feed_key_list;
}
for (auto& i : list) {
Psc::JSON cur = Psc::JSON::object();
auto odf = Global::instance()->mode_acs.data_feed_config.map.get(i.key);
if (odf.has_value()) {
const auto& df = odf.value();
i.mode_s_cache_num.update(df->msg_buffer.mode_s_msg_num);
i.mode_other_cache_num.update(df->msg_buffer.mode_other_msg_num);
cur.append({"mode_s_msg_num", i.mode_s_cache_num.to_json()});
cur.append({"mode_other_msg_num", i.mode_other_cache_num.to_json()});
}
ret.append({i.key, cur});
}
return ret;
}
Psc::JSON Data_Source::get_all_connect_feed() {
Psc::JSON ret = Psc::JSON::array();
std::vector<Cached_Source_Info> list;
{
std::lock_guard g(cdf_mtx);
list = cached_data_feed_key_list;
}
for (auto& i : list) {
Psc::JSON cur = Psc::JSON::object();
cur.append({"key", i.key});
ret.append(cur);
}
return ret;
}
std::vector<std::string> readLines(const std::string& path) {
@@ -344,9 +306,6 @@ std::string get_true_from_raw_line(Data_Source* ds, File_Data_Type data_type, in
std::string info = log_info;
std::string result = ds->last_char;
ds->last_char = "";
for (char c : info)
{
if (std::isxdigit(c)) { // 会判断字符 c 是否是十六进制数字字符
@@ -367,26 +326,6 @@ std::string get_true_from_raw_line(Data_Source* ds, File_Data_Type data_type, in
result.pop_back();
}
ret = hex2mem(result);
// if (size % 2 != 0)
// {
// std::ostringstream oss;
// size_t prefix_num = 20;
// size_t suffix_num = 20;
// auto origin_size = log_info.size();
// prefix_num = std::min(origin_size, prefix_num);
// suffix_num = std::min(origin_size, suffix_num);
// std::string prefix_data = log_info.substr(0, prefix_num);
// std::string suffix_data = log_info.substr(log_info.size() - suffix_num, suffix_num);
// oss << "index:" << index << " 长度不为偶数 不能转换成内存：" << "[内容]" << "result_size" << result.size() << " origin_size:[" << origin_size << "]" << std::endl <<
// " prefix[" << prefix_num << "]:" << prefix_data << std::endl <<
// " suffix[" << suffix_num << "]:" << suffix_data << std::endl <<
// std::endl;
// server_logger->c_debug("长度不为偶数", {}, oss.str());
// //std::cout << "result:" << log_info << std::endl << std::endl;
// return "";
// }
}
else if (data_type == File_Data_Type::BIN_Blank_One_Line_With_Escape) {
auto size = log_info.size();
@@ -526,13 +465,15 @@ std::vector<std::string> File_Data_Source::readBinaryFileAsString(const std::str
}
ucoro::awaitable<std::string> File_Data_Source::read_coro() {
void File_Data_Source::origin_data_transform_mode_data(std::string& mode_data)
{
std::lock_guard g(mtx);
assert(mode_data.size() == 0);
int ret_index = 0;
if (play_mode != Play_Mode::analysis)
{
std::optional<std::string> log_info = get_raw_line(ret_index);
co_return log_info.has_value() ? get_true_from_raw_line(this, data_type, ret_index, log_info.value()) : "";
mode_data = log_info.has_value() ? get_true_from_raw_line(this, data_type, ret_index, log_info.value()) : "";
} else
{
int num = 0;
@@ -551,7 +492,7 @@ ucoro::awaitable<std::string> File_Data_Source::read_coro() {
log_info = get_raw_line(ret_index);
}
// std::cout << "size == " << t.size() << std::endl;
co_return ret;
mode_data = ret;
}
}
@@ -590,11 +531,11 @@ void File_Data_Source::handle_mode_s(std::shared_ptr<SSR::Mode_S_Msg> msg) {
}
ucoro::awaitable<std::string> Dll_Data_Source::read_coro() {
void Dll_Data_Source::origin_data_transform_mode_data(std::string& mode_data)
{
auto start = std::chrono::steady_clock::now();
if (read_func_ptr == nullptr) co_return "";
if (read_func_ptr == nullptr) return;
auto buf = reinterpret_cast<char *>(buffer.data());
auto len = read_func_ptr(buf, buffer_size);
@@ -606,7 +547,7 @@ ucoro::awaitable<std::string> Dll_Data_Source::read_coro() {
auto end = std::chrono::steady_clock::now();
auto cost = std::chrono::duration_cast<std::chrono::microseconds>(end - start).count();
//std::cout << "Dll_Data_Source::read cost: " << cost << " us\n";
co_return "";
return;
}
auto ret = get_true_from_raw_line(this, data_type, -1, data);
@@ -614,9 +555,9 @@ ucoro::awaitable<std::string> Dll_Data_Source::read_coro() {
auto end = std::chrono::steady_clock::now();
auto cost = std::chrono::duration_cast<std::chrono::microseconds>(end - start).count();
// std::cout << "Dll_Data_Source::read cost: " << cost << " us\n";
// std::cout << "Dll_Data_Source::read cost: " << cost << " us\n";
co_return ret;
mode_data = ret;
}
@@ -629,8 +570,8 @@ void Shared_Memory_Data_Source::_close() {
sm.reset();
}
ucoro::awaitable<std::string> Shared_Memory_Data_Source::read_coro() {
void Shared_Memory_Data_Source::origin_data_transform_mode_data(std::string& mode_data)
{
auto size = sm->shm.size();
size_t length = size;
std::string data;
@@ -638,7 +579,7 @@ ucoro::awaitable<std::string> Shared_Memory_Data_Source::read_coro() {
bool ok = sm->read((uint8_t *)data.data(), length);
if (!ok) {
co_return "";
mode_data = "";
}
if (length != size) {
@@ -648,9 +589,11 @@ ucoro::awaitable<std::string> Shared_Memory_Data_Source::read_coro() {
std::cout << "read:" << data << std::endl;
}
auto ret = get_true_from_raw_line(this, data_type, -1, data);
co_return ret;
mode_data = ret;
}
void Data_Source_Config::server(Global* g) {
auto& svr = g->svr;
auto& api = g->api;
@@ -755,124 +698,5 @@ void Data_Source_Config::server(Global* g) {
g->mode_acs.source_feed_relation_config.set_need_refresh();
});
}
void Data_Source::refresh_data_feed_key_list() {
//std::cout << key << " refresh_data_feed_key_list" << std::endl;
std::vector<std::string> tmp;
for (auto& relation : Global::instance()->mode_acs.source_feed_relation_config.map.list()) {
if (!relation->enable) continue;
if (relation->type == "One_to_One_Relation") {
auto t = dynamic_cast<One_to_One_Relation*>(relation.get());
//std::cout << VAR_STR_2(t->source_key, this->key) << " refresh_data_feed_key_list" << std::endl;
if (t->source_key == this->key) {
tmp.push_back(t->feed_key);
}
} else if (relation->type == "First_Source_To_All_Feed_Relation") {
auto t = dynamic_cast<First_Source_To_All_Feed_Relation*>(relation.get());
std::shared_ptr<Data_Source> first = nullptr;
for (const auto& ds : Global::instance()->mode_acs.data_source_config.map.list()) {
if (ds->enable) {
first = ds;
break;
}
}
if (first->key == key) {
for (const auto& df : Global::instance()->mode_acs.data_feed_config.map.list()) {
if (df->enable) {
tmp.push_back(df->key);
}
}
}
}
std::sort(tmp.begin(), tmp.end());
// 去重
auto last = std::unique(tmp.begin(), tmp.end());
tmp.erase(last, tmp.end());
{
std::vector<Cached_Source_Info> tmp_info;
// 创建 tmp_info 向量，将 tmp 的内容转化为 Cached_Source_Info 对象
for (const auto& key : tmp) {
tmp_info.push_back({key});
}
{
// 进入临界区，锁住 mutex，确保线程安全
std::lock_guard<std::mutex> lock(cdf_mtx);
// 1. 删除 tmp 中没有的 cached_data_feed_key_list 元素
auto it = cached_data_feed_key_list.begin();
while (it != cached_data_feed_key_list.end()) {
if (std::find_if(tmp_info.begin(), tmp_info.end(),
[&](const Cached_Source_Info& info) {
return info.key == it->key;
}) == tmp_info.end()) {
// 如果当前元素在 tmp 中找不到，删除它
it = cached_data_feed_key_list.erase(it);
} else {
++it;
}
}
// 2. 创建 tmp 中有但 cached_data_feed_key_list 没有的元素
for (const auto& tmp_item : tmp_info) {
auto found = std::find_if(cached_data_feed_key_list.begin(), cached_data_feed_key_list.end(),
[&](const Cached_Source_Info& cached_item) {
return cached_item.key == tmp_item.key;
});
if (found == cached_data_feed_key_list.end()) {
// 如果 tmp_item 不在 cached_data_feed_key_list 中，添加它
cached_data_feed_key_list.push_back(tmp_item);
}
}
}
}
}
}
void Data_Source::push_to_feed(const std::shared_ptr<SSR::Msg>& msg) {
if (need_refresh_data_feed_key_list) {
refresh_data_feed_key_list();
need_refresh_data_feed_key_list = false;
}
auto& other_size = Global::instance()->mode_acs.mode_other_max_num;
auto& s_size = Global::instance()->mode_acs.mode_other_max_num;
std::vector<std::string> list;
for (const auto& item : cached_data_feed_key_list) {
auto& key = item.key;
auto opt_feed = Global::instance()->mode_acs.data_feed_config.map.get(key);
if (!opt_feed.has_value()) {
std::cout << "wrong data feed: " << key << std::endl;
break;
}
std::shared_ptr<Data_Feed>& feed = opt_feed.value();
if (!feed->enable) continue;
if (msg->type == SSR::Mode_Msg::T::S7 || msg->type == SSR::Mode_Msg::T::S14) {
if (feed->msg_buffer.mode_s_msg_num > s_size) {
continue;
} else {
++feed->msg_buffer.mode_s_msg_num;
}
} else {
if (feed->msg_buffer.mode_other_msg_num > other_size) {
continue;
} else {
++feed->msg_buffer.mode_other_msg_num;
}
}
std::optional<std::string> binary = convert_to_send_format(this, feed, msg);
if (binary.has_value()) {
feed->msg_buffer.push(binary.value());
}
}
}