diff --git a/tokio-executor/Cargo.toml b/tokio-executor/Cargo.toml
index 4fb77c5f3..d1bd6a189 100644
--- a/tokio-executor/Cargo.toml
+++ b/tokio-executor/Cargo.toml
@@ -21,13 +21,13 @@ keywords = ["futures", "tokio"]
 categories = ["concurrency", "asynchronous"]
 
 [features]
-blocking = ["tokio-sync", "lazy_static"]
+blocking = ["lazy_static"]
 current-thread = ["crossbeam-channel"]
 thread-pool = ["num_cpus"]
 
 [dependencies]
 futures-util-preview = { version = "=0.3.0-alpha.19", features = ["channel"] }
-tokio-sync = { version = "=0.2.0-alpha.6", optional = true, path = "../tokio-sync" }
+tokio-sync = { version = "=0.2.0-alpha.6", path = "../tokio-sync" }
 
 # current-thread dependencies
 crossbeam-channel = { version = "0.3.8", optional = true }
@@ -45,7 +45,7 @@ tokio-sync = { version = "=0.2.0-alpha.6", path = "../tokio-sync" }
 tokio-test = { version = "=0.2.0-alpha.6", path = "../tokio-test" }
 
 futures-core-preview = "=0.3.0-alpha.19"
-loom = { version = "0.2.9", features = ["futures", "checkpoint"] }
+loom = { version = "0.2.11", features = ["futures", "checkpoint"] }
 rand = "0.7"
 
 [package.metadata.docs.rs]
