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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
