feat: Add option to limit number of executor worker threads

this option is similar to `spark.executor.cores`
it sets number of tokio worker threads.

ballista/executor/Cargo.toml

@@ -51,13 +51,7 @@ log = { workspace = true }
 mimalloc = { workspace = true, default-features = false, optional = true }
 parking_lot = { workspace = true }
 tempfile = { workspace = true }
-tokio = { workspace = true, features = [
-    "macros",
-    "rt",
-    "rt-multi-thread",
-    "parking_lot",
-    "signal",
-] }
+tokio = { workspace = true, features = ["full"] }
 tokio-stream = { workspace = true, features = ["net"] }
 tonic = { workspace = true }
 tracing = { workspace = true }

ballista/executor/executor_config_spec.toml

@@ -164,4 +164,9 @@ default = "1073741824"
 name = "cache_io_concurrency"
 type = "u32"
 doc = "The number of worker threads for the runtime of caching. Default: 2"
-default = "2"
+default = "2"
+
+[[param]]
+name = "executor_cores"
+type = "usize"
+doc = "The number of worker threads. Default: number of available cores"

ballista/executor/src/bin/main.rs

@@ -32,13 +32,26 @@ use tracing_subscriber::EnvFilter;
 #[global_allocator]
 static GLOBAL: mimalloc::MiMalloc = mimalloc::MiMalloc;
 
-#[tokio::main]
-async fn main() -> Result<()> {
-    // parse command-line arguments
+fn main() -> Result<()> {
     let (opt, _remaining_args) =
         Config::including_optional_config_files(&["/etc/ballista/executor.toml"])
             .unwrap_or_exit();
 
+    let executor_cores = opt
+        .executor_cores
+        .unwrap_or_else(|| std::thread::available_parallelism().unwrap().get());
+
+    let runtime = tokio::runtime::Builder::new_multi_thread()
+        .enable_all()
+        .thread_name("ballista_executor")
+        .worker_threads(executor_cores)
+        .build()
+        .unwrap();
+
+    runtime.block_on(inner(opt))
+}
+
+async fn inner(opt: Config) -> Result<()> {
     if opt.version {
         print_version();
         std::process::exit(0);

ballista/scheduler/src/bin/main.rs

@@ -31,6 +31,7 @@ fn main() -> Result<()> {
     let runtime = tokio::runtime::Builder::new_multi_thread()
         .enable_io()
         .enable_time()
+        .thread_name("ballista_scheduler")
         .thread_stack_size(32 * 1024 * 1024) // 32MB
         .build()
         .unwrap();