nimforum mirror - Zig-style async/await

mratsim (orginal) [2021-01-07T10:28:43+01:00] view original

Async/await is an abstraction over resumable functions + storing them in a scheduler.

Closure iterators or disruptek CPS solves the resumable function part. Asyncdispatch and Chronos add a scheduler on top, they use epoll and select because async/await is mainly used for IO and the OS already have scheduling facilities for them.

If you want to participate in the future design of async in Nim feel free to read:

https://github.com/nim-lang/RFCs/issues/295 (next steps for CPS)

https://github.com/nim-lang/RFCs/issues/304 (completing the Nim async/await implementation)

Disruptek, Zevv and me are also researching, implementing and designing the ergonomics of resumable functions (not async/await) and replace/reimplement closure iterators on top. This would allow to not have that much transformation logic in Asyncdispatch and Chronos, and also hopefully significantly reduce overhead and improve composition.

So with my current design, here is how await would be translated and integrated with 2 schedulers (one based on epoll for IO and one based on threadpool):

type Awaitable[T] = concept aw
  ## An awaitable concept, for example a Future or Flowvar.
  ## Implementation left at the scheduler discretion.
  aw.get is T
  aw.isReady is bool

var ioScheduler{.threadvar.}: IOScheduler
var computeScheduler{.threadvar.}: CPUScheduler
var isRootThread{.threadvar.}: bool

proc saveContinuationAndSuspend() {.suspend.} =
  ioScheduler.enqueue bindCallerContinuation()

proc await[T](e: Awaitable[T]): T {.resumable.} =
  if not e.isReady():
    saveContinuationAndSuspend()
  return e.get()

proc continueOnThreadpool(pool: CPUScheduler) {.suspend.} =
  pool.spawn bindCallerContinuation()

proc serve(socket: Socket) =
  while true:
    let conn = await socket.connect()
    if isRootThread:
      suspendWith pool.continueOnThreadpool()
    # --- the rest is on another thread on the CPU threadpool.
    #     and the root thread can handle IO again.
    # Stream processing
    var finished = false
    while not finished:
      let size = await conn.hasData()
      var chunk = newSeq[byte](size)
      conn.read(chunk) # non-blocking read
      finished = process(chunk)
    # Now that thread still owns the socket,
    # we can return it to the main thread via channel
    # or keep its ownership.

Here is my WIP proposal: https://github.com/weavers-guild/weave-io/blob/master/design/design_2_continuations.md

Mirror of forum.nim-lang.org

7347 :: Zig-style async/await