Lely CANopen：io_tqueue 与 pheap 的实现原理、运行流程及其与 Timer、can_net 的关系

在这里插入图片描述

@[toc]

一句话结论

io_tqueue 是一个“多路逻辑定时器复用器”：

  • 上层可以同时提交很多个不同绝对截止时间的 io_tqueue_wait；
  • 中间用 pheap 维护“当前最早到期”的等待；
  • 下层始终只占用 一个 io_timer_t；
  • 底层 Timer 到期时，再回到 io_tqueue 批量完成所有 value <= now 的等待。

它与 can_net 串起来之后，又形成“两级最早截止时间压缩”：

1
2
3
4
5
6
7
8
9
很多 can_timer_t
    ↓
can_net.timer_heap
    ↓ 选出协议层最早一个
io_tqueue.queue
    ↓ 再与其他 I/O 等待一起排序
一个 io_timer_t
    ↓
Linux timerfd / poll / executor

1. 为什么需要 io_tqueue

如果没有 io_tqueue，最直观的设计是：

1
2
3
4
每一个逻辑等待
    → 一个 io_timer_t
    → 一个系统层定时器（如 Linux timerfd）
    → 一个异步等待对象

但 CANopen 运行时会同时出现很多不同类型的超时和周期事件，例如：

  • 心跳生产 / 消费；
  • SDO 超时；
  • PDO event timer；
  • SYNC 周期；
  • CAN 帧写确认超时；
  • 其他用户层异步等待。

如果每个都单独占用一个系统定时器，成本会很高。

io_tqueue 的核心思想是：

只维护“谁最早到期”，并把这个最早截止时间映射到底层唯一的 io_timer_t 上。

因此，系统里可能有很多逻辑等待，但底层只有一个真正被设置到内核的 Timer。

2. 四层对象模型

从上到下可以分成四层：

1
2
3
4
5
6
7
8
9
10
11
12
13
第 1 层：用户逻辑等待
    struct io_tqueue_wait

第 2 层：逻辑等待队列
    struct io_tqueue
    └── struct pheap queue

第 3 层：底层唯一 Timer 等待
    struct io_timer_wait wait

第 4 层：真正系统 Timer
    io_timer_t
    └── Linux 后端通常落到 timerfd

也可以画成：

1
2
3
4
5
6
7
8
9
很多 io_tqueue_wait
        ↓
io_tqueue.queue（pheap）
        ↓ 只取堆顶最早截止时间
一个 io_timer_t
        ↓
到期后触发 io_tqueue_wait_func()
        ↓
批量完成所有已过期等待

3. io_tqueue_wait：一个逻辑截止时间

io_tqueue_wait 里最关键的成员有三个：

  • value：绝对到期时间；
  • task：完成时投递到 executor 的任务；
  • _node：嵌入在等待对象里的 pnode，用于把这个等待挂入 pheap。

可以把它理解为：

1
一个等待对象 = 截止时间 + 完成回调 + 堆节点

3.1 为什么 _node 直接嵌入 io_tqueue_wait

这是典型的 侵入式数据结构 设计。

好处：

  • 不需要为堆节点单独再分配内存；
  • pheap 只负责维护“节点关系”，业务数据仍然在外部对象里；
  • 从 pnode 可以通过 structof() 很快回到 io_tqueue_wait。

3.2 value 为什么不能在入堆后修改

因为 pnode.key 指向的就是 wait->value。

一旦节点已经在堆里，如果你直接改了 value，堆序就可能被破坏。

所以正确做法只能是：

1
2
3
先 remove
再修改时间
再 insert

4. pheap：为什么适合这里

Lely 这里使用的是 pairing heap（配对堆）。

4.1 它维护什么性质

对最小堆来说，pheap 只保证：

  • root 是全局最小值；
  • 每个父节点不大于自己的子节点。

它并不保证兄弟链表有序，也不保证整棵树线性有序。

这点非常重要。可以画成：

1
2
3
4
5
      root = 10
     /    |    \
   20     30    25
  /  \           |
