Lely CANopen 事件调度机制详解：从 ev_task、Executor、std_exec 到 ev_loop、Future 与 Poll

在这里插入图片描述

@[toc]

0. 阅读前先记住六个结论

  1. ev_task 只是“可执行工作”的载体，不负责排队、线程切换或等待。
  2. ev_exec_t 是抽象 Executor 接口，本质是 C 语言虚函数表。
  3. ev_std_exec 不是另一个事件循环，而是一个适配层：只要求后端实现 post/abort/on_task_init/on_task_fini，它补齐 dispatch/defer/run。
  4. ev_loop 才拥有真正的全局任务队列、等待线程、Poll 线程和 outstanding work 状态。
  5. ev_exec_run() 的关键价值不是简单调用回调，而是建立“当前线程正在这个 Executor 内执行任务”的 TLS 上下文。
  6. Future、I/O、定时器最终都不会直接“执行用户逻辑”；它们把 ev_task 投递给 Executor，由 ev_loop 取出并运行。

可以先把整个系统压缩成下面这条链：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
异步事件 / Future ready / 用户主动 post
                    │
                    ▼
                ev_task
                    │
                    ▼
      ev_exec_dispatch/post/defer
                    │
                    ▼
            ev_std_exec 语义层
                    │
                    ▼
        ev_loop 后端队列与唤醒逻辑
                    │
                    ▼
 ev_loop_wait_one() 从队列取出 task
                    │
                    ▼
             ev_exec_run(task)
                    │
                    ▼
              task->func(task)

1. 源码范围与文件职责

文件 主要职责 阅读优先级
include/lely/ev/ev.h 前置声明 ev_task，定义抽象类型 ev_exec_t 最高
include/lely/ev/task.h 定义 C 任务对象 ev_task 与任务队列辅助函数 最高
include/lely/ev/task.hpp C++ Task、临时 TaskWrapper 中
src/ev/task.c structof() 反查、批量 post/abort 高
include/lely/ev/exec.h Executor 虚函数表及 C API 最高
src/ev/exec.c 为头文件 inline API 提供外部定义 中
include/lely/ev/exec.hpp C++ Executor 包装 中
include/lely/ev/std_exec.h 标准 Executor 适配器及精简后端接口 最高
src/ev/std_exec.c dispatch/post/defer/run/abort 的核心语义 最高
include/lely/ev/loop.h Event loop 的公开行为与 outstanding work 定义 高
src/ev/loop.c 队列、线程、Poll、条件变量、Future wait 的落地实现 最高
include/lely/ev/future.h 非阻塞 Future/Promise 契约 高
src/ev/future.c Future 状态机、等待任务队列、计数闭环 高

2. 整体分层：谁定义语义，谁保存状态，谁真正执行

Lely 的事件库不是把所有功能堆在 ev_loop 中，而是分成四层：

1
2
3
4
5
6
7
8
9
flowchart TB
    A[业务层 / CANopen / I/O / Future] --> B[ev_task 工作对象]
    B --> C[ev_exec_t 抽象 Executor]
    C --> D[ev_std_exec 标准语义适配层]
    D --> E[ev_loop 精简后端实现]
    E --> F[全局 task queue]
    E --> G[Poll / condition variable]
    F --> H[ev_exec_run]
    H --> I[task->func]

2.1 四层职责不能混淆

层 解决的问题 不负责什么
ev_task “要执行哪个函数，属于哪个 Executor” 不保存线程，不主动入队
ev_exec_t “用统一 API 提交、运行、取消任务” 不规定具体队列结构
ev_std_exec “补齐 dispatch/defer/run 的标准行为” 不拥有全局事件循环队列
ev_loop “任务排队、唤醒、Poll、停止、线程协调” 不理解 CANopen 协议业务语义

因此，下面两句话都不准确：

1
2
“Future 自己执行回调”
“Poll 返回后直接执行 CANopen 用户回调”

更准确的描述是：

1
2
3
4
5
6
Future 或 Poll 产生完成事件
    → 形成或找到一个 ev_task
    → 向 Executor 投递
    → event loop 取出 task
    → Executor 建立执行上下文
    → task->func() 运行

3. ev_exec_t：用 C 指针模拟对象和虚函数

3.1 看似简单的 typedef，实际有两层指针语义

ev.h 中的核心定义是：

1
typedef const struct ev_exec_vtbl *const ev_exec_t;

这表示：

1
2
ev_exec_t
= 指向 const ev_exec_vtbl 的 const 指针类型

公开 API 使用的通常是：

1
ev_exec_t *exec;

所以 exec 并不是“直接指向 Executor 结构体”，而是：

1
指向 concrete object 内部 vptr 字段的地址

调用方式：

1
(*exec)->post(exec, task);

可以把它类比为：

1
C++ this 指针 + 虚函数表

3.2 对象布局

以 ev_std_exec 为例：

1
2
3
4
struct ev_std_exec {
    const struct ev_exec_vtbl *exec_vptr;
    ev_std_exec_impl_t *impl;
};

调用方拿到的是：

1
&std_exec->exec_vptr

内存关系如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
 ev_exec_t *exec
        │
        ▼
┌───────────────────────────────────────┐
│ struct ev_std_exec                    │
│  ┌─────────────────────────────────┐  │
│  │ exec_vptr ───────────────┐      │  │
│  └──────────────────────────│──────┘  │
│  ┌──────────────────────────│──────┐  │
│  │ impl                     │      │  │
│  └──────────────────────────│──────┘  │
└─────────────────────────────│─────────┘
                              ▼
                    struct ev_exec_vtbl

std_exec.c 通过：

1
structof(exec, struct ev_std_exec, exec_vptr)

从 exec_vptr 字段地址反推出完整 ev_std_exec 对象地址。

3.3 为什么不是普通函数指针

单个函数指针只能表示“如何执行”，而 Executor 需要一组统一操作：

