Trio documents some of these things in detail: Trio’s core functionality — Trio 0.34.0 documentation
Re exception groups: that sounds great, and thanks for the helpful clarifications!
You can build a slightly less ergonomic version of async for in library code as well, which is what Rust’s streams do.
Hi Steffi, thanks for the detailed response. However, a few things are still not quite clear to me.
- I’m still not sure how suspension happens. Does the current implementation suspend at every
awaitpoint to allow the runtime to schedule the next coroutine with the parent set in the completion handler? - There seems to be a lot of indirection in the current design, which I think may be related to the first bullet point. From the code, it seems the resume functions synthesised for the coroutine state machine are responsible for scheduling the next coroutine in the chain. Can we not just call the next coroutine’s resume function directly instead? That way, there is no reason to make a completion token; control simply returns to the parent resume function when the child returns. This is what Python does, and from doing some searching, it is what Rust does as well.
awaitchains are normal synchronous function calls that do not involve the runtime.
As I mentioned on Discord, I don’t think the coroutine state machine should know or care about scheduling or external cleanups; I think it should only know about starting, pausing, resuming, and finishing. Here is a sample pseudocode of what I envision a state machine may look like:
struct YCoroutineState:
var state: Int8
# locals live across awaits would go here, e.g.:
# var n: Int
# var a: Int
struct XCoroutineState:
var state: Int8
var child: Optional[YCoroutineState]
# locals live across awaits would go here, e.g.:
# var n: Int
# var r: Int
# var s: Int
struct CoroutineResult:
var completed: Bool
var result: OpaquePtr # typed per-coro in a real lowering
def y_coroutine_state_machine(mut handle: YCoroutineState) -> CoroutineResult:
# Start coroutine
if handle.state == 0:
# assume some initial code already ran
# First suspension point (leaf suspend — no child)
handle.state = 1
return CoroutineResult(completed=False, result=None)
# Coroutine was paused
elif handle.state == 1:
# code after suspend, then finish
handle.state = -1
return CoroutineResult(completed=True, result=...) # e.g. a*2
return CoroutineResult(completed=False, result=None)
def x_coroutine_state_machine(mut handle: XCoroutineState) -> CoroutineResult:
# Start coroutine
if handle.state == 0:
# assume some initial code already ran
# First suspension point: r = await y(...)
handle.child = YCoroutineState(state=0)
var r = y_coroutine_state_machine(handle.child) # direct sync call
if not r.completed:
handle.state = 1
return CoroutineResult(completed=False, result=None)
# child finished on first poll — fall through past the await
# handle.r = r.result
handle.child = None
# code after await ran here, e.g. s = r + 10
# Second suspension point
handle.state = 2
return CoroutineResult(completed=False, result=None)
# Coroutine was paused while waiting on y
elif handle.state == 1:
if handle.child is not None:
var child_result = y_coroutine_state_machine(handle.child) # direct sync call
if not child_result.completed:
# still waiting — state stays 1
return CoroutineResult(completed=False, result=None)
# child done — take value, continue past the await
# handle.r = child_result.result
handle.child = None
# code after await, e.g. s = r + 10
# Second suspension point
handle.state = 2
return CoroutineResult(completed=False, result=None)
# Coroutine was paused at second suspend
elif handle.state == 2:
handle.state = -1
return CoroutineResult(completed=True, result=...) # e.g. s
return CoroutineResult(completed=False, result=None)
To allow handing back control to the caller, we need a language mechanism that pauses execution on the leaf node. In Python, this is done by having a yield statement in an awaitable. In Rust it is done by eventually returning Poll.Pending
(forwarding my message from the issue)
I’ve been building bouclette, a I/O library with io_uring and epoll backends, and a few things from that work seem worth raising early.
On Linux the fast path is completion-based and usually thread-per-core. The kernel owns the buffer while an operation is in flight, so cancellation must either wait for the kernel’s completion, or the buffer must be moved into the operation and handed back with the result (what monoio, compio and bouclette do). It would help if “safe to cancel” said which one the runtime guarantees.
Since coroutine handles must be destroyed exactly once and coroutines can fail: in Mojo 1.0/1.1 a linear value can’t stay alive across a raises call, so a linear future held across a raising run() doesn’t compile, and we had to make ours droppable. Async will need either a destructor for error paths or driving verbs that never raise.
Origins don’t fully cover this either. A handle into loop-owned state that the kernel also addresses has to hold a raw address, so nothing stops it outliving the loop, and even a tracked handle doesn’t prevent mut access to the loop while it’s alive. Some structured-scope primitive seems necessary.
And finally, I’d love the executor interface to include a hook for the scheduler to poll I/O sources (the “subsystems” idea from #3945) in the prototype rather than after stabilization, since it decides whether a library like ours can sit underneath the default runtime!
This can be handled by finally blocks.
+1 to the need for extra subsystems, but I may be biased.
Re: “Does the current implementation suspend at every await point to allow the runtime to schedule the next coroutine with the parent set in the completion handler?” The implementation of await is in the library. Scheduling is not compiler synthesized code. The coroutine stores a resume function pointer and a callback. The code that calls the resume function is in the library and the code that sets the callback function is also in the library. The function pointer representation of the resume function is subject to change though.
Thanks for writing up this design doc Steffi. I appreciate Modular giving the community some insight into where Mojo’s design is heading.
I am largely happy with the design goals that you’ve presented. I am excited to see how it shapes up.
You mention this open design question:
[How to express] APIs that are generic over sync and async.
IMO this will be a major question that needs answering after the async MVP lands. I hope we can figure out the right solution to “async generics” before the standard library (and broader ecosystem) stablises its async APIs.
I don’t have a polished solution in mind, but Owen and I have both sketched out some ideas about how we might resolve this. (e.g. Owen presented his sync ↔ async idea earlier in this thread.) When the time is right we should probably put together a detailed motivation for such a feature, including tons of code snippets comparing non-generic vs generic APIs.
Does this leave the door open for typestate-based coroutines? Ideally, I’d like to avoid having the function pointer for “simple” async constructs where it’s acting as more of a state machine abstraction. Especially for network infrastructure code, it’s very reasonable to have a limited number of states such that every coroutine in each of a group of lists or other lookup-optimized data structures is in the same state. Forcing the function pointers in there means adding the pointer in a case where the there’s no need for it.
from std.ffi.unsafe_union import UnsafeUnion
@fieldwise_init
struct States(Equatable, Hashable, ImplicitlyCopyable, Movable):
"""
a -> b;
b -> c;
c -> c;
c -> e;
b -> d;
d -> e;
"""
var _value: UInt8
comptime a = Self(0)
comptime b = Self(1)
comptime c = Self(2)
comptime d = Self(3)
comptime e = Self(4)
@fieldwise_init
struct IAmNotImportant(
Defaultable, Equatable, Hashable, ImplicitlyCopyable, Movable
):
var _value: UInt8
struct TypestateBasedCoroutine[state: States]:
comptime WithState[state: States] = TypestateBasedCoroutine[state]
var state: IAmNotImportant
comptime next_state_function = Self.next_state_a if Self.state == States.a else Self.next_state_b if Self.state == States.b else Self.next_state_c if Self.state == States.c else Self.next_state_d
def __init__(
var self, var state: IAmNotImportant
) where Self.state == States.a:
self.state = state
def next_state_a(var self) -> Self.WithState[States.b]:
...
def next_state_b(
var self,
) -> Either[Self.WithState[States.c], Self.WithState[States.e]]:
...
def next_state_c(
var self,
) -> Either[Self.WithState[States.c], Self.WithState[States.e]]:
...
def next_state_d(var self) -> Self.WithState[States.e]:
...
def __deinit__(var self) where Self.state == States.e:
...
struct TypeErasedCoroutine(Equatable, Hashable, ImplicitlyCopyable, Movable):
var state: States
var data: UnsafeUnion[
TypestateBasedCoroutine[States.a],
TypestateBasedCoroutine[States.b],
TypestateBasedCoroutine[States.c],
TypestateBasedCoroutine[States.d],
TypestateBasedCoroutine[States.e],
]
def __init__(var self, var state: States, var data: IAmNotImportant):
self.state = state
self.data = {TypestateBasedCoroutine[States.a](state=data)}
def next_state(mut self) -> Bool:
"""Returns true if the coroutine has reached its final state, false otherwise."""
if self.state == States.a:
var coroutine = self.data.get(TypestateBasedCoroutine[States.a])
var next_coroutine = coroutine.next_state_a()
self.state = States.b
self.data = {next_coroutine}
elif self.state == States.b:
var coroutine = self.data.get(TypestateBasedCoroutine[States.b])
var next_coroutine = coroutine.next_state_b()
if next_coroutine.is_left():
self.state = States.c
self.data = {next_coroutine.left()}
else:
self.state = States.e
self.data = {next_coroutine.right()}
elif self.state == States.c:
var coroutine = self.data.get(TypestateBasedCoroutine[States.c])
var next_coroutine = coroutine.next_state_c()
if next_coroutine.is_left():
self.state = States.c
self.data = {next_coroutine.left()}
else:
self.state = States.e
self.data = {next_coroutine.right()}
return True
elif self.state == States.d:
var coroutine = self.data.get(TypestateBasedCoroutine[States.d])
var next_coroutine = coroutine.next_state_d()
self.state = States.e
self.data = {next_coroutine}
return True
else:
panic("Cannot transition from state e")
return False
def __deinit__(var self):
...
The main benefit of this as the “base representation” is that library code can build the common representations (jump tables, callback function pointer, etc) from this with no loss in performance, which opens up experimentation in library land without extra compiler work.
I think this might need to be something that is considered as part of coroutines, because the two options I can think of for making “async generic” code at a syntax level are either to force all code to be “async correct” or to make it so that the generic dialect is less powerful than the async dialect and is forced to implicitly await anything with an __await__ when it is created. Zig has an interface which gets rid of function coloring, but it doesn’t appear to really allow you to write code which is fully generic over whether or not you’re doing async IO without duplicating most function call sites.