Lely CANopen ev_loop 详解

在这里插入图片描述

@[toc]

1. 先区分三个容易混淆的“停止”状态

loop.c 中至少存在三类停止或完成状态：

状态 所属范围 作用
loop->stopped 每个 ev_loop_t 整个事件循环是否处于 stopped 状态；ev_loop_stop() 会影响所有正在运行该 loop 的线程。
ev_loop_thrd.stopped 每个线程 中断该线程当前最内层的一次 run/wait 调用；返回后通常会清零，以便下次重新运行。
ctx->ready 每个等待 context 表示正在等待的 future 已经 ready，应结束本次等待。

不能把它们当作同一个状态：

1
2
3
4
5
6
7
8
loop->stopped
= loop 级停止

thread-local stopped
= 线程级、单次运行中断

ctx->ready
= 本次等待目标已经完成

ev_loop_ctx_wait_one() 的主循环实际同时检查这些条件：

1
2
3
4
5
loop 没有全局停止
并且
当前线程没有被 kill
并且
future 尚未 ready

2. ntasks：outstanding work 计数

2.1 统计边界：不等于任务链表长度

源码注释把 ntasks 描述为 pending work 计数，其语义可以理解为：

1
事件循环已知但尚未完成的工作数量

公开接口文档进一步定义 outstanding work 为：

1
2
3
4
待执行任务
+ 当前执行中的任务
+ ev_exec_on_task_init() 调用次数
- ev_exec_on_task_fini() 调用次数

ev_exec_on_task_init() 的典型意义是：

我现在还没有把任务放入队列，但未来会提交工作，事件循环不要因为暂时空队列而退出。

ev_exec_on_task_fini() 表示：

先前声明的这份 outstanding work 已经结束，不再需要阻止事件循环退出。

因此，ntasks 的关键价值是覆盖“任务尚未入队，但异步操作仍然存活”的窗口。

2.2 队列为空不代表异步工作已经结束

假设一个异步 CAN 读取操作已经启动，但当前 SocketCAN 没有数据：

1
2
3
任务队列暂时为空
SocketCAN read 正等待 I/O
未来收到 CAN 帧后才会投递完成任务

如果 event loop 只看任务队列，就可能错误判断：

1
queue 为空 → 没有工作 → stop

而 ntasks > 0 表示仍存在已登记的异步工作，因此 loop 应继续等待 Poll 或条件变量事件。

2.3 ntasks 的状态变化

1
2
3
4
5
6
7
8
stateDiagram-v2
    [*] --> Zero: 初始化为 0
    Zero --> Active: on_task_init 0→1
    Active --> Active: on_task_init / on_task_fini 后仍大于 0
    Active --> LastDone: on_task_fini 1→0
    LastDone --> StopCheck: 加锁检查任务队列
    StopCheck --> Zero: 队列仍有任务
    StopCheck --> Stopped: 队列为空，调用 ev_loop_do_stop

关键代码路径是：

1
2
3
4
5
6
7
8
on_task_init:
    ntasks 原子加一

on_task_fini:
    ntasks 原子减一
    若减前为 1，即减后为 0：
        获取 loop->mtx
        若任务队列为空：停止 loop

3. ntasks 的线程安全与原子内存序

3.1 跨线程并发更新是原子化的根本原因

ev_exec_on_task_init() 和 ev_exec_on_task_fini() 是 executor 的公开生命周期接口。调用它们的线程不一定是正在运行 ev_loop_wait_one() 的线程，也不保证已经持有 loop->mtx。

典型并发场景：

1
2
3
4
线程 A：启动异步读，调用 on_task_init()
线程 B：启动异步写，调用 on_task_init()
线程 C：某个异步操作结束，调用 on_task_fini()
线程 D：event loop 读取 ntasks 判断是否应停止

如果使用普通 size_t 且没有统一互斥锁，会出现 C 语言层面的数据竞争，行为未定义。

3.2 非原子操作会出现丢失更新

假设 ntasks == 5，两个线程同时执行普通的 ++：

1
2
3
4
线程 A 读取 5
线程 B 读取 5
线程 A 写入 6
线程 B 写入 6

正确结果应为 7，实际可能为 6。

对于 ntasks，计数错误会带来两类严重后果：

错误方向 后果
少计 loop 可能在异步操作尚未结束时提前 stop。
多计 loop 可能一直认为还有 outstanding work，无法自然退出。

