fix: cap default runtime sizes to a minimum of 2 threads (#8908)

* fix: cap default runtime sizes to a minimum of 2 threads

RuntimeOptions derived its default sizes directly from num_cpus. On
single-core machines every runtime (global, compact, query, ingest)
ended up with one worker thread, which can easily deadlock async code
(e.g. block_on combined with spawn).

Clamp all CPU-derived runtime sizes to at least 2 threads.

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

* fix: init logging before runtimes so runtime options are logged

The global runtimes were initialized before the global logging
subscriber, so the "Creating runtime ..." info logs that carry the
runtime sizes were silently dropped. Initialize logging first in all
node start paths; common-telemetry has no dependency on
common-runtime, so the reorder is safe.

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

---------

Signed-off-by: Lei, HUANG <ratuthomm@gmail.com>

config/config.md

@@ -25,7 +25,7 @@
 | `runtime` | -- | -- | The runtime options. |
 | `runtime.global_rt_size` | Integer | `8` | The number of threads to execute the runtime for global read operations. |
 | `runtime.compact_rt_size` | Integer | `4` | The number of threads to execute compact operations. |
-| `runtime.compact_rt_max_blocking_threads` | Integer | `4` | The maximum number of blocking threads for compact operations.<br/>Defaults to max(num_cpus / 2, 1). |
+| `runtime.compact_rt_max_blocking_threads` | Integer | `4` | The maximum number of blocking threads for compact operations.<br/>Defaults to max(num_cpus / 2, 2). |
 | `http` | -- | -- | The HTTP server options. |
 | `http.addr` | String | `127.0.0.1:4000` | The address to bind the HTTP server. |
 | `http.timeout` | String | `0s` | HTTP request timeout. Set to 0 to disable timeout.<br/>When Prometheus pending-row batching is enabled, a nonzero timeout less than or equal to the<br/>`prom_store.pending_rows_flush_interval` plus 1 second is adjusted to that value. |
@@ -255,7 +255,7 @@
 | `runtime` | -- | -- | The runtime options. |
 | `runtime.global_rt_size` | Integer | `8` | The number of threads to execute the runtime for global read operations. |
 | `runtime.compact_rt_size` | Integer | `4` | The number of threads to execute compact operations. |
-| `runtime.compact_rt_max_blocking_threads` | Integer | `4` | The maximum number of blocking threads for compact operations.<br/>Defaults to max(num_cpus / 2, 1). |
+| `runtime.compact_rt_max_blocking_threads` | Integer | `4` | The maximum number of blocking threads for compact operations.<br/>Defaults to max(num_cpus / 2, 2). |
 | `http` | -- | -- | The HTTP server options. |
 | `http.addr` | String | `127.0.0.1:4000` | The address to bind the HTTP server. |
 | `http.timeout` | String | `0s` | HTTP request timeout. Set to 0 to disable timeout.<br/>When Prometheus pending-row batching is enabled, a nonzero timeout less than or equal to the<br/>`prom_store.pending_rows_flush_interval` plus 1 second is adjusted to that value. |
@@ -398,7 +398,7 @@
 | `runtime` | -- | -- | The runtime options. |
 | `runtime.global_rt_size` | Integer | `8` | The number of threads to execute the runtime for global read operations. |
 | `runtime.compact_rt_size` | Integer | `4` | The number of threads to execute compact operations. |
-| `runtime.compact_rt_max_blocking_threads` | Integer | `4` | The maximum number of blocking threads for compact operations.<br/>Defaults to max(num_cpus / 2, 1). |
+| `runtime.compact_rt_max_blocking_threads` | Integer | `4` | The maximum number of blocking threads for compact operations.<br/>Defaults to max(num_cpus / 2, 2). |
 | `backend_tls` | -- | -- | TLS configuration for kv store backend (applicable for etcd, PostgreSQL, and MySQL backends)<br/>When using etcd, PostgreSQL, or MySQL as metadata store, you can configure TLS here<br/><br/>Note: if TLS is configured in both this section and the `store_addrs` connection string, the<br/>settings here will override the TLS settings in `store_addrs`. |
 | `backend_tls.mode` | String | `prefer` | TLS mode, refer to https://www.postgresql.org/docs/current/libpq-ssl.html<br/>- "disable" - No TLS<br/>- "prefer" (default) - Try TLS, fallback to plain<br/>- "require" - Require TLS<br/>- "verify_ca" - Require TLS and verify CA<br/>- "verify_full" - Require TLS and verify hostname |
 | `backend_tls.cert_path` | String | `""` | Path to client certificate file (for client authentication)<br/>Like "/path/to/client.crt" |
@@ -509,8 +509,8 @@
 | `runtime` | -- | -- | The runtime options. |
 | `runtime.global_rt_size` | Integer | `8` | The number of threads to execute the runtime for global read operations. |
 | `runtime.compact_rt_size` | Integer | `4` | The number of threads to execute compact operations. |
-| `runtime.compact_rt_max_blocking_threads` | Integer | `4` | The maximum number of blocking threads for compact operations.<br/>Defaults to max(num_cpus / 2, 1). |
-| `runtime.query_rt_size` | Integer | `7` | The number of threads to execute datanode query operations.<br/>Defaults to max(num_cpus - 1, 1). |
+| `runtime.compact_rt_max_blocking_threads` | Integer | `4` | The maximum number of blocking threads for compact operations.<br/>Defaults to max(num_cpus / 2, 2). |
+| `runtime.query_rt_size` | Integer | `7` | The number of threads to execute datanode query operations.<br/>Defaults to max(num_cpus - 1, 2). |
 | `runtime.ingest_rt_size` | Integer | `8` | The number of threads to execute datanode ingestion operations. |
 | `meta_client` | -- | -- | The metasrv client options. |
 | `meta_client.metasrv_addrs` | Array | -- | The addresses of the metasrv. |

config/datanode.example.toml

@@ -84,10 +84,10 @@ watch = false
 ## The number of threads to execute compact operations.
 #+ compact_rt_size = 4
 ## The maximum number of blocking threads for compact operations.
-## Defaults to max(num_cpus / 2, 1).
+## Defaults to max(num_cpus / 2, 2).
 #+ compact_rt_max_blocking_threads = 4
 ## The number of threads to execute datanode query operations.
-## Defaults to max(num_cpus - 1, 1).
+## Defaults to max(num_cpus - 1, 2).
 #+ query_rt_size = 7
 ## The number of threads to execute datanode ingestion operations.
 #+ ingest_rt_size = 8

config/frontend.example.toml

@@ -44,7 +44,7 @@ default_column_prefix = "greptime"
 ## The number of threads to execute compact operations.
 #+ compact_rt_size = 4
 ## The maximum number of blocking threads for compact operations.
-## Defaults to max(num_cpus / 2, 1).
+## Defaults to max(num_cpus / 2, 2).
 #+ compact_rt_max_blocking_threads = 4
 
 ## The HTTP server options.

config/metasrv.example.toml

@@ -92,7 +92,7 @@ node_max_idle_time = "24hours"
 ## The number of threads to execute compact operations.
 #+ compact_rt_size = 4
 ## The maximum number of blocking threads for compact operations.
-## Defaults to max(num_cpus / 2, 1).
+## Defaults to max(num_cpus / 2, 2).
 #+ compact_rt_max_blocking_threads = 4
 
 ## TLS configuration for kv store backend (applicable for etcd, PostgreSQL, and MySQL backends)

config/standalone.example.toml

@@ -57,7 +57,7 @@ max_concurrent_queries = 0
 ## The number of threads to execute compact operations.
 #+ compact_rt_size = 4
 ## The maximum number of blocking threads for compact operations.
-## Defaults to max(num_cpus / 2, 1).
+## Defaults to max(num_cpus / 2, 2).
 #+ compact_rt_max_blocking_threads = 4
 
 ## The HTTP server options.

src/cmd/src/datanode/builder.rs

@@ -61,9 +61,6 @@ impl InstanceBuilder {
     }
 
     async fn init(opts: &mut DatanodeOptions, plugins: &mut Plugins) -> Result<Vec<WorkerGuard>> {
-        common_runtime::init_global_runtimes(&opts.runtime);
-        common_runtime::init_datanode_runtimes(&opts.runtime);
-
         let dn_opts = &mut opts.component;
         let guard = common_telemetry::init_global_logging(
             APP_NAME,
@@ -73,6 +70,9 @@ impl InstanceBuilder {
             None,
         );
 
+        common_runtime::init_global_runtimes(&opts.runtime);
+        common_runtime::init_datanode_runtimes(&opts.runtime);
+
         crate::options::flush_dropped_plugin_warnings();
         log_versions(verbose_version(), short_version(), APP_NAME);
         maybe_activate_heap_profile(&dn_opts.memory);

src/cmd/src/flownode.rs

@@ -248,8 +248,6 @@ impl StartCommand {
     }
 
     async fn build(&self, opts: FlownodeOptions) -> Result<Instance> {
-        common_runtime::init_global_runtimes(&opts.runtime);
-
         let guard = common_telemetry::init_global_logging(
             APP_NAME,
             &opts.component.logging,
@@ -258,6 +256,8 @@ impl StartCommand {
             None,
         );
 
+        common_runtime::init_global_runtimes(&opts.runtime);
+
         crate::options::flush_dropped_plugin_warnings();
         log_versions(verbose_version(), short_version(), APP_NAME);
         maybe_activate_heap_profile(&opts.component.memory);

src/cmd/src/frontend.rs

@@ -330,8 +330,6 @@ impl StartCommand {
     }
 
     async fn build(&self, opts: FrontendOptions) -> Result<Instance> {
-        common_runtime::init_global_runtimes(&opts.runtime);
-
         let guard = common_telemetry::init_global_logging(
             APP_NAME,
             &opts.component.logging,
@@ -340,6 +338,8 @@ impl StartCommand {
             Some(&opts.component.slow_query),
         );
 
+        common_runtime::init_global_runtimes(&opts.runtime);
+
         crate::options::flush_dropped_plugin_warnings();
         log_versions(verbose_version(), short_version(), APP_NAME);
         maybe_activate_heap_profile(&opts.component.memory);

src/cmd/src/metasrv.rs

@@ -318,8 +318,6 @@ impl StartCommand {
     }
 
     pub async fn build(&self, opts: MetasrvOptions) -> Result<Instance> {
-        common_runtime::init_global_runtimes(&opts.runtime);
-
         let guard = common_telemetry::init_global_logging(
             APP_NAME,
             &opts.component.logging,
@@ -328,6 +326,8 @@ impl StartCommand {
             None,
         );
 
+        common_runtime::init_global_runtimes(&opts.runtime);
+
         crate::options::flush_dropped_plugin_warnings();
         log_versions(verbose_version(), short_version(), APP_NAME);
         maybe_activate_heap_profile(&opts.component.memory);

src/cmd/src/standalone.rs

@@ -395,8 +395,6 @@ impl StartCommand {
     #[allow(clippy::diverging_sub_expression)]
     /// Build GreptimeDB instance with the loaded options.
     pub async fn build(&self, opts: GreptimeOptions<StandaloneOptions>) -> Result<Instance> {
-        common_runtime::init_global_runtimes(&opts.runtime);
-
         let guard = common_telemetry::init_global_logging(
             APP_NAME,
             &opts.component.logging,
@@ -405,6 +403,8 @@ impl StartCommand {
             Some(&opts.component.slow_query),
         );
 
+        common_runtime::init_global_runtimes(&opts.runtime);
+
         crate::options::flush_dropped_plugin_warnings();
         log_versions(verbose_version(), short_version(), APP_NAME);
         maybe_activate_heap_profile(&opts.component.memory);

src/common/runtime/src/global.rs

@@ -27,6 +27,9 @@ use crate::{Builder, JoinHandle, Runtime};
 const GLOBAL_WORKERS: usize = 8;
 const COMPACT_WORKERS: usize = 4;
 const HB_WORKERS: usize = 2;
+/// The minimum number of worker threads for runtimes sized by CPU count.
+/// A single-threaded runtime can easily deadlock in async code.
+const MIN_RUNTIME_THREADS: usize = 2;
 
 /// The options for the global runtimes.
 #[derive(Clone, Debug, Serialize, Deserialize, PartialEq)]
@@ -44,19 +47,25 @@ pub struct RuntimeOptions {
     pub ingest_rt_size: usize,
 }
 
-impl Default for RuntimeOptions {
-    fn default() -> Self {
-        let cpus = num_cpus::get();
+impl RuntimeOptions {
+    fn with_num_cpus(cpus: usize) -> Self {
+        let cpus = usize::max(cpus, MIN_RUNTIME_THREADS);
         Self {
             global_rt_size: cpus,
-            compact_rt_size: usize::max(cpus / 2, 1),
-            compact_rt_max_blocking_threads: usize::max(cpus / 2, 1),
-            query_rt_size: usize::max(cpus.saturating_sub(1), 1),
+            compact_rt_size: usize::max(cpus / 2, MIN_RUNTIME_THREADS),
+            compact_rt_max_blocking_threads: usize::max(cpus / 2, MIN_RUNTIME_THREADS),
+            query_rt_size: usize::max(cpus.saturating_sub(1), MIN_RUNTIME_THREADS),
             ingest_rt_size: cpus,
         }
     }
 }
 
+impl Default for RuntimeOptions {
+    fn default() -> Self {
+        Self::with_num_cpus(num_cpus::get())
+    }
+}
+
 pub fn create_runtime(runtime_name: &str, thread_name: &str, worker_threads: usize) -> Runtime {
     info!(
         "Creating runtime with runtime_name: {runtime_name}, thread_name: {thread_name}, work_threads: {worker_threads}."
@@ -287,18 +296,56 @@ mod tests {
     #[test]
     fn test_datanode_runtime_options_default() {
         let options = RuntimeOptions::default();
-        let cpus = num_cpus::get();
+        let cpus = usize::max(num_cpus::get(), MIN_RUNTIME_THREADS);
 
         assert_eq!(cpus, options.global_rt_size);
-        assert_eq!(usize::max(cpus / 2, 1), options.compact_rt_size);
         assert_eq!(
-            usize::max(cpus / 2, 1),
+            usize::max(cpus / 2, MIN_RUNTIME_THREADS),
+            options.compact_rt_size
+        );
+        assert_eq!(
+            usize::max(cpus / 2, MIN_RUNTIME_THREADS),
             options.compact_rt_max_blocking_threads
         );
-        assert_eq!(usize::max(cpus.saturating_sub(1), 1), options.query_rt_size);
+        assert_eq!(
+            usize::max(cpus.saturating_sub(1), MIN_RUNTIME_THREADS),
+            options.query_rt_size
+        );
         assert_eq!(cpus, options.ingest_rt_size);
     }
 
+    #[test]
+    fn test_runtime_options_min_threads() {
+        for cpus in [0, 1, 2] {
+            let options = RuntimeOptions::with_num_cpus(cpus);
+            assert!(
+                options.global_rt_size >= MIN_RUNTIME_THREADS,
+                "global_rt_size {} < {MIN_RUNTIME_THREADS} with {cpus} cpus",
+                options.global_rt_size
+            );
+            assert!(
+                options.compact_rt_size >= MIN_RUNTIME_THREADS,
+                "compact_rt_size {} < {MIN_RUNTIME_THREADS} with {cpus} cpus",
+                options.compact_rt_size
+            );
+            assert!(
+                options.compact_rt_max_blocking_threads >= MIN_RUNTIME_THREADS,
+                "compact_rt_max_blocking_threads {} < {MIN_RUNTIME_THREADS} with {cpus} cpus",
+                options.compact_rt_max_blocking_threads
+            );
+            assert!(
+                options.query_rt_size >= MIN_RUNTIME_THREADS,
+                "query_rt_size {} < {MIN_RUNTIME_THREADS} with {cpus} cpus",
+                options.query_rt_size
+            );
+            assert!(
+                options.ingest_rt_size >= MIN_RUNTIME_THREADS,
+                "ingest_rt_size {} < {MIN_RUNTIME_THREADS} with {cpus} cpus",
+                options.ingest_rt_size
+            );
+        }
+    }
+
     #[test]
     fn test_datanode_runtimes_fallback_to_global_runtime() {
         let runtimes = GlobalRuntimes::new(