mz_ore/pool/file/sys/
linux.rs

1// Copyright Materialize, Inc. and contributors. All rights reserved.
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License in the LICENSE file at the
6// root of this repository, or online at
7//
8//     http://www.apache.org/licenses/LICENSE-2.0
9//
10// Unless required by applicable law or agreed to in writing, software
11// distributed under the License is distributed on an "AS IS" BASIS,
12// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13// See the License for the specific language governing permissions and
14// limitations under the License.
15
16//! The Linux syscalls behind the file store.
17
18use std::ffi::CString;
19use std::fs::{File, OpenOptions};
20use std::io;
21use std::os::fd::AsRawFd;
22use std::os::unix::ffi::OsStrExt;
23use std::os::unix::fs::{FileExt, OpenOptionsExt};
24use std::path::Path;
25use std::sync::atomic::{AtomicU64, Ordering};
26
27use crate::cast::CastFrom;
28
29/// Converts `path` to the NUL-terminated form libc expects. Fails with
30/// `InvalidInput` if `path` contains an interior NUL.
31fn c_path(path: &Path) -> io::Result<CString> {
32    CString::new(path.as_os_str().as_bytes())
33        .map_err(|err| io::Error::new(io::ErrorKind::InvalidInput, err))
34}
35
36/// Names the unlinked fallback file tries before `EEXIST` fails the
37/// open.
38const OPEN_NAMED_ATTEMPTS: usize = 8;
39
40/// Converts a file offset to `off_t`. Fails with `EFBIG` if it does not
41/// fit, the error the kernel reports for an offset past the file size limit.
42fn off_t(offset: u64) -> io::Result<libc::off_t> {
43    libc::off_t::try_from(offset).map_err(|_| io::Error::from_raw_os_error(libc::EFBIG))
44}
45
46/// Whether `path` is on tmpfs or ramfs.
47///
48/// NOTE: this checks the filesystem type of `path` alone. An overlay
49/// whose upper layer is tmpfs reports the overlay type and passes.
50pub(in crate::pool::file) fn is_memory_backed(path: &Path) -> io::Result<bool> {
51    let path = c_path(path)?;
52    let mut st = std::mem::MaybeUninit::<libc::statfs>::uninit();
53    // SAFETY: `path` is NUL-terminated and `st` is valid for writes of
54    // one `statfs`.
55    if unsafe { libc::statfs(path.as_ptr(), st.as_mut_ptr()) } != 0 {
56        return Err(io::Error::last_os_error());
57    }
58    // SAFETY: `statfs` succeeded, so it initialized `st`.
59    let f_type = unsafe { st.assume_init() }.f_type;
60    // From `linux/magic.h`. `f_type`'s width and signedness differ
61    // across targets, and `libc` lacks `RAMFS_MAGIC`.
62    const TMPFS_MAGIC: u32 = 0x0102_1994;
63    const RAMFS_MAGIC: u32 = 0x8584_58f6;
64    let f_type = i128::from(f_type);
65    Ok(f_type == i128::from(TMPFS_MAGIC) || f_type == i128::from(RAMFS_MAGIC))
66}
67
68/// Bytes available to unprivileged writers on the volume holding `path`.
69pub(in crate::pool::file) fn available_bytes(path: &Path) -> io::Result<u64> {
70    let path = c_path(path)?;
71    let mut st = std::mem::MaybeUninit::<libc::statvfs>::uninit();
72    // SAFETY: `path` is NUL-terminated and `st` is valid for writes of
73    // one `statvfs`.
74    if unsafe { libc::statvfs(path.as_ptr(), st.as_mut_ptr()) } != 0 {
75        return Err(io::Error::last_os_error());
76    }
77    // SAFETY: `statvfs` succeeded, so it initialized `st`.
78    let st = unsafe { st.assume_init() };
79    // The field widths differ across targets.
80    #[allow(clippy::useless_conversion)]
81    let (bavail, frsize) = (u64::from(st.f_bavail), u64::from(st.f_frsize));
82    Ok(bavail.saturating_mul(frsize))
83}
84
85/// Opens a nameless read-write file in `dir`, close-on-exec, with
86/// `O_DIRECT` if `direct`.
87pub(in crate::pool::file) fn open_anonymous(
88    dir: &Path,
89    class: usize,
90    direct: bool,
91) -> io::Result<File> {
92    let direct = if direct { libc::O_DIRECT } else { 0 };
93    // `std` opens every file `O_CLOEXEC`.
94    let tmpfile = match tmpfile_fault() {
95        Some(err) => Err(err),
96        None => OpenOptions::new()
97            .read(true)
98            .write(true)
99            .mode(0o600)
100            .custom_flags(libc::O_TMPFILE | direct)
101            .open(dir),
102    };
103    match tmpfile {
104        Err(err) if matches!(err.raw_os_error(), Some(libc::EOPNOTSUPP | libc::EISDIR)) => {
105            // The filesystem lacks `O_TMPFILE`: create a unique name and
106            // unlink it at once, leaving the same lifetime as an
107            // anonymous file. A crash between the two leaves the name
108            // behind, and a restarted container can reuse the pid, so
109            // the name carries the clock too and a collision retries.
110            static NONCE: AtomicU64 = AtomicU64::new(0);
111            let mut attempts = 0;
112            let (file, path) = loop {
113                let nanos = std::time::SystemTime::now()
114                    .duration_since(std::time::UNIX_EPOCH)
115                    .map_or(0, |d| d.as_nanos());
116                let path = dir.join(format!(
117                    ".mz-pool-extents-{}-{nanos}-{class}-{}",
118                    std::process::id(),
119                    NONCE.fetch_add(1, Ordering::Relaxed),
120                ));
121                match OpenOptions::new()
122                    .read(true)
123                    .write(true)
124                    .create_new(true)
125                    .mode(0o600)
126                    .custom_flags(direct)
127                    .open(&path)
128                {
129                    Ok(file) => break (file, path),
130                    Err(err)
131                        if err.kind() == io::ErrorKind::AlreadyExists
132                            && attempts + 1 < OPEN_NAMED_ATTEMPTS =>
133                    {
134                        attempts += 1;
135                    }
136                    Err(err) => return Err(err),
137                }
138            };
139            std::fs::remove_file(&path)?;
140            Ok(file)
141        }
142        result => result,
143    }
144}
145
146/// The injected failure of the next `O_TMPFILE` open, if any.
147fn tmpfile_fault() -> Option<io::Error> {
148    #[cfg(test)]
149    if let Some(result) = crate::pool::file::testing::fault::inject(
150        crate::pool::file::testing::fault::Op::OpenTmpfile,
151    ) {
152        return Some(result.expect_err("an OpenTmpfile fault carries an errno"));
153    }
154    None
155}
156
157/// Allocates blocks for `[offset, offset + len)`.
158pub(in crate::pool::file) fn fallocate(file: &File, offset: u64, len: usize) -> io::Result<()> {
159    #[cfg(test)]
160    if let Some(result) =
161        crate::pool::file::testing::fault::inject(crate::pool::file::testing::fault::Op::Fallocate)
162    {
163        return result.map(|_| ());
164    }
165    fallocate_mode(file, 0, offset, len)
166}
167
168/// Deallocates the blocks of `[offset, offset + len)`, keeping the file
169/// size.
170pub(in crate::pool::file) fn punch_hole(file: &File, offset: u64, len: usize) -> io::Result<()> {
171    #[cfg(test)]
172    if let Some(result) =
173        crate::pool::file::testing::fault::inject(crate::pool::file::testing::fault::Op::Punch)
174    {
175        return result.map(|_| ());
176    }
177    fallocate_mode(
178        file,
179        libc::FALLOC_FL_PUNCH_HOLE | libc::FALLOC_FL_KEEP_SIZE,
180        offset,
181        len,
182    )
183}
184
185fn fallocate_mode(file: &File, mode: libc::c_int, offset: u64, len: usize) -> io::Result<()> {
186    let offset = off_t(offset)?;
187    let len = off_t(u64::cast_from(len))?;
188    loop {
189        // SAFETY: plain syscall on an owned descriptor, no memory
190        // arguments.
191        if unsafe { libc::fallocate(file.as_raw_fd(), mode, offset, len) } == 0 {
192            return Ok(());
193        }
194        let err = io::Error::last_os_error();
195        if err.kind() != io::ErrorKind::Interrupted {
196            return Err(err);
197        }
198    }
199}
200
201/// One `pwrite` of `buf` at `offset`.
202pub(in crate::pool::file) fn pwrite(file: &File, buf: &[u8], offset: u64) -> io::Result<usize> {
203    #[cfg(test)]
204    if let Some(result) =
205        crate::pool::file::testing::fault::inject(crate::pool::file::testing::fault::Op::Write)
206    {
207        return result;
208    }
209    file.write_at(buf, offset)
210}
211
212/// One `pread` into `buf` at `offset`.
213pub(in crate::pool::file) fn pread(file: &File, buf: &mut [u8], offset: u64) -> io::Result<usize> {
214    #[cfg(test)]
215    if let Some(result) =
216        crate::pool::file::testing::fault::inject(crate::pool::file::testing::fault::Op::Read)
217    {
218        return result;
219    }
220    file.read_at(buf, offset)
221}
222
223/// Writes back `[offset, offset + len)` synchronously and drops it from
224/// the page cache.
225pub(in crate::pool::file) fn writeback_and_drop(
226    file: &File,
227    offset: u64,
228    len: usize,
229) -> io::Result<()> {
230    let range_offset = off_t(offset)?;
231    let range_len = off_t(u64::cast_from(len))?;
232    // SAFETY: plain syscall on an owned descriptor, no memory arguments.
233    let synced = unsafe {
234        libc::sync_file_range(
235            file.as_raw_fd(),
236            range_offset,
237            range_len,
238            libc::SYNC_FILE_RANGE_WAIT_BEFORE
239                | libc::SYNC_FILE_RANGE_WRITE
240                | libc::SYNC_FILE_RANGE_WAIT_AFTER,
241        )
242    };
243    if synced != 0 {
244        return Err(io::Error::last_os_error());
245    }
246    drop_cache(file, offset, len)
247}
248
249/// Drops the clean pages of `[offset, offset + len)` from the page cache.
250pub(in crate::pool::file) fn drop_cache(file: &File, offset: u64, len: usize) -> io::Result<()> {
251    let offset = off_t(offset)?;
252    let len = off_t(u64::cast_from(len))?;
253    // SAFETY: plain syscall on an owned descriptor, no memory arguments.
254    let errno =
255        unsafe { libc::posix_fadvise(file.as_raw_fd(), offset, len, libc::POSIX_FADV_DONTNEED) };
256    match errno {
257        0 => Ok(()),
258        errno => Err(io::Error::from_raw_os_error(errno)),
259    }
260}
mz_ore/pool/file/sys/linux.rs

mz_ore/pool/file/sys/
linux.rs
