Lely CANopen Future/Promise 机制原理与源码详解

在这里插入图片描述

@[toc]

1. 先给结论：Lely Future 到底是什么

Lely 的 Future/Promise 不是一个“阻塞等待结果”的线程同步工具，而是一个和 executor、task、event loop 深度结合的一次性异步完成通知与结果共享机制。

可以先记住以下五句话：

  1. ev_promise_t 是异步结果的生产端，负责把共享状态从未完成推进到完成；
  2. ev_future_t 是观察端，负责检查结果是否完成、读取结果、登记完成后的任务；
  3. Future 本身不会执行 I/O，也不会主动运行回调；真正执行回调的是 task 所属的 executor；
  4. ev_future_submit() 不阻塞，而是把 task 暂存在 Future 内部队列，Future ready 后再提交给 executor；
  5. Promise 和 Future 不是两个独立对象，它们共享同一块内存、同一个引用计数和同一个生命周期。

官方 C 接口头文件直接强调：与 C++11 的 Future/Promise 不同，Lely 提供非阻塞语义；用户不是等待 Future，而是提交一个在 Future ready 后执行的 task。

1
2
3
4
5
6
7
8
9
10
11
12
13
异步操作发起
    ↓
返回 Future 给调用方
    ↓
操作完成方通过 Promise 写入结果
    ↓
Future 变为 ready
    ↓
Future 中登记的 task 被提交到 executor
    ↓
executor/event loop 执行 task
    ↓
调用方读取结果或继续异步链

因此，理解 Lely Future 的关键不是“谁在等待”，而是：

谁持有 Promise、谁持有 Future、结果存在哪里、Future ready 后哪个 task 被投递到哪个 executor。

2. 三个源码文件各自负责什么

文件 层次 主要职责
future.h C 公共接口 定义 Promise/Future 的外部语义、引用计数接口、结果发布、task 提交、取消、组合 Future。
future.c C 底层实现 定义实际内存结构、状态机、原子同步、等待 task 队列、生命周期和 when_all/when_any。
future.hpp C++ 类型封装 提供 RAII、模板化结果、异常捕获、continuation、implicit unwrapping 和辅助函数。

推荐阅读顺序：

1
2
3
4
5
6
future.h
  ↓ 先理解对外契约
future.c
  ↓ 再理解状态、内存和并发实现
future.hpp
  ↓ 最后理解 C++ 如何组合这些底层机制

不要一开始就从 future.hpp 的模板代码往下钻，否则很容易被 AsyncTask、类型萃取和 Result<T, E> 淹没，而没有建立共享状态的核心模型。

3. 四个核心角色

3.1 Promise：结果生产端

ev_promise_t 用于完成一次异步操作。它的典型持有者包括：

  • 异步 CAN 读写操作；
  • 异步 SDO 请求对象；
  • 定时器等待操作；
  • C++ AsyncTask；
  • when_all()、when_any() 的内部聚合对象。

Promise 的核心动作只有一次：

1
int ev_promise_set(ev_promise_t *promise, void *value);

成功完成后，Promise 对应的 Future 永久进入 ready 状态，不能重新复位后再次使用。

所以 Promise 更接近：

1
一次性完成开关 + 结果发布端

而不是可重复发送消息的队列。

3.2 Future：结果观察端

ev_future_t 主要提供三种能力：

1
2
3
int ev_future_is_ready(const ev_future_t *future);
void *ev_future_get(const ev_future_t *future);
void ev_future_submit(ev_future_t *future, struct ev_task *task);

它不负责完成结果，只负责观察状态和登记后续任务。

3.3 Task：Future ready 后要执行的动作

Future 不直接保存 C 函数指针列表，而是保存 ev_task 队列。ev_task 至少包含：

1
2
3
4
5
6
struct ev_task {
    ev_exec_t *exec;
    ev_task_func_t *func;
    struct slnode _node;
    void *_data;
};

这意味着每一个等待 Future 的动作都明确指定：

1
2
3
Future ready 后
    在哪个 executor 上
    执行哪个 task function

3.4 Executor：真正执行 task 的对象

Future 只决定 task 何时具备提交条件，executor 决定 task 在哪里、何时运行。

在典型 Lely CANopen 程序中，executor 往往来自 ev_loop_t：

1
2
3
4
5
Future
  └─ 等待 task
       └─ task.exec
            └─ ev_loop 的 executor
                 └─ event loop 线程执行 task.func

因此，Future 与 event loop 的关系是：

Future 负责完成依赖，event loop 负责调度执行。

4. 最重要的内存布局：Promise 内嵌 Future

