Show HN: Parallelise pipelines of Python async iterables
github.com
github.com
I did my fair share of tornado/twisted/zeromq ioloop programming, and even though I completely agree that it gets messy real quick, I always "got it".
With async/await though, I always struggle and need a really long time to get a clear picture of what is happening.
Maybe my mental model of how it works is not optimal so I would love for people to describe how they "read" async/await code.
I typically see await very much like I would read a "yield", except that instead of yielding a value, it just yields a promise to the ioloop that will "next()" once it resolves.
Still, it seems every language is adopting it, so I guess à majority of people find it simpler to reason with, I just don't :(
Note: the same applies for async/await in JS. I can easily get my head around the "then" style, but struggle with async/await. A bit less than in python though.
async def foo():
conn = await DB.connect()
results = await conn.query("SELECT bar, baz;")
restresult = await REST.get("somewhere.else.prod.example.net/v1/BAR", id=results.id)
return restresult.xyz
class foo():
future = None
conn = None
results = None
restresult = None
def call(self):
self.future = Future()
future = DB.connect()
future.add_done_callback(self._db_conn)
return self.future
def _db_conn(self, fut):
try:
self.conn = fut.get_result()
except:
self.future.set_exception()
return
future = self.conn.query("SELECT bar, baz;")
future.add_done_callback(self._results)
def _results(self, fut):
try:
self.results = fut.get_result()
except:
self.future.set_exception()
return
future = REST.get("somewhere.else.prod.example.net/v1/BAR", id=self.results.id)
future.add_done_callback(self._restresult)
def _restresult(self, fut):
try:
self.restresult = fut.get_result()
except:
self.future.set_exception()
return
self.future.set_result(self.restresult.xyz)Async re-merges functions that have been shattered due to having to block.
Maybe another helpful way to see it is that each sync function is one atomic block that is executed in one go. On the other hand, see each « await » in an async function as a split of the function into atomic blocks that can be executed separately, allowing to do other things while waiting between each block.
This abstraction catches up because it works really well. It is very composable as well. For example adding a queue instead of naively processing a lot of promises at once is 2 lines of code with this https://www.npmjs.com//promise-queue
Now I am waiting for async generators to catch up as well, to replace streams... I recently worked with this package for example and it was so much better than plain streams https://www.npmjs.com/package/streaming-iterables
I still don't understand promises in JavaScript. I got my head around the convoluted constructor (a function whose argument is a function whose arguments are two functions), but I can't make sense of chaining and the "promise resolution procedure" [0].
However I've looked into async-await in Python and it seems straightforward. Doesn't `await foo()` just mean "suspend execution here and don't run any more code in this function until foo is done"?
[0] https://promisesaplus.com/#the-promise-resolution-procedure
And yes, forget about .then and other old school stuff like that in JS. Just use async await to deal with promises.
Async function calls work like "normal" until they hit a yield statement. When they do, they yell back up the callstack "hey, I'm going to be waiting for a while, let something else happen while I am".
When that happens, one of two things can happen. 1. Either there's nothing else to do, in which case we wait until something that has yielded to no longer be waiting. 2. Something else is done waiting and it's waiting for its turn. In which case, it's allowed to continue at this point until it either returns or yields again.
In this sense, instead of yielding a value, the function is yielding the stack to something else that can use it while it's waiting.
Async is great for when you're waiting on external processes to complete (database calls, API calls, etc). It's terrible for any CPU heavy tasks because, at least in the Python world, it can still only use a single CPU. In this way, again, specific to Python, it's use-case is much closer to multithreading than multiprocessing.
Another way to think about the machinery behind the scenes is it's a queue of callstacks. If an async call yields the primary stack, its callstack is thrown onto the queue. The queue of callstacks is then processed in order.
Hope that helps! Took me a long while to wrap my head around it and I still struggle reading it some of the time.
From a mental model perspective, I would recommend Communicating Sequential Processes - https://en.m.wikipedia.org/wiki/Communicating_sequential_pro... - and see Golang's channels and goroutines as a pretty good surface. Clojure also has this surface in the core.async library.
CSP is about asynchrony in the large- decoupled concurrent activities, using data to coordinate, that more closely reflect mental models of how the real world operates.
Asynchrony in the small, at the line-of-code/function-spec level, is terrible. Its exposure completely defeats Python's very concrete, literal pseudocode-as-code origin story. And its use prevents the machine from doing what it does well-scheduling and efficient execution.
You are not the only one. It is a shame that these tools have been added to the toolbox, and the conceptual spaghetti they lead to today is tomorrow's technical debt.
Cheers.
The first and most important is that Python made its bones with an enforced single thread execution runtime- with the GIL- that dramatically simplified the mental machinery a programmer needed to do productive things with computers. Extremely successful approach. Async/await, as keywords- core parts of the language- are a complete conceptual departure. Like Python was a very safe multipurpose Swiss Army knife, that now as part of the core has a powersaw. For people who need the powersaw capability, with the ergonomics of a simple blade, Python has always had libraries and module extensibility, and the brilliance of the dunder methods that give access to the machinery for experts who need it. Clear distinction between 95% and 5% use case tooling.
Second, as concurrency-enabling abstractions, async/await are poor because they are an opinion that what is hard about concurrency is code. This is a bad opinion. It is data that is hard.
When you are dealing with things happening in parallel, trying to understand them, you realize that the understanding part is only hard when there are dependencies between the parallel things.
In computing terms, those dependencies are expressed with data relationships. One process waits for another only because it needs some data from it, or it needs it to be ready to receive some data. Things that don't share data, in one form or another, do not care about each other.
Queues- a data structure- help model data dependencies explicitly. Async/await only model them implicitly, and poorly, at that. For every await, there is a use case for a queue that much more clearly and composably expresses the true dependency between the potentially concurrent operations.
Async/await- as conceptual flags for code blocks- are helpful optimizations for experts, where the operational ergonomics of interacting concurrent processes are well known enough. And for experts they help unify two different concurrency domains- processor-based being one, io-based being the other. This is why you see examples of use of these in state machines and ioloops, both expert level optimization constructs.
But even there, they are attributes of a code block, something that should be reachable through some dunder-token, as are other expert level machinery/magic. To add them as top level standalone keywords in a language lauded for being executable pseudocode is an astonishing misunderstanding.
Hope that is helpful.
Again, CSP is a much more approachable mental model. A Golang-like channel construct would have been a much better addition to the language in place of await. And exposing a not-directly-executable dunder flag much better than async.
Cheers.
Trio especially with its requirement of an owner for each task makes the thinking of it all entirely sequential and clear. In curio it's optional.
See the rationale here: https://vorpus.org/blog/notes-on-structured-concurrency-or-g...
On top of that, I must recommend dabeaz's concurrency from the ground up talk, where he explains it all in primitive terms and lives codes a wee co-operative yielding thing on the fly:
Given its existence, I honestly don't understand why asyncio is in use by anyone. It should be removed from stdlib as a mistake and efforts around the curio / trio ecosys should be redoubled and brought to the mainstream. It's clear now, all this time after asyncawait was brought in, that the thinking which lead to asyncio was erroneous.
I'm the author of a sort of popular curio/trio lib. I used to maintain a compatibility layer lib that allowed code to be written to use either or, just for my project. That torch has been taken up by anyio, allowing anyone who's writing something for either curio or trio to kill two birds with one stone and contrib to both sides at once. Is great.
The code executes asynchronously, and if there's a specific operation I want to wait for before continuing execution, I signal that I want to synchronously execute this block of code using await.
I understand this is not technically how async/await works, however I believe it's an easy way to read and understand how async code will execute.
For example:
async def func():
# do some stuff asynchronously
# wait to get values before continuing execution
values = await get_values()
# do more stuff with values asynchronouslyWhen you're running ordinary Python code, it is synchronous - it runs line by line and has control of the CPU. When you await on another function, that is the asynchronous part - your code is suspended until that other code completes. And then after the await is satisfied, your function starts to run synchronously again.
So I started using multiprocessing and multithreading available via concurrent.futures. I built a wrapper package called `ori`[2] that lets you build chains of threadpools and processpools without having to enter the Python asyncio kingdom.
[1]: https://journal.stuffwithstuff.com/2015/02/01/what-color-is-...
[2]: https://ori.technology.neocrym.com/en/latest/ori.poolchain/#...
"Parallelized async"
I was slightly torn on the usage of the word "parallelise", in that yes, CPU-wise, no tasks will progress in parallel.
However, the point of the library is to allow certain parts of a pipeline to progress at the same time in the real-world. In the example, they are calls to asyncio.sleep, but in real cases, they could be HTTP requests. As in, the bytes of the request/response will be going across the wire at the same time, so the total wall-clock time of (certain) pipelines will go down.
So, I claim "parallelise" _is_ appropriate.
And for writing to disk, async is basically the only way to "parallelize" such operations
I just think its inappropriate because, if I was looking for a parallelization library like multiprocessing, and I found this, I might waste 15 minutes or so before I figure out that this isn't parallel, it's just async.
Async is okay, but it does nothing to improve performance, like what parallelization does. Async provides good UX, parallelization provides good performance. Conflating the two just muddys the water for the people who like to do high-performance-computing.
Like you have a pipeline of [coro1, coro2] where coro2 is using data as it's yielded out of coro1.
But you can also just have like, coro2 yield from coro1 or similar.
I think I like it. Not sure yet. It might help composibility if you find yourself slapping together a lot of discreet little coros.
Ours is based off Twisted, which works pretty similarly to the new async stuff in python3.
The thing ours does, that I can't tell if this one supports or not, is you pass it a list of functions you want it to call (in python3 you would just pass it coroutines I suppose, but in Twisted an async function gets scheduled automatically, so you need to pass it in an uncalled state), and then you specify the "width" of the pipeline, and it runs through your list of calls, ensuring that "width" number of calls are in flight at a time.
It's a super useful little utility to have around.
An eagerly executing coroutine to consume the original generator and put it in a queue, and a new simple generator that yields values from the queue till it's told to stop.
I'm not sure if this matches the behavior of the original library for task cancellation though.
def eager_gen(gen):
end_marker = object()
queue = asyncio.Queue()
async def consumer():
try:
async for val in gen:
await queue.put(val)
except Exception as e:
await queue.put((end_marker, e))
else:
await queue.put((end_marker, None))
async def new_gen():
while True:
item = await queue.get()
if isinstance(item, tuple) and item[0] == end_marker:
break
yield item
if item[1]:
raise item[1] from None
_task = asyncio.create_task(consumer())
return new_gen()
While yes, running a coroutine in a new task is in the standard library, that is only a component of what this library does, which is:
- given an async iterable,
- iterating over it in a new task, into a buffer,
- and then returning an iterable that yields values from this buffer as it's filled.
I don't _think_ this logic is in the standard library, at least not at https://docs.python.org/3/library/asyncio-task.html#asyncio.... from what I can see.
Have you thought about making it work as a decorator?
The issue is what happens on exceptions to avoid tasks hanging around forever. Exceptions propagate “forward” in the pipeline fairly automatically. However, getting the ones “behind” the exception to notice: that was tricky. The best I came up with was the bit of state wrapped up in buffered_pipeline that notices the exceptions and cancels the tasks.
I've gotten along ok with create_task and gather. I'll need to spend some time with this implementation to see if it fits any of my existing asyncio coding patterns.
What is the mechanism used to parallelize, threads, processes or something else - it's worth mentioning in the readme.