Move CPU Bound Tasks off Tokio Threadpool

[tustvold]

Is your feature request related to a problem or challenge?

DataFusion currently performs CPU-bound work on the tokio runtime and this can lead to issues where IO tasks are starved and unable to adequately service the IO. Crucially this occurs long before the actual resources of the threadpool are exhausted, it is expected that if there is no CPU available that IO will be starved, what we want to avoid is IO getting throttled when there is resource available.

The accepted wisdom has been to run DataFusion in a separate threadpool and spawn IO off into a separate threadpool, as described here. #12393 seeks to document this, and by extension #13424 and #13690 to provide examples on how to achieve this.

Unfortunately this approach comes with a lot of both explicit and implicit complexity, and makes it hard to integrate DataFusion into existing codebases making use of tokio.

Describe the solution you'd like

The basic idea would be, rather than move the IO off to a separate runtime away from the CPU bound tasks, instead wrap the CPU bound tasks so that they can't starve the runtime.

Ultimately DF has three broad class of task:

1. Fully synchronous code - e.g. optimisation, function evaluation, planning, most physical operators
2. Mixed IO and CPU-bound - e.g. file IO, schema inference, etc...
3. Pure IO - catalog

The vast majority of code in DF falls into the fully synchronous bucket (1.), and so it is a natural response to want to special case the less common IO (2. / 3.) instead of the more common CPU-bound code. However, I'd like to posit that conceptually and practically it is much simpler to dispatch the synchronous code than the asynchronous code, and that the items falling into the second bucket are the most complex operators in DataFusion.

This would yield some quite compelling advantages:

• DataFusion could easily be integrated with existing tokio-based workloads
• Avoids the many footguns of CPU-bound async tasks

This could be achieved using the native primitives

One way to achieve this would be to use block_in_place, as unlike spawn_blocking it does not require a 'static lifetime and can therefore borrow data from the invoking context.

That being said this does come with some risks:

• It might become hard to determine what is sufficiently CPU-bound to warrant spawning
• There would potentially be a lot of callsites that would need wrapping
• It would be important to use block_in_place only at the granularity of a data batch, to ensure overheads are amortized
• Applications relying on thread count to limit concurrency might need to put in place some additional concurrency controls (e.g. a semaphore)

An arguably better solution would be to dispatch the work using an actual CPU-threadpool such as rayon.

Describe alternatives you've considered

We could instead move IO off to a separate runtime

Additional context

No response

[adriangb]

I considered something like this when looking into how to solve the problem and I think the sticking point is:

It might become hard to determine what is sufficiently CPU-bound to warrant spawning

It's relatively easy to, at a low level, know where IO calls are happening. In fact Tokio lets you configure the runtime to panic if you do IO at all (which makes it easy to be almost completely certain you didn't miss a spot).

[tustvold]

Yeah, there are tools like tokio-console and the runtime metrics that could help to find problematic codepaths, but I'm not going to pretend it is trivial. I do think it may end up being simpler, and potentially a better long term solution, than dispatching the async work.

[crepererum]

Couple of thoughts:

• runtime requirements: block_in_place requires the multi-thread runtime. That's not a blocker, but we should clearly communicate that this means that DF can no longer run under the "same thread" runtime.
• performance: Some experiments that I did suggested that block_in_place scales significantly worse that two runtimes (one for CPU work, one for IO). However this may not be the general case. I just had the impression that block_in_place is not particularity well optimized and in fact it requires the tokio scheduler to do some rather nasty thread switching.

I'm wondering if we should rather write our own async scheduler for the async compute graph that can better deal with the DF workload. For IO, we still can (and should) use tokio, but I question if shoehorning tokio into the CPU-bound workload is really worth it. And FWIW: such a scheduler can -- at least to some extend -- be pull OR push-based (requires some wiring, but not impossible).

[tustvold]

The performance implications definitely concern me, I have a nagging suspicion block_in_place spawns a thread...

An arguably better solution would be to instead spawn the CPU bound work to a threadpool like rayon, but this would require any arguments to be 'static. However, this may not be as much of an issue for sync code as it is for async.... I'll update the ticket... Such a solution would have a number of compelling properties

I'm wondering if we should rather write our own async scheduler for the async compute graph that can better deal with the DF workload

This sounds a lot like #2199, I am not entirely sure such an approach is feasible at this point.

[alamb]