future.c 中的两个核心结构如下：

1
2
3
4
5
6
7
8
9
10
11
12
struct ev_future {
    atomic_int state;
    void *value;
    mtx_t mtx;
    struct sllist queue;
};

struct ev_promise {
    atomic_size_t refcnt;
    ev_promise_dtor_t *dtor;
    struct ev_future future;
};

实际代码会根据单线程、Windows、C11 原子等构建条件选择不同字段类型，但逻辑结构不变。

对象关系不是：

1
Promise ──指针──> Future

而是：

1
2
3
4
5
6
7
8
9
10
11
12
13
一整块分配内存
┌────────────────────────────┐
│ ev_promise_t               │
│  ├─ refcnt                 │
│  ├─ dtor                   │
│  └─ ev_future_t future     │
│      ├─ state              │
│      ├─ value              │
│      ├─ mtx                │
│      └─ queue              │
├────────────────────────────┤
│ 用户共享数据区 size bytes  │
└────────────────────────────┘

ev_promise_create(size, dtor) 实际分配：

1
calloc(1, EV_PROMISE_SIZE + size);

而 ev_promise_data() 返回 Promise 结构后方的尾随数据区：

1
return (char *)promise + EV_PROMISE_SIZE;

这套布局有三个直接效果。

4.1 Promise 和 Future 共用生命周期

ev_future_acquire() 并不是增加 Future 自己的引用计数，而是先通过 structof() 从内嵌 Future 找回外层 Promise，然后增加 Promise 的 refcnt。

1
2
3
4
5
ev_future_t *
    ↓ structof
对应的 ev_promise_t
    ↓
共享 refcnt

因此 refcnt 的真实语义是：

1
2
3
所有 Promise 引用数量
+
所有 Future 引用数量

4.2 结果对象可以和控制块一次分配

C 接口可用尾随数据区存放任意共享状态；C++ Promise<T, E> 则直接在该区域 placement-new 一个：

1
util::Result<T, E>

这样控制块、Future 状态和结果对象可以在一次分配中完成。

4.3 Future 指针不是独立分配的对象

ev_promise_get_future() 返回的是：

1
&promise->future

外加一次引用计数增加。

所以不能单独 free(future)，也不能假设 Future 的地址来自独立分配。

5. Future 状态机：WAITING、SETTING、READY

future.c 定义了三个状态：

1
2
3
4
5
enum ev_future_state {
    EV_FUTURE_WAITING,
    EV_FUTURE_SETTING,
    EV_FUTURE_READY
};

状态变化只能单向进行：

1
2
3
4
5
stateDiagram-v2
    [*] --> WAITING
    WAITING --> SETTING: ev_promise_set_acquire() 成功
    SETTING --> READY: ev_promise_set_release()
    READY --> READY: 后续 set 请求失败

5.1 WAITING

含义：

  • Promise 尚未被任何完成者占有；
  • Future 尚未 ready；
  • task 可以继续通过 ev_future_submit() 加入等待队列。

5.2 SETTING

含义：

  • 某一个线程已经获得完成权；
  • 其他线程不能再完成同一个 Promise；
  • 结果可能还在构造或写入中；
  • Future 对外仍不应被视为 ready。

5.3 READY

含义：

  • value 已经写入；
  • 结果数据已经发布；
  • 原先等待的 task 已从 Future 队列移出并提交给 executor；
  • 后续新提交的 task 不再排队，会直接提交给 executor。

6. 为什么完成 Promise 要拆成 acquire 和 release 两步

简单接口是：

1
int ev_promise_set(ev_promise_t *promise, void *value);

内部等价于：

1
2
3
4
int result = ev_promise_set_acquire(promise);
if (result)
    ev_promise_set_release(promise, value);
return result;

两阶段设计不是多余封装，而是为了把“抢占完成权”和“构造结果”分开。

6.1 第一步：抢占唯一完成权

多线程下，ev_promise_set_acquire() 使用原子 compare-exchange：

1
2
期望状态：WAITING
目标状态：SETTING

只有一个线程能成功完成这个状态转换。

假设两个完成路径同时到达：

1
2
线程 A：WAITING → SETTING，成功
线程 B：发现状态已不是 WAITING，失败

因此，一次性语义由状态机保证，而不是靠调用者约定。

6.2 第二步：构造并发布结果

获得完成权后，调用者可以：

  • 填充尾随共享数据区；
  • 构造 C++ Result<T, E>；
  • 保存错误码；
  • 保存异常；
  • 决定最终传给 ev_future_get() 的地址。

完成后再调用：