1
2
3
4
5
6
7
8
9
struct ev_exec_vtbl {
    void   (*on_task_init)(ev_exec_t *exec);
    void   (*on_task_fini)(ev_exec_t *exec);
    int    (*dispatch)(ev_exec_t *exec, struct ev_task *task);
    void   (*post)(ev_exec_t *exec, struct ev_task *task);
    void   (*defer)(ev_exec_t *exec, struct ev_task *task);
    size_t (*abort)(ev_exec_t *exec, struct ev_task *task);
    void   (*run)(ev_exec_t *exec, struct ev_task *task);
};

这允许同一套上层代码连接不同 Executor：

  • polling event loop；
  • fiber executor；
  • 自定义线程池；
  • 测试用同步 Executor；
  • 其他平台后端。

3.4 exec.c 为什么只有几行

exec.h 中的 API 采用 static inline/inline 转调虚函数表。src/ev/exec.c 先定义：

1
#define LELY_EV_EXEC_INLINE extern inline

再包含 exec.h，主要作用是为 C inline 函数提供可链接的外部定义。

因此：

exec.c 几乎没有业务逻辑，真正的 Executor 行为在虚函数表对应的具体实现中。

4. ev_task：侵入式、零额外队列节点的任务对象

4.1 数据结构

1
2
3
4
5
6
struct ev_task {
    ev_exec_t *exec;
    ev_task_func_t *func;
    struct slnode _node;
    void *_data;
};
字段 含义
exec 该任务所属或将要提交到的 Executor
func 执行入口，签名为 void func(struct ev_task *)
_node 侵入式单链表节点，可直接挂到队列
_data 内部处理时可保存附加信息，不应随意假设用途

4.2 为什么使用侵入式链表

普通任务队列通常需要单独分配节点：

1
queue node → task object

Lely 直接把 _node 嵌入任务：

1
queue 直接保存 task._node

优点：

  1. 投递时不必额外 malloc；
  2. 队列操作开销固定；
  3. 适合实时性和嵌入式场景；
  4. Future 队列、event loop 队列、defer 临时队列可以复用同一节点。

代价也很明确：

同一个 ev_task 在同一时刻只能安全地位于一个链表中。

不要在任务仍在 Future 队列或 event loop 队列时再次提交同一个对象，否则 _node 链接关系会被破坏。

4.3 通过嵌入实现“继承”

典型 C 用法：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
struct read_task {
    struct ev_task task;
    int fd;
    size_t count;
};

static void
read_task_func(struct ev_task *task)
{
    struct read_task *self =
            structof(task, struct read_task, task);

    /* 使用 self->fd、self->count */
}

这是 Lely 中非常常见的对象模型：

1
2
3
基类对象嵌入派生结构
回调只拿到基类指针
通过 structof() 恢复派生对象

4.4 生命周期责任

ev_task 本身不带引用计数，也不会被 Executor 自动复制。

必须保证：

1
2
3
从提交任务开始
直到任务执行完成或被成功 abort
任务对象一直有效

栈上任务只有在调用方能保证任务在离开作用域前完成时才安全。异步提交通常应使用：

  • 长生命周期对象内嵌任务；
  • 堆对象；
  • Promise/Future 共享对象内嵌任务；
  • 驱动对象或 I/O 请求对象内嵌任务。

5. C++ Task 与 TaskWrapper

5.1 Task

task.hpp 的 Task 继承 ev_task，内部持有：

1
std::function<void()> func_;

构造函数把 C 回调入口设置为一个无捕获 lambda：

1
2
3
ev_task 回调
    → static_cast<Task *>(task)
    → 调用 func_()

Task 不自删除，因此其生命周期仍由调用方负责。

5.2 TaskWrapper

make_task_wrapper() 使用 new 创建临时任务。任务运行时：

1
2
调用 invoker_()
然后 delete self

这使 C++ API 可以直接写：

1
2
3
exec.post([] {
  // task body
});

但要记住：

临时 wrapper 是在“任务实际运行”时自删除；若在运行前被取消，必须确认当前 Lely 版本对该匿名任务的所有权约定，不能把“abort”自动等同于“delete”。

5.3 一个版本相关的源码阅读提示

在官方 v2.3.5 与当前 master 的 exec.hpp 中，Callable 版本的 Executor::dispatch() 声明返回 bool，函数体调用 dispatch(*task) 后未显式 return。

这不影响理解 Executor 主机制，但如果本地代码使用了这个重载的返回值，应检查本地版本、编译告警和上游修复情况。直接提交已有 ev_task 的重载没有这一问题。

6. Executor 七个操作的准确语义

6.1 总表

API 是否允许同步完成 当前任务结束前是否允许开始 主要用途
on_task_init 不适用 不适用 声明未来还有工作，防止 loop 提前退出
on_task_fini 不适用 不适用 释放先前声明的工作引用
dispatch 允许 允许，常在当前 Executor 上内联执行 “能立即执行就立即执行”
post 调用本身不等待完成 允许其他 worker 并发执行 普通异步投递
defer 调用本身不等待完成 若在当前任务内调用，禁止在当前任务结束前执行 continuation/避免重入
abort 不执行任务 不适用 移除尚未开始的任务
run 当前调用中执行 立即 建立 Executor 执行上下文并调用任务

6.2 dispatch

抽象契约是：

1
2
3
4
如果当前就在该 Executor 的执行上下文中
    可以直接运行任务并同步返回完成
否则
    退化为普通 post

标准 Executor 的返回值：

1
2
1：本次调用中任务已经完成
0：仅提交，尚未由本次调用完成

6.3 post

post 保证调用者不等待任务完成，但不保证任务一定等到函数返回之后才开始。

在 Lely ev_loop 后端中，post() 自己只是加锁入队；但如果另一个 worker 正在运行同一 loop，该 worker 可能在投递线程返回前就取走并执行任务。

6.4 defer

