VLink  2.0.0
A high-performance communication middleware
🪨 8. 基础库

VLink 的 base 基础库是一套轻量、高性能、无第三方强制依赖的底层工具集,为通信核心与上层应用共同提供日志、调度、并发、计时、IPC 与唯一标识等能力。除 Coroutine 依赖 C++20 协程外,其余组件以 C++17 为基线,可脱离通信层独立使用,只需链接 vlink::vlink

本章按任务定位组件,给出能力边界、高频接口与最小可运行示例;接口完整签名与语义约束以对应头文件为准。


🧭 8.1 组件索引

按任务选取组件,再进入对应小节;低频与内部工具见 附录 A

任务 组件 头文件 小节
日志输出 LoggerVLOG_* 宏) base/logger.h 8.2
承载/传递原始字节 Bytes base/bytes.h 8.3
对象与 PMR 容器复用内存池 MemoryPool / MemoryResource base/memory_pool.h 8.4
类型擦除的可调用包装 Function / MoveFunction base/functional.h 8.5
单线程串行调度回调(免锁) MessageLoop base/message_loop.h 8.6
周期/单次/高精度计时 Timer 等四种 base/timer.h 8.7
并行计算 ThreadPool / MultiLoop base/thread_pool.h / base/multi_loop.h 8.8
可取消/等待/查状态的任务 TaskHandle / Cancellation base/task_handle.h / base/cancellation.h 8.9
子进程启动与管理 Process base/process.h 8.10
对象复用、无锁队列、互斥、限流 ObjectPool / MpmcQueue / SpinLock / Semaphore / ConditionVariable base/object_pool.h8.11
跨进程同步/共享内存 SysSemaphore / SysSharemem base/sys_semaphore.h / base/sys_sharemem.h 8.12
延迟/超时任务、DAG 编排 Schedule / GraphTask base/schedule.h / base/graph_task.h 8.13
CPU 利用率测量 CpuProfiler base/cpu_profiler.h 8.14
唯一标识/随机 token Uuid base/uuid.h 8.15
协程化异步流程(C++20) Coroutinevlink::Cobase/coroutine.h 8.16

📝 8.2 Logger 日志系统

vlink::Logger 是全局单例日志器:Logger::init() 初始化一次后,可在任意位置经宏写日志。输出同时面向控制台与文件两个 Sink,二者最低输出级别独立设置。

8.2.1 快速开始

vlink::Logger::init("my_app", "/var/log/my_app.log");
VLOG_I("node started, id=", 42);
MLOG_W("temperature is {} C, threshold exceeded", 78.5);
CLOG_E("errno=%d msg=%s", errno, strerror(errno));
SLOG_D << "values: " << 42 << " temp=" << 78.5;
Singleton logger with stream / format / printf / RAII-stream entry points.
#define CLOG_E(...)
Definition: logger.h:801
#define SLOG_D
Definition: logger.h:819
#define VLOG_I(...)
Definition: logger.h:785
#define MLOG_W(...)
Definition: logger.h:811

8.2.2 级别与写法

每条日志携带一个级别。set_console_level / set_file_level 分别控制对应 Sink 的最低输出级别,低于阈值的日志被丢弃。

级别 用途
kTrace 详细内部跟踪
kDebug 开发调试信息
kInfo 正常运行信息
kWarn 异常但可恢复
kError 影响运行的错误
kFatal 写日志后抛出 Exception::RuntimeError
kOff 关闭对应 Sink

四种写法对应同一套级别短宏(_T/_D/_I/_W/_E/_F),语义等价,按习惯任选其一。

写法 宏前缀 示例 特点
流式拼接 VLOG_* VLOG_I("x=", x) 拼接式,零堆分配
占位格式化 MLOG_* MLOG_I("x={}", x) {} 占位符
C 风格 CLOG_* CLOG_I("x=%d", x) printf 风格
RAII 流 SLOG_* SLOG_I << "x=" << x 析构时提交

kFatal 级别在写日志后抛出异常,e.what() 携带日志消息,可经 try/catch 捕获。编译期级别过滤、自定义日志后端、崩溃回溯环形缓冲等进阶能力见头文件。完整示例见 examples/base/logger_basic/


📦 8.3 Bytes 字节载体

vlink::Bytes 是 VLink 的核心数据载体:小数据直接内联在对象内部,避免堆分配;超出内联容量时才从内存池取块。它既可新建并拥有数据,也可零拷贝包装外部缓冲区。Bytes 在序列化系统中的角色见 消息序列化

8.3.1 所有权模型

构造方式决定所有权与复制语义。性能敏感路径优先 shallow_copy,仅在需要独立持有数据的副本时才用 deep_copy

构造方式 是否拥有数据 复制语义 用途
Bytes::create(n) 拥有副本 新建并填充缓冲区
Bytes::shallow_copy(p, n) 指针别名 零拷贝包装外部缓冲区
Bytes::deep_copy(p, n) 拥有副本 持有外部数据的独立副本

8.3.2 高频接口