1
void ev_promise_set_release(ev_promise_t *promise, void *value);

它执行以下关键步骤：

1
2
3
4
5
6
1. future->value = value
2. 获取 future->mtx
3. 把 future->queue 整体转移到局部队列
4. release-store state = READY
5. 释放 future->mtx
6. 将局部队列中的 task 提交到各自 executor

6.3 为什么不在持锁状态下执行 executor post

Future 内部互斥锁只用于保护：

  • ready 检查与 task 入队之间的原子关系；
  • Future 等待队列的转移和移除。

它不应该覆盖 executor 的调度过程。否则 executor 可能：

  • 立即运行 task；
  • 再次访问同一个 Future；
  • 引发锁嵌套或死锁；
  • 造成不必要的长临界区。

因此，源码先把等待队列整体转移到局部变量，解锁后再提交 task。

7. ev_future_submit() 如何避免丢失完成通知

这是整套实现中最值得学习的并发细节之一。

表面上存在一个经典竞态：

1
2
3
线程 A：检查 Future 还没 ready
线程 B：把 Future 设为 ready，并唤醒当前等待者
线程 A：此后才把 task 放入等待队列

如果实现不正确，线程 A 的 task 将永久留在队列里，再也不会被提交。

Lely 用同一把 future->mtx 消除了这个窗口。

ev_future_submit() 的核心逻辑是：

1
2
3
4
5
6
7
8
9
10
11
ev_exec_on_task_init(exec);
mtx_lock(&future->mtx);

if (ev_future_is_ready(future)) {
    mtx_unlock(&future->mtx);
    ev_exec_post(exec, task);
    ev_exec_on_task_fini(exec);
} else {
    sllist_push_back(&future->queue, &task->_node);
    mtx_unlock(&future->mtx);
}

而 ev_promise_set_release() 也在同一把锁下：

1
2
转移全部等待 task
设置 READY

因此只会出现两种情况。

情况 A：submit 先获得锁

1
2
3
4
5
6
7
submit：看到未 ready
submit：task 入队
submit：解锁
set_release：获得锁
set_release：取走该 task
set_release：设置 ready
set_release：解锁并提交 task

情况 B：set_release 先获得锁

1
2
3
4
5
6
set_release：取走原等待队列
set_release：设置 ready
set_release：解锁
submit：获得锁
submit：看到 ready
submit：直接提交 task

不存在第三种“既没入队，也没直接提交”的状态。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
sequenceDiagram
    participant S as submit线程
    participant F as Future
    participant P as Promise完成线程
    participant E as Executor

    S->>F: lock(mtx)
    alt Future 尚未 ready
        S->>F: task 加入 queue
        S->>F: unlock(mtx)
        P->>F: lock(mtx)
        P->>F: 取走 queue + state=READY
        P->>F: unlock(mtx)
        P->>E: post(task)
    else Future 已 ready
        S->>F: unlock(mtx)
        S->>E: post(task)
    end

8. Future task 与 outstanding work 的闭环

ev_future_submit() 在 task 真正进入 executor 队列之前，先调用：

1
ev_exec_on_task_init(exec);

它的含义是：

这份 task 现在还不能执行，但未来会提交，请不要让 executor/event loop 因当前队列为空而提前退出。

随后会在三个出口之一调用 ev_exec_on_task_fini()：

出口 处理
Future ready，task 被 post ev_exec_post() 后 on_task_fini()
Future cancel，task 仍被 post ev_task_queue_post() 内调用 on_task_fini()
Future abort，task 不执行 ev_task_queue_abort() 内调用 on_task_fini()

这构成严格闭环：

1
2
3
4
5
6
7
submit Future waiter
    ↓
on_task_init：登记未来工作
    ↓
等待 Future 完成、取消或中止
    ↓
on_task_fini：撤销未来工作登记

对 ev_loop 而言，这份等待 Future 的 task 会计入 ntasks，即使它尚未进入 event loop 的真实执行队列。

这正是为什么：

1
loop queue 暂时为空

不等于：

1
系统已经没有异步工作

9. ev_future_get() 不会等待

C 接口：

1
void *ev_future_get(const ev_future_t *future);

它内部只做：

1
2
assert(ev_future_is_ready(future));
return future->value;

它没有条件变量，没有 sleep，没有 Poll，也不会帮调用者推进 event loop。

所以正确用法是：

1
2
3
if (ev_future_is_ready(future)) {
    void *value = ev_future_get(future);
}

或者先通过 ev_future_submit() 登记完成 task。

错误理解是：

1
ev_future_get() 会一直阻塞，直到结果完成

正确理解是：

