Allow creation of CpuPool to be fallible.

To allow error handling in the face of errors returned by thread
creation.

Author: djg

futures-cpupool/src/lib.rs

@@ -19,7 +19,7 @@
 //! # fn main() {
 //!
 //! // Create a worker thread pool with four threads
-//! let pool = CpuPool::new(4);
+//! let pool = CpuPool::new(4).unwrap();
 //!
 //! // Execute some work on the thread pool, optionally closing over data.
 //! let a = pool.spawn(long_running_future(2));
@@ -40,6 +40,7 @@
 extern crate futures;
 extern crate num_cpus;
 
+use std::io;
 use std::panic::{self, AssertUnwindSafe};
 use std::sync::{Arc, Mutex};
 use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering};
@@ -147,16 +148,24 @@ impl CpuPool {
     ///
     /// ```rust
     /// # use futures_cpupool::{Builder, CpuPool};
+    /// # use std::io;
     /// #
-    /// # fn new(size: usize) -> CpuPool {
+    /// # fn new(size: usize) -> io::Result<CpuPool> {
     /// Builder::new().pool_size(size).create()
     /// # }
     /// ```
     ///
+    /// # Errors
+    ///
+    /// This method yields an [`io::Result`] to capture any failure to
+    /// create the thread at the OS level.
+    ///
+    /// [`io::Result`]: https://doc.rust-lang.org/stable/std/io/type.Result.html
+    ///
     /// # Panics
     ///
     /// Panics if `size == 0`.
-    pub fn new(size: usize) -> CpuPool {
+    pub fn new(size: usize) -> io::Result<CpuPool> {
         Builder::new().pool_size(size).create()
     }
 
@@ -167,12 +176,20 @@ impl CpuPool {
     ///
     /// ```rust
     /// # use futures_cpupool::{Builder, CpuPool};
+    /// # use std::io;
     /// #
-    /// # fn new_num_cpus() -> CpuPool {
+    /// # fn new_num_cpus() -> io::Result<CpuPool> {
     /// Builder::new().create()
     /// # }
     /// ```
-    pub fn new_num_cpus() -> CpuPool {
+    ///
+    /// # Errors
+    ///
+    /// This method yields an [`io::Result`] to capture any failure to
+    /// create the thread at the OS level.
+    ///
+    /// [`io::Result`]: https://doc.rust-lang.org/stable/std/io/type.Result.html
+    pub fn new_num_cpus() -> io::Result<CpuPool> {
         Builder::new().create()
     }
 
@@ -398,10 +415,17 @@ impl Builder {
 
     /// Create CpuPool with configured parameters
     ///
+    /// # Errors
+    ///
+    /// This method yields an [`io::Result`] to capture any failure to
+    /// create the thread at the OS level.
+    ///
+    /// [`io::Result`]: https://doc.rust-lang.org/stable/std/io/type.Result.html
+    ///
     /// # Panics
     ///
     /// Panics if `pool_size == 0`.
-    pub fn create(&mut self) -> CpuPool {
+    pub fn create(&mut self) -> io::Result<CpuPool> {
         let (tx, rx) = mpsc::channel();
         let pool = CpuPool {
             inner: Arc::new(Inner {
@@ -424,9 +448,9 @@ impl Builder {
             if self.stack_size > 0 {
                 thread_builder = thread_builder.stack_size(self.stack_size);
             }
-            thread_builder.spawn(move || inner.work(after_start, before_stop)).unwrap();
+            thread_builder.spawn(move || inner.work(after_start, before_stop))?;
         }
-        return pool
+        Ok(pool)
     }
 }
 

futures-cpupool/tests/smoke.rs

@@ -6,15 +6,15 @@ use std::thread;
 use std::time::Duration;
 
 use futures::future::Future;
-use futures_cpupool::{CpuPool, Builder};
+use futures_cpupool::{Builder, CpuPool};
 
 fn done<T: Send + 'static>(t: T) -> Box<Future<Item = T, Error = ()> + Send> {
     Box::new(futures::future::ok(t))
 }
 
 #[test]
 fn join() {
-    let pool = CpuPool::new(2);
+    let pool = CpuPool::new(2).unwrap();
     let a = pool.spawn(done(1));
     let b = pool.spawn(done(2));
     let res = a.join(b).map(|(a, b)| a + b).wait();
@@ -24,7 +24,7 @@ fn join() {
 
 #[test]
 fn select() {
-    let pool = CpuPool::new(2);
+    let pool = CpuPool::new(2).unwrap();
     let a = pool.spawn(done(1));
     let b = pool.spawn(done(2));
     let (item1, next) = a.select(b).wait().ok().unwrap();
@@ -48,7 +48,7 @@ fn threads_go_away() {
 
     thread_local!(static FOO: A = A);
 
-    let pool = CpuPool::new(2);
+    let pool = CpuPool::new(2).unwrap();
     let _handle = pool.spawn_fn(|| {
         FOO.with(|_| ());
         Ok::<(), ()>(())
@@ -57,7 +57,7 @@ fn threads_go_away() {
 
     for _ in 0..100 {
         if CNT.load(Ordering::SeqCst) == 1 {
-            return
+            return;
         }
         thread::sleep(Duration::from_millis(10));
     }
@@ -81,10 +81,9 @@ fn lifecycle_test() {
         .pool_size(4)
         .after_start(after_start)
         .before_stop(before_stop)
-        .create();
-    let _handle = pool.spawn_fn(|| {
-        Ok::<(), ()>(())
-    });
+        .create()
+        .unwrap();
+    let _handle = pool.spawn_fn(|| Ok::<(), ()>(()));
     drop(pool);
 
     for _ in 0..100 {
@@ -99,9 +98,7 @@ fn lifecycle_test() {
 
 #[test]
 fn thread_name() {
-    let pool = Builder::new()
-        .name_prefix("my-pool-")
-        .create();
+    let pool = Builder::new().name_prefix("my-pool-").create().unwrap();
     let future = pool.spawn_fn(|| {
         assert!(thread::current().name().unwrap().starts_with("my-pool-"));
         Ok::<(), ()>(())