chore(vector-common): Move spawn helpers out of the binary crate

Moves spawn_named and num_threads next to spawn_in_current_span.
Call sites import them from vector_lib.

Author: bruceg

lib/vector-common/Cargo.toml

@@ -56,7 +56,7 @@ serde_json.workspace = true
 smallvec = { version = "1", default-features = false }
 strum.workspace = true
 stream-cancel = { version = "0.8.2", default-features = false }
-tokio = { workspace = true, features = ["macros", "rt", "time"] }
+tokio = { workspace = true, features = ["macros", "rt", "time", "tracing"] }
 tracing.workspace = true
 vrl.workspace = true
 vector-config.workspace = true

lib/vector-common/src/lib.rs

@@ -153,3 +153,40 @@ where
     use tracing::Instrument as _;
     tokio::spawn(task.in_current_span())
 }
+
+/// Spawn a task with the given name. The name is only used if
+/// built with [`tokio_unstable`][tokio_unstable].
+///
+/// [tokio_unstable]: https://docs.rs/tokio/latest/tokio/#unstable-features
+#[track_caller]
+pub fn spawn_named<T>(
+    task: impl std::future::Future<Output = T> + Send + 'static,
+    _name: &str,
+) -> tokio::task::JoinHandle<T>
+where
+    T: Send + 'static,
+{
+    #[cfg(tokio_unstable)]
+    return tokio::task::Builder::new()
+        .name(_name)
+        .spawn(task)
+        .expect("tokio task should spawn");
+
+    #[cfg(not(tokio_unstable))]
+    tokio::spawn(task)
+}
+
+/// Returns an estimate of the number of recommended threads that Vector should spawn.
+pub fn num_threads() -> usize {
+    let count = match std::thread::available_parallelism() {
+        Ok(count) => count,
+        Err(error) => {
+            tracing::warn!(
+                message = "Failed to determine available parallelism for thread count, defaulting to 1.",
+                %error
+            );
+            std::num::NonZeroUsize::MIN
+        }
+    };
+    usize::from(count)
+}

lib/vector-lib/src/lib.rs

@@ -13,8 +13,8 @@ pub use vector_common::{
     Error, NamedInternalEvent, Result, TimeZone, assert_event_data_eq, atomic, btreemap,
     byte_size_of, byte_size_of::ByteSizeOf, conversion, counter, encode_logfmt, extra_context,
     finalization, finalizer, gauge, histogram, id, impl_event_data_eq, internal_event, json_size,
-    registered_event, request_metadata, sampling, sensitive_string, shutdown,
-    spawn_in_current_span, stats, trigger,
+    num_threads, registered_event, request_metadata, sampling, sensitive_string, shutdown,
+    spawn_in_current_span, spawn_named, stats, trigger,
 };
 pub use vector_config as configurable;
 pub use vector_config::impl_generate_config_from_default;

src/app.rs