1
ev_future_get() 只是读取一个已经发布的结果

C++ Future<T, E>::get() 做了更友好的检查：

1
2
if (!*this || !is_ready())
    throw future_not_ready("get");

但它同样不会阻塞。

10. release/acquire 内存序解决了什么

多线程实现中，结果发布路径采用：

1
2
3
写入 result/value
    ↓
release-store READY

观察路径采用：

1
2
3
4
5
acquire-load state
    ↓
确认 READY
    ↓
读取 value/result

其目标是建立以下可见性关系：

1
2
3
Promise 完成线程在 READY 之前写入的结果
happens-before
Future 观察线程在 acquire 读取 READY 后访问结果

因此，观察者只要通过 acquire 语义确认 Future 已 ready，就能看到完成者在发布 ready 之前写入的数据。

需要区分两类同步：

同步对象 保护内容
原子 state 完成状态和结果发布可见性。
future->mtx 等待 task 队列，以及 ready 检查与入队之间的竞态。
原子 refcnt Promise/Future 共享控制块的生命周期。

互斥锁不能替代引用计数；引用计数也不能替代等待队列锁。三者职责不同。

11. Promise/Future 引用计数和销毁顺序

11.1 初始引用

ev_promise_create() 初始化：

1
refcnt = 1

这一个引用属于返回给调用者的 Promise。

11.2 获取 Future 会增加引用

1
ev_future_t *future = ev_promise_get_future(promise);

内部会增加同一个 refcnt。

典型状态：

1
2
3
Promise 引用：1
Future 引用：1
总 refcnt：2

11.3 Promise 和 Future 的 acquire/release 都操作同一计数

1
2
3
4
5
6
ev_promise_acquire/release
    └─ 直接修改 promise->refcnt

ev_future_acquire/release
    └─ 先从 future 找到外层 promise
       再修改 promise->refcnt

11.4 最后一个引用释放时的顺序

当 refcnt 从 1 变为 0：

1
2
3
4
1. 调用用户注册的共享数据析构函数 dtor
2. 调用 ev_future_fini()
3. 销毁 future mutex
4. 释放整个 Promise 控制块

引用计数减一使用 release 语义，最后一个释放者再执行 acquire fence，这是典型的“最后引用负责销毁”模式。

12. 一个容易忽略的设计：Future 被遗弃时仍会调度等待 task

ev_future_fini() 的行为不是直接丢弃等待队列，而是：

1
ev_task_queue_post(&future->queue);

也就是说，如果 Promise/Future 的最后一个引用被释放，而 Future 仍未 ready，之前通过 ev_future_submit() 登记的 task 仍会被提交到 executor。

这看起来反直觉，但它是为了让等待者能够观察到：

1
2
我被唤醒了
但 Future 并没有 ready

从而识别共享状态已被遗弃。

因此，一个通用 Future 完成 task 不应无条件调用 ev_future_get()，而应先判断：

1
2
3
4
5
if (ev_future_is_ready(future)) {
    // 正常完成
} else {
    // Future 被取消、遗弃或以非 ready 路径唤醒
}

这也是组合 Future 可以处理“输入 Future 在 ready 前被遗弃”的基础。

13. cancel() 和 abort() 的区别

两者都会从 Future 的等待队列中移除尚未调度的 task，但后续处理不同。

接口 task 是否执行 outstanding work 是否结束
ev_future_cancel() 是，仍提交到 task 的 executor 是
ev_future_abort() 否，直接终止 是

13.1 cancel

1
2
3
4
5
从 Future queue 移除 task
    ↓
post 到 executor
    ↓
task function 仍会运行

所以 cancel 的语义不是“保证回调不执行”，而是：

不再等待 Future，尽快调度 task，由 task 自己处理取消状态。

13.2 abort

1
2
3
4
5
从 Future queue 移除 task
    ↓
不提交 executor
    ↓
只结束 outstanding work

这才是“该 task 不再执行”。

13.3 工程风险

如果 task 内部持有需要在回调中释放的资源，则直接 abort 可能绕过该清理逻辑。资源所有权必须由外围对象或析构函数保证，不能默认依赖 task function 一定会运行。

官方还特别提示：when_all() 和 when_any() 内部提交的 task 不应由调用者 abort，否则可能导致资源泄漏。

14. when_all() 的实现思路

ev_future_when_all_n() 创建一个新的 Promise/Future，用于表示多个输入 Future 的聚合完成状态。

内部共享数据大致为：

1
2
3
4
5
6
7
struct ev_future_when_all {
    size_t idx;
    ev_promise_t *promise;
    struct ev_task task;
    size_t n;
    ev_future_t *futures[];
};

