tor_proto/stream/flow_ctrl/window/
state.rs

1//! Circuit reactor's stream window flow control.
2
3use tor_cell::relaycell::flow_ctrl::{Xoff, Xon, XonKbpsEwma};
4use tor_cell::relaycell::msg::{AnyRelayMsg, Sendme};
5use tor_cell::relaycell::{RelayCmd, RelayMsg, UnparsedRelayMsg};
6
7use crate::congestion::sendme::{
8    self, StreamRecvWindow, StreamSendWindow, cmd_counts_towards_windows,
9};
10use crate::stream::flow_ctrl::state::{FlowCtrlHooks, HalfStreamFlowCtrlHooks};
11use crate::stream::{RECV_WINDOW_INIT, STREAM_READER_BUFFER};
12use crate::{Error, Result};
13
14#[cfg(doc)]
15use crate::stream::flow_ctrl::state::StreamFlowCtrl;
16
17/// State for window-based flow control.
18#[derive(Debug)]
19pub(crate) struct WindowFlowCtrl {
20    /// Send window.
21    window: StreamSendWindow,
22}
23
24impl WindowFlowCtrl {
25    /// Returns a new sendme-window-based state.
26    // TODO: Maybe take the raw u16 and create StreamSendWindow ourselves?
27    // Unclear whether we need or want to support creating this object from a
28    // preexisting StreamSendWindow.
29    pub(crate) fn new(window: StreamSendWindow) -> Self {
30        Self { window }
31    }
32}
33
34impl FlowCtrlHooks for WindowFlowCtrl {
35    fn can_send<M: RelayMsg>(&self, msg: &M) -> bool {
36        !sendme::cmd_counts_towards_windows(msg.cmd()) || self.window.window() > 0
37    }
38
39    fn about_to_send(&mut self, msg: &AnyRelayMsg) -> Result<()> {
40        if sendme::cmd_counts_towards_windows(msg.cmd()) {
41            self.window.take().map(|_| ())
42        } else {
43            // TODO: Maybe make this an error?
44            // Ideally caller would have checked this already.
45            Ok(())
46        }
47    }
48
49    fn put_for_incoming_sendme(&mut self, msg: UnparsedRelayMsg) -> Result<()> {
50        let _sendme = msg
51            .decode::<Sendme>()
52            .map_err(|e| Error::from_bytes_err(e, "failed to decode stream sendme message"))?
53            .into_msg();
54
55        self.window.put()
56    }
57
58    fn handle_incoming_xon(&mut self, _msg: UnparsedRelayMsg) -> Result<()> {
59        let msg = "XON messages not allowed with window flow control";
60        Err(Error::CircProto(msg.into()))
61    }
62
63    fn handle_incoming_xoff(&mut self, _msg: UnparsedRelayMsg) -> Result<()> {
64        let msg = "XOFF messages not allowed with window flow control";
65        Err(Error::CircProto(msg.into()))
66    }
67
68    fn maybe_send_xon(&mut self, _rate: XonKbpsEwma, _buffer_len: usize) -> Result<Option<Xon>> {
69        let msg = "XON messages cannot be sent with window flow control";
70        Err(Error::CircProto(msg.into()))
71    }
72
73    fn maybe_send_xoff(&mut self, _buffer_len: usize) -> Result<Option<Xoff>> {
74        let msg = "XOFF messages cannot be sent with window flow control";
75        Err(Error::CircProto(msg.into()))
76    }
77
78    fn inbound_queue_max_len(&self) -> usize {
79        // SENDME-window flow control sets a maximum number of inflight DATA cells,
80        // so there's an upper limit to the number of cells we typically expect on the stream
81        STREAM_READER_BUFFER
82    }
83}
84
85/// State for window-based flow control on a half-stream.
86#[derive(Debug)]
87pub(crate) struct HalfStreamWindowFlowCtrl {
88    /// The original [`WindowFlowCtrl`] from the full stream.
89    ///
90    /// We keep this since we need to continue validating any incoming messages.
91    inner: WindowFlowCtrl,
92    /// The stream's receive window.
93    ///
94    /// When it was a full-stream, the receive window was tracked by the `DataStream`.
95    /// But since the `DataStream` has gone away, we need to track it ourselves.
96    recv_window: StreamRecvWindow,
97}
98
99impl HalfStreamWindowFlowCtrl {
100    /// Returns a new sendme-window-based state for a half-stream.
101    pub(crate) fn new(flow_ctrl: WindowFlowCtrl) -> Self {
102        Self {
103            inner: flow_ctrl,
104            // FIXME(eta): we don't copy the receive window, instead just creating a new one,
105            //             so a malicious peer can send us slightly more data than they should
106            //             be able to; see arti#230.
107            recv_window: StreamRecvWindow::new(RECV_WINDOW_INIT),
108        }
109    }
110}
111
112impl HalfStreamFlowCtrlHooks for HalfStreamWindowFlowCtrl {
113    fn handle_incoming_dropped(&mut self, msg_count: u16) -> Result<()> {
114        self.recv_window.decrement_n(msg_count)
115    }
116
117    fn handle_incoming_msg(&mut self, msg: UnparsedRelayMsg) -> Result<Option<UnparsedRelayMsg>> {
118        match msg.cmd() {
119            RelayCmd::SENDME => {
120                self.inner.put_for_incoming_sendme(msg)?;
121                Ok(None)
122            }
123            RelayCmd::XON => {
124                self.inner.handle_incoming_xon(msg)?;
125                Ok(None)
126            }
127            RelayCmd::XOFF => {
128                self.inner.handle_incoming_xoff(msg)?;
129                Ok(None)
130            }
131            cmd if cmd_counts_towards_windows(cmd) => {
132                // Discard the returned bool since we aren't sending any more SENDMEs.
133                let _ = self.recv_window.take()?;
134                Ok(Some(msg))
135            }
136            // Nothing to do here.
137            _ => Ok(Some(msg)),
138        }
139    }
140}
tor_proto/stream/flow_ctrl/window/state.rs

tor_proto/stream/flow_ctrl/window/
state.rs
