Lely CANopen IO2 CAN 与 SocketCAN 完整机制学习笔记

在这里插入图片描述

@[toc]

1. CANopen、liblely-co、io2 与 SocketCAN 的边界

Lely CANopen 协议栈本身是被动状态机。它不直接拥有 Linux socket，也不自行创建线程。Linux 上的实际 I/O 由 liblely-io2 提供，任务调度由 liblely-ev 提供。

1
2
3
4
5
6
7
8
9
10
flowchart TB
    APP[业务代码 / CANopen Master] --> CO[liblely-co / coapp]
    CO --> NET[CAN network / protocol state machine]
    NET --> IO2[liblely-io2 abstract CAN API]
    IO2 --> CHAN[Linux CanChannel]
    CHAN --> SOCK[SocketCAN raw socket]
    SOCK --> KERNEL[Linux CAN networking]
    IO2 --> EXEC[Executor]
    EXEC --> EVL[ev_loop]
    EVL --> POLL[epoll/poll backend]

可以用一句话区分：

1
2
CANopen 层决定“这是什么协议报文、状态机如何变化”；
io2 层决定“报文怎样异步进入或离开 Linux CAN socket”。

2. CanController 与 CanChannel 必须严格区分

2.1 CanController：网络接口控制面

CanController 代表一个 Linux CAN 网络接口，例如 can0、can1 或 vcan0。它主要完成：

  • 名称与接口索引转换；
  • 读取 CAN 接口能力标志；
  • 查询或设置标称位率、CAN FD 数据位率；
  • 查询总线状态；
  • 将接口置为 UP 或 DOWN；
  • 尝试设置接口发送队列长度。

它不保存 CAN 报文收发队列，也不负责执行异步读写。

2.2 CanChannel：数据面

CanChannel 代表一个可执行读写操作的 CAN 通道。Linux 实现内部拥有：

  • SocketCAN 文件描述符 fd；
  • io_poll_watch；
  • 三个内部任务 rxbuf_task/read_task/write_task；
  • 用户态单生产者单消费者环形缓冲区；
  • 待处理读队列；
  • 待发送队列；
  • 等待发送确认队列；
  • shutdown 与任务已投递状态位。

2.3 两者的对象关系

1
2
3
4
5
6
7
8
9
10
CanController("can1")
    └─ 指向 Linux 接口 can1

CanChannel(poll, exec, 256, true)
    ├─ 绑定 Poll 与 Executor
    ├─ 创建 256 槽用户态接收队列
    └─ 配置发送完成是否等待 MSG_CONFIRM

channel.open(controller)
    └─ 创建 socket，并 bind 到 controller 对应的 ifindex

3. 抽象 C 接口：两个虚函数表

include/lely/io2/can.h 定义两个抽象类型：

1
2
io_can_ctrl_t  → CAN 控制器接口
io_can_chan_t  → CAN 通道接口

它们和 ev_exec_t 一样，都是 C 语言虚函数表对象。

3.1 Controller 虚函数表

1
2
3
4
5
6
stop
stopped
restart
get_bitrate
set_bitrate
get_state

3.2 Channel 虚函数表

1
2
3
4
5
6
get_dev
get_flags
read
submit_read
write
submit_write

其中：

  • read/write 是同步阻塞接口；
  • submit_read/submit_write 是异步请求提交接口；
  • io_dev_t 提供统一的 cancel/abort/context/executor 能力。

3.3 请求对象为什么内嵌 ev_task

异步读请求的逻辑布局：

1
2
3
4
5
6
io_can_chan_read
├─ msg      输出普通 CAN 帧
├─ err      输出 CAN 错误帧
├─ tp       输出接收时间
├─ task     完成后投递给 Executor
└─ result   result + errc

异步写请求：

1
2
3
4
io_can_chan_write
├─ msg      待发送帧指针
├─ task     完成后投递给 Executor
└─ errc     完成状态

这说明“异步操作”和“完成回调”不是两个独立分配的对象，而是一个请求对象内部直接携带完成任务。

4. C++ 包装层：RAII、异常与回调对象

4.1 include/lely/io2/linux/can.hpp

Linux 平台类负责资源所有权：

1
2
3
4
5
CanController 构造 → io_can_ctrl_create_*
CanController 析构 → io_can_ctrl_destroy