diff --git a/tokio-executor/src/blocking.rs b/tokio-executor/src/blocking.rs
deleted file mode 100644
index 724872466..000000000
--- a/tokio-executor/src/blocking.rs
+++ /dev/null
@@ -1,170 +0,0 @@
-//! Thread pool for blocking operations
-
-use tokio_sync::oneshot;
-
-use lazy_static::lazy_static;
-use std::collections::VecDeque;
-use std::future::Future;
-use std::pin::Pin;
-use std::sync::{Condvar, Mutex};
-use std::task::{Context, Poll};
-use std::thread;
-use std::time::Duration;
-
-struct Pool {
-    shared: Mutex<Shared>,
-    condvar: Condvar,
-}
-
-struct Shared {
-    queue: VecDeque<Box<dyn FnOnce() + Send>>,
-    num_th: u32,
-    num_idle: u32,
-    num_notify: u32,
-}
-
-lazy_static! {
-    static ref POOL: Pool = Pool::new();
-}
-
-const MAX_THREADS: u32 = 1_000;
-const KEEP_ALIVE: Duration = Duration::from_secs(10);
-
-/// Result of a blocking operation running on the blocking thread pool.
-#[derive(Debug)]
-pub struct Blocking<T> {
-    rx: oneshot::Receiver<T>,
-}
-
-/// Run the provided function on a threadpool dedicated to blocking operations.
-pub fn run<F, R>(f: F) -> Blocking<R>
-where
-    F: FnOnce() -> R + Send + 'static,
-    R: Send + 'static,
-{
-    let (tx, rx) = oneshot::channel();
-
-    let should_spawn = {
-        let mut shared = POOL.shared.lock().unwrap();
-
-        shared.queue.push_back(Box::new(move || {
-            // The receiver may have been dropped.
-            let _ = tx.send(f());
-        }));
-
-        if shared.num_idle == 0 {
-            // No threads are able to process the task.
-
-            if shared.num_th == MAX_THREADS {
-                // At max number of threads
-                false
-            } else {
-                shared.num_th += 1;
-                true
-            }
-        } else {
-            // Notify an idle worker thread. The notification counter
-            // is used to count the needed amount of notifications
-            // exactly. Thread libraries may generate spurious
-            // wakeups, this counter is used to keep us in a
-            // consistent state.
-            shared.num_idle -= 1;
-            shared.num_notify += 1;
-            POOL.condvar.notify_one();
-            false
-        }
-    };
-
-    if should_spawn {
-        spawn_thread();
-    }
-
-    Blocking { rx }
-}
-
-impl<T> Future for Blocking<T> {
-    type Output = T;
-
-    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
-        use std::task::Poll::*;
-
-        match Pin::new(&mut self.rx).poll(cx) {
-            Ready(Ok(v)) => Ready(v),
-            Ready(Err(_)) => panic!(
-                "the blocking operation has been dropped before completing. \
-                 This should not happen and is a bug."
-            ),
-            Pending => Pending,
-        }
-    }
-}
-
-fn spawn_thread() {
-    thread::Builder::new()
-        .name("tokio-blocking-driver".to_string())
-        .spawn(|| {
-            let mut shared = POOL.shared.lock().unwrap();
-            loop {
-                // BUSY
-                while let Some(task) = shared.queue.pop_front() {
-                    drop(shared);
-                    run_task(task);
-                    shared = POOL.shared.lock().unwrap();
-                }
-
-                // IDLE
-                shared.num_idle += 1;
-
-                loop {
-                    let lock_result = POOL.condvar.wait_timeout(shared, KEEP_ALIVE).unwrap();
-                    shared = lock_result.0;
-                    let timeout_result = lock_result.1;
-
-                    if shared.num_notify != 0 {
-                        // We have received a legitimate wakeup,
-                        // acknowledge it by decrementing the counter
-                        // and transition to the BUSY state.
-                        shared.num_notify -= 1;
-                        break;
-                    }
-
-                    if timeout_result.timed_out() {
-                        // Thread exit
-                        shared.num_th -= 1;
-
-                        // num_idle should now be tracked exactly, panic
-                        // with a descriptive message if it is not the
-                        // case.
-                        shared.num_idle = shared
-                            .num_idle
-                            .checked_sub(1)
-                            .expect("num_idle underflowed on thread exit");
-                        return;
-                    }
-
-                    // Spurious wakeup detected, go back to sleep.
-                }
-            }
-        })
-        .unwrap();
-}
-
-fn run_task(f: Box<dyn FnOnce() + Send>) {
-    use std::panic::{catch_unwind, AssertUnwindSafe};
-
-    let _ = catch_unwind(AssertUnwindSafe(|| f()));
-}
-
-impl Pool {
-    fn new() -> Pool {
-        Pool {
-            shared: Mutex::new(Shared {
-                queue: VecDeque::new(),
-                num_th: 0,
-                num_idle: 0,
-                num_notify: 0,
-            }),
-            condvar: Condvar::new(),
-        }
-    }
-}
diff --git a/tokio-executor/src/blocking/builder.rs b/tokio-executor/src/blocking/builder.rs
new file mode 100644
index 000000000..3cf1aef8f
--- /dev/null
+++ b/tokio-executor/src/blocking/builder.rs
@@ -0,0 +1,57 @@
+use super::Pool;
+use crate::loom::thread;
+use std::usize;
+
+/// Builds a blocking thread pool with custom configuration values.
+pub(crate) struct Builder {
+    /// Thread name
+    name: String,
+
+    /// Thread stack size
+    stack_size: Option<usize>,
+}
+
+impl Default for Builder {
+    fn default() -> Self {
+        Builder {
+            name: "tokio-blocking-thread".to_string(),
+            stack_size: None,
+        }
+    }
+}
+
+impl Builder {
+    /// Set name of threads spawned by the pool
+    ///
+    /// If this configuration is not set, then the thread will use the system
+    /// default naming scheme.
+    pub(crate) fn name<S: Into<String>>(&mut self, val: S) -> &mut Self {
+        self.name = val.into();
+        self
+    }
+
+    /// Set the stack size (in bytes) for worker threads.
+    ///
+    /// The actual stack size may be greater than this value if the platform
+    /// specifies minimal stack size.
+    ///
+    /// The default stack size for spawned threads is 2 MiB, though this
+    /// particular stack size is subject to change in the future.
+    pub(crate) fn stack_size(&mut self, val: usize) -> &mut Self {
+        self.stack_size = Some(val);
+        self
+    }
+
+    pub(crate) fn build(self) -> Pool {
+        let mut p = Pool::default();
+        let Builder { stack_size, name } = self;
+        p.new_thread = Box::new(move || {
+            let mut b = thread::Builder::new().name(name.clone());
+            if let Some(stack_size) = stack_size {
+                b = b.stack_size(stack_size);
+            }
+            b
+        });
+        p
+    }
+}
diff --git a/tokio-executor/src/blocking/mod.rs b/tokio-executor/src/blocking/mod.rs
new file mode 100644
index 000000000..afa928555
--- /dev/null
+++ b/tokio-executor/src/blocking/mod.rs
@@ -0,0 +1,325 @@
+//! Thread pool for blocking operations
+
+use crate::loom::sync::{Arc, Condvar, Mutex};
+use crate::loom::thread;
+use tokio_sync::oneshot;
+
+use std::cell::Cell;
+use std::collections::VecDeque;
+use std::fmt;
+use std::future::Future;
+use std::ops::Deref;
+use std::pin::Pin;
+use std::task::{Context, Poll};
+use std::time::Duration;
+
+#[cfg(feature = "thread-pool")]
+mod builder;
+
+#[cfg(feature = "thread-pool")]
+pub(crate) use builder::Builder;
+
+#[derive(Clone, Copy)]
+enum State {
+    Empty,
+    Ready(*const Arc<Pool>),
+}
+
+thread_local! {
+    /// Thread-local tracking the current executor
+    static BLOCKING: Cell<State> = Cell::new(State::Empty)
+}
+
+/// Set the blocking pool for the duration of the closure
+///
+/// If a blocking pool is already set, it will be restored when the closure returns or if it
+/// panics.
+#[allow(dead_code)] // we allow dead code since this won't be called if no executors are enabled
+pub(crate) fn with_pool<F, R>(pool: &Arc<Pool>, f: F) -> R
+where
+    F: FnOnce() -> R,
+{
+    // While scary, this is safe. The function takes a `&Pool`, which guarantees
+    // that the reference lives for the duration of `with_pool`.
+    //
+    // Because we are always clearing the TLS value at the end of the
+    // function, we can cast the reference to 'static which thread-local
+    // cells require.
+    BLOCKING.with(|cell| {
+        let was = cell.replace(State::Empty);
+
+        // Ensure that the pool is removed from the thread-local context
+        // when leaving the scope. This handles cases that involve panicking.
+        struct Reset<'a>(&'a Cell<State>, State);
+
+        impl Drop for Reset<'_> {
+            fn drop(&mut self) {
+                self.0.set(self.1);
+            }
+        }
+
+        let _reset = Reset(cell, was);
+        cell.set(State::Ready(pool as *const _));
+        f()
+    })
+}
+
+pub(crate) struct Pool {
+    shared: Mutex<Shared>,
+    condvar: Condvar,
+    new_thread: Box<dyn Fn() -> thread::Builder + Send + Sync + 'static>,
+}
+
+impl fmt::Debug for Pool {
+    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
+        fmt.debug_struct("Pool").finish()
+    }
+}
+
+struct Shared {
+    queue: VecDeque<Box<dyn FnOnce() + Send>>,
+    num_th: u32,
+    num_idle: u32,
+    num_notify: u32,
+    shutdown: bool,
+}
+
+const MAX_THREADS: u32 = 1_000;
+const KEEP_ALIVE: Duration = Duration::from_secs(10);
+
+/// Result of a blocking operation running on the blocking thread pool.
+#[derive(Debug)]
+pub struct Blocking<T> {
+    rx: oneshot::Receiver<T>,
+}
+
+impl Pool {
+    /// Run the provided function on an executor dedicated to blocking operations.
+    pub(crate) fn spawn(this: &Arc<Self>, f: Box<dyn FnOnce() + Send + 'static>) {
+        let should_spawn = {
+            let mut shared = this.shared.lock().unwrap();
+
+            if shared.shutdown {
+                // no need to even push this task; it would never get picked up
+                return;
+            }
+
+            shared.queue.push_back(f);
+
+            if shared.num_idle == 0 {
+                // No threads are able to process the task.
+
+                if shared.num_th == MAX_THREADS {
+                    // At max number of threads
+                    false
+                } else {
+                    shared.num_th += 1;
+                    true
+                }
+            } else {
+                // Notify an idle worker thread. The notification counter
+                // is used to count the needed amount of notifications
+                // exactly. Thread libraries may generate spurious
+                // wakeups, this counter is used to keep us in a
+                // consistent state.
+                shared.num_idle -= 1;
+                shared.num_notify += 1;
+                this.condvar.notify_one();
+                false
+            }
+        };
+
+        if should_spawn {
+            Pool::spawn_thread(Arc::clone(this), (this.new_thread)());
+        }
+    }
+
+    // NOTE: we cannot use self here w/o arbitrary_self_types since Arc is loom::Arc
+    fn spawn_thread(this: Arc<Self>, builder: thread::Builder) {
+        builder
+            .spawn(move || {
+                let mut shared = this.shared.lock().unwrap();
+                'main: loop {
+                    // BUSY
+                    while let Some(task) = shared.queue.pop_front() {
+                        drop(shared);
+                        run_task(task);
+                        shared = this.shared.lock().unwrap();
+                        if shared.shutdown {
+                            break; // Need to increment idle before we exit
+                        }
+                    }
+
+                    // IDLE
+                    shared.num_idle += 1;
+
+                    while !shared.shutdown {
+                        let lock_result = this.condvar.wait_timeout(shared, KEEP_ALIVE).unwrap();
+                        shared = lock_result.0;
+                        let timeout_result = lock_result.1;
+
+                        if shared.num_notify != 0 {
+                            // We have received a legitimate wakeup,
+                            // acknowledge it by decrementing the counter
+                            // and transition to the BUSY state.
+                            shared.num_notify -= 1;
+                            break;
+                        }
+
+                        if timeout_result.timed_out() {
+                            break 'main;
+                        }
+
+                        // Spurious wakeup detected, go back to sleep.
+                    }
+
+                    if shared.shutdown {
+                        // Work was produced, and we "took" it (by decrementing num_notify).
+                        // This means that num_idle was decremented once for our wakeup.
+                        // But, since we are exiting, we need to "undo" that, as we'll stay idle.
+                        shared.num_idle += 1;
+                        // NOTE: Technically we should also do num_notify++ and notify again,
+                        // but since we're shutting down anyway, that won't be necessary.
+                        break;
+                    }
+                }
+
+                // Thread exit
+                shared.num_th -= 1;
+
+                // num_idle should now be tracked exactly, panic
+                // with a descriptive message if it is not the
+                // case.
+                shared.num_idle = shared
+                    .num_idle
+                    .checked_sub(1)
+                    .expect("num_idle underflowed on thread exit");
+
+                if shared.shutdown && shared.num_th == 0 {
+                    this.condvar.notify_one();
+                }
+            })
+            .unwrap();
+    }
+
+    /// Shut down all workers in the pool the next time they are idle.
+    ///
+    /// Blocks until all threads have exited.
+    pub(crate) fn shutdown(&self) {
+        let mut shared = self.shared.lock().unwrap();
+        shared.shutdown = true;
+        self.condvar.notify_all();
+
+        while shared.num_th > 0 {
+            shared = self.condvar.wait(shared).unwrap();
+        }
+    }
+}
+
+pub(crate) struct PoolWaiter(Arc<Pool>);
+
+impl From<Pool> for PoolWaiter {
+    fn from(p: Pool) -> Self {
+        Self::from(Arc::new(p))
+    }
+}
+
+impl From<Arc<Pool>> for PoolWaiter {
+    fn from(p: Arc<Pool>) -> Self {
+        Self(p)
+    }
+}
+
+impl Deref for PoolWaiter {
+    type Target = Arc<Pool>;
+    fn deref(&self) -> &Self::Target {
+        &self.0
+    }
+}
+
+impl Drop for PoolWaiter {
+    fn drop(&mut self) {
+        self.0.shutdown();
+    }
+}
+
+/// Run the provided closure on a thread where blocking is acceptable.
+///
+/// In general, issuing a blocking call or performing a lot of compute in a future without
+/// yielding is not okay, as it may prevent the executor from driving other futures forward.
+/// A closure that is run through this method will instead be run on a dedicated thread pool for
+/// such blocking tasks without holding up the main futures executor.
+///
+/// # Examples
+///
+/// ```
+/// # async fn docs() {
+/// tokio_executor::blocking::run(move || {
+///     // do some compute-heavy work or call synchronous code
+/// }).await;
+/// # }
+/// ```
+pub fn run<F, R>(f: F) -> Blocking<R>
+where
+    F: FnOnce() -> R + Send + 'static,
+    R: Send + 'static,
+{
+    let (tx, rx) = oneshot::channel();
+
+    BLOCKING.with(|current_pool| match current_pool.get() {
+        State::Ready(pool) => {
+            let pool = unsafe { &*pool };
+            Pool::spawn(
+                pool,
+                Box::new(move || {
+                    // receiver may have gone away
+                    let _ = tx.send(f());
+                }),
+            );
+        }
+        State::Empty => panic!("must be called from the context of Tokio runtime"),
+    });
+
+    Blocking { rx }
+}
+
+impl<T> Future for Blocking<T> {
+    type Output = T;
+
+    fn poll(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
+        use std::task::Poll::*;
+
+        match Pin::new(&mut self.rx).poll(cx) {
+            Ready(Ok(v)) => Ready(v),
+            Ready(Err(_)) => panic!(
+                "the blocking operation has been dropped before completing. \
+                 This should not happen and is a bug."
+            ),
+            Pending => Pending,
+        }
+    }
+}
+
+fn run_task(f: Box<dyn FnOnce() + Send>) {
+    use std::panic::{catch_unwind, AssertUnwindSafe};
+
+    let _ = catch_unwind(AssertUnwindSafe(|| f()));
+}
+
+impl Default for Pool {
+    fn default() -> Self {
+        Pool {
+            shared: Mutex::new(Shared {
+                queue: VecDeque::new(),
+                num_th: 0,
+                num_idle: 0,
+                num_notify: 0,
+                shutdown: false,
+            }),
+            condvar: Condvar::new(),
+            new_thread: Box::new(|| {
+                thread::Builder::new().name("tokio-blocking-driver".to_string())
+            }),
+        }
+    }
+}
diff --git a/tokio-executor/src/current_thread/mod.rs b/tokio-executor/src/current_thread/mod.rs
index 5daa634f7..f302b0192 100644
--- a/tokio-executor/src/current_thread/mod.rs
+++ b/tokio-executor/src/current_thread/mod.rs
@@ -17,7 +17,9 @@
 
 mod scheduler;
 
