You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "andygrove (via GitHub)" <gi...@apache.org> on 2026/09/29 01:00:36 UTC
[PR] fix: size the native runtime for standalone executors without spark.executor.cores [datafusion-comet]
andygrove opened a new pull request, #6368:
URL: https://github.com/apache/datafusion-comet/pull/6368
## Which issue does this PR close?
Closes #6292.
## Rationale for this change
`CometExecIterator.serializeCometSQLConfs` passes the core count that sizes the native Tokio runtime to native code as `spark.executor.cores`. `numDriverOrExecutorCores` took it from the thread count of a `local[N]` master, or for any other master from `spark.executor.cores`, falling back to 1.
A standalone (`spark://`) executor without `spark.executor.cores` takes every core its worker offers and runs that many tasks at once, but its `SparkConf` has no `spark.executor.cores`. `CoarseGrainedExecutorBackend.run` builds the executor's conf from the driver's `spark.*` properties and only sets `spark.executor.cores` for a non-default resource profile (and Spark 3.4 never sets it). So every such executor got a single Tokio worker, and so did `local-cluster` executors. Native plans that read no input from the JVM run entirely on Tokio workers, so all of the executor's concurrent tasks shared one thread (13.0 s instead of 5.5 s in the issue's measurement). On `main` this can also deadlock under memory pressure, when that one worker parks in Spark's `ExecutionMemoryPool.acquireMemory`.
YARN and Kubernetes were not affected: their default for `spark.executor.cores` is 1, which is also the executor's number of task slots.
## What changes are included in this PR?
- A new pure function, `CometExecIterator.numDriverOrExecutorCores(master, executorCores, availableProcessors)`, resolves the core count from the first of these that applies:
- local mode: the master's thread count (unchanged)
- `spark.executor.cores`, when set (unchanged)
- `local-cluster[N, C, M]`: `C`, the cores each worker offers
- `spark://`: `Runtime.getRuntime.availableProcessors()`. A standalone worker offers all of its machine's processors by default (`WorkerArguments.inferDefaultCores`), and without `spark.executor.cores` the master launches one executor per worker with all of its free cores (`Master.scheduleExecutorsOnWorkers`).
- `yarn` and `k8s://`: 1, their default
- any other master (for example `mesos://` on Spark 3.x, or an external cluster manager): unknown
- For an unknown master, the `SparkConf` wrapper falls back to 1 as before and logs a warning once per JVM that points at `spark.executor.cores` and `COMET_WORKER_THREADS`. The warning is limited to this case rather than every count of 1 on a cluster, because one worker is right for a one-core executor (an explicit `spark.executor.cores=1`, or the YARN and Kubernetes default), and warning on each of those executors would be noise.
- `spark.task.cpus` needs no handling: every task takes that many cores, so the core count is at least the number of task slots.
- `COMET_WORKER_THREADS` still overrides the count; the native side is unchanged.
- The "Configuring Tokio Runtime" section of the tuning guide describes the new resolution and what happens when an executor runs more tasks at once than it has worker threads.
A standalone worker started with a core count other than its machine's (`SPARK_WORKER_CORES`) still gets the machine's processor count. Fewer offered cores only leave spare threads. For more, `spark.executor.cores` or `COMET_WORKER_THREADS` sets the count, as the tuning guide says.
Open PR #6300 rewrites the worker-thread bullet in `docs/source/contributor-guide/development.md` to say that Comet uses one worker when `spark.executor.cores` is not set outside local mode. This PR leaves that file alone, so the bullet will need a follow-up once both have landed. Related to #6261, which reduces the deadlock risk independently of this change.
## How are these changes tested?
- New test `the native runtime gets a worker thread for every core the executor runs tasks on` in `CometExecIteratorLifecycleSuite` covers `local`, `local[N]`, `local[*]` and `local[N, F]` (including `local[N]` with `spark.executor.cores` set, which local mode ignores), `local-cluster` and `spark://` (single and HA master URLs) with and without `spark.executor.cores`, YARN and Kubernetes with and without it, and an unknown master. The test calls a function this PR introduces, so there is no red run against the old code, which resolved `spark://` and `local-cluster` without `spark.executor.cores` to 1.
- On the default Spark 4.1 profile, ran the new test and the two `SQLConf serde` tests in `CometExecSuite`, which go through the `SparkConf` wrapper. All passed.
- Checked the Spark behavior the fix relies on against the 3.4.3, 4.1.3 and 4.2.0 sources: executors receive every `spark.*` property of the driver, including `spark.master` (`CoarseGrainedSchedulerBackend.sparkProperties`); `spark.executor.cores` is set on the executor only for a non-default resource profile, and never on 3.4; a standalone executor gets its cores through `--cores`; and the `local-cluster` master URL format.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: size the native runtime for standalone executors without spark.executor.cores [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove merged PR #6368:
URL: https://github.com/apache/datafusion-comet/pull/6368
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: size the native runtime for standalone executors without spark.executor.cores [datafusion-comet]
Posted by "comphead (via GitHub)" <gi...@apache.org>.
comphead commented on code in PR #6368:
URL: https://github.com/apache/datafusion-comet/pull/6368#discussion_r4146856184
##########
spark/src/main/scala/org/apache/comet/CometExecIterator.scala:
##########
@@ -680,26 +680,69 @@ object CometExecIterator extends Logging {
}
}
+ private val unknownCoresWarned = new AtomicBoolean(false)
+
+ /**
+ * The number of cores this JVM runs tasks on, which sizes the native runtime's worker threads.
+ * Falls back to 1, with a warning logged once, if the master does not tell.
+ */
private def numDriverOrExecutorCores(conf: SparkConf): Int = {
- def convertToInt(threads: String): Int = {
- if (threads == "*") Runtime.getRuntime.availableProcessors() else threads.toInt
+ val master = conf.get("spark.master")
+ numDriverOrExecutorCores(
+ master,
+ conf.getOption("spark.executor.cores"),
+ Runtime.getRuntime.availableProcessors()).getOrElse {
+ if (unknownCoresWarned.compareAndSet(false, true)) {
+ logWarning(
+ s"Comet cannot tell how many cores this executor has from spark.master=$master when " +
+ "spark.executor.cores is not set, so its native runtime starts one worker thread. " +
+ "Native plans that read no input from the JVM run on the worker threads, so " +
+ "concurrent tasks share this one. Set spark.executor.cores, or the " +
+ "COMET_WORKER_THREADS environment variable, to the executor's core count. " +
+ s"${CometConf.TUNING_GUIDE}.")
+ }
+ 1
}
+ }
- // If running in local mode, get number of threads from the spark.master setting.
- // See https://spark.apache.org/docs/latest/submitting-applications.html#master-urls
- // for supported formats
+ // Master URL formats, as SparkContext parses them. See
+ // https://spark.apache.org/docs/latest/submitting-applications.html#master-urls
+ // `local[*]` means using all available cores and `local[2]` means using 2 cores.
+ private val LOCAL_N_REGEX = """local\[([0-9]+|\*)\]""".r
+ // `local[num-worker-threads, max-failures]`
+ private val LOCAL_N_FAILURES_REGEX = """local\[([0-9]+|\*)\s*,\s*([0-9]+)\]""".r
+ // `local-cluster[num-workers, cores-per-worker, memory-per-worker-mib]`
+ private val LOCAL_CLUSTER_REGEX =
+ """local-cluster\[\s*([0-9]+)\s*,\s*([0-9]+)\s*,\s*([0-9]+)\s*]""".r
- // `local[*]` means using all available cores and `local[2]` means using 2 cores.
- val LOCAL_N_REGEX = """local\[([0-9]+|\*)\]""".r
- // Also handle format `local[num-worker-threads, max-failures]
- val LOCAL_N_FAILURES_REGEX = """local\[([0-9]+|\*)\s*,\s*([0-9]+)\]""".r
+ /**
+ * The number of cores the driver (in local mode) or an executor runs tasks on, given the master
+ * URL, `spark.executor.cores` and the number of processors available to the JVM. Every task
+ * takes `spark.task.cpus` of these cores, so this is at least the number of tasks that run at
+ * once. None if it cannot be determined.
+ */
+ def numDriverOrExecutorCores(
+ master: String,
+ executorCores: Option[String],
+ availableProcessors: Int): Option[Int] = {
+ def localThreads(threads: String): Int =
+ if (threads == "*") availableProcessors else threads.toInt
- val master = conf.get("spark.master")
master match {
- case "local" => 1
- case LOCAL_N_REGEX(threads) => convertToInt(threads)
- case LOCAL_N_FAILURES_REGEX(threads, _) => convertToInt(threads)
- case _ => conf.get("spark.executor.cores", "1").toInt
+ // Local mode runs tasks on the master's threads and ignores spark.executor.cores.
+ case "local" => Some(1)
+ case LOCAL_N_REGEX(threads) => Some(localThreads(threads))
+ case LOCAL_N_FAILURES_REGEX(threads, _) => Some(localThreads(threads))
+ case _ if executorCores.isDefined => executorCores.map(_.toInt)
+ // Without spark.executor.cores, a standalone executor takes every core its worker offers,
+ // but Spark does not set spark.executor.cores on the executor. A local-cluster worker
+ // offers the master's cores per worker, and a standalone worker offers all of the
+ // machine's processors unless it is started with a different number.
+ case LOCAL_CLUSTER_REGEX(_, coresPerWorker, _) => Some(coresPerWorker.toInt)
+ case _ if master.startsWith("spark://") => Some(availableProcessors)
+ // The default of spark.executor.cores on YARN and Kubernetes.
+ case _ if master == "yarn" || master.startsWith("k8s://") => Some(1)
+ case _ => None
Review Comment:
For an unknown master this still starts a single worker, which is the situation in #6292. For `mesos://` on Spark 3.x, an executor without `spark.executor.cores` takes the cores of the offer (`MesosCoarseGrainedSchedulerBackend.executorCores` in v3.5.8), which is similar to standalone. Would `availableProcessors` be a safer fallback than 1 here? I'd expect a few idle Tokio threads to cost less than starving concurrent tasks, but I haven't measured that or looked at other external cluster managers.
##########
docs/source/user-guide/latest/tuning.md:
##########
@@ -38,11 +38,22 @@ guide is split into the following pages:
## Configuring Tokio Runtime
-Comet uses a global tokio runtime per executor process. By default it starts one worker thread per executor core
-(`spark.executor.cores`, or the thread count of `local[N]` and `local[*]` masters) and allows up to 512 blocking
-threads, which is tokio's default. If `spark.executor.cores` is not set outside local mode, Comet starts a single
-worker thread. These values can be overridden using the environment variables `COMET_WORKER_THREADS` and
-`COMET_MAX_BLOCKING_THREADS`.
+Comet uses a global tokio runtime per executor process. By default it starts one worker thread per core that the
+executor runs tasks on, and allows up to 512 blocking threads, which is tokio's default. Comet takes the number of
+cores from the first of these that applies:
+
+- the thread count of a `local`, `local[N]` or `local[*]` master, in local mode
+- `spark.executor.cores`, when it is set
+- the cores per worker `C` of a `local-cluster[N, C, M]` master
+- the number of processors available to the executor's JVM on a standalone (`spark://`) cluster, since a standalone
+ executor without `spark.executor.cores` takes every core its worker offers, and a worker offers all of its
+ machine's cores unless it is started with a different number (`SPARK_WORKER_CORES`)
+- one on YARN and Kubernetes, which is their default for `spark.executor.cores`
+
+On any other cluster manager, Comet starts a single worker thread when `spark.executor.cores` is not set, and logs a
Review Comment:
Now that #6300 has merged, the worker threads bullet in `docs/source/contributor-guide/development.md` says Comet starts one worker thread when `spark.executor.cores` is not set outside local mode. With this change that is no longer true for `spark://` and `local-cluster`. Would it be worth updating that bullet in this PR so it does not land stale?
##########
spark/src/main/scala/org/apache/comet/CometExecIterator.scala:
##########
@@ -680,26 +680,69 @@ object CometExecIterator extends Logging {
}
}
+ private val unknownCoresWarned = new AtomicBoolean(false)
+
+ /**
+ * The number of cores this JVM runs tasks on, which sizes the native runtime's worker threads.
+ * Falls back to 1, with a warning logged once, if the master does not tell.
+ */
private def numDriverOrExecutorCores(conf: SparkConf): Int = {
- def convertToInt(threads: String): Int = {
- if (threads == "*") Runtime.getRuntime.availableProcessors() else threads.toInt
+ val master = conf.get("spark.master")
+ numDriverOrExecutorCores(
+ master,
+ conf.getOption("spark.executor.cores"),
+ Runtime.getRuntime.availableProcessors()).getOrElse {
+ if (unknownCoresWarned.compareAndSet(false, true)) {
+ logWarning(
+ s"Comet cannot tell how many cores this executor has from spark.master=$master when " +
+ "spark.executor.cores is not set, so its native runtime starts one worker thread. " +
Review Comment:
Nit: the native side prefers `COMET_WORKER_THREADS` when it is set, so this warning, and the single worker it mentions, would be misleading for executors that already followed its advice. Would it make sense to skip the warning when that variable is set?
##########
spark/src/main/scala/org/apache/comet/CometExecIterator.scala:
##########
@@ -680,26 +680,69 @@ object CometExecIterator extends Logging {
}
}
+ private val unknownCoresWarned = new AtomicBoolean(false)
+
+ /**
+ * The number of cores this JVM runs tasks on, which sizes the native runtime's worker threads.
+ * Falls back to 1, with a warning logged once, if the master does not tell.
+ */
private def numDriverOrExecutorCores(conf: SparkConf): Int = {
- def convertToInt(threads: String): Int = {
- if (threads == "*") Runtime.getRuntime.availableProcessors() else threads.toInt
+ val master = conf.get("spark.master")
+ numDriverOrExecutorCores(
+ master,
+ conf.getOption("spark.executor.cores"),
+ Runtime.getRuntime.availableProcessors()).getOrElse {
+ if (unknownCoresWarned.compareAndSet(false, true)) {
+ logWarning(
+ s"Comet cannot tell how many cores this executor has from spark.master=$master when " +
+ "spark.executor.cores is not set, so its native runtime starts one worker thread. " +
+ "Native plans that read no input from the JVM run on the worker threads, so " +
+ "concurrent tasks share this one. Set spark.executor.cores, or the " +
+ "COMET_WORKER_THREADS environment variable, to the executor's core count. " +
+ s"${CometConf.TUNING_GUIDE}.")
+ }
+ 1
}
+ }
- // If running in local mode, get number of threads from the spark.master setting.
- // See https://spark.apache.org/docs/latest/submitting-applications.html#master-urls
- // for supported formats
+ // Master URL formats, as SparkContext parses them. See
+ // https://spark.apache.org/docs/latest/submitting-applications.html#master-urls
+ // `local[*]` means using all available cores and `local[2]` means using 2 cores.
+ private val LOCAL_N_REGEX = """local\[([0-9]+|\*)\]""".r
+ // `local[num-worker-threads, max-failures]`
+ private val LOCAL_N_FAILURES_REGEX = """local\[([0-9]+|\*)\s*,\s*([0-9]+)\]""".r
+ // `local-cluster[num-workers, cores-per-worker, memory-per-worker-mib]`
+ private val LOCAL_CLUSTER_REGEX =
+ """local-cluster\[\s*([0-9]+)\s*,\s*([0-9]+)\s*,\s*([0-9]+)\s*]""".r
- // `local[*]` means using all available cores and `local[2]` means using 2 cores.
- val LOCAL_N_REGEX = """local\[([0-9]+|\*)\]""".r
- // Also handle format `local[num-worker-threads, max-failures]
- val LOCAL_N_FAILURES_REGEX = """local\[([0-9]+|\*)\s*,\s*([0-9]+)\]""".r
+ /**
+ * The number of cores the driver (in local mode) or an executor runs tasks on, given the master
+ * URL, `spark.executor.cores` and the number of processors available to the JVM. Every task
+ * takes `spark.task.cpus` of these cores, so this is at least the number of tasks that run at
+ * once. None if it cannot be determined.
+ */
+ def numDriverOrExecutorCores(
+ master: String,
+ executorCores: Option[String],
+ availableProcessors: Int): Option[Int] = {
+ def localThreads(threads: String): Int =
+ if (threads == "*") availableProcessors else threads.toInt
- val master = conf.get("spark.master")
master match {
- case "local" => 1
- case LOCAL_N_REGEX(threads) => convertToInt(threads)
- case LOCAL_N_FAILURES_REGEX(threads, _) => convertToInt(threads)
- case _ => conf.get("spark.executor.cores", "1").toInt
+ // Local mode runs tasks on the master's threads and ignores spark.executor.cores.
+ case "local" => Some(1)
+ case LOCAL_N_REGEX(threads) => Some(localThreads(threads))
+ case LOCAL_N_FAILURES_REGEX(threads, _) => Some(localThreads(threads))
+ case _ if executorCores.isDefined => executorCores.map(_.toInt)
+ // Without spark.executor.cores, a standalone executor takes every core its worker offers,
+ // but Spark does not set spark.executor.cores on the executor. A local-cluster worker
+ // offers the master's cores per worker, and a standalone worker offers all of the
+ // machine's processors unless it is started with a different number.
+ case LOCAL_CLUSTER_REGEX(_, coresPerWorker, _) => Some(coresPerWorker.toInt)
+ case _ if master.startsWith("spark://") => Some(availableProcessors)
+ // The default of spark.executor.cores on YARN and Kubernetes.
+ case _ if master == "yarn" || master.startsWith("k8s://") => Some(1)
Review Comment:
Nit: once this is rebased, `isContainerSizedFromOverhead(master)` (added in #6375) tests the same `yarn` or `k8s://` condition. Would it make sense to share one check, perhaps under a neutral name, so the two cannot drift apart? The intent differs (container memory versus default cores), so keeping both is also reasonable.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] fix: size the native runtime for standalone executors without spark.executor.cores [datafusion-comet]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #6368:
URL: https://github.com/apache/datafusion-comet/pull/6368#discussion_r4302853072
##########
docs/source/user-guide/latest/tuning.md:
##########
@@ -38,11 +38,22 @@ guide is split into the following pages:
## Configuring Tokio Runtime
-Comet uses a global tokio runtime per executor process. By default it starts one worker thread per executor core
-(`spark.executor.cores`, or the thread count of `local[N]` and `local[*]` masters) and allows up to 512 blocking
-threads, which is tokio's default. If `spark.executor.cores` is not set outside local mode, Comet starts a single
-worker thread. These values can be overridden using the environment variables `COMET_WORKER_THREADS` and
-`COMET_MAX_BLOCKING_THREADS`.
+Comet uses a global tokio runtime per executor process. By default it starts one worker thread per core that the
+executor runs tasks on, and allows up to 512 blocking threads, which is tokio's default. Comet takes the number of
+cores from the first of these that applies:
+
+- the thread count of a `local`, `local[N]` or `local[*]` master, in local mode
+- `spark.executor.cores`, when it is set
+- the cores per worker `C` of a `local-cluster[N, C, M]` master
+- the number of processors available to the executor's JVM on a standalone (`spark://`) cluster, since a standalone
+ executor without `spark.executor.cores` takes every core its worker offers, and a worker offers all of its
+ machine's cores unless it is started with a different number (`SPARK_WORKER_CORES`)
+- one on YARN and Kubernetes, which is their default for `spark.executor.cores`
+
+On any other cluster manager, Comet starts a single worker thread when `spark.executor.cores` is not set, and logs a
Review Comment:
Yes, updated. The bullet now lists each master and points at `CometExecIterator.numDriverOrExecutorCores`.
##########
spark/src/main/scala/org/apache/comet/CometExecIterator.scala:
##########
@@ -680,26 +680,69 @@ object CometExecIterator extends Logging {
}
}
+ private val unknownCoresWarned = new AtomicBoolean(false)
+
+ /**
+ * The number of cores this JVM runs tasks on, which sizes the native runtime's worker threads.
+ * Falls back to 1, with a warning logged once, if the master does not tell.
+ */
private def numDriverOrExecutorCores(conf: SparkConf): Int = {
- def convertToInt(threads: String): Int = {
- if (threads == "*") Runtime.getRuntime.availableProcessors() else threads.toInt
+ val master = conf.get("spark.master")
+ numDriverOrExecutorCores(
+ master,
+ conf.getOption("spark.executor.cores"),
+ Runtime.getRuntime.availableProcessors()).getOrElse {
+ if (unknownCoresWarned.compareAndSet(false, true)) {
+ logWarning(
+ s"Comet cannot tell how many cores this executor has from spark.master=$master when " +
+ "spark.executor.cores is not set, so its native runtime starts one worker thread. " +
+ "Native plans that read no input from the JVM run on the worker threads, so " +
+ "concurrent tasks share this one. Set spark.executor.cores, or the " +
+ "COMET_WORKER_THREADS environment variable, to the executor's core count. " +
+ s"${CometConf.TUNING_GUIDE}.")
+ }
+ 1
}
+ }
- // If running in local mode, get number of threads from the spark.master setting.
- // See https://spark.apache.org/docs/latest/submitting-applications.html#master-urls
- // for supported formats
+ // Master URL formats, as SparkContext parses them. See
+ // https://spark.apache.org/docs/latest/submitting-applications.html#master-urls
+ // `local[*]` means using all available cores and `local[2]` means using 2 cores.
+ private val LOCAL_N_REGEX = """local\[([0-9]+|\*)\]""".r
+ // `local[num-worker-threads, max-failures]`
+ private val LOCAL_N_FAILURES_REGEX = """local\[([0-9]+|\*)\s*,\s*([0-9]+)\]""".r
+ // `local-cluster[num-workers, cores-per-worker, memory-per-worker-mib]`
+ private val LOCAL_CLUSTER_REGEX =
+ """local-cluster\[\s*([0-9]+)\s*,\s*([0-9]+)\s*,\s*([0-9]+)\s*]""".r
- // `local[*]` means using all available cores and `local[2]` means using 2 cores.
- val LOCAL_N_REGEX = """local\[([0-9]+|\*)\]""".r
- // Also handle format `local[num-worker-threads, max-failures]
- val LOCAL_N_FAILURES_REGEX = """local\[([0-9]+|\*)\s*,\s*([0-9]+)\]""".r
+ /**
+ * The number of cores the driver (in local mode) or an executor runs tasks on, given the master
+ * URL, `spark.executor.cores` and the number of processors available to the JVM. Every task
+ * takes `spark.task.cpus` of these cores, so this is at least the number of tasks that run at
+ * once. None if it cannot be determined.
+ */
+ def numDriverOrExecutorCores(
+ master: String,
+ executorCores: Option[String],
+ availableProcessors: Int): Option[Int] = {
+ def localThreads(threads: String): Int =
+ if (threads == "*") availableProcessors else threads.toInt
- val master = conf.get("spark.master")
master match {
- case "local" => 1
- case LOCAL_N_REGEX(threads) => convertToInt(threads)
- case LOCAL_N_FAILURES_REGEX(threads, _) => convertToInt(threads)
- case _ => conf.get("spark.executor.cores", "1").toInt
+ // Local mode runs tasks on the master's threads and ignores spark.executor.cores.
+ case "local" => Some(1)
+ case LOCAL_N_REGEX(threads) => Some(localThreads(threads))
+ case LOCAL_N_FAILURES_REGEX(threads, _) => Some(localThreads(threads))
+ case _ if executorCores.isDefined => executorCores.map(_.toInt)
+ // Without spark.executor.cores, a standalone executor takes every core its worker offers,
+ // but Spark does not set spark.executor.cores on the executor. A local-cluster worker
+ // offers the master's cores per worker, and a standalone worker offers all of the
+ // machine's processors unless it is started with a different number.
+ case LOCAL_CLUSTER_REGEX(_, coresPerWorker, _) => Some(coresPerWorker.toInt)
+ case _ if master.startsWith("spark://") => Some(availableProcessors)
+ // The default of spark.executor.cores on YARN and Kubernetes.
+ case _ if master == "yarn" || master.startsWith("k8s://") => Some(1)
+ case _ => None
Review Comment:
Agreed, and I checked the Mesos case: on Spark 3.x an executor without `spark.executor.cores` takes the cores of the offer, as a standalone one does. An unknown master now falls back to `availableProcessors` and still logs the warning once. The tuning guide says why: a few idle workers cost less than too few.
##########
spark/src/main/scala/org/apache/comet/CometExecIterator.scala:
##########
@@ -680,26 +680,69 @@ object CometExecIterator extends Logging {
}
}
+ private val unknownCoresWarned = new AtomicBoolean(false)
+
+ /**
+ * The number of cores this JVM runs tasks on, which sizes the native runtime's worker threads.
+ * Falls back to 1, with a warning logged once, if the master does not tell.
+ */
private def numDriverOrExecutorCores(conf: SparkConf): Int = {
- def convertToInt(threads: String): Int = {
- if (threads == "*") Runtime.getRuntime.availableProcessors() else threads.toInt
+ val master = conf.get("spark.master")
+ numDriverOrExecutorCores(
+ master,
+ conf.getOption("spark.executor.cores"),
+ Runtime.getRuntime.availableProcessors()).getOrElse {
+ if (unknownCoresWarned.compareAndSet(false, true)) {
+ logWarning(
+ s"Comet cannot tell how many cores this executor has from spark.master=$master when " +
+ "spark.executor.cores is not set, so its native runtime starts one worker thread. " +
Review Comment:
Good catch. The warning is now skipped when `COMET_WORKER_THREADS` is set to a number, which is the same condition the native side uses to prefer it.
##########
spark/src/main/scala/org/apache/comet/CometExecIterator.scala:
##########
@@ -680,26 +680,69 @@ object CometExecIterator extends Logging {
}
}
+ private val unknownCoresWarned = new AtomicBoolean(false)
+
+ /**
+ * The number of cores this JVM runs tasks on, which sizes the native runtime's worker threads.
+ * Falls back to 1, with a warning logged once, if the master does not tell.
+ */
private def numDriverOrExecutorCores(conf: SparkConf): Int = {
- def convertToInt(threads: String): Int = {
- if (threads == "*") Runtime.getRuntime.availableProcessors() else threads.toInt
+ val master = conf.get("spark.master")
+ numDriverOrExecutorCores(
+ master,
+ conf.getOption("spark.executor.cores"),
+ Runtime.getRuntime.availableProcessors()).getOrElse {
+ if (unknownCoresWarned.compareAndSet(false, true)) {
+ logWarning(
+ s"Comet cannot tell how many cores this executor has from spark.master=$master when " +
+ "spark.executor.cores is not set, so its native runtime starts one worker thread. " +
+ "Native plans that read no input from the JVM run on the worker threads, so " +
+ "concurrent tasks share this one. Set spark.executor.cores, or the " +
+ "COMET_WORKER_THREADS environment variable, to the executor's core count. " +
+ s"${CometConf.TUNING_GUIDE}.")
+ }
+ 1
}
+ }
- // If running in local mode, get number of threads from the spark.master setting.
- // See https://spark.apache.org/docs/latest/submitting-applications.html#master-urls
- // for supported formats
+ // Master URL formats, as SparkContext parses them. See
+ // https://spark.apache.org/docs/latest/submitting-applications.html#master-urls
+ // `local[*]` means using all available cores and `local[2]` means using 2 cores.
+ private val LOCAL_N_REGEX = """local\[([0-9]+|\*)\]""".r
+ // `local[num-worker-threads, max-failures]`
+ private val LOCAL_N_FAILURES_REGEX = """local\[([0-9]+|\*)\s*,\s*([0-9]+)\]""".r
+ // `local-cluster[num-workers, cores-per-worker, memory-per-worker-mib]`
+ private val LOCAL_CLUSTER_REGEX =
+ """local-cluster\[\s*([0-9]+)\s*,\s*([0-9]+)\s*,\s*([0-9]+)\s*]""".r
- // `local[*]` means using all available cores and `local[2]` means using 2 cores.
- val LOCAL_N_REGEX = """local\[([0-9]+|\*)\]""".r
- // Also handle format `local[num-worker-threads, max-failures]
- val LOCAL_N_FAILURES_REGEX = """local\[([0-9]+|\*)\s*,\s*([0-9]+)\]""".r
+ /**
+ * The number of cores the driver (in local mode) or an executor runs tasks on, given the master
+ * URL, `spark.executor.cores` and the number of processors available to the JVM. Every task
+ * takes `spark.task.cpus` of these cores, so this is at least the number of tasks that run at
+ * once. None if it cannot be determined.
+ */
+ def numDriverOrExecutorCores(
+ master: String,
+ executorCores: Option[String],
+ availableProcessors: Int): Option[Int] = {
+ def localThreads(threads: String): Int =
+ if (threads == "*") availableProcessors else threads.toInt
- val master = conf.get("spark.master")
master match {
- case "local" => 1
- case LOCAL_N_REGEX(threads) => convertToInt(threads)
- case LOCAL_N_FAILURES_REGEX(threads, _) => convertToInt(threads)
- case _ => conf.get("spark.executor.cores", "1").toInt
+ // Local mode runs tasks on the master's threads and ignores spark.executor.cores.
+ case "local" => Some(1)
+ case LOCAL_N_REGEX(threads) => Some(localThreads(threads))
+ case LOCAL_N_FAILURES_REGEX(threads, _) => Some(localThreads(threads))
+ case _ if executorCores.isDefined => executorCores.map(_.toInt)
+ // Without spark.executor.cores, a standalone executor takes every core its worker offers,
+ // but Spark does not set spark.executor.cores on the executor. A local-cluster worker
+ // offers the master's cores per worker, and a standalone worker offers all of the
+ // machine's processors unless it is started with a different number.
+ case LOCAL_CLUSTER_REGEX(_, coresPerWorker, _) => Some(coresPerWorker.toInt)
+ case _ if master.startsWith("spark://") => Some(availableProcessors)
+ // The default of spark.executor.cores on YARN and Kubernetes.
+ case _ if master == "yarn" || master.startsWith("k8s://") => Some(1)
Review Comment:
Shared. Both now call a private `isYarnOrKubernetes`, and `isContainerSizedFromOverhead` keeps its name and its doc comment for the memory-overhead use.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org