Discussion: async

[roee30]

Out of curiosity, I tried incorporating coroutine handling into Runtime so that async code could be run, basically implementing an Async ability. This is a very partial POC.

patch for runtime.py:

2c2,4
< 
---
> import types
> import asyncio
> import threading
35c37,39
< 
---
>         object.__setattr__(self, "loop", asyncio.new_event_loop())
>         self.loop._thread_id = threading.get_ident()
>         asyncio.events._set_running_loop(self.loop)
134a135
>             from inspect import isawaitable
139a141,148
>                         case co if isawaitable(co):
>                             if isinstance(co, types.CoroutineType):
>                                 task = self.loop.create_task(co)
>                             else:
>                                 task = co
>                             while not task.done():
>                                 self.loop._run_once()
>                             effect.send(task.result())

Makes this work:

import asyncio
from dataclasses import dataclass
from typing import Never, Awaitable, Iterable

import httpx
from stateless import Effect, Runtime, Depend, depend, catch, throws

SYNC = False

async def gather[R](cos: Iterable[Awaitable[R]]) -> list[R]:
    if SYNC:
        return [await c for c in cos]
    return await asyncio.gather(*cos)


@dataclass(frozen=True)
class HTTP:
    client: httpx.AsyncClient

    def get(self, url, *args, **kwargs) -> Awaitable[httpx.Response]:
        return self.client.get(url, *args, **kwargs)


def http_client() -> Depend[Never, HTTP]:
    client = yield httpx.AsyncClient().__aenter__()
    return HTTP(client)


def get_concurrently():
    http = yield from depend(HTTP)
    result = yield gather(http.get(f"https://{x}.com") for x in ['google', 'reddit', 'xkcd'])
    return result


runtime = Runtime().use_effect(http_client())
e = get_concurrently()
runtime.run(e)

Changing SYNC to True doubles the runtime, which verifies that the coroutines originally run concurrently.

Problems:

• doesn't propagate errors from HTTP.get() to the caller
• yielding coroutines, which are not abilities, mess with static typing
• HTTP.get() invocations can't be encapsulated in effects (functions) because gather() can't deal with plain generators. Solving this probably requires implementing a special flavor of gather() that drives each generator until it yields a coroutine, collects them, and calls gather(), repeating until all generators are exhausted

[suned]

Hello! Thanks a lot for putting effort towards this problem. Strong integration with asyncio is a very high priority for me in this project, and In fact I would like a long term solution to asyncio integration to eventually replace the existing Parallel ability.

I have experimented with a solution similar to yours, although mine was just:

import asyncio

class AsyncIO:
    def run(a: Awaitable[A]) -> A:
        return asyncio.run(a)


def needs_asyncio() -> Depend[Asyncio, ...]:
    asyncio = yield from depend(AsyncIO)
    result = asyncio.run(gather(...))
    return result

Which from what I understand functionally achieves the same thing as your solution? Correct me if I'm wrong 😅

This allows you to use async functions in effects.

But for "strong integration with asyncio", I think ideally it should also be possible to use effects in async functions. I have not come up with a good solution to this.

A long term solution I have been considering is to contribute to the effort to support yield from in async functions (which also requires supporting return values from in async generators for which there is a draft PR that may need to be revived). This will likely be a long journey though, since it will require at least one, maybe two, accepted PEPs.

But it would allow us to do something like:

AsyncEffect: TypeAlias = AsyncGenerator[Type[A] | E, Any, R]


async def async_effect() -> AsyncEffect[HTTP, HTTPError, ...]:
    http = yield from depend(HTTP)
    response = await (yield from http.get(...))
    ...

async def use_async_effect() -> AsyncEffect[HTTP, HTTPError, ...]:
    result = await (yield from async_effect())
    ...

(maybe the API will/should not be exactly this but I hope you get my point)

I'm very open to suggestions and ideas. I consider this high priority, but also quite difficult to solve well.

[roee30]

I do realize now that our solutions are functionally identical. However, mine is one step closer to integrating an event loop in Runtime itself. This is required to run multiple effects concurrently, like async functions.

There's no need to integrate effects into async functions because conceptually, async is just another effect. I think that the solution for gather() I described above actually allows full integration of async into this library. Generators, i.e., yielding values one by one, can also be considered an effect and implemented by special handling in Runtime. You don't actually need async yield from, and if you implement this you're actually several steps ahead of the standard language. And I don't think it's even that difficult to get a working POC for this. Supporting the entire async API might prove a challenge, and I hope it doesn't require reimplementing everything. There's only one way to find out...

