Add Future[A] that helps with async code

[stereobutter]

I am lately writing a lot of asynchronous code (using asyncio mostly) and while async/await is fairly straight forward and nice to work with, it hinders writing modular code with small functions and reuse of said functions. This is imho due to two issues in particular:

1. regular functions def foo(... and coroutines async def bar(... have different calling semantics. For a regular function you just do foo(...) to get your result back while for coroutines you need to await the result, i.e. use await bar(...)
2. coroutines don't compose well (without pouring await all over your code)

Let's define a couple simple functions (both regular and coroutines) to illustrate these issues:

def double(x): 
    return 2*x

async def delay(x):
    return x

async def double_async(x):
    return 2*x

Cases:

1. Composing regular functions just for reference - easy:

   double(double(1))

2. Composing regular and coroutine functions V1 - not so bad,:

   await double_async(double(1))

3. Composing regular and coroutine functions V2 - awkward with the await thrown into the middle:

   double(await double_async(1))

4. Composing two coroutine functions - awaits galore :

   await double_async(await double_async(1))

To ease this pain I propose a (monad-like) container that implements map and bind so that functions can be chained nicely

class Promise:
    
    def __init__(self, coro):
        self._coro = coro
        
    def map(self, f):
        async def coro():
            x = await self._coro
            return f(x)
        return Promise(coro())
    
    def bind(self, f):
        async def coro():
            x = await self._coro
            return await f(x)
        return Promise(coro())
    
    def __await__(self):
        return self._coro.__await__()

Usage

• Wrap a coroutine (function defined with async def) to create an instance of Promise.

  p = Promise(delay(1))
  await p  # get the result of the wrapped coroutine

• Call a regular function on an instance of Promise

  await Promise(delay(1)).map(double)

• Call a coroutine on an instance of Promise

  await Promise(delay(1)).bind(double_async)

Extended example

Since the examples above don't look too intimidating, behold:

await (
    double_async(
        double(
            await double_async(
                await delay(1)
            )
        )
    )
)

vs.:

await (
    Promise(delay(1))
    .bind(double_async)
    .map(double)
    .bind(double_async)
)

Feedback greatly appreciated!

[stereobutter]

A perhaps more interesting example:

await (
    Promise(delay([1,2,3]))
    .map(partial(map, double))   # double each entry
    .map(partial(map, double_async))   # double each entry with a coroutine
    .bind(star_gather)  # gather the result with a coroutine
)
# ==> [4, 8, 12]

where star_gather is simply

async def star_gather(list_of_coros):
    return await asyncio.gather(*list_of_coros)

[sobolevn]

Hi, @SaschaSchlemmer! Thanks a lot!

This is indeed on our list: #235 I will close that issue and keep this one to discuss this topic.

The problem with asyncio is that I don't use it at all. And I am not sure how to design an API for it. I would be happy if you can help me with that.

P.S. Considering partial, see #267

[sobolevn]

Inspirations:

• https://zio.dev/docs/overview/overview_basic_concurrency

And:

• https://github.com/gcanti/fp-ts/blob/master/docs/modules/Task.ts.md
• https://github.com/gcanti/fp-ts/blob/master/docs/modules/TaskEither.ts.md
• https://github.com/gcanti/fp-ts/blob/master/docs/modules/ReaderTask.ts.md
• https://github.com/gcanti/fp-ts/blob/master/docs/modules/ReaderTaskEither.ts.md
• https://github.com/gcanti/fp-ts/blob/master/docs/modules/TaskThese.ts.md

[stereobutter]

actually the above implementation is not tied to asyncio. It just uses python 3.6 async/await which should work with any library that supports this (trio for instance)

[sobolevn]

We also would probably need things like:

• TaskResult
• RequiresContextTaskResult

I consider all tasks to be IO bound. Is it a good idea?

[stereobutter]

I'd not consider everything IO, and I'd also avoid the name Task for something like demonstrated above; the reason being that this proposal is not about tasks (order of execution, handling failures, parallel/concurrently running tasks) and just about adding functional syntactic sugar around async and await independent of any specific library for dealing with scheduling (like asyncio, trio, twisted, etc. )

[sobolevn]

Fair enough about Task.

What's your opinion about not being IO? Are there any valid cases to create a coroutine that does not do any IO?

[stereobutter]

Yes, there are definitely use cases where coroutines are not just used for IO; in particular there is concurrent.futures.ProcessPoolExecutor which is used to run synchronous code in parallel (true parallelism with multiple processes).

[stereobutter]

I've got a couple of examples that help to better understand what this whole synchronous vs asynchronous/concurrent vs parallel business is all about (and how this would look like using Promise). For this I modified my simple functions from above as follows:

def double(x: int) -> int:
    time.sleep(1)
    return 2*x
    
async def double_async(x: int) -> int:
    await asyncio.sleep(1)
    return 2*x

---

Test cases:

1. Running computations sequentially (aka. synchronous code)

   tic = time.perf_counter()
   await (
       Promise(coro([1,2,3]))
       .map(partial(map, double))
       .map(list)
   )
   toc = time.perf_counter()
   
   print(toc-tic)  # 3 seconds

2. Running computations concurrently (aka. asynchronous code)

   tic = time.perf_counter()
   await (
       Promise(coro([1,2,3]))
       .map(partial(map, double_async))
       .bind(star_gather)
       .map(list)
   )
   toc = time.perf_counter()
   
   print(toc-tic)  # 1 second

3. Running computations in parallel (aka. multiprocessing)

   tic = time.perf_counter()
   with concurrent.futures.ProcessPoolExecutor() as pool:
       await (
           Promise(coro([1,2,3]))
          .map(partial(pool.map, double))
          .map(list)
   )
   toc = time.perf_counter()
   
   print(toc-tic)  # 1 second  (on any computer with more than 3 CPU cores)

[stereobutter]

Notice that I added another usage example at the bottom of my original post :)

[stereobutter]

Also it appears typing with mypy works out of the box (and without requiring a plugin):

from __future__ import annotations
from typing import Generic, TypeVar, Callable, Awaitable

A = TypeVar('A')
B = TypeVar('B')


class Continuation(Generic[A]):

    def __init__(self, coro: Awaitable[A]):
        self._coro = coro

    @classmethod
    def unit(cls, value: A) -> Continuation[A]:
        async def coro() -> A:
            return value
        return cls(coro())

    def map(self, f: Callable[[A], B]) -> Continuation[B]:
        async def coro() -> B:
            x = await self._coro
            return f(x)
        return Continuation(coro())

    def bind(self, f: Callable[[A], Awaitable[B]]) -> Continuation[B]:
        async def coro() -> B:
            x = await self._coro
            return await f(x)
        return Continuation(coro())

    def __await__(self):
        return self._coro.__await__()

After some googl'ing around I came to the conclusion that Promise also isn't a good name because it is used in javascript for a similar but different concept. In haskell there is Cont which is (as far as I understood) exactly what I am proposing here.

[sobolevn]

Yes, I like Continuation the most! Will you send a PR for this?

And we can start a detailed discussion from there.

[stereobutter]

Over the weekend I will probably have some spare time to work on this :) I also played around with lists of continuations (I implemented another monad ListContinuation that works over multiple continuations) because running multiple things at the same time is usually what async code is all about. I don't know what your take is on more specialized monads, monad transfomers and extensible effects for nesting monads inside one another.

