limbo_core/io/
mod.rs

1use crate::Result;
2use bitflags::bitflags;
3use cfg_block::cfg_block;
4use std::fmt;
5use std::sync::Arc;
6use std::{
7    cell::{Cell, Ref, RefCell, RefMut},
8    fmt::Debug,
9    mem::ManuallyDrop,
10    pin::Pin,
11    rc::Rc,
12};
13
14pub trait File: Send + Sync {
15    fn lock_file(&self, exclusive: bool) -> Result<()>;
16    fn unlock_file(&self) -> Result<()>;
17    fn pread(&self, pos: usize, c: Arc<Completion>) -> Result<()>;
18    fn pwrite(&self, pos: usize, buffer: Arc<RefCell<Buffer>>, c: Arc<Completion>) -> Result<()>;
19    fn sync(&self, c: Arc<Completion>) -> Result<()>;
20    fn size(&self) -> Result<u64>;
21}
22
23#[derive(Debug, Copy, Clone, PartialEq)]
24pub struct OpenFlags(i32);
25
26bitflags! {
27    impl OpenFlags: i32 {
28        const None = 0b00000000;
29        const Create = 0b0000001;
30        const ReadOnly = 0b0000010;
31    }
32}
33
34impl Default for OpenFlags {
35    fn default() -> Self {
36        Self::Create
37    }
38}
39
40pub trait IO: Clock + Send + Sync {
41    fn open_file(&self, path: &str, flags: OpenFlags, direct: bool) -> Result<Arc<dyn File>>;
42
43    fn run_once(&self) -> Result<()>;
44
45    fn wait_for_completion(&self, c: Arc<Completion>) -> Result<()>;
46
47    fn generate_random_number(&self) -> i64;
48
49    fn get_memory_io(&self) -> Arc<MemoryIO>;
50}
51
52pub type Complete = dyn Fn(Arc<RefCell<Buffer>>);
53pub type WriteComplete = dyn Fn(i32);
54pub type SyncComplete = dyn Fn(i32);
55
56pub enum Completion {
57    Read(ReadCompletion),
58    Write(WriteCompletion),
59    Sync(SyncCompletion),
60}
61
62pub struct ReadCompletion {
63    pub buf: Arc<RefCell<Buffer>>,
64    pub complete: Box<Complete>,
65    pub is_completed: Cell<bool>,
66}
67
68impl Completion {
69    pub fn is_completed(&self) -> bool {
70        match self {
71            Self::Read(r) => r.is_completed.get(),
72            Self::Write(w) => w.is_completed.get(),
73            Self::Sync(s) => s.is_completed.get(),
74        }
75    }
76
77    pub fn complete(&self, result: i32) {
78        match self {
79            Self::Read(r) => r.complete(),
80            Self::Write(w) => w.complete(result),
81            Self::Sync(s) => s.complete(result), // fix
82        }
83    }
84
85    /// only call this method if you are sure that the completion is
86    /// a ReadCompletion, panics otherwise
87    pub fn as_read(&self) -> &ReadCompletion {
88        match self {
89            Self::Read(ref r) => r,
90            _ => unreachable!(),
91        }
92    }
93}
94
95pub struct WriteCompletion {
96    pub complete: Box<WriteComplete>,
97    pub is_completed: Cell<bool>,
98}
99
100pub struct SyncCompletion {
101    pub complete: Box<SyncComplete>,
102    pub is_completed: Cell<bool>,
103}
104
105impl ReadCompletion {
106    pub fn new(buf: Arc<RefCell<Buffer>>, complete: Box<Complete>) -> Self {
107        Self {
108            buf,
109            complete,
110            is_completed: Cell::new(false),
111        }
112    }
113
114    pub fn buf(&self) -> Ref<'_, Buffer> {
115        self.buf.borrow()
116    }
117
118    pub fn buf_mut(&self) -> RefMut<'_, Buffer> {
119        self.buf.borrow_mut()
120    }
121
122    pub fn complete(&self) {
123        (self.complete)(self.buf.clone());
124        self.is_completed.set(true);
125    }
126}
127
128impl WriteCompletion {
129    pub fn new(complete: Box<WriteComplete>) -> Self {
130        Self {
131            complete,
132            is_completed: Cell::new(false),
133        }
134    }
135
136    pub fn complete(&self, bytes_written: i32) {
137        (self.complete)(bytes_written);
138        self.is_completed.set(true);
139    }
140}
141
142impl SyncCompletion {
143    pub fn new(complete: Box<SyncComplete>) -> Self {
144        Self {
145            complete,
146            is_completed: Cell::new(false),
147        }
148    }
149
150    pub fn complete(&self, res: i32) {
151        (self.complete)(res);
152        self.is_completed.set(true);
153    }
154}
155
156pub type BufferData = Pin<Vec<u8>>;
157
158pub type BufferDropFn = Rc<dyn Fn(BufferData)>;
159
160#[derive(Clone)]
161pub struct Buffer {
162    data: ManuallyDrop<BufferData>,
163    drop: BufferDropFn,
164}
165
166impl Debug for Buffer {
167    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
168        write!(f, "{:?}", self.data)
169    }
170}
171
172impl Drop for Buffer {
173    fn drop(&mut self) {
174        let data = unsafe { ManuallyDrop::take(&mut self.data) };
175        (self.drop)(data);
176    }
177}
178
179impl Buffer {
180    pub fn allocate(size: usize, drop: BufferDropFn) -> Self {
181        let data = ManuallyDrop::new(Pin::new(vec![0; size]));
182        Self { data, drop }
183    }
184
185    pub fn new(data: BufferData, drop: BufferDropFn) -> Self {
186        let data = ManuallyDrop::new(data);
187        Self { data, drop }
188    }
189
190    pub fn len(&self) -> usize {
191        self.data.len()
192    }
193
194    pub fn is_empty(&self) -> bool {
195        self.data.is_empty()
196    }
197
198    pub fn as_slice(&self) -> &[u8] {
199        &self.data
200    }
201
202    pub fn as_mut_slice(&mut self) -> &mut [u8] {
203        &mut self.data
204    }
205
206    pub fn as_ptr(&self) -> *const u8 {
207        self.data.as_ptr()
208    }
209
210    pub fn as_mut_ptr(&mut self) -> *mut u8 {
211        self.data.as_mut_ptr()
212    }
213}
214
215cfg_block! {
216    #[cfg(all(target_os = "linux", feature = "io_uring"))] {
217        mod io_uring;
218        #[cfg(feature = "fs")]
219        pub use io_uring::UringIO;
220        mod unix;
221        #[cfg(feature = "fs")]
222        pub use unix::UnixIO;
223        pub use unix::UnixIO as SyscallIO;
224        pub use unix::UnixIO as PlatformIO;
225    }
226
227    #[cfg(any(all(target_os = "linux",not(feature = "io_uring")), target_os = "macos"))] {
228        mod unix;
229        #[cfg(feature = "fs")]
230        pub use unix::UnixIO;
231        pub use unix::UnixIO as PlatformIO;
232        pub use PlatformIO as SyscallIO;
233    }
234
235    #[cfg(target_os = "windows")] {
236        mod windows;
237        pub use windows::WindowsIO as PlatformIO;
238        pub use PlatformIO as SyscallIO;
239    }
240
241    #[cfg(not(any(target_os = "linux", target_os = "macos", target_os = "windows")))] {
242        mod generic;
243        pub use generic::GenericIO as PlatformIO;
244        pub use PlatformIO as SyscallIO;
245    }
246}
247
248mod memory;
249#[cfg(feature = "fs")]
250mod vfs;
251pub use memory::MemoryIO;
252pub mod clock;
253mod common;
254pub use clock::Clock;
limbo_core/io/mod.rs

limbo_core/io/
mod.rs
