`future<T>` 与异步函数

ASCO 将“异步函数调用”和“异步任务”明确区分为两类对象：future<T> 与 join_handle<T>。

future<T>：异步函数的返回类型，表示一次异步调用的结果；co_await 该对象会执行对应协程并得到 T。
join_handle<T>：由运行时调度的任务句柄；可 co_await 以等待任务完成，或 detach() 以放弃 join。

1. `future<T>`：异步函数调用的惰性返回对象

返回 future<T> 的可调用对象满足 asco::async_function 概念，可视为 ASCO 语义下的异步函数。

#include <asco/future.h>
using namespace asco;

future<int> add_one(int x) {
    co_return x + 1;
}

1.1 惰性执行：调用不执行，`co_await` 才执行

future<T> 是惰性对象。因此：

调用 add_one(41) 仅构造 future<int>（协程句柄处于初始挂起态）。
对该对象执行 co_await 时，协程才开始运行。

future<int> example() {
    auto f = add_one(41);  // 仅构造 future
    int v = co_await f;    // 协程在此处开始执行
    co_return v;
}

1.2 等待语义：顺序依赖执行（非独立任务）

co_await future<T> 表达的是“等待一次异步调用的结果”，而不是“并发启动一个任务”。被等待的协程会作为当前任务调用链的一部分继续推进，而不是作为新的调度任务独立运行。

推论：

future 不构成并发；连续 co_await 表示顺序依赖执行。
对 future 的一次 co_await 若进入挂起路径，控制流会回到 worker，并触发一次调度公平性检查；因此频繁发生 co_await 的异步代码通常会给同一 worker 上的其他活动任务让出调度机会。
真正容易长期占用 worker 的情况，是长时间没有发生任何挂起点的连续计算代码。
异常会在 co_await 处重新抛出。

1.3 运行时约束：必须在 runtime 内等待

future<T> 必须在 runtime 上下文中 co_await：

通过 core::runtime::block_on(...) 进入运行时；或
在 spawn(...) 启动的任务内部等待。

在运行时之外直接 co_await future 会触发断言失败或 panic。

1.4 `[[nodiscard]]`：避免未执行导致的资源泄露

future<T> 标记为 [[nodiscard]]。若创建 future 但从不 co_await，该异步调用不会执行完成，通常会导致资源泄露或逻辑缺失。

2. `join_handle<T>`：运行时任务句柄

当需要并发执行异步逻辑时，使用 spawn(...) 将异步函数提交给 runtime，并获得 join_handle<T>。

2.1 `spawn`：将异步函数提交为任务

core::runtime::spawn(async_function) 接收一个返回 future<T> 的可调用对象，启动一个并发任务并返回 join_handle<T>。

#include <asco/core/runtime.h>
#include <asco/future.h>
#include <asco/yield.h>

using namespace asco;

future<int> work() {
    co_await this_task::yield();
    co_return 42;
}

future<void> parent() {
    auto h = spawn([]() -> future<int> { co_return co_await work(); });
    int v = co_await h;  // join
    (void)v;
    co_return;
}

2.2 `co_await join_handle`：挂起当前协程，等待任务完成

join_handle<T> 的等待模型与 future 不同：co_await join_handle<T> 会挂起当前协程，直到该任务完成。

正常完成：返回 T（或 void）。
任务抛异常：在 co_await 处重新抛出。

2.3 `detach()`：放弃 join

detach() 表示调用方不再等待结果，任务将独立执行直至完成/失败。

2.4 `spawn_blocking`：启动“blocking 环境”的任务

当你需要在 runtime 内部执行少量同步逻辑（例如调用一个必须同步等待的第三方 API）时，可以使用 spawn_blocking(fn) 提交一个同步函数作为任务执行。

语义要点：

spawn_blocking(fn) 不会把工作自动转移到“后台线程池”；它仍然运行在 ASCO runtime 的 worker 线程上。
该任务会被标记为“blocking 环境”（asco::this_task::is_blocking_env() == true），因此在该任务内部允许调用一些同步阻塞接口（例如 runtime::block_on(...)、blocking_lock()、blocking_acquire()）。
由于它会阻塞 worker 线程，应避免长时间阻塞操作；若 runtime 线程数过少（尤其是单线程 runtime），在 blocking 任务中再调用 block_on() 这类同步等待接口可能导致死锁。

3. `join_set<T>`：批量管理并发任务

join_set<T> 用于管理一组任务：

spawn(...)：提交任务到 set。
co_await set：按完成顺序收集结果。
join_all()：停止收集并返回已到达的结果。

更完整的语义与注意事项见：join_set<T>：批量任务收集。

示例（非 void 输出）：

#include <asco/core/runtime.h>
#include <asco/task/join_set.h>
#include <asco/yield.h>

using namespace asco;

future<int> job(int x) {
    co_await this_task::yield();
    co_return x * 2;
}

int main() {
    core::runtime rt;
    return rt.block_on([&]() -> future<int> {
        task::join_set<int> set{rt};
        for (int i = 0; i < 10; ++i) {
            set.spawn([i]() -> future<int> { co_return co_await job(i); });
        }

        int sum = 0;
        while (auto v = co_await set) {
            sum += *v;
        }

        co_await set.join_all();
        co_return sum;
    });
}

4. `co_invoke`：延长临时可调用对象的生命周期

co_invoke 的目的不是简化书写，而是保证“临时可调用对象 + 协程执行”的组合是内存安全的。

4.1 风险来源：临时可调用对象生命周期不足

如果把一个临时可调用对象（rvalue，例如临时协程 lambda）用于创建 future/join_handle，而该任务又会在之后继续执行，那么该临时对象在表达式结束后就会被销毁。

当任务后续执行仍需要访问该可调用对象的捕获状态时，就会触发未定义行为（常见表现为偶发崩溃或数据损坏）。

4.2 典型反例：临时协程 lambda

future<int> bad() {
    auto f = ([p = std::make_unique<int>(123)]() -> future<int> {
        co_await this_task::yield();
        co_return *p;
    })();

    // 临时 lambda 已销毁；协程恢复后访问 p 可能 UB。
    co_return co_await f;
}

该问题通常呈现为偶发崩溃、错误读写或内存破坏，且与根因位置不一致。

4.3 `co_invoke` 的机制（对应 `asco/invoke.h`）

co_invoke 的语义是：当你用“临时可调用对象”（尤其是临时协程 lambda）来创建 future/join_handle 时，框架会确保该临时对象的生命周期足够长，从而避免捕获状态悬垂。

4.4 不使用 `co_invoke` 的后果

若在框架/工具函数中对 Fn&& 进行天真转发并直接调用：

template<class Fn>
auto naive_invoke(Fn&& fn) {
    return std::invoke(std::forward<Fn>(fn));
}

则对临时协程可调用对象的调用可能在后续执行中访问已销毁的捕获状态，从而触发未定义行为。该风险通常不会被编译器诊断。

4.5 适用范围

框架/库作者：只要需要“接受 Fn&& 并调用”，且允许传入临时可调用对象，应使用 co_invoke 或等价的“将可调用对象绑定到任务以延长生命周期”的方案。
一般使用者：通常无需直接调用 co_invoke；spawn(...)、join_set::spawn(...) 等常用入口已经覆盖了该类生命周期风险。

5. `this_task::yield()`：协作式调度点

co_await this_task::yield() 提供一个协作式调度点：让出当前执行权，使其它可运行任务有机会执行。它常用于：

避免忙等；
提升公平性。

6. 典型入口：`runtime::block_on(async_main)`

常见工程结构为：

用户实现 future<int> async_main()。
main() 创建 core::runtime 并调用 block_on(async_main)。

补充说明：

block_on(...) 是同步阻塞接口，要求当前处于“blocking 环境”（asco::this_task::is_blocking_env() == true）。
因此它最常见的用法是在 runtime 之外（例如 main()）作为入口；在 runtime 内也只有在 spawn_blocking(...) 任务里才允许使用。

链接目标 asco::main 已提供默认 main()；使用者仅需定义 async_main()。

7. 选型摘要

在协程中等待一次异步调用：co_await future<T>
提交并发任务并等待其完成：spawn(...) -> join_handle<T>，随后 co_await
管理多个并发任务：join_set<T>

任务（asco::task）

本章介绍围绕任务批量管理与结果收集的公开接口。

这些接口关注的是任务如何被提交、结果如何被观察，以及调用方需要承担的生命周期与错误处理约束。

`join_all`：等待多个异步操作并按参数位置汇总结果

asco::task::join_all 用于同时等待多个异步操作，并在全部结束后一次性返回结果。

它适合：

你已经知道要等待的异步操作集合；
需要并发运行这些异步操作；
关心“每个参数位置对应哪个结果”，而不是“谁先完成”。

对应头文件：asco/task/join_all.h。

