kui_trace/
file.rs

1//! A recording's file as a [`Transport`] (backlog F249, ADR 0075 decisions
2//! 5 and 6): what was appended since the last poll, or the whole again when
3//! the file was rewritten — shorter than what was read, or its first bytes
4//! changed; and a [`Watch`] that checks the file every so often on a thread
5//! of its own and calls back only when its size or its modification time
6//! moved, so a reader idles at no cost.
7
8use std::fs::File;
9use std::io::{self, Read, Seek, SeekFrom};
10use std::path::{Path, PathBuf};
11use std::sync::Arc;
12use std::sync::atomic::{AtomicBool, Ordering};
13use std::time::{Duration, SystemTime};
14
15use crate::stream::{Polled, Transport};
16
17/// How many of a file's first bytes say it is the recording read so far.
18const HEAD: usize = 64;
19/// How much one poll reads at most, so a reader of a long file draws while
20/// it reads.
21const CHUNK: usize = 8 << 20;
22
23/// A recording's file, read as it grows.
24pub struct FileTransport {
25    path: PathBuf,
26    file: File,
27    offset: u64,
28    head: Vec<u8>,
29    follow: bool,
30}
31
32impl FileTransport {
33    /// The file at `path`, from its start. With `follow`, its end is
34    /// [`Polled::Idle`] — it may still be written — and not
35    /// [`Polled::Ended`].
36    pub fn open(path: impl AsRef<Path>, follow: bool) -> io::Result<Self> {
37        let path = path.as_ref().to_path_buf();
38        Ok(FileTransport {
39            file: File::open(&path)?,
40            path,
41            offset: 0,
42            head: Vec::new(),
43            follow,
44        })
45    }
46
47    pub fn path(&self) -> &Path {
48        &self.path
49    }
50
51    /// Whether the file is still the one read so far: as long as what was
52    /// read, its first bytes the same.
53    fn same(&mut self, len: u64) -> io::Result<bool> {
54        if len < self.offset {
55            return Ok(false);
56        }
57        if self.head.is_empty() {
58            return Ok(true);
59        }
60        let mut now = vec![0; self.head.len()];
61        self.file.seek(SeekFrom::Start(0))?;
62        self.file.read_exact(&mut now)?;
63        Ok(now == self.head)
64    }
65
66    fn read_from(&mut self, at: u64, buf: &mut Vec<u8>) -> io::Result<usize> {
67        self.file.seek(SeekFrom::Start(at))?;
68        let before = buf.len();
69        (&mut self.file).take(CHUNK as u64).read_to_end(buf)?;
70        let n = buf.len() - before;
71        // Reads run on from the start, so the head is filled while what
72        // is read follows straight on from it.
73        if self.head.len() < HEAD && at == self.head.len() as u64 {
74            let take = (HEAD - self.head.len()).min(n);
75            self.head.extend_from_slice(&buf[before..before + take]);
76        }
77        self.offset = at + n as u64;
78        Ok(n)
79    }
80}
81
82impl Transport for FileTransport {
83    fn poll(&mut self, buf: &mut Vec<u8>) -> io::Result<Polled> {
84        // A file replaced under its name is another inode: open the name.
85        let file = File::open(&self.path)?;
86        self.file = file;
87        let len = self.file.metadata()?.len();
88        if !self.same(len)? {
89            self.offset = 0;
90            self.head.clear();
91            self.read_from(0, buf)?;
92            return Ok(Polled::Reset);
93        }
94        if len == self.offset {
95            return Ok(if self.follow {
96                Polled::Idle
97            } else {
98                Polled::Ended
99            });
100        }
101        let n = self.read_from(self.offset, buf)?;
102        Ok(Polled::More(n))
103    }
104}
105
106/// A file watched on a thread of its own: `on_change` runs there whenever
107/// the file's size or modification time moved since the last look. The
108/// thread stops when the `Watch` drops.
109pub struct Watch {
110    stop: Arc<AtomicBool>,
111}
112
113impl Watch {
114    /// Watches `path` every `every` (ADR 0075: 250 ms), as it stands now.
115    pub fn spawn(
116        path: impl AsRef<Path>,
117        every: Duration,
118        on_change: impl Fn() + Send + 'static,
119    ) -> io::Result<Watch> {
120        let path = path.as_ref().to_path_buf();
121        let stop = Arc::new(AtomicBool::new(false));
122        let stopped = stop.clone();
123        let stamp = |p: &Path| -> Option<(u64, SystemTime)> {
124            let m = std::fs::metadata(p).ok()?;
125            Some((m.len(), m.modified().ok()?))
126        };
127        let mut last = stamp(&path);
128        std::thread::Builder::new()
129            .name("kui-trace-watch".into())
130            .spawn(move || {
131                while !stopped.load(Ordering::Relaxed) {
132                    std::thread::sleep(every);
133                    if stopped.load(Ordering::Relaxed) {
134                        break;
135                    }
136                    let now = stamp(&path);
137                    if now != last {
138                        last = now;
139                        on_change();
140                    }
141                }
142            })?;
143        Ok(Watch { stop })
144    }
145}
146
147impl Drop for Watch {
148    fn drop(&mut self) {
149        self.stop.store(true, Ordering::Relaxed);
150    }
151}
kui_trace/file.rs

kui_trace/
file.rs