方法 语义
Bytes::create(size, offset=0) 分配缓冲区,可预留前缀字节
Bytes::shallow_copy(data, size) 零拷贝包装外部缓冲区
Bytes::deep_copy(data, size) 拷贝外部数据并拥有
Bytes::from_string(str) 从字符串构造
data() / size() / empty() 访问指针、长度、是否为空
to_string() / to_string_view() 转字符串 / 视图
resize(n) / shrink_to(n) / clear() 调整大小 / 清空
auto buf = vlink::Bytes::create(64);
std::memcpy(buf.data(), payload, 64);
auto view = vlink::Bytes::shallow_copy(ext_ptr, ext_size);
auto owned = vlink::Bytes::deep_copy(ext_ptr, ext_size);
auto str_buf = vlink::Bytes::from_string("hello world");
for (uint8_t byte : buf) {
process(byte);
}
std::string s = buf.to_string();
Canonical 128-byte binary payload carrier with inline storage, multi-mode ownership and LZAV compress...

完整示例见 examples/base/bytes_basic/

8.3.3 内置工具

Bytes 静态方法提供常用数据处理,避免引入额外依赖:

  • **压缩**:Bytes::compress_data(data, size, high_ratio=false) / uncompress_data(...) / is_compress_data(...)
  • **Base64**:Bytes::encode_to_base64(buf) / decode_from_base64(str)
  • **CRC 校验**:Bytes::get_crc_32(buf)(与 ZIP/gzip/PNG 一致)、get_crc_64(buf)
  • **内存池**:Bytes::init_memory_pool() 在应用启动时调用一次,见 8.4。
auto compressed = vlink::Bytes::compress_data(raw.data(), raw.size());
if (vlink::Bytes::is_compress_data(compressed.data(), compressed.size())) {
auto original = vlink::Bytes::uncompress_data(compressed.data(), compressed.size());
}
std::string b64 = vlink::Bytes::encode_to_base64(buf);
uint32_t crc32 = vlink::Bytes::get_crc_32(buf);

🗄️ 8.4 内存池与 PMR 接入

vlink::MemoryPool 是 Bytes 的默认堆分配器,按尺寸分级复用空闲块以减少 new/delete 开销。常规用法是在应用启动时初始化一次,其余分配由 Bytes 自动经过该池。

vlink::Bytes::init_memory_pool(); // 启动时调用一次
vlink::Bytes::release_memory_pool(); // 运行期可选回收完全空闲的块
Size-class tiered memory pool with per-tier free lists and runtime statistics.
  • 池的容量分级由环境变量 VLINK_MEMORY_LEVEL0..9,默认 3)选择;VLINK_MEMORY_PREALLOC=1 在启动时预分配。两者含义见 环境变量
  • vlink::MemoryResourcestd::pmr 容器复用同一池:
#include <memory_resource>
std::pmr::vector<int> v(&vlink::MemoryResource::global_instance());
v.reserve(1024);
auto sp = vlink::MemoryResource::make_shared<State>(/*x=*/42);
PMR adapter that lets standard pmr-aware containers allocate through vlink::MemoryPool.

MemoryResource::make_shared 将对象与控制块在池内一次性分配,make_unique 仅在池内分配对象本体;两者均回退到全局池。需要隔离的私有池时,向构造函数传入 MemoryPool::Config 或 level(MemoryResource(int level, bool prealloc=false))创建独立实例。


🎯 8.5 Function 与 MoveFunction

base/functional.h 提供两个类型擦除的可调用包装器,作为热路径(消息回调、定时器、线程池任务)上 std::function 的替代:默认将一定容量内的闭包内联存储以避免堆分配,更大的闭包走内存池。承载更重闭包时使用大档别名 LargeFunction / LargeMoveFunction

类型 拷贝语义 适用场景
vlink::Function 可拷贝 需要被拷贝或共享的回调(默认选择)
vlink::MoveFunction move-only 闭包捕获 unique_ptr / packaged_task 等不可拷贝对象
vlink::Function<int(int, int)> add = [](int a, int b) { return a + b; };
int x = add(1, 2);
auto work = std::make_unique<HeavyWork>();
vlink::MoveFunction<void()> task = [w = std::move(work)]() { w->run(); };
vlink::LargeFunction<void()> jumbo = [arr = std::array<int, 50>{}]() { use(arr); };
Pool-backed type-erased callables: copyable vlink::Function and move-only vlink::MoveFunction.

边界条件:

  • 空对象被调用时抛 std::bad_function_call,与 std::function 一致。
  • std::function 可双向隐式互转;vlink::Function 可移动转换为 vlink::MoveFunction
  • MoveFunction::operator() 为非 const;需要 const 调用语义时使用 Function

🔁 8.6 MessageLoop 消息循环

vlink::MessageLoop 是 VLink 的核心任务调度器,也是 Timer、Schedule 的执行基础。其语义是单线程串行事件循环:投递到同一 loop 的所有任务在同一线程顺序执行,因此回调内访问该 loop 的私有状态无需加锁。post_task() 自身线程安全,可从任意线程调用。

8.6.1 核心接口

方法 语义
async_run() 在后台线程启动循环;run() 在当前线程阻塞运行
post_task(cb) 投递任务到 loop 线程(fire-and-forget)
invoke_task(fn, args...) -> future 投递并经 future 取返回值
quit(force=false) / wait_for_quit() 请求退出 / 等待退出完成
is_in_same_thread() 判断当前是否在 loop 线程,防止 invoke_task 死锁
loop.async_run();
loop.post_task([] { do_work(); });
std::string data = "sensor_data";
loop.post_task([data]() { process(data); });
loop.quit();
Single-threaded task dispatcher with three queue backends, timers and scheduling envelopes.