若需要使用 task::fetch_result(...) 辅助取值，还需要包含 asco/task/fetch_result.h。

1. 快速上手

#include <asco/task/fetch_result.h>
#include <asco/task/join_all.h>

using namespace asco;

future<int> calc_a() {
    co_return 1;
}

future<int> calc_b() {
    co_return 2;
}

future<int> run() {
    auto [a, b] = co_await task::join_all{calc_a, calc_b};

    co_return task::fetch_result(std::move(a)) + task::fetch_result(std::move(b));
}

要点：

join_all 会并发启动传入的异步函数（不产生新任务）。
返回值是一个 std::tuple，槽位顺序与传入参数顺序一致。
每个槽位都是 std::expected<结果类型, std::exception_ptr>；成功时取值，失败时保存异常。

2. API 语义

2.1 构造与等待

task::join_all(async_fn_1, async_fn_2, ...)

语义：

每个参数都必须是“异步函数”，即调用后返回 future<T>。
co_await task::join_all{...} 会等待所有传入任务结束。
返回值是 std::tuple<std::expected<...>, ...>。

结果顺序：

第 i 个 tuple 槽位，总是对应第 i 个传入任务。
即使任务完成先后顺序不同，返回槽位顺序也不会变化。

异常语义：

某个任务抛出异常时，异常会被保存进该槽位的 std::expected。
其他任务仍会继续运行并各自产生自己的结果槽位。
因此，join_all 不把“某个子任务失败”作为整体失败直接抛出。

2.2 `void` 任务的结果表示

当某个任务返回 future<void> 时：

对应槽位类型是 std::expected<std::monostate, std::exception_ptr>。
has_value() == true 表示该任务成功结束。
若任务抛出异常，异常仍保存在 error() 中。

2.3 `fetch_result(expected)`：取值或重抛异常

template<typename T>
T fetch_result(std::expected<T, std::exception_ptr> &&e);

语义：

若 e.has_value()，返回其中的值。
若 e 保存异常，则重抛该异常。
对 std::expected<void, std::exception_ptr>，成功时不返回值，失败时同样重抛异常。

fetch_result(...) 适合在你希望恢复“直接拿值/直接抛错”的调用风格时使用。

3. 使用约束与建议

3.1 适合固定数量的并发等待

join_all 更适合“当前这一批任务已知且固定”的场景。

如果你需要：

动态追加任务；
按完成顺序持续消费结果；
在任务尚未全部完成前逐个取回结果；

更适合使用 join_set<T>。

3.2 不提供单个任务句柄

join_all 只返回整体聚合结果，不暴露单个任务的 join_handle。

因此：

不能通过 join_all 单独取消某一个子任务；
也不能在等待过程中单独 co_await 某一个子任务。

3.3 谨慎处理跨挂起点引用生命周期

传入 join_all 的异步任务如果捕获了引用，并且这些引用会跨 co_await 被使用：

调用方必须保证被引用对象在所有相关任务结束前保持有效。

这不是 join_all 特有的限制，但在批量并发等待场景中更容易被忽略。

`join_set<T>`：按完成顺序收集并发任务

join_set<T> 用于批量启动任务，并按“任务完成顺序”持续产出结果。

它适合：

你要同时跑很多个任务；
想边完成边消费结果（streaming），而不是固定按提交顺序等待。

对应头文件：asco/task/join_set.h。

1. 快速上手

#include <asco/core/runtime.h>
#include <asco/task/join_set.h>
#include <asco/yield.h>

using namespace asco;

future<int> job(int x) {
    co_await this_task::yield();
    co_return x * 2;
}

int main() {
    core::runtime rt;
    return rt.block_on([&]() -> future<int> {
        task::join_set<int> set{rt};

        for (int i = 0; i < 10; ++i) {
            set.spawn([i]() -> future<int> { co_return co_await job(i); });
        }

        int sum = 0;
        while (auto v = co_await set) {
            // v 的到达顺序 = 完成顺序（不保证与 i 相同）
            sum += *v;
        }

        co_return sum;
    });
}

要点：

set.spawn(...) 提交任务；join_set 不提供单个任务的 join_handle。
co_await set 每次取回一个结果；当 set 中已提交任务全部产出结果后，返回 std::nullopt。

2. API 语义

2.1 构造

join_set()：使用 core::runtime::current()
join_set(core::runtime&)：显式绑定 runtime

2.2 `spawn(async_fn)`：提交异步任务

void spawn(async_function<> auto &&fn);

语义：

fn() 必须返回 future<T>（即“异步函数”）。
spawn(...) 会启动一个并发任务。
当该任务完成并产生结果后，结果会出现在 co_await set 的返回序列中（按完成顺序）。

注意：

join_set 不暴露单个任务的 join_handle，因此无法对单个任务进行 cancel() 或单独 co_await。
传入的任务应当在完成路径上总能产出一个结果；如果任务提前退出且没有产出结果，join_set 将永远等不到该条结果。

2.3 `co_await set`：取回一个结果

future<std::conditional_t<std::is_void_v<T>, bool, std::optional<T>>> operator co_await();

当 T 为非 void 类型时：

每次 co_await set 返回一个 std::optional<T>
- has_value() == true：拿到一个结果
- std::nullopt：当前 join_set 中已提交任务的结果都已收齐

当 T 为 void 时：

每次 co_await set 返回一个 bool
- true：拿到一个结果
- false：当前 join_set 中已提交任务的结果都已收齐

2.4 `join_all()`：关闭结果收集并收集剩余结果

join_all() 会关闭 join_set 内部的结果收集通道，并返回关闭后仍能收集到的剩余结果。

注意：

调用 join_all() 之后，不应再继续向该 join_set 提交新任务。
若之后仍调用 spawn(...)，任务本身仍可能被启动，但其结果不会再进入这个 join_set。

3. 常见问题与建议

3.1 任务异常如何处理？

join_set 不会把“任务失败”作为结果返回给你。为了保证 join_set 能持续产出结果：

不要让异常从任务中逃逸；
推荐把错误编码进返回值（例如 std::expected<T, E>），并始终返回一个值。

工程建议：

让任务返回 std::expected<T, E> 或类似结果类型，把错误显式作为值传回；
或在任务内部 try/catch，把错误信息转成 T 的某种错误表示。

3.2 `spawn_blocking` 并不把工作移到“后台线程”

spawn_blocking(...) 不会把工作自动转移到其它线程。

因此：

避免在其中执行长时间阻塞操作（例如长 IO、长时间 sleep）。

3.3 什么时候该用 `join_handle` 而不是 `join_set`？

需要对某个任务单独 cancel() / detach() / co_await：用 spawn(...) -> join_handle<T>。
需要批量启动、按完成顺序消费结果：用 join_set<T>。

`select`：等待首个完成的异步操作

asco::task::select 用于同时等待多个异步操作，并在其中任意一个操作完成后返回该操作的结果。

它适合：

需要从多个异步来源中选择最先完成的一项；
只关心首个完成结果；
首个结果产生后，希望未完成的其他分支收到取消请求。

对应头文件：asco/task/select.h。

若需要使用 task::fetch_result(...) 辅助取值，还需要包含 asco/task/fetch_result.h。

1. 快速上手

#include <variant>

#include <asco/task/fetch_result.h>
#include <asco/task/select.h>

using namespace asco;

future<int> from_cache();
future<int> from_remote();

future<int> run() {
    auto selected = co_await task::select{from_cache, from_remote};

    co_return std::visit(
        [](auto &branch) {
            return task::fetch_result(std::move(branch.value));
        },
        selected);
}

要点：

select 会并发启动传入的异步函数（不产生新任务）。
返回值是一个 std::variant，其中只包含首个完成分支的结果。
分支对象提供 index 与 value：index 表示传入参数的位置，value 是该分支的 std::expected<结果类型, std::exception_ptr>。

2. API 语义

2.1 构造与等待

task::select(async_fn_1, async_fn_2, ...)

语义：

每个参数都必须是“异步函数”，即调用后返回 future<T>。
co_await task::select{...} 会等待直到某一个分支率先完成。
返回结果只包含首个完成的分支。
首个分支完成后，其他尚未完成的分支会收到取消请求。

分支顺序：

返回分支的 index 对应原始参数位置，从 0 开始计数。
不要依赖多个分支几乎同时完成时的选择顺序；如果调用方需要稳定汇总全部结果，应使用 join_all。

2.2 返回值与异常

每个分支的 value 类型是 std::expected<结果类型, std::exception_ptr>。

语义：

若首个完成的分支正常返回，value.has_value() == true。
若首个完成的分支抛出异常，异常会保存在 value.error() 中。
select 不会继续等待其他分支来替代失败结果；抛异常的首个完成分支仍然是被选中的分支。

可以使用 task::fetch_result(...) 将 std::expected 恢复为“成功返回值，失败重抛异常”的调用风格。