它没有为每个输入 Future 创建一个 task，而是使用同一个 task 按顺序等待：

1
2
3
4
5
6
7
等待 futures[0]
    ↓
完成后复用同一个 task 等待 futures[1]
    ↓
完成后等待 futures[2]
    ↓
...
1
2
3
4
5
6
7
flowchart LR
    F0[Future 0] --> T[同一个 when_all task]
    T --> F1[Future 1]
    F1 --> T
    T --> F2[Future 2]
    F2 --> T
    T --> R[完成聚合 Promise]

这种设计的优势是：

  • 只需要一个 ev_task；
  • 组合逻辑简单；
  • 不需要额外的原子完成计数器；
  • 适合 event-loop 串行推进模型。

其代价是：

  • 它按顺序观察输入 Future；
  • 后面的 Future 即使已经 ready，也要等前面的检查流程推进到它；
  • 精确行为还受到输入 Future 是否被遗弃以及引用唯一性检查影响。

对外契约是：

  • 全部输入 ready 时，结果表示 ready 的数量；
  • 某个输入在 ready 前被遗弃时，结果表示对应索引；
  • 输入数量为 0 时，立即返回 ready Future，结果指针为空。

工程上不要把 when_all() 的结果简单当作业务数据，它是一个聚合状态索引/计数。

15. when_any() 的实现思路

ev_future_when_any_n() 为每一个输入 Future 创建一个独立等待项：

1
2
3
4
5
struct ev_future_when_any {
    size_t idx;
    ev_promise_t *promise;
    struct ev_task task;
};
1
2
3
Future 0 → waiter[0] ┐
Future 1 → waiter[1] ├→ 竞争完成同一个聚合 Promise
Future 2 → waiter[2] ┘

每个 waiter 被唤醒后都会尝试：

1
ev_promise_set(when->promise, &when->idx);

由于 Promise 是一次性的，只有第一个执行 WAITING → SETTING 的 waiter 能成功。

1
2
3
4
5
6
7
8
9
sequenceDiagram
    participant F0 as Future0 waiter
    participant F1 as Future1 waiter
    participant P as 聚合 Promise

    F1->>P: set(index=1)
    P-->>F1: 成功，WAITING→SETTING→READY
    F0->>P: set(index=0)
    P-->>F0: 失败，Promise 已完成

因此，when_any() 的本质是：

多个 task 竞争完成同一个 Promise，原子状态机决定唯一赢家。

对外结果为第一个 ready 或被遗弃的输入 Future 索引。

剩余 waiter 即使后来运行，也无法再次改变聚合结果；它们只需释放自己持有的 Promise 引用。

16. C++ Promise<T, E> 做了什么

C++ 层把 C 的无类型 void * 结果包装为：

1
util::Result<T, E>

16.1 构造

Promise<T, E> 创建足够大的共享数据区：

1
ev_promise_create(sizeof(result_type), dtor)

然后在尾随数据区 placement-new：

1
new (ev_promise_data(promise_)) result_type();

16.2 析构

C 层最后释放共享状态时调用注册的 dtor，显式析构 Result<T, E>。

16.3 set

C++ set() 仍遵循底层两阶段完成协议：

1
2
3
4
5
set_acquire
    ↓
给 Result<T, E> 赋值
    ↓
set_release(result 地址)

如果结果赋值抛出异常，源码仍调用 set_release(..., nullptr) 结束 Promise，然后重新抛出异常。

这意味着完成权一旦被当前调用者取得，就不能把状态退回 WAITING。

16.4 RAII

复制 Promise/Future 会 acquire 引用；析构会 release；移动操作转移裸指针并清空源对象。

所以 C++ 层解决的是：

  • 引用计数自动管理；
  • 结果类型安全；
  • 错误值统一封装；
  • 异常安全；
  • continuation 组合。

它没有改变底层状态机。

17. C++ Future<T, E> 做了什么

C++ Future 主要是薄封装：

1
2
3
class Future {
    ev_future_t *future_{nullptr};
};

对应关系如下：

C++ 接口 C 接口
is_ready() ev_future_is_ready()
get() ev_future_get()
submit() ev_future_submit()
cancel() ev_future_cancel()
abort() ev_future_abort()
拷贝构造 ev_future_acquire()
析构 ev_future_release()

get() 返回的是共享状态中 Result<T, E> 的引用，不是复制一份结果。

因此结果对象的有效期受共享状态生命周期约束：只要调用方仍在使用该引用，就必须保证至少有一个 Promise/Future 引用继续持有共享状态。