21   40         26

这里只能保证：

  • 10 是全局最小；
  • 20 >= 10，30 >= 10，25 >= 10；
  • 21 >= 20，40 >= 20，26 >= 25。

但是：

  • 兄弟 20 / 30 / 25 没有顺序含义；
  • 不能通过兄弟链顺序直接推断第二小、第三小；
  • “下一小是谁”不是简单看 root->next 或第一个 child。

4.2 为什么 pheap_first() 能是 O(1)

因为最小值总是堆根：

1
2
3
4
struct pnode *pheap_first(const struct pheap *heap)
{
    return heap->root;
}

所以“取当前最早截止时间”非常便宜，这正适合 io_tqueue 高频做的操作。

4.3 为什么插入成本也低

新节点插入时，只需要和当前根比较：

  • 如果新节点更小，新节点成为新根，旧根挂成它的子节点；
  • 否则把新节点挂到根的子节点链表。

因此插入是 O(1)。

5. 你最关心的问题：删除 root 后，下一个最小节点怎么产生？

这是理解配对堆最关键的一步。

你的判断完全对：

  • root 是全局最小值；
  • 其他节点只满足“父不大于子”；
  • 根的孩子链表 不是 一个排好序的线性序列；
  • 所以 root 删除后，不能靠“顺着兄弟链找下一个”来得到最小值。

那答案是什么？

删除 root 之后，不是“找一个现成排好序的第二名”；而是把 root 的各个子树重新做一次 meld / merge（融合），融合完成后，新根自然就是剩余节点中的全局最小值。

5.1 先看 pheap_remove() 删除 root 时做了什么

对 root 节点来说，逻辑可以概括成：

1
2
3
4
1. 把 root 从堆中摘掉
2. 取出 root 的 child 链表
3. 对这些 child 子树做配对融合（pnode_merge_pairs）
4. 融合出来的一棵树，作为新的 root

源码语义可以简化为：

1
2
3
4
5
6
7
8
9
10
if 删除的是 root:
    heap->root = NULL

child = root->child
root->child = NULL

if child 不为空:
    child = pnode_merge_pairs(child, cmp)
    heap->root = pnode_merge(heap->root, child, cmp)
    heap->root->parent = NULL

要点是：

  • root 被删掉后，原来挂在 root 下面的每一个 child，实际上都代表“一棵仍然满足堆序的子树”；
  • 每棵子树的根，都是它自己这棵子树里的最小值；
  • 把这些“子树根”不断两两融合，最终得到一棵新树；
  • 新树的根，就是整个剩余集合里的最小值。

5.2 为什么这样融合后就能得到新的全局最小值

假设删掉 root 后，剩余子树根分别是：

1
20, 21, 22, 25

它们各自还带着自己的子树，但不用先扫描全部后代，因为：

  • 20 已经是自己那棵子树里的最小值；
  • 21 也是自己那棵子树里的最小值；
  • 22 也是；
  • 25 也是。

于是只要不断比较这些“子树根”即可。

例如配对融合的直觉过程可以理解成：

1
2
3
(20, 21) 先融合  →  得到一棵以 20 为根的树
(22, 25) 先融合  →  得到一棵以 22 为根的树
再把这两棵树融合 →  得到一棵以 20 为根的树

最终新根就是 20，它就是删除原 root 后的全局最小值。

注意：

  • 这里的“20 是下一个最小值”并不是因为它恰好排在 child 链表第一个；
  • 而是因为在一轮轮 merge 之后，它在所有剩余子树根中最小。

5.3 这不是“寻找第二小”，而是“重建堆顶”

这是很容易想岔的一点。

很多人会自然地想成：

1
2
删除最小值(root)
    → 找第二小

但配对堆不是有序数组，也不是有序链表。

更准确的表述应该是：

1
2
3
删除 root
    → 把 root 的子树集合重新融合
    → 融合后的新根，就是新的全局最小值

