🪨 8. 基础库

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

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

---

🧭 8.1 组件索引

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

任务	组件	头文件	小节
日志输出	Logger（VLOG_* 宏）	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.h 等	8.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）	Coroutine（vlink::Co）	base/coroutine.h	8.16

---

📝 8.2 Logger 日志系统

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

8.2.1 快速开始

#include <vlink/base/logger.h>
 
vlink::Logger::init("my_app", "/var/log/my_app.log");
vlink::Logger::set_console_level(vlink::Logger::kInfo);
vlink::Logger::set_file_level(vlink::Logger::kDebug);
 
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;

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()	调整大小 / 清空

#include <vlink/base/bytes.h>
 
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();

完整示例见 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 自动经过该池。

#include <vlink/base/memory_pool.h>
 
vlink::Bytes::init_memory_pool();       // 启动时调用一次
vlink::Bytes::release_memory_pool();    // 运行期可选回收完全空闲的块

• 池的容量分级由环境变量 VLINK_MEMORY_LEVEL（0..9，默认 3）选择；VLINK_MEMORY_PREALLOC=1 在启动时预分配。两者含义见 环境变量。
• 经 vlink::MemoryResource 让 std::pmr 容器复用同一池：

#include <memory_resource>
#include <vlink/base/memory_resource.h>
 
std::pmr::vector<int> v(&vlink::MemoryResource::global_instance());
v.reserve(1024);
 
auto sp = vlink::MemoryResource::make_shared<State>(/*x=*/42);

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 等不可拷贝对象

#include <vlink/base/functional.h>
 
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); };

边界条件：

• 空对象被调用时抛 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 死锁

#include <vlink/base/message_loop.h>
 
vlink::MessageLoop loop;
loop.async_run();
 
loop.post_task([] { do_work(); });
 
std::string data = "sensor_data";
loop.post_task([data]() { process(data); });
 
loop.quit();
loop.wait_for_quit();

完整示例见 examples/base/message_loop_basic/。

invoke_task 经 std::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 后，定时回调与普通任务共用同一线程，回调间无需同步：

#include <vlink/base/timer.h>
 
vlink::MessageLoop loop;
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"); });

8.6.3 串行化通信回调

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

vlink::MessageLoop my_loop;
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 为必填参数（无默认值），取值自 kLowestPriority 至 kHighestPriority，常规任务约定使用优先级常量 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"); });

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

8.6.6 线程安全约束

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

常见陷阱：

• **死锁**：参见 8.6.1 的 is_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，回调在循环线程上串行触发，无需同步。

#include <vlink/base/timer.h>
#include <vlink/base/message_loop.h>
 
vlink::MessageLoop loop;
 
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() 提供）。

vlink::ElapsedTimer t(vlink::ElapsedTimer::kMicro);
t.start();
do_work();
int64_t us = t.get();

• **DeadlineTimer**（base/deadline_timer.h）：保存绝对到期时间戳，has_expired() / remaining_time() 无锁，可多线程并发读。

vlink::DeadlineTimer dt(200);
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()	判断是否在工作线程，防死锁

#include <vlink/base/thread_pool.h>
 
vlink::ThreadPool pool(8);
 
pool.post_task([] { heavy_work(); });
 
auto fut = pool.invoke_task([]() -> int { return compute_answer(); });
int result = fut.get();
 
pool.shutdown();

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

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

vlink::MessageLoop ui_loop;
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 接口，但任务被转发到内部线程池并行执行，适合既需定时器又需多线程吞吐的场景（如传感器流水线）。

#include <vlink/base/multi_loop.h>
 
vlink::MultiLoop loop(4);
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);

• 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

#include <vlink/base/cancellation.h>
#include <vlink/base/message_loop.h>
#include <vlink/base/task_handle.h>
 
vlink::MessageLoop loop;
loop.async_run();
 
auto h = loop.post_task_handle([] { heavy(); });
h.wait();
assert(h.state() == vlink::TaskExecutionState::kCompleted);
 
vlink::CancellationSource parent;
vlink::PostTaskOptions opts;
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");
}

TaskExecutionState 转移：kQueued → kRunning → 终态之一（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	协作取消的规范化异常类型

#include <vlink/base/cancellation.h>
 
vlink::CancellationSource source;
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）：