3.3 未统一使用 loop->mtx 的原因

理论上可以把所有 ntasks 读写都放进 loop->mtx，但当前实现选择原子计数，原因可以从代码行为推断为：

  1. on_task_init() 是高频轻量路径，只做原子加一，不需要锁任务队列；
  2. on_task_fini() 在减后仍不为 0 时，也不需要进入互斥锁；
  3. 只有最后一个 outstanding work 消失时，才加锁检查队列并决定是否 stop；
  4. 避免每次工作引用变化都与任务队列竞争同一把锁。

这是典型的“原子引用计数快路径 + 最后一个引用进入慢路径”的结构。

3.4 加一 relaxed、减一 release 和归零 acquire fence

Linux/POSIX 原子实现中：

1
2
3
加一：relaxed
减一：release
观察到 1→0：acquire fence

这里最基本的要求是：

  • 所有线程必须对同一个计数进行不可分割的 read-modify-write；
  • 单个变量的修改顺序必须一致；
  • 最后一个完成者在执行“归零后的停止判断”前，需要建立更强的内存顺序边界。

relaxed 加一足以完成“占一个 outstanding work 名额”；它不负责发布其他数据。

减一使用 release，并在最后一个减一线程上执行 acquire fence，是常见的最后引用释放模式：只有真正从 1 变为 0 的线程承担最终收尾工作。

需要强调：

原子操作主要解决的是 ntasks 自身的并发计数问题；任务队列、waiting/polling 链表、loop->stopped 等结构仍由 loop->mtx 保护。

3.5 不同构建配置下的计数类型

源码针对不同平台和裁剪配置提供多个实现：

配置 实现
禁用线程 普通 size_t，因为不存在并发访问。
Windows MSVC InterlockedIncrement/Decrement。
支持 C11 原子 atomic_size_t。
明确禁用原子且特定配置 普通整数，依赖该裁剪配置的使用约束。

因此，“为什么要原子”准确地说是：

在启用多线程且平台支持原子操作的正常构建中，ntasks 必须以原子方式维护；单线程构建不需要。

4. 线程局部 ev_loop_thrd 与 context 栈

4.1 TLS 实例按线程独立

定义使用 _Thread_local：

1
static + _Thread_local + struct ev_loop_thrd

三个关键词分别表示：

关键字 含义
static 变量名只在当前源文件可见。
_Thread_local 每个线程拥有独立存储实例。
静态初始化 {0, NULL} 每个线程第一次获得的实例初始为停止标志 0、context 指针 NULL。

这里的“每个线程一份”是 C TLS 的存储语义，不应理解为进程启动时预先为未来所有线程集中创建对象。对实际存在并访问该符号的线程，运行库提供该线程自己的实例，并以 { 0, NULL } 初始化。

因此，假设有三个线程调用同一个 ev_loop_t：

1
2
3
线程 A：自己的 ev_loop_thrd
线程 B：自己的 ev_loop_thrd
线程 C：自己的 ev_loop_thrd

它们之间不会共享 stopped 和 ctx 字段。

4.2 线程状态与 loop 状态相互独立

同一个线程可以先后运行不同的 loop，甚至在任务回调中嵌套运行另一个 loop。该线程仍只有一个 ev_loop_thrd。

这也是为什么结构体名称是 ev_loop_thrd，而不是 ev_loop_thread_of_loop。

4.3 两个字段的职责

1
2
stopped：中断当前线程正在执行的最内层 loop 调用
ctx：指向当前线程最内层的 ev_loop_ctx

ev_loop_self() 返回的实际上就是当前线程这份 TLS 状态的地址。这个地址被当成线程令牌传给 ev_loop_kill()。

4.4 ctx 构成嵌套调用栈

创建 context 时：

1
2
新 ctx.next = 当前 TLS ctx
TLS ctx = 新 ctx

销毁时：

1
TLS ctx = 当前 ctx.next

形成的结构是：

1
2
3
4
5
6
7
TLS ev_loop_thrd.ctx
        ↓
最内层 ctx
        ↓ next
外一层 ctx
        ↓ next
更外层 ctx
1
2
3
4
5
flowchart TD
    T[线程局部 ev_loop_thrd] --> C3[最内层 ctx]
    C3 --> C2[外层 ctx]
    C2 --> C1[最外层 ctx]
    C1 --> N[NULL]

所以 ev_loop_kill() 的公开契约能够实现：

