Compare commits

7 Commits

11 changed files with 705 additions and 39 deletions

BIN
LFtid1056.rar Normal file

Binary file not shown.

View File

@@ -20,9 +20,10 @@ $SRC_DIR/tinyxml2.cpp \
./dealMsg.cpp \
./main_thread.cpp \
./PQSMsg.cpp \
./pqdif_thread_processor.cpp \
./pqdif/PQDIF.cpp \
./pqdif/include/cjson.c "
./pqdif_semantic_ids.cpp \
./pqdif_thread_processor.cpp \
./pqdif/include/cjson.c"
INCLUDE_DIRS="-I$SRC_DIR \
-I$SRC_DIR/nlohmann \
@@ -33,20 +34,22 @@ INCLUDE_DIRS="-I$SRC_DIR \
-I./lib/libuv-v1.51.0/include \
-I./pqdif \
-I./pqdif/include \
-I. "
-I. "
LIB_DIRS="-L$LIB_DIR -L/usr/lib64 -L/usr/local/lib"
LIB_DIRS="-L$LIB_DIR -L./pqdif/lib -L/usr/lib64 -L/usr/local/lib"
LIBS="./cloudfront/lib/libcurl.so \
./cloudfront/lib/libssl.so \
./cloudfront/lib/libcrypto.so \
./cloudfront/lib/liblog4cplus.so \
./pqdif/lib/libpqdiflib.a \
./pqdif/lib/libz.a \
-lpthread -ldl -lrt \
-lstdc++fs \
-lz \
./libuv.a \
-Wl,--start-group \
./pqdif/lib/libpqdiflib.a \
./pqdif/lib/libz.a \
-Wl,--end-group \
-pthread"
# 如果有静态 rocketmq 库就加上
@@ -74,4 +77,4 @@ if [ $? -eq 0 ]; then
ldd "$OUT_DIR/$TARGET" || echo "是静态编译程序 ✔"
else
echo "❌ 编译失败"
fi
fi

View File

@@ -1310,12 +1310,14 @@ bool ClientManager::set_cloud_status(const std::string& identifier, int status)
connect_status_update(identifier, status);
std::cout << "[Device " << identifier
<< "] ****Cloud status: " << ctx->cloudstatus << " updated to: " << status << std::endl;
DIY_INFOLOG_CODE(identifier,1,LOG_CODE_COMM,"设备登录状态更新为在线");
}
else if (ctx->cloudstatus == 1 && status == 0) {
//设备从在线转换至离线,通知前台状态发生翻转
connect_status_update(identifier, status);
std::cout << "[Device " << identifier
<< "] ****Cloud status: " << ctx->cloudstatus << " updated to: " << status << std::endl;
DIY_INFOLOG_CODE(identifier,1,LOG_CODE_COMM,"设备登录状态更新为离线");
}
// 修改云前置登录状态

View File