If it's possible to patch any existing event loop (not necessarily asyncio) to have an API where the event loop stops running once any of a predetermined set of coroutines is completed, you can reuse all the event loop code and just call it whenever an effect yields a coroutine (EDIT: it's asyncio.wait(..., return_when=asyncio.FIRST_COMPLETED))

[suned]

There's no need to integrate effects into async functions because conceptually, async is just another effect.

From a purely technical perspective, I agree. There is nothing preventing implementing async as an effect that could run stateless.Effect or asyncio.Awaitable concurrently in an event loop during effect interpretation.

My motivation for supporting two-way integration with asyncio is integration with the python async eco system at large. If you are already using asyncio, you are more likely to start using stateless if we can play nice with the code you already have, and you don't need to re-write everything to Effect. But maybe I'm misunderstanding what you are saying?

But maybe having direct support for awaitables in "effect" functions would be a good first step. At least I'd be very interested in pursuing this more, I'm not very well versed with the internals of asyncio and event loops however, so this would be a long journey for me as well 😅

[roee30]

With full event loop integration, I think you could rewrite any async function anywhere as an effect, one at a time, without converting the entire call stack, which is almost as good:

• effects can call async functions with yield and get an appropriate response from Runtime
• async functions can call effects with Runtime.run()

That way, you don't have to wait for language changes.

Replacing your asyncio.run() with loop.run_until_complete() will allow cooperating with an already running event loop. In my understanding, this (really, either of these) runs the entire event loop, not just the specific coroutine, which means you don't compromise concurrency. The only challenges I see:

• Running effects concurrently. Primitively, just drive each effect until an awaitable is yielded, then call loop.run_until_complete(). More sophisticatedly, drive an effect, get an awaitable, and move to the next effect, collecting all awaitables, then call asyncio.wait() with asyncio.FIRST_COMPLETED. This is actually similar to what the event loop does. When async yield from happens, you could dispose of Runtime and just add a few abilities-handling lines to an existing event loop.
• as mentioned, static typing

[suned]

Thanks for this, I think the solution you have roughly sketched here has a lot of potential. Also, it inspired me to perhaps solve for my dream of "two way integration between with asyncio and stateless.Effect" by just adding an async run function. With your solution in mind this would mean two run functions:

class Runtime(Generic[A]):
    def run(self, effect[A, Exception, R]) -> R:
        """
        Run effects by blocking,  running asyncio coroutines yielded by effects with asyncio "somehow"
        (probably with `loop.run_until_complete)
        """

    async def run_async(self, Effect[A, Exception, R]) -> R:
        """
        Run effects async, simply awaiting asyncio coroutines yielded by effects
        """
         ...
         match effect_or_ability with:
             ...
             case co if isawaitable(co):
                 result = await co
                 effect.send(result)
         ...

I think that basically provides all the "two way" integration between asyncio and stateless that you would need.

[suned]

I was building on your suggestion, and came up with the following solution (admittedly a major change to the current api)

import asyncio
from dataclasses import dataclass
from typing import (
    Any,
    Callable,
    Coroutine,
    Generator,
    Generic,
    Never,
    Self,
    TypeVar,
    cast,
    overload,
)
from typing import Type as Type_

from stateless.errors import MissingAbilityError
from typing_extensions import ParamSpec

T_co = TypeVar('T_co', covariant=True)

E = TypeVar('E', bound=Exception)
A = TypeVar('A')
A2 = TypeVar('A2')
R = TypeVar('R')
P = ParamSpec('P')

# Effect now no longer requires abilities to be types, they
# can be anything. This is useful for our
# goal of sending awaitables to the main
# interpreter loop, since we can now
# yield awaitables for example.
Effect = Generator[A | E, Any, R]
Depend = Generator[A, Any, R]
Try = Generator[E, Any, R]
Success = Generator[Never, Any, R]

@dataclass(frozen=True)
class Handler(Generic[A]):

    @overload
    def handle(self, f: Callable[P, Depend[A, R]]) -> Callable[P, Success[R]]:
        ...

    @overload
    def handle(self, f: Callable[P, Depend[A2 | A, R]]) -> Callable[P, Depend[A2, R]]:
        ...

    @overload
    def handle(self, f: Callable[P, Effect[A, E, R]]) -> Callable[P, Try[E, R]]:
        ...

    @overload
    def handle(self, f: Callable[P, Effect[A2 | A, E, R]]) -> Callable[P, Effect[A2, E, R]]:
        ...

    def handle(self, f: Callable[P, Effect[A, E, R] | Effect[A | A2, E, R]]) -> Callable[P, Effect[Never, E, R] | Effect[A2, E, R]]:  # type: ignore
        raise NotImplementedError()

    def __and__(self, handler: "Handler[A2]") -> "Handler[A | A2]":
        raise NotImplementedError()