18. AsyncTask：把函数执行结果转成 Future

make_async_task() 会把以下内容放在 Promise 的尾随共享数据区：

  • 一个继承自 ev_task 的 AsyncTask；
  • Promise 包装对象；
  • callable 和参数；
  • 最终结果；
  • 必要时保存 inner Future。
1
2
3
4
5
6
7
8
9
10
11
Promise 控制块
┌───────────────────────────┐
│ refcnt / dtor / Future    │
├───────────────────────────┤
│ AsyncTask                 │
│  ├─ ev_task               │
│  ├─ Promise               │
│  ├─ Invoker               │
│  ├─ inner Future（可选）   │
│  └─ Result                │
└───────────────────────────┘

18.1 callable 返回普通值

执行 task 时：

1
2
3
4
5
6
调用 callable
    ↓
成功：Result 保存返回值
失败：Result 保存 exception_ptr
    ↓
完成 Promise

catch_result() 同时处理普通返回值和 void 返回值。

18.2 callable 返回 Future

这时使用另一套 AsyncTask 特化，实现 implicit unwrapping。

第一次运行 task：

1
2
3
4
5
执行 callable
    ↓
得到 inner Future
    ↓
把同一个 task 重新 submit 到 inner Future

第二次由 inner Future 唤醒 task：

1
2
3
4
5
读取 inner Future 的 Result
    ↓
提取 value()
    ↓
完成外层 Promise

所以：

1
Future<A>.then(... -> Future<B>)

最终得到的是：

1
Future<B>

而不是：

1
Future<Future<B>>

这就是源码注释所称的 implicit unwrapping。

19. then() 如何构造 continuation

Future::then() 的流程是：

1
2
3
4
auto task = make_async_task(exec, continuation, *this);
auto future = task->get_future();
submit(*task);
return future;

完整链路：

1
2
3
4
5
6
7
8
9
原 Future A
    ↓ submit continuation task
A ready
    ↓
executor 执行 continuation(A)
    ↓
结果写入 Promise B
    ↓
Future B ready

源码特意先取得返回 Future，再把 task 提交给原 Future：

1
2
先 get_future()
再 submit(task)

原因是 executor 允许 task 在 submit() 返回前开始执行。如果先 submit，task 可能立即运行并释放其内部状态，随后调用者再取 Future 就可能发生竞态。

这是一条非常通用的异步编程经验：

在提交可能立即执行的异步任务前，先取得调用方后续需要的所有引用和句柄。

20. async() 如何工作

async(exec, f, args...) 是 make_async_task() 加 executor submit 的便利封装：

1
2
3
4
5
6
7
8
9
10
11
创建 AsyncTask 和 Promise
    ↓
先取得返回 Future
    ↓
把 AsyncTask 提交到 executor
    ↓
executor 执行 callable
    ↓
完成 Promise
    ↓
返回 Future ready

它不是创建一个新线程。callable 在传入的 executor 上执行。

如果 executor 属于 ev_loop，那么 callable 通常就在运行该 loop 的线程中执行。

21. Future 与 ev_loop_wait() 的关系

虽然 Lely Future 本身不阻塞，但 ev_loop_wait(loop, future) 可以把 Future 作为 event loop 本次运行的结束条件。

其内部不是阻塞在 Future 条件变量上，而是：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
向 Future 提交一个内部唤醒 task
    ↓
继续运行 event loop
    ├─ 执行普通 task
    ├─ Poll I/O
    └─ 等待内部唤醒
    ↓
异步操作完成 Promise
    ↓
Future 内部唤醒 task 被提交到 loop
    ↓
loop 执行该 task，标记等待 context ready
    ↓
ev_loop_wait() 返回

这解决了 CANopen 异步操作中的关键问题：

1
Future 的完成本身依赖 event loop 继续处理 SocketCAN、定时器和协议状态机。

如果只让线程同步睡眠，而没有其他线程运行 event loop，异步 SDO Future 可能永远不会 ready。

22. 放到异步 CANopen 场景中理解

以异步 SDO 上传为例：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
sequenceDiagram
    participant App as 应用层
    participant SDO as SDO异步操作
    participant EvLoop as ev_loop
    participant IO as SocketCAN/Poll
    participant P as Promise
    participant F as Future

    App->>SDO: 发起异步 SDO 上传
    SDO->>P: 创建并持有 Promise
    SDO-->>App: 返回 Future
    App->>EvLoop: wait(Future) 或注册 then()
    EvLoop->>IO: 等待 CAN 响应或超时
    IO-->>EvLoop: CAN 帧可读
    EvLoop->>SDO: 执行接收与状态机 task
    SDO->>P: set(result)
    P->>F: state = READY
    F->>EvLoop: 提交等待 task
    EvLoop-->>App: wait 返回或执行 continuation