CanChannel 构造 → io_can_chan_create
CanChannel 析构 → io_can_chan_destroy

创建失败时，构造函数读取 Lely 错误码并抛出异常。

4.2 include/lely/io2/can.hpp

平台无关 C++ 层提供三种使用方式：

方式 API 完成模型
同步 read() / write() 当前线程阻塞
回调 submit_read() / submit_write() 完成任务调用 callable
Future async_read() / async_write() Promise ready 后由调用方等待或续接

回调包装器会在完成回调结束后删除自身；显式 CanChannelRead/CanChannelWrite 对象则由调用方管理生命周期。

5. CanController("can1") 的真实调用链

1
2
3
4
5
6
7
8
9
CanController(name)
  → io_can_ctrl_create_from_name(name, txlen)
  → if_nametoindex(name)
  → io_can_ctrl_create_from_index(index, txlen)
  → io_can_ctrl_init()
      ├─ 保存 ifindex 与接口名
      ├─ 尝试读取 CAN netlink 属性
      ├─ 保存 CAN/CAN FD 能力标志
      └─ 设置接口发送队列长度

5.1 默认发送队列长度

源码默认值是 128 帧。构造 Controller 时会尝试保证接口使用指定的发送队列长度。这是一个容易忽略的副作用：

1
2
构造 Controller
≠ 单纯只读对象

普通用户进程可能因缺少网络管理权限而失败。官方教程也专门提示了 Operation not permitted 与 txqueuelen 的关系。

生产环境应在服务启动前完成接口配置，例如由 systemd-networkd、NetworkManager、启动脚本或具备权限的初始化服务执行，而不是无条件依赖业务进程修改网络接口。

5.2 get_bitrate()

Linux 实现从 CAN netlink 属性读取：

1
2
nominal bitrate → 仲裁阶段位率
data bitrate    → CAN FD 数据阶段位率；经典 CAN 为 0

需要注意：初始化阶段对“不支持 CAN 属性”的部分接口进行了容忍，但 get_bitrate() 本身若无法取得属性仍会返回错误。因此，以下代码是严格校验策略，而不是适用于所有 SocketCAN 类型的通用策略：

1
controller->get_bitrate(&nominal_bitrate, &data_bitrate);

对 vcan、部分 SLCAN 或定制虚拟接口，应根据本地内核驱动能力决定是否启用位率校验。

5.3 get_state()

Linux 实现先读取接口 flags：

1
2
接口没有 IFF_UP → CAN_STATE_STOPPED
接口为 UP       → 再读取 CAN netlink state

因此 STOPPED 首先反映网络接口未启用。其他状态依赖驱动提供的 CAN 状态属性。

6. CanChannel(..., 256, true) 的真实含义

1
2
impl_->channel.reset(
    new lely::io::CanChannel(*impl_->poll, *impl_->executor, 256, true));

四个参数分别是：

参数 含义
poll 监听 SocketCAN fd 的 I/O readiness
executor 执行内部 I/O 任务与完成任务
256 Lely 用户态接收环形队列的槽数
true 启用发送确认等待 txwait

6.1 256 不是什么

它不是：

  • Linux SO_RCVBUF 字节数；
  • CAN 控制器硬件 FIFO 深度；
  • CANopen PDO 数量；
  • 一次事件循环最多处理的报文数；
  • 发送队列长度。

它控制的是：

1
struct io_can_frame rxbuf[256]

以及配套 spscring 的逻辑容量。

6.2 txwait=true 的含义

开启后，Lely 会：

  1. 启用 SocketCAN 本地回环；
  2. 启用接收本 socket 自己发送的帧；
  3. sendmsg() 成功后把写请求放入 confirm_queue；
  4. 等收到带 MSG_CONFIRM 的回送帧；
  5. 匹配成功后才完成异步写请求。

因此：

1
2
3
4
5
async_write Future ready
≠ sendmsg() 已把帧交给内核

async_write Future ready 且 txwait=true
≈ SocketCAN 已回送该帧的发送确认

若 txwait=false，成功 sendmsg() 后即可完成写请求，吞吐更高，但完成语义更弱。

7. CanChannel::open()：SocketCAN socket 的创建过程

调用链：

