Add server support for push (#327)

Closes #291, closes #185

Author: michaelbeaumont

src/frame/headers.rs

@@ -3,7 +3,7 @@ use crate::frame::{Error, Frame, Head, Kind};
 use crate::hpack;
 
 use http::header::{self, HeaderName, HeaderValue};
-use http::{uri, HeaderMap, Method, StatusCode, Uri};
+use http::{uri, HeaderMap, Method, Request, StatusCode, Uri};
 
 use bytes::{Bytes, BytesMut};
 use string::String;
@@ -300,9 +300,92 @@ impl fmt::Debug for Headers {
     }
 }
 
+// ===== util =====
+
+pub fn parse_u64(src: &[u8]) -> Result<u64, ()> {
+    if src.len() > 19 {
+        // At danger for overflow...
+        return Err(());
+    }
+
+    let mut ret = 0;
+
+    for &d in src {
+        if d < b'0' || d > b'9' {
+            return Err(());
+        }
+
+        ret *= 10;
+        ret += (d - b'0') as u64;
+    }
+
+    Ok(ret)
+}
+
 // ===== impl PushPromise =====
 
+#[derive(Debug)]
+pub enum PushPromiseHeaderError {
+    InvalidContentLength(Result<u64, ()>),
+    NotSafeAndCacheable,
+}
+
 impl PushPromise {
+    pub fn new(
+        stream_id: StreamId,
+        promised_id: StreamId,
+        pseudo: Pseudo,
+        fields: HeaderMap,
+    ) -> Self {
+        PushPromise {
+            flags: PushPromiseFlag::default(),
+            header_block: HeaderBlock {
+                fields,
+                is_over_size: false,
+                pseudo,
+            },
+            promised_id,
+            stream_id,
+        }
+    }
+
+    pub fn validate_request(req: &Request<()>) -> Result<(), PushPromiseHeaderError> {
+        use PushPromiseHeaderError::*;
+        // The spec has some requirements for promised request headers
+        // [https://httpwg.org/specs/rfc7540.html#PushRequests]
+
+        // A promised request "that indicates the presence of a request body
+        // MUST reset the promised stream with a stream error"
+        if let Some(content_length) = req.headers().get(header::CONTENT_LENGTH) {
+            let parsed_length = parse_u64(content_length.as_bytes());
+            if parsed_length != Ok(0) {
+                return Err(InvalidContentLength(parsed_length));
+            }
+        }
+        // "The server MUST include a method in the :method pseudo-header field
+        // that is safe and cacheable"
+        if !Self::safe_and_cacheable(req.method()) {
+            return Err(NotSafeAndCacheable);
+        }
+
+        Ok(())
+    }
+
+    fn safe_and_cacheable(method: &Method) -> bool {
+        // Cacheable: https://httpwg.org/specs/rfc7231.html#cacheable.methods
+        // Safe: https://httpwg.org/specs/rfc7231.html#safe.methods
+        return method == Method::GET || method == Method::HEAD;
+    }
+
+    pub fn fields(&self) -> &HeaderMap {
+        &self.header_block.fields
+    }
+
+    #[cfg(feature = "unstable")]
+    pub fn into_fields(self) -> HeaderMap {
+        self.header_block.fields
+    }
+
     /// Loads the push promise frame but doesn't actually do HPACK decoding.
     ///
     /// HPACK decoding is done in the `load_hpack` step.
@@ -401,44 +484,13 @@ impl PushPromise {
     fn head(&self) -> Head {
         Head::new(Kind::PushPromise, self.flags.into(), self.stream_id)
     }
-}
 
-impl PushPromise {
     /// Consume `self`, returning the parts of the frame
     pub fn into_parts(self) -> (Pseudo, HeaderMap) {
         (self.header_block.pseudo, self.header_block.fields)
     }
 }
 
-#[cfg(feature = "unstable")]
-impl PushPromise {
-    pub fn new(
-        stream_id: StreamId,
-        promised_id: StreamId,
-        pseudo: Pseudo,
-        fields: HeaderMap,
-    ) -> Self {
-        PushPromise {
-            flags: PushPromiseFlag::default(),
-            header_block: HeaderBlock {
-                fields,
-                is_over_size: false,
-                pseudo,
-            },
-            promised_id,
-            stream_id,
-        }
-    }
-
-    pub fn fields(&self) -> &HeaderMap {
-        &self.header_block.fields
-    }
-
-    pub fn into_fields(self) -> HeaderMap {
-        self.header_block.fields
-    }
-}
-
 impl<T> From<PushPromise> for Frame<T> {
     fn from(src: PushPromise) -> Self {
         Frame::PushPromise(src)

src/frame/mod.rs

@@ -42,7 +42,9 @@ mod window_update;
 pub use self::data::Data;
 pub use self::go_away::GoAway;
 pub use self::head::{Head, Kind};
-pub use self::headers::{Continuation, Headers, Pseudo, PushPromise};
+pub use self::headers::{
+    parse_u64, Continuation, Headers, Pseudo, PushPromise, PushPromiseHeaderError,
+};
 pub use self::ping::Ping;
 pub use self::priority::{Priority, StreamDependency};
 pub use self::reason::Reason;

src/proto/streams/prioritize.rs

@@ -111,7 +111,7 @@ impl Prioritize {
 
     pub fn schedule_send(&mut self, stream: &mut store::Ptr, task: &mut Option<Waker>) {
         // If the stream is waiting to be opened, nothing more to do.
-        if !stream.is_pending_open {
+        if stream.is_send_ready() {
             log::trace!("schedule_send; {:?}", stream.id);
             // Queue the stream
             self.pending_send.push(stream);
@@ -445,19 +445,9 @@ impl Prioritize {
             self.pending_capacity.push(stream);
         }
 
-        // If data is buffered and the stream is not pending open, then
+        // If data is buffered and the stream is send ready, then
         // schedule the stream for execution
-        //
-        // Why do we not push into pending_send when the stream is in pending_open?
-        //
-        // We allow users to call send_request() which schedules a stream to be pending_open
-        // if there is no room according to the concurrency limit (max_send_streams), and we
-        // also allow data to be buffered for send with send_data() if there is no capacity for
-        // the stream to send the data, which attempts to place the stream in pending_send.
-        // If the stream is not open, we don't want the stream to be scheduled for
-        // execution (pending_send). Note that if the stream is in pending_open, it will be
-        // pushed to pending_send when there is room for an open stream.
-        if stream.buffered_send_data > 0 && !stream.is_pending_open {
+        if stream.buffered_send_data > 0 && stream.is_send_ready() {
             // TODO: This assertion isn't *exactly* correct. There can still be
             // buffered send data while the stream's pending send queue is
             // empty. This can happen when a large data frame is in the process
@@ -766,6 +756,22 @@ impl Prioritize {
                                 stream: stream.key(),
                             }))
                         }
+                        Some(Frame::PushPromise(pp)) => {
+                            let mut pushed =
+                                stream.store_mut().find_mut(&pp.promised_id()).unwrap();
+                            pushed.is_pending_push = false;
+                            // Transition stream from pending_push to pending_open
+                            // if possible
+                            if !pushed.pending_send.is_empty() {
+                                if counts.can_inc_num_send_streams() {
+                                    counts.inc_num_send_streams(&mut pushed);
+                                    self.pending_send.push(&mut pushed);
+                                } else {
+                                    self.queue_open(&mut pushed);
+                                }
+                            }
+                            Frame::PushPromise(pp)
+                        }
                         Some(frame) => frame.map(|_| {
                             unreachable!(
                                 "Frame::map closure will only be called \

src/proto/streams/recv.rs

@@ -1,10 +1,10 @@
 use super::*;
 use crate::codec::{RecvError, UserError};
-use crate::frame::{Reason, DEFAULT_INITIAL_WINDOW_SIZE};
+use crate::frame::{PushPromiseHeaderError, Reason, DEFAULT_INITIAL_WINDOW_SIZE};
 use crate::{frame, proto};
 use std::task::Context;
 
-use http::{HeaderMap, Method, Request, Response};
+use http::{HeaderMap, Request, Response};
 
 use std::io;
 use std::task::{Poll, Waker};
@@ -178,7 +178,7 @@ impl Recv {
             use http::header;
 
             if let Some(content_length) = frame.fields().get(header::CONTENT_LENGTH) {
-                let content_length = match parse_u64(content_length.as_bytes()) {
+                let content_length = match frame::parse_u64(content_length.as_bytes()) {
                     Ok(v) => v,
                     Err(()) => {
                         proto_err!(stream: "could not parse content-length; stream={:?}", stream.id);
@@ -632,44 +632,31 @@ impl Recv {
         }
 
         let promised_id = frame.promised_id();
-        use http::header;
         let (pseudo, fields) = frame.into_parts();
         let req = crate::server::Peer::convert_poll_message(pseudo, fields, promised_id)?;
-        // The spec has some requirements for promised request headers
-        // [https://httpwg.org/specs/rfc7540.html#PushRequests]
-
-        // A promised request "that indicates the presence of a request body
-        // MUST reset the promised stream with a stream error"
-        if let Some(content_length) = req.headers().get(header::CONTENT_LENGTH) {
-            match parse_u64(content_length.as_bytes()) {
-                Ok(0) => {}
-                otherwise => {
-                    proto_err!(stream:
-                        "recv_push_promise; promised request has content-length {:?}; promised_id={:?}",
-                        otherwise,
-                        promised_id,
-                    );
-                    return Err(RecvError::Stream {
-                        id: promised_id,
-                        reason: Reason::PROTOCOL_ERROR,
-                    });
-                }
+
+        if let Err(e) = frame::PushPromise::validate_request(&req) {
+            use PushPromiseHeaderError::*;
+            match e {
+                NotSafeAndCacheable => proto_err!(
+                    stream:
+                    "recv_push_promise: method {} is not safe and cacheable; promised_id={:?}",
+                    req.method(),
+                    promised_id,
+                ),
+                InvalidContentLength(e) => proto_err!(
+                    stream:
+                    "recv_push_promise; promised request has invalid content-length {:?}; promised_id={:?}",
+                    e,
+                    promised_id,
+                ),
             }
-        }
-        // "The server MUST include a method in the :method pseudo-header field
-        // that is safe and cacheable"
-        if !Self::safe_and_cacheable(req.method()) {
-            proto_err!(
-                stream:
-                "recv_push_promise: method {} is not safe and cacheable; promised_id={:?}",
-                req.method(),
-                promised_id,
-            );
             return Err(RecvError::Stream {
                 id: promised_id,
                 reason: Reason::PROTOCOL_ERROR,
             });
         }
+
         use super::peer::PollMessage::*;
         stream
             .pending_recv
@@ -678,12 +665,6 @@ impl Recv {
         Ok(())
     }
 
-    fn safe_and_cacheable(method: &Method) -> bool {
-        // Cacheable: https://httpwg.org/specs/rfc7231.html#cacheable.methods
-        // Safe: https://httpwg.org/specs/rfc7231.html#safe.methods
-        method == Method::GET || method == Method::HEAD
-    }
-
     /// Ensures that `id` is not in the `Idle` state.
     pub fn ensure_not_idle(&self, id: StreamId) -> Result<(), Reason> {
         if let Ok(next) = self.next_stream_id {
@@ -1057,25 +1038,3 @@ impl<T> From<RecvError> for RecvHeaderBlockError<T> {
         RecvHeaderBlockError::State(err)
     }
 }
-
-// ===== util =====
-
-fn parse_u64(src: &[u8]) -> Result<u64, ()> {
-    if src.len() > 19 {
-        // At danger for overflow...
-        return Err(());
-    }
-
-    let mut ret = 0;
-
-    for &d in src {
-        if d < b'0' || d > b'9' {
-            return Err(());
-        }
-
-        ret *= 10;
-        ret += u64::from(d - b'0');
-    }
-
-    Ok(ret)
-}