Cleanup non-Stable stuff from futures::__rt

Author: Nemo157

futures-async-runtime/src/future.rs

@@ -1,45 +1,37 @@
+use std::marker::Unpin;
+use std::mem::Pin;
 use std::ops::{Generator, GeneratorState};
 
 use super::{IsResult, Reset, CTX};
 
-use futures_core::Never;
-use futures_core::task;
-use futures_core::{Poll, Async, Future};
+use futures_core::{Async, Poll, Never, task};
+use futures_stable::StableFuture;
 
-pub trait MyFuture<T: IsResult>: Future<Item=T::Ok, Error = T::Err> {}
+pub trait MyStableFuture<T: IsResult>: StableFuture<Item=T::Ok, Error = T::Err> {}
 
-impl<F, T> MyFuture<T> for F
-    where F: Future<Item = T::Ok, Error = T::Err > + ?Sized,
-          T: IsResult
+impl<F, T> MyStableFuture<T> for F
+    where F: StableFuture<Item = T::Ok, Error = T::Err> + ?Sized,
+          T: IsResult,
 {}
 
-/// Small shim to translate from a generator to a future.
-///
-/// This is the translation layer from the generator/coroutine protocol to
-/// the futures protocol.
-struct GenFuture<T>(T);
+struct GenStableFuture<T>(T);
 
-pub fn gen_move<T>(gen: T) -> impl MyFuture<T::Return>
-    where T: Generator<Yield = Async<Never>>,
-          T::Return: IsResult,
-{
-    GenFuture(gen)
-}
+impl<T> !Unpin for GenStableFuture<T> { }
 