1
2
3
4
5
6
7
8
9
CanChannel::open(controller)
  → io_can_chan_open()
      ├─ 校验请求 flags
      ├─ socket(AF_CAN, SOCK_RAW | SOCK_CLOEXEC, CAN_RAW)
      ├─ bind(sockaddr_can{ifindex})
      ├─ 可选启用错误帧过滤
      ├─ 可选启用 CAN FD
      ├─ 配置 loopback / own messages / send buffer
      └─ io_can_chan_impl_set_fd()

7.1 默认打开 flags

用户代码调用：

1
channel->open(*controller);

未显式传入 CanBusFlag，因此默认是 NONE。这意味着：

  • 不主动订阅 SocketCAN 错误帧；
  • 不启用 CAN FD；
  • 不启用 BRS。

txwait 与 CanBusFlag 是两组不同概念：

1
2
3
CanBusFlag → 总线报文能力和错误帧接收

txwait     → 写操作完成时是否等待本地发送确认

7.2 替换 fd 时的行为

open()、assign()、release() 最终都经过 io_can_chan_impl_set_fd()。替换 fd 时会：

  • 停止原 fd 的 Poll 监听；
  • 中止环形队列等待；
  • 清空用户态接收队列；
  • 交换新旧 fd；
  • 取消所有 pending read/write/confirmation 请求。

所以 open() 不是无副作用的重复操作。对已经有 pending I/O 的 channel 再次 open，会导致这些操作以取消状态完成。

8. io_can_chan_impl：Poll watch、三个内部任务与私有队列

io_can_chan_impl 不是一个简单的 socket 包装器。它把 SocketCAN fd、Poll 监听对象、Executor 任务、接收环形缓冲区和三类请求队列组合成一个异步状态机。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
io_can_chan_impl
├─ poll / ctx / exec
├─ txwait
├─ watch
├─ rxbuf_task
├─ read_task
├─ write_task
├─ rxring + rxbuf
├─ fd / flags / events
├─ shutdown
├─ rxbuf_posted / read_posted / write_posted
├─ read_queue
├─ write_queue
├─ confirm_queue
└─ current_write

8.1 初始化代码实际建立了什么

源码初始化：

1
2
3
4
5
6
7
8
9
impl->watch = (struct io_poll_watch)IO_POLL_WATCH_INIT(
        &io_can_chan_impl_watch_func);

impl->rxbuf_task = (struct ev_task)EV_TASK_INIT(
        impl->exec, &io_can_chan_impl_rxbuf_task_func);
impl->read_task = (struct ev_task)EV_TASK_INIT(
        impl->exec, &io_can_chan_impl_read_task_func);
impl->write_task = (struct ev_task)EV_TASK_INIT(
        impl->exec, &io_can_chan_impl_write_task_func);

这段代码只完成“对象与回调入口绑定”，不会立刻监听 fd，也不会立刻执行任务。

对象 类型 初始化后绑定的入口 谁触发它 主要职责
watch io_poll_watch io_can_chan_impl_watch_func() Poll 后端检测到 IN/OUT/ERR 把 readiness 转换成内部任务投递
rxbuf_task ev_task io_can_chan_impl_rxbuf_task_func() 首次读、等待确认、Poll 可读事件 recvmsg()、填充 rxring、处理 MSG_CONFIRM
read_task ev_task io_can_chan_impl_read_task_func() submit_read() 或 rxring 产生数据 用接收帧满足 read_queue
write_task ev_task io_can_chan_impl_write_task_func() submit_write() 或 Poll 可写事件 sendmsg()、重试阻塞写、进入确认队列

IO_POLL_WATCH_INIT() 会初始化整个 watch 记录：写入回调函数，将内部 fd 置为 -1，初始化红黑树节点，并把内部事件掩码清零。真正把目标 fd、事件掩码和 watch 注册到 Poll 后端的是后续的：

1
io_poll_watch(impl->poll, impl->fd, events, &impl->watch);

EV_TASK_INIT(exec, func) 则把任务固定绑定到同一个 Executor，并设置执行函数。任务只有在调用 ev_exec_post() 后才进入 Executor/event loop 的可执行队列。Poll 接口采用一次通知、再次注册的语义：一次注册在事件发生并回调后即被消费；若仍需接收后续事件，channel 必须再次调用 io_poll_watch() 注册。watch_func() 因此会从 impl->events 中移除本次已处理事件，并重新注册尚未处理的事件掩码。