完整示例见 examples/base/message_loop_basic/

invoke_taskstd::future 取回结果。约束:不可在 loop 自身线程上对其返回的 future 调用 .get()——任务等待被执行,线程却等待任务完成,构成死锁。用 is_in_same_thread() 防护:

if (!loop.is_in_same_thread()) {
auto fut = loop.invoke_task([] { return get_state(); });
auto state = fut.get();
} else {
auto state = get_state();
}

8.6.2 与 Timer 集成

Timer 绑定到 MessageLoop 后,定时回调与普通任务共用同一线程,回调间无需同步:

loop.async_run();
vlink::Timer heartbeat(&loop, 1000, vlink::Timer::kInfinite, [&] {
VLOG_I("heartbeat tick");
});
heartbeat.start();
vlink::Timer::call_once(&loop, 500, [] { VLOG_I("delayed init"); });
MessageLoop-driven periodic or one-shot timer with priority support.

8.6.3 串行化通信回调

VLink 通信回调(Subscriber、Server 等)在传输层内部线程上触发。将消息投递回自有 MessageLoop 是串行化处理的标准模式;回调入参仅在回调内有效,外带前须先复制:

my_loop.async_run();
vlink::Subscriber<MyMsg> sub("dds://my/topic");
sub.listen([&](const MyMsg& msg) {
auto copy = msg;
my_loop.post_task([copy = std::move(copy)]() { process_message(copy); });
});

回调用法详见 通信模型

8.6.4 队列类型与入队策略

构造时可选队列类型;入队策略仅在有界队列已满时影响行为。

队列类型 特点
kNormalType 默认,FIFO 无优先级
kLockfreeType 无锁,适合多生产者多消费者、低竞争
kPriorityType 支持任务优先级(数值大者先执行)
入队策略(队列满时) 行为
kOptimizationStrategy 默认,重试至多 10 次(每次间隔 1ms)后丢弃一个可丢弃任务再入队
kPopStrategy 立即丢弃一个可丢弃任务并入队新任务
kBlockStrategy 持续重试直至有空闲槽位

优先级循环用 post_task_with_priority(cb, priority),priority 为必填参数(无默认值),取值自 kLowestPrioritykHighestPriority,常规任务约定使用优先级常量 kNormalPriority;其他队列类型忽略优先级。QoS 扩展中的 Qos::Additions::Priority 是独立于 MessageLoop 优先级的另一套配置,见 QoS 配置

8.6.5 链式调度 exec_task

exec_task()post_task() 基础上支持延迟、超时与链式延续回调,参数为 Schedule::Config(见 8.13):

loop.exec_task(vlink::Schedule::Config{/*delay_ms=*/100, 0, /*sched_to=*/0, /*exec_to=*/500},
[]() -> bool { return try_connect(); })
.on_then([]() -> bool { subscribe_all(); return true; })
.on_else([] { VLOG_W("connect failed, will retry"); })
.on_execution_timeout([] { VLOG_W("connect timed out"); });
#define VLOG_W(...)
Definition: logger.h:787

任务延迟到句柄提交时投递(临时句柄在表达式结束析构即提交),保证延续回调先于任务执行注册;存入具名变量的句柄须调用 dispatch() 立即投递。

8.6.6 线程安全约束

操作 线程安全 约束
post_task / quit 可从任意线程并发调用
invoke_task 不可在同一 loop 线程上对返回 future 调 .get()
run / async_run 仅由构造 loop 的线程调用一次
多 loop 线程访问同一状态 需额外同步(mutex / 原子量)

常见陷阱:

  • **死锁**:参见 8.6.1is_in_same_thread() 防护。
  • **递归投递**:可在任务内 post_task 新任务,但不可在任务内 wait_for_idle() 等待自身。
  • **长任务阻塞**:单线程下耗时任务会阻塞全部任务(含定时器);耗时操作应下发到 ThreadPool,结果再 post_task 回来(见 8.8)。

8.6.7 可追踪任务

post_task() 为 fire-and-forget。需要取消、等待或查询状态时使用 post_task_handle(),返回 TaskHandle,见 8.9


⏱️ 8.7 定时器

VLink 提供四种定时器,Timer 用于事件循环驱动的通用定时,其余三种针对特定场景。

8.7.1 Timer:事件循环定时器

vlink::Timer 绑定到 MessageLoop,回调在循环线程上串行触发,无需同步。

vlink::Timer timer(&loop, 500, vlink::Timer::kInfinite, [] { VLOG_I("tick"); });
timer.start();
vlink::Timer count_down(&loop, 1000, /*loop_count=*/3, [] { VLOG_I("count down"); });
count_down.start();
vlink::Timer::call_once(&loop, 200, [] { VLOG_I("delayed once"); });
loop.run();
方法 语义
start() / stop() / restart() 启动 / 停止 / 重置后重启
set_interval(ms) / set_loop_count(n) 修改间隔 / 触发次数,对活跃定时器立即生效
set_strict(true) 严格模式:错过的 tick 立即补发
is_active() / get_invoke_count() 是否运行 / 已触发次数
Timer::call_once(loop, ms, cb) 静态方法,单次触发

kInfinite(-1) 表示无限重复。interval_ms 为 0 时被钳至最小保护间隔,避免空转。完整示例见 examples/base/timer/

