lmflow 设计方案

状态:成品。Rust 引擎、C ABI、C++ 糖层、可选 OpenCV adapter、18 个内置算子、 Python 绑定(pybind11)、原生 SDK 发布(各平台头文件+库)、三端文档站全部就位; 326 个测试(Rust 282 + soak 2 + doctest 3 + Python 39)全绿,TSan 硬门禁 0 竞态。Rust / C++ / Python 三种宿主的 hello_world 都输出正确;支持线程池绑核 + 实时优先级(Linux/Android),可交叉编到 Android / iOS / 鸿蒙。 定位:一个数据流图计算框架 —— 把计算描述成有向图,节点是算子(Kernel), 边上流动带时间戳的数据包(Packet)。 组成:Rust 引擎 + C ABI 门面 + C++/Python 算子。 核心概念:Graph(图)、Node(节点)、Kernel(算子)、Edge/Port(边/端口)、 Packet(数据包)、Contract(端口类型契约)、Poller/Observer(输出的拉/推)。

0. 目标与范围

0.1 做什么

• 引擎(调度、线程、边队列、拓扑、YAML 解析)用 Rust 实现。
• 对外只暴露一层 C ABI(include/lmflow/flow.h)。这是唯一的稳定接口。
• 算子可用 C++(经 flow.hpp 糖层)或 Python(经 pybind11)编写,YAML 中平等引用。
• 图的拓扑与参数由 YAML 描述,含每节点的 options。

0.2 阶段划分

批处理(batch 输入策略)已支持(见 §7.10):攒够 capacity 个包一次交给算子,process() 用 input_count / input_at 读整批,关流刷余批。用于批推理 / 窗口聚合。多输入口按时间戳对齐(与 sync 同源,见 §7.10)。内置 BatchSumKernel 为样板。

零输入口 source 节点(生成型算子)已支持(见 §7.4):内核自产数据、source_done() 自报产完；无数据时可 source_yield(delay) 协作式让出 worker。源不能挂委托执行器(否则会独占宿主线程)。内置 RangeSourceKernel 为样板。

子图(subgraph)+ 跨文件 include 已支持(见 §7.11):纯建图期变换,把带 subgraphs / node.type 的配置展平成等价扁平图,运行时引擎 / 调度器不感知子图。

反馈环(back-edge)已支持(见 §7.12):把输入口标为 back_edges 即「最新值反馈寄存器」—— 容量 1、不参与就绪 / 终止 / 对齐;去掉 back-edge 后的图仍是 DAG。未被 back-edge 打断的拓扑环仍在 init 报错。

0.3 非目标

• 不做分布式 / 跨进程;单进程内多线程。
• 不做 GPU 内存空间(LMFlowBuffer.device 已预留字段,但本版本只有 CPU)。
• ~~不支持图跑完后重跑~~ → 已支持 reset(§7.13):保留算子实例复位重跑。

1. 决策记录(ADR)

已锁定的决策及其理由 —— 避免重复讨论。

2. 分层架构

┌──────────────────────────────────────────────────────────────────┐
│  宿主(驱动图):Rust / C++ / Python                                │
│    new → init_from_yaml → add_poller → start                      │
│    loop { input.send(pkt); poller.next() } → close → wait_done    │
├─────────────────────── C ABI (include/lmflow/flow.h) ────────────────────┤ ← unsafe 边界①
│  Rust 引擎 (lmflow crate, lib + staticlib + cdylib)               │
│    graph(索引 arena) · scheduler/executor · edge(FIFO)            │
│    packet · timestamp · config(YAML) · registry · poller          │
├─────────────────────── C ABI (回调) ──────────────────────────────┤ ← unsafe 边界②
│  算子:C++(flow.hpp 糖层)   |   Python(pybind11 蹦床)            │
│        process 内调 lmflow_ctx_* 读输入 / 发输出                     │
└──────────────────────────────────────────────────────────────────┘

• 边界①:宿主 → 引擎(Rust 提供 extern "C" 函数)。
• 边界②:引擎 → 算子(算子侧提供 vtable),算子回读 → 引擎(lmflow_ctx_*)。
• 两个方向都在 C ABI 上,没有 C++ 模板过界。

3. 数据模型

3.1 Packet 与所有权

pub enum Payload {
    Native(Box<dyn Any + Send + Sync>),  // Rust 侧构造
    Builtin(BuiltinPayload),              // 引擎分配:BYTES / BUFFER / 标量 / STR
    Foreign(ForeignPayload),              // 外部构造:裸指针 + drop_fn
}

pub struct ForeignPayload {
    ptr: *mut c_void,
    drop_fn: Option<unsafe extern "C" fn(*mut c_void)>,
    type_id: u64,
}
impl Drop for ForeignPayload {
    fn drop(&mut self) { /* 调一次 drop_fn */ }
}
// 断言:同一 payload 任一时刻只被单线程访问;drop 可能发生在别的线程。
// 前提由「节点独占令牌」保证(§7.0 规则 R3)。这是本设计唯一的 Send/Sync unsafe 断言。
unsafe impl Send for ForeignPayload {}
unsafe impl Sync for ForeignPayload {}

#[derive(Clone)]
pub struct Packet {
    data: Option<Arc<Payload>>,   // None = 空包(仅时间戳,用于 bound / 关流)
    ts:   Timestamp,
}

跨界表示 LMFlowPacket{payload, type_id, timestamp, owner, drop_fn}(40 字节), 三种所有权语义由 owner 区分:

多路分发(Forward/扇出)= Arc::clone,不拷贝数据。

3.2 数据类型模型:两条路

引擎对 payload 完全不作解释,只搬引用、只按 type_id 做相等性校验。因此任意类型都能流动。

1. 任意自定义类型(推荐给纯 C++ / 纯 Rust 管线) 调用方自备指针与 drop_fn,type_id 自取(C++ 糖层默认 typeid 哈希)。 cv::Mat、自定义结构体、模型张量对象一视同仁,引擎零参与。
2. 内建类型(为跨语言而设) BYTES / I64 / F64 / BOOL / STR / BUFFER。由引擎分配、复制、释放, 免去在别的语言里构造 drop_fn;type_id 为约定常量,跨语言稳定。 对引擎没有特权,只是一套双方都认识的内存约定。

type_id 的现实限制:typeid(T).hash_code() 跨编译器/跨动态库/-fno-rtti 不保证一致。 算子与宿主分属不同编译产物时应改用稳定方案(对类型名做 FNV-1a)。糖层已把生成逻辑 收敛到单一函数 lmflow::TypeId<T>(),改一处即可。

3.3 LMFlowBuffer:一个描述符统管所有大块数值数据

typedef struct {
  void*   data;                    /* 首字节 */
  int64_t shape[8];                /* 各维元素数 */
  int64_t strides[8];              /* 各维**字节**步长 */
  int32_t ndim, dtype;
  uint32_t flags; int32_t device;  /* READONLY 标记 / 内存空间 */
  int64_t reserved[2];             /* ABI 预留,置零 */
} LMFlowBuffer;                      /* 168 字节,布局由 static_assert 钉死 */

语义与 numpy buffer protocol 一致(strides 以字节计,不要求连续),所以 numpy 与 cv::Mat 都能零拷贝包住同一块内存。

3.4 引用与写时复制(省拷贝)

payload 默认不可变共享。需要就地改写时用 CoW —— 语义等同 Arc::make_mut。

关键点(极易写错):算子拿到的输入包是借用,引擎的 Context 自己还持一份引用。 若直接 clone + make_mutable,引用数 ≥ 2,CoW 必然复制,省拷贝的意图落空。 必须先把包从输入槽取走:

lmflow::Packet p = cc.TakeInput(0);          // ← 关键:移出输入槽
LMFlowBuffer buf{};
if (LMFlowStatus st = p.MakeMutableBuffer(&buf)) return st;   // 独占 → 零拷贝
// ...原地写 buf.data...
cc.Emit(0, std::move(p));

• 线性管线(本节点是唯一消费者)→ 全程零拷贝。
• 上游是 Split 扇出、数据被别的分支共享 → 才复制一份,因此不会污染对方。 这同时消除了「共享 payload 被就地改写」的数据竞争。

线程池上的线性管线也保持零拷贝。 算子回调返回后，引擎立即释放本次 invocation 的输入槽；派发到下游后，先清除 producer 的 staging 引用，再唤醒下游 worker。这样下游 take_input + make_mutable 在唯一消费者的线性管线中稳定看到独占 payload。扇出仍然 保留共享引用并按 CoW 复制，保证兄弟分支互不污染。

CoW 生效的不变量(必须写进代码注释):

引擎在投递数据包之后,不得再保留任何额外引用。 否则引用数恒 ≥ 2,CoW 永远退化成全量拷贝 —— 而且不报错,只是变慢。 (例如「为推进时间戳边界而缓存最后一个包」这类实现会静默破坏它。)

限制:CoW 只支持引擎持有的内建 payload(BUFFER/BYTES/标量)。自定义 payload 引擎只有 drop_fn、无从复制,返回 LMFLOW_ERR_INVALID_ARG(该情形自行拷贝)。 要通用化需给 LMFlowPacket 增加 clone_fn 字段 —— 会改 ABI 布局,本版本不做。

3.5 Timestamp

UNSET < UNSTARTED < PRE_STREAM < MIN … MAX < POST_STREAM < ONE_OVER_POST_STREAM < DONE
                                 └ 普通数据区间 ┘

PRE_STREAM/POST_STREAM 是流首/流尾的单包位置;DONE 表示端口已关且不再有数据; UNSET 为默认值。算术用饱和运算,哨兵附近不会溢出回绕。 已实现于 core/src/timestamp.rs(含 9 个单测)。

提交时 timestamp == UNSET 的包,引擎自动继承当前 input_timestamp(与 Forward 一致); 在图输入口上提交 UNSET 视为非法。

3.6 Side packet:常量输入

整个 run 期间不变的任意对象,由宿主在启动前注入,算子按名字读取。

没有 side packet 就无法把「一个已经初始化好的模型」交给算子 —— options 做不到, 这是引入它的直接原因。

LMFlowStatus lmflow_graph_set_side_packet(LMFlowGraph*, const char* name, LMFlowPacket pkt);  /* start 之前 */
LMFlowPacket lmflow_ctx_side_packet(const LMFlowContext*, const char* name);                /* 借用,勿 drop */
bool       lmflow_ctx_has_side_packet(const LMFlowContext*, const char* name);

• 必须在 start 之前设置,之后返回 LMFLOW_ERR_STATE;传入即移交所有权,引擎持有到 graph 释放。
• 算子侧读到的是借用,不得 drop。

4. 对外 C 接口

权威定义见 include/lmflow/flow.h(手写,已通过 gcc -std=c11 -Wall -Wextra 与 g++ -std=c++17 双向验证)。本节只列分组,避免文档与 header 双份漂移。

接口设计要点

