1use std::collections::{HashMap, HashSet};
4use std::fmt;
5use std::io::{self, Read, Seek, SeekFrom as IoSeekFrom, Write};
6use std::path::{Component, Path, PathBuf};
7use std::sync::atomic::{AtomicBool, Ordering};
8use std::sync::{Arc, Mutex};
9use std::time::SystemTime;
10
11use vfsi_core::api::internal::OwnedReadResult as FsReadResult;
12use vfsi_core::api::{
13 DirectoryListing, ReadIntoResult as FsReadIntoResult, ResourceLimits, StreamCompletion,
14 TraversalCompletion, WriteResult as FsWriteResult,
15};
16
17use crate::traits::{validate_read_into_results, validate_read_results, validate_write_results};
18use crate::{
19 AttrMask, Capabilities, CopyFileSystem, DirEntry, DirectoryFileSystem, FileSystem,
20 LinkFileSystem, Metadata, MetadataFileSystem, MetadataQuery, MetadataUpdate,
21 NamespaceFileSystem, OpenFlags, OpenRequest, Permissions, ReadAllOptions, ReadDirOptions,
22 ReadOp, ReadResult, ReadStreamOptions, RemoveOptions, SetAttributes, VecFs, VectorFileSystem,
23 VfDir, VfError, VfFile, VfOffset, VfResult, WriteOpRef, WriteResult,
24};
25
26fn read_result(result: ReadResult) -> FsReadResult {
27 FsReadResult {
28 offset: result.offset,
29 data: result.data,
30 eof: result.eof,
31 }
32}
33
34fn write_result(result: WriteResult) -> FsWriteResult {
35 FsWriteResult {
36 offset: result.offset,
37 written: result.written,
38 stable: result.stable,
39 }
40}
41
42fn io_error(error: VfError) -> io::Error {
43 error.into()
44}
45
46fn poisoned() -> VfError {
47 VfError::client(0, crate::ERR_IO)
48}
49
50fn wrong_result_count(operation: &str, expected: usize, actual: usize) -> VfError {
51 VfError::transport_with_kind(
52 None,
53 crate::TransportKind::InvalidReply,
54 format!("{operation} backend returned {actual} results for {expected} requests"),
55 )
56}
57
58enum CleanupTarget {
59 File(VfFile, PathBuf),
60 Directory(VfDir, PathBuf),
61}
62struct PendingClose<F> {
63 target: CleanupTarget,
64 close: fn(&mut F, &CleanupTarget) -> VfResult<()>,
65}
66
67struct SharedBackend<F: FileSystem> {
70 backend: Mutex<Option<F>>,
71 pending: Mutex<Vec<PendingClose<F>>>,
72 has_pending: AtomicBool,
73}
74impl<F: FileSystem> SharedBackend<F> {
75 fn new(backend: F) -> Self {
76 Self {
77 backend: Mutex::new(Some(backend)),
78 pending: Mutex::new(Vec::new()),
79 has_pending: AtomicBool::new(false),
80 }
81 }
82 fn defer(&self, target: CleanupTarget, close: fn(&mut F, &CleanupTarget) -> VfResult<()>) {
83 self.pending
84 .lock()
85 .unwrap_or_else(|e| e.into_inner())
86 .push(PendingClose { target, close });
87 self.has_pending.store(true, Ordering::Release);
88 }
89 fn lock_without_cleanup(&self) -> Result<BackendGuard<'_, F>, ()> {
90 self.backend
91 .lock()
92 .map(|guard| BackendGuard {
93 guard: Some(guard),
94 pending: &self.pending,
95 has_pending: &self.has_pending,
96 })
97 .map_err(|_| ())
98 }
99 fn lock(&self) -> Result<BackendGuard<'_, F>, ()> {
100 let mut guard = self.lock_without_cleanup()?;
101 let _ = guard.drain_cleanup();
104 Ok(guard)
105 }
106}
107impl<F: FileSystem> Drop for SharedBackend<F> {
108 fn drop(&mut self) {
109 if let Some(backend) = self
110 .backend
111 .get_mut()
112 .unwrap_or_else(|e| e.into_inner())
113 .as_mut()
114 {
115 let pending = self.pending.get_mut().unwrap_or_else(|e| e.into_inner());
116 for close in pending.drain(..) {
117 let _ = (close.close)(backend, &close.target);
118 }
119 let callbacks = backend.take_notifications();
120 for callback in callbacks {
121 let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(callback));
122 }
123 }
124 }
125}
126struct BackendGuard<'a, F: FileSystem> {
127 guard: Option<std::sync::MutexGuard<'a, Option<F>>>,
128 pending: &'a Mutex<Vec<PendingClose<F>>>,
129 has_pending: &'a AtomicBool,
130}
131impl<F: FileSystem> std::ops::Deref for BackendGuard<'_, F> {
132 type Target = F;
133 fn deref(&self) -> &F {
134 self.guard
135 .as_ref()
136 .expect("live guard")
137 .as_ref()
138 .expect("live backend")
139 }
140}
141impl<F: FileSystem> std::ops::DerefMut for BackendGuard<'_, F> {
142 fn deref_mut(&mut self) -> &mut F {
143 self.guard
144 .as_mut()
145 .expect("live guard")
146 .as_mut()
147 .expect("live backend")
148 }
149}
150impl<F: FileSystem> BackendGuard<'_, F> {
151 fn drain_cleanup(&mut self) -> VfResult<()> {
152 if !self.has_pending.swap(false, Ordering::AcqRel) {
153 return Ok(());
154 }
155 let pending = std::mem::take(&mut *self.pending.lock().unwrap_or_else(|e| e.into_inner()));
156 let mut failed = Vec::new();
157 let mut first = None;
158 for close in pending {
159 if let Err(error) = (close.close)(self, &close.target) {
160 first.get_or_insert(error);
161 failed.push(close);
162 }
163 }
164 if !failed.is_empty() {
165 self.pending
166 .lock()
167 .unwrap_or_else(|e| e.into_inner())
168 .extend(failed);
169 self.has_pending.store(true, Ordering::Release);
170 }
171 first.map_or(Ok(()), Err)
172 }
173}
174impl<F: FileSystem> Drop for BackendGuard<'_, F> {
175 fn drop(&mut self) {
176 let callbacks = self.take_notifications();
177 drop(self.guard.take());
178 for callback in callbacks {
179 let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(callback));
182 }
183 }
184}
185
186pub struct FsClient<F: FileSystem> {
193 inner: Arc<SharedBackend<F>>,
194 limits: ResourceLimits,
195}
196
197impl<F: FileSystem> Clone for FsClient<F> {
198 fn clone(&self) -> Self {
199 Self {
200 inner: Arc::clone(&self.inner),
201 limits: self.limits,
202 }
203 }
204}
205
206impl<F: FileSystem> fmt::Debug for FsClient<F> {
207 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
208 formatter.debug_struct("FsClient").finish_non_exhaustive()
209 }
210}
211
212impl<F: FileSystem> FsClient<F> {
213 pub fn new(filesystem: F) -> Self {
214 Self {
215 inner: Arc::new(SharedBackend::new(filesystem)),
216 limits: ResourceLimits::default(),
217 }
218 }
219
220 pub fn with_limits(mut self, limits: ResourceLimits) -> Self {
223 self.limits = limits;
224 self
225 }
226
227 pub fn limits(&self) -> ResourceLimits {
228 self.limits
229 }
230
231 fn lock(&self) -> VfResult<BackendGuard<'_, F>> {
232 self.inner.lock().map_err(|_| poisoned())
233 }
234
235 pub fn into_inner(self) -> Result<F, Self> {
236 if self.drain_cleanup().is_err() {
237 return Err(self);
238 }
239 match Arc::try_unwrap(self.inner) {
240 Ok(mut shared) => Ok(shared
241 .backend
242 .get_mut()
243 .unwrap_or_else(|e| e.into_inner())
244 .take()
245 .expect("owned backend")),
246 Err(inner) => Err(Self {
247 inner,
248 limits: self.limits,
249 }),
250 }
251 }
252
253 pub fn drain_cleanup(&self) -> VfResult<()> {
256 self.inner
257 .lock_without_cleanup()
258 .map_err(|_| poisoned())?
259 .drain_cleanup()
260 }
261}
262
263impl<F: FileSystem> FsClient<F> {
264 pub fn capabilities(&self) -> VfResult<Capabilities> {
265 Ok(self.lock()?.capabilities())
266 }
267 pub fn open(&self, path: impl AsRef<Path>) -> VfResult<FsFile<F>> {
269 self.open_with(OpenRequest::new(path.as_ref(), OpenFlags::READ))
270 }
271
272 pub fn open_with(&self, request: OpenRequest) -> VfResult<FsFile<F>> {
273 let file = self.lock()?.open_one(&request)?;
274 Ok(FsFile {
275 inner: Arc::clone(&self.inner),
276 file: Some(file),
277 path: request.path,
278 })
279 }
280
281 pub fn open_options(&self) -> OpenOptions<'_, F> {
282 OpenOptions::new(self)
283 }
284
285 pub fn create(&self, path: impl AsRef<Path>) -> VfResult<FsFile<F>> {
286 self.open_with(OpenRequest::new(
287 path.as_ref(),
288 OpenFlags::WRITE | OpenFlags::CREATE | OpenFlags::TRUNCATE,
289 ))
290 }
291
292 pub fn read(&self, path: impl AsRef<Path>) -> VfResult<Vec<u8>> {
293 self.read_with_limit(path, self.limits.max_read_bytes)
294 }
295
296 pub fn read_with_limit(&self, path: impl AsRef<Path>, max_bytes: usize) -> VfResult<Vec<u8>> {
301 let path = path.as_ref();
302 let file = self.open(path)?;
303 let operation = self
304 .lock()?
305 .read_file(file.raw()?, max_bytes)
306 .map_err(|error| error.with_context("read", path))
307 .and_then(|data| {
308 if data.len() > max_bytes {
309 Err(VfError::client(0, libc::EFBIG as u32).with_context("read", path))
310 } else {
311 Ok(data)
312 }
313 });
314 let cleanup = file.close();
317 operation.and_then(|data| cleanup.map(|()| data))
318 }
319
320 pub fn read_to_string(&self, path: impl AsRef<Path>) -> VfResult<String> {
321 self.read_to_string_with_limit(path, self.limits.max_read_bytes)
322 }
323
324 pub fn read_to_string_with_limit(
326 &self,
327 path: impl AsRef<Path>,
328 max_bytes: usize,
329 ) -> VfResult<String> {
330 let path = path.as_ref();
331 String::from_utf8(self.read_with_limit(path, max_bytes)?)
332 .map_err(|_| VfError::client(0, crate::ERR_INVAL).with_context("read_to_string", path))
333 }
334
335 pub fn write(&self, path: impl AsRef<Path>, data: &[u8]) -> VfResult<()> {
336 let path = path.as_ref();
337 let mut file = self.create(path)?;
338 let mut written = 0;
339 while written < data.len() {
340 let count = file.write_native(&data[written..])?;
341 if count == 0 {
342 return Err(VfError::client(0, crate::ERR_IO).with_context("write", path));
343 }
344 written += count;
345 }
346 file.close()
347 }
348}
349
350impl<F: FileSystem> FsClient<F> {
351 pub fn read_stream(
359 &self,
360 path: impl AsRef<Path>,
361 callback: impl FnMut(u64, &[u8]) -> VfResult<bool>,
362 ) -> VfResult<StreamCompletion> {
363 self.read_stream_with_options(
364 path,
365 ReadStreamOptions::new().chunk_size(self.limits.stream_chunk_bytes),
366 callback,
367 )
368 }
369
370 pub fn read_stream_with_options(
372 &self,
373 path: impl AsRef<Path>,
374 options: ReadStreamOptions,
375 mut callback: impl FnMut(u64, &[u8]) -> VfResult<bool>,
376 ) -> VfResult<StreamCompletion> {
377 let chunk_size = options.chunk_size_bytes();
378 if chunk_size == 0 {
379 return Err(VfError::failure(0, crate::ERR_INVAL));
380 }
381
382 let file = self.open(path)?;
383 let raw_file = file.raw()?.clone();
384 let operation = (|| -> VfResult<StreamCompletion> {
385 let mut offset = 0u64;
386 loop {
387 let request = ReadOp::at(raw_file.clone(), offset, chunk_size);
388 let result = {
389 let mut backend = self.lock()?;
390 let result = backend.read_one(&request)?;
391 validate_read_results(
392 "read_stream",
393 std::slice::from_ref(&request),
394 std::slice::from_ref(&result),
395 )?;
396 result
397 };
398 let length = result.data.len();
399 if !result.data.is_empty() && !callback(offset, &result.data)? {
400 return Ok(StreamCompletion::Stopped {
401 next_offset: offset
402 .checked_add(length as u64)
403 .ok_or_else(|| VfError::client(0, crate::ERR_INVAL))?,
404 });
405 }
406 if result.eof {
407 return Ok(StreamCompletion::Complete);
408 }
409 offset = offset
410 .checked_add(length as u64)
411 .ok_or_else(|| VfError::client(0, libc::EOVERFLOW as u32))?;
412 }
413 })();
414 let close = file.close();
415 match operation {
416 Err(error) => Err(error),
417 Ok(completion) => close.map(|()| completion),
418 }
419 }
420}
421
422impl<F: MetadataFileSystem> FsClient<F> {
423 pub fn metadata(&self, path: impl AsRef<Path>) -> VfResult<Metadata> {
424 self.lock()?.metadata_path(path.as_ref(), true)
425 }
426
427 pub fn symlink_metadata(&self, path: impl AsRef<Path>) -> VfResult<Metadata> {
428 self.lock()?.metadata_path(path.as_ref(), false)
429 }
430
431 pub fn set_metadata(&self, path: impl AsRef<Path>) -> SetMetadata<'_, F> {
432 SetMetadata {
433 client: self,
434 path: path.as_ref().to_path_buf(),
435 update: MetadataUpdate::new(),
436 follow: true,
437 }
438 }
439}
440
441impl<F: DirectoryFileSystem> FsClient<F> {
442 pub fn create_dir(&self, path: impl AsRef<Path>) -> VfResult<()> {
443 self.create_dir_with_mode(path, 0o777)
444 }
445
446 pub fn create_dir_with_mode(&self, path: impl AsRef<Path>, mode: u32) -> VfResult<()> {
447 self.lock()?.create_dir_one(path.as_ref(), mode)
448 }
449
450 pub fn read_dir(&self, path: impl AsRef<Path>) -> VfResult<Vec<DirEntry>> {
451 self.read_dir_with_options(path, self.limits.directory_options())
452 }
453
454 pub fn read_dir_with_options(
456 &self,
457 path: impl AsRef<Path>,
458 options: ReadDirOptions,
459 ) -> VfResult<Vec<DirEntry>> {
460 self.lock()?.read_dir_one(path.as_ref(), options)
461 }
462
463 pub fn visit_dir(
467 &self,
468 path: impl AsRef<Path>,
469 callback: impl FnMut(DirEntry) -> VfResult<std::ops::ControlFlow<()>>,
470 ) -> VfResult<TraversalCompletion> {
471 self.visit_dir_with_options(path, self.limits.directory_options(), callback)
472 }
473
474 pub fn visit_dir_with_options(
476 &self,
477 path: impl AsRef<Path>,
478 options: ReadDirOptions,
479 callback: impl FnMut(DirEntry) -> VfResult<std::ops::ControlFlow<()>>,
480 ) -> VfResult<TraversalCompletion> {
481 self.visit_dir_with_fields(path, crate::native::metadata_mask(), options, callback)
482 }
483
484 pub fn visit_dir_with_fields(
485 &self,
486 path: impl AsRef<Path>,
487 fields: AttrMask,
488 options: ReadDirOptions,
489 mut callback: impl FnMut(DirEntry) -> VfResult<std::ops::ControlFlow<()>>,
490 ) -> VfResult<TraversalCompletion> {
491 const PAGE_SIZE: usize = 1024;
492 let path = path.as_ref();
493 let mut cursor = None;
494 let mut count = 0usize;
495 let mut path_bytes = 0usize;
496 let max_entries = if options.entry_limit() == usize::MAX {
497 0 } else {
499 options.entry_limit().saturating_add(1)
500 };
501 loop {
502 let page_size = if cursor.is_none() {
505 1
506 } else {
507 PAGE_SIZE.min(
508 options
509 .entry_limit()
510 .saturating_sub(count)
511 .saturating_add(1),
512 )
513 };
514 let (entries, next) = {
515 self.lock()?.read_dir_page_with_fields(
516 path,
517 fields,
518 cursor,
519 page_size,
520 max_entries,
521 )?
522 };
523 if entries.is_empty() && next.is_some() {
524 return Err(VfError::transport(None, "directory page made no progress"));
525 }
526 for entry in entries {
527 if count >= options.entry_limit() {
528 return Err(
529 VfError::failure(count, libc::EFBIG as u32).with_context("visit_dir", path)
530 );
531 }
532 path_bytes = path_bytes
533 .checked_add(entry.path().as_os_str().len())
534 .ok_or_else(|| {
535 VfError::failure(count, libc::EFBIG as u32).with_context("visit_dir", path)
536 })?;
537 if path_bytes > options.path_byte_limit() {
538 return Err(
539 VfError::failure(count, libc::EFBIG as u32).with_context("visit_dir", path)
540 );
541 }
542 count += 1;
543 if callback(entry)?.is_break() {
544 return Ok(TraversalCompletion::Stopped);
545 }
546 }
547 match next {
548 Some(value) => cursor = Some(value),
549 None => return Ok(TraversalCompletion::Complete),
550 }
551 }
552 }
553}
554
555impl<F: DirectoryFileSystem + MetadataFileSystem> FsClient<F> {
556 pub fn create_dir_all(&self, path: impl AsRef<Path>) -> VfResult<()> {
557 let path = path.as_ref();
558 let mut current = PathBuf::new();
559 for component in path.components() {
560 match component {
561 Component::RootDir => {
562 current.push(Path::new("/"));
563 continue;
564 }
565 Component::Normal(part) => current.push(part),
566 Component::CurDir => continue,
567 Component::ParentDir => {
568 current.push("..");
573 continue;
574 }
575 Component::Prefix(_) => return Err(VfError::client(0, crate::ERR_INVAL)),
576 }
577 match self.create_dir(¤t) {
578 Ok(()) => {}
579 Err(error) if error.err_no() == crate::ERR_EXIST => {
580 if !self.metadata(¤t)?.is_dir() {
581 return Err(VfError::client(0, crate::ERR_NOTDIR)
582 .with_context("create_dir_all", ¤t));
583 }
584 }
585 Err(error) => return Err(error),
586 }
587 }
588 Ok(())
589 }
590}
591
592impl<F: NamespaceFileSystem + MetadataFileSystem> FsClient<F> {
593 fn removal_metadata(&self, path: &Path, operation: &'static str) -> VfResult<Metadata> {
594 let mut filesystem = self.lock()?;
595 let follow = !filesystem.capabilities().contains(Capabilities::LSTAT);
596 filesystem
597 .metadata_path(path, follow)
598 .map_err(|error| error.with_context(operation, path))
599 }
600
601 pub fn remove_file(&self, path: impl AsRef<Path>) -> VfResult<()> {
602 let path = path.as_ref();
603 if self.removal_metadata(path, "remove_file")?.is_dir() {
604 return Err(VfError::client(0, crate::ERR_ISDIR).with_context("remove_file", path));
605 }
606 self.lock()?.remove_one(path, false)
607 }
608
609 pub fn remove_dir(&self, path: impl AsRef<Path>) -> VfResult<()> {
610 let path = path.as_ref();
611 if !self.removal_metadata(path, "remove_dir")?.is_dir() {
612 return Err(VfError::client(0, crate::ERR_NOTDIR).with_context("remove_dir", path));
613 }
614 self.lock()?.remove_one(path, false)
615 }
616
617 pub fn remove_dir_all(&self, path: impl AsRef<Path>) -> VfResult<()> {
618 let path = path.as_ref();
619 if !self.removal_metadata(path, "remove_dir_all")?.is_dir() {
620 return Err(VfError::client(0, crate::ERR_NOTDIR).with_context("remove_dir_all", path));
621 }
622 self.lock()?.remove_one(path, true)
623 }
624
625 pub fn remove_dir_contents(&self, path: impl AsRef<Path>) -> VfResult<()> {
627 let path = path.as_ref();
628 if !self.removal_metadata(path, "remove_dir_contents")?.is_dir() {
629 return Err(
630 VfError::client(0, crate::ERR_NOTDIR).with_context("remove_dir_contents", path)
631 );
632 }
633 self.lock()?.remove_dir_contents(path)
634 }
635
636 pub fn rename(&self, from: impl AsRef<Path>, to: impl AsRef<Path>) -> VfResult<()> {
637 self.lock()?.rename_one(from.as_ref(), to.as_ref())
638 }
639}
640
641impl<F: VecFs> FsClient<F> {
642 pub fn renamev<P: AsRef<Path>, Q: AsRef<Path>>(&self, pairs: &[(P, Q)]) -> VfResult<()> {
644 if pairs.is_empty() {
645 return Ok(());
646 }
647 let requests: Vec<_> = pairs
648 .iter()
649 .map(|(from, to)| {
650 (
651 VfFile::from_os_path(from.as_ref()),
652 VfFile::from_os_path(to.as_ref()),
653 )
654 })
655 .collect();
656 self.lock()?
657 .renamev(&requests)
658 .map_err(|error| match error.index() {
659 Some(index) if index < pairs.len() => {
660 error.with_context("renamev", pairs[index].0.as_ref())
661 }
662 Some(_) => {
663 VfError::transport(None, "rename backend returned an invalid error index")
664 }
665 None => error,
666 })
667 }
668
669 pub fn mkdirv<P: AsRef<Path>>(&self, paths: &[P]) -> VfResult<()> {
672 let mut seen = HashSet::with_capacity(paths.len());
673 for (index, path) in paths.iter().enumerate() {
674 if !seen.insert(path.as_ref()) {
675 return Err(
676 VfError::client(index, crate::ERR_INVAL).with_context("mkdirv", path.as_ref())
677 );
678 }
679 }
680 if paths.is_empty() {
681 return Ok(());
682 }
683 let dirs: Vec<_> = paths
684 .iter()
685 .map(|path| crate::VfAttrs {
686 file: VfFile::from_os_path(path.as_ref()),
687 masks: AttrMask::MODE,
688 mode: 0o777,
689 ..Default::default()
690 })
691 .collect();
692 self.lock()?.mkdirv(&dirs).map_err(|error| {
693 match error.index().and_then(|index| paths.get(index)) {
694 Some(path) => error.with_context("mkdirv", path.as_ref()),
695 None => error,
696 }
697 })
698 }
699
700 pub fn walk_events_with_options(
705 &self,
706 root: impl AsRef<Path>,
707 fields: AttrMask,
708 options: crate::WalkOptions,
709 sort_by_name: bool,
710 callback: impl FnMut(&crate::WalkEvent) -> VfResult<crate::WalkControl>,
711 ) -> VfResult<TraversalCompletion> {
712 let root = root.as_ref();
713 let metadata = self.symlink_metadata_with_fields(root, fields | AttrMask::MODE)?;
714 crate::walk_events(
715 DirEntry::new(root.to_path_buf(), metadata),
716 options,
717 sort_by_name,
718 |path, limits| {
719 let mut listings =
720 self.read_dirs_with_options(&[path], fields | AttrMask::MODE, limits)?;
721 if listings.len() != 1 {
722 return Err(VfError::transport(None, "invalid directory result count"));
723 }
724 Ok(listings.remove(0).entries)
725 },
726 callback,
727 )
728 }
729 pub fn visit_walk(
738 &self,
739 root: impl AsRef<Path>,
740 callback: impl FnMut(&DirEntry) -> VfResult<std::ops::ControlFlow<()>>,
741 ) -> VfResult<TraversalCompletion> {
742 self.visit_walk_with_options(root, self.limits.walk_options(), callback)
743 }
744
745 pub fn visit_walk_with_options(
746 &self,
747 root: impl AsRef<Path>,
748 options: crate::WalkOptions,
749 callback: impl FnMut(&DirEntry) -> VfResult<std::ops::ControlFlow<()>>,
750 ) -> VfResult<TraversalCompletion> {
751 self.visit_walk_with_fields(root, crate::native::metadata_mask(), options, callback)
752 }
753
754 pub fn visit_walk_with_fields(
755 &self,
756 root: impl AsRef<Path>,
757 fields: AttrMask,
758 options: crate::WalkOptions,
759 callback: impl FnMut(&DirEntry) -> VfResult<std::ops::ControlFlow<()>>,
760 ) -> VfResult<TraversalCompletion> {
761 let root = root.as_ref();
762 if !self.symlink_metadata(root)?.is_dir() {
763 return Err(VfError::client(0, crate::ERR_NOTDIR).with_context("visit_walk", root));
764 }
765 visit_walk_pages(
766 root,
767 options,
768 |path, cursor, page_size, max_entries| {
769 self.lock()?
770 .read_dir_page_with_fields(path, fields, cursor, page_size, max_entries)
771 },
772 callback,
773 )
774 }
775 pub fn directory_page_batch_size(&self) -> VfResult<usize> {
777 Ok(self.lock()?.directory_page_batch_size().clamp(1, 32))
778 }
779 pub fn read_dir_pages_with_fields(
781 &self,
782 paths: &[&Path],
783 fields: AttrMask,
784 cursors: Vec<Option<crate::DirPageCursor>>,
785 page_size: usize,
786 max_entries: usize,
787 ) -> VfResult<Vec<crate::DirectoryPage>> {
788 let pages = self.lock()?.listdir_pages(
789 paths,
790 fields | AttrMask::MODE | AttrMask::SIZE,
791 cursors,
792 page_size,
793 max_entries,
794 )?;
795 if pages.len() != paths.len() {
796 return Err(VfError::transport(
797 None,
798 "directory pages returned an invalid result count",
799 ));
800 }
801 pages
802 .into_iter()
803 .zip(paths)
804 .enumerate()
805 .map(|(index, ((attrs, next, children), path))| {
806 if children.len() != attrs.len()
807 || attrs.len() > page_size
808 || (attrs.is_empty() && next.is_some())
809 {
810 return Err(VfError::transport(
811 Some(index),
812 "invalid directory page progress",
813 ));
814 }
815 let mut seeds = Vec::new();
816 let entries = attrs
817 .into_iter()
818 .zip(children)
819 .map(|(attrs, child)| {
820 let entry_path = attrs
821 .file
822 .path()
823 .ok_or_else(|| {
824 VfError::transport(Some(index), "directory entry has no path")
825 })?
826 .to_path_buf();
827 if entry_path.parent() != Some(*path) {
828 return Err(VfError::transport(
829 Some(index),
830 "directory entry is outside its parent",
831 ));
832 }
833 if let Some(child) = child {
834 seeds.push((entry_path.clone(), child));
835 }
836 Ok(DirEntry::new(
837 entry_path,
838 vfsi_core::metadata_from_attrs(attrs),
839 ))
840 })
841 .collect::<VfResult<Vec<_>>>()?;
842 Ok((
843 DirectoryListing {
844 path: path.to_path_buf(),
845 entries,
846 },
847 next,
848 seeds,
849 ))
850 })
851 .collect()
852 }
853
854 pub fn read_dirs<P: AsRef<Path>>(&self, paths: &[P]) -> VfResult<Vec<DirectoryListing>> {
857 self.read_dirs_with_options(paths, AttrMask::stat(), self.limits.directory_options())
858 }
859}
860
861type DirectoryPage = (Vec<DirEntry>, Option<crate::DirPageCursor>);
862
863fn visit_walk_pages(
864 root: &Path,
865 options: crate::WalkOptions,
866 mut read_page: impl FnMut(
867 &Path,
868 Option<crate::DirPageCursor>,
869 usize,
870 usize,
871 ) -> VfResult<DirectoryPage>,
872 mut callback: impl FnMut(&DirEntry) -> VfResult<std::ops::ControlFlow<()>>,
873) -> VfResult<TraversalCompletion> {
874 let mut pending = vec![(root.to_path_buf(), 0usize)];
875 let mut count = 0usize;
876 let mut path_bytes = root.as_os_str().len();
877 if path_bytes > options.path_byte_limit() {
878 return Err(VfError::client(0, libc::EFBIG as u32));
879 }
880 while let Some((path, depth)) = pending.pop() {
881 let mut cursor = None;
882 let mut children = Vec::new();
883 loop {
884 let requested = options
885 .entry_limit()
886 .saturating_sub(count)
887 .saturating_add(1);
888 let (entries, next) = read_page(&path, cursor, requested.min(128), requested)?;
889 if entries.is_empty() && next.is_some() {
890 return Err(VfError::transport(None, "directory page made no progress")
891 .with_context("visit_walk", &path));
892 }
893 for entry in entries {
894 count = count
895 .checked_add(1)
896 .ok_or_else(|| VfError::client(0, libc::EFBIG as u32))?;
897 path_bytes = path_bytes
898 .checked_add(entry.path().as_os_str().len())
899 .ok_or_else(|| VfError::client(0, libc::EFBIG as u32))?;
900 if count > options.entry_limit() || path_bytes > options.path_byte_limit() {
901 return Err(
902 VfError::client(0, libc::EFBIG as u32).with_context("visit_walk", &path)
903 );
904 }
905 if callback(&entry)?.is_break() {
906 return Ok(TraversalCompletion::Stopped);
907 }
908 if entry.metadata().is_dir() {
909 if depth >= options.depth_limit() {
910 if !options.truncates_at_depth_limit() {
911 return Err(VfError::client(0, libc::EFBIG as u32)
912 .with_context("visit_walk", entry.path()));
913 }
914 } else {
915 children.push((entry.path().to_path_buf(), depth + 1));
916 }
917 }
918 }
919 cursor = next;
920 if cursor.is_none() {
921 break;
922 }
923 }
924 pending.extend(children.into_iter().rev());
925 }
926 Ok(TraversalCompletion::Complete)
927}
928
929impl<F: VecFs> FsClient<F> {
930 pub fn walk(&self, root: impl AsRef<Path>) -> VfResult<Vec<DirectoryListing>> {
933 self.walk_with_options(root, AttrMask::stat(), self.limits.walk_options())
934 }
935
936 pub fn ensure_empty_dir(&self, path: impl AsRef<Path>) -> VfResult<()> {
939 self.lock()?.ensure_empty_dir(path.as_ref())
940 }
941
942 pub fn remove_dir_all_with_options(
944 &self,
945 path: impl AsRef<Path>,
946 options: RemoveOptions,
947 ) -> VfResult<()> {
948 let path = path.as_ref();
949 if !self.removal_metadata(path, "remove_dir_all")?.is_dir() {
950 return Err(VfError::client(0, crate::ERR_NOTDIR).with_context("remove_dir_all", path));
951 }
952 self.lock()?
953 .rm_with_options(&[path], true, options)
954 .map_err(|error| error.with_context("remove_dir_all", path))
955 }
956
957 pub fn remove_dir_contents_with_options(
959 &self,
960 path: impl AsRef<Path>,
961 options: RemoveOptions,
962 ) -> VfResult<()> {
963 let path = path.as_ref();
964 if !self.removal_metadata(path, "remove_dir_contents")?.is_dir() {
965 return Err(
966 VfError::client(0, crate::ERR_NOTDIR).with_context("remove_dir_contents", path)
967 );
968 }
969 self.lock()?
970 .rm_contents_with_options(path, options)
971 .map_err(|error| error.with_context("remove_dir_contents", path))
972 }
973
974 pub fn open_dir_handle(&self, path: impl AsRef<Path>) -> VfResult<FsDir<F>> {
978 let path = path.as_ref();
979 let mut backend = self.lock()?;
980 let dir = backend.open_dir(path)?;
981 if !matches!(dir, VfDir::Descriptor { .. }) {
982 let _ = backend.close_dir(&dir);
983 return Err(VfError::unsupported(0).with_context("open_dir_handle", path));
984 }
985 Ok(FsDir {
986 inner: Arc::clone(&self.inner),
987 dir: Some(dir),
988 path: path.to_path_buf(),
989 })
990 }
991}
992
993pub struct FsDir<F: VecFs> {
996 inner: Arc<SharedBackend<F>>,
997 dir: Option<VfDir>,
998 path: PathBuf,
999}
1000
1001impl<F: VecFs> fmt::Debug for FsDir<F> {
1002 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
1003 formatter
1004 .debug_struct("FsDir")
1005 .field("path", &self.path)
1006 .finish_non_exhaustive()
1007 }
1008}
1009
1010impl<F: VecFs> FsDir<F> {
1011 pub fn is_closed(&self) -> bool {
1012 self.dir.is_none()
1013 }
1014 pub fn path(&self) -> &Path {
1016 &self.path
1017 }
1018
1019 pub fn remove_contents(&self) -> VfResult<()> {
1020 self.remove_contents_with_options(RemoveOptions::default())
1021 }
1022
1023 pub fn remove_contents_with_options(&self, options: RemoveOptions) -> VfResult<()> {
1024 let dir = self
1025 .dir
1026 .as_ref()
1027 .ok_or_else(|| VfError::client(0, crate::ERR_EBADF))?;
1028 self.inner
1029 .lock()
1030 .map_err(|_| poisoned())?
1031 .rm_dir_contents_with_options(dir, options)
1032 .map_err(|error| error.with_context("remove_dir_contents", &self.path))
1033 }
1034
1035 pub fn close(mut self) -> VfResult<()> {
1036 self.try_close()
1037 }
1038
1039 pub fn try_close(&mut self) -> VfResult<()> {
1041 let Some(dir) = self.dir.as_ref() else {
1042 return Ok(());
1043 };
1044 self.inner.lock().map_err(|_| poisoned())?.close_dir(dir)?;
1045 self.dir = None;
1046 Ok(())
1047 }
1048}
1049
1050impl<F: VecFs> Drop for FsDir<F> {
1051 fn drop(&mut self) {
1052 if let Some(dir) = self.dir.take() {
1053 self.inner.defer(
1054 CleanupTarget::Directory(dir, std::mem::take(&mut self.path)),
1055 |backend, target| {
1056 let CleanupTarget::Directory(dir, path) = target else {
1057 unreachable!()
1058 };
1059 backend
1060 .close_dir(dir)
1061 .map_err(|error| error.with_context("close_dir", path))
1062 },
1063 );
1064 }
1065 }
1066}
1067
1068impl<F: LinkFileSystem> FsClient<F> {
1069 pub fn symlink(&self, target: impl AsRef<Path>, link: impl AsRef<Path>) -> VfResult<()> {
1070 self.lock()?.symlink_one(target.as_ref(), link.as_ref())
1071 }
1072
1073 pub fn hard_link(&self, source: impl AsRef<Path>, link: impl AsRef<Path>) -> VfResult<()> {
1074 self.lock()?.hard_link_one(source.as_ref(), link.as_ref())
1075 }
1076
1077 pub fn read_link(&self, path: impl AsRef<Path>) -> VfResult<PathBuf> {
1078 self.lock()?.read_link_one(path.as_ref())
1079 }
1080}
1081
1082impl<F: CopyFileSystem> FsClient<F> {
1083 pub fn copy(&self, source: impl AsRef<Path>, destination: impl AsRef<Path>) -> VfResult<()> {
1084 self.lock()?.copy_one(source.as_ref(), destination.as_ref())
1085 }
1086}
1087
1088impl<F: VecFs> FsClient<F> {
1089 pub fn symlink_metadata_with_fields(
1092 &self,
1093 path: impl AsRef<Path>,
1094 fields: AttrMask,
1095 ) -> VfResult<Metadata> {
1096 let path = path.as_ref();
1097 let mut attrs = crate::VfAttrs {
1098 file: VfFile::from_os_path(path),
1099 masks: fields | AttrMask::MODE | AttrMask::SIZE,
1100 ..crate::VfAttrs::default()
1101 };
1102 self.lock()?
1103 .lgetattrsv(std::slice::from_mut(&mut attrs))
1104 .map_err(|error| error.with_context("symlink_metadata", path))?;
1105 Ok(vfsi_core::metadata_from_attrs(attrs))
1106 }
1107
1108 pub fn read_dirs_with_options<P: AsRef<Path>>(
1115 &self,
1116 paths: &[P],
1117 fields: AttrMask,
1118 options: ReadDirOptions,
1119 ) -> VfResult<Vec<DirectoryListing>> {
1120 if paths.is_empty() {
1121 return Ok(Vec::new());
1122 }
1123 let paths: Vec<&Path> = paths.iter().map(AsRef::as_ref).collect();
1124 let mut positions: HashMap<PathBuf, Vec<usize>> = HashMap::with_capacity(paths.len());
1125 let mut unique_paths = Vec::with_capacity(paths.len());
1126 for (index, path) in paths.iter().enumerate() {
1127 if let Some(indices) = positions.get_mut(*path) {
1128 indices.push(index);
1129 } else {
1130 positions.insert(path.to_path_buf(), vec![index]);
1131 unique_paths.push(*path);
1132 }
1133 }
1134 let mut listings: Vec<DirectoryListing> = paths
1135 .iter()
1136 .map(|path| DirectoryListing {
1137 path: path.to_path_buf(),
1138 entries: Vec::new(),
1139 })
1140 .collect();
1141 let mut entry_count = 0usize;
1142 let mut path_bytes = 0usize;
1143 const DIRECTORY_COHORT: usize = 32;
1146 for cohort in unique_paths.chunks(DIRECTORY_COHORT) {
1147 let mut callback_error = None;
1148 let requested = options
1149 .entry_limit()
1150 .saturating_sub(entry_count)
1151 .saturating_add(1);
1152 let result = self.lock()?.listdirv(
1153 cohort,
1154 fields | AttrMask::MODE | AttrMask::SIZE,
1155 requested,
1156 false,
1157 &mut |attributes, directory| {
1158 let Some(indices) = positions.get(directory) else {
1159 callback_error = Some(VfError::transport(
1160 None,
1161 format!(
1162 "read_dirs backend returned an unexpected directory: {}",
1163 directory.display()
1164 ),
1165 ));
1166 return false;
1167 };
1168 if !cohort.contains(&directory) {
1169 callback_error = Some(VfError::transport(
1170 None,
1171 "read_dirs backend returned a directory outside the active cohort",
1172 ));
1173 return false;
1174 }
1175 let Some(path) = attributes.file.path() else {
1176 callback_error = Some(VfError::transport(
1177 None,
1178 "read_dirs backend returned an entry without a path",
1179 ));
1180 return false;
1181 };
1182 if path.parent() != Some(directory) {
1183 callback_error = Some(VfError::transport(
1184 None,
1185 format!(
1186 "read_dirs backend returned an entry outside {}",
1187 directory.display()
1188 ),
1189 ));
1190 return false;
1191 }
1192 for &index in indices {
1193 entry_count += 1;
1194 path_bytes = path_bytes.saturating_add(path.as_os_str().len());
1195 if entry_count > options.entry_limit()
1196 || path_bytes > options.path_byte_limit()
1197 {
1198 callback_error = Some(
1199 VfError::failure(index, libc::EFBIG as u32)
1200 .with_context("read_dirs", directory),
1201 );
1202 return false;
1203 }
1204 listings[index].entries.push(DirEntry::new(
1205 path.to_path_buf(),
1206 vfsi_core::metadata_from_attrs(attributes.clone()),
1207 ));
1208 }
1209 true
1210 },
1211 );
1212 if let Some(error) = callback_error {
1213 return Err(error);
1214 }
1215 result.map_err(|error| {
1216 error.map_index(|index| {
1217 cohort
1218 .get(index)
1219 .and_then(|path| positions.get(*path))
1220 .and_then(|indices| indices.first())
1221 .copied()
1222 .unwrap_or(index)
1223 })
1224 })?;
1225 }
1226 Ok(listings)
1227 }
1228
1229 pub fn walk_with_options(
1233 &self,
1234 root: impl AsRef<Path>,
1235 fields: AttrMask,
1236 options: crate::WalkOptions,
1237 ) -> VfResult<Vec<DirectoryListing>> {
1238 let root = root.as_ref();
1239 let tree = self.lock()?.walk_with_options(
1240 root,
1241 fields | AttrMask::MODE | AttrMask::SIZE,
1242 options,
1243 &mut |_, _| {},
1244 )?;
1245 tree.into_iter()
1246 .map(|directory| {
1247 let entries = directory
1248 .entries
1249 .into_iter()
1250 .enumerate()
1251 .map(|(index, attributes)| {
1252 let path = attributes
1253 .file
1254 .path()
1255 .ok_or_else(|| {
1256 VfError::transport(
1257 None,
1258 format!("walk backend returned entry {index} without a path"),
1259 )
1260 })?
1261 .to_path_buf();
1262 Ok(DirEntry::new(
1263 path,
1264 vfsi_core::metadata_from_attrs(attributes),
1265 ))
1266 })
1267 .collect::<VfResult<Vec<_>>>()?;
1268 Ok(DirectoryListing {
1269 path: directory.path,
1270 entries,
1271 })
1272 })
1273 .collect()
1274 }
1275
1276 pub fn copyv<P: AsRef<Path>, Q: AsRef<Path>>(&self, pairs: &[(P, Q)]) -> VfResult<()> {
1279 let extents: Vec<_> = pairs
1280 .iter()
1281 .map(|(from, to)| {
1282 crate::ExtentPair::from_os_paths(from.as_ref(), 0, to.as_ref(), 0, None)
1283 })
1284 .collect();
1285 self.lock()?.copyv(&extents).map_err(|error| {
1286 error
1287 .index()
1288 .and_then(|index| pairs.get(index))
1289 .map_or(error.clone(), |(_, to)| {
1290 error.with_context("copyv", to.as_ref())
1291 })
1292 })
1293 }
1294
1295 pub fn remove_paths<P: AsRef<Path>>(&self, paths: &[P], recursive: bool) -> VfResult<()> {
1298 self.remove_paths_with_options(paths, recursive, RemoveOptions::default())
1299 }
1300
1301 pub fn remove_paths_with_options<P: AsRef<Path>>(
1303 &self,
1304 paths: &[P],
1305 recursive: bool,
1306 options: RemoveOptions,
1307 ) -> VfResult<()> {
1308 let paths: Vec<&Path> = paths.iter().map(AsRef::as_ref).collect();
1309 self.lock()?
1310 .rm_with_options(&paths, recursive, options)
1311 .map_err(|error| {
1312 error
1313 .index()
1314 .and_then(|index| paths.get(index))
1315 .map_or(error.clone(), |path| {
1316 error.with_context("remove_paths", path)
1317 })
1318 })
1319 }
1320
1321 pub fn metadata_many<P: AsRef<Path>>(
1324 &self,
1325 paths: &[P],
1326 fields: AttrMask,
1327 follow: bool,
1328 ) -> VfResult<Vec<Metadata>> {
1329 if paths.is_empty() {
1330 return Ok(Vec::new());
1331 }
1332 let mut attrs: Vec<_> = paths
1333 .iter()
1334 .map(|path| crate::VfAttrs {
1335 file: VfFile::from_os_path(path.as_ref()),
1336 masks: fields | AttrMask::MODE,
1337 ..crate::VfAttrs::default()
1338 })
1339 .collect();
1340 let result = {
1341 let mut backend = self.lock()?;
1342 if follow {
1343 backend.getattrsv(&mut attrs)
1344 } else {
1345 backend.lgetattrsv(&mut attrs)
1346 }
1347 };
1348 result.map_err(|error| match error.index() {
1349 Some(index) if index < paths.len() => {
1350 error.with_context("metadatav", paths[index].as_ref())
1351 }
1352 Some(_) => VfError::transport(None, "metadata backend returned an invalid error index"),
1353 None => error,
1354 })?;
1355 Ok(attrs
1356 .into_iter()
1357 .map(vfsi_core::metadata_from_attrs)
1358 .collect())
1359 }
1360
1361 pub fn symlink_metadatav(&self, paths: &[&Path]) -> VfResult<Vec<Metadata>> {
1363 let mut attrs: Vec<_> = paths
1364 .iter()
1365 .map(|path| crate::VfAttrs {
1366 file: VfFile::from_os_path(path),
1367 masks: AttrMask::MODE,
1368 ..crate::VfAttrs::default()
1369 })
1370 .collect();
1371 self.lock()?.lgetattrsv(&mut attrs).map_err(|error| {
1372 error
1373 .index()
1374 .and_then(|index| paths.get(index))
1375 .map_or(error.clone(), |path| {
1376 error.with_context("symlink_metadatav", path)
1377 })
1378 })?;
1379 Ok(attrs
1380 .into_iter()
1381 .map(vfsi_core::metadata_from_attrs)
1382 .collect())
1383 }
1384}
1385
1386impl<F: VectorFileSystem> FsClient<F> {
1387 pub fn openv(&self, requests: &[OpenRequest]) -> VfResult<Vec<FsFile<F>>> {
1393 let mut filesystem = self.lock()?;
1394 let files = filesystem.open_many(requests).map_err(|error| {
1395 error
1396 .index()
1397 .and_then(|index| requests.get(index))
1398 .map_or(error.clone(), |request| {
1399 error.with_context("openv", &request.path)
1400 })
1401 })?;
1402 if files.len() != requests.len() {
1403 let error = wrong_result_count("openv", requests.len(), files.len());
1404 for file in &files {
1405 let _ = filesystem.close_one(file);
1406 }
1407 return Err(error);
1408 }
1409 drop(filesystem);
1410 Ok(files
1411 .into_iter()
1412 .zip(requests)
1413 .map(|(file, request)| FsFile {
1414 inner: Arc::clone(&self.inner),
1415 file: Some(file),
1416 path: request.path.clone(),
1417 })
1418 .collect())
1419 }
1420
1421 pub fn try_closev<'a>(&self, files: impl IntoIterator<Item = &'a mut FsFile<F>>) -> VfResult<()>
1425 where
1426 F: 'a,
1427 {
1428 let mut files: Vec<_> = files.into_iter().collect();
1429 for (index, file) in files.iter().enumerate() {
1430 self.validate_owner(file, index)?;
1431 }
1432 let positions: Vec<_> = files
1433 .iter()
1434 .enumerate()
1435 .filter_map(|(index, file)| (!file.is_closed()).then_some(index))
1436 .collect();
1437 files.retain(|file| !file.is_closed());
1438 if files.is_empty() {
1439 return Ok(());
1440 }
1441 let descriptors: Vec<VfFile> = files
1442 .iter()
1443 .map(|file| file.raw().cloned())
1444 .collect::<VfResult<_>>()?;
1445 self.lock()?.close_many(&descriptors).map_err(|error| {
1446 error
1447 .index()
1448 .and_then(|index| files.get(index))
1449 .map_or(error.clone(), |file| {
1450 error.with_context("closev", file.path())
1451 })
1452 .map_index(|index| positions.get(index).copied().unwrap_or(index))
1453 })?;
1454 for file in &mut files {
1455 file.file = None;
1456 }
1457 Ok(())
1458 }
1459
1460 pub fn closev(&self, mut files: Vec<FsFile<F>>) -> VfResult<()> {
1466 self.try_closev(&mut files)
1467 }
1468
1469 pub fn readv(&self, requests: &[FsRead<'_, F>]) -> VfResult<Vec<FsReadResult>> {
1473 self.readv_with_limit(requests, self.limits.max_read_bytes)
1474 }
1475
1476 pub fn readv_with_limit(
1478 &self,
1479 requests: &[FsRead<'_, F>],
1480 max_total_bytes: usize,
1481 ) -> VfResult<Vec<FsReadResult>> {
1482 self.readv_with_limit_projected(requests, max_total_bytes, |request| request)
1483 }
1484
1485 pub fn readv_with_limit_projected<'b, T>(
1488 &self,
1489 requests: &[T],
1490 max_total_bytes: usize,
1491 project: impl for<'r> Fn(&'r T) -> &'r FsRead<'b, F>,
1492 ) -> VfResult<Vec<FsReadResult>>
1493 where
1494 F: 'b,
1495 {
1496 let mut requested = 0usize;
1497 for (index, request) in requests.iter().map(&project).enumerate() {
1498 requested = requested
1499 .checked_add(request.length)
1500 .ok_or_else(|| VfError::client(index, libc::EFBIG as u32))?;
1501 if requested > max_total_bytes {
1502 return Err(VfError::client(index, libc::EFBIG as u32)
1503 .with_context("readv", request.file.path()));
1504 }
1505 }
1506 let reads = self.read_ops(requests, &project)?;
1507 let results = self.lock()?.read_many(&reads).map_err(|error| {
1508 error
1509 .index()
1510 .and_then(|index| requests.get(index).map(&project))
1511 .map_or(error.clone(), |request| {
1512 error.with_context("readv", request.file.path())
1513 })
1514 })?;
1515 if results.len() != requests.len() {
1516 return Err(wrong_result_count("readv", requests.len(), results.len()));
1517 }
1518 validate_read_results("readv", &reads, &results).map_err(|error| {
1519 error
1520 .index()
1521 .and_then(|index| requests.get(index).map(&project))
1522 .map_or(error.clone(), |request| {
1523 error.with_context("readv", request.file.path())
1524 })
1525 })?;
1526 Ok(results.into_iter().map(read_result).collect())
1527 }
1528
1529 fn read_ops<'b, T>(
1530 &self,
1531 requests: &[T],
1532 project: impl for<'r> Fn(&'r T) -> &'r FsRead<'b, F>,
1533 ) -> VfResult<Vec<ReadOp>>
1534 where
1535 F: 'b,
1536 {
1537 let mut reads = Vec::with_capacity(requests.len());
1538 for (index, request) in requests.iter().map(&project).enumerate() {
1539 self.validate_owner(request.file, index)?;
1540 reads.push(ReadOp::new(
1541 request.file.raw()?.clone(),
1542 request.offset,
1543 request.length,
1544 ));
1545 }
1546 Ok(reads)
1547 }
1548
1549 pub fn readv_into(
1550 &self,
1551 requests: &mut [FsReadInto<'_, F>],
1552 ) -> VfResult<Vec<FsReadIntoResult>> {
1553 self.readv_into_with_limit(requests, self.limits.max_read_bytes)
1554 }
1555
1556 pub fn readv_into_with_limit(
1559 &self,
1560 requests: &mut [FsReadInto<'_, F>],
1561 max_bytes: usize,
1562 ) -> VfResult<Vec<FsReadIntoResult>> {
1563 self.readv_into_with_limit_projected(
1564 requests,
1565 max_bytes,
1566 |request| request,
1567 |request| request,
1568 )
1569 }
1570
1571 pub fn readv_into_with_limit_projected<'b, T>(
1575 &self,
1576 requests: &mut [T],
1577 max_bytes: usize,
1578 project: impl for<'r> Fn(&'r T) -> &'r FsReadInto<'b, F>,
1579 mut project_mut: impl for<'r> FnMut(&'r mut T) -> &'r mut FsReadInto<'b, F>,
1580 ) -> VfResult<Vec<FsReadIntoResult>>
1581 where
1582 F: 'b,
1583 {
1584 let mut requested = 0usize;
1585 let mut reads = Vec::with_capacity(requests.len());
1586 for (index, request) in requests.iter().map(&project).enumerate() {
1587 self.validate_owner(request.file, index)?;
1588 requested = requested
1589 .checked_add(request.buffer.len())
1590 .ok_or_else(|| VfError::client(index, libc::EFBIG as u32))?;
1591 if requested > max_bytes {
1592 return Err(VfError::client(index, libc::EFBIG as u32));
1593 }
1594 reads.push(ReadOp::new(
1595 request.file.raw()?.clone(),
1596 request.offset,
1597 request.buffer.len(),
1598 ));
1599 }
1600 let results = {
1601 let mut buffers: Vec<&mut [u8]> = requests
1602 .iter_mut()
1603 .map(&mut project_mut)
1604 .map(|request| &mut *request.buffer)
1605 .collect();
1606 self.lock()?.read_many_into(&reads, &mut buffers)
1607 }
1608 .map_err(|error| {
1609 error
1610 .index()
1611 .and_then(|index| requests.get(index).map(&project))
1612 .map_or(error.clone(), |request| {
1613 error.with_context("readv_into", request.file.path())
1614 })
1615 })?;
1616 validate_read_into_results("readv_into", &reads, &results).map_err(|error| {
1617 error
1618 .index()
1619 .and_then(|index| requests.get(index).map(&project))
1620 .map_or(error.clone(), |request| {
1621 error.with_context("readv_into", request.file.path())
1622 })
1623 })?;
1624 Ok(results
1625 .into_iter()
1626 .map(|result| FsReadIntoResult {
1627 offset: result.offset,
1628 read: result.read,
1629 eof: result.eof,
1630 })
1631 .collect())
1632 }
1633
1634 pub fn writev(&self, requests: &[FsWrite<'_, F>]) -> VfResult<Vec<FsWriteResult>> {
1635 self.writev_projected(requests, |request| request)
1636 }
1637
1638 pub fn writev_projected<'b, T>(
1641 &self,
1642 requests: &[T],
1643 project: impl for<'r> Fn(&'r T) -> &'r FsWrite<'b, F>,
1644 ) -> VfResult<Vec<FsWriteResult>>
1645 where
1646 F: 'b,
1647 {
1648 self.writev_mapped(requests, |item| {
1649 let request = project(item);
1650 FsWrite {
1651 file: request.file,
1652 offset: request.offset,
1653 data: request.data,
1654 }
1655 })
1656 }
1657
1658 #[doc(hidden)]
1661 pub fn writev_mapped<'b, T>(
1662 &self,
1663 requests: &[T],
1664 project: impl Fn(&T) -> FsWrite<'b, F>,
1665 ) -> VfResult<Vec<FsWriteResult>>
1666 where
1667 F: 'b,
1668 {
1669 let writes = self.write_ops(requests, &project)?;
1670 let results = self.lock()?.write_many(&writes).map_err(|error| {
1671 error
1672 .index()
1673 .and_then(|index| requests.get(index).map(&project))
1674 .map_or(error.clone(), |request| {
1675 error.with_context("writev", request.file.path())
1676 })
1677 })?;
1678 if results.len() != requests.len() {
1679 return Err(wrong_result_count("writev", requests.len(), results.len()));
1680 }
1681 validate_write_results("writev", &writes, &results).map_err(|error| {
1682 error
1683 .index()
1684 .and_then(|index| requests.get(index).map(&project))
1685 .map_or(error.clone(), |request| {
1686 error.with_context("writev", request.file.path())
1687 })
1688 })?;
1689 Ok(results.into_iter().map(write_result).collect())
1690 }
1691
1692 pub fn write_allv(&self, requests: &[FsWrite<'_, F>]) -> VfResult<Vec<FsWriteResult>> {
1698 self.write_allv_projected(requests, |request| request)
1699 }
1700
1701 pub fn write_allv_projected<'b, T>(
1704 &self,
1705 requests: &[T],
1706 project: impl for<'r> Fn(&'r T) -> &'r FsWrite<'b, F>,
1707 ) -> VfResult<Vec<FsWriteResult>>
1708 where
1709 F: 'b,
1710 {
1711 self.write_allv_mapped(requests, |item| {
1712 let request = project(item);
1713 FsWrite {
1714 file: request.file,
1715 offset: request.offset,
1716 data: request.data,
1717 }
1718 })
1719 }
1720
1721 #[doc(hidden)]
1724 pub fn write_allv_mapped<'b, T>(
1725 &self,
1726 requests: &[T],
1727 project: impl Fn(&T) -> FsWrite<'b, F>,
1728 ) -> VfResult<Vec<FsWriteResult>>
1729 where
1730 F: 'b,
1731 {
1732 let mut prior_by_path: HashMap<&Path, Vec<(usize, u64, u64)>> = HashMap::new();
1735 let mut blocked_by = vec![Vec::new(); requests.len()];
1736 for (index, request) in requests.iter().map(&project).enumerate() {
1737 self.validate_owner(request.file, index)?;
1738 request.file.raw().map_err(|error| {
1739 error
1740 .with_index(index)
1741 .with_context("write_allv", request.file.path())
1742 })?;
1743 let VfOffset::At(offset) = request.offset else {
1744 return Err(VfError::client(index, crate::ERR_INVAL)
1745 .with_context("write_allv", request.file.path()));
1746 };
1747 let end = offset
1748 .checked_add(request.data.len() as u64)
1749 .ok_or_else(|| {
1750 VfError::client(index, libc::EOVERFLOW as u32)
1751 .with_context("write_allv", request.file.path())
1752 })?;
1753 let prior = prior_by_path.entry(request.file.path()).or_default();
1757 for &(earlier, start, earlier_end) in prior.iter() {
1758 if offset < earlier_end && start < end {
1759 blocked_by[index].push(earlier);
1760 }
1761 }
1762 prior.push((index, offset, end));
1763 }
1764 let mut totals = vec![0usize; requests.len()];
1765 let mut stable = vec![true; requests.len()];
1766 let mut pending: Vec<usize> = (0..requests.len())
1767 .filter(|&index| !project(&requests[index]).data.is_empty())
1768 .collect();
1769 while !pending.is_empty() {
1770 let wave_indices: Vec<usize> = pending
1771 .iter()
1772 .copied()
1773 .filter(|&index| {
1774 blocked_by[index]
1775 .iter()
1776 .all(|&earlier| totals[earlier] == project(&requests[earlier]).data.len())
1777 })
1778 .collect();
1779 let wave: Vec<_> = wave_indices
1780 .iter()
1781 .map(|&index| {
1782 let request = project(&requests[index]);
1783 let VfOffset::At(offset) = request.offset else {
1784 return Err(VfError::client(index, crate::ERR_INVAL)
1785 .with_context("write_allv", request.file.path()));
1786 };
1787 let offset = offset.checked_add(totals[index] as u64).ok_or_else(|| {
1788 VfError::client(index, libc::EOVERFLOW as u32)
1789 .with_context("write_allv", request.file.path())
1790 })?;
1791 Ok(FsWrite {
1792 file: request.file,
1793 offset: VfOffset::At(offset),
1794 data: &request.data[totals[index]..],
1795 })
1796 })
1797 .collect::<VfResult<_>>()?;
1798 let results = self.writev(&wave).map_err(|error| {
1799 error.map_index(|index| wave_indices.get(index).copied().unwrap_or(index))
1800 })?;
1801 for (&index, result) in wave_indices.iter().zip(results) {
1802 if result.written == 0 {
1803 return Err(VfError::client(index, crate::ERR_IO)
1804 .with_context("write_allv", project(&requests[index]).file.path()));
1805 }
1806 totals[index] += result.written;
1807 stable[index] &= result.stable;
1808 }
1809 pending.retain(|&index| totals[index] < project(&requests[index]).data.len());
1810 }
1811 requests
1812 .iter()
1813 .map(&project)
1814 .enumerate()
1815 .map(|(index, request)| {
1816 let VfOffset::At(offset) = request.offset else {
1817 return Err(VfError::client(index, crate::ERR_INVAL)
1818 .with_context("write_allv", request.file.path()));
1819 };
1820 Ok(FsWriteResult {
1821 offset,
1822 written: totals[index],
1823 stable: stable[index],
1824 })
1825 })
1826 .collect()
1827 }
1828
1829 fn write_ops<'a, 'b, T>(
1830 &self,
1831 requests: &'a [T],
1832 project: impl Fn(&T) -> FsWrite<'b, F>,
1833 ) -> VfResult<Vec<WriteOpRef<'a>>>
1834 where
1835 F: 'b,
1836 'b: 'a,
1837 {
1838 let mut writes = Vec::with_capacity(requests.len());
1839 for (index, request) in requests.iter().map(&project).enumerate() {
1840 self.validate_owner(request.file, index)?;
1841 writes.push(WriteOpRef::new(
1842 request
1843 .file
1844 .raw()
1845 .map_err(|error| error.with_index(index))?,
1846 request.offset,
1847 request.data,
1848 ));
1849 }
1850 Ok(writes)
1851 }
1852
1853 fn validate_owner(&self, file: &FsFile<F>, index: usize) -> VfResult<()> {
1854 if Arc::ptr_eq(&self.inner, &file.inner) {
1855 Ok(())
1856 } else {
1857 Err(VfError::client(index, crate::ERR_INVAL))
1858 }
1859 }
1860}
1861
1862impl<F: VectorFileSystem + VecFs> FsClient<F> {
1863 pub fn read_files<P: AsRef<Path>>(&self, paths: &[P]) -> VfResult<Vec<Vec<u8>>> {
1870 self.read_files_with_options(
1871 paths,
1872 ReadAllOptions::new().max_total_bytes(self.limits.max_read_bytes),
1873 )
1874 }
1875
1876 pub fn read_files_with_options<P: AsRef<Path>>(
1878 &self,
1879 paths: &[P],
1880 options: ReadAllOptions,
1881 ) -> VfResult<Vec<Vec<u8>>> {
1882 let files: Vec<_> = paths
1883 .iter()
1884 .map(|path| VfFile::from_os_path(path.as_ref()))
1885 .collect();
1886 let result = self
1887 .lock()?
1888 .read_allv_with_options(&files, options)
1889 .map_err(|error| {
1890 error
1891 .index()
1892 .and_then(|index| paths.get(index))
1893 .map_or(error.clone(), |path| {
1894 error.with_context("read_files", path.as_ref())
1895 })
1896 });
1897 result.and_then(|buffers| {
1898 if buffers.len() != paths.len() {
1899 Err(wrong_result_count("read_files", paths.len(), buffers.len()))
1900 } else {
1901 Ok(buffers)
1902 }
1903 })
1904 }
1905
1906 pub fn write_files<P: AsRef<Path>, B: AsRef<[u8]>>(&self, entries: &[(P, B)]) -> VfResult<()> {
1913 let mut seen = HashSet::with_capacity(entries.len());
1914 for (index, (path, _)) in entries.iter().enumerate() {
1915 if !seen.insert(path.as_ref()) {
1916 return Err(VfError::client(index, crate::ERR_INVAL)
1917 .with_context("write_files", path.as_ref()));
1918 }
1919 }
1920 let requests: Vec<_> = entries
1921 .iter()
1922 .map(|(path, _)| {
1923 OpenRequest::new(
1924 path.as_ref(),
1925 OpenFlags::WRITE | OpenFlags::CREATE | OpenFlags::TRUNCATE,
1926 )
1927 })
1928 .collect();
1929 let files = self.openv(&requests)?;
1930 let writes: Vec<_> = files
1931 .iter()
1932 .zip(entries)
1933 .map(|(file, (_, data))| file.write_request_at(0, data.as_ref()))
1934 .collect();
1935 let result = self.write_allv(&writes);
1936 drop(writes);
1937 let close_result = self.closev(files);
1938 result?;
1939 close_result
1940 }
1941}
1942
1943pub struct OpenOptions<'a, F: FileSystem> {
1945 client: &'a FsClient<F>,
1946 flags: OpenFlags,
1947 mode: u32,
1948}
1949
1950impl<F: FileSystem> Clone for OpenOptions<'_, F> {
1951 fn clone(&self) -> Self {
1952 Self {
1953 client: self.client,
1954 flags: self.flags,
1955 mode: self.mode,
1956 }
1957 }
1958}
1959
1960impl<F: FileSystem> fmt::Debug for OpenOptions<'_, F> {
1961 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
1962 formatter
1963 .debug_struct("OpenOptions")
1964 .field("flags", &self.flags)
1965 .field("mode", &format_args!("{:#o}", self.mode))
1966 .finish()
1967 }
1968}
1969
1970impl<'a, F: FileSystem> OpenOptions<'a, F> {
1971 fn new(client: &'a FsClient<F>) -> Self {
1972 Self {
1973 client,
1974 flags: OpenFlags::empty(),
1975 mode: 0o666,
1976 }
1977 }
1978
1979 pub fn read(&mut self, enabled: bool) -> &mut Self {
1980 self.flags.set(OpenFlags::READ, enabled);
1981 self
1982 }
1983
1984 pub fn write(&mut self, enabled: bool) -> &mut Self {
1985 self.flags.set(OpenFlags::WRITE, enabled);
1986 self
1987 }
1988
1989 pub fn append(&mut self, enabled: bool) -> &mut Self {
1990 self.flags.set(OpenFlags::APPEND, enabled);
1991 self
1992 }
1993
1994 pub fn truncate(&mut self, enabled: bool) -> &mut Self {
1995 self.flags.set(OpenFlags::TRUNCATE, enabled);
1996 self
1997 }
1998
1999 pub fn create(&mut self, enabled: bool) -> &mut Self {
2000 self.flags.set(OpenFlags::CREATE, enabled);
2001 self
2002 }
2003
2004 pub fn create_new(&mut self, enabled: bool) -> &mut Self {
2005 self.flags.set(OpenFlags::CREATE_NEW, enabled);
2006 self
2007 }
2008
2009 pub fn mode(&mut self, mode: u32) -> &mut Self {
2010 self.mode = mode;
2011 self
2012 }
2013
2014 pub fn open(&self, path: impl AsRef<Path>) -> VfResult<FsFile<F>> {
2015 self.client
2016 .open_with(OpenRequest::new(path.as_ref(), self.flags).mode(self.mode))
2017 }
2018}
2019
2020impl<F: VectorFileSystem> OpenOptions<'_, F> {
2021 pub fn openv<P: AsRef<Path>>(&self, paths: &[P]) -> VfResult<Vec<FsFile<F>>> {
2022 let requests: Vec<OpenRequest> = paths
2023 .iter()
2024 .map(|path| OpenRequest::new(path.as_ref(), self.flags).mode(self.mode))
2025 .collect();
2026 self.client.openv(&requests)
2027 }
2028}
2029
2030pub struct SetMetadata<'a, F: MetadataFileSystem> {
2032 client: &'a FsClient<F>,
2033 path: PathBuf,
2034 update: MetadataUpdate,
2035 follow: bool,
2036}
2037
2038impl<F: MetadataFileSystem> Clone for SetMetadata<'_, F> {
2039 fn clone(&self) -> Self {
2040 Self {
2041 client: self.client,
2042 path: self.path.clone(),
2043 update: self.update.clone(),
2044 follow: self.follow,
2045 }
2046 }
2047}
2048
2049impl<F: MetadataFileSystem> fmt::Debug for SetMetadata<'_, F> {
2050 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
2051 formatter
2052 .debug_struct("SetMetadata")
2053 .field("path", &self.path)
2054 .field("update", &self.update)
2055 .field("follow", &self.follow)
2056 .finish_non_exhaustive()
2057 }
2058}
2059
2060impl<F: MetadataFileSystem> SetMetadata<'_, F> {
2061 pub fn permissions(&mut self, permissions: Permissions) -> &mut Self {
2062 self.update.permissions = Some(permissions);
2063 self
2064 }
2065
2066 pub fn len(&mut self, len: u64) -> &mut Self {
2067 self.update.len = Some(len);
2068 self
2069 }
2070
2071 pub fn accessed(&mut self, accessed: SystemTime) -> &mut Self {
2072 self.update.accessed = Some(accessed);
2073 self
2074 }
2075
2076 pub fn modified(&mut self, modified: SystemTime) -> &mut Self {
2077 self.update.modified = Some(modified);
2078 self
2079 }
2080
2081 pub fn follow_symlinks(&mut self, follow: bool) -> &mut Self {
2082 self.follow = follow;
2083 self
2084 }
2085
2086 pub fn apply(&self) -> VfResult<()> {
2087 self.client
2088 .lock()?
2089 .set_metadata_path(&self.path, self.update.clone(), self.follow)
2090 }
2091}
2092
2093pub struct FsFile<F: FileSystem> {
2099 inner: Arc<SharedBackend<F>>,
2100 file: Option<VfFile>,
2101 path: PathBuf,
2102}
2103
2104impl<F: FileSystem> fmt::Debug for FsFile<F> {
2105 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
2106 formatter
2107 .debug_struct("FsFile")
2108 .field("path", &self.path)
2109 .field("is_open", &self.file.is_some())
2110 .finish()
2111 }
2112}
2113
2114impl<F: FileSystem> FsFile<F> {
2115 pub fn is_closed(&self) -> bool {
2117 self.file.is_none()
2118 }
2119 fn raw(&self) -> VfResult<&VfFile> {
2120 self.file
2121 .as_ref()
2122 .ok_or_else(|| VfError::client(0, crate::ERR_EBADF))
2123 }
2124
2125 pub fn read_at(&self, buffer: &mut [u8], offset: u64) -> VfResult<usize> {
2127 self.read_into(buffer, VfOffset::At(offset))
2128 }
2129
2130 pub fn write_at(&self, buffer: &[u8], offset: u64) -> VfResult<usize> {
2132 self.write_from(buffer, VfOffset::At(offset))
2133 }
2134
2135 pub fn read_request_at(&self, offset: u64, length: usize) -> FsRead<'_, F> {
2136 FsRead {
2137 file: self,
2138 offset: VfOffset::At(offset),
2139 length,
2140 }
2141 }
2142
2143 pub fn path(&self) -> &Path {
2147 &self.path
2148 }
2149
2150 pub fn metadata(&self) -> VfResult<Metadata> {
2152 let attributes = AttrMask::MODE
2153 | AttrMask::SIZE
2154 | AttrMask::NLINK
2155 | AttrMask::FILEID
2156 | AttrMask::UID
2157 | AttrMask::GID
2158 | AttrMask::ATIME
2159 | AttrMask::MTIME
2160 | AttrMask::CTIME
2161 | AttrMask::CHANGE;
2162 self.inner
2163 .lock()
2164 .map_err(|_| poisoned())?
2165 .metadata(MetadataQuery::new(self.raw()?.clone(), attributes))
2166 .map(vfsi_core::metadata_from_attrs)
2167 .map_err(|error| error.with_context("metadata", &self.path))
2168 }
2169
2170 pub fn set_len(&self, len: u64) -> VfResult<()> {
2172 let mut update = SetAttributes::new(self.raw()?.clone());
2173 update.size = Some(len);
2174 self.inner
2175 .lock()
2176 .map_err(|_| poisoned())?
2177 .set_attributes(update)
2178 .map_err(|error| error.with_context("set_len", &self.path))
2179 }
2180
2181 pub fn set_permissions(&self, permissions: Permissions) -> VfResult<()> {
2183 let mut update = SetAttributes::new(self.raw()?.clone());
2184 update.mode = Some(permissions.mode());
2185 self.inner
2186 .lock()
2187 .map_err(|_| poisoned())?
2188 .set_attributes(update)
2189 .map_err(|error| error.with_context("set_permissions", &self.path))
2190 }
2191
2192 pub fn write_request_at<'a>(&'a self, offset: u64, data: &'a [u8]) -> FsWrite<'a, F> {
2193 FsWrite {
2194 file: self,
2195 offset: VfOffset::At(offset),
2196 data,
2197 }
2198 }
2199
2200 pub fn read_request_at_into<'a>(
2201 &'a self,
2202 offset: u64,
2203 buffer: &'a mut [u8],
2204 ) -> FsReadInto<'a, F> {
2205 FsReadInto {
2206 file: self,
2207 offset: VfOffset::At(offset),
2208 buffer,
2209 }
2210 }
2211
2212 pub fn read_native(&mut self, buffer: &mut [u8]) -> VfResult<usize> {
2213 self.read_into(buffer, VfOffset::Cur)
2214 }
2215
2216 pub fn read_to_end_with_limit(&mut self, max_bytes: usize) -> VfResult<Vec<u8>> {
2224 let mut data = Vec::new();
2225 loop {
2226 let remaining = max_bytes - data.len();
2227 if remaining == 0 {
2228 let mut probe = [0_u8; 1];
2229 if self.read_native(&mut probe)? == 0 {
2230 return Ok(data);
2231 }
2232 return Err(VfError::client(0, libc::EFBIG as u32)
2233 .with_context("read_to_end_with_limit", &self.path));
2234 }
2235 let chunk = remaining.min(crate::DEFAULT_READ_STREAM_CHUNK_BYTES);
2236 data.try_reserve(chunk).map_err(|_| {
2237 VfError::client(0, libc::ENOMEM as u32)
2238 .with_context("read_to_end_with_limit", &self.path)
2239 })?;
2240 let start = data.len();
2241 data.resize(start + chunk, 0);
2242 let count = self.read_native(&mut data[start..])?;
2243 data.truncate(start + count);
2244 if count == 0 {
2245 return Ok(data);
2246 }
2247 }
2248 }
2249
2250 pub fn write_native(&mut self, buffer: &[u8]) -> VfResult<usize> {
2251 self.write_from(buffer, VfOffset::Cur)
2252 }
2253
2254 fn read_into(&self, buffer: &mut [u8], offset: VfOffset) -> VfResult<usize> {
2255 if buffer.is_empty() {
2256 return Ok(0);
2257 }
2258 let request = ReadOp::new(self.raw()?.clone(), offset, buffer.len());
2259 let result = self
2260 .inner
2261 .lock()
2262 .map_err(|_| poisoned())?
2263 .read_one_into(&request, buffer)
2264 .map_err(|error| error.with_context("read", &self.path))?;
2265 validate_read_into_results(
2266 "read",
2267 std::slice::from_ref(&request),
2268 std::slice::from_ref(&result),
2269 )
2270 .map_err(|error| error.with_context("read", &self.path))?;
2271 Ok(result.read)
2272 }
2273
2274 fn write_from(&self, buffer: &[u8], offset: VfOffset) -> VfResult<usize> {
2275 if buffer.is_empty() {
2276 return Ok(0);
2277 }
2278 let file = self.raw()?.clone();
2279 let result = self
2280 .inner
2281 .lock()
2282 .map_err(|_| poisoned())?
2283 .write_one(WriteOpRef::new(&file, offset, buffer))
2284 .map_err(|error| error.with_context("write", &self.path))?;
2285 if result.written > buffer.len() {
2286 return Err(VfError::client(0, crate::ERR_IO).with_context("write", &self.path));
2287 }
2288 Ok(result.written)
2289 }
2290
2291 pub fn sync_data(&self) -> VfResult<()> {
2292 self.inner
2293 .lock()
2294 .map_err(|_| poisoned())?
2295 .sync_data(self.raw()?)
2296 .map_err(|error| error.with_context("sync_data", &self.path))
2297 }
2298
2299 pub fn sync_all(&self) -> VfResult<()> {
2300 self.inner
2301 .lock()
2302 .map_err(|_| poisoned())?
2303 .sync_all(self.raw()?)
2304 .map_err(|error| error.with_context("sync_all", &self.path))
2305 }
2306
2307 pub fn try_close(&mut self) -> VfResult<()> {
2313 let Some(file) = self.file.as_ref() else {
2314 return Ok(());
2315 };
2316 self.inner
2317 .lock()
2318 .map_err(|_| poisoned())?
2319 .close_one(file)
2320 .map_err(|error| error.with_context("close", &self.path))?;
2321 self.file = None;
2322 Ok(())
2323 }
2324
2325 pub fn close(mut self) -> VfResult<()> {
2329 self.try_close()
2330 }
2331
2332 pub fn seek_native(&mut self, position: IoSeekFrom) -> VfResult<u64> {
2334 self.inner
2335 .lock()
2336 .map_err(|_| poisoned())?
2337 .seek_one(self.raw()?, position)
2338 .map_err(|error| error.with_context("seek", &self.path))
2339 }
2340}
2341
2342impl<F: FileSystem> Read for FsFile<F> {
2343 fn read(&mut self, buffer: &mut [u8]) -> io::Result<usize> {
2344 self.read_native(buffer).map_err(io_error)
2345 }
2346}
2347
2348impl<F: FileSystem> Write for FsFile<F> {
2349 fn write(&mut self, buffer: &[u8]) -> io::Result<usize> {
2350 self.write_native(buffer).map_err(io_error)
2351 }
2352
2353 fn flush(&mut self) -> io::Result<()> {
2354 self.sync_data().map_err(io_error)
2355 }
2356}
2357
2358impl<F: FileSystem> Seek for FsFile<F> {
2359 fn seek(&mut self, position: IoSeekFrom) -> io::Result<u64> {
2360 self.seek_native(position).map_err(io_error)
2361 }
2362}
2363
2364pub struct FsRead<'a, F: FileSystem> {
2366 file: &'a FsFile<F>,
2367 offset: VfOffset,
2368 length: usize,
2369}
2370
2371pub struct FsWrite<'a, F: FileSystem> {
2373 file: &'a FsFile<F>,
2374 offset: VfOffset,
2375 data: &'a [u8],
2376}
2377
2378pub struct FsReadInto<'a, F: FileSystem> {
2380 file: &'a FsFile<F>,
2381 offset: VfOffset,
2382 buffer: &'a mut [u8],
2383}
2384
2385impl<F: FileSystem> Drop for FsFile<F> {
2386 fn drop(&mut self) {
2387 if let Some(file) = self.file.take() {
2388 self.inner.defer(
2389 CleanupTarget::File(file, std::mem::take(&mut self.path)),
2390 |backend, target| {
2391 let CleanupTarget::File(file, path) = target else {
2392 unreachable!()
2393 };
2394 backend
2395 .close_deferred(file)
2396 .map_err(|error| error.with_context("close", path))
2397 },
2398 );
2399 }
2400 }
2401}
2402
2403#[cfg(test)]
2404mod traversal_tests {
2405 use super::*;
2406 use std::sync::atomic::{AtomicUsize, Ordering};
2407
2408 struct Snapshot {
2411 remaining: std::vec::IntoIter<DirEntry>,
2412 live: Arc<AtomicUsize>,
2413 }
2414 impl Drop for Snapshot {
2415 fn drop(&mut self) {
2416 self.live.fetch_sub(1, Ordering::SeqCst);
2417 }
2418 }
2419 fn snapshot_pages(
2420 live: Arc<AtomicUsize>,
2421 peak: Arc<AtomicUsize>,
2422 ) -> impl FnMut(&Path, Option<crate::DirPageCursor>, usize, usize) -> VfResult<DirectoryPage>
2423 {
2424 move |path, cursor, page_size, max_entries| {
2425 let mut snapshot = match cursor {
2426 Some(cursor) => cursor.into_state::<Snapshot>()?,
2427 None => {
2428 let depth = path.components().count();
2429 let entries = (0..129)
2430 .take(max_entries)
2431 .map(|index| {
2432 let path = path.join(format!("entry-{index}"));
2433 let attrs = crate::VfAttrs {
2434 file: VfFile::from_os_path(&path),
2435 ftype: if index == 0 && depth < 5 {
2436 crate::VfType::Directory
2437 } else {
2438 crate::VfType::Regular
2439 },
2440 ..Default::default()
2441 };
2442 DirEntry::new(path, vfsi_core::metadata_from_attrs(attrs))
2443 })
2444 .collect::<Vec<_>>();
2445 peak.fetch_max(live.fetch_add(1, Ordering::SeqCst) + 1, Ordering::SeqCst);
2446 Snapshot {
2447 remaining: entries.into_iter(),
2448 live: Arc::clone(&live),
2449 }
2450 }
2451 };
2452 let entries = snapshot.remaining.by_ref().take(page_size).collect();
2453 let next = if snapshot.remaining.len() == 0 {
2454 None
2455 } else {
2456 Some(crate::DirPageCursor::new(snapshot))
2457 };
2458 Ok((entries, next))
2459 }
2460 }
2461 #[test]
2462 fn traversal_never_retains_multiple_fallback_snapshots() {
2463 let live = Arc::new(AtomicUsize::new(0));
2464 let peak = Arc::new(AtomicUsize::new(0));
2465 let mut seen = 0;
2466 let completion = visit_walk_pages(
2467 Path::new("/tree"),
2468 crate::WalkOptions::new().max_entries(1000),
2469 snapshot_pages(Arc::clone(&live), Arc::clone(&peak)),
2470 |_| {
2471 seen += 1;
2472 Ok(std::ops::ControlFlow::Continue(()))
2473 },
2474 )
2475 .unwrap();
2476 assert_eq!(completion, TraversalCompletion::Complete);
2477 assert_eq!(seen, 4 * 129);
2478 assert_eq!(live.load(Ordering::SeqCst), 0);
2479 assert_eq!(peak.load(Ordering::SeqCst), 1);
2480 }
2481 #[test]
2482 fn traversal_drops_fallback_snapshot_on_stop_error_and_limit() {
2483 for outcome in 0..3 {
2484 let live = Arc::new(AtomicUsize::new(0));
2485 let peak = Arc::new(AtomicUsize::new(0));
2486 let result = visit_walk_pages(
2487 Path::new("/tree"),
2488 crate::WalkOptions::new().max_entries(if outcome == 2 { 2 } else { 1000 }),
2489 snapshot_pages(Arc::clone(&live), peak),
2490 |_| match outcome {
2491 0 => Ok(std::ops::ControlFlow::Break(())),
2492 1 => Err(VfError::client(0, libc::EIO as u32)),
2493 _ => Ok(std::ops::ControlFlow::Continue(())),
2494 },
2495 );
2496 match outcome {
2497 0 => assert_eq!(result.unwrap(), TraversalCompletion::Stopped),
2498 1 => assert_eq!(result.unwrap_err().err_no(), libc::EIO as u32),
2499 _ => assert_eq!(result.unwrap_err().err_no(), libc::EFBIG as u32),
2500 }
2501 assert_eq!(live.load(Ordering::SeqCst), 0);
2502 }
2503 }
2504}