8.7.2 其余定时器

  • **WheelTimer**(base/wheel_timer.h):哈希时间轮,插入/删除均摊 O(1),适合数十万级并发超时(连接保活、会话超时)。回调在内部工作线程触发,通常投递回 MessageLoop 处理。
vlink::WheelTimer wheel(256, 10);
wheel.start();
auto key = wheel.add(1000, [](vlink::WheelTimer::Key k) { VLOG_I("timeout"); });
wheel.remove(key);
  • **ElapsedTimer**(base/elapsed_timer.h):高精度计时,实例支持单调时钟与 CPU 活跃时间,毫秒/微秒/纳秒精度(系统墙钟仅由静态 get_sys_timestamp() 提供)。
t.start();
do_work();
int64_t us = t.get();
  • **DeadlineTimer**(base/deadline_timer.h):保存绝对到期时间戳,has_expired() / remaining_time() 无锁,可多线程并发读。
while (!dt.has_expired()) { process_events(); }

⚙️ 8.8 ThreadPool 与 MultiLoop 并行

并行执行任务时使用 ThreadPool(纯计算并行)或 MultiLoop(兼需定时器与并行)。

8.8.1 ThreadPool

vlink::ThreadPool 维护固定数量工作线程并行执行任务,不含定时器,适合 CPU 密集型工作。

方法 语义
ThreadPool(n) 构造,指定线程数
post_task(cb) 投递任务
invoke_task(fn, args...) -> future 投递并取结果
shutdown() 关闭
is_in_work_thread() 判断是否在工作线程,防死锁
pool.post_task([] { heavy_work(); });
auto fut = pool.invoke_task([]() -> int { return compute_answer(); });
int result = fut.get();
pool.shutdown();
Fixed-size worker pool for parallel task execution.

MessageLoop 同理:不可在线程池工作线程内对 invoke_task 的 future 调 .get(),否则死锁;用 is_in_work_thread() 防护或改用 post_task()post_task_handle() 返回 TaskHandle,语义与 MessageLoop 一致(见 8.9)。

典型模式是 CPU 密集任务下发到 ThreadPool,结果投递回 MessageLoop 串行更新:

vlink::ThreadPool compute_pool(4);
ui_loop.async_run();
compute_pool.post_task([&] {
auto result = heavy_compute();
ui_loop.post_task([result]() { update_ui(result); });
});

8.8.2 MultiLoop

vlink::MultiLoop 继承 MessageLoop,保持相同的 post_task / invoke_task / exec_task 接口,但任务被转发到内部线程池并行执行,适合既需定时器又需多线程吞吐的场景(如传感器流水线)。

loop.async_run();
for (int i = 0; i < 100; ++i) {
loop.post_task([i]() { handle(i); });
}
loop.wait_for_idle(5000);
loop.quit();
loop.wait_for_quit(2000);
Multi-threaded variant of MessageLoop that forwards tasks to an internal ThreadPool.
  • N 个工作线程共享同一任务队列,任务执行顺序无保证。
  • 共享状态需调用方自行保护(SpinLock / mutex)。
  • ThreadPool,不在任务回调内对 invoke_task 的 future 同步等待,否则可能因线程池耗尽死锁。

8.8.3 三者对比

特性 MessageLoop ThreadPool MultiLoop
执行模型 单线程串行 固定线程池并行 事件循环 + 线程池并行
任务顺序 严格 FIFO 无保证 无保证
定时器 支持 不支持 支持(继承)
适用场景 有序任务派发 纯计算并行 兼需定时器与并行

🧵 8.9 可追踪任务与协作取消

post_task() 为 fire-and-forget。需要取消、等待或查询状态时使用可追踪变体 post_task_handle(),返回 TaskHandle;配合 CancellationSource 可成组取消。

8.9.1 TaskHandle

loop.async_run();
auto h = loop.post_task_handle([] { heavy(); });
h.wait();
opts.cancellation_token = parent.token();
auto h2 = loop.post_task_handle([token = opts.cancellation_token] {
while (!token.is_cancellation_requested()) { do_unit(); }
}, opts);
h2.cancel();
if (!h2.wait(/*timeout_ms=*/500)) {
VLOG_W("task did not finish in 500ms");
}
Cooperative one-shot cancellation primitives shared by VLink async building blocks.
Observable handle returned by tracked task-posting APIs of MessageLoop and ThreadPool.

TaskExecutionState 转移:kQueuedkRunning → 终态之一(kCompleted 正常返回 / kCancelled 被取消 / kDropped 溢出丢弃 / kRejected 拒收 / kFailed 抛异常)。

PostTaskOptions 字段:

字段 取值
overflow_policy kUseDispatcherStrategy(默认) / kReject / kBlock
drop_policy kDroppable(默认) / kProtected(永不被溢出丢弃)
cancellation_token 父级取消 token

边界条件:句柄析构不会取消任务,dispatcher 持续执行至终态。kLockfreeType 队列不追踪 kProtected,要保护任务不被溢出丢弃须使用 kNormalType / kPriorityType

8.9.2 协作取消

协作取消基于写端/观察者模型:写端持 CancellationSource 调用 request_cancel();工作任务持由同一 source 派生的 CancellationToken(轻量、可拷贝、可跨线程),经轮询或回调响应取消。