1
存在嵌套 event loop 时，只中断最内层 loop。

它通过 thr->ctx 直接找到栈顶 context，而不是遍历某个全局 loop 列表。

4.5 ev_loop_ctx 不是 ev_loop_t

二者是“被管理对象”和“单次运行上下文”的关系：

1
ev_loop_ctx.loop ──指向──> ev_loop_t

一个 ev_loop_t 可以同时被多个线程运行，因此可以同时存在多个 context；同一线程发生嵌套 wait/run 时，也可以在 TLS 栈上挂载多个 context。每个 context 只记录本次调用需要的状态，例如：

  • 使用哪个 loop；
  • 是否等待某个 future；
  • 当前正在 Poll 还是条件变量等待；
  • Poll 后端线程令牌；
  • 所属线程的 pstopped；
  • future 是否 ready。

4.6 context 是按需创建的

ev_loop_ctx_wait_one() 先尝试从任务队列取得真实任务。只有没有任务可立即执行，并且需要进入 Poll、条件变量等待或注册 future 完成通知时，才调用 ev_loop_ctx_create()。因此：

1
2
3
调用 wait_one/run
    ├─ 立即取到任务：可以不创建 context
    └─ 需要等待或 Poll：创建 context 并压入线程 TLS 栈

这也是 pctx 在第一次进入 ev_loop_ctx_wait_one() 时允许为 NULL 的原因。

5. pstopped：context 到线程中断标志的引用

5.1 指向当前线程 TLS 中的 stopped

context 创建时建立关系：

1
2
3
ctx->pstopped
      ↓
当前线程的 ev_loop_thrd.stopped

所以 pstopped 不是指向 loop->stopped，也不是 context 自己单独分配的停止标志。

5.2 context 保存该指针的用途

ev_loop_ctx 可能被以下执行路径访问：

  • 当前运行线程的 wait 主循环；
  • future 完成后执行的 ctx->task；
  • 其他线程调用 ev_loop_kill()；
  • loop stop 或新任务到来时的唤醒逻辑。

这些路径需要操作“该 context 所属线程的中断标志”。其他线程不能通过普通变量名取得目标线程的 TLS 实例，但 context 在创建时已经把目标 TLS 地址保存为 pstopped，因此可以定向访问。

5.3 线程级停止标志优于 context 私有标志

从现有设计可以看出，停止语义刻意绑定到线程，而不是单个 context 对象：

  1. ev_loop_self() 返回线程令牌；
  2. ev_loop_kill() 接收线程令牌；
  3. TLS 中维护最内层 context；
  4. 中断后标志会在本次 wait 返回前清零，使下一次调用能够继续运行。

如果每个 context 有独立停止字段，就需要额外处理：

  • 没有 context 时发生的 kill；
  • context 尚未创建但线程即将进入等待的窗口；
  • 嵌套 context 的停止传播规则；
  • ev_loop_self() 所代表的线程级语义。

当前方案用一个线程级标志统一这些状态。

5.4 pstopped 的同步保护

虽然指向的是线程局部对象，但该指针可能被其他线程使用。源码通过 loop->mtx 对相关访问进行同步：

  • wait 主循环在持有 loop->mtx 时读取；
  • ev_loop_kill() 持有 loop->mtx 时设置；
  • future 回调持有 loop->mtx 时检查；
  • wait 退出时持有 loop->mtx 清零。

因此，这个字段没有像 ntasks 一样采用原子类型。二者区别是：

1
2
ntasks：有无锁访问路径 → 必须原子化
pstopped 指向的值：统一在 loop->mtx 下访问 → 普通 int 即可

5.5 单次中断结束后的自动清零

线程级 stopped 的目标是中断“当前这一次”运行，而不是永久停止 loop。

若本次没有执行任务，并且检测到 thread-local stopped：

1
2
清零 stopped
返回调用者

因此下次再调用 run() 或 wait_one() 时，可以重新进入事件循环。

永久或全局停止使用的是 loop->stopped，需要通过 ev_loop_restart() 重置。

5.6 loop 级停止与线程定向中断

两个公开接口的作用范围不同：

接口 作用范围 内部状态 恢复方式
ev_loop_stop(loop) 停止该 loop 上所有正在进行及后续的 run/wait 调用 设置 loop->stopped = 1，并唤醒 waiting/polling contexts ev_loop_restart(loop)
ev_loop_kill(loop, thr) 中断目标线程当前最内层的一次 run/wait 调用 通过栈顶 context 设置目标线程 TLS stopped，再 signal 条件变量或 kill Poll wait 本次调用返回前自动清零