2.3 `void` 任务的结果表示

当被选中的分支返回 future<void> 时：

对应的 value 类型是 std::expected<std::monostate, std::exception_ptr>。
has_value() == true 表示该分支成功完成。
若该分支抛出异常，异常仍保存在 error() 中。

3. 使用约束与建议

3.1 只返回首个完成结果

select 适合“首个结果足够”的场景。

如果你需要：

等待所有分支完成；
保留每个参数位置的结果；
不希望首个完成后取消其他分支；

更适合使用 join_all。

3.2 未完成分支需要能正确响应取消

首个分支完成后，select 会请求取消未完成分支。

因此，传入的异步操作应确保在收到取消请求后能够正确收束，例如：

在需要清理资源时注册取消回调；
在长循环中适时检查当前取消状态；
避免在不可取消的等待中永久阻塞。

3.3 谨慎处理跨挂起点引用生命周期

传入 select 的异步任务如果捕获了引用，并且这些引用会跨 co_await 被使用：

调用方必须保证被引用对象在相关分支完成或取消收束前保持有效。

这不是 select 特有的限制，但在“首个完成后取消其他分支”的场景中更容易被忽略。

同步原语

本章介绍 ASCO 提供的同步原语。它们用于在多个并发执行流之间建立互斥与协调关系。

本章只描述行为与语义：你能做什么、它会怎样表现、以及常见用法。

选型建议

何时使用 `channel`

channel 适合在并发执行流之间传递值，并通过等待来表达背压（缓冲满时发送等待，缓冲空时接收等待）。

何时使用 `spinlock`

spinlock 适合保护非常短的临界区：

修改一个小的共享状态（例如计数器、指针、短容器操作）；
临界区内不进行可能长时间运行的操作。

注意：

不要在持有锁期间跨越 co_await。

何时使用 `semaphore`

semaphore 适合表达“许可”的语义：

并发限流（最多允许同时进行 N 个工作）；
生产者/消费者式的资源计数（有资源才能继续）。

acquire() 是可等待操作：当许可不足时会等待直到有许可被释放。

何时使用 `rwlock`

rwlock 适合保护“读多写少”的共享状态：

多个 reader 同时读取是安全且有价值的；
写入虽然较少，但写入时需要与所有 reader 互斥；
你希望一旦 writer 开始等待，后续 reader 不再继续插队。

如果读写比例并不偏向读取，或者临界区极短，通常 mutex 会更直接。

何时使用 `condition_variable`

condition_variable 适合表达“等某个条件成立，然后由别的任务通知我继续”：

条件本身保存在外部共享状态里；
等待方通过 predicate 判断是否可以继续；
状态改变后，由生产者调用 notify_one() / notify(n) 唤醒等待方；必要时也可以使用带 predicate 的通知重载做额外门控。

它本身不拥有被保护的数据，也不提供互斥；若 predicate 访问共享状态，调用方需要自行保证可见性与并发安全。

常见约定

用 try_* 系列 API 表达“不等待的快速路径”；失败时走其它分支。
需要等待时，优先用明确的同步原语（例如 semaphore::acquire()），避免忙等。

`sync::channel<T>`：通道

sync::channel<T> 用于在并发执行流之间传递值。

它是有界的：当缓冲区满时，发送会等待。
当缓冲区空时，接收会等待。
通道被关闭后，发送会失败；接收在“通道已关闭且缓冲已空”时结束。

头文件：asco/sync/channel.h

1. 创建通道

#include <asco/sync/channel.h>

auto [tx, rx] = asco::sync::channel<int>();

语义：

channel<T>() 返回一对端点：sender<T>（发送端）与 receiver<T>（接收端）。
sender<T>/receiver<T> 可拷贝；拷贝后的对象与原对象共享同一个通道。

2. 发送：`sender<T>::send(...)`

2.1 `T` 非 `void`

#include <asco/sync/channel.h>
#include <asco/future.h>
#include <expected>

using namespace asco;

future<void> producer(sync::sender<int> tx) {
    auto r = co_await tx.send(42);
    if (!r) {
        // 发送失败：通道已关闭，42 没有被发送
        int unsent = std::move(r.error());
        (void)unsent;
    }
    co_return;
}

语义：

send(value) 返回 future<std::expected<std::monostate, T>>。
当通道可写时：把 value 发送到通道中，并返回“成功”。
当缓冲区满时：等待直到通道可写或通道关闭。
当通道已关闭且本次发送未发生时：返回“失败”，并在 error() 中返回未发送的 value。

2.2 `T == void`

当 T 为 void 时，发送表示发送一个“事件”。

#include <asco/sync/channel.h>
#include <asco/future.h>

using namespace asco;

future<void> producer(sync::sender<void> tx) {
    bool ok = co_await tx.send();
    if (!ok) {
        // 通道已关闭，本次事件没有被发送
    }
    co_return;
}

语义：

send() 返回 future<bool>。
返回 true 表示发送成功；返回 false 表示通道已关闭且本次发送未发生。

3. 接收：`receiver<T>::recv()`

#include <asco/sync/channel.h>
#include <asco/future.h>
#include <optional>

using namespace asco;

future<void> consumer(sync::receiver<int> rx) {
    while (auto v = co_await rx.recv()) {
        int x = *v;
        (void)x;
    }
    // 循环结束：通道已关闭且缓冲已空
    co_return;
}

语义（T 非 void）：

recv() 返回 future<std::optional<T>>。
当通道中存在值时：返回 T。
当通道暂时为空且未关闭时：等待直到有值可读或通道关闭。
当通道已关闭且缓冲已空时：返回 std::nullopt。

4. 关闭通道：`stop()`

发送端与接收端都提供 stop()：

tx.stop();
rx.stop();

语义：

调用 stop() 会关闭通道。
关闭后：新的发送不会发生；接收在读完缓冲中的剩余值后结束（recv() 返回 std::nullopt）。

5. 使用建议

当需要把数据从生产者传给消费者，并让双方通过等待来表达背压时，使用 channel<T>。
若你只需要“完成通知/事件”，可以用 channel<void> 表达事件流。
当某一侧确定不再使用通道时，调用 stop() 让另一侧尽快结束等待并退出。

`sync::condition_variable`：条件变量

sync::condition_variable 用于在多个异步任务之间表达“条件未满足就等待，条件变化后由别的任务通知”。

它只维护等待队列；不绑定互斥锁，也不拥有被保护的数据。因此，条件本身应保存在调用方自己的共享状态里，并由调用方负责并发安全。

头文件：asco/sync/condition_variable.h

1. 基本模型

典型模式是：

等待方：反复检查 predicate；若条件不成立，则挂起等待通知；
通知方：先更新共享状态，再调用 notify_one() / notify(n)，或使用带 predicate 的通知重载做额外门控。

示例：

#include <atomic>
#include <asco/future.h>
#include <asco/sync/condition_variable.h>

using namespace asco;

sync::condition_variable cv;
std::atomic_bool ready{false};

future<void> waiter() {
    co_await cv.wait([&]() { return ready.load(std::memory_order::acquire); });
    // 这里只有在 ready == true 时才会继续
    co_return;
}

future<void> notifier() {
    ready.store(true, std::memory_order::release);
    cv.notify_one();
    co_return;
}

2. `wait(predicate)`：等待直到条件成立

co_await cv.wait([&]() { return ready.load(std::memory_order::acquire); });

语义：

wait(...) 返回 future<void>，需要在 ASCO runtime 上下文中 co_await。
每次进入等待循环时，都会先执行一次 predicate：
- 若返回 true：立即返回，不进入等待队列；
- 若返回 false：当前任务进入等待队列，直到被通知后再次检查 predicate。
被通知后，wait(...) 会再次检查 predicate；只有条件成立时才返回。

这意味着：

wait(...) 适合表达“我要一直等到条件真的成立”；
调用 notify_*() 只是唤醒等待方，不等于条件一定已经满足；最终是否返回，仍由 predicate 决定。

3. `wait_once(predicate)`：最多等待一次通知

bool suspended = co_await cv.wait_once([&]() { return ready.load(std::memory_order::acquire); });

返回值语义：

返回 false：predicate 一开始就成立，因此没有挂起；
返回 true：predicate 一开始不成立，当前任务确实挂起过一次，并在收到一次通知后恢复执行。

与 wait(...) 的关键区别：

wait_once(...) 在恢复后不会再次检查 predicate；
它更像一个低层原语：告诉你“这次是否挂起过一次”，而不是“条件现在是否已经成立”。

因此，若你的需求是“直到条件成立才继续”，优先使用 wait(...)；只有在你明确要自己控制重试逻辑时，才使用 wait_once(...)。

4. 通知接口

4.0 `notify_failed`

