refactor(lib): Use pin-project crate to perform pin projections

Remove all pin-related `unsafe` code from Hyper, as well as the
now-unused 'pin-utils' dependency.

Author: Aaron1011

Cargo.toml

@@ -34,7 +34,8 @@ iovec = "0.1"
 itoa = "0.4.1"
 log = "0.4"
 net2 = { version = "0.2.32", optional = true }
-pin-utils = "=0.1.0-alpha.4"
+pin-project = { version = "0.4.0-alpha.7", features = ["project_attr"] }
+
 time = "0.1"
 tokio = { version = "=0.2.0-alpha.4", optional = true, default-features = false, features = ["rt-full"] }
 tower-service = "=0.3.0-alpha.1"

src/common/drain.rs

@@ -2,6 +2,7 @@ use std::mem;
 
 use futures_util::FutureExt as _;
 use tokio_sync::{mpsc, watch};
+use pin_project::pin_project;
 
 use super::{Future, Never, Poll, Pin, task};
 
@@ -43,7 +44,9 @@ pub struct Watch {
 }
 
 #[allow(missing_debug_implementations)]
+#[pin_project]
 pub struct Watching<F, FN> {
+    #[pin]
     future: F,
     state: State<FN>,
     watch: Watch,
@@ -95,28 +98,28 @@ where
 {
     type Output = F::Output;
 
-    fn poll(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
-        let me = unsafe { self.get_unchecked_mut() };
+    fn poll(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<Self::Output> {
         loop {
-            match mem::replace(&mut me.state, State::Draining) {
+            let me = self.project();
+            match mem::replace(me.state, State::Draining) {
                 State::Watch(on_drain) => {
                     let recv = me.watch.rx.recv_ref();
                     futures_util::pin_mut!(recv);
 
                     match recv.poll_unpin(cx) {
                         Poll::Ready(None) => {
                             // Drain has been triggered!
-                            on_drain(unsafe { Pin::new_unchecked(&mut me.future) });
+                            on_drain(me.future);
                         },
                         Poll::Ready(Some(_/*State::Open*/)) |
                         Poll::Pending => {
-                            me.state = State::Watch(on_drain);
-                            return unsafe { Pin::new_unchecked(&mut me.future) }.poll(cx);
+                            *me.state = State::Watch(on_drain);
+                            return me.future.poll(cx);
                         },
                     }
                 },
                 State::Draining => {
-                    return unsafe { Pin::new_unchecked(&mut me.future) }.poll(cx);
+                    return me.future.poll(cx)
                 },
             }
         }

src/proto/h2/server.rs

@@ -1,6 +1,7 @@
 use std::error::Error as StdError;
 use std::marker::Unpin;
 
+use pin_project::{pin_project, project};
 use h2::Reason;
 use h2::server::{Builder, Connection, Handshake, SendResponse};
 use tokio_io::{AsyncRead, AsyncWrite};
@@ -199,19 +200,22 @@ where
 }
 
 #[allow(missing_debug_implementations)]
+#[pin_project]
 pub struct H2Stream<F, B>
 where
     B: Payload,
 {
     reply: SendResponse<SendBuf<B::Data>>,
+    #[pin]
     state: H2StreamState<F, B>,
 }
 
+#[pin_project]
 enum H2StreamState<F, B>
 where
     B: Payload,
 {
-    Service(F),
+    Service(#[pin] F),
     Body(PipeToSendStream<B>),
 }
 
@@ -229,20 +233,34 @@ where
     }
 }
 
+macro_rules! reply {
+    ($me:expr, $res:expr, $eos:expr) => ({
+        match $me.reply.send_response($res, $eos) {
+            Ok(tx) => tx,
+            Err(e) => {
+                debug!("send response error: {}", e);
+                $me.reply.send_reset(Reason::INTERNAL_ERROR);
+                return Poll::Ready(Err(crate::Error::new_h2(e)));
+            }
+        }
+    })
+}
+
 impl<F, B, E> H2Stream<F, B>
 where
     F: Future<Output = Result<Response<B>, E>>,
     B: Payload + Unpin,
     B::Data: Unpin,
     E: Into<Box<dyn StdError + Send + Sync>>,
 {
-    fn poll2(self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<crate::Result<()>> {
-        // Safety: State::{Service, Body} futures are never moved
-        let me = unsafe { self.get_unchecked_mut() };
+    #[project]
+    fn poll2(mut self: Pin<&mut Self>, cx: &mut task::Context<'_>) -> Poll<crate::Result<()>> {
         loop {
-            let next = match me.state {
-                H2StreamState::Service(ref mut h) => {
-                    let res = match unsafe { Pin::new_unchecked(h) }.poll(cx) {
+            let mut me = self.project();
+            #[project]
+            let next = match me.state.project() {
+                H2StreamState::Service(h) => {
+                    let res = match h.poll(cx) {
                         Poll::Ready(Ok(r)) => r,
                         Poll::Pending => {
                             // Response is not yet ready, so we want to check if the client has sent a
@@ -274,37 +292,26 @@ where
                         .expect("DATE is a valid HeaderName")
                         .or_insert_with(crate::proto::h1::date::update_and_header_value);
 
-                    macro_rules! reply {
-                        ($eos:expr) => ({
-                            match me.reply.send_response(res, $eos) {
-                                Ok(tx) => tx,
-                                Err(e) => {
-                                    debug!("send response error: {}", e);
-                                    me.reply.send_reset(Reason::INTERNAL_ERROR);
-                                    return Poll::Ready(Err(crate::Error::new_h2(e)));
-                                }
-                            }
-                        })
-                    }
+
 
                     // automatically set Content-Length from body...
                     if let Some(len) = body.size_hint().exact() {
                         headers::set_content_length_if_missing(res.headers_mut(), len);
                     }
 
                     if !body.is_end_stream() {
-                        let body_tx = reply!(false);
+                        let body_tx = reply!(me, res, false);
                         H2StreamState::Body(PipeToSendStream::new(body, body_tx))
                     } else {
-                        reply!(true);
+                        reply!(me, res, true);
                         return Poll::Ready(Ok(()));
                     }
                 },
                 H2StreamState::Body(ref mut pipe) => {
                     return Pin::new(pipe).poll(cx);
                 }
             };
-            me.state = next;
+            me.state.set(next);
         }
     }
 }