atlas_tier/direct_swap/
unix.rs

1// SPDX-License-Identifier: AGPL-3.0-only
2
3//! Unix [`DirectSwapFile`]: the real `O_DIRECT` NVMe cold tier, plus its
4//! page-aligned bounce-buffer plumbing. See `mod.rs` for the platform split.
5
6use std::fs::OpenOptions;
7use std::os::fd::{AsRawFd, OwnedFd};
8use std::os::unix::fs::OpenOptionsExt;
9use std::path::Path;
10
11use anyhow::{Result, bail};
12
13use crate::direct_swap::validate_record_bytes;
14use crate::traits::SwapStore;
15
16/// `O_DIRECT` is a Linux-only open flag (macOS has no equivalent — `F_NOCACHE`
17/// is an `fcntl`, not an open flag). The type must still compile on every unix
18/// (the workspace has a macOS/metal CI job); off Linux it simply opens buffered.
19/// That is harmless: the NVMe cold tier only ever runs on the Linux fleet.
20#[cfg(target_os = "linux")]
21const DIRECT_FLAGS: i32 = libc::O_DIRECT;
22#[cfg(not(target_os = "linux"))]
23const DIRECT_FLAGS: i32 = 0;
24
25/// O_DIRECT fixed-stride swap file on NVMe (the peer's cold tier). `record_bytes`
26/// MUST be a 4 KiB multiple (O_DIRECT) — the SSM snapshot blob (66,846,720 B =
27/// 16,320 × 4 KiB) already is. Records are addressed by `disk_slot` at
28/// `disk_slot * record_bytes`; the file grows sparsely as slots are allocated.
29///
30/// Buffers passed to read/write must be page-aligned for O_DIRECT; `read_record`
31/// / `write_record` stage through an internal aligned bounce (a full extra
32/// record memcpy) when they aren't. Both in-tree callers are aligned: the peer
33/// passes its mmap'd arena, and [`crate::Residency`]'s scratch is a
34/// `PageAlignedBuf` — it used to be a plain `Vec`, which took the bounce every
35/// single time.
36///
37/// **A 0-byte swap file is not evidence of a broken write.** Records are only
38/// written when the residency's hot arena has no free slot, so the first
39/// `num_slots` PUTs of distinct keys never touch this file; with the default
40/// 64-slot SSM arena the first `pwrite` happens on spill #65. The size then
41/// jumps to `(highest disk_slot + 1) × record_bytes` — and never shrinks, since
42/// `discard_record` is the documented no-op (freed records are reused by index,
43/// their blocks are never punched back out).
44pub struct DirectSwapFile {
45    fd: OwnedFd,
46    record_bytes: usize,
47    /// Page-aligned bounce for callers whose buffer isn't O_DIRECT-aligned.
48    bounce: AlignedBuf,
49}
50
51impl DirectSwapFile {
52    pub fn create(path: &Path, record_bytes: usize) -> Result<Self> {
53        validate_record_bytes(record_bytes)?;
54        let f = OpenOptions::new()
55            .read(true)
56            .write(true)
57            .create(true)
58            .truncate(true)
59            .custom_flags(DIRECT_FLAGS)
60            .open(path)
61            .map_err(|e| anyhow::anyhow!("open O_DIRECT {}: {e}", path.display()))?;
62        Ok(Self {
63            fd: OwnedFd::from(f),
64            record_bytes,
65            bounce: AlignedBuf::new(record_bytes),
66        })
67    }
68
69    fn offset(&self, disk_slot: usize) -> libc::off_t {
70        (disk_slot as u64 * self.record_bytes as u64) as libc::off_t
71    }
72}
73
74impl SwapStore for DirectSwapFile {
75    fn record_bytes(&self) -> usize {
76        self.record_bytes
77    }
78
79    fn write_record(&mut self, disk_slot: usize, bytes: &[u8]) -> Result<()> {
80        if bytes.len() != self.record_bytes {
81            bail!(
82                "write_record: {} bytes, expected {}",
83                bytes.len(),
84                self.record_bytes
85            );
86        }
87        let off = self.offset(disk_slot);
88        let src = if is_aligned(bytes.as_ptr()) {
89            bytes.as_ptr()
90        } else {
91            self.bounce.as_mut_slice().copy_from_slice(bytes);
92            self.bounce.ptr()
93        };
94        let n = unsafe {
95            libc::pwrite(
96                self.fd.as_raw_fd(),
97                src as *const libc::c_void,
98                self.record_bytes,
99                off,
100            )
101        };
102        if n != self.record_bytes as isize {
103            bail!("pwrite record {disk_slot} returned {n}, errno {}", errno());
104        }
105        Ok(())
106    }
107
108    fn read_record(&self, disk_slot: usize, out: &mut [u8]) -> Result<()> {
109        if out.len() != self.record_bytes {
110            bail!(
111                "read_record: {} bytes, expected {}",
112                out.len(),
113                self.record_bytes
114            );
115        }
116        let off = self.offset(disk_slot);
117        if is_aligned(out.as_ptr()) {
118            let n = unsafe {
119                libc::pread(
120                    self.fd.as_raw_fd(),
121                    out.as_mut_ptr() as *mut libc::c_void,
122                    self.record_bytes,
123                    off,
124                )
125            };
126            if n != self.record_bytes as isize {
127                bail!("pread record {disk_slot} returned {n}, errno {}", errno());
128            }
129        } else {
130            // Stage through the aligned bounce, then copy out. `&self` — the
131            // bounce is interior; take a raw ptr (single-threaded peer loop).
132            let bp = self.bounce.ptr();
133            let n = unsafe {
134                libc::pread(
135                    self.fd.as_raw_fd(),
136                    bp as *mut libc::c_void,
137                    self.record_bytes,
138                    off,
139                )
140            };
141            if n != self.record_bytes as isize {
142                bail!(
143                    "pread(bounce) record {disk_slot} returned {n}, errno {}",
144                    errno()
145                );
146            }
147            unsafe {
148                std::ptr::copy_nonoverlapping(bp, out.as_mut_ptr(), self.record_bytes);
149            }
150        }
151        Ok(())
152    }
153}
154
155fn errno() -> i32 {
156    std::io::Error::last_os_error().raw_os_error().unwrap_or(0)
157}
158
159fn is_aligned(p: *const u8) -> bool {
160    (p as usize) & 0xfff == 0
161}
162
163/// A page-aligned heap buffer (posix_memalign) for O_DIRECT staging.
164struct AlignedBuf {
165    ptr: *mut u8,
166    len: usize,
167}
168unsafe impl Send for AlignedBuf {}
169impl AlignedBuf {
170    fn new(len: usize) -> Self {
171        let mut p: *mut libc::c_void = std::ptr::null_mut();
172        let rc = unsafe { libc::posix_memalign(&mut p, 4096, len) };
173        assert!(
174            rc == 0 && !p.is_null(),
175            "posix_memalign({len}) failed rc={rc}"
176        );
177        Self {
178            ptr: p as *mut u8,
179            len,
180        }
181    }
182    fn ptr(&self) -> *mut u8 {
183        self.ptr
184    }
185    fn as_mut_slice(&mut self) -> &mut [u8] {
186        unsafe { std::slice::from_raw_parts_mut(self.ptr, self.len) }
187    }
188}
189impl Drop for AlignedBuf {
190    fn drop(&mut self) {
191        unsafe { libc::free(self.ptr as *mut libc::c_void) }
192    }
193}
atlas_tier/direct_swap/unix.rs

atlas_tier/direct_swap/
unix.rs
