#[cfg(feature = "qlog")]
use std::path::Path;
use std::{
fmt,
net::SocketAddr,
num::{NonZeroU8, NonZeroU32},
sync::Arc,
};
use crate::{
ConnectionId, Duration, INITIAL_MTU, Instant, MAX_UDP_PAYLOAD, Side, VarInt,
VarIntBoundsExceeded, address_discovery, congestion, connection::qlog::QlogSink,
};
#[cfg(feature = "qlog")]
use crate::{QlogFactory, QlogFileFactory};
const DEFAULT_CONCURRENT_MULTIPATH_PATHS_WHEN_ENABLED_: NonZeroU32 = {
match NonZeroU32::new(12) {
Some(v) => v,
None => panic!("to enable multipath this must be positive, which clearly it is"),
}
};
#[cfg(doc)]
pub const DEFAULT_CONCURRENT_MULTIPATH_PATHS_WHEN_ENABLED: NonZeroU32 =
DEFAULT_CONCURRENT_MULTIPATH_PATHS_WHEN_ENABLED_;
#[derive(Clone)]
pub struct TransportConfig {
pub(crate) max_concurrent_bidi_streams: VarInt,
pub(crate) max_concurrent_uni_streams: VarInt,
pub(crate) max_idle_timeout: Option<VarInt>,
pub(crate) stream_receive_window: VarInt,
pub(crate) receive_window: VarInt,
pub(crate) send_window: u64,
pub(crate) send_fairness: bool,
pub(crate) packet_threshold: u32,
pub(crate) time_threshold: f32,
pub(crate) initial_rtt: Duration,
pub(crate) initial_mtu: u16,
pub(crate) min_mtu: u16,
pub(crate) mtu_discovery_config: Option<MtuDiscoveryConfig>,
pub(crate) pad_to_mtu: bool,
pub(crate) ack_frequency_config: Option<AckFrequencyConfig>,
pub(crate) persistent_congestion_threshold: u32,
pub(crate) keep_alive_interval: Option<Duration>,
pub(crate) crypto_buffer_size: usize,
pub(crate) allow_spin: bool,
pub(crate) datagram_receive_buffer_size: Option<usize>,
pub(crate) datagram_send_buffer_size: usize,
#[cfg(test)]
pub(crate) deterministic_packet_numbers: bool,
pub(crate) congestion_controller_factory: Arc<dyn congestion::ControllerFactory + Send + Sync>,
pub(crate) enable_segmentation_offload: bool,
pub(crate) address_discovery_role: address_discovery::Role,
pub(crate) max_concurrent_multipath_paths: Option<NonZeroU32>,
pub(crate) default_path_max_idle_timeout: Option<Duration>,
pub(crate) default_path_keep_alive_interval: Option<Duration>,
pub(crate) max_remote_nat_traversal_addresses: Option<NonZeroU8>,
#[cfg(feature = "qlog")]
pub(crate) qlog_factory: Option<Arc<dyn QlogFactory>>,
}
impl TransportConfig {
pub fn max_concurrent_bidi_streams(&mut self, value: VarInt) -> &mut Self {
self.max_concurrent_bidi_streams = value;
self
}
pub fn max_concurrent_uni_streams(&mut self, value: VarInt) -> &mut Self {
self.max_concurrent_uni_streams = value;
self
}
pub fn max_idle_timeout(&mut self, value: Option<IdleTimeout>) -> &mut Self {
self.max_idle_timeout = value.map(|t| t.0);
self
}
pub fn stream_receive_window(&mut self, value: VarInt) -> &mut Self {
self.stream_receive_window = value;
self
}
pub fn receive_window(&mut self, value: VarInt) -> &mut Self {
self.receive_window = value;
self
}
pub fn send_window(&mut self, value: u64) -> &mut Self {
self.send_window = value;
self
}
pub fn send_fairness(&mut self, value: bool) -> &mut Self {
self.send_fairness = value;
self
}
pub fn packet_threshold(&mut self, value: u32) -> &mut Self {
self.packet_threshold = value;
self
}
pub fn time_threshold(&mut self, value: f32) -> &mut Self {
self.time_threshold = value;
self
}
pub fn initial_rtt(&mut self, value: Duration) -> &mut Self {
self.initial_rtt = value;
self
}
pub fn initial_mtu(&mut self, value: u16) -> &mut Self {
self.initial_mtu = value.max(INITIAL_MTU);
self
}
pub(crate) fn get_initial_mtu(&self) -> u16 {
self.initial_mtu.max(self.min_mtu)
}
pub fn min_mtu(&mut self, value: u16) -> &mut Self {
self.min_mtu = value.max(INITIAL_MTU);
self
}
pub fn mtu_discovery_config(&mut self, value: Option<MtuDiscoveryConfig>) -> &mut Self {
self.mtu_discovery_config = value;
self
}
pub fn pad_to_mtu(&mut self, value: bool) -> &mut Self {
self.pad_to_mtu = value;
self
}
pub fn ack_frequency_config(&mut self, value: Option<AckFrequencyConfig>) -> &mut Self {
self.ack_frequency_config = value;
self
}
pub fn persistent_congestion_threshold(&mut self, value: u32) -> &mut Self {
self.persistent_congestion_threshold = value;
self
}
pub fn keep_alive_interval(&mut self, value: Option<Duration>) -> &mut Self {
self.keep_alive_interval = value;
self
}
pub fn crypto_buffer_size(&mut self, value: usize) -> &mut Self {
self.crypto_buffer_size = value;
self
}
pub fn allow_spin(&mut self, value: bool) -> &mut Self {
self.allow_spin = value;
self
}
pub fn datagram_receive_buffer_size(&mut self, value: Option<usize>) -> &mut Self {
self.datagram_receive_buffer_size = value;
self
}
pub fn datagram_send_buffer_size(&mut self, value: usize) -> &mut Self {
self.datagram_send_buffer_size = value;
self
}
#[cfg(test)]
pub(crate) fn deterministic_packet_numbers(&mut self, enabled: bool) -> &mut Self {
self.deterministic_packet_numbers = enabled;
self
}
pub fn congestion_controller_factory(
&mut self,
factory: Arc<dyn congestion::ControllerFactory + Send + Sync + 'static>,
) -> &mut Self {
self.congestion_controller_factory = factory;
self
}
pub fn enable_segmentation_offload(&mut self, enabled: bool) -> &mut Self {
self.enable_segmentation_offload = enabled;
self
}
pub fn send_observed_address_reports(&mut self, enabled: bool) -> &mut Self {
self.address_discovery_role.send_reports_to_peers(enabled);
self
}
pub fn receive_observed_address_reports(&mut self, enabled: bool) -> &mut Self {
self.address_discovery_role
.receive_reports_from_peers(enabled);
self
}
pub fn max_concurrent_multipath_paths(&mut self, max_concurrent: u32) -> &mut Self {
self.max_concurrent_multipath_paths = NonZeroU32::new(max_concurrent);
self
}
pub fn default_path_max_idle_timeout(&mut self, timeout: Option<Duration>) -> &mut Self {
self.default_path_max_idle_timeout = timeout;
self
}
pub fn default_path_keep_alive_interval(&mut self, interval: Option<Duration>) -> &mut Self {
self.default_path_keep_alive_interval = interval;
self
}
pub(crate) fn get_initial_max_path_id(&self) -> Option<crate::PathId> {
self.max_concurrent_multipath_paths
.map(|nonzero_concurrent| nonzero_concurrent.get() - 1)
.map(Into::into)
}
pub fn set_max_remote_nat_traversal_addresses(&mut self, max_addresses: u8) -> &mut Self {
self.max_remote_nat_traversal_addresses = NonZeroU8::new(max_addresses);
if max_addresses != 0 && self.max_concurrent_multipath_paths.is_none() {
self.max_concurrent_multipath_paths(
DEFAULT_CONCURRENT_MULTIPATH_PATHS_WHEN_ENABLED_.get(),
);
}
self
}
#[cfg(feature = "qlog")]
pub fn qlog_factory(&mut self, factory: Arc<dyn QlogFactory>) -> &mut Self {
self.qlog_factory = Some(factory);
self
}
#[cfg(feature = "qlog")]
pub fn qlog_from_env(&mut self, prefix: &str) -> &mut Self {
self.qlog_factory(Arc::new(QlogFileFactory::from_env().with_prefix(prefix)))
}
#[cfg(feature = "qlog")]
pub fn qlog_from_path(&mut self, path: impl AsRef<Path>, prefix: &str) -> &mut Self {
self.qlog_factory(Arc::new(
QlogFileFactory::new(path.as_ref().to_owned()).with_prefix(prefix),
))
}
pub(crate) fn create_qlog_sink(
&self,
side: Side,
remote: SocketAddr,
initial_dst_cid: ConnectionId,
now: Instant,
) -> QlogSink {
#[cfg(not(feature = "qlog"))]
let sink = {
let _ = (side, remote, initial_dst_cid, now);
QlogSink::default()
};
#[cfg(feature = "qlog")]
let sink = {
if let Some(config) = self
.qlog_factory
.as_ref()
.and_then(|factory| factory.for_connection(side, remote, initial_dst_cid, now))
{
QlogSink::new(config, initial_dst_cid, side, now)
} else {
QlogSink::default()
}
};
sink
}
}
impl Default for TransportConfig {
fn default() -> Self {
const EXPECTED_RTT: u32 = 100; const MAX_STREAM_BANDWIDTH: u32 = 12500 * 1000; const STREAM_RWND: u32 = MAX_STREAM_BANDWIDTH / 1000 * EXPECTED_RTT;
Self {
max_concurrent_bidi_streams: 100u32.into(),
max_concurrent_uni_streams: 100u32.into(),
max_idle_timeout: Some(VarInt(30_000)),
stream_receive_window: STREAM_RWND.into(),
receive_window: VarInt::MAX,
send_window: (8 * STREAM_RWND).into(),
send_fairness: true,
packet_threshold: 3,
time_threshold: 9.0 / 8.0,
initial_rtt: Duration::from_millis(333), initial_mtu: INITIAL_MTU,
min_mtu: INITIAL_MTU,
mtu_discovery_config: Some(MtuDiscoveryConfig::default()),
pad_to_mtu: false,
ack_frequency_config: None,
persistent_congestion_threshold: 3,
keep_alive_interval: None,
crypto_buffer_size: 16 * 1024,
allow_spin: true,
datagram_receive_buffer_size: Some(STREAM_RWND as usize),
datagram_send_buffer_size: 1024 * 1024,
#[cfg(test)]
deterministic_packet_numbers: false,
congestion_controller_factory: Arc::new(congestion::CubicConfig::default()),
enable_segmentation_offload: true,
address_discovery_role: address_discovery::Role::default(),
max_concurrent_multipath_paths: None,
default_path_max_idle_timeout: None,
default_path_keep_alive_interval: None,
max_remote_nat_traversal_addresses: None,
#[cfg(feature = "qlog")]
qlog_factory: None,
}
}
}
impl fmt::Debug for TransportConfig {
fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
let Self {
max_concurrent_bidi_streams,
max_concurrent_uni_streams,
max_idle_timeout,
stream_receive_window,
receive_window,
send_window,
send_fairness,
packet_threshold,
time_threshold,
initial_rtt,
initial_mtu,
min_mtu,
mtu_discovery_config,
pad_to_mtu,
ack_frequency_config,
persistent_congestion_threshold,
keep_alive_interval,
crypto_buffer_size,
allow_spin,
datagram_receive_buffer_size,
datagram_send_buffer_size,
#[cfg(test)]
deterministic_packet_numbers: _,
congestion_controller_factory: _,
enable_segmentation_offload,
address_discovery_role,
max_concurrent_multipath_paths,
default_path_max_idle_timeout,
default_path_keep_alive_interval,
max_remote_nat_traversal_addresses,
#[cfg(feature = "qlog")]
qlog_factory,
} = self;
let mut s = fmt.debug_struct("TransportConfig");
s.field("max_concurrent_bidi_streams", max_concurrent_bidi_streams)
.field("max_concurrent_uni_streams", max_concurrent_uni_streams)
.field("max_idle_timeout", max_idle_timeout)
.field("stream_receive_window", stream_receive_window)
.field("receive_window", receive_window)
.field("send_window", send_window)
.field("send_fairness", send_fairness)
.field("packet_threshold", packet_threshold)
.field("time_threshold", time_threshold)
.field("initial_rtt", initial_rtt)
.field("initial_mtu", initial_mtu)
.field("min_mtu", min_mtu)
.field("mtu_discovery_config", mtu_discovery_config)
.field("pad_to_mtu", pad_to_mtu)
.field("ack_frequency_config", ack_frequency_config)
.field(
"persistent_congestion_threshold",
persistent_congestion_threshold,
)
.field("keep_alive_interval", keep_alive_interval)
.field("crypto_buffer_size", crypto_buffer_size)
.field("allow_spin", allow_spin)
.field("datagram_receive_buffer_size", datagram_receive_buffer_size)
.field("datagram_send_buffer_size", datagram_send_buffer_size)
.field("enable_segmentation_offload", enable_segmentation_offload)
.field("address_discovery_role", address_discovery_role)
.field(
"max_concurrent_multipath_paths",
max_concurrent_multipath_paths,
)
.field(
"default_path_max_idle_timeout",
default_path_max_idle_timeout,
)
.field(
"default_path_keep_alive_interval",
default_path_keep_alive_interval,
)
.field(
"max_remote_nat_traversal_addresses",
max_remote_nat_traversal_addresses,
);
#[cfg(feature = "qlog")]
s.field("qlog_factory", &qlog_factory.is_some());
s.finish_non_exhaustive()
}
}
#[derive(Clone, Debug)]
pub struct AckFrequencyConfig {
pub(crate) ack_eliciting_threshold: VarInt,
pub(crate) max_ack_delay: Option<Duration>,
pub(crate) reordering_threshold: VarInt,
}
impl AckFrequencyConfig {
pub fn ack_eliciting_threshold(&mut self, value: VarInt) -> &mut Self {
self.ack_eliciting_threshold = value;
self
}
pub fn max_ack_delay(&mut self, value: Option<Duration>) -> &mut Self {
self.max_ack_delay = value;
self
}
pub fn reordering_threshold(&mut self, value: VarInt) -> &mut Self {
self.reordering_threshold = value;
self
}
}
impl Default for AckFrequencyConfig {
fn default() -> Self {
Self {
ack_eliciting_threshold: VarInt(1),
max_ack_delay: None,
reordering_threshold: VarInt(2),
}
}
}
#[derive(Clone, Debug)]
pub struct MtuDiscoveryConfig {
pub(crate) interval: Duration,
pub(crate) upper_bound: u16,
pub(crate) minimum_change: u16,
pub(crate) black_hole_cooldown: Duration,
}
impl MtuDiscoveryConfig {
pub fn interval(&mut self, value: Duration) -> &mut Self {
self.interval = value;
self
}
pub fn upper_bound(&mut self, value: u16) -> &mut Self {
self.upper_bound = value.min(MAX_UDP_PAYLOAD);
self
}
pub fn black_hole_cooldown(&mut self, value: Duration) -> &mut Self {
self.black_hole_cooldown = value;
self
}
pub fn minimum_change(&mut self, value: u16) -> &mut Self {
self.minimum_change = value;
self
}
}
impl Default for MtuDiscoveryConfig {
fn default() -> Self {
Self {
interval: Duration::from_secs(600),
upper_bound: 1452,
black_hole_cooldown: Duration::from_secs(60),
minimum_change: 20,
}
}
}
#[derive(Default, Copy, Clone, Eq, Hash, Ord, PartialEq, PartialOrd)]
pub struct IdleTimeout(VarInt);
impl From<VarInt> for IdleTimeout {
fn from(inner: VarInt) -> Self {
Self(inner)
}
}
impl std::convert::TryFrom<Duration> for IdleTimeout {
type Error = VarIntBoundsExceeded;
fn try_from(timeout: Duration) -> Result<Self, Self::Error> {
let inner = VarInt::try_from(timeout.as_millis())?;
Ok(Self(inner))
}
}
impl fmt::Debug for IdleTimeout {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
self.0.fmt(f)
}
}