类型 角色
CancellationSource 写端,发出取消请求(副本共享同一状态)
CancellationToken 只读观察者,可轮询、可注册回调
CancellationRegistration RAII 槽,持有一个已注册回调
Exception::OperationCancelled 协作取消的规范化异常类型
auto token = source.token();
auto reg = token.register_callback([] { VLOG_I("cancellation observed"); });
std::thread worker([token]() {
while (!token.is_cancellation_requested()) {
token.throw_if_cancellation_requested();
do_unit_of_work();
}
});
source.request_cancel();
worker.join();

成组取消(一个 source 派生多个 token):

for (int i = 0; i < 8; ++i) {
opts.cancellation_token = group.token();
pool.post_task_handle([t = opts.cancellation_token, i] {
while (!t.is_cancellation_requested()) { work(i); }
}, opts);
}
if (some_global_failure) {
group.request_cancel();
}

语义约束:

  • request_cancel() 一次性触发:首次返回 true 并触发全部回调,后续返回 false
  • 注册回调时 token 已取消,则回调在 register_callback 内同步执行。
  • 回调抛出的异常被捕获并记日志,不向外传播。
  • Exception::OperationCancelledwhat() 固定为 "vlink operation cancelled")专表协作取消已被观察,通用错误应使用 Exception::RuntimeError

🚀 8.10 Process 子进程管理

vlink::Process 是跨平台子进程管理类,接口风格参照 Qt QProcess,用于启动、监控、通信与终止外部程序,支持 stdout/stderr 管道捕获、stdin 写入与异步回调,覆盖 Linux、macOS、Windows、QNX。

8.10.1 核心接口

方法 语义
start(program, args) 异步启动子进程
start_command(cmdline) 用完整命令行字符串启动
wait_for_started/finished/ready_read(ms) 阻塞等待(默认 3000ms,kInfinite 为无限)
read_all_output(str) / read_all_error(str) 读取全部 stdout / stderr
read_line_stdout(line) / can_read_line_stdout() 按行读取 stdout
write(str) / close_write_channel() 写入 stdin / 关闭写通道
terminate() / kill() / close(force) 优雅终止 / 强制结束 / 关闭
get_state() / is_running() / get_exit_code() 状态查询
register_finished_callback(cb)注册异步回调(建议在 start() 前完成以避免竞态)
Process::execute(prog, args, ms) 静态:同步执行并返回退出码
Process::start_detached(prog, args) 静态:启动完全分离的子进程

I/O 通道模式(set_process_mode)共五种:默认 kSeparateMode(stdout/stderr 各自缓冲)、kMergedMode(stderr 并入 stdout 管道)、kForwardedMode(stdout/stderr 均继承父进程,不捕获)、kForwardedOutputMode(stdout 继承父进程、stderr 捕获)、kForwardedErrorMode(stdout 捕获、stderr 继承父进程)。回调在内部监控线程触发,访问共享数据须注意线程安全。Process 不可拷贝、不可移动。

8.10.2 捕获子进程输出

proc.start("/bin/echo", {"hello", "vlink"});
proc.wait_for_finished(3000);
std::string output;
proc.read_all_output(output);
Portable child-process driver with asynchronous I/O notifications.

8.10.3 异步监听并写入 stdin

std::string line;
while (proc.can_read_line_stdout()) {
proc.read_line_stdout(line);
VLOG_I("stdout: ", line);
}
});
MLOG_I("process finished, exit code: {}", code);
});
proc.start("/bin/cat");
proc.wait_for_started(3000);
proc.write("hello from parent\n");
proc.wait_for_finished(3000);
#define MLOG_I(...)
Definition: logger.h:809

🔧 8.11 并发工具

需要在多线程间互斥、传输数据、复用对象或计数通知时,从下表选取。

组件 场景
ObjectPool 复用对象、减少热路径堆分配
MpmcQueue 固定容量、无锁、高吞吐的线程间数据传输
SpinLock 极短临界区(纳秒级)互斥
Semaphore 进程内计数信号量、限流
ConditionVariable 条件等待(单调时钟,规避时钟跳变)

8.11.1 ObjectPool 对象池

ObjectPool<T> 线程安全地回收并重用对象,减少堆分配。必须经 std::make_shared 创建,因内部归还逻辑依赖指向池本身的 weak_ptr

