From fcb60bb48f82f38f67557238d008b26988890166 Mon Sep 17 00:00:00 2001 From: Warren Zhu Date: Sun, 18 Oct 2026 05:30:39 +0000 Subject: [PATCH] Default auron.tokio.worker.threads.per.cpu to 1 to size per-task runtimes by task CPUs Which issue does this PR close? Closes # (to be filed: per-task Tokio runtime defaults to all host cores, causing thread explosion and slower short tasks) Rationale for this change Each native task builds its own multi-threaded Tokio runtime (native-engine/auron/src/rt.rs) sized auron.tokio.worker.threads.per.cpu * spark.task.cpus. The current default of 0 leaves the worker count unset, so Tokio uses every host core for every task, regardless of spark.task.cpus. Each worker thread also attaches to the JVM on start to set the classloader and task context. When a task finishes, its runtime is torn down with shutdown_background(), which does not wait for the threads to exit. On many-core hosts with many short tasks, threads from finished tasks accumulate faster than they exit. That wastes CPU and memory, adds per-task thread start/stop overhead, and can exhaust the OS thread limit: - TPC-H SF10 Q4 with 3000 shuffle partitions on a 64-core host (local[16]): the JVM peaked at ~74,000 threads (auron-native-stage-*), and the user's total thread count reached 131,041 of a 131,072 limit. - One run failed with "OS can't spawn worker thread: Resource temporarily unavailable (os error 11)". Another hung: all final-stage tasks were blocked in JniBridge.nextBatch while native threads sat idle. What changes are included in this PR? - AuronConfiguration.TOKIO_WORKER_THREADS_PER_CPU: default 0 -> 1, so a task's runtime gets spark.task.cpus worker threads. The description now explains that 0 means "all host cores for every task". - rt.rs: the native fallbacks used when the conf can't be read become 1 instead of 0, consistent with the JVM-side defaults. Users who relied on the old behavior can set spark.auron.tokio.worker.threads.per.cpu=0. Are there any user-facing changes? Yes. The default number of Tokio worker threads per task changes from "all host cores" to spark.task.cpus. There are no API changes. How was this patch tested? 64-core host, Spark 3.5.8 / Scala 2.13, local[16], release build, using the dev/auron-it TPC-DS harness at SF10. Both configs used the same jar and differ only in this setting; runs were interleaved (old, new, old, new) and per-query medians compared. TPC-DS SF10, 101 queries per.cpu=0 (old) per.cpu=1 (new) total time 1010.7 s 919.6 s (-9%) total excluding q72 ~412 s ~308 s (-25%) median per-query ratio 0.72 geomean per-query ratio 0.663 queries >15% faster/slower 31 / 0 q72 (~600 s, dominated by spill I/O) is unchanged and makes up ~60% of the total. Same-config run-to-run geomean noise was 1-4%. Largest gains: q77 10.8 s -> 2.0 s, q76 2.2 s -> 0.6 s, q83 2.0 s -> 0.7 s, q64 11.9 s -> 6.7 s. Short queries gain most, because the saved cost is per task. Thread usage on TPC-H SF10 Q4 with 3000 shuffle partitions: ~74,000 JVM threads with the old default (crash and hang observed) vs a peak of 227 with the new default, and the query completes in 47 s. Correctness: a --result-check run with the new default matched vanilla Spark on 99/103 queries. The other four fail identically with the old default: q14a/q14b (existing Scala 2.13 ClassCastException, #2386) and q65/q71 (same rows, different order, because of ties in their ORDER BY). cargo fmt --check and spotless:check (auron-core) pass. Was this patch authored or co-authored using generative AI tooling? Yes. Generated-by: Claude Opus 5.5 --- .../apache/auron/configuration/AuronConfiguration.java | 8 +++++--- native-engine/auron/src/rt.rs | 4 ++-- 2 files changed, 7 insertions(+), 5 deletions(-) diff --git a/auron-core/src/main/java/org/apache/auron/configuration/AuronConfiguration.java b/auron-core/src/main/java/org/apache/auron/configuration/AuronConfiguration.java index 3cd06d26f..6f22e2b69 100644 --- a/auron-core/src/main/java/org/apache/auron/configuration/AuronConfiguration.java +++ b/auron-core/src/main/java/org/apache/auron/configuration/AuronConfiguration.java @@ -43,9 +43,11 @@ public abstract class AuronConfiguration { .withKey("auron.tokio.worker.threads.per.cpu") .withCategory("Runtime Configuration") .withDescription( - "Number of Tokio worker threads to create per CPU core (spark.task.cpus). Set to 0 for automatic detection " - + "based on available CPU cores. This setting controls the thread pool size for Tokio-based asynchronous operations.") - .withDefaultValue(0); + "Number of Tokio worker threads to create per CPU core (spark.task.cpus). Each task creates its own " + + "Tokio runtime, so the default of 1 sizes it to the task's CPU allocation. Set to 0 to use " + + "all available CPU cores of the host for every task, which can create a very large number " + + "of threads on many-core machines.") + .withDefaultValue(1); public static final ConfigOption SUGGESTED_BATCH_MEM_SIZE = new ConfigOption<>(Integer.class) .withKey("auron.suggested.batch.memSize") diff --git a/native-engine/auron/src/rt.rs b/native-engine/auron/src/rt.rs index 932ae3136..4a8dad86c 100644 --- a/native-engine/auron/src/rt.rs +++ b/native-engine/auron/src/rt.rs @@ -105,8 +105,8 @@ impl NativeExecutionRuntime { ); let num_worker_threads = { - let worker_threads_per_cpu = TOKIO_WORKER_THREADS_PER_CPU.value().unwrap_or(0); - let task_cpus = TASK_CPUS.value().unwrap_or(0); + let worker_threads_per_cpu = TOKIO_WORKER_THREADS_PER_CPU.value().unwrap_or(1); + let task_cpus = TASK_CPUS.value().unwrap_or(1); worker_threads_per_cpu * task_cpus };