limit concurrent_task to executor_cores

ballista/executor/executor_config_spec.toml

@@ -75,7 +75,6 @@ doc = "Directory for temporary IPC files"
 abbr = "c"
 name = "concurrent_tasks"
 type = "usize"
-default = "0" # defaults to all available cores if left as zero
 doc = "Max concurrent tasks."
 
 [[param]]
@@ -167,6 +166,7 @@ doc = "The number of worker threads for the runtime of caching. Default: 2"
 default = "2"
 
 [[param]]
+abbr = "e"
 name = "executor_cores"
 type = "usize"
 doc = "The number of worker threads. Default: number of available cores"

ballista/executor/src/bin/main.rs

@@ -37,14 +37,10 @@ fn main() -> Result<()> {
         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)
+        .worker_threads(opt.executor_cores_or_default())
         .build()
         .unwrap();
 

ballista/executor/src/config.rs

@@ -23,6 +23,15 @@ use crate::executor_process::ExecutorProcessConfig;
 // #[allow(clippy::all)] to silence clippy warnings from the generated code
 include!(concat!(env!("OUT_DIR"), "/executor_configure_me_config.rs"));
 
+impl Config {
+    /// returns executor cores if setup or number of available
+    /// cpu cores
+    pub fn executor_cores_or_default(&self) -> usize {
+        self.executor_cores
+            .unwrap_or_else(|| std::thread::available_parallelism().unwrap().get())
+    }
+}
+
 impl TryFrom<Config> for ExecutorProcessConfig {
     type Error = BallistaError;
 
@@ -35,6 +44,8 @@ impl TryFrom<Config> for ExecutorProcessConfig {
             opt.bind_port
         );
 
+        let concurrent_tasks = opt.executor_cores_or_default();
+
         Ok(ExecutorProcessConfig {
             special_mod_log_level: opt.log_level_setting,
             external_host: opt.external_host,
@@ -44,7 +55,7 @@ impl TryFrom<Config> for ExecutorProcessConfig {
             scheduler_host: opt.scheduler_host,
             scheduler_port: opt.scheduler_port,
             scheduler_connect_timeout_seconds: opt.scheduler_connect_timeout_seconds,
-            concurrent_tasks: opt.concurrent_tasks,
+            concurrent_tasks,
             task_scheduling_policy: opt.task_scheduling_policy,
             work_dir: opt.work_dir,
             log_dir: opt.log_dir,