vlink::CancellationSource group;
for (int i = 0; i < 8; ++i) {
    vlink::PostTaskOptions opts;
    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::OperationCancelled（what() 固定为 "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 捕获子进程输出

#include <vlink/base/process.h>
 
vlink::Process proc;
proc.set_process_mode(vlink::Process::kSeparateMode);
 
proc.start("/bin/echo", {"hello", "vlink"});
proc.wait_for_finished(3000);
 
std::string output;
proc.read_all_output(output);

8.10.3 异步监听并写入 stdin

vlink::Process proc;
proc.set_process_mode(vlink::Process::kSeparateMode);
 
proc.register_ready_read_stdout_callback([&proc] {
    std::string line;
    while (proc.can_read_line_stdout()) {
        proc.read_line_stdout(line);
        VLOG_I("stdout: ", line);
    }
});
proc.register_finished_callback([](int code, vlink::Process::ExitStatus status) {
    MLOG_I("process finished, exit code: {}", code);
});
 
proc.start("/bin/cat");
proc.wait_for_started(3000);
proc.write("hello from parent\n");
proc.close_write_channel();
proc.wait_for_finished(3000);

---

🔧 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*（裸指针）	否	手动控制归还时机

#include <vlink/base/object_pool.h>
 
auto pool = std::make_shared<vlink::ObjectPool<Buffer>>(
    [] { return std::make_unique<Buffer>(4096); },
    /*initial=*/4, /*max=*/16,
    [](Buffer& b) { b.clear(); },
    vlink::ObjectPool<Buffer>::kPolicyRelease);
 
{
    auto buf = pool->get();
    buf->data[0] = 0x42;
}
 
Buffer* raw = pool->borrow();
raw->data[0] = 0xFF;
pool->give_back(raw);

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

8.11.2 MpmcQueue 无锁队列

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

#include <vlink/base/mpmc_queue.h>
 
vlink::MpmcQueue<int> q(1024);
 
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();

容量须不小于 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 或耗时不确定时不应使用。

#include <vlink/base/spin_lock.h>
 
vlink::SpinLock lock;
 
{
    vlink::SpinLockGuard guard(lock);
    ++counter;
}
 
{
    std::lock_guard guard(lock);
}

指标	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）

#include <vlink/base/semaphore.h>
 
vlink::Semaphore sem(0);
 
bool ok = sem.acquire(1, 100);
sem.release(1);
 
sem.reset(true);

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

8.11.5 ConditionVariable 单调时钟条件变量

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

#include <mutex>
#include <vlink/base/condition_variable.h>
 
std::mutex mtx;
vlink::ConditionVariable cv;
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();

配合 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 配对示例

#include <vlink/base/sys_semaphore.h>
#include <vlink/base/sys_sharemem.h>
 
struct SharedMessage { uint32_t seq; uint32_t length; char payload[256]; };
 
vlink::SysSharemem shm;
shm.create("/vlink_demo_shm", sizeof(SharedMessage));
 
vlink::SysSemaphore sem(0);
sem.attach("/vlink_demo_sem");
 
auto* msg = static_cast<SharedMessage*>(shm.data());
msg->seq = 1;
sem.release();
 
vlink::SysSharemem shm2;
shm2.attach("/vlink_demo_shm", vlink::SysSharemem::kReadOnly);
vlink::SysSemaphore sem2;
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);
}

这是最底层的 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	执行超过此时间则触发超时

#include <vlink/base/schedule.h>
 
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()); });
 
loop.exec_task(vlink::Schedule::Config{},
               []() -> bool { return try_connect(); })
    .on_then([]() -> bool { start_session(); return true; })
    .on_else([] { retry_later(); });

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)。添加边前进行环路预检，成环则拒绝并记日志。

#include <vlink/base/graph_task.h>
#include <vlink/base/message_loop.h>
 
vlink::MessageLoop engine;
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();

执行必须从根节点（无前驱）发起。执行策略：kPolicyOnce（默认，每次 execute 最多执行一次）、kPolicyMultiple、kPolicyWaitAll（等待所有前驱完成）。

---

📊 8.14 CPU 利用率测量

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

#include <vlink/base/cpu_profiler.h>
#include <vlink/base/cpu_profiler_guard.h>
 
vlink::CpuProfiler profiler;
 
profiler.begin();
heavy_computation();
profiler.end();
 
double pct = profiler.get();
double reset_pct = profiler.restart();
 
void process_frame() {
    vlink::CpuProfilerGuard guard(&profiler);
    work();
}

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)	生成随机十六进制串

#include <vlink/base/uuid.h>
 
vlink::Uuid id = vlink::Uuid::generate_random();
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();

边界条件：from_string / is_valid 的 const 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;
}
 
vlink::MessageLoop loop;
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 中较常用的函数：

#include <vlink/base/utils.h>
 
std::string name = vlink::Utils::get_app_name();
int32_t pid      = vlink::Utils::get_pid();
vlink::Utils::set_thread_name("worker");
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);

---

🔖 相关文档

• 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/）