feat: remove dedicated runtime for grpc, mysql and pg protocols

.github/workflows/develop.yml

@@ -141,6 +141,7 @@ jobs:
     runs-on: ubuntu-latest
     timeout-minutes: 60
     strategy:
+      fail-fast: false
       matrix:
         target: [ "fuzz_create_table", "fuzz_alter_table", "fuzz_create_database", "fuzz_create_logical_table", "fuzz_alter_logical_table", "fuzz_insert", "fuzz_insert_logical_table" ]
     steps:

config/config.md

@@ -16,9 +16,8 @@
 | `enable_telemetry` | Bool | `true` | Enable telemetry to collect anonymous usage data. |
 | `default_timezone` | String | `None` | The default timezone of the server. |
 | `runtime` | -- | -- | The runtime options. |
-| `runtime.read_rt_size` | Integer | `8` | The number of threads to execute the runtime for global read operations. |
-| `runtime.write_rt_size` | Integer | `8` | The number of threads to execute the runtime for global write operations. |
-| `runtime.bg_rt_size` | Integer | `4` | The number of threads to execute the runtime for global background operations. |
+| `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 the runtime for global write operations. |
 | `http` | -- | -- | The HTTP server options. |
 | `http.addr` | String | `127.0.0.1:4000` | The address to bind the HTTP server. |
 | `http.timeout` | String | `30s` | HTTP request timeout. Set to 0 to disable timeout. |
@@ -170,9 +169,8 @@
 | --- | -----| ------- | ----------- |
 | `default_timezone` | String | `None` | The default timezone of the server. |
 | `runtime` | -- | -- | The runtime options. |
-| `runtime.read_rt_size` | Integer | `8` | The number of threads to execute the runtime for global read operations. |
-| `runtime.write_rt_size` | Integer | `8` | The number of threads to execute the runtime for global write operations. |
-| `runtime.bg_rt_size` | Integer | `4` | The number of threads to execute the runtime for global background operations. |
+| `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 the runtime for global write operations. |
 | `heartbeat` | -- | -- | The heartbeat options. |
 | `heartbeat.interval` | String | `18s` | Interval for sending heartbeat messages to the metasrv. |
 | `heartbeat.retry_interval` | String | `3s` | Interval for retrying to send heartbeat messages to the metasrv. |
@@ -262,9 +260,8 @@
 | `store_key_prefix` | String | `""` | If it's not empty, the metasrv will store all data with this key prefix. |
 | `enable_region_failover` | Bool | `false` | Whether to enable region failover.<br/>This feature is only available on GreptimeDB running on cluster mode and<br/>- Using Remote WAL<br/>- Using shared storage (e.g., s3). |
 | `runtime` | -- | -- | The runtime options. |
-| `runtime.read_rt_size` | Integer | `8` | The number of threads to execute the runtime for global read operations. |
-| `runtime.write_rt_size` | Integer | `8` | The number of threads to execute the runtime for global write operations. |
-| `runtime.bg_rt_size` | Integer | `4` | The number of threads to execute the runtime for global background operations. |
+| `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 the runtime for global write operations. |
 | `procedure` | -- | -- | Procedure storage options. |
 | `procedure.max_retry_times` | Integer | `12` | Procedure max retry time. |
 | `procedure.retry_delay` | String | `500ms` | Initial retry delay of procedures, increases exponentially |
@@ -338,9 +335,8 @@
 | `grpc.tls.key_path` | String | `None` | Private key file path. |
 | `grpc.tls.watch` | Bool | `false` | Watch for Certificate and key file change and auto reload.<br/>For now, gRPC tls config does not support auto reload. |
 | `runtime` | -- | -- | The runtime options. |
-| `runtime.read_rt_size` | Integer | `8` | The number of threads to execute the runtime for global read operations. |
-| `runtime.write_rt_size` | Integer | `8` | The number of threads to execute the runtime for global write operations. |
-| `runtime.bg_rt_size` | Integer | `4` | The number of threads to execute the runtime for global background operations. |
+| `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 the runtime for global write operations. |
 | `heartbeat` | -- | -- | The heartbeat options. |
 | `heartbeat.interval` | String | `3s` | Interval for sending heartbeat messages to the metasrv. |
 | `heartbeat.retry_interval` | String | `3s` | Interval for retrying to send heartbeat messages to the metasrv. |

