conmonrs/container_log/
line_buffer.rs

1//! Collecting the output into whole lines, see [`LineBuffer`].
2
3use super::{Line, Text, strip_line_ending};
4use crate::container_io::{Pipe, READ_BUFFER_SIZE};
5use bytes::BytesMut;
6
7/// Collects the output of both pipes into whole lines for journald, which
8/// writes one entry per line.
9///
10/// A read chunk may end anywhere, in the middle of a line, between "\r" and
11/// "\n" or inside a multi byte character. Keeping the incomplete tail until
12/// the next chunk of the same pipe arrives avoids split lines and garbled
13/// characters. A line which grows beyond [`LineBuffer::MAX_LINE`] is written
14/// as a partial one, cut at a character boundary, and so is a line which
15/// waits too long for its ending, see [`LineBuffer::flush`].
16#[derive(Debug, Default)]
17pub(super) struct LineBuffer {
18    pub(super) stdout: BytesMut,
19    stderr: BytesMut,
20}
21
22impl LineBuffer {
23    /// Longer lines get written as partial entries.
24    pub(super) const MAX_LINE: usize = 16 * 1024;
25
26    fn tail(&mut self, pipe: Pipe) -> &mut BytesMut {
27        match pipe {
28            Pipe::StdOut => &mut self.stdout,
29            Pipe::StdErr => &mut self.stderr,
30        }
31    }
32
33    /// Whether a line of the pipe waits for its ending.
34    pub(super) fn has_pending(&self, pipe: Pipe) -> bool {
35        !match pipe {
36            Pipe::StdOut => &self.stdout,
37            Pipe::StdErr => &self.stderr,
38        }
39        .is_empty()
40    }
41
42    /// Add a chunk and return the lines it completes. The whole lines of the
43    /// chunk borrow it, only the end of a line which started in an earlier
44    /// chunk and the start of one which continues in a later one get copied
45    /// into the buffer of the pipe.
46    pub(super) fn push<'a>(&mut self, pipe: Pipe, bytes: &'a [u8]) -> Vec<Line<'a>> {
47        let mut lines = vec![];
48        let mut ends = memchr::memchr_iter(b'\n', bytes).map(|end| end + 1);
49        let mut start = 0;
50        let tail = self.tail(pipe);
51        if !tail.is_empty() {
52            let end = ends.next();
53            start = end.unwrap_or(bytes.len());
54            tail.extend_from_slice(&bytes[..start]);
55            if end.is_some() {
56                let line = tail.split().freeze();
57                lines.push(Line {
58                    text: Text::Buffered(line.slice(..strip_line_ending(&line).len())),
59                    partial: false,
60                });
61            }
62        }
63
64        for end in ends {
65            lines.push(Line {
66                text: Text::Chunk(strip_line_ending(&bytes[start..end])),
67                partial: false,
68            });
69            start = end;
70        }
71        tail.extend_from_slice(&bytes[start..]);
72        lines.extend(Self::cut_long(tail));
73        lines
74    }
75
76    /// Cut a line which grew too long without its ending, see
77    /// [`LineBuffer::cut`], and release the buffer once empty.
78    fn cut_long(tail: &mut BytesMut) -> Option<Line<'static>> {
79        let line = (tail.len() >= Self::MAX_LINE).then(|| Self::cut(tail));
80        Self::shrink(tail);
81        line
82    }
83
84    /// Cut the tail at a point which keeps an unfinished multi byte character
85    /// and a "\r" waiting for its "\n" in the buffer. Invalid bytes are not
86    /// held back, they would stay in front forever.
87    fn cut(tail: &mut BytesMut) -> Line<'static> {
88        let mut end = Self::complete_len(tail);
89        if tail[..end].ends_with(b"\r") {
90            end -= 1;
91        }
92        Line {
93            text: Text::Buffered(tail.split_to(end).freeze()),
94            partial: true,
95        }
96    }
97
98    /// The length of the bytes without the unfinished multi byte character
99    /// they end with, if any. Only the last character matters, which starts
100    /// within the last three bytes if it is unfinished, an invalid byte in
101    /// front of it does not.
102    fn complete_len(bytes: &[u8]) -> usize {
103        let window = bytes.len().saturating_sub(3);
104        let Some(start) = bytes[window..]
105            .iter()
106            .rposition(|b| b & 0xc0 != 0x80)
107            .map(|i| window + i)
108        else {
109            return bytes.len();
110        };
111        match str::from_utf8(&bytes[start..]) {
112            Err(e) if e.error_len().is_none() => start + e.valid_up_to(),
113            _ => bytes.len(),
114        }
115    }
116
117    /// A line longer than a read chunk leaves a large buffer behind, release
118    /// it once drained. An idle container keeps at most a chunk of it.
119    fn shrink(tail: &mut BytesMut) {
120        if tail.is_empty() && tail.capacity() > READ_BUFFER_SIZE {
121            *tail = BytesMut::new();
122        }
123    }
124
125    /// Return the line which waits for its ending as a partial one. The
126    /// output paused, a prompt or progress output without a newline would
127    /// otherwise stay invisible.
128    pub(super) fn flush(&mut self, pipe: Pipe) -> Vec<Line<'static>> {
129        // Like a long line, it keeps an unfinished multi byte character and
130        // a "\r" back for the next chunk, see [`LineBuffer::cut`].
131        let tail = self.tail(pipe);
132        if tail.is_empty() {
133            return vec![];
134        }
135        let line = Self::cut(tail);
136        Self::shrink(tail);
137        if line.text.is_empty() {
138            return vec![];
139        }
140        vec![line]
141    }
142
143    /// The output of the pipe ended, return the line which still waits for
144    /// its ending as a partial one.
145    pub(super) fn finish(&mut self, pipe: Pipe) -> Vec<Line<'static>> {
146        let tail = self.tail(pipe);
147        if tail.is_empty() {
148            return vec![];
149        }
150        // Without a newline, so there is no line ending to strip.
151        let text = Text::Buffered(tail.split().freeze());
152        Self::shrink(tail);
153        vec![Line {
154            text,
155            partial: true,
156        }]
157    }
158}
159
160#[cfg(test)]
161mod tests;
conmonrs/container_log/line_buffer.rs

conmonrs/container_log/
line_buffer.rs