8.2 四个回调载体的连接关系

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
flowchart LR
    Sock["SocketCAN fd"] --> Poller["io_poll_t / epoll backend"]
    Poller --> Watch["watch.func<br/>io_can_chan_impl_watch_func"]
    Watch -->|"IO_EVENT_IN / ERR"| RxTask["rxbuf_task"]
    Watch -->|"IO_EVENT_OUT / ERR"| WrTask["write_task"]

    RdReq["read_queue"] --> RdTask["read_task"]
    RdTask -->|"rxring empty"| RxTask
    RxTask --> Ring["rxring + rxbuf"]
    Ring -->|"consumer signal"| RdTask

    WrReq["write_queue"] --> WrTask
    WrTask -->|"txwait = true"| Confirm["confirm_queue"]
    Confirm --> RxTask

    RxTask --> Exec["Executor"]
    RdTask --> Exec
    WrTask --> Exec

图中最重要的边界是：

1
2
3
watch 负责“事件转任务”；
三个 ev_task 负责“在 Executor 上推进状态机”；
用户请求 task 负责“完成通知与 Future/回调”。

8.3 三个内部任务不是三个线程

三个 ev_task 都绑定到 impl->exec，它们只是可重复投递的工作对象：

1
2
3
rxbuf_task.func = io_can_chan_impl_rxbuf_task_func
read_task.func  = io_can_chan_impl_read_task_func
write_task.func = io_can_chan_impl_write_task_func

实际在哪个 OS 线程执行，取决于哪个线程正在驱动对应的 ev_loop。若多个 worker 同时运行同一个 loop，任务可以由不同 worker 取出；channel 通过 mutex、队列状态和 posted 标志保护内部状态。

8.4 posted 标志保证内部任务不会重复入队

标志 对应任务 置 1 的典型位置 清零或更新位置
rxbuf_posted rxbuf_task 需要读取 socket 或等待确认时 rxbuf_task 结束前根据后续工作更新
read_posted read_task 首个 read 请求入队、ring consumer 被唤醒时 read_task 结束前更新
write_posted write_task 首个 write 请求入队、fd 可写时 write_task 结束前更新

同一个 ev_task 内嵌同一个侵入式链表节点，因此不能在尚未出队时再次 post。posted 标志就是每个内部任务的“单实例在途锁”。

8.5 请求队列、内部任务和 Poll 的统一状态机

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
flowchart TD
    Init["初始化 watch 与三个 ev_task"] --> Idle["等待业务请求"]

    Idle -->|"submit_read"| ReadQ["请求进入 read_queue"]
    ReadQ --> PostRead["post read_task"]
    PostRead --> RunRead["read_task 检查 rxring"]
    RunRead -->|"有帧"| CompleteRead["完成 read 请求"]
    RunRead -->|"无帧"| NeedRx["post rxbuf_task"]

    NeedRx --> RunRx["rxbuf_task 调用 recvmsg"]
    RunRx -->|"收到普通帧"| CommitRing["提交到 rxring"]
    CommitRing --> SignalRead["consumer signal"]
    SignalRead --> PostRead
    RunRx -->|"EAGAIN"| WatchIn["注册 IO_EVENT_IN"]

    Idle -->|"submit_write"| WriteQ["请求进入 write_queue"]
    WriteQ --> PostWrite["post write_task"]
    PostWrite --> RunWrite["write_task 调用 sendmsg"]
    RunWrite -->|"EAGAIN"| WatchOut["注册 IO_EVENT_OUT"]
    RunWrite -->|"成功且 txwait=false"| CompleteWrite["完成 write 请求"]
    RunWrite -->|"成功且 txwait=true"| ConfirmQ["请求进入 confirm_queue"]
    ConfirmQ --> NeedRx
    RunRx -->|"MSG_CONFIRM"| MatchConfirm["匹配 confirm_queue"]
    MatchConfirm --> CompleteWrite

    WatchIn --> PollWait["event loop 进入 Poll 等待"]
    WatchOut --> PollWait
    PollWait --> Ready["fd readiness"]
    Ready --> WatchFunc["watch_func"]
    WatchFunc -->|"IN / ERR"| NeedRx
    WatchFunc -->|"OUT / ERR"| PostWrite

8.6 为什么内部任务没有单独调用 on_task_init()