config/datanode.example.toml

@@ -73,11 +73,9 @@ watch = false
 ## The runtime options.
 [runtime]
 ## The number of threads to execute the runtime for global read operations.
-read_rt_size = 8
+global_rt_size = 8
 ## The number of threads to execute the runtime for global write operations.
-write_rt_size = 8
-## The number of threads to execute the runtime for global background operations.
-bg_rt_size = 4
+compact_rt_size = 4
 
 ## The heartbeat options.
 [heartbeat]

config/frontend.example.toml

@@ -5,11 +5,9 @@ default_timezone = "UTC"
 ## The runtime options.
 [runtime]
 ## The number of threads to execute the runtime for global read operations.
-read_rt_size = 8
+global_rt_size = 8
 ## The number of threads to execute the runtime for global write operations.
-write_rt_size = 8
-## The number of threads to execute the runtime for global background operations.
-bg_rt_size = 4
+compact_rt_size = 4
 
 ## The heartbeat options.
 [heartbeat]

config/metasrv.example.toml

@@ -34,11 +34,9 @@ enable_region_failover = false
 ## The runtime options.
 [runtime]
 ## The number of threads to execute the runtime for global read operations.
-read_rt_size = 8
+global_rt_size = 8
 ## The number of threads to execute the runtime for global write operations.
-write_rt_size = 8
-## The number of threads to execute the runtime for global background operations.
-bg_rt_size = 4
+compact_rt_size = 4
 
 ## Procedure storage options.
 [procedure]

config/standalone.example.toml

@@ -11,11 +11,9 @@ default_timezone = "UTC"
 ## The runtime options.
 [runtime]
 ## The number of threads to execute the runtime for global read operations.
-read_rt_size = 8
+global_rt_size = 8
 ## The number of threads to execute the runtime for global write operations.
-write_rt_size = 8
-## The number of threads to execute the runtime for global background operations.
-bg_rt_size = 4
+compact_rt_size = 4
 
 ## The HTTP server options.
 [http]

docs/how-to/how-to-write-fuzz-tests.md

