hiasync

一个单线程下调度异步任务的运行时, 不依赖任何底层系统接口，仅提供在同步环境下调用异步函数的能力.

版本说明

v0.2.1

新增： async fn task_force_abort(task_id: u64)

v0.2.0

新增:

1. JoinHandle<T>::task_id

新增两个强制结束异步任务的接口:

1. Runtime::task_force_abort
2. JoinHandle<T>::force_abort

与此对应的是JoinHandle::join的返回值从T变更为Option<T>. JoinSet::wait_all和JoinSet::wait_any的返回值也由(u64, T)变更为(u64, Option<T>).

JoinHandle<T>的泛型参数类型从Future变更为Future::Output

v0.1.6

增加两个异步接口:

1. async fn task_resume(task_id: u64);
2. async fn notify_events(events: &[Event<'static>]);

v0.1.5

1. 新增task_self/task_suspend/task_resume方便支持业务层自己的异步事件通知机制.

进一步将异步事件数据的传递机制和Runtime解耦, 业务层按需自定义.

async fn send_event(ctx: Rc<RefCell<MsgContext>>, msg: MsgRequest) -> msgResponse {
    non_block_send(&msg);
    // 将消息的key和当前task关联起来
    let task_id = task_self().await;
    ctx.insert(msg.key(), task_id);
    task_suspend().await;
    ctx.get_response(msg.key())
}

fn notify_response(ctx: Rc<RefCell<MsgContext>>, msg: MsgResponse, rt: &mut Runtime) {
    if let Some(task_id) = ctx.get_task_id(msg.key()) {
        ctx.set_response(msg);
        rt.task_resume(task_id);
    }
}

v0.1.4

1. 新增JoinHandle::abort接口
2. 新增JoinSet, 管理多个并发的子异步任务. 参见lib.rs中的测试用例.

v0.1.3

新增partial_or. 消除or中存在的异步任务可能被强制终止的场景.

v0.1.2

新增and, or组合多个Future.

async fn foo() {
    // 收到任意一个消息后返回
    let (r1, r2) = wait_event::<Msg>(MSG_ID1).or(wait_event::<Msg>)(MSG_ID2)).await;
    if let Some(r1) = r1 {
        // 如果收到MsG_ID1
    }
    if let Some(r2) = r2 {
        // 如果收到MsG_ID2
    }
}

样例

主要应用于将非阻塞接口实现的同步函数改造为异步函数的场景.

如下是一个例子:

fn process_msg1(msg: Msg) {
    // ...
    nonblock_send_msg(new_request);
}

fn process_msg2(msg: Msg) {
    // ...
    nonblock_send_msg(new_request);
}

fn main() {
    while let Ok(msg) = recv_msg() {
        match msg.id {
            MSG_ID_1 => process_msg1(msg),
            MSG_ID_2 => process_msg2(msg),
            _ => //...
        }
    }
}

我们希望将process_msg改造为异步函数，可以如下实现:

use hiasync::{wait_event, Event, Runtime};

async fn process_msg1() {
    let msg = wait_event::<Msg>(MSG_ID_1).await;
    //...
    nonblock_send_msg(new_request);
}

async fn process_msg2() {
    let msg = wait_event::<Msg>(MSG_ID_2).await;
    //...
    nonblock_send_msg(new_request);
}

fn main() {
    let mut rt = Runtime::new();
    let _ = rt.spawn(process_msg1());
    let _ = rt.spawn(process_msg2());
    while let Ok(msg) = recv_msg() {
        rt.sched_events(&[Event::new(msg.id, &msg)]);
    }
}