Restructure proto

The existing code has been moved out and is being copied back piece / by
piece while restructuring the code to (hopefully) be more manageable.

src/proto/streams/mod.rs

@@ -0,0 +1,263 @@
+mod recv;
+mod send;
+
+use self::recv::Recv;
+use self::send::Send;
+
+use {frame, Peer, StreamId, ConnectionError};
+use proto::*;
+use error::Reason::*;
+use error::User::*;
+
+use ordermap::{OrderMap, Entry};
+
+// TODO: All the VecDeques should become linked lists using the state::Stream
+// values.
+#[derive(Debug)]
+pub struct Streams<P> {
+    /// State related to managing the set of streams.
+    inner: Inner<P>,
+
+    /// Streams
+    streams: StreamMap,
+}
+
+type StreamMap = OrderMap<StreamId, state::Stream>;
+
+/// Fields needed to manage state related to managing the set of streams. This
+/// is mostly split out to make ownership happy.
+///
+/// TODO: better name
+#[derive(Debug)]
+struct Inner<P> {
+    /// Manages state transitions initiated by receiving frames
+    recv: Recv<P>,
+
+    /// Manages state transitions initiated by sending frames
+    send: Send<P>,
+}
+
+#[derive(Debug)]
+pub struct Config {
+    /// Maximum number of remote initiated streams
+    pub max_remote_initiated: Option<usize>,
+
+    /// Initial window size of remote initiated streams
+    pub init_remote_window_sz: WindowSize,
+
+    /// Maximum number of locally initiated streams
+    pub max_local_initiated: Option<usize>,
+
+    /// Initial window size of locally initiated streams
+    pub init_local_window_sz: WindowSize,
+}
+
+impl<P: Peer> Streams<P> {
+    pub fn new(config: Config) -> Self {
+        Streams {
+            inner: Inner {
+                recv: Recv::new(&config),
+                send: Send::new(&config),
+            },
+            streams: OrderMap::default(),
+        }
+    }
+
+    pub fn recv_headers(&mut self, frame: frame::Headers)
+        -> Result<Option<frame::Headers>, ConnectionError>
+    {
+        let id = frame.stream_id();
+
+        let state = match self.streams.entry(id) {
+            Entry::Occupied(e) => e.into_mut(),
+            Entry::Vacant(e) => {
+                // Trailers cannot open a stream. Trailers are header frames
+                // that do not contain pseudo headers. Requests MUST contain a
+                // method and responses MUST contain a status. If they do not,t
+                // hey are considered to be malformed.
+                if frame.is_trailers() {
+                    return Err(ProtocolError.into());
+                }
+
+                match try!(self.inner.recv.open(id)) {
+                    Some(state) => e.insert(state),
+                    None => return Ok(None),
+                }
+            }
+        };
+
+        if frame.is_trailers() {
+            try!(self.inner.recv.recv_trailers(state, frame.is_end_stream()));
+        } else {
+            try!(self.inner.recv.recv_headers(state, frame.is_end_stream()));
+        }
+
+        if state.is_closed() {
+            self.inner.dec_num_streams(id);
+        }
+
+        Ok(Some(frame))
+    }
+
+    pub fn recv_data(&mut self, frame: &frame::Data)
+        -> Result<(), ConnectionError>
+    {
+        let id = frame.stream_id();
+
+        let state = match self.streams.get_mut(&id) {
+            Some(state) => state,
+            None => return Err(ProtocolError.into()),
+        };
+
+        // Ensure there's enough capacity on the connection before acting on the
+        // stream.
+        try!(self.inner.recv.recv_data(frame, state));
+
+        if state.is_closed() {
+            self.inner.dec_num_streams(id);
+        }
+
+        Ok(())
+    }
+
+    pub fn recv_reset(&mut self, _frame: &frame::Reset)
+        -> Result<(), ConnectionError>
+    {
+        unimplemented!();
+    }
+
+    pub fn recv_window_update(&mut self, frame: frame::WindowUpdate)
+        -> Result<(), ConnectionError> {
+        let id = frame.stream_id();
+
+        if id.is_zero() {
+            try!(self.inner.send.recv_connection_window_update(frame));
+        } else {
+            // The remote may send window updates for streams that the local now
+            // considers closed. It's ok...
+            if let Some(state) = self.streams.get_mut(&id) {
+                try!(self.inner.send.recv_stream_window_update(frame, state));
+            }
+        }
+
+        Ok(())
+    }
+
+    pub fn recv_push_promise(&mut self, _frame: frame::PushPromise)
+        -> Result<(), ConnectionError>
+    {
+        unimplemented!();
+    }
+
+    pub fn send_headers(&mut self, frame: &frame::Headers)
+        -> Result<(), ConnectionError>
+    {
+        let id = frame.stream_id();
+
+        trace!("send_headers; id={:?}", id);
+
+        let state = match self.streams.entry(id) {
+            Entry::Occupied(e) => e.into_mut(),
+            Entry::Vacant(e) => {
+                // Trailers cannot open a stream. Trailers are header frames
+                // that do not contain pseudo headers. Requests MUST contain a
+                // method and responses MUST contain a status. If they do not,t
+                // hey are considered to be malformed.
+                if frame.is_trailers() {
+                    // TODO: Should this be a different error?
+                    return Err(UnexpectedFrameType.into());
+                }
+
+                let state = try!(self.inner.send.open(id));
+                e.insert(state)
+            }
+        };
+
+        if frame.is_trailers() {
+            try!(self.inner.send.send_trailers(state, frame.is_end_stream()));
+        } else {
+            try!(self.inner.send.send_headers(state, frame.is_end_stream()));
+        }
+
+        if state.is_closed() {
+            self.inner.dec_num_streams(id);
+        }
+
+        Ok(())
+    }
+
+    pub fn send_data<B: Buf>(&mut self, frame: &frame::Data<B>)
+        -> Result<(), ConnectionError>
+    {
+        let id = frame.stream_id();
+
+        let state = match self.streams.get_mut(&id) {
+            Some(state) => state,
+            None => return Err(UnexpectedFrameType.into()),
+        };
+
+        // Ensure there's enough capacity on the connection before acting on the
+        // stream.
+        try!(self.inner.send.send_data(frame, state));
+
+        if state.is_closed() {
+            self.inner.dec_num_streams(id);
+        }
+
+        Ok(())
+    }
+
+    pub fn poll_window_update(&mut self)
+        -> Poll<WindowUpdate, ConnectionError>
+    {
+        self.inner.send.poll_window_update(&mut self.streams)
+    }
+
+    pub fn expand_window(&mut self, id: StreamId, sz: WindowSize)
+        -> Result<(), ConnectionError>
+    {
+        if id.is_zero() {
+            try!(self.inner.recv.expand_connection_window(sz));
+        } else {
+            if let Some(state) = self.streams.get_mut(&id) {
+                try!(self.inner.recv.expand_stream_window(id, sz, state));
+            }
+        }
+
+        Ok(())
+    }
+
+    pub fn send_pending_refusal<T, B>(&mut self, dst: &mut Codec<T, B>)
+        -> Poll<(), ConnectionError>
+        where T: AsyncWrite,
+              B: Buf,
+    {
+        self.inner.recv.send_pending_refusal(dst)
+    }
+
+    pub fn send_pending_window_updates<T, B>(&mut self, dst: &mut Codec<T, B>)
+        -> Poll<(), ConnectionError>
+        where T: AsyncWrite,
+              B: Buf,
+    {
+        try_ready!(self.inner.recv.send_connection_window_update(dst));
+        try_ready!(self.inner.recv.send_stream_window_update(&mut self.streams, dst));
+
+        Ok(().into())
+    }
+}
+
+impl<P: Peer> Inner<P> {
+    fn dec_num_streams(&mut self, id: StreamId) {
+        if self.is_local_init(id) {
+            self.send.dec_num_streams();
+        } else {
+            self.recv.dec_num_streams();
+        }
+    }
+
+    fn is_local_init(&self, id: StreamId) -> bool {
+        assert!(!id.is_zero());
+        P::is_server() == id.is_server_initiated()
+    }
+}