-use self::scheduler::Scheduler;
+use self::scheduler::{Scheduler, TickArgs};
+#[cfg(feature = "blocking")]
+use crate::blocking::{Pool, PoolWaiter};
 use crate::park::{Park, ParkThread, Unpark};
 use crate::{EnterError, Executor, SpawnError, TypedExecutor};
 
@@ -52,6 +54,10 @@ pub struct CurrentThread<P: Park = ParkThread> {
     /// Receiver for futures spawned from other threads
     spawn_receiver: crossbeam_channel::Receiver<Pin<Box<dyn Future<Output = ()> + Send + 'static>>>,
 
+    /// Handle to pool for handling blocking tasks
+    #[cfg(feature = "blocking")]
+    blocking: PoolWaiter,
+
     /// The thread-local ID assigned to this executor.
     id: u64,
 }
@@ -150,9 +156,16 @@ impl<T: fmt::Debug> Error for BlockError<T> {}
 
 /// This is mostly split out to make the borrow checker happy.
 struct Borrow<'a, U> {
+    spawner: BorrowSpawner<'a, U>,
+    #[cfg(feature = "blocking")]
+    blocking: &'a PoolWaiter,
+}
+
+/// As is this.
+struct BorrowSpawner<'a, U> {
     id: u64,
-    scheduler: &'a mut Scheduler<U>,
     num_futures: &'a atomic::AtomicUsize,
+    scheduler: &'a mut Scheduler<U>,
 }
 
 trait SpawnLocal {
@@ -269,6 +282,9 @@ impl<P: Park> CurrentThread<P> {
                 id,
             },
             spawn_receiver,
+
+            #[cfg(feature = "blocking")]
+            blocking: PoolWaiter::from(Pool::default()),
         }
     }
 
