Improve parallel scaling of e-matching - #954
Conversation
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #954 +/- ##
==========================================
+ Coverage 86.59% 86.77% +0.17%
==========================================
Files 95 95
Lines 29676 30572 +896
==========================================
+ Hits 25699 26528 +829
- Misses 3977 4044 +67 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
Merging this PR will degrade performance by 22.84%
|
| Mode | Benchmark | BASE |
HEAD |
Efficiency | |
|---|---|---|---|---|---|
| ❌ | Simulation | tests[luminal-llama] |
504.2 ms | 653.4 ms | -22.84% |
Tip
Investigate this regression by commenting @codspeedbot fix this regression on this PR, or directly use the CodSpeed MCP with your agent.
Comparing ezr-better-parallel (90ebda4) with main (53b9721)
Footnotes
-
227 benchmarks were skipped, so the baseline results were used instead. If they were deleted from the codebase, click here and archive them to remove them from the performance reports. ↩
| // immutable plans before any worker starts. A slot acquires and refreshes | ||
| // its Arc through the regular catalog helper on first cached use. The | ||
| // sidecars are dropped before `merge_all` resets catalog entries. | ||
| let prepared_plans = rule_set |
There was a problem hiding this comment.
nit: call them prepared_indexes instead?
| } | ||
| match plan { | ||
| Plan::SinglePlan(plan) => { | ||
| match (plan, prepared_plan) { |
| let prober = | ||
| self.get_column_index(atoms, binding_info, scan.atom, scan.column, prepared); | ||
| let size = prober.len(); | ||
| // The two-way hot path historically breaks ties in favor of the |
There was a problem hiding this comment.
I wonder if it's cleaner to instead change run_plan to match the natural tie-breaking behavior here.
| for i in start..instrs.len() { | ||
| if matches!( | ||
| &instrs[i], | ||
| &instrs[order.get(i)], |
There was a problem hiding this comment.
is it true that i == order.get(i) at this point?
| // Count how many times each atom has been refined so far. | ||
| for ins in instrs[..range.start].iter() { | ||
| match ins { | ||
| for position in 0..range.start { |
There was a problem hiding this comment.
i also don't quite understand this change
There was a problem hiding this comment.
This change is bad! good catch. It's fixed in a follow-up and I can try and backport it if the rest looks good.
| // serially. Besides avoiding an intermediate key copy, this keeps | ||
| // related nested probes on one worker. Buffers that cannot construct an | ||
| // independent partition (the in-place executors) decline this path. | ||
| if !stages.instrs.is_empty() |
There was a problem hiding this comment.
Is the idea that if the top level has enough entries, we will only do top-level parallelism? This coarser-grained parallelism can be technically more sensitive to skews but it does not show up in practice and it avoids some overheads with nested parallelism?
Have you measured the percentage of hits on this path versus the nested parallelism path? I feel this top-level parallelism may overshadow the default nested parallelism paths for most of the rules.
| /// coarse partition also passes `index_shard`; recursive calls clear it so | ||
| /// only the first intersection is restricted to that physical shard. | ||
| #[allow(clippy::too_many_arguments)] | ||
| fn run_plan<'buf, A: NumericId + 'buf, BUF: ActionBuffer<'buf, A>>( |
There was a problem hiding this comment.
maybe document the assumption that index_shard.some() implies no more parallelism at this level.
| action: ActionId, | ||
| bindings: &DenseIdMap<Variable, Value>, | ||
| mut to_exec_state: impl FnMut() -> ExecutionState<'scope>, | ||
| ) { |
There was a problem hiding this comment.
should we set self.needs_flush = true; here? (was comparing this and ScopedActionBuffer::push_bindings)
| &mut self.batches, | ||
| self.rule_set, | ||
| self.match_counter.as_ref(), | ||
| ); |
There was a problem hiding this comment.
should we set self.needs_flush = false;?
|
From running Update: numbers from nightly are mostly consistent with the above
But the packed trie PR #959 shows some significant speedups.
|
|
I think the llama slowdown is due to the buggy variable ordering heuristic that codex snuck in; it's reverted in the packed-trie branch. |
Add private worker queues and scheduler metrics, partition generic joins by physical index shard, and retain prepared index handles for recursive probes.
a384d25 to
90ebda4
Compare
3 participants
(branched off of
perf-shared-trie-nodes)This change improves parallel e-matching scaling with a few optimizations:
Arc::clonecalls with index construction: handles on shared indexes are grabbed before the plan starts running to avoid big clone calls. This alone was a large win.spawn_localmethod onto the thread pool that allows for work to be appended to a local deque first, only migrating to the global queue if sufficient threads are stalled. This appears to help cases where there isn't a lot of parallelism to expose. The thread pool essentially sprayed data randomly across threads, which makes cache locality much harder to achieve. (The original morsel-driven parallelism paper talks about this too). Local spawns maintain locality while still improving parallel utilization if there's a lot of skew.To evaluate this I had codex write up a benchmark using the dataset and queries from the Honeycomb paper. This is just a single query over a large dataset, so it's a helpful test case for this part of the code: other egglog benchmarks have complex schedules and many rules that we parallelize across sometimes.
Here are the overall results on my m4 max laptop:
(Note that M4 max only has 12 P-cores, so some amount of flattening after 12 is expected. Still, I suspect there's more to do here)