Add indexing scheduling plan density optimization - #6796
Conversation
Codex Review SummaryThis comment shows the latest Codex review activity on this pull request.
ℹ️ About Codex in GitHubYour team has set up Codex to review pull requests in this repo. Reviews are triggered when you
Codex reacts with 👀 while any review is running, comments if it has suggestions, and reacts with 👍 once all reviews finish with no findings. |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 53240078fd
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| let can_optimize_plan = is_plan_eligible_for_optimization( | ||
| &indexers, | ||
| &indexer_statuses, | ||
| is_locality_aware, | ||
| &mut self.state, |
There was a problem hiding this comment.
When an underfilled plan becomes stable after a normal apply, the periodic ControlPlaneLoop calls control_running_plan, whose equal-node/equal-task path performs no rebuild, so this eligibility check is never revisited when the 30-second deadline expires. The event-driven rebuild that created the plan generally occurs before the new plan is stable, meaning a quiet cluster can remain underfilled indefinitely; schedule a repair rebuild from the stable control-loop path when the cooldown is due.
Useful? React with 👍 / 👎.
| let combined_num_shards = indexing_tasks[*donor_task_ord].shard_ids.len() | ||
| + indexing_tasks[*receiver_task_ord].shard_ids.len(); | ||
| if combined_num_shards <= max_num_shards { |
There was a problem hiding this comment.
When the two smallest pipelines exceed the capacity in aggregate, this code makes no repair even when redistribution can reduce the pipeline count. For example, with capacity 10 and three pipelines containing 6 shards each, moving 4 shards produces [10, 2, 6], after which the latter two can merge into an optimal [10, 8]; the current implementation permanently stalls at [6, 6, 6], so the advertised hill-climbing convergence fails for valid layouts.
Useful? React with 👍 / 👎.
| .map(|(task_ord, _)| task_ord) | ||
| .collect(); | ||
| task_ords.sort_by_key(|task_ord| (indexing_tasks[*task_ord].shard_ids.len(), *task_ord)); | ||
| let [donor_task_ord, receiver_task_ord, ..] = task_ords.as_slice() else { |
There was a problem hiding this comment.
O_O I did not know we could do that. That's so cool!
There was a problem hiding this comment.
so you only consider merging opportunities right?
if max_num_shards == 3, and you have 3 pipelines with 2 shards, density won't get improved?
There was a problem hiding this comment.
Yes, I considered this. The approach is then to always collapse pipelines so that one is full. ie. 2 + 2 should leave 3 + 1, and then the next pass will fill the 1.
There were two drawbacks to this:
The first, it will take longer for the plan to settle, and it will reset more pipelines. It will also create partially full pipelines for some period of time, ie. to go from pipelines of 2/2/2 to 3/3, you first need to go to 3/1/2, then 3/3, which means you're also producing more small splits until that part is done. Because indexers are shuffled (important to not hammer one indexer/keep one shard from getting indexed for too long), it could be some number of minutes before it's re-addressed.
The second is that it's more code, and a little more complexity as a result.
We can do this. The density percentage (% of pipelines in our plan compared to the mathematical minimum) difference was about 10% . (ie. 240 pipelines vs 260 pipelines). The net result is ~two less compactions and write amplification. Perhaps it's worth it.
(thinking out loud).
a6ce9cf to
57e9995
Compare
045039d to
3d77337
Compare
2 participants
Description
Introduces a hill climbing density optimization pass for the scheduling algorithm. Copied from the Readme:
The idea behind optimization is simple: if we're able to perform one single action that improves the plan, over time, we'll eventually
converge on a plan that's quite good. This is a hill climbing algorithm - we don't attempt to solve the optimization problem in one shot,
because 1. It's very complex, and the algorithm to do so is (dangerously) exponential, and 2. it's very disruptive - we'd likely reset a lot of pipelines at once.
Optimization is gated on three things: locality being enabled, the plan being stable, and a 30 second cooldown.
The stability requirement is because resetting a pipeline, which optimization does, is expensive, as
we throw away uncommitted work for the shards in that pipeline. So we try to perform optimization only on a stable cluster.
This works doubly well with deployments, because deployments, which are already resetting many pipelines in their own right, also leave the cluster
in a chaotic state. Optimization can kick in once the plan is stable (and the deployment finished).
Optimization itself works in a naive, greedy, simple way. For each source, find two pipelines that are running under capacity; rearrange the shards each is indexing
so that one pipeline fills up.
If we do this enough times, eventually, we'll end up with the minimal number of pipelines, and optimal density.
How was this PR tested?
Unit tests and extensive cluster tests.