Bump version to 0.1.12 (#718)

Also bumps the following sub-crates:

* tokio-fs (0.1.4)
* tokio-io (0.1.10)
* tokio-signal (0.2.6)
* tokio-threadpool (0.1.8)
* tokio-uds (0.2.3)

CHANGELOG.md

@@ -1,6 +1,18 @@
 This changelog only applies to the `tokio` crate proper. Each sub crate
 maintains its own changelog tracking changes made in each respective sub crate.
 
+# 0.1.12 (October 23, 2018)
+
+* runtime: expose `keep_alive` on runtime builder (#676).
+* runtime: create a reactor per worker thread (#660).
+* codec: fix panic in `LengthDelimitedCodec` (#682).
+* io: re-export `tokio_io::io::read` function (#689).
+* runtime: check for executor re-entry in more places (#708).
+
+# 0.1.11 (September 28, 2018)
+
+* Fix `tokio-async-await` dependency (#675).
+
 # 0.1.10 (September 27, 2018)
 
 * Fix minimal versions

CONTRIBUTING.md

@@ -109,7 +109,7 @@ and dependencies in the Tokio repository.
 Even tiny pull requests (e.g., one character pull request fixing a typo in API
 documentation) are greatly appreciated. Before making a large change, it is
 usually a good idea to first open an issue describing the change to solicit
-feedback and guidance. This will increasethe likelihood of the PR getting
+feedback and guidance. This will increase the likelihood of the PR getting
 merged.
 
 ### Tests
@@ -380,8 +380,8 @@ left). When doing so, it is courteous to give the original contributor credit
 for the work they started (either by preserving their name and email address in
 the commit log, or by using an `Author: ` meta-data tag in the commit.
 
-_Adapted from the [Node.js contributing guide][node]_
+_Adapted from the [Node.js contributing guide][node]_.
 
-[node]: https://github.com/nodejs/node/blob/master/CONTRIBUTING.md.
+[node]: https://github.com/nodejs/node/blob/master/CONTRIBUTING.md
 [hiding-a-comment]: https://help.github.com/articles/managing-disruptive-comments/#hiding-a-comment
 [documentation test]: https://doc.rust-lang.org/rustdoc/documentation-tests.html

Cargo.toml

@@ -6,11 +6,11 @@ name = "tokio"
 # - Update CHANGELOG.md.
 # - Update doc URL.
 # - Create "v0.1.x" git tag.
-version = "0.1.10"
+version = "0.1.12"
 authors = ["Carl Lerche <[email protected]>"]
 license = "MIT"
 readme = "README.md"
-documentation = "https://docs.rs/tokio/0.1.10/tokio/"
+documentation = "https://docs.rs/tokio/0.1.12/tokio/"
 repository = "https://github.com/tokio-rs/tokio"
 homepage = "https://tokio.rs"
 description = """
@@ -54,6 +54,7 @@ appveyor = { repository = "carllerche/tokio", id = "s83yxhy9qeb58va7" }
 
 [dependencies]
 bytes = "0.4"
+num_cpus = "1.8.0"
 tokio-codec = { version = "0.1.0", path = "tokio-codec" }
 tokio-current-thread = { version = "0.1.3", path = "tokio-current-thread" }
 tokio-io = { version = "0.1.6", path = "tokio-io" }
@@ -70,12 +71,12 @@ futures = "0.1.20"
 # Needed until `reactor` is removed from `tokio`.
 mio = "0.6.14"
 
-[target.'cfg(unix)'.dependencies]
-tokio-uds = { version = "0.2.1", path = "tokio-uds" }
-
 # Needed for async/await preview support
 tokio-async-await = { version = "0.1.0", path = "tokio-async-await", optional = true }
 
+[target.'cfg(unix)'.dependencies]
+tokio-uds = { version = "0.2.1", path = "tokio-uds" }
+
 [dev-dependencies]
 env_logger = { version = "0.5", default-features = false }
 flate2 = { version = "1", features = ["tokio"] }
@@ -88,3 +89,20 @@ serde = "1.0"
 serde_derive = "1.0"
 serde_json = "1.0"
 time = "0.1"
+
+[patch.crates-io]
+tokio = { path = "." }
+tokio-async-await  = { path = "./tokio-async-await" }
+tokio-codec = { path = "./tokio-codec" }
+tokio-current-thread = { path = "./tokio-current-thread" }
+tokio-executor = { path = "./tokio-executor" }
+tokio-fs = { path = "./tokio-fs" }
+tokio-io = { path = "./tokio-io" }
+tokio-reactor = { path = "./tokio-reactor" }
+tokio-signal = { path = "./tokio-signal" }
+tokio-tcp = { path = "./tokio-tcp" }
+tokio-threadpool = { path = "./tokio-threadpool" }
+tokio-timer = { path = "./tokio-timer" }
+tokio-tls = { path = "./tokio-tls" }
+tokio-udp = { path = "./tokio-udp" }
+tokio-uds = { path = "./tokio-uds" }

examples/hello_world.rs

@@ -1,59 +1,45 @@
 //! Hello world server.
 //!
-//! A simple server that accepts connections, writes "hello world\n", and closes
+//! A simple client that opens a TCP stream, writes "hello world\n", and closes
 //! the connection.
 //!
 //! You can test this out by running:
 //!
-//!     cargo run --example hello_world
+//!     ncat -l 6142
 //!
 //! And then in another terminal run:
 //!
-//!     telnet localhost 6142
-//!
+//!     cargo run --example hello_world
 
 #![deny(warnings)]
 
 extern crate tokio;
 
 use tokio::io;
-use tokio::net::TcpListener;
+use tokio::net::TcpStream;
 use tokio::prelude::*;
 
 pub fn main() {
     let addr = "127.0.0.1:6142".parse().unwrap();
 
-    // Bind a TCP listener to the socket address.
+    // Open a TCP stream to the socket address.
     //
-    // Note that this is the Tokio TcpListener, which is fully async.
-    let listener = TcpListener::bind(&addr).unwrap();
-
-    // The server task asynchronously iterates over and processes each
-    // incoming connection.
-    let server = listener.incoming().for_each(|socket| {
-        println!("accepted socket; addr={:?}", socket.peer_addr().unwrap());
-
-        let connection = io::write_all(socket, "hello world\n")
-            .then(|res| {
-                println!("wrote message; success={:?}", res.is_ok());
-                Ok(())
-            });
-
-        // Spawn a new task that processes the socket:
-        tokio::spawn(connection);
-
-        Ok(())
+    // Note that this is the Tokio TcpStream, which is fully async.
+    let client = TcpStream::connect(&addr).and_then(|stream| {
+        println!("created stream");
+        io::write_all(stream, "hello world\n").then(|result| {
+            println!("wrote to stream; success={:?}", result.is_ok());
+            Ok(())
+        })
     })
     .map_err(|err| {
         // All tasks must have an `Error` type of `()`. This forces error
         // handling and helps avoid silencing failures.
         //
         // In our example, we are only going to log the error to STDOUT.
-        println!("accept error = {:?}", err);
+        println!("connection error = {:?}", err);
     });
 
-    println!("server running on localhost:6142");
-
     // Start the Tokio runtime.
     //
     // The Tokio is a pre-configured "out of the box" runtime for building
@@ -63,8 +49,7 @@ pub fn main() {
     // This function blocks until the runtime reaches an idle state. Idle is
     // defined as all spawned tasks have completed and all I/O resources (TCP
     // sockets in our case) have been dropped.
-    //
-    // In our example, we have not defined a shutdown strategy, so this will
-    // block until `ctrl-c` is pressed at the terminal.
-    tokio::run(server);
+    println!("About to create the stream and write to it...");
+    tokio::run(client);
+    println!("Stream has been created and written to.");
 }

src/codec/length_delimited.rs

@@ -578,6 +578,10 @@ impl Encoder for LengthDelimitedCodec {
             "provided length would overflow after adjustment",
         ))?;
 
+        // Reserve capacity in the destination buffer to fit the frame and
+        // length field (plus adjustment).
+        dst.reserve(self.builder.length_field_len + n);
+
         if self.builder.length_field_is_big_endian {
             dst.put_uint_be(n as u64, self.builder.length_field_len);
         } else {

src/io.rs

@@ -68,6 +68,7 @@ pub use tokio_io::io::{
     Flush,
     lines,
     Lines,
+    read,
     read_exact,
     ReadExact,
     read_to_end,

src/lib.rs

@@ -1,4 +1,4 @@
-#![doc(html_root_url = "https://docs.rs/tokio/0.1.10")]
+#![doc(html_root_url = "https://docs.rs/tokio/0.1.12")]
 #![deny(missing_docs, warnings, missing_debug_implementations)]
 #![cfg_attr(feature = "async-await-preview", feature(
         async_await,
@@ -76,6 +76,7 @@ extern crate bytes;
 #[macro_use]
 extern crate futures;
 extern crate mio;
+extern crate num_cpus;
 extern crate tokio_current_thread;
 extern crate tokio_io;
 extern crate tokio_executor;

src/runtime/builder.rs

@@ -3,10 +3,12 @@ use runtime::{Inner, Runtime};
 use reactor::Reactor;
 
 use std::io;
+use std::sync::Mutex;
+use std::time::Duration;
 
+use num_cpus;
 use tokio_reactor;
 use tokio_threadpool::Builder as ThreadPoolBuilder;
-use tokio_threadpool::park::DefaultPark;
 use tokio_timer::clock::{self, Clock};
 use tokio_timer::timer::{self, Timer};
 
@@ -50,6 +52,9 @@ pub struct Builder {
     /// Thread pool specific builder
     threadpool_builder: ThreadPoolBuilder,
 
+    /// The number of worker threads
+    core_threads: usize,
+
     /// The clock to use
     clock: Clock,
 }
@@ -60,11 +65,15 @@ impl Builder {
     ///
     /// Configuration methods can be chained on the return value.
     pub fn new() -> Builder {
+        let core_threads = num_cpus::get().max(1);
+
         let mut threadpool_builder = ThreadPoolBuilder::new();
         threadpool_builder.name_prefix("tokio-runtime-worker-");
+        threadpool_builder.pool_size(core_threads);
 
         Builder {
             threadpool_builder,
+            core_threads,
             clock: Clock::new(),
         }
     }
@@ -79,7 +88,8 @@ impl Builder {
     #[deprecated(
         since="0.1.9",
         note="use the `core_threads`, `blocking_threads`, `name_prefix`, \
-              and `stack_size` functions on `runtime::Builder`, instead")]
+              `keep_alive`, and `stack_size` functions on `runtime::Builder`, \
+              instead")]
     #[doc(hidden)]
     pub fn threadpool_builder(&mut self, val: ThreadPoolBuilder) -> &mut Self {
         self.threadpool_builder = val;
@@ -108,6 +118,7 @@ impl Builder {
     /// # }
     /// ```
     pub fn core_threads(&mut self, val: usize) -> &mut Self {
+        self.core_threads = val;
         self.threadpool_builder.pool_size(val);
         self
     }
@@ -142,6 +153,37 @@ impl Builder {
         self
     }
 
+    /// Set the worker thread keep alive duration for threads in the `Runtime`'s
+    /// thread pool.
+    ///
+    /// If set, a worker thread will wait for up to the specified duration for
+    /// work, at which point the thread will shutdown. When work becomes
+    /// available, a new thread will eventually be spawned to replace the one
+    /// that shut down.
+    ///
+    /// When the value is `None`, the thread will wait for work forever.
+    ///
+    /// The default value is `None`.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// # extern crate tokio;
+    /// # extern crate futures;
+    /// # use tokio::runtime;
+    /// use std::time::Duration;
+    ///
+    /// # pub fn main() {
+    /// let mut rt = runtime::Builder::new()
+    ///     .keep_alive(Some(Duration::from_secs(30)))
+    ///     .build();
+    /// # }
+    /// ```
+    pub fn keep_alive(&mut self, val: Option<Duration>) -> &mut Self {
+        self.threadpool_builder.keep_alive(val);
+        self
+    }
+
     /// Set name prefix of threads spawned by the `Runtime`'s thread pool.
     ///
     /// Thread name prefix is used for generating thread names. For example, if
@@ -210,44 +252,50 @@ impl Builder {
     /// # }
     /// ```
     pub fn build(&mut self) -> io::Result<Runtime> {
-        use std::collections::HashMap;
-        use std::sync::{Arc, Mutex};
+        // TODO(stjepang): Once we remove the `threadpool_builder` method, remove this line too.
+        self.threadpool_builder.pool_size(self.core_threads);
+
+        let mut reactor_handles = Vec::new();
+        let mut timer_handles = Vec::new();
+        let mut timers = Vec::new();
+
+        for _ in 0..self.core_threads {
+            // Create a new reactor.
+            let reactor = Reactor::new()?;
+            reactor_handles.push(reactor.handle());
+
+            // Create a new timer.
+            let timer = Timer::new_with_now(reactor, self.clock.clone());
+            timer_handles.push(timer.handle());
+            timers.push(Mutex::new(Some(timer)));
+        }
 
         // Get a handle to the clock for the runtime.
-        let clock1 = self.clock.clone();
-        let clock2 = clock1.clone();
+        let clock = self.clock.clone();
 
-        let timers = Arc::new(Mutex::new(HashMap::<_, timer::Handle>::new()));
-        let t1 = timers.clone();
-
-        // Spawn a reactor on a background thread.
-        let reactor = Reactor::new()?.background()?;
-
-        // Get a handle to the reactor.
-        let reactor_handle = reactor.handle().clone();
+        // Get a handle to the first reactor.
+        let reactor = reactor_handles[0].clone();
 
         let pool = self.threadpool_builder
             .around_worker(move |w, enter| {
-                let timer_handle = t1.lock().unwrap()
-                    .get(w.id()).unwrap()
-                    .clone();
+                let index = w.id().to_usize();
 
-                tokio_reactor::with_default(&reactor_handle, enter, |enter| {
-                    clock::with_default(&clock1, enter, |enter| {
-                        timer::with_default(&timer_handle, enter, |_| {
+                tokio_reactor::with_default(&reactor_handles[index], enter, |enter| {
+                    clock::with_default(&clock, enter, |enter| {
+                        timer::with_default(&timer_handles[index], enter, |_| {
                             w.run();
                         });
                     })
                 });
             })
             .custom_park(move |worker_id| {
-                // Create a new timer
-                let timer = Timer::new_with_now(DefaultPark::new(), clock2.clone());
+                let index = worker_id.to_usize();
 
-                timers.lock().unwrap()
-                    .insert(worker_id.clone(), timer.handle());
-
-                timer
+                timers[index]
+                    .lock()
+                    .unwrap()
+                    .take()
+                    .unwrap()
             })
             .build();
 

src/runtime/mod.rs

@@ -121,7 +121,7 @@ pub use self::builder::Builder;
 pub use self::shutdown::Shutdown;
 pub use self::task_executor::TaskExecutor;
 
-use reactor::{Background, Handle};
+use reactor::Handle;
 
 use std::io;
 
@@ -152,8 +152,8 @@ pub struct Runtime {
 
 #[derive(Debug)]
 struct Inner {
-    /// Reactor running on a background thread.
-    reactor: Background,
+    /// A handle to one of the per-worker reactors.
+    reactor: Handle,
 
     /// Task execution pool.
     pool: threadpool::ThreadPool,
@@ -209,11 +209,13 @@ struct Inner {
 pub fn run<F>(future: F)
 where F: Future<Item = (), Error = ()> + Send + 'static,
 {
-    let mut runtime = Runtime::new().unwrap();
+    // Check enter before creating a new Runtime...
+    let mut entered = enter().expect("nested tokio::run");
+    let mut runtime = Runtime::new().expect("failed to start new Runtime");
     runtime.spawn(future);
-    enter().expect("nested tokio::run")
+    entered
         .block_on(runtime.shutdown_on_idle())
-        .unwrap();
+        .expect("shutdown cannot error")
 }
 
 impl Runtime {
@@ -254,6 +256,7 @@ impl Runtime {
     #[deprecated(since = "0.1.5", note = "use `reactor` instead")]
     #[doc(hidden)]
     pub fn handle(&self) -> &Handle {
+        #[allow(deprecated)]
         self.reactor()
     }
 
@@ -275,8 +278,9 @@ impl Runtime {
     ///
     /// // use `reactor_handle`
     /// ```
+    #[deprecated(since = "0.1.11", note = "there is now a reactor per worker thread")]
     pub fn reactor(&self) -> &Handle {
-        self.inner().reactor.handle()
+        &self.inner().reactor
     }
 
     /// Return a handle to the runtime's executor.
@@ -360,9 +364,10 @@ impl Runtime {
         R: Send + 'static,
         E: Send + 'static,
     {
+        let mut entered = enter().expect("nested block_on");
         let (tx, rx) = futures::sync::oneshot::channel();
         self.spawn(future.then(move |r| tx.send(r).map_err(|_| unreachable!())));
-        rx.wait().unwrap()
+        entered.block_on(rx).unwrap()
     }
 
     /// Run a future to completion on the Tokio runtime, then wait for all
@@ -385,9 +390,16 @@ impl Runtime {
         R: Send + 'static,
         E: Send + 'static,
     {
-        let res = self.block_on(future);
-        self.shutdown_on_idle().wait().unwrap();
-        res
+        let mut entered = enter().expect("nested block_on_all");
+        let (tx, rx) = futures::sync::oneshot::channel();
+        self.spawn(future.then(move |r| tx.send(r).map_err(|_| unreachable!())));
+        let block = rx
+            .map_err(|_| unreachable!())
+            .and_then(move |r| {
+                self.shutdown_on_idle()
+                    .map(move |()| r)
+            });
+        entered.block_on(block).unwrap()
     }
 
     /// Signals the runtime to shutdown once it becomes idle.
@@ -424,16 +436,7 @@ impl Runtime {
     /// [mod]: index.html
     pub fn shutdown_on_idle(mut self) -> Shutdown {
         let inner = self.inner.take().unwrap();
-
-        let inner = Box::new({
-            let pool = inner.pool;
-            let reactor = inner.reactor;
-
-            pool.shutdown_on_idle().and_then(|_| {
-                reactor.shutdown_on_idle()
-            })
-        });
-
+        let inner = inner.pool.shutdown_on_idle();
         Shutdown { inner }
     }
 

src/runtime/shutdown.rs

@@ -1,4 +1,5 @@
 use runtime::Inner;
+use tokio_threadpool as threadpool;
 
 use std::fmt;
 
@@ -6,23 +7,12 @@ use futures::{Future, Poll};
 
 /// A future that resolves when the Tokio `Runtime` is shut down.
 pub struct Shutdown {
-    pub(super) inner: Box<Future<Item = (), Error = ()> + Send>,
+    pub(super) inner: threadpool::Shutdown,
 }
 
 impl Shutdown {
     pub(super) fn shutdown_now(inner: Inner) -> Self {
-        let inner = Box::new({
-            let pool = inner.pool;
-            let reactor = inner.reactor;
-
-            pool.shutdown_now().and_then(|_| {
-                reactor.shutdown_now()
-                    .then(|_| {
-                        Ok(())
-                    })
-            })
-        });
-
+        let inner = inner.pool.shutdown_now();
         Shutdown { inner }
     }
 }

tests/length_delimited.rs

@@ -5,7 +5,7 @@ extern crate bytes;
 use tokio::io::{AsyncRead, AsyncWrite};
 use tokio::codec::*;
 
-use bytes::Bytes;
+use bytes::{Bytes, BytesMut, BufMut};
 use futures::{Stream, Sink, Poll};
 use futures::Async::*;
 
@@ -483,6 +483,22 @@ fn write_zero() {
     assert!(io.get_ref().calls.is_empty());
 }
 
+#[test]
+fn encode_overflow() {
+    // Test reproducing tokio-rs/tokio#681.
+    let mut codec = length_delimited::Builder::new().new_codec();
+    let mut buf = BytesMut::with_capacity(1024);
+
+    // Put some data into the buffer without resizing it to hold more.
+    let some_as = std::iter::repeat(b'a')
+        .take(1024)
+        .collect::<Vec<_>>();
+    buf.put_slice(&some_as[..]);
+
+    // Trying to encode the length header should resize the buffer if it won't fit.
+    codec.encode(Bytes::from("hello"), &mut buf).unwrap();
+}
+
 // ===== Test utils =====
 
 fn would_block() -> io::Error {

tests/runtime.rs

@@ -391,14 +391,78 @@ mod from_block_on_all {
     }
 }
 
-#[test]
-fn run_in_run() {
+mod nested_enter {
+    use super::*;
+    use tokio::runtime::current_thread;
     use std::panic;
 
-    tokio::run(lazy(|| {
-        panic::catch_unwind(|| {
-            tokio::run(lazy(|| { Ok::<(), ()>(()) }))
-        }).unwrap_err();
-        Ok::<(), ()>(())
-    }));
+    fn test<F1, F2>(first: F1, nested: F2)
+    where
+        F1: Fn(Box<Future<Item=(), Error=()> + Send>) + Send + 'static,
+        F2: Fn(Box<Future<Item=(), Error=()> + Send>) + panic::UnwindSafe + Send + 'static,
+    {
+        let panicked = Arc::new(Mutex::new(false));
+        let panicked2 = panicked.clone();
+
+        // Since this is testing panics in other threads, printing about panics
+        // is noisy and can give the impression that the test is ignoring panics.
+        //
+        // It *is* ignoring them, but on purpose.
+        let prev_hook = panic::take_hook();
+        panic::set_hook(Box::new(|info| {
+            let s = info.to_string();
+            if s.starts_with("panicked at 'nested ")
+                || s.starts_with("panicked at 'Multiple executors at once")
+            {
+                // expected, noop
+            } else {
+                println!("{}", s);
+            }
+        }));
+
+        first(Box::new(lazy(move || {
+            panic::catch_unwind(move || {
+                nested(Box::new(lazy(|| { Ok::<(), ()>(()) })))
+            }).expect_err("nested should panic");
+            *panicked2.lock().unwrap() = true;
+            Ok::<(), ()>(())
+        })));
+
+        panic::set_hook(prev_hook);
+
+        assert!(*panicked.lock().unwrap(), "nested call should have panicked");
+    }
+
+    fn threadpool_new() -> Runtime {
+        Runtime::new().expect("rt new")
+    }
+
+    #[test]
+    fn run_in_run() {
+        test(tokio::run, tokio::run);
+    }
+
+    #[test]
+    fn threadpool_block_on_in_run() {
+        test(tokio::run, |fut| {
+            let mut rt = threadpool_new();
+            rt.block_on(fut).unwrap();
+        });
+    }
+
+    #[test]
+    fn threadpool_block_on_all_in_run() {
+        test(tokio::run, |fut| {
+            let rt = threadpool_new();
+            rt.block_on_all(fut).unwrap();
+        });
+    }
+
+    #[test]
+    fn current_thread_block_on_all_in_run() {
+        test(tokio::run, |fut| {
+            current_thread::block_on_all(fut).unwrap();
+        });
+    }
 }
+

tokio-fs/CHANGELOG.md

@@ -1,3 +1,7 @@
+# 0.1.4 (October 23, 2018)
+
+* Provide `File::from_std` (#696).
+
 # 0.1.3 (August 6, 2018)
 
 * Add async equivalents to most of `std::fs` (#494).

tokio-fs/Cargo.toml

@@ -4,14 +4,15 @@ name = "tokio-fs"
 # When releasing to crates.io:
 # - Update html_root_url.
 # - Update CHANGELOG.md.
+# - Update doc URL.
 # - Create "v0.1.x" git tag.
-version = "0.1.3"
+version = "0.1.4"
 authors = ["Carl Lerche <[email protected]>"]
 license = "MIT"
 readme = "README.md"
 repository = "https://github.com/tokio-rs/tokio"
 homepage = "https://tokio.rs"
-documentation = "https://docs.rs/tokio-fs/0.1"
+documentation = "https://docs.rs/tokio-fs/0.1.4/tokio_fs"
 description = """
 Filesystem API for Tokio.
 """

tokio-fs/src/file/metadata.rs

@@ -8,7 +8,7 @@ use std::io;
 
 const POLL_AFTER_RESOLVE: &str = "Cannot poll MetadataFuture after it resolves";
 
-/// Future returned by `File::metadata` and resolves to a `(Metadata, File)` instance.
+/// Future returned by `File::metadata` and resolves to a `(File, Metadata)` instance.
 #[derive(Debug)]
 pub struct MetadataFuture {
     file: Option<File>,

tokio-fs/src/file/mod.rs

@@ -82,7 +82,7 @@ impl File {
     /// Convert a [`std::fs::File`][std] to a `tokio_fs::File`.
     ///
     /// [std]: https://doc.rust-lang.org/std/fs/struct.File.html
-    pub(crate) fn from_std(std: StdFile) -> File {
+    pub fn from_std(std: StdFile) -> File {
         File { std: Some(std) }
     }
 

tokio-fs/src/lib.rs

@@ -1,3 +1,6 @@
+#![deny(missing_docs, missing_debug_implementations, warnings)]
+#![doc(html_root_url = "https://docs.rs/tokio-fs/0.1.4")]
+
 //! Asynchronous file and standard stream adaptation.
 //!
 //! This module contains utility methods and adapter types for input/output to
@@ -25,9 +28,6 @@
 //! [`AsyncRead`]: https://docs.rs/tokio-io/0.1/tokio_io/trait.AsyncRead.html
 //! [tokio-threadpool]: https://docs.rs/tokio-threadpool/0.1/tokio_threadpool
 
-#![deny(missing_docs, missing_debug_implementations, warnings)]
-#![doc(html_root_url = "https://docs.rs/tokio-fs/0.1.3")]
-
 #[macro_use]
 extern crate futures;
 extern crate tokio_io;

tokio-io/CHANGELOG.md

@@ -1,3 +1,8 @@
+# 0.1.10 (October 23, 2018)
+
+* Expose inner codec from `Framed` (#686).
+* Implement AsyncRead::prepare_uninitialized_buffer for Take and Chain (#678).
+
 # 0.1.9 (September 27, 2018)
 
 * Fix bug in `AsyncRead::split()` (#655).

tokio-io/Cargo.toml

@@ -6,12 +6,12 @@ name = "tokio-io"
 # - Update CHANGELOG.md.
 # - Update doc URL.
 # - Create "v0.1.x" git tag.
-version = "0.1.9"
+version = "0.1.10"
 authors = ["Carl Lerche <[email protected]>"]
 license = "MIT"
 repository = "https://github.com/tokio-rs/tokio"
 homepage = "https://tokio.rs"
-documentation = "https://docs.rs/tokio-io/0.1.9/tokio_io"
+documentation = "https://docs.rs/tokio-io/0.1.10/tokio_io"
 description = """
 Core I/O primitives for asynchronous I/O in Rust.
 """

tokio-io/src/_tokio_codec/framed.rs

@@ -97,6 +97,24 @@ impl<T, U> Framed<T, U> {
         &mut self.inner.get_mut().get_mut().0
     }
 
+    /// Returns a reference to the underlying codec wrapped by
+    /// `Frame`.
+    ///
+    /// Note that care should be taken to not tamper with the underlying codec
+    /// as it may corrupt the stream of frames otherwise being worked with.
+    pub fn codec(&self) -> &U {
+        &self.inner.get_ref().get_ref().1
+    }
+
+    /// Returns a mutable reference to the underlying codec wrapped by
+    /// `Frame`.
+    ///
+    /// Note that care should be taken to not tamper with the underlying codec
+    /// as it may corrupt the stream of frames otherwise being worked with.
+    pub fn codec_mut(&mut self) -> &mut U {
+        &mut self.inner.get_mut().get_mut().1
+    }
+
     /// Consumes the `Frame`, returning its underlying I/O stream.
     ///
     /// Note that care should be taken to not tamper with the underlying stream

tokio-io/src/async_read.rs

@@ -13,18 +13,18 @@ use split::{ReadHalf, WriteHalf};
 /// **non-blocking**. All non-blocking I/O objects must return an error when
 /// bytes are unavailable instead of blocking the current thread.
 ///
-/// Specifically, this means that the `read` function will return one of the
-/// following:
+/// Specifically, this means that the `poll_read` function will return one of
+/// the following:
 ///
 /// * `Ok(Async::Ready(n))` means that `n` bytes of data was immediately read
 ///   and placed into the output buffer, where `n` == 0 implies that EOF has
 ///   been reached.
 ///
-/// * `Ok(Async::NotReady)` means that no data was read into the buffer 
+/// * `Ok(Async::NotReady)` means that no data was read into the buffer
 ///   provided. The I/O object is not currently readable but may become readable
 ///   in the future. Most importantly, **the current future's task is scheduled
-///   to get unparked when the object is readable**. This means that like 
-///   `Future::poll` you'll receive a notification when the I/O object is 
+///   to get unparked when the object is readable**. This means that like
+///   `Future::poll` you'll receive a notification when the I/O object is
 ///   readable again.
 ///
 /// * `Err(e)` for other errors are standard I/O errors coming from the
@@ -43,13 +43,13 @@ pub trait AsyncRead: std_io::Read {
     /// zero out buffers can be expensive.
     ///
     /// This function does any necessary work to prepare an uninitialized buffer
-    /// to be safe to pass to `read`. If `read` guarantees to never attempt read
-    /// data out of the supplied buffer, then `prepare_uninitialized_buffer`
+    /// to be safe to pass to `read`. If `read` guarantees to never attempt to
+    /// read data out of the supplied buffer, then `prepare_uninitialized_buffer`
     /// doesn't need to do any work.
     ///
     /// If this function returns `true`, then the memory has been zeroed out.
     /// This allows implementations of `AsyncRead` which are composed of
-    /// multiple sub implementations to efficiently implement
+    /// multiple subimplementations to efficiently implement
     /// `prepare_uninitialized_buffer`.
     ///
     /// This function isn't actually `unsafe` to call but `unsafe` to implement.
@@ -74,7 +74,7 @@ pub trait AsyncRead: std_io::Read {
     /// On success, returns `Ok(Async::Ready(num_bytes_read))`.
     ///
     /// If no data is available for reading, the method returns
-    /// `Ok(Async::Pending)` and arranges for the current task (via
+    /// `Ok(Async::NotReady)` and arranges for the current task (via
     /// `cx.waker()`) to receive a notification when the object becomes
     /// readable or is closed.
     fn poll_read(&mut self, buf: &mut [u8]) -> Poll<usize, std_io::Error> {
@@ -87,7 +87,7 @@ pub trait AsyncRead: std_io::Read {
         }
     }
 
-    /// Pull some bytes from this source into the specified `Buf`, returning
+    /// Pull some bytes from this source into the specified `BufMut`, returning
     /// how many bytes were read.
     ///
     /// The `buf` provided will have bytes read into it and the internal cursor
@@ -115,7 +115,7 @@ pub trait AsyncRead: std_io::Read {
     }
 
     /// Provides a `Stream` and `Sink` interface for reading and writing to this
-    /// `Io` object, using `Decode` and `Encode` to read and write the raw data.
+    /// I/O object, using `Decode` and `Encode` to read and write the raw data.
     ///
     /// Raw I/O objects work with byte sequences, but higher-level code usually
     /// wants to batch these into meaningful chunks, called "frames". This

tokio-io/src/async_write.rs

@@ -10,8 +10,8 @@ use AsyncRead;
 /// **nonblocking**. All non-blocking I/O objects must return an error when
 /// bytes cannot be written instead of blocking the current thread.
 ///
-/// Specifically, this means that the `write` function will return one of the
-/// following:
+/// Specifically, this means that the `poll_write` function will return one of
+/// the following:
 ///
 /// * `Ok(Async::Ready(n))` means that `n` bytes of data was immediately
 ///   written.
@@ -19,7 +19,7 @@ use AsyncRead;
 /// * `Ok(Async::NotReady)` means that no data was written from the buffer
 ///   provided. The I/O object is not currently writable but may become writable
 ///   in the future. Most importantly, **the current future's task is scheduled
-///   to get unparked when the object is readable**. This means that like
+///   to get unparked when the object is writable**. This means that like
 ///   `Future::poll` you'll receive a notification when the I/O object is
 ///   writable again.
 ///
@@ -40,7 +40,7 @@ pub trait AsyncWrite: std_io::Write {
     /// On success, returns `Ok(Async::Ready(num_bytes_written))`.
     ///
     /// If the object is not ready for writing, the method returns
-    /// `Ok(Async::Pending)` and arranges for the current task (via
+    /// `Ok(Async::NotReady)` and arranges for the current task (via
     /// `cx.waker()`) to receive a notification when the object becomes
     /// readable or is closed.
     fn poll_write(&mut self, buf: &[u8]) -> Poll<usize, std_io::Error> {
@@ -59,7 +59,7 @@ pub trait AsyncWrite: std_io::Write {
     /// On success, returns `Ok(Async::Ready(()))`.
     ///
     /// If flushing cannot immediately complete, this method returns
-    /// `Ok(Async::Pending)` and arranges for the current task (via
+    /// `Ok(Async::NotReady)` and arranges for the current task (via
     /// `cx.waker()`) to receive a notification when the object can make
     /// progress towards flushing.
     fn poll_flush(&mut self) -> Poll<(), std_io::Error> {
@@ -172,17 +172,22 @@ impl AsyncWrite for std_io::Sink {
     }
 }
 
-// TODO: Implement `prepare_uninitialized_buffer` for `io::Take`.
-// This is blocked on rust-lang/rust#27269
 impl<T: AsyncRead> AsyncRead for std_io::Take<T> {
+    unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [u8]) -> bool {
+        self.get_ref().prepare_uninitialized_buffer(buf)
+    }
 }
 
-// TODO: Implement `prepare_uninitialized_buffer` when upstream exposes inner
-// parts
 impl<T, U> AsyncRead for std_io::Chain<T, U>
     where T: AsyncRead,
           U: AsyncRead,
 {
+    unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [u8]) -> bool {
+        let (t, u) = self.get_ref();
+        // We don't need to execute the second initializer if the first one
+        // already zeroed the buffer out.
+        t.prepare_uninitialized_buffer(buf) || u.prepare_uninitialized_buffer(buf)
+    }
 }
 
 impl<T: AsyncWrite> AsyncWrite for std_io::BufWriter<T> {

tokio-io/src/lib.rs

@@ -1,5 +1,5 @@
 #![deny(missing_docs, missing_debug_implementations, warnings)]
-#![doc(html_root_url = "https://docs.rs/tokio-io/0.1.9")]
+#![doc(html_root_url = "https://docs.rs/tokio-io/0.1.10")]
 
 //! Core I/O traits and combinators when working with Tokio.
 //!

tokio-reactor/Cargo.toml

@@ -33,3 +33,4 @@ tokio-io = { version = "0.1.6", path = "../tokio-io" }
 [dev-dependencies]
 num_cpus = "1.8.0"
 tokio = { version = "0.1.7", path = ".." }
+tokio-io-pool = "0.1.4"

tokio-reactor/benches/basic.rs

@@ -6,56 +6,134 @@ extern crate mio;
 extern crate num_cpus;
 extern crate test;
 extern crate tokio;
+extern crate tokio_io_pool;
 extern crate tokio_reactor;
 
-use std::sync::mpsc;
+const NUM_YIELD: usize = 500;
+const TASKS_PER_CPU: usize = 100;
 
-use futures::{future, Async};
-use self::test::Bencher;
-use tokio_reactor::Registration;
+mod threadpool {
+    use super::*;
+    use std::sync::mpsc;
 
-const NUM_YIELD: usize = 1_000;
-const TASKS_PER_CPU: usize = 50;
+    use test::Bencher;
+    use futures::{future, Async};
+    use tokio_reactor::Registration;
+    use tokio::runtime::Runtime;
 
-#[bench]
-fn notify_many(b: &mut Bencher) {
-    let mut rt = tokio::runtime::Runtime::new().unwrap();
+    #[bench]
+    fn notify_many(b: &mut Bencher) {
+        let mut rt = Runtime::new().unwrap();
+        let tasks = TASKS_PER_CPU * num_cpus::get();
 
-    let tasks = TASKS_PER_CPU * num_cpus::get();
-    let (tx, rx) = mpsc::channel();
+        b.iter(|| {
+            let (tx, rx) = mpsc::channel();
 
-    b.iter(|| {
-        for _ in 0..tasks {
-            let (r, s) = mio::Registration::new2();
-            let registration = Registration::new();
-            registration.register(&r).unwrap();
+            rt.block_on::<_, (), ()>(future::lazy(move || {
+                for _ in 0..tasks {
+                    let tx = tx.clone();
 
-            let mut rem = NUM_YIELD;
-            let mut r = Some(r);
-            let tx = tx.clone();
+                    tokio::spawn(future::lazy(move || {
+                        let (r, s) = mio::Registration::new2();
+                        let registration = Registration::new();
+                        registration.register(&r).unwrap();
 
-            rt.spawn(future::poll_fn(move || {
-                loop {
-                    let is_ready = registration.poll_read_ready().unwrap().is_ready();
+                        let mut rem = NUM_YIELD;
+                        let mut r = Some(r);
+                        let tx = tx.clone();
 
-                    if is_ready {
-                        rem -= 1;
+                        tokio::spawn(future::poll_fn(move || {
+                            loop {
+                                let is_ready = registration.poll_read_ready().unwrap().is_ready();
 
-                        if rem == 0 {
-                            r.take().unwrap();
-                            tx.send(()).unwrap();
-                            return Ok(Async::Ready(()));
-                        }
-                    } else {
-                        s.set_readiness(mio::Ready::readable()).unwrap();
-                        return Ok(Async::NotReady);
-                    }
+                                if is_ready {
+                                    rem -= 1;
+
+                                    if rem == 0 {
+                                        r.take().unwrap();
+                                        tx.send(()).unwrap();
+                                        return Ok(Async::Ready(()));
+                                    }
+                                } else {
+                                    s.set_readiness(mio::Ready::readable()).unwrap();
+                                    return Ok(Async::NotReady);
+                                }
+                            }
+                        }));
+
+                        Ok(())
+                    }));
                 }
-            }));
-        }
 
-        for _ in 0..tasks {
-            rx.recv().unwrap();
-        }
-    });
+                Ok(())
+            })).unwrap();
+
+            for _ in 0..tasks {
+                rx.recv().unwrap();
+            }
+        })
+    }
+}
+
+mod io_pool {
+    use super::*;
+    use std::sync::mpsc;
+
+    use futures::{future, Async};
+    use test::Bencher;
+    use tokio_io_pool::Runtime;
+    use tokio_reactor::Registration;
+
+    #[bench]
+    fn notify_many(b: &mut Bencher) {
+        let mut rt = Runtime::new();
+        let tasks = TASKS_PER_CPU * num_cpus::get();
+
+        b.iter(|| {
+            let (tx, rx) = mpsc::channel();
+
+            rt.block_on::<_, (), ()>(future::lazy(move || {
+                for _ in 0..tasks {
+                    let tx = tx.clone();
+
+                    tokio::spawn(future::lazy(move || {
+                        let (r, s) = mio::Registration::new2();
+                        let registration = Registration::new();
+                        registration.register(&r).unwrap();
+
+                        let mut rem = NUM_YIELD;
+                        let mut r = Some(r);
+                        let tx = tx.clone();
+
+                        tokio::spawn(future::poll_fn(move || {
+                            loop {
+                                let is_ready = registration.poll_read_ready().unwrap().is_ready();
+
+                                if is_ready {
+                                    rem -= 1;
+
+                                    if rem == 0 {
+                                        r.take().unwrap();
+                                        tx.send(()).unwrap();
+                                        return Ok(Async::Ready(()));
+                                    }
+                                } else {
+                                    s.set_readiness(mio::Ready::readable()).unwrap();
+                                    return Ok(Async::NotReady);
+                                }
+                            }
+                        }));
+
+                        Ok(())
+                    }));
+                }
+
+                Ok(())
+            })).unwrap();
+
+            for _ in 0..tasks {
+                rx.recv().unwrap();
+            }
+        })
+    }
 }

tokio-signal/CHANGELOG.md

@@ -1,41 +1,32 @@
-# Changelog
-All notable changes to this project will be documented in this file.
+## Unreleased
 
-The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/)
-and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html).
+## 0.2.6 - (October 26, 2018)
+### Changed
+* Use the `signal-hook` crate for managing signal registrations
 
-## [Unreleased]
-
-## [0.2.5] - 2018-08-29
+## 0.2.5 - (September 29, 2018)
 ### Fixes
 * Fix a possible starvation when polling multiple `Signal` instances outside of
 a tokio reactor (e.g. by using `Future::wait`)
 
-## [0.2.4] - 2018-08-25
+## 0.2.4 - (August 25, 2018)
 ### Fixes
 * Actually make `unix::bsd` public
 
-## [0.2.3] - 2018-08-25
+## 0.2.3 - (August 25, 2018)
 ### Features
-* Exposes `SIGINFO` on BSD-based operating systems. (#46)
+* Exposes `SIGINFO` on BSD-based operating systems.
 
-## [0.2.2] - 2018-08-14
+## 0.2.2 - (August 14, 2018)
 ### Fixes
 * Fix starvation of `Signal`s whenever a `Signal` instance is dropped
 * Fix starvation of individual `Signal`s based on their creation order
 
-## [0.2.1] - 2018-05-27
+## 0.2.1 - (May 27, 2018)
 ### Fixes
 * Bump minimum supported version of `mio` to 0.6.14
 
-## 0.2.0 - 2018-05-07
+## 0.2.0 - (May 7, 2018)
 #### Features
- * Uses `tokio` instead of `tokio_core` (#24)
- * Supports all 33 signals on FreeBSD (#27)
-
-[Unreleased]: https://github.com/alexcrichton/tokio-process/compare/0.2.5...HEAD
-[0.2.5]: https://github.com/alexcrichton/tokio-signal/compare/0.2.4...0.2.5
-[0.2.4]: https://github.com/alexcrichton/tokio-signal/compare/0.2.3...0.2.4
-[0.2.3]: https://github.com/alexcrichton/tokio-signal/compare/0.2.2...0.2.3
-[0.2.2]: https://github.com/alexcrichton/tokio-signal/compare/0.2.1...0.2.2
-[0.2.1]: https://github.com/alexcrichton/tokio-signal/compare/0.2.0...0.2.1
+ * Uses `tokio` instead of `tokio_core`
+ * Supports all 33 signals on FreeBSD

tokio-signal/Cargo.toml

@@ -1,11 +1,17 @@
 [package]
 name = "tokio-signal"
-version = "0.2.5"
+
+# When releasing to crates.io:
+# - Update html_root_url.
+# - Update CHANGELOG.md.
+# - Update doc URL.
+# - Create "v0.2.x" git tag.
+version = "0.2.6"
 authors = ["Alex Crichton <[email protected]>"]
 license = "MIT"
 repository = "https://github.com/tokio-rs/tokio"
 homepage = "https://github.com/tokio-rs/tokio"
-documentation = "https://docs.rs/tokio-signal/0.2"
+documentation = "https://docs.rs/tokio-signal/0.2.6/tokio_signal"
 description = """
 An implementation of an asynchronous Unix signal handling backed futures.
 """

tokio-threadpool/CHANGELOG.md

@@ -1,3 +1,9 @@
+# 0.1.8 (October 23, 2018)
+
+* Assign spawned tasks to random worker (#660).
+* Worker threads no longer shutdown (#692).
+* Reduce atomic ops in notifier (#702).
+
 # 0.1.7 (September 27, 2018)
 
 * Add ThreadPool::spawn_handle (#602, #604).

tokio-threadpool/Cargo.toml

@@ -5,8 +5,8 @@ name = "tokio-threadpool"
 # - Update CHANGELOG.md.
 # - Update doc URL.
 # - Create "v0.1.x" git tag.
-version = "0.1.7"
-documentation = "https://docs.rs/tokio-threadpool/0.1.7/tokio_threadpool"
+version = "0.1.8"
+documentation = "https://docs.rs/tokio-threadpool/0.1.8/tokio_threadpool"
 repository = "https://github.com/tokio-rs/tokio"
 homepage = "https://github.com/tokio-rs/tokio"
 license = "MIT"

tokio-threadpool/src/builder.rs

@@ -170,14 +170,14 @@ impl Builder {
         self
     }
 
-    /// Set the worker thread keep alive duration
+    /// Set the thread keep alive duration
     ///
-    /// If set, a worker thread will wait for up to the specified duration for
-    /// work, at which point the thread will shutdown. When work becomes
-    /// available, a new thread will eventually be spawned to replace the one
-    /// that shut down.
+    /// If set, a thread that has completed a `blocking` call will wait for up
+    /// to the specified duration to become a worker thread again. Once the
+    /// duration elapses, the thread will shutdown.
     ///
-    /// When the value is `None`, the thread will wait for work forever.
+    /// When the value is `None`, the thread will wait to become a worker
+    /// thread forever.
     ///
     /// The default value is `None`.
     ///

tokio-threadpool/src/lib.rs

@@ -1,4 +1,4 @@
-#![doc(html_root_url = "https://docs.rs/tokio-threadpool/0.1.7")]
+#![doc(html_root_url = "https://docs.rs/tokio-threadpool/0.1.8")]
 #![deny(warnings, missing_docs, missing_debug_implementations)]
 
 //! A work-stealing based thread pool for executing futures.

tokio-threadpool/src/notifier.rs

@@ -3,7 +3,7 @@ use task::Task;
 
 use std::mem;
 use std::ops;
-use std::sync::{Arc, Weak};
+use std::sync::Arc;
 
 use futures::executor::Notify;
 
@@ -13,7 +13,7 @@ use futures::executor::Notify;
 /// to poll the future again.
 #[derive(Debug)]
 pub(crate) struct Notifier {
-    pub inner: Weak<Pool>,
+    pub inner: Arc<Pool>,
 }
 
 /// A guard that ensures that the inner value gets forgotten.
@@ -38,9 +38,7 @@ impl Notify for Notifier {
                 // Bump the ref count
                 let task = task.clone();
 
-                if let Some(inner) = self.inner.upgrade() {
-                    let _ = inner.submit(task, &inner);
-                }
+                let _ = self.inner.submit(task, &self.inner);
             }
         }
     }

tokio-threadpool/src/pool/mod.rs

@@ -317,6 +317,16 @@ impl Pool {
 
         // All workers are active, so pick a random worker and submit the
         // task to it.
+        self.submit_to_random(task, inner);
+    }
+
+    /// Submit a task to a random worker
+    ///
+    /// Called from outside of the scheduler, this function is how new tasks
+    /// enter the system.
+    pub fn submit_to_random(&self, task: Arc<Task>, inner: &Arc<Pool>) {
+        debug_assert_eq!(*self, **inner);
+
         let len = self.workers.len();
         let idx = self.rand_usize() % len;
 

tokio-threadpool/src/sender.rs

@@ -161,7 +161,7 @@ impl<'a> tokio_executor::Executor for &'a Sender {
         // Create a new task for the future
         let task = Arc::new(Task::new(future));
 
-        self.inner.submit(task, &self.inner);
+        self.inner.submit_to_random(task, &self.inner);
 
         Ok(())
     }

tokio-threadpool/src/worker/mod.rs

@@ -26,7 +26,7 @@ use std::rc::Rc;
 use std::sync::atomic::Ordering::{AcqRel, Acquire};
 use std::sync::Arc;
 use std::thread;
-use std::time::{Duration, Instant};
+use std::time::Duration;
 
 /// Thread worker
 ///
@@ -222,7 +222,7 @@ impl Worker {
 
         // Get the notifier.
         let notify = Arc::new(Notifier {
-            inner: Arc::downgrade(&self.inner),
+            inner: self.inner.clone(),
         });
 
         let mut first = true;
@@ -667,39 +667,14 @@ impl Worker {
 
         trace!("    -> starting to sleep; idx={}", self.id.0);
 
-        let sleep_until = self.inner.config.keep_alive
-            .map(|dur| Instant::now() + dur);
-
         // The state has been transitioned to sleeping, we can now wait by
         // calling the parker. This is done in a loop as condvars can wakeup
         // spuriously.
-        'sleep:
         loop {
-            let mut drop_thread = false;
-
-            match sleep_until {
-                Some(when) => {
-                    let now = Instant::now();
-
-                    if when >= now {
-                        drop_thread = true;
-                    }
-
-                    let dur = when - now;
-
-                    unsafe {
-                        (*self.entry().park.get())
-                            .park_timeout(dur)
-                            .unwrap();
-                    }
-                }
-                None => {
-                    unsafe {
-                        (*self.entry().park.get())
-                            .park()
-                            .unwrap();
-                    }
-                }
+            unsafe {
+                (*self.entry().park.get())
+                    .park()
+                    .unwrap();
             }
 
             trace!("    -> wakeup; idx={}", self.id.0);
@@ -710,10 +685,7 @@ impl Worker {
             // If the worker has been notified, transition back to running.
             match state.lifecycle() {
                 Sleeping => {
-                    if !drop_thread {
-                        // This goes back to the outer loop.
-                        continue 'sleep;
-                    }
+                    // Still sleeping. Park again.
                 }
                 Notified | Signaled => {
                     // Transition back to running
@@ -738,80 +710,6 @@ impl Worker {
                     unreachable!();
                 }
             }
-
-            // The thread has reached the maximum permitted sleep duration.
-            // It is now going to begin to shutdown.
-            //
-            // Doing this requires first releasing the thread to the backup
-            // stack. Because the moment the worker state is transitioned to
-            // `Shutdown`, other threads **expect** the thread's backup
-            // entry to be available on the backup stack.
-            //
-            // However, it is possible that the worker is notified between
-            // us pushing the backup entry onto the backup stack and
-            // transitioning the worker to `Shutdown`. If this happens, the
-            // current thread lost the token to run the backup entry and has
-            // to shutdown no matter what.
-            //
-            // To deal with this, the worker is transitioned to another
-            // thread. This is a pretty rare condition.
-            //
-            // If pushing on the backup stack fails, then the pool is being
-            // terminated and the thread should just shutdown
-            let backup_push_err = self.inner.release_backup(self.backup_id).is_err();
-
-            if backup_push_err {
-                debug_assert!({
-                    let state: State = self.entry().state.load(Acquire).into();
-                    state.lifecycle() != Sleeping
-                });
-
-                self.should_finalize.set(true);
-
-                return true;
-            }
-
-            loop {
-                let mut next = state;
-                next.set_lifecycle(Shutdown);
-
-                let actual: State = self.entry().state.compare_and_swap(
-                    state.into(), next.into(), AcqRel).into();
-
-                if actual == state {
-                    // Transitioned to a shutdown state
-                    return false;
-                }
-
-                match actual.lifecycle() {
-                    Sleeping => {
-                        state = actual;
-                    }
-                    Notified | Signaled => {
-                        // Transition back to running
-                        loop {
-                            let mut next = state;
-                            next.set_lifecycle(Running);
-
-                            let actual = self.entry().state.compare_and_swap(
-                                state.into(), next.into(), AcqRel).into();
-
-                            if actual == state {
-                                self.inner.spawn_thread(self.id.clone(), &self.inner);
-                                return false;
-                            }
-
-                            state = actual;
-                        }
-                    }
-                    Shutdown | Running => {
-                        // To get here, the block above transitioned the state to
-                        // `Sleeping`. No other thread can concurrently
-                        // transition to `Shutdown` or `Running`.
-                        unreachable!();
-                    }
-                }
-            }
         }
     }
 
@@ -892,4 +790,12 @@ impl WorkerId {
     pub(crate) fn new(idx: usize) -> WorkerId {
         WorkerId(idx)
     }
+
+    /// Returns this identifier represented as an integer.
+    ///
+    /// Worker identifiers in a single thread pool are guaranteed to correspond to integers in the
+    /// range `0..pool_size`.
+    pub fn to_usize(&self) -> usize {
+        self.0
+    }
 }

tokio-threadpool/tests/threadpool.rs

@@ -204,50 +204,6 @@ fn drop_threadpool_drops_futures() {
     }
 }
 
-#[test]
-fn thread_shutdown_timeout() {
-    use std::sync::Mutex;
-
-    let _ = ::env_logger::try_init();
-
-    let (shutdown_tx, shutdown_rx) = mpsc::channel();
-    let (complete_tx, complete_rx) = mpsc::channel();
-
-    let t = Mutex::new(shutdown_tx);
-
-    let pool = Builder::new()
-        .keep_alive(Some(Duration::from_millis(200)))
-        .around_worker(move |w, _| {
-            w.run();
-            // There could be multiple threads here
-            let _ = t.lock().unwrap().send(());
-        })
-        .build();
-    let tx = pool.sender().clone();
-
-    let t = complete_tx.clone();
-    tx.spawn(lazy(move || {
-        t.send(()).unwrap();
-        Ok(())
-    })).unwrap();
-
-    // The future completes
-    complete_rx.recv().unwrap();
-
-    // The thread shuts down eventually
-    shutdown_rx.recv().unwrap();
-
-    // Futures can still be run
-    tx.spawn(lazy(move || {
-        complete_tx.send(()).unwrap();
-        Ok(())
-    })).unwrap();
-
-    complete_rx.recv().unwrap();
-
-    pool.shutdown().wait().unwrap();
-}
-
 #[test]
 fn many_oneshot_futures() {
     const NUM: usize = 10_000;

tokio-uds/CHANGELOG.md

@@ -1,3 +1,11 @@
+# 0.2.3 (October 23, 2018)
+
+* Fix build on NetBSD (#715).
+
+# 0.2.2 (September 27, 2018)
+
+* Fix bug in `UdsStream::read_buf` (#672).
+
 # 0.2.1 (August 19, 2018)
 
 * Re-export `ConnectFuture` (#430).

tokio-uds/Cargo.toml

@@ -2,14 +2,15 @@
 name = "tokio-uds"
 # When releasing to crates.io:
 # - Update html_root_url.
+# - Update doc url
 # - Update CHANGELOG.md.
 # - Create "v0.2.x" git tag.
-version = "0.2.1"
+version = "0.2.3"
 authors = ["Carl Lerche <[email protected]>"]
 license = "MIT"
 repository = "https://github.com/tokio-rs/tokio"
 homepage = "https://github.com/tokio-rs/tokio"
-documentation = "https://docs.rs/tokio-uds/0.2.1/tokio_uds/"
+documentation = "https://docs.rs/tokio-uds/0.2.3/tokio_uds/"
 description = """
 Unix Domain sockets for Tokio
 """

tokio-uds/src/lib.rs

@@ -1,11 +1,11 @@
+#![cfg(unix)]
+#![doc(html_root_url = "https://docs.rs/tokio-uds/0.2.3")]
+#![deny(missing_docs, warnings, missing_debug_implementations)]
+
 //! Unix Domain Sockets for Tokio.
 //!
 //! This crate provides APIs for using Unix Domain Sockets with Tokio.
 
-#![cfg(unix)]
-#![doc(html_root_url = "https://docs.rs/tokio-uds/0.2.1")]
-#![deny(missing_docs, warnings, missing_debug_implementations)]
-
 extern crate bytes;
 #[macro_use]
 extern crate futures;

tokio-uds/src/stream.rs

@@ -196,7 +196,7 @@ impl<'a> AsyncRead for &'a UnixStream {
             if r == -1 {
                 let e = io::Error::last_os_error();
                 if e.kind() == io::ErrorKind::WouldBlock {
-                    self.io.clear_write_ready()?;
+                    self.io.clear_read_ready(Ready::readable())?;
                     Ok(Async::NotReady)
                 } else {
                     Err(e)

tokio-uds/src/ucred.rs

@@ -12,7 +12,7 @@ pub struct UCred {
 #[cfg(any(target_os = "linux", target_os = "android"))]
 pub use self::impl_linux::get_peer_cred;
 
-#[cfg(any(target_os = "dragonfly", target_os = "macos", target_os = "ios", target_os = "freebsd", target_os = "openbsd"))]
+#[cfg(any(target_os = "dragonfly", target_os = "macos", target_os = "ios", target_os = "freebsd", target_os = "netbsd", target_os = "openbsd"))]
 pub use self::impl_macos::get_peer_cred;
 
 #[cfg(any(target_os = "linux", target_os = "android"))]
@@ -61,7 +61,7 @@ pub mod impl_linux {
     }
 }
 
-#[cfg(any(target_os = "dragonfly", target_os = "macos", target_os = "ios", target_os = "freebsd", target_os = "openbsd"))]
+#[cfg(any(target_os = "dragonfly", target_os = "macos", target_os = "ios", target_os = "freebsd", target_os = "netbsd", target_os = "openbsd"))]
 pub mod impl_macos {
     use libc::getpeereid;
     use std::{io, mem};
@@ -95,6 +95,7 @@ mod test {
 
     #[test]
     #[cfg_attr(target_os = "freebsd", ignore = "Requires FreeBSD 12.0 or later. https://bugs.freebsd.org/bugzilla/show_bug.cgi?id=176419")]
+    #[cfg_attr(target_os = "netbsd", ignore = "NetBSD does not support getpeereid() for sockets created by socketpair()")]
     fn test_socket_pair() {
         let (a, b) = UnixStream::pair().unwrap();
         let cred_a = a.peer_cred().unwrap();