Show HN: Parallelise pipelines of Python async iterables (github.com)

"Parallelized async"

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:

https://youtu.be/MCs5OvhV9S4

    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)

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

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 asynchronously

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.

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.