也就是说：

  • 算法并不直接维护“第二名指针”；
  • 它维护的是“堆序结构”；
  • 删除 root 后，通过合并剩余子树恢复一个新的最小堆。

这就是为什么 pheap_remove() 的代价要比 pheap_first() 和 pheap_insert() 高。

6. pheap_remove(root) 运行流程图

下面专门用流程图展示“root 到期后被删除，下一个最小节点如何上来”。

1
2
3
4
5
6
7
8
9
10
11
12
flowchart TD
    A["当前堆的 root 是全局最小值"] --> B["删除 root"]
    B --> C["取出 root 的所有 child 子树"]
    C --> D{"child 是否为空"}
    D -- 是 --> E["heap.root 置空，堆变空"]
    D -- 否 --> F["把这些 child 看作多个小堆"]
    F --> G["调用 pnode_merge_pairs，沿 sibling 链逐步融合"]
    G --> H["不断比较两棵子树的根"]
    H --> I["较小根保留为父，较大根挂为 child"]
    I --> J["逐步融合成一棵新树"]
    J --> K["新树的根成为新的 heap.root"]
    K --> L["新的 root 就是剩余节点中的全局最小值"]

再补一张“不是线性有序”的示意图：

1
2
3
4
5
6
7
8
9
10
11
flowchart TD
    R["old root = 10"] --> C1["20"]
    R --> C2["30"]
    R --> C3["25"]
    C1 --> C11["21"]
    C1 --> C12["40"]
    C3 --> C31["26"]

    X["删除 10 后"] --> Y["不能顺着兄弟顺序直接找第二小"]
    Y --> Z["需要把 20、30、25 这些子树根重新融合"]
    Z --> N["new root = min of {20, 30, 25}"]

6.1 pnode_merge_pairs() 的实际融合顺序

需要特别注意：当前 Lely 源码中的 pnode_merge_pairs() 并不是先把所有兄弟节点分成独立的两两分组，再执行第二轮反向合并。它沿着 sibling 链从左向右执行连续融合：

1
2
3
4
5
6
while (node->next) {
    struct pnode *next = node->next->next;
    node->next->next = NULL;
    node = pnode_merge(node, node->next, cmp);
    node->next = next;
}

假设删除旧 root 后得到的子树根链为：

1
20 -> 30 -> 25 -> 40

实际执行过程如下：

1
2
3
4
5
6
7
flowchart LR
    A["初始 sibling 链<br/>20 → 30 → 25 → 40"] --> B["第 1 轮<br/>merge 20 与 30<br/>融合根为 20"]
    B --> C["剩余处理链<br/>20 → 25 → 40"]
    C --> D["第 2 轮<br/>merge 20 与 25<br/>融合根为 20"]
    D --> E["剩余处理链<br/>20 → 40"]
    E --> F["第 3 轮<br/>merge 20 与 40<br/>融合根为 20"]
    F --> G["返回节点 20<br/>成为新的 heap.root"]

每轮 pnode_merge() 都会比较两棵子树的根：

  • 较小的根继续作为父节点；
  • 较大的根挂到较小根的 child 链；
  • 融合结果仍然是一棵满足最小堆序的子树。

因此，在处理到第 k 个子树根后，变量 node 始终是“目前已经处理过的所有子树中的最小根”。当 sibling 链全部处理完成时，返回的 node 自然就是剩余全部节点中的全局最小值。

这个过程的关键不是维护一条完整排序链，而是持续维护下面这个不变量：

1
2
每完成一轮 merge：
当前融合结果的 root = 已处理子树根中的最小值

7. io_tqueue 的内部状态

io_tqueue 的关键成员可以概括成：

  • timer：底层复用的唯一 io_timer_t；
  • exec：默认 executor；
  • wait：唯一的底层 io_timer_wait；
  • submitted：当前是否已经向底层提交了这个唯一 wait；
  • next：当前底层 Timer 被设置到的绝对截止时间；
  • queue：存放所有逻辑等待的 pheap；
  • shutdown：是否关闭。

它的核心状态关系是：

