[DISCUSS] Distinguish between "performance" and "efficiency" cores

[alamb]

Is your feature request related to a problem or challenge?

Modern Macs / Apple Silicon (and Intel 12th gen processors) make a distinction between Performance (P) and Efficiency (E) cores.

https://developer.apple.com/news/?id=vk3m204o

Recent Apple Silicon like A13 Bionic has both high-performance cores (P cores) and high-efficiency cores (E cores). These different core types allow you to deliver apps that have both great performance and great battery life. To take full advantage of their performance and efficiency, you can provide the operating system (OS) with information about how to execute your app in the most optimal way. From there, the OS uses semantic information to make better scheduling and performance control decisions.

Intel 12th gen chips have this feature too, for example:
https://www.intel.com/content/www/us/en/gaming/resources/how-hybrid-design-works.html

Intel® Core™ desktop processors integrate two types of cores into a single die: powerful Performance-cores (P-cores) and flexible Efficient-cores (E-cores). Both types of core have a different role.

DataFusion currently evenly partitions work between cores, and does not distinguish or take advantage of the different kinds of cores. This could lead to less than optimal throughput as the (slower) efficiency cores would be given as much work as the faster power cores, leading to the power cores being idle sometimes

@pepijnve may be seen this manifesting as inconsistent behavior / profiling on #16398

Describe the solution you'd like

Ensure DataFusion runs on architectures with this hybrid P/E design efficiently.

I don't know what this means exactly

Describe alternatives you've considered

Note sure. Maybe we can follow the ideas from Mac:

• https://developer.apple.com/news/?id=vk3m204o

We may also have to have support in tokio 🤔

Additional context

@pepijnve brought this up on a PR where we are seeing performance variability

• Use Tokio's task budget consistently, better APIs to support task cancellation #16398 (comment)

🤔 now that I write '10-core MacBook', I'm wondering if the 10 core part is where my variability is coming from. That's 6 performance and 4 efficiency cores. Ideally DataFusion keeps the CPU bound work on the perf cores, and uses the efficiency ones for IO. I had been wondering about that and NUMA effects already. A topic for a different thread though.

[alamb]

A good quite quite https://developer.apple.com/news/?id=vk3m204o

Statically pre-assigning pieces of a parallel workload to worker threads will leave threads idle before the end of the execution. This is because not all cores are equal and so worker threads will not make identical progress. Instead, subdivide parallel problems into a large number of pieces and use a work-stealing algorithm to balance these pieces across threads to keep all workers busy.

[pepijnve]

What the article describes is pretty much what datafusion is already doing I guess. What I observed, but only at a glance, is that all cores are kept busy when the benchmarks are running. The hypothesis I had was that perhaps unfortunate placement of work (number crunching in efficiency and waiting for io on performance for instance) could lead to suboptimal performance. I’m not super familiar with Instruments, but maybe it has a way to make the io vs cpu work visible in the profiler somehow.

[alamb]

I think Datafusion is statitically partitioning the work N ways where N is the number of CPUs

It doesn't take into account that the cores are not symmetric (AMP rather than SMP) 🤔

[pepijnve]

The guide behind the developer login wall has much more detailed information if you're interested https://developer.apple.com/download/apple-silicon-cpu-optimization-guide/

This reminds me a bit of the morsel paper. I wonder if you could achieve similar results with a repartition strategy that takes queue length into account or some other metric that approximates load per thread. In other words, try to send data down the lane that's processing the fastest.

You might risk getting a rather uneven data distribution between the partitions that way though. How do the push engines deal with that? If I understood it correctly from the paper they also keep data in separate per-thread storage before merging into something global.

[pepijnve]

I don't think I'm legally allowed to copy/paste from that guide, but I think I can paraphrase.

Things might actually be pretty reasonable as is today. In a sense, every Tokio task 'tick' is a discrete chunk of work. There are actually lots of tasks in flight at the same time when a query is running, definitely more than the number unless the query is quite simple. This might be sufficient already. The yielding work that was done may actually help in this regard, since it shortens the length of each 'tick'. This ends up creating more schedulable chunks of work.

Making the partitioning strategies more load aware somehow might still help though to try to avoid congestion in partitions that for whatever reason are going a bit slower than the others.
I will say that I have no understanding yet at this point how the hash based partitioning percolates through the pipeline. Is that something that's essentially a local decision for the repartition operator or does that have consequences further down the line in parent operators as well?

[pepijnve]

I was just scanning through https://www.microsoft.com/en-us/research/wp-content/uploads/2024/12/Extensible-Query-Optimizers-in-Practice.pdf. Seems my intuition is not too far off

Finally, we note that while the Exchange operator can simplify
the implementation of parallelism, it can still be beneficial to design
and implement multi-threaded operators. For example, with Exchange
operator, each thread would build its own hash table in the Hash Join
operator, which could result in hash tables of variable sizes due to
data skew. In contrast, if the Hash Join operator is designed to be
multi-threaded, it is possible to build a single hash table that supports
concurrent operations, which avoids the above issue with data skew.

I think that's kind of what the morsel paper suggests as well. Avoid processing imbalance by partitioning more dynamically and compensate for data skew by coalescing in a multi-threaded build side. Easy to write that sentence, actually building it is a different matter.

[alamb]

I think that's kind of what the morsel paper suggests as well. Avoid processing imbalance by partitioning more dynamically and compensate for data skew by coalescing in a multi-threaded build side. Easy to write that sentence, actually building it is a different matter.

Yeah, I agree -- in theory building multi-threaded operators could potentially better manage the asymmetry, but in practice I think it gets very tricky. For example, in DuckDB they use thread-local storage to store state per thread (rather than have a fully concurrent hash table); That design does not obviously have any better skew avoidance properties

You might risk getting a rather uneven data distribution between the partitions that way though. How do the push engines deal with that? If I understood it correctly from the paper they also keep data in separate per-thread storage before merging into something global.

100% agree -- I don't know how

[alamb]

Making the partitioning strategies more load aware somehow might still help though to try to avoid congestion in partitions that for whatever reason are going a bit slower than the others.

This might be interesting for RoundRobin partitioning, but it is not clear to me how it would work for hash partitioning

I will say that I have no understanding yet at this point how the hash based partitioning percolates through the pipeline. Is that something that's essentially a local decision for the repartition operator or does that have consequences further down the line in parent operators as well?

I think it has consequences farther down (for example, the last phase of a two phase grouping relies on the fact that data is partitioned into non overlapping partitions for correctness)