获取方式 返回 自动归还 场景
get() unique_ptr<T, PoolDeleter> 单一所有权(默认)
get_shared() shared_ptr<T> 共享所有权
borrow() T*(裸指针) 手动控制归还时机
auto pool = std::make_shared<vlink::ObjectPool<Buffer>>(
[] { return std::make_unique<Buffer>(4096); },
/*initial=*/4, /*max=*/16,
[](Buffer& b) { b.clear(); },
{
auto buf = pool->get();
buf->data[0] = 0x42;
}
Buffer* raw = pool->borrow();
raw->data[0] = 0xFF;
pool->give_back(raw);
Thread-safe generic object pool with RAII handles and a tunable reset policy.

边界条件:池耗尽(全部对象借出且达到 max_size)时 get()std::runtime_error,应设置合理上限并准备降级路径,经 pool->stats() 监控。Policy 控制获取/归还时是否调用重置回调(kPolicyNone / kPolicyRelease(默认) / kPolicyAcquire / kPolicyBoth)。

8.11.2 MpmcQueue 无锁队列

MpmcQueue<T> 是固定容量、无锁、缓存行对齐的多生产者多消费者环形队列。

q.push(42);
int val;
q.pop(val);
bool ok = q.try_push(42);
ok = q.try_pop(val);
size_t cap = q.capacity();
size_t sz = q.size();
bool e = q.empty();
q.notify_to_quit();
Bounded lock-free multi-producer multi-consumer ring buffer with optional cv blocking.

容量须不小于 1(否则构造抛 std::invalid_argument),经验值取预期突发峰值的 2 至 4 倍。默认的 push / pop 以自旋方式阻塞;需要条件变量唤醒式的阻塞收发时,按 kConditionBehavior 行为调用(push<vlink::MpmcQueue<int>::kConditionBehavior>(...) 配合 wait_not_empty() / wait_not_full()),否则 cv 通知是纯开销。

8.11.3 SpinLock 自旋锁

SpinLock 适用于极短临界区(数条指令,如更新关联变量或计数器),此时 std::mutex 的上下文切换开销大于等待本身。临界区含 I/O 或耗时不确定时不应使用。

{
vlink::SpinLockGuard guard(lock);
++counter;
}
{
std::lock_guard guard(lock);
}
Cache-line aligned adaptive spin lock with exponential back-off.
指标 SpinLock std::mutex
等待方式 用户态自旋 内核态阻塞
适合时长 极短(纳秒至微秒) 任意
上下文切换
递归 死锁 可用 recursive_mutex

8.11.4 Semaphore 信号量

进程内计数信号量,采用 P/V(acquire/release)语义,内部使用单调时钟以规避系统时钟跳变。

方法 语义
Semaphore(n) 构造,初始计数 n
acquire(n, ms) 获取 n 个许可,最多等待 ms 毫秒(kInfinite 无限)
release(n) 释放 n 个许可
reset(true) 恢复初始计数并中断所有等待者(acquire 返回 false)
bool ok = sem.acquire(1, 100);
sem.release(1);
sem.reset(true);
Counting semaphore confined to a single process.

release() 内部获取 mutex,非 async-signal-safe,不可在信号处理函数中调用。跨进程同步使用 SysSemaphore(见 8.12)。

8.11.5 ConditionVariable 单调时钟条件变量

vlink::ConditionVariablestd::condition_variable 的就地替换,内部锁定 CLOCK_MONOTONIC,避免系统时钟被调整时 wait_for/wait_until 提前唤醒或永久等待。接口与标准库一致(wait / wait_for / wait_until / notify_one / notify_all),非 POSIX 平台直接别名到标准库版本。

#include <mutex>
std::mutex mtx;
bool ready = false;
{
std::unique_lock lock(mtx);
cv.wait_for(lock, std::chrono::milliseconds(200), [&] { return ready; });
}
{
std::lock_guard lock(mtx);
ready = true;
}
cv.notify_one();
Monotonic-clock condition variable replacement immune to system clock jumps.

配合 SpinLock 等非 std::mutex 锁时使用 ConditionVariableAny

8.11.6 选型决策

需要任务调度?
+-- 串行执行 ---------> MessageLoop
+-- 并行 + 定时器 ----> MultiLoop
+-- 纯并行计算 -------> ThreadPool
需要线程间数据传输?
+-- 固定容量 + 高吞吐 ----> MpmcQueue
+-- 对象重用 -------------> ObjectPool
需要互斥保护?
+-- 极短临界区 -----------> SpinLock
+-- 一般临界区 -----------> std::mutex
+-- 条件等待 -------------> vlink::ConditionVariable
需要信号通知 / 限流?
+-------------------------> Semaphore

自定义并发数据结构时,将不同线程频繁写入的原子变量按缓存行对齐分布,规避伪共享。


🔗 8.12 跨进程 IPC

base 层提供两个直接封装操作系统 IPC 的类,用于跨进程同步与数据共享,不依赖任何第三方库。

头文件 功能
SysSemaphore base/sys_semaphore.h 命名计数信号量
SysSharemem base/sys_sharemem.h 命名共享内存区域

8.12.1 SysSemaphore

命名的跨进程计数信号量,多个进程以相同名称访问同一内核对象。

class SysSemaphore final {
public:
static constexpr int kInfinite{-1};
explicit SysSemaphore(size_t count = 0);
bool attach(const std::string& name);
bool detach(bool force = true);
bool acquire(size_t n = 1, int timeout_ms = kInfinite);
void release(size_t n = 1);
bool is_attached() const;
size_t get_count() const;
};

POSIX 上名称须以 / 开头(如 /vlink_ready)。timeout_ms=0 为非阻塞尝试。析构默认 detach(false),仅关闭句柄而不删除内核对象。

8.12.2 SysSharemem

命名共享内存:create() 分配并映射新区域,attach() 仅映射已存在的区域,实现零拷贝数据共享。

class SysSharemem final {
public:
enum Mode : uint8_t { kReadOnly = 0, kReadWrite = 1 };
bool create(const std::string& name, size_t size, Mode mode = kReadWrite);
bool attach(const std::string& name, Mode mode = kReadWrite);
bool detach(bool force = true);
void* data();
const void* data() const;
size_t size() const;
};

kReadOnly 映射上写入未定义(data() 在只读模式返回 nullptr)。共享内存本身不带同步语义,须配合 SysSemaphore 使用。受支持的 POSIX/Windows 后端会对 create() 新建区域零初始化。

8.12.3 配对示例

struct SharedMessage { uint32_t seq; uint32_t length; char payload[256]; };
shm.create("/vlink_demo_shm", sizeof(SharedMessage));
sem.attach("/vlink_demo_sem");
auto* msg = static_cast<SharedMessage*>(shm.data());
msg->seq = 1;
sem.release();
shm2.attach("/vlink_demo_shm", vlink::SysSharemem::kReadOnly);
sem2.attach("/vlink_demo_sem");
if (sem2.acquire(1, 5000)) {
const vlink::SysSharemem& ro = shm2; // 只读模式须经 const data() 取指针
const auto* m = static_cast<const SharedMessage*>(ro.data());
MLOG_I("received seq={}", m->seq);
}
Cross-process named semaphore backed by the host kernel.
Named cross-process shared-memory region.

这是最底层的 OS IPC 原语。需要完整的发布/订阅与零拷贝传输时,直接使用 VLink 的 shm:// 传输后端(见 传输后端与 URL),其内部即建立在此类原语之上。POSIX 命名对象在进程崩溃后不会自动删除,启动时应检查并清理残留。


🗂️ 8.13 任务调度

8.13.1 Schedule 调度包装

vlink::Schedule 不单独构造,而经 MessageLoop::exec_task() 使用:将回调包入 Config,支持延迟、优先级、调度超时与执行超时,并返回 RAII 句柄链式注册延续回调。任务在句柄提交时才投递(临时句柄在表达式结束析构即提交,或显式调用 dispatch()),从而保证所有延续回调先于任务执行注册完成;存入具名变量的句柄须调用 dispatch() 立即投递。

Config 字段 含义
delay_ms 任务发布前的延迟
priority 调度优先级(kPriorityType 循环)
schedule_timeout_ms 未在此时间内启动则触发超时
execution_timeout_ms 执行超过此时间则触发超时
loop.exec_task(vlink::Schedule::Config{/*delay_ms=*/100, 0, 0, /*exec_to=*/500},
[] { expensive_op(); })
.on_execution_timeout([] { VLOG_W("task took too long"); })
.on_catch([](std::exception& e) { VLOG_E("exception: ", e.what()); });
[]() -> bool { return try_connect(); })
.on_then([]() -> bool { start_session(); return true; })
.on_else([] { retry_later(); });
#define VLOG_E(...)
Definition: logger.h:789
Fluent task-scheduling wrapper used by MessageLoop::exec_task() and family.

