From bba04bc49204f70001ba807066eafafef4db44aa Mon Sep 17 00:00:00 2001 From: ByteBaker <42913098+ByteBaker@users.noreply.github.com> Date: Fri, 11 Jul 2025 12:51:52 +0530 Subject: [PATCH] chore: correct `tokio::spawn` usage There was code of the form `tokio::spawn(async move { foo().await }` While this works without issues, it's redundant, as `tokio::spawn` needs a future as the argument. Therefore the new code now has only `tokio::spawn(foo())`. + perform `cargo fmt` --- src/common/infra/cluster/etcd.rs | 2 +- src/common/infra/config.rs | 2 +- src/config/src/meta/dashboards/v3/mod.rs | 2 +- src/config/src/meta/dashboards/v5/mod.rs | 8 +- src/config/src/meta/self_reporting/error.rs | 5 +- src/infra/src/db/etcd.rs | 2 +- .../m20250107_160900_delete_bad_dashboards.rs | 14 ++-- src/job/alert_manager.rs | 12 +-- src/job/compactor.rs | 22 +++--- src/job/files/mod.rs | 6 +- src/job/flatten_compactor.rs | 2 +- src/job/mod.rs | 76 +++++++++---------- src/job/stats.rs | 6 +- src/main.rs | 14 +--- src/report_server/src/server.rs | 4 +- src/service/alerts/scheduler/worker.rs | 1 - src/service/compact/flatten.rs | 3 +- src/service/metadata/distinct_values.rs | 2 +- src/service/pipeline/batch_execution.rs | 31 ++++---- src/service/self_reporting/queues.rs | 18 +++-- 20 files changed, 113 insertions(+), 119 deletions(-) diff --git a/src/common/infra/cluster/etcd.rs b/src/common/infra/cluster/etcd.rs index 8b087c4d6c..3867c5d240 100644 --- a/src/common/infra/cluster/etcd.rs +++ b/src/common/infra/cluster/etcd.rs @@ -76,7 +76,7 @@ async fn register() -> Result<()> { let locker = dist_lock::lock("/nodes/register", cfg.limit.node_heartbeat_ttl as u64).await?; // 2. watch node list - tokio::task::spawn(async move { super::watch_node_list().await }); + tokio::task::spawn(super::watch_node_list()); // 3. get node list let node_list = match super::list_nodes().await { diff --git a/src/common/infra/config.rs b/src/common/infra/config.rs index b35411a639..051ae03fbd 100644 --- a/src/common/infra/config.rs +++ b/src/common/infra/config.rs @@ -179,7 +179,7 @@ mod tests { #[tokio::test] async fn test_update_cache() { let (tx, rx) = mpsc::channel(100); - tokio::spawn(async move { update_cache(rx).await }); + tokio::spawn(update_cache(rx)); tx.send(infra::db::nats::NatsEvent::Connected) .await .unwrap(); diff --git a/src/config/src/meta/dashboards/v3/mod.rs b/src/config/src/meta/dashboards/v3/mod.rs index 86c8a8a604..338e3726f1 100644 --- a/src/config/src/meta/dashboards/v3/mod.rs +++ b/src/config/src/meta/dashboards/v3/mod.rs @@ -222,7 +222,7 @@ pub struct PanelConfig { pub struct DrillDown { #[serde(skip_serializing_if = "Option::is_none")] name: Option, - #[serde(skip_serializing_if = "Option::is_none", rename="type")] + #[serde(skip_serializing_if = "Option::is_none", rename = "type")] type_field: Option, #[serde(skip_serializing_if = "Option::is_none")] target_blank: Option, diff --git a/src/config/src/meta/dashboards/v5/mod.rs b/src/config/src/meta/dashboards/v5/mod.rs index f79a31178c..3730a93220 100644 --- a/src/config/src/meta/dashboards/v5/mod.rs +++ b/src/config/src/meta/dashboards/v5/mod.rs @@ -318,7 +318,7 @@ pub struct ColorCfg { #[derive(Debug, Clone, PartialEq, Hash, Serialize, Deserialize, ToSchema)] #[serde(rename_all = "camelCase")] pub struct Mapping { - #[serde(skip_serializing_if = "Option::is_none", rename="type")] + #[serde(skip_serializing_if = "Option::is_none", rename = "type")] typee: Option, #[serde(skip_serializing_if = "Option::is_none")] value: Option, @@ -328,7 +328,7 @@ pub struct Mapping { to: Option, #[serde(skip_serializing_if = "Option::is_none")] pattern: Option, - #[serde(skip_serializing_if = "Option::is_none", rename="match")] + #[serde(skip_serializing_if = "Option::is_none", rename = "match")] matchh: Option, #[serde(skip_serializing_if = "Option::is_none")] color: Option, @@ -341,7 +341,7 @@ pub struct Mapping { pub struct DrillDown { #[serde(skip_serializing_if = "Option::is_none")] name: Option, - #[serde(skip_serializing_if = "Option::is_none", rename="type")] + #[serde(skip_serializing_if = "Option::is_none", rename = "type")] type_field: Option, #[serde(skip_serializing_if = "Option::is_none")] target_blank: Option, @@ -356,7 +356,7 @@ pub struct DrillDown { pub struct MarkLine { #[serde(skip_serializing_if = "Option::is_none")] name: Option, - #[serde(skip_serializing_if = "Option::is_none", rename="type")] + #[serde(skip_serializing_if = "Option::is_none", rename = "type")] typee: Option, #[serde(skip_serializing_if = "Option::is_none")] value: Option, diff --git a/src/config/src/meta/self_reporting/error.rs b/src/config/src/meta/self_reporting/error.rs index 694b9f456b..c2fb0830be 100644 --- a/src/config/src/meta/self_reporting/error.rs +++ b/src/config/src/meta/self_reporting/error.rs @@ -48,7 +48,10 @@ pub struct PipelineError { pub pipeline_name: String, #[serde(skip_serializing_if = "Option::is_none")] pub error: Option, - #[serde(skip_serializing_if = "HashMap::is_empty", serialize_with = "serialize_values_only")] + #[serde( + skip_serializing_if = "HashMap::is_empty", + serialize_with = "serialize_values_only" + )] pub node_errors: HashMap, } diff --git a/src/infra/src/db/etcd.rs b/src/infra/src/db/etcd.rs index 584b5abd77..5de8c0cd2c 100644 --- a/src/infra/src/db/etcd.rs +++ b/src/infra/src/db/etcd.rs @@ -52,7 +52,7 @@ pub async fn init() { return; } // enable keep alive for auth token - tokio::task::spawn(async move { keep_alive_connection().await }); + tokio::task::spawn(keep_alive_connection()); } pub struct Etcd { diff --git a/src/infra/src/table/migration/m20250107_160900_delete_bad_dashboards.rs b/src/infra/src/table/migration/m20250107_160900_delete_bad_dashboards.rs index 156d1bb2c8..dd15577c45 100644 --- a/src/infra/src/table/migration/m20250107_160900_delete_bad_dashboards.rs +++ b/src/infra/src/table/migration/m20250107_160900_delete_bad_dashboards.rs @@ -687,7 +687,7 @@ mod v3 { pub struct DrillDown { #[serde(skip_serializing_if = "Option::is_none")] name: Option, - #[serde(skip_serializing_if = "Option::is_none", rename="type")] + #[serde(skip_serializing_if = "Option::is_none", rename = "type")] type_field: Option, #[serde(skip_serializing_if = "Option::is_none")] target_blank: Option, @@ -1026,7 +1026,7 @@ mod v4 { pub struct DrillDown { #[serde(skip_serializing_if = "Option::is_none")] name: Option, - #[serde(skip_serializing_if = "Option::is_none", rename="type")] + #[serde(skip_serializing_if = "Option::is_none", rename = "type")] type_field: Option, #[serde(skip_serializing_if = "Option::is_none")] target_blank: Option, @@ -1041,7 +1041,7 @@ mod v4 { pub struct MarkLine { #[serde(skip_serializing_if = "Option::is_none")] name: Option, - #[serde(skip_serializing_if = "Option::is_none", rename="type")] + #[serde(skip_serializing_if = "Option::is_none", rename = "type")] typee: Option, #[serde(skip_serializing_if = "Option::is_none")] value: Option, @@ -1430,7 +1430,7 @@ mod v5 { #[derive(Deserialize)] #[serde(rename_all = "camelCase")] pub struct Mapping { - #[serde(skip_serializing_if = "Option::is_none", rename="type")] + #[serde(skip_serializing_if = "Option::is_none", rename = "type")] typee: Option, #[serde(skip_serializing_if = "Option::is_none")] value: Option, @@ -1440,7 +1440,7 @@ mod v5 { to: Option, #[serde(skip_serializing_if = "Option::is_none")] pattern: Option, - #[serde(skip_serializing_if = "Option::is_none", rename="match")] + #[serde(skip_serializing_if = "Option::is_none", rename = "match")] matchh: Option, #[serde(skip_serializing_if = "Option::is_none")] color: Option, @@ -1453,7 +1453,7 @@ mod v5 { pub struct DrillDown { #[serde(skip_serializing_if = "Option::is_none")] name: Option, - #[serde(skip_serializing_if = "Option::is_none", rename="type")] + #[serde(skip_serializing_if = "Option::is_none", rename = "type")] type_field: Option, #[serde(skip_serializing_if = "Option::is_none")] target_blank: Option, @@ -1468,7 +1468,7 @@ mod v5 { pub struct MarkLine { #[serde(skip_serializing_if = "Option::is_none")] name: Option, - #[serde(skip_serializing_if = "Option::is_none", rename="type")] + #[serde(skip_serializing_if = "Option::is_none", rename = "type")] typee: Option, #[serde(skip_serializing_if = "Option::is_none")] value: Option, diff --git a/src/job/alert_manager.rs b/src/job/alert_manager.rs index c7398cbb4b..f82177600e 100644 --- a/src/job/alert_manager.rs +++ b/src/job/alert_manager.rs @@ -60,14 +60,14 @@ pub async fn run() -> Result<(), anyhow::Error> { .await?; } - tokio::task::spawn(async move { run_schedule_jobs().await }); - tokio::task::spawn(async move { watch_timeout_jobs().await }); + tokio::task::spawn(run_schedule_jobs()); + tokio::task::spawn(watch_timeout_jobs()); for i in 0..cfg.limit.search_job_workers { - tokio::task::spawn(async move { run_search_jobs(i).await }); + tokio::task::spawn(run_search_jobs(i)); } - tokio::task::spawn(async move { run_check_running_search_jobs().await }); - tokio::task::spawn(async move { run_delete_jobs_by_retention().await }); - tokio::task::spawn(async move { run_delete_jobs().await }); + tokio::task::spawn(run_check_running_search_jobs()); + tokio::task::spawn(run_delete_jobs_by_retention()); + tokio::task::spawn(run_delete_jobs()); Ok(()) } diff --git a/src/job/compactor.rs b/src/job/compactor.rs index 73213b7ec9..b82b7f10ad 100644 --- a/src/job/compactor.rs +++ b/src/job/compactor.rs @@ -46,19 +46,19 @@ pub async fn run() -> Result<(), anyhow::Error> { log::error!("[COMPACTOR::JOB] start merge job scheduler error: {e}"); } - tokio::task::spawn(async move { run_generate_job().await }); - tokio::task::spawn(async move { run_generate_old_data_job().await }); + tokio::task::spawn(run_generate_job()); + tokio::task::spawn(run_generate_old_data_job()); #[cfg(feature = "enterprise")] - tokio::task::spawn(async move { run_generate_downsampling_job().await }); - tokio::task::spawn(async move { run_merge(scheduler.tx()).await }); - tokio::task::spawn(async move { run_retention().await }); - tokio::task::spawn(async move { run_delay_deletion().await }); - tokio::task::spawn(async move { run_sync_to_db().await }); + tokio::task::spawn(run_generate_downsampling_job()); + tokio::task::spawn(run_merge(scheduler.tx())); + tokio::task::spawn(run_retention()); + tokio::task::spawn(run_delay_deletion()); + tokio::task::spawn(run_sync_to_db()); #[cfg(feature = "enterprise")] - tokio::task::spawn(async move { run_downsampling_sync_to_db().await }); - tokio::task::spawn(async move { run_check_running_jobs().await }); - tokio::task::spawn(async move { run_clean_done_jobs().await }); - tokio::task::spawn(async move { run_compactor_pending_jobs_metric().await }); + tokio::task::spawn(run_downsampling_sync_to_db()); + tokio::task::spawn(run_check_running_jobs()); + tokio::task::spawn(run_clean_done_jobs()); + tokio::task::spawn(run_compactor_pending_jobs_metric()); Ok(()) } diff --git a/src/job/files/mod.rs b/src/job/files/mod.rs index 73321c0056..f3e2f6f439 100644 --- a/src/job/files/mod.rs +++ b/src/job/files/mod.rs @@ -31,9 +31,9 @@ pub async fn run() -> Result<(), anyhow::Error> { // load pending delete files to memory cache crate::service::db::file_list::local::load_pending_delete().await?; - tokio::task::spawn(async move { parquet::run().await }); - tokio::task::spawn(async move { broadcast::run().await }); - tokio::task::spawn(async move { clean_empty_dirs().await }); + tokio::task::spawn(parquet::run()); + tokio::task::spawn(broadcast::run()); + tokio::task::spawn(clean_empty_dirs()); Ok(()) } diff --git a/src/job/flatten_compactor.rs b/src/job/flatten_compactor.rs index bbcd57cc44..a9c8fbd885 100644 --- a/src/job/flatten_compactor.rs +++ b/src/job/flatten_compactor.rs @@ -58,7 +58,7 @@ pub async fn run() -> Result<(), anyhow::Error> { }); } - tokio::task::spawn(async move { run_generate(tx).await }); + tokio::task::spawn(run_generate(tx)); Ok(()) } diff --git a/src/job/mod.rs b/src/job/mod.rs index 6964148e37..f32675ee31 100644 --- a/src/job/mod.rs +++ b/src/job/mod.rs @@ -97,7 +97,7 @@ pub async fn init() -> Result<(), anyhow::Error> { && (LOCAL_NODE.is_ingester() || LOCAL_NODE.is_querier() || LOCAL_NODE.is_alert_manager()) { // Try to download the mmdb files, if its not disabled. - tokio::task::spawn(async move { mmdb_downloader::run().await }); + tokio::task::spawn(mmdb_downloader::run()); } db::user::cache().await.expect("user cache failed"); @@ -111,9 +111,9 @@ pub async fn init() -> Result<(), anyhow::Error> { .expect("organization settings cache sync failed"); // watch org users - tokio::task::spawn(async move { db::user::watch().await }); - tokio::task::spawn(async move { db::org_users::watch().await }); - tokio::task::spawn(async move { db::organization::watch().await }); + tokio::task::spawn(db::user::watch()); + tokio::task::spawn(db::org_users::watch()); + tokio::task::spawn(db::organization::watch()); // check version db::metas::version::set() @@ -125,13 +125,13 @@ pub async fn init() -> Result<(), anyhow::Error> { // Auth auditing should be done by router also #[cfg(feature = "enterprise")] - tokio::task::spawn(async move { self_reporting::run_audit_publish().await }); + tokio::task::spawn(self_reporting::run_audit_publish()); #[cfg(feature = "cloud")] - tokio::task::spawn(async move { self_reporting::cloud_events::flush_cloud_events().await }); + tokio::task::spawn(self_reporting::cloud_events::flush_cloud_events()); #[cfg(feature = "enterprise")] { - tokio::task::spawn(async move { db::ofga::watch().await }); + tokio::task::spawn(db::ofga::watch()); db::ofga::cache().await.expect("ofga model cache failed"); o2_openfga::authorizer::authz::init_open_fga().await; // RBAC model @@ -142,7 +142,7 @@ pub async fn init() -> Result<(), anyhow::Error> { } } - tokio::task::spawn(async move { promql_self_consume::run().await }); + tokio::task::spawn(promql_self_consume::run()); // Router doesn't need to initialize job if LOCAL_NODE.is_router() && LOCAL_NODE.is_single_role() { return Ok(()); @@ -150,27 +150,27 @@ pub async fn init() -> Result<(), anyhow::Error> { // telemetry run if cfg.common.telemetry_enabled && LOCAL_NODE.is_querier() { - tokio::task::spawn(async move { telemetry::run().await }); + tokio::task::spawn(telemetry::run()); } - tokio::task::spawn(async move { self_reporting::run().await }); + tokio::task::spawn(self_reporting::run()); // cache short_urls - tokio::task::spawn(async move { db::short_url::watch().await }); + tokio::task::spawn(db::short_url::watch()); db::short_url::cache() .await .expect("short url cache failed"); // initialize metadata watcher - tokio::task::spawn(async move { db::schema::watch().await }); - tokio::task::spawn(async move { db::functions::watch().await }); - tokio::task::spawn(async move { db::compact::retention::watch().await }); - tokio::task::spawn(async move { db::metrics::watch_prom_cluster_leader().await }); - tokio::task::spawn(async move { db::alerts::templates::watch().await }); - tokio::task::spawn(async move { db::alerts::destinations::watch().await }); - tokio::task::spawn(async move { db::alerts::realtime_triggers::watch().await }); - tokio::task::spawn(async move { db::alerts::alert::watch().await }); - tokio::task::spawn(async move { db::organization::org_settings_watch().await }); + tokio::task::spawn(db::schema::watch()); + tokio::task::spawn(db::functions::watch()); + tokio::task::spawn(db::compact::retention::watch()); + tokio::task::spawn(db::metrics::watch_prom_cluster_leader()); + tokio::task::spawn(db::alerts::templates::watch()); + tokio::task::spawn(db::alerts::destinations::watch()); + tokio::task::spawn(db::alerts::realtime_triggers::watch()); + tokio::task::spawn(db::alerts::alert::watch()); + tokio::task::spawn(db::organization::org_settings_watch()); // pipeline not used on compactors if LOCAL_NODE.is_ingester() || LOCAL_NODE.is_querier() || LOCAL_NODE.is_alert_manager() { @@ -179,10 +179,10 @@ pub async fn init() -> Result<(), anyhow::Error> { #[cfg(feature = "enterprise")] if LOCAL_NODE.is_ingester() || LOCAL_NODE.is_querier() { - tokio::task::spawn(async move { db::session::watch().await }); + tokio::task::spawn(db::session::watch()); } if LOCAL_NODE.is_ingester() || LOCAL_NODE.is_querier() || LOCAL_NODE.is_alert_manager() { - tokio::task::spawn(async move { db::enrichment_table::watch().await }); + tokio::task::spawn(db::enrichment_table::watch()); } tokio::task::yield_now().await; @@ -235,31 +235,29 @@ pub async fn init() -> Result<(), anyhow::Error> { } } - tokio::task::spawn(async move { files::run().await }); - tokio::task::spawn(async move { stats::run().await }); - tokio::task::spawn(async move { compactor::run().await }); - tokio::task::spawn(async move { flatten_compactor::run().await }); - tokio::task::spawn(async move { metrics::run().await }); - tokio::task::spawn(async move { promql::run().await }); - tokio::task::spawn(async move { alert_manager::run().await }); - tokio::task::spawn(async move { file_downloader::run().await }); + tokio::task::spawn(files::run()); + tokio::task::spawn(stats::run()); + tokio::task::spawn(compactor::run()); + tokio::task::spawn(flatten_compactor::run()); + tokio::task::spawn(metrics::run()); + tokio::task::spawn(promql::run()); + tokio::task::spawn(alert_manager::run()); + tokio::task::spawn(file_downloader::run()); if LOCAL_NODE.is_compactor() { - tokio::task::spawn(async move { file_list_dump::run().await }); + tokio::task::spawn(file_list_dump::run()); } // load metrics disk cache - tokio::task::spawn(async move { crate::service::promql::search::init().await }); + tokio::task::spawn(crate::service::promql::search::init()); // start pipeline data retention #[cfg(feature = "enterprise")] - tokio::task::spawn( - async move { o2_enterprise::enterprise::pipeline::pipeline_job::run().await }, - ); + tokio::task::spawn(o2_enterprise::enterprise::pipeline::pipeline_job::run()); #[cfg(feature = "enterprise")] - tokio::task::spawn(async move { cipher::run().await }); + tokio::task::spawn(cipher::run()); #[cfg(feature = "enterprise")] - tokio::task::spawn(async move { db::keys::watch().await }); + tokio::task::spawn(db::keys::watch()); // additional for cloud #[cfg(feature = "cloud")] @@ -279,8 +277,8 @@ pub async fn init() -> Result<(), anyhow::Error> { log::info!("Job initialization complete"); // Syslog server start - tokio::task::spawn(async move { db::syslog::watch().await }); - tokio::task::spawn(async move { db::syslog::watch_syslog_settings().await }); + tokio::task::spawn(db::syslog::watch()); + tokio::task::spawn(db::syslog::watch_syslog_settings()); let start_syslog = *SYSLOG_ENABLED.read(); if start_syslog { diff --git a/src/job/stats.rs b/src/job/stats.rs index b8e2130c6e..54cf669772 100644 --- a/src/job/stats.rs +++ b/src/job/stats.rs @@ -19,9 +19,9 @@ use tokio::time; use crate::service::{compact::stats::update_stats_from_file_list, db}; pub async fn run() -> Result<(), anyhow::Error> { - tokio::task::spawn(async move { update_node_memory_usage().await }); - tokio::task::spawn(async move { file_list_update_stats().await }); - tokio::task::spawn(async move { cache_stream_stats().await }); + tokio::task::spawn(update_node_memory_usage()); + tokio::task::spawn(file_list_update_stats()); + tokio::task::spawn(cache_stream_stats()); Ok(()) } diff --git a/src/main.rs b/src/main.rs index 430f75313f..3061103d28 100644 --- a/src/main.rs +++ b/src/main.rs @@ -359,7 +359,7 @@ async fn main() -> Result<(), anyhow::Error> { .expect("Deferred jobs failed to init"); if cfg.log.events_enabled { - tokio::task::spawn(async move { zo_logger::send_logs().await }); + tokio::task::spawn(zo_logger::send_logs()); } if cfg.common.telemetry_enabled { tokio::task::spawn(async move { @@ -753,9 +753,7 @@ async fn init_http_server() -> Result<(), anyhow::Error> { .disable_signals() .run(); let handle = server.handle(); - tokio::task::spawn(async move { - graceful_shutdown(handle).await; - }); + tokio::task::spawn(graceful_shutdown(handle)); server.await?; Ok(()) } @@ -861,9 +859,7 @@ async fn init_http_server_without_tracing() -> Result<(), anyhow::Error> { .disable_signals() .run(); let handle = server.handle(); - tokio::task::spawn(async move { - graceful_shutdown(handle).await; - }); + tokio::task::spawn(graceful_shutdown(handle)); server.await?; Ok(()) } @@ -1104,9 +1100,7 @@ async fn init_script_server() -> Result<(), anyhow::Error> { .disable_signals() .run(); let handle = server.handle(); - tokio::task::spawn(async move { - graceful_shutdown(handle).await; - }); + tokio::task::spawn(graceful_shutdown(handle)); server.await?; log::info!("HTTP server stopped"); diff --git a/src/report_server/src/server.rs b/src/report_server/src/server.rs index ecf3e3c536..e96d957f45 100644 --- a/src/report_server/src/server.rs +++ b/src/report_server/src/server.rs @@ -40,9 +40,7 @@ pub async fn spawn_server() -> Result<(), anyhow::Error> { .run(); let handle = server.handle(); - tokio::task::spawn(async move { - graceful_shutdown(handle).await; - }); + tokio::task::spawn(graceful_shutdown(handle)); server.await?; log::info!("Report server stopped"); Ok(()) diff --git a/src/service/alerts/scheduler/worker.rs b/src/service/alerts/scheduler/worker.rs index 7f5be8fa89..fae3b1c26f 100644 --- a/src/service/alerts/scheduler/worker.rs +++ b/src/service/alerts/scheduler/worker.rs @@ -140,7 +140,6 @@ impl SchedulerWorker { if let Err(e) = handler_job.await { return Err(anyhow::anyhow!("Error in handler: {}", e)); } - Ok(()) } } diff --git a/src/service/compact/flatten.rs b/src/service/compact/flatten.rs index 38880f598a..65b2950ced 100644 --- a/src/service/compact/flatten.rs +++ b/src/service/compact/flatten.rs @@ -62,8 +62,7 @@ pub async fn run_generate(worker_tx: mpsc::Sender) -> Result<(), anyhow infra::schema::get_settings(&org_id, &stream_name, stream_type) .await .unwrap_or_default(); - let defined_schema_fields = - stream_setting.defined_schema_fields; + let defined_schema_fields = stream_setting.defined_schema_fields; if defined_schema_fields.is_empty() { continue; } diff --git a/src/service/metadata/distinct_values.rs b/src/service/metadata/distinct_values.rs index 6a41c8de58..794abd3016 100644 --- a/src/service/metadata/distinct_values.rs +++ b/src/service/metadata/distinct_values.rs @@ -109,7 +109,7 @@ impl Default for DistinctValues { impl DistinctValues { pub fn new() -> Self { - tokio::task::spawn(async move { run_flush().await }); + tokio::task::spawn(run_flush()); Self { channel: handle_channel(), shutdown: Arc::new(AtomicBool::new(false)), diff --git a/src/service/pipeline/batch_execution.rs b/src/service/pipeline/batch_execution.rs index e590a8a81f..6d620980d5 100644 --- a/src/service/pipeline/batch_execution.rs +++ b/src/service/pipeline/batch_execution.rs @@ -220,7 +220,7 @@ impl ExecutablePipeline { } // Spawn tasks for each node - let mut node_tasks = Vec::new(); + let mut node_tasks = Vec::with_capacity(self.sorted_nodes.len()); for (idx, node_id) in self.sorted_nodes.iter().enumerate() { let pl_id_cp = self.id.to_string(); let org_id_cp = org_id.to_string(); @@ -239,22 +239,19 @@ impl ExecutablePipeline { // WARN: Do not change. Processing node can only be done in a task, as the internals of // remote wal writer depends on the task id. - let task = tokio::spawn(async move { - process_node( - pl_id_cp, - idx, - org_id_cp, - node, - node_receiver, - child_senders, - vrl_runtime, - result_sender_cp, - error_sender_cp, - pipeline_name, - stream_name, - ) - .await - }); + let task = tokio::spawn(process_node( + pl_id_cp, + idx, + org_id_cp, + node, + node_receiver, + child_senders, + vrl_runtime, + result_sender_cp, + error_sender_cp, + pipeline_name, + stream_name, + )); node_tasks.push(task); } diff --git a/src/service/self_reporting/queues.rs b/src/service/self_reporting/queues.rs index 00755f6eb1..c0d61ffc31 100644 --- a/src/service/self_reporting/queues.rs +++ b/src/service/self_reporting/queues.rs @@ -56,9 +56,12 @@ fn initialize_usage_queue() -> ReportingQueue { for thread_id in 0..cfg.limit.usage_reporting_thread_num { let msg_receiver = msg_receiver.clone(); - tokio::task::spawn(async move { - self_reporting_ingest_job(thread_id, msg_receiver, batch_size, timeout).await - }); + tokio::task::spawn(self_reporting_ingest_job( + thread_id, + msg_receiver, + batch_size, + timeout, + )); } ReportingQueue::new(msg_sender) @@ -81,9 +84,12 @@ fn initialize_error_queue() -> ReportingQueue { for thread_id in 0..cfg.limit.usage_reporting_thread_num { let msg_receiver = msg_receiver.clone(); - tokio::task::spawn(async move { - self_reporting_ingest_job(thread_id, msg_receiver, batch_size, timeout).await - }); + tokio::task::spawn(self_reporting_ingest_job( + thread_id, + msg_receiver, + batch_size, + timeout, + )); } ReportingQueue::new(msg_sender)