@@ -600,7 +600,7 @@ pub fn build_runtime(
     rt_builder.max_blocking_threads(20_000);
     rt_builder.enable_all().thread_name(thread_name);
 
-    let threads = threads.unwrap_or_else(crate::num_threads);
+    let threads = threads.unwrap_or_else(vector_lib::num_threads);
     if threads == 0 {
         error!("The `threads` argument must be greater or equal to 1.");
         return Err(exitcode::CONFIG);

src/lib.rs

@@ -243,37 +243,3 @@ pub fn get_hostname() -> std::io::Result<String> {
 }
 
 pub(crate) use vector_lib::spawn_in_current_span;
-
-/// Spawn a task with the given name. The name is only used if
-/// built with [`tokio_unstable`][tokio_unstable].
-///
-/// [tokio_unstable]: https://docs.rs/tokio/latest/tokio/#unstable-features
-#[track_caller]
-pub(crate) fn spawn_named<T>(
-    task: impl std::future::Future<Output = T> + Send + 'static,
-    _name: &str,
-) -> tokio::task::JoinHandle<T>
-where
-    T: Send + 'static,
-{
-    #[cfg(tokio_unstable)]
-    return tokio::task::Builder::new()
-        .name(_name)
-        .spawn(task)
-        .expect("tokio task should spawn");
-
-    #[cfg(not(tokio_unstable))]
-    tokio::spawn(task)
-}
-
-/// Returns an estimate of the number of recommended threads that Vector should spawn.
-pub fn num_threads() -> usize {
-    let count = match std::thread::available_parallelism() {
-        Ok(count) => count,
-        Err(error) => {
-            warn!(message = "Failed to determine available parallelism for thread count, defaulting to 1.", %error);
-            std::num::NonZeroUsize::MIN
-        }
-    };
-    usize::from(count)
-}

src/sources/aws_s3/sqs.rs

@@ -353,7 +353,7 @@ impl Ingestor {
             max_number_of_messages: config.max_number_of_messages as i32,
             client_concurrency: config
                 .client_concurrency
-                .map_or_else(crate::num_threads, NonZeroUsize::get),
+                .map_or_else(vector_lib::num_threads, NonZeroUsize::get),
             visibility_timeout_secs: config.visibility_timeout_secs as i32,
             delete_message: config.delete_message,
             delete_failed_message: config.delete_failed_message,

src/sources/aws_sqs/config.rs

@@ -119,7 +119,7 @@ impl SourceConfig for AwsSqsConfig {
                 poll_secs: self.poll_secs,
                 concurrency: self
                     .client_concurrency
-                    .map_or_else(crate::num_threads, NonZeroUsize::get),
+                    .map_or_else(vector_lib::num_threads, NonZeroUsize::get),
                 visibility_timeout_secs: self.visibility_timeout_secs,
                 delete_message: self.delete_message,
                 acknowledgements,

src/sources/util/net/tcp/mod.rs

@@ -178,7 +178,7 @@ where
             let shutdown_clone = cx.shutdown.clone();
 
             let request_limiter =
-                RequestLimiter::new(MAX_IN_FLIGHT_EVENTS_TARGET, crate::num_threads());
+                RequestLimiter::new(MAX_IN_FLIGHT_EVENTS_TARGET, vector_lib::num_threads());
 
             listener
                 .accept_stream_limited(max_connections)

src/topology/builder.rs

@@ -37,6 +37,7 @@ use vector_lib::{
     latency::LatencyRecorder,
     schema::Definition,
     source_sender::{DEFAULT_CHUNK_SIZE_EVENTS, SourceSenderItem},
+    spawn_named,
     transform::update_runtime_schema_definition,
 };
 use vector_lib::{gauge, internal_event::GaugeName};
@@ -59,7 +60,6 @@ use crate::{
     event::{EventArray, EventContainer},
     internal_events::EventsReceived,
     shutdown::SourceShutdownCoordinator,
-    spawn_named,
     topology::task::TaskError,
     transforms::{SyncTransform, TaskTransform, Transform, TransformOutputs, TransformOutputsBuf},
     utilization::{
@@ -121,7 +121,7 @@ pub(crate) fn set_ready_array_capacity(size: usize) {
 pub(crate) const TOPOLOGY_BUFFER_SIZE: NonZeroUsize = NonZeroUsize::new(100).unwrap();
 
 static TRANSFORM_CONCURRENCY_LIMIT: LazyLock<usize> = LazyLock::new(|| {
-    crate::app::worker_threads().map_or_else(crate::num_threads, NonZeroUsize::get)
+    crate::app::worker_threads().map_or_else(vector_lib::num_threads, NonZeroUsize::get)
 });
 
 const INTERNAL_SOURCES: [&str; 2] = ["internal_logs", "internal_metrics"];

src/topology/running.rs

@@ -18,6 +18,7 @@ use vector_lib::{
     gauge,
     internal_event::GaugeName,
     shutdown::ShutdownSignal,
+    spawn_named,
     tap::topology::{TapOutput, TapResource, WatchRx, WatchTx},
     trigger::DisabledTrigger,
 };
@@ -37,7 +38,6 @@ use crate::{
     event::EventArray,
     shutdown::SourceShutdownCoordinator,
     signal::ShutdownError,
-    spawn_named,
     utilization::UtilizationRegistry,
 };