use core::marker::PhantomData;
use core::ops::Deref;
use std::fmt;
use dope::runtime::profile::Throughput;
use o3::buffer::Pooled;
use crate::egress::Egress;
use crate::frame::{
self, Data, ErrorCode, Flags, FrameBuf, GoAway, HEADER_LEN, ParseError, Ping, Priority,
RstStream, SettingId, WindowUpdate,
};
use crate::ingress::{Ingress, PendingHeaders, PendingKind};
use crate::role::Role;
use crate::stream::{self, Side, Stream, StreamId, TransitionError};
use crate::stream_registry::{StreamClass, StreamRecord, StreamRegistry};
use crate::tuning::Tuning;
use crate::validate::Validate;
use crate::{flow, hpack};
pub const CLIENT_PREFACE: &[u8] = b"PRI * HTTP/2.0\r\n\r\nSM\r\n\r\n";
#[derive(Copy, Clone, Debug, PartialEq, Eq)]
pub struct Settings {
pub header_table_size: u32,
pub enable_push: bool,
pub max_concurrent_streams: Option<u32>,
pub initial_window_size: u32,
pub max_frame_size: u32,
pub max_header_list_size: Option<u32>,
}
impl Settings {
pub const DEFAULT: Self = Self {
header_table_size: 4096,
enable_push: true,
max_concurrent_streams: None,
initial_window_size: 65_535,
max_frame_size: 16_384,
max_header_list_size: None,
};
pub fn apply(&mut self, id: SettingId, value: u32) -> Result<(), ConnError> {
match id {
SettingId::HeaderTableSize => {
self.header_table_size = value;
}
SettingId::EnablePush => match value {
0 => self.enable_push = false,
1 => self.enable_push = true,
_ => return Err(ConnError::BadSettings),
},
SettingId::MaxConcurrentStreams => {
self.max_concurrent_streams = Some(value);
}
SettingId::InitialWindowSize => {
if value > 0x7fff_ffff {
return Err(ConnError::FlowControl);
}
self.initial_window_size = value;
}
SettingId::MaxFrameSize => {
if !(16_384..=16_777_215).contains(&value) {
return Err(ConnError::BadSettings);
}
self.max_frame_size = value;
}
SettingId::MaxHeaderListSize => {
self.max_header_list_size = Some(value);
}
}
Ok(())
}
pub fn encode(&self, out: &mut impl FrameBuf) {
Self::push_param(out, SettingId::HeaderTableSize, self.header_table_size);
Self::push_param(
out,
SettingId::EnablePush,
if self.enable_push { 1 } else { 0 },
);
if let Some(v) = self.max_concurrent_streams {
Self::push_param(out, SettingId::MaxConcurrentStreams, v);
}
Self::push_param(out, SettingId::InitialWindowSize, self.initial_window_size);
Self::push_param(out, SettingId::MaxFrameSize, self.max_frame_size);
if let Some(v) = self.max_header_list_size {
Self::push_param(out, SettingId::MaxHeaderListSize, v);
}
}
fn push_param(out: &mut impl FrameBuf, id: SettingId, value: u32) {
out.extend_from_slice(&(id as u16).to_be_bytes());
out.extend_from_slice(&value.to_be_bytes());
}
pub(crate) fn param_count(&self) -> usize {
2 + self.max_concurrent_streams.is_some() as usize
+ 2
+ self.max_header_list_size.is_some() as usize
}
}
impl Default for Settings {
fn default() -> Self {
Self::DEFAULT
}
}
#[derive(Copy, Clone, Debug, PartialEq, Eq)]
pub struct Config {
pub local_settings: Settings,
pub recv_window_target: u32,
pub stream_capacity: usize,
pub event_capacity: usize,
pub data_capacity: usize,
pub header_capacity: usize,
pub inbound_capacity: usize,
pub outbound_capacity: usize,
}
impl Default for Config {
fn default() -> Self {
Self {
local_settings: Settings {
initial_window_size: <Throughput as Tuning>::STREAM_RECV_WINDOW,
..Settings::DEFAULT
},
recv_window_target: <Throughput as Tuning>::CONN_RECV_WINDOW,
stream_capacity: <Throughput as Tuning>::MAX_ACTIVE_STREAMS,
event_capacity: DEFAULT_EVENT_CAPACITY,
data_capacity: DEFAULT_DATA_EVENTS,
header_capacity: DEFAULT_HEADER_EVENTS,
inbound_capacity: DEFAULT_INBOUND_CAPACITY,
outbound_capacity: DEFAULT_OUTBOUND_CAPACITY,
}
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum ConnError {
BadPreface,
BadSettings,
ParseError(ParseError),
Protocol,
StreamClosed,
FlowControl,
GoAwayReceived(ErrorCode),
StreamLimit,
Hpack(hpack::DecoderError),
BadStream,
Continuation,
FrameSize,
StreamGoneAway,
HeaderListTooLarge,
Overload,
}
impl From<ParseError> for ConnError {
fn from(e: ParseError) -> Self {
ConnError::ParseError(e)
}
}
impl From<flow::Error> for ConnError {
fn from(e: flow::Error) -> Self {
match e {
flow::Error::ZeroIncrement => ConnError::Protocol,
flow::Error::Overflow => ConnError::FlowControl,
flow::Error::Stalled => ConnError::FlowControl,
}
}
}
impl From<hpack::DecoderError> for ConnError {
fn from(e: hpack::DecoderError) -> Self {
match e {
hpack::DecoderError::HeaderListTooLarge => ConnError::HeaderListTooLarge,
other => ConnError::Hpack(other),
}
}
}
impl From<&ConnError> for ErrorCode {
fn from(e: &ConnError) -> Self {
match e {
ConnError::BadPreface
| ConnError::Protocol
| ConnError::BadStream
| ConnError::Continuation
| ConnError::BadSettings
| ConnError::StreamGoneAway => ErrorCode::ProtocolError,
ConnError::StreamClosed => ErrorCode::StreamClosed,
ConnError::ParseError(ParseError::FrameSize)
| ConnError::ParseError(ParseError::BadLength) => ErrorCode::FrameSize,
ConnError::ParseError(_) => ErrorCode::ProtocolError,
ConnError::FlowControl => ErrorCode::FlowControl,
ConnError::FrameSize => ErrorCode::FrameSize,
ConnError::Hpack(_) => ErrorCode::Compression,
ConnError::GoAwayReceived(c) => *c,
ConnError::StreamLimit => ErrorCode::RefusedStream,
ConnError::HeaderListTooLarge | ConnError::Overload => ErrorCode::EnhanceYourCalm,
}
}
}
#[derive(Clone, Debug, PartialEq, Eq)]
pub enum Event {
PrefaceComplete,
SettingsApplied,
SettingsAck,
Ping {
ack: bool,
opaque: [u8; 8],
},
GoAway {
last_stream_id: StreamId,
error: ErrorCode,
debug: DataPayload,
},
Headers {
stream_id: StreamId,
headers: hpack::HeaderBlock,
end_stream: bool,
trailing: bool,
},
Data {
stream_id: StreamId,
data: DataPayload,
end_stream: bool,
},
StreamReset {
stream_id: StreamId,
error: ErrorCode,
},
PushPromise {
stream_id: StreamId,
promised_stream_id: StreamId,
headers: hpack::HeaderBlock,
},
}
#[derive(Clone)]
pub struct DataPayload(Pooled);
impl Deref for DataPayload {
type Target = [u8];
fn deref(&self) -> &Self::Target {
self.0.as_slice()
}
}
impl AsRef<[u8]> for DataPayload {
fn as_ref(&self) -> &[u8] {
self
}
}
impl DataPayload {
pub fn into_pooled(self) -> Pooled {
self.0
}
}
impl PartialEq for DataPayload {
fn eq(&self, other: &Self) -> bool {
self.as_ref() == other.as_ref()
}
}
impl PartialEq<[u8]> for DataPayload {
fn eq(&self, other: &[u8]) -> bool {
self.as_ref() == other
}
}
impl<const N: usize> PartialEq<[u8; N]> for DataPayload {
fn eq(&self, other: &[u8; N]) -> bool {
self.as_ref() == other
}
}
impl<const N: usize> PartialEq<&[u8; N]> for DataPayload {
fn eq(&self, other: &&[u8; N]) -> bool {
self.as_ref() == *other
}
}
impl Eq for DataPayload {}
impl fmt::Debug for DataPayload {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.debug_struct("DataPayload")
.field("len", &self.len())
.finish()
}
}
const DEFAULT_MAX_HEADER_LIST_SIZE: u32 = 16_384;
const DEFAULT_MAX_ACTIVE_STREAMS: usize = 256;
const DEFAULT_INBOUND_CAPACITY: usize = 1 << 20;
const DEFAULT_OUTBOUND_CAPACITY: usize = 1 << 20;
const DEFAULT_EVENT_CAPACITY: usize = 1 << 13;
const DEFAULT_DATA_EVENTS: usize = 64;
const DEFAULT_HEADER_EVENTS: usize = 64;
const MAX_RESET_STREAMS: u32 = 100;
const MAX_CONTINUATION_FRAMES: u32 = 64;
pub struct Conn<R: Role> {
role: PhantomData<R>,
ingress: Ingress,
egress: Egress,
initial_settings_sent: bool,
peer_settings_received: bool,
goaway_sent: bool,
goaway_received: Option<ErrorCode>,
local_settings: Settings,
peer_settings: Settings,
send_window: flow::Window,
recv_window: flow::Window,
recv_window_target: u32,
streams: StreamRegistry<R>,
conn_pending_release: u32,
peer_reset_count: u32,
send_window_opened: bool,
}
impl<R: Role> Conn<R> {
pub fn new() -> Self {
Self::with_tuning::<Throughput>()
}
pub fn with_tuning<P: Tuning>() -> Self {
Self::with_config(Config {
local_settings: Settings {
initial_window_size: P::STREAM_RECV_WINDOW,
..Settings::DEFAULT
},
recv_window_target: P::CONN_RECV_WINDOW,
stream_capacity: P::MAX_ACTIVE_STREAMS,
event_capacity: DEFAULT_EVENT_CAPACITY,
data_capacity: DEFAULT_DATA_EVENTS,
header_capacity: DEFAULT_HEADER_EVENTS,
inbound_capacity: DEFAULT_INBOUND_CAPACITY,
outbound_capacity: DEFAULT_OUTBOUND_CAPACITY,
})
}
pub fn with_local_settings(local: Settings, recv_window_target: u32) -> Self {
let stream_capacity = local
.max_concurrent_streams
.map_or(DEFAULT_MAX_ACTIVE_STREAMS, |limit| (limit as usize).max(1));
Self::with_config(Config {
local_settings: local,
recv_window_target,
stream_capacity,
event_capacity: DEFAULT_EVENT_CAPACITY,
data_capacity: DEFAULT_DATA_EVENTS,
header_capacity: DEFAULT_HEADER_EVENTS,
inbound_capacity: DEFAULT_INBOUND_CAPACITY,
outbound_capacity: DEFAULT_OUTBOUND_CAPACITY,
})
}
pub fn with_config(config: Config) -> Self {
let Config {
local_settings: local,
recv_window_target,
stream_capacity,
event_capacity,
data_capacity,
header_capacity,
inbound_capacity,
outbound_capacity,
} = config;
assert!(stream_capacity > 0, "stream capacity must be positive");
assert!(event_capacity > 0, "event capacity must be positive");
assert!(data_capacity > 0, "data capacity must be positive");
assert!(header_capacity > 0, "header capacity must be positive");
assert!(inbound_capacity > 0, "inbound capacity must be positive");
assert!(outbound_capacity > 0, "outbound capacity must be positive");
let capacity = u32::try_from(stream_capacity).expect("stream capacity overflow");
let peer = Settings::DEFAULT;
let mut local = local;
if R::IS_SERVER {
local.enable_push = false;
}
local.max_concurrent_streams = Some(
local
.max_concurrent_streams
.map_or(capacity, |limit| limit.min(capacity)),
);
if local.max_header_list_size.is_none() {
local.max_header_list_size = Some(DEFAULT_MAX_HEADER_LIST_SIZE);
}
let initial_outbound = HEADER_LEN
+ local.param_count() * 6
+ if R::PREFACE_SENDS_FIRST {
CLIENT_PREFACE.len()
} else {
0
}
+ if recv_window_target > flow::Window::INITIAL as u32 {
HEADER_LEN + 4
} else {
0
};
assert!(
outbound_capacity >= initial_outbound,
"outbound capacity is too small"
);
let header_list_cap = local.max_header_list_size.unwrap() as usize;
let mut conn = Self {
role: PhantomData,
ingress: Ingress::new(
inbound_capacity,
event_capacity,
data_capacity,
local.max_frame_size as usize,
header_capacity,
local.header_table_size as usize,
header_list_cap,
R::PREFACE_SENDS_FIRST,
),
egress: Egress::new(
outbound_capacity,
local.header_table_size as usize,
header_list_cap,
),
initial_settings_sent: false,
peer_settings_received: false,
goaway_sent: false,
goaway_received: None,
local_settings: local,
peer_settings: peer,
send_window: flow::Window::new(),
recv_window: flow::Window::new(),
recv_window_target,
streams: StreamRegistry::new(stream_capacity),
conn_pending_release: 0,
peer_reset_count: 0,
send_window_opened: false,
};
if R::PREFACE_SENDS_FIRST {
if conn
.egress
.raw_mut()
.try_extend_from_slice(CLIENT_PREFACE)
.is_err()
{
unreachable!();
}
conn.ingress.complete_preface();
}
conn.emit_initial_settings();
conn.initial_settings_sent = true;
let bump = recv_window_target.saturating_sub(flow::Window::INITIAL as u32);
if bump > 0
&& conn.recv_window.increase(bump).is_ok()
&& conn.emit_window_update(StreamId::CONNECTION, bump).is_err()
{
unreachable!();
}
conn
}
pub fn local_settings(&self) -> &Settings {
&self.local_settings
}
pub fn peer_settings(&self) -> &Settings {
&self.peer_settings
}
pub fn send_window(&self) -> flow::Window {
self.send_window
}
pub fn recv_window(&self) -> flow::Window {
self.recv_window
}
pub fn goaway_received(&self) -> Option<ErrorCode> {
self.goaway_received
}
pub fn goaway_sent(&self) -> bool {
self.goaway_sent
}
pub fn outbound(&self) -> &[u8] {
self.egress.first()
}
pub fn outbound_slices(&self) -> (&[u8], &[u8]) {
self.egress.slices()
}
pub fn drain_outbound(&mut self, n: usize) {
self.egress.drain(n);
}
pub fn drain_into(&mut self, write_buf: &mut [u8]) -> usize {
self.egress.drain_into(write_buf)
}
pub fn poll_event(&mut self) -> Option<Event> {
self.ingress.poll_event()
}
fn push_event(&mut self, event: Event) -> Result<(), ConnError> {
self.ingress.push_event(event)
}
fn ensure_event_capacity(&self) -> Result<(), ConnError> {
self.ingress.ensure_event_capacity()
}
fn prepare_outbound(&mut self, additional: usize) -> Result<(), ConnError> {
self.egress.reserve(additional)
}
pub fn take_window_opened(&mut self) -> bool {
std::mem::take(&mut self.send_window_opened)
}
pub fn has_stream(&self, id: StreamId) -> bool {
self.stream(id).is_some()
}
pub fn stream_state(&self, id: StreamId) -> Option<stream::State> {
self.stream(id).map(|record| record.stream.state)
}
pub fn active_count(&self) -> usize {
self.streams.active_count()
}
pub fn tracked_closed_count(&self) -> usize {
self.streams.reset_count()
}
pub fn stream_send_window(&self, id: StreamId) -> Option<flow::Window> {
self.stream(id).map(|record| record.send_window)
}
pub fn stream_recv_window(&self, id: StreamId) -> Option<flow::Window> {
self.stream(id).map(|record| record.recv_window)
}
pub fn ping(&mut self, opaque: [u8; 8]) -> Result<(), ConnError> {
self.prepare_outbound(HEADER_LEN + opaque.len())?;
let frame = Ping { ack: false, opaque };
frame.encode(self.egress.raw_mut());
Ok(())
}
pub fn goaway(&mut self, error: ErrorCode, debug: &[u8]) -> Result<(), ConnError> {
self.prepare_outbound(HEADER_LEN + 8 + debug.len())?;
let frame = GoAway {
last_stream_id: StreamId(self.streams.last_peer_id()),
error,
debug,
};
frame.encode(self.egress.raw_mut());
self.goaway_sent = true;
Ok(())
}
pub fn reset_stream(&mut self, stream_id: StreamId, error: ErrorCode) -> Result<(), ConnError> {
if !self.has_stream(stream_id) {
return Err(ConnError::BadStream);
}
self.prepare_outbound(HEADER_LEN + 4)?;
RstStream { stream_id, error }.encode(self.egress.raw_mut());
self.advance_stream(stream_id, stream::Event::RstStream, Side::Local)
.map_err(|_| ConnError::Protocol)?;
Ok(())
}
pub fn send_data(
&mut self,
stream_id: StreamId,
data: &[u8],
end_stream: bool,
) -> Result<usize, ConnError> {
self.send_data_parts(stream_id, data, &[], end_stream)
}
pub fn send_data_parts(
&mut self,
stream_id: StreamId,
first: &[u8],
second: &[u8],
end_stream: bool,
) -> Result<usize, ConnError> {
if !self.has_stream(stream_id) {
return Err(ConnError::BadStream);
}
let len = first
.len()
.checked_add(second.len())
.ok_or(ConnError::FrameSize)?;
let max_frame = self.peer_settings.max_frame_size as usize;
let avail = {
let record = self
.streams
.get_mut(stream_id)
.ok_or(ConnError::BadStream)?;
flow::Pair {
conn: &mut self.send_window,
stream: &mut record.send_window,
}
.available()
};
if avail == 0 && len != 0 {
return Ok(0);
}
let send_n = len.min(avail).min(max_frame);
self.prepare_outbound(HEADER_LEN + send_n)?;
if send_n > 0 {
let record = self
.streams
.get_mut(stream_id)
.ok_or(ConnError::BadStream)?;
let mut pair = flow::Pair {
conn: &mut self.send_window,
stream: &mut record.send_window,
};
pair.consume(send_n).map_err(ConnError::from)?;
}
let last_chunk = send_n == len;
let es = end_stream && last_chunk;
Data::encode_parts(stream_id, es, first, second, send_n, self.egress.raw_mut());
self.advance_stream(
stream_id,
stream::Event::Data { end_stream: es },
Side::Local,
)
.map_err(|_| ConnError::Protocol)?;
Ok(send_n)
}
pub fn send_trailers(
&mut self,
stream_id: StreamId,
headers: &[hpack::Header<'_>],
) -> Result<(), ConnError> {
if !self.has_stream(stream_id) {
return Err(ConnError::BadStream);
}
self.emit_headers(stream_id, headers.iter().copied(), true)?;
self.advance_stream(
stream_id,
stream::Event::Headers { end_stream: true },
Side::Local,
)
.map_err(|_| ConnError::Protocol)?;
Ok(())
}
pub fn ingest(&mut self, bytes: &[u8]) -> Result<(), ConnError> {
self.ingress.append(bytes)?;
self.resume()
}
pub fn resume(&mut self) -> Result<(), ConnError> {
self.drive()?;
if self.egress.over_capacity() {
return Err(ConnError::Overload);
}
Ok(())
}
fn emit_initial_settings(&mut self) {
self.egress.initial_settings(&self.local_settings);
}
fn emit_settings_ack(&mut self) -> Result<(), ConnError> {
self.egress.settings_ack()
}
fn emit_window_update(&mut self, stream_id: StreamId, increment: u32) -> Result<(), ConnError> {
self.egress.window_update(stream_id, increment)
}
fn emit_rst(&mut self, stream_id: StreamId, error: ErrorCode) -> Result<(), ConnError> {
self.egress.reset(stream_id, error)
}
fn rst_evict(&mut self, stream_id: StreamId, error: ErrorCode) -> Result<(), ConnError> {
self.emit_rst(stream_id, error)?;
self.mark_reset(stream_id);
self.evict_stream(stream_id);
Ok(())
}
fn emit_headers<'a, I>(
&mut self,
stream_id: StreamId,
headers: I,
end_stream: bool,
) -> Result<(), ConnError>
where
I: IntoIterator<Item = hpack::Header<'a>>,
{
let max_frame = self.peer_settings.max_frame_size as usize;
self.egress
.headers(stream_id, headers, end_stream, max_frame)
}
fn advance_stream(
&mut self,
id: StreamId,
ev: stream::Event,
side: Side,
) -> Result<(), TransitionError> {
let stream = &mut self.stream_mut(id).ok_or(TransitionError::Protocol)?.stream;
let next = match side {
Side::Local => stream.state.send(ev)?,
Side::Remote => stream.state.recv(ev)?,
};
stream.state = next;
if next == stream::State::Closed {
if matches!(ev, stream::Event::RstStream) {
self.mark_reset(id);
}
self.evict_stream(id);
}
Ok(())
}
fn mark_reset(&mut self, id: StreamId) {
self.streams.mark_reset(id);
}
fn is_peer_initiated(id: StreamId) -> bool {
StreamRegistry::<R>::is_peer_initiated(id)
}
fn stream(&self, id: StreamId) -> Option<&StreamRecord> {
self.streams.get(id)
}
fn stream_mut(&mut self, id: StreamId) -> Option<&mut StreamRecord> {
self.streams.get_mut(id)
}
fn track_stream(&mut self, stream: Stream) -> Result<(), Stream> {
self.streams.insert(
stream,
self.peer_settings.initial_window_size,
self.local_settings.initial_window_size,
)
}
fn can_track_peer_stream(&self) -> bool {
self.streams.can_accept_peer(
self.local_settings
.max_concurrent_streams
.unwrap_or(u32::MAX) as usize,
)
}
fn can_track_local_stream(&self) -> bool {
self.streams.can_open_local(
self.peer_settings
.max_concurrent_streams
.map_or(usize::MAX, |limit| limit as usize),
)
}
fn reserve_promised_stream(&mut self, id: StreamId) -> Result<bool, ConnError> {
if !Self::is_peer_initiated(id) || id.0 <= self.streams.last_peer_id() {
return Err(ConnError::Protocol);
}
if !self.can_track_peer_stream() {
self.emit_rst(id, ErrorCode::RefusedStream)?;
self.streams.observe_peer(id);
return Ok(false);
}
self.track_stream(Stream::reserve_remote(id))
.map_err(|_| ConnError::Protocol)?;
self.streams.observe_peer(id);
Ok(true)
}
fn classify_stream(&self, id: StreamId) -> StreamClass {
self.streams.classify(id)
}
fn evict_stream(&mut self, id: StreamId) {
self.streams.remove(id);
}
fn decode_recv_block(&mut self) -> Result<(hpack::HeaderBlock, bool), ConnError> {
self.ingress.decode_headers()
}
fn drive(&mut self) -> Result<(), ConnError> {
if !self.ingress.accept_preface()? {
return Ok(());
}
loop {
let Some(header) = self
.ingress
.next_frame(self.local_settings.max_frame_size)?
else {
return Ok(());
};
let total = HEADER_LEN + header.length as usize;
if self.ingress.has_pending_headers() && header.kind != frame::Type::Continuation {
return Err(ConnError::Continuation);
}
let emits_event = match header.kind {
frame::Type::Settings
| frame::Type::Ping
| frame::Type::GoAway
| frame::Type::Data
| frame::Type::RstStream => true,
frame::Type::Headers | frame::Type::PushPromise | frame::Type::Continuation => {
header.flags.has(Flags::END_HEADERS)
}
frame::Type::WindowUpdate | frame::Type::Priority => false,
};
if emits_event {
self.ensure_event_capacity()?;
}
match header.kind {
frame::Type::Settings => {
let ack = header.flags.has(Flags::ACK);
if !header.stream_id.is_zero() {
return Err(ParseError::Protocol.into());
}
if (ack && header.length != 0) || !header.length.is_multiple_of(6) {
return Err(ParseError::FrameSize.into());
}
if ack {
self.peer_settings_received = true;
self.push_event(Event::SettingsAck)?;
} else {
self.prepare_outbound(HEADER_LEN)?;
let mut next_settings = self.peer_settings;
let mut encoder_size = None;
let prev_iws = self.peer_settings.initial_window_size as i64;
let mut offset = 0;
while offset < header.length as usize {
let mut chunk = [0; 6];
let copied = self.ingress.copy(HEADER_LEN + offset, &mut chunk);
debug_assert!(copied);
let id_raw = u16::from_be_bytes([chunk[0], chunk[1]]);
let val = u32::from_be_bytes([chunk[2], chunk[3], chunk[4], chunk[5]]);
if let Some(id) = SettingId::from_u16(id_raw) {
next_settings.apply(id, val)?;
if id == SettingId::HeaderTableSize {
encoder_size = Some(val as usize);
}
}
offset += 6;
}
let new_iws = next_settings.initial_window_size as i64;
let delta = new_iws - prev_iws;
if delta != 0 {
let delta32 = delta as i32;
for record in self.streams.values_mut() {
record
.send_window
.adjust_initial(delta32)
.map_err(ConnError::from)?;
}
}
self.peer_settings = next_settings;
if let Some(max_size) = encoder_size {
self.egress.set_header_table_size(max_size);
}
self.ingress.consume(total);
self.emit_settings_ack()?;
self.push_event(Event::SettingsApplied)?;
if delta > 0 {
self.send_window_opened = true;
}
continue;
}
}
frame::Type::Ping => {
let mut payload = [0; 8];
if header.length != payload.len() as u32 {
return Err(ParseError::FrameSize.into());
}
let copied = self.ingress.copy(HEADER_LEN, &mut payload);
debug_assert!(copied);
let parsed = Ping::parse(header, &payload)?;
if !parsed.ack {
let pong = Ping {
ack: true,
opaque: parsed.opaque,
};
self.prepare_outbound(HEADER_LEN + 8)?;
self.ingress.consume(total);
pong.encode(self.egress.raw_mut());
self.push_event(Event::Ping {
ack: false,
opaque: parsed.opaque,
})?;
continue;
}
self.push_event(Event::Ping {
ack: true,
opaque: parsed.opaque,
})?;
}
frame::Type::GoAway => {
if header.length < 8 {
return Err(ParseError::FrameSize.into());
}
let mut prefix = [0; 8];
let copied = self.ingress.copy(HEADER_LEN, &mut prefix);
debug_assert!(copied);
let parsed = GoAway::parse(header, &prefix)?;
let debug_len = header.length as usize - prefix.len();
let debug = self.ingress.data(HEADER_LEN + prefix.len(), debug_len)?;
self.goaway_received = Some(parsed.error);
self.push_event(Event::GoAway {
last_stream_id: parsed.last_stream_id,
error: parsed.error,
debug: DataPayload(debug),
})?;
}
frame::Type::WindowUpdate => {
let mut payload = [0; 4];
if header.length != payload.len() as u32 {
return Err(ParseError::FrameSize.into());
}
let copied = self.ingress.copy(HEADER_LEN, &mut payload);
debug_assert!(copied);
let parsed = WindowUpdate::parse(header, &payload)?;
self.ingress.consume(total);
self.handle_window_update_frame(parsed)?;
continue;
}
frame::Type::Headers => {
if header.stream_id.is_zero() {
return Err(ParseError::Protocol.into());
}
self.prepare_outbound(HEADER_LEN + 4)?;
self.ingress.clear_headers();
let (mut start, mut len) = self.ingress.unpadded_payload(header)?;
if header.flags.has(Flags::PRIORITY) {
if len < 5 {
return Err(ParseError::FrameSize.into());
}
let mut priority = [0; 5];
let copied = self.ingress.copy(start, &mut priority);
debug_assert!(copied);
let _ = frame::PriorityFields::parse(&priority)?;
start += priority.len();
len -= priority.len();
}
if len > self.ingress.header_remaining() {
return Err(ConnError::HeaderListTooLarge);
}
self.ingress.extend_headers(start, len)?;
let sid = header.stream_id;
let end_stream = header.flags.has(Flags::END_STREAM);
let end_headers = header.flags.has(Flags::END_HEADERS);
self.ingress.consume(total);
self.handle_headers_frame(sid, end_stream, end_headers)?;
continue;
}
frame::Type::Data => {
if header.stream_id.is_zero() {
return Err(ParseError::Protocol.into());
}
self.prepare_outbound(2 * (HEADER_LEN + 4))?;
let (start, len) = self.ingress.unpadded_payload(header)?;
let data = self.ingress.data(start, len)?;
let stream_id = header.stream_id;
let end_stream = header.flags.has(Flags::END_STREAM);
let payload = DataPayload(data);
self.ingress.consume(total);
self.handle_data_frame(stream_id, end_stream, payload)?;
continue;
}
frame::Type::Continuation => {
if header.stream_id.is_zero() {
return Err(ParseError::Protocol.into());
}
self.prepare_outbound(HEADER_LEN + 4)?;
let len = header.length as usize;
if len > self.ingress.header_remaining() {
return Err(ConnError::HeaderListTooLarge);
}
self.ingress.extend_headers(HEADER_LEN, len)?;
let stream_id = header.stream_id;
let end_headers = header.flags.has(Flags::END_HEADERS);
self.ingress.consume(total);
self.handle_continuation_frame(stream_id, end_headers, len)?;
continue;
}
frame::Type::RstStream => {
let mut payload = [0; 4];
if header.length != payload.len() as u32 {
return Err(ParseError::FrameSize.into());
}
let copied = self.ingress.copy(HEADER_LEN, &mut payload);
debug_assert!(copied);
let parsed = RstStream::parse(header, &payload)?;
self.ingress.consume(total);
self.handle_rst_frame(parsed)?;
continue;
}
frame::Type::PushPromise => {
if header.stream_id.is_zero() {
return Err(ParseError::Protocol.into());
}
self.prepare_outbound(HEADER_LEN + 4)?;
self.ingress.clear_headers();
let (start, len) = self.ingress.unpadded_payload(header)?;
if len < 4 {
return Err(ParseError::FrameSize.into());
}
let mut promised = [0; 4];
let copied = self.ingress.copy(start, &mut promised);
debug_assert!(copied);
let promised = StreamId::from_u32_masked(u32::from_be_bytes(promised));
let block_start = start + 4;
let block_len = len - 4;
if block_len > self.ingress.header_remaining() {
return Err(ConnError::HeaderListTooLarge);
}
self.ingress.extend_headers(block_start, block_len)?;
let stream_id = header.stream_id;
let end_headers = header.flags.has(Flags::END_HEADERS);
self.ingress.consume(total);
self.handle_push_promise_frame(stream_id, promised, end_headers)?;
continue;
}
frame::Type::Priority => {
let mut payload = [0; 5];
if header.length != payload.len() as u32 {
return Err(ParseError::FrameSize.into());
}
let copied = self.ingress.copy(HEADER_LEN, &mut payload);
debug_assert!(copied);
let _ = Priority::parse(header, &payload)?;
}
}
self.ingress.consume(total);
}
}
fn handle_headers_frame(
&mut self,
stream_id: StreamId,
end_stream: bool,
end_headers: bool,
) -> Result<(), ConnError> {
if stream_id.is_zero() {
return Err(ConnError::Protocol);
}
match self.classify_stream(stream_id) {
StreamClass::Connection => return Err(ConnError::Protocol),
StreamClass::ClosedEnd => return Err(ConnError::StreamClosed),
StreamClass::ClosedRst => {
self.emit_rst(stream_id, ErrorCode::StreamClosed)?;
return Ok(());
}
StreamClass::Idle => {
let peer_init = if R::IS_SERVER {
stream_id.is_client()
} else {
stream_id.is_server()
};
if !peer_init {
return Err(ConnError::Protocol);
}
if self.goaway_sent {
self.emit_rst(stream_id, ErrorCode::RefusedStream)?;
return Ok(());
}
if !self.can_track_peer_stream() {
self.emit_rst(stream_id, ErrorCode::RefusedStream)?;
self.streams.observe_peer(stream_id);
return Ok(());
}
if self.track_stream(Stream::new(stream_id)).is_err() {
self.emit_rst(stream_id, ErrorCode::RefusedStream)?;
self.streams.observe_peer(stream_id);
return Ok(());
}
self.streams.observe_peer(stream_id);
}
StreamClass::Active => {}
}
let trailing = self.is_trailing(stream_id);
if end_headers {
let (headers, over_limit) = self.decode_recv_block()?;
let valid = if R::IS_SERVER {
Validate::request(&headers, trailing)
} else {
Validate::response(&headers, trailing)
};
if valid.is_err() || over_limit {
self.rst_evict(stream_id, ErrorCode::ProtocolError)?;
return Ok(());
}
match self.advance_stream(
stream_id,
stream::Event::Headers { end_stream },
Side::Remote,
) {
Ok(()) => {}
Err(TransitionError::Protocol) => return Err(ConnError::Protocol),
Err(TransitionError::StreamClosed) => {
self.rst_evict(stream_id, ErrorCode::StreamClosed)?;
return Ok(());
}
}
if let Some(record) = self.stream_mut(stream_id) {
record.stream.peer_headers_received = true;
}
self.push_event(Event::Headers {
stream_id,
headers,
end_stream,
trailing,
})?;
} else {
self.ingress.start_headers(PendingHeaders {
stream_id,
kind: PendingKind::Headers {
end_stream,
trailing,
},
continuations: 0,
});
}
Ok(())
}
fn is_trailing(&self, stream_id: StreamId) -> bool {
self.stream(stream_id)
.map(|record| record.stream.peer_headers_received)
.unwrap_or(false)
}
fn handle_data_frame(
&mut self,
stream_id: StreamId,
end_stream: bool,
payload: DataPayload,
) -> Result<(), ConnError> {
match self.classify_stream(stream_id) {
StreamClass::Connection => return Err(ConnError::Protocol),
StreamClass::Idle => return Err(ConnError::Protocol),
StreamClass::ClosedEnd => return Err(ConnError::StreamClosed),
StreamClass::ClosedRst => {
self.emit_rst(stream_id, ErrorCode::StreamClosed)?;
return Ok(());
}
StreamClass::Active => {}
}
let n = payload.len();
self.recv_window
.consume(n)
.map_err(|_| ConnError::FlowControl)?;
{
self.stream_mut(stream_id)
.ok_or(ConnError::Protocol)?
.recv_window
.consume(n)
.map_err(|_| ConnError::FlowControl)?;
}
self.replenish_recv(stream_id, n)?;
match self.advance_stream(stream_id, stream::Event::Data { end_stream }, Side::Remote) {
Ok(()) => {}
Err(TransitionError::Protocol) => return Err(ConnError::Protocol),
Err(TransitionError::StreamClosed) => {
self.rst_evict(stream_id, ErrorCode::StreamClosed)?;
return Ok(());
}
}
self.push_event(Event::Data {
stream_id,
data: payload,
end_stream,
})?;
Ok(())
}
fn replenish_recv(&mut self, stream_id: StreamId, n: usize) -> Result<(), ConnError> {
if n == 0 {
return Ok(());
}
let n32 = u32::try_from(n).map_err(|_| ConnError::FlowControl)?;
let conn_threshold = (self.recv_window_target / 2).max(1);
self.conn_pending_release = self.conn_pending_release.saturating_add(n32);
if self.conn_pending_release >= conn_threshold {
let inc = self.conn_pending_release;
self.conn_pending_release = 0;
self.recv_window.increase(inc).map_err(ConnError::from)?;
self.emit_window_update(StreamId::CONNECTION, inc)?;
}
let stream_threshold = (self.local_settings.initial_window_size / 2).max(1);
let stream_increment = if let Some(record) = self.stream_mut(stream_id) {
record.pending_release = record.pending_release.saturating_add(n32);
if record.pending_release >= stream_threshold {
let increment = record.pending_release;
record.pending_release = 0;
record
.recv_window
.increase(increment)
.map_err(ConnError::from)?;
Some(increment)
} else {
None
}
} else {
None
};
if let Some(increment) = stream_increment {
self.emit_window_update(stream_id, increment)?;
}
Ok(())
}
fn handle_window_update_frame(&mut self, parsed: WindowUpdate) -> Result<(), ConnError> {
if parsed.stream_id.is_zero() {
self.send_window
.increase(parsed.increment)
.map_err(ConnError::from)?;
self.send_window_opened = true;
return Ok(());
}
match self.classify_stream(parsed.stream_id) {
StreamClass::Connection => Err(ConnError::Protocol),
StreamClass::Idle => Err(ConnError::Protocol),
StreamClass::ClosedRst | StreamClass::ClosedEnd => Ok(()),
StreamClass::Active => {
self.stream_mut(parsed.stream_id)
.ok_or(ConnError::Protocol)?
.send_window
.increase(parsed.increment)
.map_err(ConnError::from)?;
self.send_window_opened = true;
Ok(())
}
}
}
fn handle_continuation_frame(
&mut self,
stream_id: StreamId,
end_headers: bool,
fragment_len: usize,
) -> Result<(), ConnError> {
let pending = self
.ingress
.pending_headers_mut()
.ok_or(ConnError::Continuation)?;
if pending.stream_id != stream_id {
return Err(ConnError::Continuation);
}
if fragment_len == 0 && !end_headers {
return Err(ConnError::Continuation);
}
pending.continuations = pending.continuations.saturating_add(1);
if pending.continuations > MAX_CONTINUATION_FRAMES {
return Err(ConnError::Overload);
}
if end_headers {
let pending = self.ingress.take_pending_headers().unwrap();
let (headers, over_limit) = self.decode_recv_block()?;
match pending.kind {
PendingKind::Headers {
end_stream,
trailing,
} => {
let valid = if R::IS_SERVER {
Validate::request(&headers, trailing)
} else {
Validate::response(&headers, trailing)
};
if valid.is_err() || over_limit {
self.rst_evict(pending.stream_id, ErrorCode::ProtocolError)?;
return Ok(());
}
match self.advance_stream(
pending.stream_id,
stream::Event::Headers { end_stream },
Side::Remote,
) {
Ok(()) => {}
Err(TransitionError::Protocol) => {
return Err(ConnError::Protocol);
}
Err(TransitionError::StreamClosed) => {
self.rst_evict(pending.stream_id, ErrorCode::StreamClosed)?;
return Ok(());
}
}
if let Some(record) = self.stream_mut(pending.stream_id) {
record.stream.peer_headers_received = true;
}
self.push_event(Event::Headers {
stream_id: pending.stream_id,
headers,
end_stream,
trailing,
})?;
}
PendingKind::PushPromise { promised } => {
let valid = Validate::request(&headers, false);
if valid.is_err() || over_limit {
self.rst_evict(promised, ErrorCode::ProtocolError)?;
return Ok(());
}
if !self.reserve_promised_stream(promised)? {
return Ok(());
}
self.push_event(Event::PushPromise {
stream_id: pending.stream_id,
promised_stream_id: promised,
headers,
})?;
}
}
}
Ok(())
}
fn handle_rst_frame(&mut self, r: RstStream) -> Result<(), ConnError> {
match self.classify_stream(r.stream_id) {
StreamClass::Connection => return Err(ConnError::Protocol),
StreamClass::Idle => return Err(ConnError::Protocol),
StreamClass::ClosedRst | StreamClass::ClosedEnd => return Ok(()),
StreamClass::Active => {}
}
let peer_reset_count = self.peer_reset_count.saturating_add(1);
if peer_reset_count > MAX_RESET_STREAMS {
return Err(ConnError::Overload);
}
self.advance_stream(r.stream_id, stream::Event::RstStream, Side::Remote)
.map_err(|_| ConnError::Protocol)?;
self.push_event(Event::StreamReset {
stream_id: r.stream_id,
error: r.error,
})?;
self.peer_reset_count = peer_reset_count;
Ok(())
}
fn handle_push_promise_frame(
&mut self,
stream_id: StreamId,
promised: StreamId,
end_headers: bool,
) -> Result<(), ConnError> {
if R::IS_SERVER {
return Err(ConnError::Protocol);
}
if end_headers {
let (headers, over_limit) = self.decode_recv_block()?;
if Validate::request(&headers, false).is_err() || over_limit {
self.emit_rst(promised, ErrorCode::ProtocolError)?;
return Ok(());
}
if !self.reserve_promised_stream(promised)? {
return Ok(());
}
self.push_event(Event::PushPromise {
stream_id,
promised_stream_id: promised,
headers,
})?;
} else {
self.ingress.start_headers(PendingHeaders {
stream_id,
kind: PendingKind::PushPromise { promised },
continuations: 0,
});
}
Ok(())
}
}
impl Conn<crate::role::ClientRole> {
pub fn start_request(
&mut self,
headers: &[hpack::Header<'_>],
end_stream: bool,
) -> Result<StreamId, ConnError> {
if self.goaway_received.is_some() || self.goaway_sent {
return Err(ConnError::StreamGoneAway);
}
if !self.can_track_local_stream() {
return Err(ConnError::StreamLimit);
}
let id = self.streams.next_local_id().ok_or(ConnError::StreamLimit)?;
self.track_stream(Stream::new(id))
.map_err(|_| ConnError::StreamLimit)?;
self.emit_headers(id, headers.iter().copied(), end_stream)?;
self.advance_stream(id, stream::Event::Headers { end_stream }, Side::Local)
.map_err(|_| ConnError::Protocol)?;
Ok(id)
}
}
impl Conn<crate::role::ServerRole> {
pub fn send_response<'a, I>(
&mut self,
stream_id: StreamId,
headers: I,
end_stream: bool,
) -> Result<(), ConnError>
where
I: IntoIterator<Item = hpack::Header<'a>>,
{
if !self.has_stream(stream_id) {
return Err(ConnError::BadStream);
}
self.emit_headers(stream_id, headers, end_stream)?;
self.advance_stream(
stream_id,
stream::Event::Headers { end_stream },
Side::Local,
)
.map_err(|_| ConnError::Protocol)?;
Ok(())
}
}
impl<R: Role> Default for Conn<R> {
fn default() -> Self {
Self::new()
}
}