[GLUTEN-12616][CORE] Guard SparkResourceUtil.getTaskSlots against non-positive task cpus#12617
[GLUTEN-12616][CORE] Guard SparkResourceUtil.getTaskSlots against non-positive task cpus#12617LuciferYang wants to merge 2 commits into
Conversation
…-positive task cpus getTaskSlots computed executorCores / taskCores with no guard. GlutenDriverPlugin.init reads the slot count and divides by it (and four other callers use it as a denominator), so two invalid configs crash there with an opaque ArithmeticException: - spark.task.cpus > spark.executor.cores makes the quotient 0, so a caller's offHeapSize / taskSlots throws. - spark.task.cpus = 0 makes getTaskSlots itself throw; it reads the value with raw conf.getInt, which bypasses Spark's CPUS_PER_TASK.checkValue(_ > 0). The plugin runs before createTaskScheduler, so Gluten throws before Spark's own validation (validateTaskCpusLargeEnough / CPUS_PER_TASK) can report the real misconfiguration with a clear message. Return a single slot for taskCores <= 0 and floor the quotient at 1 otherwise, deferring to Spark for the error text. Add SparkResourceUtilSuite covering both guarded branches and the normal paths.
|
Run Gluten Clickhouse CI on x86 |
There was a problem hiding this comment.
Pull request overview
This PR hardens SparkResourceUtil.getTaskSlots so Gluten plugin initialization never triggers a divide-by-zero or returns 0 task slots for invalid spark.task.cpus settings, letting Spark’s own validation surface the proper user-facing error.
Changes:
- Guard
getTaskSlotsagainstspark.task.cpus <= 0by returning 1. - Floor
executorCores / taskCoresat 1 to avoid returning 0 whenspark.task.cpus > executor cores. - Add a new
SparkResourceUtilSuitecovering the guarded behaviors and normal cases.
Reviewed changes
Copilot reviewed 2 out of 2 changed files in this pull request and generated 1 comment.
| File | Description |
|---|---|
| gluten-core/src/main/scala/org/apache/spark/util/SparkResourceUtil.scala | Adds guards to prevent / by zero and zero-slot results from invalid CPU-per-task configs. |
| gluten-core/src/test/scala/org/apache/spark/util/SparkResourceUtilSuite.scala | Introduces unit tests for getTaskSlots behavior across invalid and typical configurations. |
💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.
| test("getTaskSlots returns one core per slot by default") { | ||
| val conf = new SparkConf(false).set("spark.master", "local[1]") | ||
| assert(SparkResourceUtil.getTaskSlots(conf) == 1) | ||
| } |
There was a problem hiding this comment.
Good catch. local[1] didn't actually distinguish the default behavior since 1 slot comes out regardless of the logic under test. Changed it to local[8] with spark.task.cpus unset, asserting 8 slots, so it now validates one-slot-per-core when task cpus defaults to 1.
…e master The default-behavior test used local[1], which yields 1 slot regardless of the logic under test. Use local[8] with spark.task.cpus unset and assert 8 slots, so it validates one-slot-per-core when task cpus defaults to 1.
|
Run Gluten Clickhouse CI on x86 |
| val executorCores = SparkResourceUtil.getExecutorCores(conf) | ||
| val taskCores = conf.getInt("spark.task.cpus", 1) | ||
| executorCores / taskCores | ||
| if (taskCores <= 0) { |
There was a problem hiding this comment.
the task cores has been checked while setting the configs, do we need this check again?
What changes were proposed in this pull request?
SparkResourceUtil.getTaskSlotscomputedexecutorCores / taskCoreswith no guard.GlutenDriverPlugin.initreads the slot count and divides by it, and four other callers (MemoryTargets,ColumnarShuffleWriter, and the Celeborn and Uniffle writers) use it as a denominator too. Two invalid configs crash there with an opaque error.When
spark.task.cpus > spark.executor.cores, the integer division gives 0, so a caller'soffHeapSize / taskSlotsthrowsArithmeticException: / by zero. Whenspark.task.cpus = 0,getTaskSlotsitself throws, because it reads the value with rawconf.getInt, which skips Spark'sCPUS_PER_TASK.checkValue(_ > 0)(that check runs only on the typedconf.get(CPUS_PER_TASK)).The plugin runs at the
PluginContainerstep inSparkContext, beforecreateTaskScheduler. So Gluten throws before Spark's own validation (validateTaskCpusLargeEnoughand theCPUS_PER_TASKcheck) can report the real problem, and the user gets a GlutenArithmeticExceptionstack trace instead of Spark's clear message. Both configs are invalid and Spark rejects them regardless, so the job does not start either way. The point is that Gluten should not misattribute the failure to itself with a divide-by-zero.getTaskSlotsnow returns a single slot whentaskCores <= 0, and floors the quotient at 1 otherwise, so it never divides by a non-positive value and never returns 0. Gluten then defers to Spark for the error message.How was this patch tested?
Added
SparkResourceUtilSuitewith four tests:task.cpusgreater than executor cores floors to one slot,task.cpus=0does not divide by zero,8 / 2gives 4, and the default is one slot per core. The first two fail on the current code (they hit the two divide-by-zero paths) and pass after the fix.Closes #12616