[SPARK-43086][CORE] Add configurable bin-packing task placement

[starcatmeow]

What changes were proposed in this pull request?

This PR adds an opt-in bin-packing task placement strategy to TaskSchedulerImpl.

It introduces spark.scheduler.taskPlacement.strategy with two values:

• SPREAD, the default, preserves the existing behavior of cycling through shuffled
  executor offers one task at a time.
• BIN_PACK fills an eligible executor before moving to the next one.

BIN_PACK applies only during the NO_PREF and ANY scheduling passes.
PROCESS_LOCAL, NODE_LOCAL, and RACK_LOCAL retain Spark's existing shuffled offer
order, so locality continues to take precedence over packing.

For bin-packed passes, executors with running tasks or tasks assigned earlier in the
current resourceOffers call are considered before idle executors. Each group is ordered
lexicographically by executor ID. The scheduler sorts offer indices once per
resourceOffers call, leaving shuffledOffers, availableCpus, availableResources,
and the output task buffers in their existing shared index space. It then performs a
stable busy/idle partition for each TaskSet and bin-packed pass so that later TaskSets
observe assignments made earlier in the scheduling round.

The PR also documents the new configuration and adds tests for the configuration,
placement order, locality, multiple TaskSets, custom resources, and barrier tasks.

Why are the changes needed?

SPARK-43086 describes a resource
efficiency issue when dynamic allocation is used with the existing spread placement
behavior.

Dynamic allocation removes an executor only after it becomes idle. When a stage has
fewer tasks than the available cluster slots, spreading those tasks across executors can
keep more executors busy than necessary and delay scale-down.

For example, if four executors have two task slots each, a four-task stage can place one
task on each of four executors with SPREAD. With BIN_PACK, it can instead place two
tasks on each of two executors, allowing the other two executors to become idle and be
removed.

The new strategy is opt-in to avoid changing Spark's existing default placement and
latency behavior.

Does this PR introduce any user-facing change?

Yes. It adds the following configuration:

spark.scheduler.taskPlacement.strategy=BIN_PACK

Supported values are:

• SPREAD: preserves the existing placement behavior and remains the default.
• BIN_PACK: fills busy executors first and then idle executors, using lexicographic
  executor ID order within each group, during the NO_PREF and ANY passes.

Applications that do not set this configuration retain the existing behavior.
Locality-specific placement for PROCESS_LOCAL, NODE_LOCAL, and RACK_LOCAL is
unchanged.

How was this patch tested?

The scheduler unit suite passed:

build/sbt 'core/testOnly org.apache.spark.scheduler.TaskSchedulerImplSuite'

All 122 tests passed. The added coverage includes:

• the default SPREAD behavior;
• configuration parsing and invalid values;
• deterministic BIN_PACK executor ordering;
• running and current-round assigned executors before idle executors;
• re-evaluating busy state between TaskSets;
• preservation of PROCESS_LOCAL and NODE_LOCAL placement;
• bin packing at ANY locality;
• custom resource limits;
• barrier tasks and provisional barrier assignments.

Scalastyle and a full package build also passed:

build/sbt core/scalastyle
build/sbt package

The two strategies were additionally tested with real executor JVMs using:

• local-cluster[4,2,1024];
• four initial and maximum executors;
• two cores per executor;
• dynamic allocation and shuffle tracking enabled;
• one minimum executor;
• a four-second executor idle timeout;
• sequential stages containing 8, 4, and 2 NO_PREF tasks;
• twelve-second tasks, allowing idle executors to be removed while a stage was running.

The steady executor counts observed during each stage were:

Strategy	8-task stage	4-task stage	2-task stage
SPREAD	4	4	2
BIN_PACK	4	2	1

The Spark Event Timeline showed that SPREAD first removed executors during the
two-task stage, while BIN_PACK removed two executors during the four-task stage and
another executor during the two-task stage.

SPREAD:

BIN_PACK:

Was this patch authored or co-authored using generative AI tooling?

Generated-by: OpenAI Codex (GPT-5)

[sunchao]

LGTM

[sunchao]

cc @viirya @cloud-fan @dongjoon-hyun @peter-toth

[viirya]

I traced the scheduler paths independently and this holds up — including a few spots worth spelling out since it's a placement change.

• SPREAD is behavior-preserving. With useBinPacking = false the new inner do { ... } while (useBinPacking && launchedTask) runs the body exactly once per offer index, so resourceOfferSingleTaskSet still does a single pass launching at most one task per offer, and the outer do/while(launchedTaskAtCurrentMaxLocality) still drives the round-robin. Default behavior is unchanged.
• The busy/idle predicate is well chosen. isExecutorBusy(execId) || availableCpus(i) < shuffledOffers(i).cores — the first term catches executors made busy by prior rounds or earlier addRunningTasks in this round, and the second catches provisional barrier assignments in the current round (which decrement availableCpus but aren't registered as running until the whole barrier set launches). The comment captures this well, and the "treats barrier locality assignments as busy" test pins it.
• Shared index space stays intact. Only the iteration order is reordered via offerIndicesByExecutorId; shuffledOffers / availableCpus / availableResources / tasks are all still addressed by the original index i, so there's no re-indexing hazard, and the sort happens once per resourceOffers call.
• Delay-scheduling accounting is unaffected. hasScheduleDelayReject only fires at maxLocality == ANY with pending-but-unschedulable tasks. BIN_PACK just moves the repeated probing of a single offer from the outer loop into the inner loop; the final no-launch probe is what can set the reject in both cases, so noDelayScheduleRejects reaches the same result.
• Locality is preserved — BIN_PACK only applies at NO_PREF/ANY, and the PROCESS_LOCAL/NODE_LOCAL tests confirm it.

Two non-blocking notes:

1. Executor IDs are sorted lexicographically (sortBy(_.executorId) on the string), so e.g. "10" sorts before "2". That's fine for correctness — bin-packing only needs a stable order — and the doc does say "lexicographic executor ID order," so it's deliberate. Just flagging that since executor IDs are numeric-looking, the ordering can read as surprising; a half-sentence noting it's lexicographic purely for determinism would help the next reader.
2. Under BIN_PACK the inner loop already fills each offer to exhaustion, so the enclosing do/while(launchedTaskAtCurrentMaxLocality) will make one extra scan that launches nothing before exiting. Harmless, just a minor redundant pass — not worth changing.

Docs and test coverage are thorough. LGTM.

[pan3793]

this one looks very similar to #56957

[cloud-fan]

So the motivation seems to make auto-scale more efficient? Can we have a proper design discussion for the end to end auto-scale workflow? Why changing the scheduler is the best design choice? Just for example, one idea can be: when the auto scaler predicate an upcoming idle time window, it can pick some executors and mark them as decommissioned, then scheduler will not assign new tasks to these executors, and auto scaler can kill them after current tasks are done.