角色不要混淆：

角色 在 SDO 场景中的对象
异步操作生产端 SDO 客户端状态机及其 Promise。
结果观察端 应用层持有的 Future。
I/O 驱动 SocketCAN + Poll。
调度器 ev_loop executor。
结果内容 SDO 结果、错误码或相关数据。

Future 不传输 CAN 帧，也不实现 SDO 协议。它只负责把“SDO 操作已经完成”这一事实及结果传给观察者。

23. 常见错误理解

23.1 把 Future 当消息队列

错误：

1
Promise 可以不断写结果，Future 可以不断取消息

正确：

1
2
一个 Promise 只能成功完成一次
一个 Future 只观察这一次完成

23.2 认为 get() 会阻塞

错误：

1
直接 get，没完成就等

正确：

1
2
C 版要求调用前已经 ready
C++ 版未 ready 会抛 future_not_ready

23.3 认为 Promise 和 Future 有两个引用计数

错误：

1
Promise 管 Promise，Future 管 Future

正确：

1
二者共享外层 Promise 控制块中的一个 refcnt

23.4 认为 cancel 后 task 不会执行

错误：

1
cancel == 丢弃 task

正确：

1
2
cancel 会把 task 提交到 executor
abort 才不会执行 task

23.5 认为 Future ready 后回调在完成线程直接执行

不一定。

Promise 完成线程只负责把 task 提交到 task 指定的 executor。task 实际在哪个线程运行，由 executor 的实现和当前运行环境决定。

23.6 认为 async() 一定创建线程

错误。

Lely async() 只是把 callable 包成 task 并提交给指定 executor。executor 可以是 event loop，也可以是其他实现。

24. 调试 Future 问题时建议观察的变量

24.1 future->state

1
2
3
WAITING：尚无人完成
SETTING：某个线程正在构造/发布结果
READY：结果已发布

如果长期停在 SETTING，通常说明某条路径成功调用了 set_acquire()，却没有调用 set_release()。

24.2 promise->refcnt

用于定位：

  • 提前释放；
  • Future 永不销毁；
  • continuation 引用环；
  • 聚合 Future 内部引用未释放。

不要仅凭某一瞬间的 refcnt 在多线程环境中做严格逻辑判断，但可以用于断点和生命周期分析。

24.3 future->queue

如果 Future 已 ready，但 queue 中仍残留 task，需要检查：

  • 实际是否执行到 set_release()；
  • 是否使用了不同的 Future 对象；
  • task 是否被重复挂入链表；
  • 是否有内存破坏。

24.4 future->value

READY 时应指向有效结果，除非调用路径明确使用 NULL 表示空结果或结果构造失败。

24.5 task 的 exec 和 func

Future 只负责提交。如果 task 不执行，还要继续检查：

  • executor 是否仍在运行；
  • event loop 是否 stopped；
  • task 是否被 executor abort；
  • func 是否为空或对象生命周期已结束。

24.6 event loop 的 ntasks

Future waiter 已调用 on_task_init() 后，即使尚未进入 loop queue，也应作为 outstanding work 存在。

如果 ntasks 提前归零，检查 Future task 的 init/fini 是否配对。

25. 建议的源码单步顺序

第一次调试时，不建议直接从完整 CANopen 主站流程开始。可以按以下顺序建立断点。

25.1 最小 Promise/Future

1
2
3
4
5
6
7
8
9
10
11
12
13
ev_promise_create
    ↓
ev_promise_get_future
    ↓
ev_future_submit
    ↓
ev_promise_set_acquire
    ↓
ev_promise_set_release
    ↓
ev_task_queue_post
    ↓
task.func

25.2 引用计数

1
2
3
4
ev_promise_acquire/release
ev_future_acquire/release
ev_promise_fini
ev_future_fini

25.3 组合 Future

1
2
3
4
5
ev_future_when_all_n
ev_future_when_all_func

ev_future_when_any_n
ev_future_when_any_func

25.4 C++ continuation

1
2
3
4
5
Promise<T,E>::set
Future<T,E>::then
make_async_task
AsyncTask::task function
Future<T,E>::get

25.5 接入 CANopen

最后再从具体异步 SDO/PDO/定时器 API 返回的 Future 开始，追踪是谁持有 Promise，以及在哪条完成或错误路径调用 set。

26. 最小 C 接口示例