@@ -289,7 +305,7 @@ impl<P: Park> CurrentThread<P> {
     where
         F: Future<Output = ()> + 'static,
     {
-        self.borrow().spawn_local(Box::pin(future), false);
+        self.borrow().spawner.spawn_local(Box::pin(future), false);
         self
     }
 
@@ -353,9 +369,13 @@ impl<P: Park> CurrentThread<P> {
 
     fn borrow(&mut self) -> Borrow<'_, P::Unpark> {
         Borrow {
-            id: self.id,
-            scheduler: &mut self.scheduler,
-            num_futures: &*self.num_futures,
+            spawner: BorrowSpawner {
+                id: self.id,
+                scheduler: &mut self.scheduler,
+                num_futures: &*self.num_futures,
+            },
+            #[cfg(feature = "blocking")]
+            blocking: &self.blocking,
         }
     }
 
@@ -383,6 +403,8 @@ impl<P: Park> Drop for CurrentThread<P> {
         // which sets LSB (as above) do make Handle::spawn stop working, and then runs until
         // num_futures.load() == 1.
         let _ = pending;
+
+        // We will wait for any blocking ops by virtue of dropping `blocking`.
     }
 }
 
@@ -391,7 +413,7 @@ impl Executor for CurrentThread {
         &mut self,
         future: Pin<Box<dyn Future<Output = ()> + Send>>,
     ) -> Result<(), SpawnError> {
-        self.borrow().spawn_local(future, false);
+        self.borrow().spawner.spawn_local(future, false);
         Ok(())
     }
 }
@@ -401,7 +423,7 @@ where
     T: Future<Output = ()> + 'static,
 {
     fn spawn(&mut self, future: T) -> Result<(), SpawnError> {
-        self.borrow().spawn_local(Box::pin(future), false);
+        self.borrow().spawner.spawn_local(Box::pin(future), false);
         Ok(())
     }
 }
@@ -434,7 +456,10 @@ impl<P: Park> Entered<'_, P> {
     where
         F: Future<Output = ()> + 'static,
     {
-        self.executor.borrow().spawn_local(Box::pin(future), false);
+        self.executor
+            .borrow()
+            .spawner
+            .spawn_local(Box::pin(future), false);
         self
     }
 
@@ -574,19 +599,28 @@ impl<P: Park> Entered<'_, P> {
         // FIXME: Slightly ugly but needed to make the borrow checker happy
         let (mut borrow, spawn_receiver) = (
             Borrow {
-                id: self.executor.id,
-                scheduler: &mut self.executor.scheduler,
-                num_futures: &*self.executor.num_futures,
+                spawner: BorrowSpawner {
+                    id: self.executor.id,
+                    scheduler: &mut self.executor.scheduler,
+                    num_futures: &*self.executor.num_futures,
+                },
+                #[cfg(feature = "blocking")]
+                blocking: &self.executor.blocking,
             },
             &mut self.executor.spawn_receiver,
         );
 
         while let Ok(future) = spawn_receiver.try_recv() {
-            borrow.spawn_local(future, true);
+            borrow.spawner.spawn_local(future, true);
         }
 
         // After any pending futures were scheduled, do the actual tick
-        borrow.scheduler.tick(borrow.id, borrow.num_futures)
+        borrow.spawner.scheduler.tick(TickArgs {
+            id: borrow.spawner.id,
+            num_futures: borrow.spawner.num_futures,
+            #[cfg(feature = "blocking")]
+            blocking: borrow.blocking,
+        })
     }
 }
 
@@ -741,13 +775,27 @@ impl<U: Unpark> Borrow<'_, U> {
         F: FnOnce() -> R,
     {
         CURRENT.with(|current| {
-            current.id.set(Some(self.id));
-            current.set_spawn(self, || f())
+            current.id.set(Some(self.spawner.id));
+
+            let Borrow {
+                ref mut spawner,
+                #[cfg(all(feature = "blocking", not(loom)))]
+                ref blocking,
+                ..
+            } = self;
+
+            current.set_spawn(spawner, || {
+                #[cfg(all(feature = "blocking", not(loom)))]
+                let res = crate::blocking::with_pool(blocking, || f());
+                #[cfg(any(not(feature = "blocking"), loom))]
+                let res = f();
+                res
+            })
         })
     }
 }
 
-impl<U: Unpark> SpawnLocal for Borrow<'_, U> {
+impl<U: Unpark> SpawnLocal for BorrowSpawner<'_, U> {
     fn spawn_local(&mut self, future: Pin<Box<dyn Future<Output = ()>>>, already_counted: bool) {
         if !already_counted {
             // NOTE: we have a borrow of the Runtime, so we know that it isn't shut down.
diff --git a/tokio-executor/src/current_thread/scheduler.rs b/tokio-executor/src/current_thread/scheduler.rs
index c688b7d37..0793a0670 100644
--- a/tokio-executor/src/current_thread/scheduler.rs
+++ b/tokio-executor/src/current_thread/scheduler.rs
@@ -1,4 +1,4 @@
-use super::Borrow;
+use super::{Borrow, BorrowSpawner};
 use crate::park::Unpark;
 
 use std::cell::UnsafeCell;
@@ -124,6 +124,13 @@ pub(crate) struct Scheduled<'a, U> {
     done: &'a mut bool,
 }
 
+pub(super) struct TickArgs<'a> {
+    pub(super) id: u64,
+    pub(super) num_futures: &'a AtomicUsize,
+    #[cfg(feature = "blocking")]
+    pub(super) blocking: &'a crate::blocking::PoolWaiter,
+}
+
 impl<U> Scheduler<U>
 where
     U: Unpark,
@@ -199,7 +206,7 @@ where
     ///
     /// This function should be called whenever the caller is notified via a
     /// wakeup.
-    pub(crate) fn tick(&mut self, eid: u64, num_futures: &AtomicUsize) -> bool {
+    pub(super) fn tick(&mut self, args: TickArgs<'_>) -> bool {
         let mut ret = false;
         let tick = self.inner.tick_num.fetch_add(1, SeqCst).wrapping_add(1);
 
@@ -265,9 +272,13 @@ where
                 let node = self.nodes.remove(node);
 
                 let mut borrow = Borrow {
-                    id: eid,
-                    scheduler: self,
-                    num_futures,
+                    spawner: BorrowSpawner {
+                        id: args.id,
+                        scheduler: self,
+                        num_futures: args.num_futures,
+                    },
+                    #[cfg(feature = "blocking")]
+                    blocking: args.blocking,
                 };
 
                 let mut bomb = Bomb {
@@ -315,7 +326,7 @@ where
 
                     if borrow.enter(|| scheduled.tick()) {
                         // we have a borrow of the Runtime, so we know it's not shut down
-                        borrow.num_futures.fetch_sub(2, SeqCst);
+                        borrow.spawner.num_futures.fetch_sub(2, SeqCst);
                     }
                 }
 
@@ -323,7 +334,7 @@ where
                     // The future is not done, push it back into the "all
                     // node" list.
                     let node = bomb.node.take().unwrap();
-                    bomb.borrow.scheduler.nodes.push_back(node);
+                    bomb.borrow.spawner.scheduler.nodes.push_back(node);
                 }
             }
         }
diff --git a/tokio-executor/src/global.rs b/tokio-executor/src/global.rs
index 2dfe35ac9..11ede54bb 100644
--- a/tokio-executor/src/global.rs
+++ b/tokio-executor/src/global.rs
@@ -220,24 +220,3 @@ unsafe fn hide_lt<'a>(p: *mut (dyn Executor + 'a)) -> *mut (dyn Executor + 'stat
     #[allow(clippy::transmute_ptr_to_ptr)]
     mem::transmute(p)
 }