class Ability(Generic[T_co]):
    def __iter__(self: Self) -> Generator[Self, T_co, T_co]:
        try:
            v = yield self
        except MissingAbilityError:
            raise MissingAbilityError(self) from None
        return v

@dataclass(frozen=True)
class Need(Ability[T_co]):
    t: Type_[T_co]


@dataclass(frozen=True)
class Async(Ability[Any]):
    awaitable: Coroutine[Any, Any, Any]


T2 = TypeVar('T2')

def need(t: Type_[T2]) -> Depend[Need[T2], T2]:
    v = yield from Need(t)
    return v


def wait(a: Coroutine[Any, Any, R]) -> Depend[Async, R]:
    v = yield from Async(a)
    # We dont want `Async` to be generic since we don't
    # want to specify handlers for e.g `Async[int]` and `Async[str]`
    # separately. They should be handled by the same handler.
    # Unfortunately that breaks the pattern with `Ability`
    # so `v` here is `Any`
    return v


# `run` requires that all
# abilities are handled.
def run(effect: Try[Exception, R]) -> R:
    raise NotImplementedError()


# `run_async` does not need to have the `Async` ability handled
# since it "translates" this ability into asyncio compatible
# coroutines.
async def run_async(effect: Effect[Async, Exception, R]) -> R:
    raise NotImplementedError()


async def j() -> int:
    raise NotImplementedError()


def example() -> Depend[Need[str] | Need[int] | Async, tuple[str, int, int]]:
    s = yield from need(str)
    i = yield from need(int)
    v = yield from wait(j())
    return s, i, v



def handle_need(v: T2) -> Handler[Need[T2]]:
    raise NotImplementedError()


def handle_async() -> Handler[Async]:
    raise NotImplementedError()


handler = handle_need('') & handle_need(0) & handle_async()
e = handler.handle(example)()
v = asyncio.run(run_async(e))

It builds on my recent changes that enables partially providing abilities,Handler here works similarly to Abilities in that PR, so you can check out the readme there for some guidance also.

Obviously, just the types and overall API stubs, but I think the approach is sound (of course problems will appear when I start on the actual implementation).

This is probably a bit much to take in, but if you are up for it I would like to hear your thoughts.

[roee30]

It looks pretty good, although I can't tell if you'll run into type-checking issues. My only comment is that I don't quite understand why you need run_async(). run() already cooperates with the event loop, so they boil down to the same thing, unnecessarily duplicating the API (unless I'm missing something).

[suned]

although I can't tell if you'll run into type-checking issues.

I've started with the implementation, so far so good. The only thing I've run into is that I cant specify an abstract method on Handler to enforce types for handling abilities because it would require higher kinded types. But I don't think its a huge problem.

I don't quite understand why you need run_async(). run() already cooperates with the event loop, so they boil down to the same thing,

If we run coroutines like in your implementation (or some variation thereof) ie:

if isinstance(co, types.CoroutineType):
    task = self.loop.create_task(co)
else:
    task = co
while not task.done():
    self.loop._run_once()
effect.send(task.result())

My understanding is that this is blocking. So it would not be good to do this in an async function.

Lets say you have a fastapi application for example and you did.

from stateless import run

from fastapi import App

app = App()

@app.post(...)
async def some_operation(...) -> Response:
      # blocking, not good
      value = run(some_effect())
      return Response(value)

I'm thinking it would be better to do:

from stateless import run_async

@app.post(...)
async def some_operation(...) -> Response:
      value = await run_async(some_effect())
      return Response(value)

Albeit I'm not very clear on whether calling loop._run_once() inside an async function would be considered blocking.

[roee30]

Never mind my original implementation, it's not good. Call async.wait(... FIRST_COMPLETED) in the runner with all currently relevant awaitables from all running effects. However, neither implementation blocks any tasks that belong to the loop you're running, since the entire loop is executing.

In more detail: since await is equivalent to yield from, which doesn't actually yield but instead delegates yielding to another function, the next yield point in either case is the first yield point of some_effect() (actual yield statement). It is at this point that control is passed ("yielded") to your runner, which then has the responsibility to cooperate with the event loop. If you then choose to relinquish all control to the event loop (e.g., with await), no one will be driving your effects. Rather, the event loop must be given control only in a limited context, i.e., asyncio.wait(), and not completely, as part of the runner's job of interleaving effects and coroutines.

In less detail: await run_async() and run() are the same in this context.