天工开物 #8 Async Rust 的实现

绝大多数第一次接触 Async Rust 的开发者所写的 Hello World 程序是下面这样的：

async fn say_world() {
    println!("world");
}
 
#[tokio::main]
async fn main() {
    // Calling `say_world()` does not execute the body of `say_world()`.
    let op = say_world();
    // This println! comes first
    println!("hello");
    // Calling `.await` on `op` starts executing `say_world`.
    op.await;
}

目前最流行的 Rust Async Runtime tokio^[1] 告诉你，只要在 main 函数前加上 #[tokio::main] 这个咒语，再把 main 函数定义成一个 async fn 函数，就可以在程序的任何地方用 async fn 和 .await 语法享受到并发编程的免费午餐了。

但是，这样的解释显然不足以满足相信 Rust 尽可能显式表达所有开销的哲学的人；也就是说，到底 async fn 里的逻辑是怎么被执行的，代码中出现 .await 的时候，Rust 程序实际在执行什么动作。

本文基于具体的异步运行时和并发库代码，介绍支持 Rust 并发模型工作的核心概念。

Async Runtime

Rust 语言是一门系统级编程语言，这意味着 Rust 程序不像 Node.js 程序那样自带运行时。这不仅意味着没有全局管理内存的垃圾回收器，也意味着没有一个全局默认的异步任务运行时。

例如，以下 Node.js 代码会打印出 Hello world 字串：

(async function() { console.log('Hello world'); })()

而相似的 Rust 代码却不会有任何输出：

async { println!("Hello world") };

Rust 编译器提示：

note: futures do nothing unless you `.await` or poll them

如果在上述代码后加入 .await 调用：

async { println!("42") }.await;

编译会直接报错：

`await` is only allowed inside `async` functions and blocks

编译器提示 .await 只能在 async 代码块里使用，我们尝试把 main 函数改成 async 函数：

async fn main() {
    async { println!("Hello world") }.await
}

编译器同样报错：

`main` function is not allowed to be `async`

我们回过头看 tokio 的示例，在添加了 #[tokio::main] 以后，似乎 main 函数就可以写成 async fn main 了，也能用 .await 触发 Future 计算了。

查看 tokio 的文档，可以知道 #[tokio:main] 实际上是一个过程宏，它会把 async fn main 展开成：

fn main() {
    tokio::runtime::Builder::new_multi_thread()
        .enable_all()
        .build()
        .unwrap()
        .block_on(async {
            println!("Hello world");
        });
}

由此，我们知道，在 Rust 程序里要想执行 async 块里的并发代码，我们需要先构造一个 Async Runtime 的实例，再调用类似 block_on 的方法来驱动并发计算执行。

Simplest Runtime

tokio 为我们提供了一个功能完备的异步运行时。然而，tokio 的实现也是极其复杂的。要想理解 Rust 异步运行时的执行逻辑，我们可以先实现一个最简单的异步运行时：

struct ThreadWaker(Thread);
impl Wake for ThreadWaker {
    fn wake(self: Arc) {
        self.0.unpark();
    }
}
 
fn block_on(fut: F) -> F::Output {
    let mut fut = pin!(fut);
    let t = thread::current();
    let waker = Arc::new(ThreadWaker(t)).into();
    let mut cx = Context::from_waker(&waker);
    loop {
        match fut.as_mut().poll(&mut cx) {
            Poll::Ready(res) => return res,
            Poll::Pending => thread::park(),
        }
    }
}

应用代码如下：

fn main() {
    block_on(async { println!("Hello world") });
}

这时，我们编译运行代码，可以看到屏幕正常输出 Hello world 字串。

我们来看一看上述代码实际执行了什么动作：

1. async 代码块构造了一个并发计算闭包，返回一个 Future trait 的实例。

2. 这个 Future 被传到 block_on 函数里，用 pin! 宏构造了符合 poll 方法要求的 Pin 实例。

3. block_on 函数内，构造了一个 ThreadWaker 实例，持有当前线程的引用。

4. 从 ThreadWaker 示例调用一系列构造器构造出 poll 方法需要的 Context 实例。

5. 调用 poll 方法，在本示例中，首次调用就会执行 println 动作并返回 Poll::Ready(()) 结果，于是 block_on 函数返回。