defer 的核心语义不是“低优先级”，而是：

1
2
3
4
5
如果当前正在这个 Executor 的某个任务中
    把新任务放到当前执行上下文的临时 defer 队列
    当前任务返回后再转交后端
否则
    直接退化为 post

它主要解决回调重入和 continuation 顺序问题。

6.5 run

run 不是通常意义上的“启动事件循环”。它的对象是单个任务：

1
ev_exec_run(exec, task);

其职责是：

  1. 标记当前线程正在 exec 上执行；
  2. 为本次任务建立 defer 队列；
  3. 调用 task->func(task)；
  4. 任务结束后，把 defer 队列转交后端；
  5. 恢复上一层嵌套执行上下文。

这是理解 std_exec.c 的核心。

7. ev_std_exec：为什么要在 Executor 与 event loop 之间再加一层

7.1 精简后端接口

ev_std_exec_impl_vtbl 只要求后端提供四个操作：

1
2
3
4
5
6
struct ev_std_exec_impl_vtbl {
    void   (*on_task_init)(ev_std_exec_impl_t *impl);
    void   (*on_task_fini)(ev_std_exec_impl_t *impl);
    void   (*post)(ev_std_exec_impl_t *impl, struct ev_task *task);
    size_t (*abort)(ev_std_exec_impl_t *impl, struct ev_task *task);
};

而对外的完整 Executor 有七个操作。

因此：

1
2
3
ev_std_exec
= 用 post + abort + work-count hooks
  实现完整 Executor 语义的通用适配器

这样 ev_loop 不必重复实现复杂的：

  • 当前线程是否在该 Executor 上执行；
  • dispatch 是否可以内联；
  • defer 应该挂在哪一层；
  • 嵌套 run 的 defer 队列如何恢复。

7.2 两套虚函数表

1
2
3
4
5
flowchart LR
    API[ev_exec_t API] --> V1[ev_exec_vtbl 完整接口]
    V1 --> S[ev_std_exec]
    S --> V2[ev_std_exec_impl_vtbl 精简接口]
    V2 --> L[ev_loop backend]

调用 ev_exec_post() 时的实际链路：

1
2
3
4
5
6
7
8
9
ev_exec_post(exec, task)
    ↓
(*exec)->post(exec, task)
    ↓
ev_std_exec_post()
    ↓
(*exec->impl)->post(exec->impl, task)
    ↓
ev_loop_std_exec_impl_post()

8. std_exec.c 最关键的数据结构：线程局部执行上下文

8.1 TLS 根节点

1
static _Thread_local struct ev_exec_node *ev_exec_list;

每个线程拥有独立的 ev_exec_list。

它回答一个问题：

1
当前线程此刻正在执行哪些 Executor 的任务？

8.2 两级结构

1
2
3
4
5
6
7
8
9
10
struct ev_task_node {
    struct ev_task_node *next;
    struct sllist *queue;
};

struct ev_exec_node {
    struct ev_exec_node *next;
    ev_exec_t *exec;
    struct ev_task_node *queue;
};

关系如下：

1
2
3
4
5
6
7
8
9
10
11
线程 TLS ev_exec_list
    │
    ├─ Executor A
    │    ├─ 最内层 run 的 defer queue
    │    ├─ 外一层 run 的 defer queue
    │    └─ ...
    │
    ├─ Executor B
    │    └─ 当前 run 的 defer queue
    │
    └─ ...

8.3 为什么一个 Executor 还需要多层 queue

同一线程可能出现嵌套 ev_exec_run()：

1
2
3
run(task A)
    └─ 任务 A 内部显式 run(task B)
           └─ 任务 B defer(task C)

任务 C 应属于 B 这一层的 continuation，而不是直接混到 A 的 defer 队列中。

所以同一个 ev_exec_node 下还有 ev_task_node 栈。

8.4 不同 Executor 的嵌套

1
2
Executor A 正在运行 task A
    └─ task A 调用 Executor B.run(task B)

线程 TLS 中会同时出现 A、B 两个 ev_exec_node。ev_exec_find() 按 Executor 地址查找对应节点。

9. ev_std_exec_run()：完整执行边界

其核心过程可简化为：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
void run(exec, task)
{
    建立本次 defer queue;
    把当前 exec 标记进 TLS;

    task->func(task);

    从 TLS 弹出本次执行上下文;

    while (defer queue 非空) {
        impl->post(task);
        on_task_fini();
    }
}

9.1 为什么 defer 入队时要 on_task_init()

defer 任务暂时只存在于线程栈上的局部队列中，还没有进入 ev_loop->queue。

若不登记 outstanding work：

1
2
3
其他 loop 线程看到全局队列为空
并且 ntasks == 0
    → 可能停止 loop

因此 ev_std_exec_defer() 在加入局部队列前执行：

1
ev_std_exec_on_task_init(exec);

9.2 为什么 flush 时必须先 post，再 fini

源码顺序是：

1
2
impl->post(impl, task);
ev_std_exec_on_task_fini(exec);

不能随意交换。

正确的状态迁移：

1
2
3
4
5
6
任务在 defer 临时队列
    │ 由 on_task_init 保活
    ▼
先进入 event loop 全局队列
    ▼
再释放临时 work 引用

若先 fini：

1
2
3
4
ntasks 可能先变 0
全局队列此时仍为空
loop 被 stop
随后才 post

这会形成提前停止窗口。

同样的“先 post，后 fini”原则也出现在：

  • ev_task_queue_post()；
  • Future ready 后转移等待任务；
  • Future cancel 后提交任务。

10. dispatch、post、defer 在 std_exec 中怎样实现

10.1 dispatch

1
2
3
查找当前线程 TLS 中是否存在该 exec
    ├─ 不存在：调用 post，返回 0
    └─ 存在：直接 task->func(task)，返回 1

关键细节：

dispatch() 的内联分支直接调用 task->func()，不会再创建新的 ev_exec_run() defer 边界。

