Switch to stable futures

Author: tomaka

core/Cargo.toml

@@ -21,7 +21,7 @@ log = "0.4"
 multiaddr = { package = "parity-multiaddr", version = "0.5.0", path = "../misc/multiaddr" }
 multihash = { package = "parity-multihash", version = "0.1.0", path = "../misc/multihash" }
 multistream-select = { version = "0.4.0", path = "../misc/multistream-select" }
-futures = "0.1"
+futures-preview = { version = "0.3.0-alpha.17", features = ["compat"] }
 parking_lot = "0.8"
 protobuf = "2.3"
 quick-error = "1.2"
@@ -30,8 +30,6 @@ rw-stream-sink = { version = "0.1.1", path = "../misc/rw-stream-sink" }
 libsecp256k1 = { version = "0.2.2", optional = true }
 sha2 = "0.8.0"
 smallvec = "0.6"
-tokio-executor = "0.1.4"
-tokio-io = "0.1"
 wasm-timer = "0.1"
 unsigned-varint = "0.2"
 void = "1"

core/src/either.rs

@@ -37,7 +37,7 @@
 
 /// Multi-address re-export.
 pub use multiaddr;
-pub use multistream_select::Negotiated;
+pub type Negotiated<T> = futures::compat::Compat01As03<multistream_select::Negotiated<T>>;
 
 mod keys_proto;
 mod peer_id;

core/src/lib.rs

@@ -19,10 +19,9 @@
 // DEALINGS IN THE SOFTWARE.
 
 use crate::{Endpoint, muxing::StreamMuxer};
-use futures::prelude::*;
+use futures::{prelude::*, io::Initializer};
 use parking_lot::Mutex;
-use std::{io, sync::atomic::{AtomicBool, Ordering}};
-use tokio_io::{AsyncRead, AsyncWrite};
+use std::{io, pin::Pin, sync::atomic::{AtomicBool, Ordering}, task::Context, task::Poll};
 
 /// Implementation of `StreamMuxer` that allows only one substream on top of a connection,
 /// yielding the connection itself.
@@ -62,67 +61,67 @@ pub struct OutboundSubstream {}
 
 impl<TSocket> StreamMuxer for SingletonMuxer<TSocket>
 where
