更新:1、优化数据记录的性能,满足实时运行要求;

Signed-off-by: waiwaylee <waiwaylee@foxmail.com>
This commit is contained in:
2026-10-20 17:31:50 +08:00
parent 3dee734d89
commit 17a3d998a6
14 changed files with 2029 additions and 210 deletions
+25 -6
View File
@@ -1,7 +1,6 @@
#include "data_process.h"
#include "rdp_log.h"
#include "rdp_record.h"
#include "rdp_record_entity.h"
#include <cstring>
#include <vector>
#include <cmath>
@@ -26,6 +25,11 @@ int Data_Process::data_preprocess(struct DataRev Data_Input[150])
{
data_num=min(Data_Input[0].Point_Sum, 150);
//std::cout << "TARGET azi beam idx:" <<Data_Input[0].Beam_index_aiz;
#ifdef RDP_RECORD_ENABLE
//记录输入点迹:处理线程内先收集源结构体,循环结束后一次入队(构造实体与写库由记录写线程完成)
std::vector<PointRecv> rec_points;
rec_points.reserve(static_cast<std::size_t>(data_num));
#endif
for (int i=0; i<data_num;i++)
{
PointRecv Data_buffer_temp;
@@ -54,14 +58,18 @@ int Data_Process::data_preprocess(struct DataRev Data_Input[150])
std::memcpy(Data_buffer_temp.speed_dim, Data_Input[i].speed_dim, sizeof(Data_buffer_temp.speed_dim));
std::memcpy(Data_buffer_temp.range_dim, Data_Input[i].range_dim, sizeof(Data_buffer_temp.range_dim));
//数据记录示例①:把本次接收的输入点迹写入表 RdpRecPointTws(字段为示例,可按需增删)。
//记录功能关闭时该语句展开为空操作(实参不求值),交付构建不产生任何代码,
//用法与新增记录点配方见 rdp_record.h 与 docs/data_record_debug.md。
RDP_RECORD(RdpRecPoint(Data_buffer_temp));
#ifdef RDP_RECORD_ENABLE
rec_points.push_back(Data_buffer_temp);
#endif
Data_buffer.push_back(Data_buffer_temp);
}
#ifdef RDP_RECORD_ENABLE
//判圈之前(所有 return 之前)把本帧收集的点迹一次入队
RDP_RECORD_POINTS(rec_points);
#endif
#ifdef PHASE_SCANNING
// 相扫模式下,判断前后两帧波位号的大小,来判断是否扫描一周完成
if(Data_Input[0].Beam_index_aiz <= last_beam_num) //TWS
@@ -101,6 +109,11 @@ int Data_Process::data_preprocess(struct DataRev Data_Input[150])
data_num = min(max(Data_Input[0].Point_Sum, 0), 150);
TAS_track_idx = Data_Input[0].TAS_track_index;
#ifdef RDP_RECORD_ENABLE
//记录输入点迹:处理线程内先收集源结构体,循环结束后一次入队(构造实体与写库由记录写线程完成)
std::vector<PointRecv> rec_points;
rec_points.reserve(static_cast<std::size_t>(data_num));
#endif
for (int i=0; i<data_num;i++)
{
PointRecv Data_buffer_temp;
@@ -129,7 +142,9 @@ int Data_Process::data_preprocess(struct DataRev Data_Input[150])
std::memcpy(Data_buffer_temp.speed_dim, Data_Input[i].speed_dim, sizeof(Data_buffer_temp.speed_dim));
std::memcpy(Data_buffer_temp.range_dim, Data_Input[i].range_dim, sizeof(Data_buffer_temp.range_dim));
RDP_RECORD(RdpRecPoint(Data_buffer_temp));
#ifdef RDP_RECORD_ENABLE
rec_points.push_back(Data_buffer_temp);
#endif
Data_buffer_tas.push_back(Data_buffer_temp);
@@ -140,6 +155,10 @@ int Data_Process::data_preprocess(struct DataRev Data_Input[150])
// << "data_num:" << data_num;
}
//circle_id++ 之前把本帧收集的点迹一次入队
#ifdef RDP_RECORD_ENABLE
RDP_RECORD_POINTS(rec_points);
#endif
circle_id++;
return 3;
}
+11 -2
View File
@@ -3,7 +3,6 @@
#include <cstring>
#include <vector>
#include "rdp_record.h"
#include "rdp_record_entity.h"
using namespace std;
@@ -224,11 +223,21 @@ int Dot_Coh::dot_coh_process_buff( std::vector <PointRecv> *data_input,
//2.data_input_buff数据输出给point_recv
#ifdef RDP_RECORD_ENABLE
//记录凝聚后点迹:先收集,输出循环结束后一次入队
std::vector<PointRecv> rec_coh;
rec_coh.reserve(data_input_buff.size());
#endif
for (size_t i=0;i<data_input_buff.size();i++)
{
(*point_recv).push_back(data_input_buff[i]);
RDP_RECORD(RdpRecPointCohesionTWS(data_input_buff[i]));
#ifdef RDP_RECORD_ENABLE
rec_coh.push_back(data_input_buff[i]);
#endif
}
#ifdef RDP_RECORD_ENABLE
RDP_RECORD_COHESION_TWS(rec_coh);
#endif
std::vector<PointRecv>().swap(data_input_buff);
+11 -2
View File
@@ -3,7 +3,6 @@
#include <cstring>
#include <vector>
#include "rdp_record.h"
#include "rdp_record_entity.h"
using namespace std;
int Dot_Coh_TAS::dot_coh_tas_process(std::vector <PointRecv> *data_input, //输入点迹
@@ -67,11 +66,21 @@ int Dot_Coh_TAS::dot_coh_tas_process(std::vector <PointRecv> *dat
}
}
#ifdef RDP_RECORD_ENABLE
//记录凝聚后点迹:先收集,输出循环结束后一次入队
std::vector<PointRecv> rec_coh;
rec_coh.reserve(data_input->size());
#endif
for (size_t i=0;i<data_input->size();i++)
{
(*point_recv_tas).push_back((*data_input)[i]);
RDP_RECORD(RdpRecPointCohesionTAS((*data_input)[i]));
#ifdef RDP_RECORD_ENABLE
rec_coh.push_back((*data_input)[i]);
#endif
}
#ifdef RDP_RECORD_ENABLE
RDP_RECORD_COHESION_TAS(rec_coh);
#endif
return 1;
}
+44
View File
@@ -46,6 +46,50 @@
//记录数据库存放目录(相对宿主进程当前工作目录,不存在时自动创建)
#define RDP_RECORD_DB_DIR ".\\rdp_records\\"
//数据记录性能参数(V1.5.8 优化:批量事务 + 异步写线程 + WAL)
//说明:以下宏只被 rdp_record.cpp 使用;用 #ifndef 保护,便于用编译选项临时覆盖做测试。
//队列字节上限:处理线程入队总量超过该值时整批丢弃(不阻塞处理线程)
#ifndef RDP_RECORD_QUEUE_CAP_BYTES
#define RDP_RECORD_QUEUE_CAP_BYTES (64u * 1024u * 1024u)
#endif
//队列批数上限(防止小批过多导致 deque 开销/延迟)
#ifndef RDP_RECORD_QUEUE_MAX_BATCHES
#define RDP_RECORD_QUEUE_MAX_BATCHES 4096u
#endif
//写线程每次事务提交的目标行数(累积到该行数或等待超时即提交)
#ifndef RDP_RECORD_WRITE_BATCH_ROWS
#define RDP_RECORD_WRITE_BATCH_ROWS 4096u
#endif
//写线程为凑批而等待的最长时间(毫秒)
#ifndef RDP_RECORD_WRITE_MAX_WAIT_MS
#define RDP_RECORD_WRITE_MAX_WAIT_MS 50u
#endif
//关联信息单次入队分块上限(⑥ 单圈行数可能很大,分块限制处理线程内存峰值;
//取值应大于常见单圈候选数,避免同一圈内多次重新分配;131072 行 Asso_Info 约 11 MB)
#ifndef RDP_RECORD_ASSO_CHUNK_ROWS
#define RDP_RECORD_ASSO_CHUNK_ROWS 131072u
#endif
//SQLite 页缓存大小(KB,负数表示 KB;默认 8 MB)
#ifndef RDP_RECORD_SQLITE_CACHE_KB
#define RDP_RECORD_SQLITE_CACHE_KB 8192
#endif
//SQLite 忙等待超时(毫秒)
#ifndef RDP_RECORD_SQLITE_BUSY_TIMEOUT_MS
#define RDP_RECORD_SQLITE_BUSY_TIMEOUT_MS 2000
#endif
//是否尝试开启 WAL(1=尝试;失败/不支持时自动回退 DELETE 日志模式)
#ifndef RDP_RECORD_SQLITE_WAL
#define RDP_RECORD_SQLITE_WAL 1
#endif
//连续写入错误达到该次数后全局熔断(停止记录,避免持续消耗写线程)
#ifndef RDP_RECORD_MAX_CONSECUTIVE_ERRORS
#define RDP_RECORD_MAX_CONSECUTIVE_ERRORS 10
#endif
//队列丢弃日志的限频间隔(每丢弃多少行提示一次)
#ifndef RDP_RECORD_DROP_LOG_INTERVAL
#define RDP_RECORD_DROP_LOG_INTERVAL 1024u
#endif
//#define MECHANICAL_SCANNING
//#define PHASE_SCANNING
+795 -77
View File
@@ -2,19 +2,45 @@
// 本文件仅在构建系统定义 RDP_RECORD_ENABLE(CMake 选项 RDP_DATA_RECORD 开启)时参与编译,
// 且以 C++20 编译(ormpp 要求);雷达处理代码仍为 C++17,二者通过 rdp_record.h 的
// 「仅声明 + 显式实例化」跨标准边界,详见 docs/data_record_debug.md。
//
// 性能模型(V1.5.8 优化):
// 处理线程:记录点只做「收集源结构体 → 一次入队」(拷贝 + 互斥队列 push,不阻塞);
// 写线程 :从有界队列取批 → 构造记录实体(数组转文本在这里)→ ormpp 批量事务写 SQLite;
// 数据库 :WAL + synchronous=OFF + 8MB 页缓存 + busy_timeout;建表去 AUTOINCREMENT;
// 队列满 :整批丢弃并计数(实时性优先),Shutdown 时排空并打印统计。
#ifdef RDP_RECORD_ENABLE
//注意:windows.h 必须在业务/实体头文件之前包含 —— struct.h 在全局作用域有
//"using namespace std;",若之后再包含 windows.h,其 byte 类型会与 std::byte 产生歧义。
#ifdef _WIN32
#ifndef NOMINMAX
#define NOMINMAX
#endif
#include <windows.h> //仅用于把记录写线程优先级降到 BELOW_NORMAL,避免与雷达处理线程争抢 CPU
#endif
#include "rdp_record.h"
#include "rdp_record_entity.h"
#include "parameters.h"
#include "rdp_log.h"
#include <atomic>
#include <chrono>
#include <climits>
#include <condition_variable>
#include <cstdint>
#include <deque>
#include <direct.h>
#include <io.h>
#include <iterator>
#include <map>
#include <memory>
#include <mutex>
#include <string>
#include <thread>
#include <time.h>
#include <tuple>
#include <utility>
#include <vector>
//第三方库(ormpp / iguana)模板代码在 /W3 下可能产生告警,按 third_party/ormpp/UPSTREAM.md
@@ -31,27 +57,86 @@
namespace
{
//============================== 基础类型 ==============================
enum class BatchKind : std::uint8_t
{
Point, //RdpRecPoint(输入点迹)
CohesionTws, //RdpRecPointCohesionTWS
CohesionTas, //RdpRecPointCohesionTAS
TempTrack, //RdpRecTempTrack
TrustTrack, //RdpRecTrustTrack
AssoInfo //RdpRecAssoInfo
};
//一批入队数据(一种记录表的一批源结构体;处理线程收集后整体 move 进队列)
struct Batch
{
BatchKind kind = BatchKind::Point;
std::vector<PointRecv> points; //Point/CohesionTws/CohesionTas 共用
std::vector<Temp_track> temp_tracks;
std::vector<Trust_Track> trust_tracks;
std::vector<Asso_Info> asso_infos;
};
//写线程内的待提交缓冲(按表分别累积,达到批大小或超时后各表分别提交)
struct Pending
{
std::vector<PointRecv> points;
std::vector<PointRecv> cohesion_tws;
std::vector<PointRecv> cohesion_tas;
std::vector<Temp_track> temp_tracks;
std::vector<Trust_Track> trust_tracks;
std::vector<Asso_Info> asso_infos;
};
//统计计数器(处理线程/写线程并发读写)
struct Counters
{
std::atomic<std::uint64_t> enqueued_rows{0};
std::atomic<std::uint64_t> dropped_rows{0};
std::atomic<std::uint64_t> written_rows{0};
std::atomic<std::uint64_t> transactions{0};
std::atomic<std::uint64_t> write_errors{0};
std::atomic<std::uint64_t> queue_peak_bytes{0};
std::atomic<std::uint64_t> queue_peak_rows{0};
};
//记录模块内部状态(Pimpl:ormpp / SQLite 类型不出现在 rdp_record.h 中)
struct Record_State
{
ormpp::dbng<ormpp::sqlite> db; //SQLite 连接(单连接,写入时串行使用)
std::string path; //当前 db 文件全路径
bool ready; //是否已成功打开并完成初始化
bool failed; //初始化失败:后续写入全部忽略,不再反复报错
std::mutex mtx; //写入串行化(宿主多线程调用时保护单连接)
ormpp::dbng<ormpp::sqlite> db; //SQLite 连接(写线程独占;遗留同步接口经 db_mtx 串行)
std::string path; //当前 db 文件全路径
std::atomic<bool> ready{false};
std::atomic<bool> failed{false};
Record_State() : ready(false), failed(false) {}
std::mutex db_mtx; //串行化数据库访问(写线程 + 遗留同步写)
std::map<std::string, int> tables; //表状态:0 未建 / 1 可用 / 2 失败(db_mtx 保护)
std::mutex q_mtx; //队列锁
std::condition_variable q_cv; //队列条件变量
std::deque<Batch> queue; //有界入队队列
std::size_t queued_bytes = 0;
std::size_t queued_rows = 0;
std::size_t queued_batches = 0;
bool stopping = false;
std::thread writer; //记录写线程
Counters counters;
std::atomic<int> consecutive_errors{0};
~Record_State();
};
std::unique_ptr<Record_State>& State()
{
//函数内静态:宿主未调用 Data_Process_Factory::Destroy() 时由析构兜底关库
//函数内静态:宿主未调用 Data_Process_Factory::Destroy() 时由析构兜底停止写线程/关库
static std::unique_ptr<Record_State> s;
return s;
}
//============================== 通用小工具 ==============================
//当前本地时间字符串,格式 "YYYYMMDD_HHMMSS"(与 rdp_log 的日志文件命名风格一致)
bool TimeStamp(char* buf, size_t size)
bool TimeStamp(char* buf, std::size_t size)
{
time_t now = time(NULL);
struct tm local;
@@ -62,39 +147,513 @@ namespace
return strftime(buf, size, "%Y%m%d_%H%M%S", &local) != 0;
}
//写入失败提示(「每个实体只提示一次」由调用方的静态标志控制)
void ReportError(Record_State& st, const char* what)
//写入失败提示
void ReportError(Record_State& st, const char* what, const std::string& detail)
{
RDP_LOG << "[rdp_record] " << what << " failed: " << st.db.get_last_error() << std::endl;
RDP_LOG << "[rdp_record] " << what << " failed (" << detail << "): "
<< st.db.get_last_error() << std::endl;
}
//确保实体的数据表存在:每个实体类型只尝试建表一次(失败后该实体永久跳过)
//注意:ormpp 的 ormpp_auto_key 定义在全局命名空间(third_party/ormpp/ormpp/entity.hpp),
// 传该参数可在建表时登记自增主键,使后续插入自动跳过 id 列
template <typename T>
bool EnsureTable(Record_State& st)
void LogStats(Record_State& st, const char* tag)
{
static const bool created = st.db.create_datatable<T>(::ormpp_auto_key{"id"});
return created;
RDP_LOG << "[rdp_record] stats(" << tag << "): enqueued_rows="
<< st.counters.enqueued_rows.load()
<< ", written_rows=" << st.counters.written_rows.load()
<< ", dropped_rows=" << st.counters.dropped_rows.load()
<< ", transactions=" << st.counters.transactions.load()
<< ", write_errors=" << st.counters.write_errors.load()
<< ", queue_peak_bytes=" << st.counters.queue_peak_bytes.load()
<< ", queue_peak_rows=" << st.counters.queue_peak_rows.load()
<< std::endl;
}
//写入前置处理:自动建表;返回 true 表示可以向该表写入
template <typename T>
bool PrepareWrite(Record_State& st)
std::string TableName()
{
static bool reported = false;
if (EnsureTable<T>(st))
//ormpp 以「短结构体名」作为表名
return std::string(ormpp::get_short_struct_name<T>());
}
//============================== 记录批次辅助 ==============================
template <typename T>
std::size_t BatchRowsOf(const std::vector<T>& v)
{
return v.size();
}
std::size_t BatchRows(const Batch& b)
{
return b.points.size() + b.temp_tracks.size() + b.trust_tracks.size() + b.asso_infos.size();
}
std::size_t BatchBytes(const Batch& b)
{
//近似统计(用于有界队列的字节上限;容器容量 + 对象本身)
std::size_t bytes = sizeof(Batch);
bytes += b.points.capacity() * sizeof(PointRecv);
bytes += b.temp_tracks.capacity() * sizeof(Temp_track);
bytes += b.trust_tracks.capacity() * sizeof(Trust_Track);
bytes += b.asso_infos.capacity() * sizeof(Asso_Info);
return bytes;
}
std::size_t PendingRows(const Pending& p)
{
return p.points.size() + p.cohesion_tws.size() + p.cohesion_tas.size()
+ p.temp_tracks.size() + p.trust_tracks.size() + p.asso_infos.size();
}
void ClearPending(Pending& p)
{
p.points.clear();
p.cohesion_tws.clear();
p.cohesion_tas.clear();
p.temp_tracks.clear();
p.trust_tracks.clear();
p.asso_infos.clear();
}
template <typename T>
void AppendMove(std::vector<T>& dst, std::vector<T>& src)
{
if (src.empty())
{
return;
}
if (dst.empty())
{
dst = std::move(src);
return;
}
dst.insert(dst.end(), std::make_move_iterator(src.begin()), std::make_move_iterator(src.end()));
src.clear();
}
void AppendBatch(Pending& p, Batch& b)
{
switch (b.kind)
{
case BatchKind::Point: AppendMove(p.points, b.points); break;
case BatchKind::CohesionTws: AppendMove(p.cohesion_tws, b.points); break;
case BatchKind::CohesionTas: AppendMove(p.cohesion_tas, b.points); break;
case BatchKind::TempTrack: AppendMove(p.temp_tracks, b.temp_tracks); break;
case BatchKind::TrustTrack: AppendMove(p.trust_tracks, b.trust_tracks); break;
case BatchKind::AssoInfo: AppendMove(p.asso_infos, b.asso_infos); break;
}
}
//============================== 数据库初始化/建表 ==============================
//应用 SQLite PRAGMA(失败只记录,不阻止记录功能;调试用途)
void ApplyPragmas(Record_State& st)
{
#if RDP_RECORD_SQLITE_WAL
try
{
//PRAGMA journal_mode 会返回一行,ormpp 的 execute() 只接受不返回结果行的语句,
//因此用 query_s<std::tuple<...>> 消费返回行。
auto rows = st.db.query_s<std::tuple<std::string>>("PRAGMA journal_mode=WAL;");
std::string mode = rows.empty() ? std::string("<none>") : std::get<0>(rows[0]);
RDP_LOG << "[rdp_record] journal_mode=" << mode << std::endl;
}
catch (...)
{
RDP_LOG << "[rdp_record] set journal_mode=WAL failed, fallback to default journal" << std::endl;
}
#endif
if (!st.db.execute("PRAGMA synchronous=OFF;"))
{
RDP_LOG << "[rdp_record] set synchronous=OFF failed" << std::endl;
}
const std::string cache = "PRAGMA cache_size=-" + std::to_string(RDP_RECORD_SQLITE_CACHE_KB) + ";";
if (!st.db.execute(cache))
{
RDP_LOG << "[rdp_record] set cache_size failed" << std::endl;
}
try
{
//busy_timeout 也返回一行
st.db.query_s<std::tuple<int>>(
"PRAGMA busy_timeout=" + std::to_string(RDP_RECORD_SQLITE_BUSY_TIMEOUT_MS) + ";");
}
catch (...)
{
RDP_LOG << "[rdp_record] set busy_timeout failed" << std::endl;
}
}
//建表 + 去掉 AUTOINCREMENT(本项目每次运行新建库、只增不删,普通 INTEGER PRIMARY KEY 即可;
//去 AUTOINCREMENT 可省掉每行对 sqlite_sequence 的读写。ormpp 的 DDL 硬编码 AUTOINCREMENT,
//这里先用 ormpp 建表(同时登记自增主键映射),再按 sqlite_master 中的 DDL 去关键字重建空表。)
template <typename T>
bool CreateTable(Record_State& st)
{
if (!st.db.create_datatable<T>(::ormpp_auto_key{"id"}))
{
return false;
}
const std::string table = TableName<T>();
std::string ddl;
try
{
auto rows = st.db.query_s<std::tuple<std::string>>(
"select sql from sqlite_master where type='table' and name='" + table + "'");
if (!rows.empty())
{
ddl = std::get<0>(rows[0]);
}
}
catch (...)
{
return true; //查询失败不影响记录,保留 ormpp 建出的表
}
if (ddl.find("AUTOINCREMENT") == std::string::npos)
{
return true; //ormpp 版本变化后不再带 AUTOINCREMENT,直接使用
}
std::string stripped = ddl;
const std::string keyword = "AUTOINCREMENT";
for (std::size_t pos = stripped.find(keyword); pos != std::string::npos;
pos = stripped.find(keyword))
{
stripped.erase(pos, keyword.size());
}
if (st.db.execute("DROP TABLE \"" + table + "\";") &&
st.db.execute(stripped))
{
RDP_LOG << "[rdp_record] table " << table << " created without AUTOINCREMENT" << std::endl;
return true;
}
if (!reported)
//回退:若重建失败(极端情况),尽量恢复一张可用表,保证记录功能继续
if (st.db.create_datatable<T>(::ormpp_auto_key{"id"}))
{
reported = true;
ReportError(st, "create table");
RDP_LOG << "[rdp_record] warning: keep AUTOINCREMENT for " << table << std::endl;
return true;
}
return false;
}
//写入前置处理(调用方需持有 db_mtx):按表缓存建表结果,失败的表永久跳过
template <typename T>
bool PrepareWrite(Record_State& st)
{
const std::string table = TableName<T>();
std::map<std::string, int>::iterator it = st.tables.find(table);
const int state = (it == st.tables.end()) ? 0 : it->second;
if (state == 2)
{
return false;
}
if (state == 1)
{
return true;
}
if (CreateTable<T>(st))
{
st.tables[table] = 1;
return true;
}
st.tables[table] = 2;
ReportError(st, "create table", table);
return false;
}
//批量写入实体(一次 BEGIN/COMMIT 事务;调用方无需持锁)
template <typename T>
bool InsertEntities(Record_State& st, const std::vector<T>& v)
{
if (v.empty())
{
return true;
}
std::lock_guard<std::mutex> lock(st.db_mtx);
if (st.failed.load())
{
return false;
}
if (!PrepareWrite<T>(st))
{
return false;
}
const int rows = st.db.insert(v);
if (rows == INT_MIN)
{
st.counters.write_errors.fetch_add(1);
st.tables[TableName<T>()] = 2;
const int errors = st.consecutive_errors.fetch_add(1) + 1;
ReportError(st, "insert", TableName<T>());
if (errors >= RDP_RECORD_MAX_CONSECUTIVE_ERRORS)
{
st.failed.store(true);
RDP_LOG << "[rdp_record] too many consecutive write errors, recording stopped" << std::endl;
}
return false;
}
st.consecutive_errors.store(0);
st.counters.written_rows.fetch_add(v.size());
st.counters.transactions.fetch_add(1);
return true;
}
//按批大小分块构造实体并提交(实体构造 = 数组转文本,全部在写线程完成)
template <typename Entity, typename Source>
bool FlushKind(Record_State& st, std::vector<Source>& src)
{
if (src.empty())
{
return true;
}
const std::size_t chunk = (RDP_RECORD_WRITE_BATCH_ROWS == 0)
? 4096
: static_cast<std::size_t>(RDP_RECORD_WRITE_BATCH_ROWS);
bool ok = true;
for (std::size_t pos = 0; pos < src.size() && !st.failed.load();)
{
std::size_t n = src.size() - pos;
if (n > chunk)
{
n = chunk;
}
std::vector<Entity> entities;
entities.reserve(n);
for (std::size_t i = 0; i < n; ++i)
{
entities.emplace_back(src[pos + i]); //Entity(Source&) 构造:数组转文本在此完成
}
if (!InsertEntities<Entity>(st, entities))
{
ok = false;
}
pos += n;
}
src.clear();
return ok;
}
bool FlushPending(Record_State& st, Pending& p)
{
bool ok = true;
ok &= FlushKind<RdpRecPoint>(st, p.points);
ok &= FlushKind<RdpRecPointCohesionTWS>(st, p.cohesion_tws);
ok &= FlushKind<RdpRecPointCohesionTAS>(st, p.cohesion_tas);
ok &= FlushKind<RdpRecTempTrack>(st, p.temp_tracks);
ok &= FlushKind<RdpRecTrustTrack>(st, p.trust_tracks);
ok &= FlushKind<RdpRecAssoInfo>(st, p.asso_infos);
ClearPending(p);
return ok;
}
//============================== 入队/写线程 ==============================
void LogDroppedRows(std::uint64_t total_dropped, std::uint64_t newly_dropped)
{
const std::uint64_t interval = (RDP_RECORD_DROP_LOG_INTERVAL == 0)
? 0
: static_cast<std::uint64_t>(RDP_RECORD_DROP_LOG_INTERVAL);
if (interval == 0)
{
return;
}
if ((total_dropped / interval) != ((total_dropped - newly_dropped) / interval))
{
RDP_LOG << "[rdp_record] queue full, dropped rows total=" << total_dropped << std::endl;
}
}
void EnqueueBatch(Batch&& b)
{
const std::size_t rows = BatchRows(b);
if (rows == 0)
{
return;
}
std::unique_ptr<Record_State>& sp = State();
if (sp.get() == 0)
{
return;
}
if (!sp->ready || sp->failed.load())
{
sp->counters.dropped_rows.fetch_add(rows);
return;
}
const std::size_t bytes = BatchBytes(b);
std::lock_guard<std::mutex> lock(sp->q_mtx);
if (sp->stopping || sp->failed.load())
{
const std::uint64_t total = sp->counters.dropped_rows.fetch_add(rows) + rows;
LogDroppedRows(total, rows);
return;
}
if (sp->queued_batches + 1 > static_cast<std::size_t>(RDP_RECORD_QUEUE_MAX_BATCHES) ||
sp->queued_bytes + bytes > static_cast<std::size_t>(RDP_RECORD_QUEUE_CAP_BYTES))
{
const std::uint64_t total = sp->counters.dropped_rows.fetch_add(rows) + rows;
LogDroppedRows(total, rows);
return;
}
sp->queue.push_back(std::move(b));
sp->queued_batches += 1;
sp->queued_bytes += bytes;
sp->queued_rows += rows;
sp->counters.enqueued_rows.fetch_add(rows);
if (sp->counters.queue_peak_bytes.load() < sp->queued_bytes)
{
sp->counters.queue_peak_bytes.store(sp->queued_bytes);
}
if (sp->counters.queue_peak_rows.load() < sp->queued_rows)
{
sp->counters.queue_peak_rows.store(sp->queued_rows);
}
sp->q_cv.notify_one();
}
void DropQueued(Record_State& st)
{
std::size_t rows = 0;
{
std::lock_guard<std::mutex> lock(st.q_mtx);
for (std::size_t i = 0; i < st.queue.size(); ++i)
{
rows += BatchRows(st.queue[i]);
}
st.queue.clear();
st.queued_bytes = 0;
st.queued_rows = 0;
st.queued_batches = 0;
}
if (rows > 0)
{
const std::uint64_t total = st.counters.dropped_rows.fetch_add(rows) + rows;
RDP_LOG << "[rdp_record] writer stopped, dropped queued rows=" << rows
<< ", total dropped=" << total << std::endl;
}
}
void WriterLoopBody(Record_State& st)
{
Pending pending;
for (;;)
{
std::deque<Batch> local;
{
std::unique_lock<std::mutex> lock(st.q_mtx);
st.q_cv.wait(lock, [&st]() { return !st.queue.empty() || st.stopping; });
if (st.queue.empty() && st.stopping)
{
break; //停止且队列已空:退出(pending 已在上一轮提交)
}
local.swap(st.queue);
st.queued_bytes = 0;
st.queued_rows = 0;
st.queued_batches = 0;
}
for (std::size_t i = 0; i < local.size(); ++i)
{
AppendBatch(pending, local[i]);
}
local.clear();
//凑批:未达目标且未停止时,短暂等待更多数据,减少事务次数
while (!st.stopping && !st.failed.load() &&
PendingRows(pending) < static_cast<std::size_t>(RDP_RECORD_WRITE_BATCH_ROWS))
{
std::unique_lock<std::mutex> lock(st.q_mtx);
const bool more = st.q_cv.wait_for(
lock,
std::chrono::milliseconds(RDP_RECORD_WRITE_MAX_WAIT_MS),
[&st]() { return !st.queue.empty() || st.stopping; });
if (!more)
{
break; //超时:提交现有数据
}
if (st.queue.empty())
{
break; //停止且队列已空
}
local.swap(st.queue);
st.queued_bytes = 0;
st.queued_rows = 0;
st.queued_batches = 0;
lock.unlock();
for (std::size_t i = 0; i < local.size(); ++i)
{
AppendBatch(pending, local[i]);
}
local.clear();
}
FlushPending(st, pending);
if (st.failed.load())
{
DropQueued(st);
return;
}
}
FlushPending(st, pending);
}
void WriterLoop(Record_State& st)
{
#ifdef _WIN32
//记录写线程是后台任务:把线程 CPU 优先级降到 BELOW_NORMAL,避免与雷达处理线程争抢 CPU;
//I/O 保持正常优先级(否则写入积压会拉长 Shutdown 排空时间)。队列有界,写线程变慢时
//由丢弃策略兜底,不影响处理线程。
SetThreadPriority(GetCurrentThread(), THREAD_PRIORITY_BELOW_NORMAL);
#endif
try
{
WriterLoopBody(st);
}
catch (...)
{
//写线程内任何异常都不能抛出 DLL 边界;记录失败即降级
st.failed.store(true);
RDP_LOG << "[rdp_record] writer thread exception, recording stopped" << std::endl;
DropQueued(st);
}
}
//============================== 生命周期辅助 ==============================
Record_State::~Record_State()
{
{
std::lock_guard<std::mutex> lock(q_mtx);
stopping = true;
}
q_cv.notify_all();
if (writer.joinable())
{
writer.join();
}
std::lock_guard<std::mutex> lock(db_mtx);
try
{
db.disconnect();
}
catch (...)
{
}
}
}
namespace rdp_record
@@ -104,65 +663,143 @@ namespace rdp_record
std::unique_ptr<Record_State>& sp = State();
if (sp.get() != 0)
{
return sp->ready; //已经初始化过(含失败的幂等返回)
if (sp->ready)
{
return true; //已经初始化
}
if (sp->failed)
{
return false; //初始化失败(本次进程内不再重试)
}
sp.reset(); //上一次已 Shutdown:允许重新初始化(宿主销毁/重建工厂的场景)
}
sp.reset(new Record_State());
Record_State& st = *sp;
//创建记录目录(目录已存在时 _mkdir 返回 EEXIST,忽略;逻辑与 rdp_log.cpp 一致)
std::string dir = RDP_RECORD_DB_DIR;
while (dir.size() > 0 && (dir[dir.size() - 1] == '\\' || dir[dir.size() - 1] == '/'))
try
{
dir.erase(dir.size() - 1);
//创建记录目录(目录已存在时 _mkdir 返回 EEXIST,忽略;逻辑与 rdp_log.cpp 一致)
std::string dir = RDP_RECORD_DB_DIR;
while (dir.size() > 0 && (dir[dir.size() - 1] == '\\' || dir[dir.size() - 1] == '/'))
{
dir.erase(dir.size() - 1);
}
if (dir.size() > 0)
{
_mkdir(dir.c_str());
}
//按当前日期+时间生成记录库文件名:rdp_record_YYYYMMDD_HHMMSS.db
char stamp[32] = {0};
if (!TimeStamp(stamp, sizeof(stamp)))
{
RDP_LOG << "[rdp_record] cannot get local time, recording disabled" << std::endl;
st.failed = true;
return false;
}
st.path = RDP_RECORD_DB_DIR;
st.path += "rdp_record_";
st.path += stamp;
st.path += ".db";
//同一秒内重复 Init(Destroy→GetB 重连)或同目录多进程启动时,避免复用同一个库文件
if (_access(st.path.c_str(), 0) == 0)
{
const std::string base = st.path.substr(0, st.path.size() - 3); //去掉 ".db"
for (int suffix = 1; suffix < 1000; ++suffix)
{
const std::string candidate = base + "_" + std::to_string(suffix) + ".db";
if (_access(candidate.c_str(), 0) != 0)
{
st.path = candidate;
break;
}
}
}
if (!st.db.connect(st.path))
{
RDP_LOG << "[rdp_record] cannot open database: " << st.path << std::endl;
st.failed = true;
return false;
}
ApplyPragmas(st);
//启动记录写线程(处理线程之后只入队,不直接访问数据库)
st.writer = std::thread(WriterLoop, std::ref(st));
st.ready = true;
RDP_LOG << "[rdp_record] enabled, database: " << st.path
<< " (async writer, write_batch_rows=" << RDP_RECORD_WRITE_BATCH_ROWS
<< ", queue_cap_bytes=" << RDP_RECORD_QUEUE_CAP_BYTES << ")" << std::endl;
return true;
}
if (dir.size() > 0)
catch (const std::exception& e)
{
_mkdir(dir.c_str());
RDP_LOG << "[rdp_record] init exception: " << e.what() << std::endl;
}
catch (...)
{
RDP_LOG << "[rdp_record] init exception" << std::endl;
}
//按当前日期+时间生成记录库文件名:rdp_record_YYYYMMDD_HHMMSS.db
char stamp[32] = {0};
if (!TimeStamp(stamp, sizeof(stamp)))
st.failed = true;
st.ready = false;
try
{
RDP_LOG << "[rdp_record] cannot get local time, recording disabled" << std::endl;
st.failed = true;
return false;
st.db.disconnect();
}
st.path = RDP_RECORD_DB_DIR;
st.path += "rdp_record_";
st.path += stamp;
st.path += ".db";
if (!st.db.connect(st.path))
catch (...)
{
RDP_LOG << "[rdp_record] cannot open database: " << st.path << std::endl;
st.failed = true;
return false;
}
//调试用途的性能取舍:关闭同步刷盘(提交时不再 fsync),崩溃时可能丢失尾部数据;
//如需崩溃安全可改为 PRAGMA synchronous=NORMAL(或在 sqlite3 层自行开启 WAL)。
//注意:ormpp 的 execute() 只接受「不返回结果行」的语句,故不在此设置 journal_mode。
st.db.execute("PRAGMA synchronous=OFF;");
st.ready = true;
RDP_LOG << "[rdp_record] enabled, database: " << st.path << std::endl;
return true;
return false;
}
void Shutdown()
{
std::unique_ptr<Record_State>& sp = State();
if (sp.get() == 0 || !sp->ready)
if (sp.get() == 0)
{
return;
}
std::lock_guard<std::mutex> lock(sp->mtx);
RDP_LOG << "[rdp_record] shutdown, database: " << sp->path << std::endl;
sp->db.disconnect();
sp->ready = false; //path 保留,便于宿主在关库后仍能取到本次记录库路径
Record_State& st = *sp;
if (!st.ready)
{
sp.reset();
return;
}
//1) 停止接收新数据;2) 等待写线程排空队列/pending 并退出;3) 检查点/关库;4) 打印统计
{
std::lock_guard<std::mutex> lock(st.q_mtx);
st.stopping = true;
}
st.q_cv.notify_all();
if (st.writer.joinable())
{
st.writer.join();
}
LogStats(st, "shutdown");
try
{
std::lock_guard<std::mutex> lock(st.db_mtx);
//WAL 模式下把 -wal 合并回主库(尽力而为;失败不影响关库)
st.db.query_s<std::tuple<int, int, int>>("PRAGMA wal_checkpoint(TRUNCATE);");
st.db.disconnect();
}
catch (...)
{
}
RDP_LOG << "[rdp_record] shutdown, database: " << st.path << std::endl;
st.ready = false;
//保留状态对象(连接已断开、写线程已退出):下次 Init() 检测到 !ready && !failed 时复位并重新建库。
//不在 Shutdown 内直接释放,避免宿主在 Destroy 的同时仍有记录调用时出现悬空指针(防御性处理)。
}
std::string db_path()
@@ -171,28 +808,49 @@ namespace rdp_record
return (sp.get() == 0) ? std::string() : sp->path;
}
Stats stats()
{
Stats s;
std::unique_ptr<Record_State>& sp = State();
if (sp.get() == 0)
{
return s;
}
s.enqueued_rows = sp->counters.enqueued_rows.load();
s.dropped_rows = sp->counters.dropped_rows.load();
s.written_rows = sp->counters.written_rows.load();
s.transactions = sp->counters.transactions.load();
s.write_errors = sp->counters.write_errors.load();
s.queue_peak_bytes = sp->counters.queue_peak_bytes.load();
s.queue_peak_rows = sp->counters.queue_peak_rows.load();
return s;
}
//============================== 遗留同步接口(实体) ==============================
//说明:会阻塞调用线程,常规记录点请使用上面各表的异步入队接口;仅保留给测试/特殊场景。
template <typename T>
void Write(const T& v)
{
std::unique_ptr<Record_State>& sp = State();
if (sp.get() == 0 || !sp->ready || sp->failed)
if (sp.get() == 0 || !sp->ready || sp->failed.load())
{
return; //未初始化 / 初始化失败 / 已关库:静默忽略
}
std::lock_guard<std::mutex> lock(sp->mtx);
std::lock_guard<std::mutex> lock(sp->db_mtx);
if (!PrepareWrite<T>(*sp))
{
return;
}
//写入失败时每个实体只提示一次:不重试、不抛出、不影响雷达数据处理
static bool reported = false;
if (sp->db.insert(v) == INT_MIN && !reported)
if (sp->db.insert(v) == INT_MIN)
{
reported = true;
ReportError(*sp, "insert");
sp->counters.write_errors.fetch_add(1);
sp->tables[TableName<T>()] = 2;
ReportError(*sp, "insert", TableName<T>());
return;
}
sp->counters.written_rows.fetch_add(1);
sp->counters.transactions.fetch_add(1);
}
template <typename T>
@@ -204,24 +862,84 @@ namespace rdp_record
}
std::unique_ptr<Record_State>& sp = State();
if (sp.get() == 0 || !sp->ready || sp->failed)
if (sp.get() == 0 || !sp->ready || sp->failed.load())
{
return;
}
InsertEntities<T>(*sp, v);
}
std::lock_guard<std::mutex> lock(sp->mtx);
if (!PrepareWrite<T>(*sp))
//============================== 异步入队接口(源结构体) ==============================
void WritePoints(std::vector<PointRecv> v)
{
if (v.empty())
{
return;
}
Batch b;
b.kind = BatchKind::Point;
b.points = std::move(v);
EnqueueBatch(std::move(b));
}
//写入失败时每个实体只提示一次:不重试、不抛出、不影响雷达数据处理
static bool reported = false;
if (sp->db.insert(v) == INT_MIN && !reported)
void WriteCohesionTWS(std::vector<PointRecv> v)
{
if (v.empty())
{
reported = true;
ReportError(*sp, "insert");
return;
}
Batch b;
b.kind = BatchKind::CohesionTws;
b.points = std::move(v);
EnqueueBatch(std::move(b));
}
void WriteCohesionTAS(std::vector<PointRecv> v)
{
if (v.empty())
{
return;
}
Batch b;
b.kind = BatchKind::CohesionTas;
b.points = std::move(v);
EnqueueBatch(std::move(b));
}
void WriteTempTracks(std::vector<Temp_track> v)
{
if (v.empty())
{
return;
}
Batch b;
b.kind = BatchKind::TempTrack;
b.temp_tracks = std::move(v);
EnqueueBatch(std::move(b));
}
void WriteTrustTracks(std::vector<Trust_Track> v)
{
if (v.empty())
{
return;
}
Batch b;
b.kind = BatchKind::TrustTrack;
b.trust_tracks = std::move(v);
EnqueueBatch(std::move(b));
}
void WriteAssoInfos(std::vector<Asso_Info> v)
{
if (v.empty())
{
return;
}
Batch b;
b.kind = BatchKind::AssoInfo;
b.asso_infos = std::move(v);
EnqueueBatch(std::move(b));
}
}
+65 -18
View File
@@ -5,42 +5,88 @@
// 功能(完整说明见 docs/data_record_debug.md):
// 1) 把结构体数据按「一个记录实体一张表」的方式写入 SQLite 数据库,结构体到数据表的映射由
// 内置的 ormpp 完成(third_party/ormpp,见该目录 UPSTREAM.md);
// 2) 整个功能由构建开关控制:CMake 选项 RDP_DATA_RECORD(AUTO/ON/OFF)开启时由构建系统定义
// 2) 性能模型(V1.5.8 优化后):
// - 记录点只做「收集源结构体 → 一次入队」,在雷达处理线程上只有拷贝 + 队列 push 的开销;
// - 实体构造(X/P 等数组转文本)与 SQLite 写入由独立的写线程完成,批量事务提交;
// - 队列有界,写线程跟不上时整批丢弃并计数,**绝不阻塞雷达处理线程**;
// 3) 整个功能由构建开关控制:CMake 选项 RDP_DATA_RECORD(AUTO/ON/OFF)开启时由构建系统定义
// RDP_RECORD_ENABLE,rdp_record.cpp 与 sqlite3 amalgamation 才参与编译;关闭(交付/Release
// 构建默认)时本模块完全不编译,RDP_RECORD / RDP_RECORD_BATCH 展开为 ((void)0)、实参不求值,
// 不留任何代码与运行开销;
// 3) 生命周期:Data_Process_Factory::GetB() 调用 rdp_record::Init()(建目录、开库),
// Data_Process_Factory::Destroy() 调用 rdp_record::Shutdown()(关库);Init 之前、初始化
// 失败之后或 Shutdown 之后的写入一律静默忽略,不影响雷达数据处理;
// 4) 记录点用法(新增数据 = 加结构体 + 注册 + 一行记录点,配方见 docs/data_record_debug.md):
// 单条:RDP_RECORD(RdpRecPointTws{0, 值1, 值2, ...});
// 批量:RDP_RECORD_BATCH(recs); // recs 为 std::vector<某记录实体>
// 构建默认)时本模块完全不编译,所有记录宏展开为 ((void)0)、实参不求值,不留任何开销;
// 4) 生命周期:Data_Process_Factory::GetB() 调用 rdp_record::Init()(建目录、开库、启动写线程),
// Data_Process_Factory::Destroy() 调用 rdp_record::Shutdown()(停止入队、排空队列、关库);
// Init 之前、初始化失败之后或 Shutdown 之后的写入一律静默忽略,不影响雷达数据处理;
// 5) 记录点用法(新增数据 = 加记录实体 + 注册 + 一行批量记录点,配方见 docs/data_record_debug.md):
// 帧/圈内收集:std::vector<源结构体> recs;
// 记录点: RDP_RECORD_POINTS(recs); // 按表选择对应宏,内部 std::move
// 需要语句/局部变量的记录点请整体用 #ifdef RDP_RECORD_ENABLE ... #endif 包裹,
// 保证关闭记录功能时既无代码也无「变量未使用」告警;
// 5) 记录实体类型必须先在 rdp_record_entity.h 中声明并注册,否则链接期报 LNK2019
// 6) 记录实体类型必须先在 rdp_record_entity.h 中声明并注册,否则链接期报 LNK2019
// (extern template 已阻止 C++17 调用方隐式实例化)。
#include <cstdint>
#include <string>
#include <utility>
#include <vector>
//记录源结构体(定义在 struct.h;此处只做声明,避免业务头文件依赖 struct.h 的全部内容)
struct PointRecv;
struct Temp_track;
struct Trust_Track;
struct Asso_Info;
namespace rdp_record
{
bool Init(); //初始化记录库(由 Data_Process_Factory::GetB() 调用)
void Shutdown(); //关闭记录库(由 Data_Process_Factory::Destroy() 调用)
std::string db_path(); //当前记录库文件全路径(未启用或未成功打开时返回空串)
//记录统计(仅用于调试/验收;Shutdown 时自动打印,也可随时读取)
struct Stats
{
std::uint64_t enqueued_rows = 0; //已入队行数
std::uint64_t dropped_rows = 0; //队列满/停止后丢弃的行数
std::uint64_t written_rows = 0; //已成功写入数据库的行数
std::uint64_t transactions = 0; //提交事务数
std::uint64_t write_errors = 0; //写入失败次数
std::uint64_t queue_peak_bytes = 0; //队列峰值字节数
std::uint64_t queue_peak_rows = 0; //队列峰值行数
};
//写入单条记录实体(T 为记录实体类型,须在 rdp_record.cpp 中显式实例化)
bool Init(); //初始化记录库并启动写线程(由 Data_Process_Factory::GetB() 调用)
void Shutdown(); //停止写线程、排空队列并关闭记录库(由 Data_Process_Factory::Destroy() 调用)
std::string db_path(); //当前记录库文件全路径(未启用或未成功打开时返回空串)
Stats stats(); //当前统计快照(任何时刻可调用,线程安全)
//异步入队接口(处理线程只拷贝源结构体并送入有界队列;参数按值接收,调用点用 std::move)
//仅在记录功能开启(RDP_RECORD_ENABLE)时有实现;关闭构建由对应宏展开为空。
void WritePoints(std::vector<PointRecv> v); //RdpRecPoint(输入点迹,TWS/TAS 共用)
void WriteCohesionTWS(std::vector<PointRecv> v); //RdpRecPointCohesionTWS
void WriteCohesionTAS(std::vector<PointRecv> v); //RdpRecPointCohesionTAS
void WriteTempTracks(std::vector<Temp_track> v); //RdpRecTempTrack(临时航迹)
void WriteTrustTracks(std::vector<Trust_Track> v); //RdpRecTrustTrack(可靠航迹)
void WriteAssoInfos(std::vector<Asso_Info> v); //RdpRecAssoInfo(点航关联信息)
//遗留接口:同步写入记录实体(在调用线程内直接写库,会阻塞;仅供测试/特殊场景使用)。
//T 为记录实体类型,须在 rdp_record.cpp 中显式实例化。
template <typename T> void Write(const T& v);
//批量写入同一记录实体的多条记录(内部使用事务)
template <typename T> void Write(const std::vector<T>& v);
}
#ifdef RDP_RECORD_ENABLE
//记录宏:把指定记录实体写入数据库(首次写入该实体时自动建表)
//按记录表批量入队(参数为源结构体 vector,宏内部 std::move;调用后 vector 为空)
#define RDP_RECORD_POINTS(v) ::rdp_record::WritePoints(std::move(v))
#define RDP_RECORD_COHESION_TWS(v) ::rdp_record::WriteCohesionTWS(std::move(v))
#define RDP_RECORD_COHESION_TAS(v) ::rdp_record::WriteCohesionTAS(std::move(v))
#define RDP_RECORD_TEMP_TRACKS(v) ::rdp_record::WriteTempTracks(std::move(v))
#define RDP_RECORD_TRUST_TRACKS(v) ::rdp_record::WriteTrustTracks(std::move(v))
#define RDP_RECORD_ASSO_INFOS(v) ::rdp_record::WriteAssoInfos(std::move(v))
//遗留宏:记录实体单条/批量同步写(会阻塞调用线程;新记录点请使用上面的批量入队接口)
#define RDP_RECORD(...) ::rdp_record::Write(__VA_ARGS__)
//批量记录宏:把 std::vector<记录实体> 一次性写入(每批一个事务)
#define RDP_RECORD_BATCH(...) ::rdp_record::Write(__VA_ARGS__)
#else
//记录功能关闭:记录模块不参与编译,宏展开为空(实参不求值)
//记录功能关闭:记录模块不参与编译,所有宏展开为空(实参不求值)
#define RDP_RECORD_POINTS(v) ((void)0)
#define RDP_RECORD_COHESION_TWS(v) ((void)0)
#define RDP_RECORD_COHESION_TAS(v) ((void)0)
#define RDP_RECORD_TEMP_TRACKS(v) ((void)0)
#define RDP_RECORD_TRUST_TRACKS(v) ((void)0)
#define RDP_RECORD_ASSO_INFOS(v) ((void)0)
#define RDP_RECORD(...) ((void)0)
#define RDP_RECORD_BATCH(...) ((void)0)
namespace rdp_record
@@ -48,6 +94,7 @@ namespace rdp_record
inline bool Init() { return false; }
inline void Shutdown() {}
inline std::string db_path() { return std::string(); }
inline Stats stats() { return Stats(); }
template <typename T> inline void Write(const T&) {}
template <typename T> inline void Write(const std::vector<T>&) {}
}
+53 -5
View File
@@ -8,7 +8,6 @@
#include <Eigen/Dense>
#include "rdp_record.h"
#include "rdp_record_entity.h"
using namespace Eigen;
using namespace std;
@@ -68,6 +67,11 @@ int Track_Asso:: track_asso_process(std::vector<PointRecv> *point_
std::vector<PointRecv>().swap(point_process);
//输出航迹
#ifdef RDP_RECORD_ENABLE
//记录可靠航迹:先收集,输出循环结束后一次入队
std::vector<Trust_Track> rec_trust;
rec_trust.reserve(trust_track->size());
#endif
for (size_t i=0;i<trust_track->size();i++)
{ // 输出更新航迹条件:
if((*trust_track)[i].Track_Update_Flag==1){
@@ -160,10 +164,16 @@ int Track_Asso:: track_asso_process(std::vector<PointRecv> *point_
std::memcpy(Trust_Track_Output[*Trust_track_num_Output-1][0].speed_dim, (*trust_track)[i].speed_dim, sizeof(Trust_Track_Output[*Trust_track_num_Output-1][0].speed_dim));
std::memcpy(Trust_Track_Output[*Trust_track_num_Output-1][0].range_dim, (*trust_track)[i].range_dim, sizeof(Trust_Track_Output[*Trust_track_num_Output-1][0].range_dim));
RDP_RECORD(RdpRecTrustTrack((*trust_track)[i]));
#ifdef RDP_RECORD_ENABLE
rec_trust.push_back((*trust_track)[i]);
#endif
}
}
#ifdef RDP_RECORD_ENABLE
RDP_RECORD_TRUST_TRACKS(rec_trust);
#endif
return 0;
}
@@ -436,6 +446,19 @@ void Track_Asso:: model_filter(std::vector <Trust_Track> *trust_track,struct Ra
std::vector <Asso_Info> associated_info;
#ifdef RDP_RECORD_ENABLE
//记录未关联上的点航信息:收集到分块上限即入队,函数末尾再 flush 剩余
//按本圈候选数(点迹数×航迹数,封顶到分块上限)预留,避免收集期间多次扩容
std::vector<Asso_Info> rec_asso;
{
std::size_t est = point_process.size() * trust_track->size();
if (est > static_cast<std::size_t>(RDP_RECORD_ASSO_CHUNK_ROWS))
{
est = static_cast<std::size_t>(RDP_RECORD_ASSO_CHUNK_ROWS);
}
rec_asso.reserve(est);
}
#endif
//遍历所有点迹航迹 计算点迹和航迹的统计距离
for (size_t loop_of_track = 0; loop_of_track<trust_track->size();loop_of_track++)
@@ -551,7 +574,13 @@ void Track_Asso:: model_filter(std::vector <Trust_Track> *trust_track,struct Ra
{
associated_info.push_back(associated_info_tmp);
}else{
RDP_RECORD(RdpRecAssoInfo(associated_info_tmp));
#ifdef RDP_RECORD_ENABLE
rec_asso.push_back(associated_info_tmp);
if (rec_asso.size() >= RDP_RECORD_ASSO_CHUNK_ROWS)
{
RDP_RECORD_ASSO_INFOS(rec_asso);
}
#endif
}
}
}
@@ -707,7 +736,13 @@ void Track_Asso:: model_filter(std::vector <Trust_Track> *trust_track,struct Ra
{
if(associated_info[ii].track_index==track_index)
{
RDP_RECORD(RdpRecAssoInfo(associated_info[ii]));
#ifdef RDP_RECORD_ENABLE
rec_asso.push_back(associated_info[ii]);
if (rec_asso.size() >= RDP_RECORD_ASSO_CHUNK_ROWS)
{
RDP_RECORD_ASSO_INFOS(rec_asso);
}
#endif
associated_info.erase(associated_info.begin()+ii);
associated_num=static_cast<int>(associated_info.size());
}
@@ -719,7 +754,13 @@ void Track_Asso:: model_filter(std::vector <Trust_Track> *trust_track,struct Ra
{
if(associated_info[ii].point_index==point_index)
{
RDP_RECORD(RdpRecAssoInfo(associated_info[ii]));
#ifdef RDP_RECORD_ENABLE
rec_asso.push_back(associated_info[ii]);
if (rec_asso.size() >= RDP_RECORD_ASSO_CHUNK_ROWS)
{
RDP_RECORD_ASSO_INFOS(rec_asso);
}
#endif
associated_info.erase(associated_info.begin()+ii);
associated_num=static_cast<int>(associated_info.size());
}
@@ -813,6 +854,13 @@ void Track_Asso:: model_filter(std::vector <Trust_Track> *trust_track,struct Ra
}
}
#ifdef RDP_RECORD_ENABLE
//函数返回前把剩余未入队的关联信息 flush 掉(宏内部 std::move,调用后 rec_asso 为空)
if (!rec_asso.empty())
{
RDP_RECORD_ASSO_INFOS(rec_asso);
}
#endif
}
void Track_Asso:: model_output(std::vector <Trust_Track> *trust_track)
+38 -4
View File
@@ -8,7 +8,6 @@
#include <Eigen/Dense>
#include "rdp_record.h"
#include "rdp_record_entity.h"
using namespace Eigen;
using namespace std;
@@ -41,6 +40,11 @@ long long latest_timestamp //最新时间戳
std::vector<PointRecv>().swap(point_process);
//输出航迹
#ifdef RDP_RECORD_ENABLE
//记录可靠航迹:先收集,输出循环结束后一次入队
std::vector<Trust_Track> rec_trust;
rec_trust.reserve(trust_track->size());
#endif
for (size_t i=0;i<trust_track->size();i++)
{ // 输出更新航迹条件:
if((*trust_track)[i].Track_Index == tas_track_idx && (*trust_track)[i].Track_Update_Flag == 1)
@@ -119,11 +123,17 @@ long long latest_timestamp //最新时间戳
(*trust_track)[i].Elevation = Trust_Track_Output[*Trust_track_num_Output-1][0].Elevation;
(*trust_track)[i].Velocity = Trust_Track_Output[*Trust_track_num_Output-1][0].Range_V;
RDP_RECORD(RdpRecTrustTrack((*trust_track)[i]));
#ifdef RDP_RECORD_ENABLE
rec_trust.push_back((*trust_track)[i]);
#endif
}
}
#ifdef RDP_RECORD_ENABLE
RDP_RECORD_TRUST_TRACKS(rec_trust);
#endif
return 0;
}
@@ -252,6 +262,11 @@ void Track_Asso_Tas::model_filter( std::vector<Trust_Track> *t
)
{
std::vector <Asso_Info> associated_info;
#ifdef RDP_RECORD_ENABLE
//记录未关联上的点航信息:收集到分块上限即入队,函数末尾再 flush 剩余
std::vector<Asso_Info> rec_asso;
rec_asso.reserve(point_process.size());
#endif
//航迹信息
double X1[6] = {0},X2[6] = {0},X3[6] = {0},P1[6][6] = {{0}},P2[6][6] = {{0}},P3[6][6] = {{0}};
@@ -360,7 +375,13 @@ void Track_Asso_Tas::model_filter( std::vector<Trust_Track> *t
{
associated_info.push_back(associated_info_tmp);
}else{
RDP_RECORD(RdpRecAssoInfo(associated_info_tmp));
#ifdef RDP_RECORD_ENABLE
rec_asso.push_back(associated_info_tmp);
if (rec_asso.size() >= RDP_RECORD_ASSO_CHUNK_ROWS)
{
RDP_RECORD_ASSO_INFOS(rec_asso);
}
#endif
}
}
@@ -496,7 +517,13 @@ void Track_Asso_Tas::model_filter( std::vector<Trust_Track> *t
for(auto& info : associated_info)
{
RDP_RECORD(RdpRecAssoInfo(info));
#ifdef RDP_RECORD_ENABLE
rec_asso.push_back(info);
if (rec_asso.size() >= RDP_RECORD_ASSO_CHUNK_ROWS)
{
RDP_RECORD_ASSO_INFOS(rec_asso);
}
#endif
}
}
//未关联上,航迹外推
@@ -540,6 +567,13 @@ void Track_Asso_Tas::model_filter( std::vector<Trust_Track> *t
}
}
#ifdef RDP_RECORD_ENABLE
//函数返回前把剩余未入队的关联信息 flush 掉(宏内部 std::move,调用后 rec_asso 为空)
if (!rec_asso.empty())
{
RDP_RECORD_ASSO_INFOS(rec_asso);
}
#endif
}
void Track_Asso_Tas::model_output(std::vector<Trust_Track> *trust_track, int tas_track_idx )
+3 -2
View File
@@ -7,7 +7,6 @@
#include <iostream>
#include<iomanip>
#include "rdp_record.h"
#include "rdp_record_entity.h"
using namespace std;
int Track_Init::track_init_process_logic( std::vector <PointRecv> *point_recv, //输入点迹
@@ -109,11 +108,13 @@ int Track_Init::track_init_process_logic( std::vector <PointRecv>
#ifdef RDP_RECORD_ENABLE
//记录临时航迹
std::vector<Temp_track> rec_temp;
for(auto& vec_t : *temp_track){
for(auto& t : vec_t){
RDP_RECORD(RdpRecTempTrack(t));
rec_temp.push_back(t);
}
}
RDP_RECORD_TEMP_TRACKS(rec_temp);
#endif
//临时航迹满足起始长度 转为可靠航迹