我们把这个例子稍微写复杂一点，把异步代码块从简单打印一个字符串，改为一个类似 Timer 的逻辑：

struct MyFuture {
    id: u32,
    start: Instant,
    duration: Duration,
}
 
impl Future for MyFuture {
    type Output = ();
 
    fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll {
        println!("Polling {} {}", self.id, chrono::Utc::now());
        let now = Instant::now();
        let expect = self.start + self.duration;
        let id = self.id;
        if expect > now {
            let duration = expect - now;
            let waker = cx.waker().clone();
            thread::spawn(move || {
                sleep(duration);
                println!("Wake up {id} {}", chrono::Utc::now());
                waker.wake();
            });
            println!("Pending {id} {}", chrono::Utc::now());
            Poll::Pending
        } else {
            println!("Ready {id} {}", chrono::Utc::now());
            Poll::Ready(())
        }
    }
}
fn main() {
    println!("Start {}", chrono::Utc::now());
    block_on(MyFuture {
        start: Instant::now(),
        duration: Duration::from_secs(10),
    });
    println!("Finish {}", chrono::Utc::now());
}

输出内容如下：

Start 2023-11-05 09:53:03.202626 UTC
Polling 1 2023-11-05 09:53:03.203152 UTC
Pending 1 2023-11-05 09:53:03.203222 UTC
Wake up 1 2023-11-05 09:53:13.206258 UTC
Polling 1 2023-11-05 09:53:13.206370 UTC
Ready 1 2023-11-05 09:53:13.206396 UTC
Finish 2023-11-05 09:53:13.206407 UTC

可以看到，两次 Polling 之间间隔了十秒钟。实际执行的内容如下：

1. 同上 block_on 函数第一次调用 poll 方法，这次 poll 返回了 Poll::Pending 结果，于是 block_on 阻塞在 thread::park() 上。该方法用同步原语信号量阻塞当前线程，并等待其他线程调用 unpark 唤醒内部信号量。

2. poll 方法返回 Pending 前，从 Context 里取得 Waker 实例，并启动一个新线程，在十秒后调用 waker.wake() 方法。

3. wake 方法调用即调用 self.0.unpark() 方法，于是第一步里阻塞的线程恢复执行，在 loop 中再次调用 MyFuture 的 poll 方法，此时时间已经超过预计等待时间，方法返回 Poll::Ready(()) 结果，block_on 方法也返回。

Simplest Runtime 在 crates.io 上有一个完整的实现 pollster^[2] 运行时。这个运行时的核心代码^[3]同样极其简短，不超过一百行。

Waker API

上面这个例子解释了在 Rust 的 Future API 中神秘的 Context 参数和它唯一传递的 Waker 实例的作用：Waker 实际上就是一个回调闭包，使用 wake 或 wake_by_ref 执行通知异步运行时可以再次调度当前 Future 计算。

waker-fn^[4] 库提供了闭包到 Waker API 的适配器，这直观地展示了 Waker 就是一个回调闭包：

pub fn waker_fn'static>(f: F) -> Waker {
    Waker::from(Arc::new(Helper(f)))
}
struct Helper(F);
implstatic> Wake for Helper {
    fn wake(self: Arc) {
        (self.0)();
    }
 
    fn wake_by_ref(self: &Arc) {
        (self.0)();
    }
}

那么，为什么 Rust 要在 FnOnce 和 Fn 之外设计 Waker API 呢？我们看到 Waker 的定义：

pub struct Waker {
    waker: RawWaker,
}
 
pub struct RawWaker {
    data: *const (),
    vtable: &'static RawWakerVTable,
}
pub struct RawWakerVTable {
    clone: unsafe fn(*const ()) -> RawWaker,
    wake: unsafe fn(*const ()),
    wake_by_ref: unsafe fn(*const ()),
    drop: unsafe fn(*const ()),
}
impl Waker {
    pub fn wake(self) {
        let wake = self.waker.vtable.wake;
        let data = self.waker.data;
        crate::mem::forget(self);
        unsafe { (wake)(data) };
    }
    pub fn wake_by_ref(&self) {
        unsafe { (self.waker.vtable.wake_by_ref)(self.waker.data) }
    }
}