1
2
3
queue 里可以有很多逻辑等待
但底层 wait 永远最多只有一个 submitted
next 永远代表“当前底层 Timer 目标截止时间”

8. 初始化流程：io_tqueue_init()

初始化主要做了几件事：

  1. 保存 timer；
  2. 选择默认 exec；
  3. 初始化唯一底层 io_timer_wait，回调设为 io_tqueue_wait_func()；
  4. 初始化互斥锁（若启用线程）；
  5. submitted = 0，next = {0,0}；
  6. pheap_init(&tq->queue, &timespec_cmp)；
  7. 把 io_tqueue 注册到 io_ctx。

8.1 与 Timer 的约束关系

io_tqueue.h 明确要求：

io_tqueue 生命周期内，外部不要再直接对这个 timer 调用 io_timer_settime()。

因为 io_tqueue 自己要独占控制这个 Timer 的截止时间。

否则就会破坏：

1
堆顶截止时间  <->  底层 Timer 当前设置值

这个关键对应关系。

9. 提交等待：io_tqueue_submit_wait()

这是整个机制最重要的入口。

9.1 提交时做什么

当一个新的 io_tqueue_wait 被提交时，流程大致是：

1
2
3
4
5
6
1. 若 wait.task.exec 为空，则使用 tq->exec
2. 初始化 executor 对 task 的生命周期跟踪
3. 若 tq 已 shutdown，则立即按 canceled 完成
4. 否则把 wait->_node 插入 tq->queue
5. 如果它比 tq->next 更早：重设底层 Timer
6. 如果当前还没有 submitted 的底层 wait：提交 tq->wait

可以压缩成一张图：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
flowchart TD
    A["提交一个 io_tqueue_wait"] --> B{"task.exec 是否为空"}
    B -- 是 --> C["使用 tq->exec"]
    B -- 否 --> D["保留用户指定的 exec"]
    C --> E
    D --> E
    E{"tq 是否 shutdown"}
    E -- 是 --> F["直接按 canceled 完成"]
    E -- 否 --> G["初始化 wait 节点，并把 key 指向 wait.value"]
    G --> H["把 wait 插入 tq.queue"]
    H --> I{"wait.value 是否早于 tq.next<br/>或 tq.next 还未设置"}
    I -- 是 --> J["把底层 Timer 重设到 wait.value"]
    I -- 否 --> K["底层 Timer 保持不变"]
    J --> L
    K --> L
    L{"当前是否还没有提交底层 tq.wait"}
    L -- 是 --> M["提交唯一的底层 io_timer_wait"]
    L -- 否 --> N["结束"]
    M --> N

9.2 为什么不是每次提交都要重新设置 Timer

因为 io_tqueue 只关心 最早截止时间。

例如当前：

1
tq.next = 100ms

新来一个等待：

  • 若到期时间是 150ms：不会影响当前最早者，不必重设底层 Timer；
  • 若到期时间是 80ms：它变成新堆顶，需要立即把底层 Timer 改到 80ms。

所以条件是：

1
新等待是否早于当前 tq.next

而不是“是否新插入了一个节点”。

10. Timer 到期：io_tqueue_wait_func()

底层唯一 io_timer_wait 到期后，会执行 io_tqueue_wait_func()。

这个函数做的事可以概括为：

1
2
3
4
5
6
1. 读取当前时间 now
2. 清掉 submitted 标记，并清空 tq->next
3. 从堆顶开始，反复弹出所有 value <= now 的等待
4. 如果堆还不空，把底层 Timer 设到新的最早截止时间
5. 如有需要，再次提交唯一底层 wait
6. 把本轮到期等待逐个 post 给 executor

10.1 为什么一次会弹出多个等待

因为底层 Timer 唤醒时，可能已经有多个逻辑等待同时到期了。

例如堆里有：

1
100ms, 100ms, 102ms, 120ms

当前 now = 105ms，那本轮就要一次性完成：

1
100ms, 100ms, 102ms

只剩下 120ms 重新编程到底层 Timer。

