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, Backend, Capabilities, DirEntry, FileSystem, Metadata, MetadataQuery, MetadataUpdate,
20 OpenFlags, OpenRequest, Permissions, ReadAllOptions, ReadDirOptions, ReadOp, ReadResult,
21 ReadStreamOptions, RemoveOptions, VfDir, VfError, VfFile, VfOffset, VfResult, WriteOpRef,
22 WriteResult,
23};
24
25fn read_result(result: ReadResult) -> FsReadResult {
26 FsReadResult {
27 offset: result.offset,
28 data: result.data,
29 eof: result.eof,
30 }
31}
32
33fn write_result(result: WriteResult) -> FsWriteResult {
34 FsWriteResult {
35 offset: result.offset,
36 written: result.written,
37 stable: result.stable,
38 }
39}
40
41fn io_error(error: VfError) -> io::Error {
42 error.into()
43}
44
45fn poisoned() -> VfError {
46 VfError::client(0, crate::ERR_IO)
47}
48
49fn wrong_result_count(operation: &str, expected: usize, actual: usize) -> VfError {
50 VfError::transport_with_kind(
51 None,
52 crate::TransportKind::InvalidReply,
53 format!("{operation} backend returned {actual} results for {expected} requests"),
54 )
55}
56
57enum CleanupTarget {
58 File(VfFile, PathBuf),
59 Directory(VfDir, PathBuf),
60}
61struct PendingClose<F> {
62 target: CleanupTarget,
63 close: fn(&mut F, &CleanupTarget) -> VfResult<()>,
64}
65
66struct SharedBackend<F: FileSystem> {
69 backend: Mutex<Option<F>>,
70 pending: Mutex<Vec<PendingClose<F>>>,
71 has_pending: AtomicBool,
72}
73impl<F: FileSystem> SharedBackend<F> {
74 fn new(backend: F) -> Self {
75 Self {
76 backend: Mutex::new(Some(backend)),
77 pending: Mutex::new(Vec::new()),
78 has_pending: AtomicBool::new(false),
79 }
80 }
81 fn defer(&self, target: CleanupTarget, close: fn(&mut F, &CleanupTarget) -> VfResult<()>) {
82 self.pending
83 .lock()
84 .unwrap_or_else(|e| e.into_inner())
85 .push(PendingClose { target, close });
86 self.has_pending.store(true, Ordering::Release);
87 }
88 fn lock_without_cleanup(&self) -> Result<BackendGuard<'_, F>, ()> {
89 self.backend
90 .lock()
91 .map(|guard| BackendGuard {
92 guard: Some(guard),
93 pending: &self.pending,
94 has_pending: &self.has_pending,
95 })
96 .map_err(|_| ())
97 }
98 fn lock(&self) -> Result<BackendGuard<'_, F>, ()> {
99 let mut guard = self.lock_without_cleanup()?;
100 let _ = guard.drain_cleanup();
103 Ok(guard)
104 }
105}
106impl<F: FileSystem> Drop for SharedBackend<F> {
107 fn drop(&mut self) {
108 if let Some(backend) = self
109 .backend
110 .get_mut()
111 .unwrap_or_else(|e| e.into_inner())
112 .as_mut()
113 {
114 let pending = self.pending.get_mut().unwrap_or_else(|e| e.into_inner());
115 for close in pending.drain(..) {
116 let _ = (close.close)(backend, &close.target);
117 }
118 let callbacks = backend.take_notifications();
119 for callback in callbacks {
120 let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(callback));
121 }
122 }
123 }
124}
125struct BackendGuard<'a, F: FileSystem> {
126 guard: Option<std::sync::MutexGuard<'a, Option<F>>>,
127 pending: &'a Mutex<Vec<PendingClose<F>>>,
128 has_pending: &'a AtomicBool,
129}
130impl<F: FileSystem> std::ops::Deref for BackendGuard<'_, F> {
131 type Target = F;
132 fn deref(&self) -> &F {
133 self.guard
134 .as_ref()
135 .expect("live guard")
136 .as_ref()
137 .expect("live backend")
138 }
139}
140impl<F: FileSystem> std::ops::DerefMut for BackendGuard<'_, F> {
141 fn deref_mut(&mut self) -> &mut F {
142 self.guard
143 .as_mut()
144 .expect("live guard")
145 .as_mut()
146 .expect("live backend")
147 }
148}
149impl<F: FileSystem> BackendGuard<'_, F> {
150 fn drain_cleanup(&mut self) -> VfResult<()> {
151 if !self.has_pending.swap(false, Ordering::AcqRel) {
152 return Ok(());
153 }
154 let pending = std::mem::take(&mut *self.pending.lock().unwrap_or_else(|e| e.into_inner()));
155 let mut failed = Vec::new();
156 let mut first = None;
157 for close in pending {
158 if let Err(error) = (close.close)(self, &close.target) {
159 first.get_or_insert(error);
160 failed.push(close);
161 }
162 }
163 if !failed.is_empty() {
164 self.pending
165 .lock()
166 .unwrap_or_else(|e| e.into_inner())
167 .extend(failed);
168 self.has_pending.store(true, Ordering::Release);
169 }
170 first.map_or(Ok(()), Err)
171 }
172}
173impl<F: FileSystem> Drop for BackendGuard<'_, F> {
174 fn drop(&mut self) {
175 let callbacks = self.take_notifications();
176 drop(self.guard.take());
177 for callback in callbacks {
178 let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(callback));
181 }
182 }
183}
184
185pub struct FsClient<F: FileSystem> {
192 inner: Arc<SharedBackend<F>>,
193 limits: ResourceLimits,
194}
195
196impl<F: FileSystem> Clone for FsClient<F> {
197 fn clone(&self) -> Self {
198 Self {
199 inner: Arc::clone(&self.inner),
200 limits: self.limits,
201 }
202 }
203}
204
205impl<F: FileSystem> fmt::Debug for FsClient<F> {
206 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
207 formatter.debug_struct("FsClient").finish_non_exhaustive()
208 }
209}
210
211impl<F: FileSystem> FsClient<F> {
212 pub fn new(filesystem: F) -> Self {
213 Self {
214 inner: Arc::new(SharedBackend::new(filesystem)),
215 limits: ResourceLimits::default(),
216 }
217 }
218
219 pub fn with_limits(mut self, limits: ResourceLimits) -> Self {
222 self.limits = limits;
223 self
224 }
225
226 pub fn limits(&self) -> ResourceLimits {
227 self.limits
228 }
229
230 fn lock(&self) -> VfResult<BackendGuard<'_, F>> {
231 self.inner.lock().map_err(|_| poisoned())
232 }
233
234 pub fn into_inner(self) -> Result<F, Self> {
235 if self.drain_cleanup().is_err() {
236 return Err(self);
237 }
238 match Arc::try_unwrap(self.inner) {
239 Ok(mut shared) => Ok(shared
240 .backend
241 .get_mut()
242 .unwrap_or_else(|e| e.into_inner())
243 .take()
244 .expect("owned backend")),
245 Err(inner) => Err(Self {
246 inner,
247 limits: self.limits,
248 }),
249 }
250 }
251
252 pub fn drain_cleanup(&self) -> VfResult<()> {
255 self.inner
256 .lock_without_cleanup()
257 .map_err(|_| poisoned())?
258 .drain_cleanup()
259 }
260}
261
262impl<F: FileSystem> FsClient<F> {
263 pub fn capabilities(&self) -> VfResult<Capabilities> {
264 Ok(self.lock()?.capabilities())
265 }
266 pub fn open(&self, path: impl AsRef<Path>) -> VfResult<FsFile<F>> {
268 self.open_with(OpenRequest::new(path.as_ref(), OpenFlags::READ))
269 }
270
271 pub fn open_with(&self, request: OpenRequest) -> VfResult<FsFile<F>> {
272 let file = self.lock()?.open_impl(&request)?;
273 Ok(FsFile {
274 inner: Arc::clone(&self.inner),
275 file: Some(file),
276 path: request.path,
277 })
278 }
279
280 pub fn open_options(&self) -> OpenOptions<'_, F> {
281 OpenOptions::new(self)
282 }
283
284 pub fn create(&self, path: impl AsRef<Path>) -> VfResult<FsFile<F>> {
285 self.open_with(OpenRequest::new(
286 path.as_ref(),
287 OpenFlags::WRITE | OpenFlags::CREATE | OpenFlags::TRUNCATE,
288 ))
289 }
290
291 pub fn read(&self, path: impl AsRef<Path>) -> VfResult<Vec<u8>> {
292 self.read_with_limit(path, self.limits.max_read_bytes)
293 }
294
295 pub fn read_with_limit(&self, path: impl AsRef<Path>, max_bytes: usize) -> VfResult<Vec<u8>> {
300 let path = path.as_ref();
301 let file = self.open(path)?;
302 let operation = self
303 .lock()?
304 .read_file_impl(file.raw()?, max_bytes)
305 .map_err(|error| error.with_context("read", path))
306 .and_then(|data| {
307 if data.len() > max_bytes {
308 Err(VfError::client(0, libc::EFBIG as u32).with_context("read", path))
309 } else {
310 Ok(data)
311 }
312 });
313 let cleanup = file.close();
316 operation.and_then(|data| cleanup.map(|()| data))
317 }
318
319 pub fn read_to_string(&self, path: impl AsRef<Path>) -> VfResult<String> {
320 self.read_to_string_with_limit(path, self.limits.max_read_bytes)
321 }
322
323 pub fn read_to_string_with_limit(
325 &self,
326 path: impl AsRef<Path>,
327 max_bytes: usize,
328 ) -> VfResult<String> {
329 let path = path.as_ref();
330 String::from_utf8(self.read_with_limit(path, max_bytes)?)
331 .map_err(|_| VfError::client(0, crate::ERR_INVAL).with_context("read_to_string", path))
332 }
333
334 pub fn write(&self, path: impl AsRef<Path>, data: &[u8]) -> VfResult<()> {
335 let path = path.as_ref();
336 let mut file = self.create(path)?;
337 let mut written = 0;
338 while written < data.len() {
339 let count = file.write_native(&data[written..])?;
340 if count == 0 {
341 return Err(VfError::client(0, crate::ERR_IO).with_context("write", path));
342 }
343 written += count;
344 }
345 file.close()
346 }
347}
348
349impl<F: FileSystem> FsClient<F> {
350 pub fn read_stream(
358 &self,
359 path: impl AsRef<Path>,
360 callback: impl FnMut(u64, &[u8]) -> VfResult<bool>,
361 ) -> VfResult<StreamCompletion> {
362 self.read_stream_with_options(
363 path,
364 ReadStreamOptions::new().chunk_size(self.limits.stream_chunk_bytes),
365 callback,
366 )
367 }
368
369 pub fn read_stream_with_options(
371 &self,
372 path: impl AsRef<Path>,
373 options: ReadStreamOptions,
374 mut callback: impl FnMut(u64, &[u8]) -> VfResult<bool>,
375 ) -> VfResult<StreamCompletion> {
376 let chunk_size = options.chunk_size_bytes();
377 if chunk_size == 0 {
378 return Err(VfError::failure(0, crate::ERR_INVAL));
379 }
380
381 let file = self.open(path)?;
382 let raw_file = file.raw()?.clone();
383 let operation = (|| -> VfResult<StreamCompletion> {
384 let mut offset = 0u64;
385 loop {
386 let request = ReadOp::at(raw_file.clone(), offset, chunk_size);
387 let result = {
388 let mut backend = self.lock()?;
389 let result = backend.read_impl(&request)?;
390 validate_read_results(
391 "read_stream",
392 std::slice::from_ref(&request),
393 std::slice::from_ref(&result),
394 )?;
395 result
396 };
397 let length = result.data.len();
398 if !result.data.is_empty() && !callback(offset, &result.data)? {
399 return Ok(StreamCompletion::Stopped {
400 next_offset: offset
401 .checked_add(length as u64)
402 .ok_or_else(|| VfError::client(0, crate::ERR_INVAL))?,
403 });
404 }
405 if result.eof {
406 return Ok(StreamCompletion::Complete);
407 }
408 offset = offset
409 .checked_add(length as u64)
410 .ok_or_else(|| VfError::client(0, libc::EOVERFLOW as u32))?;
411 }
412 })();
413 let close = file.close();
414 match operation {
415 Err(error) => Err(error),
416 Ok(completion) => close.map(|()| completion),
417 }
418 }
419}
420
421impl<F: Backend> FsClient<F> {
422 pub fn metadata(&self, path: impl AsRef<Path>) -> VfResult<Metadata> {
423 self.lock()?.metadata_path_impl(path.as_ref(), true)
424 }
425
426 pub fn symlink_metadata(&self, path: impl AsRef<Path>) -> VfResult<Metadata> {
427 self.lock()?.metadata_path_impl(path.as_ref(), false)
428 }
429
430 pub fn set_metadata(&self, path: impl AsRef<Path>) -> SetMetadata<'_, F> {
431 SetMetadata {
432 client: self,
433 path: path.as_ref().to_path_buf(),
434 update: MetadataUpdate::new(),
435 follow: true,
436 }
437 }
438}
439
440impl<F: Backend> FsClient<F> {
441 pub fn create_dir(&self, path: impl AsRef<Path>) -> VfResult<()> {
442 self.create_dir_with_mode(path, 0o777)
443 }
444
445 pub fn create_dir_with_mode(&self, path: impl AsRef<Path>, mode: u32) -> VfResult<()> {
446 self.lock()?.create_dir_impl(path.as_ref(), mode)
447 }
448
449 pub fn read_dir(&self, path: impl AsRef<Path>) -> VfResult<Vec<DirEntry>> {
450 self.read_dir_with_options(path, self.limits.directory_options())
451 }
452
453 pub fn read_dir_with_options(
455 &self,
456 path: impl AsRef<Path>,
457 options: ReadDirOptions,
458 ) -> VfResult<Vec<DirEntry>> {
459 self.lock()?.read_dir_impl(path.as_ref(), options)
460 }
461
462 pub fn visit_dir(
466 &self,
467 path: impl AsRef<Path>,
468 callback: impl FnMut(DirEntry) -> VfResult<std::ops::ControlFlow<()>>,
469 ) -> VfResult<TraversalCompletion> {
470 self.visit_dir_with_options(path, self.limits.directory_options(), callback)
471 }
472
473 pub fn visit_dir_with_options(
475 &self,
476 path: impl AsRef<Path>,
477 options: ReadDirOptions,
478 callback: impl FnMut(DirEntry) -> VfResult<std::ops::ControlFlow<()>>,
479 ) -> VfResult<TraversalCompletion> {
480 self.visit_dir_with_fields(path, crate::native::metadata_mask(), options, callback)
481 }
482
483 pub fn visit_dir_with_fields(
484 &self,
485 path: impl AsRef<Path>,
486 fields: AttrMask,
487 options: ReadDirOptions,
488 mut callback: impl FnMut(DirEntry) -> VfResult<std::ops::ControlFlow<()>>,
489 ) -> VfResult<TraversalCompletion> {
490 const PAGE_SIZE: usize = 1024;
491 let path = path.as_ref();
492 let mut cursor = None;
493 let mut count = 0usize;
494 let mut path_bytes = 0usize;
495 let max_entries = if options.entry_limit() == usize::MAX {
496 0 } else {
498 options.entry_limit().saturating_add(1)
499 };
500 loop {
501 let page_size = if cursor.is_none() {
504 1
505 } else {
506 PAGE_SIZE.min(
507 options
508 .entry_limit()
509 .saturating_sub(count)
510 .saturating_add(1),
511 )
512 };
513 let (entries, next) = {
514 self.lock()?.read_dir_page_with_fields_impl(
515 path,
516 fields,
517 cursor,
518 page_size,
519 max_entries,
520 )?
521 };
522 if entries.is_empty() && next.is_some() {
523 return Err(VfError::transport(None, "directory page made no progress"));
524 }
525 for entry in entries {
526 if count >= options.entry_limit() {
527 return Err(
528 VfError::failure(count, libc::EFBIG as u32).with_context("visit_dir", path)
529 );
530 }
531 path_bytes = path_bytes
532 .checked_add(entry.path().as_os_str().len())
533 .ok_or_else(|| {
534 VfError::failure(count, libc::EFBIG as u32).with_context("visit_dir", path)
535 })?;
536 if path_bytes > options.path_byte_limit() {
537 return Err(
538 VfError::failure(count, libc::EFBIG as u32).with_context("visit_dir", path)
539 );
540 }
541 count += 1;
542 if callback(entry)?.is_break() {
543 return Ok(TraversalCompletion::Stopped);
544 }
545 }
546 match next {
547 Some(value) => cursor = Some(value),
548 None => return Ok(TraversalCompletion::Complete),
549 }
550 }
551 }
552}
553
554impl<F: Backend> FsClient<F> {
555 pub fn create_dir_all(&self, path: impl AsRef<Path>) -> VfResult<()> {
556 let path = path.as_ref();
557 let mut current = PathBuf::new();
558 for component in path.components() {
559 match component {
560 Component::RootDir => {
561 current.push(Path::new("/"));
562 continue;
563 }
564 Component::Normal(part) => current.push(part),
565 Component::CurDir => continue,
566 Component::ParentDir => {
567 current.push("..");
572 continue;
573 }
574 Component::Prefix(_) => return Err(VfError::client(0, crate::ERR_INVAL)),
575 }
576 match self.create_dir(¤t) {
577 Ok(()) => {}
578 Err(error) if error.err_no() == crate::ERR_EXIST => {
579 if !self.metadata(¤t)?.is_dir() {
580 return Err(VfError::client(0, crate::ERR_NOTDIR)
581 .with_context("create_dir_all", ¤t));
582 }
583 }
584 Err(error) => return Err(error),
585 }
586 }
587 Ok(())
588 }
589}
590
591impl<F: Backend> FsClient<F> {
592 fn removal_metadata(&self, path: &Path, operation: &'static str) -> VfResult<Metadata> {
593 let mut filesystem = self.lock()?;
594 let follow = !filesystem.capabilities().contains(Capabilities::LSTAT);
595 filesystem
596 .metadata_path_impl(path, follow)
597 .map_err(|error| error.with_context(operation, path))
598 }
599
600 pub fn remove_file(&self, path: impl AsRef<Path>) -> VfResult<()> {
601 let path = path.as_ref();
602 if self.removal_metadata(path, "remove_file")?.is_dir() {
603 return Err(VfError::client(0, crate::ERR_ISDIR).with_context("remove_file", path));
604 }
605 self.lock()?.remove_impl(path, false)
606 }
607
608 pub fn remove_dir(&self, path: impl AsRef<Path>) -> VfResult<()> {
609 let path = path.as_ref();
610 if !self.removal_metadata(path, "remove_dir")?.is_dir() {
611 return Err(VfError::client(0, crate::ERR_NOTDIR).with_context("remove_dir", path));
612 }
613 self.lock()?.remove_impl(path, false)
614 }
615
616 pub fn remove_dir_all(&self, path: impl AsRef<Path>) -> VfResult<()> {
617 let path = path.as_ref();
618 if !self.removal_metadata(path, "remove_dir_all")?.is_dir() {
619 return Err(VfError::client(0, crate::ERR_NOTDIR).with_context("remove_dir_all", path));
620 }
621 self.lock()?.remove_impl(path, true)
622 }
623
624 pub fn remove_dir_contents_impl(&self, path: impl AsRef<Path>) -> VfResult<()> {
626 let path = path.as_ref();
627 if !self
628 .removal_metadata(path, "remove_dir_contents_impl")?
629 .is_dir()
630 {
631 return Err(
632 VfError::client(0, crate::ERR_NOTDIR).with_context("remove_dir_contents", path)
633 );
634 }
635 self.lock()?.remove_dir_contents_impl(path)
636 }
637
638 pub fn rename(&self, from: impl AsRef<Path>, to: impl AsRef<Path>) -> VfResult<()> {
639 self.lock()?.rename_impl(from.as_ref(), to.as_ref())
640 }
641}
642
643impl<F: Backend> FsClient<F> {
644 pub fn vrename<P: AsRef<Path>, Q: AsRef<Path>>(&self, pairs: &[(P, Q)]) -> VfResult<()> {
646 if pairs.is_empty() {
647 return Ok(());
648 }
649 let requests: Vec<_> = pairs
650 .iter()
651 .map(|(from, to)| {
652 (
653 VfFile::from_os_path(from.as_ref()),
654 VfFile::from_os_path(to.as_ref()),
655 )
656 })
657 .collect();
658 self.lock()?
659 .vrename_impl(&requests)
660 .map_err(|error| match error.index() {
661 Some(index) if index < pairs.len() => {
662 error.with_context("vrename", pairs[index].0.as_ref())
663 }
664 Some(_) => {
665 VfError::transport(None, "rename backend returned an invalid error index")
666 }
667 None => error,
668 })
669 }
670
671 pub fn vrename_with_options<P: AsRef<Path>, Q: AsRef<Path>>(
673 &self,
674 pairs: &[(P, Q)],
675 options: vfsi_core::api::RenameOptions,
676 ) -> VfResult<()> {
677 if pairs.is_empty() {
678 return Ok(());
679 }
680 let requests: Vec<_> = pairs
681 .iter()
682 .map(|(from, to)| {
683 (
684 VfFile::from_os_path(from.as_ref()),
685 VfFile::from_os_path(to.as_ref()),
686 )
687 })
688 .collect();
689 self.lock()?
690 .vrename_with_options_impl(&requests, options)
691 .map_err(|error| match error.index() {
692 Some(index) if index < pairs.len() => {
693 error.with_context("vrename_with_options", pairs[index].0.as_ref())
694 }
695 Some(_) => {
696 VfError::transport(None, "rename backend returned an invalid error index")
697 }
698 None => error,
699 })
700 }
701
702 pub fn vmkdir_default<P: AsRef<Path>>(&self, paths: &[P]) -> VfResult<()> {
705 let directories: Vec<_> = paths.iter().map(|path| (path.as_ref(), 0o777)).collect();
706 self.vmkdir(&directories)
707 }
708
709 pub fn vmkdir<P: AsRef<Path>>(&self, directories: &[(P, u32)]) -> VfResult<()> {
711 let mut seen = HashSet::with_capacity(directories.len());
712 for (index, (path, _)) in directories.iter().enumerate() {
713 if !seen.insert(path.as_ref()) {
714 return Err(VfError::client(index, crate::ERR_INVAL)
715 .with_context("vmkdir_default", path.as_ref()));
716 }
717 }
718 if directories.is_empty() {
719 return Ok(());
720 }
721 let dirs: Vec<_> = directories
722 .iter()
723 .map(|(path, mode)| crate::VfAttrs {
724 file: VfFile::from_os_path(path.as_ref()),
725 masks: AttrMask::MODE,
726 mode: *mode,
727 ..Default::default()
728 })
729 .collect();
730 self.lock()?
731 .vmkdir_impl(&dirs)
732 .map_err(|error| match error.index() {
733 Some(index) if index < directories.len() => {
734 error.with_context("vmkdir_default", directories[index].0.as_ref())
735 }
736 Some(_) => {
737 VfError::transport(None, "mkdir backend returned an invalid error index")
738 }
739 None => error,
740 })
741 }
742
743 pub fn vsymlink<P: AsRef<Path>, Q: AsRef<Path>>(&self, pairs: &[(P, Q)]) -> VfResult<()> {
745 if pairs.is_empty() {
746 return Ok(());
747 }
748 let targets: Vec<_> = pairs.iter().map(|(target, _)| target.as_ref()).collect();
749 let links: Vec<_> = pairs.iter().map(|(_, link)| link.as_ref()).collect();
750 self.lock()?
751 .vsymlink_impl(&targets, &links)
752 .map_err(|error| link_error(error, &links, "vsymlink"))
753 }
754
755 pub fn vreadlink<P: AsRef<Path>>(&self, paths: &[P]) -> VfResult<Vec<PathBuf>> {
757 if paths.is_empty() {
758 return Ok(Vec::new());
759 }
760 let paths: Vec<_> = paths.iter().map(AsRef::as_ref).collect();
761 let targets = self
762 .lock()?
763 .vreadlink_impl(&paths)
764 .map_err(|error| link_error(error, &paths, "vreadlink"))?;
765 if targets.len() != paths.len() {
766 return Err(VfError::transport(
767 None,
768 "readlink backend returned an invalid result count",
769 ));
770 }
771 Ok(targets
772 .into_iter()
773 .map(crate::native::bytes_to_path)
774 .collect())
775 }
776
777 pub fn vhardlink<P: AsRef<Path>, Q: AsRef<Path>>(&self, pairs: &[(P, Q)]) -> VfResult<()> {
779 if pairs.is_empty() {
780 return Ok(());
781 }
782 let sources: Vec<_> = pairs.iter().map(|(source, _)| source.as_ref()).collect();
783 let links: Vec<_> = pairs.iter().map(|(_, link)| link.as_ref()).collect();
784 self.lock()?
785 .vhardlink_impl(&sources, &links)
786 .map_err(|error| link_error(error, &links, "vhardlink"))
787 }
788
789 pub fn walk_events_with_options(
794 &self,
795 root: impl AsRef<Path>,
796 fields: AttrMask,
797 options: crate::WalkOptions,
798 sort_by_name: bool,
799 callback: impl FnMut(&crate::WalkEvent) -> VfResult<crate::WalkControl>,
800 ) -> VfResult<TraversalCompletion> {
801 let root = root.as_ref();
802 let metadata = self.symlink_metadata_with_fields(root, fields | AttrMask::MODE)?;
803 crate::walk_events(
804 DirEntry::new(root.to_path_buf(), metadata),
805 options,
806 sort_by_name,
807 |path, limits| {
808 let mut listings =
809 self.read_dirs_with_options(&[path], fields | AttrMask::MODE, limits)?;
810 if listings.len() != 1 {
811 return Err(VfError::transport(None, "invalid directory result count"));
812 }
813 Ok(listings.remove(0).entries)
814 },
815 callback,
816 )
817 }
818 pub fn visit_walk(
827 &self,
828 root: impl AsRef<Path>,
829 callback: impl FnMut(&DirEntry) -> VfResult<std::ops::ControlFlow<()>>,
830 ) -> VfResult<TraversalCompletion> {
831 self.visit_walk_with_options(root, self.limits.walk_options(), callback)
832 }
833
834 pub fn visit_walk_with_options(
835 &self,
836 root: impl AsRef<Path>,
837 options: crate::WalkOptions,
838 callback: impl FnMut(&DirEntry) -> VfResult<std::ops::ControlFlow<()>>,
839 ) -> VfResult<TraversalCompletion> {
840 self.visit_walk_with_fields(root, crate::native::metadata_mask(), options, callback)
841 }
842
843 pub fn visit_walk_with_fields(
844 &self,
845 root: impl AsRef<Path>,
846 fields: AttrMask,
847 options: crate::WalkOptions,
848 callback: impl FnMut(&DirEntry) -> VfResult<std::ops::ControlFlow<()>>,
849 ) -> VfResult<TraversalCompletion> {
850 let root = root.as_ref();
851 if !self.symlink_metadata(root)?.is_dir() {
852 return Err(VfError::client(0, crate::ERR_NOTDIR).with_context("visit_walk", root));
853 }
854 visit_walk_pages(
855 root,
856 options,
857 |path, cursor, page_size, max_entries| {
858 self.lock()?.read_dir_page_with_fields_impl(
859 path,
860 fields,
861 cursor,
862 page_size,
863 max_entries,
864 )
865 },
866 callback,
867 )
868 }
869 pub fn directory_page_batch_size(&self) -> VfResult<usize> {
871 Ok(self.lock()?.directory_page_batch_size().clamp(1, 32))
872 }
873 pub fn read_dir_pages_with_fields(
875 &self,
876 paths: &[&Path],
877 fields: AttrMask,
878 cursors: Vec<Option<crate::DirPageCursor>>,
879 page_size: usize,
880 max_entries: usize,
881 ) -> VfResult<Vec<crate::DirectoryPage>> {
882 let pages = self.lock()?.vlistdir_pages_impl(
883 paths,
884 fields | AttrMask::MODE | AttrMask::SIZE,
885 cursors,
886 page_size,
887 max_entries,
888 )?;
889 if pages.len() != paths.len() {
890 return Err(VfError::transport(
891 None,
892 "directory pages returned an invalid result count",
893 ));
894 }
895 pages
896 .into_iter()
897 .zip(paths)
898 .enumerate()
899 .map(|(index, ((attrs, next, children), path))| {
900 if children.len() != attrs.len()
901 || attrs.len() > page_size
902 || (attrs.is_empty() && next.is_some())
903 {
904 return Err(VfError::transport(
905 Some(index),
906 "invalid directory page progress",
907 ));
908 }
909 let mut seeds = Vec::new();
910 let entries = attrs
911 .into_iter()
912 .zip(children)
913 .map(|(attrs, child)| {
914 let entry_path = attrs
915 .file
916 .path()
917 .ok_or_else(|| {
918 VfError::transport(Some(index), "directory entry has no path")
919 })?
920 .to_path_buf();
921 if entry_path.parent() != Some(*path) {
922 return Err(VfError::transport(
923 Some(index),
924 "directory entry is outside its parent",
925 ));
926 }
927 if let Some(child) = child {
928 seeds.push((entry_path.clone(), child));
929 }
930 Ok(DirEntry::new(
931 entry_path,
932 vfsi_core::metadata_from_attrs(attrs),
933 ))
934 })
935 .collect::<VfResult<Vec<_>>>()?;
936 Ok((
937 DirectoryListing {
938 path: path.to_path_buf(),
939 entries,
940 },
941 next,
942 seeds,
943 ))
944 })
945 .collect()
946 }
947
948 pub fn read_dirs<P: AsRef<Path>>(&self, paths: &[P]) -> VfResult<Vec<DirectoryListing>> {
951 self.read_dirs_with_options(paths, AttrMask::stat(), self.limits.directory_options())
952 }
953}
954
955type DirectoryPage = (Vec<DirEntry>, Option<crate::DirPageCursor>);
956
957fn visit_walk_pages(
958 root: &Path,
959 options: crate::WalkOptions,
960 mut read_page: impl FnMut(
961 &Path,
962 Option<crate::DirPageCursor>,
963 usize,
964 usize,
965 ) -> VfResult<DirectoryPage>,
966 mut callback: impl FnMut(&DirEntry) -> VfResult<std::ops::ControlFlow<()>>,
967) -> VfResult<TraversalCompletion> {
968 let mut pending = vec![(root.to_path_buf(), 0usize)];
969 let mut count = 0usize;
970 let mut path_bytes = root.as_os_str().len();
971 if path_bytes > options.path_byte_limit() {
972 return Err(VfError::client(0, libc::EFBIG as u32));
973 }
974 while let Some((path, depth)) = pending.pop() {
975 let mut cursor = None;
976 let mut children = Vec::new();
977 loop {
978 let requested = options
979 .entry_limit()
980 .saturating_sub(count)
981 .saturating_add(1);
982 let (entries, next) = read_page(&path, cursor, requested.min(128), requested)?;
983 if entries.is_empty() && next.is_some() {
984 return Err(VfError::transport(None, "directory page made no progress")
985 .with_context("visit_walk", &path));
986 }
987 for entry in entries {
988 count = count
989 .checked_add(1)
990 .ok_or_else(|| VfError::client(0, libc::EFBIG as u32))?;
991 path_bytes = path_bytes
992 .checked_add(entry.path().as_os_str().len())
993 .ok_or_else(|| VfError::client(0, libc::EFBIG as u32))?;
994 if count > options.entry_limit() || path_bytes > options.path_byte_limit() {
995 return Err(
996 VfError::client(0, libc::EFBIG as u32).with_context("visit_walk", &path)
997 );
998 }
999 if callback(&entry)?.is_break() {
1000 return Ok(TraversalCompletion::Stopped);
1001 }
1002 if entry.metadata().is_dir() {
1003 if depth >= options.depth_limit() {
1004 if !options.truncates_at_depth_limit() {
1005 return Err(VfError::client(0, libc::EFBIG as u32)
1006 .with_context("visit_walk", entry.path()));
1007 }
1008 } else {
1009 children.push((entry.path().to_path_buf(), depth + 1));
1010 }
1011 }
1012 }
1013 cursor = next;
1014 if cursor.is_none() {
1015 break;
1016 }
1017 }
1018 pending.extend(children.into_iter().rev());
1019 }
1020 Ok(TraversalCompletion::Complete)
1021}
1022
1023impl<F: Backend> FsClient<F> {
1024 pub fn walk(&self, root: impl AsRef<Path>) -> VfResult<Vec<DirectoryListing>> {
1027 self.walk_with_options(root, AttrMask::stat(), self.limits.walk_options())
1028 }
1029
1030 pub fn ensure_empty_dir(&self, path: impl AsRef<Path>) -> VfResult<()> {
1033 self.lock()?.ensure_empty_dir_impl(path.as_ref())
1034 }
1035
1036 pub fn remove_dir_all_with_options(
1038 &self,
1039 path: impl AsRef<Path>,
1040 options: RemoveOptions,
1041 ) -> VfResult<()> {
1042 let path = path.as_ref();
1043 if !self.removal_metadata(path, "remove_dir_all")?.is_dir() {
1044 return Err(VfError::client(0, crate::ERR_NOTDIR).with_context("remove_dir_all", path));
1045 }
1046 self.lock()?
1047 .remove_paths_with_options_impl(&[path], true, options)
1048 .map_err(|error| error.with_context("remove_dir_all", path))
1049 }
1050
1051 pub fn remove_dir_contents_with_options(
1053 &self,
1054 path: impl AsRef<Path>,
1055 options: RemoveOptions,
1056 ) -> VfResult<()> {
1057 let path = path.as_ref();
1058 if !self
1059 .removal_metadata(path, "remove_dir_contents_impl")?
1060 .is_dir()
1061 {
1062 return Err(
1063 VfError::client(0, crate::ERR_NOTDIR).with_context("remove_dir_contents", path)
1064 );
1065 }
1066 self.lock()?
1067 .remove_dir_contents_path_with_options_impl(path, options)
1068 .map_err(|error| error.with_context("remove_dir_contents", path))
1069 }
1070
1071 pub fn open_dir_handle(&self, path: impl AsRef<Path>) -> VfResult<FsDir<F>> {
1075 let path = path.as_ref();
1076 let mut backend = self.lock()?;
1077 let dir = backend.open_dir_impl(path)?;
1078 if !matches!(dir, VfDir::Descriptor { .. }) {
1079 let _ = backend.close_dir_impl(&dir);
1080 return Err(VfError::unsupported(0).with_context("open_dir_handle", path));
1081 }
1082 Ok(FsDir {
1083 inner: Arc::clone(&self.inner),
1084 dir: Some(dir),
1085 path: path.to_path_buf(),
1086 })
1087 }
1088}
1089
1090pub struct FsDir<F: Backend> {
1093 inner: Arc<SharedBackend<F>>,
1094 dir: Option<VfDir>,
1095 path: PathBuf,
1096}
1097
1098impl<F: Backend> fmt::Debug for FsDir<F> {
1099 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
1100 formatter
1101 .debug_struct("FsDir")
1102 .field("path", &self.path)
1103 .finish_non_exhaustive()
1104 }
1105}
1106
1107impl<F: Backend> FsDir<F> {
1108 pub fn is_closed(&self) -> bool {
1109 self.dir.is_none()
1110 }
1111 pub fn path(&self) -> &Path {
1113 &self.path
1114 }
1115
1116 pub fn remove_contents(&self) -> VfResult<()> {
1117 self.remove_contents_with_options(RemoveOptions::default())
1118 }
1119
1120 pub fn remove_contents_with_options(&self, options: RemoveOptions) -> VfResult<()> {
1121 let dir = self
1122 .dir
1123 .as_ref()
1124 .ok_or_else(|| VfError::client(0, crate::ERR_EBADF))?;
1125 self.inner
1126 .lock()
1127 .map_err(|_| poisoned())?
1128 .remove_dir_contents_handle_with_options_impl(dir, options)
1129 .map_err(|error| error.with_context("remove_dir_contents", &self.path))
1130 }
1131
1132 pub fn close(mut self) -> VfResult<()> {
1133 self.try_close()
1134 }
1135
1136 pub fn try_close(&mut self) -> VfResult<()> {
1138 let Some(dir) = self.dir.as_ref() else {
1139 return Ok(());
1140 };
1141 self.inner
1142 .lock()
1143 .map_err(|_| poisoned())?
1144 .close_dir_impl(dir)?;
1145 self.dir = None;
1146 Ok(())
1147 }
1148}
1149
1150impl<F: Backend> Drop for FsDir<F> {
1151 fn drop(&mut self) {
1152 if let Some(dir) = self.dir.take() {
1153 self.inner.defer(
1154 CleanupTarget::Directory(dir, std::mem::take(&mut self.path)),
1155 |backend, target| {
1156 let CleanupTarget::Directory(dir, path) = target else {
1157 unreachable!()
1158 };
1159 backend
1160 .close_dir_impl(dir)
1161 .map_err(|error| error.with_context("close_dir", path))
1162 },
1163 );
1164 }
1165 }
1166}
1167
1168impl<F: Backend> FsClient<F> {
1169 pub fn symlink(&self, target: impl AsRef<Path>, link: impl AsRef<Path>) -> VfResult<()> {
1170 self.lock()?.symlink_impl(target.as_ref(), link.as_ref())
1171 }
1172
1173 pub fn hard_link(&self, source: impl AsRef<Path>, link: impl AsRef<Path>) -> VfResult<()> {
1174 self.lock()?.hard_link_impl(source.as_ref(), link.as_ref())
1175 }
1176
1177 pub fn read_link(&self, path: impl AsRef<Path>) -> VfResult<PathBuf> {
1178 self.lock()?.read_link_impl(path.as_ref())
1179 }
1180}
1181
1182impl<F: Backend> FsClient<F> {
1183 pub fn copy(&self, source: impl AsRef<Path>, destination: impl AsRef<Path>) -> VfResult<()> {
1184 self.lock()?
1185 .copy_impl(source.as_ref(), destination.as_ref())
1186 }
1187}
1188
1189impl<F: Backend> FsClient<F> {
1190 pub fn symlink_metadata_with_fields(
1193 &self,
1194 path: impl AsRef<Path>,
1195 fields: AttrMask,
1196 ) -> VfResult<Metadata> {
1197 let path = path.as_ref();
1198 let mut attrs = crate::VfAttrs {
1199 file: VfFile::from_os_path(path),
1200 masks: fields | AttrMask::MODE | AttrMask::SIZE,
1201 ..crate::VfAttrs::default()
1202 };
1203 self.lock()?
1204 .vgetattrs_nofollow_impl(std::slice::from_mut(&mut attrs))
1205 .map_err(|error| error.with_context("symlink_metadata", path))?;
1206 Ok(vfsi_core::metadata_from_attrs(attrs))
1207 }
1208
1209 pub fn read_dirs_with_options<P: AsRef<Path>>(
1216 &self,
1217 paths: &[P],
1218 fields: AttrMask,
1219 options: ReadDirOptions,
1220 ) -> VfResult<Vec<DirectoryListing>> {
1221 if paths.is_empty() {
1222 return Ok(Vec::new());
1223 }
1224 let paths: Vec<&Path> = paths.iter().map(AsRef::as_ref).collect();
1225 let mut positions: HashMap<PathBuf, Vec<usize>> = HashMap::with_capacity(paths.len());
1226 let mut unique_paths = Vec::with_capacity(paths.len());
1227 for (index, path) in paths.iter().enumerate() {
1228 if let Some(indices) = positions.get_mut(*path) {
1229 indices.push(index);
1230 } else {
1231 positions.insert(path.to_path_buf(), vec![index]);
1232 unique_paths.push(*path);
1233 }
1234 }
1235 let mut listings: Vec<DirectoryListing> = paths
1236 .iter()
1237 .map(|path| DirectoryListing {
1238 path: path.to_path_buf(),
1239 entries: Vec::new(),
1240 })
1241 .collect();
1242 let mut entry_count = 0usize;
1243 let mut path_bytes = 0usize;
1244 const DIRECTORY_COHORT: usize = 32;
1247 for cohort in unique_paths.chunks(DIRECTORY_COHORT) {
1248 let mut callback_error = None;
1249 let requested = options
1250 .entry_limit()
1251 .saturating_sub(entry_count)
1252 .saturating_add(1);
1253 let result = self.lock()?.vlistdirs_impl(
1254 cohort,
1255 fields | AttrMask::MODE | AttrMask::SIZE,
1256 requested,
1257 false,
1258 &mut |attributes, directory| {
1259 let Some(indices) = positions.get(directory) else {
1260 callback_error = Some(VfError::transport(
1261 None,
1262 format!(
1263 "read_dirs backend returned an unexpected directory: {}",
1264 directory.display()
1265 ),
1266 ));
1267 return false;
1268 };
1269 if !cohort.contains(&directory) {
1270 callback_error = Some(VfError::transport(
1271 None,
1272 "read_dirs backend returned a directory outside the active cohort",
1273 ));
1274 return false;
1275 }
1276 let Some(path) = attributes.file.path() else {
1277 callback_error = Some(VfError::transport(
1278 None,
1279 "read_dirs backend returned an entry without a path",
1280 ));
1281 return false;
1282 };
1283 if path.parent() != Some(directory) {
1284 callback_error = Some(VfError::transport(
1285 None,
1286 format!(
1287 "read_dirs backend returned an entry outside {}",
1288 directory.display()
1289 ),
1290 ));
1291 return false;
1292 }
1293 for &index in indices {
1294 entry_count += 1;
1295 path_bytes = path_bytes.saturating_add(path.as_os_str().len());
1296 if entry_count > options.entry_limit()
1297 || path_bytes > options.path_byte_limit()
1298 {
1299 callback_error = Some(
1300 VfError::failure(index, libc::EFBIG as u32)
1301 .with_context("read_dirs", directory),
1302 );
1303 return false;
1304 }
1305 listings[index].entries.push(DirEntry::new(
1306 path.to_path_buf(),
1307 vfsi_core::metadata_from_attrs(attributes.clone()),
1308 ));
1309 }
1310 true
1311 },
1312 );
1313 if let Some(error) = callback_error {
1314 return Err(error);
1315 }
1316 result.map_err(|error| {
1317 error.map_index(|index| {
1318 cohort
1319 .get(index)
1320 .and_then(|path| positions.get(*path))
1321 .and_then(|indices| indices.first())
1322 .copied()
1323 .unwrap_or(index)
1324 })
1325 })?;
1326 }
1327 Ok(listings)
1328 }
1329
1330 pub fn walk_with_options(
1334 &self,
1335 root: impl AsRef<Path>,
1336 fields: AttrMask,
1337 options: crate::WalkOptions,
1338 ) -> VfResult<Vec<DirectoryListing>> {
1339 let root = root.as_ref();
1340 let tree = self.lock()?.walk_with_options_impl(
1341 root,
1342 fields | AttrMask::MODE | AttrMask::SIZE,
1343 options,
1344 &mut |_, _| {},
1345 )?;
1346 tree.into_iter()
1347 .map(|directory| {
1348 let entries = directory
1349 .entries
1350 .into_iter()
1351 .enumerate()
1352 .map(|(index, attributes)| {
1353 let path = attributes
1354 .file
1355 .path()
1356 .ok_or_else(|| {
1357 VfError::transport(
1358 None,
1359 format!("walk backend returned entry {index} without a path"),
1360 )
1361 })?
1362 .to_path_buf();
1363 Ok(DirEntry::new(
1364 path,
1365 vfsi_core::metadata_from_attrs(attributes),
1366 ))
1367 })
1368 .collect::<VfResult<Vec<_>>>()?;
1369 Ok(DirectoryListing {
1370 path: directory.path,
1371 entries,
1372 })
1373 })
1374 .collect()
1375 }
1376
1377 pub fn vcopy<P: AsRef<Path>, Q: AsRef<Path>>(&self, pairs: &[(P, Q)]) -> VfResult<()> {
1380 let extents: Vec<_> = pairs
1381 .iter()
1382 .map(|(from, to)| {
1383 crate::ExtentPair::from_os_paths(from.as_ref(), 0, to.as_ref(), 0, None)
1384 })
1385 .collect();
1386 self.lock()?
1387 .vcopy_impl(&extents, vfsi_core::CopyOption::new())
1388 .map_err(|error| {
1389 error
1390 .index()
1391 .and_then(|index| pairs.get(index))
1392 .map_or(error.clone(), |(_, to)| {
1393 error.with_context("vcopy", to.as_ref())
1394 })
1395 })
1396 }
1397
1398 pub fn vremove_native<P: AsRef<Path>>(&self, paths: &[P], recursive: bool) -> VfResult<()> {
1401 self.vremove_with_options_native(paths, recursive, RemoveOptions::default())
1402 }
1403
1404 pub fn vremove_with_options_native<P: AsRef<Path>>(
1406 &self,
1407 paths: &[P],
1408 recursive: bool,
1409 options: RemoveOptions,
1410 ) -> VfResult<()> {
1411 let paths: Vec<&Path> = paths.iter().map(AsRef::as_ref).collect();
1412 self.lock()?
1413 .remove_paths_with_options_impl(&paths, recursive, options)
1414 .map_err(|error| {
1415 error
1416 .index()
1417 .and_then(|index| paths.get(index))
1418 .map_or(error.clone(), |path| {
1419 error.with_context("vremove_native", path)
1420 })
1421 })
1422 }
1423
1424 pub fn vgetattrs_native<P: AsRef<Path>>(
1427 &self,
1428 paths: &[P],
1429 fields: AttrMask,
1430 follow: bool,
1431 ) -> VfResult<Vec<Metadata>> {
1432 if paths.is_empty() {
1433 return Ok(Vec::new());
1434 }
1435 let mut attrs: Vec<_> = paths
1436 .iter()
1437 .map(|path| crate::VfAttrs {
1438 file: VfFile::from_os_path(path.as_ref()),
1439 masks: fields | AttrMask::MODE,
1440 ..crate::VfAttrs::default()
1441 })
1442 .collect();
1443 let result = {
1444 let mut backend = self.lock()?;
1445 if follow {
1446 backend.vgetattrs_impl(&mut attrs)
1447 } else {
1448 backend.vgetattrs_nofollow_impl(&mut attrs)
1449 }
1450 };
1451 result.map_err(|error| match error.index() {
1452 Some(index) if index < paths.len() => {
1453 error.with_context("metadatav", paths[index].as_ref())
1454 }
1455 Some(_) => VfError::transport(None, "metadata backend returned an invalid error index"),
1456 None => error,
1457 })?;
1458 Ok(attrs
1459 .into_iter()
1460 .map(vfsi_core::metadata_from_attrs)
1461 .collect())
1462 }
1463
1464 pub fn vsymlink_metadata_native(&self, paths: &[&Path]) -> VfResult<Vec<Metadata>> {
1466 let mut attrs: Vec<_> = paths
1467 .iter()
1468 .map(|path| crate::VfAttrs {
1469 file: VfFile::from_os_path(path),
1470 masks: AttrMask::MODE,
1471 ..crate::VfAttrs::default()
1472 })
1473 .collect();
1474 self.lock()?
1475 .vgetattrs_nofollow_impl(&mut attrs)
1476 .map_err(|error| {
1477 error
1478 .index()
1479 .and_then(|index| paths.get(index))
1480 .map_or(error.clone(), |path| {
1481 error.with_context("vsymlink_metadata_native", path)
1482 })
1483 })?;
1484 Ok(attrs
1485 .into_iter()
1486 .map(vfsi_core::metadata_from_attrs)
1487 .collect())
1488 }
1489}
1490
1491impl<F: Backend> FsClient<F> {
1492 pub fn vopen(&self, requests: &[OpenRequest]) -> VfResult<Vec<FsFile<F>>> {
1498 let mut filesystem = self.lock()?;
1499 let files = filesystem.vopen_impl(requests).map_err(|error| {
1500 error
1501 .index()
1502 .and_then(|index| requests.get(index))
1503 .map_or(error.clone(), |request| {
1504 error.with_context("vopen", &request.path)
1505 })
1506 })?;
1507 if files.len() != requests.len() {
1508 let error = wrong_result_count("vopen", requests.len(), files.len());
1509 for file in &files {
1510 let _ = filesystem.close_impl(file);
1511 }
1512 return Err(error);
1513 }
1514 drop(filesystem);
1515 Ok(files
1516 .into_iter()
1517 .zip(requests)
1518 .map(|(file, request)| FsFile {
1519 inner: Arc::clone(&self.inner),
1520 file: Some(file),
1521 path: request.path.clone(),
1522 })
1523 .collect())
1524 }
1525
1526 pub fn vclose<'a>(&self, files: impl IntoIterator<Item = &'a mut FsFile<F>>) -> VfResult<()>
1530 where
1531 F: 'a,
1532 {
1533 let mut files: Vec<_> = files.into_iter().collect();
1534 for (index, file) in files.iter().enumerate() {
1535 self.validate_owner(file, index)?;
1536 }
1537 let positions: Vec<_> = files
1538 .iter()
1539 .enumerate()
1540 .filter_map(|(index, file)| (!file.is_closed()).then_some(index))
1541 .collect();
1542 files.retain(|file| !file.is_closed());
1543 if files.is_empty() {
1544 return Ok(());
1545 }
1546 let descriptors: Vec<VfFile> = files
1547 .iter()
1548 .map(|file| file.raw().cloned())
1549 .collect::<VfResult<_>>()?;
1550 self.lock()?.vclose_impl(&descriptors).map_err(|error| {
1551 error
1552 .index()
1553 .and_then(|index| files.get(index))
1554 .map_or(error.clone(), |file| {
1555 error.with_context("vclose_owned", file.path())
1556 })
1557 .map_index(|index| positions.get(index).copied().unwrap_or(index))
1558 })?;
1559 for file in &mut files {
1560 file.file = None;
1561 }
1562 Ok(())
1563 }
1564
1565 pub fn vclose_owned(&self, mut files: Vec<FsFile<F>>) -> VfResult<()> {
1571 self.vclose(&mut files)
1572 }
1573
1574 pub fn vread_native(&self, requests: &[FsRead<'_, F>]) -> VfResult<Vec<FsReadResult>> {
1578 self.vread_with_limit_native(requests, self.limits.max_read_bytes)
1579 }
1580
1581 pub fn vread_with_limit_native(
1583 &self,
1584 requests: &[FsRead<'_, F>],
1585 max_total_bytes: usize,
1586 ) -> VfResult<Vec<FsReadResult>> {
1587 self.vread_with_limit_projected_native(requests, max_total_bytes, |request| request)
1588 }
1589
1590 pub fn vread_with_limit_projected_native<'b, T>(
1593 &self,
1594 requests: &[T],
1595 max_total_bytes: usize,
1596 project: impl for<'r> Fn(&'r T) -> &'r FsRead<'b, F>,
1597 ) -> VfResult<Vec<FsReadResult>>
1598 where
1599 F: 'b,
1600 {
1601 let mut requested = 0usize;
1602 for (index, request) in requests.iter().map(&project).enumerate() {
1603 requested = requested
1604 .checked_add(request.length)
1605 .ok_or_else(|| VfError::client(index, libc::EFBIG as u32))?;
1606 if requested > max_total_bytes {
1607 return Err(VfError::client(index, libc::EFBIG as u32)
1608 .with_context("vread_native", request.file.path()));
1609 }
1610 }
1611 let reads = self.read_ops(requests, &project)?;
1612 let results = self.lock()?.vread_impl(&reads).map_err(|error| {
1613 error
1614 .index()
1615 .and_then(|index| requests.get(index).map(&project))
1616 .map_or(error.clone(), |request| {
1617 error.with_context("vread_native", request.file.path())
1618 })
1619 })?;
1620 if results.len() != requests.len() {
1621 return Err(wrong_result_count(
1622 "vread_native",
1623 requests.len(),
1624 results.len(),
1625 ));
1626 }
1627 validate_read_results("vread_native", &reads, &results).map_err(|error| {
1628 error
1629 .index()
1630 .and_then(|index| requests.get(index).map(&project))
1631 .map_or(error.clone(), |request| {
1632 error.with_context("vread_native", request.file.path())
1633 })
1634 })?;
1635 Ok(results.into_iter().map(read_result).collect())
1636 }
1637
1638 fn read_ops<'b, T>(
1639 &self,
1640 requests: &[T],
1641 project: impl for<'r> Fn(&'r T) -> &'r FsRead<'b, F>,
1642 ) -> VfResult<Vec<ReadOp>>
1643 where
1644 F: 'b,
1645 {
1646 let mut reads = Vec::with_capacity(requests.len());
1647 for (index, request) in requests.iter().map(&project).enumerate() {
1648 self.validate_owner(request.file, index)?;
1649 reads.push(ReadOp::new(
1650 request.file.raw()?.clone(),
1651 request.offset,
1652 request.length,
1653 ));
1654 }
1655 Ok(reads)
1656 }
1657
1658 pub fn vread_into_native(
1659 &self,
1660 requests: &mut [FsReadInto<'_, F>],
1661 ) -> VfResult<Vec<FsReadIntoResult>> {
1662 self.vread_into_with_limit_native(requests, self.limits.max_read_bytes)
1663 }
1664
1665 pub fn vread_into_with_limit_native(
1668 &self,
1669 requests: &mut [FsReadInto<'_, F>],
1670 max_bytes: usize,
1671 ) -> VfResult<Vec<FsReadIntoResult>> {
1672 self.vread_into_with_limit_projected_native(
1673 requests,
1674 max_bytes,
1675 |request| request,
1676 |request| request,
1677 )
1678 }
1679
1680 pub fn vread_into_with_limit_projected_native<'b, T>(
1684 &self,
1685 requests: &mut [T],
1686 max_bytes: usize,
1687 project: impl for<'r> Fn(&'r T) -> &'r FsReadInto<'b, F>,
1688 mut project_mut: impl for<'r> FnMut(&'r mut T) -> &'r mut FsReadInto<'b, F>,
1689 ) -> VfResult<Vec<FsReadIntoResult>>
1690 where
1691 F: 'b,
1692 {
1693 let mut requested = 0usize;
1694 let mut reads = Vec::with_capacity(requests.len());
1695 for (index, request) in requests.iter().map(&project).enumerate() {
1696 self.validate_owner(request.file, index)?;
1697 requested = requested
1698 .checked_add(request.buffer.len())
1699 .ok_or_else(|| VfError::client(index, libc::EFBIG as u32))?;
1700 if requested > max_bytes {
1701 return Err(VfError::client(index, libc::EFBIG as u32));
1702 }
1703 reads.push(ReadOp::new(
1704 request.file.raw()?.clone(),
1705 request.offset,
1706 request.buffer.len(),
1707 ));
1708 }
1709 let results = {
1710 let mut buffers: Vec<&mut [u8]> = requests
1711 .iter_mut()
1712 .map(&mut project_mut)
1713 .map(|request| &mut *request.buffer)
1714 .collect();
1715 self.lock()?.vread_into_impl(&reads, &mut buffers)
1716 }
1717 .map_err(|error| {
1718 error
1719 .index()
1720 .and_then(|index| requests.get(index).map(&project))
1721 .map_or(error.clone(), |request| {
1722 error.with_context("vread_into_native", request.file.path())
1723 })
1724 })?;
1725 validate_read_into_results("vread_into_native", &reads, &results).map_err(|error| {
1726 error
1727 .index()
1728 .and_then(|index| requests.get(index).map(&project))
1729 .map_or(error.clone(), |request| {
1730 error.with_context("vread_into_native", request.file.path())
1731 })
1732 })?;
1733 Ok(results
1734 .into_iter()
1735 .map(|result| FsReadIntoResult {
1736 offset: result.offset,
1737 read: result.read,
1738 eof: result.eof,
1739 })
1740 .collect())
1741 }
1742
1743 pub fn vwrite_native(&self, requests: &[FsWrite<'_, F>]) -> VfResult<Vec<FsWriteResult>> {
1744 self.vwrite_projected_native(requests, |request| request)
1745 }
1746
1747 pub fn vwrite_projected_native<'b, T>(
1750 &self,
1751 requests: &[T],
1752 project: impl for<'r> Fn(&'r T) -> &'r FsWrite<'b, F>,
1753 ) -> VfResult<Vec<FsWriteResult>>
1754 where
1755 F: 'b,
1756 {
1757 self.vwrite_mapped_native(requests, |item| {
1758 let request = project(item);
1759 FsWrite {
1760 file: request.file,
1761 offset: request.offset,
1762 data: request.data,
1763 }
1764 })
1765 }
1766
1767 #[doc(hidden)]
1770 pub fn vwrite_mapped_native<'b, T>(
1771 &self,
1772 requests: &[T],
1773 project: impl Fn(&T) -> FsWrite<'b, F>,
1774 ) -> VfResult<Vec<FsWriteResult>>
1775 where
1776 F: 'b,
1777 {
1778 let writes = self.write_ops(requests, &project)?;
1779 let results = self.lock()?.vwrite_impl(&writes).map_err(|error| {
1780 error
1781 .index()
1782 .and_then(|index| requests.get(index).map(&project))
1783 .map_or(error.clone(), |request| {
1784 error.with_context("vwrite_native", request.file.path())
1785 })
1786 })?;
1787 if results.len() != requests.len() {
1788 return Err(wrong_result_count(
1789 "vwrite_native",
1790 requests.len(),
1791 results.len(),
1792 ));
1793 }
1794 validate_write_results("vwrite_native", &writes, &results).map_err(|error| {
1795 error
1796 .index()
1797 .and_then(|index| requests.get(index).map(&project))
1798 .map_or(error.clone(), |request| {
1799 error.with_context("vwrite_native", request.file.path())
1800 })
1801 })?;
1802 Ok(results.into_iter().map(write_result).collect())
1803 }
1804
1805 pub fn vwrite_all_native(&self, requests: &[FsWrite<'_, F>]) -> VfResult<Vec<FsWriteResult>> {
1811 self.vwrite_all_projected_native(requests, |request| request)
1812 }
1813
1814 pub fn vwrite_all_projected_native<'b, T>(
1817 &self,
1818 requests: &[T],
1819 project: impl for<'r> Fn(&'r T) -> &'r FsWrite<'b, F>,
1820 ) -> VfResult<Vec<FsWriteResult>>
1821 where
1822 F: 'b,
1823 {
1824 self.vwrite_all_mapped_native(requests, |item| {
1825 let request = project(item);
1826 FsWrite {
1827 file: request.file,
1828 offset: request.offset,
1829 data: request.data,
1830 }
1831 })
1832 }
1833
1834 #[doc(hidden)]
1837 pub fn vwrite_all_mapped_native<'b, T>(
1838 &self,
1839 requests: &[T],
1840 project: impl Fn(&T) -> FsWrite<'b, F>,
1841 ) -> VfResult<Vec<FsWriteResult>>
1842 where
1843 F: 'b,
1844 {
1845 let mut prior_by_path: HashMap<&Path, Vec<(usize, u64, u64)>> = HashMap::new();
1848 let mut blocked_by = vec![Vec::new(); requests.len()];
1849 for (index, request) in requests.iter().map(&project).enumerate() {
1850 self.validate_owner(request.file, index)?;
1851 request.file.raw().map_err(|error| {
1852 error
1853 .with_index(index)
1854 .with_context("vwrite_all_native", request.file.path())
1855 })?;
1856 let VfOffset::At(offset) = request.offset else {
1857 return Err(VfError::client(index, crate::ERR_INVAL)
1858 .with_context("vwrite_all_native", request.file.path()));
1859 };
1860 let end = offset
1861 .checked_add(request.data.len() as u64)
1862 .ok_or_else(|| {
1863 VfError::client(index, libc::EOVERFLOW as u32)
1864 .with_context("vwrite_all_native", request.file.path())
1865 })?;
1866 let prior = prior_by_path.entry(request.file.path()).or_default();
1870 for &(earlier, start, earlier_end) in prior.iter() {
1871 if offset < earlier_end && start < end {
1872 blocked_by[index].push(earlier);
1873 }
1874 }
1875 prior.push((index, offset, end));
1876 }
1877 let mut totals = vec![0usize; requests.len()];
1878 let mut stable = vec![true; requests.len()];
1879 let mut pending: Vec<usize> = (0..requests.len())
1880 .filter(|&index| !project(&requests[index]).data.is_empty())
1881 .collect();
1882 while !pending.is_empty() {
1883 let wave_indices: Vec<usize> = pending
1884 .iter()
1885 .copied()
1886 .filter(|&index| {
1887 blocked_by[index]
1888 .iter()
1889 .all(|&earlier| totals[earlier] == project(&requests[earlier]).data.len())
1890 })
1891 .collect();
1892 let wave: Vec<_> = wave_indices
1893 .iter()
1894 .map(|&index| {
1895 let request = project(&requests[index]);
1896 let VfOffset::At(offset) = request.offset else {
1897 return Err(VfError::client(index, crate::ERR_INVAL)
1898 .with_context("vwrite_all_native", request.file.path()));
1899 };
1900 let offset = offset.checked_add(totals[index] as u64).ok_or_else(|| {
1901 VfError::client(index, libc::EOVERFLOW as u32)
1902 .with_context("vwrite_all_native", request.file.path())
1903 })?;
1904 Ok(FsWrite {
1905 file: request.file,
1906 offset: VfOffset::At(offset),
1907 data: &request.data[totals[index]..],
1908 })
1909 })
1910 .collect::<VfResult<_>>()?;
1911 let results = self.vwrite_native(&wave).map_err(|error| {
1912 error.map_index(|index| wave_indices.get(index).copied().unwrap_or(index))
1913 })?;
1914 for (&index, result) in wave_indices.iter().zip(results) {
1915 if result.written == 0 {
1916 return Err(VfError::client(index, crate::ERR_IO)
1917 .with_context("vwrite_all_native", project(&requests[index]).file.path()));
1918 }
1919 totals[index] += result.written;
1920 stable[index] &= result.stable;
1921 }
1922 pending.retain(|&index| totals[index] < project(&requests[index]).data.len());
1923 }
1924 requests
1925 .iter()
1926 .map(&project)
1927 .enumerate()
1928 .map(|(index, request)| {
1929 let VfOffset::At(offset) = request.offset else {
1930 return Err(VfError::client(index, crate::ERR_INVAL)
1931 .with_context("vwrite_all_native", request.file.path()));
1932 };
1933 Ok(FsWriteResult {
1934 offset,
1935 written: totals[index],
1936 stable: stable[index],
1937 })
1938 })
1939 .collect()
1940 }
1941
1942 fn write_ops<'a, 'b, T>(
1943 &self,
1944 requests: &'a [T],
1945 project: impl Fn(&T) -> FsWrite<'b, F>,
1946 ) -> VfResult<Vec<WriteOpRef<'a>>>
1947 where
1948 F: 'b,
1949 'b: 'a,
1950 {
1951 let mut writes = Vec::with_capacity(requests.len());
1952 for (index, request) in requests.iter().map(&project).enumerate() {
1953 self.validate_owner(request.file, index)?;
1954 writes.push(WriteOpRef::new(
1955 request
1956 .file
1957 .raw()
1958 .map_err(|error| error.with_index(index))?,
1959 request.offset,
1960 request.data,
1961 ));
1962 }
1963 Ok(writes)
1964 }
1965
1966 fn validate_owner(&self, file: &FsFile<F>, index: usize) -> VfResult<()> {
1967 if Arc::ptr_eq(&self.inner, &file.inner) {
1968 Ok(())
1969 } else {
1970 Err(VfError::client(index, crate::ERR_INVAL))
1971 }
1972 }
1973}
1974
1975impl<F: Backend> FsClient<F> {
1976 pub fn read_files<P: AsRef<Path>>(&self, paths: &[P]) -> VfResult<Vec<Vec<u8>>> {
1983 self.read_files_with_options(
1984 paths,
1985 ReadAllOptions::new().max_total_bytes(self.limits.max_read_bytes),
1986 )
1987 }
1988
1989 pub fn read_files_with_options<P: AsRef<Path>>(
1991 &self,
1992 paths: &[P],
1993 options: ReadAllOptions,
1994 ) -> VfResult<Vec<Vec<u8>>> {
1995 let files: Vec<_> = paths
1996 .iter()
1997 .map(|path| VfFile::from_os_path(path.as_ref()))
1998 .collect();
1999 let result = self
2000 .lock()?
2001 .vread_all_with_options_impl(&files, options)
2002 .map_err(|error| {
2003 error
2004 .index()
2005 .and_then(|index| paths.get(index))
2006 .map_or(error.clone(), |path| {
2007 error.with_context("read_files", path.as_ref())
2008 })
2009 });
2010 result.and_then(|buffers| {
2011 if buffers.len() != paths.len() {
2012 Err(wrong_result_count("read_files", paths.len(), buffers.len()))
2013 } else {
2014 Ok(buffers)
2015 }
2016 })
2017 }
2018
2019 pub fn write_files<P: AsRef<Path>, B: AsRef<[u8]>>(&self, entries: &[(P, B)]) -> VfResult<()> {
2026 let mut seen = HashSet::with_capacity(entries.len());
2027 for (index, (path, _)) in entries.iter().enumerate() {
2028 if !seen.insert(path.as_ref()) {
2029 return Err(VfError::client(index, crate::ERR_INVAL)
2030 .with_context("write_files", path.as_ref()));
2031 }
2032 }
2033 let requests: Vec<_> = entries
2034 .iter()
2035 .map(|(path, _)| {
2036 OpenRequest::new(
2037 path.as_ref(),
2038 OpenFlags::WRITE | OpenFlags::CREATE | OpenFlags::TRUNCATE,
2039 )
2040 })
2041 .collect();
2042 let files = self.vopen(&requests)?;
2043 let writes: Vec<_> = files
2044 .iter()
2045 .zip(entries)
2046 .map(|(file, (_, data))| file.write_request_at(0, data.as_ref()))
2047 .collect();
2048 let result = self.vwrite_all_native(&writes);
2049 drop(writes);
2050 let close_result = self.vclose_owned(files);
2051 result?;
2052 close_result
2053 }
2054}
2055
2056pub struct OpenOptions<'a, F: FileSystem> {
2058 client: &'a FsClient<F>,
2059 flags: OpenFlags,
2060 mode: u32,
2061}
2062
2063impl<F: FileSystem> Clone for OpenOptions<'_, F> {
2064 fn clone(&self) -> Self {
2065 Self {
2066 client: self.client,
2067 flags: self.flags,
2068 mode: self.mode,
2069 }
2070 }
2071}
2072
2073impl<F: FileSystem> fmt::Debug for OpenOptions<'_, F> {
2074 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
2075 formatter
2076 .debug_struct("OpenOptions")
2077 .field("flags", &self.flags)
2078 .field("mode", &format_args!("{:#o}", self.mode))
2079 .finish()
2080 }
2081}
2082
2083impl<'a, F: FileSystem> OpenOptions<'a, F> {
2084 fn new(client: &'a FsClient<F>) -> Self {
2085 Self {
2086 client,
2087 flags: OpenFlags::empty(),
2088 mode: 0o666,
2089 }
2090 }
2091
2092 pub fn read(&mut self, enabled: bool) -> &mut Self {
2093 self.flags.set(OpenFlags::READ, enabled);
2094 self
2095 }
2096
2097 pub fn write(&mut self, enabled: bool) -> &mut Self {
2098 self.flags.set(OpenFlags::WRITE, enabled);
2099 self
2100 }
2101
2102 pub fn append(&mut self, enabled: bool) -> &mut Self {
2103 self.flags.set(OpenFlags::APPEND, enabled);
2104 self
2105 }
2106
2107 pub fn truncate(&mut self, enabled: bool) -> &mut Self {
2108 self.flags.set(OpenFlags::TRUNCATE, enabled);
2109 self
2110 }
2111
2112 pub fn create(&mut self, enabled: bool) -> &mut Self {
2113 self.flags.set(OpenFlags::CREATE, enabled);
2114 self
2115 }
2116
2117 pub fn create_new(&mut self, enabled: bool) -> &mut Self {
2118 self.flags.set(OpenFlags::CREATE_NEW, enabled);
2119 self
2120 }
2121
2122 pub fn mode(&mut self, mode: u32) -> &mut Self {
2123 self.mode = mode;
2124 self
2125 }
2126
2127 pub fn open(&self, path: impl AsRef<Path>) -> VfResult<FsFile<F>> {
2128 self.client
2129 .open_with(OpenRequest::new(path.as_ref(), self.flags).mode(self.mode))
2130 }
2131}
2132
2133impl<F: Backend> OpenOptions<'_, F> {
2134 pub fn vopen<P: AsRef<Path>>(&self, paths: &[P]) -> VfResult<Vec<FsFile<F>>> {
2135 let requests: Vec<OpenRequest> = paths
2136 .iter()
2137 .map(|path| OpenRequest::new(path.as_ref(), self.flags).mode(self.mode))
2138 .collect();
2139 self.client.vopen(&requests)
2140 }
2141}
2142
2143pub struct SetMetadata<'a, F: Backend> {
2145 client: &'a FsClient<F>,
2146 path: PathBuf,
2147 update: MetadataUpdate,
2148 follow: bool,
2149}
2150
2151impl<F: Backend> Clone for SetMetadata<'_, F> {
2152 fn clone(&self) -> Self {
2153 Self {
2154 client: self.client,
2155 path: self.path.clone(),
2156 update: self.update.clone(),
2157 follow: self.follow,
2158 }
2159 }
2160}
2161
2162impl<F: Backend> fmt::Debug for SetMetadata<'_, F> {
2163 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
2164 formatter
2165 .debug_struct("SetMetadata")
2166 .field("path", &self.path)
2167 .field("update", &self.update)
2168 .field("follow", &self.follow)
2169 .finish_non_exhaustive()
2170 }
2171}
2172
2173impl<F: Backend> SetMetadata<'_, F> {
2174 pub fn permissions(&mut self, permissions: Permissions) -> &mut Self {
2175 self.update.permissions = Some(permissions);
2176 self
2177 }
2178
2179 pub fn len(&mut self, len: u64) -> &mut Self {
2180 self.update.len = Some(len);
2181 self
2182 }
2183
2184 pub fn accessed(&mut self, accessed: SystemTime) -> &mut Self {
2185 self.update.accessed = Some(accessed);
2186 self
2187 }
2188
2189 pub fn modified(&mut self, modified: SystemTime) -> &mut Self {
2190 self.update.modified = Some(modified);
2191 self
2192 }
2193
2194 pub fn follow_symlinks(&mut self, follow: bool) -> &mut Self {
2195 self.follow = follow;
2196 self
2197 }
2198
2199 pub fn uid(&mut self, uid: u32) -> &mut Self {
2200 self.update.uid = Some(uid);
2201 self
2202 }
2203 pub fn gid(&mut self, gid: u32) -> &mut Self {
2204 self.update.gid = Some(gid);
2205 self
2206 }
2207 pub fn apply(&self) -> VfResult<()> {
2208 self.client
2209 .vsetattrs(&[(&self.path, self.update.clone())], self.follow)
2210 }
2211}
2212
2213pub struct FsFile<F: FileSystem> {
2219 inner: Arc<SharedBackend<F>>,
2220 file: Option<VfFile>,
2221 path: PathBuf,
2222}
2223
2224impl<F: FileSystem> fmt::Debug for FsFile<F> {
2225 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
2226 formatter
2227 .debug_struct("FsFile")
2228 .field("path", &self.path)
2229 .field("is_open", &self.file.is_some())
2230 .finish()
2231 }
2232}
2233
2234impl<F: FileSystem> FsFile<F> {
2235 pub fn is_closed(&self) -> bool {
2237 self.file.is_none()
2238 }
2239 fn raw(&self) -> VfResult<&VfFile> {
2240 self.file
2241 .as_ref()
2242 .ok_or_else(|| VfError::client(0, crate::ERR_EBADF))
2243 }
2244
2245 pub fn read_at(&self, buffer: &mut [u8], offset: u64) -> VfResult<usize> {
2247 self.read_into(buffer, VfOffset::At(offset))
2248 }
2249
2250 pub fn write_at(&self, buffer: &[u8], offset: u64) -> VfResult<usize> {
2252 self.write_from(buffer, VfOffset::At(offset))
2253 }
2254
2255 pub fn read_request_at(&self, offset: u64, length: usize) -> FsRead<'_, F> {
2256 FsRead {
2257 file: self,
2258 offset: VfOffset::At(offset),
2259 length,
2260 }
2261 }
2262
2263 pub fn path(&self) -> &Path {
2267 &self.path
2268 }
2269
2270 pub fn metadata(&self) -> VfResult<Metadata> {
2272 let attributes = AttrMask::MODE
2273 | AttrMask::SIZE
2274 | AttrMask::NLINK
2275 | AttrMask::FILEID
2276 | AttrMask::UID
2277 | AttrMask::GID
2278 | AttrMask::ATIME
2279 | AttrMask::MTIME
2280 | AttrMask::CTIME
2281 | AttrMask::CHANGE;
2282 self.inner
2283 .lock()
2284 .map_err(|_| poisoned())?
2285 .metadata_impl(MetadataQuery::new(self.raw()?.clone(), attributes))
2286 .map(vfsi_core::metadata_from_attrs)
2287 .map_err(|error| error.with_context("metadata", &self.path))
2288 }
2289
2290 pub fn truncate(&self, len: u64) -> VfResult<()> {
2292 self.set_metadata_update(MetadataUpdate::new().len(len))
2293 }
2294
2295 pub fn chmod(&self, permissions: Permissions) -> VfResult<()> {
2297 self.set_metadata_update(MetadataUpdate::new().permissions(permissions))
2298 }
2299
2300 fn set_metadata_update(&self, update: MetadataUpdate) -> VfResult<()> {
2301 let client = FsClient {
2302 inner: Arc::clone(&self.inner),
2303 limits: ResourceLimits::default(),
2304 };
2305 client.vsetattrs(&[(vfsi_core::MetadataTarget::File(self), update)], true)
2306 }
2307
2308 pub fn write_request_at<'a>(&'a self, offset: u64, data: &'a [u8]) -> FsWrite<'a, F> {
2309 FsWrite {
2310 file: self,
2311 offset: VfOffset::At(offset),
2312 data,
2313 }
2314 }
2315
2316 pub fn read_request_at_into<'a>(
2317 &'a self,
2318 offset: u64,
2319 buffer: &'a mut [u8],
2320 ) -> FsReadInto<'a, F> {
2321 FsReadInto {
2322 file: self,
2323 offset: VfOffset::At(offset),
2324 buffer,
2325 }
2326 }
2327
2328 pub fn read_native(&mut self, buffer: &mut [u8]) -> VfResult<usize> {
2329 self.read_into(buffer, VfOffset::Cur)
2330 }
2331
2332 pub fn read_to_end_with_limit(&mut self, max_bytes: usize) -> VfResult<Vec<u8>> {
2340 let mut data = Vec::new();
2341 loop {
2342 let remaining = max_bytes - data.len();
2343 if remaining == 0 {
2344 let mut probe = [0_u8; 1];
2345 if self.read_native(&mut probe)? == 0 {
2346 return Ok(data);
2347 }
2348 return Err(VfError::client(0, libc::EFBIG as u32)
2349 .with_context("read_to_end_with_limit", &self.path));
2350 }
2351 let chunk = remaining.min(crate::DEFAULT_READ_STREAM_CHUNK_BYTES);
2352 data.try_reserve(chunk).map_err(|_| {
2353 VfError::client(0, libc::ENOMEM as u32)
2354 .with_context("read_to_end_with_limit", &self.path)
2355 })?;
2356 let start = data.len();
2357 data.resize(start + chunk, 0);
2358 let count = self.read_native(&mut data[start..])?;
2359 data.truncate(start + count);
2360 if count == 0 {
2361 return Ok(data);
2362 }
2363 }
2364 }
2365
2366 pub fn write_native(&mut self, buffer: &[u8]) -> VfResult<usize> {
2367 self.write_from(buffer, VfOffset::Cur)
2368 }
2369
2370 fn read_into(&self, buffer: &mut [u8], offset: VfOffset) -> VfResult<usize> {
2371 if buffer.is_empty() {
2372 return Ok(0);
2373 }
2374 let request = ReadOp::new(self.raw()?.clone(), offset, buffer.len());
2375 let result = self
2376 .inner
2377 .lock()
2378 .map_err(|_| poisoned())?
2379 .read_into_impl(&request, buffer)
2380 .map_err(|error| error.with_context("read", &self.path))?;
2381 validate_read_into_results(
2382 "read",
2383 std::slice::from_ref(&request),
2384 std::slice::from_ref(&result),
2385 )
2386 .map_err(|error| error.with_context("read", &self.path))?;
2387 Ok(result.read)
2388 }
2389
2390 fn write_from(&self, buffer: &[u8], offset: VfOffset) -> VfResult<usize> {
2391 if buffer.is_empty() {
2392 return Ok(0);
2393 }
2394 let file = self.raw()?.clone();
2395 let result = self
2396 .inner
2397 .lock()
2398 .map_err(|_| poisoned())?
2399 .write_impl(WriteOpRef::new(&file, offset, buffer))
2400 .map_err(|error| error.with_context("write", &self.path))?;
2401 if result.written > buffer.len() {
2402 return Err(VfError::client(0, crate::ERR_IO).with_context("write", &self.path));
2403 }
2404 Ok(result.written)
2405 }
2406
2407 pub fn sync_data(&self) -> VfResult<()> {
2408 self.inner
2409 .lock()
2410 .map_err(|_| poisoned())?
2411 .sync_data(self.raw()?)
2412 .map_err(|error| error.with_context("sync_data", &self.path))
2413 }
2414
2415 pub fn sync_all(&self) -> VfResult<()> {
2416 self.inner
2417 .lock()
2418 .map_err(|_| poisoned())?
2419 .sync_all(self.raw()?)
2420 .map_err(|error| error.with_context("sync_all", &self.path))
2421 }
2422
2423 pub fn try_close(&mut self) -> VfResult<()> {
2429 let Some(file) = self.file.as_ref() else {
2430 return Ok(());
2431 };
2432 self.inner
2433 .lock()
2434 .map_err(|_| poisoned())?
2435 .close_impl(file)
2436 .map_err(|error| error.with_context("close", &self.path))?;
2437 self.file = None;
2438 Ok(())
2439 }
2440
2441 pub fn close(mut self) -> VfResult<()> {
2445 self.try_close()
2446 }
2447
2448 pub fn seek_native(&mut self, position: IoSeekFrom) -> VfResult<u64> {
2450 self.inner
2451 .lock()
2452 .map_err(|_| poisoned())?
2453 .seek_impl(self.raw()?, position)
2454 .map_err(|error| error.with_context("seek", &self.path))
2455 }
2456}
2457
2458impl<F: FileSystem> Read for FsFile<F> {
2459 fn read(&mut self, buffer: &mut [u8]) -> io::Result<usize> {
2460 self.read_native(buffer).map_err(io_error)
2461 }
2462}
2463
2464impl<F: FileSystem> Write for FsFile<F> {
2465 fn write(&mut self, buffer: &[u8]) -> io::Result<usize> {
2466 self.write_native(buffer).map_err(io_error)
2467 }
2468
2469 fn flush(&mut self) -> io::Result<()> {
2470 self.sync_data().map_err(io_error)
2471 }
2472}
2473
2474impl<F: FileSystem> Seek for FsFile<F> {
2475 fn seek(&mut self, position: IoSeekFrom) -> io::Result<u64> {
2476 self.seek_native(position).map_err(io_error)
2477 }
2478}
2479
2480pub struct FsRead<'a, F: FileSystem> {
2482 file: &'a FsFile<F>,
2483 offset: VfOffset,
2484 length: usize,
2485}
2486
2487pub struct FsWrite<'a, F: FileSystem> {
2489 file: &'a FsFile<F>,
2490 offset: VfOffset,
2491 data: &'a [u8],
2492}
2493
2494pub struct FsReadInto<'a, F: FileSystem> {
2496 file: &'a FsFile<F>,
2497 offset: VfOffset,
2498 buffer: &'a mut [u8],
2499}
2500
2501impl<F: FileSystem> Drop for FsFile<F> {
2502 fn drop(&mut self) {
2503 if let Some(file) = self.file.take() {
2504 self.inner.defer(
2505 CleanupTarget::File(file, std::mem::take(&mut self.path)),
2506 |backend, target| {
2507 let CleanupTarget::File(file, path) = target else {
2508 unreachable!()
2509 };
2510 backend
2511 .close_deferred(file)
2512 .map_err(|error| error.with_context("close", path))
2513 },
2514 );
2515 }
2516 }
2517}
2518
2519impl<F: FileSystem> FsClient<F> {
2520 pub fn vstatfs<P: vfsi_core::MetadataOperand<FsFile<F>>>(
2522 &self,
2523 targets: &[P],
2524 ) -> VfResult<Vec<crate::FilesystemStats>> {
2525 use vfsi_core::MetadataTarget;
2526 if targets.is_empty() {
2527 return Ok(Vec::new());
2528 }
2529 let mut files = Vec::with_capacity(targets.len());
2530 let mut paths = Vec::with_capacity(targets.len());
2531 for (index, target) in targets.iter().enumerate() {
2532 let (raw, path) = match target.metadata_target() {
2533 MetadataTarget::Path(path) => (VfFile::from_os_path(path), path),
2534 MetadataTarget::File(file) => {
2535 if !Arc::ptr_eq(&self.inner, &file.inner) {
2536 return Err(VfError::client(index, crate::ERR_INVAL)
2537 .with_context("vstatfs", &file.path));
2538 }
2539 (
2540 file.raw()
2541 .map_err(|e| e.with_index(index).with_context("vstatfs", &file.path))?
2542 .clone(),
2543 file.path.as_path(),
2544 )
2545 }
2546 };
2547 files.push(raw);
2548 paths.push(path);
2549 }
2550 let results = self
2551 .lock()?
2552 .vstatfs_impl(&files)
2553 .map_err(|error| match error.index() {
2554 Some(index) if index < paths.len() => error.with_context("vstatfs", paths[index]),
2555 Some(_) => {
2556 VfError::transport(None, "statfs backend returned an invalid error index")
2557 }
2558 None => error,
2559 })?;
2560 if results.len() != targets.len() {
2561 return Err(VfError::transport(
2562 None,
2563 "statfs backend returned an invalid result count",
2564 ));
2565 }
2566 Ok(results)
2567 }
2568 pub fn vsetattrs<P: vfsi_core::MetadataOperand<FsFile<F>>>(
2570 &self,
2571 updates: &[(P, MetadataUpdate)],
2572 follow_symlinks: bool,
2573 ) -> VfResult<()> {
2574 use vfsi_core::MetadataTarget;
2575 if updates.is_empty() {
2576 return Ok(());
2577 }
2578 let mut paths = Vec::with_capacity(updates.len());
2579 let mut attrs = Vec::with_capacity(updates.len());
2580 for (index, (target, update)) in updates.iter().enumerate() {
2581 let (raw, path) = match target.metadata_target() {
2582 MetadataTarget::Path(path) => (VfFile::from_os_path(path), path),
2583 MetadataTarget::File(file) => {
2584 if !Arc::ptr_eq(&self.inner, &file.inner) {
2585 return Err(VfError::client(index, crate::ERR_INVAL)
2586 .with_context("vsetattrs", &file.path));
2587 }
2588 (
2589 file.raw()
2590 .map_err(|e| e.with_index(index).with_context("vsetattrs", &file.path))?
2591 .clone(),
2592 file.path.as_path(),
2593 )
2594 }
2595 };
2596 if update.uid == Some(u32::MAX) || update.gid == Some(u32::MAX) {
2597 return Err(
2598 VfError::client(index, crate::ERR_INVAL).with_context("vsetattrs", path)
2599 );
2600 }
2601 paths.push(path);
2602 let mut attributes = crate::SetAttributes::new(raw);
2603 attributes.mode = update.permissions.map(crate::Permissions::mode);
2604 attributes.size = update.len;
2605 attributes.uid = update.uid;
2606 attributes.gid = update.gid;
2607 attributes.atime = update
2608 .accessed
2609 .map(crate::native::system_time_parts)
2610 .transpose()
2611 .map_err(|e| e.with_index(index).with_context("vsetattrs", path))?;
2612 attributes.mtime = update
2613 .modified
2614 .map(crate::native::system_time_parts)
2615 .transpose()
2616 .map_err(|e| e.with_index(index).with_context("vsetattrs", path))?;
2617 attrs.push(attributes);
2618 }
2619 self.lock()?
2620 .vsetattrs_impl(attrs, follow_symlinks)
2621 .map_err(|error| match error.index() {
2622 Some(index) if index < updates.len() => {
2623 error.with_context("vsetattrs", paths[index])
2624 }
2625 Some(_) => {
2626 VfError::transport(None, "setattrs backend returned an invalid error index")
2627 }
2628 None => error,
2629 })
2630 }
2631}
2632
2633fn link_error(error: VfError, paths: &[&Path], operation: &'static str) -> VfError {
2634 match error.index() {
2635 Some(index) if index < paths.len() => error.with_context(operation, paths[index]),
2636 Some(_) => VfError::transport(None, "link backend returned an invalid error index"),
2637 None => error,
2638 }
2639}
2640
2641#[cfg(test)]
2642mod traversal_tests {
2643 use super::*;
2644 use std::sync::atomic::{AtomicUsize, Ordering};
2645
2646 struct Snapshot {
2649 remaining: std::vec::IntoIter<DirEntry>,
2650 live: Arc<AtomicUsize>,
2651 }
2652 impl Drop for Snapshot {
2653 fn drop(&mut self) {
2654 self.live.fetch_sub(1, Ordering::SeqCst);
2655 }
2656 }
2657 fn snapshot_pages(
2658 live: Arc<AtomicUsize>,
2659 peak: Arc<AtomicUsize>,
2660 ) -> impl FnMut(&Path, Option<crate::DirPageCursor>, usize, usize) -> VfResult<DirectoryPage>
2661 {
2662 move |path, cursor, page_size, max_entries| {
2663 let mut snapshot = match cursor {
2664 Some(cursor) => cursor.into_state::<Snapshot>()?,
2665 None => {
2666 let depth = path.components().count();
2667 let entries = (0..129)
2668 .take(max_entries)
2669 .map(|index| {
2670 let path = path.join(format!("entry-{index}"));
2671 let attrs = crate::VfAttrs {
2672 file: VfFile::from_os_path(&path),
2673 ftype: if index == 0 && depth < 5 {
2674 crate::VfType::Directory
2675 } else {
2676 crate::VfType::Regular
2677 },
2678 ..Default::default()
2679 };
2680 DirEntry::new(path, vfsi_core::metadata_from_attrs(attrs))
2681 })
2682 .collect::<Vec<_>>();
2683 peak.fetch_max(live.fetch_add(1, Ordering::SeqCst) + 1, Ordering::SeqCst);
2684 Snapshot {
2685 remaining: entries.into_iter(),
2686 live: Arc::clone(&live),
2687 }
2688 }
2689 };
2690 let entries = snapshot.remaining.by_ref().take(page_size).collect();
2691 let next = if snapshot.remaining.len() == 0 {
2692 None
2693 } else {
2694 Some(crate::DirPageCursor::new(snapshot))
2695 };
2696 Ok((entries, next))
2697 }
2698 }
2699 #[test]
2700 fn traversal_never_retains_multiple_fallback_snapshots() {
2701 let live = Arc::new(AtomicUsize::new(0));
2702 let peak = Arc::new(AtomicUsize::new(0));
2703 let mut seen = 0;
2704 let completion = visit_walk_pages(
2705 Path::new("/tree"),
2706 crate::WalkOptions::new().max_entries(1000),
2707 snapshot_pages(Arc::clone(&live), Arc::clone(&peak)),
2708 |_| {
2709 seen += 1;
2710 Ok(std::ops::ControlFlow::Continue(()))
2711 },
2712 )
2713 .unwrap();
2714 assert_eq!(completion, TraversalCompletion::Complete);
2715 assert_eq!(seen, 4 * 129);
2716 assert_eq!(live.load(Ordering::SeqCst), 0);
2717 assert_eq!(peak.load(Ordering::SeqCst), 1);
2718 }
2719 #[test]
2720 fn traversal_drops_fallback_snapshot_on_stop_error_and_limit() {
2721 for outcome in 0..3 {
2722 let live = Arc::new(AtomicUsize::new(0));
2723 let peak = Arc::new(AtomicUsize::new(0));
2724 let result = visit_walk_pages(
2725 Path::new("/tree"),
2726 crate::WalkOptions::new().max_entries(if outcome == 2 { 2 } else { 1000 }),
2727 snapshot_pages(Arc::clone(&live), peak),
2728 |_| match outcome {
2729 0 => Ok(std::ops::ControlFlow::Break(())),
2730 1 => Err(VfError::client(0, libc::EIO as u32)),
2731 _ => Ok(std::ops::ControlFlow::Continue(())),
2732 },
2733 );
2734 match outcome {
2735 0 => assert_eq!(result.unwrap(), TraversalCompletion::Stopped),
2736 1 => assert_eq!(result.unwrap_err().err_no(), libc::EIO as u32),
2737 _ => assert_eq!(result.unwrap_err().err_no(), libc::EFBIG as u32),
2738 }
2739 assert_eq!(live.load(Ordering::SeqCst), 0);
2740 }
2741 }
2742}