因此，内联 dispatch 的任务与当前外层任务共享当前最内层 defer queue。

例如：

1
2
3
4
5
6
outer task
    dispatch(inner task)
        inner task defer(D)
    outer task 继续执行
outer task 返回
    才 flush D

D 不是在 inner task 返回后立即 flush，而是在当前 ev_exec_run() 边界结束后 flush。

10.2 post

1
2
如果 task->exec 为空，绑定当前 exec
调用后端 impl->post()

对 ev_loop 后端而言就是进入全局 FIFO 单链表。

10.3 defer

1
2
3
4
5
6
查找当前线程是否正在该 exec 上运行
    ├─ 否：退化为 post
    └─ 是：
         绑定 task->exec
         on_task_init()
         加到当前最内层 defer queue 尾部

10.4 abort

标准 Executor 先检查当前线程该 Executor 的局部 defer queue：

1
2
3
4
找到待取消任务
    → 从 defer queue 删除
    → on_task_fini()
    → 不再调用后端 abort

若未找到，或要求取消全部任务，则再调用后端 impl->abort()。

这体现了一个重要事实：

1
2
3
尚未开始的任务可能位于两处：
1. 当前线程的 defer 临时队列
2. 后端 event loop 全局队列

11. 一个最直观的执行顺序示例

假设单线程 event loop 正在运行 outer：

1
2
3
4
5
void outer() {
  exec.defer(C);
  exec.dispatch(A);
  exec.post(B);
}

执行过程：

1
2
3
4
5
6
7
8
9
10
11
1. outer 开始
2. defer(C)
   C 进入 outer 的局部 defer queue
3. dispatch(A)
   A 立即内联执行
4. post(B)
   B 立即进入 event loop 全局 queue
5. outer 返回
6. ev_std_exec_run() flush C
   C 被追加到 event loop 全局 queue
7. event loop 后续依次执行 B、C

单线程且没有其他已有任务时，顺序通常为：

1
outer → A → B → C

多线程时，B 进入全局队列后可能被其他 worker 并发执行，甚至在 outer 返回前完成；C 仍不会在 outer 的当前 run 边界结束前被提交。

12. ev_loop 怎样接入 ev_std_exec

12.1 event loop 内嵌标准 Executor

struct ev_loop 中包含：

1
2
const struct ev_std_exec_impl_vtbl *impl_vptr;
struct ev_std_exec exec;

初始化时：

1
2
3
4
5
loop->impl_vptr = &ev_loop_std_exec_impl_vtbl

loop->exec
    通过 ev_std_exec_init()
    连接到 &loop->impl_vptr

所以：

1
ev_loop_get_exec(loop)

返回的是 event loop 内嵌 Executor 的接口地址，而不是新建对象。

12.2 event loop 只实现四个后端操作

1
2
3
4
5
6
7
static const struct ev_std_exec_impl_vtbl
        ev_loop_std_exec_impl_vtbl = {
    on_task_init,
    on_task_fini,
    post,
    abort
};

其余 dispatch/defer/run 都由 ev_std_exec 提供。

13. event loop 后端四个操作

13.1 on_task_init：增加额外 work 引用

多线程正常构建中：

1
2
atomic_fetch_add_explicit(&loop->ntasks, 1,
                          memory_order_relaxed);

其语义不是“队列中新增一个任务”，而是：

1
2
还有一份尚未进入 loop queue 的异步工作
请不要因队列暂时为空而退出

13.2 on_task_fini：释放 work 引用

当最后一个引用从 1 变 0：

  1. 建立 acquire 屏障；
  2. 获取 loop->mtx；
  3. 再检查真实任务队列；
  4. 队列为空时调用 ev_loop_do_stop()。

这里采用典型的：

1
2
原子快路径
+ 最后一个引用进入加锁慢路径

13.3 post：加入全局队列并唤醒一个线程

1
2
3
4
5
6
lock loop->mtx
记录投递前是否为空
push_back(loop->queue, task)
若发生 空 → 非空
    ev_loop_kill_any(loop, 1)
unlock

只在空队列首次获得任务时唤醒，可减少无效 signal/kill。

13.4 abort

后端 abort 从 loop->queue 移除指定任务或全部真实任务。

若启用了 poll_task，这个伪任务会被保留或重新放回队列，因为它是 event loop 的调度哨兵，不是普通用户任务。

14. poll_task：不是用户任务，而是避免 Poll 饥饿的调度哨兵

当 poll_task 启用时，event loop 队列中会长期存在一个特殊 loop->task。

其目的：

1
2
3
即使 CPU 任务持续不断
也定期给 Poll 一次机会
避免 I/O readiness 长期得不到处理

关键规则：

  • ev_loop_empty() 认为“队列中仅有 poll_task”仍然等价于没有真实任务；
  • wait 循环弹出它后不调用 task->func()；
  • 本轮结束时重新把它放回队列；
  • abort-all 时必须保留它。

因此，不能把：

1
sllist_empty(loop->queue)

简单等同于：

1
没有业务工作

源码专门通过 ev_loop_empty() 统一判断。

15. ev_loop_wait_one()：任务真正开始运行的桥

核心路径：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
持有 loop->mtx
    ↓
从 loop->queue 头部弹出一个真实 task
    ↓
释放 loop->mtx
    ↓
ev_exec_run(task->exec, task)
    ↓
ev_std_exec_run()
    ↓
建立 TLS 执行上下文
    ↓
task->func(task)
    ↓
flush defer queue
    ↓
重新取得 loop->mtx

为什么执行任务前必须释放 loop->mtx：

  1. 用户回调可能再次 post/defer/abort；
  2. 回调可能启动新的异步操作；
  3. 回调可能调用 Future、I/O、CANopen API；
  4. 长时间持锁会阻塞其他 worker 和 I/O 完成线程；
  5. 若回调内部再次访问 loop，会产生死锁风险。