RawWaker 中的 data 字段，内容就是 Waker 本身。这跟我们常见的 Safe Rust 有很大的区别。它使用了等价于 void* 的 *const () 以及一系列方法引用构成 VTable 来组成 RawWaker 进而构成 Waker 实例，从而在类型签名中把类型参数或生命周期标记给抹掉。

这样，Rust 的 Async API 可以直接传递 struct Waker 的实例，而不是 Box 或 Box 等。但是，后者编译时会被 Rust 编译器在生命周期和是否 Send 等类型约束上不停折磨，甚至有些代码表达不出来。

由于我们此时知道 Waker 对应的回调一定是可用的，所以 Rust 的 Async API 选择使用 VTable 和 unsafe 来实现躲过编译器检查的 Box 的类型。

Simplest Parallel Runtime

上述异步运行时实现里，我们观察到了异步代码被 poll 的过程，但是这跟常见的通过并发实现并行，同时执行多个任务还是有所出入。

我们对 block_on 的代码稍作改动，把阻塞在当前线程的行为改为每次传入一个新的 Future 实例，就启动一个新线程来驱动执行，同时，将函数名改为 spawn 以体现出这种行为的改变：

fn spawn(fut: F) -> Receiver
    where F: Future + Send + 'static,
          F::Output: Send,
{
    let (tx, rx) = mpsc::channel();
    thread::spawn(move || {
        let mut fut = pin!(fut);
        let t = thread::current();
        let waker = Arc::new(ThreadWaker(t)).into();
        let mut cx = Context::from_waker(&waker);
        loop {
            match fut.as_mut().poll(&mut cx) {
                Poll::Ready(res) => {
                    let _ = tx.send(res);
                    break;
                }
                Poll::Pending => thread::park(),
            }
        }
    });
    rx
}

这时，我们再编写一个同时启动两个任务的应用程序：

fn main() {
    println!("Start {}", chrono::Utc::now());
    let r1 = spawn(MyFuture {
        id: 1,
        start: Instant::now(),
        duration: Duration::from_secs(10),
    });
    let r2 = spawn(MyFuture {
        id: 2,
        start: Instant::now(),
        duration: Duration::from_secs(10),
    });
    let h1 = thread::spawn(move || {
        r1.recv().unwrap();
        println!("Finish 1 {}", chrono::Utc::now());
    });
    let h2 = thread::spawn(move || {
        r2.recv().unwrap();
        println!("Finish 2 {}", chrono::Utc::now());
    });
    h1.join().unwrap();
    h2.join().unwrap();
}

可以看到以下输出：

Start 2023-11-05 10:46:31.880514 UTC
Polling 1 2023-11-05 10:46:31.880631 UTC
Polling 2 2023-11-05 10:46:31.880650 UTC
Pending 1 2023-11-05 10:46:31.880659 UTC
Pending 2 2023-11-05 10:46:31.880672 UTC
Wake up 2 2023-11-05 10:46:41.881091 UTC
Wake up 1 2023-11-05 10:46:41.881091 UTC
Polling 2 2023-11-05 10:46:41.881206 UTC
Ready 2 2023-11-05 10:46:41.881232 UTC
Polling 1 2023-11-05 10:46:41.881261 UTC
Ready 1 2023-11-05 10:46:41.881276 UTC
Finish 2 2023-11-05 10:46:41.881330 UTC
Finish 1 2023-11-05 10:46:41.881377 UTC

这样，我们就实现了一个最简单的 thread per task 并发模型。

Send + 'static

上面 thread per task 的例子里，spawn 函数里多了 Send 或 'static 约束，这是因为我们在不同线程间传递 Future 实例，同时可能在线程间传递 Future 计算的结果。

这并不是 Rust Async Runtime 自身约束。如前所述，如果 Future 在当前线程构造，随后使用 block_on 在当前线程阻塞等待，Future 没有跨线程传递过，也就不需要满足 Send 约束。同样，异步运行时在能够保证计算一定不会跨线程进行时，异步计算的代码块即 Future 也不必满足 Send 约束。

在 tokio 的实现里，这种同一线程的计算是用 LocalSet 来抽象的：

let runtime = tokio::runtime::Runtime::new().unwrap();
runtime.block_on(async {
    let local = tokio::task::LocalSet::new();
    let rc = Rc::new(42);
    local.run_until(async move {
        println!("rc={rc}");
    }).await;
});