It's relatively easy to, at a low level, know where IO calls are happening. In fact Tokio lets you configure the runtime to panic if you do IO at all (which makes it easy to be almost completely certain you didn't miss a spot)

Yes, agree broadly speaking with @adriangb

Almost all of DataFusion's code is CPU bound and as @tustvold and others have observed before the locations where IO is done is (relatively) isolated in comparison.

I can try to take another shot at #13690 to more fully annotate IO in DataFusion.

[tustvold]

I can try to take another shot at #13690 to more fully annotate IO in DataFusion.

IMO the key challenge is async is an abstraction designed for IO concurrency, not CPU parallelism. This is what ultimately makes it so difficult to disentangle in a coherent way. Whilst we can dispatch the "IO" off, the boundary inevitably ends up fuzzy, an async task doesn't have a single start and end, inevitably leading to complex shenanigans around lifetimes, buffering, etc...

I'm increasingly of the opinion that having CPU-bound futures fundamentally breaks their abstraction, and rather than fighting this, DF would be better served by avoiding such things in the first place. Instead it can use async / tokio for what it is designed for, and move the CPU bound work somewhere else. I don't actually think this would be that bad, most kernels are synchronous and could be offloaded without too much ceremony. It would also be largely mechanical, and something that could easily be actioned by a broad contributor base.

I'd be happy to write up an example of what this might look like if there is interest in taking such a direction

Edit: It is also potentially worth highlighting that this would naturally open the door to parallelism beyond that afforded by the plan, i.e. morsel driven parallelism.

[alamb]

I'm increasingly of the opinion that having CPU-bound futures fundamentally breaks their abstraction, and rather than fighting this, DF would be better served by avoiding such things in the first place.

LOL I feel like this is not a new opinion (e.g. #2199) I agreed to disagree with you on this 😆

I realize the current use of async tasks is non ideal, but it seems to be working just fine for performance.

So, attempting to be pragmatic, I am going to spend my effort improving the current situation rather than trying to undertake a larger refactoring, unless there is something that is not possible (however non ideal) in the current design.

[tustvold]

I realize the current use of async tasks is non ideal, but it seems to be working just fine for performance.

Yes the current model works perfectly fine when there is little to no IO and sufficient plan parallelism to exploit, I don't dispute this.

There is likely more going on here, but I can't help notice DF doesn't perform nearly as well on "cold" data

LOL I feel like this is not a new opinion (e.g. #2199) I agreed to disagree with you on this 😆

Sure I have made no secret of disliking DF's use of async, and that we have disagreed on this in the past, however, I guess my hope is that we might have reached critical mass of pain to actually do something about it. This is no longer me just moaning about a theoretical issue.

Ultimately moving the CPU bound work off tokio isn't that complex, and the DF community has successfully pulled off far more complex initiatives (e.g. StringView). Doing so would simplify integrating DF into existing codebases, eliminate whole classes of footgun, and allow for greater parallelism than is currently possible.

Edit:

unless there is something that is not possible (however non ideal) in the current design

I don't think there is a good story for intermixing IO with DF, and there is plenty of empirical evidence to this effect. We at Influx were aware of the need to run separate threadpools, and yet we still lost months to this class of issue. Blessing an approach we empirically know caused significant pain, in spite of significant investment to avoid the issue, seems unfortunate.

[crepererum]

I realize the current use of async tasks is non ideal, but it seems to be working just fine for performance.

It works in an artificial benchmark that likely runs on an oversized machine. So like most benchmarks, it's a rather biased PoV.

Yes the current model works perfectly fine when there is little to no IO and sufficient plan parallelism to exploit, I don't dispute this.

💯

[alamb]

Ultimately moving the CPU bound work off tokio isn't that complex, and the DF community has successfully pulled off far more complex initiatives (e.g. StringView).

This is the core challenge in my mind. I would love to believe this project isn't that complex, but I have yet to see someone show how this would work practically with DataFusion (as well as trying myself). I am more than willing to coordinate this work once we have a pattern to apply.

Doing so would simplify integrating DF into existing codebases, eliminate whole classes of footgun, and allow for greater parallelism than is currently possible.

I agree with the first two points (more parallelism I am not sure). I would very much like this to happen and am willing to invest a significant amount of effort to to do. I just don't know how to do it.

[tustvold]

I think moving to a push-based execution model as #2199 attempted is likely intractable at this stage. However, I think there is a compromise where just the kernels are dispatched, I can probably try to find some time to write up something more concrete in the next few days if there is interest