16. 队列为空后：Poll 还是条件变量

当没有真实任务可执行，但仍有 outstanding work 时：

1
2
3
4
5
当前线程有 Poll 资格
    → ev_poll_wait()

当前线程没有 Poll 资格
    → cnd_wait()

16.1 Poll 线程

负责等待外部事件，例如：

  • SocketCAN fd 可读/可写；
  • timerfd；
  • signal fd/pipe；
  • 平台 I/O completion。

Poll 返回后，后端回调通常继续向 Executor post() 任务。

16.2 非 Poll worker

在 npoll 已达到上限时，其他 worker 使用条件变量等待内部事件：

  • 新任务入队；
  • loop stop；
  • ev_loop_kill()；
  • Future ready。

16.3 为什么优先唤醒 waiting worker

ev_loop_kill_any() 先查 waiting 链表，再查 polling 链表。

这样可以：

  • 让条件变量 worker 处理新入队任务；
  • 尽量保持 Poll 线程继续等待外部 I/O；
  • 减少不必要地打断 epoll_pwait()。

17. outstanding work：不能只看任务队列长度

公开文档将 outstanding work 定义为：

1
2
3
4
待执行任务
+ 当前执行任务
+ on_task_init 调用次数
- on_task_fini 调用次数

在 loop.c 的停止判断中，最关键的两个观测量是：

1
2
ev_loop_empty(loop)
ev_loop_ntasks(loop)

其中 ntasks 主要承担“尚未进入全局队列的工作引用”，例如：

  • Future 等待队列中的任务；
  • 当前任务的 defer 临时队列；
  • 正在等待 I/O 完成、未来才会提交回调的异步操作；
  • 其他显式 work guard。

17.1 为什么计数必须严格成对

少调用一次 fini：

1
2
loop 永远认为还有工作
run() 无法自然退出

多调用一次 fini：

1
2
计数下溢或提前归零
loop 可能提前 stop

17.2 计数迁移原则

任务在多个“未执行位置”间迁移时，应保持连续保活：

1
2
3
4
5
6
临时队列 / Future queue
    │ 由 on_task_init 保活
    ▼
先 post 到 Executor 后端
    ▼
再 on_task_fini

这个顺序是整个系统正确性的关键不变量之一。

18. task.c 两个批处理函数为什么重要

18.1 ev_task_queue_post()

1
2
3
逐个从临时 queue 取任务
    → ev_exec_post(task->exec, task)
    → ev_exec_on_task_fini(task->exec)

用途：

1
2
把“由 work 引用保护的等待队列”
安全迁移到真正 Executor 队列

18.2 ev_task_queue_abort()

1
2
3
逐个丢弃任务
    → 不调用 task->func
    → 只调用 on_task_fini

用途：

1
2
任务不再执行
但必须释放此前占用的 outstanding work

18.3 为什么 post 在 fini 前

再次强调：

1
2
ev_exec_post(exec, task);
ev_exec_on_task_fini(exec);

这是为了让任务在任何时刻至少满足一个条件：

1
2
要么由额外 work 引用保活
要么已经位于 Executor 的真实队列中

19. Future 怎样连接 Executor

Lely Future 是非阻塞 Future：

1
2
Future::get() 不负责等待
Future ready 后提交预先注册的 task

19.1 ev_future_submit()

核心流程：

1
2
3
4
5
6
7
8
9
取得 task->exec
    ↓
on_task_init(exec)
    ↓
lock future->mtx
    ↓
Future 已 ready？
    ├─ 是：unlock → post(task) → on_task_fini()
    └─ 否：task 加入 future->queue → unlock

为什么注册 Future 等待时就要 on_task_init()：

1
2
3
任务此时不在 event loop queue
但它未来一定可能被提交
loop 必须继续存活

19.2 Promise 完成

ev_promise_set_release()：

  1. 写入结果指针；
  2. 锁定 Future；
  3. 把等待任务整体转移到局部队列；
  4. release-store 状态为 READY；
  5. 解锁；
  6. 调用 ev_task_queue_post()。

因此每个等待任务完成如下迁移：

1
2
3
4
5
6
Future queue
  由 on_task_init 保活
        ↓
Executor queue
        ↓
on_task_fini 释放 Future 阶段引用

19.3 cancel 与 abort

操作 从 Future queue 移除 是否继续执行任务 work 引用
ev_future_cancel() 是 是，立即 post 释放
ev_future_abort() 是 否 释放

所以：

1
2
cancel ≠ 保证回调不执行
abort 才是不再提交该等待任务

20. ev_loop_wait(loop, future) 并不是睡眠等待 Future

当 loop 等待一个 Future 时，会创建 ev_loop_ctx，其中内嵌：

1
struct ev_task task;

该任务的 Executor 是：

1
ev_loop_get_exec(loop)

然后：

1
ev_future_submit(future, &ctx->task);

Future ready 后：

1
2
3
4
5
6
7
8
9
10
11
12
13
Promise 完成
    ↓
Future 把 ctx->task post 到 loop Executor
    ↓
loop 取出 ctx->task
    ↓
ev_loop_ctx_task_func()
    ↓
ctx->ready = 1
    ↓
唤醒 Poll 或 cnd wait
    ↓
wait 主循环观察 ready 并返回

因此 ev_loop_wait() 的真实含义是：

1
2
持续驱动同一个 event loop
直到指定 Future ready 或其他退出条件成立

它在等待期间仍会：

  • 运行其他任务；
  • 处理 SocketCAN；
  • 推进 SDO/NMT 状态机；
  • 处理定时器；
  • 执行超时回调。