8.13.2 GraphTask 有向无环图调度

vlink::GraphTask 实现 DAG 任务调度:每个节点经 precede() / succeed() 声明依赖,再 execute() 提交到任意兼容引擎(MessageLoop / MultiLoop / ThreadPool)。

工厂方法 回调签名 用途
create(name, cb) void() 普通工作任务
create_condition(name, cb, n) int() 条件分支(返回值选择分支)

依赖声明:A -- > B 等价 A->precede(B)(A 先执行、B 后执行);A -- < B 等价 A->succeed(B)。添加边前进行环路预检,成环则拒绝并记日志。

engine.async_run();
auto load = vlink::GraphTask::create("load", [] { load_data(); });
auto proc = vlink::GraphTask::create("proc", [] { process(); });
auto save = vlink::GraphTask::create("save", [] { save_data(); });
auto clean = vlink::GraphTask::create("clean", [] { cleanup(); });
load -- > proc -- > save;
proc -- > clean;
load->execute(&engine);
std::string dot = load->export_to_dot();
Directed acyclic task graph with condition branching, cycle guard and DOT export.

执行必须从根节点(无前驱)发起。执行策略:kPolicyOnce(默认,每次 execute 最多执行一次)、kPolicyMultiplekPolicyWaitAll(等待所有前驱完成)。


📊 8.14 CPU 利用率测量

vlink::CpuProfiler 测量一段代码的 CPU 活跃时间占总墙钟时间的百分比,用于量化节点与操作的 CPU 利用率。

profiler.begin();
heavy_computation();
profiler.end();
double pct = profiler.get();
double reset_pct = profiler.restart();
void process_frame() {
vlink::CpuProfilerGuard guard(&profiler);
work();
}
Per-instance CPU utilisation sampler with an env-driven global enable gate.
Stack-based RAII bracket that pairs every CpuProfiler::begin with a guaranteed end.

get() 返回百分比(0.0 表示尚无数据;多核下若 begin/end 区间累计活跃时间超过墙钟时间,可能超过 100),restart() 取值后清零。CpuProfilerGuard 在构造时 begin、析构时 end,异常安全。

与通信节点集成:所有节点内部持有可选的 CpuProfiler,开启全局开关后自动采集,经 Node::get_cpu_usage() 取值,-1.0 表示未启用。

vlink::Subscriber<SensorMsg> sub("dds://sensors/lidar");
double cpu = sub.get_cpu_usage();

全局开关由环境变量 VLINK_PROFILER_ENABLE=1 控制(首次读取后缓存,见 环境变量)。节点集成细节见 通信模型。热路径可先判 CpuProfiler::is_global_enabled() 再决定是否构造 Guard。


🆔 8.15 Uuid 唯一标识

vlink::Uuid 是 RFC 4122 128 位唯一标识符值类型:可平凡复制、负载内联,提供 v4 随机生成及项目级随机字节/十六进制工具。