• lmflow_ctx_forward 专为直通存在:算子不碰引用计数,引擎内部共享同一 Arc。
• 输入口句柄化(LMFlowInput*),热路径免去按名字 hash+strcmp。
• 阻塞接口一律有非阻塞/超时兄弟:try_send / try_next / next_timeout / wait_done_timeout。 生产代码建议一律用带超时版本 —— 算子逻辑有误会让图静止而非结束,无超时等待就是永久挂起。
• lmflow_last_error() 是线程局部的:算子在工作线程失败时,其文本不会出现在宿主线程。 要拿那条信息用 lmflow_graph_last_error(graph)。
• lmflow_graph_dump 返回线程局部缓冲,多线程同时调用不会互相踩踏。
• lmflow_graph_to_dot_view(g, view) 导出 Graphviz DOT(dot -Tsvg 可渲染),支持 TOPOLOGY / COMPACT / DIAGNOSTICS 三档。返回值同 dump(线程局部,不得 free)。 统计视图可在图运行期间随时调用(读原子快照)。
• 日志回调:引擎保证调用时不持有任何内部锁,故回调里可安全抢 GIL / 加锁。

5. 算子

5.1 C++ 算子(flow.hpp 糖层)

糖层 header-only、不属于 ABI、零运行开销,100% 建立在 flow.h 上。 核心是 KernelAdapter<T> 把虚函数桥成 C ABI vtable,并在蹦床里 try/catch 挡住 C++ 异常穿越 FFI(异常穿 extern "C" 是 UB)。

class PassThroughKernel : public lmflow::Kernel {
 public:
  static void GetContract(lmflow::Contract& c) { c.InputSetAny(0); c.OutputSetAny(0); }
  lmflow::Status Process(lmflow::Context& cc) override {
    cc.Forward(0, 0);                     // 零拷贝直通
    return lmflow::Status::Ok();
  }
};
LMFLOW_REGISTER_KERNEL(PassThroughKernel)    // 或 LMFLOW_REGISTER_KERNEL_AS(T, "别名")