src/proto/streams/recv.rs

@@ -0,0 +1,235 @@
+use {frame, Peer, ConnectionError};
+use proto::*;
+use super::{Config, StreamMap};
+
+use error::Reason::*;
+
+use std::collections::VecDeque;
+use std::marker::PhantomData;
+
+#[derive(Debug)]
+pub struct Recv<P> {
+    /// Maximum number of remote initiated streams
+    max_streams: Option<usize>,
+
+    /// Current number of remote initiated streams
+    num_streams: usize,
+
+    /// Initial window size of remote initiated streams
+    init_window_sz: WindowSize,
+
+    /// Connection level flow control governing received data
+    flow_control: state::FlowControl,
+
+    pending_window_updates: VecDeque<StreamId>,
+
+    /// Refused StreamId, this represents a frame that must be sent out.
+    refused: Option<StreamId>,
+
+    _p: PhantomData<P>,
+}
+
+impl<P: Peer> Recv<P> {
+    pub fn new(config: &Config) -> Self {
+        Recv {
+            max_streams: config.max_remote_initiated,
+            num_streams: 0,
+            init_window_sz: config.init_remote_window_sz,
+            flow_control: state::FlowControl::new(config.init_remote_window_sz),
+            pending_window_updates: VecDeque::new(),
+            refused: None,
+            _p: PhantomData,
+        }
+    }
+
+    /// Update state reflecting a new, remotely opened stream
+    ///
+    /// Returns the stream state if successful. `None` if refused
+    pub fn open(&mut self, id: StreamId) -> Result<Option<state::Stream>, ConnectionError> {
+        assert!(self.refused.is_none());
+
+        try!(self.ensure_can_open(id));
+
+        if let Some(max) = self.max_streams {
+            if max <= self.num_streams {
+                self.refused = Some(id);
+                return Ok(None);
+            }
+        }
+
+        // Increment the number of remote initiated streams
+        self.num_streams += 1;
+
+        Ok(Some(state::Stream::default()))
+    }
+
+    /// Transition the stream state based on receiving headers
+    pub fn recv_headers(&mut self, state: &mut state::Stream, eos: bool)
+        -> Result<(), ConnectionError>
+    {
+        state.recv_open(self.init_window_sz, eos)
+    }
+
+    pub fn recv_trailers(&mut self, _state: &mut state::Stream, _eos: bool)
+        -> Result<(), ConnectionError>
+    {
+        unimplemented!();
+    }
+
+    pub fn recv_data(&mut self,
+                     frame: &frame::Data,
+                     state: &mut state::Stream)
+        -> Result<(), ConnectionError>
+    {
+        let sz = frame.payload().len();
+
+        if sz > MAX_WINDOW_SIZE as usize {
+            unimplemented!();
+        }
+
+        let sz = sz as WindowSize;
+
+        match state.recv_flow_control() {
+            Some(flow) => {
+                // Ensure there's enough capacity on the connection before
+                // acting on the stream.
+                try!(self.flow_control.ensure_window(sz, FlowControlError));
+
+                // Claim the window on the stream
+                try!(flow.claim_window(sz, FlowControlError));
+
+                // Claim the window on the connection.
+                self.flow_control.claim_window(sz, FlowControlError)
+                    .expect("local connection flow control error");
+            }
+            None => return Err(ProtocolError.into()),
+        }
+
+        if frame.is_end_stream() {
+            try!(state.recv_close());
+        }
+
+        Ok(())
+    }
+
+    pub fn dec_num_streams(&mut self) {
+        self.num_streams -= 1;
+    }
+
+    /// Returns true if the remote peer can initiate a stream with the given ID.
+    fn ensure_can_open(&self, id: StreamId) -> Result<(), ConnectionError> {
+        if !P::is_server() {
+            // Remote is a server and cannot open streams. PushPromise is
+            // registered by reserving, so does not go through this path.
+            return Err(ProtocolError.into());
+        }
+
+        // Ensure that the ID is a valid server initiated ID
+        if !id.is_client_initiated() {
+            return Err(ProtocolError.into());
+        }
+
+        Ok(())
+    }
+
+    /// Send any pending refusals.
+    pub fn send_pending_refusal<T, B>(&mut self, dst: &mut Codec<T, B>)
+        -> Poll<(), ConnectionError>
+        where T: AsyncWrite,
+              B: Buf,
+    {
+        if let Some(stream_id) = self.refused.take() {
+            let frame = frame::Reset::new(stream_id, RefusedStream);
+
+            match dst.start_send(frame.into())? {
+                AsyncSink::Ready => {
+                    self.reset(stream_id, RefusedStream);
+                    return Ok(Async::Ready(()));
+                }
+                AsyncSink::NotReady(_) => {
+                    self.refused = Some(stream_id);
+                    return Ok(Async::NotReady);
+                }
+            }
+        }
+
+        Ok(Async::Ready(()))
+    }
+
+    pub fn expand_connection_window(&mut self, sz: WindowSize)
+        -> Result<(), ConnectionError>
+    {
+        // TODO: handle overflow
+        self.flow_control.expand_window(sz);
+
+        Ok(())
+    }
+
+    pub fn expand_stream_window(&mut self,
+                                id: StreamId,
+                                sz: WindowSize,
+                                state: &mut state::Stream)
+        -> Result<(), ConnectionError>
+    {
+        // TODO: handle overflow
+        if let Some(flow) = state.recv_flow_control() {
+            flow.expand_window(sz);
+            self.pending_window_updates.push_back(id);
+        }
+
+        Ok(())
+    }
+
+    /// Send connection level window update
+    pub fn send_connection_window_update<T, B>(&mut self, dst: &mut Codec<T, B>)
+        -> Poll<(), ConnectionError>
+        where T: AsyncWrite,
+              B: Buf,
+    {
+        if let Some(incr) = self.flow_control.peek_window_update() {
+            let frame = frame::WindowUpdate::new(StreamId::zero(), incr);
+
+            if dst.start_send(frame.into())?.is_ready() {
+                assert_eq!(Some(incr), self.flow_control.apply_window_update());
+            } else {
+                return Ok(Async::NotReady);
+            }
+        }
+
+        Ok(().into())
+    }
+
+    /// Send stream level window update
+    pub fn send_stream_window_update<T, B>(&mut self,
+                                           streams: &mut StreamMap,
+                                           dst: &mut Codec<T, B>)
+        -> Poll<(), ConnectionError>
+        where T: AsyncWrite,
+              B: Buf,
+    {
+        while let Some(id) = self.pending_window_updates.pop_front() {
+            let flow = streams.get_mut(&id)
+                .and_then(|state| state.recv_flow_control());
+
+
+            if let Some(flow) = flow {
+                if let Some(incr) = flow.peek_window_update() {
+                    let frame = frame::WindowUpdate::new(id, incr);
+
+                    if dst.start_send(frame.into())?.is_ready() {
+                        assert_eq!(Some(incr), flow.apply_window_update());
+                    } else {
+                        self.pending_window_updates.push_front(id);
+                        return Ok(Async::NotReady);
+                    }
+                }
+            }
+        }
+
+        Ok(().into())
+    }
+
+    fn reset(&mut self, _stream_id: StreamId, _reason: Reason) {
+        unimplemented!();
+    }
+}