submit_read() 和 submit_write() 会对“用户请求自身的 completion task”调用 ev_exec_on_task_init()。请求随后可能停留在 read_queue、write_queue、confirm_queue 或系统调用过程中，这个 outstanding-work 引用保证 event loop 在等待外部 CAN 事件时不会因全局任务队列暂时为空而退出。

三个内部任务是推进这些请求的执行工具；它们通过普通 ev_exec_post() 投递，不另行持有长期 work 引用。完成请求时，辅助函数遵守：

1
2
先 post 用户 completion task
后 on_task_fini 释放请求的 work 引用

因此从提交到完成，请求始终满足至少一个条件：位于私有队列、正在执行、已经进入 Executor 队列，或由 outstanding-work 引用保活。

9. 异步读完整机制

9.1 提交阶段

1
2
3
4
5
submit_read(request)
  ├─ 若 request.task.exec 为空，绑定 channel 默认 executor
  ├─ ev_exec_on_task_init(exec)
  ├─ request 加入 read_queue
  └─ 必要时 post 内部 read_task

这里的 on_task_init() 非常关键：用户读请求已经进入 channel 私有队列，但尚未进入 event loop 的可执行队列。它必须用 outstanding work 引用保持 loop 存活。

9.2 read_task 阶段

read_task 尝试从 rxring 取帧：

1
2
3
4
5
6
7
8
rxring 有帧
  → 填充 read->msg / read->err / read->tp
  → 设置 result
  → 请求移到局部完成队列

rxring 无帧
  → 在 spscring 注册 consumer wait
  → 确保 rxbuf_task 或 Poll 正在等待 fd 可读

9.3 Poll 与 rxbuf_task

若 SocketCAN 非阻塞读取返回 EAGAIN/EWOULDBLOCK：

1
2
3
4
5
rxbuf_task 不持续忙轮询
  → io_poll_watch(fd, IO_EVENT_IN)
  → event loop 进入 Poll
  → fd 可读时 watch_func 被调用
  → watch_func post rxbuf_task

watch_func 本身只做轻量状态更新和任务投递，不在 Poll 回调中直接执行完整读循环。

9.4 rxbuf_task 对两类帧的分流

1
2
3
4
5
6
7
普通接收帧
  → 写入 rxring
  → 后续由 read_task 交给用户读请求

MSG_CONFIRM 帧
  → 不进入普通接收 ring
  → 用于完成 confirm_queue 中的写请求

9.5 完成阶段

完成的 read 请求先移入局部队列，解锁后统一执行：

1
2
3
ev_task_queue_post(queue)
  → ev_exec_post(request.task)
  → ev_exec_on_task_fini(request.task.exec)

这正是前一篇 Executor/Future 笔记中的“先 post，后 fini”。

10. 异步写完整机制

10.1 提交阶段

1
2
3
4
5
6
submit_write(request)
  ├─ 校验 CAN FD/BRS flags 是否被 channel 支持
  ├─ 绑定默认 executor
  ├─ on_task_init()
  ├─ 加入 write_queue
  └─ 必要时 post write_task

10.2 write_task 发送阶段

1
2
3
write_queue 首元素
  → 转换 Lely can_msg 为 Linux can_frame/canfd_frame
  → sendmsg()

结果有三类：

结果 行为
成功，txwait=false 立即完成请求
成功，txwait=true 请求进入 confirm_queue
EAGAIN/EWOULDBLOCK 放回 write_queue 头部，监听 IO_EVENT_OUT
其他错误 以对应 errno 完成请求

10.3 为什么设置最小发送缓冲区

channel 初始化 socket 时会把 SO_SNDBUF 设为最小值。源码意图是让发送压力表现为：

1
阻塞或 EAGAIN

而不是直接返回 ENOBUFS。这样异步写状态机可以把请求留在队列中，并等待 IO_EVENT_OUT 后重试。

10.4 写确认匹配

发送确认使用 can_msg_cmp() 比较：

1
2
3
4
id
flags
length
data bytes

收到 MSG_CONFIRM 后，在 confirm_queue 中查找首个完全相等的写请求。

匹配到某个请求时：

  • 该请求标记成功；
  • 它之前仍未确认的请求标记 EIO；
  • 完成任务统一投递给 Executor。

