forked/h2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

chore: cargo fmt, clippy

Browse Source
This commit is contained in:
Gurwinder Singh
2019-08-17 09:07:32 +05:30
committed by Sean McArthur
parent e72d6dc189
commit f46840f3fa
43 changed files with 785 additions and 870 deletions
Download Patch File Download Diff File Expand all files Collapse all files

47
src/proto/connection.rs
View File

@@ -96,11 +96,11 @@ where
Connection {
state: State::Open,
error: None,
codec: codec,
codec,
go_away: GoAway::new(),
ping_pong: PingPong::new(),
settings: Settings::new(),
streams: streams,
streams,
_phantom: PhantomData,
}
}
@@ -210,11 +210,11 @@ where
// This will also handle flushing `self.codec`
ready!(self.streams.poll_complete(cx, &mut self.codec))?;
if self.error.is_some() || self.go_away.should_close_on_idle() {
if !self.streams.has_streams() {
self.go_away_now(Reason::NO_ERROR);
continue;
}
if (self.error.is_some() || self.go_away.should_close_on_idle())
&& !self.streams.has_streams()
{
self.go_away_now(Reason::NO_ERROR);
continue;
}
return Poll::Pending;
@@ -289,25 +289,22 @@ where
// The order here matters:
// - poll_go_away may buffer a graceful shutdown GOAWAY frame
// - If it has, we've also added a PING to be sent in poll_ready
match ready!(self.poll_go_away(cx)?) {
Some(reason) => {
if self.go_away.should_close_now() {
if self.go_away.is_user_initiated() {
// A user initiated abrupt shutdown shouldn't return
// the same error back to the user.
return Poll::Ready(Ok(()));
} else {
return Poll::Ready(Err(RecvError::Connection(reason)));
}
if let Some(reason) = ready!(self.poll_go_away(cx)?) {
if self.go_away.should_close_now() {
if self.go_away.is_user_initiated() {
// A user initiated abrupt shutdown shouldn't return
// the same error back to the user.
return Poll::Ready(Ok(()));
} else {
return Poll::Ready(Err(RecvError::Connection(reason)));
}
// Only NO_ERROR should be waiting for idle
debug_assert_eq!(
reason,
Reason::NO_ERROR,
"graceful GOAWAY should be NO_ERROR"
);
}
None => (),
// Only NO_ERROR should be waiting for idle
debug_assert_eq!(
reason,
Reason::NO_ERROR,
"graceful GOAWAY should be NO_ERROR"
);
}
ready!(self.poll_ready(cx))?;
@@ -364,7 +361,7 @@ where
}
None => {
log::trace!("codec closed");
self.streams.recv_eof(false).ok().expect("mutex poisoned");
self.streams.recv_eof(false).expect("mutex poisoned");
return Poll::Ready(Ok(()));
}
}

2
src/proto/go_away.rs
View File

@@ -137,7 +137,7 @@ impl GoAway {
}
let reason = frame.reason();
dst.buffer(frame.into()).ok().expect("invalid GOAWAY frame");
dst.buffer(frame.into()).expect("invalid GOAWAY frame");
return Poll::Ready(Some(Ok(reason)));
} else if self.should_close_now() {

9
src/proto/peer.rs
View File

@@ -17,7 +17,9 @@ pub(crate) trait Peer {
fn is_server() -> bool;
fn convert_poll_message(
pseudo: Pseudo, fields: HeaderMap, stream_id: StreamId
pseudo: Pseudo,
fields: HeaderMap,
stream_id: StreamId,
) -> Result<Self::Poll, RecvError>;
fn is_local_init(id: StreamId) -> bool {
@@ -54,7 +56,10 @@ impl Dyn {
}
pub fn convert_poll_message(
&self, pseudo: Pseudo, fields: HeaderMap, stream_id: StreamId
&self,
pseudo: Pseudo,
fields: HeaderMap,
stream_id: StreamId,
) -> Result<PollMessage, RecvError> {
if self.is_server() {
crate::server::Peer::convert_poll_message(pseudo, fields, stream_id)

10
src/proto/settings.rs
View File

@@ -1,7 +1,7 @@
use crate::codec::RecvError;
use crate::frame;
use crate::proto::*;
use std::task::{Poll, Context};
use std::task::{Context, Poll};
#[derive(Debug)]
pub(crate) struct Settings {
@@ -13,9 +13,7 @@ pub(crate) struct Settings {
impl Settings {
pub fn new() -> Self {
Settings {
pending: None,
}
Settings { pending: None }
}
pub fn recv_settings(&mut self, frame: frame::Settings) {
@@ -52,9 +50,7 @@ impl Settings {
let frame = frame::Settings::ack();
// Buffer the settings frame
dst.buffer(frame.into())
.ok()
.expect("invalid settings frame");
dst.buffer(frame.into()).expect("invalid settings frame");
log::trace!("ACK sent; applying settings");

30
src/proto/streams/buffer.rs
View File

@@ -27,17 +27,13 @@ struct Slot<T> {
impl<T> Buffer<T> {
pub fn new() -> Self {
Buffer {
slab: Slab::new(),
}
Buffer { slab: Slab::new() }
}
}
impl Deque {
pub fn new() -> Self {
Deque {
indices: None,
}
Deque { indices: None }
}
pub fn is_empty(&self) -> bool {
@@ -45,42 +41,36 @@ impl Deque {
}
pub fn push_back<T>(&mut self, buf: &mut Buffer<T>, value: T) {
let key = buf.slab.insert(Slot {
value,
next: None,
});
let key = buf.slab.insert(Slot { value, next: None });
match self.indices {
Some(ref mut idxs) => {
buf.slab[idxs.tail].next = Some(key);
idxs.tail = key;
},
}
None => {
self.indices = Some(Indices {
head: key,
tail: key,
});
},
}
}
}
pub fn push_front<T>(&mut self, buf: &mut Buffer<T>, value: T) {
let key = buf.slab.insert(Slot {
value,
next: None,
});
let key = buf.slab.insert(Slot { value, next: None });
match self.indices {
Some(ref mut idxs) => {
buf.slab[key].next = Some(idxs.head);
idxs.head = key;
},
}
None => {
self.indices = Some(Indices {
head: key,
tail: key,
});
},
}
}
}
@@ -97,8 +87,8 @@ impl Deque {
self.indices = Some(idxs);
}
return Some(slot.value);
},
Some(slot.value)
}
None => None,
}
}

22
src/proto/streams/counts.rs
View File

@@ -133,16 +133,18 @@ impl Counts {
// TODO: move this to macro?
pub fn transition_after(&mut self, mut stream: store::Ptr, is_reset_counted: bool) {
log::trace!("transition_after; stream={:?}; state={:?}; is_closed={:?}; \
pending_send_empty={:?}; buffered_send_data={}; \
num_recv={}; num_send={}",
stream.id,
stream.state,
stream.is_closed(),
stream.pending_send.is_empty(),
stream.buffered_send_data,
self.num_recv_streams,
self.num_send_streams);
log::trace!(
"transition_after; stream={:?}; state={:?}; is_closed={:?}; \
pending_send_empty={:?}; buffered_send_data={}; \
num_recv={}; num_send={}",
stream.id,
stream.state,
stream.is_closed(),
stream.pending_send.is_empty(),
stream.buffered_send_data,
self.num_recv_streams,
self.num_send_streams
);
if stream.is_closed() {
if !stream.is_pending_reset_expiration() {

2
src/proto/streams/flow_control.rs
View File

@@ -200,7 +200,6 @@ impl PartialEq<WindowSize> for Window {
}
}
impl PartialEq<Window> for WindowSize {
fn eq(&self, other: &Window) -> bool {
other.eq(self)
@@ -227,7 +226,6 @@ impl PartialOrd<Window> for WindowSize {
}
}
impl ::std::ops::SubAssign<WindowSize> for Window {
fn sub_assign(&mut self, other: WindowSize) {
self.0 -= other as i32;

2
src/proto/streams/mod.rs
View File

@@ -12,7 +12,7 @@ mod streams;
pub(crate) use self::prioritize::Prioritized;
pub(crate) use self::recv::Open;
pub(crate) use self::send::PollReset;
pub(crate) use self::streams::{StreamRef, OpaqueStreamRef, Streams};
pub(crate) use self::streams::{OpaqueStreamRef, StreamRef, Streams};
use self::buffer::Buffer;
use self::counts::Counts;

104
src/proto/streams/prioritize.rs
View File

@@ -1,5 +1,5 @@
use super::*;
use super::store::Resolve;
use super::*;
use crate::frame::{Reason, StreamId};
@@ -8,9 +8,9 @@ use crate::codec::UserError::*;
use bytes::buf::Take;
use futures::ready;
use std::{cmp, fmt, mem};
use std::io;
use std::task::{Context, Poll, Waker};
use std::{cmp, fmt, mem};
/// # Warning
///
@@ -81,7 +81,6 @@ impl Prioritize {
let mut flow = FlowControl::new();
flow.inc_window(config.remote_init_window_sz)
.ok()
.expect("invalid initial window size");
flow.assign_capacity(config.remote_init_window_sz);
@@ -92,7 +91,7 @@ impl Prioritize {
pending_send: store::Queue::new(),
pending_capacity: store::Queue::new(),
pending_open: store::Queue::new(),
flow: flow,
flow,
last_opened_id: StreamId::ZERO,
in_flight_data_frame: InFlightData::Nothing,
}
@@ -203,9 +202,7 @@ impl Prioritize {
// The stream has no capacity to send the frame now, save it but
// don't notify the connection task. Once additional capacity
// becomes available, the frame will be flushed.
stream
.pending_send
.push_back(buffer, frame.into());
stream.pending_send.push_back(buffer, frame.into());
}
Ok(())
@@ -216,7 +213,8 @@ impl Prioritize {
&mut self,
capacity: WindowSize,
stream: &mut store::Ptr,
counts: &mut Counts) {
counts: &mut Counts,
) {
log::trace!(
"reserve_capacity; stream={:?}; requested={:?}; effective={:?}; curr={:?}",
stream.id,
@@ -338,8 +336,8 @@ impl Prioritize {
&mut self,
inc: WindowSize,
store: &mut R,
counts: &mut Counts)
where
counts: &mut Counts,
) where
R: Resolve,
{
log::trace!("assign_connection_capacity; inc={}", inc);
@@ -419,11 +417,7 @@ impl Prioritize {
// TODO: Should prioritization factor into this?
let assign = cmp::min(conn_available, additional);
log::trace!(
" assigning; stream={:?}, capacity={}",
stream.id,
assign,
);
log::trace!(" assigning; stream={:?}, capacity={}", stream.id, assign,);
// Assign the capacity to the stream
stream.assign_capacity(assign);
@@ -440,16 +434,16 @@ impl Prioritize {
stream.send_flow.has_unavailable()
);
if stream.send_flow.available() < stream.requested_send_capacity {
if stream.send_flow.has_unavailable() {
// The stream requires additional capacity and the stream's
// window has available capacity, but the connection window
// does not.
//
// In this case, the stream needs to be queued up for when the
// connection has more capacity.
self.pending_capacity.push(stream);
}
if stream.send_flow.available() < stream.requested_send_capacity
&& stream.send_flow.has_unavailable()
{
// The stream requires additional capacity and the stream's
// window has available capacity, but the connection window
// does not.
//
// In this case, the stream needs to be queued up for when the
// connection has more capacity.
self.pending_capacity.push(stream);
}
// If data is buffered and the stream is not pending open, then
@@ -515,26 +509,26 @@ impl Prioritize {
if let Frame::Data(ref frame) = frame {
self.in_flight_data_frame = InFlightData::DataFrame(frame.payload().stream);
}
dst.buffer(frame).ok().expect("invalid frame");
dst.buffer(frame).expect("invalid frame");
// Ensure the codec is ready to try the loop again.
ready!(dst.poll_ready(cx))?;
// Because, always try to reclaim...
self.reclaim_frame(buffer, store, dst);
},
}
None => {
// Try to flush the codec.
ready!(dst.flush(cx))?;
// This might release a data frame...
if !self.reclaim_frame(buffer, store, dst) {
return Poll::Ready(Ok(()))
return Poll::Ready(Ok(()));
}
// No need to poll ready as poll_complete() does this for
// us...
},
}
}
}
}
@@ -603,11 +597,12 @@ impl Prioritize {
/// Push the frame to the front of the stream's deque, scheduling the
/// stream if needed.
fn push_back_frame<B>(&mut self,
frame: Frame<B>,
buffer: &mut Buffer<Frame<B>>,
stream: &mut store::Ptr)
{
fn push_back_frame<B>(
&mut self,
frame: Frame<B>,
buffer: &mut Buffer<Frame<B>>,
stream: &mut store::Ptr,
) {
// Push the frame to the front of the stream's deque
stream.pending_send.push_front(buffer, frame);
@@ -665,8 +660,11 @@ impl Prioritize {
loop {
match self.pending_send.pop(store) {
Some(mut stream) => {
log::trace!("pop_frame; stream={:?}; stream.state={:?}",
stream.id, stream.state);
log::trace!(
"pop_frame; stream={:?}; stream.state={:?}",
stream.id,
stream.state
);
// It's possible that this stream, besides having data to send,
// is also queued to send a reset, and thus is already in the queue
@@ -675,8 +673,11 @@ impl Prioritize {
// To be safe, we just always ask the stream.
let is_pending_reset = stream.is_pending_reset_expiration();
log::trace!(" --> stream={:?}; is_pending_reset={:?};",
stream.id, is_pending_reset);
log::trace!(
" --> stream={:?}; is_pending_reset={:?};",
stream.id,
is_pending_reset
);
let frame = match stream.pending_send.pop_front(buffer) {
Some(Frame::Data(mut frame)) => {
@@ -715,9 +716,7 @@ impl Prioritize {
// happen if the remote reduced the stream
// window. In this case, we need to buffer the
// frame and wait for a window update...
stream
.pending_send
.push_front(buffer, frame.into());
stream.pending_send.push_front(buffer, frame.into());
continue;
}
@@ -726,7 +725,8 @@ impl Prioritize {
let len = cmp::min(sz, max_len);
// Only send up to the stream's window capacity
let len = cmp::min(len, stream_capacity.as_size() as usize) as WindowSize;
let len =
cmp::min(len, stream_capacity.as_size() as usize) as WindowSize;
// There *must* be be enough connection level
// capacity at this point.
@@ -761,20 +761,18 @@ impl Prioritize {
frame.set_end_stream(false);
}
Frame::Data(frame.map(|buf| {
Prioritized {
inner: buf.take(len),
end_of_stream: eos,
stream: stream.key(),
}
Frame::Data(frame.map(|buf| Prioritized {
inner: buf.take(len),
end_of_stream: eos,
stream: stream.key(),
}))
},
Some(frame) => frame.map(|_|
}
Some(frame) => frame.map(|_| {
unreachable!(
"Frame::map closure will only be called \
on DATA frames."
)
),
)
}),
None => {
if let Some(reason) = stream.state.get_scheduled_reset() {
stream.state.set_reset(reason);
@@ -814,7 +812,7 @@ impl Prioritize {
counts.transition_after(stream, is_pending_reset);
return Some(frame);
},
}
None => return None,
}
}

140
src/proto/streams/recv.rs
View File

@@ -1,15 +1,15 @@
use std::task::Context;
use super::*;
use crate::{frame, proto};
use crate::codec::{RecvError, UserError};
use crate::frame::{Reason, DEFAULT_INITIAL_WINDOW_SIZE};
use crate::{frame, proto};
use std::task::Context;
use http::{HeaderMap, Response, Request, Method};
use futures::ready;
use http::{HeaderMap, Method, Request, Response};
use std::io;
use std::time::{Duration, Instant};
use std::task::{Poll, Waker};
use std::time::{Duration, Instant};
#[derive(Debug)]
pub(super) struct Recv {
@@ -98,7 +98,7 @@ impl Recv {
Recv {
init_window_sz: config.local_init_window_sz,
flow: flow,
flow,
in_flight_data: 0 as WindowSize,
next_stream_id: Ok(next_stream_id.into()),
pending_window_updates: store::Queue::new(),
@@ -186,8 +186,9 @@ impl Recv {
return Err(RecvError::Stream {
id: stream.id,
reason: Reason::PROTOCOL_ERROR,
}.into())
},
}
.into());
}
};
stream.content_length = ContentLength::Remaining(content_length);
@@ -215,7 +216,7 @@ impl Recv {
let mut res = frame::Headers::new(
stream.id,
frame::Pseudo::response(::http::StatusCode::REQUEST_HEADER_FIELDS_TOO_LARGE),
HeaderMap::new()
HeaderMap::new(),
);
res.set_end_stream();
Err(RecvHeaderBlockError::Oversize(Some(res)))
@@ -226,7 +227,9 @@ impl Recv {
let stream_id = frame.stream_id();
let (pseudo, fields) = frame.into_parts();
let message = counts.peer().convert_poll_message(pseudo, fields, stream_id)?;
let message = counts
.peer()
.convert_poll_message(pseudo, fields, stream_id)?;
// Push the frame onto the stream's recv buffer
stream
@@ -246,9 +249,7 @@ impl Recv {
/// Called by the server to get the request
///
/// TODO: Should this fn return `Result`?
pub fn take_request(&mut self, stream: &mut store::Ptr)
-> Request<()>
{
pub fn take_request(&mut self, stream: &mut store::Ptr) -> Request<()> {
use super::peer::PollMessage::*;
match stream.pending_recv.pop_front(&mut self.buffer) {
@@ -261,20 +262,19 @@ impl Recv {
pub fn poll_pushed(
&mut self,
cx: &Context,
stream: &mut store::Ptr
stream: &mut store::Ptr,
) -> Poll<Option<Result<(Request<()>, store::Key), proto::Error>>> {
use super::peer::PollMessage::*;
let mut ppp = stream.pending_push_promises.take();
let pushed = ppp.pop(stream.store_mut()).map(
|mut pushed| match pushed.pending_recv.pop_front(&mut self.buffer) {
Some(Event::Headers(Server(headers))) =>
(headers, pushed.key()),
let pushed = ppp.pop(stream.store_mut()).map(|mut pushed| {
match pushed.pending_recv.pop_front(&mut self.buffer) {
Some(Event::Headers(Server(headers))) => (headers, pushed.key()),
// When frames are pushed into the queue, it is verified that
// the first frame is a HEADERS frame.
_ => panic!("Headers not set on pushed stream")
_ => panic!("Headers not set on pushed stream"),
}
);
});
stream.pending_push_promises = ppp;
if let Some(p) = pushed {
Poll::Ready(Some(Ok(p)))
@@ -301,14 +301,14 @@ impl Recv {
// If the buffer is not empty, then the first frame must be a HEADERS
// frame or the user violated the contract.
match stream.pending_recv.pop_front(&mut self.buffer) {
Some(Event::Headers(Client(response))) => Poll::Ready(Ok(response.into())),
Some(Event::Headers(Client(response))) => Poll::Ready(Ok(response)),
Some(_) => panic!("poll_response called after response returned"),
None => {
stream.state.ensure_recv_open()?;
stream.recv_task = Some(cx.waker().clone());
Poll::Pending
},
}
}
}
@@ -341,11 +341,7 @@ impl Recv {
}
/// Releases capacity of the connection
pub fn release_connection_capacity(
&mut self,
capacity: WindowSize,
task: &mut Option<Waker>,
) {
pub fn release_connection_capacity(&mut self, capacity: WindowSize, task: &mut Option<Waker>) {
log::trace!(
"release_connection_capacity; size={}, connection in_flight_data={}",
capacity,
@@ -386,7 +382,6 @@ impl Recv {
// Assign capacity to stream
stream.recv_flow.assign_capacity(capacity);
if stream.recv_flow.unclaimed_capacity().is_some() {
// Queue the stream for sending the WINDOW_UPDATE frame.
self.pending_window_updates.push(stream);
@@ -400,11 +395,7 @@ impl Recv {
}
/// Release any unclaimed capacity for a closed stream.
pub fn release_closed_capacity(
&mut self,
stream: &mut store::Ptr,
task: &mut Option<Waker>,
) {
pub fn release_closed_capacity(&mut self, stream: &mut store::Ptr, task: &mut Option<Waker>) {
debug_assert_eq!(stream.ref_count, 0);
if stream.in_flight_recv_data == 0 {
@@ -417,10 +408,7 @@ impl Recv {
stream.in_flight_recv_data,
);
self.release_connection_capacity(
stream.in_flight_recv_data,
task,
);
self.release_connection_capacity(stream.in_flight_recv_data, task);
stream.in_flight_recv_data = 0;
self.clear_recv_buffer(stream);
@@ -485,9 +473,7 @@ impl Recv {
return false;
}
stream
.pending_recv
.is_empty()
stream.pending_recv.is_empty()
}
pub fn recv_data(
@@ -522,7 +508,6 @@ impl Recv {
stream.recv_flow.window_size()
);
if is_ignoring_frame {
log::trace!(
"recv_data; frame ignored on locally reset {:?} for some time",
@@ -609,7 +594,7 @@ impl Recv {
// the capacity as available to be reclaimed. When the available
// capacity meets a threshold, a WINDOW_UPDATE is then sent.
self.release_connection_capacity(sz, &mut None);
return Ok(());
Ok(())
}
pub fn consume_connection_window(&mut self, sz: WindowSize) -> Result<(), RecvError> {
@@ -670,7 +655,7 @@ impl Recv {
// 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) => {},
Ok(0) => {}
otherwise => {
proto_err!(stream:
"recv_push_promise; promised request has content-length {:?}; promised_id={:?}",
@@ -681,7 +666,7 @@ impl Recv {
id: promised_id,
reason: Reason::PROTOCOL_ERROR,
});
},
}
}
}
// "The server MUST include a method in the :method pseudo-header field
@@ -699,7 +684,9 @@ impl Recv {
});
}
use super::peer::PollMessage::*;
stream.pending_recv.push_back(&mut self.buffer, Event::Headers(Server(req)));
stream
.pending_recv
.push_back(&mut self.buffer, Event::Headers(Server(req)));
stream.notify_recv();
Ok(())
}
@@ -707,14 +694,17 @@ impl Recv {
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;
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 {
if id >= next {
log::debug!("stream ID implicitly closed, PROTOCOL_ERROR; stream={:?}", id);
log::debug!(
"stream ID implicitly closed, PROTOCOL_ERROR; stream={:?}",
id
);
return Err(Reason::PROTOCOL_ERROR);
}
}
@@ -726,7 +716,9 @@ impl Recv {
/// Handle remote sending an explicit RST_STREAM.
pub fn recv_reset(&mut self, frame: frame::Reset, stream: &mut Stream) {
// Notify the stream
stream.state.recv_reset(frame.reason(), stream.is_pending_send);
stream
.state
.recv_reset(frame.reason(), stream.is_pending_send);
stream.notify_send();
stream.notify_recv();
@@ -777,10 +769,7 @@ impl Recv {
pub fn may_have_created_stream(&self, id: StreamId) -> bool {
if let Ok(next_id) = self.next_stream_id {
// Peer::is_local_init should have been called beforehand
debug_assert_eq!(
id.is_server_initiated(),
next_id.is_server_initiated(),
);
debug_assert_eq!(id.is_server_initiated(), next_id.is_server_initiated(),);
id < next_id
} else {
true
@@ -788,9 +777,7 @@ impl Recv {
}
/// Returns true if the remote peer can reserve a stream with the given ID.
pub fn ensure_can_reserve(&self)
-> Result<(), RecvError>
{
pub fn ensure_can_reserve(&self) -> Result<(), RecvError> {
if !self.is_push_enabled {
proto_err!(conn: "recv_push_promise: push is disabled");
return Err(RecvError::Connection(Reason::PROTOCOL_ERROR));
@@ -800,11 +787,7 @@ impl Recv {
}
/// Add a locally reset stream to queue to be eventually reaped.
pub fn enqueue_reset_expiration(
&mut self,
stream: &mut store::Ptr,
counts: &mut Counts,
) {
pub fn enqueue_reset_expiration(&mut self, stream: &mut store::Ptr, counts: &mut Counts) {
if !stream.state.is_local_reset() || stream.is_pending_reset_expiration() {
return;
}
@@ -843,9 +826,7 @@ impl Recv {
let frame = frame::Reset::new(stream_id, Reason::REFUSED_STREAM);
// Buffer the frame
dst.buffer(frame.into())
.ok()
.expect("invalid RST_STREAM frame");
dst.buffer(frame.into()).expect("invalid RST_STREAM frame");
}
self.refused = None;
@@ -864,11 +845,12 @@ impl Recv {
}
}
pub fn clear_queues(&mut self,
clear_pending_accept: bool,
store: &mut Store,
counts: &mut Counts)
{
pub fn clear_queues(
&mut self,
clear_pending_accept: bool,
store: &mut Store,
counts: &mut Counts,
) {
self.clear_stream_window_update_queue(store, counts);
self.clear_all_reset_streams(store, counts);
@@ -921,7 +903,7 @@ impl Recv {
/// Send connection level window update
fn send_connection_window_update<T, B>(
&mut self,
cx: &mut Context,
cx: &mut Context,
dst: &mut Codec<T, Prioritized<B>>,
) -> Poll<io::Result<()>>
where
@@ -936,13 +918,11 @@ impl Recv {
// Buffer the WINDOW_UPDATE frame
dst.buffer(frame.into())
.ok()
.expect("invalid WINDOW_UPDATE frame");
// Update flow control
self.flow
.inc_window(incr)
.ok()
.expect("unexpected flow control state");
}
@@ -992,14 +972,12 @@ impl Recv {
// Buffer it
dst.buffer(frame.into())
.ok()
.expect("invalid WINDOW_UPDATE frame");
// Update flow control
stream
.recv_flow
.inc_window(incr)
.ok()
.expect("unexpected flow control state");
}
})
@@ -1010,7 +988,11 @@ impl Recv {
self.pending_accept.pop(store).map(|ptr| ptr.key())
}
pub fn poll_data(&mut self, cx: &Context, stream: &mut Stream) -> Poll<Option<Result<Bytes, proto::Error>>> {
pub fn poll_data(
&mut self,
cx: &Context,
stream: &mut Stream,
) -> Poll<Option<Result<Bytes, proto::Error>>> {
// TODO: Return error when the stream is reset
match stream.pending_recv.pop_front(&mut self.buffer) {
Some(Event::Data(payload)) => Poll::Ready(Some(Ok(payload))),
@@ -1030,7 +1012,7 @@ impl Recv {
// No more data frames
Poll::Ready(None)
},
}
None => self.schedule_recv(cx, stream),
}
}
@@ -1047,12 +1029,16 @@ impl Recv {
stream.pending_recv.push_front(&mut self.buffer, event);
Poll::Pending
},
}
None => self.schedule_recv(cx, stream),
}
}
fn schedule_recv<T>(&mut self, cx: &Context, stream: &mut Stream) -> Poll<Option<Result<T, proto::Error>>> {
fn schedule_recv<T>(
&mut self,
cx: &Context,
stream: &mut Stream,
) -> Poll<Option<Result<T, proto::Error>>> {
if stream.state.ensure_recv_open()? {
// Request to get notified once more frames arrive
stream.recv_task = Some(cx.waker().clone());
@@ -1112,7 +1098,7 @@ fn parse_u64(src: &[u8]) -> Result<u64, ()> {
}
ret *= 10;
ret += (d - b'0') as u64;
ret += u64::from(d - b'0');
}
Ok(ret)

8
src/proto/streams/send.rs
View File

@@ -325,13 +325,7 @@ impl Send {
if let Err(e) = self.prioritize.recv_stream_window_update(sz, stream) {
log::debug!("recv_stream_window_update !!; err={:?}", e);
self.send_reset(
Reason::FLOW_CONTROL_ERROR.into(),
buffer,
stream,
counts,
task,
);
self.send_reset(Reason::FLOW_CONTROL_ERROR, buffer, stream, counts, task);
return Err(e);
}

148
src/proto/streams/state.rs
View File

@@ -1,7 +1,7 @@
use std::io;
use crate::codec::{RecvError, UserError};
use crate::codec::UserError::*;
use crate::codec::{RecvError, UserError};
use crate::frame::Reason;
use crate::proto::{self, PollReset};
@@ -94,37 +94,40 @@ impl State {
let local = Streaming;
self.inner = match self.inner {
Idle => if eos {
HalfClosedLocal(AwaitingHeaders)
} else {
Open {
local,
remote: AwaitingHeaders,
Idle => {
if eos {
HalfClosedLocal(AwaitingHeaders)
} else {
Open {
local,
remote: AwaitingHeaders,
}
}
},
}
Open {
local: AwaitingHeaders,
remote,
} => if eos {
HalfClosedLocal(remote)
} else {
Open {
local,
remote,
} => {
if eos {
HalfClosedLocal(remote)
} else {
Open { local, remote }
}
},
HalfClosedRemote(AwaitingHeaders) => if eos {
Closed(Cause::EndStream)
} else {
HalfClosedRemote(local)
},
}
HalfClosedRemote(AwaitingHeaders) => {
if eos {
Closed(Cause::EndStream)
} else {
HalfClosedRemote(local)
}
}
_ => {
// All other transitions result in a protocol error
return Err(UnexpectedFrameType);
},
}
};
return Ok(());
Ok(())
}
/// Opens the receive-half of the stream when a HEADERS frame is received.
@@ -146,7 +149,7 @@ impl State {
remote,
}
}
},
}
ReservedRemote => {
initial = true;
@@ -155,31 +158,32 @@ impl State {
} else {
HalfClosedLocal(Streaming)
}
},
}
Open {
local,
remote: AwaitingHeaders,
} => if eos {
HalfClosedRemote(local)
} else {
Open {
local,
remote,
} => {
if eos {
HalfClosedRemote(local)
} else {
Open { local, remote }
}
},
HalfClosedLocal(AwaitingHeaders) => if eos {
Closed(Cause::EndStream)
} else {
HalfClosedLocal(remote)
},
}
HalfClosedLocal(AwaitingHeaders) => {
if eos {
Closed(Cause::EndStream)
} else {
HalfClosedLocal(remote)
}
}
state => {
// All other transitions result in a protocol error
proto_err!(conn: "recv_open: in unexpected state {:?}", state);
return Err(RecvError::Connection(Reason::PROTOCOL_ERROR));
},
}
};
return Ok(initial);
Ok(initial)
}
/// Transition from Idle -> ReservedRemote
@@ -188,7 +192,7 @@ impl State {
Idle => {
self.inner = ReservedRemote;
Ok(())
},
}
state => {
proto_err!(conn: "reserve_remote: in unexpected state {:?}", state);
Err(RecvError::Connection(Reason::PROTOCOL_ERROR))
@@ -199,19 +203,17 @@ impl State {
/// Indicates that the remote side will not send more data to the local.
pub fn recv_close(&mut self) -> Result<(), RecvError> {
match self.inner {
Open {
local, ..
} => {
Open { local, .. } => {
// The remote side will continue to receive data.
log::trace!("recv_close: Open => HalfClosedRemote({:?})", local);
self.inner = HalfClosedRemote(local);
Ok(())
},
}
HalfClosedLocal(..) => {
log::trace!("recv_close: HalfClosedLocal => Closed");
self.inner = Closed(Cause::EndStream);
Ok(())
},
}
state => {
proto_err!(conn: "recv_close: in unexpected state {:?}", state);
Err(RecvError::Connection(Reason::PROTOCOL_ERROR))
@@ -228,7 +230,7 @@ impl State {
match self.inner {
// If the stream is already in a `Closed` state, do nothing,
// provided that there are no frames still in the send queue.
Closed(..) if !queued => {},
Closed(..) if !queued => {}
// A notionally `Closed` stream may still have queued frames in
// the following cases:
//
@@ -246,11 +248,12 @@ impl State {
state => {
log::trace!(
"recv_reset; reason={:?}; state={:?}; queued={:?}",
reason, state, queued
reason,
state,
queued
);
self.inner = Closed(Cause::Proto(reason));
},
}
}
}
@@ -259,20 +262,20 @@ impl State {
use crate::proto::Error::*;
match self.inner {
Closed(..) => {},
Closed(..) => {}
_ => {
log::trace!("recv_err; err={:?}", err);
self.inner = Closed(match *err {
Proto(reason) => Cause::LocallyReset(reason),
Io(..) => Cause::Io,
});
},
}
}
}
pub fn recv_eof(&mut self) {
match self.inner {
Closed(..) => {},
Closed(..) => {}
s => {
log::trace!("recv_eof; state={:?}", s);
self.inner = Closed(Cause::Io);
@@ -283,17 +286,15 @@ impl State {
/// Indicates that the local side will not send more data to the local.
pub fn send_close(&mut self) {
match self.inner {
Open {
remote, ..
} => {
Open { remote, .. } => {
// The remote side will continue to receive data.
log::trace!("send_close: Open => HalfClosedLocal({:?})", remote);
self.inner = HalfClosedLocal(remote);
},
}
HalfClosedRemote(..) => {
log::trace!("send_close: HalfClosedRemote => Closed");
self.inner = Closed(Cause::EndStream);
},
}
state => panic!("send_close: unexpected state {:?}", state),
}
}
@@ -343,8 +344,7 @@ impl State {
pub fn is_send_streaming(&self) -> bool {
match self.inner {
Open {
local: Streaming,
..
local: Streaming, ..
} => true,
HalfClosedRemote(Streaming) => true,
_ => false,
@@ -368,8 +368,7 @@ impl State {
pub fn is_recv_streaming(&self) -> bool {
match self.inner {
Open {
remote: Streaming,
..
remote: Streaming, ..
} => true,
HalfClosedLocal(Streaming) => true,
_ => false,
@@ -407,12 +406,11 @@ impl State {
pub fn ensure_recv_open(&self) -> Result<bool, proto::Error> {
// TODO: Is this correct?
match self.inner {
Closed(Cause::Proto(reason)) |
Closed(Cause::LocallyReset(reason)) |
Closed(Cause::Scheduled(reason)) => Err(proto::Error::Proto(reason)),
Closed(Cause::Proto(reason))
| Closed(Cause::LocallyReset(reason))
| Closed(Cause::Scheduled(reason)) => Err(proto::Error::Proto(reason)),
Closed(Cause::Io) => Err(proto::Error::Io(io::ErrorKind::BrokenPipe.into())),
Closed(Cause::EndStream) |
HalfClosedRemote(..) => Ok(false),
Closed(Cause::EndStream) | HalfClosedRemote(..) => Ok(false),
_ => Ok(true),
}
}
@@ -420,15 +418,15 @@ impl State {
/// Returns a reason if the stream has been reset.
pub(super) fn ensure_reason(&self, mode: PollReset) -> Result<Option<Reason>, crate::Error> {
match self.inner {
Closed(Cause::Proto(reason)) |
Closed(Cause::LocallyReset(reason)) |
Closed(Cause::Scheduled(reason)) => Ok(Some(reason)),
Closed(Cause::Proto(reason))
| Closed(Cause::LocallyReset(reason))
| Closed(Cause::Scheduled(reason)) => Ok(Some(reason)),
Closed(Cause::Io) => Err(proto::Error::Io(io::ErrorKind::BrokenPipe.into()).into()),
Open { local: Streaming, .. } |
HalfClosedRemote(Streaming) => match mode {
PollReset::AwaitingHeaders => {
Err(UserError::PollResetAfterSendResponse.into())
},
Open {
local: Streaming, ..
}
| HalfClosedRemote(Streaming) => match mode {
PollReset::AwaitingHeaders => Err(UserError::PollResetAfterSendResponse.into()),
PollReset::Streaming => Ok(None),
},
_ => Ok(None),
@@ -438,9 +436,7 @@ impl State {
impl Default for State {
fn default() -> State {
State {
inner: Inner::Idle,
}
State { inner: Inner::Idle }
}
}

30
src/proto/streams/store.rs
View File

@@ -118,9 +118,7 @@ impl Store {
use self::indexmap::map::Entry::*;
match self.ids.entry(id) {
Occupied(e) => Entry::Occupied(OccupiedEntry {
ids: e,
}),
Occupied(e) => Entry::Occupied(OccupiedEntry { ids: e }),
Vacant(e) => Entry::Vacant(VacantEntry {
ids: e,
slab: &mut self.slab,
@@ -143,10 +141,7 @@ impl Store {
};
f(Ptr {
key: Key {
index,
stream_id,
},
key: Key { index, stream_id },
store: self,
})?;
@@ -167,10 +162,7 @@ impl Store {
impl Resolve for Store {
fn resolve(&mut self, key: Key) -> Ptr {
Ptr {
key: key,
store: self,
}
Ptr { key, store: self }
}
}
@@ -267,14 +259,14 @@ where
// Update the tail pointer
idxs.tail = stream.key();
},
}
None => {
log::trace!(" -> first entry");
self.indices = Some(store::Indices {
head: stream.key(),
tail: stream.key(),
});
},
}
}
true
@@ -356,7 +348,7 @@ impl<'a> Ptr<'a> {
impl<'a> Resolve for Ptr<'a> {
fn resolve(&mut self, key: Key) -> Ptr {
Ptr {
key: key,
key,
store: &mut *self.store,
}
}
@@ -388,10 +380,7 @@ impl<'a> OccupiedEntry<'a> {
pub fn key(&self) -> Key {
let stream_id = *self.ids.key();
let index = *self.ids.get();
Key {
index,
stream_id,
}
Key { index, stream_id }
}
}
@@ -406,9 +395,6 @@ impl<'a> VacantEntry<'a> {
// Insert the handle in the ID map
self.ids.insert(index);
Key {
index,
stream_id,
}
Key { index, stream_id }
}
}

24
src/proto/streams/stream.rs
View File

@@ -1,8 +1,8 @@
use super::*;
use std::task::{Context, Waker};
use std::time::Instant;
use std::usize;
use std::task::{Context, Waker};
/// Tracks Stream related state
///
@@ -133,23 +133,17 @@ pub(super) struct NextOpen;
pub(super) struct NextResetExpire;
impl Stream {
pub fn new(
id: StreamId,
init_send_window: WindowSize,
init_recv_window: WindowSize,
) -> Stream {
pub fn new(id: StreamId, init_send_window: WindowSize, init_recv_window: WindowSize) -> Stream {
let mut send_flow = FlowControl::new();
let mut recv_flow = FlowControl::new();
recv_flow
.inc_window(init_recv_window)
.ok()
.expect("invalid initial receive window");
recv_flow.assign_capacity(init_recv_window);
send_flow
.inc_window(init_send_window)
.ok()
.expect("invalid initial send window size");
Stream {
@@ -161,7 +155,7 @@ impl Stream {
// ===== Fields related to sending =====
next_pending_send: None,
is_pending_send: false,
send_flow: send_flow,
send_flow,
requested_send_capacity: 0,
buffered_send_data: 0,
send_task: None,
@@ -175,7 +169,7 @@ impl Stream {
// ===== Fields related to receiving =====
next_pending_accept: None,
is_pending_accept: false,
recv_flow: recv_flow,
recv_flow,
in_flight_recv_data: 0,
next_window_update: None,
is_pending_window_update: false,
@@ -247,8 +241,12 @@ impl Stream {
self.send_capacity_inc = true;
self.send_flow.assign_capacity(capacity);
log::trace!(" assigned capacity to stream; available={}; buffered={}; id={:?}",
self.send_flow.available(), self.buffered_send_data, self.id);
log::trace!(
" assigned capacity to stream; available={}; buffered={}; id={:?}",
self.send_flow.available(),
self.buffered_send_data,
self.id
);
// Only notify if the capacity exceeds the amount of buffered data
if self.send_flow.available() > self.buffered_send_data {
@@ -265,7 +263,7 @@ impl Stream {
None => return Err(()),
},
ContentLength::Head => return Err(()),
_ => {},
_ => {}
}
Ok(())