task: add LocalSet API for running !Send futures (#1733)

## Motivation

In earlier versions of `tokio`, the `current_thread::Runtime` type could
be used to run `!Send` futures. However, PR #1716 merged the
current-thread and threadpool runtimes into a single type, which can no
longer run `!Send` futures. There is still a need in some cases to
support futures that don't implement `Send`, and the `tokio-compat`
crate requires this in order to provide APIs that existed in `tokio`
0.1.

## Solution

This branch implements the API described by @carllerche in
https://github.com/tokio-rs/tokio/pull/1716#issuecomment-549496309. It
adds a new `LocalSet` type and `spawn_local` function to `tokio::task`.
The `LocalSet` type is used to group together a set of tasks which must
run on the same thread and don't implement `Send`. These are available
when a new "rt-util" feature flag is enabled.

Currently, the local task set is run by passing it a reference to a
`Runtime` and a future to `block_on`. In the future, we may also want
to investigate allowing spawned futures to construct their own local
task sets, which would be executed on the worker that the future is
executing on. 

In order to implement the new API, I've made some internal changes to
the `task` module and `Schedule` trait to support scheduling both `Send`
and `!Send` futures.

Signed-off-by: Eliza Weisman <[email protected]>

tokio/tests/rt_common.rs

@@ -592,4 +592,67 @@ rt_test! {
         assert_eq!(buf, b"hello");
         tx.send(()).unwrap();
     }
+
+    mod local_set {
+        use tokio::task;
+        use super::*;
+
+        #[test]
+        fn block_on_socket() {
+            let mut rt = rt();
+            let local = task::LocalSet::new();
+
+            local.block_on(&mut rt, async move {
+                let (tx, rx) = oneshot::channel();
+
+                let mut listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+                let addr = listener.local_addr().unwrap();
+
+                task::spawn_local(async move {
+                    let _ = listener.accept().await;
+                    tx.send(()).unwrap();
+                });
+
+                TcpStream::connect(&addr).await.unwrap();
+                rx.await.unwrap();
+            });
+        }
+
+        #[test]
+        fn client_server_block_on() {
+            let mut rt = rt();
+            let (tx, rx) = mpsc::channel();
+
+            let local = task::LocalSet::new();
+
+            local.block_on(&mut rt, async move { client_server_local(tx).await });
+
+            assert_ok!(rx.try_recv());
+            assert_err!(rx.try_recv());
+        }
+
+        async fn client_server_local(tx: mpsc::Sender<()>) {
+            let mut server = assert_ok!(TcpListener::bind("127.0.0.1:0").await);
+
+            // Get the assigned address
+            let addr = assert_ok!(server.local_addr());
+
+            // Spawn the server
+            task::spawn_local(async move {
+                // Accept a socket
+                let (mut socket, _) = server.accept().await.unwrap();
+
+                // Write some data
+                socket.write_all(b"hello").await.unwrap();
+            });
+
+            let mut client = TcpStream::connect(&addr).await.unwrap();
+
+            let mut buf = vec![];
+            client.read_to_end(&mut buf).await.unwrap();
+
+            assert_eq!(buf, b"hello");
+            tx.send(()).unwrap();
+        }
+    }
 }