如果把 local.run_until 换成 tokio::spawn 那么 Rust 编译器就会抱怨被 move 进去的 Rc 实例不是 Send 的。

LocalSet 的底层实现是持有一个 LocalState 实例，其中指示了持有该 LocalSet 的线程的 ThreadId 是什么。tokio 运行时调度过程里，如果拿出来的作业对应的 ThreadId 跟当前 ThreadId 不同，则不执行该作业，而是放回 LocalSet 的任务队列里。

glommio^[5] 是一个 thread per code 的异步运行时，它只提供对应到单个线程的 LocalExecutorBuilder 构造器：

let handle = LocalExecutorBuilder::default().spawn(|| async move { ... })?;

虽然顶层 async 块内部，可以使用 executor().create_task_queue(...) 创建任务队列，再调用 glommio::spawn_local_into(fut, tq) 把任务提交到任务队列内并发调度，但是同一个 spawn 内的所有任务，都将在同一个线程内执行。因此，所有 spawn 到 glommio 调度器上的 Future 都不必是 Send 的。

上述 LocalExecutorBuilder 的 spawn 方法最终会调用以下代码：

pub fn spawn(self, fut_gen: G) -> ...
where
    G: FnOnce() -> F + Send + 'static,
    F: Future + 'static,
    T: Send + 'static,
{
    std::thread::Builder::new()
        .spawn(move || {
            let mut le = LocalExecutor::new(...);
            le.init();
            le.run(async move { Ok(fut_gen().await) })
        })
}

其中 LocalExecutor 的 run 方法定义如下：

pub fn run(&self, future: impl Future) -> T { ... }

可以看到，这跟其他运行时的 block_on 函数非常相似。实际上，它们确实都是在当前线程里不停 Poll 来驱动异步计算进行。

RawTask

block_on 函数驱动的运行时，提交上来的 Future 实例以 local 变量的方式存储在当前线程的栈上，我们在当前线程里循环阻塞的 poll 这个 Future 实例。这种情况下，我们没有把 Future 提交到什么任务队列中，也就不需要额外的存储逻辑。

上面介绍的 thread per task 运行时只是把 block_on 放到多个 thread 里，每个 thread 拿住唯一的 Future 实例不停地 poll 获取结果。可以说，thread per task 运行时针对每个 task spawn 出来的线程的栈，就是存储该 Future 实例的地方。

但是，对于 tokio 这样有任务队列的运行时来说，情况就有所不同。这时调度器需要考虑如何存储未执行或执行后返回 Pending 的作业，以等待空闲线程取出执行。

实际上，tokio 仿照标准库用 RawWaker 封装 Waker 的方法，设计了一个 RawTask 结构用以存储待执行的 Future 实例：

#[derive(Clone)]
pub(crate) struct RawTask {
    ptr: NonNull
,
}
 
#[repr(C)]
pub(crate) struct Header {
    pub(super) state: State,
    pub(super) queue_next: UnsafeCell>>,
    pub(super) vtable: &'static Vtable,
    pub(super) owner_id: UnsafeCell>,
}
pub(super) struct Vtable {
    pub(super) poll: unsafe fn(NonNull
),
    pub(super) schedule: unsafe fn(NonNull
),
    pub(super) dealloc: unsafe fn(NonNull
),
    pub(super) try_read_output: unsafe fn(NonNull
, *mut (), &Waker),
    pub(super) drop_join_handle_slow: unsafe fn(NonNull
),
    pub(super) drop_abort_handle: unsafe fn(NonNull
),
    pub(super) shutdown: unsafe fn(NonNull
),
    pub(super) trailer_offset: usize,
    pub(super) scheduler_offset: usize,
    pub(super) id_offset: usize,
}

在 tokio multi-thread 运行时的 spawn 方法调用时，底层会依次调用方法构造 RawTask 实例：

• src/runtime/scheduler/multi_thread/handle.rs/Handle::spawn

• src/runtime/scheduler/multi_thread/handle.rs/Handle::bind_new_task

• src/runtime/task/list.rs/OwnedTasks::bind

• src/runtime/task/mod.rs/new_task

// task::new_task
let raw = RawTask::new::(fut, scheduler, id);
 
// RawTask::new
let ptr = Box::into_raw(Cell::<_, S>::new(fut, scheduler, State::new(), id));
let ptr = unsafe { NonNull::new_unchecked(ptr as *mut Header) };
RawTask { ptr }
 
