gsadmin/noq
1
0
Fork 0
mirror of https://github.com/n0-computer/noq.git synced 2026-10-03 12:40:46 +00:00
Code Issues Packages Projects Releases Wiki Activity

Extensive bugfixing, HTTP/0.9 server

Browse Source
This commit is contained in:
Benjamin Saunders
2018-04-25 19:38:55 -07:00
parent 59876386ab
commit 769eb512d3
6 changed files with 264 additions and 144 deletions
Download Patch File Download Diff File Expand all files Collapse all files
quicr/Cargo.toml
+1
View File
@@ -14,6 +14,7 @@ futures = "0.1.21"
fnv = "1.0.6"
openssl = "0.10.6"
failure = "0.1"
bytes = "0.4.6"
[dev-dependencies]
tokio = "0.1.5"
quicr/examples/client.rs
-3
View File
@@ -59,11 +59,9 @@ fn run(log: Logger) -> Result<()> {
endpoint.connect(&remote, url.host_str().map(|x| x.as_bytes()))
.map_err(|e| format_err!("failed to connect: {}", e))
.and_then(|(conn, _)| {
eprintln!("connected!");
conn.open_bi().map_err(|e| format_err!("failed to open stream: {}", e))
})
.and_then(|(send, recv)| {
eprintln!("opened a stream");
tokio::io::write_all(send, request.as_bytes()).map_err(|e| format_err!("failed to send request: {}", e))
.map(move |(send, _)| (send, recv))
})
@@ -73,7 +71,6 @@ fn run(log: Logger) -> Result<()> {
.map(|data| {
io::stdout().write_all(&data).unwrap();
io::stdout().flush().unwrap();
eprintln!("done")
})
).map_err(|e| e.into_inner().unwrap())?;
quicr/examples/server.rs
+63 -17
View File
@@ -14,6 +14,9 @@ use std::net::UdpSocket;
use std::fs::File;
use std::io::Read;
use std::fmt;
use std::path::{self, Path, PathBuf};
use std::str;
use std::rc::Rc;
use futures::{Future, Stream};
use tokio::executor::current_thread::{self, CurrentThread};
@@ -62,6 +65,10 @@ fn main() {
}
fn run(log: Logger) -> Result<()> {
let root = ::std::env::args().nth(1).ok_or(format_err!("missing root argument"))?;
let root = Rc::new(Path::new(&root).to_owned());
if !root.exists() { bail!("root path does not exist"); }
let socket = UdpSocket::bind("[::]:4433")?;
let mut protocols = Vec::new();
const PROTO: &[u8] = b"hq-11";
@@ -96,25 +103,41 @@ fn run(log: Logger) -> Result<()> {
let mut executor = CurrentThread::new_with_park(timer);
executor.spawn(incoming.for_each(move |conn| {
info!(log, "got connection");
let quicr::NewConnection { incoming, address, .. } = conn;
info!(log, "got connection"; "remote" => %address);
let root = root.clone();
let log = log.clone();
let log2 = log.clone();
current_thread::spawn(
conn.incoming.into_future()
.map_err(|_| unreachable!())
.and_then(|(stream, _)| match stream {
Some(quicr::NewStream::Bi(send, recv)) => Ok((send, recv)),
Some(quicr::NewStream::Uni(_)) => unreachable!(),
None => Err(format_err!("no request submitted")),
incoming
.map_err(move |e| info!(log2, "connection terminated"; "remote" => %address, "reason" => %e))
.and_then(|stream| { match stream {
quicr::NewStream::Bi(send, recv) => Ok((send, recv)),
quicr::NewStream::Uni(_) => unreachable!(),
}})
.for_each(move |(send, recv)| {
let root = root.clone();
let log = log.clone();
let log2 = log.clone();
let log3 = log.clone();
current_thread::spawn(
recv.read_to_end(64 * 1024)
.map_err(|e| format_err!("failed reading request: {}", e))
.map(move |data| (send, data))
.and_then(move |(send, req)| {
info!(log, "got request"; "remote" => %address);
let resp = process_request(&root, &req).unwrap_or_else(move |e| {
error!(log, "failed to process request"; "reaosn" => %e.pretty());
format!("failed to process request: {}\n", e.pretty()).into_bytes().into()
});
tokio::io::write_all(send, resp).map_err(|e| format_err!("failed to send response: {}", e))
})
.and_then(|(send, _)| tokio::io::shutdown(send).map_err(|e| format_err!("failed to shutdown stream: {}", e)))
.map(move |_| info!(log3, "request complete"; "remote" => %address))
.map_err(move |e| error!(log2, "request failed"; "reason" => %e.pretty()))
);
Ok(())
})
.and_then(|(send, recv)| recv.read_to_end(64 * 1024)
.map_err(|e| format_err!("failed reading request: {}", e))
.map(move |data| (send, data)))
.and_then(move |(send, _)| {
eprintln!("processing request");
tokio::io::write_all(send, b"hello\n").map_err(|e| format_err!("failed to send response: {}", e))
})
.and_then(|(send, _)| tokio::io::shutdown(send).map_err(|e| format_err!("failed to shutdown stream: {}", e)))
.map(|_| eprintln!("done"))
.map_err(|e| eprintln!("failed: {}", e))
);
Ok(())
}));
@@ -123,3 +146,26 @@ fn run(log: Logger) -> Result<()> {
Ok(())
}
fn process_request(root: &Path, x: &[u8]) -> Result<Box<[u8]>> {
if x.len() < 4 || &x[0..4] != b"GET " { bail!("missing GET"); }
if x[4..].len() < 2 || &x[x.len()-2..] != b"\r\n" { bail!("missing \\r\\n"); }
let path = str::from_utf8(&x[4..x.len()-2]).context("path is malformed UTF-8")?;
let path = Path::new(&path);
let mut real_path = PathBuf::from(root);
let mut components = path.components();
match components.next() {
Some(path::Component::RootDir) => {}
_ => { bail!("path must be absolute"); }
}
for c in components {
match c {
path::Component::Normal(x) => { real_path.push(x); }
x => { bail!("illegal component in path: {:?}", x); }
}
}
let mut file = File::open(real_path)?;
let mut data = Vec::new();
file.read_to_end(&mut data).context("failed reading file")?;
Ok(data.into())
}
quicr/src/lib.rs
+130 -69
View File
@@ -10,6 +10,7 @@ extern crate fnv;
extern crate openssl;
#[macro_use]
extern crate failure;
extern crate bytes;
use std::{io, mem};
use std::net::{SocketAddr, SocketAddrV6};
@@ -29,10 +30,11 @@ use futures::task::{self, Task};
use futures::stream::FuturesUnordered;
use fnv::{FnvHashMap, FnvHashSet};
use openssl::ssl;
use bytes::Bytes;
use quicr::{Directionality, StreamId, ConnectionHandle};
pub use quicr::{Config, ListenConfig, PersistentState, ConnectionError, TransportError, ReadError, WriteError};
pub use quicr::{Config, ListenConfig, PersistentState, ConnectionError, TransportError};
#[derive(Debug, Fail)]
pub enum Error {
@@ -66,14 +68,15 @@ struct Pending {
blocked_writers: FnvHashMap<StreamId, Task>,
blocked_readers: FnvHashMap<StreamId, Task>,
connecting: Option<oneshot::Sender<Option<ConnectionError>>>,
uni_opening: VecDeque<oneshot::Sender<Result<StreamId, ConnectionLost>>>,
bi_opening: VecDeque<oneshot::Sender<Result<StreamId, ConnectionLost>>>,
uni_opening: VecDeque<oneshot::Sender<Result<StreamId, ConnectionError>>>,
bi_opening: VecDeque<oneshot::Sender<Result<StreamId, ConnectionError>>>,
cancel_loss_detect: Option<oneshot::Sender<()>>,
cancel_idle: Option<oneshot::Sender<()>>,
incoming_streams: VecDeque<StreamId>,
incoming_streams_reader: Option<Task>,
remote_recv_streams: FnvHashSet<StreamId>,
finishing: FnvHashMap<StreamId, oneshot::Sender<()>>,
finishing: FnvHashMap<StreamId, oneshot::Sender<Option<ConnectionError>>>,
error: Option<ConnectionError>,
}
impl Pending {
@@ -89,7 +92,33 @@ impl Pending {
incoming_streams_reader: None,
remote_recv_streams: FnvHashSet::default(),
finishing: FnvHashMap::default(),
error: None,
}}
pub fn fail(mut self, reason: ConnectionError) {
self.error = Some(reason.clone());
for (_, writer) in self.blocked_writers.drain() {
writer.notify()
}
for (_, reader) in self.blocked_readers.drain() {
reader.notify()
}
if let Some(c) = self.connecting.take() {
let _ = c.send(Some(reason.clone()));
}
for x in self.uni_opening.drain(..) {
let _ = x.send(Err(reason.clone()));
}
for x in self.bi_opening.drain(..) {
let _ = x.send(Err(reason.clone()));
}
if let Some(x) = self.incoming_streams_reader.take() {
x.notify();
}
for (_, x) in self.finishing.drain() {
let _ = x.send(Some(reason.clone()));
}
}
}
#[derive(Clone)]
@@ -173,7 +202,6 @@ impl Future for Driver {
match event {
Connected { address, protocol } => {
if let Some(c) = endpoint.pending.get_mut(&connection).unwrap().connecting.take() {
// Graceful close should be handled by drop impl
let _ = c.send(None);
} else {
let conn = Rc::new(ConnectionInner { endpoint: Endpoint(self.0.clone()), conn: connection });
@@ -186,10 +214,7 @@ impl Future for Driver {
}
}
ConnectionLost { reason } => {
if let Some(c) = endpoint.pending.get_mut(&connection).unwrap().connecting.take() {
// Graceful close should be handled by drop impl
let _ = c.send(Some(reason));
}
endpoint.pending.remove(&connection).unwrap().fail(reason);
}
StreamWritable { stream } => {
if let Some(writer) = endpoint.pending.get_mut(&connection).unwrap().blocked_writers.remove(&stream) {
@@ -224,7 +249,7 @@ impl Future for Driver {
}
StreamFinished { stream } => {
let _ = endpoint.pending.get_mut(&connection).unwrap()
.finishing.remove(&stream).unwrap().send(());
.finishing.remove(&stream).unwrap().send(None);
}
}
}
@@ -257,37 +282,48 @@ impl Future for Driver {
endpoint.outgoing.push_front((destination, packet));
}
}
TimerStart { connection, timer: timer@quicr::Timer::Close, time } => {
let instant = endpoint.epoch + duration_micros(time);
endpoint.timers.push(Timer {
conn: connection,
ty: timer,
delay: endpoint.timer.delay(instant),
cancel: None,
});
}
TimerStart { connection, timer, time } => {
// Loss detection and idle timers start before the connection is established
let pending = endpoint.pending.entry(connection).or_insert_with(|| Pending::new(None));
use quicr::Timer::*;
let mut cancel = match timer {
LossDetection => Some(&mut pending.cancel_loss_detect),
Idle => Some(&mut pending.cancel_idle),
Close => None
LossDetection => &mut pending.cancel_loss_detect,
Idle => &mut pending.cancel_idle,
Close => unreachable!()
};
let instant = endpoint.epoch + duration_micros(time);
if let Some(cancel) = cancel.as_mut().and_then(|x| x.take()) {
if let Some(cancel) = cancel.take() {
let _ = cancel.send(());
}
let (send, recv) = oneshot::channel();
if let Some(cancel) = cancel { *cancel = Some(send); }
*cancel = Some(send);
trace!(endpoint.log, "timer start"; "timer" => ?timer, "time" => ?duration_micros(time));
endpoint.timers.push(Timer {
conn: connection,
ty: timer,
delay: endpoint.timer.delay(instant),
cancel: recv,
cancel: Some(recv),
});
}
TimerStop { connection, timer } => {
trace!(endpoint.log, "timer stop"; "timer" => ?timer);
let pending = endpoint.pending.get_mut(&connection).unwrap();
use quicr::Timer::*;
match timer {
LossDetection => { pending.cancel_loss_detect.take().map(|x| x.send(()).unwrap()); }
Idle => { pending.cancel_idle.take().map(|x| x.send(())); }
Close => { unreachable!() }
// If a connection was lost, we already canceled its loss/idle timers.
if let Some(pending) = endpoint.pending.get_mut(&connection) {
use quicr::Timer::*;
match timer {
LossDetection => { pending.cancel_loss_detect.take().map(|x| x.send(()).unwrap()); }
Idle => { pending.cancel_idle.take().map(|x| x.send(())); }
Close => { unreachable!() }
}
}
}
}
@@ -328,12 +364,8 @@ struct ConnectionInner {
pub struct Connection(Rc<ConnectionInner>);
#[derive(Copy, Clone, Debug, Fail)]
#[fail(display = "connection lost")]
pub struct ConnectionLost;
impl Connection {
pub fn open_uni(&self) -> Box<Future<Item=SendStream, Error=ConnectionLost>> {
pub fn open_uni(&self) -> Box<Future<Item=SendStream, Error=ConnectionError>> {
let (send, recv) = oneshot::channel();
{
let mut endpoint = self.0.endpoint.0.borrow_mut();
@@ -353,7 +385,7 @@ impl Connection {
)
}
pub fn open_bi(&self) -> Box<Future<Item=(SendStream, RecvStream), Error=ConnectionLost>> {
pub fn open_bi(&self) -> Box<Future<Item=(SendStream, RecvStream), Error=ConnectionError>> {
let (send, recv) = oneshot::channel();
{
let mut endpoint = self.0.endpoint.0.borrow_mut();
@@ -386,6 +418,7 @@ impl Drop for ConnectionInner {
fn drop(&mut self) {
let endpoint = &mut *self.endpoint.0.borrow_mut();
endpoint.inner.close(micros_from(endpoint.epoch.elapsed()), self.conn, 0, (&[][..]).into());
endpoint.pending.remove(&self.conn);
endpoint.driver.as_ref().map(|x| x.notify());
}
}
@@ -394,7 +427,7 @@ pub struct SendStream {
endpoint: Endpoint,
conn: Rc<ConnectionInner>,
stream: StreamId,
finishing: Option<oneshot::Receiver<()>>,
finishing: Option<oneshot::Receiver<Option<ConnectionError>>>,
stop_reason: Option<u16>,
finished: bool,
}
@@ -428,14 +461,17 @@ impl SendStream {
pub fn poll_write(&mut self, buf: &[u8]) -> Poll<usize, WriteError> {
let mut endpoint = self.endpoint.0.borrow_mut();
use quicr::WriteError::*;
let n = match endpoint.inner.write(self.conn.conn, self.stream, buf.into()) {
Ok(()) => buf.len(),
Err((ref unwritten, WriteError::Blocked)) if unwritten.len() < buf.len() => buf.len() - unwritten.len(),
Err((_, WriteError::Blocked)) => {
endpoint.pending.get_mut(&self.conn.conn).unwrap().blocked_writers.insert(self.stream, task::current());
Err((ref unwritten, Blocked)) if unwritten.len() < buf.len() => buf.len() - unwritten.len(),
Err((_, Blocked)) => {
let pending = endpoint.pending.get_mut(&self.conn.conn).unwrap();
if let Some(ref x) = pending.error { return Err(WriteError::ConnectionClosed(x.clone())); }
pending.blocked_writers.insert(self.stream, task::current());
return Ok(Async::NotReady);
}
Err((_, WriteError::Stopped { error_code })) => {
Err((_, Stopped { error_code })) => {
self.stop_reason = Some(error_code);
return Err(WriteError::Stopped { error_code });
}
@@ -444,7 +480,7 @@ impl SendStream {
Ok(Async::Ready(n))
}
pub fn poll_finish(&mut self) -> Async<()> {
pub fn poll_finish(&mut self) -> Poll<(), ConnectionError> {
let mut endpoint = self.endpoint.0.borrow_mut();
if self.finishing.is_none() {
endpoint.inner.finish(self.conn.conn, self.stream);
@@ -453,8 +489,14 @@ impl SendStream {
endpoint.pending.get_mut(&self.conn.conn).unwrap().finishing.insert(self.stream, send);
}
let r = self.finishing.as_mut().unwrap().poll().unwrap();
if let Async::Ready(()) = r { self.finished = true; }
r
match r {
Async::Ready(None) => {
self.finished = true;
Ok(Async::Ready(()))
}
Async::Ready(Some(e)) => Err(e),
Async::NotReady => Ok(Async::NotReady),
}
}
pub fn reset(&self, error_code: u16) {
@@ -470,7 +512,7 @@ impl io::Write for SendStream {
Ok(Async::Ready(n)) => Ok(n),
Ok(Async::NotReady) => Err(io::Error::new(io::ErrorKind::WouldBlock, "stream blocked")),
Err(WriteError::Stopped { .. }) => Err(io::Error::new(io::ErrorKind::ConnectionReset, "peer stopped this stream")),
Err(WriteError::Blocked) => unreachable!(),
Err(WriteError::ConnectionClosed(e)) => Err(io::Error::new(io::ErrorKind::ConnectionAborted, format!("connection closed: {}", e))),
}
}
@@ -479,7 +521,7 @@ impl io::Write for SendStream {
impl AsyncWrite for SendStream {
fn shutdown(&mut self) -> Poll<(), io::Error> {
Ok(self.poll_finish())
self.poll_finish().map_err(|e| io::Error::new(io::ErrorKind::ConnectionAborted, format!("connection closed: {}", e)))
}
}
@@ -493,6 +535,16 @@ impl Drop for SendStream {
}
}
#[derive(Debug, Fail, Clone)]
pub enum WriteError {
/// The peer is no longer accepting data on this stream.
#[fail(display = "sending stopped by peer: error {}", error_code)]
Stopped { error_code: u16 },
/// The connection was closed.
#[fail(display = "connection closed: {}", _0)]
ConnectionClosed(ConnectionError),
}
impl Drop for RecvStream {
fn drop(&mut self) {
let endpoint = &mut *self.endpoint.0.borrow_mut();
@@ -504,24 +556,25 @@ impl Drop for RecvStream {
}
impl RecvStream {
pub fn poll_read_unordered(&mut self) -> Poll<(Box<[u8]>, u64), ReadError> {
pub fn poll_read_unordered(&mut self) -> Poll<(Bytes, u64), ReadError> {
let endpoint = &mut *self.endpoint.0.borrow_mut();
use ReadError::*;
use quicr::ReadError::*;
let pending = endpoint.pending.get_mut(&self.conn.conn).unwrap();
match endpoint.inner.read_unordered(self.conn.conn, self.stream) {
Ok((bytes, offset)) => Ok(Async::Ready((bytes.to_vec().into(), offset))),
Ok((bytes, offset)) => Ok(Async::Ready((bytes, offset))),
Err(Blocked) => {
if let Some(ref x) = pending.error { return Err(ReadError::ConnectionClosed(x.clone())); }
pending.blocked_readers.insert(self.stream, task::current());
Ok(Async::NotReady)
}
Err(e@Reset { .. }) => {
Err(Reset { error_code }) => {
pending.remote_recv_streams.remove(&self.stream);
Err(e)
Err(ReadError::Reset { error_code })
}
Err(e@Finished) => {
Err(Finished) => {
pending.remote_recv_streams.remove(&self.stream);
self.recvd = true;
Err(e)
Err(ReadError::Finished)
}
}
}
@@ -531,19 +584,33 @@ impl RecvStream {
}
}
#[derive(Debug, Fail, Clone)]
pub enum ReadError {
/// The peer abandoned transmitting data on this stream.
#[fail(display = "stream reset by peer: error {}", error_code)]
Reset { error_code: u16 },
/// The data on this stream has been fully delivered and no more will be transmitted.
#[fail(display = "the stream has been completely received")]
Finished,
/// The connection was closed.
#[fail(display = "connection closed: {}", _0)]
ConnectionClosed(ConnectionError),
}
struct Timer {
conn: ConnectionHandle,
ty: quicr::Timer,
delay: Delay,
cancel: oneshot::Receiver<()>,
cancel: Option<oneshot::Receiver<()>>,
}
impl Future for Timer {
type Item = Option<(ConnectionHandle, quicr::Timer)>;
type Error = (); // FIXME
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
if self.ty != quicr::Timer::Close {
if let Async::Ready(()) = self.cancel.poll().unwrap() {
if let Some(ref mut cancel) = self.cancel {
if let Ok(Async::NotReady) = cancel.poll() {}
else {
return Ok(Async::Ready(None));
}
}
@@ -567,10 +634,11 @@ pub enum NewStream {
impl FuturesStream for IncomingStreams {
type Item = NewStream;
type Error = (); // FIXME
type Error = ConnectionError;
fn poll(&mut self) -> Poll<Option<Self::Item>, Self::Error> {
let mut endpoint = self.endpoint.0.borrow_mut();
let pending = endpoint.pending.get_mut(&self.conn.conn).unwrap();
// TODO: Yield error
let pending = if let Some(x) = endpoint.pending.get_mut(&self.conn.conn) { x } else { return Ok(Async::Ready(None)); };
if let Some(x) = pending.incoming_streams.pop_front() {
let recv = RecvStream::new(self.endpoint.clone(), self.conn.clone(), x);
let stream = if x.directionality() == Directionality::Uni {
@@ -580,8 +648,12 @@ impl FuturesStream for IncomingStreams {
};
return Ok(Async::Ready(Some(stream)));
}
pending.incoming_streams_reader = Some(task::current());
return Ok(Async::NotReady);
if let Some(ref x) = pending.error {
Err(x.clone())
} else {
pending.incoming_streams_reader = Some(task::current());
Ok(Async::NotReady)
}
}
}
@@ -596,30 +668,19 @@ impl Future for ReadToEnd {
type Item = Box<[u8]>;
type Error = ReadError;
fn poll(&mut self) -> Poll<Self::Item, Self::Error> {
let endpoint = &mut *self.stream.endpoint.0.borrow_mut();
use ReadError::*;
let pending = endpoint.pending.get_mut(&self.stream.conn.conn).unwrap();
loop {
match endpoint.inner.read_unordered(self.stream.conn.conn, self.stream.stream) {
Ok((data, offset)) => {
match self.stream.poll_read_unordered() {
Ok(Async::Ready((data, offset))) => {
let len = self.buffer.len().max(offset as usize + data.len());
if len > self.size_limit { return Err(Finished); }
if len > self.size_limit { return Err(ReadError::Finished); }
self.buffer.resize(len, 0);
self.buffer[offset as usize..offset as usize+data.len()].copy_from_slice(&data);
}
Err(Blocked) => {
pending.blocked_readers.insert(self.stream.stream, task::current());
return Ok(Async::NotReady);
}
Err(e@Reset { .. }) => {
pending.remote_recv_streams.remove(&self.stream.stream);
return Err(e);
}
Err(Finished) => {
self.stream.recvd = true;
pending.remote_recv_streams.remove(&self.stream.stream);
Ok(Async::NotReady) => { return Ok(Async::NotReady); }
Err(ReadError::Finished) => {
return Ok(Async::Ready(mem::replace(&mut self.buffer, Vec::new()).into()));
}
Err(e) => { return Err(e); }
}
}
}
src/endpoint.rs
+69 -54
View File
@@ -778,6 +778,7 @@ impl Endpoint {
let rs = stream.recv_mut().unwrap();
if let Some(final_offset) = rs.final_offset() {
if end > final_offset || (frame.fin && end != final_offset) {
debug!(self.log, "final offset error"; "frame end" => end, "final offset" => final_offset);
self.events.push_back((conn, Event::ConnectionLost { reason: TransportError::FINAL_OFFSET_ERROR.into() }));
return State::closed(TransportError::FINAL_OFFSET_ERROR);
}
@@ -785,6 +786,9 @@ impl Endpoint {
let prev_end = rs.limit();
let new_bytes = end.saturating_sub(prev_end);
if end > rs.max_data || data_recvd + new_bytes > max_data {
debug!(self.log, "flow control error";
"stream" => frame.id.0, "recvd" => data_recvd, "new bytes" => new_bytes,
"max data" => max_data, "end" => end, "stream max data" => rs.max_data);
self.events.push_back((conn, Event::ConnectionLost { reason: TransportError::FLOW_CONTROL_ERROR.into() }));
return State::closed(TransportError::FLOW_CONTROL_ERROR);
}
@@ -868,6 +872,8 @@ impl Endpoint {
if let Some(stream) = self.connections[conn.0].streams.get_mut(&id) {
let ss = stream.send_mut().unwrap();
if offset > ss.max_data {
trace!(self.log, "stream limit increased"; "stream" => id.0,
"old" => ss.max_data, "new" => offset, "current offset" => ss.offset);
if ss.offset == ss.max_data {
self.events.push_back((conn, Event::StreamWritable { stream: id }));
}
@@ -1045,31 +1051,38 @@ impl Endpoint {
fn reset_idle_timeout(&mut self, now: u64, conn: ConnectionHandle) {
let dt = cmp::min(self.config.idle_timeout, self.connections[conn.0].params.idle_timeout) as u64 * 1000000;
self.io.push_back(Io::TimerStart { time: now + dt, connection: conn, timer: Timer::Idle });
self.connections[conn.0].set_idle = Some(Some(now + dt));
}
/// Returns whether anything was sent
fn flush_pending(&mut self, now: u64, conn: ConnectionHandle) {
let mut timer = None;
let mut sent = false;
while let Some((packet, t)) = self.connections[conn.0].next_packet(&self.log, &self.config, now) {
timer = t.or(timer);
while let Some(packet) = self.connections[conn.0].next_packet(&self.log, &self.config, now) {
self.io.push_back(Io::Transmit {
destination: self.connections[conn.0].remote,
packet: packet.into(),
});
sent = true;
}
if let Some(time) = timer {
self.io.push_back(Io::TimerStart {
connection: conn,
timer: Timer::LossDetection,
time
});
}
if sent {
self.reset_idle_timeout(now, conn);
}
{
let c = &mut self.connections[conn.0];
if let Some(setting) = c.set_idle.take() {
if let Some(time) = setting {
self.io.push_back(Io::TimerStart { connection: conn, timer: Timer::Idle, time });
} else {
self.io.push_back(Io::TimerStop { connection: conn, timer: Timer::Idle });
}
}
if let Some(setting) = c.set_loss_detection.take() {
if let Some(time) = setting {
self.io.push_back(Io::TimerStart { connection: conn, timer: Timer::LossDetection, time });
} else {
self.io.push_back(Io::TimerStop { connection: conn, timer: Timer::LossDetection });
}
}
}
}
fn forget(&mut self, conn: ConnectionHandle) {
@@ -1105,6 +1118,7 @@ impl Endpoint {
self.events.push_back((conn, Event::ConnectionLost {
reason: ConnectionError::TimedOut,
}));
self.dirty_conns.insert(conn); // Ensure the loss detection timer cancellation goes through
}
Timer::LossDetection => {
if self.connections[conn.0].handshake_sent != 0 {
@@ -1146,14 +1160,7 @@ impl Endpoint {
self.reset_idle_timeout(now, conn);
self.connections[conn.0].rto_count += 1;
}
let alarm = self.connections[conn.0].compute_loss_detection_alarm(&self.config);
if alarm != u64::max_value() {
self.io.push_back(Io::TimerStart {
connection: conn,
timer: Timer::LossDetection,
time: alarm,
});
}
self.connections[conn.0].set_loss_detection_alarm(&self.config);
self.dirty_conns.insert(conn);
}
}
@@ -1173,19 +1180,7 @@ impl Endpoint {
fn on_ack_received(&mut self, now: u64, conn: ConnectionHandle, ack: frame::Ack) {
trace!(self.log, "got ack"; "ranges" => ?ack.iter().collect::<Vec<_>>());
let was_blocked = self.connections[conn.0].blocked();
let time = self.connections[conn.0].on_ack_received(&self.config, now, ack);
if time == u64::max_value() {
self.io.push_back(Io::TimerStop {
connection: conn,
timer: Timer::LossDetection,
});
} else {
self.io.push_back(Io::TimerStart {
connection: conn,
timer: Timer::LossDetection,
time,
});
}
self.connections[conn.0].on_ack_received(&self.config, now, ack);
if was_blocked && !self.connections[conn.0].blocked() {
for stream in self.connections[conn.0].blocked_streams.drain() {
self.events.push_back((conn, Event::StreamWritable { stream }));
@@ -1203,9 +1198,13 @@ impl Endpoint {
pub fn write(&mut self, conn: ConnectionHandle, stream: StreamId, mut data: Bytes) -> Result<(), (Bytes, WriteError)> {
if self.connections[conn.0].state.as_ref().unwrap().is_closed() { return Err((data, WriteError::Blocked)); }
assert!(stream.directionality() == Directionality::Bi || stream.initiator() == self.connections[conn.0].side);
if self.connections[conn.0].blocked() { return Err((data, WriteError::Blocked)); }
if self.connections[conn.0].blocked() {
self.connections[conn.0].blocked_streams.insert(stream);
return Err((data, WriteError::Blocked));
}
let (stop_reason, stream_budget) = {
let ss = self.connections[conn.0].streams.get_mut(&stream).expect("stream already closed").send_mut().unwrap();
trace!(self.log, "writing"; "stream" => stream.0, "stream max data" => ss.max_data, "stream offset" => ss.offset);
(match ss.state {
stream::SendState::ResetSent { ref mut stop_reason }
| stream::SendState::ResetRecvd { ref mut stop_reason } => stop_reason.take(),
@@ -1219,15 +1218,19 @@ impl Endpoint {
}
let conn_budget = self.connections[conn.0].max_data - self.connections[conn.0].data_sent;
let result = if conn_budget < data.len() as u64 {
let result = if conn_budget < stream_budget && conn_budget < data.len() as u64 {
self.connections[conn.0].blocked_streams.insert(stream);
Err((data.split_off(conn_budget as usize), WriteError::Blocked))
} else if stream_budget < data.len() as u64 {
Err((data.split_off(stream_budget as usize), WriteError::Blocked))
} else { Ok(()) };
if !data.is_empty() {
trace!(self.log, "queuing"; "stream" => stream.0, "len" => data.len());
self.connections[conn.0].transmit(stream, data);
self.dirty_conns.insert(conn);
} else {
trace!(self.log, "stream blocked"; "stream" => stream.0);
}
result
}
@@ -1266,6 +1269,7 @@ impl Endpoint {
/// # Panics
/// - when applied to a stream that does not have an active incoming channel
pub fn read_unordered(&mut self, conn: ConnectionHandle, stream: StreamId) -> Result<(Bytes, u64), ReadError> {
self.dirty_conns.insert(conn); // May need to send flow control frames after reading
match self.connections[conn.0].read(stream) {
x@Err(ReadError::Finished) | x@Err(ReadError::Reset { .. }) => {
self.connections[conn.0].maybe_cleanup(stream);
@@ -1320,7 +1324,7 @@ impl Endpoint {
fn close_common(&mut self, now: u64, conn: ConnectionHandle) {
trace!(self.log, "connection closed");
self.io.push_back(Io::TimerStop { connection: conn, timer: Timer::LossDetection });
self.connections[conn.0].set_loss_detection = Some(None);
self.io.push_back(Io::TimerStart {
connection: conn,
timer: Timer::Close,
@@ -1347,6 +1351,7 @@ impl Endpoint {
packet: self.connections[conn.0].make_close(&reason),
});
self.reset_idle_timeout(now, conn);
self.dirty_conns.insert(conn);
}
self.connections[conn.0].state = Some(match self.connections[conn.0].state.take().unwrap() {
State::Handshake(_) => State::HandshakeFailed(state::HandshakeFailed { reason, alert: None, app_closed: true }),
@@ -1501,6 +1506,11 @@ struct Connection {
/// Set iff we have received a non-ack frame since the last ack-only packet we sent
permit_ack_only: bool,
// Timer updates: None if no change, Some(None) to stop, Some(Some(_)) to reset
set_idle: Option<Option<u64>>,
set_loss_detection: Option<Option<u64>>,
//
// Stream states
//
@@ -1664,6 +1674,9 @@ impl Connection {
pending_acks: RangeSet::new(),
permit_ack_only: false,
set_idle: None,
set_loss_detection: None,
streams,
next_uni_stream: 0,
next_bi_stream: match side { Side::Client => 1, Side::Server => 0 },
@@ -1683,7 +1696,7 @@ impl Connection {
}
/// Returns new loss detection alarm time, if applicable
fn on_packet_sent(&mut self, config: &Config, now: u64, packet_number: u64, packet: SentPacket) -> Option<u64> {
fn on_packet_sent(&mut self, config: &Config, now: u64, packet_number: u64, packet: SentPacket) {
self.largest_sent_packet = packet_number;
let bytes = packet.bytes;
let handshake = packet.handshake;
@@ -1697,14 +1710,12 @@ impl Connection {
self.time_of_last_sent_handshake_packet = now;
}
self.bytes_in_flight += bytes as u64;
Some(self.compute_loss_detection_alarm(config))
} else {
None
self.set_loss_detection_alarm(config);
}
}
/// Returns new loss detection alarm time
fn on_ack_received(&mut self, config: &Config, now: u64, ack: frame::Ack) -> u64 {
/// Updates set_loss_detection
fn on_ack_received(&mut self, config: &Config, now: u64, ack: frame::Ack) {
self.largest_acked_packet = cmp::max(self.largest_acked_packet, ack.largest); // TODO: Validate
if let Some(info) = self.sent_packets.get(&ack.largest).cloned() {
self.latest_rtt = now - info.time;
@@ -1721,7 +1732,7 @@ impl Connection {
}
}
self.detect_lost_packets(config, now, ack.largest);
self.compute_loss_detection_alarm(config)
self.set_loss_detection_alarm(config);
}
fn update_rtt(&mut self, ack_delay: u64, ack_only: bool) {
@@ -1848,9 +1859,10 @@ impl Connection {
fn in_recovery(&self, packet: u64) -> bool { packet <= self.end_of_recovery }
fn compute_loss_detection_alarm(&self, config: &Config) -> u64 {
fn set_loss_detection_alarm(&mut self, config: &Config) {
if self.bytes_in_flight == 0 {
return u64::max_value();
self.set_loss_detection = Some(None);
return;
}
let mut alarm_duration: u64;
@@ -1864,7 +1876,8 @@ impl Connection {
alarm_duration = cmp::max(alarm_duration + self.max_ack_delay,
config.min_tlp_timeout);
alarm_duration = alarm_duration * 2u64.pow(self.handshake_count);
return self.time_of_last_sent_handshake_packet + alarm_duration;
self.set_loss_detection = Some(Some(self.time_of_last_sent_handshake_packet + alarm_duration));
return;
}
if self.loss_time != 0 {
@@ -1880,7 +1893,7 @@ impl Connection {
alarm_duration = cmp::min(alarm_duration, tlp_duration);
}
}
self.time_of_last_sent_retransmittable_packet + alarm_duration
self.set_loss_detection = Some(Some(self.time_of_last_sent_retransmittable_packet + alarm_duration));
}
/// Retransmit time-out
@@ -1937,7 +1950,7 @@ impl Connection {
});
}
fn next_packet(&mut self, log: &Logger, config: &Config, now: u64) -> Option<(Vec<u8>, Option<u64>)> {
fn next_packet(&mut self, log: &Logger, config: &Config, now: u64) -> Option<Vec<u8>> {
let is_handshake;
match *self.state.as_ref().unwrap() {
ref x if x.is_closed() => { return None; }
@@ -2074,6 +2087,7 @@ impl Connection {
if pending.max_uni_stream_id && buf.len() + 9 < max_size {
pending.max_uni_stream_id = false;
sent.max_uni_stream_id = true;
trace!(log, "MAX_STREAM_ID (unidirectional)");
buf.write(frame::Type::MAX_STREAM_ID);
buf.write(StreamId::new(!self.side, Directionality::Uni, self.max_remote_uni_stream));
}
@@ -2082,6 +2096,7 @@ impl Connection {
if pending.max_bi_stream_id && buf.len() + 9 < max_size {
pending.max_bi_stream_id = false;
sent.max_bi_stream_id = true;
trace!(log, "MAX_STREAM_ID (bidirectional)");
buf.write(frame::Type::MAX_STREAM_ID);
buf.write(StreamId::new(!self.side, Directionality::Bi, self.max_remote_bi_stream));
}
@@ -2093,12 +2108,13 @@ impl Connection {
continue;
}
let len = cmp::min(stream.data.len(), max_size as usize - buf.len() - 25);
trace!(log, "STREAM"; "id" => stream.id.0, "off" => stream.offset, "len" => len, "fin" => stream.fin);
let data = stream.data.split_to(len);
let fin = stream.fin && stream.data.is_empty();
trace!(log, "STREAM"; "id" => stream.id.0, "off" => stream.offset, "len" => len, "fin" => fin);
let frame = frame::Stream {
id: stream.id,
offset: stream.offset,
fin: stream.fin && stream.data.is_empty(),
fin: fin,
data: data,
};
frame.encode(true, &mut buf);
@@ -2118,7 +2134,7 @@ impl Connection {
}
self.encrypt(is_handshake, number, &mut buf, header_len);
let timer = self.on_packet_sent(config, now, number, SentPacket {
self.on_packet_sent(config, now, number, SentPacket {
acks,
time: now, bytes: if ack_only { 0 } else { buf.len() as u16 },
handshake: is_handshake,
@@ -2129,8 +2145,7 @@ impl Connection {
// needlessly prevents us from ACKing the next packet if it's ACK-only, but saves the need for subtler logic to
// avoid double-transmitting acks all the time.
self.permit_ack_only = false;
Some((buf, timer))
Some(buf)
}
fn encrypt(&self, handshake: bool, number: u64, buf: &mut Vec<u8>, header_len: u16) {
@@ -2183,7 +2198,7 @@ impl Connection {
self.max_bi_streams = params.initial_max_streams_bidi as u64;
self.max_uni_streams = params.initial_max_streams_uni as u64;
self.max_data = params.initial_max_data as u64;
for i in match self.side { Side::Client => 0..config.max_remote_bi_streams, Side::Server => 1..(config.max_remote_bi_streams+1) } {
for i in match self.side { Side::Client => 0..config.max_remote_bi_streams, Side::Server => 0..(config.max_remote_bi_streams+1) } {
let id = StreamId::new(!self.side, Directionality::Bi, i as u64);
self.streams.get_mut(&id).unwrap().send_mut().unwrap().max_data = params.initial_max_stream_data as u64;
}
src/frame.rs
+1 -1
View File
@@ -118,7 +118,7 @@ impl Frame {
ConnectionClose(_) => Type::CONNECTION_CLOSE,
ApplicationClose(_) => Type::APPLICATION_CLOSE,
MaxData(_) => Type::MAX_DATA,
MaxStreamData { .. } => Type::MAX_STREAM_ID,
MaxStreamData { .. } => Type::MAX_STREAM_DATA,
MaxStreamId(_) => Type::MAX_STREAM_ID,
Ping => Type::PING,
Blocked { .. } => Type::BLOCKED,