所以不能表述为“需要整个线程停止时直接使用 stopped”。stopped 是内部标志；对外应使用 ev_loop_kill(loop, ev_loop_self() 返回的线程令牌)。该接口不会终止操作系统线程，只会让目标线程当前最内层的 event-loop 调用返回。

6. cnd：非 Poll 线程的休眠与唤醒

6.1 每个 context 独立持有条件变量

ev_loop_ctx 中包含：

1
2
cnd_t cond
waiting 标志

在 Linux pthread 兼容实现中：

1
2
3
4
cnd_t              → pthread_cond_t
cnd_wait()         → pthread_cond_wait()
cnd_timedwait()    → pthread_cond_timedwait()
cnd_signal()       → pthread_cond_signal()

它不是 CANopen 条件，也不是 Poll 的事件对象，而是纯线程同步原语。

6.2 进入条件变量等待的条件

只有同时满足以下条件时，线程才会进入 cnd_wait()：

  1. 当前没有真实任务可执行；
  2. context 已经创建；
  3. 当前线程不能进入 ev_poll_wait()；
  4. 任务队列仍为空；
  5. loop 尚未停止、线程未被 kill、future 未 ready。

“不能进入 Poll”通常是因为：

1
2
3
loop 配置了 npoll 限制
并且
已有足够数量的其他线程正在 ev_poll_wait()

POSIX reactor 场景通常建议 npoll == 1。这意味着：

1
2
一个线程负责 epoll_pwait
其他 loop 线程没有任务时用条件变量睡眠

6.3 Poll 并发数量限制

技术上 npoll == 0 可以允许无限数量线程同时 poll，但在 POSIX readiness reactor 模型中，多线程同时等待同一个事件源通常会增加：

  • 唤醒竞争；
  • 锁竞争；
  • 同一批 readiness 事件的调度复杂度；
  • 不必要的上下文切换。

Lely 的 C API 文档明确建议 POSIX reactor 使用单 polling thread。

因此条件变量承担“非 polling worker 的休眠机制”。

6.4 cnd_wait() 的原子解锁与重新加锁

调用前线程持有 loop->mtx。条件变量等待会完成一个原子组合动作：

1
2
释放 loop->mtx
并进入 cond 等待

被唤醒后：

1
2
重新获得 loop->mtx
然后 cnd_wait 返回

这里“释放锁并睡眠”必须是相对于 signal 过程原子的，否则会出现经典丢失唤醒：

1
2
3
4
5
等待线程：检查无任务
等待线程：准备睡眠但尚未真正睡眠
投递线程：加入任务并 signal
等待线程：此后才睡眠
结果：任务已存在，但等待线程永久睡眠

条件变量和互斥锁的组合就是为解决这个窗口。

6.5 条件变量的唤醒来源

ev_loop_ctx_kill() 根据 context 当前状态选择唤醒方式：

1
2
3
4
5
6
7
8
ctx 正在 cnd 等待
    → cnd_signal(&ctx->cond)

ctx 正在 Poll 等待
    → ev_poll_kill(loop->poll, ctx->thr)

ctx 两者都不是
    → 只更新状态，后续循环自行观察

触发 ev_loop_ctx_kill() 的主要来源包括：

  • 新任务从空队列变为非空；
  • future 变为 ready；
  • ev_loop_stop()；
  • 其他线程调用 ev_loop_kill()。

6.6 每个 context 独立条件变量的价值

它可以做到定向唤醒：

1
2
只唤醒被选中的 waiting context
而不是 broadcast 唤醒所有 worker

loop->waiting 链表按 LIFO 方式组织。ev_loop_kill_any() 取一个 waiting context，然后 signal 它自己的 cond。

这减少了惊群效应。

6.7 虚假唤醒与循环复核

条件变量允许虚假唤醒，因此正确写法必须始终在循环中重新检查状态，而不能认为 cnd_wait() 返回就必然有任务。

Lely 的结构正是：

1
2
3
4
5
while 条件仍允许继续等待:
    检查 stop / killed / future
    检查任务队列
    检查 poll 资格
    必要时再次 wait

所以虚假唤醒不会破坏逻辑，只会多执行一轮条件检查。