// Cell::new
let vtable = raw::vtable::();
let ptr = Box::new(Cell {
    header: new_header(
        state,
        vtable,
    ),
    core: Core {
        scheduler,
        stage: CoreStage {
            stage: UnsafeCell::new(Stage::Running(fut)),
        },
        task_id,
    },
    trailer: Trailer::new(),
});

概括地说，就是把 Future 计算作业和调度器 scheduler 都塞到 state 里，存一个 void* 指针指向到这一整坨内容，连同取回这个状态时可以如何 poll/schedule 的 VTable 也在里面。

最后，这个 RawTask 被塞到任务队列里等待被取出后 poll 运行：

// OwnedTasks::bind
lock.list.push_front(task);

其他运行时，包括 smol^[6] 和 yatp^[7] 乃至以上的 glommio 在向 local 任务队列 spawn 作业时，都有类似的逻辑。在相应代码库里搜索 RawTask 结构就可以了解相应的逻辑。

Future 执行

如果你分别用过 tokio 的 multi-thread 运行时和 current-thread 运行时，你就会发现前者 spawn 作业后，不用 block_on 以某种形式被调用就会自动运行，而后者 spawn 出来的作业，如果你不用 block_on 驱动，那么作业永远也不会执行。

这是因为，单纯的 spawn 部分只完成了把 RawTask 塞入到任务队列中的动作，从任务队列中取出任务执行还需要另一端逻辑来完成。

显式的 block_on 肯定是一种办法，而 tokio 的 multi-thread 运行时之所以能够“自动”运行，是因为它启动了一个多线程的线程池，线程池里的线程默认的工作就是不断从任务队列里取出待运行作业执行。

其中，Launch::launch 是初始化代码的关键节点。它会逐步调用到 Spawner::spawn_task 和 Spawner::spawn_thread 方法，最终启动一个新线程执行 move || run(worker) 闭包的逻辑。

run 函数在新线程里开始执行后，会逐步执行到一个 while !core.is_shutdown 的循环。这个循环的工作就是在 Runtime 被 shutdown 前不断获取任务队列中的作业驱动运行。

RawTask 重新调度

再次以 tokio 为例，如果第一次 poll Future 返回 Pending 结果，那么 Future 需要保存传入 poll 的 Context 实例，从而在可以继续运行作业时将 RawTask 再次提交到调度器上。

前面我们写到 RawTask 的 VTable 里有 poll 和 schedule 方法，这两个方法就是为了这个目的设计的。

其中，RawTask 的 poll 是由上一节提到的 run 函数驱动的：

• src/runtime/scheduler/multi_thread/worker.rs/run

• src/runtime/scheduler/multi_thread/worker.rs/Context::run

• src/runtime/scheduler/multi_thread/worker.rs/Context::run_task

• src/runtime/task/mod.rs/LocalNotified::run

// LocalNotified::run
let raw: RawTask = self.task.raw;
raw.poll();
 
// Harness::poll
self.poll_inner()
 
// Harness::poll_inner
let header_ptr = self.header_ptr();
let waker_ref = waker_ref::(&header_ptr);
let cx = Context::from_waker(&waker_ref);
let res = poll_future(self.core(), cx);

unsafe fn wake_by_ref(ptr: *const ()) {
    let ptr = NonNull::new_unchecked(ptr as *mut Header);
    let raw = RawTask::from_raw(ptr);
    raw.wake_by_ref();
}
 
// RawTask::wake_by_ref
pub(super) fn wake_by_ref(&self) {
    self.schedule();
}

pub(crate) struct Reactor {
    pub(crate) poller: Poller,
    /// Registered sources.
    sources: Mutex>>,
    events: Mutex,
}
 
/// A registered source of I/O events.
pub(crate) struct Source {
    /// This source's registration into the reactor.
    registration: Registration,
    /// The key of this source obtained during registration.
    key: usize,
    /// Inner state with registered wakers.
    state: Mutex<[Direction; 2]>,
}
 
pub enum Registration {
    Fd(RawFd),
    Signal(Signal),
    Process(Child),
}
 
/// A read or write direction.
struct Direction {
    waker: Option,
    wakers: Slab>,
}
 