带 predicate 的通知重载在失败时返回 std::expected<..., notify_failed>，错误码含义如下：

notify_failed::predicate_false：通知侧的 predicate 返回 false，因此这次通知被拒绝，没有唤醒任何等待方。
notify_failed::no_waiters：当前没有等待方，因此没有执行通知。

4.1 `notify_one()`

bool woke = cv.notify_one();

语义：

若当前存在等待方：唤醒一个，并返回 true；
若等待队列为空：返回 false。

4.2 `notify_one(predicate)`

auto result = cv.notify_one([&]() { return ready.load(std::memory_order::acquire); });

语义：

返回 std::expected<void, notify_failed>。
若没有等待方：返回 std::unexpected{notify_failed::no_waiters}。
若有等待方但 predicate 返回 false：返回 std::unexpected{notify_failed::predicate_false}，且不会唤醒等待方。
若有等待方且 predicate 返回 true：唤醒一个等待方，并返回成功结果。

4.3 `notify(n)` 与 `notify()`

std::size_t woke = cv.notify(3);
std::size_t all = cv.notify();

语义：

唤醒最多 n 个等待方；
返回值为“本次实际唤醒的数量”；
当等待方数量少于 n 时，只会唤醒现有等待方。
notify() 等价于“尽可能唤醒全部等待方”。

4.4 `notify(predicate, n)` 与 `notify(predicate)`

auto result = cv.notify([&]() { return ready.load(std::memory_order::acquire); }, 3);
auto all = cv.notify([&]() { return ready.load(std::memory_order::acquire); });

语义：

返回 std::expected<std::size_t, notify_failed>。
若没有等待方：返回 std::unexpected{notify_failed::no_waiters}。
若有等待方但 predicate 返回 false：返回 std::unexpected{notify_failed::predicate_false}，且不会唤醒等待方。
若 predicate 返回 true：唤醒最多 n 个等待方，并返回实际唤醒数量。
notify(predicate) 等价于“在 predicate 允许时，尽可能唤醒全部等待方”。

5. 与 `std::condition_variable` 的区别

sync::condition_variable 和传统线程条件变量的使用习惯相似，但有两个重要区别：

它是异步的：等待通过 co_await 完成，而不是阻塞线程；
它不与外部 mutex 绑定：共享状态的保护和可见性由调用方自己负责。

因此，predicate 最好只读取：

原子变量；或
已由其它同步原语安全保护的状态。

6. 使用建议

先更新共享状态，再调用 notify_*()；不要先通知后改状态。
需要“直到条件成立”为止时，使用 wait(...)。
需要“如果条件不成立，就先睡一次，醒来后我自己决定下一步”时，使用 wait_once(...)。
若多个等待方共享同一个条件，使用 notify(n) 控制一次唤醒多少个任务。
若通知本身也需要门控，使用带 predicate 的 notify_*() 重载；predicate 应尽量短小、无阻塞、无副作用。
当前实现尚未提供任务取消支持；设计等待协议时，需要考虑任务可能长期等待的路径。

`sync::mutex`：互斥锁

sync::mutex 用于在多个并发执行流之间建立互斥：同一时刻最多只有一个执行流进入临界区。

它提供两种形态：

sync::mutex<>：只提供互斥，不绑定数据。
sync::mutex<T>：把一个值 T 与互斥锁绑定在一起，通过 guard 直接访问该值。

头文件：asco/sync/mutex.h

1. `mutex<>`（无数据）

1.1 `lock()`：异步获取

#include <asco/sync/mutex.h>
#include <asco/future.h>

using namespace asco;

sync::mutex<> m;

future<void> f() {
    auto g = co_await m.lock();
    // 临界区
    co_return;
} // g 析构后自动解锁

语义：

lock() 返回 future<guard>；需要在 runtime 上下文中 co_await。
当锁空闲时：立即获取并返回 guard。
当锁被占用时：等待直到锁被释放。
guard 不可拷贝、可移动；guard 析构时自动解锁。

1.2 `try_lock()`：立即尝试获取

if (auto g = m.try_lock()) {
    // 获取成功
} else {
    // 获取失败（不等待）
}

语义：

若锁空闲：返回有效 guard。
若锁被占用：返回空 guard。

1.3 `blocking_lock()`：同步阻塞获取

auto g = m.blocking_lock();

语义：

同步阻塞直到获得锁，并返回 guard。
仅允许在“blocking 环境”中调用；否则会触发 panic。
- runtime 之外：允许（此时会阻塞当前线程）。
- runtime 之内：仅当当前任务由 spawn_blocking(...) 启动（即 asco::this_task::is_blocking_env() == true）时允许。

2. `mutex<T>`（保护一个值）

mutex<T> 将一个值 T 与互斥锁绑定。

#include <asco/sync/mutex.h>
#include <asco/future.h>

using namespace asco;

sync::mutex<int> counter{0};

future<void> inc() {
    auto g = co_await counter.lock();
    ++(*g);
    co_return;
}

语义：

lock() 返回 future<mutex<T>::guard>。
通过 *g / g-> 访问被保护的 T。
若 guard 为空（例如 try_lock() 失败返回的 guard），对其解引用会触发 panic。

3. guard 的移动语义

guard 不可拷贝，但可以移动。
被移动后的 guard 变为空（if (g) 为假）。

4. 使用建议

try_lock() 适合实现“不等待的快速路径”；失败时走其它分支。
互斥锁的临界区尽量保持短小，避免把长时间运行的工作放在持锁期间。
在普通异步任务中不要使用 blocking_lock()；请使用 co_await lock()。
只有当你确实处于“blocking 环境”（例如同步代码、或 spawn_blocking 任务内）时，才考虑 blocking_lock()。

`sync::rwlock`：读写锁

sync::rwlock 用于保护“读多写少”的共享状态。

多个 reader 可以同时持有读锁；
writer 必须独占；
一旦有 writer 开始等待，后续 reader 不会再继续插队。
reader 可以尝试把自己持有的读锁原子地升级为写锁。

它提供两种形态：

sync::rwlock<>：只提供读写锁语义，不绑定数据。
sync::rwlock<T>：把一个值 T 与读写锁绑定，通过 guard 直接访问该值。

头文件：asco/sync/rwlock.h

1. `rwlock<>`（无数据）

1.1 `write()` / `try_write()`：获取写锁

#include <asco/sync/rwlock.h>
#include <asco/future.h>

using namespace asco;

sync::rwlock<> lock;

future<void> mutate() {
    auto g = co_await lock.write();
    // 写临界区
    co_return;
}

void maybe_mutate() {
    if (auto g = lock.try_write()) {
        // 立即拿到写锁
    }
}

语义：

write() 返回 future<write_guard>，需要在 runtime 上下文中 co_await。
只有在“没有 reader、也没有 writer”的情况下，writer 才能进入。
若存在活跃 reader 或 writer：write() 等待，try_write() 直接失败。
一旦某个 writer 开始等待，后续 reader 不再继续进入。这让 writer 在已有 reader 退出后优先获得锁，避免被新 reader 长期饿死。
write_guard 不可拷贝、可移动；析构时自动释放写锁。

1.2 `read()` / `try_read()`：获取读锁

#include <asco/sync/rwlock.h>
#include <asco/future.h>

using namespace asco;

sync::rwlock<> lock;

future<void> inspect() {
    auto g = co_await lock.read();
    // 读临界区
    co_return;
}

void fast_path() {
    if (auto g = lock.try_read()) {
        // 立即拿到读锁
    }
}

语义：

read() 返回 future<read_guard>，需要在 runtime 上下文中 co_await。
若当前没有 writer 持有或等待：reader 立即进入。
多个 reader 可以同时成功获取。
若当前已有 writer 持有，或已经有 writer 在等待：新的 reader 会等待；try_read() 则直接失败。
read_guard 不可拷贝、可移动；析构时自动释放读锁。

1.3 `read_guard::upgrade()`：读锁升级为写锁

#include <asco/future.h>
#include <asco/sync/rwlock.h>
#include <utility>

using namespace asco;

future<void> maybe_promote(sync::rwlock<> &lock) {
    auto g = co_await lock.read();

    // ... 先读取共享状态

    auto wg = co_await std::move(g).upgrade();
    if (!wg) {
        // 升级失败：当前读视图不再有效，需要自行决定是否重试
        co_return;
    }

    // 现在已经持有写锁
    co_return;
}

语义：

upgrade() 定义在 read_guard 上，返回 future<write_guard>。
它会消耗当前 read_guard，因此调用形式应为 std::move(g).upgrade()。
若当前 reader 已经是最后一个 reader：会直接升级为 writer。
若当前没有别的升级者，但仍有其它 reader：会等待直到其它 reader 释放，然后把当前 guard 升级为写锁。
若此时已经有另一个 upgrade() 在进行中：当前 upgrade() 会失败，并返回空的 write_guard。
升级开始后，会像等待中的 writer 一样阻止新的 reader 继续进入。
若 guard 为空，upgrade() 返回空的 write_guard。

