1use std::collections::{HashMap, HashSet};
4use std::fmt;
5use std::io::{self, Read, Seek, SeekFrom as IoSeekFrom, Write};
6use std::path::{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 WriteResult as FsWriteResult,
15};
16
17use crate::traits::{validate_read_into_results, validate_read_results, validate_write_results};
18use crate::{
19 AttrMask, Attrs, Backend, Capabilities, DirEntry, FileSystem, MetadataQuery, MetadataUpdate,
20 OpenFlags, OpenOp, Permissions, ReadAllOptions, ReadOp, ReadResult, RemoveOptions,
21 StreamOptions, VfDir, VfError, VfFile, VfOffset, VfResult, WriteOpRef, WriteResult,
22};
23
24fn read_result(result: ReadResult) -> FsReadResult {
25 FsReadResult {
26 offset: result.offset,
27 data: result.data,
28 eof: result.eof,
29 }
30}
31
32fn write_result(result: WriteResult) -> FsWriteResult {
33 FsWriteResult {
34 offset: result.offset,
35 written: result.written,
36 stable: result.stable,
37 }
38}
39
40fn io_error(error: VfError) -> io::Error {
41 error.into()
42}
43
44fn poisoned() -> VfError {
45 VfError::client(0, crate::ERR_IO)
46}
47
48fn wrong_result_count(operation: &str, expected: usize, actual: usize) -> VfError {
49 VfError::transport_with_kind(
50 None,
51 crate::TransportKind::InvalidReply,
52 format!("{operation} backend returned {actual} results for {expected} requests"),
53 )
54}
55
56enum CleanupTarget {
57 File(VfFile, PathBuf),
58 Directory(VfDir, PathBuf),
59}
60struct PendingClose<F> {
61 target: CleanupTarget,
62 close: fn(&mut F, &CleanupTarget) -> VfResult<()>,
63}
64
65struct SharedBackend<F: FileSystem> {
68 backend: Mutex<Option<F>>,
69 pending: Mutex<Vec<PendingClose<F>>>,
70 has_pending: AtomicBool,
71}
72impl<F: FileSystem> SharedBackend<F> {
73 fn new(backend: F) -> Self {
74 Self {
75 backend: Mutex::new(Some(backend)),
76 pending: Mutex::new(Vec::new()),
77 has_pending: AtomicBool::new(false),
78 }
79 }
80 fn defer(&self, target: CleanupTarget, close: fn(&mut F, &CleanupTarget) -> VfResult<()>) {
81 self.pending
82 .lock()
83 .unwrap_or_else(|e| e.into_inner())
84 .push(PendingClose { target, close });
85 self.has_pending.store(true, Ordering::Release);
86 }
87 fn lock_without_cleanup(&self) -> Result<BackendGuard<'_, F>, ()> {
88 self.backend
89 .lock()
90 .map(|guard| BackendGuard {
91 guard: Some(guard),
92 pending: &self.pending,
93 has_pending: &self.has_pending,
94 })
95 .map_err(|_| ())
96 }
97 fn lock(&self) -> Result<BackendGuard<'_, F>, ()> {
98 let mut guard = self.lock_without_cleanup()?;
99 let _ = guard.drain_cleanup();
102 Ok(guard)
103 }
104}
105impl<F: FileSystem> Drop for SharedBackend<F> {
106 fn drop(&mut self) {
107 if let Some(backend) = self
108 .backend
109 .get_mut()
110 .unwrap_or_else(|e| e.into_inner())
111 .as_mut()
112 {
113 let pending = self.pending.get_mut().unwrap_or_else(|e| e.into_inner());
114 for close in pending.drain(..) {
115 let _ = (close.close)(backend, &close.target);
116 }
117 let callbacks = backend.take_notifications();
118 for callback in callbacks {
119 let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(callback));
120 }
121 }
122 }
123}
124struct BackendGuard<'a, F: FileSystem> {
125 guard: Option<std::sync::MutexGuard<'a, Option<F>>>,
126 pending: &'a Mutex<Vec<PendingClose<F>>>,
127 has_pending: &'a AtomicBool,
128}
129impl<F: FileSystem> std::ops::Deref for BackendGuard<'_, F> {
130 type Target = F;
131 fn deref(&self) -> &F {
132 self.guard
133 .as_ref()
134 .expect("live guard")
135 .as_ref()
136 .expect("live backend")
137 }
138}
139impl<F: FileSystem> std::ops::DerefMut for BackendGuard<'_, F> {
140 fn deref_mut(&mut self) -> &mut F {
141 self.guard
142 .as_mut()
143 .expect("live guard")
144 .as_mut()
145 .expect("live backend")
146 }
147}
148impl<F: FileSystem> BackendGuard<'_, F> {
149 fn drain_cleanup(&mut self) -> VfResult<()> {
150 if !self.has_pending.swap(false, Ordering::AcqRel) {
151 return Ok(());
152 }
153 let pending = std::mem::take(&mut *self.pending.lock().unwrap_or_else(|e| e.into_inner()));
154 let mut failed = Vec::new();
155 let mut first = None;
156 for close in pending {
157 if let Err(error) = (close.close)(self, &close.target) {
158 first.get_or_insert(error);
159 failed.push(close);
160 }
161 }
162 if !failed.is_empty() {
163 self.pending
164 .lock()
165 .unwrap_or_else(|e| e.into_inner())
166 .extend(failed);
167 self.has_pending.store(true, Ordering::Release);
168 }
169 first.map_or(Ok(()), Err)
170 }
171}
172impl<F: FileSystem> Drop for BackendGuard<'_, F> {
173 fn drop(&mut self) {
174 let callbacks = self.take_notifications();
175 drop(self.guard.take());
176 for callback in callbacks {
177 let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(callback));
180 }
181 }
182}
183
184pub struct FsClient<F: FileSystem> {
191 inner: Arc<SharedBackend<F>>,
192 limits: ResourceLimits,
193}
194
195impl<F: FileSystem> Clone for FsClient<F> {
196 fn clone(&self) -> Self {
197 Self {
198 inner: Arc::clone(&self.inner),
199 limits: self.limits,
200 }
201 }
202}
203
204impl<F: FileSystem> fmt::Debug for FsClient<F> {
205 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
206 formatter.debug_struct("FsClient").finish_non_exhaustive()
207 }
208}
209
210impl<F: FileSystem> FsClient<F> {
211 pub fn new(filesystem: F) -> Self {
212 Self {
213 inner: Arc::new(SharedBackend::new(filesystem)),
214 limits: ResourceLimits::default(),
215 }
216 }
217
218 pub fn with_limits(mut self, limits: ResourceLimits) -> Self {
221 self.limits = limits;
222 self
223 }
224
225 pub fn limits(&self) -> ResourceLimits {
226 self.limits
227 }
228
229 fn lock(&self) -> VfResult<BackendGuard<'_, F>> {
230 self.inner.lock().map_err(|_| poisoned())
231 }
232
233 pub fn into_inner(self) -> Result<F, Self> {
234 if self.drain_cleanup().is_err() {
235 return Err(self);
236 }
237 match Arc::try_unwrap(self.inner) {
238 Ok(mut shared) => Ok(shared
239 .backend
240 .get_mut()
241 .unwrap_or_else(|e| e.into_inner())
242 .take()
243 .expect("owned backend")),
244 Err(inner) => Err(Self {
245 inner,
246 limits: self.limits,
247 }),
248 }
249 }
250
251 pub fn drain_cleanup(&self) -> VfResult<()> {
254 self.inner
255 .lock_without_cleanup()
256 .map_err(|_| poisoned())?
257 .drain_cleanup()
258 }
259}
260
261impl<F: FileSystem> FsClient<F> {
262 pub(crate) fn capabilities(&self) -> VfResult<Capabilities> {
263 Ok(self.lock()?.capabilities())
264 }
265 pub(crate) fn open(&self, path: impl AsRef<Path>) -> VfResult<FsFile<F>> {
267 self.open_with_native(OpenOp::new(path.as_ref(), OpenFlags::READ))
268 }
269
270 pub(crate) fn open_with_native(&self, request: OpenOp) -> VfResult<FsFile<F>> {
271 let file = self.lock()?.open_impl(&request)?;
272 Ok(FsFile {
273 inner: Arc::clone(&self.inner),
274 file: Some(file),
275 path: request.path,
276 })
277 }
278
279 pub fn open_options(&self) -> OpenOptions<'_, F> {
280 OpenOptions::new(self)
281 }
282}
283
284impl<F: FileSystem> FsClient<F> {
285 pub(crate) fn read_stream_with_options(
292 &self,
293 path: impl AsRef<Path>,
294 options: StreamOptions,
295 mut callback: impl FnMut(u64, &[u8]) -> VfResult<bool>,
296 ) -> VfResult<StreamCompletion> {
297 let chunk_size = options.chunk_size_bytes();
298 if chunk_size == 0 {
299 return Err(VfError::failure(0, crate::ERR_INVAL));
300 }
301
302 let file = self.open(path)?;
303 let raw_file = file.raw()?.clone();
304 let operation = (|| -> VfResult<StreamCompletion> {
305 let mut offset = 0u64;
306 loop {
307 let request = ReadOp::at(raw_file.clone(), offset, chunk_size);
308 let result = {
309 let mut backend = self.lock()?;
310 let result = backend.read_impl(&request)?;
311 validate_read_results(
312 "read_stream",
313 std::slice::from_ref(&request),
314 std::slice::from_ref(&result),
315 )?;
316 result
317 };
318 let length = result.data.len();
319 if !result.data.is_empty() && !callback(offset, &result.data)? {
320 return Ok(StreamCompletion::Stopped {
321 next_offset: offset
322 .checked_add(length as u64)
323 .ok_or_else(|| VfError::client(0, crate::ERR_INVAL))?,
324 });
325 }
326 if result.eof {
327 return Ok(StreamCompletion::Complete);
328 }
329 offset = offset
330 .checked_add(length as u64)
331 .ok_or_else(|| VfError::client(0, libc::EOVERFLOW as u32))?;
332 }
333 })();
334 let close = file.close();
335 match operation {
336 Err(error) => Err(error),
337 Ok(completion) => close.map(|()| completion),
338 }
339 }
340}
341
342impl<F: Backend> FsClient<F> {
343 pub fn set_metadata(&self, path: impl AsRef<Path>) -> SetMetadata<'_, F> {
344 SetMetadata {
345 client: self,
346 path: path.as_ref().to_path_buf(),
347 update: MetadataUpdate::new(),
348 follow: true,
349 }
350 }
351}
352
353impl<F: Backend> FsClient<F> {
354 fn removal_metadata(&self, path: &Path, operation: &'static str) -> VfResult<Attrs> {
355 let mut filesystem = self.lock()?;
356 let follow = !filesystem.capabilities().contains(Capabilities::LSTAT);
357 filesystem
358 .metadata_path_impl(path, follow)
359 .map_err(|error| error.with_context(operation, path))
360 }
361}
362
363impl<F: Backend> FsClient<F> {
364 pub(crate) fn vrename<P: AsRef<Path>, Q: AsRef<Path>>(
368 &self,
369 pairs: &[(P, Q)],
370 options: vfsi_core::api::RenameOptions,
371 ) -> VfResult<()> {
372 if pairs.is_empty() {
373 return Ok(());
374 }
375 let requests: Vec<_> = pairs
376 .iter()
377 .map(|(from, to)| {
378 (
379 VfFile::from_os_path(from.as_ref()),
380 VfFile::from_os_path(to.as_ref()),
381 )
382 })
383 .collect();
384 self.lock()?
385 .vrename_with_options_impl(&requests, options)
386 .map_err(|error| match error.index() {
387 Some(index) if index < pairs.len() => {
388 error.with_context("vrename", pairs[index].0.as_ref())
389 }
390 Some(_) => {
391 VfError::transport(None, "rename backend returned an invalid error index")
392 }
393 None => error,
394 })
395 }
396
397 pub(crate) fn vmkdir<P: AsRef<Path>>(&self, directories: &[(P, u32)]) -> VfResult<()> {
399 let mut seen = HashSet::with_capacity(directories.len());
400 for (index, (path, _)) in directories.iter().enumerate() {
401 if !seen.insert(path.as_ref()) {
402 return Err(
403 VfError::client(index, crate::ERR_INVAL).with_context("vmkdir", path.as_ref())
404 );
405 }
406 }
407 if directories.is_empty() {
408 return Ok(());
409 }
410 let dirs: Vec<_> = directories
411 .iter()
412 .map(|(path, mode)| crate::VfAttrs {
413 file: VfFile::from_os_path(path.as_ref()),
414 masks: AttrMask::MODE,
415 mode: *mode,
416 ..Default::default()
417 })
418 .collect();
419 self.lock()?
420 .vmkdir_impl(&dirs)
421 .map_err(|error| match error.index() {
422 Some(index) if index < directories.len() => {
423 error.with_context("vmkdir", directories[index].0.as_ref())
424 }
425 Some(_) => {
426 VfError::transport(None, "mkdir backend returned an invalid error index")
427 }
428 None => error,
429 })
430 }
431
432 pub(crate) fn vsymlink<P: AsRef<Path>, Q: AsRef<Path>>(
434 &self,
435 pairs: &[(P, Q)],
436 ) -> VfResult<()> {
437 if pairs.is_empty() {
438 return Ok(());
439 }
440 let targets: Vec<_> = pairs.iter().map(|(target, _)| target.as_ref()).collect();
441 let links: Vec<_> = pairs.iter().map(|(_, link)| link.as_ref()).collect();
442 self.lock()?
443 .vsymlink_impl(&targets, &links)
444 .map_err(|error| link_error(error, &links, "vsymlink"))
445 }
446
447 pub(crate) fn vreadlink<P: AsRef<Path>>(&self, paths: &[P]) -> VfResult<Vec<PathBuf>> {
449 if paths.is_empty() {
450 return Ok(Vec::new());
451 }
452 let paths: Vec<_> = paths.iter().map(AsRef::as_ref).collect();
453 let targets = self
454 .lock()?
455 .vreadlink_impl(&paths)
456 .map_err(|error| link_error(error, &paths, "vreadlink"))?;
457 if targets.len() != paths.len() {
458 return Err(VfError::transport(
459 None,
460 "readlink backend returned an invalid result count",
461 ));
462 }
463 Ok(targets
464 .into_iter()
465 .map(crate::native::bytes_to_path)
466 .collect())
467 }
468
469 pub(crate) fn vhardlink<P: AsRef<Path>, Q: AsRef<Path>>(
471 &self,
472 pairs: &[(P, Q)],
473 ) -> VfResult<()> {
474 if pairs.is_empty() {
475 return Ok(());
476 }
477 let sources: Vec<_> = pairs.iter().map(|(source, _)| source.as_ref()).collect();
478 let links: Vec<_> = pairs.iter().map(|(_, link)| link.as_ref()).collect();
479 self.lock()?
480 .vhardlink_impl(&sources, &links)
481 .map_err(|error| link_error(error, &links, "vhardlink"))
482 }
483
484 #[doc(hidden)]
490 pub fn directory_page_batch_size(&self) -> VfResult<usize> {
491 Ok(self.lock()?.directory_page_batch_size().clamp(1, 32))
492 }
493 #[doc(hidden)]
495 pub fn read_dir_pages_with_fields(
496 &self,
497 paths: &[&Path],
498 fields: AttrMask,
499 cursors: Vec<Option<crate::DirPageCursor>>,
500 page_size: usize,
501 max_entries: usize,
502 ) -> VfResult<Vec<crate::DirectoryPage>> {
503 let pages = self.lock()?.vlistdir_pages_impl(
504 paths,
505 fields | AttrMask::MODE | AttrMask::SIZE,
506 cursors,
507 page_size,
508 max_entries,
509 )?;
510 if pages.len() != paths.len() {
511 return Err(VfError::transport(
512 None,
513 "directory pages returned an invalid result count",
514 ));
515 }
516 pages
517 .into_iter()
518 .zip(paths)
519 .enumerate()
520 .map(|(index, ((attrs, next, children), path))| {
521 if children.len() != attrs.len()
522 || attrs.len() > page_size
523 || (attrs.is_empty() && next.is_some())
524 {
525 return Err(VfError::transport(
526 Some(index),
527 "invalid directory page progress",
528 ));
529 }
530 let mut seeds = Vec::new();
531 let entries = attrs
532 .into_iter()
533 .zip(children)
534 .map(|(attrs, child)| {
535 let entry_path = attrs
536 .file
537 .path()
538 .ok_or_else(|| {
539 VfError::transport(Some(index), "directory entry has no path")
540 })?
541 .to_path_buf();
542 if entry_path.parent() != Some(*path) {
543 return Err(VfError::transport(
544 Some(index),
545 "directory entry is outside its parent",
546 ));
547 }
548 if let Some(child) = child {
549 seeds.push((entry_path.clone(), child));
550 }
551 Ok(DirEntry::new(
552 entry_path,
553 vfsi_core::metadata_from_attrs(attrs),
554 ))
555 })
556 .collect::<VfResult<Vec<_>>>()?;
557 Ok((
558 DirectoryListing {
559 path: path.to_path_buf(),
560 entries,
561 },
562 next,
563 seeds,
564 ))
565 })
566 .collect()
567 }
568}
569
570impl<F: Backend> FsClient<F> {
571 pub fn ensure_empty_dir(&self, path: impl AsRef<Path>) -> VfResult<()> {
574 self.lock()?.ensure_empty_dir_impl(path.as_ref())
575 }
576
577 pub(crate) fn remove_dir_contents_with_options(
579 &self,
580 path: impl AsRef<Path>,
581 options: RemoveOptions,
582 ) -> VfResult<()> {
583 let path = path.as_ref();
584 if !self
585 .removal_metadata(path, "remove_dir_contents_impl")?
586 .is_dir()
587 {
588 return Err(
589 VfError::client(0, crate::ERR_NOTDIR).with_context("remove_dir_contents", path)
590 );
591 }
592 self.lock()?
593 .remove_dir_contents_path_with_options_impl(path, options)
594 .map_err(|error| error.with_context("remove_dir_contents", path))
595 }
596
597 pub fn open_dir_handle(&self, path: impl AsRef<Path>) -> VfResult<FsDir<F>> {
601 let path = path.as_ref();
602 let mut backend = self.lock()?;
603 let dir = backend.open_dir_impl(path)?;
604 if !matches!(dir, VfDir::Descriptor { .. }) {
605 let _ = backend.close_dir_impl(&dir);
606 return Err(VfError::unsupported(0).with_context("open_dir_handle", path));
607 }
608 Ok(FsDir {
609 inner: Arc::clone(&self.inner),
610 dir: Some(dir),
611 path: path.to_path_buf(),
612 })
613 }
614}
615
616pub struct FsDir<F: Backend> {
619 inner: Arc<SharedBackend<F>>,
620 dir: Option<VfDir>,
621 path: PathBuf,
622}
623
624impl<F: Backend> fmt::Debug for FsDir<F> {
625 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
626 formatter
627 .debug_struct("FsDir")
628 .field("path", &self.path)
629 .finish_non_exhaustive()
630 }
631}
632
633impl<F: Backend> FsDir<F> {
634 pub fn is_closed(&self) -> bool {
635 self.dir.is_none()
636 }
637 pub fn path(&self) -> &Path {
639 &self.path
640 }
641
642 pub fn remove_contents(&self) -> VfResult<()> {
643 self.remove_contents_with_options(RemoveOptions::default())
644 }
645
646 pub fn remove_contents_with_options(&self, options: RemoveOptions) -> VfResult<()> {
647 let dir = self
648 .dir
649 .as_ref()
650 .ok_or_else(|| VfError::client(0, crate::ERR_EBADF))?;
651 self.inner
652 .lock()
653 .map_err(|_| poisoned())?
654 .remove_dir_contents_handle_with_options_impl(dir, options)
655 .map_err(|error| error.with_context("remove_dir_contents", &self.path))
656 }
657
658 pub fn close(mut self) -> VfResult<()> {
659 self.try_close()
660 }
661
662 pub fn try_close(&mut self) -> VfResult<()> {
664 let Some(dir) = self.dir.as_ref() else {
665 return Ok(());
666 };
667 self.inner
668 .lock()
669 .map_err(|_| poisoned())?
670 .close_dir_impl(dir)?;
671 self.dir = None;
672 Ok(())
673 }
674}
675
676impl<F: Backend> Drop for FsDir<F> {
677 fn drop(&mut self) {
678 if let Some(dir) = self.dir.take() {
679 self.inner.defer(
680 CleanupTarget::Directory(dir, std::mem::take(&mut self.path)),
681 |backend, target| {
682 let CleanupTarget::Directory(dir, path) = target else {
683 unreachable!()
684 };
685 backend
686 .close_dir_impl(dir)
687 .map_err(|error| error.with_context("close_dir", path))
688 },
689 );
690 }
691 }
692}
693
694impl<F: Backend> FsClient<F> {
695 pub(crate) fn vcopy<P: AsRef<Path>, Q: AsRef<Path>>(
698 &self,
699 pairs: &[(P, Q)],
700 options: vfsi_core::api::CopyOption,
701 ) -> VfResult<()> {
702 let extents: Vec<_> = pairs
703 .iter()
704 .map(|(from, to)| {
705 crate::ExtentPair::from_os_paths(from.as_ref(), 0, to.as_ref(), 0, None)
706 })
707 .collect();
708 self.lock()?.vcopy_impl(&extents, options).map_err(|error| {
709 error
710 .index()
711 .and_then(|index| pairs.get(index))
712 .map_or(error.clone(), |(_, to)| {
713 error.with_context("vcopy", to.as_ref())
714 })
715 })
716 }
717
718 #[doc(hidden)]
721 pub fn vremove_native<P: AsRef<Path>>(&self, paths: &[P], recursive: bool) -> VfResult<()> {
722 self.vremove_with_options_native(paths, recursive, RemoveOptions::default())
723 }
724
725 #[doc(hidden)]
727 pub fn vremove_with_options_native<P: AsRef<Path>>(
728 &self,
729 paths: &[P],
730 recursive: bool,
731 options: RemoveOptions,
732 ) -> VfResult<()> {
733 let paths: Vec<&Path> = paths.iter().map(AsRef::as_ref).collect();
734 self.lock()?
735 .remove_paths_with_options_impl(&paths, recursive, options)
736 .map_err(|error| {
737 error
738 .index()
739 .and_then(|index| paths.get(index))
740 .map_or(error.clone(), |path| {
741 error.with_context("vremove_native", path)
742 })
743 })
744 }
745
746 #[doc(hidden)]
749 pub fn vgetattrs_native<P: AsRef<Path>>(
750 &self,
751 paths: &[P],
752 fields: AttrMask,
753 follow: bool,
754 ) -> VfResult<Vec<Attrs>> {
755 if paths.is_empty() {
756 return Ok(Vec::new());
757 }
758 let mut attrs: Vec<_> = paths
759 .iter()
760 .map(|path| crate::VfAttrs {
761 file: VfFile::from_os_path(path.as_ref()),
762 masks: fields | AttrMask::MODE,
763 ..crate::VfAttrs::default()
764 })
765 .collect();
766 let result = {
767 let mut backend = self.lock()?;
768 if follow {
769 backend.vgetattrs_impl(&mut attrs)
770 } else {
771 backend.vgetattrs_nofollow_impl(&mut attrs)
772 }
773 };
774 result.map_err(|error| match error.index() {
775 Some(index) if index < paths.len() => {
776 error.with_context("metadatav", paths[index].as_ref())
777 }
778 Some(_) => VfError::transport(None, "metadata backend returned an invalid error index"),
779 None => error,
780 })?;
781 Ok(attrs
782 .into_iter()
783 .map(vfsi_core::metadata_from_attrs)
784 .collect())
785 }
786
787 #[doc(hidden)]
789 pub fn vsymlink_attrs_native(&self, paths: &[&Path]) -> VfResult<Vec<Attrs>> {
790 let mut attrs: Vec<_> = paths
791 .iter()
792 .map(|path| crate::VfAttrs {
793 file: VfFile::from_os_path(path),
794 masks: AttrMask::MODE,
795 ..crate::VfAttrs::default()
796 })
797 .collect();
798 self.lock()?
799 .vgetattrs_nofollow_impl(&mut attrs)
800 .map_err(|error| {
801 error
802 .index()
803 .and_then(|index| paths.get(index))
804 .map_or(error.clone(), |path| {
805 error.with_context("vsymlink_attrs_native", path)
806 })
807 })?;
808 Ok(attrs
809 .into_iter()
810 .map(vfsi_core::metadata_from_attrs)
811 .collect())
812 }
813}
814
815impl<F: Backend> FsClient<F> {
816 pub(crate) fn vopen(&self, requests: &[OpenOp]) -> VfResult<Vec<FsFile<F>>> {
822 let mut filesystem = self.lock()?;
823 let files = filesystem.vopen_impl(requests).map_err(|error| {
824 error
825 .index()
826 .and_then(|index| requests.get(index))
827 .map_or(error.clone(), |request| {
828 error.with_context("vopen", &request.path)
829 })
830 })?;
831 if files.len() != requests.len() {
832 let error = wrong_result_count("vopen", requests.len(), files.len());
833 for file in &files {
834 let _ = filesystem.close_impl(file);
835 }
836 return Err(error);
837 }
838 drop(filesystem);
839 Ok(files
840 .into_iter()
841 .zip(requests)
842 .map(|(file, request)| FsFile {
843 inner: Arc::clone(&self.inner),
844 file: Some(file),
845 path: request.path.clone(),
846 })
847 .collect())
848 }
849
850 #[doc(hidden)]
854 pub fn vclose<'a>(&self, files: impl IntoIterator<Item = &'a mut FsFile<F>>) -> VfResult<()>
855 where
856 F: 'a,
857 {
858 let mut files: Vec<_> = files.into_iter().collect();
859 for (index, file) in files.iter().enumerate() {
860 self.validate_owner(file, index)?;
861 }
862 let positions: Vec<_> = files
863 .iter()
864 .enumerate()
865 .filter_map(|(index, file)| (!file.is_closed()).then_some(index))
866 .collect();
867 files.retain(|file| !file.is_closed());
868 if files.is_empty() {
869 return Ok(());
870 }
871 let descriptors: Vec<VfFile> = files
872 .iter()
873 .map(|file| file.raw().cloned())
874 .collect::<VfResult<_>>()?;
875 self.lock()?.vclose_impl(&descriptors).map_err(|error| {
876 error
877 .index()
878 .and_then(|index| files.get(index))
879 .map_or(error.clone(), |file| {
880 error.with_context("vclose_owned", file.path())
881 })
882 .map_index(|index| positions.get(index).copied().unwrap_or(index))
883 })?;
884 for file in &mut files {
885 file.file = None;
886 }
887 Ok(())
888 }
889
890 #[doc(hidden)]
896 pub fn vclose_owned(&self, mut files: Vec<FsFile<F>>) -> VfResult<()> {
897 self.vclose(&mut files)
898 }
899
900 #[doc(hidden)]
904 pub fn vread_native(&self, requests: &[FsRead<'_, F>]) -> VfResult<Vec<FsReadResult>> {
905 self.vread_with_limit_native(requests, self.limits.max_read_bytes)
906 }
907
908 #[doc(hidden)]
910 pub fn vread_with_limit_native(
911 &self,
912 requests: &[FsRead<'_, F>],
913 max_total_bytes: usize,
914 ) -> VfResult<Vec<FsReadResult>> {
915 self.vread_with_limit_projected_native(requests, max_total_bytes, |request| request)
916 }
917
918 #[doc(hidden)]
921 pub fn vread_with_limit_projected_native<'b, T>(
922 &self,
923 requests: &[T],
924 max_total_bytes: usize,
925 project: impl for<'r> Fn(&'r T) -> &'r FsRead<'b, F>,
926 ) -> VfResult<Vec<FsReadResult>>
927 where
928 F: 'b,
929 {
930 let mut requested = 0usize;
931 for (index, request) in requests.iter().map(&project).enumerate() {
932 requested = requested
933 .checked_add(request.length)
934 .ok_or_else(|| VfError::client(index, libc::EFBIG as u32))?;
935 if requested > max_total_bytes {
936 return Err(VfError::client(index, libc::EFBIG as u32)
937 .with_context("vread_native", request.file.path()));
938 }
939 }
940 let reads = self.read_ops(requests, &project)?;
941 let results = self.lock()?.vread_impl(&reads).map_err(|error| {
942 error
943 .index()
944 .and_then(|index| requests.get(index).map(&project))
945 .map_or(error.clone(), |request| {
946 error.with_context("vread_native", request.file.path())
947 })
948 })?;
949 if results.len() != requests.len() {
950 return Err(wrong_result_count(
951 "vread_native",
952 requests.len(),
953 results.len(),
954 ));
955 }
956 validate_read_results("vread_native", &reads, &results).map_err(|error| {
957 error
958 .index()
959 .and_then(|index| requests.get(index).map(&project))
960 .map_or(error.clone(), |request| {
961 error.with_context("vread_native", request.file.path())
962 })
963 })?;
964 Ok(results.into_iter().map(read_result).collect())
965 }
966
967 fn read_ops<'b, T>(
968 &self,
969 requests: &[T],
970 project: impl for<'r> Fn(&'r T) -> &'r FsRead<'b, F>,
971 ) -> VfResult<Vec<ReadOp>>
972 where
973 F: 'b,
974 {
975 let mut reads = Vec::with_capacity(requests.len());
976 for (index, request) in requests.iter().map(&project).enumerate() {
977 self.validate_owner(request.file, index)?;
978 reads.push(ReadOp::new(
979 request.file.raw()?.clone(),
980 request.offset,
981 request.length,
982 ));
983 }
984 Ok(reads)
985 }
986
987 #[doc(hidden)]
988 pub fn vread_into_native(
989 &self,
990 requests: &mut [FsReadInto<'_, F>],
991 ) -> VfResult<Vec<FsReadIntoResult>> {
992 self.vread_into_with_limit_native(requests, self.limits.max_read_bytes)
993 }
994
995 #[doc(hidden)]
998 pub fn vread_into_with_limit_native(
999 &self,
1000 requests: &mut [FsReadInto<'_, F>],
1001 max_bytes: usize,
1002 ) -> VfResult<Vec<FsReadIntoResult>> {
1003 self.vread_into_with_limit_projected_native(
1004 requests,
1005 max_bytes,
1006 |request| request,
1007 |request| request,
1008 )
1009 }
1010
1011 #[doc(hidden)]
1015 pub fn vread_into_with_limit_projected_native<'b, T>(
1016 &self,
1017 requests: &mut [T],
1018 max_bytes: usize,
1019 project: impl for<'r> Fn(&'r T) -> &'r FsReadInto<'b, F>,
1020 mut project_mut: impl for<'r> FnMut(&'r mut T) -> &'r mut FsReadInto<'b, F>,
1021 ) -> VfResult<Vec<FsReadIntoResult>>
1022 where
1023 F: 'b,
1024 {
1025 let mut requested = 0usize;
1026 let mut reads = Vec::with_capacity(requests.len());
1027 for (index, request) in requests.iter().map(&project).enumerate() {
1028 self.validate_owner(request.file, index)?;
1029 requested = requested
1030 .checked_add(request.buffer.len())
1031 .ok_or_else(|| VfError::client(index, libc::EFBIG as u32))?;
1032 if requested > max_bytes {
1033 return Err(VfError::client(index, libc::EFBIG as u32));
1034 }
1035 reads.push(ReadOp::new(
1036 request.file.raw()?.clone(),
1037 request.offset,
1038 request.buffer.len(),
1039 ));
1040 }
1041 let results = {
1042 let mut buffers: Vec<&mut [u8]> = requests
1043 .iter_mut()
1044 .map(&mut project_mut)
1045 .map(|request| &mut *request.buffer)
1046 .collect();
1047 self.lock()?.vread_into_impl(&reads, &mut buffers)
1048 }
1049 .map_err(|error| {
1050 error
1051 .index()
1052 .and_then(|index| requests.get(index).map(&project))
1053 .map_or(error.clone(), |request| {
1054 error.with_context("vread_into_native", request.file.path())
1055 })
1056 })?;
1057 validate_read_into_results("vread_into_native", &reads, &results).map_err(|error| {
1058 error
1059 .index()
1060 .and_then(|index| requests.get(index).map(&project))
1061 .map_or(error.clone(), |request| {
1062 error.with_context("vread_into_native", request.file.path())
1063 })
1064 })?;
1065 Ok(results
1066 .into_iter()
1067 .map(|result| FsReadIntoResult {
1068 offset: result.offset,
1069 read: result.read,
1070 eof: result.eof,
1071 })
1072 .collect())
1073 }
1074
1075 #[doc(hidden)]
1076 pub fn vwrite_native(&self, requests: &[FsWrite<'_, F>]) -> VfResult<Vec<FsWriteResult>> {
1077 self.vwrite_projected_native(requests, |request| request)
1078 }
1079
1080 #[doc(hidden)]
1083 pub fn vwrite_projected_native<'b, T>(
1084 &self,
1085 requests: &[T],
1086 project: impl for<'r> Fn(&'r T) -> &'r FsWrite<'b, F>,
1087 ) -> VfResult<Vec<FsWriteResult>>
1088 where
1089 F: 'b,
1090 {
1091 self.vwrite_mapped_native(requests, |item| {
1092 let request = project(item);
1093 FsWrite {
1094 file: request.file,
1095 offset: request.offset,
1096 data: request.data,
1097 }
1098 })
1099 }
1100
1101 #[doc(hidden)]
1104 #[doc(hidden)]
1105 pub fn vwrite_mapped_native<'b, T>(
1106 &self,
1107 requests: &[T],
1108 project: impl Fn(&T) -> FsWrite<'b, F>,
1109 ) -> VfResult<Vec<FsWriteResult>>
1110 where
1111 F: 'b,
1112 {
1113 let writes = self.write_ops(requests, &project)?;
1114 let results = self.lock()?.vwrite_impl(&writes).map_err(|error| {
1115 error
1116 .index()
1117 .and_then(|index| requests.get(index).map(&project))
1118 .map_or(error.clone(), |request| {
1119 error.with_context("vwrite_native", request.file.path())
1120 })
1121 })?;
1122 if results.len() != requests.len() {
1123 return Err(wrong_result_count(
1124 "vwrite_native",
1125 requests.len(),
1126 results.len(),
1127 ));
1128 }
1129 validate_write_results("vwrite_native", &writes, &results).map_err(|error| {
1130 error
1131 .index()
1132 .and_then(|index| requests.get(index).map(&project))
1133 .map_or(error.clone(), |request| {
1134 error.with_context("vwrite_native", request.file.path())
1135 })
1136 })?;
1137 Ok(results.into_iter().map(write_result).collect())
1138 }
1139
1140 #[doc(hidden)]
1146 pub fn vwrite_all_native(&self, requests: &[FsWrite<'_, F>]) -> VfResult<Vec<FsWriteResult>> {
1147 self.vwrite_all_projected_native(requests, |request| request)
1148 }
1149
1150 #[doc(hidden)]
1153 pub fn vwrite_all_projected_native<'b, T>(
1154 &self,
1155 requests: &[T],
1156 project: impl for<'r> Fn(&'r T) -> &'r FsWrite<'b, F>,
1157 ) -> VfResult<Vec<FsWriteResult>>
1158 where
1159 F: 'b,
1160 {
1161 self.vwrite_all_mapped_native(requests, |item| {
1162 let request = project(item);
1163 FsWrite {
1164 file: request.file,
1165 offset: request.offset,
1166 data: request.data,
1167 }
1168 })
1169 }
1170
1171 #[doc(hidden)]
1174 #[doc(hidden)]
1175 pub fn vwrite_all_mapped_native<'b, T>(
1176 &self,
1177 requests: &[T],
1178 project: impl Fn(&T) -> FsWrite<'b, F>,
1179 ) -> VfResult<Vec<FsWriteResult>>
1180 where
1181 F: 'b,
1182 {
1183 let mut prior_by_path: HashMap<&Path, Vec<(usize, u64, u64)>> = HashMap::new();
1186 let mut blocked_by = vec![Vec::new(); requests.len()];
1187 for (index, request) in requests.iter().map(&project).enumerate() {
1188 self.validate_owner(request.file, index)?;
1189 request.file.raw().map_err(|error| {
1190 error
1191 .with_index(index)
1192 .with_context("vwrite_all_native", request.file.path())
1193 })?;
1194 let VfOffset::At(offset) = request.offset else {
1195 return Err(VfError::client(index, crate::ERR_INVAL)
1196 .with_context("vwrite_all_native", request.file.path()));
1197 };
1198 let end = offset
1199 .checked_add(request.data.len() as u64)
1200 .ok_or_else(|| {
1201 VfError::client(index, libc::EOVERFLOW as u32)
1202 .with_context("vwrite_all_native", request.file.path())
1203 })?;
1204 let prior = prior_by_path.entry(request.file.path()).or_default();
1208 for &(earlier, start, earlier_end) in prior.iter() {
1209 if offset < earlier_end && start < end {
1210 blocked_by[index].push(earlier);
1211 }
1212 }
1213 prior.push((index, offset, end));
1214 }
1215 let mut totals = vec![0usize; requests.len()];
1216 let mut stable = vec![true; requests.len()];
1217 let mut pending: Vec<usize> = (0..requests.len())
1218 .filter(|&index| !project(&requests[index]).data.is_empty())
1219 .collect();
1220 while !pending.is_empty() {
1221 let wave_indices: Vec<usize> = pending
1222 .iter()
1223 .copied()
1224 .filter(|&index| {
1225 blocked_by[index]
1226 .iter()
1227 .all(|&earlier| totals[earlier] == project(&requests[earlier]).data.len())
1228 })
1229 .collect();
1230 let wave: Vec<_> = wave_indices
1231 .iter()
1232 .map(|&index| {
1233 let request = project(&requests[index]);
1234 let VfOffset::At(offset) = request.offset else {
1235 return Err(VfError::client(index, crate::ERR_INVAL)
1236 .with_context("vwrite_all_native", request.file.path()));
1237 };
1238 let offset = offset.checked_add(totals[index] as u64).ok_or_else(|| {
1239 VfError::client(index, libc::EOVERFLOW as u32)
1240 .with_context("vwrite_all_native", request.file.path())
1241 })?;
1242 Ok(FsWrite {
1243 file: request.file,
1244 offset: VfOffset::At(offset),
1245 data: &request.data[totals[index]..],
1246 })
1247 })
1248 .collect::<VfResult<_>>()?;
1249 let results = self.vwrite_native(&wave).map_err(|error| {
1250 error.map_index(|index| wave_indices.get(index).copied().unwrap_or(index))
1251 })?;
1252 for (&index, result) in wave_indices.iter().zip(results) {
1253 if result.written == 0 {
1254 return Err(VfError::client(index, crate::ERR_IO)
1255 .with_context("vwrite_all_native", project(&requests[index]).file.path()));
1256 }
1257 totals[index] += result.written;
1258 stable[index] &= result.stable;
1259 }
1260 pending.retain(|&index| totals[index] < project(&requests[index]).data.len());
1261 }
1262 requests
1263 .iter()
1264 .map(&project)
1265 .enumerate()
1266 .map(|(index, request)| {
1267 let VfOffset::At(offset) = request.offset else {
1268 return Err(VfError::client(index, crate::ERR_INVAL)
1269 .with_context("vwrite_all_native", request.file.path()));
1270 };
1271 Ok(FsWriteResult {
1272 offset,
1273 written: totals[index],
1274 stable: stable[index],
1275 })
1276 })
1277 .collect()
1278 }
1279
1280 fn write_ops<'a, 'b, T>(
1281 &self,
1282 requests: &'a [T],
1283 project: impl Fn(&T) -> FsWrite<'b, F>,
1284 ) -> VfResult<Vec<WriteOpRef<'a>>>
1285 where
1286 F: 'b,
1287 'b: 'a,
1288 {
1289 let mut writes = Vec::with_capacity(requests.len());
1290 for (index, request) in requests.iter().map(&project).enumerate() {
1291 self.validate_owner(request.file, index)?;
1292 writes.push(WriteOpRef::new(
1293 request
1294 .file
1295 .raw()
1296 .map_err(|error| error.with_index(index))?,
1297 request.offset,
1298 request.data,
1299 ));
1300 }
1301 Ok(writes)
1302 }
1303
1304 fn validate_owner(&self, file: &FsFile<F>, index: usize) -> VfResult<()> {
1305 if Arc::ptr_eq(&self.inner, &file.inner) {
1306 Ok(())
1307 } else {
1308 Err(VfError::client(index, crate::ERR_INVAL))
1309 }
1310 }
1311}
1312
1313impl<F: Backend> FsClient<F> {
1314 #[doc(hidden)]
1318 pub fn read_files_native<P: AsRef<Path>>(
1319 &self,
1320 paths: &[P],
1321 options: ReadAllOptions,
1322 ) -> VfResult<Vec<Vec<u8>>> {
1323 let files: Vec<_> = paths
1324 .iter()
1325 .map(|path| VfFile::from_os_path(path.as_ref()))
1326 .collect();
1327 let result = self
1328 .lock()?
1329 .vread_all_with_options_impl(&files, options)
1330 .map_err(|error| {
1331 error
1332 .index()
1333 .and_then(|index| paths.get(index))
1334 .map_or(error.clone(), |path| {
1335 error.with_context("read_files", path.as_ref())
1336 })
1337 });
1338 result.and_then(|buffers| {
1339 if buffers.len() != paths.len() {
1340 Err(wrong_result_count("read_files", paths.len(), buffers.len()))
1341 } else {
1342 Ok(buffers)
1343 }
1344 })
1345 }
1346}
1347
1348pub struct OpenOptions<'a, F: FileSystem> {
1350 client: &'a FsClient<F>,
1351 flags: OpenFlags,
1352 mode: u32,
1353}
1354
1355impl<F: FileSystem> Clone for OpenOptions<'_, F> {
1356 fn clone(&self) -> Self {
1357 Self {
1358 client: self.client,
1359 flags: self.flags,
1360 mode: self.mode,
1361 }
1362 }
1363}
1364
1365impl<F: FileSystem> fmt::Debug for OpenOptions<'_, F> {
1366 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
1367 formatter
1368 .debug_struct("OpenOptions")
1369 .field("flags", &self.flags)
1370 .field("mode", &format_args!("{:#o}", self.mode))
1371 .finish()
1372 }
1373}
1374
1375impl<'a, F: FileSystem> OpenOptions<'a, F> {
1376 fn new(client: &'a FsClient<F>) -> Self {
1377 Self {
1378 client,
1379 flags: OpenFlags::empty(),
1380 mode: 0o666,
1381 }
1382 }
1383
1384 pub fn read(&mut self, enabled: bool) -> &mut Self {
1385 self.flags.set(OpenFlags::READ, enabled);
1386 self
1387 }
1388
1389 pub fn write(&mut self, enabled: bool) -> &mut Self {
1390 self.flags.set(OpenFlags::WRITE, enabled);
1391 self
1392 }
1393
1394 pub fn append(&mut self, enabled: bool) -> &mut Self {
1395 self.flags.set(OpenFlags::APPEND, enabled);
1396 self
1397 }
1398
1399 pub fn truncate(&mut self, enabled: bool) -> &mut Self {
1400 self.flags.set(OpenFlags::TRUNCATE, enabled);
1401 self
1402 }
1403
1404 pub fn create(&mut self, enabled: bool) -> &mut Self {
1405 self.flags.set(OpenFlags::CREATE, enabled);
1406 self
1407 }
1408
1409 pub fn create_new(&mut self, enabled: bool) -> &mut Self {
1410 self.flags.set(OpenFlags::CREATE_NEW, enabled);
1411 self
1412 }
1413
1414 pub fn mode(&mut self, mode: u32) -> &mut Self {
1415 self.mode = mode;
1416 self
1417 }
1418
1419 pub fn open(&self, path: impl AsRef<Path>) -> VfResult<FsFile<F>> {
1420 self.client
1421 .open_with_native(OpenOp::new(path.as_ref(), self.flags).mode(self.mode))
1422 }
1423}
1424
1425impl<F: Backend> OpenOptions<'_, F> {
1426 pub fn vopen<P: AsRef<Path>>(&self, paths: &[P]) -> VfResult<Vec<FsFile<F>>> {
1427 let requests: Vec<OpenOp> = paths
1428 .iter()
1429 .map(|path| OpenOp::new(path.as_ref(), self.flags).mode(self.mode))
1430 .collect();
1431 self.client.vopen(&requests)
1432 }
1433}
1434
1435pub struct SetMetadata<'a, F: Backend> {
1437 client: &'a FsClient<F>,
1438 path: PathBuf,
1439 update: MetadataUpdate,
1440 follow: bool,
1441}
1442
1443impl<F: Backend> Clone for SetMetadata<'_, F> {
1444 fn clone(&self) -> Self {
1445 Self {
1446 client: self.client,
1447 path: self.path.clone(),
1448 update: self.update.clone(),
1449 follow: self.follow,
1450 }
1451 }
1452}
1453
1454impl<F: Backend> fmt::Debug for SetMetadata<'_, F> {
1455 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
1456 formatter
1457 .debug_struct("SetMetadata")
1458 .field("path", &self.path)
1459 .field("update", &self.update)
1460 .field("follow", &self.follow)
1461 .finish_non_exhaustive()
1462 }
1463}
1464
1465impl<F: Backend> SetMetadata<'_, F> {
1466 pub fn permissions(&mut self, permissions: Permissions) -> &mut Self {
1467 self.update.permissions = Some(permissions);
1468 self
1469 }
1470
1471 pub fn len(&mut self, len: u64) -> &mut Self {
1472 self.update.len = Some(len);
1473 self
1474 }
1475
1476 pub fn accessed(&mut self, accessed: SystemTime) -> &mut Self {
1477 self.update.accessed = Some(accessed);
1478 self
1479 }
1480
1481 pub fn modified(&mut self, modified: SystemTime) -> &mut Self {
1482 self.update.modified = Some(modified);
1483 self
1484 }
1485
1486 pub fn follow_symlinks(&mut self, follow: bool) -> &mut Self {
1487 self.follow = follow;
1488 self
1489 }
1490
1491 pub fn uid(&mut self, uid: u32) -> &mut Self {
1492 self.update.uid = Some(uid);
1493 self
1494 }
1495 pub fn gid(&mut self, gid: u32) -> &mut Self {
1496 self.update.gid = Some(gid);
1497 self
1498 }
1499 pub fn apply(&self) -> VfResult<()> {
1500 self.client
1501 .vsetattrs(&[(&self.path, self.update.clone())], self.follow)
1502 }
1503}
1504
1505pub struct FsFile<F: FileSystem> {
1511 inner: Arc<SharedBackend<F>>,
1512 file: Option<VfFile>,
1513 path: PathBuf,
1514}
1515
1516impl<F: FileSystem> fmt::Debug for FsFile<F> {
1517 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
1518 formatter
1519 .debug_struct("FsFile")
1520 .field("path", &self.path)
1521 .field("is_open", &self.file.is_some())
1522 .finish()
1523 }
1524}
1525
1526impl<F: FileSystem> FsFile<F> {
1527 pub fn is_closed(&self) -> bool {
1529 self.file.is_none()
1530 }
1531 fn raw(&self) -> VfResult<&VfFile> {
1532 self.file
1533 .as_ref()
1534 .ok_or_else(|| VfError::client(0, crate::ERR_EBADF))
1535 }
1536
1537 pub fn read_at(&self, buffer: &mut [u8], offset: u64) -> VfResult<usize> {
1539 self.read_into(buffer, VfOffset::At(offset))
1540 }
1541
1542 pub fn write_at(&self, buffer: &[u8], offset: u64) -> VfResult<usize> {
1544 self.write_from(buffer, VfOffset::At(offset))
1545 }
1546
1547 pub fn read_request_at(&self, offset: u64, length: usize) -> FsRead<'_, F> {
1548 FsRead {
1549 file: self,
1550 offset: VfOffset::At(offset),
1551 length,
1552 }
1553 }
1554
1555 pub fn path(&self) -> &Path {
1559 &self.path
1560 }
1561
1562 pub fn attrs(&self) -> VfResult<Attrs> {
1564 let attributes = AttrMask::MODE
1565 | AttrMask::SIZE
1566 | AttrMask::NLINK
1567 | AttrMask::FILEID
1568 | AttrMask::UID
1569 | AttrMask::GID
1570 | AttrMask::ATIME
1571 | AttrMask::MTIME
1572 | AttrMask::CTIME
1573 | AttrMask::CHANGE;
1574 self.inner
1575 .lock()
1576 .map_err(|_| poisoned())?
1577 .metadata_impl(MetadataQuery::new(self.raw()?.clone(), attributes))
1578 .map(vfsi_core::metadata_from_attrs)
1579 .map_err(|error| error.with_context("metadata", &self.path))
1580 }
1581
1582 pub fn truncate(&self, len: u64) -> VfResult<()> {
1584 self.set_metadata_update(MetadataUpdate::new().len(len))
1585 }
1586
1587 pub fn chmod(&self, permissions: Permissions) -> VfResult<()> {
1589 self.set_metadata_update(MetadataUpdate::new().permissions(permissions))
1590 }
1591
1592 fn set_metadata_update(&self, update: MetadataUpdate) -> VfResult<()> {
1593 let client = FsClient {
1594 inner: Arc::clone(&self.inner),
1595 limits: ResourceLimits::default(),
1596 };
1597 client.vsetattrs(&[(vfsi_core::MetadataTarget::File(self), update)], true)
1598 }
1599
1600 pub fn write_request_at<'a>(&'a self, offset: u64, data: &'a [u8]) -> FsWrite<'a, F> {
1601 FsWrite {
1602 file: self,
1603 offset: VfOffset::At(offset),
1604 data,
1605 }
1606 }
1607
1608 pub fn read_request_at_into<'a>(
1609 &'a self,
1610 offset: u64,
1611 buffer: &'a mut [u8],
1612 ) -> FsReadInto<'a, F> {
1613 FsReadInto {
1614 file: self,
1615 offset: VfOffset::At(offset),
1616 buffer,
1617 }
1618 }
1619
1620 pub fn read_native(&mut self, buffer: &mut [u8]) -> VfResult<usize> {
1621 self.read_into(buffer, VfOffset::Cur)
1622 }
1623
1624 pub fn read_to_end_with_limit(&mut self, max_bytes: usize) -> VfResult<Vec<u8>> {
1632 let mut data = Vec::new();
1633 loop {
1634 let remaining = max_bytes - data.len();
1635 if remaining == 0 {
1636 let mut probe = [0_u8; 1];
1637 if self.read_native(&mut probe)? == 0 {
1638 return Ok(data);
1639 }
1640 return Err(VfError::client(0, libc::EFBIG as u32)
1641 .with_context("read_to_end_with_limit", &self.path));
1642 }
1643 let chunk = remaining.min(crate::DEFAULT_READ_STREAM_CHUNK_BYTES);
1644 data.try_reserve(chunk).map_err(|_| {
1645 VfError::client(0, libc::ENOMEM as u32)
1646 .with_context("read_to_end_with_limit", &self.path)
1647 })?;
1648 let start = data.len();
1649 data.resize(start + chunk, 0);
1650 let count = self.read_native(&mut data[start..])?;
1651 data.truncate(start + count);
1652 if count == 0 {
1653 return Ok(data);
1654 }
1655 }
1656 }
1657
1658 pub fn write_native(&mut self, buffer: &[u8]) -> VfResult<usize> {
1659 self.write_from(buffer, VfOffset::Cur)
1660 }
1661
1662 fn read_into(&self, buffer: &mut [u8], offset: VfOffset) -> VfResult<usize> {
1663 if buffer.is_empty() {
1664 return Ok(0);
1665 }
1666 let request = ReadOp::new(self.raw()?.clone(), offset, buffer.len());
1667 let result = self
1668 .inner
1669 .lock()
1670 .map_err(|_| poisoned())?
1671 .read_into_impl(&request, buffer)
1672 .map_err(|error| error.with_context("read", &self.path))?;
1673 validate_read_into_results(
1674 "read",
1675 std::slice::from_ref(&request),
1676 std::slice::from_ref(&result),
1677 )
1678 .map_err(|error| error.with_context("read", &self.path))?;
1679 Ok(result.read)
1680 }
1681
1682 fn write_from(&self, buffer: &[u8], offset: VfOffset) -> VfResult<usize> {
1683 if buffer.is_empty() {
1684 return Ok(0);
1685 }
1686 let file = self.raw()?.clone();
1687 let result = self
1688 .inner
1689 .lock()
1690 .map_err(|_| poisoned())?
1691 .write_impl(WriteOpRef::new(&file, offset, buffer))
1692 .map_err(|error| error.with_context("write", &self.path))?;
1693 if result.written > buffer.len() {
1694 return Err(VfError::client(0, crate::ERR_IO).with_context("write", &self.path));
1695 }
1696 Ok(result.written)
1697 }
1698
1699 pub fn sync_data(&self) -> VfResult<()> {
1700 self.inner
1701 .lock()
1702 .map_err(|_| poisoned())?
1703 .sync_data(self.raw()?)
1704 .map_err(|error| error.with_context("sync_data", &self.path))
1705 }
1706
1707 pub fn sync_all(&self) -> VfResult<()> {
1708 self.inner
1709 .lock()
1710 .map_err(|_| poisoned())?
1711 .sync_all(self.raw()?)
1712 .map_err(|error| error.with_context("sync_all", &self.path))
1713 }
1714
1715 pub fn try_close(&mut self) -> VfResult<()> {
1721 let Some(file) = self.file.as_ref() else {
1722 return Ok(());
1723 };
1724 self.inner
1725 .lock()
1726 .map_err(|_| poisoned())?
1727 .close_impl(file)
1728 .map_err(|error| error.with_context("close", &self.path))?;
1729 self.file = None;
1730 Ok(())
1731 }
1732
1733 pub fn close(mut self) -> VfResult<()> {
1737 self.try_close()
1738 }
1739
1740 pub fn seek_native(&mut self, position: IoSeekFrom) -> VfResult<u64> {
1742 self.inner
1743 .lock()
1744 .map_err(|_| poisoned())?
1745 .seek_impl(self.raw()?, position)
1746 .map_err(|error| error.with_context("seek", &self.path))
1747 }
1748}
1749
1750impl<F: FileSystem> Read for FsFile<F> {
1751 fn read(&mut self, buffer: &mut [u8]) -> io::Result<usize> {
1752 self.read_native(buffer).map_err(io_error)
1753 }
1754}
1755
1756impl<F: FileSystem> Write for FsFile<F> {
1757 fn write(&mut self, buffer: &[u8]) -> io::Result<usize> {
1758 self.write_native(buffer).map_err(io_error)
1759 }
1760
1761 fn flush(&mut self) -> io::Result<()> {
1762 self.sync_data().map_err(io_error)
1763 }
1764}
1765
1766impl<F: FileSystem> Seek for FsFile<F> {
1767 fn seek(&mut self, position: IoSeekFrom) -> io::Result<u64> {
1768 self.seek_native(position).map_err(io_error)
1769 }
1770}
1771
1772pub struct FsRead<'a, F: FileSystem> {
1774 file: &'a FsFile<F>,
1775 offset: VfOffset,
1776 length: usize,
1777}
1778
1779pub struct FsWrite<'a, F: FileSystem> {
1781 file: &'a FsFile<F>,
1782 offset: VfOffset,
1783 data: &'a [u8],
1784}
1785
1786pub struct FsReadInto<'a, F: FileSystem> {
1788 file: &'a FsFile<F>,
1789 offset: VfOffset,
1790 buffer: &'a mut [u8],
1791}
1792
1793impl<F: FileSystem> Drop for FsFile<F> {
1794 fn drop(&mut self) {
1795 if let Some(file) = self.file.take() {
1796 self.inner.defer(
1797 CleanupTarget::File(file, std::mem::take(&mut self.path)),
1798 |backend, target| {
1799 let CleanupTarget::File(file, path) = target else {
1800 unreachable!()
1801 };
1802 backend
1803 .close_deferred(file)
1804 .map_err(|error| error.with_context("close", path))
1805 },
1806 );
1807 }
1808 }
1809}
1810
1811impl<F: FileSystem> FsClient<F> {
1812 pub(crate) fn vstatfs<P: vfsi_core::MetadataOperand<FsFile<F>>>(
1814 &self,
1815 targets: &[P],
1816 ) -> VfResult<Vec<crate::FilesystemStats>> {
1817 use vfsi_core::MetadataTarget;
1818 if targets.is_empty() {
1819 return Ok(Vec::new());
1820 }
1821 let mut files = Vec::with_capacity(targets.len());
1822 let mut paths = Vec::with_capacity(targets.len());
1823 for (index, target) in targets.iter().enumerate() {
1824 let (raw, path) = match target.metadata_target() {
1825 MetadataTarget::Path(path) => (VfFile::from_os_path(path), path),
1826 MetadataTarget::File(file) => {
1827 if !Arc::ptr_eq(&self.inner, &file.inner) {
1828 return Err(VfError::client(index, crate::ERR_INVAL)
1829 .with_context("vstatfs", &file.path));
1830 }
1831 (
1832 file.raw()
1833 .map_err(|e| e.with_index(index).with_context("vstatfs", &file.path))?
1834 .clone(),
1835 file.path.as_path(),
1836 )
1837 }
1838 };
1839 files.push(raw);
1840 paths.push(path);
1841 }
1842 let results = self
1843 .lock()?
1844 .vstatfs_impl(&files)
1845 .map_err(|error| match error.index() {
1846 Some(index) if index < paths.len() => error.with_context("vstatfs", paths[index]),
1847 Some(_) => {
1848 VfError::transport(None, "statfs backend returned an invalid error index")
1849 }
1850 None => error,
1851 })?;
1852 if results.len() != targets.len() {
1853 return Err(VfError::transport(
1854 None,
1855 "statfs backend returned an invalid result count",
1856 ));
1857 }
1858 Ok(results)
1859 }
1860 pub(crate) fn vsetattrs<P: vfsi_core::MetadataOperand<FsFile<F>>>(
1862 &self,
1863 updates: &[(P, MetadataUpdate)],
1864 follow_symlinks: bool,
1865 ) -> VfResult<()> {
1866 use vfsi_core::MetadataTarget;
1867 if updates.is_empty() {
1868 return Ok(());
1869 }
1870 let mut paths = Vec::with_capacity(updates.len());
1871 let mut attrs = Vec::with_capacity(updates.len());
1872 for (index, (target, update)) in updates.iter().enumerate() {
1873 let (raw, path) = match target.metadata_target() {
1874 MetadataTarget::Path(path) => (VfFile::from_os_path(path), path),
1875 MetadataTarget::File(file) => {
1876 if !Arc::ptr_eq(&self.inner, &file.inner) {
1877 return Err(VfError::client(index, crate::ERR_INVAL)
1878 .with_context("vsetattrs", &file.path));
1879 }
1880 (
1881 file.raw()
1882 .map_err(|e| e.with_index(index).with_context("vsetattrs", &file.path))?
1883 .clone(),
1884 file.path.as_path(),
1885 )
1886 }
1887 };
1888 if update.uid == Some(u32::MAX) || update.gid == Some(u32::MAX) {
1889 return Err(
1890 VfError::client(index, crate::ERR_INVAL).with_context("vsetattrs", path)
1891 );
1892 }
1893 paths.push(path);
1894 let mut attributes = crate::SetAttributes::new(raw);
1895 attributes.mode = update.permissions.map(crate::Permissions::mode);
1896 attributes.size = update.len;
1897 attributes.uid = update.uid;
1898 attributes.gid = update.gid;
1899 attributes.atime = update
1900 .accessed
1901 .map(crate::native::system_time_parts)
1902 .transpose()
1903 .map_err(|e| e.with_index(index).with_context("vsetattrs", path))?;
1904 attributes.mtime = update
1905 .modified
1906 .map(crate::native::system_time_parts)
1907 .transpose()
1908 .map_err(|e| e.with_index(index).with_context("vsetattrs", path))?;
1909 attrs.push(attributes);
1910 }
1911 self.lock()?
1912 .vsetattrs_impl(attrs, follow_symlinks)
1913 .map_err(|error| match error.index() {
1914 Some(index) if index < updates.len() => {
1915 error.with_context("vsetattrs", paths[index])
1916 }
1917 Some(_) => {
1918 VfError::transport(None, "setattrs backend returned an invalid error index")
1919 }
1920 None => error,
1921 })
1922 }
1923}
1924
1925fn link_error(error: VfError, paths: &[&Path], operation: &'static str) -> VfError {
1926 match error.index() {
1927 Some(index) if index < paths.len() => error.with_context(operation, paths[index]),
1928 Some(_) => VfError::transport(None, "link backend returned an invalid error index"),
1929 None => error,
1930 }
1931}