-impl<T> Future for GenFuture<T>
+impl<T> StableFuture for GenStableFuture<T>
     where T: Generator<Yield = Async<Never>>,
           T::Return: IsResult,
 {
     type Item = <T::Return as IsResult>::Ok;
     type Error = <T::Return as IsResult>::Err;
 
-    fn poll(&mut self, ctx: &mut task::Context) -> Poll<Self::Item, Self::Error> {
+    fn poll(mut self: Pin<Self>, ctx: &mut task::Context) -> Poll<Self::Item, Self::Error> {
         CTX.with(|cell| {
             let _r = Reset::new(ctx, cell);
-            // Because we are controlling the creation of our underlying
-            // generator, we know that this is definitely a movable generator
-            // so calling resume is always safe.
-            match unsafe { self.0.resume() } {
+            let this: &mut Self = unsafe { Pin::get_mut(&mut self) };
+            // This is an immovable generator, but since we're only accessing
+            // it via a Pin this is safe.
+            match unsafe { this.0.resume() } {
                 GeneratorState::Yielded(Async::Pending)
                     => Ok(Async::Pending),
                 GeneratorState::Yielded(Async::Ready(mu))
@@ -50,3 +42,10 @@ impl<T> Future for GenFuture<T>
         })
     }
 }
+
+pub fn gen_future<'a, T>(gen: T) -> impl MyStableFuture<T::Return> + 'a
+    where T: Generator<Yield = Async<Never>> + 'a,
+          T::Return: IsResult,
+{
+    GenStableFuture(gen)
+}

futures-async-runtime/src/lib.rs

@@ -22,8 +22,6 @@ if_nightly_and_std! {
 
     mod future;
     mod stream;
-    mod pinned_future;
-    mod pinned_stream;
 
     use std::cell::Cell;
     use std::mem;
@@ -32,8 +30,6 @@ if_nightly_and_std! {
 
     pub use self::future::*;
     pub use self::stream::*;
-    pub use self::pinned_future::*;
-    pub use self::pinned_stream::*;
 
     pub use futures_core::{Async, Future, Stream};
     pub use futures_stable::{StableFuture, StableStream};

futures-async-runtime/src/pinned_future.rs

@@ -1,55 +1,58 @@
+use std::mem::Pin;
 use std::ops::{Generator, GeneratorState};
-use std::marker::PhantomData;
+use std::marker::{PhantomData, Unpin};
 
-use futures_core::task;
-use futures_core::{Poll, Async, Stream};
+use futures_core::{Poll, Async, task};
+use futures_stable::StableStream;
 
 use super::{CTX, Reset, IsResult};
 
-pub trait MyStream<T, U: IsResult<Ok=()>>: Stream<Item=T, Error=U::Err> {}
+pub trait MyStableStream<T, U: IsResult<Ok=()>>: StableStream<Item=T, Error=U::Err> {}
 
-impl<F, T, U> MyStream<T, U> for F
-    where F: Stream<Item = T, Error = U::Err> + ?Sized,
+impl<F, T, U> MyStableStream<T, U> for F
+    where F: StableStream<Item = T, Error = U::Err> + ?Sized,
           U: IsResult<Ok=()>
 {}
 
 /// Small shim to translate from a generator to a stream.
-struct GenStream<U, T> {
+struct GenStableStream<U, T> {
     gen: T,
     done: bool,
     phantom: PhantomData<U>,
 }
 
-pub fn gen_stream<T, U>(gen: T) -> impl MyStream<U, T::Return>
+impl<U, T> !Unpin for GenStableStream<U, T> { }
+
+pub fn gen_stream<T, U>(gen: T) -> impl MyStableStream<U, T::Return>
     where T: Generator<Yield = Async<U>>,
           T::Return: IsResult<Ok = ()>,
 {
-    GenStream { gen, done: false, phantom: PhantomData }
+    GenStableStream { gen, done: false, phantom: PhantomData }
 }
 
-impl<U, T> Stream for GenStream<U, T>
+impl<U, T> StableStream for GenStableStream<U, T>
     where T: Generator<Yield = Async<U>>,
           T::Return: IsResult<Ok = ()>,
 {
     type Item = U;
     type Error = <T::Return as IsResult>::Err;
 
-    fn poll_next(&mut self, ctx: &mut task::Context) -> Poll<Option<Self::Item>, Self::Error> {
+    fn poll_next(mut self: Pin<Self>, ctx: &mut task::Context) -> Poll<Option<Self::Item>, Self::Error> {
         CTX.with(|cell| {
             let _r = Reset::new(ctx, cell);
-            if self.done { return Ok(Async::Ready(None)) }
-            // Because we are controlling the creation of our underlying
-            // generator, we know that this is definitely a movable generator
-            // so calling resume is always safe.
-            match unsafe { self.gen.resume() } {
+            let this: &mut Self = unsafe { Pin::get_mut(&mut self) };
+            if this.done { return Ok(Async::Ready(None)) }
+            // This is an immovable generator, but since we're only accessing
+            // it via a Pin this is safe.
+            match unsafe { this.gen.resume() } {
                 GeneratorState::Yielded(Async::Ready(e)) => {
                     Ok(Async::Ready(Some(e)))
                 }
                 GeneratorState::Yielded(Async::Pending) => {
                     Ok(Async::Pending)
                 }
                 GeneratorState::Complete(e) => {
-                    self.done = true;
+                    this.done = true;
                     e.into_result().map(|()| Async::Ready(None))
                 }
             }

futures-async-runtime/src/pinned_stream.rs

@@ -255,7 +255,7 @@ if_nightly! {
 
         let attribute = Attribute::from(args.0.into_iter().map(|arg| arg.0));
 
-        async_inner(attribute, function, quote_cs! { ::futures::__rt::gen_pinned }, |output, lifetimes| {
+        async_inner(attribute, function, quote_cs! { ::futures::__rt::gen_future }, |output, lifetimes| {
             // TODO: can we lift the restriction that `futures` must be at the root of
             //       the crate?
             let output_span = first_last(&output);
@@ -313,7 +313,7 @@ if_nightly! {
         let item_ty = item_ty.expect("#[async_stream] requires item type to be specified");
         let attribute = Attribute::from(args);
 
-        async_inner(attribute, function, quote_cs! { ::futures::__rt::gen_stream_pinned }, |output, lifetimes| {
+        async_inner(attribute, function, quote_cs! { ::futures::__rt::gen_stream }, |output, lifetimes| {
             let return_ty = match attribute {
                 Attribute::NONE => quote_cs! {
                     impl ::futures::__rt::MyStableStream<!, !> + #(#lifetimes +)*
@@ -348,7 +348,7 @@ if_nightly! {
         let expr = ExpandAsyncFor.fold_expr(expr);
 
         let mut tokens = quote_cs! {
-            ::futures::__rt::gen_pinned
+            ::futures::__rt::gen_future
         };
 
         // Use some manual token construction here instead of `quote_cs!` to ensure
@@ -377,7 +377,7 @@ if_nightly! {
         let expr = ExpandAsyncFor.fold_expr(expr);
 
         let mut tokens = quote_cs! {
-            ::futures::__rt::gen_stream_pinned
+            ::futures::__rt::gen_stream
         };
 
         // Use some manual token construction here instead of `quote_cs!` to ensure