这意味着 upgrade() 的成功语义是“当前读视图被原子地提升为写锁”；一旦做不到这个保证，API 会失败，而不是等待到某个未来时刻再给你一个写锁。

2. `rwlock<T>`（保护一个值）

rwlock<T> 把一个值 T 与读写锁绑定：

写锁返回可写 guard。
读锁返回只读 guard；

#include <asco/sync/rwlock.h>
#include <asco/future.h>

using namespace asco;

struct config {
    int version;
    bool enabled;
};

sync::rwlock<config> cfg{config{1, true}};

future<void> read_cfg() {
    auto g = co_await cfg.read();
    int version = g->version;
    bool enabled = g->enabled;
    (void)version;
    (void)enabled;
    co_return;
}

future<void> update_cfg() {
    auto g = co_await cfg.write();
    g->version += 1;
    g->enabled = false;
    co_return;
}

语义：

co_await lock.write() 返回 rwlock<T>::write_guard，通过 *g / g-> 读写访问 T。
co_await lock.read() 返回 rwlock<T>::read_guard，通过 *g / g-> 只读访问 T。
co_await std::move(read_guard).upgrade() 返回 rwlock<T>::write_guard，用于尝试把当前读锁原子升级成写锁；失败时返回空 guard。
try_read() / try_write() 分别返回对应 guard；失败时返回空 guard。
若 guard 为空（例如 try_*() 失败），对其解引用会触发 panic。

3. 调度与公平性

rwlock 的行为重点不是“严格公平”，而是“writer 不被新 reader 持续插队”：

已经持有读锁的 reader 可以并发完成；
等待中的 writer 会阻止后续 reader 再进入；
被唤醒的具体等待方顺序不保证是 FIFO。

如果你的场景是：

读操作远多于写操作；
读操作之间可以安全并发；
写操作希望在已有 reader 完成后尽快获得独占权限；

那么 rwlock 比 mutex 更合适。

4. 使用建议

用 try_read() / try_write() 表达“不等待的快速路径”；失败时自行走降级逻辑。
如果你一开始就知道自己要修改共享状态，优先直接使用 write() / try_write()；upgrade() 适合先读后判定是否需要写入的路径。
需要从读路径进入写路径时，用 std::move(g).upgrade()；不要先手动释放读锁再去重新 write()，否则中间会出现竞争窗口。
upgrade() 失败时，原 read_guard 已经被消费；此时应丢弃先前读到的状态，并从头重新执行“读取-校验-升级/写入”流程。
不要假设支持“写锁降级为读锁”；当前 API 没有提供这种能力。
同一把 rwlock 上同时只能有一个升级者继续等待；后续并发 upgrade() 会快速失败并返回空 guard。
如果读写比例并不明显偏向“读多写少”，或者临界区本身非常短，优先考虑语义更直接的 mutex。

`sync::spinlock`：自旋锁

sync::spinlock 提供线程间互斥：同一时刻最多只有一个执行流持有锁。

它适用于：

临界区很短、竞争不激烈的场景。

头文件：asco/sync/spinlock.h

1. `spinlock<>`（无数据）

#include <asco/sync/spinlock.h>

asco::sync::spinlock<> lock;

{
    auto g = lock.lock();
    // 临界区
}
// g 析构后自动解锁

语义：

lock.lock() 会获取锁，并返回一个守卫对象（guard）。
guard 的生命周期内锁处于持有状态；guard 析构时解锁。

2. `spinlock<T>`（保护一个值）

spinlock<T> 用于把某个值 T 与一把锁绑定在一起。

#include <asco/sync/spinlock.h>
#include <vector>

asco::sync::spinlock<std::vector<int>> xs;

{
    auto g = xs.lock();
    g->push_back(1);
    g->push_back(2);
}

语义：

xs.lock() 返回一个 guard。
通过 *g / g-> 访问被保护的 T。

3. 使用建议

只在临界区很短时使用自旋锁；临界区越长，对其它执行流的影响越大。
不要在持有自旋锁期间执行可能长时间运行的操作（例如长循环、阻塞调用）。
不要在持有自旋锁期间跨越 co_await；建议在 co_await 前释放锁，恢复后再重新获取。

`sync::semaphore<N>`：信号量

sync::semaphore<N> 表示一个“最多拥有 N 个许可（permit）”的计数信号量。

当许可数大于 0 时，acquire()/try_acquire() 会消耗 1 个许可并成功返回。
当许可数为 0 时，acquire() 会等待，直到有许可被释放。

常用别名：

sync::binary_semaphore：sync::semaphore<1>
sync::unlimited_semaphore：许可上限非常大

头文件：asco/sync/semaphore.h

1. 构造与计数

sync::semaphore<3> sem{2};

语义：

初始许可数为 min(N, count)。
get_count() 返回当前许可数的一个瞬时值（可用于观测与调试；并发访问下不保证与后续操作之间的时序关系）。

2. 获取许可

2.1 `try_acquire()`：非等待获取

bool ok = sem.try_acquire();

语义：

若当前许可数大于 0：消耗 1 个许可并返回 true。
若当前许可数为 0：不等待，直接返回 false。

2.2 `acquire()`：等待获取

co_await sem.acquire();

语义：

若当前许可数大于 0：消耗 1 个许可并返回。
若当前许可数为 0：挂起等待，直到有其它执行流释放许可。
支持任务取消。

acquire() 返回 future<void>，需要在 ASCO runtime 上下文中 co_await。

2.3 `blocking_acquire()`：同步等待获取

sem.blocking_acquire();

语义：

同步地等待并获取 1 个许可。
仅允许在“blocking 环境”中调用；否则会触发 panic。
- runtime 之外：允许（此时会阻塞当前线程）。
- runtime 之内：仅当当前任务由 spawn_blocking(...) 启动（即 asco::this_task::is_blocking_env() == true）时允许。

3. 释放许可

3.1 `release(n = 1)`

std::size_t released = sem.release(5);

语义：

向信号量增加最多 n 个许可。
许可数不会超过上限 N。
返回值为“本次实际增加的许可数”。

当存在等待 acquire() 的执行流时，释放许可会让其中最多 released 个等待方继续执行；被唤醒的具体等待方不作保证。

4. 典型用法

4.1 `binary_semaphore`：一次只允许一个进入

#include <asco/sync/semaphore.h>
#include <asco/future.h>

using namespace asco;

future<void> f() {
    sync::binary_semaphore sem{1};
    co_await sem.acquire();
    // 临界区
    sem.release();
    co_return;
}

4.2 限流：最多并发 `N` 个任务

#include <asco/sync/semaphore.h>
#include <asco/core/runtime.h>
#include <asco/future.h>

using namespace asco;

future<void> limited(sync::semaphore<8> &sem) {
    co_await sem.acquire();
    // ... 执行受限工作
    sem.release();
    co_return;
}

5. 使用建议

使用 try_acquire() 实现“尽力而为”的快速路径；失败时走其它分支。
使用 acquire() 表达“必须获得许可才能继续”。
不要在未获得许可时调用 release() 来“抵消”；这会破坏许可语义。

时间与计时器（asco::time）

asco::time 提供基于单调时钟的“睡眠（sleep）”与“周期 tick（interval）”工具，用于在协程中进行延时与周期调度。

睡眠：见 sleep：sleep_until 与 sleep_for
周期：见 interval：asco::time::interval

时间基准

计时基于 std::chrono::steady_clock（单调递增，不受系统时间校准影响）。
所有等待接口语义上都以“目标时间点是否已到”为准：如果目标已过去，应当立即完成而不是等待。

使用建议

需要“尽快触发一次然后每隔 N 时间触发”：使用 interval，并在循环里 co_await it.tick()。
需要“延迟到某个绝对时间点”：使用 sleep_until。
需要“延迟一段相对时间”：使用 sleep_for。

睡眠（sleep）：`sleep_until` 与 `sleep_for`

本页介绍 asco::time 里的睡眠等待工具，用于在协程中进行延时。

时间基准

计时基于 std::chrono::steady_clock（单调递增，不受系统时间校准影响）。
等待接口语义上以“目标时间点是否已到”为准：如果目标已过去，应当立即完成而不是等待。

`sleep_until`：睡到某个时间点

签名：future<void> sleep_until(std::chrono::steady_clock::time_point tp)

语义：

等待直到 tp 到达。
若 tp 在调用时已经是过去时间点，应立即返回。
可被取消：当调用方协程在等待期间收到取消请求时，等待会以取消方式结束（通常表现为对任务/句柄 co_await 时抛出 core::coroutine_cancelled）。

`sleep_for`：睡一段时间