@@ -105,7 +105,7 @@ use tests_fuzz::utils::{init_greptime_connections, Connections};
 
 fuzz_target!(|input: FuzzInput| {
     common_telemetry::init_default_ut_logging();
-    common_runtime::block_on_write(async {
+    common_runtime::block_on_global(async {
         let Connections { mysql } = init_greptime_connections().await;
             let mut rng = ChaChaRng::seed_from_u64(input.seed);
             let columns = rng.gen_range(2..30);

src/cmd/tests/load_config_test.rs

@@ -46,9 +46,8 @@ fn test_load_datanode_example_config() {
 
     let expected = GreptimeOptions::<DatanodeOptions> {
         runtime: RuntimeOptions {
-            read_rt_size: 8,
-            write_rt_size: 8,
-            bg_rt_size: 4,
+            global_rt_size: 8,
+            compact_rt_size: 4,
         },
         component: DatanodeOptions {
             node_id: Some(42),
@@ -119,9 +118,8 @@ fn test_load_frontend_example_config() {
             .unwrap();
     let expected = GreptimeOptions::<FrontendOptions> {
         runtime: RuntimeOptions {
-            read_rt_size: 8,
-            write_rt_size: 8,
-            bg_rt_size: 4,
+            global_rt_size: 8,
+            compact_rt_size: 4,
         },
         component: FrontendOptions {
             default_timezone: Some("UTC".to_string()),
@@ -167,9 +165,8 @@ fn test_load_metasrv_example_config() {
             .unwrap();
     let expected = GreptimeOptions::<MetasrvOptions> {
         runtime: RuntimeOptions {
-            read_rt_size: 8,
-            write_rt_size: 8,
-            bg_rt_size: 4,
+            global_rt_size: 8,
+            compact_rt_size: 4,
         },
         component: MetasrvOptions {
             selector: SelectorType::LeaseBased,
@@ -200,9 +197,8 @@ fn test_load_standalone_example_config() {
             .unwrap();
     let expected = GreptimeOptions::<StandaloneOptions> {
         runtime: RuntimeOptions {
-            read_rt_size: 8,
-            write_rt_size: 8,
-            bg_rt_size: 4,
+            global_rt_size: 8,
+            compact_rt_size: 4,
         },
         component: StandaloneOptions {
             default_timezone: Some("UTC".to_string()),

src/common/datasource/src/file_format/csv.rs

@@ -185,7 +185,7 @@ impl FileFormat for CsvFormat {
         let schema_infer_max_record = self.schema_infer_max_record;
         let has_header = self.has_header;
 
-        common_runtime::spawn_blocking_read(move || {
+        common_runtime::spawn_blocking_global(move || {
             let reader = SyncIoBridge::new(decoded);
 
             let (schema, _records_read) =

src/common/datasource/src/file_format/json.rs

@@ -101,7 +101,7 @@ impl FileFormat for JsonFormat {
 
         let schema_infer_max_record = self.schema_infer_max_record;
 
-        common_runtime::spawn_blocking_read(move || {
+        common_runtime::spawn_blocking_global(move || {
             let mut reader = BufReader::new(SyncIoBridge::new(decoded));
 
             let iter = ValueIter::new(&mut reader, schema_infer_max_record);

src/common/greptimedb-telemetry/src/lib.rs

@@ -72,7 +72,7 @@ impl GreptimeDBTelemetryTask {
         match self {
             GreptimeDBTelemetryTask::Enable((task, _)) => {
                 print_anonymous_usage_data_disclaimer();
-                task.start(common_runtime::bg_runtime())
+                task.start(common_runtime::global_runtime())
             }
             GreptimeDBTelemetryTask::Disable => Ok(()),
         }

src/common/grpc/src/channel_manager.rs

@@ -225,7 +225,7 @@ impl ChannelManager {
         }
 
         let pool = self.pool.clone();
-        let _handle = common_runtime::spawn_bg(async {
+        let _handle = common_runtime::spawn_global(async {
             recycle_channel_in_loop(pool, RECYCLE_CHANNEL_INTERVAL_SECS).await;
         });
         info!(

src/common/macro/src/admin_fn.rs

@@ -228,7 +228,7 @@ fn build_struct(
                         .create_mutable_vector(rows_num);
 
                     if columns_num == 0 {
-                        let result = common_runtime::block_on_read(async move {
+                        let result = common_runtime::block_on_global(async move {
                             #fn_name(handler, query_ctx, &[]).await
                         })?;
 
@@ -239,7 +239,7 @@ fn build_struct(
                                 .map(|vector| vector.get_ref(i))
                                 .collect();
 
-                            let result = common_runtime::block_on_read(async move {
+                            let result = common_runtime::block_on_global(async move {
                                 #fn_name(handler, query_ctx, &args).await
                             })?;
 

src/common/procedure/src/local.rs

@@ -461,7 +461,7 @@ impl LocalManager {
 
         let tracing_context = TracingContext::from_current_span();
 
-        let _handle = common_runtime::spawn_bg(async move {
+        let _handle = common_runtime::spawn_global(async move {
             // Run the root procedure.
             // The task was moved to another runtime for execution.
             // In order not to interrupt tracing, a span needs to be created to continue tracing the current task.
@@ -593,7 +593,7 @@ impl ProcedureManager for LocalManager {
         let task_inner = self.build_remove_outdated_meta_task();
 
         task_inner
-            .start(common_runtime::bg_runtime())
+            .start(common_runtime::global_runtime())
             .context(StartRemoveOutdatedMetaTaskSnafu)?;
 
         *task = Some(task_inner);

src/common/procedure/src/local/runner.rs

@@ -393,7 +393,7 @@ impl Runner {
         // Add the id of the subprocedure to the metadata.
         self.meta.push_child(procedure_id);
 
-        let _handle = common_runtime::spawn_bg(async move {
+        let _handle = common_runtime::spawn_global(async move {
             // Run the root procedure.
             runner.run().await
         });