katui_rt/
pacing.rs

1//! Output pacing: writes frames in step with what the terminal has
2//! actually consumed, so nothing queues up behind a slow terminal.
3
4use std::io;
5use std::time::Duration;
6
7use katui_present::session;
8use katui_proto::kitty::{self, IdSpace};
9use katui_proto::probe::{Capabilities, ProbeSource};
10use katui_scene::CellSize;
11
12use crate::events::EventPump;
13
14/// A batch size well inside what a multiplexer buffers: rmux 0.10 drops
15/// graphics commands past about 64 KB in flight, and 32 KB batches still
16/// lost the odd tile.
17const MULTIPLEXER_BATCH: usize = 16 * 1024;
18
19/// Bytes per pane cell in one batch. tmux discards its whole output once
20/// it holds eight bytes per cell of the client terminal, which is at
21/// least as big as our pane, so we keep each batch at half of that.
22const BATCH_BYTES_PER_CELL: usize = 4;
23
24/// The first graphics query id a run waits on; each query takes its own
25/// id, because a second client attached to a multiplexer answers too and
26/// its answer must not stand in for the one being waited on.
27const QUERY_IDS_FROM: u32 = 1_000_000;
28const QUERY_IDS_UNTIL: u32 = 0x00f0_0000;
29
30/// How long a run waits for one answer before it stops asking.
31const ANSWER_WAIT: Duration = Duration::from_millis(500);
32
33/// How a run keeps its output in step with the terminal.
34#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
35pub enum Pacing {
36    /// The frame clock alone, for terminals that answer no queries.
37    #[default]
38    Timer,
39    /// Wait for the terminal after every frame.
40    EachFrame,
41    /// Wait after every batch of commands, sized for a multiplexer.
42    Batches,
43    /// Wait for the device attributes answer after every frame, for a
44    /// terminal without kitty graphics that answered the probe.
45    DeviceAttributes,
46}
47
48impl Pacing {
49    /// The pacing `caps` can support.
50    #[must_use]
51    pub fn for_terminal(caps: &Capabilities) -> Self {
52        let answers = caps.trust == ProbeSource::Probed;
53        match (caps.kitty, caps.multiplexed()) {
54            (false, false) if answers => Self::DeviceAttributes,
55            (false, _) => Self::Timer,
56            (true, false) => Self::EachFrame,
57            (true, true) => Self::Batches,
58        }
59    }
60}
61
62/// Delivers frames under a [`Pacing`], dropping to the timer when the
63/// terminal stops answering.
64#[derive(Debug)]
65pub(crate) struct Pacer {
66    configured: Pacing,
67    multiplexed: bool,
68    pacing: Pacing,
69    next_query: u32,
70    space: IdSpace,
71}
72
73impl Pacer {
74    pub(crate) fn new(pacing: Pacing, multiplexed: bool, space: IdSpace) -> Self {
75        Self {
76            configured: pacing,
77            multiplexed,
78            pacing,
79            next_query: QUERY_IDS_FROM,
80            space,
81        }
82    }
83
84    /// Go back to the configured pacing: a pane that lost focus stops
85    /// getting answers, which says nothing about the terminal.
86    pub(crate) fn restore(&mut self) {
87        self.pacing = self.configured;
88    }
89
90    fn take_query_id(&mut self) -> u32 {
91        let local = self.next_query;
92        self.next_query = if local + 1 < QUERY_IDS_UNTIL {
93            local + 1
94        } else {
95            QUERY_IDS_FROM
96        };
97        self.space.image(local).0
98    }
99
100    /// Write `frame`; `false` when this frame lost the terminal's answers.
101    /// The frame is always written whole.
102    pub(crate) fn deliver<W: io::Write, M: Send + 'static>(
103        &mut self,
104        out: &mut W,
105        frame: &[u8],
106        size: CellSize,
107        pump: &mut EventPump<M>,
108    ) -> io::Result<bool> {
109        if frame.is_empty() {
110            return Ok(true);
111        }
112        if self.pacing == Pacing::DeviceAttributes {
113            return self.deliver_with_device_attributes(out, frame, pump);
114        }
115        let parts = if self.pacing == Pacing::Batches {
116            kitty::split_at_commands(frame, batch_limit(size))
117        } else {
118            vec![frame]
119        };
120        let mut answering = true;
121        for part in parts {
122            let mut batch = part.to_vec();
123            let query = (self.pacing != Pacing::Timer && answering).then(|| self.take_query_id());
124            if let Some(query) = query {
125                session::terminal_query(&mut batch, query)?;
126            }
127            if self.multiplexed {
128                batch = kitty::wrap_for_passthrough(&batch);
129            }
130            out.write_all(&batch)?;
131            out.flush()?;
132            if let Some(query) = query {
133                answering = pump.await_answer(query, ANSWER_WAIT);
134            }
135        }
136        if !answering {
137            self.pacing = Pacing::Timer;
138        }
139        Ok(answering)
140    }
141
142    fn deliver_with_device_attributes<W: io::Write, M: Send + 'static>(
143        &mut self,
144        out: &mut W,
145        frame: &[u8],
146        pump: &mut EventPump<M>,
147    ) -> io::Result<bool> {
148        out.write_all(frame)?;
149        out.write_all(DEVICE_ATTRIBUTES_QUERY)?;
150        out.flush()?;
151        let answering = pump.await_device_attributes(ANSWER_WAIT);
152        if !answering {
153            self.pacing = Pacing::Timer;
154        }
155        Ok(answering)
156    }
157}
158
159const DEVICE_ATTRIBUTES_QUERY: &[u8] = b"\x1b[c";
160
161fn batch_limit(size: CellSize) -> usize {
162    let cells = usize::from(size.columns) * usize::from(size.rows);
163    MULTIPLEXER_BATCH.min(cells * BATCH_BYTES_PER_CELL)
164}
165
166#[cfg(test)]
167mod tests {
168    use katui_proto::probe::{Capabilities, Identity};
169    use katui_scene::CellSize;
170
171    use super::Pacing;
172
173    #[test]
174    fn multiplexers_get_batches_and_plain_terminals_no_queries() {
175        let plain = Capabilities::default();
176        assert_eq!(Pacing::for_terminal(&plain), Pacing::Timer);
177        let kitty = Capabilities {
178            kitty: true,
179            ..Capabilities::default()
180        };
181        assert_eq!(Pacing::for_terminal(&kitty), Pacing::EachFrame);
182        let muxed = Capabilities {
183            identity: Some(Identity {
184                name: "rmux".to_owned(),
185                version: None,
186            }),
187            ..kitty
188        };
189        assert_eq!(Pacing::for_terminal(&muxed), Pacing::Batches);
190        let answering_plain = Capabilities {
191            trust: katui_proto::probe::ProbeSource::Probed,
192            ..Capabilities::default()
193        };
194        assert_eq!(
195            Pacing::for_terminal(&answering_plain),
196            Pacing::DeviceAttributes,
197            "a terminal that answered the probe paces on device attributes"
198        );
199    }
200
201    #[test]
202    fn a_multiplexed_pacer_asks_with_ids_of_its_own_space() {
203        let space = katui_proto::kitty::IdSpace::of_process(4242);
204        let mut pacer = super::Pacer::new(Pacing::Batches, true, space);
205        let first = pacer.take_query_id();
206        assert_eq!(first >> 24, space.image(0).0 >> 24);
207        assert_ne!(pacer.take_query_id(), first);
208    }
209
210    #[test]
211    fn a_small_pane_gets_batches_under_the_multiplexer_discard_limit() {
212        let small = CellSize {
213            columns: 80,
214            rows: 24,
215        };
216        assert!(super::batch_limit(small) * 2 <= 80 * 24 * 8);
217        let large = CellSize {
218            columns: 300,
219            rows: 90,
220        };
221        assert_eq!(super::batch_limit(large), super::MULTIPLEXER_BATCH);
222    }
223}
katui_rt/pacing.rs

katui_rt/
pacing.rs
