1mod interface;
47
48#[cfg(any(target_os = "linux", target_os = "macos"))]
49mod posix;
50
51#[cfg(target_os = "windows")]
52mod windows;
53
54use crate::{
55 ack::{AckTicket, Completion, SyncTrigger, TEpoch},
56 error::{BindModule, ErrCode, FrozenError, FrozenResult},
57};
58use interface::FileInterface;
59use std::sync::{Arc, Condvar, Mutex, RwLock, atomic};
60
61#[cfg(any(target_os = "linux", target_os = "macos"))]
62pub(in crate::file) type PlatformFile = posix::POSIXFile;
63
64#[cfg(target_os = "windows")]
65pub(in crate::file) type PlatformFile = windows::WINFile;
66
67pub(in crate::file) mod err {
69 use super::{ErrCode, FrozenError, FrozenResult};
70
71 const ERRDOMAIN: u8 = 0x08;
73
74 pub const HCF: ErrCode = ErrCode::new(0x02, "hault and catch fire");
76
77 pub const UNK: ErrCode = ErrCode::new(0x04, "unknown error");
79
80 pub const NSP: ErrCode = ErrCode::new(0x08, "not enough space available on the storage device");
82
83 pub const SYN: ErrCode = ErrCode::new(0x0A, "failed to sync/flush data to storage device");
85
86 pub const WRT: ErrCode = ErrCode::new(0x0C, "missing permissions for write");
88
89 pub const RED: ErrCode = ErrCode::new(0x0E, "missing permissions for read");
91
92 pub const INV: ErrCode = ErrCode::new(0x10, "invalid path to file");
94
95 pub const CPT: ErrCode = ErrCode::new(0x12, "file is either invalid or corrupted");
97
98 pub const GRW: ErrCode = ErrCode::new(0x14, "unable to zero-extend file");
100
101 pub const LEX: ErrCode =
103 ErrCode::new(0x18, "failed to obtain lock, as no more locks available");
104
105 pub const PRM: ErrCode = ErrCode::new(0x1A, "missing permissions for IO");
107
108 pub const LCK: ErrCode =
110 ErrCode::new(0x1C, "failed to obtain exclusive lock as file may already opened");
111
112 pub const EXS: ErrCode = ErrCode::new(0x1E, "file already exists");
114
115 #[inline]
116 pub(in crate::file) fn raw_error<R, E: std::fmt::Display>(
117 code: ErrCode,
118 error: E,
119 ) -> FrozenResult<R> {
120 let err = FrozenError::unbound_raw(ERRDOMAIN, code, error);
121 Err(err)
122 }
123
124 #[inline]
125 pub(in crate::file) fn default_error<R>(code: ErrCode) -> FrozenResult<R> {
126 let err = FrozenError::unbound(ERRDOMAIN, code, "");
127 Err(err)
128 }
129
130 #[inline]
131 pub(in crate::file) fn make_error(code: ErrCode) -> FrozenError {
132 FrozenError::unbound(ERRDOMAIN, code, "")
133 }
134
135 #[inline]
136 pub(in crate::file) fn make_raw_error<E: std::fmt::Display>(
137 code: ErrCode,
138 error: E,
139 ) -> FrozenError {
140 FrozenError::unbound_raw(ERRDOMAIN, code, error)
141 }
142}
143
144#[cfg(any(target_os = "linux", target_os = "macos"))]
146pub type FileId = libc::c_int;
147
148#[cfg(target_os = "windows")]
155pub type FileId = isize;
156
157#[derive(Debug, Clone)]
178pub struct FileCfg {
179 pub module_id: u8,
181
182 pub path: std::path::PathBuf,
187
188 pub buffer_size: usize,
197
198 pub initial_available_buffers: usize,
202
203 pub sync_interval: Option<std::time::Duration>,
209}
210
211#[derive(Debug)]
212struct FileInner {
213 cfg: FileCfg,
214 file: RwLock<Option<PlatformFile>>,
215 current_length: atomic::AtomicUsize,
216 completion: Arc<Completion>,
217 sync_condvar: Condvar,
218 sync_mutex: Mutex<bool>,
219 shutdown: atomic::AtomicBool,
220}
221
222impl SyncTrigger for FileInner {
223 fn trigger_sync(&self) -> FrozenResult<()> {
224 if self.cfg.sync_interval.is_some() {
225 {
226 let mut guard = self.sync_mutex.lock().unwrap_or_else(|e| e.into_inner());
227 *guard = true;
228 }
229
230 self.sync_condvar.notify_one();
231 Ok(())
232 } else {
233 self.sync_internal()
234 }
235 }
236}
237
238impl FileInner {
239 fn sync_internal(&self) -> FrozenResult<()> {
240 let guard = self.file.write().unwrap_or_else(|e| e.into_inner());
241 let file = match guard.as_ref() {
242 Some(f) => f,
243 None => return Ok(()),
244 };
245
246 let target_epoch = self.completion.read_current_epoch();
247 let durable_epoch = self.completion.read_durable_epoch();
248
249 if target_epoch == durable_epoch {
250 return Ok(());
251 }
252
253 match file.sync() {
254 Ok(()) => {
255 self.completion.mark_epoch_as_durable(target_epoch);
256 self.completion.del_err();
257 self.completion.notify_all_listeners();
258
259 Ok(())
260 }
261 Err(e) => {
262 let e = e.with_module(self.cfg.module_id);
263 self.completion.set_err(e.clone());
264 self.completion.notify_all_listeners();
265
266 Err(e)
267 }
268 }
269 }
270}
271
272#[derive(Debug)]
274pub struct File {
275 cfg: FileCfg,
276 inner: Arc<FileInner>,
277 bg_thread: Option<std::thread::JoinHandle<()>>,
278}
279
280unsafe impl Send for File {}
281unsafe impl Sync for File {}
282
283impl File {
284 fn from_platform_file(
285 cfg: FileCfg,
286 file: PlatformFile,
287 current_length: usize,
288 ) -> FrozenResult<Self> {
289 let completion = Arc::new(Completion::default());
290 let sync_interval = cfg.sync_interval;
291
292 let inner = Arc::new(FileInner {
293 cfg: cfg.clone(),
294 file: RwLock::new(Some(file)),
295 current_length: atomic::AtomicUsize::new(current_length),
296 completion: completion.clone(),
297 sync_condvar: Condvar::new(),
298 sync_mutex: Mutex::new(false),
299 shutdown: atomic::AtomicBool::new(false),
300 });
301
302 let weak_inner: std::sync::Weak<dyn SyncTrigger> =
303 Arc::downgrade(&inner) as std::sync::Weak<dyn SyncTrigger>;
304 completion.set_sync_trigger(weak_inner);
305
306 let bg_thread = if let Some(interval) = sync_interval {
307 let worker_inner = Arc::clone(&inner);
308 let handle =
309 std::thread::Builder::new().name("frozen-sync-worker".into()).spawn(move || {
310 let mut guard =
311 worker_inner.sync_mutex.lock().unwrap_or_else(|e| e.into_inner());
312
313 while !worker_inner.shutdown.load(atomic::Ordering::Acquire) {
314 if *guard {
315 *guard = false;
316 drop(guard);
317
318 let _ = worker_inner.sync_internal();
319 guard =
320 worker_inner.sync_mutex.lock().unwrap_or_else(|e| e.into_inner());
321
322 continue;
323 }
324
325 let (new_guard, _) = worker_inner
326 .sync_condvar
327 .wait_timeout(guard, interval)
328 .unwrap_or_else(|e| e.into_inner());
329
330 guard = new_guard;
331 if worker_inner.shutdown.load(atomic::Ordering::Acquire) {
332 break;
333 }
334
335 *guard = false;
336 drop(guard);
337
338 let _ = worker_inner.sync_internal();
339 guard = worker_inner.sync_mutex.lock().unwrap_or_else(|e| e.into_inner());
340 }
341 });
342
343 let handle = match handle {
344 Ok(h) => h,
345 Err(spawn_err) => {
346 let mut file_guard = inner.file.write().unwrap_or_else(|e| e.into_inner());
347 if let Some(f) = file_guard.take() {
348 let mut err = err::make_raw_error(err::HCF, spawn_err);
349 if let Err(close_err) = f.close() {
350 err.add_suppressed(close_err);
351 }
352
353 return Err(err.with_module(cfg.module_id));
354 }
355
356 return err::raw_error(err::HCF, spawn_err).with_module(cfg.module_id);
357 }
358 };
359
360 Some(handle)
361 } else {
362 None
363 };
364
365 Ok(Self { cfg, inner, bg_thread })
366 }
367
368 pub fn new(cfg: FileCfg) -> FrozenResult<Self> {
401 let mid = cfg.module_id;
402 if cfg.buffer_size == 0 || cfg.initial_available_buffers == 0 {
403 return err::default_error(err::INV).with_module(mid);
404 }
405
406 let init_len = match cfg.buffer_size.checked_mul(cfg.initial_available_buffers) {
407 Some(len) => len,
408 None => return err::default_error(err::INV).with_module(mid),
409 };
410
411 let file = PlatformFile::create(&cfg.path).with_module(mid)?;
412 if let Err(mut e) = file.flock() {
413 if let Err(unlink_err) = file.unlink(&cfg.path) {
414 e.add_suppressed(unlink_err);
415 }
416
417 return Err(e.with_module(mid));
418 }
419
420 if let Err(mut e) = file.grow(0, init_len) {
421 if let Err(unlink_err) = file.unlink(&cfg.path) {
422 e.add_suppressed(unlink_err);
423 }
424
425 return Err(e.with_module(mid));
426 }
427
428 if let Err(mut e) = file.sync() {
429 if let Err(unlink_err) = file.unlink(&cfg.path) {
430 e.add_suppressed(unlink_err);
431 }
432
433 return Err(e.with_module(mid));
434 }
435
436 Self::from_platform_file(cfg, file, init_len)
437 }
438
439 pub fn open(cfg: FileCfg) -> FrozenResult<Self> {
473 let mid = cfg.module_id;
474
475 if cfg.buffer_size == 0 || cfg.initial_available_buffers == 0 {
476 return err::default_error(err::INV).with_module(mid);
477 }
478
479 let init_len = match cfg.buffer_size.checked_mul(cfg.initial_available_buffers) {
480 Some(len) => len,
481 None => return err::default_error(err::INV).with_module(mid),
482 };
483
484 let file = PlatformFile::open(&cfg.path).with_module(mid)?;
485
486 if let Err(mut e) = file.flock() {
487 if let Err(close_err) = file.close() {
488 e.add_suppressed(close_err);
489 }
490
491 return Err(e.with_module(mid));
492 }
493
494 let curr_len = match file.length() {
495 Ok(len) => len,
496 Err(mut e) => {
497 if let Err(close_err) = file.close() {
498 e.add_suppressed(close_err);
499 }
500
501 return Err(e.with_module(mid));
502 }
503 };
504
505 if curr_len < init_len || curr_len % cfg.buffer_size != 0 {
506 let mut e = err::make_error(err::CPT);
507 if let Err(close_err) = file.close() {
508 e.add_suppressed(close_err);
509 }
510
511 return Err(e.with_module(mid));
512 }
513
514 Self::from_platform_file(cfg, file, curr_len)
515 }
516
517 pub fn open_or_create(cfg: FileCfg) -> FrozenResult<Self> {
554 let mid = cfg.module_id;
555
556 if cfg.buffer_size == 0 || cfg.initial_available_buffers == 0 {
557 return err::default_error(err::INV).with_module(mid);
558 }
559
560 let init_len = match cfg.buffer_size.checked_mul(cfg.initial_available_buffers) {
561 Some(len) => len,
562 None => return err::default_error(err::INV).with_module(mid),
563 };
564
565 let file = PlatformFile::new(&cfg.path).with_module(mid)?;
566 if let Err(mut e) = file.flock() {
567 if let Err(close_err) = file.close() {
568 e.add_suppressed(close_err);
569 }
570
571 return Err(e.with_module(mid));
572 }
573
574 let curr_len = match file.length() {
575 Ok(len) => len,
576 Err(mut e) => {
577 if let Err(close_err) = file.close() {
578 e.add_suppressed(close_err);
579 }
580
581 return Err(e.with_module(mid));
582 }
583 };
584
585 if curr_len == 0 {
586 if let Err(mut e) = file.grow(0, init_len) {
587 if let Err(close_err) = file.close() {
588 e.add_suppressed(close_err);
589 }
590
591 return Err(e.with_module(mid));
592 }
593
594 if let Err(mut e) = file.sync() {
595 if let Err(close_err) = file.close() {
596 e.add_suppressed(close_err);
597 }
598
599 return Err(e.with_module(mid));
600 }
601
602 Self::from_platform_file(cfg, file, init_len)
603 } else {
604 if curr_len < init_len || curr_len % cfg.buffer_size != 0 {
605 let mut e = err::make_error(err::CPT);
606 if let Err(close_err) = file.close() {
607 e.add_suppressed(close_err);
608 }
609
610 return Err(e.with_module(mid));
611 }
612
613 Self::from_platform_file(cfg, file, curr_len)
614 }
615 }
616
617 #[inline(always)]
658 pub fn read(&self, buf: &mut [u8], index: usize) -> FrozenResult<()> {
659 let mid = self.cfg.module_id;
660 if buf.is_empty() {
661 return Ok(());
662 }
663
664 if buf.len() % self.cfg.buffer_size != 0 {
665 return err::default_error(err::INV).with_module(mid);
666 }
667
668 let offset = match index.checked_mul(self.cfg.buffer_size) {
669 Some(off) => off,
670 None => return err::default_error(err::INV).with_module(mid),
671 };
672
673 if offset.checked_add(buf.len()).is_none_or(|end| end > self.length()) {
674 return err::default_error(err::HCF).with_module(mid);
675 }
676
677 let guard = self.inner.file.read().unwrap_or_else(|e| e.into_inner());
678 let file = match guard.as_ref() {
679 Some(f) => f,
680 None => return err::default_error(err::INV).with_module(mid),
681 };
682
683 file.pread(buf, offset).with_module(mid)
684 }
685
686 #[inline(always)]
722 pub fn write(&self, buf: &[u8], index: usize) -> FrozenResult<AckTicket> {
723 let mid = self.cfg.module_id;
724 if buf.is_empty() {
725 let current = self.inner.completion.read_current_epoch();
726 return Ok(AckTicket::new(current, self.inner.completion.clone()));
727 }
728
729 if buf.len() % self.cfg.buffer_size != 0 {
730 return err::default_error(err::INV).with_module(mid);
731 }
732
733 let offset = match index.checked_mul(self.cfg.buffer_size) {
734 Some(off) => off,
735 None => return err::default_error(err::INV).with_module(mid),
736 };
737
738 if offset.checked_add(buf.len()).is_none_or(|end| end > self.length()) {
739 return err::default_error(err::HCF).with_module(mid);
740 }
741
742 let guard = self.inner.file.read().unwrap_or_else(|e| e.into_inner());
743 let file = match guard.as_ref() {
744 Some(f) => f,
745 None => return err::default_error(err::INV).with_module(mid),
746 };
747
748 file.pwrite(buf, offset).with_module(mid)?;
749
750 let epoch = self.inner.completion.increment_current_epoch();
751 Ok(AckTicket::new(epoch, self.inner.completion.clone()))
752 }
753
754 pub fn grow(&self, count: usize) -> FrozenResult<()> {
781 let mid = self.cfg.module_id;
782 if count == 0 {
783 return Ok(());
784 }
785
786 let len_to_add = match self.cfg.buffer_size.checked_mul(count) {
787 Some(len) => len,
788 None => return err::default_error(err::GRW).with_module(mid),
789 };
790
791 let guard = self.inner.file.write().unwrap_or_else(|e| e.into_inner());
792 let file = match guard.as_ref() {
793 Some(f) => f,
794 None => return err::default_error(err::INV).with_module(mid),
795 };
796
797 let curr_len = self.inner.current_length.load(atomic::Ordering::Acquire);
798 file.grow(curr_len, len_to_add).with_module(mid)?;
799
800 self.inner.current_length.fetch_add(len_to_add, atomic::Ordering::Release);
801 self.inner.completion.increment_current_epoch();
802
803 Ok(())
804 }
805
806 #[inline]
830 pub fn sync(&self) -> FrozenResult<()> {
831 self.inner.sync_internal()
832 }
833
834 #[cfg(target_os = "linux")]
861 pub fn sync_range(&self, index: usize, count: usize) -> FrozenResult<()> {
862 if count == 0 {
863 return Ok(());
864 }
865
866 let mid = self.cfg.module_id;
867 let offset = match index.checked_mul(self.cfg.buffer_size) {
868 Some(off) => off,
869 None => return err::default_error(err::INV).with_module(mid),
870 };
871 let len_to_sync = match count.checked_mul(self.cfg.buffer_size) {
872 Some(len) => len,
873 None => return err::default_error(err::INV).with_module(mid),
874 };
875
876 let guard = self.inner.file.read().unwrap_or_else(|e| e.into_inner());
877 let file = match guard.as_ref() {
878 Some(f) => f,
879 None => return err::default_error(err::INV).with_module(mid),
880 };
881
882 file.sync_range(offset, len_to_sync).with_module(mid)
883 }
884
885 #[inline]
905 pub fn total_buffers(&self) -> FrozenResult<usize> {
906 let curr_len = self.length();
907 let buffer_size = self.cfg.buffer_size;
908
909 if crate::hints::unlikely(curr_len % buffer_size != 0) {
910 return err::default_error(err::CPT).with_module(self.cfg.module_id);
911 }
912
913 Ok(curr_len / buffer_size)
914 }
915
916 #[inline]
937 pub fn cfg(&self) -> &FileCfg {
938 &self.cfg
939 }
940
941 #[inline]
961 pub fn length(&self) -> usize {
962 self.inner.current_length.load(atomic::Ordering::Acquire)
963 }
964
965 #[inline]
988 pub fn current_epoch(&self) -> TEpoch {
989 self.inner.completion.read_current_epoch()
990 }
991
992 #[inline]
1016 pub fn durable_epoch(&self) -> TEpoch {
1017 self.inner.completion.read_durable_epoch()
1018 }
1019
1020 #[inline]
1041 pub fn completion(&self) -> &Arc<Completion> {
1042 &self.inner.completion
1043 }
1044
1045 #[inline]
1069 pub fn exists(&self) -> FrozenResult<bool> {
1070 PlatformFile::exists(&self.cfg.path).with_module(self.cfg.module_id)
1071 }
1072
1073 pub fn delete(mut self) -> FrozenResult<()> {
1107 let mid = self.cfg.module_id;
1108 self.inner.shutdown.store(true, atomic::Ordering::Release);
1109 {
1110 let mut guard = self.inner.sync_mutex.lock().unwrap_or_else(|e| e.into_inner());
1111 *guard = true;
1112 }
1113 self.inner.sync_condvar.notify_all();
1114
1115 if let Some(handle) = self.bg_thread.take() {
1116 let _ = handle.join();
1117 }
1118
1119 let mut guard = self.inner.file.write().unwrap_or_else(|e| e.into_inner());
1120 let file = match guard.take() {
1121 Some(f) => f,
1122 None => return err::default_error(err::INV).with_module(mid),
1123 };
1124
1125 file.unlink(&self.cfg.path).with_module(mid)
1126 }
1127
1128 #[inline]
1151 pub fn fd(&self) -> FileId {
1152 let guard = self.inner.file.read().unwrap_or_else(|e| e.into_inner());
1153 match guard.as_ref() {
1154 Some(f) => f.fd(),
1155 None => PlatformFile::CLOSED_ID,
1156 }
1157 }
1158}
1159
1160impl Drop for File {
1161 fn drop(&mut self) {
1162 self.inner.shutdown.store(true, atomic::Ordering::Release);
1163 {
1164 let mut guard = self.inner.sync_mutex.lock().unwrap_or_else(|e| e.into_inner());
1165 *guard = true;
1166 }
1167 self.inner.sync_condvar.notify_all();
1168
1169 if let Some(handle) = self.bg_thread.take() {
1170 let _ = handle.join();
1171 }
1172
1173 let is_closed = {
1174 let guard = self.inner.file.read().unwrap_or_else(|e| e.into_inner());
1175 match guard.as_ref() {
1176 Some(f) => f.is_closed(),
1177 None => true,
1178 }
1179 };
1180
1181 if is_closed {
1182 return;
1183 }
1184
1185 let _ = self.inner.sync_internal();
1186 }
1187}
1188
1189#[cfg(test)]
1190mod tests {
1191 use super::*;
1192 use std::sync::{Arc, Barrier};
1193
1194 const MID: u8 = 0;
1195 const BUFFER_SIZE: usize = 0x10;
1196 const INIT_BUFFERS: usize = 4;
1197
1198 fn tmp_path() -> (tempfile::TempDir, FileCfg) {
1199 let dir = tempfile::tempdir().unwrap();
1200 let path = dir.path().join("tmp_file");
1201 let cfg = FileCfg {
1202 module_id: MID,
1203 path,
1204 buffer_size: BUFFER_SIZE,
1205 initial_available_buffers: INIT_BUFFERS,
1206 sync_interval: None,
1207 };
1208
1209 (dir, cfg)
1210 }
1211
1212 mod file_new {
1213 use super::*;
1214
1215 #[test]
1216 fn ok_new_creates_file_with_correct_length() {
1217 let (_dir, cfg) = tmp_path();
1218 let file = File::new(cfg.clone()).unwrap();
1219
1220 let expected_len = BUFFER_SIZE * INIT_BUFFERS;
1221 assert_eq!(file.length(), expected_len);
1222 assert_eq!(file.cfg().buffer_size, BUFFER_SIZE);
1223 assert_ne!(file.fd(), PlatformFile::CLOSED_ID);
1224 assert!(cfg.path.exists());
1225 }
1226
1227 #[test]
1228 fn err_new_when_already_exists() {
1229 let (_dir, cfg) = tmp_path();
1230 let file = File::new(cfg.clone()).unwrap();
1231 drop(file);
1232
1233 let err = File::new(cfg).unwrap_err();
1234 assert_eq!(err.reason, err::EXS.reason);
1235 }
1236
1237 #[test]
1238 fn err_new_invalid_cfg() {
1239 let (_dir, mut cfg) = tmp_path();
1240
1241 cfg.buffer_size = 0;
1242 let err = File::new(cfg.clone()).unwrap_err();
1243 assert_eq!(err.reason, err::INV.reason);
1244
1245 cfg.buffer_size = BUFFER_SIZE;
1246 cfg.initial_available_buffers = 0;
1247 let err = File::new(cfg).unwrap_err();
1248 assert_eq!(err.reason, err::INV.reason);
1249 }
1250
1251 #[test]
1252 fn err_new_cfg_overflow() {
1253 let (_dir, mut cfg) = tmp_path();
1254 cfg.buffer_size = usize::MAX;
1255 cfg.initial_available_buffers = 2;
1256 let err = File::new(cfg).unwrap_err();
1257 assert_eq!(err.reason, err::INV.reason);
1258 }
1259
1260 #[test]
1261 fn err_new_missing_parent_dir() {
1262 let (_dir, mut cfg) = tmp_path();
1263 cfg.path = cfg.path.join("missing/sub/dir/file.db");
1264
1265 let err = File::new(cfg).unwrap_err();
1266 assert_eq!(err.reason, err::INV.reason);
1267 }
1268
1269 #[test]
1270 fn err_new_cleans_up_on_grow_failure() {
1271 let (_dir, mut cfg) = tmp_path();
1272 cfg.buffer_size = 1 << 0x3F;
1275 cfg.initial_available_buffers = 1;
1276
1277 let err = File::new(cfg.clone()).unwrap_err();
1278 assert_eq!(err.reason, err::GRW.reason);
1279
1280 assert!(!cfg.path.exists());
1282
1283 cfg.buffer_size = BUFFER_SIZE;
1285 cfg.initial_available_buffers = INIT_BUFFERS;
1286 let file = File::new(cfg);
1287 assert!(file.is_ok());
1288 }
1289 }
1290
1291 mod file_open {
1292 use super::*;
1293
1294 #[test]
1295 fn ok_open_existing_valid_file() {
1296 let (_dir, cfg) = tmp_path();
1297
1298 {
1299 let file = File::new(cfg.clone()).unwrap();
1300 assert_eq!(file.length(), BUFFER_SIZE * INIT_BUFFERS);
1301 }
1302
1303 let file = File::open(cfg).unwrap();
1304 assert_eq!(file.length(), BUFFER_SIZE * INIT_BUFFERS);
1305 }
1306
1307 #[test]
1308 fn err_open_missing_file() {
1309 let (_dir, cfg) = tmp_path();
1310
1311 let err = File::open(cfg).unwrap_err();
1312 assert_eq!(err.reason, err::INV.reason);
1313 }
1314
1315 #[test]
1316 fn err_open_invalid_cfg() {
1317 let (_dir, mut cfg) = tmp_path();
1318
1319 cfg.buffer_size = 0;
1320 let err = File::open(cfg.clone()).unwrap_err();
1321 assert_eq!(err.reason, err::INV.reason);
1322
1323 cfg.buffer_size = BUFFER_SIZE;
1324 cfg.initial_available_buffers = 0;
1325 let err = File::open(cfg).unwrap_err();
1326 assert_eq!(err.reason, err::INV.reason);
1327 }
1328
1329 #[test]
1330 fn err_open_cfg_overflow() {
1331 let (_dir, mut cfg) = tmp_path();
1332 cfg.buffer_size = usize::MAX;
1333 cfg.initial_available_buffers = 2;
1334 let err = File::open(cfg).unwrap_err();
1335 assert_eq!(err.reason, err::INV.reason);
1336 }
1337
1338 #[test]
1339 fn err_open_when_file_smaller_than_init_len() {
1340 let (_dir, cfg) = tmp_path();
1341
1342 std::fs::write(&cfg.path, []).unwrap();
1343
1344 let err = File::open(cfg).unwrap_err();
1345 assert_eq!(err.reason, err::CPT.reason);
1346 }
1347
1348 #[test]
1349 fn err_open_when_file_not_buffer_multiple() {
1350 let (_dir, cfg) = tmp_path();
1351
1352 let non_aligned_len = (BUFFER_SIZE * INIT_BUFFERS) + 3;
1353 std::fs::write(&cfg.path, vec![0u8; non_aligned_len]).unwrap();
1354
1355 let err = File::open(cfg).unwrap_err();
1356 assert_eq!(err.reason, err::CPT.reason);
1357 }
1358 }
1359
1360 mod file_open_or_create {
1361 use super::*;
1362
1363 #[test]
1364 fn ok_creates_when_missing() {
1365 let (_dir, cfg) = tmp_path();
1366 assert!(!cfg.path.exists());
1367
1368 let file = File::open_or_create(cfg.clone()).unwrap();
1369 assert!(cfg.path.exists());
1370 assert_eq!(file.length(), BUFFER_SIZE * INIT_BUFFERS);
1371 assert!(file.exists().unwrap());
1372 }
1373
1374 #[test]
1375 fn ok_opens_when_existing() {
1376 let (_dir, cfg) = tmp_path();
1377 let file = File::new(cfg.clone()).unwrap();
1378 let data = [0x5Au8; BUFFER_SIZE];
1379 file.write(&data, 0).unwrap();
1380 file.sync().unwrap();
1381 drop(file);
1382
1383 let opened = File::open_or_create(cfg.clone()).unwrap();
1384 assert_eq!(opened.length(), BUFFER_SIZE * INIT_BUFFERS);
1385 let mut buf = [0u8; BUFFER_SIZE];
1386 opened.read(&mut buf, 0).unwrap();
1387 assert_eq!(buf, data);
1388 }
1389
1390 #[test]
1391 fn err_open_or_create_invalid_cfg() {
1392 let (_dir, mut cfg) = tmp_path();
1393 cfg.buffer_size = 0;
1394 let err = File::open_or_create(cfg.clone()).unwrap_err();
1395 assert_eq!(err.reason, err::INV.reason);
1396
1397 cfg.buffer_size = BUFFER_SIZE;
1398 cfg.initial_available_buffers = 0;
1399 let err = File::open_or_create(cfg).unwrap_err();
1400 assert_eq!(err.reason, err::INV.reason);
1401 }
1402
1403 #[test]
1404 fn err_open_or_create_cfg_overflow() {
1405 let (_dir, mut cfg) = tmp_path();
1406 cfg.buffer_size = usize::MAX;
1407 cfg.initial_available_buffers = 2;
1408 let err = File::open_or_create(cfg).unwrap_err();
1409 assert_eq!(err.reason, err::INV.reason);
1410 }
1411
1412 #[test]
1413 fn err_open_or_create_corrupt_existing() {
1414 let (_dir, cfg) = tmp_path();
1415 let non_aligned = (BUFFER_SIZE * INIT_BUFFERS) + 1;
1416 std::fs::write(&cfg.path, vec![0u8; non_aligned]).unwrap();
1417
1418 let err = File::open_or_create(cfg).unwrap_err();
1419 assert_eq!(err.reason, err::CPT.reason);
1420 }
1421
1422 #[test]
1423 fn err_open_or_create_when_locked() {
1424 let (_dir, cfg) = tmp_path();
1425 let file = File::open_or_create(cfg.clone()).unwrap();
1426
1427 let err = File::open_or_create(cfg).unwrap_err();
1428 assert_eq!(err.reason, err::LCK.reason);
1429 drop(file);
1430 }
1431 }
1432
1433 mod file_toctou_and_concurrency {
1434 use super::*;
1435
1436 #[test]
1437 fn err_concurrent_new_exact_same_path() {
1438 let (_dir, cfg) = tmp_path();
1439 let barrier = Arc::new(Barrier::new(2));
1440
1441 let t1 = {
1442 let cfg = cfg.clone();
1443 let b = barrier.clone();
1444 std::thread::spawn(move || {
1445 b.wait();
1446 File::new(cfg)
1447 })
1448 };
1449
1450 let t2 = {
1451 let b = barrier;
1452 std::thread::spawn(move || {
1453 b.wait();
1454 File::new(cfg)
1455 })
1456 };
1457
1458 let r1 = t1.join().unwrap();
1459 let r2 = t2.join().unwrap();
1460
1461 match (r1, r2) {
1465 (Ok(f), Err(e)) | (Err(e), Ok(f)) => {
1466 assert_eq!(e.reason, err::EXS.reason);
1467 assert_eq!(f.length(), BUFFER_SIZE * INIT_BUFFERS);
1468 }
1469 (Ok(_), Ok(_)) => {
1470 panic!("TOCTOU violation: both concurrent File::new succeeded on same path!");
1471 }
1472 (Err(e1), Err(e2)) => {
1473 panic!("Unexpected failure of both threads: {:?}, {:?}", e1, e2);
1474 }
1475 }
1476 }
1477
1478 #[test]
1479 fn err_concurrent_new_and_open() {
1480 let (_dir, cfg) = tmp_path();
1481 let barrier = Arc::new(Barrier::new(2));
1482
1483 let t_new = {
1484 let cfg = cfg.clone();
1485 let b = barrier.clone();
1486 std::thread::spawn(move || {
1487 b.wait();
1488 File::new(cfg)
1489 })
1490 };
1491
1492 let t_open = {
1493 let b = barrier;
1494 std::thread::spawn(move || {
1495 b.wait();
1496 File::open(cfg)
1497 })
1498 };
1499
1500 let r_new = t_new.join().unwrap();
1501 let r_open = t_open.join().unwrap();
1502
1503 let _file = r_new.expect("File::new should succeed");
1504
1505 if let Err(e) = r_open {
1508 assert!(
1509 e.reason == err::INV.reason
1510 || e.reason == err::LCK.reason
1511 || e.reason == err::CPT.reason,
1512 "unexpected error: {:?}",
1513 e
1514 );
1515 }
1516 }
1517
1518 #[test]
1519 fn err_concurrent_open_same_file() {
1520 let (_dir, cfg) = tmp_path();
1521
1522 {
1523 let file = File::new(cfg.clone()).unwrap();
1524 assert_eq!(file.length(), BUFFER_SIZE * INIT_BUFFERS);
1525 }
1526
1527 let barrier = Arc::new(Barrier::new(2));
1528
1529 let t1 = {
1530 let cfg = cfg.clone();
1531 let b = barrier.clone();
1532 std::thread::spawn(move || {
1533 b.wait();
1534 File::open(cfg)
1535 })
1536 };
1537
1538 let t2 = {
1539 let b = barrier;
1540 std::thread::spawn(move || {
1541 b.wait();
1542 File::open(cfg)
1543 })
1544 };
1545
1546 let r1 = t1.join().unwrap();
1547 let r2 = t2.join().unwrap();
1548
1549 match (r1, r2) {
1553 (Ok(_), Err(e)) | (Err(e), Ok(_)) => {
1554 assert_eq!(e.reason, err::LCK.reason, "error must be LCK: {:?}", e);
1555 }
1556 (Ok(_), Ok(_)) => {
1557 panic!(
1558 "Locking violation: both concurrent File::open acquired exclusive lock!"
1559 );
1560 }
1561 (Err(e1), Err(e2)) => {
1562 panic!("Unexpected failure of both threads: {:?}, {:?}", e1, e2);
1563 }
1564 }
1565 }
1566
1567 #[test]
1568 fn err_new_while_file_open() {
1569 let (_dir, cfg) = tmp_path();
1570 let file = File::new(cfg.clone()).unwrap();
1571
1572 let err = File::new(cfg).unwrap_err();
1573 assert_eq!(err.reason, err::EXS.reason);
1574
1575 drop(file);
1576 }
1577
1578 #[test]
1579 fn ok_reopen_after_drop() {
1580 let (_dir, cfg) = tmp_path();
1581 let file = File::new(cfg.clone()).unwrap();
1582 drop(file);
1583
1584 let reopened = File::open(cfg).unwrap();
1585 assert_eq!(reopened.length(), BUFFER_SIZE * INIT_BUFFERS);
1586 }
1587
1588 #[test]
1589 fn err_concurrent_open_during_and_after_delete() {
1590 let (_dir, cfg) = tmp_path();
1591 let file = File::new(cfg.clone()).unwrap();
1592 file.grow(2).unwrap();
1593
1594 let running = Arc::new(atomic::AtomicBool::new(true));
1595
1596 let opener = {
1598 let cfg = cfg.clone();
1599 let running = running.clone();
1600 std::thread::spawn(move || {
1601 let mut attempts = 0;
1602 while running.load(atomic::Ordering::Relaxed) || attempts < 0x32 {
1603 attempts += 1;
1604 let res = File::open(cfg.clone());
1605 if let Err(err) = res {
1611 assert!(
1612 err.reason == err::LCK.reason
1613 || err.reason == err::INV.reason
1614 || err.reason == err::PRM.reason,
1615 "Unexpected error reason: {:?}",
1616 err
1617 );
1618 }
1619 std::thread::yield_now();
1620 }
1621 })
1622 };
1623
1624 std::thread::sleep(std::time::Duration::from_millis(5));
1626 file.delete().unwrap();
1627 running.store(false, atomic::Ordering::Relaxed);
1628
1629 opener.join().unwrap();
1630
1631 let err = File::open(cfg).unwrap_err();
1633 assert_eq!(err.reason, err::INV.reason);
1634 }
1635
1636 #[test]
1637 fn ok_concurrent_new_fails_before_delete_and_succeeds_after() {
1638 let (_dir, cfg) = tmp_path();
1639 let file = File::new(cfg.clone()).unwrap();
1640 file.grow(2).unwrap();
1641
1642 let deleted = Arc::new(atomic::AtomicBool::new(false));
1643
1644 let creator = {
1645 let cfg = cfg.clone();
1646 let deleted = deleted.clone();
1647 std::thread::spawn(move || {
1648 let mut created = None;
1649 while created.is_none() {
1652 match File::new(cfg.clone()) {
1653 Ok(f) => {
1654 assert!(
1655 deleted.load(atomic::Ordering::Acquire),
1656 "File::new succeeded before file.delete completed!"
1657 );
1658 created = Some(f);
1659 }
1660 Err(e) => {
1661 assert!(
1662 e.reason == err::EXS.reason || e.reason == err::PRM.reason,
1663 "Unexpected error reason during race: {:?}",
1664 e
1665 );
1666 std::thread::yield_now();
1667 }
1668 }
1669 }
1670 created.unwrap()
1671 })
1672 };
1673
1674 std::thread::sleep(std::time::Duration::from_millis(0x0A));
1675 deleted.store(true, atomic::Ordering::Release);
1676 file.delete().unwrap();
1677
1678 let new_file = creator.join().unwrap();
1679 assert!(new_file.exists().unwrap());
1680 new_file.delete().unwrap();
1681 }
1682
1683 #[test]
1684 fn ok_concurrent_open_or_create_same_file() {
1685 let (_dir, cfg) = tmp_path();
1686 let num_threads = 8;
1687 let barrier = Arc::new(Barrier::new(num_threads));
1688 let mut handles = vec![];
1689
1690 for _ in 0..num_threads {
1691 let c = cfg.clone();
1692 let b = barrier.clone();
1693 handles.push(std::thread::spawn(move || {
1694 b.wait();
1695 File::open_or_create(c)
1696 }));
1697 }
1698
1699 let mut opened_files = vec![];
1700 let mut lock_failures = 0;
1701
1702 for h in handles {
1703 match h.join().unwrap() {
1704 Ok(f) => {
1705 assert_eq!(f.length(), BUFFER_SIZE * INIT_BUFFERS);
1706 opened_files.push(f);
1707 }
1708 Err(e) => {
1709 assert_eq!(e.reason, err::LCK.reason, "error must be LCK: {:?}", e);
1710 lock_failures += 1;
1711 }
1712 }
1713 }
1714
1715 assert_eq!(opened_files.len(), 1);
1717 assert_eq!(lock_failures, num_threads - 1);
1718 assert!(cfg.path.exists());
1719 }
1720
1721 #[test]
1722 fn ok_file_send_and_sync() {
1723 fn assert_send_sync<T: Send + Sync>() {}
1724 assert_send_sync::<File>();
1725 }
1726 }
1727
1728 mod file_write_read {
1729 use super::*;
1730
1731 #[test]
1732 fn ok_single_buffer_write_read() {
1733 let (_dir, cfg) = tmp_path();
1734 let file = File::new(cfg).unwrap();
1735
1736 let data = [0xABu8; BUFFER_SIZE];
1737 file.write(&data, 2).unwrap();
1738 file.sync().unwrap();
1739
1740 let mut buf = [0u8; BUFFER_SIZE];
1741 file.read(&mut buf, 2).unwrap();
1742 assert_eq!(buf, data);
1743 }
1744
1745 #[test]
1746 fn ok_multi_buffer_write_read() {
1747 let (_dir, cfg) = tmp_path();
1748 let file = File::new(cfg).unwrap();
1749
1750 let multi_buf_data = [0x77u8; BUFFER_SIZE * 2];
1751 file.write(&multi_buf_data, 1).unwrap();
1752 file.sync().unwrap();
1753
1754 let mut read_buf = [0u8; BUFFER_SIZE * 2];
1755 file.read(&mut read_buf, 1).unwrap();
1756 assert_eq!(read_buf, multi_buf_data);
1757
1758 let mut chunk1 = [0u8; BUFFER_SIZE];
1759 let mut chunk2 = [0u8; BUFFER_SIZE];
1760 file.read(&mut chunk1, 1).unwrap();
1761 file.read(&mut chunk2, 2).unwrap();
1762 assert_eq!(chunk1, [0x77u8; BUFFER_SIZE]);
1763 assert_eq!(chunk2, [0x77u8; BUFFER_SIZE]);
1764 }
1765
1766 #[test]
1767 fn ok_empty_buffer_noop() {
1768 let (_dir, cfg) = tmp_path();
1769 let file = File::new(cfg).unwrap();
1770
1771 assert!(file.write(&[], 0).is_ok());
1772 assert!(file.read(&mut [], 0).is_ok());
1773 }
1774
1775 #[test]
1776 fn err_write_unaligned_buffer_size() {
1777 let (_dir, cfg) = tmp_path();
1778 let file = File::new(cfg).unwrap();
1779
1780 let data = [0u8; BUFFER_SIZE - 1];
1781 let err = file.write(&data, 0).unwrap_err();
1782 assert_eq!(err.reason, err::INV.reason);
1783
1784 let data_multi_unaligned = [0u8; (BUFFER_SIZE * 2) + 1];
1785 let err = file.write(&data_multi_unaligned, 0).unwrap_err();
1786 assert_eq!(err.reason, err::INV.reason);
1787 }
1788
1789 #[test]
1790 fn err_read_unaligned_buffer_size() {
1791 let (_dir, cfg) = tmp_path();
1792 let file = File::new(cfg).unwrap();
1793
1794 let mut buf = [0u8; BUFFER_SIZE - 1];
1795 let err = file.read(&mut buf, 0).unwrap_err();
1796 assert_eq!(err.reason, err::INV.reason);
1797
1798 let mut buf_multi_unaligned = [0u8; (BUFFER_SIZE * 2) + 3];
1799 let err = file.read(&mut buf_multi_unaligned, 0).unwrap_err();
1800 assert_eq!(err.reason, err::INV.reason);
1801 }
1802
1803 #[test]
1804 fn err_read_past_eof() {
1805 let (_dir, cfg) = tmp_path();
1806 let file = File::new(cfg).unwrap();
1807
1808 let mut buf = [0u8; BUFFER_SIZE];
1809 let err = file.read(&mut buf, INIT_BUFFERS).unwrap_err();
1810 assert_eq!(err.reason, err::HCF.reason);
1811
1812 let mut multi_buf = [0u8; BUFFER_SIZE * 2];
1813 let err = file.read(&mut multi_buf, INIT_BUFFERS - 1).unwrap_err();
1814 assert_eq!(err.reason, err::HCF.reason);
1815 }
1816
1817 #[test]
1818 fn err_write_past_eof() {
1819 let (_dir, cfg) = tmp_path();
1820 let file = File::new(cfg).unwrap();
1821
1822 let data = [1u8; BUFFER_SIZE];
1823 let err = file.write(&data, INIT_BUFFERS).unwrap_err();
1824 assert_eq!(err.reason, err::HCF.reason);
1825
1826 let multi_data = [1u8; BUFFER_SIZE * 2];
1827 let err = file.write(&multi_data, INIT_BUFFERS - 1).unwrap_err();
1828 assert_eq!(err.reason, err::HCF.reason);
1829 }
1830
1831 #[test]
1832 fn ok_concurrent_non_overlapping_writes() {
1833 let (_dir, mut cfg) = tmp_path();
1834 cfg.initial_available_buffers = 0x10;
1835 let file = Arc::new(File::new(cfg).unwrap());
1836
1837 let mut handles = vec![];
1838 for i in 0..4 {
1839 let f = file.clone();
1840 handles.push(std::thread::spawn(move || {
1841 let data = [i as u8 + 0x0A; BUFFER_SIZE * 2];
1842 f.write(&data, i * 2).unwrap();
1843 }));
1844 }
1845
1846 for h in handles {
1847 h.join().unwrap();
1848 }
1849
1850 file.sync().unwrap();
1851
1852 for i in 0..4 {
1853 let mut buf = [0u8; BUFFER_SIZE * 2];
1854 file.read(&mut buf, i * 2).unwrap();
1855 assert_eq!(buf, [i as u8 + 0x0A; BUFFER_SIZE * 2]);
1856 }
1857 }
1858
1859 #[test]
1860 fn ok_concurrent_readers() {
1861 let (_dir, cfg) = tmp_path();
1862 let file = Arc::new(File::new(cfg).unwrap());
1863
1864 let data = [0x42u8; BUFFER_SIZE * 2];
1865 file.write(&data, 0).unwrap();
1866 file.sync().unwrap();
1867
1868 let mut handles = vec![];
1869 for _ in 0..8 {
1870 let f = file.clone();
1871 let expected = data;
1872 handles.push(std::thread::spawn(move || {
1873 let mut buf = [0u8; BUFFER_SIZE * 2];
1874 f.read(&mut buf, 0).unwrap();
1875 assert_eq!(buf, expected);
1876 }));
1877 }
1878
1879 for h in handles {
1880 h.join().unwrap();
1881 }
1882 }
1883 }
1884
1885 mod file_grow {
1886 use super::*;
1887
1888 #[test]
1889 fn ok_grow_updates_length() {
1890 let (_dir, cfg) = tmp_path();
1891 let file = File::new(cfg).unwrap();
1892 assert_eq!(file.length(), BUFFER_SIZE * INIT_BUFFERS);
1893 assert_eq!(file.total_buffers().unwrap(), INIT_BUFFERS);
1894
1895 file.grow(0x20).unwrap();
1896 assert_eq!(file.length(), BUFFER_SIZE * (INIT_BUFFERS + 0x20));
1897 assert_eq!(file.total_buffers().unwrap(), INIT_BUFFERS + 0x20);
1898 }
1899
1900 #[test]
1901 fn ok_grow_zero_count() {
1902 let (_dir, cfg) = tmp_path();
1903 let file = File::new(cfg).unwrap();
1904 file.grow(0).unwrap();
1905 assert_eq!(file.length(), BUFFER_SIZE * INIT_BUFFERS);
1906 }
1907
1908 #[test]
1909 fn ok_grow_and_multi_buffer_write_read() {
1910 let (_dir, cfg) = tmp_path();
1911 let file = File::new(cfg).unwrap();
1912
1913 file.grow(4).unwrap();
1914 let new_index = INIT_BUFFERS;
1915
1916 let data = [0x55u8; BUFFER_SIZE * 3];
1917 file.write(&data, new_index).unwrap();
1918 file.sync().unwrap();
1919
1920 let mut buf = [0u8; BUFFER_SIZE * 3];
1921 file.read(&mut buf, new_index).unwrap();
1922 assert_eq!(buf, data);
1923 }
1924
1925 #[test]
1926 fn ok_concurrent_grow_and_write() {
1927 let (_dir, cfg) = tmp_path();
1928 let file = Arc::new(File::new(cfg).unwrap());
1929
1930 let writer = {
1931 let f = file.clone();
1932 std::thread::spawn(move || {
1933 for i in 0..INIT_BUFFERS {
1934 let data = [i as u8; BUFFER_SIZE];
1935 f.write(&data, i).unwrap();
1936 }
1937 })
1938 };
1939
1940 let chunks_to_grow = 0x20;
1941 let grower = {
1942 let f = file.clone();
1943 std::thread::spawn(move || {
1944 f.grow(chunks_to_grow).unwrap();
1945 })
1946 };
1947
1948 writer.join().unwrap();
1949 grower.join().unwrap();
1950
1951 file.sync().unwrap();
1952 assert_eq!(file.length(), BUFFER_SIZE * (INIT_BUFFERS + chunks_to_grow));
1953
1954 for i in 0..INIT_BUFFERS {
1955 let mut buf = [0u8; BUFFER_SIZE];
1956 file.read(&mut buf, i).unwrap();
1957 assert_eq!(buf, [i as u8; BUFFER_SIZE]);
1958 }
1959 }
1960
1961 #[test]
1962 #[cfg(target_os = "linux")]
1963 fn ok_sync_range() {
1964 let (_dir, cfg) = tmp_path();
1965 let file = File::new(cfg).unwrap();
1966
1967 let data = [0x33u8; BUFFER_SIZE * 2];
1968 file.write(&data, 0).unwrap();
1969 file.sync_range(0, 2).unwrap();
1970 file.sync().unwrap();
1971
1972 let mut buf = [0u8; BUFFER_SIZE * 2];
1973 file.read(&mut buf, 0).unwrap();
1974 assert_eq!(buf, data);
1975 }
1976
1977 #[test]
1978 #[cfg(target_os = "linux")]
1979 fn ok_sync_range_zero_count() {
1980 let (_dir, cfg) = tmp_path();
1981 let file = File::new(cfg).unwrap();
1982 assert!(file.sync_range(0, 0).is_ok());
1983 assert!(file.sync_range(5, 0).is_ok());
1984 }
1985
1986 #[test]
1987 #[cfg(target_os = "linux")]
1988 fn err_sync_range_index_overflow() {
1989 let (_dir, cfg) = tmp_path();
1990 let file = File::new(cfg).unwrap();
1991 let err1 = file.sync_range(usize::MAX, 1).unwrap_err();
1992 assert_eq!(err1.reason, err::INV.reason);
1993
1994 let err2 = file.sync_range(0, usize::MAX).unwrap_err();
1995 assert_eq!(err2.reason, err::INV.reason);
1996 }
1997
1998 #[test]
1999 fn ok_concurrent_multiple_growers() {
2000 let (_dir, cfg) = tmp_path();
2001 let file = Arc::new(File::new(cfg).unwrap());
2002 let num_threads = 4;
2003 let chunks_per_thread = 4;
2004 let barrier = Arc::new(Barrier::new(num_threads));
2005 let mut handles = vec![];
2006
2007 for _ in 0..num_threads {
2008 let f = file.clone();
2009 let b = barrier.clone();
2010 handles.push(std::thread::spawn(move || {
2011 b.wait();
2012 f.grow(chunks_per_thread).unwrap();
2013 }));
2014 }
2015
2016 for h in handles {
2017 h.join().unwrap();
2018 }
2019
2020 let expected_total_chunks = INIT_BUFFERS + (num_threads * chunks_per_thread);
2021 assert_eq!(file.length(), BUFFER_SIZE * expected_total_chunks);
2022 assert_eq!(file.total_buffers().unwrap(), expected_total_chunks);
2023 assert_eq!(file.current_epoch(), num_threads as TEpoch);
2024 }
2025
2026 #[test]
2027 fn ok_concurrent_readers_during_grow() {
2028 let (_dir, cfg) = tmp_path();
2029 let file = Arc::new(File::new(cfg).unwrap());
2030 let initial_data = [0x5Au8; BUFFER_SIZE];
2031 file.write(&initial_data, 0).unwrap();
2032 file.sync().unwrap();
2033
2034 let running = Arc::new(atomic::AtomicBool::new(true));
2035 let num_readers = 4;
2036 let mut readers = vec![];
2037
2038 for _ in 0..num_readers {
2039 let f = file.clone();
2040 let r = running.clone();
2041 let expected = initial_data;
2042 readers.push(std::thread::spawn(move || {
2043 let mut buf = [0u8; BUFFER_SIZE];
2044 while r.load(atomic::Ordering::Relaxed) {
2045 f.read(&mut buf, 0).unwrap();
2046 assert_eq!(buf, expected);
2047 std::thread::yield_now();
2048 }
2049 }));
2050 }
2051
2052 for _ in 0..5 {
2054 file.grow(2).unwrap();
2055 std::thread::sleep(std::time::Duration::from_millis(5));
2056 }
2057
2058 running.store(false, atomic::Ordering::Relaxed);
2059 for r in readers {
2060 r.join().unwrap();
2061 }
2062
2063 assert_eq!(file.length(), BUFFER_SIZE * (INIT_BUFFERS + 0x0A));
2064 assert_eq!(file.total_buffers().unwrap(), INIT_BUFFERS + 0x0A);
2065 }
2066
2067 #[test]
2068 fn err_total_buffers_unaligned_file() {
2069 let (_dir, cfg) = tmp_path();
2070 let file = File::new(cfg).unwrap();
2071 assert_eq!(file.total_buffers().unwrap(), INIT_BUFFERS);
2072
2073 file.inner.current_length.fetch_add(1, atomic::Ordering::SeqCst);
2075
2076 let err = file.total_buffers().unwrap_err();
2077 assert_eq!(err.reason, err::CPT.reason);
2078 }
2079 }
2080
2081 mod file_delete_exists {
2082 use super::*;
2083
2084 #[test]
2085 fn ok_exists_true_on_created_file() {
2086 let (_dir, cfg) = tmp_path();
2087 let file = File::new(cfg.clone()).unwrap();
2088
2089 assert!(file.exists().unwrap());
2090 assert!(cfg.path.exists());
2091 }
2092
2093 #[test]
2094 fn ok_delete_removes_file() {
2095 let (_dir, cfg) = tmp_path();
2096 let file = File::new(cfg.clone()).unwrap();
2097 let path = cfg.path.clone();
2098
2099 assert!(file.exists().unwrap());
2100 file.delete().unwrap();
2101
2102 assert!(!path.exists());
2103 assert!(File::new(cfg).is_ok());
2104 }
2105
2106 #[test]
2107 fn ok_open_and_delete() {
2108 let (_dir, cfg) = tmp_path();
2109 let file = File::new(cfg.clone()).unwrap();
2110 drop(file);
2111
2112 let opened = File::open(cfg.clone()).unwrap();
2113 assert!(opened.exists().unwrap());
2114 opened.delete().unwrap();
2115
2116 assert!(!cfg.path.exists());
2117 }
2118
2119 #[test]
2120 fn err_open_after_delete_grown_file() {
2121 let (_dir, cfg) = tmp_path();
2122 let file = File::new(cfg.clone()).unwrap();
2123
2124 file.grow(2).unwrap();
2125 assert_eq!(file.length(), (INIT_BUFFERS + 2) * BUFFER_SIZE);
2126
2127 file.delete().unwrap();
2128 assert!(!cfg.path.exists());
2129
2130 let err = File::open(cfg).unwrap_err();
2131 assert_eq!(err.reason, err::INV.reason);
2132 }
2133
2134 #[test]
2135 #[cfg(unix)]
2136 fn ok_exists_false_after_external_removal() {
2137 let (_dir, cfg) = tmp_path();
2138 let file = File::new(cfg.clone()).unwrap();
2139 assert!(file.exists().unwrap());
2140
2141 std::fs::remove_file(&cfg.path).unwrap();
2142 assert!(!file.exists().unwrap());
2143 }
2144
2145 #[test]
2146 #[cfg(windows)]
2147 fn ok_exists_false_after_external_removal() {
2148 let (_dir, cfg) = tmp_path();
2149 let file = File::new(cfg.clone()).unwrap();
2150 assert!(file.exists().unwrap());
2151 drop(file);
2152
2153 std::fs::remove_file(&cfg.path).unwrap();
2154 assert!(!PlatformFile::exists(&cfg.path).unwrap());
2155 }
2156
2157 #[test]
2158 #[cfg(unix)]
2159 fn err_delete_when_unlinked_externally() {
2160 let (_dir, cfg) = tmp_path();
2161 let file = File::new(cfg.clone()).unwrap();
2162
2163 std::fs::remove_file(&cfg.path).unwrap();
2164 let err = file.delete().unwrap_err();
2165 assert_eq!(err.reason, err::INV.reason);
2166 }
2167
2168 #[test]
2169 #[cfg(windows)]
2170 fn err_delete_when_unlinked_externally() {
2171 let (_dir, cfg) = tmp_path();
2172 let file = File::new(cfg.clone()).unwrap();
2173
2174 std::fs::remove_file(&cfg.path).unwrap();
2180 let err = file.delete().unwrap_err();
2181 assert!(
2182 err.reason == err::INV.reason || err.reason == err::PRM.reason,
2183 "Unexpected error reason: {:?}",
2184 err
2185 );
2186 }
2187
2188 #[test]
2189 #[cfg(unix)]
2190 fn err_delete_permission_denied() {
2191 use std::os::unix::fs::PermissionsExt;
2192
2193 let dir = tempfile::tempdir().unwrap();
2194 let sub_dir = dir.path().join("readonly_dir");
2195 std::fs::create_dir(&sub_dir).unwrap();
2196
2197 let path = sub_dir.join("victim.db");
2198 let cfg = FileCfg {
2199 module_id: MID,
2200 path,
2201 buffer_size: BUFFER_SIZE,
2202 initial_available_buffers: INIT_BUFFERS,
2203 sync_interval: None,
2204 };
2205
2206 let file = File::new(cfg).unwrap();
2207 std::fs::set_permissions(&sub_dir, std::fs::Permissions::from_mode(0o555)).unwrap();
2208
2209 let err = file.delete().unwrap_err();
2210 std::fs::set_permissions(&sub_dir, std::fs::Permissions::from_mode(0o755)).unwrap();
2211
2212 assert_eq!(err.reason, err::PRM.reason);
2213 }
2214
2215 #[test]
2216 #[cfg(windows)]
2217 fn err_delete_permission_denied() {
2218 let (_dir, cfg) = tmp_path();
2219 let file = File::new(cfg.clone()).unwrap();
2220
2221 let orig_perms = std::fs::metadata(&cfg.path).unwrap().permissions();
2222 let mut read_only = orig_perms.clone();
2223 read_only.set_readonly(true);
2224 std::fs::set_permissions(&cfg.path, read_only).unwrap();
2225
2226 let err = file.delete().unwrap_err();
2227
2228 let _ = std::fs::set_permissions(&cfg.path, orig_perms);
2229
2230 assert_eq!(err.reason, err::PRM.reason);
2231 }
2232
2233 #[test]
2234 fn ok_delete_and_recreate_cycle() {
2235 let (_dir, cfg) = tmp_path();
2236
2237 for i in 0..3 {
2238 let file = File::new(cfg.clone()).unwrap();
2239 let data = [i as u8; BUFFER_SIZE];
2240 file.write(&data, 0).unwrap();
2241 assert!(file.exists().unwrap());
2242
2243 file.delete().unwrap();
2244 assert!(!cfg.path.exists());
2245 }
2246 }
2247 }
2248
2249 mod file_drop {
2250 use super::*;
2251
2252 #[test]
2253 fn ok_drop_persists_written_data() {
2254 let (_dir, cfg) = tmp_path();
2255 let data = [0x7Au8; BUFFER_SIZE * 2];
2256
2257 {
2258 let file = File::new(cfg.clone()).unwrap();
2259 file.write(&data, 0).unwrap();
2260 drop(file);
2261 }
2262
2263 {
2264 let opened = File::open(cfg).unwrap();
2265 let mut buf = [0u8; BUFFER_SIZE * 2];
2266 opened.read(&mut buf, 0).unwrap();
2267 assert_eq!(buf, data);
2268 }
2269 }
2270
2271 #[test]
2272 fn ok_drop_releases_exclusive_lock() {
2273 let (_dir, cfg) = tmp_path();
2274
2275 let file = File::new(cfg.clone()).unwrap();
2276 let err = File::open(cfg.clone()).unwrap_err();
2277 assert_eq!(err.reason, err::LCK.reason);
2278
2279 drop(file);
2280
2281 let opened = File::open(cfg);
2282 assert!(opened.is_ok());
2283 }
2284 }
2285
2286 mod file_sync_and_durability {
2287 use super::*;
2288 use std::time::{Duration, Instant};
2289
2290 #[test]
2291 fn ok_manual_sync_advances_epoch_and_tickets() {
2292 let (_dir, cfg) = tmp_path();
2293 let file = File::new(cfg).unwrap();
2294 let data = [0x55u8; BUFFER_SIZE];
2295
2296 assert_eq!(file.current_epoch(), 0);
2297 assert_eq!(file.durable_epoch(), 0);
2298
2299 let t1 = file.write(&data, 0).unwrap();
2300 assert_eq!(t1.epoch(), 1);
2301 assert!(!t1.is_durable());
2302 assert_eq!(file.current_epoch(), 1);
2303 assert_eq!(file.durable_epoch(), 0);
2304
2305 let t2 = file.write(&data, 1).unwrap();
2306 assert_eq!(t2.epoch(), 2);
2307 assert!(!t2.is_durable());
2308 assert_eq!(file.current_epoch(), 2);
2309 assert_eq!(file.durable_epoch(), 0);
2310
2311 file.sync().unwrap();
2312 assert!(t1.is_durable());
2313 assert!(t2.is_durable());
2314 assert_eq!(file.durable_epoch(), 2);
2315 }
2316
2317 #[test]
2318 fn ok_ticket_force_without_background_thread() {
2319 let (_dir, cfg) = tmp_path();
2320 let file = File::new(cfg).unwrap();
2321 let data = [0xAAu8; BUFFER_SIZE];
2322
2323 let ticket = file.write(&data, 0).unwrap();
2324 assert!(!ticket.is_durable());
2325
2326 let durable = ticket.force().unwrap();
2327 assert_eq!(durable, ticket.epoch());
2328 assert!(ticket.is_durable());
2329 assert_eq!(file.durable_epoch(), ticket.epoch());
2330 }
2331
2332 #[test]
2333 fn ok_background_sync_advances_epoch() {
2334 let (_dir, mut cfg) = tmp_path();
2335 cfg.sync_interval = Some(Duration::from_millis(0x19));
2336
2337 let file = File::new(cfg).unwrap();
2338 let data = [0x33u8; BUFFER_SIZE];
2339
2340 let ticket = file.write(&data, 0).unwrap();
2341 let epoch = ticket.wait().unwrap();
2342 assert_eq!(epoch, ticket.epoch());
2343 assert!(ticket.is_durable());
2344 assert!(file.durable_epoch() >= ticket.epoch());
2345 }
2346
2347 #[test]
2348 fn ok_ticket_force_with_background_thread() {
2349 let (_dir, mut cfg) = tmp_path();
2350 cfg.sync_interval = Some(Duration::from_secs(0x0A));
2352
2353 let file = File::new(cfg).unwrap();
2354 let data = [0x44u8; BUFFER_SIZE];
2355
2356 let ticket = file.write(&data, 0).unwrap();
2357 assert!(!ticket.is_durable());
2358
2359 let start = Instant::now();
2360 let epoch = ticket.force().unwrap();
2361 let elapsed = start.elapsed();
2362
2363 assert_eq!(epoch, ticket.epoch());
2364 assert!(ticket.is_durable());
2365 assert!(elapsed < Duration::from_secs(2));
2366 }
2367
2368 #[test]
2369 fn ok_drop_with_background_thread_persists_data() {
2370 let (_dir, mut cfg) = tmp_path();
2371 cfg.sync_interval = Some(Duration::from_secs(0x1E));
2372 let data = [0x88u8; BUFFER_SIZE];
2373
2374 {
2375 let file = File::new(cfg.clone()).unwrap();
2376 let _ticket = file.write(&data, 0).unwrap();
2377 drop(file);
2379 }
2380
2381 {
2382 let opened = File::open(cfg).unwrap();
2383 let mut buf = [0u8; BUFFER_SIZE];
2384 opened.read(&mut buf, 0).unwrap();
2385 assert_eq!(buf, data);
2386 }
2387 }
2388
2389 #[test]
2390 fn ok_delete_with_background_thread() {
2391 let (_dir, mut cfg) = tmp_path();
2392 cfg.sync_interval = Some(Duration::from_millis(0x14));
2393
2394 let file = File::new(cfg.clone()).unwrap();
2395 let data = [0x99u8; BUFFER_SIZE];
2396 let _ = file.write(&data, 0).unwrap();
2397
2398 assert!(file.exists().unwrap());
2399 file.delete().unwrap();
2400 assert!(!cfg.path.exists());
2401 }
2402
2403 #[test]
2404 fn ok_concurrent_writes_distinct_epochs() {
2405 let (_dir, cfg) = tmp_path();
2406 let file = Arc::new(File::new(cfg).unwrap());
2407 let num_threads = 4;
2408 let mut handles = Vec::new();
2409
2410 for i in 0..num_threads {
2411 let f = file.clone();
2412 handles.push(std::thread::spawn(move || {
2413 let data = [i as u8; BUFFER_SIZE];
2414 f.write(&data, i).unwrap()
2415 }));
2416 }
2417
2418 let mut epochs = Vec::new();
2419 for h in handles {
2420 let ticket = h.join().unwrap();
2421 epochs.push(ticket.epoch());
2422 }
2423
2424 epochs.sort();
2425 epochs.dedup();
2426 assert_eq!(epochs.len(), num_threads);
2427 }
2428
2429 #[test]
2430 fn err_ticket_wait_when_file_deleted() {
2431 let (_dir, cfg) = tmp_path();
2432 let file = File::new(cfg).unwrap();
2433 let data = [0x11u8; BUFFER_SIZE];
2434 let ticket = file.write(&data, 0).unwrap();
2435
2436 file.delete().unwrap();
2437
2438 let res = ticket.force();
2440 assert!(res.is_err());
2441 }
2442
2443 #[test]
2444 fn err_ticket_wait_surfaces_sync_failure() {
2445 let (_dir, cfg) = tmp_path();
2446 let file = File::new(cfg).unwrap();
2447 let data = [0x22u8; BUFFER_SIZE];
2448 let ticket = file.write(&data, 0).unwrap();
2449
2450 let sync_err = err::make_error(err::SYN);
2452 file.completion().set_err(sync_err.clone());
2453 file.completion().notify_all_listeners();
2454
2455 let err_wait = ticket.wait().unwrap_err();
2456 assert_eq!(err_wait.reason, err::SYN.reason);
2457
2458 let err_force = ticket.force().unwrap_err();
2459 assert_eq!(err_force.reason, err::SYN.reason);
2460 }
2461
2462 #[test]
2463 fn ok_concurrent_ticket_force_storm() {
2464 let (_dir, mut cfg) = tmp_path();
2465 cfg.initial_available_buffers = 0x20;
2466 cfg.sync_interval = Some(Duration::from_millis(0x0A));
2467 let file = Arc::new(File::new(cfg).unwrap());
2468 let num_threads = 0x10;
2469 let barrier = Arc::new(Barrier::new(num_threads));
2470 let mut handles = Vec::new();
2471
2472 for i in 0..num_threads {
2473 let f = file.clone();
2474 let b = barrier.clone();
2475 handles.push(std::thread::spawn(move || {
2476 b.wait();
2477 let data = [i as u8; BUFFER_SIZE];
2478 let ticket = f.write(&data, i).unwrap();
2479 let durable = ticket.force().unwrap();
2480 assert_eq!(durable, ticket.epoch());
2481 assert!(ticket.is_durable());
2482 }));
2483 }
2484
2485 for h in handles {
2486 h.join().unwrap();
2487 }
2488
2489 assert!(file.durable_epoch() >= num_threads as TEpoch);
2490 }
2491 }
2492
2493 mod module_binding {
2494 use super::*;
2495
2496 #[test]
2497 fn ok_multiple_files_distinct_module_ids() {
2498 let dir = tempfile::tempdir().unwrap();
2499 let cfg1 = FileCfg {
2500 module_id: 0x11,
2501 path: dir.path().join("file1.db"),
2502 buffer_size: BUFFER_SIZE,
2503 initial_available_buffers: INIT_BUFFERS,
2504 sync_interval: None,
2505 };
2506 let cfg2 = FileCfg {
2507 module_id: 0x22,
2508 path: dir.path().join("file2.db"),
2509 buffer_size: BUFFER_SIZE,
2510 initial_available_buffers: INIT_BUFFERS,
2511 sync_interval: None,
2512 };
2513
2514 let file1 = File::new(cfg1.clone()).unwrap();
2515 let file2 = File::new(cfg2.clone()).unwrap();
2516
2517 let mut invalid_buf = [0u8; BUFFER_SIZE - 1];
2519 let err1 = file1.read(&mut invalid_buf, 0).unwrap_err();
2520 assert_eq!(err1.module, 0x11);
2521 assert_eq!(err1.reason, err::INV.reason);
2522
2523 let err1_oob = file1.read(&mut [0u8; BUFFER_SIZE], 0x3E7).unwrap_err();
2524 assert_eq!(err1_oob.module, 0x11);
2525 assert_eq!(err1_oob.reason, err::HCF.reason);
2526
2527 let err2 = file2.read(&mut invalid_buf, 0).unwrap_err();
2529 assert_eq!(err2.module, 0x22);
2530 assert_eq!(err2.reason, err::INV.reason);
2531
2532 let err2_oob = file2.write(&[0u8; BUFFER_SIZE], 0x3E7).unwrap_err();
2533 assert_eq!(err2_oob.module, 0x22);
2534 assert_eq!(err2_oob.reason, err::HCF.reason);
2535
2536 let err1_exs = File::new(cfg1).unwrap_err();
2538 assert_eq!(err1_exs.module, 0x11);
2539 assert_eq!(err1_exs.reason, err::EXS.reason);
2540
2541 let err2_exs = File::new(cfg2).unwrap_err();
2542 assert_eq!(err2_exs.module, 0x22);
2543 assert_eq!(err2_exs.reason, err::EXS.reason);
2544 }
2545 }
2546}