下面的示例只展示对象关系，executor 创建过程需按实际平台补充。

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
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
#include <lely/ev/future.h>
#include <lely/ev/task.h>
#include <lely/util/util.h>
#include <stdio.h>
#include <stdlib.h>

struct result_data {
    int value;
};

struct wait_task {
    struct ev_task task;
    ev_future_t *future;
};

static void
on_ready(struct ev_task *task)
{
    struct wait_task *wait = structof(task, struct wait_task, task);

    if (ev_future_is_ready(wait->future)) {
        const struct result_data *result = ev_future_get(wait->future);
        printf("result = %d\n", result->value);
    } else {
        puts("future was canceled or abandoned");
    }

    ev_future_release(wait->future);
    free(wait);
}

int
start_example(ev_exec_t *exec)
{
    ev_promise_t *promise = ev_promise_create(sizeof(struct result_data), NULL);
    if (!promise)
        return -1;

    struct wait_task *wait = malloc(sizeof(*wait));
    if (!wait) {
        ev_promise_release(promise);
        return -1;
    }

    wait->task = (struct ev_task)EV_TASK_INIT(exec, &on_ready);
    wait->future = ev_promise_get_future(promise);
    ev_future_submit(wait->future, &wait->task);

    if (ev_promise_set_acquire(promise)) {
        struct result_data *result = ev_promise_data(promise);
        result->value = 42;
        ev_promise_set_release(promise, result);
    }

    ev_promise_release(promise);
    return 0;
}

示例把等待 task 放在堆上，并由 task function 在执行后释放，以保证对象生命周期覆盖 executor 的异步调度。生产代码还应结合具体异步操作处理分配失败、取消和 executor 关闭流程。

27. 最小 C++ 接口示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
#include <lely/ev/future.hpp>

using lely::ev::Future;
using lely::ev::Promise;

Future<int> start_operation()
{
    Promise<int> promise;
    auto future = promise.get_future();

    // 实际程序应把 promise 移交给异步操作对象。
    promise.set(lely::util::success(42));

    return future;
}

使用 continuation：

1
2
3
4
5
auto future = start_operation();

auto next = future.then(exec, [](Future<int> completed) {
    return completed.get().value() * 2;
});

这里 next 的类型是保存 continuation 返回值的 Future；continuation 中的异常会被捕获为 std::exception_ptr 错误结果。

28. 一张总图串起全部机制

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
flowchart TD
    A[异步操作创建 Promise] --> B[Promise 内嵌 Future]
    B --> C[调用方取得 Future 引用]
    C --> D[ev_future_submit 登记 task]
    D --> E[on_task_init 增加 outstanding work]
    E --> F{Promise 是否完成}
    F -- 否 --> G[task 保存在 Future queue]
    F -- 是 --> H[task 直接 post 到 executor]
    G --> I[完成方 set_acquire 抢到完成权]
    I --> J[写共享结果]
    J --> K[set_release 发布 READY]
    K --> L[转移 Future queue]
    L --> H
    H --> M[on_task_fini 撤销 outstanding work]
    M --> N[executor/event loop 执行 task]
    N --> O[检查 ready 并读取 Result]

29. 最终理解框架

学完这三个文件后，应该能回答以下问题：

  1. 为什么 Lely Future 是非阻塞的？
  2. Promise 和 Future 为什么共享同一个引用计数？
  3. WAITING → SETTING → READY 为什么需要三个状态，而不是布尔值？
  4. ev_future_submit() 如何避免 task 丢失？
  5. 为什么等待 Future 的 task 要先调用 on_task_init()？
  6. Future 被遗弃时，为什么仍要提交等待 task？
  7. cancel 和 abort 为什么语义不同？
  8. when_any() 如何用一次性 Promise 决定唯一赢家？
  9. C++ then() 如何把 continuation 变成新的 Future？
  10. continuation 返回 Future 时，为什么不会得到嵌套 Future？
  11. Future、executor、event loop 和 SocketCAN 各自负责哪一部分？

如果这些问题都能从对象关系、状态机和 task 调度路径解释清楚，就已经掌握了 Lely Future/Promise 的核心设计。

30. 官方源码与参考资料

30.1 本文核心源码

  1. Lely src/ev/future.c
  2. Lely include/lely/ev/future.h
  3. Lely include/lely/ev/future.hpp

30.2 关联源码

  1. Lely include/lely/ev/task.h
  2. Lely src/ev/task.c
  3. Lely include/lely/ev/exec.h
  4. Lely src/ev/loop.c

30.3 官方说明

  1. Lely CANopen Library overview
  2. Lely CANopen Build configuration