-
-#[cfg(test)]
-mod tests {
-    use super::{with_default, DefaultExecutor};
-
-    #[test]
-    fn default_executor_is_send_and_sync() {
-        fn assert_send_sync<T: Send + Sync>() {}
-
-        assert_send_sync::<DefaultExecutor>();
-    }
-
-    #[test]
-    #[should_panic]
-    fn nested_default_executor_status() {
-        let _enter = super::super::enter().unwrap();
-        let mut executor = DefaultExecutor::current();
-
-        let _result = with_default(&mut executor, || ());
-    }
-}
diff --git a/tokio-executor/src/lib.rs b/tokio-executor/src/lib.rs
index 0b0cd3822..d1f0f52dc 100644
--- a/tokio-executor/src/lib.rs
+++ b/tokio-executor/src/lib.rs
@@ -83,6 +83,8 @@ mod typed;
 #[cfg(feature = "thread-pool")]
 mod util;
 
+#[cfg(all(not(feature = "blocking"), feature = "thread-pool"))]
+mod blocking;
 #[cfg(feature = "blocking")]
 pub mod blocking;
 
diff --git a/tokio-executor/src/loom/mod.rs b/tokio-executor/src/loom/mod.rs
index 033651355..bbfac05cb 100644
--- a/tokio-executor/src/loom/mod.rs
+++ b/tokio-executor/src/loom/mod.rs
@@ -21,5 +21,7 @@ pub(crate) mod std {
 }
 
 pub(crate) use self::std::sync;
+#[cfg(any(feature = "blocking", feature = "thread-pool"))]
+pub(crate) use self::std::thread;
 #[cfg(feature = "thread-pool")]
-pub(crate) use self::std::{alloc, cell, rand, sys, thread};
+pub(crate) use self::std::{alloc, cell, rand, sys};
diff --git a/tokio-executor/src/loom/std/mod.rs b/tokio-executor/src/loom/std/mod.rs
index 850399fe6..c051bdacf 100644
--- a/tokio-executor/src/loom/std/mod.rs
+++ b/tokio-executor/src/loom/std/mod.rs
@@ -73,5 +73,5 @@ pub(crate) mod sys {
     }
 }
 
-#[cfg(feature = "thread-pool")]
+#[cfg(any(feature = "blocking", feature = "thread-pool"))]
 pub(crate) use std::thread;
diff --git a/tokio-executor/src/thread_pool/builder.rs b/tokio-executor/src/thread_pool/builder.rs
index 921b52d7c..394bda313 100644
--- a/tokio-executor/src/thread_pool/builder.rs
+++ b/tokio-executor/src/thread_pool/builder.rs
@@ -3,26 +3,28 @@ use crate::loom::sys::num_cpus;
 use crate::loom::thread;
 use crate::park::Park;
 use crate::thread_pool::park::DefaultPark;
-use crate::thread_pool::{shutdown, worker, Spawner, ThreadPool};
+use crate::thread_pool::{shutdown, worker, worker::Worker, Spawner, ThreadPool};
 
 use std::{fmt, usize};
 
 /// Builds a thread pool with custom configuration values.
 pub struct Builder {
-    /// Number of threads to spawn
+    /// Number of worker threads to spawn
     pool_size: usize,
 
-    /// Thread name prefix
-    name_prefix: String,
+    /// Thread name
+    name: String,
 
     /// Thread stack size
     stack_size: Option<usize>,
 
     /// Around worker callback
-    around_worker: Option<Arc<Callback>>,
+    around_worker: Option<Callback>,
 }
 
