reactor: Remove extra semi colon (#1616)

* reactor: Remove extra semi colon

* fmt

Cargo.toml

@@ -9,7 +9,6 @@ members = [
   "tokio-fs",
   "tokio-futures",
   "tokio-io",
-  "tokio-macros",
   "tokio-reactor",
   "tokio-signal",
   "tokio-sync",

azure-pipelines.yml

@@ -91,7 +91,7 @@ jobs:
 - template: ci/azure-check-minrust.yml
   parameters:
     name: minrust
-    rust_version: 1.26.0
+    rust_version: 1.31.0
 
 - template: ci/azure-tsan.yml
   parameters:

ci/azure-tsan.yml

@@ -12,7 +12,7 @@ jobs:
   steps:
   - template: azure-install-rust.yml
     parameters:
-      rust_version: nightly-2018-11-18
+      rust_version: nightly-2019-07-17
 
   - template: azure-patch-crates.yml
   - script: |

ci/tsan

@@ -35,3 +35,10 @@ race:WorkerEntry::set_next_sleeper
 # This ignores a false positive caused by `thread::park()`/`thread::unpark()`.
 # See: https://github.com/rust-lang/rust/pull/54806#issuecomment-436193353
 race:pthread_cond_destroy
+
+# Recent rand dependency updates and seeding changes have introduced
+# lazy_static's and other racy code. See:
+# https://github.com/tokio-rs/tokio/pull/1358#issuecomment-516172383
+race:RandomState*::build_hasher
+race:lazy_static::
+race:c2_chacha::guts

tokio-buf/tests/buf_stream.rs

@@ -4,4 +4,4 @@ use tokio_buf::BufStream;
 
 // Ensures that `BufStream` can be a trait object
 #[allow(dead_code)]
-fn obj(_: &mut BufStream<Item = u32, Error = ()>) {}
+fn obj(_: &mut dyn BufStream<Item = u32, Error = ()>) {}

tokio-current-thread/src/lib.rs

@@ -64,7 +64,7 @@ pub struct CurrentThread<P: Park = ParkThread> {
     spawn_handle: Handle,
 
     /// Receiver for futures spawned from other threads
-    spawn_receiver: mpsc::Receiver<Box<Future<Item = (), Error = ()> + Send + 'static>>,
+    spawn_receiver: mpsc::Receiver<Box<dyn Future<Item = (), Error = ()> + Send + 'static>>,
 
     /// The thread-local ID assigned to this executor.
     id: u64,
@@ -186,11 +186,15 @@ struct Borrow<'a, U: 'a> {
 }
 
 trait SpawnLocal {
-    fn spawn_local(&mut self, future: Box<Future<Item = (), Error = ()>>, already_counted: bool);
+    fn spawn_local(
+        &mut self,
+        future: Box<dyn Future<Item = (), Error = ()>>,
+        already_counted: bool,
+    );
 }
 
 struct CurrentRunner {
-    spawn: Cell<Option<*mut SpawnLocal>>,
+    spawn: Cell<Option<*mut dyn SpawnLocal>>,
     id: Cell<Option<u64>>,
 }
 
@@ -424,7 +428,7 @@ impl<P: Park> Drop for CurrentThread<P> {
 impl tokio_executor::Executor for CurrentThread {
     fn spawn(
         &mut self,
-        future: Box<Future<Item = (), Error = ()> + Send>,
+        future: Box<dyn Future<Item = (), Error = ()> + Send>,
     ) -> Result<(), SpawnError> {
         self.borrow().spawn_local(future, false);
         Ok(())
@@ -629,7 +633,7 @@ impl<'a, P: Park> fmt::Debug for Entered<'a, P> {
 /// Handle to spawn a future on the corresponding `CurrentThread` instance
 #[derive(Clone)]
 pub struct Handle {
-    sender: mpsc::Sender<Box<Future<Item = (), Error = ()> + Send + 'static>>,
+    sender: mpsc::Sender<Box<dyn Future<Item = (), Error = ()> + Send + 'static>>,
     num_futures: Arc<atomic::AtomicUsize>,
     shut_down: Cell<bool>,
     notify: executor::NotifyHandle,
@@ -731,7 +735,7 @@ impl TaskExecutor {
     /// Spawn a future onto the current `CurrentThread` instance.
     pub fn spawn_local(
         &mut self,
-        future: Box<Future<Item = (), Error = ()>>,
+        future: Box<dyn Future<Item = (), Error = ()>>,
     ) -> Result<(), SpawnError> {
         CURRENT.with(|current| match current.spawn.get() {
             Some(spawn) => {
@@ -746,7 +750,7 @@ impl TaskExecutor {
 impl tokio_executor::Executor for TaskExecutor {
     fn spawn(
         &mut self,
-        future: Box<Future<Item = (), Error = ()> + Send>,
+        future: Box<dyn Future<Item = (), Error = ()> + Send>,
     ) -> Result<(), SpawnError> {
         self.spawn_local(future)
     }
@@ -791,7 +795,11 @@ impl<'a, U: Unpark> Borrow<'a, U> {
 }
 
 impl<'a, U: Unpark> SpawnLocal for Borrow<'a, U> {
-    fn spawn_local(&mut self, future: Box<Future<Item = (), Error = ()>>, already_counted: bool) {
+    fn spawn_local(
+        &mut self,
+        future: Box<dyn Future<Item = (), Error = ()>>,
+        already_counted: bool,
+    ) {
         if !already_counted {
             // NOTE: we have a borrow of the Runtime, so we know that it isn't shut down.
             // NOTE: += 2 since LSB is the shutdown bit
@@ -804,7 +812,7 @@ impl<'a, U: Unpark> SpawnLocal for Borrow<'a, U> {
 // ===== impl CurrentRunner =====
 
 impl CurrentRunner {
-    fn set_spawn<F, R>(&self, spawn: &mut SpawnLocal, f: F) -> R
+    fn set_spawn<F, R>(&self, spawn: &mut dyn SpawnLocal, f: F) -> R
     where
         F: FnOnce() -> R,
     {
@@ -819,14 +827,14 @@ impl CurrentRunner {
 
         let _reset = Reset(self);
 
-        let spawn = unsafe { hide_lt(spawn as *mut SpawnLocal) };
+        let spawn = unsafe { hide_lt(spawn as *mut dyn SpawnLocal) };
         self.spawn.set(Some(spawn));
 
         f()
     }
 }
 
-unsafe fn hide_lt<'a>(p: *mut (SpawnLocal + 'a)) -> *mut (SpawnLocal + 'static) {
+unsafe fn hide_lt<'a>(p: *mut (dyn SpawnLocal + 'a)) -> *mut (dyn SpawnLocal + 'static) {
     use std::mem;
     mem::transmute(p)
 }

tokio-current-thread/src/scheduler.rs

@@ -125,7 +125,7 @@ enum Dequeue<U> {
 }
 
 /// Wraps a spawned boxed future
-struct Task(Spawn<Box<Future<Item = (), Error = ()>>>);
+struct Task(Spawn<Box<dyn Future<Item = (), Error = ()>>>);
 
 /// A task that is scheduled. `turn` must be called
 pub struct Scheduled<'a, U: 'a> {
@@ -171,7 +171,7 @@ where
         self.inner.clone().into()
     }
 
-    pub fn schedule(&mut self, item: Box<Future<Item = (), Error = ()>>) {
+    pub fn schedule(&mut self, item: Box<dyn Future<Item = (), Error = ()>>) {
         // Get the current scheduler tick
         let tick_num = self.inner.tick_num.load(SeqCst);
 
@@ -359,7 +359,7 @@ impl<'a, U: Unpark> Scheduled<'a, U> {
 }
 
 impl Task {
-    pub fn new(future: Box<Future<Item = (), Error = ()> + 'static>) -> Self {
+    pub fn new(future: Box<dyn Future<Item = (), Error = ()> + 'static>) -> Self {
         Task(executor::spawn(future))
     }
 }
@@ -687,8 +687,8 @@ unsafe impl<U: Unpark> UnsafeNotify for ArcNode<U> {
     }
 }
 
-unsafe fn hide_lt<U: Unpark>(p: *mut ArcNode<U>) -> *mut UnsafeNotify {
-    mem::transmute(p as *mut UnsafeNotify)
+unsafe fn hide_lt<U: Unpark>(p: *mut ArcNode<U>) -> *mut dyn UnsafeNotify {
+    mem::transmute(p as *mut dyn UnsafeNotify)
 }
 
 impl<U: Unpark> Node<U> {

tokio-current-thread/tests/current_thread.rs

@@ -22,7 +22,7 @@ use futures::sync::oneshot;
 
 mod from_block_on_all {
     use super::*;
-    fn test<F: Fn(Box<Future<Item = (), Error = ()>>) + 'static>(spawn: F) {
+    fn test<F: Fn(Box<dyn Future<Item = (), Error = ()>>) + 'static>(spawn: F) {
         let cnt = Rc::new(Cell::new(0));
         let c = cnt.clone();
 
@@ -102,7 +102,7 @@ fn spawn_many() {
 mod does_not_set_global_executor_by_default {
     use super::*;
 
-    fn test<F: Fn(Box<Future<Item = (), Error = ()> + Send>) -> Result<(), E> + 'static, E>(
+    fn test<F: Fn(Box<dyn Future<Item = (), Error = ()> + Send>) -> Result<(), E> + 'static, E>(
         spawn: F,
     ) {
         block_on_all(lazy(|| {
@@ -127,7 +127,7 @@ mod does_not_set_global_executor_by_default {
 mod from_block_on_future {
     use super::*;
 
-    fn test<F: Fn(Box<Future<Item = (), Error = ()>>)>(spawn: F) {
+    fn test<F: Fn(Box<dyn Future<Item = (), Error = ()>>)>(spawn: F) {
         let cnt = Rc::new(Cell::new(0));
 
         let mut tokio_current_thread = CurrentThread::new();
@@ -181,8 +181,8 @@ mod outstanding_tasks_are_dropped_when_executor_is_dropped {
 
     fn test<F, G>(spawn: F, dotspawn: G)
     where
-        F: Fn(Box<Future<Item = (), Error = ()>>) + 'static,
-        G: Fn(&mut CurrentThread, Box<Future<Item = (), Error = ()>>),
+        F: Fn(Box<dyn Future<Item = (), Error = ()>>) + 'static,
+        G: Fn(&mut CurrentThread, Box<dyn Future<Item = (), Error = ()>>),
     {
         let mut rc = Rc::new(());
 
@@ -383,8 +383,8 @@ mod and_turn {
 
     fn test<F, G>(spawn: F, dotspawn: G)
     where
-        F: Fn(Box<Future<Item = (), Error = ()>>) + 'static,
-        G: Fn(&mut CurrentThread, Box<Future<Item = (), Error = ()>>),
+        F: Fn(Box<dyn Future<Item = (), Error = ()>>) + 'static,
+        G: Fn(&mut CurrentThread, Box<dyn Future<Item = (), Error = ()>>),
     {
         let cnt = Rc::new(Cell::new(0));
         let c = cnt.clone();
@@ -445,7 +445,6 @@ mod and_turn {
             },
         );
     }
-
 }
 
 mod in_drop {
@@ -459,7 +458,7 @@ mod in_drop {
     }
 
     struct MyFuture {
-        _data: Box<Any>,
+        _data: Box<dyn Any>,
     }
 
     impl Future for MyFuture {
@@ -473,8 +472,8 @@ mod in_drop {
 
     fn test<F, G>(spawn: F, dotspawn: G)
     where
-        F: Fn(Box<Future<Item = (), Error = ()>>) + 'static,
-        G: Fn(&mut CurrentThread, Box<Future<Item = (), Error = ()>>),
+        F: Fn(Box<dyn Future<Item = (), Error = ()>>) + 'static,
+        G: Fn(&mut CurrentThread, Box<dyn Future<Item = (), Error = ()>>),
     {
         let mut tokio_current_thread = CurrentThread::new();
 
@@ -520,7 +519,6 @@ mod in_drop {
             },
         );
     }
-
 }
 
 #[test]

tokio-executor/src/enter.rs

@@ -11,7 +11,7 @@ thread_local!(static ENTERED: Cell<bool> = Cell::new(false));
 ///
 /// For more details, see [`enter` documentation](fn.enter.html)
 pub struct Enter {
-    on_exit: Vec<Box<Callback>>,
+    on_exit: Vec<Box<dyn Callback>>,
     permanent: bool,
 }
 

tokio-executor/src/executor.rs

@@ -94,7 +94,7 @@ pub trait Executor {
     /// ```
     fn spawn(
         &mut self,
-        future: Box<Future<Item = (), Error = ()> + Send>,
+        future: Box<dyn Future<Item = (), Error = ()> + Send>,
     ) -> Result<(), SpawnError>;
 
     /// Provides a best effort **hint** to whether or not `spawn` will succeed.
@@ -140,7 +140,7 @@ pub trait Executor {
 impl<E: Executor + ?Sized> Executor for Box<E> {
     fn spawn(
         &mut self,
-        future: Box<Future<Item = (), Error = ()> + Send>,
+        future: Box<dyn Future<Item = (), Error = ()> + Send>,
     ) -> Result<(), SpawnError> {
         (**self).spawn(future)
     }

tokio-executor/src/global.rs

@@ -37,7 +37,7 @@ impl DefaultExecutor {
     }
 
     #[inline]
-    fn with_current<F: FnOnce(&mut Executor) -> R, R>(f: F) -> Option<R> {
+    fn with_current<F: FnOnce(&mut dyn Executor) -> R, R>(f: F) -> Option<R> {
         EXECUTOR.with(
             |current_executor| match current_executor.replace(State::Active) {
                 State::Ready(executor_ptr) => {
@@ -57,7 +57,7 @@ enum State {
     // default executor not defined
     Empty,
     // default executor is defined and ready to be used
-    Ready(*mut Executor),
+    Ready(*mut dyn Executor),
     // default executor is currently active (used to detect recursive calls)
     Active,
 }
@@ -72,7 +72,7 @@ thread_local! {
 impl super::Executor for DefaultExecutor {
     fn spawn(
         &mut self,
-        future: Box<Future<Item = (), Error = ()> + Send>,
+        future: Box<dyn Future<Item = (), Error = ()> + Send>,
     ) -> Result<(), SpawnError> {
         DefaultExecutor::with_current(|executor| executor.spawn(future))
             .unwrap_or_else(|| Err(SpawnError::shutdown()))
@@ -210,7 +210,7 @@ where
     })
 }
 
-unsafe fn hide_lt<'a>(p: *mut (Executor + 'a)) -> *mut (Executor + 'static) {
+unsafe fn hide_lt<'a>(p: *mut (dyn Executor + 'a)) -> *mut (dyn Executor + 'static) {
     use std::mem;
     mem::transmute(p)
 }

tokio-executor/src/lib.rs

@@ -1,7 +1,5 @@
 #![deny(missing_docs, missing_debug_implementations, warnings)]
 #![doc(html_root_url = "https://docs.rs/tokio-executor/0.1.8")]
-// Our MSRV doesn't allow us to fix these warnings yet
-#![allow(rust_2018_idioms)]
 
 //! Task execution related traits and utilities.
 //!

tokio-executor/src/park.rs

@@ -128,13 +128,13 @@ pub trait Unpark: Sync + Send + 'static {
     fn unpark(&self);
 }
 
-impl Unpark for Box<Unpark> {
+impl Unpark for Box<dyn Unpark> {
     fn unpark(&self) {
         (**self).unpark()
     }
 }
 
-impl Unpark for Arc<Unpark> {
+impl Unpark for Arc<dyn Unpark> {
     fn unpark(&self) {
         (**self).unpark()
     }

tokio-executor/tests/executor.rs

@@ -10,7 +10,7 @@ mod out_of_executor_context {
 
     fn test<F, E>(spawn: F)
     where
-        F: Fn(Box<Future<Item = (), Error = ()> + Send>) -> Result<(), E>,
+        F: Fn(Box<dyn Future<Item = (), Error = ()> + Send>) -> Result<(), E>,
     {
         let res = spawn(Box::new(lazy(|| Ok(()))));
         assert!(res.is_err());

tokio-fs/Cargo.toml

@@ -27,9 +27,8 @@ tokio-threadpool = "0.1.3"
 tokio-io = "0.1.6"
 
 [dev-dependencies]
-rand = "0.6"
-tempfile = "3"
-tempdir = "0.3"
+rand = "0.7"
+tempfile = "~3.1.0"
 tokio-io = "0.1.6"
 tokio-codec = "0.1.0"
 tokio = "0.1.7"

tokio-fs/examples/std-echo.rs

@@ -14,7 +14,7 @@ use futures::{Future, Sink, Stream};
 
 use std::io;
 
-pub fn main() -> Result<(), Box<std::error::Error>> {
+pub fn main() -> Result<(), Box<dyn std::error::Error>> {
     let pool = Builder::new().pool_size(1).build();
 
     pool.spawn({

tokio-fs/tests/dir.rs

@@ -1,18 +1,18 @@
 extern crate futures;
-extern crate tempdir;
+extern crate tempfile;
 extern crate tokio_fs;
 
 use futures::{Future, Stream};
 use std::fs;
 use std::sync::{Arc, Mutex};
-use tempdir::TempDir;
+use tempfile::tempdir;
 use tokio_fs::*;
 
 mod pool;
 
 #[test]
 fn create() {
-    let base_dir = TempDir::new("base").unwrap();
+    let base_dir = tempdir().unwrap();
     let new_dir = base_dir.path().join("foo");
 
     pool::run({ create_dir(new_dir.clone()) });
@@ -22,7 +22,7 @@ fn create() {
 
 #[test]
 fn create_all() {
-    let base_dir = TempDir::new("base").unwrap();
+    let base_dir = tempdir().unwrap();
     let new_dir = base_dir.path().join("foo").join("bar");
 
     pool::run({ create_dir_all(new_dir.clone()) });
@@ -32,7 +32,7 @@ fn create_all() {
 
 #[test]
 fn remove() {
-    let base_dir = TempDir::new("base").unwrap();
+    let base_dir = tempdir().unwrap();
     let new_dir = base_dir.path().join("foo");
 
     fs::create_dir(new_dir.clone()).unwrap();
@@ -44,7 +44,7 @@ fn remove() {
 
 #[test]
 fn read() {
-    let base_dir = TempDir::new("base").unwrap();
+    let base_dir = tempdir().unwrap();
 
     let p = base_dir.path();
     fs::create_dir(p.join("aa")).unwrap();

tokio-fs/tests/link.rs

@@ -1,19 +1,19 @@
 extern crate futures;
-extern crate tempdir;
+extern crate tempfile;
 extern crate tokio_fs;
 
 use futures::Future;
 use std::fs;
 use std::io::prelude::*;
 use std::io::BufReader;
-use tempdir::TempDir;
+use tempfile::tempdir;
 use tokio_fs::*;
 
 mod pool;
 
 #[test]
 fn test_hard_link() {
-    let dir = TempDir::new("base").unwrap();
+    let dir = tempdir().unwrap();
     let src = dir.path().join("src.txt");
     let dst = dir.path().join("dst.txt");
 
@@ -38,7 +38,7 @@ fn test_hard_link() {
 #[cfg(unix)]
 #[test]
 fn test_symlink() {
-    let dir = TempDir::new("base").unwrap();
+    let dir = tempdir().unwrap();
     let src = dir.path().join("src.txt");
     let dst = dir.path().join("dst.txt");
 

tokio-io/src/lib.rs

@@ -21,10 +21,10 @@ use std::io as std_io;
 use futures::{Future, Stream};
 
 /// A convenience typedef around a `Future` whose error component is `io::Error`
-pub type IoFuture<T> = Box<Future<Item = T, Error = std_io::Error> + Send>;
+pub type IoFuture<T> = Box<dyn Future<Item = T, Error = std_io::Error> + Send>;
 
 /// A convenience typedef around a `Stream` whose error component is `io::Error`
-pub type IoStream<T> = Box<Stream<Item = T, Error = std_io::Error> + Send>;
+pub type IoStream<T> = Box<dyn Stream<Item = T, Error = std_io::Error> + Send>;
 
 /// A convenience macro for working with `io::Result<T>` from the `Read` and
 /// `Write` traits.
@@ -65,6 +65,6 @@ pub use self::async_write::AsyncWrite;
 
 fn _assert_objects() {
     fn _assert<T>() {}
-    _assert::<Box<AsyncRead>>();
-    _assert::<Box<AsyncWrite>>();
+    _assert::<Box<dyn AsyncRead>>();
+    _assert::<Box<dyn AsyncWrite>>();
 }

tokio-reactor/CHANGELOG.md

@@ -1,3 +1,9 @@
+# 0.1.10 (September 25, 2019)
+
+### Changed
+- Upgrade to parking_lot 0.9.0 (#1298 backport)
+- The minimum supported rust version (MSRV) is now 1.31.0. (#1358)
+
 # 0.1.9 (March 1, 2019)
 
 ### Added

tokio-reactor/Cargo.toml

@@ -8,13 +8,13 @@ name = "tokio-reactor"
 #   - README.md
 # - Update CHANGELOG.md.
 # - Create "v0.1.x" git tag.
-version = "0.1.9"
+version = "0.1.10"
 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-reactor/0.1.9/tokio_reactor"
+documentation = "https://docs.rs/tokio-reactor/0.1.10/tokio_reactor"
 description = """
 Event loop that drives Tokio I/O resources.
 """
@@ -27,7 +27,7 @@ lazy_static = "1.0.2"
 log = "0.4.1"
 mio = "0.6.14"
 num_cpus = "1.8.0"
-parking_lot = "0.7.0"
+parking_lot = "0.9.0"
 slab = "0.4.0"
 tokio-executor = "0.1.1"
 tokio-io = "0.1.6"
@@ -36,4 +36,4 @@ tokio-sync = "0.1.1"
 [dev-dependencies]
 num_cpus = "1.8.0"
 tokio = "0.1.7"
-tokio-io-pool = "0.1.4"
+tokio-io-pool = "=0.1.4"

tokio-reactor/README.md

@@ -2,7 +2,7 @@
 
 Event loop that drives Tokio I/O resources.
 
-[Documentation](https://docs.rs/tokio-reactor/0.1.9/tokio_reactor)
+[Documentation](https://docs.rs/tokio-reactor/0.1.10/tokio_reactor)
 
 ## Overview
 
@@ -25,10 +25,10 @@ are building a custom I/O resource.
 
 [`mio`]: http://github.com/carllerche/mio
 [`futures`]: http://github.com/rust-lang-nursery/futures-rs
-[`Reactor`]: https://docs.rs/tokio-reactor/0.1.9/tokio_reactor/struct.Reactor.html
-[`Handle`]: https://docs.rs/tokio-reactor/0.1.9/tokio_reactor/struct.Handle.html
-[`Registration`]: https://docs.rs/tokio-reactor/0.1.9/tokio_reactor/struct.Registration.html
-[`PollEvented`]: https://docs.rs/tokio-reactor/0.1.9/tokio_reactor/struct.PollEvented.html
+[`Reactor`]: https://docs.rs/tokio-reactor/0.1.10/tokio_reactor/struct.Reactor.html
+[`Handle`]: https://docs.rs/tokio-reactor/0.1.10/tokio_reactor/struct.Handle.html
+[`Registration`]: https://docs.rs/tokio-reactor/0.1.10/tokio_reactor/struct.Registration.html
+[`PollEvented`]: https://docs.rs/tokio-reactor/0.1.10/tokio_reactor/struct.PollEvented.html
 [`tokio`]: ../
 
 ## License

tokio-reactor/src/lib.rs

@@ -1,4 +1,4 @@
-#![doc(html_root_url = "https://docs.rs/tokio-reactor/0.1.9")]
+#![doc(html_root_url = "https://docs.rs/tokio-reactor/0.1.10")]
 #![deny(missing_docs, warnings, missing_debug_implementations)]
 
 //! Event loop that drives Tokio I/O resources.
@@ -631,7 +631,7 @@ impl HandlePriv {
     }
 
     unsafe fn from_usize(val: usize) -> HandlePriv {
-        let inner = mem::transmute::<usize, Weak<Inner>>(val);;
+        let inner = mem::transmute::<usize, Weak<Inner>>(val);
         HandlePriv { inner }
     }
 
@@ -652,7 +652,7 @@ impl Inner {
     /// Register an I/O resource with the reactor.
     ///
     /// The registration token is returned.
-    fn add_source(&self, source: &Evented) -> io::Result<usize> {
+    fn add_source(&self, source: &dyn Evented) -> io::Result<usize> {
         // Get an ABA guard value
         let aba_guard = self.next_aba_guard.fetch_add(1 << TOKEN_SHIFT, Relaxed);
 
@@ -690,7 +690,7 @@ impl Inner {
     }
 
     /// Deregisters an I/O resource from the reactor.
-    fn deregister_source(&self, source: &Evented) -> io::Result<()> {
+    fn deregister_source(&self, source: &dyn Evented) -> io::Result<()> {
         self.io.deregister(source)
     }
 

tokio-signal/examples/ctrl-c.rs

@@ -7,7 +7,7 @@ use futures::{Future, Stream};
 /// how many signals to handle before exiting
 const STOP_AFTER: u64 = 10;
 
-fn main() -> Result<(), Box<std::error::Error>> {
+fn main() -> Result<(), Box<dyn std::error::Error>> {
     // tokio_signal provides a convenience builder for Ctrl+C
     // this even works cross-platform: linux and windows!
     //

tokio-signal/examples/multiple.rs

@@ -11,7 +11,7 @@ mod platform {
     use futures::{Future, Stream};
     use tokio_signal::unix::{Signal, SIGINT, SIGTERM};
 
-    pub fn main() -> Result<(), Box<::std::error::Error>> {
+    pub fn main() -> Result<(), Box<dyn std::error::Error>> {
         // Create a stream for each of the signals we'd like to handle.
         let sigint = Signal::new(SIGINT).flatten_stream();
         let sigterm = Signal::new(SIGTERM).flatten_stream();
@@ -39,7 +39,6 @@ mod platform {
         }
         Ok(())
     }
-
 }
 
 #[cfg(not(unix))]
@@ -49,6 +48,6 @@ mod platform {
     }
 }
 
-fn main() -> Result<(), Box<std::error::Error>> {
+fn main() -> Result<(), Box<dyn std::error::Error>> {
     platform::main()
 }

tokio-signal/examples/sighup-example.rs

@@ -9,7 +9,7 @@ mod platform {
     use futures::{Future, Stream};
     use tokio_signal::unix::{Signal, SIGHUP};
 
-    pub fn main() -> Result<(), Box<::std::error::Error>> {
+    pub fn main() -> Result<(), Box<dyn std::error::Error>> {
         // on Unix, we can listen to whatever signal we want, in this case: SIGHUP
         let stream = Signal::new(SIGHUP).flatten_stream();
 
@@ -38,7 +38,6 @@ mod platform {
         ::tokio::runtime::current_thread::block_on_all(future)?;
         Ok(())
     }
-
 }
 
 #[cfg(not(unix))]
@@ -48,6 +47,6 @@ mod platform {
     }
 }
 
-fn main() -> Result<(), Box<std::error::Error>> {
+fn main() -> Result<(), Box<dyn std::error::Error>> {
     platform::main()
 }

tokio-signal/src/lib.rs

@@ -86,9 +86,9 @@ pub mod unix;
 pub mod windows;
 
 /// A future whose error is `io::Error`
-pub type IoFuture<T> = Box<Future<Item = T, Error = io::Error> + Send>;
+pub type IoFuture<T> = Box<dyn Future<Item = T, Error = io::Error> + Send>;
 /// A stream whose error is `io::Error`
-pub type IoStream<T> = Box<Stream<Item = T, Error = io::Error> + Send>;
+pub type IoStream<T> = Box<dyn Stream<Item = T, Error = io::Error> + Send>;
 
 /// Creates a stream which receives "ctrl-c" notifications sent to a process.
 ///
@@ -125,7 +125,7 @@ pub fn ctrl_c_handle(handle: &Handle) -> IoFuture<IoStream<()>> {
         let handle = handle.clone();
         Box::new(future::lazy(move || {
             unix::Signal::with_handle(unix::libc::SIGINT, &handle)
-                .map(|x| Box::new(x.map(|_| ())) as Box<Stream<Item = _, Error = _> + Send>)
+                .map(|x| Box::new(x.map(|_| ())) as Box<dyn Stream<Item = _, Error = _> + Send>)
         }))
     }
 

tokio-sync/Cargo.toml

@@ -24,7 +24,7 @@ fnv = "1.0.6"
 futures = "0.1.19"
 
 [dev-dependencies]
-env_logger = { version = "0.5", default-features = false }
+env_logger = { version = "0.6", default-features = false }
 tokio = { version = "0.1.15", path = "../tokio" }
 tokio-mock-task = "0.1.1"
 loom = { version = "0.1.1", features = ["futures"] }

tokio-tcp/Cargo.toml

@@ -28,6 +28,6 @@ iovec = "0.1"
 futures = "0.1.19"
 
 [dev-dependencies]
-env_logger = { version = "0.5", default-features = false }
+env_logger = { version = "0.6", default-features = false }
 net2 = "0.2"
 tokio = "0.1.13"

tokio-test/src/macros.rs

@@ -127,5 +127,4 @@ mod tests {
         let mut fut = future::ok::<(), ()>(());
         assert_ready_eq!(fut.poll(), ());
     }
-
 }

tokio-threadpool/CHANGELOG.md

@@ -1,3 +1,11 @@
+# 0.1.16 (September 25, 2019)
+
+### Changed
+- Remove last non-dev dependency on rand crate by seeding PRNG via libstd
+  `RandomState` (#1324 backport)
+- Upgrade (dev-only dependency) rand to 0.7.0 (#1302 backport)
+- The minimum supported rust version (MSRV) is now 1.31.0 (#1358)
+
 # 0.1.15 (June 2, 2019)
 
 ### Changed

tokio-threadpool/Cargo.toml

@@ -8,8 +8,8 @@ name = "tokio-threadpool"
 #   - README.md
 # - Update CHANGELOG.md.
 # - Create "v0.1.x" git tag.
-version = "0.1.15"
-documentation = "https://docs.rs/tokio-threadpool/0.1.14/tokio_threadpool"
+version = "0.1.16"
+documentation = "https://docs.rs/tokio-threadpool/0.1.16/tokio_threadpool"
 repository = "https://github.com/tokio-rs/tokio"
 homepage = "https://github.com/tokio-rs/tokio"
 license = "MIT"
@@ -27,12 +27,13 @@ crossbeam-deque = "0.7.0"
 crossbeam-queue = "0.1.0"
 crossbeam-utils = "0.6.4"
 num_cpus = "1.2"
-rand = "0.6"
 slab = "0.4.1"
 log = "0.4"
+lazy_static = "1"
 
 [dev-dependencies]
-env_logger = "0.5"
+rand = "0.7"
+env_logger = { version = "0.6", default-features = false }
 
 # For comparison benchmarks
 futures-cpupool = "0.1.7"

tokio-threadpool/README.md

@@ -3,7 +3,7 @@
 A library for scheduling execution of futures concurrently across a pool of
 threads.
 
-[Documentation](https://docs.rs/tokio-threadpool/0.1.15/tokio_threadpool)
+[Documentation](https://docs.rs/tokio-threadpool/0.1.16/tokio_threadpool)
 
 ### Why not Rayon?
 

tokio-threadpool/src/builder.rs

@@ -67,7 +67,7 @@ pub struct Builder {
     max_blocking: usize,
 
     /// Generates the `Park` instances
-    new_park: Box<Fn(&WorkerId) -> BoxPark>,
+    new_park: Box<dyn Fn(&WorkerId) -> BoxPark>,
 }
 
 impl Builder {
@@ -223,7 +223,7 @@ impl Builder {
     /// ```
     pub fn panic_handler<F>(&mut self, f: F) -> &mut Self
     where
-        F: Fn(Box<Any + Send>) + Send + Sync + 'static,
+        F: Fn(Box<dyn Any + Send>) + Send + Sync + 'static,
     {
         self.config.panic_handler = Some(Arc::new(f));
         self

tokio-threadpool/src/callback.rs

@@ -7,7 +7,7 @@ use tokio_executor::Enter;
 
 #[derive(Clone)]
 pub(crate) struct Callback {
-    f: Arc<Fn(&Worker, &mut Enter) + Send + Sync>,
+    f: Arc<dyn Fn(&Worker, &mut Enter) + Send + Sync>,
 }
 
 impl Callback {

tokio-threadpool/src/config.rs

@@ -13,9 +13,9 @@ pub(crate) struct Config {
     pub name_prefix: Option<String>,
     pub stack_size: Option<usize>,
     pub around_worker: Option<Callback>,
-    pub after_start: Option<Arc<Fn() + Send + Sync>>,
-    pub before_stop: Option<Arc<Fn() + Send + Sync>>,
-    pub panic_handler: Option<Arc<Fn(Box<Any + Send>) + Send + Sync>>,
+    pub after_start: Option<Arc<dyn Fn() + Send + Sync>>,
+    pub before_stop: Option<Arc<dyn Fn() + Send + Sync>>,
+    pub panic_handler: Option<Arc<dyn Fn(Box<dyn Any + Send>) + Send + Sync>>,
 }
 
 /// Max number of workers that can be part of a pool. This is the most that can

tokio-threadpool/src/lib.rs

@@ -1,7 +1,5 @@
-#![doc(html_root_url = "https://docs.rs/tokio-threadpool/0.1.15")]
+#![doc(html_root_url = "https://docs.rs/tokio-threadpool/0.1.16")]
 #![deny(warnings, missing_docs, missing_debug_implementations)]
-// Our MSRV doesn't allow us to fix these warnings yet
-#![allow(rust_2018_idioms)]
 
 //! A work-stealing based thread pool for executing futures.
 //!
@@ -86,8 +84,9 @@ extern crate crossbeam_queue;
 extern crate crossbeam_utils;
 #[macro_use]
 extern crate futures;
+#[macro_use]
+extern crate lazy_static;
 extern crate num_cpus;
-extern crate rand;
 extern crate slab;
 
 #[macro_use]

tokio-threadpool/src/park/boxed.rs

@@ -3,8 +3,8 @@ use tokio_executor::park::{Park, Unpark};
 use std::error::Error;
 use std::time::Duration;
 
-pub(crate) type BoxPark = Box<Park<Unpark = BoxUnpark, Error = ()> + Send>;
-pub(crate) type BoxUnpark = Box<Unpark>;
+pub(crate) type BoxPark = Box<dyn Park<Unpark = BoxUnpark, Error = ()> + Send>;
+pub(crate) type BoxUnpark = Box<dyn Unpark>;
 
 pub(crate) struct BoxedPark<T>(T);
 

tokio-threadpool/src/pool/mod.rs

@@ -17,6 +17,8 @@ use worker::{self, Worker, WorkerId};
 use futures::Poll;
 
 use std::cell::Cell;
+use std::collections::hash_map::RandomState;
+use std::hash::{BuildHasher, Hash, Hasher};
 use std::num::Wrapping;
 use std::sync::atomic::AtomicUsize;
 use std::sync::atomic::Ordering::{AcqRel, Acquire};
@@ -25,7 +27,6 @@ use std::thread;
 
 use crossbeam_deque::Injector;
 use crossbeam_utils::CachePadded;
-use rand;
 
 #[derive(Debug)]
 pub(crate) struct Pool {
@@ -420,11 +421,7 @@ impl Pool {
     /// Uses a thread-local random number generator based on XorShift.
     pub fn rand_usize(&self) -> usize {
         thread_local! {
-            static RNG: Cell<Wrapping<u32>> = {
-                // The initial seed must be non-zero.
-                let init = rand::random::<u32>() | 1;
-                Cell::new(Wrapping(init))
-            }
+            static RNG: Cell<Wrapping<u32>> = Cell::new(Wrapping(prng_seed()));
         }
 
         RNG.with(|rng| {
@@ -448,3 +445,31 @@ impl PartialEq for Pool {
 
 unsafe impl Send for Pool {}
 unsafe impl Sync for Pool {}
+
+// Return a thread-specific, 32-bit, non-zero seed value suitable for a 32-bit
+// PRNG. This uses one libstd RandomState for a default hasher and hashes on
+// the current thread ID to obtain an unpredictable, collision resistant seed.
+fn prng_seed() -> u32 {
+    // This obtains a small number of random bytes from the host system (for
+    // example, on unix via getrandom(2)) in order to seed an unpredictable and
+    // HashDoS resistant 64-bit hash function (currently: `SipHasher13` with
+    // 128-bit state). We only need one of these, to make the seeds for all
+    // process threads different via hashed IDs, collision resistant, and
+    // unpredictable.
+    lazy_static! {
+        static ref RND_STATE: RandomState = RandomState::new();
+    }
+
+    // Hash the current thread ID to produce a u32 value
+    let mut hasher = RND_STATE.build_hasher();
+    thread::current().id().hash(&mut hasher);
+    let hash: u64 = hasher.finish();
+    let seed = (hash as u32) ^ ((hash >> 32) as u32);
+
+    // Ensure non-zero seed (Xorshift yields only zero's for that seed)
+    if seed == 0 {
+        0x9b4e_6d25 // misc bits, could be any non-zero
+    } else {
+        seed
+    }
+}

tokio-threadpool/src/sender.rs

@@ -131,7 +131,7 @@ impl tokio_executor::Executor for Sender {
 
     fn spawn(
         &mut self,
-        future: Box<Future<Item = (), Error = ()> + Send>,
+        future: Box<dyn Future<Item = (), Error = ()> + Send>,
     ) -> Result<(), SpawnError> {
         let mut s = &*self;
         tokio_executor::Executor::spawn(&mut s, future)
@@ -157,7 +157,7 @@ impl<'a> tokio_executor::Executor for &'a Sender {
 
     fn spawn(
         &mut self,
-        future: Box<Future<Item = (), Error = ()> + Send>,
+        future: Box<dyn Future<Item = (), Error = ()> + Send>,
     ) -> Result<(), SpawnError> {
         self.prepare_for_spawn()?;
 

tokio-threadpool/src/task/mod.rs

@@ -60,7 +60,7 @@ pub(crate) enum Run {
     Complete,
 }
 
-type BoxFuture = Box<Future<Item = (), Error = ()> + Send + 'static>;
+type BoxFuture = Box<dyn Future<Item = (), Error = ()> + Send + 'static>;
 
 // ===== impl Task =====
 

tokio-threadpool/tests/threadpool.rs

@@ -18,7 +18,9 @@ use std::time::Duration;
 
 thread_local!(static FOO: Cell<u32> = Cell::new(0));
 
-fn ignore_results<F: Future + Send + 'static>(f: F) -> Box<Future<Item = (), Error = ()> + Send> {
+fn ignore_results<F: Future + Send + 'static>(
+    f: F,
+) -> Box<dyn Future<Item = (), Error = ()> + Send> {
     Box::new(f.map(|_| ()).map_err(|_| ()))
 }
 

tokio-timer/Cargo.toml

@@ -28,6 +28,6 @@ crossbeam-utils = "0.6.0"
 slab = "0.4.1"
 
 [dev-dependencies]
-rand = "0.6"
+rand = "0.7"
 tokio-mock-task = "0.1.0"
 tokio = "0.1.7"

tokio-timer/src/clock/clock.rs

@@ -17,7 +17,7 @@ use std::time::Instant;
 /// [`Instant::now`]: https://doc.rust-lang.org/std/time/struct.Instant.html#method.now
 #[derive(Default, Clone)]
 pub struct Clock {
-    now: Option<Arc<Now>>,
+    now: Option<Arc<dyn Now>>,
 }
 
 thread_local! {

tokio-timer/src/throttle.rs

@@ -158,7 +158,7 @@ impl<T: StdError + 'static> StdError for ThrottleError<T> {
     // FIXME(taiki-e): When the minimum support version of tokio reaches Rust 1.30,
     // replace this with Error::source.
     #[allow(deprecated)]
-    fn cause(&self) -> Option<&StdError> {
+    fn cause(&self) -> Option<&dyn StdError> {
         match self.0 {
             Either::A(ref err) => Some(err),
             Either::B(ref err) => Some(err),

tokio-timer/src/timer/mod.rs

@@ -162,7 +162,7 @@ pub(crate) struct Inner {
     process: AtomicStack,
 
     /// Unparks the timer thread.
-    unpark: Box<Unpark>,
+    unpark: Box<dyn Unpark>,
 }
 
 /// Maximum number of timeouts the system can handle concurrently.
@@ -426,7 +426,7 @@ impl<T, N> Drop for Timer<T, N> {
 // ===== impl Inner =====
 
 impl Inner {
-    fn new(start: Instant, unpark: Box<Unpark>) -> Inner {
+    fn new(start: Instant, unpark: Box<dyn Unpark>) -> Inner {
         Inner {
             num: AtomicUsize::new(0),
             elapsed: AtomicU64::new(0),

tokio-tls/Cargo.toml

@@ -31,7 +31,7 @@ tokio-io = "0.1.7"
 [dev-dependencies]
 tokio = "0.1"
 cfg-if = "0.1"
-env_logger = { version = "0.5", default-features = false }
+env_logger = { version = "0.6", default-features = false }
 
 [target.'cfg(all(not(target_os = "macos"), not(windows), not(target_os = "ios")))'.dev-dependencies]
 openssl = "0.10"

tokio-tls/examples/download-rust-lang.rs

@@ -12,7 +12,7 @@ use native_tls::TlsConnector;
 use tokio::net::TcpStream;
 use tokio::runtime::Runtime;
 
-fn main() -> Result<(), Box<std::error::Error>> {
+fn main() -> Result<(), Box<dyn std::error::Error>> {
     let mut runtime = Runtime::new()?;
     let addr = "www.rust-lang.org:443"
         .to_socket_addrs()?

tokio-tls/examples/echo.rs

@@ -8,7 +8,7 @@ use tokio::io;
 use tokio::net::TcpListener;
 use tokio::prelude::*;
 
-fn main() -> Result<(), Box<std::error::Error>> {
+fn main() -> Result<(), Box<dyn std::error::Error>> {
     // Bind the server's socket
     let addr = "127.0.0.1:12345".parse()?;
     let tcp = TcpListener::bind(&addr)?;

tokio-udp/CHANGELOG.md

@@ -1,6 +1,15 @@
+# 0.1.5 (August 30, 2019)
+
+* Allow `UdpFramed::new` to revert to previous behavior for decoding frames (#1517)
+* Fix `UdpFramed` decoding to repeatedly call `Decoder::decode_eof (#1517)
+
+# 0.1.4 (August 28, 2019)
+
+* Fix `UdpFramed`'s ability to decode multiple frames in one datagram (#1444)
+
 # 0.1.3 (November 21, 2018)
 
-* Add `RecvDgram::into_parts` (#710).
+* Add `RecvDgram::into_parts` (#710)
 
 # 0.1.2 (August 23, 2018)
 

tokio-udp/Cargo.toml

@@ -8,10 +8,10 @@ name = "tokio-udp"
 #   - README.md
 # - Update CHANGELOG.md.
 # - Create "v0.1.x" git tag.
-version = "0.1.3"
+version = "0.1.5"
 authors = ["Carl Lerche <[email protected]>"]
 license = "MIT"
-documentation = "https://docs.rs/tokio-udp/0.1.3/tokio_udp"
+documentation = "https://docs.rs/tokio-udp/0.1.5/tokio_udp"
 repository = "https://github.com/tokio-rs/tokio"
 homepage = "https://tokio.rs"
 description = """
@@ -29,4 +29,4 @@ log = "0.4"
 futures = "0.1.19"
 
 [dev-dependencies]
-env_logger = { version = "0.5", default-features = false }
+env_logger = { version = "0.6", default-features = false }

tokio-udp/README.md

@@ -2,7 +2,7 @@
 
 UDP bindings for `tokio`.
 
-[Documentation](https://docs.rs/tokio-udp/0.1.3/tokio_udp/)
+[Documentation](https://docs.rs/tokio-udp/0.1.5/tokio_udp/)
 
 ## License
 

tokio-udp/src/frame.rs

@@ -33,6 +33,9 @@ pub struct UdpFramed<C> {
     wr: BytesMut,
     out_addr: SocketAddr,
     flushed: bool,
+    is_readable: bool,
+    repeat_decode: bool,
+    current_addr: Option<SocketAddr>,
 }
 
 impl<C: Decoder> Stream for UdpFramed<C> {
@@ -42,19 +45,59 @@ impl<C: Decoder> Stream for UdpFramed<C> {
     fn poll(&mut self) -> Poll<Option<(Self::Item)>, Self::Error> {
         self.rd.reserve(INITIAL_RD_CAPACITY);
 
-        let (n, addr) = unsafe {
-            // Read into the buffer without having to initialize the memory.
-            let (n, addr) = try_ready!(self.socket.poll_recv_from(self.rd.bytes_mut()));
-            self.rd.advance_mut(n);
-            (n, addr)
-        };
-        trace!("received {} bytes, decoding", n);
-        let frame_res = self.codec.decode(&mut self.rd);
-        self.rd.clear();
-        let frame = frame_res?;
-        let result = frame.map(|frame| (frame, addr)); // frame -> (frame, addr)
-        trace!("frame decoded from buffer");
-        Ok(Async::Ready(result))
+        if self.repeat_decode {
+            loop {
+                // Are there are still bytes left in the read buffer to decode?
+                if self.is_readable {
+                    // Use deocde_eof since every datagram contains its own
+                    // eof which is just the end of the datagram. This supports
+                    // the lines use case where there may not be a terminating
+                    // delimiter and thus you may never get the end of the frame.
+                    // This is generally fine for most implementations of codec
+                    // since by default this will defer to calling decode.
+                    if let Some(frame) = self.codec.decode_eof(&mut self.rd)? {
+                        trace!("frame decoded from buffer");
+
+                        let current_addr = self
+                            .current_addr
+                            .expect("will always be set before this line is called");
+
+                        return Ok(Async::Ready(Some((frame, current_addr))));
+                    }
+
+                    // if this line has been reached then decode has returned `None`.
+                    self.is_readable = false;
+                    self.rd.clear();
+                }
+
+                // We're out of data. Try and fetch more data to decode
+                let (n, addr) = unsafe {
+                    // Read into the buffer without having to initialize the memory.
+                    let (n, addr) = try_ready!(self.socket.poll_recv_from(self.rd.bytes_mut()));
+                    self.rd.advance_mut(n);
+                    (n, addr)
+                };
+
+                self.current_addr = Some(addr);
+                self.is_readable = true;
+
+                trace!("received {} bytes, decoding", n);
+            }
+        } else {
+            let (n, addr) = unsafe {
+                // Read into the buffer without having to initialize the memory.
+                let (n, addr) = try_ready!(self.socket.poll_recv_from(self.rd.bytes_mut()));
+                self.rd.advance_mut(n);
+                (n, addr)
+            };
+            trace!("received {} bytes, decoding", n);
+            let frame_res = self.codec.decode(&mut self.rd);
+            self.rd.clear();
+            let frame = frame_res?;
+            let result = frame.map(|frame| (frame, addr)); // frame -> (frame, addr)
+            trace!("frame decoded from buffer");
+            Ok(Async::Ready(result))
+        }
     }
 }
 
@@ -126,6 +169,27 @@ impl<C> UdpFramed<C> {
             rd: BytesMut::with_capacity(INITIAL_RD_CAPACITY),
             wr: BytesMut::with_capacity(INITIAL_WR_CAPACITY),
             flushed: true,
+            is_readable: false,
+            repeat_decode: false,
+            current_addr: None,
+        }
+    }
+
+    /// Create a new `UdpFramed` backed by the given socket and codec. That will
+    /// continue to call `decode_eof` until the decoder has cleared the entire buffer.
+    ///
+    /// See struct level documentation for more details.
+    pub fn with_decode(socket: UdpSocket, codec: C, repeat_decode: bool) -> UdpFramed<C> {
+        UdpFramed {
+            socket: socket,
+            codec: codec,
+            out_addr: SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::new(0, 0, 0, 0), 0)),
+            rd: BytesMut::with_capacity(INITIAL_RD_CAPACITY),
+            wr: BytesMut::with_capacity(INITIAL_WR_CAPACITY),
+            flushed: true,
+            is_readable: false,
+            repeat_decode,
+            current_addr: None,
         }
     }
 

tokio-udp/src/lib.rs

@@ -1,4 +1,4 @@
-#![doc(html_root_url = "https://docs.rs/tokio-tcp/0.1.3")]
+#![doc(html_root_url = "https://docs.rs/tokio-tcp/0.1.5")]
 #![deny(missing_docs, warnings, missing_debug_implementations)]
 
 //! UDP bindings for `tokio`.

tokio-udp/tests/udp.rs

@@ -1,3 +1,5 @@
+#![allow(deprecated, unused_must_use)]
+
 extern crate futures;
 extern crate tokio_codec;
 extern crate tokio_udp;
@@ -12,7 +14,7 @@ use std::net::SocketAddr;
 use futures::{Future, Poll, Sink, Stream};
 
 use bytes::{BufMut, BytesMut};
-use tokio_codec::{Decoder, Encoder};
+use tokio_codec::{Decoder, Encoder, LinesCodec};
 use tokio_udp::{UdpFramed, UdpSocket};
 
 macro_rules! t {
@@ -247,7 +249,7 @@ impl Encoder for ByteCodec {
 }
 
 #[test]
-fn send_framed() {
+fn send_framed_byte_codec() {
     drop(env_logger::try_init());
 
     let mut a_soc = t!(UdpSocket::bind(&t!("127.0.0.1:0".parse())));
@@ -288,3 +290,193 @@ fn send_framed() {
         assert_eq!(a_addr, addr);
     }
 }
+
+#[test]
+fn send_framed_lines_codec() {
+    drop(env_logger::try_init());
+
+    let a_soc = t!(UdpSocket::bind(&t!("127.0.0.1:0".parse())));
+    let b_soc = t!(UdpSocket::bind(&t!("127.0.0.1:0".parse())));
+    let a_addr = t!(a_soc.local_addr());
+    let b_addr = t!(b_soc.local_addr());
+
+    let a = UdpFramed::new(a_soc, ByteCodec);
+    let b = UdpFramed::with_decode(b_soc, LinesCodec::new(), true);
+
+    let msg = b"1\r\n2\r\n3\r\n".to_vec();
+
+    let send = a.send((msg.clone(), b_addr));
+    t!(send.wait());
+
+    let mut recv = Stream::wait(b).map(|e| e.unwrap());
+
+    assert_eq!(recv.next(), Some(("1".to_string(), a_addr)));
+    assert_eq!(recv.next(), Some(("2".to_string(), a_addr)));
+    assert_eq!(recv.next(), Some(("3".to_string(), a_addr)));
+}
+
+#[test]
+fn recv_framed_codec_errs() {
+    drop(env_logger::try_init());
+
+    #[derive(Debug)]
+    struct LinesCodecMaxLen {
+        max_len: usize,
+        codec: LinesCodec,
+    }
+
+    impl LinesCodecMaxLen {
+        fn new(max_len: usize) -> Self {
+            Self {
+                max_len,
+                codec: LinesCodec::new(),
+            }
+        }
+    }
+
+    impl Decoder for LinesCodecMaxLen {
+        type Item = String;
+        type Error = io::Error;
+
+        fn decode(&mut self, buf: &mut BytesMut) -> Result<Option<String>, io::Error> {
+            let opt_string = self.codec.decode_eof(buf)?;
+            match opt_string {
+                None => Ok(None),
+                Some(string) => {
+                    if string.len() > self.max_len {
+                        Err(io::Error::new(io::ErrorKind::InvalidData, "Too big"))
+                    } else {
+                        Ok(Some(string))
+                    }
+                }
+            }
+        }
+    }
+
+    let a_soc = t!(UdpSocket::bind(&t!("127.0.0.1:0".parse())));
+    let b_soc = t!(UdpSocket::bind(&t!("127.0.0.1:0".parse())));
+    let a_addr = t!(a_soc.local_addr());
+    let b_addr = t!(b_soc.local_addr());
+
+    {
+        let a = UdpFramed::new(a_soc, ByteCodec);
+        let b = UdpFramed::new(b_soc, LinesCodecMaxLen::new(/*max_len*/ 1));
+
+        let msg = b"hello world".to_vec(); // hello world is too big
+
+        let send = a.send((msg.clone(), b_addr));
+        let a = t!(send.wait());
+
+        let msg = b"1\r\n".to_vec(); // fits ok
+        let send = a.send((msg.clone(), b_addr));
+        t!(send.wait());
+
+        let mut b = Stream::wait(b);
+
+        let hello_world = b.next().unwrap();
+        assert!(hello_world.is_err()); // first one is too big
+
+        let mut recv = b.map(|e| e.unwrap());
+
+        // and then we restore the state and continue receiving
+        assert_eq!(recv.next(), Some(("1".to_string(), a_addr)));
+    }
+}
+
+#[test]
+fn send_framed_lines_codec_with_non_terminating_frame() {
+    drop(env_logger::try_init());
+
+    let a_soc = t!(UdpSocket::bind(&t!("127.0.0.1:0".parse())));
+    let b_soc = t!(UdpSocket::bind(&t!("127.0.0.1:0".parse())));
+    let a_addr = t!(a_soc.local_addr());
+    let b_addr = t!(b_soc.local_addr());
+
+    let a = UdpFramed::new(a_soc, ByteCodec);
+    let b = UdpFramed::with_decode(b_soc, LinesCodec::new(), true);
+
+    // This has no terminating delimiter thus we want to return the rest of the
+    // frame and this tests that if decode fails, we try to decode_eof.
+    let msg = b"1\r\n2\r\n3".to_vec();
+
+    let send = a.send((msg.clone(), b_addr));
+    t!(send.wait());
+
+    let mut recv = Stream::wait(b).map(|e| e.unwrap());
+
+    assert_eq!(recv.next(), Some(("1".to_string(), a_addr)));
+    assert_eq!(recv.next(), Some(("2".to_string(), a_addr)));
+    assert_eq!(recv.next(), Some(("3".to_string(), a_addr)));
+}
+
+#[test]
+fn recv_multi_framed_lines_codec_errs() {
+    drop(env_logger::try_init());
+
+    #[derive(Debug)]
+    struct LinesCodecMaxLen {
+        max_len: usize,
+        codec: LinesCodec,
+    }
+
+    impl LinesCodecMaxLen {
+        fn new(max_len: usize) -> Self {
+            Self {
+                max_len,
+                codec: LinesCodec::new(),
+            }
+        }
+    }
+
+    impl Decoder for LinesCodecMaxLen {
+        type Item = String;
+        type Error = io::Error;
+
+        fn decode(&mut self, buf: &mut BytesMut) -> Result<Option<String>, io::Error> {
+            return self.codec.decode(buf);
+        }
+
+        fn decode_eof(&mut self, buf: &mut BytesMut) -> Result<Option<String>, io::Error> {
+            let opt_string = self.codec.decode_eof(buf)?;
+            match opt_string {
+                None => Ok(None),
+                Some(string) => {
+                    if string.len() > self.max_len {
+                        Err(io::Error::new(io::ErrorKind::InvalidData, "Too big"))
+                    } else {
+                        Ok(Some(string))
+                    }
+                }
+            }
+        }
+    }
+
+    let a_soc = t!(UdpSocket::bind(&t!("127.0.0.1:0".parse())));
+    let b_soc = t!(UdpSocket::bind(&t!("127.0.0.1:0".parse())));
+    let a_addr = t!(a_soc.local_addr());
+    let b_addr = t!(b_soc.local_addr());
+
+    let a = UdpFramed::new(a_soc, ByteCodec);
+    let b = UdpFramed::with_decode(b_soc, LinesCodecMaxLen::new(/*max_len*/ 1), true);
+
+    let msg = b"hello world".to_vec(); // hello world is too big
+
+    let send = a.send((msg.clone(), b_addr));
+    let a = t!(send.wait());
+
+    let msg = b"1\r\n2\r\n3\r\n".to_vec();
+    let send = a.send((msg.clone(), b_addr));
+    t!(send.wait());
+
+    let mut b = Stream::wait(b);
+
+    let hello_world = b.next().unwrap();
+    assert!(hello_world.is_err()); // first one is too big
+
+    let mut recv = b.map(|e| e.unwrap());
+
+    // and then we restore the state and continue receiving
+    assert_eq!(recv.next(), Some(("1".to_string(), a_addr)));
+    assert_eq!(recv.next(), Some(("2".to_string(), a_addr)));
+    assert_eq!(recv.next(), Some(("3".to_string(), a_addr)));
+}

tokio-uds/Cargo.toml

@@ -33,4 +33,4 @@ tokio-io = "0.1.6"
 
 [dev-dependencies]
 tokio = "0.1.6"
-tempfile = "3"
+tempfile = "~3.1.0"

tokio/Cargo.toml

@@ -82,8 +82,8 @@ mio = { version = "0.6.14", optional = true }
 tokio-uds = { version = "0.2.1", optional = true }
 
 [dev-dependencies]
-env_logger = { version = "0.5", default-features = false }
-flate2 = { version = "1", features = ["tokio"] }
+env_logger = { version = "0.6", default-features = false }
+flate2 = { version = ">=1.0.2, <1.0.10", features = ["tokio"] }
 futures-cpupool = "0.1"
 http = "0.1"
 httparse = "1.0"

tokio/examples/chat-combinator-current-thread.rs

@@ -41,7 +41,7 @@ use std::io::BufReader;
 use std::iter;
 use std::rc::Rc;
 
-fn main() -> Result<(), Box<std::error::Error>> {
+fn main() -> Result<(), Box<dyn std::error::Error>> {
     let mut runtime = Runtime::new().unwrap();
 
     // Create the TCP listener we'll accept connections on.

tokio/examples/chat-combinator.rs

@@ -34,7 +34,7 @@ use std::io::BufReader;
 use std::iter;
 use std::sync::{Arc, Mutex};
 
-fn main() -> Result<(), Box<std::error::Error>> {
+fn main() -> Result<(), Box<dyn std::error::Error>> {
     // Create the TCP listener we'll accept connections on.
     let addr = env::args().nth(1).unwrap_or("127.0.0.1:8080".to_string());
     let addr = addr.parse()?;

tokio/examples/chat.rs

@@ -422,7 +422,7 @@ fn process(socket: TcpStream, state: Arc<Mutex<Shared>>) {
     tokio::spawn(connection);
 }
 
-pub fn main() -> Result<(), Box<std::error::Error>> {
+pub fn main() -> Result<(), Box<dyn std::error::Error>> {
     // Create the shared state. This is how all the peers communicate.
     //
     // The server task will hold a handle to this. For every new client, the

tokio/examples/connect.rs

@@ -29,7 +29,7 @@ use std::thread;
 use futures::sync::mpsc;
 use tokio::prelude::*;
 
-fn main() -> Result<(), Box<std::error::Error>> {
+fn main() -> Result<(), Box<dyn std::error::Error>> {
     // Determine if we're going to run in TCP or UDP mode
     let mut args = env::args().skip(1).collect::<Vec<_>>();
     let tcp = match args.iter().position(|a| a == "--udp") {
@@ -133,8 +133,8 @@ mod tcp {
 
     pub fn connect(
         addr: &SocketAddr,
-        stdin: Box<Stream<Item = Vec<u8>, Error = io::Error> + Send>,
-    ) -> Result<Box<Stream<Item = BytesMut, Error = io::Error> + Send>, Box<Error>> {
+        stdin: Box<dyn Stream<Item = Vec<u8>, Error = io::Error> + Send>,
+    ) -> Result<Box<dyn Stream<Item = BytesMut, Error = io::Error> + Send>, Box<dyn Error>> {
         let tcp = TcpStream::connect(addr);
 
         // After the TCP connection has been established, we set up our client
@@ -185,8 +185,8 @@ mod udp {
 
     pub fn connect(
         &addr: &SocketAddr,
-        stdin: Box<Stream<Item = Vec<u8>, Error = io::Error> + Send>,
-    ) -> Result<Box<Stream<Item = BytesMut, Error = io::Error> + Send>, Box<Error>> {
+        stdin: Box<dyn Stream<Item = Vec<u8>, Error = io::Error> + Send>,
+    ) -> Result<Box<dyn Stream<Item = BytesMut, Error = io::Error> + Send>, Box<dyn Error>> {
         // We'll bind our UDP socket to a local IP/port, but for now we
         // basically let the OS pick both of those.
         let addr_to_bind = if addr.ip().is_ipv4() {

tokio/examples/echo-udp.rs

@@ -50,7 +50,7 @@ impl Future for Server {
     }
 }
 
-fn main() -> Result<(), Box<std::error::Error>> {
+fn main() -> Result<(), Box<dyn std::error::Error>> {
     let addr = env::args().nth(1).unwrap_or("127.0.0.1:8080".to_string());
     let addr = addr.parse::<SocketAddr>()?;
 

tokio/examples/echo.rs

@@ -30,7 +30,7 @@ use tokio::prelude::*;
 use std::env;
 use std::net::SocketAddr;
 
-fn main() -> Result<(), Box<std::error::Error>> {
+fn main() -> Result<(), Box<dyn std::error::Error>> {
     // Allow passing an address to listen on as the first argument of this
     // program, but otherwise we'll just set up our TCP listener on
     // 127.0.0.1:8080 for connections.

tokio/examples/hello_world.rs

@@ -19,7 +19,7 @@ use tokio::io;
 use tokio::net::TcpStream;
 use tokio::prelude::*;
 
-pub fn main() -> Result<(), Box<std::error::Error>> {
+pub fn main() -> Result<(), Box<dyn std::error::Error>> {
     let addr = "127.0.0.1:6142".parse()?;
 
     // Open a TCP stream to the socket address.

tokio/examples/manual-runtime.rs

@@ -60,7 +60,7 @@ fn run<F: Future<Item = (), Error = ()>>(f: F) -> Result<(), IoError> {
     Ok(())
 }
 
-fn main() -> Result<(), Box<std::error::Error>> {
+fn main() -> Result<(), Box<dyn std::error::Error>> {
     run(future::lazy(|| {
         // Here comes the application logic. It can spawn further tasks by tokio_current_thread::spawn().
         // It also can use the default reactor and create timeouts.

tokio/examples/print_each_packet.rs

@@ -65,7 +65,7 @@ use tokio_codec::BytesCodec;
 use std::env;
 use std::net::SocketAddr;
 
-fn main() -> Result<(), Box<std::error::Error>> {
+fn main() -> Result<(), Box<dyn std::error::Error>> {
     // Allow passing an address to listen on as the first argument of this
     // program, but otherwise we'll just set up our TCP listener on
     // 127.0.0.1:8080 for connections.

tokio/examples/proxy.rs

@@ -33,7 +33,7 @@ use tokio::io::{copy, shutdown};
 use tokio::net::{TcpListener, TcpStream};
 use tokio::prelude::*;
 
-fn main() -> Result<(), Box<std::error::Error>> {
+fn main() -> Result<(), Box<dyn std::error::Error>> {
     let listen_addr = env::args().nth(1).unwrap_or("127.0.0.1:8081".to_string());
     let listen_addr = listen_addr.parse::<SocketAddr>()?;
 

tokio/examples/tinydb.rs

@@ -83,7 +83,7 @@ enum Response {
     },
 }
 
-fn main() -> Result<(), Box<std::error::Error>> {
+fn main() -> Result<(), Box<dyn std::error::Error>> {
     // Parse the address we're going to run this server on
     // and set up our TCP listener to accept connections.
     let addr = env::args().nth(1).unwrap_or("127.0.0.1:8080".to_string());

tokio/examples/tinyhttp.rs

@@ -34,7 +34,7 @@ use bytes::BytesMut;
 use http::header::HeaderValue;
 use http::{Request, Response, StatusCode};
 
-fn main() -> Result<(), Box<std::error::Error>> {
+fn main() -> Result<(), Box<dyn std::error::Error>> {
     // Parse the arguments, bind the TCP socket we'll be listening to, spin up
     // our worker threads, and start shipping sockets to those worker threads.
     let addr = env::args().nth(1).unwrap_or("127.0.0.1:8080".to_string());
@@ -82,7 +82,7 @@ fn process(socket: TcpStream) {
 /// This function is a map from and HTTP request to a future of a response and
 /// represents the various handling a server might do. Currently the contents
 /// here are pretty uninteresting.
-fn respond(req: Request<()>) -> Box<Future<Item = Response<String>, Error = io::Error> + Send> {
+fn respond(req: Request<()>) -> Box<dyn Future<Item = Response<String>, Error = io::Error> + Send> {
     let f = future::lazy(move || {
         let mut response = Response::builder();
         let body = match req.uri().path() {

tokio/examples/udp-client.rs

@@ -35,13 +35,13 @@ use std::net::SocketAddr;
 use tokio::net::UdpSocket;
 use tokio::prelude::*;
 
-fn get_stdin_data() -> Result<Vec<u8>, Box<std::error::Error>> {
+fn get_stdin_data() -> Result<Vec<u8>, Box<dyn std::error::Error>> {
     let mut buf = Vec::new();
     stdin().read_to_end(&mut buf)?;
     Ok(buf)
 }
 
-fn main() -> Result<(), Box<std::error::Error>> {
+fn main() -> Result<(), Box<dyn std::error::Error>> {
     let remote_addr: SocketAddr = env::args()
         .nth(1)
         .unwrap_or("127.0.0.1:8080".into())

tokio/examples/udp-codec.rs

@@ -19,7 +19,7 @@ use tokio::net::{UdpFramed, UdpSocket};
 use tokio::prelude::*;
 use tokio_codec::BytesCodec;
 
-fn main() -> Result<(), Box<std::error::Error>> {
+fn main() -> Result<(), Box<dyn std::error::Error>> {
     let _ = env_logger::init();
 
     let addr: SocketAddr = "127.0.0.1:0".parse()?;

tokio/src/runtime/current_thread/runtime.rs

@@ -105,7 +105,7 @@ impl Error for RunError {
     // FIXME(taiki-e): When the minimum support version of tokio reaches Rust 1.30,
     // replace this with Error::source.
     #[allow(deprecated)]
-    fn cause(&self) -> Option<&Error> {
+    fn cause(&self) -> Option<&dyn Error> {
         self.inner.cause()
     }
 }

tokio/src/runtime/threadpool/builder.rs

@@ -128,7 +128,7 @@ impl Builder {
     /// ```
     pub fn panic_handler<F>(&mut self, f: F) -> &mut Self
     where
-        F: Fn(Box<Any + Send>) + Send + Sync + 'static,
+        F: Fn(Box<dyn Any + Send>) + Send + Sync + 'static,
     {
         self.threadpool_builder.panic_handler(f);
         self

tokio/src/runtime/threadpool/task_executor.rs

@@ -67,7 +67,7 @@ where T: Future<Item = (), Error = ()> + Send + 'static,
 }
 
 impl ::executor::Executor for TaskExecutor {
-    fn spawn(&mut self, future: Box<Future<Item = (), Error = ()> + Send>)
+    fn spawn(&mut self, future: Box<dyn Future<Item = (), Error = ()> + Send>)
         -> Result<(), ::executor::SpawnError>
     {
         self.inner.spawn(future)

tokio/tests/reactor.rs

@@ -30,7 +30,7 @@ fn test_drop_on_notify() {
 
     struct MyNotify;
 
-    type Task = Mutex<Spawn<Box<Future<Item = (), Error = ()>>>>;
+    type Task = Mutex<Spawn<Box<dyn Future<Item = (), Error = ()>>>>;
 
     impl Notify for MyNotify {
         fn notify(&self, _: usize) {
@@ -66,7 +66,7 @@ fn test_drop_on_notify() {
             .incoming()
             .for_each(|_| Ok(()))
             .map_err(|_| panic!())
-    }) as Box<Future<Item = (), Error = ()>>;
+    }) as Box<dyn Future<Item = (), Error = ()>>;
 
     let task = Arc::new(Mutex::new(spawn(task)));
     let notify = Arc::new(MyNotify);

tokio/tests/runtime.rs

@@ -25,7 +25,7 @@ macro_rules! t {
     };
 }
 
-fn create_client_server_future() -> Box<Future<Item = (), Error = ()> + Send> {
+fn create_client_server_future() -> Box<dyn Future<Item = (), Error = ()> + Send> {
     let server = t!(TcpListener::bind(&"127.0.0.1:0".parse().unwrap()));
     let addr = t!(server.local_addr());
     let client = TcpStream::connect(&addr);
@@ -84,7 +84,7 @@ mod runtime_single_threaded_block_on_all {
 
     fn test<F>(spawn: F)
     where
-        F: Fn(Box<Future<Item = (), Error = ()> + Send>),
+        F: Fn(Box<dyn Future<Item = (), Error = ()> + Send>),
     {
         let cnt = Arc::new(Mutex::new(0));
         let c = cnt.clone();
@@ -133,7 +133,10 @@ mod runtime_single_threaded_racy {
     use super::*;
     fn test<F>(spawn: F)
     where
-        F: Fn(tokio::runtime::current_thread::Handle, Box<Future<Item = (), Error = ()> + Send>),
+        F: Fn(
+            tokio::runtime::current_thread::Handle,
+            Box<dyn Future<Item = (), Error = ()> + Send>,
+        ),
     {
         let (trigger, exit) = futures::sync::oneshot::channel();
         let (handle_tx, handle_rx) = ::std::sync::mpsc::channel();
@@ -218,7 +221,7 @@ fn block_on_timer() {
     use std::time::{Duration, Instant};
     use tokio::timer::{Delay, Error};
 
-    fn after_1s<T>(x: T) -> Box<Future<Item = T, Error = Error> + Send>
+    fn after_1s<T>(x: T) -> Box<dyn Future<Item = T, Error = Error> + Send>
     where
         T: Send + 'static,
     {
@@ -235,7 +238,7 @@ mod from_block_on {
 
     fn test<F>(spawn: F)
     where
-        F: Fn(Box<Future<Item = (), Error = ()> + Send>) + Send + 'static,
+        F: Fn(Box<dyn Future<Item = (), Error = ()> + Send>) + Send + 'static,
     {
         let cnt = Arc::new(Mutex::new(0));
         let c = cnt.clone();
@@ -317,7 +320,7 @@ mod many {
     const ITER: usize = 200;
     fn test<F>(spawn: F)
     where
-        F: Fn(&mut Runtime, Box<Future<Item = (), Error = ()> + Send>),
+        F: Fn(&mut Runtime, Box<dyn Future<Item = (), Error = ()> + Send>),
     {
         let cnt = Arc::new(Mutex::new(0));
         let mut runtime = Runtime::new().unwrap();
@@ -360,7 +363,7 @@ mod from_block_on_all {
 
     fn test<F>(spawn: F)
     where
-        F: Fn(Box<Future<Item = (), Error = ()> + Send>) + Send + 'static,
+        F: Fn(Box<dyn Future<Item = (), Error = ()> + Send>) + Send + 'static,
     {
         let cnt = Arc::new(Mutex::new(0));
         let c = cnt.clone();
@@ -414,8 +417,8 @@ mod nested_enter {
 
     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,
+        F1: Fn(Box<dyn Future<Item = (), Error = ()> + Send>) + Send + 'static,
+        F2: Fn(Box<dyn Future<Item = (), Error = ()> + Send>) + panic::UnwindSafe + Send + 'static,
     {
         let panicked = Arc::new(Mutex::new(false));
         let panicked2 = panicked.clone();