方法 语义
Uuid::generate_random() 生成 v4 随机 UUID
id.to_string() / to_compact_string() 转规范 / 紧凑字符串
Uuid::from_string(s) 解析(接受规范/紧凑/带花括号形式),返回 optional
Uuid::is_valid(s) 校验字符串
Uuid::random_bytes(n) 生成 n 字节随机数据
Uuid::random_hex(n) 生成随机十六进制串
std::string s = id.to_string();
auto parsed = vlink::Uuid::from_string("47ac10b8-58cc-4a3c-8c5b-0e778899aabb");
if (parsed) { use(*parsed); }
std::string token = vlink::Uuid::random_hex();
Value-typed RFC 4122 UUID and project random-bytes primitive.

边界条件:from_string / is_validconst char* 重载为 null-safe(传 nullptr 返回失败)。可注入 std::mt19937 引擎获得确定性结果,也可作为 std::unordered_set 的键。底层 std::mt19937 不是密码学安全随机源,random_hex / random_bytes 适用于短期会话标识、关联 ID 与 proxy auth-token,不适用于长期密钥——后者应使用 OpenSSL RAND_bytes 等 CSPRNG。完整签名见头文件 base/uuid.h


🧬 8.16 Coroutine 协程

vlink::Coroutine(别名 vlink::Co)基于 C++20 stackless 协程,将所有挂起与恢复绕回 MessageLoop,因此协程体语句(除 await_future 等待瞬间外)均在 loop 线程上运行,共享状态无需加锁。头文件 <vlink/base/coroutine.h>;构建需 ENABLE_CXX_STD_20=ON 且工具链同时声明 __cpp_impl_coroutine__cpp_lib_coroutine(GCC 10+ / Clang 14+ / MSVC 19.x+),满足后框架自动启用协程支持。

8.16.1 任务定义与启动

vlink::Co::Task<int> compute(vlink::MessageLoop& loop) {
co_await vlink::Co::yield(loop);
co_await vlink::Co::delay_ms(loop, 100);
co_return 42;
}
loop.async_run();
auto t = compute(loop);
vlink::Co::co_spawn(loop, std::move(t), [](int v) { VLOG_I("done v=", v); });
  • co_await awaiter 为唯一挂起点;co_return value 设置返回值并结束。
  • Task<T>(亦写作 Task<> 表示 Task<void>)是协程返回句柄,可 co_await、可移动、不可拷贝。
  • 协程参数会被拷贝进协程帧,捕获状态务必经函数参数传入;切勿把带捕获的协程 lambda 作为临时表达式直接传给 co_spawn(如 co_spawn(loop, []()->Task<>{...}())),lambda 在整表达式结束即析构,捕获引用会悬空。

8.16.2 Awaiter 与编排

Awaiter 语义
vlink::Co::schedule(loop) 切到指定 loop 线程继续执行
vlink::Co::yield(loop) 协作让出(等价同 loop 的 schedule)
vlink::Co::delay_ms(loop, ms) 非阻塞睡眠 ms 毫秒
vlink::Co::await_future(loop, fut) 等待 std::future<T>,不在 loop 线程阻塞 .get()
vlink::Co::await_graph(loop, graph) 等待 GraphTask DAG 全部完成
vlink::Co::Task<void> orchestrate(vlink::MessageLoop& loop) {
co_await vlink::Co::when_all(loop, make_tasks());
size_t winner = co_await vlink::Co::when_any(loop, make_tasks());
co_await vlink::Co::sequence(loop, make_tasks());
}

协程内异常沿 co_await 链向外传播,co_spawn 在顶层捕获并记日志。MessageLoop 析构时挂起的协程进入失败分支而非崩溃。


📎 附录 A 低频/内部工具

下列组件多为内部或低频使用,需要时查阅头文件,本章不展开。

组件 头文件 功能
Format base/format.h 轻量 {} 占位符格式化器(MLOG_* 内部使用)
FastStream base/fast_stream.h 高性能输出流(Logger 内部引擎)
Utils base/utils.h 进程/线程/网络/信号等跨平台工具函数
Helpers base/helpers.h 字符串/数字/哈希/转义等无状态工具
Quantize base/quantize.h 线性量化/反量化(紧凑容器复用)
Uint128 base/uint128.h 可移植 128 位无符号整数
CachedTimestamp base/cached_timestamp.h 低开销线程安全格式化时间戳(Logger 内部使用)
Plugin base/plugin.h 类型安全的动态插件加载器
Exception base/exception.h VLink 异常类型
LoggerPluginInterface base/logger_plugin_interface.h 自定义日志后端纯虚接口

Utils 中较常用的函数:

std::string name = vlink::Utils::get_app_name();
int32_t pid = vlink::Utils::get_pid();
std::string val = vlink::Utils::get_env("MY_VAR", "default");
vlink::Utils::register_terminate_signal([](int sig) { app.shutdown(); });
auto ipv4_list = vlink::Utils::get_all_ipv4_address(/*filter_available=*/true);
Portable host-system utility surface used across the VLink runtime.

🔖 相关文档

  • 03-serialization —— Bytes 在序列化系统中的角色
  • 02-communication —— 通信回调中如何使用 MessageLoop、节点与 CpuProfiler 的集成
  • 04-transport —— shm:// 等传输后端(建立在 IPC 原语之上)
  • 05-qos —— QoS 优先级与 MessageLoop 优先级的关系
  • 13-integration —— VLINK_MEMORY_LEVEL / VLINK_PROFILER_ENABLE 等环境变量
  • 01-started —— 配套示例(含 examples/base/