10.2 io_tqueue_wait_func() 主流程图

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
flowchart TD
    A["底层 io_timer_wait 到期"] --> B["读取当前时间 now"]
    B --> C["submitted 清零；next 清零"]
    C --> D["查看当前堆顶"]
    D --> E{"堆顶 wait.value 是否小于等于 now"}
    E -- 是 --> F["删除堆顶节点"]
    F --> G["把对应 wait 放入完成队列"]
    G --> D
    E -- 否 --> H{"堆是否为空"}
    H -- 是 --> I["本轮不再设置 Timer"]
    H -- 否 --> J["取新的堆顶作为 tq.next"]
    J --> K["把底层 Timer 设到 tq.next"]
    K --> L["重新提交唯一底层 io_timer_wait"]
    I --> M["锁外 post 所有到期任务"]
    L --> M
    M --> N["结束"]

10.3 这时 pheap_remove(root) 正好派上用场

注意这里的弹出动作不是别的，正是：

1
取堆顶 → 删除 root → 新 root 自动成为剩余最小值

因此 io_tqueue_wait_func() 之所以能写成：

1
2
3
4
5
while ((node = pheap_first(&tq->queue))) {
    if (now < wait->value) break;
    pheap_remove(&tq->queue, node);
    ...
}

就是因为每次删掉当前 root 之后，配对堆已经帮它把“下一个最小节点”重建成新的 root 了。

这正好呼应了你前面的问题。

11. 取消、abort 与 shutdown

11.1 cancel

cancel 的语义是：

  • 如果等待还在队列里，就把它从堆里摘掉；
  • 然后以 ERRNUM_CANCELED 作为完成结果，正常投递它的 completion task。

11.2 abort

abort 的语义是：

  • 也会把等待从堆里摘掉；
  • 但 不会 投递 completion task。

11.3 堆空时为何要取消底层 wait

当从堆里移除节点后，如果发现：

1
pheap_empty(&tq->queue) && tq->submitted

那么说明：

  • 逻辑等待已经没有了；
  • 但底层唯一 io_timer_wait 可能还在挂着。

这时会调用：

1
io_timer_cancel_wait(tq->timer, &tq->wait)

避免底层继续无意义等待。

11.4 shutdown

关闭时会：

  1. 标记 shutdown = 1；
  2. 若底层 wait 已提交，则尝试 io_timer_abort_wait()；
  3. 然后把所有逻辑等待统一 cancel 掉。

12. io_tqueue 与 io_timer_t 的关系

二者不是替代关系，而是上下层关系。

12.1 io_timer_t 负责什么

io_timer_t 负责：

  • 持有具体系统 Timer 资源；
  • 提供 io_timer_settime()；
  • 提供 io_timer_submit_wait() / cancel_wait() / abort_wait()；
  • 在 Linux 后端通常进一步映射到 timerfd + io_poll。

12.2 io_tqueue 负责什么

io_tqueue 负责：

  • 把很多逻辑等待统一排序；
  • 始终找出当前最早截止时间；
  • 控制底层 Timer 应该被设成什么绝对时间；
  • 底层到期后批量回灌上层任务。

所以职责边界可以写成：

1
2
io_tqueue = 逻辑调度器
io_timer_t = 单个真实 Timer 的异步封装

12.3 为什么这里用绝对时间

Lely 这里普遍采用绝对截止时间，而不是“从现在开始延时 N ms”。

好处是：

  • 避免连续重设时的累计漂移；
  • 重新编程时只需维护“最早绝对 deadline”；
  • 和 timespec_cmp() 及堆排序天然配合。

13. io_tqueue 与 can_net 的关系

这里是 CANopen 学习里非常关键的一环。

13.1 can_net 自己也有一个 pheap

can_net 内部也维护一个协议层定时器堆：

1
can_net.timer_heap

其中放的是很多 can_timer_t，例如：

  • 协议超时；
  • 周期发送；
  • 其他网络状态机的时间事件。

它的职责是：

在“协议层很多定时器”中，选出最早的一个。

