1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
|
//! Declare the lowest level of stream: a stream that operates on raw
//! cells.
use std::pin::Pin;
use std::task::{Context, Poll};
use futures::Stream;
use pin_project::pin_project;
use tor_async_utils::peekable_stream::UnobtrusivePeekableStream;
use tor_cell::relaycell::{RelayCmd, UnparsedRelayMsg};
use tracing::debug;
use crate::congestion::sendme;
use crate::stream::StreamTarget;
use crate::stream::queue::StreamQueueReceiver;
use crate::{Error, Result};
/// The read part of a stream on a particular circuit.
///
/// This [`Stream`](Stream) will return incoming messages for this Tor stream, excluding flow control
/// related messages like SENDME, XON, and XOFF.
///
/// To avoid ambiguity, the following uses "stream" to refer to the `futures::Stream`, not the Tor
/// stream.
///
/// If the stream ends unexpectedly (before an END message), the stream will return an error.
/// After the stream returns an END message or an error, this stream will be "terminated" and future
/// [`poll_next`](Stream::poll_next) calls will return `None`.
// I think it would be better to *not* return an error if the stream ends before an END message is
// received, and just return `None`. The caller will know if it received an END message or not, so
// returning an error isn't very useful and is maybe unexpected.
#[derive(Debug)]
#[pin_project]
pub struct StreamReceiver {
/// The underlying `StreamTarget` for this stream.
///
/// A reader has this target in order to:
/// * Make the reactor send SENDME messages.
/// * Tell the reactor when there is a protocol error.
/// * Keep the stream alive at least until the StreamReceiver
/// is dropped.
pub(crate) target: StreamTarget,
/// Channel to receive stream messages from the reactor.
#[pin]
pub(crate) receiver: StreamQueueReceiver,
/// Congestion control receive window for this stream.
///
/// Having this here means we're only going to update it when the end consumer of this stream
/// actually reads things, meaning we don't ask for more data until it's actually needed (as
/// opposed to having the reactor assume we're always reading, and potentially overwhelm itself
/// with having to buffer data).
pub(crate) recv_window: sendme::StreamRecvWindow,
/// Whether or not this stream has ended.
pub(crate) ended: bool,
}
impl StreamReceiver {
/// Try to read the next relay message from this stream.
fn poll_next_inner(
mut self: Pin<&mut Self>,
cx: &mut Context<'_>,
) -> Result<Poll<UnparsedRelayMsg>> {
let msg = match self.as_mut().project().receiver.poll_next(cx) {
Poll::Ready(Some(msg)) => msg,
Poll::Ready(None) => {
// The channel is indicating that it has terminated, likely from a dropped sender.
// But if we're here, it means we never received an END cell.
//
// We don't know here if the sender was dropped because either:
// - The stream map dropped its open entry (see `StreamMap::terminate()`).
// For example if `DataStream::close()` was called.
// - The circuit closed and the entire stream map was dropped.
//
// We only know conclusively that the stream is closed.
return Err(Error::NotConnected);
}
Poll::Pending => return Ok(Poll::Pending),
};
if sendme::cell_counts_towards_windows(&msg) && self.recv_window.take()? {
if let Err(e) = self.target.send_sendme() {
if matches!(e, Error::CircuitClosed) {
// If the tunnel has closed, sending a message to the tunnel reactor may fail.
// But this is okay. We still want the user to be able to continue reading the
// remaining queued data for this stream, and if the tunnel has closed it
// wouldn't make sense to send a SENDME message anyways.
debug!("Failed to send stream-level SENDME. Ignoring: {e}");
} else {
// This error is unexpected. Let's return it to the user.
return Err(e);
}
}
self.recv_window.put();
}
Ok(Poll::Ready(msg))
}
/// Shut down this stream.
pub fn protocol_error(&mut self) {
self.target.protocol_error();
}
/// Is the stream currently empty?
///
/// This method is inherently subject to race conditions. More data may arrive even before this
/// method returns, so a result of `true` might have been outdated before the method even
/// returned.
///
/// This takes a `&mut` so that we can peek the stream.
///
/// We provide an `is_empty` method rather than implementing [`UnobtrusivePeekableStream`]
/// directly since `UnobtrusivePeekableStream` allows you to mutate the peeked item, which could
/// break any accounting we do here in `StreamReceiver` (like stream sendme accounting). Also
/// the stream types are incompatible (the inner receiver returns items of `UnparsedRelayMsg`,
/// but this [`StreamReceiver`] returns items of `Result<UnparsedRelayMsg>`).
pub(crate) fn is_empty(&mut self) -> bool {
// The `StreamQueueReceiver` gives us two ways of checking if the queue is empty:
// `unobtrusive_peek().is_none()` and `approx_stream_bytes() == 0`. The peek seems like a
// better approach, so we do that here.
// TODO(arti#534): Should reconsider using `unobtrusive_peek()`. What we really want to know
// is if there is more stream data in the queue. But peeking only tells us if there are more
// messages. There could be more messages, but none of them data messages.
let peek_is_none = Pin::new(&mut self.receiver).unobtrusive_peek().is_none();
// if the peek says that the stream is empty, assert that `approx_stream_bytes()` shows 0
// bytes
#[cfg(debug_assertions)]
if peek_is_none {
assert_eq!(self.receiver.approx_stream_bytes(), 0);
} else {
// if the peek is not empty it doesn't mean that approx_stream_bytes() != 0,
// since there may be messages that contain no stream data
}
peek_is_none
}
}
impl Stream for StreamReceiver {
type Item = Result<UnparsedRelayMsg>;
fn poll_next(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
if self.ended {
// Prevent reading more messages from streams after they've ended. `None` indicates that
// the stream is complete/terminated.
return Poll::Ready(None);
}
match self.as_mut().poll_next_inner(cx) {
Ok(Poll::Pending) => Poll::Pending,
Ok(Poll::Ready(msg)) => {
if msg.cmd() == RelayCmd::END {
// We return the END cell, and future polls will return `None`.
self.ended = true;
}
Poll::Ready(Some(Ok(msg)))
}
Err(e) => {
// We return the error, and future polls will return `None`.
self.ended = true;
Poll::Ready(Some(Err(e)))
}
}
}
}
|