6.8 条件变量与 Poll 的职责边界

等待方式 等待对象 唤醒机制
ev_poll_wait() 外部 I/O、timerfd、signal pipe 等 Poll 事件 ev_poll_kill() 或底层 I/O 事件
cnd_wait() loop 内部任务或状态变化 cnd_signal()

可以把它们理解为：

1
2
Poll：等外部世界
cnd：等同一进程内其他线程

7. Future/Promise：一次性异步结果与 event loop 的连接点

7.1 Future 不是消息队列

Lely 的 Future/Promise 可以宽泛地看成一种“异步完成通知机制”，但更准确的定义是：

Future/Promise 是围绕一份共享状态建立的一次性异步结果同步机制。

它与消息队列的区别如下：

机制 主要语义
消息队列 发送方可以连续写入多条消息，接收方逐条取出；通常是多次生产、多次消费。
Future/Promise Promise 最多成功完成一次；Future 观察同一个 ready 状态，并在 ready 后取得一次异步操作的结果。
Event-loop task queue 保存待执行的 ev_task，解决“在哪个 executor 上执行回调”；它不等同于 Future 的结果存储区。

因此，不宜把 Future 理解为“发送方把请求消息放进 loop 队列，接收方执行后再沿队列返回结果”。Future 自身不负责传输请求，也不负责执行 I/O；它只保存或关联异步操作的完成状态、结果值以及等待完成通知的任务。

7.2 Promise、Future、异步操作和 Loop 的角色

对象 准确角色
异步操作发起方 调用 AsyncRead()、AsyncWrite()、异步 SDO 等接口，并取得 Future。
ev_promise_t 异步操作完成方持有的写端。操作成功、失败或取消后，通过 Promise 发布结果并将共享状态置为 ready。
ev_future_t 观察端。用于检查 ready、取得结果，或登记“Future ready 后应投递”的任务。
ev_loop_t 任务执行器和 I/O 驱动器。它执行 executor 队列中的任务，并在队列暂时为空时进入 Poll 或条件变量等待。
Poll/I/O 组件 驱动真正的 SocketCAN、timerfd、信号或其他异步事件；事件完成后通常投递任务，任务再推进异步操作。

ev_loop_t 不是固定的“发送方”或“接收方”。同一个 loop 既可能执行发起异步操作的任务，也可能执行 I/O 完成任务、Promise 完成后的通知任务以及其他应用任务。

7.3 Promise 与 Future 共享一次性结果状态

Promise 用于存入一个稍后由 Future 异步取得的值，其语义类似 single-shot event。Future 则提供对异步操作结果的访问。

典型状态变化为：

1
2
3
4
5
stateDiagram-v2
    [*] --> Waiting
    Waiting --> Setting: promise acquires completion right
    Setting --> Ready: store result and publish ready
    Ready --> Ready: later completion attempts are rejected

Promise 完成后，Future 可以：

  1. 通过 ev_future_is_ready() 判断是否完成；
  2. 在 ready 后通过 ev_future_get() 取得 Promise 保存的结果；
  3. 通过 ev_future_submit() 登记一个任务，使其在 Future ready 后被投递到指定 executor。

这里的结果值与 executor task 是两个不同概念：结果保存在 Future/Promise 的共享状态中；task 只负责在完成时执行通知或后续处理。

7.4 ev_loop_wait(loop, future) 如何接入 Future

当 ev_loop_ctx_create() 收到非空 Future 时，它会：

1
2
3
1. 获取 Future 引用并保存到 ctx->future
2. 初始化 ctx->task，使其 executor 指向当前 loop
3. 调用 ev_future_submit(future, &ctx->task)

这一步不是把“异步请求”提交给 Future，而是把一个“Future 完成后唤醒本次 wait”的内部任务登记到 Future。

当异步操作完成并满足 Promise 后：

1
2
3
4
5
6
7
8
9
10
11
Promise 发布结果并置 Future ready
    ↓
Future 将 ctx->task 投递到 loop executor
    ↓
loop 从任务队列取出并执行 ctx->task
    ↓
ev_loop_ctx_task_func() 设置 ctx->ready = 1
    ↓
ev_loop_ctx_kill(ctx, 0) 唤醒 Poll 或条件变量等待
    ↓
ev_loop_wait() 看到 ctx->ready 后结束本次等待