签名：future<void> sleep_for(duration)（duration 满足 util::types::duration_type）

语义：

等待至少给定的 duration。
若 duration <= 0，应立即返回。
与 sleep_until 一样可被取消。

何时使用

需要“延迟到某个绝对时间点”：用 sleep_until。
需要“延迟一段相对时间”：用 sleep_for。

周期 tick（interval）：`asco::time::interval`

本页介绍 asco::time::interval，用于在协程中以固定周期进行调度。

构造

interval(duration)

duration 为周期长度（满足 util::types::duration_type）。

成员

future<void> tick()

语义

第一次 tick()：以构造时刻为基准；若调用时尚未到达“构造时刻 + duration”，则等待到该时间点。
第 N 次 tick() 的目标时间点始终是“构造时刻 + N × duration”。
若调用时已经超过该目标时间点，则本次 tick() 立即返回。
因此，当调用方长期未轮询时，后续若干次 tick() 可能连续立即返回，直到重新追上当前时间。
可被取消：当 tick() 内部正在等待下一次到期时，取消应使等待以取消方式结束。

何时使用

需要“尽快触发一次然后每隔 N 时间触发”：用 interval，并在循环里 co_await it.tick()。

并发数据结构与算法

本章记录 asco 中与并发相关的数据结构实现与使用约定。

hash_map<K, V>：并发哈希表

`asco::concurrency::hash_map<K, V>`：并发哈希表

hash_map<K, V> 提供在并发场景下可用的键值存储，并以“显式失败码 + 可重试”作为主要控制流：当遇到并发冲突或正在重整（rehash）时，操作会返回错误码提示你让出执行权后重试。

类型要求

K 必须可比较（==）且可被 std::hash<K> 哈希。
V 需要满足可移动/可安全搬运（以及 nothrow 析构）。
当 V = void 时，可作为集合（set）语义使用：只关心 key 是否存在。

并发语义

1) 线程安全边界

多个任务/线程可以并发调用 try_insert/try_get/try_contains/try_remove/try_rehash 以及 insert/get/remove/contains/size。
在并发下，部分调用可能失败但不破坏容器状态；调用方需要按错误码进行重试或退避。

2) `guard` 的语义（`try_get()` 返回）

当 V != void 且 try_get(key) 成功时返回一个 guard。
在 guard 生命周期内：该元素不会被 try_remove() 或 try_rehash() 销毁/搬迁，从而保证通过 guard 取得的引用不会悬垂。
不要长时间持有 guard：长持有会让 try_remove()（对同一 key）以及 try_rehash() 更容易失败或等待。

3) value 的并发读写

hash_map 仅保证元素生命周期安全；不保证对同一个 value 的并发修改一定无数据竞争。
若需要并发写同一个 V，请让 V 自身具备同步语义（如原子/锁/无锁结构），或由外部加锁。

API 与错误码

`size()`

返回：std::size_t。

返回当前元素数量的一个快照。
在并发插入/删除过程中，该值可能不是线性一致（可能偏旧）；用于统计、监控或启发式决策。

`try_insert(key, value)`

返回：std::expected<std::monostate, insert_failed>。

key_repeated：key 已存在。
rehash_needed：建议调用 try_rehash() 后重试插入。
rehashing：当前有线程/任务正在执行 try_rehash()。
retry：遇到瞬态并发冲突，建议退避后重试。

当 V = void 时，还提供 try_insert(key)（不带 value），表示“插入 key”。

`try_get(key)`

当 V != void 时：返回 std::expected<guard, get_failed>。

当 V = void 时：返回 std::expected<std::monostate, get_failed>。成功表示 key 存在。

none：key 不存在。
rehashing：当前正在 try_rehash()。
retry：遇到瞬态并发冲突，建议退避后重试。

`try_contains(key)`

返回：std::expected<bool, contains_failed>。

成功时返回 true/false，表示 key 是否存在。
成功时返回 true/false，表示 key 是否存在。
try_contains 不提供元素生命周期保护；返回 true 也不意味着后续操作一定不会因并发而失败。
失败时返回 contains_failed：
- rehashing：当前正在 try_rehash()。
- retry：遇到瞬态并发冲突，建议退避后重试。

`try_remove(key)`

当 V != void 时：返回 std::expected<V, remove_failed>。

当 V = void 时：返回 std::expected<std::monostate, remove_failed>。

none：key 不存在。
guard_protecting：有 guard 正在保护该元素，暂时无法删除。
rehashing：当前正在 try_rehash()。
retry：遇到瞬态并发冲突，建议退避后重试。
thrown：容器内部存在由异常路径引入的不可移除状态（见下文“异常行为”）。

`try_rehash()`

try_rehash() 尝试扩容并重整内部结构：

返回 true：本次成功执行。
返回 false：未能开始（例如已有其他 try_rehash() 在进行）。
当 try_rehash() 进行中，try_insert/try_get/try_remove 可能返回 rehashing。
try_rehash() 可能需要等待相关 guard 释放；因此长时间持有 guard 会显著影响 try_rehash() 的完成时机。

注意：容器刻意不提供“不带 try_ 的 rehash()”。在高并发下，如果 rehash() 被设计为“最终必然成功”，很容易在短时间内连续触发多次扩容，导致容量被快速放大、浪费内存。

该容器更推荐的模式是：把 rehash 作为一次性纠偏动作（例如在插入返回 rehash_needed 时执行一次 try_rehash()，随后立刻重试插入）。只要当前容量已经足够，后续操作就不需要再触发 rehash。

便捷接口（非 `try_`）

hash_map 额外提供一组“不带 try_ 前缀”的便捷接口：它们会在内部循环调用 try_*，对 rehashing/retry 做忙等重试（使用 cpu_relax()）。

在协程环境中，若你希望让出执行权而不是忙等，优先使用 try_* 并在失败时 co_await asco::this_task::yield()。

`insert(key, value)` / `insert(key)`

当 V != void：bool insert(const K&, V&&)。
当 V = void：bool insert(const K&)。
成功返回 true。
若 key 已存在返回 false。
遇到 rehash_needed 时，会执行一次 try_rehash() 并立即重试插入；容量足够后就会停止扩容并完成插入。

`get(key)`（仅 `V != void`）

返回 guard：

若 key 存在，返回一个有效 guard。
若 key 不存在，返回空 guard（可用 if (g) { ... } 判断）。

`remove(key)`

当 V != void：返回 std::optional<V>。成功删除返回 V；key 不存在返回 std::nullopt。
当 V = void：返回 bool。成功删除返回 true；key 不存在返回 false。

注意：若内部遇到 remove_failed::thrown，会触发 panic（表示出现了异常遗留状态，需由更高层处理）。

`contains(key)`

返回 bool：key 存在返回 true，不存在返回 false。该接口同样会在 rehashing/retry 时忙等重试。

异常行为

try_insert() 可能因 K/V 的构造或移动而抛异常。
在发生异常后：
- 该次插入不会返回成功。
- 后续 try_remove() 可能返回 thrown（表示出现了需要调用方关注的异常遗留状态）。

建议：在高可靠场景中优先使用 nothrow 构造/移动的类型；或在捕获异常后采用更高层的恢复策略（例如重新创建容器并回放数据）。

示例（协程风格）

插入：处理 `rehash_needed` 与可重试错误

for (;;) {
    auto r = m.try_insert(k, v);
    if (r) break;

    switch (r.error()) {
    case decltype(m)::insert_failed::key_repeated:
        co_return; // 或者做更新逻辑
    case decltype(m)::insert_failed::rehash_needed:
        (void)m.try_rehash();
        break;
    case decltype(m)::insert_failed::rehashing:
    case decltype(m)::insert_failed::retry:
        co_await asco::this_task::yield();
        break;
    }
}

删除：处理 `guard_protecting`

for (;;) {
    auto r = m.try_remove(k);
    if (r || r.error() == decltype(m)::remove_failed::none) break;

    if (r.error() == decltype(m)::remove_failed::guard_protecting
        || r.error() == decltype(m)::remove_failed::retry
        || r.error() == decltype(m)::remove_failed::rehashing) {
        co_await asco::this_task::yield();
        continue;
    }

    // remove_failed::thrown 等：按业务决定如何处理
    co_return;
}

判断存在性：处理可重试错误

for (;;) {
    auto r = m.try_contains(k);
    if (r) {
        if (*r) {
            // exists
        } else {
            // not exists
        }
        break;
    }
    co_await asco::this_task::yield();
}

进阶

本章收录一些偏工程实践与进阶语义的主题：

任务取消机制
任务本地存储（Task-local storage）
asco::core::daemon：后台守护线程基类
测试框架
贡献代码

任务取消机制

ASCO 的“任务取消”主要面向由 runtime 调度的任务（join_handle<T>）。它的目标是：

