1use std::collections::{HashMap, HashSet};
4use std::fmt;
5use std::path::{Path, PathBuf};
6use std::sync::atomic::{AtomicBool, Ordering};
7use std::sync::{Arc, Mutex};
8
9use vfsi_core::api::internal::OwnedReadResult as FsReadResult;
10use vfsi_core::api::{
11 DirectoryListing, ReadIntoResult as FsReadIntoResult, ResourceLimits, StreamCompletion,
12 WriteResult as FsWriteResult,
13};
14
15use crate::backend::{HandleBackend, VectorBackend};
16use crate::traits::{validate_read_into_results, validate_read_results, validate_write_results};
17use crate::{
18 AttrMask, Attrs, Capabilities, DirEntry, OpenFlags, OpenOp, ReadAllOptions, ReadOp, ReadResult,
19 RemoveOptions, StreamOptions, VfDir, VfError, VfFile, VfOffset, VfResult, WriteOp, WriteResult,
20};
21
22fn read_result(result: ReadResult) -> FsReadResult {
23 FsReadResult {
24 offset: result.offset,
25 data: result.data,
26 eof: result.eof,
27 }
28}
29
30fn write_result(result: WriteResult) -> FsWriteResult {
31 FsWriteResult {
32 offset: result.offset,
33 written: result.written,
34 stable: result.stable,
35 }
36}
37
38fn poisoned() -> VfError {
39 VfError::client(0, crate::ERR_IO)
40}
41
42fn wrong_result_count(operation: &str, expected: usize, actual: usize) -> VfError {
43 VfError::transport_with_kind(
44 None,
45 crate::TransportKind::InvalidReply,
46 format!("{operation} backend returned {actual} results for {expected} requests"),
47 )
48}
49
50enum CleanupTarget {
51 File(VfFile, PathBuf),
52 Directory(VfDir, PathBuf),
53}
54struct PendingClose<F> {
55 target: CleanupTarget,
56 close: fn(&mut F, &CleanupTarget) -> VfResult<()>,
57}
58
59struct SharedBackend<F: HandleBackend> {
62 backend: Mutex<Option<F>>,
63 pending: Mutex<Vec<PendingClose<F>>>,
64 has_pending: AtomicBool,
65}
66impl<F: HandleBackend> SharedBackend<F> {
67 fn new(backend: F) -> Self {
68 Self {
69 backend: Mutex::new(Some(backend)),
70 pending: Mutex::new(Vec::new()),
71 has_pending: AtomicBool::new(false),
72 }
73 }
74 fn defer(&self, target: CleanupTarget, close: fn(&mut F, &CleanupTarget) -> VfResult<()>) {
75 self.pending
76 .lock()
77 .unwrap_or_else(|e| e.into_inner())
78 .push(PendingClose { target, close });
79 self.has_pending.store(true, Ordering::Release);
80 }
81 fn lock_without_cleanup(&self) -> Result<BackendGuard<'_, F>, ()> {
82 self.backend
83 .lock()
84 .map(|guard| BackendGuard {
85 guard: Some(guard),
86 pending: &self.pending,
87 has_pending: &self.has_pending,
88 })
89 .map_err(|_| ())
90 }
91 fn lock(&self) -> Result<BackendGuard<'_, F>, ()> {
92 let mut guard = self.lock_without_cleanup()?;
93 let _ = guard.drain_cleanup();
96 Ok(guard)
97 }
98}
99impl<F: HandleBackend> Drop for SharedBackend<F> {
100 fn drop(&mut self) {
101 if let Some(backend) = self
102 .backend
103 .get_mut()
104 .unwrap_or_else(|e| e.into_inner())
105 .as_mut()
106 {
107 let pending = self.pending.get_mut().unwrap_or_else(|e| e.into_inner());
108 for close in pending.drain(..) {
109 let _ = (close.close)(backend, &close.target);
110 }
111 let callbacks = backend.take_notifications();
112 for callback in callbacks {
113 let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(callback));
114 }
115 }
116 }
117}
118struct BackendGuard<'a, F: HandleBackend> {
119 guard: Option<std::sync::MutexGuard<'a, Option<F>>>,
120 pending: &'a Mutex<Vec<PendingClose<F>>>,
121 has_pending: &'a AtomicBool,
122}
123impl<F: HandleBackend> std::ops::Deref for BackendGuard<'_, F> {
124 type Target = F;
125 fn deref(&self) -> &F {
126 self.guard
127 .as_ref()
128 .expect("live guard")
129 .as_ref()
130 .expect("live backend")
131 }
132}
133impl<F: HandleBackend> std::ops::DerefMut for BackendGuard<'_, F> {
134 fn deref_mut(&mut self) -> &mut F {
135 self.guard
136 .as_mut()
137 .expect("live guard")
138 .as_mut()
139 .expect("live backend")
140 }
141}
142impl<F: HandleBackend> BackendGuard<'_, F> {
143 fn drain_cleanup(&mut self) -> VfResult<()> {
144 if !self.has_pending.swap(false, Ordering::AcqRel) {
145 return Ok(());
146 }
147 let pending = std::mem::take(&mut *self.pending.lock().unwrap_or_else(|e| e.into_inner()));
148 let mut failed = Vec::new();
149 let mut first = None;
150 for close in pending {
151 if let Err(error) = (close.close)(self, &close.target) {
152 first.get_or_insert(error);
153 failed.push(close);
154 }
155 }
156 if !failed.is_empty() {
157 self.pending
158 .lock()
159 .unwrap_or_else(|e| e.into_inner())
160 .extend(failed);
161 self.has_pending.store(true, Ordering::Release);
162 }
163 first.map_or(Ok(()), Err)
164 }
165}
166impl<F: HandleBackend> Drop for BackendGuard<'_, F> {
167 fn drop(&mut self) {
168 let callbacks = self.take_notifications();
169 drop(self.guard.take());
170 for callback in callbacks {
171 let _ = std::panic::catch_unwind(std::panic::AssertUnwindSafe(callback));
174 }
175 }
176}
177
178pub struct FsClient<F: HandleBackend> {
185 inner: Arc<SharedBackend<F>>,
186 limits: ResourceLimits,
187}
188
189impl<F: HandleBackend> Clone for FsClient<F> {
190 fn clone(&self) -> Self {
191 Self {
192 inner: Arc::clone(&self.inner),
193 limits: self.limits,
194 }
195 }
196}
197
198impl<F: HandleBackend> fmt::Debug for FsClient<F> {
199 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
200 formatter.debug_struct("FsClient").finish_non_exhaustive()
201 }
202}
203
204impl<F: HandleBackend> FsClient<F> {
205 pub fn new(filesystem: F) -> Self {
206 Self {
207 inner: Arc::new(SharedBackend::new(filesystem)),
208 limits: ResourceLimits::default(),
209 }
210 }
211
212 pub fn with_limits(mut self, limits: ResourceLimits) -> Self {
215 self.limits = limits;
216 self
217 }
218
219 pub fn limits(&self) -> ResourceLimits {
220 self.limits
221 }
222
223 fn lock(&self) -> VfResult<BackendGuard<'_, F>> {
224 self.inner.lock().map_err(|_| poisoned())
225 }
226
227 pub fn into_inner(self) -> Result<F, Self> {
228 if self.drain_cleanup().is_err() {
229 return Err(self);
230 }
231 match Arc::try_unwrap(self.inner) {
232 Ok(mut shared) => Ok(shared
233 .backend
234 .get_mut()
235 .unwrap_or_else(|e| e.into_inner())
236 .take()
237 .expect("owned backend")),
238 Err(inner) => Err(Self {
239 inner,
240 limits: self.limits,
241 }),
242 }
243 }
244
245 pub fn drain_cleanup(&self) -> VfResult<()> {
248 self.inner
249 .lock_without_cleanup()
250 .map_err(|_| poisoned())?
251 .drain_cleanup()
252 }
253}
254
255impl<F: HandleBackend> FsClient<F> {
256 pub(crate) fn capabilities(&self) -> VfResult<Capabilities> {
257 Ok(self.lock()?.capabilities())
258 }
259 pub(crate) fn open(&self, path: impl AsRef<Path>) -> VfResult<FsFile<F>> {
261 self.open_with_native(OpenOp::new(path.as_ref(), OpenFlags::READ))
262 }
263
264 pub(crate) fn open_with_native(&self, request: OpenOp) -> VfResult<FsFile<F>> {
265 vfsi_core::internal::validate_open_requests(std::slice::from_ref(&request))?;
266 let file = self.lock()?.open_impl(&request)?;
267 Ok(FsFile {
268 inner: Arc::clone(&self.inner),
269 file: Some(file),
270 append: request.flags().contains(OpenFlags::APPEND),
271 path: request.into_path(),
272 })
273 }
274}
275
276impl<F: HandleBackend> FsClient<F> {
277 pub(crate) fn read_stream_with_options(
284 &self,
285 path: impl AsRef<Path>,
286 options: StreamOptions,
287 mut callback: impl FnMut(u64, &[u8]) -> VfResult<std::ops::ControlFlow<()>>,
288 ) -> VfResult<StreamCompletion> {
289 let chunk_size = options.chunk_size_bytes();
290 let file = self.open(path)?;
291 let raw_file = file.raw()?.clone();
292 let operation = (|| -> VfResult<StreamCompletion> {
293 let mut offset = 0u64;
294 loop {
295 let request = ReadOp::at(raw_file.clone(), offset, chunk_size);
296 let result = {
297 let mut backend = self.lock()?;
298 let result = backend.read_impl(&request)?;
299 validate_read_results(
300 "read_stream",
301 std::slice::from_ref(&request),
302 std::slice::from_ref(&result),
303 )?;
304 result
305 };
306 let length = result.data.len();
307 if !result.data.is_empty() && callback(offset, &result.data)?.is_break() {
308 return Ok(StreamCompletion::Stopped {
309 next_offset: offset
310 .checked_add(length as u64)
311 .ok_or_else(|| VfError::client(0, crate::ERR_INVAL))?,
312 });
313 }
314 if result.eof {
315 return Ok(StreamCompletion::Complete);
316 }
317 offset = offset
318 .checked_add(length as u64)
319 .ok_or_else(|| VfError::client(0, libc::EOVERFLOW as u32))?;
320 }
321 })();
322 let close = file.close();
323 match operation {
324 Err(error) => Err(error),
325 Ok(completion) => close.map(|()| completion),
326 }
327 }
328}
329
330impl<F: VectorBackend> FsClient<F> {
331 fn removal_metadata(&self, path: &Path, operation: &'static str) -> VfResult<Attrs> {
332 let mut filesystem = self.lock()?;
333 let follow = !filesystem.capabilities().contains(Capabilities::LSTAT);
334 filesystem
335 .metadata_path_impl(path, follow)
336 .map_err(|error| error.with_context(operation, path))
337 }
338}
339
340impl<F: VectorBackend> FsClient<F> {
341 pub(crate) fn vrename<P: AsRef<Path>, Q: AsRef<Path>>(
345 &self,
346 pairs: &[(P, Q)],
347 options: vfsi_core::api::RenameOptions,
348 ) -> VfResult<()> {
349 if pairs.is_empty() {
350 return Ok(());
351 }
352 let requests: Vec<_> = pairs
353 .iter()
354 .map(|(from, to)| {
355 (
356 VfFile::from_os_path(from.as_ref()),
357 VfFile::from_os_path(to.as_ref()),
358 )
359 })
360 .collect();
361 self.lock()?
362 .vrename_with_options_impl(&requests, options)
363 .map_err(|error| match error.index() {
364 Some(index) if index < pairs.len() => {
365 error.with_context("vrename", pairs[index].0.as_ref())
366 }
367 Some(_) => {
368 VfError::transport(None, "rename backend returned an invalid error index")
369 }
370 None => error,
371 })
372 }
373
374 pub(crate) fn vmkdir<P: AsRef<Path>>(
376 &self,
377 directories: &[vfsi_core::MkDirOp<P>],
378 ) -> VfResult<()> {
379 let mut seen = HashSet::with_capacity(directories.len());
380 for (index, op) in directories.iter().enumerate() {
381 if !seen.insert(op.path()) {
382 return Err(
383 VfError::client(index, crate::ERR_INVAL).with_context("vmkdir", op.path())
384 );
385 }
386 }
387 if directories.is_empty() {
388 return Ok(());
389 }
390 let dirs: Vec<_> = directories
391 .iter()
392 .map(|op| crate::VfAttrs {
393 file: VfFile::from_os_path(op.path()),
394 masks: AttrMask::MODE,
395 mode: op.mode(),
396 ..Default::default()
397 })
398 .collect();
399 self.lock()?
400 .vmkdir_impl(&dirs)
401 .map_err(|error| match error.index() {
402 Some(index) if index < directories.len() => {
403 error.with_context("vmkdir", directories[index].path())
404 }
405 Some(_) => {
406 VfError::transport(None, "mkdir backend returned an invalid error index")
407 }
408 None => error,
409 })
410 }
411
412 pub(crate) fn vsymlink<P: AsRef<Path>, Q: AsRef<Path>>(
414 &self,
415 pairs: &[(P, Q)],
416 ) -> VfResult<()> {
417 if pairs.is_empty() {
418 return Ok(());
419 }
420 let targets: Vec<_> = pairs.iter().map(|(target, _)| target.as_ref()).collect();
421 let links: Vec<_> = pairs.iter().map(|(_, link)| link.as_ref()).collect();
422 self.lock()?
423 .vsymlink_impl(&targets, &links)
424 .map_err(|error| link_error(error, &links, "vsymlink"))
425 }
426
427 pub(crate) fn vreadlink<P: AsRef<Path>>(&self, paths: &[P]) -> VfResult<Vec<PathBuf>> {
429 if paths.is_empty() {
430 return Ok(Vec::new());
431 }
432 let paths: Vec<_> = paths.iter().map(AsRef::as_ref).collect();
433 let targets = self
434 .lock()?
435 .vreadlink_impl(&paths)
436 .map_err(|error| link_error(error, &paths, "vreadlink"))?;
437 if targets.len() != paths.len() {
438 return Err(VfError::transport(
439 None,
440 "readlink backend returned an invalid result count",
441 ));
442 }
443 Ok(targets
444 .into_iter()
445 .map(crate::backend::bytes_to_path)
446 .collect())
447 }
448
449 pub(crate) fn vhardlink<P: AsRef<Path>, Q: AsRef<Path>>(
451 &self,
452 pairs: &[(P, Q)],
453 ) -> VfResult<()> {
454 if pairs.is_empty() {
455 return Ok(());
456 }
457 let sources: Vec<_> = pairs.iter().map(|(source, _)| source.as_ref()).collect();
458 let links: Vec<_> = pairs.iter().map(|(_, link)| link.as_ref()).collect();
459 self.lock()?
460 .vhardlink_impl(&sources, &links)
461 .map_err(|error| link_error(error, &links, "vhardlink"))
462 }
463
464 #[doc(hidden)]
470 pub fn directory_page_batch_size(&self) -> VfResult<usize> {
471 Ok(self.lock()?.directory_page_batch_size().clamp(1, 32))
472 }
473 #[doc(hidden)]
475 pub fn read_dir_pages_with_fields(
476 &self,
477 paths: &[&Path],
478 fields: AttrMask,
479 cursors: Vec<Option<crate::DirPageCursor>>,
480 page_size: usize,
481 max_entries: usize,
482 follow_symlinks: bool,
483 ) -> VfResult<Vec<crate::DirectoryPage>> {
484 let pages = self.lock()?.vlistdir_pages_impl(
485 paths,
486 fields | AttrMask::MODE | AttrMask::SIZE,
487 cursors,
488 page_size,
489 max_entries,
490 follow_symlinks,
491 )?;
492 if pages.len() != paths.len() {
493 return Err(VfError::transport(
494 None,
495 "directory pages returned an invalid result count",
496 ));
497 }
498 pages
499 .into_iter()
500 .zip(paths)
501 .enumerate()
502 .map(|(index, ((attrs, next, children), path))| {
503 if children.len() != attrs.len()
504 || attrs.len() > page_size
505 || (attrs.is_empty() && next.is_some())
506 {
507 return Err(VfError::transport(
508 Some(index),
509 "invalid directory page progress",
510 ));
511 }
512 let mut seeds = Vec::new();
513 let entries = attrs
514 .into_iter()
515 .zip(children)
516 .map(|(attrs, child)| {
517 let entry_path = attrs
518 .file
519 .path()
520 .ok_or_else(|| {
521 VfError::transport(Some(index), "directory entry has no path")
522 })?
523 .to_path_buf();
524 if entry_path.parent() != Some(*path) {
525 return Err(VfError::transport(
526 Some(index),
527 "directory entry is outside its parent",
528 ));
529 }
530 if let Some(child) = child {
531 seeds.push((entry_path.clone(), child));
532 }
533 Ok(DirEntry::new(
534 entry_path,
535 vfsi_core::metadata_from_attrs(attrs),
536 ))
537 })
538 .collect::<VfResult<Vec<_>>>()?;
539 Ok((
540 DirectoryListing {
541 path: path.to_path_buf(),
542 entries,
543 },
544 next,
545 seeds,
546 ))
547 })
548 .collect()
549 }
550}
551
552impl<F: VectorBackend> FsClient<F> {
553 pub fn ensure_empty_dir(&self, path: impl AsRef<Path>) -> VfResult<()> {
556 self.lock()?.ensure_empty_dir_impl(path.as_ref())
557 }
558
559 pub(crate) fn remove_dir_contents_impl(
561 &self,
562 path: impl AsRef<Path>,
563 options: RemoveOptions,
564 ) -> VfResult<()> {
565 let path = path.as_ref();
566 if !self
567 .removal_metadata(path, "remove_dir_contents_impl")?
568 .is_dir()
569 {
570 return Err(
571 VfError::client(0, crate::ERR_NOTDIR).with_context("remove_dir_contents", path)
572 );
573 }
574 self.lock()?
575 .remove_dir_contents_path_with_options_impl(path, options)
576 .map_err(|error| error.with_context("remove_dir_contents", path))
577 }
578
579 pub(crate) fn vopen_dirs_impl<P: AsRef<Path>>(&self, paths: &[P]) -> VfResult<Vec<FsDir<F>>> {
580 if paths.is_empty() {
581 return Ok(Vec::new());
582 }
583 let mut backend = self.lock()?;
584 let mut output = Vec::with_capacity(paths.len());
585 for (index, path) in paths.iter().enumerate() {
586 let path = path.as_ref();
587 let dir = backend.open_dir_impl(path).map_err(|error| {
588 crate::application::vector_index(error, index).with_context("vopen_dirs", path)
589 })?;
590 if !matches!(dir, VfDir::Descriptor { .. }) {
591 let _ = backend.close_dir_impl(&dir);
592 return Err(VfError::unsupported(index).with_context("vopen_dirs", path));
593 }
594 output.push(FsDir {
595 inner: Arc::clone(&self.inner),
596 dir: Some(dir),
597 path: path.to_path_buf(),
598 });
599 }
600 Ok(output)
601 }
602
603 pub(crate) fn vremove_dir_contents_impl(
605 &self,
606 dirs: &[&FsDir<F>],
607 options: RemoveOptions,
608 ) -> VfResult<()> {
609 for (index, dir) in dirs.iter().enumerate() {
610 if !Arc::ptr_eq(&self.inner, &dir.inner) || dir.is_closed() {
611 return Err(VfError::client(index, crate::ERR_EBADF)
612 .with_context("vremove_dir_contents", &dir.path));
613 }
614 }
615 if dirs.is_empty() {
616 return Ok(());
617 }
618 let mut backend = self.lock()?;
619 let mut first_error = None;
620 for (index, dir) in dirs.iter().enumerate() {
621 if let Err(error) = backend.remove_dir_contents_handle_with_options_impl(
622 dir.dir.as_ref().expect("preflighted"),
623 options,
624 ) {
625 let error = crate::application::vector_index(error, index)
626 .with_context("vremove_dir_contents", &dir.path);
627 if error.is_transport() || !options.continues_on_error() {
628 return Err(error);
629 }
630 first_error.get_or_insert(error);
631 }
632 }
633 first_error.map_or(Ok(()), Err)
634 }
635}
636
637pub struct FsDir<F: VectorBackend> {
640 inner: Arc<SharedBackend<F>>,
641 dir: Option<VfDir>,
642 path: PathBuf,
643}
644
645impl<F: VectorBackend> fmt::Debug for FsDir<F> {
646 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
647 formatter
648 .debug_struct("FsDir")
649 .field("path", &self.path)
650 .finish_non_exhaustive()
651 }
652}
653
654impl<F: VectorBackend> FsDir<F> {
655 pub fn is_closed(&self) -> bool {
656 self.dir.is_none()
657 }
658 pub fn path(&self) -> &Path {
660 &self.path
661 }
662
663 pub fn close(mut self) -> VfResult<()> {
664 self.try_close()
665 }
666
667 pub fn try_close(&mut self) -> VfResult<()> {
669 let Some(dir) = self.dir.as_ref() else {
670 return Ok(());
671 };
672 self.inner
673 .lock()
674 .map_err(|_| poisoned())?
675 .close_dir_impl(dir)?;
676 self.dir = None;
677 Ok(())
678 }
679}
680
681impl<F: VectorBackend> Drop for FsDir<F> {
682 fn drop(&mut self) {
683 if let Some(dir) = self.dir.take() {
684 self.inner.defer(
685 CleanupTarget::Directory(dir, std::mem::take(&mut self.path)),
686 |backend, target| {
687 let CleanupTarget::Directory(dir, path) = target else {
688 unreachable!()
689 };
690 backend
691 .close_dir_impl(dir)
692 .map_err(|error| error.with_context("close_dir", path))
693 },
694 );
695 }
696 }
697}
698
699impl<F: VectorBackend> FsClient<F> {
700 pub(crate) fn vcopy<P: AsRef<Path>, Q: AsRef<Path>>(
703 &self,
704 pairs: &[(P, Q)],
705 options: vfsi_core::api::CopyOption,
706 ) -> VfResult<()> {
707 let extents: Vec<_> = pairs
708 .iter()
709 .map(|(from, to)| {
710 crate::ExtentPair::from_os_paths(from.as_ref(), 0, to.as_ref(), 0, None)
711 })
712 .collect();
713 self.lock()?.vcopy_impl(&extents, options).map_err(|error| {
714 error
715 .index()
716 .and_then(|index| pairs.get(index))
717 .map_or(error.clone(), |(_, to)| {
718 error.with_context("vcopy", to.as_ref())
719 })
720 })
721 }
722
723 #[doc(hidden)]
726 pub fn vremove_native<P: AsRef<Path>>(&self, paths: &[P], recursive: bool) -> VfResult<()> {
727 self.vremove_impl(paths, recursive, RemoveOptions::default())
728 }
729
730 #[doc(hidden)]
732 pub fn vremove_impl<P: AsRef<Path>>(
733 &self,
734 paths: &[P],
735 recursive: bool,
736 options: RemoveOptions,
737 ) -> VfResult<()> {
738 let paths: Vec<&Path> = paths.iter().map(AsRef::as_ref).collect();
739 self.lock()?
740 .remove_paths_with_options_impl(&paths, recursive, options)
741 .map_err(|error| {
742 error
743 .index()
744 .and_then(|index| paths.get(index))
745 .map_or(error.clone(), |path| {
746 error.with_context("vremove_native", path)
747 })
748 })
749 }
750
751 #[doc(hidden)]
754 pub fn vgetattrs_native<P: vfsi_core::AsTarget<FsFile<F>>>(
755 &self,
756 targets: &[P],
757 fields: AttrMask,
758 follow: bool,
759 ) -> VfResult<Vec<Attrs>> {
760 use vfsi_core::Target;
761 if targets.is_empty() {
762 return Ok(Vec::new());
763 }
764 let mut paths = Vec::with_capacity(targets.len());
765 let mut attrs = Vec::with_capacity(targets.len());
766 let mut policies = Vec::with_capacity(targets.len());
767 for (index, target) in targets.iter().enumerate() {
768 let (raw, path, follows) = match target.as_target() {
769 Target::Path(path) => (VfFile::from_os_path(path), path, follow),
770 Target::File(file) => {
771 self.validate_owner(file, index)
772 .map_err(|e| e.with_context("vgetattrs", &file.path))?;
773 (
774 file.raw()
775 .map_err(|e| e.with_index(index).with_context("vgetattrs", &file.path))?
776 .clone(),
777 file.path.as_path(),
778 true,
779 )
780 }
781 };
782 paths.push(path);
783 policies.push(follows);
784 attrs.push(crate::VfAttrs {
785 file: raw,
786 masks: fields | AttrMask::MODE,
787 ..crate::VfAttrs::default()
788 });
789 }
790 let mut backend = self.lock()?;
791 let mut start = 0;
792 while start < attrs.len() {
793 let end = start
794 + policies[start..]
795 .iter()
796 .take_while(|p| **p == policies[start])
797 .count();
798 let result = if policies[start] {
799 backend.vgetattrs_impl(&mut attrs[start..end])
800 } else {
801 backend.vgetattrs_nofollow_impl(&mut attrs[start..end])
802 };
803 result.map_err(|error| match error.index() {
804 Some(index) if index < end - start => error
805 .with_index(start + index)
806 .with_context("vgetattrs", paths[start + index]),
807 Some(_) => {
808 VfError::transport(None, "metadata backend returned an invalid error index")
809 }
810 None => error,
811 })?;
812 start = end;
813 }
814 Ok(attrs
815 .into_iter()
816 .map(vfsi_core::metadata_from_attrs)
817 .collect())
818 }
819}
820
821impl<F: VectorBackend> FsClient<F> {
822 pub(crate) fn vopen(&self, requests: &[OpenOp]) -> VfResult<Vec<FsFile<F>>> {
828 vfsi_core::internal::validate_open_requests(requests)?;
829 let mut filesystem = self.lock()?;
830 let files = filesystem.vopen_impl(requests).map_err(|error| {
831 error
832 .index()
833 .and_then(|index| requests.get(index))
834 .map_or(error.clone(), |request| {
835 error.with_context("vopen", request.path())
836 })
837 })?;
838 if files.len() != requests.len() {
839 let error = wrong_result_count("vopen", requests.len(), files.len());
840 for file in &files {
841 let _ = filesystem.close_impl(file);
842 }
843 return Err(error);
844 }
845 drop(filesystem);
846 Ok(files
847 .into_iter()
848 .zip(requests)
849 .map(|(file, request)| FsFile {
850 inner: Arc::clone(&self.inner),
851 file: Some(file),
852 append: request.flags().contains(OpenFlags::APPEND),
853 path: request.path().to_path_buf(),
854 })
855 .collect())
856 }
857
858 #[doc(hidden)]
862 pub fn vclose<'a>(&self, files: impl IntoIterator<Item = &'a mut FsFile<F>>) -> VfResult<()>
863 where
864 F: 'a,
865 {
866 let mut files: Vec<_> = files.into_iter().collect();
867 for (index, file) in files.iter().enumerate() {
868 self.validate_owner(file, index)?;
869 }
870 let positions: Vec<_> = files
871 .iter()
872 .enumerate()
873 .filter_map(|(index, file)| (!file.is_closed()).then_some(index))
874 .collect();
875 files.retain(|file| !file.is_closed());
876 if files.is_empty() {
877 return Ok(());
878 }
879 let descriptors: Vec<VfFile> = files
880 .iter()
881 .map(|file| file.raw().cloned())
882 .collect::<VfResult<_>>()?;
883 self.lock()?.vclose_impl(&descriptors).map_err(|error| {
884 error
885 .index()
886 .and_then(|index| files.get(index))
887 .map_or(error.clone(), |file| {
888 error.with_context("vclose_owned", file.path())
889 })
890 .map_index(|index| positions.get(index).copied().unwrap_or(index))
891 })?;
892 for file in &mut files {
893 file.file = None;
894 }
895 Ok(())
896 }
897
898 #[doc(hidden)]
904 pub fn vclose_owned(&self, mut files: Vec<FsFile<F>>) -> VfResult<()> {
905 self.vclose(&mut files)
906 }
907
908 #[doc(hidden)]
912 pub fn vread_native(&self, requests: &[FsRead<'_, F>]) -> VfResult<Vec<FsReadResult>> {
913 self.vread_with_limit_native(requests, self.limits.read_byte_limit())
914 }
915
916 #[doc(hidden)]
918 pub fn vread_with_limit_native(
919 &self,
920 requests: &[FsRead<'_, F>],
921 max_total_bytes: usize,
922 ) -> VfResult<Vec<FsReadResult>> {
923 self.vread_with_limit_projected_native(requests, max_total_bytes, |request| request)
924 }
925
926 #[doc(hidden)]
929 pub fn vread_with_limit_projected_native<'b, T>(
930 &self,
931 requests: &[T],
932 max_total_bytes: usize,
933 project: impl for<'r> Fn(&'r T) -> &'r FsRead<'b, F>,
934 ) -> VfResult<Vec<FsReadResult>>
935 where
936 F: 'b,
937 {
938 let mut requested = 0usize;
939 for (index, request) in requests.iter().map(&project).enumerate() {
940 requested = requested
941 .checked_add(request.length)
942 .ok_or_else(|| VfError::client(index, libc::EFBIG as u32))?;
943 if requested > max_total_bytes {
944 return Err(VfError::client(index, libc::EFBIG as u32)
945 .with_context("vread_native", request.file.path()));
946 }
947 }
948 let reads = self.read_ops(requests, &project)?;
949 let results = self.lock()?.vread_impl(&reads).map_err(|error| {
950 error
951 .index()
952 .and_then(|index| requests.get(index).map(&project))
953 .map_or(error.clone(), |request| {
954 error.with_context("vread_native", request.file.path())
955 })
956 })?;
957 if results.len() != requests.len() {
958 return Err(wrong_result_count(
959 "vread_native",
960 requests.len(),
961 results.len(),
962 ));
963 }
964 validate_read_results("vread_native", &reads, &results).map_err(|error| {
965 error
966 .index()
967 .and_then(|index| requests.get(index).map(&project))
968 .map_or(error.clone(), |request| {
969 error.with_context("vread_native", request.file.path())
970 })
971 })?;
972 Ok(results.into_iter().map(read_result).collect())
973 }
974
975 fn read_ops<'b, T>(
976 &self,
977 requests: &[T],
978 project: impl for<'r> Fn(&'r T) -> &'r FsRead<'b, F>,
979 ) -> VfResult<Vec<ReadOp>>
980 where
981 F: 'b,
982 {
983 let mut reads = Vec::with_capacity(requests.len());
984 for (index, request) in requests.iter().map(&project).enumerate() {
985 self.validate_owner(request.file, index)?;
986 reads.push(ReadOp::new(
987 request.file.raw()?.clone(),
988 request.offset,
989 request.length,
990 ));
991 }
992 Ok(reads)
993 }
994
995 #[doc(hidden)]
996 pub fn vread_into_native(
997 &self,
998 requests: &mut [FsReadInto<'_, F>],
999 ) -> VfResult<Vec<FsReadIntoResult>> {
1000 self.vread_into_with_limit_native(requests, self.limits.read_byte_limit())
1001 }
1002
1003 #[doc(hidden)]
1006 pub fn vread_into_with_limit_native(
1007 &self,
1008 requests: &mut [FsReadInto<'_, F>],
1009 max_bytes: usize,
1010 ) -> VfResult<Vec<FsReadIntoResult>> {
1011 let mut requested = 0usize;
1012 let mut reads = Vec::with_capacity(requests.len());
1013 for (index, request) in requests.iter().enumerate() {
1014 self.validate_owner(request.file, index)?;
1015 requested = requested
1016 .checked_add(request.buffer.len())
1017 .ok_or_else(|| VfError::client(index, libc::EFBIG as u32))?;
1018 if requested > max_bytes {
1019 return Err(VfError::client(index, libc::EFBIG as u32));
1020 }
1021 reads.push(ReadOp::new(
1022 request.file.raw()?.clone(),
1023 request.offset,
1024 request.buffer.len(),
1025 ));
1026 }
1027 let results = {
1028 let mut buffers: Vec<&mut [u8]> = requests
1029 .iter_mut()
1030 .map(|request| &mut *request.buffer)
1031 .collect();
1032 self.lock()?.vread_into_impl(&reads, &mut buffers)
1033 }
1034 .map_err(|error| {
1035 error
1036 .index()
1037 .and_then(|index| requests.get(index))
1038 .map_or(error.clone(), |request| {
1039 error.with_context("vread_into_native", request.file.path())
1040 })
1041 })?;
1042 validate_read_into_results("vread_into_native", &reads, &results).map_err(|error| {
1043 error
1044 .index()
1045 .and_then(|index| requests.get(index))
1046 .map_or(error.clone(), |request| {
1047 error.with_context("vread_into_native", request.file.path())
1048 })
1049 })?;
1050 Ok(results
1051 .into_iter()
1052 .map(|result| FsReadIntoResult {
1053 offset: result.offset,
1054 read: result.read,
1055 eof: result.eof,
1056 })
1057 .collect())
1058 }
1059
1060 #[doc(hidden)]
1061 pub fn vwrite_native(&self, requests: &[FsWrite<'_, F>]) -> VfResult<Vec<FsWriteResult>> {
1062 self.vwrite_projected_native(requests, |request| request)
1063 }
1064
1065 #[doc(hidden)]
1068 pub fn vwrite_projected_native<'b, T>(
1069 &self,
1070 requests: &[T],
1071 project: impl for<'r> Fn(&'r T) -> &'r FsWrite<'b, F>,
1072 ) -> VfResult<Vec<FsWriteResult>>
1073 where
1074 F: 'b,
1075 {
1076 self.vwrite_mapped_native(requests, |item| {
1077 let request = project(item);
1078 *request
1079 })
1080 }
1081
1082 #[doc(hidden)]
1085 #[doc(hidden)]
1086 pub fn vwrite_mapped_native<'b, T>(
1087 &self,
1088 requests: &[T],
1089 project: impl Fn(&T) -> FsWrite<'b, F>,
1090 ) -> VfResult<Vec<FsWriteResult>>
1091 where
1092 F: 'b,
1093 {
1094 let writes = self.write_ops(requests, &project)?;
1095 let results = self.lock()?.vwrite_impl(&writes).map_err(|error| {
1096 error
1097 .index()
1098 .and_then(|index| requests.get(index).map(&project))
1099 .map_or(error.clone(), |request| {
1100 error.with_context("vwrite_native", request.file().path())
1101 })
1102 })?;
1103 if results.len() != requests.len() {
1104 return Err(wrong_result_count(
1105 "vwrite_native",
1106 requests.len(),
1107 results.len(),
1108 ));
1109 }
1110 validate_write_results("vwrite_native", &writes, &results).map_err(|error| {
1111 error
1112 .index()
1113 .and_then(|index| requests.get(index).map(&project))
1114 .map_or(error.clone(), |request| {
1115 error.with_context("vwrite_native", request.file().path())
1116 })
1117 })?;
1118 Ok(results.into_iter().map(write_result).collect())
1119 }
1120
1121 #[doc(hidden)]
1127 pub fn vwrite_all_native(&self, requests: &[FsWrite<'_, F>]) -> VfResult<Vec<FsWriteResult>> {
1128 self.vwrite_all_projected_native(requests, |request| request)
1129 }
1130
1131 #[doc(hidden)]
1134 pub fn vwrite_all_projected_native<'b, T>(
1135 &self,
1136 requests: &[T],
1137 project: impl for<'r> Fn(&'r T) -> &'r FsWrite<'b, F>,
1138 ) -> VfResult<Vec<FsWriteResult>>
1139 where
1140 F: 'b,
1141 {
1142 self.vwrite_all_mapped_native(requests, |item| {
1143 let request = project(item);
1144 *request
1145 })
1146 }
1147
1148 #[doc(hidden)]
1151 #[doc(hidden)]
1152 pub fn vwrite_all_mapped_native<'b, T>(
1153 &self,
1154 requests: &[T],
1155 project: impl Fn(&T) -> FsWrite<'b, F>,
1156 ) -> VfResult<Vec<FsWriteResult>>
1157 where
1158 F: 'b,
1159 {
1160 let mut prior_by_path: HashMap<&Path, Vec<(usize, u64, u64, bool)>> = HashMap::new();
1163 let mut blocked_by = vec![Vec::new(); requests.len()];
1164 let mut offsets = Vec::with_capacity(requests.len());
1165 for (index, request) in requests.iter().map(&project).enumerate() {
1166 self.validate_owner(request.file(), index)?;
1167 request.file().raw().map_err(|error| {
1168 error
1169 .with_index(index)
1170 .with_context("vwrite_all_native", request.file().path())
1171 })?;
1172 let VfOffset::At(offset) = request.offset() else {
1173 return Err(VfError::client(index, crate::ERR_INVAL)
1174 .with_context("vwrite_all_native", request.file().path()));
1175 };
1176 let end = offset
1177 .checked_add(request.data().len() as u64)
1178 .ok_or_else(|| {
1179 VfError::client(index, libc::EOVERFLOW as u32)
1180 .with_context("vwrite_all_native", request.file().path())
1181 })?;
1182 offsets.push(offset);
1183 let prior = prior_by_path.entry(request.file().path()).or_default();
1188 for &(earlier, start, earlier_end, append) in prior.iter() {
1189 if append || request.file().append || (offset < earlier_end && start < end) {
1190 blocked_by[index].push(earlier);
1191 }
1192 }
1193 prior.push((index, offset, end, request.file().append));
1194 }
1195 let mut totals = vec![0usize; requests.len()];
1196 let mut stable = vec![true; requests.len()];
1197 let mut pending: Vec<usize> = (0..requests.len())
1198 .filter(|&index| !project(&requests[index]).data().is_empty())
1199 .collect();
1200 while !pending.is_empty() {
1201 let wave_indices: Vec<usize> = pending
1202 .iter()
1203 .copied()
1204 .filter(|&index| {
1205 blocked_by[index]
1206 .iter()
1207 .all(|&earlier| totals[earlier] == project(&requests[earlier]).data().len())
1208 })
1209 .collect();
1210 let wave: Vec<_> = wave_indices
1211 .iter()
1212 .map(|&index| {
1213 let request = project(&requests[index]);
1214 let VfOffset::At(offset) = request.offset() else {
1215 return Err(VfError::client(index, crate::ERR_INVAL)
1216 .with_context("vwrite_all_native", request.file().path()));
1217 };
1218 let offset = offset.checked_add(totals[index] as u64).ok_or_else(|| {
1219 VfError::client(index, libc::EOVERFLOW as u32)
1220 .with_context("vwrite_all_native", request.file().path())
1221 })?;
1222 Ok(FsWrite::new(
1223 request.file(),
1224 VfOffset::At(offset),
1225 &request.data()[totals[index]..],
1226 ))
1227 })
1228 .collect::<VfResult<_>>()?;
1229 let results = self.vwrite_native(&wave).map_err(|error| {
1230 error.map_index(|index| wave_indices.get(index).copied().unwrap_or(index))
1231 })?;
1232 for (&index, result) in wave_indices.iter().zip(results) {
1233 if result.written == 0 {
1234 return Err(VfError::client(index, crate::ERR_IO).with_context(
1235 "vwrite_all_native",
1236 project(&requests[index]).file().path(),
1237 ));
1238 }
1239 totals[index] += result.written;
1240 if project(&requests[index]).file().append {
1241 offsets[index] = result
1245 .offset
1246 .checked_add(result.written as u64)
1247 .and_then(|end| end.checked_sub(totals[index] as u64))
1248 .ok_or_else(|| {
1249 VfError::transport_with_kind(
1250 Some(index),
1251 crate::TransportKind::InvalidReply,
1252 "append completion offset cannot be represented",
1253 )
1254 })?;
1255 }
1256 stable[index] &= result.stable;
1257 }
1258 pending.retain(|&index| totals[index] < project(&requests[index]).data().len());
1259 }
1260 Ok(offsets
1261 .into_iter()
1262 .enumerate()
1263 .map(|(index, offset)| FsWriteResult {
1264 offset,
1265 written: totals[index],
1266 stable: stable[index],
1267 })
1268 .collect())
1269 }
1270
1271 fn write_ops<'a, 'b, T>(
1272 &self,
1273 requests: &'a [T],
1274 project: impl Fn(&T) -> FsWrite<'b, F>,
1275 ) -> VfResult<Vec<WriteOp<&'a VfFile, &'a [u8]>>>
1276 where
1277 F: 'b,
1278 'b: 'a,
1279 {
1280 let mut writes = Vec::with_capacity(requests.len());
1281 for (index, request) in requests.iter().map(&project).enumerate() {
1282 self.validate_owner(request.file(), index)?;
1283 writes.push(WriteOp::new(
1284 request
1285 .file()
1286 .raw()
1287 .map_err(|error| error.with_index(index))?,
1288 if request.file().append {
1291 VfOffset::End
1292 } else {
1293 request.offset()
1294 },
1295 request.data(),
1296 ));
1297 }
1298 Ok(writes)
1299 }
1300
1301 fn validate_owner(&self, file: &FsFile<F>, index: usize) -> VfResult<()> {
1302 if Arc::ptr_eq(&self.inner, &file.inner) {
1303 Ok(())
1304 } else {
1305 Err(VfError::client(index, crate::ERR_INVAL))
1306 }
1307 }
1308}
1309
1310impl<F: VectorBackend> FsClient<F> {
1311 #[doc(hidden)]
1315 pub fn read_files_native<P: AsRef<Path>>(
1316 &self,
1317 paths: &[P],
1318 options: ReadAllOptions,
1319 ) -> VfResult<Vec<Vec<u8>>> {
1320 let files: Vec<_> = paths
1321 .iter()
1322 .map(|path| VfFile::from_os_path(path.as_ref()))
1323 .collect();
1324 let result = self
1325 .lock()?
1326 .vread_all_with_options_impl(&files, options)
1327 .map_err(|error| {
1328 error
1329 .index()
1330 .and_then(|index| paths.get(index))
1331 .map_or(error.clone(), |path| {
1332 error.with_context("read_files", path.as_ref())
1333 })
1334 });
1335 result.and_then(|buffers| {
1336 if buffers.len() != paths.len() {
1337 Err(wrong_result_count("read_files", paths.len(), buffers.len()))
1338 } else {
1339 Ok(buffers)
1340 }
1341 })
1342 }
1343}
1344
1345pub struct FsFile<F: HandleBackend> {
1352 inner: Arc<SharedBackend<F>>,
1353 file: Option<VfFile>,
1354 append: bool,
1356 path: PathBuf,
1357}
1358
1359impl<F: HandleBackend> fmt::Debug for FsFile<F> {
1360 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
1361 formatter
1362 .debug_struct("FsFile")
1363 .field("path", &self.path)
1364 .field("is_open", &self.file.is_some())
1365 .finish()
1366 }
1367}
1368
1369impl<F: HandleBackend> FsFile<F> {
1370 pub fn is_closed(&self) -> bool {
1372 self.file.is_none()
1373 }
1374 fn raw(&self) -> VfResult<&VfFile> {
1375 self.file
1376 .as_ref()
1377 .ok_or_else(|| VfError::client(0, crate::ERR_EBADF))
1378 }
1379
1380 pub fn read_request_at(&self, offset: u64, length: usize) -> FsRead<'_, F> {
1381 FsRead {
1382 file: self,
1383 offset: VfOffset::At(offset),
1384 length,
1385 }
1386 }
1387
1388 pub fn path(&self) -> &Path {
1392 &self.path
1393 }
1394
1395 pub fn write_request_at<'a>(&'a self, offset: u64, data: &'a [u8]) -> FsWrite<'a, F> {
1396 FsWrite::new(self, VfOffset::At(offset), data)
1397 }
1398
1399 pub fn read_request_at_into<'a>(
1400 &'a self,
1401 offset: u64,
1402 buffer: &'a mut [u8],
1403 ) -> FsReadInto<'a, F> {
1404 FsReadInto {
1405 file: self,
1406 offset: VfOffset::At(offset),
1407 buffer,
1408 }
1409 }
1410
1411 pub fn try_close(&mut self) -> VfResult<()> {
1417 let Some(file) = self.file.as_ref() else {
1418 return Ok(());
1419 };
1420 self.inner
1421 .lock()
1422 .map_err(|_| poisoned())?
1423 .close_impl(file)
1424 .map_err(|error| error.with_context("close", &self.path))?;
1425 self.file = None;
1426 Ok(())
1427 }
1428
1429 pub fn close(mut self) -> VfResult<()> {
1433 self.try_close()
1434 }
1435}
1436
1437pub struct FsRead<'a, F: HandleBackend> {
1439 file: &'a FsFile<F>,
1440 offset: VfOffset,
1441 length: usize,
1442}
1443
1444pub type FsWrite<'a, F> = vfsi_core::internal::WriteRequest<&'a FsFile<F>, &'a [u8], VfOffset, ()>;
1446
1447pub struct FsReadInto<'a, F: HandleBackend> {
1449 file: &'a FsFile<F>,
1450 offset: VfOffset,
1451 buffer: &'a mut [u8],
1452}
1453
1454impl<F: HandleBackend> Drop for FsFile<F> {
1455 fn drop(&mut self) {
1456 if let Some(file) = self.file.take() {
1457 self.inner.defer(
1458 CleanupTarget::File(file, std::mem::take(&mut self.path)),
1459 |backend, target| {
1460 let CleanupTarget::File(file, path) = target else {
1461 unreachable!()
1462 };
1463 backend
1464 .close_deferred(file)
1465 .map_err(|error| error.with_context("close", path))
1466 },
1467 );
1468 }
1469 }
1470}
1471
1472impl<F: HandleBackend> FsClient<F> {
1473 pub(crate) fn vfsync_impl(
1475 &self,
1476 files: &[&FsFile<F>],
1477 mode: vfsi_core::api::SyncMode,
1478 ) -> VfResult<()> {
1479 if files.is_empty() {
1480 return Ok(());
1481 }
1482 let mut raw = Vec::with_capacity(files.len());
1483 for (index, file) in files.iter().enumerate() {
1484 if !Arc::ptr_eq(&self.inner, &file.inner) {
1485 return Err(
1486 VfError::client(index, crate::ERR_INVAL).with_context("vfsync", &file.path)
1487 );
1488 }
1489 raw.push(
1490 file.raw()
1491 .map_err(|e| e.with_index(index).with_context("vfsync", &file.path))?
1492 .clone(),
1493 );
1494 }
1495 self.lock()?
1496 .vfsync_impl(&raw, mode)
1497 .map_err(|error| match error.index() {
1498 Some(index) if index < files.len() => {
1499 error.with_context("vfsync", &files[index].path)
1500 }
1501 Some(_) => {
1502 VfError::transport(None, "fsync backend returned an invalid error index")
1503 }
1504 None => error,
1505 })
1506 }
1507 pub(crate) fn vstatfs<P: vfsi_core::AsTarget<FsFile<F>>>(
1509 &self,
1510 targets: &[P],
1511 ) -> VfResult<Vec<crate::FilesystemStats>> {
1512 use vfsi_core::Target;
1513 if targets.is_empty() {
1514 return Ok(Vec::new());
1515 }
1516 let mut files = Vec::with_capacity(targets.len());
1517 let mut paths = Vec::with_capacity(targets.len());
1518 for (index, target) in targets.iter().enumerate() {
1519 let (raw, path) = match target.as_target() {
1520 Target::Path(path) => (VfFile::from_os_path(path), path),
1521 Target::File(file) => {
1522 if !Arc::ptr_eq(&self.inner, &file.inner) {
1523 return Err(VfError::client(index, crate::ERR_INVAL)
1524 .with_context("vstatfs", &file.path));
1525 }
1526 (
1527 file.raw()
1528 .map_err(|e| e.with_index(index).with_context("vstatfs", &file.path))?
1529 .clone(),
1530 file.path.as_path(),
1531 )
1532 }
1533 };
1534 files.push(raw);
1535 paths.push(path);
1536 }
1537 let results = self
1538 .lock()?
1539 .vstatfs_impl(&files)
1540 .map_err(|error| match error.index() {
1541 Some(index) if index < paths.len() => error.with_context("vstatfs", paths[index]),
1542 Some(_) => {
1543 VfError::transport(None, "statfs backend returned an invalid error index")
1544 }
1545 None => error,
1546 })?;
1547 if results.len() != targets.len() {
1548 return Err(VfError::transport(
1549 None,
1550 "statfs backend returned an invalid result count",
1551 ));
1552 }
1553 Ok(results)
1554 }
1555 pub(crate) fn vsetattrs<P: vfsi_core::AsTarget<FsFile<F>>>(
1557 &self,
1558 updates: &[vfsi_core::SetAttrsOp<P>],
1559 ) -> VfResult<()> {
1560 use vfsi_core::Target;
1561 if updates.is_empty() {
1562 return Ok(());
1563 }
1564 let mut paths = Vec::with_capacity(updates.len());
1565 let mut attrs = Vec::with_capacity(updates.len());
1566 for (index, op) in updates.iter().enumerate() {
1567 let target = op.target();
1568 let (raw, path) = match target.as_target() {
1569 Target::Path(path) => (Target::Path(path), path),
1570 Target::File(file) => {
1571 if !Arc::ptr_eq(&self.inner, &file.inner) {
1572 return Err(VfError::client(index, crate::ERR_INVAL)
1573 .with_context("vsetattrs", &file.path));
1574 }
1575 (
1576 Target::File(file.raw().map_err(|e| {
1577 e.with_index(index).with_context("vsetattrs", &file.path)
1578 })?),
1579 file.path.as_path(),
1580 )
1581 }
1582 };
1583 if op.requested_uid() == Some(u32::MAX) || op.requested_gid() == Some(u32::MAX) {
1584 return Err(
1585 VfError::client(index, crate::ERR_INVAL).with_context("vsetattrs", path)
1586 );
1587 }
1588 paths.push(path);
1589 for time in [op.requested_accessed(), op.requested_modified()]
1590 .into_iter()
1591 .flatten()
1592 {
1593 crate::backend::system_time_parts(time)
1594 .map_err(|e| e.with_index(index).with_context("vsetattrs", path))?;
1595 }
1596 attrs.push(op.with_target(raw));
1597 }
1598 let map_error = |error: VfError, start: usize, count: usize| match error.index() {
1599 Some(index) if index < count => error
1600 .with_index(start + index)
1601 .with_context("vsetattrs", paths[start + index]),
1602 Some(_) => VfError::transport(None, "setattrs backend returned an invalid error index"),
1603 None => error,
1604 };
1605 let mut backend = self.lock()?;
1606 let follow = updates[0].follows_symlinks();
1607 if updates.iter().all(|op| op.follows_symlinks() == follow) {
1608 return backend
1609 .vsetattrs_impl(&attrs)
1610 .map_err(|error| map_error(error, 0, updates.len()));
1611 }
1612 let mut start = 0;
1613 while start < updates.len() {
1614 let follow = updates[start].follows_symlinks();
1615 let count = updates[start..]
1616 .iter()
1617 .take_while(|op| op.follows_symlinks() == follow)
1618 .count();
1619 backend
1620 .vsetattrs_impl(&attrs[start..start + count])
1621 .map_err(|error| map_error(error, start, count))?;
1622 start += count;
1623 }
1624 Ok(())
1625 }
1626}
1627
1628fn link_error(error: VfError, paths: &[&Path], operation: &'static str) -> VfError {
1629 match error.index() {
1630 Some(index) if index < paths.len() => error.with_context(operation, paths[index]),
1631 Some(_) => VfError::transport(None, "link backend returned an invalid error index"),
1632 None => error,
1633 }
1634}