perf(runtime): Use all available parallelism

Performance work showed a drop-off in performance scaling when
concurrency exceeds 16. Which is the number of threads we were giving
tokio. Now we use all the cores on the machine which is the tokio default
(and also what Go does).

Also ensure in most cases we only use a single tokio runtime. I still contend
the secondary runtime is an unnecessary complication, but this way we preserve
optionality (in case we choose to pin the secondary runtime to a single
core), but also simplify the performance characteristics.

Author: grahamking

launch/dynamo-run/src/main.rs

@@ -62,8 +62,9 @@ fn main() -> anyhow::Result<()> {
 
     // max_worker_threads and max_blocking_threads from env vars or config file.
     let rt_config = dynamo_runtime::RuntimeConfig::from_settings()?;
+    tracing::debug!("Runtime config: {rt_config}");
 
-    // One per process. Wraps a Runtime with holds two tokio runtimes.
+    // One per process. Wraps a Runtime with holds one or two tokio runtimes.
     let worker = dynamo_runtime::Worker::from_config(rt_config)?;
 
     worker.execute(wrapper)

lib/runtime/src/config.rs

@@ -1,17 +1,5 @@
 // SPDX-FileCopyrightText: Copyright (c) 2024-2025 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
 // SPDX-License-Identifier: Apache-2.0
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-// http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
 
 use super::Result;
 use derive_builder::Builder;
@@ -20,6 +8,7 @@ use figment::{
     Figment,
 };
 use serde::{Deserialize, Serialize};
+use std::fmt;
 use validator::Validate;
 
 /// Default HTTP server host
@@ -66,35 +55,58 @@ impl Default for WorkerConfig {
 pub struct RuntimeConfig {
     /// Number of async worker threads
     /// If set to 1, the runtime will run in single-threaded mode
+    /// Set this at runtime with environment variable DYN_RUNTIME_NUM_WORKER_THREADS. Defaults to
+    /// number of cores.
     #[validate(range(min = 1))]
-    #[builder(default = "16")]
     #[builder_field_attr(serde(skip_serializing_if = "Option::is_none"))]
-    pub num_worker_threads: usize,
+    pub num_worker_threads: Option<usize>,
 
     /// Maximum number of blocking threads
     /// Blocking threads are used for blocking operations, this value must be greater than 0.
+    /// Set this at runtime with environment variable DYN_RUNTIME_MAX_BLOCKING_THREADS. Defaults to
+    /// 512.
     #[validate(range(min = 1))]
     #[builder(default = "512")]
     #[builder_field_attr(serde(skip_serializing_if = "Option::is_none"))]
     pub max_blocking_threads: usize,
 
     /// HTTP server host for health and metrics endpoints
+    /// Set this at runtime with environment variable DYN_RUNTIME_HTTP_SERVER_HOST
     #[builder(default = "DEFAULT_HTTP_SERVER_HOST.to_string()")]
     #[builder_field_attr(serde(skip_serializing_if = "Option::is_none"))]
     pub http_server_host: String,
 
     /// HTTP server port for health and metrics endpoints
     /// If set to 0, the system will assign a random available port
+    /// Set this at runtime with environment variable DYN_RUNTIME_HTTP_SERVER_PORT
     #[builder(default = "DEFAULT_HTTP_SERVER_PORT")]
     #[builder_field_attr(serde(skip_serializing_if = "Option::is_none"))]
     pub http_server_port: u16,
 
     /// Health and metrics HTTP server enabled
+    /// Set this at runtime with environment variable DYN_RUNTIME_HTTP_ENABLED
     #[builder(default = "false")]
     #[builder_field_attr(serde(skip_serializing_if = "Option::is_none"))]
     pub http_enabled: bool,
 }
 
+impl fmt::Display for RuntimeConfig {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        // If None, it defaults to "number of cores", so we indicate that.
+        match self.num_worker_threads {
+            Some(val) => write!(f, "num_worker_threads={val}, ")?,
+            None => write!(f, "num_worker_threads=default (num_cores), ")?,
+        }
+
+        write!(f, "max_blocking_threads={}, ", self.max_blocking_threads)?;
+        write!(f, "http_server_host={}, ", self.http_server_host)?;
+        write!(f, "http_server_port={}, ", self.http_server_port)?;
+        write!(f, "http_enabled={}", self.http_enabled)?;
+
+        Ok(())
+    }
+}
+
 impl RuntimeConfig {
     pub fn builder() -> RuntimeConfigBuilder {
         RuntimeConfigBuilder::default()
@@ -138,7 +150,7 @@ impl RuntimeConfig {
 
     pub fn single_threaded() -> Self {
         RuntimeConfig {
-            num_worker_threads: 1,
+            num_worker_threads: Some(1),
             max_blocking_threads: 1,
             http_server_host: DEFAULT_HTTP_SERVER_HOST.to_string(),
             http_server_port: DEFAULT_HTTP_SERVER_PORT,
@@ -147,20 +159,24 @@ impl RuntimeConfig {
     }
 
     /// Create a new default runtime configuration
-    pub(crate) fn create_runtime(&self) -> Result<tokio::runtime::Runtime> {
-        Ok(tokio::runtime::Builder::new_multi_thread()
-            .worker_threads(self.num_worker_threads)
+    pub(crate) fn create_runtime(&self) -> std::io::Result<tokio::runtime::Runtime> {
+        tokio::runtime::Builder::new_multi_thread()
+            .worker_threads(
+                self.num_worker_threads
+                    .unwrap_or_else(|| std::thread::available_parallelism().unwrap().get()),
+            )
             .max_blocking_threads(self.max_blocking_threads)
             .enable_all()
-            .build()?)
+            .build()
     }
 }
 
 impl Default for RuntimeConfig {
     fn default() -> Self {
+        let num_cores = std::thread::available_parallelism().unwrap().get();
         Self {
-            num_worker_threads: 16,
-            max_blocking_threads: 16,
+            num_worker_threads: Some(num_cores),
+            max_blocking_threads: num_cores,
             http_server_host: DEFAULT_HTTP_SERVER_HOST.to_string(),
             http_server_port: DEFAULT_HTTP_SERVER_PORT,
             http_enabled: false,
@@ -240,7 +256,7 @@ mod tests {
             ],
             || {
                 let config = RuntimeConfig::from_settings()?;
-                assert_eq!(config.num_worker_threads, 24);
+                assert_eq!(config.num_worker_threads, Some(24));
                 assert_eq!(config.max_blocking_threads, 32);
                 Ok(())
             },

lib/runtime/src/runtime.rs

@@ -47,6 +47,7 @@ impl Runtime {
         let secondary = match secondary {
             Some(secondary) => secondary,
             None => {
+                tracing::debug!("Created secondary runtime with single thread");
                 RuntimeType::Shared(Arc::new(RuntimeConfig::single_threaded().create_runtime()?))
             }
         };
@@ -60,26 +61,26 @@ impl Runtime {
     }
 
     pub fn from_current() -> Result<Runtime> {
-        let handle = tokio::runtime::Handle::current();
-        let primary = RuntimeType::External(handle.clone());
-        let secondary = RuntimeType::External(handle);
-        Runtime::new(primary, Some(secondary))
+        Runtime::from_handle(tokio::runtime::Handle::current())
     }
 
     pub fn from_handle(handle: tokio::runtime::Handle) -> Result<Runtime> {
-        let runtime = RuntimeType::External(handle);
-        Runtime::new(runtime, None)
+        let primary = RuntimeType::External(handle.clone());
+        let secondary = RuntimeType::External(handle);
+        Runtime::new(primary, Some(secondary))
     }
 
     /// Create a [`Runtime`] instance from the settings
     /// See [`config::RuntimeConfig::from_settings`]
     pub fn from_settings() -> Result<Runtime> {
         let config = config::RuntimeConfig::from_settings()?;
-        let owned = RuntimeType::Shared(Arc::new(config.create_runtime()?));
-        Runtime::new(owned, None)
+        let runtime = Arc::new(config.create_runtime()?);
+        let primary = RuntimeType::Shared(runtime.clone());
+        let secondary = RuntimeType::External(runtime.handle().clone());
+        Runtime::new(primary, Some(secondary))
     }
 
-    /// Create a [`Runtime`] with a single-threaded primary async tokio runtime
+    /// Create a [`Runtime`] with two single-threaded async tokio runtime
     pub fn single_threaded() -> Result<Runtime> {
         let config = config::RuntimeConfig::single_threaded();
         let owned = RuntimeType::Shared(Arc::new(config.create_runtime()?));