由外部请求取消一个正在运行/挂起的任务；
任务内可以观察当前取消状态，并在取消发生时执行清理或通知；
取消发生并被处理后，该任务会被终止，不再继续执行。

术语：本文的“任务”指 spawn(...) 产生的 runtime 任务（join_handle）。

当前公开语义同时包含两层行为：

每个任务或嵌套异步操作都有自己的取消状态与取消回调；

当外层任务等待内部异步操作时，外层取消会一并终止那些仍未完成的内部操作。

1. 取消的组成

2. 如何取消一个任务：`join_handle::cancel()`

外部取消的标准方式是：先对 join_handle<T> 调用 cancel()，再在需要时通过 co_await 该句柄观察取消结果：

#include <asco/future.h>
#include <asco/yield.h>

using namespace asco;

future<void> example_cancel(join_handle<void> &h) {
    h.cancel();

    bool cancelled = false;
    try {
        co_await h;  // join
    } catch (core::coroutine_cancelled &) {
        cancelled = true;
    }

    // cancelled == true
    (void)cancelled;
}

行为要点：

取消被任务观察并收束后，会执行已注册的取消回调（cancel_callback），并结束该任务后续执行。
被取消的 join_handle 在 co_await 时会抛出 asco::core::coroutine_cancelled。

3. 任务内如何响应取消

3.1 首选：注册取消回调（`cancel_callback`）

如果你的代码需要在取消发生时执行某些动作（比如设置 flag、通知别的协程/线程），可以注册回调：

#include <atomic>
#include <asco/cancellation.h>
#include <asco/this_task.h>

using namespace asco;

future<void> with_cancel_callback(std::atomic_bool &flag) {
    core::cancel_callback cb{[&]() { flag.store(true, std::memory_order::release); }};

    while (!flag.load(std::memory_order::acquire)) {
        co_await this_task::yield();
    }

    co_return;
}

语义：

回调在取消请求被处理时执行。
回调的常见用途是：触发一次“通知/标记/释放资源/恢复状态”的动作。
回调总是绑定到“当前正在执行的这段异步操作”；如果代码运行在 task::join_all(...) 等嵌套等待内部操作的场景中，回调只影响它所在的那一项操作。

回调顺序与生命周期：

cancel_callback 的预期用法，是在当前任务内作为局部对象注册一个取消回调。
注册与销毁应发生在同一任务上下文中；它不适合作为跨任务、跨所有权边界传递的通用对象使用。
在这种局部 RAII 用法下，对象析构时会自动注销对应回调。
取消回调以“后注册先执行”的顺序调用；这里说的是同一任务中按嵌套作用域注册的多个回调。
不要把 cancel_callback 当成可以长期保存、放入共享状态、容器或堆对象中并任意延后销毁的通用注册对象使用。

3.2 补充：查询取消请求（`cancel_requested()`）

如果你希望在协程的安全点主动结束逻辑，可以轮询当前取消状态：

#include <asco/this_task.h>

using namespace asco;

future<void> worker_loop() {
    auto &token = this_task::get_current_cancel_token();

    while (true) {
        if (token.cancel_requested()) {
            co_return;
        }
        co_await this_task::yield();
    }
}

语义：

当 cancel_requested() 为 true 时，表示当前这段异步操作已被请求取消。
该写法适合：你需要在循环/阶段边界按自己的方式收尾并退出。

4. 层级化取消

某些组合等待场景会在当前任务内部同时推进多个异步操作，例如 task::join_all(...)。

行为语义：

如果外层任务在等待这些内部操作时被取消，仍未完成的内部操作也会结束。
内部操作中注册的 cancel_callback 仍会执行，因此对应的清理逻辑可以放在回调里。
对于需要可靠清理的代码，优先使用 cancel_callback，不要只依赖循环里偶尔轮询一次 cancel_requested()。

使用建议：

当子任务需要释放资源、唤醒等待者或设置完成标志时，优先注册 cancel_callback。
当你只需要在本段异步逻辑的阶段边界主动退出时，再轮询 get_current_cancel_token().cancel_requested()。

5. 常见坑与建议

this_task::get_current_cancel_token() 只能在 runtime 中调用；不在 runtime 会 panic。
取消回调适合做“通知/打点/设置标志/清理资源/恢复状态”，不要在回调里做复杂阻塞操作。
cancel_callback 绑定的是当前这段异步操作；不要把它当成跨任务、跨上下文复用的通用注册句柄。
如果你需要让自定义 awaiter 支持取消，可以在挂起前注册 cancel_callback，并在回调里安排唤醒或中断。
任何一个挂起点都有可能执行取消，如果一个异步函数需要支持取消，应确保所有挂起点都能正确响应取消请求，如注册 cancel_callback 或查询 cancel_requested()。

任务本地存储（Task-local storage）

Task-local storage（下文简称 TLS）用于为“单个任务（task）”保存一份上下文数据，并在该任务的整个协程调用链中随时访问。

它和 thread_local 的核心区别是：

thread_local 绑定线程；task_local 绑定任务（由 ASCO runtime 调度执行）。
task_local 在同一任务内跨 co_await / yield() 保持一致与可见。

典型用途：请求/追踪上下文、每任务统计、逻辑上的“隐式参数”等。

快速开始

1）定义 TLS 类型

asco 的 TLS 以“类型”为键：一个任务里只存放一份 TLS 对象。如果你需要多份数据，把它们聚合到一个结构体里即可。

struct my_tls {
    int request_id{};
    std::string trace_id;
};

2）在 `spawn()` / `block_on()` 时提供初始值

创建任务时把 TLS 实例作为第二个参数传入：

#include <asco/core/runtime.h>

auto h = asco::spawn(
    []() -> asco::future<void> {
        // ...
        co_return;
    },
    my_tls{.request_id = 42, .trace_id = "abc"});

如果不需要 TLS，直接使用 spawn(fn) 即可。

3）在任务内部访问与修改

在任务（协程）内部，通过 asco::this_task::task_local<T>() 获取 TLS 的引用：

#include <asco/this_task.h>

auto &tls = asco::this_task::task_local<my_tls>();
tls.request_id += 1;

同一任务内多次调用 task_local<T>() 会返回同一对象的引用；写入会在后续 co_await / yield() 后仍然可见。

语义说明

同一任务内：引用稳定、跨挂起保持

在同一个任务中：

task_local<T>() 返回的引用在该任务生命周期内保持稳定（同一对象）。
对 TLS 的修改在 co_await / this_task::yield() 之后仍然能读到。

不同任务之间：相互隔离、不会继承

TLS 不会从父任务自动继承到子任务。

也就是说：父任务 spawn() 了一个子任务，子任务要想使用 TLS，必须在创建子任务时显式传入它自己的 TLS 初始值；子任务对 TLS 的修改不会影响父任务。

如果你希望“父子任务共享状态”，应显式使用共享对象（例如 std::shared_ptr<state>）并把它作为 TLS 的成员或直接作为任务参数传递。

类型必须匹配

TLS 是按类型存取的：

你在 spawn(fn, tls_value) 里传入的 TLS 类型是 T，那么任务内部必须用 task_local<T>() 访问。
如果使用了不同的类型访问，会触发断言失败（类型安全检查）。

因此，一个任务只会有一份 TLS：想存多种值，请把它们放进同一个结构体。

若当前没有正在运行的任务，task_local<T>() 会触发 panic（提示“当前没有正在运行的任务”）。
不要在 runtime 之外调用它。

析构

TLS 会在任务执行期间一直存在，并且至少存活到任务结束。

析构时机不做精确保证，但可以依赖以下语义：

若你在任务结束后仍持有对应的 join_handle，TLS 可能继续存活，从而延迟资源回收。
若你提前销毁 join_handle，TLS 也不会因此在任务尚未结束时提前析构。

建议：不要把 TLS 析构发生的“确切时刻”当作业务语义；若需要确定性的清理，应在任务函数内部显式管理资源生命周期。

常见模式与建议

把 TLS 当作“每任务上下文”，不要用它做跨任务共享。
需要传播上下文时，显式把上下文作为 spawn() 的第二个参数传给新任务。

最小示例

下面示例展示：初始值来自 spawn()，修改跨 yield() 保持，且同一任务内引用稳定。

struct tls_int { int value; };

auto h = asco::spawn(
    []() -> asco::future<int> {
        auto &tls = asco::this_task::task_local<tls_int>();
        tls.value += 1;
        co_await asco::this_task::yield();
        co_return asco::this_task::task_local<tls_int>().value;
    },
    tls_int{41});

int result = co_await h; // result == 42

`asco::core::daemon`：后台守护线程基类

asco::core::daemon 用于把一个“循环执行的后台工作”封装为一个对象。

daemon 在内部启动一个后台线程。
线程按固定生命周期运行：init() → 重复调用 run_once(...) → shutdown()。
对象析构时会请求线程停止，并等待线程退出。