21. 从异步 SDO 到 Future ready 的完整链路

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
sequenceDiagram
    participant App as 应用程序
    participant Co as CANopen SDO 客户端
    participant Io as SocketCAN/I-O
    participant Poller as ev_poll
    participant EvLoop as ev_loop
    participant Exec as ev_std_exec
    participant Fut as Future/Promise

    App->>Co: 发起异步 SDO，取得 Future
    Co->>Io: 注册 CAN 接收/超时工作
    Co->>EvLoop: on_task_init 或异步工作保活
    App->>EvLoop: wait(future) / run()
    EvLoop->>Poller: ev_poll_wait()
    Io-->>Poller: CAN fd ready
    Poller->>EvLoop: post 接收任务
    EvLoop->>Exec: ev_exec_run(task)
    Exec->>Co: 执行任务，推进 SDO 状态机
    Co->>Fut: Promise set success/error
    Fut->>EvLoop: post ctx.task
    EvLoop->>Exec: run ctx.task
    Exec->>EvLoop: ctx.ready = 1 并唤醒等待
    EvLoop-->>App: wait 返回
    App->>Fut: 读取 Result

Mermaid 兼容性说明：参与者的内部标识使用 EvLoop，显示名称仍通过 as ev_loop 保持不变。不要将 Loop/loop 用作 sequence diagram 的参与者标识，因为 loop ... end 是 Mermaid 的循环控制块语法，部分解析器会按关键字处理并产生解析错误。

这条链路解释了一个常见问题：

如果没有其他线程驱动 loop，当前线程也没有调用 run()/wait()，SDO Future 不会凭空完成，因为 CAN I/O 和超时任务无人执行。

22. 从 Executor::post() 到任务执行的完整调用链

22.1 C++ 入口

1
2
3
4
auto exec = loop.get_executor();
exec.post([] {
  // user task
});

22.2 调用链

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
Executor::post(callable)
    ↓
make_task_wrapper()
    ↓
Executor::post(ev_task&)
    ↓
ev_exec_post()
    ↓
ev_std_exec_post()
    ↓
ev_loop_std_exec_impl_post()
    ↓
loop->queue.push_back(task)
    ↓
ev_loop_kill_any()
    ↓
某个 run/wait 线程醒来
    ↓
ev_loop_ctx_wait_one()
    ↓
ev_exec_run(task->exec, task)
    ↓
ev_std_exec_run()
    ↓
TaskWrapper invoker
    ↓
delete wrapper

23. 从 defer() 到任务执行的完整调用链

当正在 loop task 内调用 defer()：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
ev_exec_defer()
    ↓
ev_std_exec_defer()
    ↓
在 TLS 找到当前 exec_node
    ↓
on_task_init()
    ↓
加入本次 run 的局部 defer queue
    ↓
当前 task->func 返回
    ↓
ev_std_exec_run() drain defer queue
    ↓
impl->post(task)
    ↓
on_task_fini()
    ↓
任务进入 loop 全局队列

当不在该 Executor 的执行上下文中调用：

1
defer() → post()

因此，不要把 defer() 理解为固定延迟或定时器 API。

24. 从 dispatch() 到任务执行的完整调用链

24.1 当前不在该 Executor 上

1
2
3
4
5
6
7
dispatch(task)
    ↓
TLS 中找不到 exec
    ↓
post(task)
    ↓
返回 0

24.2 当前正在该 Executor 上

1
2
3
4
5
6
7
dispatch(task)
    ↓
TLS 中找到 exec
    ↓
直接 task->func(task)
    ↓
返回 1

这会产生同步重入。若不希望当前调用栈中立即进入下一个回调，应使用 defer() 或 post()。

25. 多线程模型下的准确理解

假设：

1
2
npoll = 1
线程 A、B 都调用 loop.run()

可能状态：

1
2
线程 A：进入 ev_poll_wait()
线程 B：进入 cnd_wait()

线程 C post(task)：

1
2
3
4
loop queue 空 → 非空
优先 signal waiting 线程 B
B 醒来执行 task
A 保持 Poll

若没有 waiting worker：

1
2
3
4
post(task)
    → ev_poll_kill(A)
    → A 的 Poll wait 被中断
    → A 回到 loop 并执行 task

25.1 多线程中的 post 可并发执行

即使 post() 本身不内联调用任务，另一个 worker 也可以立即取走任务。

因此下面写法存在竞态：

1
2
exec.post([&] { use(value); });
value = new_value;  // 不能假设一定先于回调

需要通过锁、原子变量、消息所有权或把数据封装进任务对象来保证同步。

25.2 defer 只约束当前执行边界

defer() 保证：

1
不会在当前 task 返回前，通过当前 defer 路径开始执行

但它被 flush 到全局队列后，仍可能由任意 worker 执行。

它不是线程亲和性机制。

26. C 语言最小任务示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
#include <lely/ev/exec.h>
#include <lely/ev/task.h>
#include <lely/util/util.h>

struct app_task {
    struct ev_task task;
    int value;
};

static void
app_task_func(struct ev_task *task)
{
    struct app_task *self =
            structof(task, struct app_task, task);

    /* 处理 self->value。 */
}

static void
submit_app_task(ev_exec_t *exec, struct app_task *task, int value)
{
    task->value = value;
    task->task = (struct ev_task)EV_TASK_INIT(exec, app_task_func);
    ev_exec_post(exec, &task->task);
}

注意：

  1. struct app_task 必须保持有效直到任务完成；
  2. 未完成前不能再次提交同一个 _node；
  3. 若任务可能被 abort，调用方仍需负责对象生命周期；
  4. 回调内可通过 task->exec 获取所属 Executor。

27. C++ 最小示例

1
2
3
4
5
6
7
8
9
10
#include <lely/ev/loop.hpp>

lely::ev::Loop loop;
auto exec = loop.get_executor();

exec.post([] {
  // 异步任务
});

loop.run();

若任务需要稳定对象生命周期，优先显式定义 lely::ev::Task 或把任务嵌入业务对象，而不是无条件依赖匿名临时 wrapper。

28. dispatch/post/defer 选择表

