Optional removal the FIFO constraint for individual nodes to enable parallel execution

[oersted]

Bonobo is running all nodes in parallel using the current execution strategy, which means that no node will block another node, but for one given node, it uses a first-in-first-out, one at a time, execution strategy.

This is often the desired behaviour, but for a few cases, this is not necessary and even not wanted.

Example cases includes everything where the FIFO property is not required per-se, and where the amount of time necessary to process something is quite dependant on a non-local resource (http request, flakky webservice, humans, ...).

I will attempt to implement this feature, but I am very new to the code base (I discovered Bonobo yesterday), so I will certainly need advice and help. Any assistance will be much appreciated.

I've had a brief discussion with @hartym in Slack, and we have decided to create this issue to continue the ongoing discussion. He has kindly written out his first thoughts in the RFC, I will use it as my foundation: https://github.com/python-bonobo/bonobo/wiki/%5BRFC%5D-Transformation-Features#parallel

[oersted]

@hartym What if, instead of changing the execution of certain marked nodes to be parallel at the executor level, we left the node execution as-is, and we instead implemented parallelism at a graph level?

The same effect can be achieved by duplicating a node n times, the current execution strategy would already execute them in parallel. The only change needed would be to have a single shared input queue for these parallel nodes or to have a simple "load balancer" that would distribute inputs homogeneously between the input queues. The first option is probably more optimal performance wise.

What do you think?

[oersted]

I have just finished my first run through the code. I think I now have a decent grasp on how things work at the core, AKA what happens when you do bonobo.run(graph).

Anyways, now I have another question. The concurrent.futures module already gives us a pool of threads or processes. What is stopping us from submiting n copies of the _runner function in ExecutorStrategy.get_starter? Wouldn't that loop the node in n parallel? I assume it's not so simple, I just want to know what are the obstacles.

Also, @hartym could you further elaborate on your argument for coroutine-s? I still don't completely grasp how they could be used here. Thanks.

[hartym]

@oersted

About "Just duplicating nodes".

Would work, but has a few drawbacks, some of them make me think it must not be implemented like that.

1. You would get n contexts for 1 logical node execution. Which means that you'd have duplicated (or at least split) statistics, and more importantly, you can't rely anymore on execution-wide variables.
2. In case of non parallel strategies, you would not benefit from the "io wait" time (not very important though, naive strategy is not really real world compliant).
3. It would create unnecessary copies of basically the same object, which is not very big, but still feels not the right thing to do.
4. Dispatching the queue reads between nodes may be problematic, and maybe someone wants to have decorators or other modifiers that work at a node level thinking that this node will have only one context. If you duplicate it, bad things will happen even if I'm not very certain in which way yet.
5. It creates a lot of threads (or subprocesses, or ...), and at some point, it can have a cost and become a problem.

Implementing the queue input would be trivial though, as it's threadsafe queues I think you're certain that only one node would read an item. But you'd have to handle the "ending" cases (unrecoverable errors, queue's done, etc.)

About coroutine usage.

I think coroutines may be a good fit there, because it's already a mechanism that make code asynchronous (and FOFO as opposed to FIFO).

It works by having an event loop and "futures", so using it to react on "work done" event make me feels it's a good fit.

Let's say a context has an event loop. Instead of having a loop() function that calls self.step() while the node runs, it could behave by appending tasks to the context's event loop. Append it n times, depending on the max concurrency allowed by user. Once a task is done, it would update the statistics and push the result to the output queue as it's done today. Still have only one context. Allows tasks to finish in the order they want. As a bonus, I feel like using the async keywords make it easy to read, if you're willing to use asynchronous versions of the calls, while not having on the user side to fully understand how to build and execute its own event loop. If the user does not want to go async, then it could be used anyway using a sub-executor (like threads, etc.) and the user would have only a decorator to set it up.

I probably overlook a bit some problems that may arise, but I feel like it's the right way to go. Let's exchange on specific points you'd think about on slack.

[hartym]

One more thing though, I think that this would not "change the execution of certain marked nodes to be parallel at the executor level", instead, all could be done in the NodeExecutionContext.

And I also think that it would not make it parallel, but instead asynchronous, hence the suggestion of looking towards asyncio.

[oersted]

Good, thanks. I assumed it wouldn't be so simple, I just wanted to check these possibilities as I don't have the bigger picture. I understand and agree.

About your last comment. Yep, I wrote that before having a good understanding of the code, it would be done in NodeExecutionContext, yes.

I have also understood how it would be built with coroutines now. I don't have any more time today, we can talk later this week about implementation details.

[hartym]

@oersted Great, let me know if I can be of any help :)

[hartym]

Prior discussion in #185 too.