头文件：asco/core/daemon.h

1. 生命周期

1.1 启动：`start()`

start() 启动后台线程，并返回一个 init_waiter。

语义：

后台线程开始后会先调用 init()。
init_waiter 析构时会等待 init() 完成。

典型用法（在派生类构造函数中启动，并等待初始化完成）：

#include <asco/core/daemon.h>

struct my_daemon : asco::core::daemon {
    my_daemon() : daemon("my-daemon") {
        auto _ = start();
        // 这里开始可以认为 init() 已经完成
    }

    bool init() override;
    bool run_once(std::stop_token &st) override;
    void shutdown() override;
};

start() 是受保护成员函数，只能在 daemon 的派生类内部调用。

1.2 停止：析构函数

当 daemon 对象析构时：

会请求后台线程停止（通过 std::stop_token 传递 stop 请求）。
会调用一次 awake()，用于唤醒正在等待的后台线程。
会等待后台线程退出。

2. 线程执行逻辑：`init()` / `run_once()` / `shutdown()`

2.1 `init()`

virtual bool init();

语义：

在后台线程开始工作前调用一次。
返回 true 表示初始化成功；返回 false 表示启动失败。
当 init() 返回 false 时，后台线程不会进入 run_once() 循环，并会调用 shutdown()。

2.2 `run_once(std::stop_token &st)`

virtual bool run_once(std::stop_token &st);

语义：

在后台线程中反复调用。
当 st.stop_requested() 为 true 时，调用方应尽快结束当前轮次并返回。
返回 true 表示继续下一轮；返回 false 表示结束循环。

2.3 `shutdown()`

virtual void shutdown();

语义：

在线程退出前调用一次，用于资源释放与收尾。
无论 init() 失败还是正常退出循环，都会调用 shutdown()。

3. 唤醒与等待：`awake()` 与 `sleep_until_awake*`

3.1 `awake()`

void awake();

语义：

使一次等待中的 sleep_until_awake* 结束等待。
可由任何线程调用。

3.2 等待：`sleep_until_awake...`

daemon 提供一组等待函数，供 run_once() 在“无事可做时”进入等待：

sleep_until_awake()：等待直到被 awake() 唤醒。
sleep_until_awake_for(duration)：等待直到被唤醒或超时。
sleep_until_awake_before(time_point)：等待直到被唤醒或到达指定时间点。

语义：

这些函数在当前线程内等待。
发生唤醒或超时后返回。
这些函数不返回“唤醒原因”；如需区分原因，请在 run_once() 中自行检查条件。

4. 使用建议

在 run_once() 中优先使用 sleep_until_awake* 进入等待；这样析构时的停止请求能更快生效。
run_once() 应避免无限阻塞：若必须等待外部事件，建议设置超时并周期性检查 st.stop_requested()。
awake() 表示“有新工作/状态变化”；调用 awake() 前后如何更新共享状态由派生类自行约定。

测试框架

ASCO 提供面向协程与异步代码的轻量级测试框架。测试目标链接 asco::test 后即可使用；如需运行时、任务调度、同步原语或时间组件，通常还应同时链接 asco::core。

当前实现特性：

测试用例为协程，返回 future<asco::test::test_result>。
单个测试可执行文件可注册多个用例。
测试主程序创建带 timer 的多线程 runtime，并通过 join_set 并发执行全部已注册用例。
输出顺序取决于用例完成顺序。
框架统计通过、失败、忽略数量；存在失败时进程返回非 0。
asco::test 提供测试 main()，并自动定义 ASCO_TESTING。

核心接口

公开接口位于 asco/test/test.h：

ASCO_TEST(name, ...)：声明并注册测试用例。
ASCO_CHECK(expr, fmt, ...)：断言失败时立即结束当前测试。
ASCO_CHECK_WITH_FAILCALLBACK(callback, expr, fmt, ...)：断言失败前执行清理或取消逻辑。
ASCO_SUCCESS()：返回成功。
ASCO_IGNORE_TEST：将测试结果标记为忽略。

编写测试用例

基本结构

#include <asco/test/test.h>

using namespace asco;

ASCO_TEST(my_first_test) {
    ASCO_CHECK(1 + 1 == 2, "math broken: {}", 1 + 1);
    ASCO_SUCCESS();
}

name 作为测试名输出。
ASCO_TEST 在静态初始化阶段完成注册。
测试函数通常以 ASCO_SUCCESS() 结束。

ASCO_CHECK

ASCO_CHECK(expr, fmt, ...) 在 expr 为假时立即返回失败结果。错误信息包含：

std::format(fmt, ...) 生成的消息。
当前源码位置，包括文件名、行号、列号。

示例：

ASCO_CHECK(size == expected, "size mismatch: got={}, expected={}", size, expected);

ASCO_CHECK_WITH_FAILCALLBACK

ASCO_CHECK_WITH_FAILCALLBACK(callback, expr, fmt, ...) 与 ASCO_CHECK 等价，但在断言失败前先执行 callback()。适用于超时、后台任务或需要回收资源的场景。

auto h = spawn([&]() -> future<void> {
    co_await time::sleep_for(5s);
});

ASCO_CHECK_WITH_FAILCALLBACK(
    [&]() { h.cancel(); },
    co_await wait_until([&]() { return h.await_ready(); }, 4096),
    "sleep task did not finish in time");

ASCO_IGNORE_TEST

ASCO_TEST(flake_case, ASCO_IGNORE_TEST) {
    ASCO_SUCCESS();
}

一项测试标记了 ASCO_IGNORE_TEST 后统计结果中记为“忽略”，不参与通过/失败统计。

panic 与异常处理

在测试环境下，asco::panic(...) 会抛出 asco::panicked，可用于验证错误路径：

#include <asco/panic.h>

ASCO_TEST(expect_panic) {
    bool caught = false;
    try {
        asco::panic("boom");
    } catch (asco::panicked &e) {
        (void)e;
        caught = true;
    }

    ASCO_CHECK(caught, "panic should throw asco::panicked under ASCO_TESTING");
    ASCO_SUCCESS();
}

测试主程序的异常处理策略：

捕获 asco::panicked 并记为失败，输出 panic: ...。
捕获其他异常并记为失败，输出“发生异常”。

应按 asco::panicked & 捕获，不应依赖 std::exception &。

异步测试模式

测试用例本身是协程，可直接：

co_await 任意 ASCO future 或 awaitable。
通过 spawn(...) 启动并发任务。
使用 co_await this_task::yield() 主动让出执行权。
等待 join_handle 完成。

当前异步工具并不完善，建议为异步条件提供有界等待，避免测试永久挂起：

#include <functional>
#include <asco/this_task.h>

template<class Pred>
asco::future<bool> wait_until(Pred &&pred, std::size_t max_spins = 4096) {
    for (std::size_t i = 0; i < max_spins; ++i) {
        if (std::invoke(pred)) {
            co_return true;
        }
        co_await asco::this_task::yield();
    }
    co_return std::invoke(pred);
}

ASCO_CHECK(co_await wait_until([&] { return ready.load(); }), "condition did not become true");

实践建议：

优先使用 yield 配合条件轮询，避免固定睡眠。
后台任务应设置边界，并在失败路径执行取消或清理。
用例并发执行时，不应共享未同步的全局可变状态。

CMake 与 CTest 接入

最小接入方式

add_executable(test_channel channel.cpp)
target_link_libraries(test_channel PRIVATE asco::core asco::test)

add_test(NAME test_channel COMMAND test_channel)

无需自定义 main()。
无需手工定义 ASCO_TESTING。

当前仓库的组织方式

当前仓库将多个测试源文件聚合为单一测试目标：

add_executable(tests
    cancellation.cpp
    hash_map.cpp
    sync/mutex.cpp
    sync/semaphore.cpp
    task_local.cpp
    time.cpp
)
target_link_libraries(tests PRIVATE asco::core asco::test)

add_test(tests tests)

CTest 层面当前仅注册一个测试项 tests。
具体用例由该可执行文件内部统一注册并运行。

运行测试

通过 CTest：

ctest --test-dir build --output-on-failure

按名称过滤当前仓库的测试目标：

ctest --test-dir build -R '^tests$' --output-on-failure

直接运行测试可执行文件：

./build/tests/tests

典型输出：

[通过] semaphore_basic_try_acquire_release
[失败] sleep_for_waits_at_least_duration: ...
[忽略] hash_map_concurrent_stress
测试结果：10 通过，1 失败，2 忽略

基准测试辅助工具

asco/test/bench.h 提供 asco::test::bench_context，当前用于 benchmarks/channel.cpp。

asco::test::bench_context bench{"channel_e2e_latency", warmup, measure};

auto head = bench.get_span();
if (bench.commit(head)) {
    // 达到 warmup + measure 次提交
}