需求 建议接口 原因
当前已在 Executor 上，允许立即同步调用 dispatch 减少入队和上下文切换
必须异步排队 post 不在当前调用栈内直接执行
必须等当前任务完整返回后再提交 continuation defer 避免当前 task 内重入
Future ready 后运行回调 ev_future_submit 自动处理等待队列与 work 引用
只想直接调用函数，不需要 Executor 语义 直接函数调用 不应滥用 ev_exec_run

28.1 常见误用

误用一：把 dispatch 当作普通异步 post

当前已在 Executor 上时，它会同步重入。

误用二：把 defer 当作毫秒级延迟

它不使用时钟，不提供 deadline。

误用三：手工调用 on_task_fini，但没有对应 init

会破坏 outstanding work 计数。

误用四：Future 等待任务使用空 task->exec

ev_future_submit() 要求任务已绑定 Executor。

误用五：重复提交同一个 task

侵入式 _node 不能同时挂在多个队列。

29. 三类“队列”必须分清

队列 所属对象 何时进入 如何离开
event loop 全局 queue ev_loop post() loop worker 弹出并 run()
defer 临时 queue TLS ev_task_node 当前任务内 defer() 当前 ev_exec_run() 返回时 flush
Future 等待 queue ev_future ev_future_submit() 且尚未 ready Promise ready、cancel 或 abort

一个任务在不同队列间迁移时仍使用同一个 _node，所以迁移必须先从旧队列移除，再加入新队列。

30. 三类状态也必须分清

状态 范围 含义
loop->stopped 整个 event loop 后续 run/wait 立即停止，需 restart
ev_loop_thrd.stopped 当前 OS 线程 中断该线程最内层一次 loop 调用
ctx->ready 一次 Future wait context 等待目标已 ready

Executor 的 defer TLS 又是另一套状态：

1
2
ev_exec_list
= 当前线程正在执行哪些 Executor 及其嵌套 run 层级

它与 ev_loop_thrd 的 context 栈职责不同：

TLS 状态 管什么
ev_exec_list Executor 执行上下文、dispatch/defer/run
ev_loop_thrd loop wait/run 嵌套、中断与 Future wait context

31. 一张对象关系总图

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
flowchart TB
    subgraph LoopObject[ev_loop_t]
        LV[impl_vptr]
        SE[struct ev_std_exec exec]
        Q[global task queue]
        NT[ntasks]
        W[waiting contexts]
        P[polling contexts]
    end

    subgraph StdExec[ev_std_exec]
        EV[exec_vptr]
        IMPL[impl pointer]
    end

    subgraph ThreadTLS[每线程 TLS]
        EL[ev_exec_list]
        LT[ev_loop_thrd]
        DQ[defer queue stack]
        LC[loop ctx stack]
    end

    T[ev_task] -->|exec| EV
    EV -->|full vtable| StdExec
    IMPL -->|reduced vtable| LV
    LV --> LoopObject
    Q -->|pop| T
    EL --> DQ
    LT --> LC
    F[ev_future queue] -->|ready: post then fini| Q
    NT -->|0 and queue empty| STOP[loop stop]
    P --> POLL[ev_poll_wait]
    W --> CND[cnd_wait]

32. 一张完整调用链总图

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
flowchart TD
    S[事件发生] --> K{来源}
    K -->|用户提交| POST[ev_exec_post]
    K -->|当前任务 continuation| DEF[ev_exec_defer]
    K -->|Future ready| FQ[ev_task_queue_post]
    K -->|I/O ready| IO[I/O watch callback]

    DEF --> TLSQ[TLS defer queue]
    TLSQ -->|current run returns| POST
    FQ --> POST
    IO --> POST

    POST --> STD[ev_std_exec_post]
    STD --> LPOST[ev_loop impl post]
    LPOST --> GQ[loop global queue]
    GQ --> WAKE[wake cnd waiter or poller]
    WAKE --> WAIT[ev_loop_wait_one]
    WAIT --> POP[pop task]
    POP --> RUN[ev_exec_run]
    RUN --> SRUN[ev_std_exec_run]
    SRUN --> FUNC[task func]
    FUNC -->|may dispatch| INLINE[inline task func]
    FUNC -->|may defer| TLSQ
    FUNC -->|may post| POST

33. 调试时建议重点观察的变量

变量 观察目的
task->exec 任务是否绑定到预期 Executor
task->_node.next 是否可能重复入队或链表损坏
loop->queue 当前真实待执行任务
loop->ntasks 是否存在未进入 queue 的保活工作
loop->stopped run 不执行是否因为 loop 已停止
loop->npolling Poll 线程数量是否达到上限
ev_exec_list 当前线程是否处于该 Executor 上
exec_node->queue 当前最内层 defer queue
future->queue Future 是否仍持有等待任务
future->state WAITING/SETTING/READY
ctx->ready ev_loop_wait(future) 是否收到 ready 任务

33.1 任务“提交了但不执行”排查顺序

  1. task->exec 是否为 NULL；
  2. task 是否真的进入 loop->queue；
  3. loop 是否已经 stopped；
  4. 是否有线程正在调用 run/wait；
  5. 是否误把任务留在 Future queue；
  6. on_task_init/fini 是否失衡；
  7. 是否重复使用同一 _node；
  8. worker 是否全部阻塞在非预期锁中；
  9. Poll/condition variable 是否被正确唤醒；
  10. 回调是否被 abort 移除。

34. 与 CANopen 业务层的对应关系

Lely CANopen 上层异步接口通常可以映射为：

CANopen 行为 Event/Future 层动作
发起 SDO 请求 创建请求对象、Promise/Future、注册 I/O/超时
收到 SDO 响应 Poll 回调 post 接收任务
推进协议状态机 loop 执行 task
SDO 成功或失败 Promise set，Future READY
AsyncWrite() 返回 返回 Future 句柄，不阻塞
Wait(future) coroutine 或 loop 持续驱动，直到 ready
NMT/Heartbeat 定时事件 timer readiness → post task → 状态机回调
PDO 接收 CAN fd ready → receive task → OD/driver callback