[sobolevn]

Related https://github.com/correl/typesafe-monads/blob/master/monads/future.py

[stereobutter]

Sry for not responding for a while, been very busy; seems to me the code you linked does basically the same and adds some features my proposal didn't have as well.

[sobolevn]

I have created an issue in the typesafe-monads repo. Maybe we can collaborate on this one?

[stereobutter]

That sounds like a good idea

[sobolevn]

I also like the idea that Future[A] contains IO[A] inside as a value. And you cannot construct pure Future instance. That's how it is in fp-ts: https://gcanti.github.io/fp-ts/modules/Task.ts.html

And that's how Python really works. You cannot just can a coroutine. You have to use an executor like asyncio.run or do some magic. It is not pure.

I also like this idea, because it will simplify the implementation. No need for separate Future vs FutureIO vs FutureResult vs FutureIOResult. There would be only two types: FutureResult and Future for computations that can and cannot fail.

[sobolevn]

Ok, maybe Future is not the best name https://docs.python.org/3/library/asyncio-future.html#asyncio.Future

[sobolevn]

Really interesting project https://github.com/h2non/paco

[sobolevn]

We would also need to rework:

• @safe decorator, related safe decorator generates incorrent signature #333
• @maybe decorator
• @impure decorator

[sobolevn]

After playing around asyncio and Future a bit, I would say that a correct way to handle this would be:

1. Drop all async functions we have
2. Forbid to use async keyword in wemake-python-styleguide and ban import asyncio
3. Close this issue
4. Burn everything with fire! 🔥

[sobolevn]

Related #355

[sobolevn]

These methods are left to finish RequiresContextFutureResult:

• def bind_future(self, function): ...
• def bind_future_result(self, function): ...
• def bind_async_future(self, function): ...
• def bind_async_future_result(self, function): ...
• def bind_context_ioresult(self, function): ...
• def bind_async(self, function): ...
• def bind_awaitable(self, function): ...
• We also need to provide ContextFutureResult with .ask method

[sobolevn]

Unit methods:

• .from_io (both to ReaderIOResult and ReaderFutureResult)
• .from_failed_io (both to ReaderIOResult and ReaderFutureResult)
• .from_future
• .from_failed_future
• .from_future_result
• .from_ioresult_context

[sobolevn]

Pointfree:

• bind_async_future
• bind_async_future_result