src/proto/streams/send.rs

@@ -0,0 +1,206 @@
+use {frame, Peer, ConnectionError};
+use proto::*;
+use super::{Config, StreamMap};
+
+use error::User::*;
+
+use bytes::Buf;
+
+use std::collections::VecDeque;
+use std::marker::PhantomData;
+
+#[derive(Debug)]
+pub struct Send<P> {
+    /// Maximum number of locally initiated streams
+    max_streams: Option<usize>,
+
+    /// Current number of locally initiated streams
+    num_streams: usize,
+
+    /// Initial window size of locally initiated streams
+    init_window_sz: WindowSize,
+
+    /// Connection level flow control governing sent data
+    flow_control: state::FlowControl,
+
+    /// Holds the list of streams on which local window updates may be sent.
+    // XXX It would be cool if this didn't exist.
+    pending_window_updates: VecDeque<StreamId>,
+
+    /// When `poll_window_update` is not ready, then the calling task is saved to
+    /// be notified later. Access to poll_window_update must not be shared across tasks,
+    /// as we only track a single task (and *not* i.e. a task per stream id).
+    blocked: Option<task::Task>,
+
+    _p: PhantomData<P>,
+}
+
+impl<P: Peer> Send<P> {
+    pub fn new(config: &Config) -> Self {
+        Send {
+            max_streams: config.max_local_initiated,
+            num_streams: 0,
+            init_window_sz: config.init_local_window_sz,
+            flow_control: state::FlowControl::new(config.init_local_window_sz),
+            pending_window_updates: VecDeque::new(),
+            blocked: None,
+            _p: PhantomData,
+        }
+    }
+
+    /// Update state reflecting a new, locally opened stream
+    ///
+    /// Returns the stream state if successful. `None` if refused
+    pub fn open(&mut self, id: StreamId) -> Result<state::Stream, ConnectionError> {
+        try!(self.ensure_can_open(id));
+
+        if let Some(max) = self.max_streams {
+            if max <= self.num_streams {
+                return Err(Rejected.into());
+            }
+        }
+
+        // Increment the number of locally initiated streams
+        self.num_streams += 1;
+
+        Ok(state::Stream::default())
+    }
+
+    pub fn send_headers(&mut self, state: &mut state::Stream, eos: bool)
+        -> Result<(), ConnectionError>
+    {
+        state.send_open(self.init_window_sz, eos)
+    }
+
+    pub fn send_trailers(&mut self, _state: &mut state::Stream, _eos: bool)
+        -> Result<(), ConnectionError>
+    {
+        unimplemented!();
+    }
+
+    pub fn send_data<B: Buf>(&mut self,
+                             frame: &frame::Data<B>,
+                             state: &mut state::Stream)
+        -> Result<(), ConnectionError>
+    {
+        let sz = frame.payload().remaining();
+
+        if sz > MAX_WINDOW_SIZE as usize {
+            // TODO: handle overflow
+            unimplemented!();
+        }
+
+        let sz = sz as WindowSize;
+
+        // Make borrow checker happy
+        loop {
+            match state.send_flow_control() {
+                Some(flow) => {
+                    try!(self.flow_control.ensure_window(sz, FlowControlViolation));
+
+                    // Claim the window on the stream
+                    try!(flow.claim_window(sz, FlowControlViolation));
+
+                    // Claim the window on the connection
+                    self.flow_control.claim_window(sz, FlowControlViolation)
+                        .expect("local connection flow control error");
+
+                    break;
+                }
+                None => {}
+            }
+
+            if state.is_closed() {
+                return Err(InactiveStreamId.into())
+            } else {
+                return Err(UnexpectedFrameType.into())
+            }
+        }
+
+        if frame.is_end_stream() {
+            try!(state.send_close());
+        }
+
+        Ok(())
+    }
+
+    /// Get pending window updates
+    pub fn poll_window_update(&mut self, streams: &mut StreamMap)
+        -> Poll<WindowUpdate, ConnectionError>
+    {
+        // This biases connection window updates, which probably makes sense.
+        //
+        // TODO: We probably don't want to expose connection level updates
+        if let Some(incr) = self.flow_control.apply_window_update() {
+            return Ok(Async::Ready(WindowUpdate::new(StreamId::zero(), incr)));
+        }
+
+        // TODO this should probably account for stream priority?
+        let update = self.pending_window_updates.pop_front()
+            .and_then(|id| {
+                streams.get_mut(&id)
+                    .and_then(|state| state.send_flow_control())
+                    .and_then(|flow| flow.apply_window_update())
+                    .map(|incr| WindowUpdate::new(id, incr))
+            });
+
+        if let Some(update) = update {
+            return Ok(Async::Ready(update));
+        }
+
+        // Update the task.
+        //
+        // TODO: Extract this "gate" logic
+        self.blocked = Some(task::current());
+
+        return Ok(Async::NotReady);
+    }
+
+    pub fn recv_connection_window_update(&mut self, frame: frame::WindowUpdate)
+        -> Result<(), ConnectionError>
+    {
+        // TODO: Handle invalid increment
+        self.flow_control.expand_window(frame.size_increment());
+
+        if let Some(task) = self.blocked.take() {
+            task.notify();
+        }
+
+        Ok(())
+    }
+
+    pub fn recv_stream_window_update(&mut self,
+                                     frame: frame::WindowUpdate,
+                                     state: &mut state::Stream)
+        -> Result<(), ConnectionError>
+    {
+        if let Some(flow) = state.send_flow_control() {
+            // TODO: Handle invalid increment
+            flow.expand_window(frame.size_increment());
+        }
+
+        if let Some(task) = self.blocked.take() {
+            task.notify();
+        }
+
+        Ok(())
+    }
+
+    pub fn dec_num_streams(&mut self) {
+        self.num_streams -= 1;
+    }
+
+    /// Returns true if the local actor can initiate a stream with the given ID.
+    fn ensure_can_open(&self, id: StreamId) -> Result<(), ConnectionError> {
+        if P::is_server() {
+            // Servers cannot open streams. PushPromise must first be reserved.
+            return Err(UnexpectedFrameType.into());
+        }
+
+        if !id.is_client_initiated() {
+            return Err(InvalidStreamId.into());
+        }
+
+        Ok(())
+    }
+}