impl Direction {
    fn drain_into(&mut self, dst: &mut Vec) {
        if let Some(w) = self.waker.take() {
            dst.push(w);
        }
        for (_, opt) in self.wakers.iter_mut() {
            if let Some(w) = opt.take() {
                dst.push(w);
            }
        }
    }
}

pub(crate) fn init() {
    let _ = unparker();
}
 
fn unparker() -> &'static parking::Unparker {
    thread::Builder::new()
        .name("async-io".to_string())
        .spawn(move || main_loop(parker))
        .expect("cannot spawn async-io thread");
}
fn main_loop(parker: parking::Parker) {
    loop {
        reactor_lock.react(None).ok();
    }
}
impl ReactorLock<'_> {
    pub(crate) fn react(&mut self, timeout: Option) -> io::Result<()> {
        let mut wakers = Vec::new();
        self.events.clear();
        let res = match self.reactor.poller.wait(&mut self.events, timeout) {
            Ok(_) => {
                let sources = self.reactor.sources.lock().unwrap();
                for ev in self.events.iter() {
                    if let Some(source) = sources.get(ev.key) {
                        let mut state = source.state.lock().unwrap();
                        for &(dir, emitted) in &[(WRITE, ev.writable), (READ, ev.readable)] {
                            if emitted {
                                state[dir].tick = tick;
                                state[dir].drain_into(&mut wakers);
                            }
                        }
                    }
                }
            }
        }
        for waker in wakers {
            panic::catch_unwind(|| waker.wake()).ok();
        }
        res
    }
}

// Async
pub fn new_nonblocking(io: T) -> io::Result> {
    let registration = unsafe { Registration::new(io.as_fd()) };
    Ok(Async {
        source: Reactor::get().insert_io(registration)?,
        io: Some(io),
    })
}
 
// Source
fn poll_ready(&self, dir: usize, cx: &mut Context<'_>) -> Poll> {
    state[dir].waker = Some(cx.waker().clone());
}

pub(crate) struct ScheduledIo {
    pub(super) linked_list_pointers: UnsafeCell>,
    readiness: AtomicUsize,
    waiters: Mutex,
}
 
type WaitList = LinkedList::Target>;
#[derive(Debug, Default)]
struct Waiters {
    /// List of all current waiters.
    list: WaitList,
    /// Waker used for AsyncRead.
    reader: Option,
    /// Waker used for AsyncWrite.
    writer: Option,
}

pub(crate) fn token(&self) -> mio::Token {
    mio::Token(self as *const _ as usize)
}
 
pub(super) fn add_source(...) -> io::Result> {
    let scheduled_io = self.registrations.allocate(&mut self.synced.lock())?;
    let token = scheduled_io.token();
    self.registry.register(source, token, interest.to_mio())?;
    Ok(scheduled_io)
}

fn turn(...) {
    let events = &mut self.events;
    self.poll.poll(events, max_wait);
    for event in events.iter() {
        let token = event.token();
        let ready = Ready::from_mio(event);
        let ptr: *const ScheduledIo = token.0 as *const _;
        let io: &ScheduledIo = unsafe { &*ptr };
        io.set_readiness(Tick::Set(self.tick), |curr| curr | ready);
        io.wake(ready);
    }
}

pub(crate) fn react(&mut self, timeout: Option) -> io::Result<()> {
    let next_timer = self.reactor.process_timers(&mut wakers);
    let timeout = match (next_timer, timeout) {
        (None, None) => None,
        (Some(t), None) | (None, Some(t)) => Some(t),
        (Some(a), Some(b)) => Some(a.min(b)),
    };
    let res = match self.reactor.poller.wait(&mut self.events, timeout) { ... }
}

pub(crate) fn new_timeout(...) -> Sleep {
    let handle = scheduler::Handle::current();
    let entry = TimerEntry::new(&handle, deadline);
    let inner = Inner {};
    Sleep { inner, entry }
}

match ::std::future::IntoFuture::into_future(foo()) {
    mut __awaitee => loop {
        match unsafe { ::std::future::Future::poll(
            <::std::pin::Pin>::new_unchecked(&mut __awaitee),
            ::std::future::get_context(task_context),
        )} {
            ::std::task::Poll::Ready(result) => break result,
            ::std::task::Poll::Pending => {}
        }
        task_context = yield ();
    }
}