chore(core): Drop use of num_cpus crate (#13197)

As of the release of Rust 1.61, [`std::thread::available_parallelism` now
takes cgroup quotas into account](rust-lang/rust#92697).
This puts `available_parallelism` into feature parity with `num_cpus`
for our purposes, allowing us to drop our use of this crate.

Author: bruceg

Cargo.lock

@@ -266,7 +266,6 @@ nats = { version = "0.20.1", default-features = false, optional = true }
 nkeys = { version = "0.2.0", default-features = false, optional = true }
 nom = { version = "7.1.1", default-features = false, optional = true }
 notify = { version = "4.0.17", default-features = false }
-num_cpus = { version = "1.13.1", default-features = false }
 once_cell = { version = "1.12", default-features = false }
 openssl = { version = "0.10.40", default-features = false, features = ["vendored"] }
 openssl-probe = { version = "0.1.5", default-features = false }

Cargo.toml

@@ -159,3 +159,9 @@ where
     #[cfg(not(tokio_unstable))]
     tokio::spawn(task)
 }
+
+pub fn num_threads() -> usize {
+    std::thread::available_parallelism()
+        .expect("Could not determine available parallelism")
+        .into()
+}

src/lib.rs

@@ -110,7 +110,7 @@ const fn default_true() -> bool {
 }
 
 fn default_client_concurrency() -> u32 {
-    cmp::max(1, num_cpus::get() as u32)
+    cmp::max(1, crate::num_threads() as u32)
 }
 
 #[derive(Debug, Snafu)]

src/sources/aws_s3/sqs.rs

@@ -143,7 +143,7 @@ const fn default_poll_secs() -> u32 {
 }
 
 fn default_client_concurrency() -> u32 {
-    cmp::max(1, num_cpus::get() as u32)
+    cmp::max(1, crate::num_threads() as u32)
 }
 
 const fn default_visibility_timeout_secs() -> u32 {

src/sources/aws_sqs/config.rs

@@ -155,7 +155,8 @@ where
             let connection_gauge = OpenGauge::new();
             let shutdown_clone = cx.shutdown.clone();
 
-            let request_limiter = RequestLimiter::new(MAX_IN_FLIGHT_EVENTS_TARGET, num_cpus::get());
+            let request_limiter =
+                RequestLimiter::new(MAX_IN_FLIGHT_EVENTS_TARGET, crate::num_threads());
 
             listener
                 .accept_stream_limited(max_connections)

src/sources/util/tcp/mod.rs

@@ -61,7 +61,7 @@ static TRANSFORM_CONCURRENCY_LIMIT: Lazy<usize> = Lazy::new(|| {
     crate::app::WORKER_THREADS
         .get()
         .map(std::num::NonZeroUsize::get)
-        .unwrap_or_else(num_cpus::get)
+        .unwrap_or_else(crate::num_threads)
 });
 
 pub(self) async fn load_enrichment_tables<'a>(

src/topology/builder.rs

@@ -36,7 +36,7 @@ fn fork_test<T: std::future::Future<Output = ()>>(test_name: &'static str, fut:
             // Since we are spawning the runtime from within a forked process, use one worker less
             // to account for the additional process.
             // This adjustment mainly serves to not overload CI workers with low resources.
-            let rt = runtime_constrained(std::cmp::max(1, num_cpus::get() - 1));
+            let rt = runtime_constrained(std::cmp::max(1, crate::num_threads() - 1));
             rt.block_on(fut);
         },
     )