• 静态 GetContract 可选,有则经 SFINAE 自动接线。
• Packet::Get<T>() 带 type_id 校验;TryGet<T>() 类型不符返回 nullptr(绝不 UB)。
• Context 禁拷贝/移动,防止算子把只在回调期有效的句柄留存。
• 写算子时的断言宏:LMFLOW_RET_CHECK(cc, cond) / LMFLOW_RET_CHECK_MSG(cc, cond, msg) —— 不成立就带着失败的表达式原文 + file:line 返回算子失败。 为什么需要它:C ABI 跨界只能过一个 int32_t(ADR #1),错误文本必须另经 lmflow_ctx_set_error 存进 Context。也就是说「返回失败」与「说明原因」在本框架里是 两件事,直接 return Status::Error() 会让引擎只拿到一个码、原因为空。这两个宏把它们 绑成一个动作,让人难以漏掉文本。内置 CastKernel 是用法样板。
• 异常路径也带原因:KernelAdapter 的 catch 会把 std::exception::what() 写进 Context (非 std 异常则给出固定说明)。此前异常这条路只剩一个错误码、诊断信息为零。
• 注册:所有 C++ 算子统一使用 LMFLOW_REGISTER_KERNEL 自注册。静态库中的注册对象 可能被链接器裁剪，因此 lmflow::lmflow / lmflow::kernels 在各平台自动使用 whole-archive / force-load；自定义静态算子库也必须采用等价链接方式。
• 算子与引擎解耦:注册表(src/kernel.rs)是唯一一张语言无关的 name → vtable 表, C++(flow.hpp)、Python(pybind11)、Rust(trait Kernel)三条路都汇入同一个 kernel::register,引擎不知道算子是什么语言写的。内置的 18 个 C++ 算子因此只是捆绑的 算子库、不是引擎的一部分 —— 它们放在 crate 之外(lmflow/cpp/,见 §11),不随发布的 crate 分发；由 CMake 构建为独立 lmflow::kernels 组件。
• 引擎不隐式安装任何算子。Rust 宿主显式调用 register_kernel；原生 SDK 与 Python 通过链接可选的 lmflow::kernels 获得 C++ 算子。注册来源只有“宿主代码”或“链接组件”， 不再存在建图时修改全局注册表的隐藏路径。
• 内置算子清单见 cpp/kernels/，每个翻译单元各自注册。其中张量前处理组(纯数值 BUFFER): Cast(dtype 转换)、Affine(x*scale+shift)、Clamp、Reduce(→F64 标量)—— 统一走 double 做 dtype 分派,要求连续缓冲。含 F16:buffer_util.hpp 自带 binary16 ↔ double 的软件转换(见 §5.3),故 dtype: f16 可直接作输入或输出。示例见 examples/python/preprocess/preprocess.py(u8 图 → f32 → 归一化 → clamp)。

5.3 F16(binary16)转换

张量前处理组支持 LMFLOW_DTYPE_F16,转换由 cpp/kernels/buffer_util.hpp 里自己实现的 half_to_float / f64_to_half 承担。刻意不用 _Float16,也不用 F16C / NEON 内建:

• _Float16 不是所有目标编译器都有(MSVC 就没有可移植的 half 类型);
• 内建指令要按架构分派 + 运行期探测;
• 张量前处理不在最内层推理热路径上,这点转换成本换来「任意编译器 / 架构上行为逐位一致」 是值得的 —— 而且正因为不依赖编译器,舍入行为才能被测试钉死。

舍入是就近取整、平局取偶(IEEE 默认),double → half 直接从 double 位模式做而 不经 float 中转 —— 否则会双重舍入(两次各取整一次,极少数入参偏一个 ulp)。 上溢到 ±inf、下溢到非规格数 / ±0(保号)、inf/NaN 均按 IEEE 处理。

验证:cpp/tests/buffer_util_test.cc —— 期望值全是硬编码的 IEEE 位模式(不与任何编译器 内建类型对照,故在 MSVC / arm64 上同样有效),含全部 65536 个位模式的 half → double → half 往返、上溢/下溢临界点(65519 → 最大有限值 vs 65520 → inf)、以及相邻 half 的精确中点 (平局取偶最容易写错的地方)。CI 在 -O0 与 -O2 两档都跑(浮点位操作对优化敏感)。 开发时另用 GCC 的 _Float16 作 oracle 穷举对照过一次(65536 个 half→float + 46 万个 double→half,含全部相邻 half 中点),逐位一致 —— 但该对照不进仓库,因为它不可移植。

5.2 Python 算子(pybind11,已实现)

@lmflow.kernel("PyDouble")
class PyDouble(lmflow.Kernel):
    @staticmethod
    def get_contract(c):
        c.input_set_any(0); c.output_set_any(0)
    def open(self, cc):    self.factor = cc.option_int("factor", 2)
    def process(self, cc): cc.emit(0, cc.input(0).as_int() * self.factor)

与 C++ 算子在 YAML 里平等引用 —— 引擎不知道算子是什么语言写的。 细节见 §8 专章。约束:只收发内建类型;回调期持 GIL;异常转错误码。

5.2b Rust 算子(trait Kernel,已实现)

Rust 也能一等公民地写算子:实现 lmflow::Kernel(get_contract/open/process/close)+ 运行期 register_kernel::<T>("Name"),YAML 用 kernel: Name 引用。安全的 KernelCtx/ KernelContract 包装内部 Context/Contract(不裸调 C ABI);panic / Err 都被 catch_unwind 兜成图错误、不穿越边界。这是C ABI vtable 之上的糖(和 flow.hpp 的 KernelAdapter<T> 对 C++ 做的一样),不是引擎绕过 vtable 的原生快路(见 ADR)。见 core/src/kernel_api.rs。

外部 Rust 工程:cargo add lmflow → use lmflow::{Graph, Packet, Timestamp, Kernel, register_kernel}。 引擎就是这一个 crate lmflow(包名 = 库名 = lmflow → liblmflow.a,C ABI/CMake/Python 共用同一库)。

5.3 内置算子清单(cpp/kernels/,一文件一算子)

既是可用算子,也是 API 覆盖用例。

类型约定:捆绑算子一律用内建类型(LMFLOW_TYPE_I64 等)声明契约, 而不是 InputSet<int>(C++ 的 typeid)。原因是后者 Python/Go 侧无从产生同样的 标识 —— 若用它,Python 送来的整数会被类型校验拒绝。这条是「一套算子三语言可用」 的前提,flow.hpp 为此提供了 InputSetBuiltin。

6. 图与拓扑

6.1 Rust 侧结构:索引 arena

核心决定:不还原自引用裸指针对象图,所有实体存在 Graph 的 Vec 里, 相互引用用 usize id。这既契合数据流本身的整数 id 语义,也避开借用检查器的自引用难题。

pub type NodeId = usize;
pub type EdgeId = usize;

pub struct Edge {
    name: String,
    producer: Option<NodeId>,           // None = 图输入口喂入
    consumers: Vec<(NodeId, usize)>,    // (下游节点, 它的第几个输入口)
    queue: Mutex<VecDeque<Packet>>,
    bounded: bool,                      // 仅图输入口为 true(§7.5)
    closed: AtomicBool,
}

pub struct Node {
    name: String,
    kernel: KernelInstance,             // C++/Python 算子黑盒
    inputs:  Vec<EdgeId>,
    outputs: Vec<EdgeId>,
    ctx: Mutex<Context>,
    sched: Mutex<NodeSched>,
}

pub struct Graph {
    nodes: Vec<Node>, edges: Vec<Edge>,
    edge_by_name: HashMap<String, EdgeId>,
    graph_inputs: HashMap<String, EdgeId>,
    graph_outputs: HashMap<String, EdgeId>,
    executors: HashMap<String, Executor>,
    pollers: Vec<Arc<Poller>>,
    state: Mutex<GraphState>,
}

用 HashMap<name, EdgeId> 直接拿 id,避免「名字排序名次」与「存储下标」混用这类错误。

6.2 init_from_yaml 的校验清单

任一不通过即返回错误,并可由 lmflow_last_error() 取原因:

1. 端口名引用不到上游生产者;
2. 同一端口名有多个生产者;
3. 图输入口与某节点的输出口同名;
4. 拓扑成环 —— 未被 back-edge 打断的环无法终止,故直接拒绝;把反馈输入口标为 back_edges 即可合法成环(§7.12);
5. 节点的 executor 名未在 executors: 中定义;
6. 零输入口节点 —— B 阶段的就绪规则对空集恒为真,会被无限调度成自旋(§7.4);
7. max_in_flight > 1 却落在只有一个线程的执行器上(单线程池、或 0 线程的委托执行器, 并行度恒为 1)→ 报错,不静默降级。默认池按核数开,故默认节点配它通常合法。

第 7 条尤其重要:宁可报错也不静默忽略,否则用户以为开了并行、实际没有。

6.3 图的生命周期状态机

      new()            init_from_yaml()          start()
 ─────────────► Created ──────────────► Initialized ──────────► Running
                                             │                     │
                    add_poller / observe ────┘        close_all_inputs()
                    input()(取句柄)                  或 cancel()
                                                                   ▼
                                                                Draining
                                                                   │ 全部节点关闭
                                                                   ▼
                                                               Terminated ──► free()

• 其它组合返回 LMFLOW_ERR_STATE(如 start 两次、未 start 就 send)。
• add_poller / observe 必须在 start 之前,否则可能丢失已产出的包。
• 重跑已支持(§7.13):Terminated 后可 reset 复位重跑(保留算子实例);或 free 释放。

6.4 端口的命名与定位

有两套标识符,分工明确:

声明语法(input_ports / output_ports 的元素):

input_ports: ["frames",                    # 无 tag(归入空 tag "")
              "VIDEO:cam0",                # 有 tag,index 自动
              "MASK:0:m0", "MASK:1:m1"]    # 有 tag,index 显式

三种定位方式:

cc.InputId("VIDEO")       // → 1    按 tag(推荐:语义稳定)
cc.InputId("MASK", 1)     // → 3
cc.InputIndex("frames")   // → 0    按边名(通用/路由类算子偶尔需要)
cc.Input(0)               //        直接序号,最省事但依赖声明顺序

规则:

• tag 约定大写字母 / 数字 / 下划线,不含 :;空 tag 表示「无标签」。
• 同一算子、同一 tag 下 index 必须从 0 连续、不得重复(否则 init 报错)。
• 扁平序号 = YAML 声明顺序(第 0 个声明即序号 0)。

最后一条是有意与常见做法分道(ADR #17)。按 tag 字典序分组的方案下,混用 有标签与无标签端口时 Index(0) 拿到的不一定是你写的第一个 —— 这是个只在运行时 才暴露的陷阱。改成声明顺序后它就不存在了。

7. 执行模型(B 阶段详细规格)

7.0 三条硬规则(整套并发设计的地基)

R1 —— 调用算子期间不持任何引擎锁。 回调前必须放掉所有锁,否则算子内部回调 lmflow_ctx_*(或算子自身阻塞)会死锁。 这条决定了 §7.3 的 staging 设计。

R2 —— 锁序恒为 node.sched → {edge.queue, blocked_flush_nodes},禁止反向。 推包路径必须「锁边 → push → 解锁边 → 再唤醒下游节点」。全局无环,故无死锁。

blocked_flush_nodes 是「哪些节点有 blocked_flush」的索引,必须与 sched.blocked_flush 在同一把 node.sched 锁内一起更新。否则会出现「索引已空、节点仍有 blocked_flush」 这类不一致快照,而 wait_done / is_idle 正是读这个索引来判定死锁的 —— 不一致就会 把正常排空误报成卡死。

只读该索引的地方(is_idle、wait_done、wait_until_idle、resume_blocked_flushes) 一律「取快照后立即释放,再去锁 sched」。注意这依赖临时 MutexGuard 在语句末释放: let v: Vec<_> = self.blocked_flush_nodes.lock()...collect(); 出了这条语句锁就没了。 同一条语言规则曾经反过来咬过一次(见 §附录的 pump_step 自锁死),所以别把 .lock() 的结果绑成活得更久的局部变量,那会把只读路径变成反向锁序。

R3 —— running == true 是节点的独占令牌。 一旦某线程把节点置为 running,它即独占该节点的 Context 与「从输入边弹包」的权利, 之后无需再持 node.sched 锁。这是 §3.1 中 Send/Sync 断言的依据。

7.1 M2:同步 push(无线程,先验证 FFI 闭合)

• send → push 到图输入边 → 驱动其消费者。
• drive(node):就绪则每个输入口弹一个包装进 Context → 调算子 → 把 staging 分发到下游边 → 递归驱动。
• 为防深图/环形爆栈,用显式工作栈(Vec<NodeId>)而非真递归。
• hello_world 全程跑在调用线程上,零并发即正确。
• ⚠ M2 下 executors: 配置不生效(没有线程池),文档须写明,别让人以为配了就有用。

7.2 M3:节点调度状态机(合并唤醒,防丢唤醒)

struct NodeSched {
    opened: bool, closed: bool,
    running: bool,   // 已有任务在跑(独占令牌,R3)
    rescan: bool,    // 运行期间又来了包 —— 跑完必须重扫,否则丢唤醒
}

fn try_claim(node) -> bool {
    let mut s = nodes[node].sched.lock();            // R2:先 node
    if !s.opened || s.closed { return false; }
    if s.running { s.rescan = true; return false; }  // 合并唤醒,不重复入队
    if !inputs_ready(node) { return false; }         // 内部按需短暂锁 edge
    s.running = true; true                            // 拿到独占令牌
}

fn run(node) {                        // 由 R3,以下全程不持 node 锁
    let ctx = pop_one_from_each_input(node);
    let st  = kernel.process(ctx);     // ★ 零锁调用算子(R1)
    if st != OK { record_error(st); discard_staging(node); return finish(node); }
    dispatch_staging_to_downstream(node);   // 锁边 push → 解锁 → 唤醒下游(R2)
    finish(node);
}

fn finish(node) {
    let again = { let mut s = lock(); s.running = false;
                  let a = s.rescan; s.rescan = false; a };
    if again || inputs_ready(node) { schedule_if_claimed(node); }
    maybe_close(node);                 // §7.6
}

为什么必须有 rescan:若上游在本节点 running 期间 push 了包,try_claim 会因 running 返回 false;若不记下这次唤醒,包就永远躺在队列里没人处理 —— 经典丢唤醒。 (这也是调度态需要 Scheduling / SchedulingPending 两态区分的原因。)

7.3 emit 走 staging —— 让 R1 成立

算子在 process 里调 emit/forward 不直接写下游边(那要在回调期持边锁,违反 R1), 而是写进本节点 Context 内的暂存区;process 返回后引擎统一分发。

struct Context {
    inputs:  Vec<Option<Packet>>,   // None = 已被 take_input 取走
    staging: Vec<Vec<Packet>>,      // 每个输出口一个暂存队列
    input_ts: Timestamp,
}

• emit(i, pkt) → staging[i].push(接管 pkt)。
• forward(in, out) → staging[out].push(inputs[in].clone())(仅 Arc::clone)。
• take_input(i) → inputs[i].take(),所有权移交算子(CoW 的前提,§3.4)。
• 分发时对每条输出边按消费者数 Arc::clone。

⚠ 裸 C ABI 用户注意:take_input 后若既不 emit 也不 drop 就早退,会泄漏。 C++/Python 侧有 RAII/GC 兜住,裸 C 没有。

7.4 就绪判定 + 零输入 source 节点

B 规则:每个输入口都有 ≥1 包 → 取各口队首组成一次 Process(多输入按时间戳对齐,见 §7.2)。

零输入节点 = source(生成型算子),由内核自产数据。它套不上上面的规则(空集「恒真」既会自旋、又会「空真即关」),故单独处理:

• 就绪:未自报完成即「可产出」;try_claim 给它一个单调递增的时间戳(认领序号),auto-emit 继承 → 下游单调。并发度受 max_in_flight 个槽约束;finish 后自我重挂续产。
• 协作等待:source_yield(delay) 表示“本次正常返回、暂时无数据，请至少 delay 后再调我”。本次输出先完成按序刷新并释放 context 槽，随后 Source 进入执行器延迟队列；等待期间不占 worker。delay=0 是协作式让出，source_done() 与 yield 同时出现时以完成为准。非 Source 调用它是运行期错误。
• 定速:两条路。① 内核用 source_yield(delay) 表达外部数据暂不可用；② 声明式 rate: N(节点字段,Hz)保证相邻两次 process 至少隔 1/N 秒。rate 与显式 yield 共用延迟调度路径，取两者要求的较长等待，不在 worker 内 sleep。rate 只对源有效，且须为正的有限值。source 必须挂线程池执行器(挂委托执行器仍可能由用户回调独占宿主线程，建图期拒绝)。
• 阻塞边界:引擎无法把一个已经阻塞在用户 process() 内的 Source 自动变成非阻塞；读 socket、等锁或自行 sleep 仍会占住当前 worker。需要共享小线程池时，内核必须采用 source_yield 或 rate 的协作式模型。采用协作模型后，多个 Source 可以共享单线程池，不再按 Source 数量拒绝建图。
• 完成:内核调 source_done() → 引擎停止再触发本节点、关其输出边(复用关流级联,§7.6)→ 下游收流 → 图正常终止。无限源(永不 source_done)由 host cancel() 停。

内置 RangeSourceKernel(产 0..count 后 source_done)为样板。

7.5 背压策略

只有图输入口是限流点;图内部的边不对生产者施加背压。

⚠ 本表曾经是错的,而且错在一条安全属性上:它写着「图输入口有界(max_queue_size), 满时 send 阻塞至有空位」。实际上 max_queue_size 在全引擎只被 warn_if_over_soft_limit 读一次,send 从不看它 —— 它对图输入口和内部边同样只是告警。真正让 send 等待的是 全局包数水位,而它默认为 0(不限)。也就是说:**不显式配置 max_queued_packets 或节点级 input_queues.packets 时,普通内部队列没有任何硬上界, 只有 depth 100 时的一条 WARN。 (flow.h 的对应段落曾有同样的错误描述,已一并修正。)

Poller 队列属于全局水位统计:每个 Poller 是独立订阅者,因此同一包投给 N 个 Poller 会计 N 个队列槽(引用计数共享 payload,但每个订阅者都能独立滞留它)。默认 add_poller 保持历史兼容——无容量限制;需要严格约束输出滞留时用 bounded Poller:

• block:满时阻塞产出该边的线程,直到宿主取走一包。无损,但宿主必须从另一线程并发排水, 且带等待上界(PollerOptions::block_timeout,默认 5s;with_block_timeout(None) 可解除)。 到点仍无进展 → 记录图错误并放弃该包,wait_done / wait_until_idle 返回该错误。

  为什么必须有上界(实测,不是假想):block 是在派发路径内部原地等的,而派发可能 就跑在宿主自己的线程上(委托执行器,或 send 直接派发)。那时宿主既是生产者又是唯一 消费者 —— 它卡在 poller 的 push 里,永远走不到 poller.next()。实测 capacity=2、送 5 个包、 然后 close_all_inputs(); wait_done() 这个仓库里通用的顺序式写法会永久挂死;摘掉上界 后连 send 都回不来(而 send 没有超时变体,宿主侧无从自救)。

  为什么是「上界」而不是「死锁检测」:多线程宿主下「是否还有人能排水」不可判定 —— 可能另有一个线程正要排水。任何「宿主正阻塞即判死锁」的判据都会误杀合法的并发排水写法 (本仓库的 block_waits_until_the_host_drains_a_slot 就是那种写法)。所以这里不做推断, 只给一个可配置上界 + 说清前置条件与替代方案的报错。

  回归钉在 tests/poller_backpressure.rs:block_does_not_hang_a_sequential_host(顺序式 必须返回而非挂死)与 block_without_timeout_still_waits_for_a_concurrent_drainer (证明上界是可选的,没把 block 改成有损)。摘掉上界,前者立刻挂死。

• drop_oldest:丢最旧,保留最近 capacity 包;
• drop_newest:保留已有积压,拒绝新包;
• latest:容量固定为 1,永远只保留最新值。

有损策略都有独立 dropped_count 与指数退避 WARN。Poller pop、overflow 丢弃、reset 清空均同步扣减全局 packet/byte 计数。已注册 type descriptor 的自定义 Foreign payload 按固定对象 size 计量;这是浅尺寸,不包含 std::vector 等对象内部堆分配。

为什么内部边不能直接阻塞 worker —— 否则「扇出后再汇合」的合法 DAG 会死锁:

        ┌─► B(慢) ─┐
   A ──►┤           ├──► D
        └─► C(快) ─┘

C 迅速填满 D 的输入队列而阻塞;D 却要等 B 那一路才能消费;B 又在等 A 推进; 而 A 已阻塞在 C 上 —— 循环等待,且不需要环形拓扑就会发生。

input_queues 因此不在 dispatch 里等待。当前调用完成后若下游容量不足, 其 context 槽与 staging 保留在节点调度状态里,worker 立即返回线程池;该节点暂停认领 新输入。任一下游从输入队列弹包后重试 pending flush。限制的是节点继续产出, 不是占住线程等待,diamond 不形成线程循环等待。

input_queues.packets 是节点所有正向输入口的默认上限; input_queues.ports.video: {packets: 2} 按端口名覆盖。未列出的端口继承节点默认值, 显式 0 表示该端口不限。容量只按包数判定。

queued_bytes / peak_queued_bytes / total_queued_bytes 仍使用 Packet::byte_size() 记录 payload 浅尺寸,但只作诊断,不参与背压。它不包含 std::vector 等自定义对象内部 另行分配的堆内存,也不试图按共享 Arc 的唯一物理内存去重。

这些无损容量与 fixed_size 互斥:前者暂停生产者,后者有损丢最旧。一次调用向同一输出口 emit 的批量若自身大于有效包数容量会明确报错,避免永远无法满足的等待。若时间戳对齐 导致下游永远不能消费、所有 worker 又已空闲,等待接口会报告 internal backpressure cannot make progress,而不是永久挂起。

内部背压状态转换也会进入统一日志回调。第 1、2、4、8… 次阻塞打 WARN,包含生产者、 消费节点/端口、有效容量、已排队/已预留/本次待入队包数;对应阻塞解除时打 INFO 并报告 持续时间。指数限频避免长期慢消费者刷屏。若最终无法推进,等待接口的错误会列出每条 producer -> consumer.port 阻塞关系及容量、排队、预留和已阻塞时间。

阻塞 send 命中全局水位时怎么等 —— 分两种执行模型:

• 图跑在委托执行器(交还宿主线程):send 就地 pump_step() 自己推进一步(跑一个节点), 把水位压下去再继续。

• 图跑在线程池:主线程无任务可推,此时 send 应等池排水(wait_activity_since), 由工作线程消费降水位后唤醒。只有当全图彻底空转(无在飞任务)水位却仍下不去 (例如下游无人消费、包堆在出口)——那才是真卡死,send 返回 WOULD_BLOCK 而非永久阻塞。

  ⚠ 曾有缺陷:早期实现里 send 只会 pump_step(),而它只跑主线程任务;于是池图上 一撞水位就误报 WOULD_BLOCK(阻塞 send 退化成了报错)。已修正为「推不动且池仍在跑就等排水」, 并有回归测试 blocking_send_applies_backpressure_on_pool_instead_of_erroring 守卫。

全局水位与 Poller 背压现在也使用同一套诊断规则:第 1、2、4、8… 次阻塞打 WARN, 恢复打 INFO。Input::backpressure_stats() 提供图输入端口的当前等待者、事件数和累计时长; Poller::backpressure_stats() 同时提供策略、容量、积压、丢包与 Block 等待统计。 graph.dump() 会输出 watermark 与 poller 行。Poller 有损策略的 WARN 统一包含输出口、 策略、容量、当前积压和累计丢包数。reset 会把这些运行期统计全部清零。 graph.to_dot_with_stats() / lmflow_graph_to_dot_view(g, LMFLOW_DOT_DIAGNOSTICS) 会把相同信息直接画进拓扑: 图标题显示唯一的全局 queued/limit;图输入口只显示该句柄的等待次数/时长,避免把全局 水位误读成每端口容量。消费者输入边显示 queue/capacity/reservation,输出端用 Poller 虚拟节点显示策略、容量、积压与丢包。当前阻塞用红色粗线,历史阻塞或丢包用橙色。 多输入节点在节点框内列紧凑端口表(queued/capacity,reservation,state);同步/批处理 策略中,若一个口正在背压而同组另一个口为空且仍开放,后者以黄色 WAITING 标成疑似根因。 immediate 不做这种推断。正常边只保留端口名,避免大图被零值统计淹没。持续时间按 µs / ms / s 自动换算;标题同时显示本轮 start 后的快照时长。图内诊断图例解释 红/黄/橙和 Poller 虚线,SVG 的 tooltip 保留节点、输入边和 Poller 的完整快照,使默认 标签保持紧凑而悬停仍能查看详细数据。同一次统计图导出只取一个时间点,标题、节点、 输入队列、图输入和 Poller 的持续时间不会因格式化先后产生漂移。

DOT 视图分三档:Topology 只画静态拓扑,Compact 加节点状态和核心吞吐/延迟统计, Diagnostics 再加逐端口队列、背压、Poller 和诊断图例。旧的 to_dot() 与 to_dot_with_stats() 分别映射到 Topology / Diagnostics。统计视图中节点边框表示 CREATED / IDLE / RUNNING / CLOSED / ERROR,填充色继续专用于延迟热力图。 Compact / Diagnostics 的执行器图例同时显示当前 queued、running/线程容量、peak queued 与 completed；Topology 保持静态执行器形状。统计直接进入 DOT，不新增独立 Executor stats 公共 API。 标题汇总 running/error/blocked/waiting/dropped 热点;Compact 对未运行且无异常的节点 省略零值吞吐与队列字段。节点、算子、命名空间、执行器和端口的可见标签按固定字符数 截断,完整名称放进 SVG tooltip。节点按 executor 分组并以状态优先级稳定排序,同时调小 nodesep/ranksep;不启用 concentrate,避免 Graphviz 把语义不同的多端口边合并。

统计视图还维护视图私有的区间基线:首次导出统计本轮 start 至当前,之后按同一 Compact / Diagnostics 视图的相邻两次导出计算增量,两种视图互不污染,reset 清空基线。节点保留累计值,同时显示区间 processed、吞吐率、in/out 增量与区间平均耗时; 诊断边和 Poller 显示区间背压事件/时长与丢包增量。端口直接标注推断原因:消费者队列满、 对齐输入缺失、下游排水慢、全局 packet 水位或 Poller 消费慢/丢包。该基线只服务 DOT, 不新增宿主查询 API。

Diagnostics 在同一快照上再做 Top 5 综合排序:节点按错误、当前背压/等待、区间丢包、 区间平均耗时与积压排序;输入口按当前阻塞、对齐等待、区间丢包、区间背压和积压排序。 当前 queue-full / aligned-input 根因会触发一次只读反向遍历,把其生产者以及所有上游输入边 标成紫色 PRESSURE PATH;直接根因仍保留红/黄,避免传播链盖掉原因。该遍历只读既有拓扑, 不进入调度路径。examples/python/live_diagnostics 展示宿主每 100ms 导出 DOT、可选调用 Graphviz 生成 SVG 的完整循环。

线程池任务的全局在飞计数必须在认领仍持节点调度锁时递增,不能等到提交执行器时才加。 否则两步之间存在「节点已有 ready 调用、全局计数仍为 0」的假空闲窗口,并发关流时 wait_done 可能误报 nodes not closed。回归测试 wait_done_does_not_mistake_claimed_pool_work_for_idle 高频覆盖该窗口。

7.6 关流与终止

• close_all_inputs → 所有图输入边标 closed,并唤醒其消费者(触发排空)。
• 节点可关条件:所有输入边 closed 且队列空 且 !running。 → 调算子 close(零锁,R1)→ 标 closed → 关自己所有输出边 → 递归下游。
• 检查点:finish(node) 末尾 和 上游关流事件 —— 两处都要查,否则末尾节点可能永不关闭。
• 终止判定:GraphState{ open_nodes, err } + Condvar;节点关闭时 open_nodes -= 1, 归零则 notify_all。wait_done 等这个条件。
• poller_next_status:所在边 closed 且队空 → 返回 LMFLOW_ERR_CLOSED;否则阻塞在 poller 自己的 condvar 上。
• cancel 不是抢占:停止调度新任务、丢弃在途包、唤醒所有等待者,但已在执行中的 算子回调不会被中断。故 cancel 返回后可能仍有一个算子在跑,须 wait_done 确认静止。

7.7 错误路径

节点级策略 on_error(YAML 节点字段,默认 abort;未知值建图期明确报错,不静默):

abort(默认,历史行为)

• 算子返回非 0 → record_error → 置 AtomicBool has_error(快路径)+ 存首个错误(含文本)。
• 失败时丢弃该次的 staging,不传播半成品输出。
• 此后 try_claim 一律返回 false(停止调度)、send 返回错误、所有 poller 被唤醒返回 false、 wait_done 返回该错误码,lmflow_graph_last_error 可取文本。

skip(长跑实时管线用)

• 丢掉出错的那一个包,不置 has_error,其余包照常流过,图能正常终止。
• 必须推进下游时间戳边界,否则下游会永远等这一刻 —— 那等于把「一帧出错」升级成 「整图卡死」,比 abort 还糟。实现上不新写机制:清空 staging 后照常走刷新路径, 于是落到「无产出」分支自动 propagate_bound(input_ts + 1),与 Filter 丢包时同一条路。
• 有损行为绝不静默:计入 LMFlowNodeStats.errors,并打 WARN(指数退避,避免每帧都错 时刷爆日志 —— 与边的 note_dropped 同法)。

on_error 到底管哪些失败(容易误解,写清)。它只作用于 Process 路径,两个触发点:

1. 算子还没跑就失败 —— 输入包的 type_id 与 GetContract 声明不符,引擎侧 check_input_types 拦下,process 根本没被调用。
2. process 返回非 0 —— 这一个条件涵盖:Rust 算子返回 Err、Rust 算子 panic (catch_unwind → PANIC)、C++ 算子返回失败 Status、C++ 算子抛异常 (catch (...) → KERNEL)、以及 create 失败(构造 panic / 抛异常 → self 为 null)。 所以「算子失败」不只是「返回失败码」,panic / 异常也算。

不受 on_error 影响:

一句话:on_error 管的是逐包失败(这一帧坏、下一帧可能好),不管一次性的生命周期失败 (打不开就是打不开)。

什么时候用哪个。 能在算子内处理的错误,就在算子内处理 —— 捕获后返回成功但 不产出即可,引擎会自动推进边界(与上面 skip 走的是同一条路)。这样你能按错误类型 精确决定哪些可恢复。on_error: skip 是给你管不到的失败用的: 引擎侧的契约类型校验失败、以及算子的 panic / C++ 异常(那些没法「返回成功」); 以及你不拥有源码的第三方算子 —— 用 YAML 就能加固,不必改算子。

7.8 竞态 / 死锁自查表

7.9 执行器与线程归属

图在 YAML 里定义命名线程池,节点按名字选择在哪个池上执行:

executors:
  - name: "cpu"
    type: "ThreadPoolExecutor"
    num_threads: 4
  - name: "io"
    type: "ThreadPoolExecutor"
    num_threads: 1
nodes:
  - { name: "decode", kernel: "Decoder",  executor: "io"  }
  - { name: "detect", kernel: "Detector", executor: "cpu" }
  - { name: "draw",   kernel: "Overlay" }        # 未指定 → 默认执行器(default)

CPU 亲和力(绑核) —— 池可选 affinity: [核号...],worker i 绑到 affinity[i % len] 号核:

executors:
  - { name: "rt", type: "ThreadPoolExecutor", num_threads: 2, affinity: [2, 3] }

• 语义是 sched_setaffinity 的硬绑核:worker 线程只能在所列核上跑,减少迁移抖动、利于 NUMA/实时。
• affinity 长度 == num_threads 即 1:1 独占绑核;短于线程数则按 i % len 轮转复用。
• 仅 Linux 内核系生效(target_os = linux,含 Android=android、OpenHarmony/鸿蒙标准系统=linux); iOS/macOS/Windows 无 sched_setaffinity(Apple 不允许硬绑核,由系统管核),静默降级为不绑,不影响正确性。
• 只用 libc(glibc/Bionic/musl)已链接的 sched_setaffinity 符号(extern "C" 声明),不引入 libc crate,守住「零外部 crate 依赖」。
• ⚠ 亲和力 ≠ 优先级:它只限制「能在哪些核跑」,不改变线程分到多少 CPU 时间。

实时优先级 —— 池可选 priority: 1..=99,把 worker 切到 SCHED_FIFO 实时调度:

executors:
  - { name: "rt", type: "ThreadPoolExecutor", num_threads: 2, affinity: [2, 3], priority: 20 }

• 0(默认)= 普通分时(SCHED_OTHER),不动;1..=99 = 该 RT 优先级的 SCHED_FIFO(Linux/Android)。
• iOS/macOS 无应用级 SCHED_FIFO —— 映射成 Apple 的 QoS class:priority>0 → USER_INITIATED (「用户在等结果」,正合推理),顶格(>=90)→ USER_INTERACTIVE。这是 Darwin 上表达线程重要性的正道。
• 尽力而为:设实时调度需 CAP_SYS_NICE/root,拿不到就静默降级为普通分时,不影响正确性。
• 与绑核配合是刻意的:实时线程只在被绑的核上抢占,万一算子死循环也只拖垮那几个核、不拖垮整机。 worker 空闲时阻塞在 condvar 上(让出 CPU),故 idle 的 RT 线程不会空转饿死别人。

执行器类型 —— executors[].type 目前两种:

默认(节点未写 executor)= 默认执行器,它是一个按 CPU 核数开线程的线程池(ADR #16):

• 它完全由引擎持有,名字是 default,恒在 executors 的下标 0。和别的执行器完全同构: DOT 标 @default、线程名 default-0、日志里指名道姓。
• 不可配 —— 不绑核、不设实时优先级,YAML 里碰不到它。executors 里写的一律是宿主 自己的执行器,故 default 是保留名,声明它会报错(否则一张图里出现两个 default)。
• ⚠ 代价:默认执行不再零并发、执行顺序不再确定;且 Python 算子会在 worker 线程上 抢 GIL(见 §8.2)。线程池上的线性 CoW 由 invocation 完成路径保证零拷贝；扇出共享时仍 按 CoW 复制以隔离兄弟分支(见 §3.4)。

想控制线程数 / 绑核 / 优先级,或想要宿主线程语义 —— 都是「自己声明一个,把节点指过去」:

executors:
  - { name: "cpu",  type: "ThreadPoolExecutor", num_threads: 4, affinity: [0, 1, 2, 3] }
  - { name: "host", type: "DelegatingExecutor" }        # 不拥有线程,交还宿主线程
nodes:
  - { name: "detect", kernel: "Detector", executor: "cpu"  }
  - { name: "draw",   kernel: "Overlay",  executor: "host" }   # 零并发、顺序确定
  - { name: "post",   kernel: "Post" }                          # 不写 → 默认池

挂在 DelegatingExecutor 上就重新拿到:零并发、执行顺序确定、断点调试直观。 任务由实际进入引擎阻塞接口或调用 pump_step() 的宿主线程执行；通常由 Python 主线程负责推进时，Python 算子就在主线程运行，且同一张图内不会互相争抢 GIL(§8.2)。 多个宿主线程同时进入也不会把委托任务跑并发；多个委托执行器则按轮询顺序抽取， 避免前一个队列持续繁忙时饿死后面的队列。

⚠ 委托任务的执行时机 —— 引擎不能凭空占用宿主线程,只能在宿主进入引擎时借用它。 因此挂在 DelegatingExecutor 上的节点,其任务在宿主调用下列阻塞接口期间被抽取执行:

lmflow_graph_wait_done / _timeout
lmflow_graph_wait_until_idle / _timeout
lmflow_poller_next_status / _timeout
lmflow_input_send(阻塞等待空位时)

• 若宿主只 send 而从不调用上述任一接口,委托执行器上的节点不会推进。 自己掌握事件循环、不愿长阻塞的宿主可安装 Graph::set_wakeup_callback / lmflow_graph_set_wakeup_callback:回调从任意引擎线程发出,只负责向 Qt/libuv/自研 loop 投递事件；事件循环线程收到后反复 pump_step() 到 false。通知按边沿合并,不会每个 委托任务都跨线程唤醒一次。Python 直接用 await graph.run_async()。 每张图只有一个回调槽,再装是替换而非追加;通知也是全图的(只说"某处有进展", 不说哪个端口)。所以一个宿主里若有多处要被唤醒,须自己建一层分发:一个持有者装回调、 广播给各等待者,各自重查条件。第二次安装会静默冲掉前一个,症状是图停止推进、无错无日志。
• 反之,这些接口在等待期间一律抽取并执行委托任务,故不会因此死锁。
• 源节点(0 输入)不能挂委托执行器:用户 process 仍可能阻塞并独占宿主线程、 拖垮全图(建图期报错)。线程池上的 Source 应使用 source_yield / rate 协作等待； 协作 Source 不占等待中的 worker，因此多个 Source 可共享单线程池。
• 节点引用了未定义的 executor 名字 → init 阶段报错(§6.2 第 5 条)。

⚠ 实现注记:委托执行器的待办队列由它自己持有,不是 GraphInner 上的特例字段。 于是 Node::executor 只是一个普通下标,派任务的地方(dispatch_task)对两种执行器 一视同仁 —— 宿主线程在调度器眼里就是又一个执行器。

但 pump_step 的另一半 try_advance_closing 与执行器无关、不能省:maybe_close 不向下游递归,已排空的 sink 只能靠宿主 pump 的全量扫描去 close 并置 Terminated。 纯线程池的图今天也依赖它。

7.10 输入策略(节点级可插拔)

「多个输入口如何凑成一次 Process」+「队列满了怎么办」被抽成可插拔策略, 而不是写死在引擎里 —— 实时丢帧与(A 阶段的)时间戳对齐因此共用同一扩展点。

- name: "detect"
  kernel: "Detector"
  input_ports: ["frames"]
  input_policy: { type: "fixed_size", capacity: 2 }

sync_set 用于「A、B 该配对、C 独立」这类图 —— 只能全对齐或全不齐是不够的。配置用端口名给出分组, 须完整划分全部输入口(每口恰属一组;要独立就单独成组):

input_policy: { type: sync_set, sets: [["left", "right"], ["imu"]] }

默认 sync 已经按时间戳对齐全部输入口。只有需要“组间互不等待”时才应使用 sync_set;单个 set 覆盖全部输入口与默认 sync 等价,建议直接省略。某一组触发时, 其它组的输入不会作为残留值传入,而是继续留在各自队列中等待后续调用。

引擎内部把「就绪判定」的返回从单个时间戳升级为触发计划(时间戳 + 参与口):认领时只对参与口 弹包、推进 bound,其余口原样不动。sync/immediate/fixed_size 的参与口是"全部",行为不变。

多路选择(Mux)不做成输入策略,而是内置 MuxKernel:控制口(输入 0,I64 选择器)的值选择 把哪个数据口(输入 1..)转发到输出,配默认 sync 用。理由:push 模型下"只要求选中那一路"的 策略会让未选中口越积越多、甚至因陈旧时间戳卡死;sync 全对齐 + kernel 读控制转发既不积压也不卡, 且引擎不解释 payload(读控制值的是 kernel,守住 ADR #6)。

fixed_size 是有意的有损策略,且不阻塞上游 —— 因此与「内部边不背压」(§7.5) 并不冲突,而是其配套的内存约束手段:摄像头 30fps 而算子只跑 10fps 时, 无界队列会让内存无限增长,丢旧帧才是正确取舍。

batch 把多个时间戳攒成一批一次交给算子(与「单包/次」相反):process() 里 input_count(i) 给出本次批大小、input_at(i, k) 取第 k 个包 —— 单包策略下 input_count 恒 0/1,故同一套访问器 两种交付通用。攒够 capacity 即触发,关流时不足一批也刷出(不丢数据),批的输出时间戳继承批内 末包(下游单调)。用于批推理、窗口聚合;内置 BatchSumKernel 为样板。

多输入口:一批 = capacity 个「对齐元组」(ADR #41)。对齐规则与 sync 完全同源 —— 把单包时的那一轮对齐连续跑 capacity 轮,每轮取各口游标处的全局最小时间戳。因此 各口取数可以不同:某口在某个对齐时间戳上没有包,该轮就不取它(input_count(i) 本就是 按口计数的,算子侧无需改动)。

刻意不做成「各口各自数够 capacity 个」:那会把 0 号口的第 k 个与 1 号口的第 k 个配成 一对,而它们未必是同一帧 —— 图像批与掩码批就此错位,且不报任何错。静默的错误配对是 本项目明确拒绝的失败模式。

不足一批时只有所有正向口都已关闭才刷余量(不可能再来数据了);只要还有口开着就继续等 —— 提前交付就是过早切批,同样是静默的语义偏差。反馈口(back_edges)不参与对齐,每次触发读一次 最新值,与其它策略一致(多口 batch 才让 batch + back_edges 成为可能:单口时凑不出 「至少一个正向口 + 一个反馈口」)。

实现上,就绪判定期已把各口时间戳前缀快照出来并算好每口取数,认领期照计划批量弹出 —— 每口仍只拿一次队列锁,保住 ADR #36。

丢包绝不静默:除 lmflow_graph_dropped_count(port) 累计计数外,首次丢弃还会打一条 WARN 日志。任何有损行为都必须可观测,否则「跑通了」和「悄悄丢了一半」无法区分。

7.11 子图(subgraph)与跨文件 include

子图把一张小图打包成「一个节点」复用;include 把子图库拆进独立 YAML。两者都是纯建图期变换(GraphConfig → 展平的 GraphConfig,见 src/expand.rs),插在 parse 与 check_supported/build 之间。运行时引擎 / 调度器完全不感知子图 —— 连边纯按端口名字符串,展开只是多产出些节点和名字。

定义与实例化:

subgraphs:
  Denoise:                 # 子图名
    nodes:
      - { name: a, kernel: BlurKernel,    input_ports: [sin], output_ports: [mid] }
      - { name: b, kernel: SharpenKernel, input_ports: [mid], output_ports: [sout] }
    input_ports: [sin]     # 边界输入口
    output_ports: [sout]   # 边界输出口
nodes:
  - { name: d, type: Denoise, input_ports: [raw], output_ports: [clean] }   # 实例:type 填子图名
  - { name: s, kernel: ScaleKernel, input_ports: [clean], output_ports: [out] }
input_ports: [raw]
output_ports: [out]

展开规则:实例节点 d(type: Denoise)被内联替换成子图内部节点:

• 命名空间:内部节点名 → d/a、d/b;内部边 → d/<边名>(如 d/mid)。用 / 分隔(: 是端口 tag 分隔符,不能用)。同一子图实例化多次(d、e)各自命名空间,内部边不串。
• 边界按位置重映射:子图 input_ports[i] ↔ 实例 input_ports[i],output_ports 同(数目须一致,不齐报错)。上例 sin→raw、sout→clean,于是展平后 d/a(raw→d/mid)、d/b(d/mid→clean),与手写扁平图完全等价。
• 递归 + 环检测:子图内部节点还能是子图实例(type:),递归展开;A→…→A 报错。
• kernel 与 type 二选一:实例节点填 type、算子节点填 kernel,两个都给或都不给都报错。

跨文件 include:

# main.yml
include: ["lib.yml"]       # 相对本文件目录;可多个;可递归(被引文件也能 include)
nodes:
  - { name: d, type: Denoise, input_ports: [raw], output_ports: [clean] }   # Denoise 定义在 lib.yml

• 只并入被引文件的 subgraphs(它是子图库);其 nodes / executors / ports 忽略。子图内部节点按名引用主图的 executor。
• 子图重名(跨不同文件)报错;同一文件被引多次(菱形)去重、不报错。
• 相对路径需要基准目录,故 include 仅 from_yaml_file 支持;from_yaml(文本入口)遇 include 明确报错。C ABI 的 lmflow_graph_init_from_yaml_file 自动受益。

未知子图名 / 边界数不齐 / 环 / kernel+type 冲突,都在 init 时报错(LMFLOW_ERR_INVALID_ARG),不静默。

7.12 反馈环(back-edge)—— 最新值反馈寄存器

默认图必须是 DAG(check_acyclic 拒环)。要让边成环(反馈控制、循环状态、流驱动迭代),把消费反馈的那个输入口列进节点的 back_edges:

nodes:
  - name: acc
    kernel: FeedbackAddKernel
    input_ports: [in, out]      # in = 正向;out = 反馈(消费本节点自己的输出边 out)
    output_ports: [out]
    back_edges: [out]           # 把 out 口标为反馈寄存器 → 自环合法
input_ports: [in]
output_ports: [out]

语义(最新值反馈寄存器):被标记的输入口是一个容量 1、留最新值、消费一次的队列,且

• 不参与就绪判定 —— 只有正向输入能触发节点(核心不变式:反馈包永不自激重跑);
• 不参与终止判定 —— 节点靠正向输入排空即可关闭,关闭级联绕环拆解;
• 不参与时间戳对齐 —— 反馈包拿什么时间戳都不污染 sync_align;
• 首拍(尚无反馈)该口为空,内核自处理(如按 0)。

为什么安全(三个危害各自化解):

约束:非 source 节点必须至少留一个正向输入(否则永不触发,init 报错);back_edges 名字须是本节点输入口;sync_set 分组不得含反馈口;未被 back_edges 打断的拓扑环仍报错。纯建图期 + 调度器局部改动,无新增阻塞 / 锁。内置 FeedbackAddKernel(out = 正向 + 反馈)为样板,自环即运行累加。

7.13 重跑(reset)—— 保留算子实例的复位

Terminated 后 reset 可把图复位为可再次 start,保留已 open 的算子实例 —— 省掉每会话重建图 + 重跑 open(如重新加载模型)的开销。处理完视频 A 再处理 B,不必重建整张图。

前提:图须 Terminated 且静止(is_idle,没有 worker 还在算子里),否则返回 Error::State。宿主通常先 wait_done()。这条静止依据与 Drop / start 用的是同一个 —— in_flight == 0 且 main_queue 空 ⇒ 没有 worker 在 run_node 中途,故复位无并发。

不碰线程池:worker 随图存活、静止时都 park 在 condvar 上、stop 仍为 false,下一轮 start 直接复用;shutdown + join 只发生在 Drop。这也意味着 reset 不付重建线程的代价。

字段分三类:

最易漏的三个(都有专门测试钉住):

1. Edge::last_sent 必须回 unset() —— 否则单调性校验会拒掉下一轮从图输入口发的第一个包(时间戳通常又从小开始)。
2. GraphShared::{error, has_error, cancelled} —— record_error 只「首因生效、不覆盖」,没有反向清除路径;不清则 reset 后的图带着旧错误出生,start 的 try_claim 立刻被挡回。为此新增 reset_run_state()。
3. input_bounds 必须回 pre_stream()(不是上一轮 close 推到的 done())—— 否则 readiness / 对齐会认为每个空口「已到流尾」,语义崩坏。

opened 的保留由 start 侧配合:start 见 opened == true 就跳过 open(只重灌 side packet + 复位槽)。这正是「不重跑 open」= 不重载模型的价值所在。

C ABI:lmflow_graph_reset;Python:Graph.reset()。四层同一语义。

8. Python 接口(已实现)

8.1 形态

pybind11 模块 lmflow._lmflow(python/src/bindings.cc)链接 liblmflow, 只调用 include/lmflow/flow.h 这一层 C ABI —— 和 C++ 算子走同一条路。 Python 包 lmflow(python/lmflow/__init__.py)在其上提供 @kernel 装饰器、 Kernel 基类、Graph 上下文管理器与类型常量。 Python 既可注册算子,也可驱动图。

构建:pip install .(scikit-build-core 驱动仓库根 CMake:cargo 编引擎 + pybind11 扩展; pybind11 由 third_party/pybind11 子模块提供)。或直接走 CMake:cmake -B build -DLMFLOW_BUILD_PYTHON=ON。

@lmflow.kernel("PyOffsetKernel")
class PyOffsetKernel(lmflow.Kernel):
    @staticmethod
    def get_contract(c): c.input_set_any(0); c.output_set_any(0)
    def open(self, cc):    self.offset = cc.option_int("offset", 0)
    def process(self, cc): cc.emit(0, lmflow.Packet.from_int(cc.input(0).as_int() + self.offset))

with lmflow.Graph.from_yaml(CONFIG) as graph:        # with 是硬要求,见 8.3
    poller = graph.add_poller("out"); graph.start()
    source = graph.input("in")
    for i in range(10):
        source.send(lmflow.Packet.from_int(i), ts=i)
        print(poller.next(timeout=5.0).as_int())
    graph.close_all_inputs(); graph.wait_done(timeout=5.0)

同一张图里 C++ 算子与 Python 算子平等引用 —— 引擎不区分算子语言。

8.2 GIL

• ⚠ 默认会有 GIL 争抢:默认执行器是线程池(ADR #16),Python 算子在引擎工作线程上 被回调、期间持 GIL ⇒ 多个 Python 算子之间无法真并行。
• 想回到「同图 Python 算子不互相争抢 GIL」:声明一个 DelegatingExecutor, 把 Python 算子指过去 —— 它们在负责推进的宿主线程上串行执行；通常由 Python 主线程调用阻塞接口或 pump_step() 时，就是在 Python 主线程执行(§7.9)。 C++ 算子仍可放进线程池,两者真并行 (见 examples/python/opencv_pipeline/opencv_pipeline.py)。
• 重计算仍应优先写成 C++ 算子 —— 那是唯一能在池里真并行的路。
• 所有可能阻塞的接口必须释放 GIL(poller.next / wait_done / send), 否则工作线程拿不到 GIL → 直接死锁。pybind11 用 py::call_guard<py::gil_scoped_release>()。
• 引擎线程由 Rust 创建、Python 并不认识:每次 gil_scoped_acquire 会建立 thread state, 线程退出时须清理,否则泄漏。

8.3 解释器生命周期(崩溃来源)

图必须在解释器开始销毁之前停掉,否则工作线程可能回调进正在析构的解释器 → 崩溃。 因此 Graph 实现上下文管理器,__del__ 只作兜底、不保证时机。文档必须强调 with。

8.4 数据类型限制

Python 算子只能收发内建类型(整数/浮点/布尔/字符串/bytes/LMFlowBuffer)。 理由见 ADR #9:保持 payload 的语言中立性。表达结构化数据:

LMFLOW_TYPE_HOST_OBJECT(7 号)已预留但未启用。若将来开放,应利用「引擎在 init 阶段已知每个节点算子语言」这一点,直接拒绝「原生对象端口接到异语言算子」的拓扑。

8.5 零拷贝的正确写法

send(cv2.imread(...)) 通过 external buffer adoption 零拷贝进入图。绑定层持有 ndarray 引用并把传入数组临时标成只读；最后一个 Packet 引用归零时，释放回调先获取 GIL，再恢复 原 writeable 标志并 Py_DECREF。所有可能等待引擎线程的 Python API 都必须释放 GIL， 否则工作线程最终释放 ndarray 时会与宿主互等。

调用方不得通过另一个共享同一 allocation 的 numpy alias 并发修改数据。Python-owned buffer 以 READONLY 进入引擎，因此算子请求可写视图时会 CoW，不会改坏原数组。

如果数据本来就要由宿主生成，仍可让引擎分配缓冲并拿 numpy view 就地写入:

def process(self, cc):
    src = cc.input(0).as_numpy()                       # 零拷贝只读视图
    packet, dst = cc.new_buffer((h, w, 3), np.uint8)   # 引擎分配,dst 指向引擎内存
    cv2.resize(src, (w, h), dst=dst)                   # 直接写进去
    cc.emit(0, packet)

就地改写走 CoW,同样要先 take_input:

packet = cc.take_input(0)     # ← 先取走,否则上下文仍持引用 → CoW 必然复制
img = packet.make_mutable()   # 独占 → 零拷贝可写 numpy view
cv2.GaussianBlur(img, (5, 5), 0, dst=img)
cc.emit(0, packet)

8.6 其它必须处理的点

• Ctrl-C:阻塞调用释放 GIL 后停在 Rust 里,Python 信号处理不会运行 ⇒ Ctrl-C 无响应。 实现上须按短超时轮询,周期性回 Python 检查信号。
• 异常:Python 异常不得穿越 FFI;蹦床里捕获并转 LMFLOW_ERR_KERNEL,原文本走日志。
• fork:引擎线程启动后 fork 会得到残缺状态;文档禁止(含 multiprocessing 默认 fork 启动法)。
• 静态库双份注册表:见 §8.5 标题下 —— Python 模块必须链动态库,否则算子注册表 会有两份互不可见的副本。同时对 cdylib 做符号可见性控制,只导出 flow_*。

9. FFI 边界与安全

• Rust 内部用 Result<T, Error>;? 取代错误码宏。
• 每个 extern "C" 导出函数用 catch_unwind 包裹,panic → LMFLOW_ERR_PANIC (Rust panic 穿越 FFI 是 UB)。
• 算子回调方向:C++ 异常 / Python 异常同样不得逃出,由糖层 / pybind11 蹃床转错误码。
• ABI 版本:LMFLOW_ABI_VERSION + lmflow_abi_version();lmflow_graph_new 内部校验, 不匹配返回 NULL 并置错误。动态链接时 header 与 .so 不一致会导致布局错乱。
• 布局一致性:cpp/abi_assert.cc 的 static_assert 与 core/tests/abi_layout.rs 钉在同一组常量上,任一侧改字段而忘同步 → 构建失败(已实测能拦住)。
• ABI 演进:LMFlowBuffer 留了 reserved 供未来加字段(最可能是 GPU 内存空间); 一旦改变既有布局或删除/改变既有导出符号,必须提升 LMFLOW_ABI_VERSION,所有既有 二进制都要重编。

10. 构建与集成

Rust core 与 C++ kernels 分离构建

• lmflow(lmflow/core/):crate-type = ["lib", "staticlib"]。 lib 给 Rust host/测试；staticlib 是 CMake 合成纯 core/完整动态库的唯一原料。
• lmflow/core 没有 build.rs、没有 C++ build dependency，也没有 kernels feature。
• CMake 构建纯 Rust lmflow::core_static、可选 lmflow::kernels，并按 LMFLOW_BUILD_SHARED_LIBS 组合 lmflow::core 与统一消费目标 lmflow::lmflow， 并避免污染宿主工程的全局共享库设置。
• 引擎语义、C ABI、内存、策略和并发测试全部使用 Rust 测试 kernels，留在 lmflow/core/tests；官方 C++ kernels 由 CMake/C++ 与 Python 端到端测试覆盖。
• 示例宿主:cargo run --manifest-path lmflow/examples/rust/hello_world/Cargo.toml。
• C ABI 的验证:Rust 集成测试直接 unsafe 调 extern "C" 函数,无需 C 语言 main。

Python 部分由同一 CMake 编排(ADR #4)

• pyproject.toml(scikit-build-core)→ CMake 编 pybind11 模块 → 静态链接 lmflow::core_static 与 lmflow::kernels。
• 两步:cargo build 出库 → pip install -e . 编扩展。
• 外部 C++ 宿主:自带构建系统,链 liblmflow + include/,不进本仓库构建。

依赖:serde/serde_yaml/serde_json;构建期 cc;Python 侧 pybind11。 crossbeam-channel、parking_lot 在 M3 引入。

11. 目录结构

lm-flow/                          仓库根
├── lmflow/                       第一方源码
│   ├── core/                     引擎 crate `lmflow`(包名=库名=lmflow → liblmflow.a)
│   │   │                         **默认纯 Rust**:不编译也不捆绑任何 C++
│   │   ├── Cargo.toml · Cargo.lock
│   │   ├── src/                  timestamp / packet / edge / node / graph / scheduler / ffi /
│   │   │                         kernel_api(Rust 算子糖)/ builtin(自带默认 Rust 算子)…
│   │   ├── tests/                纯 Rust 单元/集成测试
│   │   └── benches/              dispatch.rs(纯 Rust Criterion)
│   ├── include/                  公共头(消费者 #include "lmflow/xxx.h")
│   │   └── lmflow/
│   │       ├── flow.h            C ABI —— 唯一稳定接口(权威定义)
│   │       ├── flow.hpp          C++ 算子糖层(header-only,非 ABI)
│   │       └── flow_platform_log.hpp  可选:引擎日志接平台日志(logcat/os_log/HiLog)
│   ├── adapters/
│   │   └── opencv/               可选 OpenCV adapter(CMake target `lmflow::opencv`)
│   │       ├── include/lmflow/opencv.hpp
│   │       └── tests/opencv_test.cc
│   ├── cpp/                      C++ 侧(**非引擎**,且在 crate 之外 → 不随发布的 crate 分发)
│   │   ├── kernels/              18 个内置 C++ 算子(一文件一算子、自注册)
│   │   ├── abi_assert.cc         跨界结构体布局的编译期校验
│   │   └── tests/                flow_hpp_test.cc / buffer_util_test.cc + CMakeLists
│   ├── python/                   src/bindings.cc(pybind11)+ lmflow 包 + CMakeLists
│   └── examples/                 examples/<lang>/<name>/:cpp · python · rust · android · ios · harmonyos
├── third_party/pybind11/         vendored 子模块(仅构建 Python wheel 用)
├── cmake/                        engine.cmake · install-sdk.cmake · find_package 配置
├── CMakeLists.txt                顶层构建(驱动 cargo;C/C++ SDK + Python 扩展)
├── pyproject.toml                Python wheel(scikit-build-core → 同一份 CMake)
└── docs/
    ├── design.md                 本文档
    └── web/                      文档站源码(见 §11.1)

11.1 文档站

https://laomou.github.io/lm-flow/,由 .github/workflows/docs.yml 在每次推 main 时构建部署。 一个站点覆盖三端,三端平等:

三条关键取舍:

1. C/C++ 端手写,不上 Doxygen。 flow.h 的 131 个函数声明没有一个前置 Doxygen 可识别 注释(全是普通 /* */),且大量说明是「章节横幅」而非贴着符号(一条注释同时说明 5 个 lmflow_packet_as_*)—— 换标记也绑不到符号上,Doxygen 只会产出空壳签名索引。手写指南是 唯一能真正解释 ABI 契约(所有权三态、指针生命周期、锁规则)的形式。
2. doc 注释语言按「谁会读」划分,而非按语言划分:面向用户的入口(crate 首页、Graph / Packet / Kernel / KernelCtx / KernelContract / register_kernel / Timestamp / Contract、builtin 与 ffi 模块头、Python docstring)用英文;引擎内部的实现注释与 不变量论证仍用中文。
3. 零新依赖。 build.py 只用 markdown2 + pygments,而这两个都是 pdoc 的依赖 —— workflow 本来就要 pip install pdoc,故依赖清单一行未增。刻意不引 MkDocs / Sphinx: 为两个手写页面搭一套静态站生成器,还要让它接管整个 site/(把 rustdoc / pdoc 产出当 "静态资源"塞进去),是纯粹的摩擦。

cargo doc 的产出根目录没有 index.html(真入口是 lmflow/index.html),故 build.py 额外写一个 site/rust/index.html 重定向壳,rustdoc 拷贝步骤用 cp -R .../doc/. site/rust/ 只合并内容、不覆盖它。workflow 末尾有一步产物齐全校验 —— 宁可构建失败,也不发半成品站点。

cargo test --doc 此前验证不了任何东西(4 个围栏块全是 ignore /text);现在 crate 首页与 kernel_api 模块头的 3 个示例都是可运行 doctest,RUSTDOCFLAGS: -D warnings 也一并 把失效的 intra-doc 链接变成硬错误。

12. 里程碑

M3 之所以排在 M4/M5 之后:当年默认执行器是宿主主线程(旧 ADR #16),所以 M2 的 同步执行本就是产品的默认行为,而不是临时脚手架 —— 线程池当时是「显式 opt-in」的增量。 后来 ADR #16 反转:默认改成线程池,宿主线程执行降级为显式的 DelegatingExecutor (M2 那条路径至今仍在,只是不再是默认)。

13. 测试策略(已落地 326 个:Rust 282 + soak 2 + doctest 3 + Python 39;另有 3 个独立 C++ 测试)

13.1 原始策略与补充

• Rust 单测:timestamp 边界(已有 9 个)、edge FIFO、就绪判定、YAML 解析、注册表。
• ABI 布局:abi_layout.rs + abi_assert.cc 双向钉死(已实测能拦住破坏)。
• C ABI 冒烟:Rust 集成测试直接调 extern "C",覆盖边界①。
• 端到端:hello_world 输出序列断言(0..9,时间戳对应)。
• 并发(本设计的核心风险):TSan 是硬门禁 —— 164 个测试 0 条竞态报告 (RUSTFLAGS=-Zsanitizer=thread cargo +nightly test -Zbuild-std --lib --tests; 排除 doctest 是因 -Zbuild-std 下 rustdoc 会因 sanitizer ABI 不一致而编不过,属工具链限制)。 覆盖 max_in_flight 并行路径:乱序完成仍按序、并行下取消/销毁不漏释放、多输入对齐叠加并行。 Miri 尚未跑(FFI 大量 extern "C" 与外部 C++ 符号,Miri 无法执行)。
• 死锁回归 / diamond 内存曲线:tests/soak.rs::watermark_bounds_memory_in_diamond_topology —— 一条图输入边被两个节点消费(真正的内部扇出),慢分支睡 200µs,两路再汇到一个 sync 节点的两个输入口。同时验证活性(内部边无界 ⇒ 不形成循环等待)与内存(增长受水位 约束而非受吞吐约束)。 > 本条此前是假的:文档一直声称有这个测试,而全仓库每个多输入节点都是从图输入口 > 喂的,没有任何一处是内部边扇出后汇合 —— 也就是说 ADR #11 用来拒绝内部边背压的论证, > 自己从未被测过。现已补上,实测数据见 §13.4。
• CI(.github/workflows/ci.yml):headers(纯 C/C++ -Werror)、rust(fmt/clippy -D warnings/test/示例输出比对)、 external-host(header 声明与 .a 导出符号对齐 + 编译运行外部 C++ 宿主)、 tsan(硬门禁)、sanitizers-extra(ASan/Miri,暂不门禁)、python(真编扩展 + 跑测 + 示例输出比对)。
• 跨平台矩阵(见 §14)尚未加入。

13.2 实现阶段真实抓到的缺陷

这些不是假想 —— 都是 MVP 过程中被测试或工具抓出来并修掉的,已各自留下回归测试:

13.3 原策略清单

13.4 长跑内存曲线压测(soak)

tests/soak.rs —— 默认 #[ignore](不拖慢常规套件),CI 的 rust job 里以约 1 GiB 规模显式跑一遍;本地可用 LMFLOW_SOAK_PACKETS 放大。

为什么需要它,以及它证明了什么水位读数证明不了的事。 此前全局水位只有功能测试 (「撞到水位会转成背压」),而「功能对、内存仍在涨」是完全可能的。做负对照实测过: 把消费算子改成不释放包(堆起来),total_queued_bytes 峰值依旧稳稳停在 4 MiB 水位上、丢包断言也照样通过 —— 而 RSS 跟着总吞吐一路涨了 233 MiB,曲线是干净的线性斜坡。 也就是说:水位读数正常并不能推出内存有界,这个盲区只有量真实 RSS 才能盖住。

断言的形式很关键 —— 不是「内存小」,而是:

RSS 的增长受水位约束,而不是受总吞吐量约束。

因为该上界与包数 n 无关,泄漏与正常的差距就是两个数量级,不存在阈值调参的模糊地带。 实测(Linux,/proc/self/status 的 VmRSS):

另外三条辅助断言各有职责:水位真的被压到过(否则一次「跑得太快、从不积压」的运行 也会绿,却什么都没证明)、软水位的超出量有界(允许 Relaxed 快照滞后几个包,但不该无界)、 无丢包。RSS 读数是 Linux 专属;其它平台自动跳过该条,其余断言仍然有效。

扇出 + 汇合(diamond)拓扑 —— watermark_bounds_memory_in_diamond_topology。 上面那条是单节点线性图,而 ADR #11 唯一要保护的形状恰恰是 diamond,此前零覆盖 (§13.1 曾声称有这个测试,实际没有)。拓扑:

   in ──┬─► slow(睡 200µs) ─► s ─┐
        └─► fast ───────────► f ─┴─► join(默认 sync,按时间戳对齐两路)

一条图输入边被两个节点消费 = 真正的内部扇出;两路再汇到 join 的两个输入口。 水位按个数(max_queued_packets: 64)而非字节 —— 因为 Payload::byte_size() 对 Native / Foreign 计 0,字节水位只对内建 payload 有效,而个数水位对所有形态成立。 注意扇出后每包占 2 个队列槽(on_enqueue 是按消费者调用的)。

与线性图的 4364 / 4364 数值一致,且同样与吞吐无关;2.5 GiB 跑完 5.54 秒、未挂住, sync 汇合把每个时间戳都对齐处理掉、无丢失、无错配。

这条测试同时是 ADR #11 的活性回归:内部边一旦被加上阻塞式硬上界,慢分支会让 join 永不触发,wait_done_timeout 就会在这里超时 —— 那正是 §7.5 描述的循环等待。

13.5 type_id 的跨编译器行为(ADR #22 的验证)

type_id 有两个此前完全没有测试的面,而它们的失效都是静默的:

1. 哈希算法在两侧各有一份手写实现 —— C++ 在 flow.hpp 的 Fnv1a + NormalizeTypeId, Rust 在 packet::fnv1a_type_id。任何一侧动了 offset basis、乘子、或那个 < 16 的 内建区规避分支,同一个类型在两边就会算出不同 id,契约校验形同虚设 —— 不报任何错。
2. LMFLOW_DECLARE_TYPE_NAME 是跨工具链的唯一逃生口,却只在一个示例里用过。 若它哪天悄悄退化成 no-op,type_id 就会静默退回修饰名 —— 而修饰名 MSVC 与 GCC/Clang 不同,于是跨编译器混用算子时类型校验静默失配。

对应的钉法:

两处期望值都由第三方实现(Python)独立算出,不是从任一侧抄来的 —— 否则就是拿实现验证 自己。为让编译期断言能穿透整条链,NormalizeTypeId 从 inline 改成了 constexpr。

上面这些都还抓不到编译器默认名称变化 —— 它们钉的是哈希函数,在任何编译器上都 同样通过。真正的默认身份来自 typeid(T).name(),故测试分别钉住两套已支持 ABI:

#ifdef _MSC_VER
assert(std::strcmp(typeid(int).name(), "int") == 0);
assert(lmflow::TypeId<int>() == 3143511548502526014ULL);
#else
assert(std::strcmp(typeid(int).name(), "i") == 0);
assert(lmflow::TypeId<int>() == 12638195996648667684ULL);
#endif

分歧比「修饰方案不同」更彻底:Itanium ABI(GCC/Clang)下是 "i",而 MSVC 的 type_info::name() 返回未修饰的可读名 —— "int"、"struct Foo"(修饰形式在 raw_name())。故 FNV("i") 与 FNV("int") 毫不相干。两套断言保证各自 ABI 内不会 静默漂移,但不宣称默认 id 能跨 ABI 对齐;跨工具链传自定义类型仍必须改用 LMFLOW_DECLARE_TYPE_NAME。

「证明宏真生效」那条做过负对照:把 LMFLOW_DECLARE_TYPE_NAME 改成 no-op,测试立即断言失败。 只断言「id == 某常量」是不够的 —— 那条单独看,一个 no-op 宏也可能巧合通过。

13.6 多配置生成器下的 cargo profile(构建正确性)

CMAKE_BUILD_TYPE 只对单配置生成器有意义(Unix Makefiles、单配置 Ninja);多配置 生成器(Ninja Multi-Config、Visual Studio)的配置是构建期由 --config 决定的,配置期 读那个变量通常为空。cmake/engine.cmake 原先在配置期做 if(CMAKE_BUILD_TYPE MATCHES ...), 于是在多配置生成器下会静默退到 debug:

cmake -B b -G "Ninja Multi-Config"
cmake --build b --config Release --target flow_engine
→ Finished `dev` profile [unoptimized + debuginfo]      # 要的是 Release

后果是 C++ 侧按 Release 编、却链进一个未优化的 debug 引擎,而且不给任何提示 —— 正是本项目明确拒绝的静默失效。这条在 Linux 上就能复现,不是 Windows 专属问题 (它是做 Windows 侦查时被发现的:windows-latest 默认生成器是多配置的 Visual Studio)。

改法:用生成器表达式让 profile 跟随实际构建的配置;imported target 改用 IMPORTED_LOCATION_<CONFIG> 并把 RelWithDebInfo/MinSizeRel 映到 release (cargo 只有 dev/release 两档);COMMAND_EXPAND_LISTS 让 Debug 下 --release 展开为无参数而不是空字符串参数(cargo 会拒空参数)。单配置生成器行为不变。

CI 守卫在 cmake-sdk job 里:配置一次 Ninja Multi-Config,对 Release / RelWithDebInfo / Debug 三个配置做 ninja -n 干跑,断言 profile 与 --config 一致,并额外确认真实的 --release flag 在生成的构建文件里。只配置 + 干跑,不真编译,秒级。 做过负对照:把 engine.cmake 改回原样,守卫立即失败。

14. 风险登记

15. 待决项的处置

上一版列出的 7 项未决事项已全部拍板,理由并入 ADR(§1):

另外明确不做的:程序化构图 API(ADR #25)、observer 注册后的移除(登记于 header)。

15.1 结构体前向兼容的两种做法(有意的不一致)

15.2 实现阶段的验证结果

15.3 当前实现的已知边界

诚实记录,避免误以为已完成。本节只写当下仍然成立的边界 —— 一旦某条被实现,就从这里删掉 (否则它会反过来低报完成度:曾有一版这里还写着「线程池尚未实现」,而线程池早已落地并进了 TSan 门禁)。

引擎语义上的边界

• 无通用节点 timer:执行器已有 Source 专用延迟唤醒(source_yield / rate)， 但非源节点仍不能请求“每 N 毫秒给我一次 tick”。未设 rate、未调用 source_yield 且持续产出的源仍会尽快运行，可能持续向下游施压。
• LMFLOW_TYPE_HOST_OBJECT 未启用(ADR #26):跨语言算子只收发内建类型。 它现在是主动拒绝,不再是「没人生产它」的默认状态 —— 契约声明它在建图期报错; 图输入、算子输出和 side packet 带它都在各自进入数据流的边界报错,且不会先派给 poller / observer。所有报错都给出替代方案(数值集合用 BUFFER,任意元数据用 STR 装 JSON)。 钉在 tests/host_object.rs,含一条负对照:把运行期判断挪到 want == 0 短路之后, 声明 any 的端口那条测试立刻失败 —— 那正是最容易漏的配置。
• Packet::new(Rust 原生值)的 type_id 是 NONE,不参与跨语言类型校验 —— 它是有意如此,不是待补的功能:Rust 的 std::any::TypeId 与 C++ 修饰名哈希是两套 身份空间,自动映射会造出「看着能跨语言、实际只与自己一致」的 id,那是静默失配, 比现在的明确报错糟得多。出路两条:内建 payload 用 Packet::from_i64 / from_f64 / from_builtin(它们带正确 type_id,也正是跨语言算子该交换的东西,见 ADR #9); 自定义类型优先实现 unsafe InteropType 后用 Packet::from_interop;底层逃生口 Packet::new_interop 是 unsafe,要求调用方自行证明 Rust/C++ ABI 布局一致,且拒绝 0..15 内建保留 id。双方通常以 C++ 侧 LMFLOW_DECLARE_TYPE_NAME 的稳定名字对齐。 仍留在本节是因为用户确实会撞上:Packet::new 是最自然的名字,却是唯一过不了类型 契约的构造函数,且失败在运行期第一个包而非建图期。缓解是把出路写进错误本身 —— got == NONE 时错误会点名 Packet::new 并列出该改用哪几个构造函数(有配对测试钉住: 一条断言提示出现,一条证明它指的路真的可行)。

验证覆盖上的边界

• Windows/MSVC 已进入原生 CI,不再只是交叉 cargo check:同一 job 运行纯 Rust 与 纯 Rust core 测试，并用 cl.exe 独立编译 C11/C++17 公共头、18 个算子、ABI 断言和纯头测试;再用 Visual Studio 生成器分别构建/测试 Debug 与 Release、安装 SDK, 通过临时最小外部工程验证 find_package(lmflow)、静态链接与 ABI 版本,最后构建并运行 Python 扩展。为此 CMake 按 MSVC 产物名选择 lmflow.lib,传播 rustc 实测给出的 kernel32/ntdll/userenv/ws2_32/dbghelp,并用 CMP0091 把所有配置的 CRT 固定为 /MD, 避免 Rust 的 /defaultlib:msvcrt 与 CMake Debug 默认 /MDd 混用。
• TSan 在 windows-msvc 上不可用(仅 Linux/macOS)。并发安全仍由 Linux TSan 硬门禁 负责;Windows job 验证平台编译、链接、运行和绑定,但不宣称提供等价的竞态检测。
• Miri 跑不动:FFI 里大量 extern "C" 与外部 C++ 符号,Miri 无法执行,故只作 advisory (continue-on-error),不是门禁。ASan 同样是 advisory(build-std + C++ 侧易误报)。
• pool4 类多线程基准在本机噪声达 ±13~25%,不能用于归因单次改动(见 benches/dispatch.rs 文件头;那里也记了本机系统调用被放大约 5 倍这件事)。