engramdb_io/
backend.rs

1//! IO 后端抽象：同一 badge 读路径可插拔为「同步 preadv（mac/通用）」或
2//! 「io_uring（Linux，M1.5）」。上层（BadgeGather）只依赖 `IoBackend`。
3//!
4//! 后端语义承诺（与设计 §8 一致）：
5//! - `read_exact_at`：定位读并填满 buf（EOF 为 Err）；
6//! - `read_at`：尽力读（返回实际字节数）。
7
8use std::fs::File;
9use std::io;
10
11/// 跨平台定位读：unix=pread（`read_at`），windows=OVERLAPPED-free 的 `seek_read`
12/// （逻辑等价 pread：设置文件指针后读回塞回原指针语义由 OS 处理）。
13pub fn platform_read_at(f: &File, buf: &mut [u8], off: u64) -> io::Result<usize> {
14    platform_read_at_impl(f, buf, off)
15}
16
17#[cfg(unix)]
18fn platform_read_at_impl(f: &File, buf: &mut [u8], off: u64) -> io::Result<usize> {
19    use std::os::unix::fs::FileExt;
20    f.read_at(buf, off)
21}
22
23#[cfg(windows)]
24fn platform_read_at_impl(f: &File, buf: &mut [u8], off: u64) -> io::Result<usize> {
25    use std::os::windows::fs::FileExt;
26    f.seek_read(buf, off)
27}
28
29#[cfg(not(any(unix, windows)))]
30fn platform_read_at_impl(f: &File, buf: &mut [u8], off: u64) -> io::Result<usize> {
31    let mut f2 = f.try_clone()?;
32    use std::io::{Read, Seek, SeekFrom};
33    f2.seek(SeekFrom::Start(off))?;
34    f2.read(buf)
35}
36
37/// 跨平台"填满读"（EOF 报错），供非 trait 调用方复用。
38pub fn platform_read_exact_at(f: &File, buf: &mut [u8], off: u64) -> io::Result<()> {
39    let mut done = 0usize;
40    while done < buf.len() {
41        let n = platform_read_at(f, &mut buf[done..], off + done as u64)?;
42        if n == 0 {
43            return Err(io::Error::new(
44                io::ErrorKind::UnexpectedEof,
45                "read_exact_at: eof inside buffer",
46            ));
47        }
48        done += n;
49    }
50    Ok(())
51}
52
53/// 标准库定位读后端（unix=pread / windows=seek_read，语义等同；名随历史保留 preadv）。
54pub struct PreadvBackend;
55
56impl IoBackend for PreadvBackend {
57    fn read_exact_at(&self, f: &File, buf: &mut [u8], off: u64) -> io::Result<()> {
58        platform_read_exact_at(f, buf, off)
59    }
60
61    fn read_at(&self, f: &File, buf: &mut [u8], off: u64) -> io::Result<usize> {
62        platform_read_at(f, buf, off)
63    }
64}
65
66pub trait IoBackend: Send + Sync + 'static {
67    fn read_exact_at(&self, f: &File, buf: &mut [u8], off: u64) -> io::Result<()>;
68    fn read_at(&self, f: &File, buf: &mut [u8], off: u64) -> io::Result<usize>;
69
70    /// 批量定位读（默认逐条回退——逐条语义等价，覆盖者可批式提交）。
71    /// 每个 req 必须以 `read_exact_at` 语义填满（长度不足报错）。
72    fn read_many(&self, f: &File, reqs: &mut [(u64, &mut [u8])]) -> io::Result<()> {
73        for (off, buf) in reqs.iter_mut() {
74            self.read_exact_at(f, buf, *off)?;
75        }
76        Ok(())
77    }
78}
79
80/// 平台默认后端。**保持 preadv**（可插拔实验：显式传 UringBackend）。
81/// Linux 的 io_uring 真实现（常驻 ring + 有界提交）M2 落地：UringBackend 提供
82/// per-call 提交语义（正确性/平台能力验证）；批量提交面（batch API）留后端演进。
83pub fn default_backend() -> Box<dyn IoBackend> {
84    Box::new(PreadvBackend)
85}
86
87// ---------- Linux: io_uring（M2）----------
88
89/// io_uring 后端：每线程一个常驻 ring（IoUring 非 Send），每次 read 提交并等待 1 个
90/// 完成。语义与 preadv 一致（尽力读/填满读），复用 trait 无需上层改动。
91/// 注意：per-call 提交在并行度足够（8t 各自 ring）时仍有优势（并法输入复制、多路竞争
92/// 展开），完全批量化（一次 submit N 个 SQE）留 batch API 演进（M1.5+）。
93#[cfg(target_os = "linux")]
94pub struct UringBackend;
95
96/// 批量 io_uring 后端：read_many = 一次 submit N SQE + 整体 wait（逐请求 user_data 索回）。
97/// M2 的"正路"实现——per-call UringBackend 已被 benchmark 判为无增益（-17%）。
98#[cfg(target_os = "linux")]
99pub struct UringBatchBackend;
100
101#[cfg(target_os = "linux")]
102impl IoBackend for UringBatchBackend {
103    fn read_exact_at(&self, f: &File, buf: &mut [u8], off: u64) -> io::Result<()> {
104        platform_read_exact_at(f, buf, off)
105    }
106    fn read_at(&self, f: &File, buf: &mut [u8], off: u64) -> io::Result<usize> {
107        platform_read_at(f, buf, off)
108    }
109    fn read_many(&self, f: &File, reqs: &mut [(u64, &mut [u8])]) -> io::Result<()> {
110        let _ = uring_batch_read(f, reqs)?;
111        Ok(())
112    }
113}
114
115#[cfg(target_os = "linux")]
116fn uring_batch_read(f: &File, reqs: &mut [(u64, &mut [u8])]) -> io::Result<usize> {
117    const DEPTH: u32 = 256;
118    use std::os::fd::AsRawFd;
119    URING.with(|sl| {
120        let mut r = sl.borrow_mut();
121        if r.is_none() {
122            let ring = io_uring::IoUring::new(DEPTH)
123                .map_err(|e| io::Error::other(format!("IoUring::new: {e}")))?;
124            *r = Some(ring);
125        }
126        let ring = r.as_mut().unwrap();
127        let fd = f.as_raw_fd();
128        let mut total = 0usize;
129        for chunk in reqs.chunks_mut(DEPTH as usize) {
130            let expect: Vec<(usize, usize)> = chunk
131                .iter()
132                .enumerate()
133                .map(|(i, (_, b))| (i, b.len()))
134                .collect();
135            for (i, (off, buf)) in chunk.iter_mut().enumerate() {
136                let sqe = io_uring::opcode::Read::new(
137                    io_uring::types::Fd(fd),
138                    buf.as_mut_ptr(),
139                    buf.len() as u32,
140                )
141                .offset(*off);
142                let sqe = sqe.build().user_data(i as u64);
143                unsafe {
144                    ring.submission()
145                        .push(&sqe)
146                        .map_err(|e| io::Error::other(format!("SQ push: {e}")))?;
147                }
148            }
149            let n = chunk.len();
150            ring.submit_and_wait(n)
151                .map_err(|e| io::Error::other(format!("submit_and_wait: {e}")))?;
152            for _ in 0..n {
153                let cqe = ring
154                    .completion()
155                    .next()
156                    .ok_or_else(|| io::Error::new(io::ErrorKind::UnexpectedEof, "no cqe"))?;
157                let res = cqe.result();
158                let idx = cqe.user_data() as usize;
159                let elen = expect.get(idx).map(|x| x.1).unwrap_or(0);
160                if res < 0 {
161                    return Err(io::Error::from_raw_os_error(-res));
162                }
163                if (res as usize) != elen {
164                    return Err(io::Error::new(
165                        io::ErrorKind::UnexpectedEof,
166                        format!("read_many: short read {res} < {elen}"),
167                    ));
168                }
169                total += res as usize;
170            }
171        }
172        Ok(total)
173    })
174}
175
176#[cfg(target_os = "linux")]
177thread_local! {
178    static URING: std::cell::RefCell<Option<io_uring::IoUring>> =
179        const { std::cell::RefCell::new(None) };
180}
181
182#[cfg(target_os = "linux")]
183fn submit_uring_read(f: &std::fs::File, buf: &mut [u8], off: u64) -> io::Result<usize> {
184    use std::os::fd::AsRawFd;
185    URING.with(|sl| {
186        let mut r = sl.borrow_mut();
187        if r.is_none() {
188            let ring = io_uring::IoUring::new(256)
189                .map_err(|e| io::Error::other(format!("IoUring::new: {e}")))?;
190            *r = Some(ring);
191        }
192        let ring = r.as_mut().unwrap();
193        let fd = f.as_raw_fd();
194        let sqe = io_uring::opcode::Read::new(
195            io_uring::types::Fd(fd),
196            buf.as_mut_ptr(),
197            buf.len() as u32,
198        )
199        .offset(off);
200        let sqe = sqe.build().user_data(0);
201        unsafe {
202            ring.submission()
203                .push(&sqe)
204                .map_err(|e| io::Error::other(format!("SQ push: {e}")))?;
205        }
206        ring.submit_and_wait(1)
207            .map_err(|e| io::Error::other(format!("submit_and_wait: {e}")))?;
208        let cqe = ring
209            .completion()
210            .next()
211            .ok_or_else(|| io::Error::new(io::ErrorKind::UnexpectedEof, "no cqe"))?;
212        let res = cqe.result();
213        if res < 0 {
214            return Err(io::Error::from_raw_os_error(-res));
215        }
216        Ok(res as usize)
217    })
218}
219
220#[cfg(target_os = "linux")]
221impl IoBackend for UringBackend {
222    fn read_at(&self, f: &File, buf: &mut [u8], off: u64) -> io::Result<usize> {
223        submit_uring_read(f, buf, off)
224    }
225
226    fn read_exact_at(&self, f: &File, buf: &mut [u8], off: u64) -> io::Result<()> {
227        let mut done = 0usize;
228        while done < buf.len() {
229            let n = self.read_at(f, &mut buf[done..], off + done as u64)?;
230            if n == 0 {
231                return Err(io::Error::new(
232                    io::ErrorKind::UnexpectedEof,
233                    "read_exact_at: eof inside buffer",
234                ));
235            }
236            done += n;
237        }
238        Ok(())
239    }
240}
241
242#[cfg(test)]
243mod tests {
244    use super::*;
245
246    #[test]
247    fn preadv_roundtrip() {
248        let dir = std::env::temp_dir().join("engramdb-backend-test");
249        let _ = std::fs::remove_dir_all(&dir);
250        std::fs::create_dir_all(&dir).unwrap();
251        let p = dir.join("t.bin");
252        std::fs::write(&p, (0u8..255).collect::<Vec<u8>>()).unwrap();
253        let f = std::fs::File::open(&p).unwrap();
254        let b = PreadvBackend;
255        let mut buf = [0u8; 4];
256        b.read_exact_at(&f, &mut buf, 10).unwrap();
257        assert_eq!(&buf, &[10, 11, 12, 13]);
258        let mut p2 = [0u8; 2];
259        let n = b.read_at(&f, &mut p2, 253).unwrap();
260        assert_eq!(n, 2);
261        assert_eq!(&p2, &[253, 254]);
262        let _ = std::fs::remove_dir_all(&dir);
263    }
264
265    #[test]
266    fn default_backend_constructs() {
267        let _ = default_backend();
268    }
269
270    #[cfg(target_os = "linux")]
271    #[test]
272    fn uring_roundtrip_and_semantics() {
273        let dir = std::env::temp_dir().join("engramdb-uring-test");
274        let _ = std::fs::remove_dir_all(&dir);
275        std::fs::create_dir_all(&dir).unwrap();
276        let p = dir.join("t.bin");
277        std::fs::write(&p, (0usize..512).map(|i| i as u8).collect::<Vec<u8>>()).unwrap();
278        let f = std::fs::File::open(&p).unwrap();
279        let b = UringBackend;
280        let mut buf = [0u8; 4];
281        b.read_exact_at(&f, &mut buf, 10).unwrap();
282        assert_eq!(&buf, &[10, 11, 12, 13]);
283        // 跨两次提交偏移正确
284        let mut p2 = [0u8; 2];
285        let n = b.read_at(&f, &mut p2, 253).unwrap();
286        assert_eq!(n, 2);
287        assert_eq!(&p2, &[253, 254]);
288        // EOF 语义：read_at 返回 0？（尝试超出文件尾）
289        let mut p3 = [0u8; 4];
290        let n3 = b.read_at(&f, &mut p3, 512).unwrap();
291        assert_eq!(n3, 0);
292        // read_exact_at EOF -> Err
293        let r = b.read_exact_at(&f, &mut p3, 511);
294        assert!(r.is_err());
295        let _ = std::fs::remove_dir_all(&dir);
296    }
297}
engramdb_io/backend.rs

engramdb_io/
backend.rs
