tor_cell/relaycell/
flow_ctrl.rs

1//! Cells for flow control (excluding "sendme" cells).
2
3use std::num::NonZero;
4
5use derive_deftly::Deftly;
6use tor_bytes::{EncodeResult, Error, Reader, Writer};
7use tor_memquota::derive_deftly_template_HasMemoryCost;
8
9use crate::relaycell::msg::Body;
10
11/// An `XON` relay message.
12#[derive(Clone, Debug, Deftly)]
13#[derive_deftly(HasMemoryCost)]
14pub struct Xon {
15    /// Cell `version` field.
16    version: FlowCtrlVersion,
17    /// Cell `kBps_ewma` field.
18    kbytes_per_sec_ewma: XonKBpsEwma,
19}
20
21/// An `XOFF` relay message.
22#[derive(Clone, Debug, Deftly)]
23#[derive_deftly(HasMemoryCost)]
24pub struct Xoff {
25    /// Cell `version` field.
26    version: FlowCtrlVersion,
27}
28
29impl Xon {
30    /// Construct a new [`Xon`] cell.
31    pub fn new(version: FlowCtrlVersion, kbytes_per_sec_ewma: XonKBpsEwma) -> Self {
32        Self {
33            version,
34            kbytes_per_sec_ewma,
35        }
36    }
37
38    /// Return the version.
39    pub fn version(&self) -> FlowCtrlVersion {
40        self.version
41    }
42
43    /// Return the rate limit in KB/s (1000 bytes per second).
44    pub fn kbytes_per_sec_ewma(&self) -> XonKBpsEwma {
45        self.kbytes_per_sec_ewma
46    }
47}
48
49impl Body for Xon {
50    fn decode_from_reader(r: &mut Reader<'_>) -> tor_bytes::Result<Self> {
51        let version = r.take_u8()?;
52
53        let version = match FlowCtrlVersion::new(version) {
54            Ok(x) => x,
55            Err(UnrecognizedVersionError) => {
56                return Err(Error::InvalidMessage("Unrecognized XON version.".into()));
57            }
58        };
59
60        let kbytes_per_sec_ewma = XonKBpsEwma::decode(r.take_u32()?);
61
62        Ok(Self::new(version, kbytes_per_sec_ewma))
63    }
64
65    fn encode_onto<W: Writer + ?Sized>(self, w: &mut W) -> EncodeResult<()> {
66        w.write_u8(*self.version);
67        w.write_u32(self.kbytes_per_sec_ewma.encode());
68        Ok(())
69    }
70}
71
72impl Xoff {
73    /// Construct a new [`Xoff`] cell.
74    pub fn new(version: FlowCtrlVersion) -> Self {
75        Self { version }
76    }
77
78    /// Return the version.
79    pub fn version(&self) -> FlowCtrlVersion {
80        self.version
81    }
82}
83
84impl Body for Xoff {
85    fn decode_from_reader(r: &mut Reader<'_>) -> tor_bytes::Result<Self> {
86        let version = r.take_u8()?;
87
88        let version = match FlowCtrlVersion::new(version) {
89            Ok(x) => x,
90            Err(UnrecognizedVersionError) => {
91                return Err(Error::InvalidMessage("Unrecognized XOFF version.".into()));
92            }
93        };
94
95        Ok(Self::new(version))
96    }
97
98    fn encode_onto<W: Writer + ?Sized>(self, w: &mut W) -> EncodeResult<()> {
99        w.write_u8(*self.version);
100        Ok(())
101    }
102}
103
104/// A recognized flow control version.
105#[derive(Copy, Clone, Debug, Deftly)]
106#[derive_deftly(HasMemoryCost)]
107pub struct FlowCtrlVersion(u8);
108
109impl FlowCtrlVersion {
110    /// Version 0, which is currently the only known version.
111    pub const V0: Self = Self(0);
112
113    /// If `version` is a recognized XON/XOFF version, returns a new [`FlowCtrlVersion`].
114    pub const fn new(version: u8) -> Result<Self, UnrecognizedVersionError> {
115        if version != 0 {
116            return Err(UnrecognizedVersionError);
117        }
118
119        Ok(Self(version))
120    }
121}
122
123impl TryFrom<u8> for FlowCtrlVersion {
124    type Error = UnrecognizedVersionError;
125
126    fn try_from(x: u8) -> Result<Self, Self::Error> {
127        Self::new(x)
128    }
129}
130
131impl std::ops::Deref for FlowCtrlVersion {
132    type Target = u8;
133
134    fn deref(&self) -> &Self::Target {
135        &self.0
136    }
137}
138
139/// The XON/XOFF cell version was not recognized.
140#[derive(Clone, Debug)]
141#[non_exhaustive]
142pub struct UnrecognizedVersionError;
143
144/// The `kBps_ewma` field of an XON cell.
145#[derive(Copy, Clone, Debug, PartialEq, Eq, Deftly)]
146#[derive_deftly(HasMemoryCost)]
147#[allow(clippy::exhaustive_enums)]
148pub enum XonKBpsEwma {
149    /// Stream is rate limited to the value in KB/s (1000 bytes per second).
150    Limited(NonZero<u32>),
151    /// Stream is not rate limited.
152    Unlimited,
153}
154
155impl XonKBpsEwma {
156    /// Decode the `kBps_ewma` field of an XON cell.
157    fn decode(kbytes_per_sec_ewma: u32) -> Self {
158        // prop-324:
159        // > In `xon_cell`, a zero value for `kBps_ewma` means that the stream's rate is unlimited.
160        match NonZero::new(kbytes_per_sec_ewma) {
161            Some(x) => Self::Limited(x),
162            None => Self::Unlimited,
163        }
164    }
165
166    /// Encode as the `kBps_ewma` field of an XON cell.
167    fn encode(&self) -> u32 {
168        // prop-324:
169        // > In `xon_cell`, a zero value for `kBps_ewma` means that the stream's rate is unlimited.
170        match self {
171            Self::Limited(x) => x.get(),
172            Self::Unlimited => 0,
173        }
174    }
175}
176
177impl std::fmt::Display for XonKBpsEwma {
178    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
179        match self {
180            Self::Limited(rate) => write!(f, "{rate} KB/s"),
181            Self::Unlimited => write!(f, "unlimited"),
182        }
183    }
184}
tor_cell/relaycell/flow_ctrl.rs

tor_cell/relaycell/
flow_ctrl.rs