这一实现隐含了发送确认顺序假设。若调试中出现偶发 EIO，应同时检查：

  • 是否重复发送完全相同的帧；
  • SocketCAN own-message 的顺序；
  • channel 是否在发送期间被 reopen/release；
  • 是否关闭了 txwait 或改变了 socket 选项；
  • 是否存在并发 cancel。

11. CAN 帧转换层

src/io2/linux/can_msg.h 在 Lely 与 Linux 结构之间转换。

11.1 经典 CAN

1
2
3
4
5
Lely CAN_FLAG_IDE ↔ Linux CAN_EFF_FLAG
Lely CAN_FLAG_RTR ↔ Linux CAN_RTR_FLAG
11 位 ID          ↔ CAN_SFF_MASK
29 位 ID          ↔ CAN_EFF_MASK
长度上限          ↔ CAN_MAX_LEN

RTR 帧只携带 DLC，不复制数据区。

11.2 CAN FD

1
2
3
4
CAN_FLAG_FDF ↔ canfd_frame
CAN_FLAG_BRS ↔ CANFD_BRS
CAN_FLAG_ESI ↔ CANFD_ESI
长度上限     ↔ CANFD_MAX_LEN

经典 CAN 转换函数会拒绝带 FDF 标志的报文；CAN FD 转换函数也会拒绝没有 FDF 标志的报文。

11.3 错误帧独立处理

Linux CAN_ERR_FLAG 报文不会被普通 can_frame2can_msg() 接受，而是走单独的 can_frame2can_err() 路径。是否能收到错误帧取决于 channel open 时是否启用 CanBusFlag::ERR。

12. Future 怎样接到 CAN channel 与 event loop

src/io2/can.c 使用 Promise 的内部数据区保存完整异步请求：

1
2
3
Promise allocation
├─ promise pointer
└─ io_can_chan_read 或 io_can_chan_write

12.1 async_read()

1
2
3
4
5
6
io_can_chan_async_read()
  → ev_promise_create()
  → 从 promise 数据区取得 async_read 对象
  → IO_CAN_CHAN_READ_INIT 初始化 read request 与 completion task
  → io_can_chan_submit_read()
  → 返回 future

读请求的 completion task 最终执行：

1
2
3
4
5
io_can_chan_async_read_func(task)
  → 从 task 反查 io_can_chan_read
  → 从 read 反查 io_can_chan_async_read
  → ev_promise_set(promise, &read->r)
  → ev_promise_release(promise)

C++ 返回类型是：

1
ev::Future<int, int>

结果含义：

1
2
3
result = 1   收到普通 CAN 帧
result = 0   收到 CAN 错误帧
result = -1  失败或取消，错误号位于 errc

12.2 async_write()

写路径与读路径使用相同的 Promise/Future 包装模式：

1
2
3
4
5
io_can_chan_async_write()
  → 创建 promise/future
  → 初始化 io_can_chan_write 与 completion task
  → io_can_chan_submit_write()
  → 返回 future

完成任务运行时：

1
2
3
4
io_can_chan_async_write_func(task)
  → 从 task 反查 io_can_chan_write
  → ev_promise_set(promise, &write->errc)
  → ev_promise_release(promise)

C++ 返回类型是：

1
ev::Future<void, int>

当 txwait=true 时，Future 只有在匹配到 SocketCAN MSG_CONFIRM 后才 ready；当 txwait=false 时，sendmsg() 成功即可进入完成路径。

12.3 与 event loop 的完整连接

下面的时序图使用 Evl 作为 Mermaid participant 标识，避免使用保留关键字 loop 导致解析错误。

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
34
35
36
37
38
sequenceDiagram
    participant App as Application
    participant Chan as CanChannel
    participant Exec as Executor
    participant Evl as ev_loop
    participant Poller as io_poll
    participant Sock as SocketCAN fd
    participant Done as Completion task
    participant Prm as Promise/Future

    App->>Chan: async_read() or async_write()
    Chan->>Exec: on_task_init(request task)
    Chan->>Chan: enqueue request in private queue
    Chan->>Exec: post internal task
    Exec->>Evl: enqueue internal task
    Evl->>Exec: run internal task
    Exec->>Sock: recvmsg() or sendmsg()

    opt EAGAIN or EWOULDBLOCK
        Exec->>Poller: io_poll_watch(fd, IN or OUT, watch)
        Exec-->>Evl: internal task returns
        Evl->>Poller: wait for I/O readiness
        Sock-->>Poller: fd becomes ready
        Poller->>Chan: watch_func(watch, events)
        Chan->>Exec: post rxbuf_task or write_task
        Exec->>Evl: enqueue internal task again
        Evl->>Exec: run internal task again
        Exec->>Sock: retry recvmsg() or sendmsg()
    end

    Sock-->>Exec: frame, error, or confirmation
    Exec->>Chan: update ring and request queues
    Chan->>Exec: post completion task
    Chan->>Exec: on_task_fini(request task)
    Exec->>Evl: enqueue completion task
    Evl->>Done: run completion task
    Done->>Prm: ev_promise_set()
    Prm-->>App: Future becomes ready

