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
|
use std::collections::{HashMap, VecDeque};
use std::time::Duration;
use zellij_utils::channels;
use zellij_utils::errors::{prelude::*, ContextType, PtyWriteContext};
use crate::route::NotificationEnd;
use crate::thread_bus::Bus;
// we separate these instruction to a different thread because some programs get deadlocked if
// you write into their STDIN while reading from their STDOUT (I'm looking at you, vim)
// while the same has not been observed to happen with resizes, it could conceivably happen and we have this
// here anyway, so
#[derive(Debug, Clone)]
pub enum PtyWriteInstruction {
Write(Vec<u8>, u32, Option<NotificationEnd>),
ResizePty(u32, u16, u16, Option<u16>, Option<u16>),
StartCachingResizes,
ApplyCachedResizes,
Exit,
}
impl From<&PtyWriteInstruction> for PtyWriteContext {
fn from(tty_write_instruction: &PtyWriteInstruction) -> Self {
match *tty_write_instruction {
PtyWriteInstruction::Write(..) => PtyWriteContext::Write,
PtyWriteInstruction::ResizePty(..) => PtyWriteContext::ResizePty,
PtyWriteInstruction::ApplyCachedResizes => PtyWriteContext::ApplyCachedResizes,
PtyWriteInstruction::StartCachingResizes => PtyWriteContext::StartCachingResizes,
PtyWriteInstruction::Exit => PtyWriteContext::Exit,
}
}
}
/// Maximum pending bytes per terminal before dropping the buffer.
///
/// Exposed (but not `pub`) so tests in this module can reference it.
#[allow(dead_code)]
const MAX_PENDING_BYTES: usize = 10 * 1024 * 1024;
/// When pending writes exist, poll for new instructions with this timeout
/// before retrying the drain. Keeps the thread responsive to new instructions
/// while still retrying stuck terminals.
const PENDING_DRAIN_TIMEOUT: Duration = Duration::from_millis(10);
/// A chunk of bytes waiting to be written to a terminal's stdin.
struct PendingWrite {
bytes: Vec<u8>,
offset: usize,
_completion: Option<NotificationEnd>,
}
pub(crate) fn pty_writer_main(bus: Bus<PtyWriteInstruction>) -> Result<()> {
let err_context = || "failed to write to pty".to_string();
let mut pending: HashMap<u32, VecDeque<PendingWrite>> = HashMap::new();
loop {
// If we have pending writes, use a short timeout so we can keep
// draining. Otherwise, block until a new instruction arrives.
let has_pending = pending.values().any(|q| !q.is_empty());
let event = if has_pending {
match bus.recv_timeout(PENDING_DRAIN_TIMEOUT) {
Ok(pair) => Some(pair),
Err(channels::RecvTimeoutError::Timeout) => None,
Err(channels::RecvTimeoutError::Disconnected) => return Ok(()),
}
} else {
Some(bus.recv().with_context(err_context)?)
};
let mut os_input = bus
.os_input
.clone()
.context("no OS input API found")
.with_context(err_context)?;
// Process the instruction if we received one
if let Some((event, mut err_ctx)) = event {
err_ctx.add_call(ContextType::PtyWrite((&event).into()));
match event {
PtyWriteInstruction::Write(bytes, terminal_id, completion) => {
let queue = pending.entry(terminal_id).or_default();
let queued: usize = queue
.iter()
.map(|w| w.bytes.len().saturating_sub(w.offset))
.sum();
if queued + bytes.len() > MAX_PENDING_BYTES {
log::error!(
"dropping write buffer for terminal {} \
({} + {} bytes exceeds {} limit)",
terminal_id,
queued,
bytes.len(),
MAX_PENDING_BYTES,
);
queue.clear();
} else {
queue.push_back(PendingWrite {
bytes,
offset: 0,
_completion: completion,
});
}
},
PtyWriteInstruction::ResizePty(
terminal_id,
columns,
rows,
width_in_pixels,
height_in_pixels,
) => {
os_input
.set_terminal_size_using_terminal_id(
terminal_id,
columns,
rows,
width_in_pixels,
height_in_pixels,
)
.with_context(err_context)
.non_fatal();
},
PtyWriteInstruction::StartCachingResizes => {
// we do this because there are some logic traps inside the screen/tab/layout code
// the cause multiple resizes to be sent to the pty - while the last one is always
// the correct one, many programs and shells debounce those (I guess due to the
// trauma of dealing with GUI resizes of the controlling terminal window), and this
// then causes glitches and missing redraws
// so we do this to play nice and always only send the last resize instruction to
// each pane
// the logic for this happens in the main Screen event loop
os_input.cache_resizes();
},
PtyWriteInstruction::ApplyCachedResizes => {
os_input.apply_cached_resizes();
},
PtyWriteInstruction::Exit => {
return Ok(());
},
}
}
// Drain pending writes — one pass across all terminals.
// For each terminal, write as much as the kernel accepts without
// blocking, then move on to the next terminal.
let terminal_ids: Vec<u32> = pending.keys().copied().collect();
for tid in terminal_ids {
let queue = match pending.get_mut(&tid) {
Some(q) if !q.is_empty() => q,
_ => continue,
};
while let Some(front) = queue.front_mut() {
let remaining = front.bytes.get(front.offset..).unwrap_or_default();
if remaining.is_empty() {
queue.pop_front();
continue;
}
match os_input.write_to_tty_stdin(tid, remaining) {
Ok(0) => break, // EAGAIN — move to next terminal
Ok(n) => {
front.offset = front.offset.saturating_add(n);
if front.offset >= front.bytes.len() {
queue.pop_front();
}
},
Err(e) => {
// Terminal error (EBADF, EIO, etc) — clear its queue
Err::<(), _>(e).with_context(err_context).non_fatal();
queue.clear();
break;
},
}
}
}
pending.retain(|_, q| !q.is_empty());
}
}
#[cfg(test)]
#[path = "./unit/pty_writer_tests.rs"]
mod pty_writer_tests;
|