ctx->task 的职责是结束等待，不是读取和返回业务结果。业务结果仍由调用方或上层包装在 Future ready 后通过 ev_future_get() 或相应的 C++ Future 接口取得。

7.5 等待 Future 时仍要持续驱动 Loop

ev_loop_wait(loop, future) 不是单纯阻塞在 Future 上。只要 Future 尚未 ready，它仍会继续推进 event loop：

1
2
3
4
Future 尚未 ready
    ├─ queue 有真实任务：执行一个任务
    ├─ 当前 context 可以 Poll：调用 ev_poll_wait() 等待 I/O
    └─ 当前 context 不可 Poll：调用 cnd_wait() 等待内部唤醒

这正是异步 CANopen 操作需要的行为。例如，异步 SDO 请求的完成依赖：

1
2
3
4
5
6
7
8
9
SocketCAN 收到响应或定时器到期
    ↓
Poll 回调投递 I/O 任务
    ↓
loop 执行任务并推进 SDO 状态机
    ↓
异步操作完成方满足 Promise
    ↓
Future ready，wait 返回

如果线程只是同步睡眠而不再驱动同一个 loop，且没有其他线程负责执行该 loop，那么依赖 CAN I/O 或超时任务的 Future 可能无法变为 ready。

7.6 更准确的完整流程

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
sequenceDiagram
    participant Caller as AsyncCaller
    participant Operation as AsyncOperation
    participant EL as EventLoop
    participant Poller as PollBackend
    participant Device as SocketCAN
    participant Promise as PromiseFuture

    Caller->>Operation: start async request
    Operation-->>Caller: return Future
    Caller->>EL: wait on Future
    EL->>Promise: submit internal wake task
    EL->>Poller: wait for I/O
    Device-->>Poller: response or readiness event
    Poller-->>EL: post completion task
    EL->>Operation: run completion task
    Operation->>Promise: publish result and set ready
    Promise-->>EL: post Future-ready task
    EL->>EL: set context ready and wake wait
    EL-->>Caller: wait returns
    Caller->>Promise: obtain result

因此，更准确的理解是：

1
2
3
4
5
6
异步操作发起方启动请求并取得 Future；
Loop 持续执行任务和驱动 I/O；
异步操作完成方通过 Promise 发布结果；
Future 把完成通知任务投递到 Loop；
Loop 执行该通知任务并结束 wait；
调用方再从 Future 取得结果。

7.7 run() 与 wait(future) 的区别

调用 内部 Future 主要返回条件
ev_loop_run(loop) NULL loop stopped，或 outstanding work 归零触发 stop。
ev_loop_wait(loop, future) 指定 Future Future ready、loop stopped、线程被中断或发生错误。
ev_loop_wait_until(loop, future, deadline) 指定 Future 在上述条件外增加绝对超时。
ev_loop_poll(loop) NULL 非阻塞处理当前可立即执行的工作。

Future 在 wait() 中是“本次 event-loop 运行的结束条件之一”，而不是 loop 队列中的普通业务任务。

8. 任务投递后的 Poll/cnd 唤醒路径

8.1 任务投递路径

任务被投递到 event loop 时：

1
2
3
4
5
获取 loop->mtx
判断投递前队列是否为空
把 task 放到 queue 尾部
若此前为空，调用 ev_loop_kill_any(loop, 1)
释放 loop->mtx

只在“空 → 非空”转换时唤醒线程，避免每次追加任务都重复发唤醒。

8.2 waiting context 的优先级

ev_loop_kill_any() 的顺序是：

1
2
3
4
先找 waiting 链表中的 context
若存在：通过 cnd_signal 唤醒
否则在允许时找 polling context
若存在：通过 ev_poll_kill 唤醒

原因很直接：

  • 条件变量线程本来就在等内部任务；
  • 唤醒它不需要打断正在处理外部 I/O 的 poller；
  • 可让 poller 继续服务 I/O readiness。

8.3 完整时序

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
sequenceDiagram
    participant P as 投递线程
    participant L as ev_loop
    participant W as 条件变量等待线程
    participant R as Poll 等待线程

    P->>L: ev_exec_post(task)
    L->>L: lock + queue push
    alt 存在 waiting context
        L->>W: cnd_signal(ctx.cond)
        W->>W: 重新取得 loop.mtx
        W->>L: 从 queue 取任务
    else 没有 waiting，但存在 polling context
        L->>R: ev_poll_kill()
        R->>R: 底层 Poll wait 被中断
        R->>L: 返回并从 queue 取任务
    end