这里需要把“事件循环”和“Poll”分开理解：

1
2
3
4
5
Executor / ev_loop：运行 ev_task；
Poll：等待 fd readiness 并调用 watch 回调；
watch_func：只把 readiness 转换成新的内部 task；
内部 task：执行 recvmsg/sendmsg 并推进队列状态；
completion task：设置结果、回调用户代码或完成 Promise。

12.4 event loop 为什么能在等待 CAN 帧时保持运行

1
2
3
4
5
6
7
8
9
10
11
12
13
14
flowchart TD
    Submit["提交异步请求"] --> Guard["on_task_init 增加 outstanding work"]
    Guard --> PrivateQ["请求停留在 channel 私有队列"]
    PrivateQ --> Internal["内部 task 尝试 I/O"]
    Internal -->|"EAGAIN"| Register["注册 Poll watch"]
    Register --> Empty["event loop 可执行队列暂时为空"]
    Empty --> Decision{"outstanding work 是否为 0"}
    Decision -->|"否"| WaitIO["ev_loop 调用 Poll 等待 I/O"]
    WaitIO --> Ready["fd ready"]
    Ready --> WatchCb["watch_func post 内部 task"]
    WatchCb --> Internal
    Internal -->|"完成"| PostDone["post completion task"]
    PostDone --> Fini["on_task_fini 释放 work 引用"]
    Fini --> RunDone["event loop 执行 completion task"]

关键点：Poll watch 本身不是 Future，也不是用户回调。event loop 之所以不会提前退出，根本原因是未完成请求仍持有 on_task_init() 建立的 outstanding-work 引用。

12.5 读路径的任务与队列迁移

1
2
3
4
5
6
7
8
9
10
11
12
flowchart LR
    AR["async_read"] --> RQ["read_queue"]
    RQ --> RT["read_task"]
    RT -->|"rxring 有数据"| RC["read completion task"]
    RT -->|"rxring 为空"| RX["rxbuf_task"]
    RX -->|"recvmsg EAGAIN"| WI["watch IN"]
    WI --> WF["watch_func"]
    WF --> RX
    RX -->|"普通 CAN 帧"| Ring["rxring"]
    Ring --> CS["consumer signal"]
    CS --> RT
    RC --> PR["Promise set"]

12.6 写路径的任务与队列迁移

1
2
3
4
5
6
7
8
9
10
11
12
flowchart LR
    AW["async_write"] --> WQ["write_queue"]
    WQ --> WT["write_task"]
    WT -->|"sendmsg EAGAIN"| WO["watch OUT"]
    WO --> WF["watch_func"]
    WF --> WT
    WT -->|"success and txwait=false"| WC["write completion task"]
    WT -->|"success and txwait=true"| CQ["confirm_queue"]
    CQ --> RX["rxbuf_task"]
    RX -->|"MSG_CONFIRM"| Match["match queued message"]
    Match --> WC
    WC --> PR["Promise set"]

12.7 三类 task 不要混淆

task 类型 典型对象 是否长期复用 作用
channel 内部任务 rxbuf_task/read_task/write_task 是 推进 SocketCAN I/O 状态机
请求 completion task io_can_chan_read.task、io_can_chan_write.task 每个请求一个 发布结果，执行回调或设置 Promise
Future 等待任务 调用方通过 Future API 注册的 task 由调用方/Future 管理 Future ready 后继续上层状态机

Promise ready 不等于直接在 recvmsg() 或 sendmsg() 的调用栈中运行上层 CANopen 逻辑。请求 completion task 先进入 Executor 队列；如果 Future 上还有等待任务，Future 再按其自身规则把等待任务提交给相应 Executor。