13.2 io_can_net 会创建 io_tqueue

I/O 桥接层 io_can_net 会创建一个 io_tqueue，并至少使用它来管理：

  • wait_next：等待下一个协议层截止时间；
  • wait_confirm：等待 CAN 帧写确认相关超时。

于是关系变成：

1
2
3
4
5
6
7
8
9
协议层多个 can_timer_t
    ↓
can_net.timer_heap
    ↓
协议层最早截止时间
    ↓
io_tqueue.queue
    ↓ 再与其他 I/O 等待合并排序
一个 io_timer_t

13.3 can_net.next 怎样进入 io_tqueue

can_net 会根据自己堆顶 can_timer_t 的开始时间，更新：

1
net->next

随后 io_can_net 这一桥接层会把这个时间包装进 wait_next.value，再提交到 io_tqueue。

因此：

1
2
can_net 只负责决定“协议层下一次该何时唤醒”
io_tqueue 负责把这个唤醒点和其他 I/O 定时等待一起复用到底层 Timer

13.4 到期后如何回灌给 can_net

底层 Timer 到期以后：

  1. io_tqueue_wait_func() 发现 wait_next 已到期；
  2. wait_next.task 被投递；
  3. 对应的回调里再触发 can_net 继续处理协议层时间推进；
  4. can_net 可能重新算出新的 next；
  5. 再次经由 io_tqueue_submit_wait() 进入队列。

这就形成了一个闭环。

14. 两级堆的准确含义

很容易把 can_net.timer_heap 和 io_tqueue.queue 混成一个东西，但其实它们分层非常明确。

14.1 第一层：can_net.timer_heap

作用：

  • 管协议层事件；
  • 元素是 can_timer_t；
  • 决定“CANopen 协议下一次什么时候要跑”。

14.2 第二层：io_tqueue.queue

作用：

  • 管 I/O 层逻辑等待；
  • 元素是 io_tqueue_wait；
  • 决定“底层唯一 Timer 下次什么时候真正要唤醒”。

14.3 为什么不能直接只有一层

因为这两层处理的是不同抽象：

  • can_net.timer_heap 关心的是协议状态机；
  • io_tqueue.queue 关心的是异步 I/O 等待复用。

所以可以把它们理解成：

1
2
第一层：协议事件调度
第二层：I/O 定时复用

15. 最终心智模型

把整套机制压缩成一句最容易记忆的话：

pheap 负责让“当前最早截止时间”始终出现在 root；当 root 被删除时，通过重新融合 root 的各个子树，把新的全局最小值重建成新的 root；io_tqueue 就利用这个特性，把很多逻辑等待压缩到底层唯一的 io_timer_t 上；而 can_net 又在它之上先做了一层协议定时器筛选。

最后给一个完整总图：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
flowchart TD
    A[很多 can_timer_t] --> B[can_net.timer_heap]
    B --> C[选出协议层最早截止时间]
    C --> D[封装为 wait_next]
    D --> E[io_tqueue.queue]
    F[wait_confirm / 其他逻辑等待] --> E
    E --> G[pheap_first = 当前全局最早等待]
    G --> H[io_timer_settime 到 tq.next]
    H --> I[唯一 io_timer_wait 已提交]
    I --> J[底层 Timer 到期]
    J --> K[io_tqueue_wait_func]
    K --> L[循环取堆顶并删除所有已过期 root]
    L --> M[每删一次 root 都通过 merge 重建新 root]
    M --> N[剩余堆顶重新设置到底层 Timer]
    N --> O[到期任务投递到 executor]
    O --> P[协议层 / I/O 层继续运行]

参考资料

Lely 官方源码

  • lely-core/include/lely/io2/tqueue.h
  • lely-core/src/io2/tqueue.c
  • lely-core/include/lely/util/pheap.h
  • lely-core/src/util/pheap.c
  • lely-core/src/io2/can_net.c
  • lely-core/src/can/net.c

前置学习笔记

  • Lely CANopen Timer 的实现原理与运行流程.md