9. ev_poll_wait() 的虚函数分派链

9.1 抽象接口分派

ev_poll_t 不是一个固定的 Linux io_poll 类型，而是一个抽象 Poll 接口。它的虚函数表包含：

1
2
3
self
wait
kill

ev_poll_wait() 做的事情等价于：

1
调用当前 poll 对象虚函数表中的 wait 指针

因此：

1
2
3
ev_poll_wait
不是具体后端函数
而是统一分派入口

9.2 Linux io::Poll 后端注册

Linux io_poll 初始化时：

1
poll_vptr = io_poll_poll_vtbl

该虚函数表的 wait 成员指向：

1
io_poll_poll_wait

io_poll_get_poll() 返回的不是新对象，而是 io_poll 内部 poll_vptr 字段的地址。ev_loop 保存这个抽象接口指针。

完整对象关系：

1
2
3
4
5
6
flowchart LR
    A[ev_loop.poll] --> B[io_poll.poll_vptr 的地址]
    B --> C[io_poll_poll_vtbl]
    C --> D[self = io_poll_poll_self]
    C --> E[wait = io_poll_poll_wait]
    C --> F[kill = io_poll_poll_kill]

9.3 完整调用链

1
2
3
4
5
6
7
8
9
ev_loop_ctx_wait_one_until()
  ↓
ev_poll_wait(loop->poll, msec)
  ↓
(*loop->poll)->wait(loop->poll, msec)
  ↓
io_poll_poll_wait(..., msec)
  ↓
epoll_pwait(epfd, events, maxevents, msec, signal_mask)

因此，在该对象组合下可以确定：

如果 loop->poll 来自 Linux io_poll_get_poll()，执行的就是 io_poll_poll_wait()。

但要加两个限定：

  1. 它是通过函数指针间接调用，不是 loop.c 对 io_poll_poll_wait() 的直接静态调用；
  2. 换成 Windows Poll 或自定义 ev_poll_t 实现后，会调用对应后端的 wait。

9.4 与 POSIX poll(2) 的区别

名称容易产生误导：

1
2
3
ev_poll_wait：Lely 抽象接口
io_poll_poll_wait：Linux io2 后端函数
epoll_pwait：实际 Linux 系统调用

它没有调用传统 POSIX poll()。

9.5 msec 在两个 wait 函数中的含义

调用场景 msec
无超时 wait，队列为空 -1，无限阻塞
无超时 wait，但已有任务 0，只检查一次 I/O
until，有未来 deadline 且队列为空 deadline 剩余毫秒
until，deadline 为 NULL 0，完全非阻塞
until，队列已非空 0，避免阻塞任务执行

注意：ev_poll_wait() 的返回值表示 Poll 调用成功或失败，并不等于“执行了几个 loop task”。Poll 后端可能在内部处理多个 I/O watch 回调，这些回调通常再把任务投递到 loop 队列；loop 下一轮才执行对应 task。

10. 从 Loop::run() 到 SocketCAN 事件的完整流程

结合前一篇 io_poll 分析，可以把完整运行链路串起来：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
1. C++ 调用 loop.run()
2. loop.hpp 转调 ev_loop_run()
3. ev_loop_run() 转调 ev_loop_wait(loop, NULL)
4. ev_loop_wait() 在外层循环中重复调用 ev_loop_ctx_wait_one()
5. 每次 wait_one 最多主动执行一个真实队列任务；若当前没有任务但 ntasks > 0，loop 不会 stop
6. 若当前线程有 Poll 资格，调用 ev_poll_wait(..., -1)
7. 虚函数分派到 io_poll_poll_wait()
8. io_poll_poll_wait() 在 epoll_pwait() 阻塞
9. SocketCAN fd 可读，epoll 返回
10. io_poll 找到对应 io_poll_watch 并调用 watch 回调
11. CanChannel watch 回调向 executor 投递接收任务
12. 投递使 loop queue 从空变为非空，并中断 Poll wait 或唤醒 cnd waiter
13. ev_loop_ctx_wait_one() 重新检查 queue
14. 弹出并执行 CanChannel 接收任务
15. 读取 CAN 帧，完成异步操作并进入 CanNet/CANopen 上层
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
sequenceDiagram
    participant App as Application
    participant Ev as EventLoop
    participant Vtbl as PollVtable
    participant Io as IoPoll
    participant Kernel as EpollWait
    participant Can as SocketCAN
    participant Exec as ExecutorQueue

    App->>Ev: call run
    Ev->>Ev: queue empty and outstanding work exists
    Ev->>Vtbl: wait with infinite timeout
    Vtbl->>Io: call backend wait
    Io->>Kernel: enter epoll wait
    Can-->>Kernel: file descriptor becomes readable
    Kernel-->>Io: return input event
    Io->>Exec: callback posts receive task
    Exec->>Ev: queue changes from empty to nonempty
    Ev->>Exec: pop one task
    Exec->>Can: read CAN frame