CANopen 协议本身是“被动、异步”的，线程与 I/O 调度由 liblely-ev/liblely-io2 提供。理解 Executor 后，再读 SDO、NMT、PDO 代码会清晰很多。

35. 推荐源码阅读顺序

不要直接从 1000 多行的 loop.c 从头看到尾。建议按下面顺序：

  1. ev.h：看懂 ev_exec_t；
  2. task.h：看懂任务对象和侵入式节点；
  3. exec.h：掌握七个操作契约；
  4. std_exec.h：理解“完整接口/精简后端”两层虚表；
  5. std_exec.c：重点读 run/dispatch/defer 与 TLS；
  6. task.c：理解 post-then-fini 迁移原则；
  7. loop.c：先看后端四函数，再看 wait_one；
  8. future.c：重点读 submit/set_release/cancel/abort；
  9. poll.h 与 Linux poll 后端；
  10. 最后再追到 SocketCAN、Timer、SDO/NMT。

建议第一次阅读时只围绕三条调用链做笔记：

1
2
3
4
5
post → queue → run

defer → TLS queue → post → queue → run

Future submit → Future queue → post → queue → run

36. 最终心智模型

可以把 Lely 事件机制想成一个物流系统：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
ev_task
= 包裹

ev_exec_t
= 统一寄件接口

ev_std_exec
= 根据“立即送、普通送、当前任务后再送”处理寄件语义

ev_loop->queue
= 中央分拣队列

ev_exec_run
= 建立本次派送工作上下文

Future queue
= 等待某条件后才能转运的暂存区

defer queue
= 当前处理单结束后再转运的线程本地暂存区

on_task_init/fini
= 暂存区占用的未完成工作凭证

Poll
= 等外部世界产生新包裹

condition variable
= 等进程内部其他线程送来新包裹

最重要的不变量是：

1
2
3
4
5
一份尚未完成的异步工作，必须满足至少一个条件：

1. 已经位于可执行队列；
2. 正在执行；
3. 由 on_task_init/fini 的 outstanding work 引用保活。

而任务从 Future/defer 暂存区迁移到 Executor 队列时必须遵守：

1
2
先 post
后 fini

一旦掌握这两个原则，Lely 中大部分“为什么 loop 没退出”“为什么任务没执行”“为什么要有 ntasks”“为什么还要 std_exec”的问题就能统一解释。

37. 官方源码索引

37.1 稳定版本 v2.3.5

  1. ev.h
    https://gitlab.com/lely_industries/lely-core/-/blob/v2.3.5/include/lely/ev/ev.h
  2. task.h
    https://gitlab.com/lely_industries/lely-core/-/blob/v2.3.5/include/lely/ev/task.h
  3. task.hpp
    https://gitlab.com/lely_industries/lely-core/-/blob/v2.3.5/include/lely/ev/task.hpp
  4. task.c
    https://gitlab.com/lely_industries/lely-core/-/blob/v2.3.5/src/ev/task.c
  5. exec.h
    https://gitlab.com/lely_industries/lely-core/-/blob/v2.3.5/include/lely/ev/exec.h
  6. exec.hpp
    https://gitlab.com/lely_industries/lely-core/-/blob/v2.3.5/include/lely/ev/exec.hpp
  7. exec.c
    https://gitlab.com/lely_industries/lely-core/-/blob/v2.3.5/src/ev/exec.c
  8. std_exec.h
    https://gitlab.com/lely_industries/lely-core/-/blob/v2.3.5/include/lely/ev/std_exec.h
  9. std_exec.c
    https://gitlab.com/lely_industries/lely-core/-/blob/v2.3.5/src/ev/std_exec.c
  10. loop.h
    https://gitlab.com/lely_industries/lely-core/-/blob/v2.3.5/include/lely/ev/loop.h
  11. loop.c
    https://gitlab.com/lely_industries/lely-core/-/blob/v2.3.5/src/ev/loop.c
  12. future.h
    https://gitlab.com/lely_industries/lely-core/-/blob/v2.3.5/include/lely/ev/future.h
  13. future.c
    https://gitlab.com/lely_industries/lely-core/-/blob/v2.3.5/src/ev/future.c
  14. future.hpp
    https://gitlab.com/lely_industries/lely-core/-/blob/v2.3.5/include/lely/ev/future.hpp

37.2 官方文档

  1. Library overview
    https://opensource.lely.com/canopen/docs/overview/
  2. Build configuration
    https://opensource.lely.com/canopen/docs/configuration/
  3. C++ tutorial
    https://opensource.lely.com/canopen/docs/cpp-tutorial/
  4. Lely CANopen homepage
    https://opensource.lely.com/canopen/
  5. Mermaid sequence diagram syntax
    https://mermaid.js.org/syntax/sequenceDiagram.html

38. 建议配套实验

为了把源码机制真正学会，建议依次做四个小实验：

实验一：验证 dispatch/post/defer 顺序

在一个 loop task 内分别调用三种接口，打印：

1
2
3
4
线程 ID
任务名称
进入时间
离开时间

先单线程，再双线程运行 loop。

实验二：故意制造 work 引用不平衡

仅在测试程序中：

1
调用 on_task_init 但不 fini

观察 run() 是否无法自然退出。然后补上 fini 验证恢复。

实验三：Future ready 前后提交任务

分别在：

1
2
Promise set 前 submit
Promise set 后 submit

观察两条路径都能进入 Executor，但前者经过 Future queue，后者直接 post。

实验四：npoll = 1 双 worker

记录哪个线程进入 Poll，哪个线程进入条件变量，以及普通任务入队后优先唤醒哪个线程。

完成这四个实验后，再追踪 SocketCAN 接收、SDO Future 和定时器，理解会稳定得多。