chore: replace num_cpus with available_parallelism (#2946)

Signed-off-by: tison <[email protected]>

Author: tisonkun

futures-executor/Cargo.toml

@@ -13,13 +13,12 @@ Executors for asynchronous tasks based on the futures-rs library.
 [features]
 default = ["std"]
 std = ["futures-core/std", "futures-task/std", "futures-util/std"]
-thread-pool = ["std", "num_cpus"]
+thread-pool = ["std"]
 
 [dependencies]
 futures-core = { path = "../futures-core", version = "=1.0.0-alpha.0", default-features = false }
 futures-task = { path = "../futures-task", version = "=0.4.0-alpha.0", default-features = false }
 futures-util = { path = "../futures-util", version = "=0.4.0-alpha.0", default-features = false }
-num_cpus = { version = "1.8.0", optional = true }
 
 [dev-dependencies]
 futures = { path = "../futures", features = ["thread-pool"] }

futures-executor/src/thread_pool.rs

@@ -6,7 +6,6 @@ use futures_task::{waker_ref, ArcWake};
 use futures_task::{FutureObj, Spawn, SpawnError};
 use futures_util::future::FutureExt;
 use std::boxed::Box;
-use std::cmp;
 use std::fmt;
 use std::format;
 use std::io;
@@ -190,13 +189,8 @@ impl ThreadPoolBuilder {
     ///
     /// See the other methods on this type for details on the defaults.
     pub fn new() -> Self {
-        Self {
-            pool_size: cmp::max(1, num_cpus::get()),
-            stack_size: 0,
-            name_prefix: None,
-            after_start: None,
-            before_stop: None,
-        }
+        let pool_size = thread::available_parallelism().map_or(1, |p| p.get());
+        Self { pool_size, stack_size: 0, name_prefix: None, after_start: None, before_stop: None }
     }
 
     /// Set size of a future ThreadPool
@@ -283,7 +277,7 @@ impl ThreadPoolBuilder {
             let before_stop = self.before_stop.clone();
             let mut thread_builder = thread::Builder::new();
             if let Some(ref name_prefix) = self.name_prefix {
-                thread_builder = thread_builder.name(format!("{}{}", name_prefix, counter));
+                thread_builder = thread_builder.name(format!("{name_prefix}{counter}"));
             }
             if self.stack_size > 0 {
                 thread_builder = thread_builder.stack_size(self.stack_size);

futures-util/Cargo.toml

@@ -16,8 +16,8 @@ std = ["alloc", "futures-core/std", "futures-task/std", "slab/std"]
 alloc = ["futures-core/alloc", "futures-task/alloc", "slab"]
 async-await = []
 async-await-macro = ["async-await", "futures-macro"]
-compat = ["std", "futures_01"]
-io-compat = ["io", "compat", "tokio-io"]
+compat = ["std", "futures_01", "libc"]
+io-compat = ["io", "compat", "tokio-io", "libc"]
 sink = ["futures-sink"]
 io = ["std", "futures-io", "memchr"]
 channel = ["std", "futures-channel"]
@@ -44,6 +44,9 @@ tokio-io = { version = "0.1.9", optional = true }
 pin-project-lite = "0.2.6"
 spin = { version = "0.10.0", optional = true }
 
+# INDIRECT DEPENDENCYS BUT ONLY FOR SPECIFIC MINIMAL VERSIONS
+libc = { version = "0.2.26", optional = true }
+
 [dev-dependencies]
 futures = { path = "../futures", features = ["async-await", "thread-pool"] }
 futures-test = { path = "../futures-test" }

futures-util/src/lock/mutex.rs

@@ -553,8 +553,8 @@ mod tests {
     fn test_mutex_guard_debug_not_recurse() {
         let mutex = Mutex::new(42);
         let guard = mutex.try_lock().unwrap();
-        let _ = format!("{:?}", guard);
+        let _ = format!("{guard:?}");
         let guard = MutexGuard::map(guard, |n| n);
-        let _ = format!("{:?}", guard);
+        let _ = format!("{guard:?}");
     }
 }

futures-util/src/stream/futures_unordered/ready_to_run_queue.rs

@@ -69,7 +69,7 @@ impl<Fut> ReadyToRunQueue<Fut> {
                 return Dequeue::Data(tail);
             }
 
-            if self.head.load(Acquire) as *const _ != tail {
+            if !core::ptr::eq(self.head.load(Acquire), tail) {
                 return Dequeue::Inconsistent;
             }
 

futures-util/src/stream/stream/try_for_each_concurrent.rs

@@ -1,6 +1,5 @@
 use crate::stream::{FuturesUnordered, StreamExt};
 use core::fmt;
-use core::mem;
 use core::num::NonZeroUsize;
 use core::pin::Pin;
 use futures_core::future::{FusedFuture, Future};
@@ -113,7 +112,7 @@ where
                     // Empty the stream and futures so that we know
                     // the future has completed.
                     this.stream.set(None);
-                    drop(mem::replace(this.futures, FuturesUnordered::new()));
+                    drop(core::mem::take(this.futures));
                     return Poll::Ready(Err(e));
                 }
             }

futures/tests/oneshot.rs

@@ -68,11 +68,11 @@ fn oneshot_drop_rx() {
 #[test]
 fn oneshot_debug() {
     let (tx, rx) = oneshot::channel::<i32>();
-    assert_eq!(format!("{:?}", tx), "Sender { complete: false }");
-    assert_eq!(format!("{:?}", rx), "Receiver { complete: false }");
+    assert_eq!(format!("{tx:?}"), "Sender { complete: false }");
+    assert_eq!(format!("{rx:?}"), "Receiver { complete: false }");
     drop(rx);
-    assert_eq!(format!("{:?}", tx), "Sender { complete: true }");
+    assert_eq!(format!("{tx:?}"), "Sender { complete: true }");
     let (tx, rx) = oneshot::channel::<i32>();
     drop(tx);
-    assert_eq!(format!("{:?}", rx), "Receiver { complete: true }");
+    assert_eq!(format!("{rx:?}"), "Receiver { complete: true }");
 }