-    TSocket: AsyncRead + AsyncWrite,
+    TSocket: AsyncRead + AsyncWrite + Unpin,
 {
     type Substream = Substream;
     type OutboundSubstream = OutboundSubstream;
     type Error = io::Error;
 
-    fn poll_inbound(&self) -> Poll<Self::Substream, io::Error> {
+    fn poll_inbound(&self) -> Poll<Result<Self::Substream, io::Error>> {
         match self.endpoint {
-            Endpoint::Dialer => return Ok(Async::NotReady),
+            Endpoint::Dialer => return Poll::Pending,
             Endpoint::Listener => {}
         }
 
         if !self.substream_extracted.swap(true, Ordering::Relaxed) {
-            Ok(Async::Ready(Substream {}))
+            Poll::Ready(Ok(Substream {}))
         } else {
-            Ok(Async::NotReady)
+            Poll::Pending
         }
     }
 
     fn open_outbound(&self) -> Self::OutboundSubstream {
         OutboundSubstream {}
     }
 
-    fn poll_outbound(&self, _: &mut Self::OutboundSubstream) -> Poll<Self::Substream, io::Error> {
+    fn poll_outbound(&self, _: &mut Self::OutboundSubstream) -> Poll<Result<Self::Substream, io::Error>> {
         match self.endpoint {
-            Endpoint::Listener => return Ok(Async::NotReady),
+            Endpoint::Listener => return Poll::Pending,
             Endpoint::Dialer => {}
         }
 
         if !self.substream_extracted.swap(true, Ordering::Relaxed) {
-            Ok(Async::Ready(Substream {}))
+            Poll::Ready(Ok(Substream {}))
         } else {
-            Ok(Async::NotReady)
+            Poll::Pending
         }
     }
 
     fn destroy_outbound(&self, _: Self::OutboundSubstream) {
     }
 
-    unsafe fn prepare_uninitialized_buffer(&self, buf: &mut [u8]) -> bool {
-        self.inner.lock().prepare_uninitialized_buffer(buf)
+    unsafe fn initializer(&self) -> Initializer {
+        self.inner.lock().initializer()
     }
 
-    fn read_substream(&self, _: &mut Self::Substream, buf: &mut [u8]) -> Poll<usize, io::Error> {
-        let res = self.inner.lock().poll_read(buf);
-        if let Ok(Async::Ready(_)) = res {
+    fn read_substream(&self, cx: &mut Context, _: &mut Self::Substream, buf: &mut [u8]) -> Poll<Result<usize, io::Error>> {
+        let res = AsyncRead::poll_read(Pin::new(&mut *self.inner.lock()), cx, buf);
+        if let Poll::Ready(Ok(_)) = res {
             self.remote_acknowledged.store(true, Ordering::Release);
         }
         res
     }
 
-    fn write_substream(&self, _: &mut Self::Substream, buf: &[u8]) -> Poll<usize, io::Error> {
-        self.inner.lock().poll_write(buf)
+    fn write_substream(&self, cx: &mut Context, _: &mut Self::Substream, buf: &[u8]) -> Poll<Result<usize, io::Error>> {
+        AsyncWrite::poll_write(Pin::new(&mut *self.inner.lock()), cx, buf)
     }
 
-    fn flush_substream(&self, _: &mut Self::Substream) -> Poll<(), io::Error> {
-        self.inner.lock().poll_flush()
+    fn flush_substream(&self, cx: &mut Context, _: &mut Self::Substream) -> Poll<Result<(), io::Error>> {
+        AsyncWrite::poll_flush(Pin::new(&mut *self.inner.lock()), cx)
     }
 
-    fn shutdown_substream(&self, _: &mut Self::Substream) -> Poll<(), io::Error> {
-        self.inner.lock().shutdown()
+    fn shutdown_substream(&self, cx: &mut Context, _: &mut Self::Substream) -> Poll<Result<(), io::Error>> {
+        AsyncWrite::poll_close(Pin::new(&mut *self.inner.lock()), cx)
     }
 
     fn destroy_substream(&self, _: Self::Substream) {
@@ -132,12 +131,12 @@ where
         self.remote_acknowledged.load(Ordering::Acquire)
     }
 
-    fn close(&self) -> Poll<(), io::Error> {
+    fn close(&self, cx: &mut Context) -> Poll<Result<(), io::Error>> {
         // The `StreamMuxer` trait requires that `close()` implies `flush_all()`.
-        self.flush_all()
+        self.flush_all(cx)
     }
 
-    fn flush_all(&self) -> Poll<(), io::Error> {
-        self.inner.lock().poll_flush()
+    fn flush_all(&self, cx: &mut Context) -> Poll<Result<(), io::Error>> {
+        AsyncWrite::poll_flush(Pin::new(&mut *self.inner.lock()), cx)
     }
 }

core/src/muxing.rs

@@ -29,7 +29,7 @@ use crate::{
 };
 use fnv::FnvHashMap;
 use futures::prelude::*;
-use std::{error, fmt, hash::Hash, mem};
+use std::{error, fmt, hash::Hash, mem, pin::Pin, task::Context, task::Poll};
 
 pub use crate::nodes::tasks::StartTakeOver;
 
@@ -323,7 +323,7 @@ where
     pub fn add_reach_attempt<TFut, TMuxer>(&mut self, future: TFut, handler: THandler)
         -> ReachAttemptId
     where
-        TFut: Future<Item = (TConnInfo, TMuxer), Error = TReachErr> + Send + 'static,
+        TFut: Future<Output = Result<(TConnInfo, TMuxer), TReachErr>> + Unpin + Send + 'static,
         THandler: IntoNodeHandler<TConnInfo> + Send + 'static,
         THandler::Handler: NodeHandler<Substream = Substream<TMuxer>, InEvent = TInEvent, OutEvent = TOutEvent, Error = THandlerErr> + Send + 'static,
         <THandler::Handler as NodeHandler>::OutboundOpenInfo: Send + 'static,
@@ -358,17 +358,19 @@ where
     }
 
     /// Sends an event to all nodes.
-    #[must_use]
-    pub fn start_broadcast(&mut self, event: &TInEvent) -> AsyncSink<()>
+    ///
+    /// Must be called only after a successful call to `poll_ready_broadcast`.
+    pub fn start_broadcast(&mut self, event: &TInEvent)
     where
         TInEvent: Clone
     {
         self.inner.start_broadcast(event)
     }
 
+    /// Wait until we have enough room in senders to broadcast an event.
     #[must_use]
-    pub fn complete_broadcast(&mut self) -> Async<()> {
-        self.inner.complete_broadcast()
+    pub fn poll_ready_broadcast(&mut self, cx: &mut Context) -> Poll<()> {
+        self.inner.poll_ready_broadcast(cx)
     }
 
     /// Adds an existing connection to a node to the collection.
@@ -447,13 +449,13 @@ where
     /// > **Note**: we use a regular `poll` method instead of implementing `Stream` in order to
     /// > remove the `Err` variant, but also because we want the `CollectionStream` to stay
     /// > borrowed if necessary.
-    pub fn poll(&mut self) -> Async<CollectionEvent<'_, TInEvent, TOutEvent, THandler, TReachErr, THandlerErr, TUserData, TConnInfo, TPeerId>>
+    pub fn poll(self: Pin<&mut Self>, cx: &mut Context) -> Poll<CollectionEvent<'_, TInEvent, TOutEvent, THandler, TReachErr, THandlerErr, TUserData, TConnInfo, TPeerId>>
     where
         TConnInfo: Clone,   // TODO: Clone shouldn't be necessary
     {
-        let item = match self.inner.poll() {
-            Async::Ready(item) => item,
-            Async::NotReady => return Async::NotReady,
+        let item = match self.inner.poll(cx) {
+            Poll::Ready(item) => item,
+            Poll::Pending => return Poll::Pending,
         };
 
         match item {
@@ -463,7 +465,7 @@ where
 
                 match (user_data, result, handler) {
                     (TaskState::Pending, tasks::Error::Reach(err), Some(handler)) => {
-                        Async::Ready(CollectionEvent::ReachError {
+                        Poll::Ready(CollectionEvent::ReachError {
                             id: ReachAttemptId(id),
                             error: err,
                             handler,
@@ -482,7 +484,7 @@ where
                         debug_assert!(_handler.is_none());
                         let _node_task_id = self.nodes.remove(conn_info.peer_id());
                         debug_assert_eq!(_node_task_id, Some(id));
-                        Async::Ready(CollectionEvent::NodeClosed {
+                        Poll::Ready(CollectionEvent::NodeClosed {
                             conn_info,
                             error: err,
                             user_data,
@@ -497,8 +499,8 @@ where
             tasks::Event::NodeReached { task, conn_info } => {
                 let id = task.id();
                 drop(task);
-                Async::Ready(CollectionEvent::NodeReached(CollectionReachEvent {
-                    parent: self,
+                Poll::Ready(CollectionEvent::NodeReached(CollectionReachEvent {
+                    parent: &mut *self,
                     id,
                     conn_info: Some(conn_info),
                 }))
@@ -512,7 +514,7 @@ where
                                  self.tasks is switched to the Connected state; QED"),
                 };
                 drop(task);
-                Async::Ready(CollectionEvent::NodeEvent {
+                Poll::Ready(CollectionEvent::NodeEvent {
                     // TODO: normally we'd build a `PeerMut` manually here, but the borrow checker
                     //       doesn't like it
                     peer: self.peer_mut(&conn_info.peer_id())
@@ -616,14 +618,15 @@ where
         }
     }
 
-    /// Sends an event to the given node.
-    pub fn start_send_event(&mut self, event: TInEvent) -> StartSend<TInEvent, ()> {
+    /// Begin sending an event to the given node. Must be called only after a successful call to
+    /// `poll_ready_event`.
+    pub fn start_send_event(&mut self, event: TInEvent) {
         self.inner.start_send_event(event)
     }
 
-    /// Complete sending an event message initiated by `start_send_event`.
-    pub fn complete_send_event(&mut self) -> Poll<(), ()> {
-        self.inner.complete_send_event()
+    /// Make sure we are ready to accept an event to be sent with `start_send_event`.
+    pub fn poll_ready_event(&mut self, cx: &mut Context) -> Poll<()> {
+        self.inner.poll_ready_event(cx)
     }
 
     /// Closes the connections to this node. Returns the user data.
@@ -664,7 +667,7 @@ where
     }
 
     /// Complete a take over initiated by `start_take_over`.
-    pub fn complete_take_over(&mut self) -> Poll<(), ()> {
+    pub fn complete_take_over(&mut self) -> Poll<()> {
         self.inner.complete_take_over()
     }
 }

core/src/muxing/singleton.rs

@@ -75,14 +75,14 @@ fn collection_stream_reaches_the_nodes() {
         poll_count += 1;
         let event = cs.poll();
         match poll_count {
-            1 => assert_matches!(event, Async::NotReady),
+            1 => assert_matches!(event, Poll::Pending),
             2 => {
                 assert_matches!(event, Async::Ready(CollectionEvent::NodeReached(_)));
                 return Ok(Async::Ready(())); // stop
             }
             _ => unreachable!()
         }
-        Ok(Async::NotReady)
+        Poll::Pending
     });
     rt.block_on(fut).unwrap();
 }
@@ -102,8 +102,8 @@ fn accepting_a_node_yields_new_entry() {
             let event = cs.poll();
             match poll_count {
                 1 => {
-                    assert_matches!(event, Async::NotReady);
-                    return Ok(Async::NotReady)
+                    assert_matches!(event, Poll::Pending);
+                    return Poll::Pending
                 }
                 2 => {
                     assert_matches!(event, Async::Ready(CollectionEvent::NodeReached(reach_ev)) => {
@@ -150,7 +150,7 @@ fn events_in_a_node_reaches_the_collection_stream() {
     let cs_fut = cs.clone();
     rt.block_on(future::poll_fn(move || -> Poll<_, ()> {
         let mut cs = cs_fut.lock();
-        assert_matches!(cs.poll(), Async::NotReady);
+        assert_matches!(cs.poll(), Poll::Pending);
         Ok(Async::Ready(()))
     })).expect("tokio works");
 
@@ -246,7 +246,7 @@ fn task_closed_with_error_while_task_is_pending_yields_reach_error() {
     let cs_fut = cs.clone();
     rt.block_on(future::poll_fn(move || -> Poll<_, ()> {
         let mut cs = cs_fut.lock();
-        assert_matches!(cs.poll(), Async::NotReady);
+        assert_matches!(cs.poll(), Poll::Pending);
         Ok(Async::Ready(()))
     })).expect("tokio works");
 
@@ -289,7 +289,7 @@ fn task_closed_with_error_when_task_is_connected_yields_node_error() {
     let cs_fut = cs.clone();
     rt.block_on(future::poll_fn(move || -> Poll<_, ()> {
         let mut cs = cs_fut.lock();
-        assert_matches!(cs.poll(), Async::NotReady);
+        assert_matches!(cs.poll(), Poll::Pending);
         // send an event so the Handler errors in two polls
         Ok(cs.complete_broadcast())
     })).expect("tokio works");
@@ -351,7 +351,7 @@ fn interrupting_an_established_connection_is_err() {
     let cs_fut = cs.clone();
     rt.block_on(future::poll_fn(move || -> Poll<_, ()> {
         let mut cs = cs_fut.lock();
-        assert_matches!(cs.poll(), Async::NotReady);
+        assert_matches!(cs.poll(), Poll::Pending);
         // send an event so the Handler errors in two polls
         Ok(Async::Ready(()))
     })).expect("tokio works");