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
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
//! Queued write side of one client connection.
use std::collections::VecDeque;
use std::sync::{Arc, Condvar, Mutex, MutexGuard};
use std::thread;
use crate::constants::MAX_CLIENT_BACKLOG_BYTES;
use crate::pal::ids::ConnId;
use crate::pal::transport::Transport;
use crate::protocol::Message;
/// Everything the supervisor writes to one client, delivered by its own thread.
///
/// A pipe write blocks while the peer is not draining, so writing to a client
/// directly would let a wedged client hold whichever supervisor path made the
/// write: the output pump, the steal that is trying to replace it, or the exit
/// teardown. Queueing here keeps those paths free and confines the block to the
/// thread that owns the connection.
///
/// Messages are written in the order they were queued, which is what keeps
/// `Attached` ahead of output and `AppExited` behind it.
/// Ref: docs/transport.md.
pub(crate) struct Outbox<T: Transport> {
transport: T,
conn: ConnId,
state: Mutex<OutboxState>,
/// Signals a queued message or a state change.
changed: Condvar,
/// Taken by whoever flushes, so the writer is joined exactly once.
writer: Mutex<Option<thread::JoinHandle<()>>>,
}
/// Queue and lifecycle of one outbox.
#[derive(Debug, Default)]
struct OutboxState {
queued: VecDeque<Message>,
/// Payload bytes waiting to be written, the measure of how far behind the
/// client has fallen.
pending_bytes: usize,
/// The connection has been given up: nothing more is queued or written.
abandoned: bool,
/// No further messages will be queued; the writer stops once it drains.
finished: bool,
}
impl<T: Transport + Clone> Outbox<T> {
/// Starts the writer thread that owns every write to `conn`.
pub(crate) fn start(transport: T, conn: ConnId) -> Arc<Self> {
let outbox = Arc::new(Self {
transport,
conn,
state: Mutex::new(OutboxState::default()),
changed: Condvar::new(),
writer: Mutex::new(None),
});
let writer = thread::spawn({
let outbox = Arc::clone(&outbox);
move || outbox.write_loop()
});
*outbox
.writer
.lock()
.expect("the writer handle is only taken by a flush, never across a panic") =
Some(writer);
outbox
}
/// Queues one message. Never blocks.
///
/// A client that falls far enough behind is abandoned instead: `dure` keeps
/// no screen buffer, so output that cannot be delivered has no later value,
/// and the user recovers the session with a fresh `dure resume`.
pub(crate) fn send(&self, message: Message) {
let overflowed = {
let mut state = self.lock();
if state.abandoned || state.finished {
return;
}
state.pending_bytes = state.pending_bytes.saturating_add(payload_len(&message));
state.queued.push_back(message);
self.changed.notify_all();
state.pending_bytes > MAX_CLIENT_BACKLOG_BYTES
};
if overflowed {
self.abandon();
}
}
/// Gives up on the connection, discarding whatever is still queued.
pub(crate) fn abandon(&self) {
{
let mut state = self.lock();
if state.abandoned {
return;
}
state.abandoned = true;
state.queued.clear();
state.pending_bytes = 0;
self.changed.notify_all();
}
// Cancels a write already in flight, which is what releases a writer
// thread blocked on a client that stopped reading.
self.transport.disconnect(self.conn);
}
/// Stops accepting messages and lets the writer drop the connection once it
/// has written what is already queued.
pub(crate) fn finish(&self) {
let mut state = self.lock();
state.finished = true;
self.changed.notify_all();
}
/// Waits for the writer to stop and drop the connection.
///
/// This delivers nothing itself: [`Outbox::finish`] or [`Outbox::abandon`]
/// is what tells the writer to stop, and this only waits for that to have
/// happened.
///
/// Only the final exit-status delivery waits: by then the session owns no
/// store record, job, or pseudoconsole, so a client that never drains its
/// pipe delays nothing beyond this process outliving it.
pub(crate) fn wait_for_writer(&self) {
let mut writer = self
.writer
.lock()
.expect("the writer handle is only taken by this wait, never across a panic");
if let Some(handle) = writer.take() {
handle
.join()
.expect("the outbox writer thread cannot panic");
}
}
/// Whether this outbox has given up on its connection.
///
/// Lets a test observe abandonment without waiting for the writer, which
/// is the thread abandonment exists to release.
#[cfg(test)]
pub(crate) fn is_abandoned(&self) -> bool {
self.lock().abandoned
}
fn lock(&self) -> MutexGuard<'_, OutboxState> {
self.state
.lock()
.expect("the outbox lock is only held for queue bookkeeping, never across a panic")
}
fn write_loop(&self) {
while let Some(message) = self.take_next() {
if self.transport.send(self.conn, &message).is_err() {
break;
}
}
self.transport.disconnect(self.conn);
}
/// Next message to write, or `None` once the outbox is done.
fn take_next(&self) -> Option<Message> {
let mut state = self.lock();
loop {
if state.abandoned {
return None;
}
if let Some(message) = state.queued.pop_front() {
state.pending_bytes = state.pending_bytes.saturating_sub(payload_len(&message));
return Some(message);
}
if state.finished {
return None;
}
state = self
.changed
.wait(state)
.expect("the outbox lock is only held for queue bookkeeping, never across a panic");
}
}
}
/// Bytes a message contributes to the backlog measure.
fn payload_len(message: &Message) -> usize {
match message {
Message::Output(data) | Message::Input(data) => data.len(),
_ => 0,
}
}
#[cfg(test)]
#[cfg_attr(coverage_nightly, coverage(off))]
mod tests {
use testing::with_watchdog;
use super::*;
use crate::SessionId;
use crate::constants::CONNECT_TIMEOUT;
use crate::pal::transport::MemoryTransport;
/// A connected pair on an in-memory pipe, as supervisor and client ends.
fn pair() -> (MemoryTransport, ConnId, ConnId) {
let transport = MemoryTransport::new();
let name = transport.pipe_name("outbox");
let listener = transport.listen(&name).unwrap();
let client = transport.connect(&name, CONNECT_TIMEOUT).unwrap();
let server = transport.accept(listener).unwrap();
transport.close_listener(listener);
(transport, server, client)
}
#[test]
fn queued_messages_arrive_in_order() {
with_watchdog(|| {
let (transport, server, client) = pair();
let outbox = Outbox::start(transport.clone(), server);
outbox.send(Message::Attached {
session_id: SessionId::MIN,
});
outbox.send(Message::Output(b"hello".to_vec()));
outbox.send(Message::AppExited { status: 7 });
outbox.finish();
outbox.wait_for_writer();
assert!(matches!(
transport.recv(client).unwrap(),
Message::Attached { .. }
));
assert_eq!(
transport.recv(client).unwrap(),
Message::Output(b"hello".to_vec())
);
assert_eq!(
transport.recv(client).unwrap(),
Message::AppExited { status: 7 }
);
});
}
#[test]
fn abandoning_drops_the_queue_and_the_connection() {
with_watchdog(|| {
let (transport, server, client) = pair();
let outbox = Outbox::start(transport.clone(), server);
outbox.abandon();
outbox.wait_for_writer();
// Repeat abandonment is how an overflow and an explicit give-up can
// both land on the same outbox.
outbox.abandon();
outbox.send(Message::Output(b"lost".to_vec()));
transport.recv(client).unwrap_err();
});
}
#[test]
fn a_client_that_never_drains_is_abandoned_rather_than_blocking_the_sender() {
with_watchdog(|| {
let (transport, server, _client) = pair();
// The peer stops draining its pipe, so every write to it blocks.
transport.stall(server);
let outbox = Outbox::start(transport.clone(), server);
let chunk = vec![0_u8; 64 * 1024];
outbox.send(Message::Output(chunk.clone()));
// The writer is now parked on the wedged client, which is the
// state abandonment has to get it out of.
transport.wait_for_stalled_send(server);
// One extra round covers the message the writer already took off
// the queue and is blocked on.
let rounds = MAX_CLIENT_BACKLOG_BYTES.div_euclid(chunk.len()) + 2;
for _ in 0..rounds {
outbox.send(Message::Output(chunk.clone()));
}
// Read before the teardown below, because the teardown would
// abandon it too.
let abandoned_itself = outbox.is_abandoned();
// Teardown the harness owns, so the writer is released whether or
// not the code under test released it. Without this a mutation of
// the backlog check would park this test forever instead of
// failing it, and mutation runs have no watchdog.
transport.resume(server);
outbox.abandon();
outbox.wait_for_writer();
assert!(
abandoned_itself,
"a backlog past the cap must give up on the client on its own"
);
});
}
#[test]
fn a_client_within_the_backlog_cap_keeps_everything_queued_for_it() {
with_watchdog(|| {
// A burst a responsive client can plausibly fall behind by, and the
// scale the cap exists to sit well above. Deliberately an absolute
// size rather than a fraction of the cap, so a cap that no longer
// clears an ordinary burst fails here.
const BURST_BYTES: usize = 2 * 1024 * 1024;
const {
assert!(
BURST_BYTES < MAX_CLIENT_BACKLOG_BYTES,
"the cap must leave room for an ordinary burst"
);
}
let (transport, server, client) = pair();
// The peer stops draining, so the queue accumulates instead of
// being written out as fast as it is filled.
transport.stall(server);
let outbox = Outbox::start(transport.clone(), server);
let chunk = vec![0_u8; 64 * 1024];
let rounds = BURST_BYTES.div_euclid(chunk.len());
for _ in 0..rounds {
outbox.send(Message::Output(chunk.clone()));
}
transport.resume(server);
outbox.finish();
outbox.wait_for_writer();
for _ in 0..rounds {
assert_eq!(
transport.recv(client).unwrap(),
Message::Output(chunk.clone())
);
}
});
}
#[test]
fn finishing_stops_accepting_further_messages() {
with_watchdog(|| {
let (transport, server, client) = pair();
let outbox = Outbox::start(transport.clone(), server);
outbox.finish();
outbox.send(Message::Output(b"late".to_vec()));
outbox.wait_for_writer();
transport.recv(client).unwrap_err();
});
}
#[test]
fn only_payload_messages_count_towards_the_backlog() {
assert_eq!(payload_len(&Message::Output(b"abcd".to_vec())), 4);
assert_eq!(payload_len(&Message::Input(b"ab".to_vec())), 2);
assert_eq!(payload_len(&Message::Displaced), 0);
assert_eq!(payload_len(&Message::AppExited { status: 0 }), 0);
}
}