11. npoll = 1 的典型多线程运行

假设：

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

可能出现：

1
2
线程 A：取得 Poll 资格，进入 epoll_pwait()
线程 B：无 Poll 资格，进入 cnd_wait()

此时另一个线程 C 投递一个普通任务：

1
2
3
4
线程 C：queue 空→非空
线程 C：优先 signal 线程 B 的 cond
线程 B：醒来并执行任务
线程 A：继续等待 I/O

如果没有 waiting 线程，才会中断 Poll 线程 A：

1
2
3
线程 C：ev_poll_kill(A 的 poll thread token)
线程 A：epoll_pwait 被信号打断
线程 A：返回 loop，执行新任务

这种设计让：

  • Poll 线程尽量持续负责外部 I/O；
  • 非 Poll worker 优先处理内部任务；
  • 必要时仍能中断 Poll，保证任务投递不会长期等待。

12. 关键对象关系

对象/字段 归属 是否共享 保护方式 主要职责
ev_loop_t 每个 loop 多线程共享 loop->mtx + 部分原子 任务队列、停止状态、poller/waiter 管理
loop->queue 每个 loop 多线程共享 loop->mtx 待执行任务队列
loop->ntasks 每个 loop 多线程共享 原子操作 outstanding work 计数
loop->stopped 每个 loop 多线程共享 loop->mtx loop 全局停止状态
ev_loop_thrd 每个线程 不在线程间共享实例 TLS；其字段跨线程访问时受 loop->mtx 保护 线程级中断和嵌套 context 栈顶
ev_loop_ctx 按需创建的一次 wait/run context 可被 future/kill 路径访问 loop->mtx + refcnt 连接 loop、线程、Poll/cond 和可选 future；不是 loop 对象
ctx->pstopped context 指向线程 TLS 指向目标线程实例 loop->mtx 检查或设置该线程的单次中断标志
ctx->cond 每个 context 目标线程等待、其他线程 signal 条件变量 + loop->mtx 非 Poll 线程休眠与唤醒
ctx->thr 每个 context 保存 Poll 后端线程令牌 后端定义 通过 ev_poll_kill() 中断具体 Poll wait
ctx->future 可选，每个 context 与 promise 共享异步完成状态 future mutex/atomic state + refcount future ready 时投递 ctx->task 并结束本次 wait
ev_promise_t 每个异步结果生产端 可跨线程共享 future mutex/atomic state + refcount 一次性写入结果并发布 ready
loop->poll 每个 loop 可共享后端 Poll ev_poll 接口契约 对外部事件进行抽象等待

13. 参考资料

13.1 Lely 官方源码

  1. Lely src/ev/loop.c
  2. Lely include/lely/ev/loop.h
  3. Lely include/lely/ev/loop.hpp
  4. Lely include/lely/ev/poll.h
  5. Lely Linux src/io2/linux/poll.c
  6. Lely pthread 兼容实现 src/libc/threads-pthread.c
  7. Lely executor API include/lely/ev/exec.h
  8. Lely standard executor src/ev/std_exec.c
  9. Lely Future API include/lely/ev/future.h
  10. Lely Future implementation src/ev/future.c
  11. Lely library overview

13.2 线程与语言规范

  1. The Open Group: pthread_cond_wait() / pthread_cond_timedwait()
  2. The Open Group: pthread_cond_signal()
  3. WG14 N1570: C11 draft

13.3 关联文章

本文的章节结构、结论先行、流程图、状态表和源码索引方式参考了已上传的《Lely canopen io_poll 机制原理与完整运行流程详解》。两篇文章可以连续阅读：

1
2
ev_loop：决定何时执行 task、何时等待、怎样协调多个线程
io_poll：决定怎样等待 Linux fd、怎样分发 epoll readiness