-type Callback = Box<dyn Fn(usize, &mut dyn FnMut()) + Send + Sync>;
+// The Arc<Box<_>> is needed because loom doesn't support Arc<T> where T: !Sized
+// loom doesn't support that because it requires CoerceUnsized, which is unstable
+type Callback = Arc<Box<dyn Fn(usize, &mut dyn FnMut()) + Send + Sync>>;
 
 impl Builder {
     /// Returns a new thread pool builder initialized with default configuration
@@ -30,7 +32,7 @@ impl Builder {
     pub fn new() -> Builder {
         Builder {
             pool_size: num_cpus(),
-            name_prefix: "tokio-runtime-worker-".to_string(),
+            name: "tokio-runtime-worker".to_string(),
             stack_size: None,
             around_worker: None,
         }
@@ -57,11 +59,7 @@ impl Builder {
         self
     }
 
-    /// Set name prefix of threads spawned by the scheduler
-    ///
-    /// Thread name prefix is used for generating thread names. For example, if
-    /// prefix is `my-pool-`, then threads in the pool will get names like
-    /// `my-pool-1` etc.
+    /// Set name of threads spawned by the scheduler
     ///
     /// If this configuration is not set, then the thread will use the system
     /// default naming scheme.
@@ -72,11 +70,11 @@ impl Builder {
     /// use tokio_executor::thread_pool::Builder;
     ///
     /// let thread_pool = Builder::new()
-    ///     .name_prefix("my-pool-")
+    ///     .name("my-pool")
     ///     .build();
     /// ```
-    pub fn name_prefix<S: Into<String>>(&mut self, val: S) -> &mut Self {
-        self.name_prefix = val.into();
+    pub fn name<S: Into<String>>(&mut self, val: S) -> &mut Self {
+        self.name = val.into();
         self
     }
 
@@ -156,20 +154,11 @@ impl Builder {
     {
         let (shutdown_tx, shutdown_rx) = shutdown::channel();
 
-        let (pool, workers) = worker::create_set(self.pool_size, |i| BoxedPark::new(build_park(i)));
-
-        // Spawn threads for each worker
-        for (idx, mut worker) in workers.into_iter().enumerate() {
-            let around_worker = self.around_worker.clone();
+        let around_worker = self.around_worker.as_ref().map(Arc::clone);
+        let launch_worker = move |worker: Worker<BoxedPark<P>>| {
             let shutdown_tx = shutdown_tx.clone();
-
-            let mut th = thread::Builder::new().name(format!("{}{}", self.name_prefix, idx));
-
-            if let Some(stack) = self.stack_size {
-                th = th.stack_size(stack);
-            }
-
-            let res = th.spawn(move || {
+            let around_worker = around_worker.as_ref().map(Arc::clone);
+            Box::new(move || {
                 struct AbortOnPanic;
 
                 impl Drop for AbortOnPanic {
@@ -182,27 +171,44 @@ impl Builder {
                 }
 
                 let _abort_on_panic = AbortOnPanic;
-
-                if let Some(cb) = around_worker {
-                    cb(idx, &mut || worker.run());
+                if let Some(cb) = around_worker.as_ref() {
+                    let idx = worker.id();
+                    let mut f = Some(move || worker.run());
+                    cb(idx, &mut || {
+                        (f.take()
+                            .expect("around_thread callback called closure twice"))(
+                        )
+                    })
                 } else {
-                    worker.run();
+                    worker.run()
                 }
 
-                // Worker must be dropped before the `shutdown_tx`
-                drop(worker);
-
                 // Dropping the handle must happen __after__ the callback
                 drop(shutdown_tx);
-            });
+            }) as Box<dyn FnOnce() + Send + 'static>
+        };
 
-            if let Err(err) = res {
-                panic!("failed to spawn worker thread: {:?}", err);
-            }
+        let mut blocking = crate::blocking::Builder::default();
+        blocking.name(self.name.clone());
+        if let Some(ss) = self.stack_size {
+            blocking.stack_size(ss);
+        }
+        let blocking = Arc::new(blocking.build());
+
+        let (pool, workers) = worker::create_set::<_, BoxedPark<P>>(
+            self.pool_size,
+            |i| BoxedPark::new(build_park(i)),
+            blocking.clone(),
+        );
+
+        // Spawn threads for each worker
+        for worker in workers {
+            crate::blocking::Pool::spawn(&blocking, launch_worker(worker))
         }
 
         let spawner = Spawner::new(pool);
-        ThreadPool::from_parts(spawner, shutdown_rx)
+        let blocking = crate::blocking::PoolWaiter::from(blocking);
+        ThreadPool::from_parts(spawner, shutdown_rx, blocking)
     }
 }
 
@@ -216,7 +222,7 @@ impl fmt::Debug for Builder {
     fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
         fmt.debug_struct("Builder")
             .field("pool_size", &self.pool_size)
-            .field("name_prefix", &self.name_prefix)
+            .field("name", &self.name)
             .field("stack_size", &self.stack_size)
             .finish()
     }
diff --git a/tokio-executor/src/thread_pool/pool.rs b/tokio-executor/src/thread_pool/pool.rs
index 9df368a88..6b1e302dd 100644
--- a/tokio-executor/src/thread_pool/pool.rs
+++ b/tokio-executor/src/thread_pool/pool.rs
@@ -1,3 +1,4 @@
+use crate::blocking::PoolWaiter;
 use crate::thread_pool::{shutdown, Builder, JoinHandle, Spawner};
 use crate::Executor;
 
@@ -10,6 +11,9 @@ pub struct ThreadPool {
 
     /// Shutdown waiter
     shutdown_rx: shutdown::Receiver,
+
+    /// Shutdown valve for Pool
+    blocking: PoolWaiter,
 }
 
 impl ThreadPool {
@@ -18,10 +22,15 @@ impl ThreadPool {
         Builder::new().build()
     }
 
-    pub(super) fn from_parts(spawner: Spawner, shutdown_rx: shutdown::Receiver) -> ThreadPool {
+    pub(super) fn from_parts(
+        spawner: Spawner,
+        shutdown_rx: shutdown::Receiver,
+        blocking: PoolWaiter,
+    ) -> ThreadPool {
         ThreadPool {
             spawner,
             shutdown_rx,
+            blocking,
         }
     }
 
@@ -60,7 +69,7 @@ impl ThreadPool {
     {
         crate::global::with_threadpool(self, || {
             let mut enter = crate::enter().expect("attempting to block while on a Tokio executor");
-            enter.block_on(future)
+            crate::blocking::with_pool(self.spawner.blocking_pool(), || enter.block_on(future))
         })
     }
 
@@ -69,6 +78,7 @@ impl ThreadPool {
         if self.spawner.workers().close() {
             self.shutdown_rx.wait();
         }
+        self.blocking.shutdown();
     }
 }
 
diff --git a/tokio-executor/src/thread_pool/set.rs b/tokio-executor/src/thread_pool/set.rs
index 3002223cd..2ff92d75f 100644
--- a/tokio-executor/src/thread_pool/set.rs
+++ b/tokio-executor/src/thread_pool/set.rs
@@ -3,6 +3,7 @@
 //! - Attempt to spin.
 
 use crate::loom::rand::seed;
+use crate::loom::sync::Arc;
 use crate::park::Unpark;
 use crate::task::{self, Task};
 use crate::thread_pool::{current, queue, BoxFuture, Idle, JoinHandle, Owned, Shared};
@@ -27,6 +28,9 @@ where
 
     /// Coordinates idle workers
     idle: Idle,
+
+    /// Pool where blocking tasks should be spawned.
+    pub(crate) blocking: Arc<crate::blocking::Pool>,
 }
 
 unsafe impl<P: Unpark> Send for Set<P> {}
@@ -37,7 +41,11 @@ where
     P: Unpark,
 {
     /// Create a new worker set using the provided queues.
-    pub(crate) fn new<F>(num_workers: usize, mut mk_unpark: F) -> Self
+    pub(crate) fn new<F>(
+        num_workers: usize,
+        mut mk_unpark: F,
+        blocking: Arc<crate::blocking::Pool>,
+    ) -> Self
     where
         F: FnMut(usize) -> P,
     {
@@ -62,6 +70,7 @@ where
             owned: owned.into_boxed_slice(),
             inject,
             idle: Idle::new(num_workers),
+            blocking,
         }
     }
 
@@ -104,6 +113,10 @@ where
         self.schedule(task);
     }
 
+    pub(super) fn blocking_pool(&self) -> &Arc<crate::blocking::Pool> {
+        &self.blocking
+    }
+
     pub(crate) fn schedule(&self, task: Task<Shared<P>>) {
         current::get(|current_worker| match current_worker.as_member(self) {
             Some(worker) => {
diff --git a/tokio-executor/src/thread_pool/spawner.rs b/tokio-executor/src/thread_pool/spawner.rs
index d723514f0..79eff2739 100644
--- a/tokio-executor/src/thread_pool/spawner.rs
+++ b/tokio-executor/src/thread_pool/spawner.rs
@@ -44,6 +44,10 @@ impl Spawner {
         self.workers.spawn_background(future);
     }
 
+    pub(super) fn blocking_pool(&self) -> &Arc<crate::blocking::Pool> {
+        self.workers.blocking_pool()
+    }
+
     /// Reference to the worker set. Used by `ThreadPool` to initiate shutdown.
     pub(super) fn workers(&self) -> &worker::Set<Box<dyn Unpark>> {
         &*self.workers
diff --git a/tokio-executor/src/thread_pool/tests/worker.rs b/tokio-executor/src/thread_pool/tests/worker.rs
index 6ff0c7788..c41633506 100644
--- a/tokio-executor/src/thread_pool/tests/worker.rs
+++ b/tokio-executor/src/thread_pool/tests/worker.rs
@@ -10,7 +10,9 @@ macro_rules! pool {
     }};
     (! $n:expr) => {{
         let mut mock_park = crate::tests::mock_park::MockPark::new();
-        let (pool, workers) = thread_pool::create_pool($n, |index| mock_park.mk_park(index));
+        let blocking = std::sync::Arc::new(crate::blocking::Pool::default());
+        let (pool, workers) =
+            thread_pool::create_pool($n, |index| mock_park.mk_park(index), blocking);
         (pool, workers, mock_park)
     }};
 }
diff --git a/tokio-executor/src/thread_pool/worker.rs b/tokio-executor/src/thread_pool/worker.rs
index bdf430d37..97571beb2 100644
--- a/tokio-executor/src/thread_pool/worker.rs
+++ b/tokio-executor/src/thread_pool/worker.rs
@@ -16,23 +16,19 @@ pub(crate) struct Worker<P: Park + 'static> {
     park: P,
 }
 
-struct Entry<P: 'static> {
-    pool: Arc<Set<P>>,
-    index: usize,
-}
-
 pub(crate) fn create_set<F, P>(
     pool_size: usize,
     mk_park: F,
+    blocking: Arc<crate::blocking::Pool>,
 ) -> (Arc<Set<P::Unpark>>, Vec<Worker<P>>)
 where
-    P: Park,
+    P: Send + Park,
     F: FnMut(usize) -> P,
 {
     // Create the parks...
     let parks: Vec<_> = (0..pool_size).map(mk_park).collect();
 
-    let mut pool = Arc::new(Set::new(pool_size, |i| parks[i].unpark()));
+    let mut pool = Arc::new(Set::new(pool_size, |i| parks[i].unpark(), blocking));
 
     // Establish the circular link between the individual worker state
     // structure and the container.
@@ -42,7 +38,10 @@ where
     let workers = parks
         .into_iter()
         .enumerate()
-        .map(|(index, park)| Worker::new(pool.clone(), index, park))
+        .map(|(index, park)| {
+            // unsafe is safe because we call Worker::new only once with each index in the pool
+            unsafe { Worker::new(pool.clone(), index, park) }
+        })
         .collect();
 
     (pool, workers)
@@ -56,28 +55,39 @@ const GLOBAL_POLL_INTERVAL: u16 = 61;
 
 impl<P> Worker<P>
 where
-    P: Park + 'static,
+    P: Send + Park,
 {
-    pub(super) fn new(pool: Arc<Set<P::Unpark>>, index: usize, park: P) -> Self {
+    // unsafe because new may only be called once for each index in pool's set
+    pub(super) unsafe fn new(pool: Arc<Set<P::Unpark>>, index: usize, park: P) -> Self {
         Worker {
-            entry: Entry { pool, index },
+            entry: Entry::new(pool, index),
             park,
         }
     }
 
-    pub(super) fn run(&mut self) {
-        let mut executor = &*self.entry.pool;
-        let entry = &self.entry;
+    pub(super) fn run(mut self) {
+        let pool = Arc::clone(&self.entry.pool);
+        let pool = &pool;
+        let index = self.entry.index;
+
+        let mut executor = &**pool;
+        let entry = &mut self.entry;
         let park = &mut self.park;
 
+        let blocking = &executor.blocking;
+
         // Track the current worker
-        current::set(&entry.pool, entry.index, || {
+        current::set(&pool, index, || {
             let _enter = crate::enter().expect("executor already running on thread");
 
             crate::with_default(&mut executor, || {
-                entry.run(park);
+                crate::blocking::with_pool(blocking, || entry.run(park))
             })
-        })
+        });
+    }
+
+    pub(super) fn id(&self) -> usize {
+        self.entry.index
     }
 
     #[cfg(test)]
@@ -96,11 +106,21 @@ where
     }
 }
 
+struct Entry<P: 'static> {
+    pool: Arc<Set<P>>,
+    index: usize,
+}
+
 impl<P> Entry<P>
 where
     P: Unpark,
 {
-    fn run(&self, park: &mut impl Park<Unpark = P>) {
+    // unsafe because Entry::owned assumes there is only one instance of the Entry
+    unsafe fn new(pool: Arc<Set<P>>, index: usize) -> Self {
+        Entry { pool, index }
+    }
+
+    fn run(&mut self, park: &mut impl Park<Unpark = P>) {
         while self.is_running() {
             if self.tick(park) {
                 self.park(park);
@@ -110,12 +130,12 @@ where
         self.shutdown(park);
     }
 
-    fn is_running(&self) -> bool {
+    fn is_running(&mut self) -> bool {
         self.owned().is_running.get()
     }
 
     /// Returns `true` if the worker needs to park
-    fn tick(&self, park: &mut impl Park<Unpark = P>) -> bool {
+    fn tick(&mut self, park: &mut impl Park<Unpark = P>) -> bool {
         // Process all pending tasks in the local queue.
         if !self.process_local_queue(park) {
             return false;
@@ -140,7 +160,7 @@ where
     ///
     /// Returns `false` if processing was interrupted due to the pool shutting
     /// down.
-    fn process_local_queue(&self, park: &mut impl Park<Unpark = P>) -> bool {
+    fn process_local_queue(&mut self, park: &mut impl Park<Unpark = P>) -> bool {
         debug_assert!(self.is_running());
 
         loop {
@@ -171,7 +191,7 @@ where
         }
     }
 
-    fn steal_work(&self) -> Option<Task<Shared<P>>> {
+    fn steal_work(&mut self) -> Option<Task<Shared<P>>> {
         let num_workers = self.pool.len();
         let start = self.owned().rand.fastrand_n(num_workers as u32);
 
@@ -185,17 +205,16 @@ where
 
     /// Runs maintenance work such as free pending tasks and check the pool's
     /// state.
-    fn maintenance(&self) {
+    fn maintenance(&mut self) {
         // Free any completed tasks
         self.drain_tasks_pending_drop();
 
         // Update the pool state cache
-        self.owned()
-            .is_running
-            .set(!self.owned().work_queue.is_closed());
+        let closed = self.owned().work_queue.is_closed();
+        self.owned().is_running.set(!closed)
     }
 
-    fn search_for_work(&self) -> bool {
+    fn search_for_work(&mut self) -> bool {
         debug_assert!(self.is_searching());
 
         if let Some(task) = self.steal_work() {
@@ -208,7 +227,7 @@ where
         }
     }
 
-    fn transition_to_searching(&self) -> bool {
+    fn transition_to_searching(&mut self) -> bool {
         if self.is_searching() {
             return true;
         }
@@ -218,7 +237,7 @@ where
         ret
     }
 
-    fn transition_from_searching(&self) {
+    fn transition_from_searching(&mut self) {
         debug_assert!(self.is_searching());
 
         self.owned().is_searching.set(false);
@@ -231,11 +250,13 @@ where
     }
 
     /// Returns `true` if the worker must check for any work.
-    fn transition_to_parked(&self) -> bool {
+    fn transition_to_parked(&mut self) -> bool {
+        let idx = self.index;
+        let is_searching = self.is_searching();
         let ret = self
             .set()
             .idle()
-            .transition_worker_to_parked(self.index, self.is_searching());
+            .transition_worker_to_parked(idx, is_searching);
 
         // The worker is no longer searching. Setting this is the local cache
         // only.
@@ -249,7 +270,7 @@ where
     }
 
     /// Returns `true` if the transition happened.
-    fn transition_from_parked(&self) -> bool {
+    fn transition_from_parked(&mut self) -> bool {
         if self.owned().did_submit_task.get() || !self.is_running() {
             // Remove the worker from the sleep set.
             self.set().idle().unpark_worker_by_id(self.index);
@@ -270,7 +291,7 @@ where
         }
     }
 
-    fn run_task(&self, task: Task<Shared<P>>) {
+    fn run_task(&mut self, task: Task<Shared<P>>) {
         if self.is_searching() {
             self.transition_from_searching();
         }
@@ -281,13 +302,13 @@ where
         }
     }
 
-    fn final_work_sweep(&self) {
+    fn final_work_sweep(&mut self) {
         if !self.owned().work_queue.is_empty() {
             self.set().notify_work();
         }
     }
 
-    fn park(&self, park: &mut impl Park<Unpark = P>) {
+    fn park(&mut self, park: &mut impl Park<Unpark = P>) {
         if self.transition_to_parked() {
             // We are the final searching worker, check if any work arrived
             // before parking
@@ -309,7 +330,7 @@ where
         }
     }
 
-    fn park_light(&self, park: &mut impl Park<Unpark = P>) {
+    fn park_light(&mut self, park: &mut impl Park<Unpark = P>) {
         // When tasks are submitted locally (from the parker), defer any
         // notifications in hopes that the curent worker will grab those tasks.
         self.owned().defer_notification.set(true);
@@ -326,7 +347,7 @@ where
         }
     }
 
-    fn drain_tasks_pending_drop(&self) {
+    fn drain_tasks_pending_drop(&mut self) {
         for task in self.shared().pending_drop.drain() {
             unsafe {
                 let owned = &mut *self.set().owned()[self.index].get();
@@ -340,7 +361,7 @@ where
     ///
     /// Once the shutdown flag has been observed, it is guaranteed that no
     /// further tasks may be pushed into the global queue.
-    fn shutdown(&self, park: &mut impl Park<Unpark = P>) {
+    fn shutdown(&mut self, park: &mut impl Park<Unpark = P>) {
         // Transition all tasks owned by the worker to canceled.
         self.owned().owned_tasks.shutdown();
 
@@ -369,13 +390,13 @@ where
     }
 
     /// Increment the tick, returning the value from before the increment.
-    fn tick_fetch_inc(&self) -> u16 {
+    fn tick_fetch_inc(&mut self) -> u16 {
         let tick = self.owned().tick.get();
         self.owned().tick.set(tick.wrapping_add(1));
         tick
     }
 
-    fn is_searching(&self) -> bool {
+    fn is_searching(&mut self) -> bool {
         self.owned().is_searching.get()
     }
 
@@ -387,7 +408,7 @@ where
         &self.set().shared()[self.index]
     }
 
-    fn owned(&self) -> &Owned<P> {
+    fn owned(&mut self) -> &Owned<P> {
         // safety: we own the slot
         unsafe { &*self.set().owned()[self.index].get() }
     }
diff --git a/tokio-executor/tests/global.rs b/tokio-executor/tests/global.rs
new file mode 100644
index 000000000..5aef0597d
--- /dev/null
+++ b/tokio-executor/tests/global.rs
@@ -0,0 +1,17 @@
+use tokio_executor::{with_default, DefaultExecutor};
+
+#[test]
+fn default_executor_is_send_and_sync() {
+    fn assert_send_sync<T: Send + Sync>() {}
+
+    assert_send_sync::<DefaultExecutor>();
+}
+
+#[test]
+#[should_panic]
+fn nested_default_executor_status() {
+    let _enter = tokio_executor::enter().unwrap();
+    let mut executor = DefaultExecutor::current();
+
+    let _result = with_default(&mut executor, || ());
+}
diff --git a/tokio/src/runtime/threadpool/builder.rs b/tokio/src/runtime/threadpool/builder.rs
index df0ff337c..fd6bbbb7f 100644
--- a/tokio/src/runtime/threadpool/builder.rs
+++ b/tokio/src/runtime/threadpool/builder.rs
@@ -5,8 +5,8 @@ use crate::timer::timer::{self, Timer};
 use tokio_executor::thread_pool;
 use tokio_net::driver::{self, Reactor};
 
-use std::{fmt, io};
 use std::sync::{Arc, Mutex};
+use std::{fmt, io};
 
 /// Builds Tokio Runtime with custom configuration values.
 ///
@@ -32,7 +32,7 @@ use std::sync::{Arc, Mutex};
 ///     let runtime = Builder::new()
 ///         .clock(Clock::system())
 ///         .num_threads(4)
-///         .name_prefix("my-custom-name-")
+///         .name("my-custom-name")
 ///         .stack_size(3 * 1024 * 1024)
 ///         .build()
 ///         .unwrap();
@@ -67,7 +67,7 @@ impl Builder {
 
         let mut thread_pool_builder = thread_pool::Builder::new();
         thread_pool_builder
-            .name_prefix("tokio-runtime-worker-")
+            .name("tokio-runtime-worker")
             .num_threads(num_threads);
 
         Builder {
@@ -110,13 +110,9 @@ impl Builder {
         self
     }
 
-    /// Set name prefix of threads spawned by the `Runtime`'s thread pool.
+    /// Set name of threads spawned by the `Runtime`'s thread pool.
     ///
-    /// Thread name prefix is used for generating thread names. For example, if
-    /// prefix is `my-pool-`, then threads in the pool will get names like
-    /// `my-pool-1` etc.
-    ///
-    /// The default prefix is "tokio-runtime-worker-".
+    /// The default name is "tokio-runtime-worker".
     ///
     /// # Examples
     ///
@@ -125,12 +121,12 @@ impl Builder {
     ///
     /// # pub fn main() {
     /// let rt = runtime::Builder::new()
-    ///     .name_prefix("my-pool-")
+    ///     .name("my-pool")
     ///     .build();
     /// # }
     /// ```
-    pub fn name_prefix<S: Into<String>>(&mut self, val: S) -> &mut Self {
-        self.thread_pool_builder.name_prefix(val);
+    pub fn name<S: Into<String>>(&mut self, val: S) -> &mut Self {
+        self.thread_pool_builder.name(val);
         self
     }
 
@@ -177,7 +173,8 @@ impl Builder {
     /// # }
     /// ```
     pub fn after_start<F>(&mut self, f: F) -> &mut Self
-        where F: Fn() + Send + Sync + 'static
+    where
+        F: Fn() + Send + Sync + 'static,
     {
         self.after_start = Some(Arc::new(f));
         self
@@ -201,7 +198,8 @@ impl Builder {
     /// # }
     /// ```
     pub fn before_stop<F>(&mut self, f: F) -> &mut Self
-        where F: Fn() + Send + Sync + 'static
+    where
+        F: Fn() + Send + Sync + 'static,
     {
         self.before_stop = Some(Arc::new(f));
         self
@@ -263,13 +261,7 @@ impl Builder {
                     }
                 })
             })
-            .build_with_park(move |index| {
-                timers[index]
-                    .lock()
-                    .unwrap()
-                    .take()
-                    .unwrap()
-            });
+            .build_with_park(move |index| timers[index].lock().unwrap().take().unwrap());
 
         Ok(Runtime {
             inner: Some(Inner {