@@ -10,6 +10,7 @@
#include <queue>
#include "dealMsg.h"
#include "PQSMsg.h"
// <20><><EFBFBD><EFBFBD><EFBFBD><EFBFBD>Ϣ<EFBFBD>
struct PointInfo {
std::string point_id; // <20><><EFBFBD><EFBFBD>ID

View File

@@ -103,6 +103,10 @@ static std::map<std::string, RunningInformation> g_runninginfo_cache;
static std::mutex g_versioninfo_cache_mtx;
static std::map<std::string, DeviceVersionInfo> g_versioninfo_cache;
//装置升级
std::map<std::string, std::string> g_upgrade_file_map;
std::mutex g_upgrade_file_mutex;
//补招
std::list<JournalRecall> g_StatisticLackList; //日志补招结构类链表
std::mutex g_StatisticLackList_list_mutex; //补招队列数据锁
@@ -232,6 +236,14 @@ int TEST_PORT = 11000; //用于当前进程登录测试shell的端口
std::string G_TEST_LIST = ""; //测试用的发送实际数据的终端列表
std::vector<std::string> TESTARRAY; //解析的列表数组
// 日志限流配置
int G_LOG_RATE_RESET_SEC = 3600; // 1小时重置
int G_LOG_RATE_LIMIT_SEC = 60; // 进入限流后60秒1条
int G_LOG_RATE_KEEP_ALL_MS = 60000; // 间隔 >= 60000ms全部保留
int G_LOG_RATE_KEEP_BURST_MS = 1000; // 间隔 >= 1000ms按二级策略
int G_LOG_RATE_KEEP_BURST_COUNT = 60; // 二级保留前60条
int G_LOG_RATE_KEEP_HIGHFREQ_COUNT = 10; // 高频保留前10条
//////////////////////////////////////////////////////////////////////////////////////////////////////////////////其他文件定义的函数引用声明
bool enqueue_direct_download(const std::string& dev_id,
@@ -409,6 +421,37 @@ void loadConfig(const std::string& filename) {
intMap["RocketMq.TestPort"] = &TEST_PORT;
strMap["RocketMq.TestList"] = &G_TEST_LIST;
// ==================== 新增:日志限流配置 ====================
intMap["LogRate.ResetSec"] = &G_LOG_RATE_RESET_SEC;
intMap["LogRate.LimitSec"] = &G_LOG_RATE_LIMIT_SEC;
intMap["LogRate.KeepAllMs"] = &G_LOG_RATE_KEEP_ALL_MS;
intMap["LogRate.KeepBurstMs"] = &G_LOG_RATE_KEEP_BURST_MS;
intMap["LogRate.KeepBurstCount"] = &G_LOG_RATE_KEEP_BURST_COUNT;
intMap["LogRate.KeepHighFreqCount"] = &G_LOG_RATE_KEEP_HIGHFREQ_COUNT;
// ==========================================================
// ==================== 新增:日志限流默认值保护 ====================
if (G_LOG_RATE_RESET_SEC <= 0)
G_LOG_RATE_RESET_SEC = 3600;
if (G_LOG_RATE_LIMIT_SEC <= 0)
G_LOG_RATE_LIMIT_SEC = 60;
if (G_LOG_RATE_KEEP_ALL_MS <= 0)
G_LOG_RATE_KEEP_ALL_MS = 60000;
if (G_LOG_RATE_KEEP_BURST_MS <= 0)
G_LOG_RATE_KEEP_BURST_MS = 1000;
if (G_LOG_RATE_KEEP_BURST_COUNT < 0)
G_LOG_RATE_KEEP_BURST_COUNT = 60;
if (G_LOG_RATE_KEEP_HIGHFREQ_COUNT < 0)
G_LOG_RATE_KEEP_HIGHFREQ_COUNT = 10;
// ================================================================
// 2. 打开并逐行解析 INI 文件
std::ifstream fin(filename);
if (!fin.is_open()) {
@@ -581,6 +624,15 @@ void printConfig() {
printStr("WEB_FILEUPLOAD", WEB_FILEUPLOAD);
printStr("WEB_FILEDOWNLOAD", WEB_FILEDOWNLOAD);
std::cout << "\n// 日志限流配置\n";
printInt("G_LOG_RATE_RESET_SEC", G_LOG_RATE_RESET_SEC);
printInt("G_LOG_RATE_LIMIT_SEC", G_LOG_RATE_LIMIT_SEC);
printInt("G_LOG_RATE_KEEP_ALL_MS", G_LOG_RATE_KEEP_ALL_MS);
printInt("G_LOG_RATE_KEEP_BURST_MS", G_LOG_RATE_KEEP_BURST_MS);
printInt("G_LOG_RATE_KEEP_BURST_COUNT", G_LOG_RATE_KEEP_BURST_COUNT);
printInt("G_LOG_RATE_KEEP_HIGHFREQ_COUNT",G_LOG_RATE_KEEP_HIGHFREQ_COUNT);
std::cout << "-------------------------------------\n";
}
@@ -1641,12 +1693,12 @@ void parse_terminal_from_data(trigger_update_xml_t& trigger_update_xml,
//work_terminal.dev_key = get_value("devKey");
//work_terminal.dev_series = get_value("series");
work_terminal.processNo = get_value("processNo");
//work_terminal.addr_str = get_value("ip");
work_terminal.addr_str = get_value("ip");
//work_terminal.port = get_value("port");
//work_terminal.timestamp = get_value("updateTime");
work_terminal.Righttime = get_value("Righttime");
work_terminal.mac = get_value("mac");
work_terminal.mac = get_value("ip");
for (tinyxml2::XMLElement* monitor = root->FirstChildElement("monitorData");
monitor;
@@ -6594,8 +6646,43 @@ void on_device_response_minimal(int response_code,
try {
if(dev->isbusy == 2){
//获取升级文件
std::string file_path;
{
std::lock_guard<std::mutex> lk(g_upgrade_file_mutex);
auto it = g_upgrade_file_map.find(dev->terminal_id);
if (it != g_upgrade_file_map.end()) {
file_path = it->second;
}
}
if (file_path.empty()) {
std::cerr << "upgrade file not found for devid=" << dev->terminal_id << std::endl;
send_reply_to_cloud(static_cast<int>(ResponseCode::NOT_FOUND), id, device_state_int, dev->guid, dev->mac);
dev->guid.clear();
dev->busytype = 0;
dev->isbusy = 0;
dev->busytimecount = 0;
return;
}
std::cout << "upgrade file path=" << file_path << std::endl;
// 读取升级文件
std::vector<unsigned char> file_data = read_file_as_bytes("pqs_arm2.bin");
std::vector<unsigned char> file_data = read_file_as_bytes(file_path);
if (file_data.empty()) {
std::cerr << "read file failed: " << file_path << std::endl;
send_reply_to_cloud(static_cast<int>(ResponseCode::NOT_FOUND), id, device_state_int, dev->guid, dev->mac);
dev->guid.clear();
dev->busytype = 0;
dev->isbusy = 0;
dev->busytimecount = 0;
return;
}
std::cout << "file size=" << file_data.size() << std::endl;
// 下发升级指令
ClientManager::instance().send_upgrade_action_to_device(id, file_data, 10240);

View File

@@ -390,7 +390,7 @@ void Fileupload_test()
// 下载文件:从远端路径下载到本地,并返回本地文件路径
// 入参dev设备、remote_path远端完整路径
// 返回:本地保存路径(失败返回空字符串)
std::string getfilefromweb(const std::string& devid, const std::string& remote_path)
std::string getfilefromweb(const std::string& devid, const std::string& remote_path,int type)
{
try {
terminal_dev* dev = nullptr;
@@ -428,7 +428,17 @@ std::string getfilefromweb(const std::string& devid, const std::string& remote_p
//【3】构造本地保存路径
std::string mac = sanitize(normalize_mac(dev->addr_str));
std::string save_dir = std::string(FRONT_PATH) + "/bin/upload/" + mac + "/";
std::string save_dir;
if(type == 1) {
// 升级文件放在专门的upgrade目录下
save_dir = std::string(FRONT_PATH) + "/bin/upgrade/" + mac + "/";
}
else {
// 普通文件放在upload目录下
save_dir = std::string(FRONT_PATH) + "/bin/upload/" + mac + "/";
}
if (!create_directory_recursive(save_dir)) {
std::cerr << "[getfile][ERROR] create dir failed: " << save_dir << std::endl;
@@ -824,7 +834,7 @@ int terminal_ledger_web(std::map<std::string, terminal_dev>& terminal_dev_map,
dev.processNo = safe_str(item, "node");
dev.maxProcessNum = safe_str(item, "maxProcessNum");
//dev.mac = safe_str(item, "mac");//添加mac
dev.mac = safe_str(item, "ip");//添加mac
if (item.contains("monitorData") && item["monitorData"].is_array()) {
for (auto& mon : item["monitorData"]) {

View File

@@ -687,7 +687,7 @@ bool update_qvvr_file_download(const std::string& filename_with_mac, const std::
//上送文件列表接口
bool send_file_list(terminal_dev* dev, const std::vector<tag_dir_info> &FileList);
std::string getfilefromweb(const std::string& devid, const std::string& remote_path);
std::string getfilefromweb(const std::string& devid, const std::string& remote_path,int type);
//提取mac
std::string normalize_mac(const std::string& mac);

View File

@@ -50,6 +50,14 @@ extern std::string subdir;
//日志主题
extern std::string G_LOG_TOPIC;
// 日志限流配置
extern int G_LOG_RATE_RESET_SEC;
extern int G_LOG_RATE_LIMIT_SEC;
extern int G_LOG_RATE_KEEP_ALL_MS;
extern int G_LOG_RATE_KEEP_BURST_MS;
extern int G_LOG_RATE_KEEP_BURST_COUNT;
extern int G_LOG_RATE_KEEP_HIGHFREQ_COUNT;
////////////////////////////////////////////////////////////////////////////////////////////////////
const int LOGTYPE_DEFAULT = LOG_CODE_OTHER;
@@ -223,10 +231,26 @@ void refresh_log_level_cache_locked()
class SendAppender : public Appender {
private:
struct RateState {
/*struct RateState {
uint64_t hit_count = 0; // 同一条日志累计命中次数
std::chrono::steady_clock::time_point last_emit =
std::chrono::steady_clock::time_point::min();
};*/
struct RateState {
uint64_t pass_count; // 当前周期内已放行条数
uint64_t suppressed_count; // 当前被抑制条数
std::chrono::steady_clock::time_point last_emit;
std::chrono::steady_clock::time_point last_seen;
std::chrono::steady_clock::time_point last_reset;
bool has_emit;
RateState()
: pass_count(0),
suppressed_count(0),
last_emit(),
last_seen(),
last_reset(),
has_emit(false) {}
};
static std::unordered_map<std::string, RateState> s_rate_map; //频率map
@@ -240,7 +264,7 @@ private:
}
// 前 3 次1 秒一次;第 3 次起300 秒一次,一小时恢复
static bool should_emit(const std::string& key) {
/*static bool should_emit(const std::string& key) {
using namespace std::chrono;
const auto now = steady_clock::now();
@@ -271,6 +295,96 @@ private:
return true;
}
return false;
}*/
static bool should_emit(const std::string& key, uint64_t& suppressed_before_emit) {
using namespace std::chrono;
const auto now = steady_clock::now();
suppressed_before_emit = 0;
std::lock_guard<std::mutex> lk(s_rate_mutex);
RateState& st = s_rate_map[key];
const int RESET_SEC = G_LOG_RATE_RESET_SEC ; // 1小时重置
const int LIMIT_SEC = G_LOG_RATE_LIMIT_SEC ; // 进入限流后多久发1条 主要控制中频和高频那些,低频的都直接放行了
// 初始化 / 强制每小时重置
if (st.last_reset.time_since_epoch().count() == 0) {
st.last_reset = now;
} else {
auto since_reset = duration_cast<seconds>(now - st.last_reset).count();
if (since_reset >= RESET_SEC) { //重置周期
st = RateState();
st.last_reset = now;
}
}
// 计算当前频率档位按“本次与上次看到该key的间隔”判断
// >=60秒/条:全部保留
// [1秒, 60秒)保留前60条然后1分钟1条
// <1秒保留前10条然后1分钟1条
int allow_burst = 0;
if (st.last_seen.time_since_epoch().count() == 0) {
// 第一次看到,先按“全部保留”处理
allow_burst = -1;
} else {
auto gap_ms = duration_cast<milliseconds>(now - st.last_seen).count();
if (gap_ms >= G_LOG_RATE_KEEP_ALL_MS) { //什么时候不需要限流 //低频 //如果这里设置的很低,就不会限流
allow_burst = -1; // 全部保留
} else if (gap_ms >= G_LOG_RATE_KEEP_BURST_MS) {
allow_burst = G_LOG_RATE_KEEP_BURST_COUNT; // 前60条 //中频 //如果这里设置的比低频低,也不会生效
} else {
allow_burst = G_LOG_RATE_KEEP_HIGHFREQ_COUNT; // 前10条 //高频
}
}
st.last_seen = now;
// 档位1全部保留
if (allow_burst == -1) {
suppressed_before_emit = st.suppressed_count;
st.suppressed_count = 0;
st.pass_count++;
st.last_emit = now;
st.has_emit = true;
return true;
}
// 档位2/3先放前N条
if (st.pass_count < (uint64_t)allow_burst) {
suppressed_before_emit = st.suppressed_count;
st.suppressed_count = 0;
st.pass_count++;
st.last_emit = now;
st.has_emit = true;
return true;
}
// 超过前N条后进入 1分钟1条
if (!st.has_emit) {
suppressed_before_emit = st.suppressed_count;
st.suppressed_count = 0;
st.pass_count++;
st.last_emit = now;
st.has_emit = true;
return true;
}
auto elapsed = duration_cast<seconds>(now - st.last_emit).count();
if (elapsed >= LIMIT_SEC) {
suppressed_before_emit = st.suppressed_count;
st.suppressed_count = 0;
st.pass_count++;
st.last_emit = now;
st.has_emit = true;
return true;
}
// 本条被抑制
st.suppressed_count++;
return false;
}
@@ -396,8 +510,17 @@ protected:
// ★新增:限频判断(同一条日志前 5 次 1 秒一次;之后 300 秒一次)
const std::string key = make_key(logger_name, level, code, msg);
if (!should_emit(key)) {
return;
uint64_t suppressed_before_emit = 0;
if (!should_emit(key, suppressed_before_emit)) return;
// 如果本次输出前压掉过日志,则在 log 文本后追加统计
std::string final_msg = msg;
if (suppressed_before_emit > 0) {
std::ostringstream suppressed_oss;
suppressed_oss << msg << " 【已过滤重复同类日志 "
<< suppressed_before_emit
<< " 条】";
final_msg = suppressed_oss.str();
}
std::ostringstream oss;
@@ -409,7 +532,7 @@ protected:
<< "\",\"grade\":\"" << get_level_str(level)
// ★建议code 用数字(不是字符串)
<< "\",\"code\":" << code
<< ",\"log\":\"" << escape_json(msg) << "\"}";
<< ",\"log\":\"" << escape_json(final_msg) << "\"}";
queue_data_t connect_info;
connect_info.strTopic = G_LOG_TOPIC;

View File

@@ -77,6 +77,9 @@ extern std::atomic<int> INITFLAG;
//测试用的终端数组
extern std::vector<std::string> TESTARRAY;
extern std::map<std::string, std::string> g_upgrade_file_map;
extern std::mutex g_upgrade_file_mutex;
////////////////////////////////////////////////////////////////////////////////////////////////////////外部文件函数声明
extern void execute_bash(std::string fun,int process_num,std::string type);
@@ -767,9 +770,10 @@ bool parseJsonMessageUD(const std::string& json_str, const std::string& output_d
}
json_data.processNo = std::to_string(procNo);
//int procNum = item.value("maxProcessNum", -1);
//json_data.maxProcessNum = std::to_string(procNum);
int procNum = item.value("maxProcessNum", -1);
json_data.maxProcessNum = std::to_string(procNum);
json_data.addr_str = item.value("ip", "");
json_data.mac = item.value("ip", "");
//json_data.port = item.value("port", "");
//json_data.timestamp = item.value("updateTime", "");
@@ -2595,7 +2599,7 @@ bool parsemsg(const std::string& devid, const std::string& guid, const nlohmann:
return true;
}
case 1115: { // 升级
case 1120: { // 辅助升级
parsed.ok = true;
std::cout << "[parsemsg] upgrade device, devid=" << devid
@@ -2605,14 +2609,14 @@ bool parsemsg(const std::string& devid, const std::string& guid, const nlohmann:
int ret = recordguid(devid, guid, static_cast<int>(DeviceState::SET_PREUPGRADE), 2);
if (-1 == ret) {
send_reply_to_queue(guid, static_cast<int>(ResponseCode::NOT_FOUND),
"未找到该装置,升级指令执行失败");
DIY_ERRORLOG_CODE(devid, 1, LOG_CODE_CLOUD, "未找到该装置,升级指令执行失败");
"未找到该装置,辅助升级指令执行失败");
DIY_ERRORLOG_CODE(devid, 1, LOG_CODE_CLOUD, "未找到该装置,辅助升级指令执行失败");
return true;
}
else if (ret > 0) {
send_reply_to_queue(guid, static_cast<int>(ResponseCode::BUSY),
"该装置正忙,升级指令执行失败");
DIY_WARNLOG_CODE(devid, 1, LOG_CODE_CLOUD, "该装置正忙,升级指令执行失败");
"该装置正忙,辅助升级指令执行失败");
DIY_WARNLOG_CODE(devid, 1, LOG_CODE_CLOUD, "该装置正忙,辅助升级指令执行失败");
return true;
}
else {
@@ -2621,6 +2625,63 @@ bool parsemsg(const std::string& devid, const std::string& guid, const nlohmann:
}
}
//辅助升级就是文件替换
//ClientManager::instance().set_preupgrade_action_to_device(devid, "");//尝试装置升级指令!第一步校验
return true;
}
case 1115: {
if (!msgObj.contains("Name") || !msgObj["Name"].is_string()) return false;
parsed.ok = true;
std::cout << "[parsemsg] update, devid=" << devid
<< ", guid=" << guid << std::endl;
//【1】recordguid
{
int ret = recordguid(devid, guid, static_cast<int>(DeviceState::SET_PREUPGRADE), 2);
if (-1 == ret) {
send_reply_to_queue(guid, static_cast<int>(ResponseCode::NOT_FOUND),
"未找到该装置,装置升级指令执行失败");
DIY_ERRORLOG_CODE(devid, 1, LOG_CODE_CLOUD, "未找到该装置,装置升级指令执行失败");
return true;
}
else if (ret > 0) {
send_reply_to_queue(guid, static_cast<int>(ResponseCode::BUSY),
"该装置正忙,装置升级指令执行失败");
DIY_WARNLOG_CODE(devid, 1, LOG_CODE_CLOUD, "该装置正忙,装置升级指令执行失败");
return true;
}
else {
std::cout << "记录装置状态成功,准备执行装置升级" << std::endl;
DIY_INFOLOG_CODE(devid, 1, LOG_CODE_CLOUD, "记录装置状态成功,准备执行装置升级");
}
}
//【2】取参数加 sanitize
std::string remote_path = sanitize(msgObj["Name"].get<std::string>()); // 云端路径
std::cout << "[parsemsg][1115] remote=" << remote_path
<< std::endl;
//【3】先下载到本地
std::string local_path = getfilefromweb(devid, remote_path,1);//升级文件下载到本地的路径和普通文件不一样放在专门的upgrade目录下
if (local_path.empty()) {
send_reply_to_queue(guid, static_cast<int>(ResponseCode::BAD_REQUEST),
"文件上送失败,下载源文件失败: " + remote_path);
DIY_ERRORLOG_CODE(devid, 1, LOG_CODE_CLOUD, "装置升级失败,下载源文件失败");
return true;
}
{
std::lock_guard<std::mutex> lk(g_upgrade_file_mutex);
g_upgrade_file_map[devid] = local_path;
}
DIY_INFOLOG_CODE(devid, 1, LOG_CODE_CLOUD, "升级文件下载成功,准备执行装置升级");
ClientManager::instance().set_preupgrade_action_to_device(devid, "");//尝试装置升级指令!第一步校验
return true;
}
@@ -2665,7 +2726,7 @@ bool parsemsg(const std::string& devid, const std::string& guid, const nlohmann:
<< ", dest=" << dest_file_path << std::endl;
//【3】先下载到本地
std::string local_path = getfilefromweb(devid, remote_path);
std::string local_path = getfilefromweb(devid, remote_path,0);//普通文件下载到本地的路径
if (local_path.empty()) {
send_reply_to_queue(guid, static_cast<int>(ResponseCode::BAD_REQUEST),
@@ -2849,6 +2910,10 @@ static int get_cloud_type_by_state(int type)
case DeviceState::SET_RIGHTTIME_2: return 1113; // 主动对时
case DeviceState::SET_CTRL: return 1114; // 控制/重启
case DeviceState::SET_PREUPGRADE: return 1115; // 升级预校验
case DeviceState::SEND_FILE: return 1116; // 文件上送
case DeviceState::DEL_FILE: return 1117; // 文件删除
case DeviceState::SEND_MENU: return 1118; // 目录创建
case DeviceState::DEL_MENU: return 1119; // 目录删除
default:
return type; // 其他未映射的,保持原值,避免影响现有逻辑

View File

@@ -15,7 +15,6 @@
#include "cloudfront/code/rocketmq.h" //lnk20250708
#include "cloudfront/code/log4.h" //lnk20250924
#include "client2.h"
#include "cloudfront/code/log4.h"
using namespace std;
@@ -245,6 +244,8 @@ void process_received_message(string mac, string id,const char* data, size_t len
//装置主动上送报文 暂态事件报文/暂态波形文件报文
if (udata[8] == static_cast<unsigned char>(MsgResponseType::Response_Event)) {
//处理主动上送的暂态事件报文
std::cout << "GET: MsgResponseType::Response_Event";
DIY_INFOLOG_CODE(id, 1, static_cast<int>(LogCode::LOG_CODE_TRANSIENT), "收到装置主动上送的暂态事件信息报文");
NewTaglogbuffer event = NewTaglogbuffer::createFromData(parser.RecvData.data(), parser.RecvData.size());
//获取测点id
@@ -366,6 +367,8 @@ void process_received_message(string mac, string id,const char* data, size_t len
}
else if (udata[8] == static_cast<unsigned char>(MsgResponseType::Response_ActiveSOEInfo)) {
//处理主动上送的波形文件信息报文
std::cout << "GET: MsgResponseType::Response_ActiveSOEInfo";
DIY_INFOLOG_CODE(id, 1, static_cast<int>(LogCode::LOG_CODE_TRANSIENT), "收到装置主动上送的暂态波形文件信息报文");
unsigned char file_type = udata[12];//录波文件类型数 cfg dat hdr 1-3
unsigned char line_id = udata[13];//录波测点 1-6
const uint8_t* data_ptr = parser.RecvData.data() + 2;//数据体去除前两位
@@ -847,8 +850,8 @@ void process_received_message(string mac, string id,const char* data, size_t len
<< ", Error: " << strerror(errno) << std::endl;
// 文件保存失败,通知云端
/*on_device_response_minimal(static_cast<int>(ResponseCode::BAD_REQUEST),
id, 0, static_cast<int>(DeviceState::READING_EVENTFILE));*/
//on_device_response_minimal(static_cast<int>(ResponseCode::BAD_REQUEST),
// id, 0, static_cast<int>(DeviceState::READING_EVENTFILE));
}
//当前文件下载完毕,调整为空闲处理下一项工作(如果这里后续有新文件等待下载,一般已经存入等待队列等候处理了,调成空闲状态后直接就会开始新文件的下载工作)
@@ -988,7 +991,7 @@ void process_received_message(string mac, string id,const char* data, size_t len
// 当前帧被拒收,文件上送失败
std::cout << "*** send file 0x41 fail ***! " << mac << std::endl;
on_device_response_minimal(static_cast<int>(ResponseCode::REJECTED_BUSY), id, 0, static_cast<int>(DeviceState::SEND_FILE));
on_device_response_minimal(static_cast<int>(ResponseCode::FORBIDDEN), id, 0, static_cast<int>(DeviceState::SEND_FILE));
ClientManager::instance().change_device_state(id, DeviceState::IDLE);
}
@@ -1016,7 +1019,7 @@ void process_received_message(string mac, string id,const char* data, size_t len
// 当前帧被拒收,文件删除失败
std::cout << "*** del file 0x41 fail ***! " << mac << std::endl;
on_device_response_minimal(static_cast<int>(ResponseCode::REJECTED_BUSY), id, 0, static_cast<int>(DeviceState::DEL_FILE));
on_device_response_minimal(static_cast<int>(ResponseCode::FORBIDDEN), id, 0, static_cast<int>(DeviceState::DEL_FILE));
ClientManager::instance().change_device_state(id, DeviceState::IDLE);
}
@@ -1044,7 +1047,7 @@ void process_received_message(string mac, string id,const char* data, size_t len
// 当前帧被拒收,创建目录失败
std::cout << "*** send menu 0x41 fail ***! " << mac << std::endl;
on_device_response_minimal(static_cast<int>(ResponseCode::REJECTED_BUSY), id, 0, static_cast<int>(DeviceState::SEND_MENU));
on_device_response_minimal(static_cast<int>(ResponseCode::FORBIDDEN), id, 0, static_cast<int>(DeviceState::SEND_MENU));
ClientManager::instance().change_device_state(id, DeviceState::IDLE);
}
@@ -1072,7 +1075,7 @@ void process_received_message(string mac, string id,const char* data, size_t len
// 当前帧被拒收,删除目录失败
std::cout << "*** del menu 0x41 fail ***! " << mac << std::endl;
on_device_response_minimal(static_cast<int>(ResponseCode::REJECTED_BUSY), id, 0, static_cast<int>(DeviceState::DEL_MENU));
on_device_response_minimal(static_cast<int>(ResponseCode::FORBIDDEN), id, 0, static_cast<int>(DeviceState::DEL_MENU));
ClientManager::instance().change_device_state(id, DeviceState::IDLE);
}

View File

@@ -20,6 +20,7 @@
#include <cstdlib>
#include <limits>
#include <utility>
#include <new>
// PQDIF 解析库
#include "pqdif/PQDIF.h"
@@ -37,6 +38,10 @@ namespace {
constexpr int kMaxPqdifFilesPerScan = 1;
constexpr size_t kParsedCacheLimit = 128;
// 大文件流式阈值:估算展开点数超过该值时,不再构造 expanded_stat_points
// 而是按通道逐步聚合成时间桶并直接组装 Base64避免单文件中间对象占用过大内存。
constexpr size_t kPqdifLargeFileStreamingPointThreshold = 800000;
const char* kPqdRootDir = "download";
const char* kDoneRootDir = "download_done";
const char* kFailRootDir = "download_fail";
@@ -5072,6 +5077,250 @@ namespace {
return stat_select_best_metric_sources(out);
}
size_t stat_estimate_candidate_point_count_for_streaming(const PqdifLogicalFile& lf)
{
size_t total = 0;
for (const auto& obs : lf.observations)
{
for (const auto& ch : obs.channel_instances)
{
if (!pqdif_sem::IsQuantityTypeValueLog(ch.quantity_type_id.value))
continue;
for (const auto& si : ch.series_instances)
{
if (pqdif_sem::IsValueTypeTime(si.value_type_id.value))
continue;
const PqdifSeriesInstance* resolved = stat_resolve_shared_series(obs, si);
if (resolved == nullptr || resolved->values.count <= 0)
continue;
total += static_cast<size_t>(resolved->values.count);
if (total > kPqdifLargeFileStreamingPointThreshold)
return total;
}
}
}
return total;
}
struct StatStreamingSourceBucket
{
StatMetricId metric_id = StatMetricId::Unknown;
StatMetricSourceKey key;
StatMetricSourceStats stats;
std::map<time_t, AggregatedStatValues> values_by_time;
std::map<time_t, std::string> time_text_by_time;
};
void stat_streaming_set_value(AggregatedStatValues& agg, const ExpandedStatPoint& p)
{
if (agg.source_observation_index < 0)
{
agg.source_observation_index = p.observation_index;
agg.source_channel_instance_index = p.channel_instance_index;
agg.source_series_instance_index = p.series_instance_index;
agg.source_channel_name = p.channel_name;
agg.quality = StatMetricQuality::Normal;
agg.quality_reason = "ok";
}
// 同一来源同一时间同一种统计值如果重复,保留首次写入,避免一次性覆盖造成结果不稳定。
if (stat_has_value_kind(agg, p.stat_kind))
return;
agg.source_series_instance_index = p.series_instance_index;
switch (p.stat_kind)
{
case StatValueKind::Min:
agg.has_min = true;
agg.min_value = p.value;
break;
case StatValueKind::Max:
agg.has_max = true;
agg.max_value = p.value;
break;
case StatValueKind::Avg:
agg.has_avg = true;
agg.avg_value = p.value;
break;
case StatValueKind::P95:
agg.has_p95 = true;
agg.p95_value = p.value;
break;
default:
break;
}
}
void stat_streaming_add_point(
std::map<StatMetricId, std::map<StatMetricSourceKey, StatStreamingSourceBucket>>& by_metric,
const ExpandedStatPoint& p)
{
if (p.metric_id == StatMetricId::Unknown || p.stat_kind == StatValueKind::Unknown)
return;
const StatMetricSourceKey key = stat_make_source_key(p);
StatStreamingSourceBucket& src = by_metric[p.metric_id][key];
if (src.metric_id == StatMetricId::Unknown)
{
src.metric_id = p.metric_id;
src.key = key;
}
src.stats.add(p);
AggregatedStatValues& agg = src.values_by_time[p.timestamp];
stat_streaming_set_value(agg, p);
if (src.time_text_by_time.find(p.timestamp) == src.time_text_by_time.end())
src.time_text_by_time[p.timestamp] = p.timestamp_text;
}
std::vector<TimeAggregatedStatBucket> stat_build_aggregated_buckets_streaming(
const PqdifLogicalFile& lf,
ParsedConnectionKind connection_kind,
int& selected_observation_index,
std::string& selected_observation_name,
size_t& out_raw_point_count,
size_t& out_selected_metric_count,
size_t& out_selected_dynamic_metric_count)
{
selected_observation_index = -1;
selected_observation_name.clear();
out_raw_point_count = 0;
out_selected_metric_count = 0;
out_selected_dynamic_metric_count = 0;
const PqdifObservationRecord* primary = stat_select_primary_statistical_observation(lf);
if (primary != nullptr)
{
selected_observation_index = primary->observation_index;
selected_observation_name = primary->observation_name;
}
std::map<StatMetricId, std::map<StatMetricSourceKey, StatStreamingSourceBucket>> by_metric;
// 大文件模式:遍历全部 observations但每次只临时展开一个通道随后立刻压缩到
// metric/source/timestamp/kind 聚合结构中,不保留全量 ExpandedStatPoint。
for (const auto& obs : lf.observations)
{
for (const auto& ch : obs.channel_instances)
{
std::vector<ExpandedStatPoint> points = stat_expand_channel_points(lf, connection_kind, obs, ch);
out_raw_point_count += points.size();
for (const auto& p : points)
stat_streaming_add_point(by_metric, p);
}
}
std::map<time_t, TimeAggregatedStatBucket> global_buckets;
size_t duplicate_metric_count = 0;
for (auto& metric_pair : by_metric)
{
const StatMetricId metric_id = metric_pair.first;
auto& sources = metric_pair.second;
if (sources.empty())
continue;
bool has_selected = false;
StatMetricSourceKey best_key;
int best_score = std::numeric_limits<int>::min();
for (auto& src_pair : sources)
{
int score = stat_metric_source_score(src_pair.second.stats);
// 大文件流式模式会遍历全部 observations。为尽量保持原逻辑普通核心指标在分数接近时
// 优先使用主统计 observation谐波等只存在于其他 observation 时仍能正常选中。
if (selected_observation_index >= 0 && src_pair.first.observation_index == selected_observation_index)
score += 25;
if (!has_selected || score > best_score)
{
has_selected = true;
best_score = score;
best_key = src_pair.first;
}
}
if (!has_selected)
continue;
++out_selected_metric_count;
if (stat_is_dynamic_metric(metric_id))
++out_selected_dynamic_metric_count;
if (sources.size() > 1)
++duplicate_metric_count;
StatStreamingSourceBucket& selected_src = sources[best_key];
for (auto& time_pair : selected_src.values_by_time)
{
TimeAggregatedStatBucket& bucket = global_buckets[time_pair.first];
if (bucket.timestamp == 0)
{
bucket.timestamp = time_pair.first;
const auto tit = selected_src.time_text_by_time.find(time_pair.first);
bucket.timestamp_text = (tit == selected_src.time_text_by_time.end()) ? format_time_text(time_pair.first) : tit->second;
}
bucket.metrics[metric_id] = time_pair.second;
}
}
std::vector<TimeAggregatedStatBucket> out;
out.reserve(global_buckets.size());
for (auto& kv : global_buckets)
out.push_back(std::move(kv.second));
if (pqdif_log_enabled(PqdifLogLevel::Core))
{
std::cout << "========== PQDIF LARGE FILE STREAMING SUMMARY ==========" << std::endl;
std::cout << "raw_points=" << out_raw_point_count
<< ", candidate_metric_count=" << by_metric.size()
<< ", selected_metric_count=" << out_selected_metric_count
<< ", selected_dynamic_metrics=" << out_selected_dynamic_metric_count << "/" << stat_all_dynamic_metric_order().size()
<< ", duplicate_metric_count=" << duplicate_metric_count
<< ", buckets=" << out.size()
<< ", selected_observation_index=" << selected_observation_index
<< ", selected_observation_name=" << selected_observation_name
<< std::endl;
std::cout << "mode=streaming_no_expanded_points; detail=per-channel expand then immediate aggregate" << std::endl;
std::cout << "========================================================" << std::endl;
}
return out;
}
void stat_print_aggregated_metric_line(StatMetricId metric_id, const AggregatedStatValues* agg);
void dump_grouped_bucket_streaming_preview(const ParsedPqdifFile& parsed_file)
{
std::cout << "========== GROUPED STAT STREAMING CORE SUMMARY ==========" << std::endl;
std::cout << "connection_kind=" << stat_connection_kind_name(parsed_file.connection_kind)
<< ", selected_observation_index=" << parsed_file.selected_observation_index
<< ", selected_observation_name=" << parsed_file.selected_observation_name
<< ", expanded_points=SKIPPED_BY_STREAMING"
<< ", buckets=" << parsed_file.aggregated_stat_buckets.size()
<< ", core_metric_slots=" << stat_core_metric_print_order().size()
<< ", dynamic_spectrum_slots=" << stat_all_dynamic_metric_order().size()
<< std::endl;
const size_t bucket_limit = std::min<size_t>(parsed_file.aggregated_stat_buckets.size(), 3);
for (size_t i = 0; i < bucket_limit; ++i)
{
const auto& b = parsed_file.aggregated_stat_buckets[i];
std::cout << " [BUCKET " << i << "] time=" << b.timestamp_text
<< ", metric_count_present=" << b.metrics.size()
<< std::endl;
for (const auto metric_id : stat_core_metric_print_order())
{
const auto it = b.metrics.find(metric_id);
if (it != b.metrics.end())
stat_print_aggregated_metric_line(metric_id, &it->second);
}
}
std::cout << "=========================================================" << std::endl;
}
bool pqdif_probe_text_looks_like_flicker(const std::string& text)
{
const std::string key = normalize_key(text);
@@ -7496,9 +7745,41 @@ namespace {
<< ", total_base64_chars=" << file_batch.total_base64_chars
<< std::endl;
// 按你的要求,入队前完整打印保存对象内部结构:文件批次 -> 时间点 -> Max/Min/Avg/P95 子记录 -> Base64 内容
// 日志会很长;确认完成后可以临时注释掉这一行,或改成 if (pqdif_log_enabled(PqdifLogLevel::Info)) 包裹
//pqdif_dump_stat_base64_file_batch_full(file_batch);
// 完整 Base64 内容日志非常长,尤其大文件会明显放大内存和 I/O 压力
// 现在仅在 Debug/Trace 级别打印完整对象Core/Info 只打印摘要和前几条子记录
if (pqdif_log_enabled(PqdifLogLevel::Debug))
{
pqdif_dump_stat_base64_file_batch_full(file_batch);
}
else
{
std::cout << " [BASE64 FILE BATCH COMPACT] file=" << file_batch.pqdif_file_path
<< ", time_points=" << file_batch.time_point_count
<< ", records=" << file_batch.total_record_count
<< ", total_float_count=" << file_batch.total_float_count
<< ", total_placeholder_count=" << file_batch.total_placeholder_count
<< ", total_base64_chars=" << file_batch.total_base64_chars
<< std::endl;
size_t shown = 0;
for (const auto& tp : file_batch.time_points)
{
for (const auto& rec : tp.records)
{
if (shown >= 4)
break;
std::cout << " [BASE64 SAVED SAMPLE] time=" << rec.timestamp_text
<< ", kind=" << rec.value_kind_name
<< ", floats=" << rec.float_count
<< ", placeholders=" << rec.placeholder_count
<< ", base64_len=" << rec.base64_payload.size()
<< std::endl;
++shown;
}
if (shown >= 4)
break;
}
}
push_pqdif_stat_base64_file_batch(std::move(file_batch));
@@ -7548,6 +7829,54 @@ namespace {
// 后续指标判断需要先区分星型 / 角型两套规则。
parsed_file.connection_kind = stat_classify_connection_kind(parsed_file.logical_file);
// 2) 根据估算展开点数决定是否启用“大文件流式模式”。
// 普通文件仍走原 expanded_stat_points -> grouped buckets 流程;
// 大文件则直接按通道聚合为 buckets不再保存全量 expanded_stat_points。
const size_t estimated_candidate_points =
stat_estimate_candidate_point_count_for_streaming(parsed_file.logical_file);
const bool use_large_file_streaming =
estimated_candidate_points > kPqdifLargeFileStreamingPointThreshold;
if (use_large_file_streaming)
{
std::cout << "[PQDIF][MEMORY][STREAMING] large file detected, use streaming bucket/base64 build. "
<< "estimated_candidate_points=" << estimated_candidate_points
<< ", threshold=" << kPqdifLargeFileStreamingPointThreshold
<< std::endl;
size_t raw_point_count = 0;
size_t selected_metric_count = 0;
size_t selected_dynamic_metric_count = 0;
parsed_file.aggregated_stat_buckets = stat_build_aggregated_buckets_streaming(
parsed_file.logical_file,
parsed_file.connection_kind,
parsed_file.selected_observation_index,
parsed_file.selected_observation_name,
raw_point_count,
selected_metric_count,
selected_dynamic_metric_count);
if (pqdif_is_trace_log_enabled())
dump_semantic_probe(parsed_file);
dump_grouped_bucket_streaming_preview(parsed_file);
// 3) 大文件仍完整生成 Base64 文件批次,只是跳过 expanded_stat_points 中间缓存。
pqdif_build_and_queue_base64_records(parsed_file);
// 大文件模式不再把完整 ParsedPqdifFile 放入解析缓存,避免 logical_file + buckets 在内存中长期驻留。
// 后续业务请从 PqdifStatBase64FileBatch 队列读取最终 Base64 结果。
std::cout << "[PQDIF][MEMORY][STREAMING] skip parsed_file cache for large file, "
<< "base64 batch has been queued. file=" << file_path.string()
<< std::endl;
std::cout << "[PQDIF] processed ok(streaming): " << file_path.string()
<< ", cache_size=" << GetParsedPqdifCacheSize()
<< std::endl;
return true;
}
// 2) 只筛选“统计类 observation”并对其展开目标统计指标。
// 当前阶段不处理全部 observation避免不同 observation 混入同一时间聚合结果。
parsed_file.expanded_stat_points = stat_expand_selected_statistical_observation(
@@ -7655,14 +7984,50 @@ namespace {
<< ", max_per_scan=" << kMaxPqdifFilesPerScan
<< std::endl;
const bool ok = process_single_pqdif_file(item.path, item.mac);
bool ok = false;
bool caught_exception = false;
std::string exception_text;
try
{
ok = process_single_pqdif_file(item.path, item.mac);
}
catch (const std::bad_alloc& ex)
{
ok = false;
caught_exception = true;
exception_text = ex.what();
std::cout << "[PQDIF][OOM] std::bad_alloc while processing file="
<< item.path.string()
<< ", what=" << exception_text
<< std::endl;
}
catch (const std::exception& ex)
{
ok = false;
caught_exception = true;
exception_text = ex.what();
std::cout << "[PQDIF][ERROR] exception while processing file="
<< item.path.string()
<< ", what=" << exception_text
<< std::endl;
}
catch (...)
{
ok = false;
caught_exception = true;
exception_text = "unknown";
std::cout << "[PQDIF][ERROR] unknown exception while processing file="
<< item.path.string()
<< std::endl;
}
const fs::path target_root = ok ? fs::path(kDoneRootDir) : fs::path(kFailRootDir);
const fs::path dst = target_root / item.mac / item.path.filename();
// 处理完成后移动文件,避免下一轮 scan_once() 重复解析同一个 PQDIF。
// 解析成功download/<mac>/<file>.pqd -> download_done/<mac>/<file>.pqd
// 解析失败download/<mac>/<file>.pqd -> download_fail/<mac>/<file>.pqd
// 解析失败或解析过程中出现异常download/<mac>/<file>.pqd -> download_fail/<mac>/<file>.pqd
//
// 调试时如果想让文件保留在 download 目录中、方便反复解析同一个文件,
// 可以临时注释掉下面这个 if 块;调试结束后建议恢复,否则每一轮都会重复解析。
@@ -7672,6 +8037,13 @@ namespace {
<< item.path.string() << " -> " << dst.string()
<< std::endl;
}
else if (caught_exception)
{
std::cout << "[PQDIF] exception file moved to fail dir: "
<< dst.string()
<< ", reason=" << exception_text
<< std::endl;
}
cleanup_backup_dir(fs::path(kDoneRootDir) / item.mac, kBackupLimit);
cleanup_backup_dir(fs::path(kFailRootDir) / item.mac, kBackupLimit);