1use std::collections::{HashMap, HashSet};
4use std::fmt;
5use std::io::{self, Read, Seek, SeekFrom as IoSeekFrom, Write};
6use std::path::{Component, Path, PathBuf};
7use std::sync::{Arc, Mutex};
8use std::time::SystemTime;
9
10use crate::traits::{validate_read_into_results, validate_read_results, validate_write_results};
11use crate::{
12 AttrMask, Capabilities, CopyFileSystem, DEFAULT_READ_MAX_BYTES, DirEntry, DirectoryFileSystem,
13 FileSystem, LinkFileSystem, Metadata, MetadataFileSystem, MetadataQuery, MetadataUpdate,
14 NamespaceFileSystem, OpenFlags, OpenRequest, Permissions, ReadAllOptions, ReadDirOptions,
15 ReadOp, ReadResult, ReadStreamOptions, SetAttributes, VecFs, VectorFileSystem, VfDir, VfError,
16 VfFile, VfOffset, VfResult, WriteOpRef, WriteResult,
17};
18
19pub const DEFAULT_READV_MAX_TOTAL_BYTES: usize = DEFAULT_READ_MAX_BYTES;
21
22#[derive(Clone, Debug, PartialEq, Eq)]
24pub struct DirectoryListing {
25 pub path: PathBuf,
26 pub entries: Vec<DirEntry>,
27}
28
29fn io_error(error: VfError) -> io::Error {
30 error.into()
31}
32
33fn poisoned() -> VfError {
34 VfError::client(0, crate::ERR_IO)
35}
36
37fn wrong_result_count(operation: &str, expected: usize, actual: usize) -> VfError {
38 VfError::transport(
39 None,
40 format!("{operation} backend returned {actual} results for {expected} requests"),
41 )
42}
43
44pub struct FsClient<F> {
46 inner: Arc<Mutex<F>>,
47}
48
49impl<F> Clone for FsClient<F> {
50 fn clone(&self) -> Self {
51 Self {
52 inner: Arc::clone(&self.inner),
53 }
54 }
55}
56
57impl<F> fmt::Debug for FsClient<F> {
58 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
59 formatter.debug_struct("FsClient").finish_non_exhaustive()
60 }
61}
62
63impl<F> FsClient<F> {
64 pub fn new(filesystem: F) -> Self {
65 Self {
66 inner: Arc::new(Mutex::new(filesystem)),
67 }
68 }
69
70 fn lock(&self) -> VfResult<std::sync::MutexGuard<'_, F>> {
71 self.inner.lock().map_err(|_| poisoned())
72 }
73
74 pub fn into_inner(self) -> Result<F, Self> {
75 match Arc::try_unwrap(self.inner) {
76 Ok(mutex) => match mutex.into_inner() {
77 Ok(filesystem) => Ok(filesystem),
78 Err(poisoned) => Ok(poisoned.into_inner()),
79 },
80 Err(inner) => Err(Self { inner }),
81 }
82 }
83}
84
85impl<F: FileSystem> FsClient<F> {
86 pub fn open(&self, path: impl AsRef<Path>) -> VfResult<FsFile<F>> {
88 self.open_with(OpenRequest::new(path.as_ref(), OpenFlags::READ))
89 }
90
91 pub fn open_with(&self, request: OpenRequest) -> VfResult<FsFile<F>> {
92 let file = self.lock()?.open_one(&request)?;
93 Ok(FsFile {
94 inner: Arc::clone(&self.inner),
95 file: Some(file),
96 path: request.path,
97 })
98 }
99
100 pub fn open_options(&self) -> OpenOptions<'_, F> {
101 OpenOptions::new(self)
102 }
103
104 pub fn create(&self, path: impl AsRef<Path>) -> VfResult<FsFile<F>> {
105 self.open_with(OpenRequest::new(
106 path.as_ref(),
107 OpenFlags::WRITE | OpenFlags::CREATE | OpenFlags::TRUNCATE,
108 ))
109 }
110
111 pub fn read(&self, path: impl AsRef<Path>) -> VfResult<Vec<u8>> {
112 self.read_with_limit(path, DEFAULT_READ_MAX_BYTES)
113 }
114
115 pub fn read_with_limit(&self, path: impl AsRef<Path>, max_bytes: usize) -> VfResult<Vec<u8>> {
120 let path = path.as_ref();
121 let mut file = self.open(path)?;
122 let mut output = Vec::new();
123 let mut chunk = vec![0; 64 * 1024];
124 loop {
125 let remaining = max_bytes.saturating_sub(output.len());
126 let request = if remaining == 0 {
127 1
128 } else {
129 remaining.min(chunk.len())
130 };
131 let read = file.read_native(&mut chunk[..request])?;
132 if read == 0 {
133 break;
134 }
135 if read > remaining {
136 return Err(VfError::failure(0, libc::EFBIG as u32).with_context("read", path));
137 }
138 output.extend_from_slice(&chunk[..read]);
139 }
140 file.close()?;
141 Ok(output)
142 }
143
144 pub fn read_to_string(&self, path: impl AsRef<Path>) -> VfResult<String> {
145 self.read_to_string_with_limit(path, DEFAULT_READ_MAX_BYTES)
146 }
147
148 pub fn read_to_string_with_limit(
150 &self,
151 path: impl AsRef<Path>,
152 max_bytes: usize,
153 ) -> VfResult<String> {
154 let path = path.as_ref();
155 String::from_utf8(self.read_with_limit(path, max_bytes)?)
156 .map_err(|_| VfError::client(0, crate::ERR_INVAL).with_context("read_to_string", path))
157 }
158
159 pub fn write(&self, path: impl AsRef<Path>, data: &[u8]) -> VfResult<()> {
160 let path = path.as_ref();
161 let mut file = self.create(path)?;
162 let mut written = 0;
163 while written < data.len() {
164 let count = file.write_native(&data[written..])?;
165 if count == 0 {
166 return Err(VfError::client(0, crate::ERR_IO).with_context("write", path));
167 }
168 written += count;
169 }
170 file.close()
171 }
172}
173
174impl<F: FileSystem> FsClient<F> {
175 pub fn read_stream(
183 &self,
184 path: impl AsRef<Path>,
185 callback: impl FnMut(u64, &[u8]) -> VfResult<bool>,
186 ) -> VfResult<()> {
187 self.read_stream_with_options(path, ReadStreamOptions::default(), callback)
188 }
189
190 pub fn read_stream_with_options(
192 &self,
193 path: impl AsRef<Path>,
194 options: ReadStreamOptions,
195 mut callback: impl FnMut(u64, &[u8]) -> VfResult<bool>,
196 ) -> VfResult<()> {
197 let chunk_size = options.chunk_size_bytes();
198 if chunk_size == 0 {
199 return Err(VfError::failure(0, crate::ERR_INVAL));
200 }
201
202 let file = self.open(path)?;
203 let raw_file = file.raw()?.clone();
204 let operation = (|| -> VfResult<()> {
205 let mut offset = 0u64;
206 loop {
207 let request = ReadOp::at(raw_file.clone(), offset, chunk_size);
208 let result = {
209 let mut backend = self.lock()?;
210 let result = backend.read_one(&request)?;
211 validate_read_results(
212 "read_stream",
213 std::slice::from_ref(&request),
214 std::slice::from_ref(&result),
215 )?;
216 result
217 };
218 let length = result.data.len();
219 if !result.data.is_empty() && !callback(offset, &result.data)? {
220 return Ok(());
221 }
222 if result.eof {
223 return Ok(());
224 }
225 offset = offset
226 .checked_add(length as u64)
227 .ok_or_else(|| VfError::client(0, libc::EOVERFLOW as u32))?;
228 }
229 })();
230 let close = file.close();
231 match operation {
232 Err(error) => Err(error),
233 Ok(()) => close,
234 }
235 }
236}
237
238impl<F: MetadataFileSystem> FsClient<F> {
239 pub fn metadata(&self, path: impl AsRef<Path>) -> VfResult<Metadata> {
240 self.lock()?.metadata_path(path.as_ref(), true)
241 }
242
243 pub fn symlink_metadata(&self, path: impl AsRef<Path>) -> VfResult<Metadata> {
244 self.lock()?.metadata_path(path.as_ref(), false)
245 }
246
247 pub fn set_metadata(&self, path: impl AsRef<Path>) -> SetMetadata<'_, F> {
248 SetMetadata {
249 client: self,
250 path: path.as_ref().to_path_buf(),
251 update: MetadataUpdate::new(),
252 follow: true,
253 }
254 }
255}
256
257impl<F: DirectoryFileSystem> FsClient<F> {
258 pub fn create_dir(&self, path: impl AsRef<Path>) -> VfResult<()> {
259 self.create_dir_with_mode(path, 0o777)
260 }
261
262 pub fn create_dir_with_mode(&self, path: impl AsRef<Path>, mode: u32) -> VfResult<()> {
263 self.lock()?.create_dir_one(path.as_ref(), mode)
264 }
265
266 pub fn read_dir(&self, path: impl AsRef<Path>) -> VfResult<Vec<DirEntry>> {
267 self.read_dir_with_options(path, ReadDirOptions::default())
268 }
269
270 pub fn read_dir_with_options(
272 &self,
273 path: impl AsRef<Path>,
274 options: ReadDirOptions,
275 ) -> VfResult<Vec<DirEntry>> {
276 self.lock()?.read_dir_one(path.as_ref(), options)
277 }
278
279 pub fn visit_dir(
283 &self,
284 path: impl AsRef<Path>,
285 callback: impl FnMut(DirEntry) -> VfResult<bool>,
286 ) -> VfResult<()> {
287 self.visit_dir_with_options(path, ReadDirOptions::default(), callback)
288 }
289
290 pub fn visit_dir_with_options(
292 &self,
293 path: impl AsRef<Path>,
294 options: ReadDirOptions,
295 mut callback: impl FnMut(DirEntry) -> VfResult<bool>,
296 ) -> VfResult<()> {
297 self.lock()?
298 .visit_dir_one(path.as_ref(), options, &mut callback)
299 }
300}
301
302impl<F: DirectoryFileSystem + MetadataFileSystem> FsClient<F> {
303 pub fn create_dir_all(&self, path: impl AsRef<Path>) -> VfResult<()> {
304 let path = path.as_ref();
305 let mut current = PathBuf::new();
306 for component in path.components() {
307 match component {
308 Component::RootDir => {
309 current.push(Path::new("/"));
310 continue;
311 }
312 Component::Normal(part) => current.push(part),
313 Component::CurDir => continue,
314 Component::ParentDir => {
315 current.push("..");
320 continue;
321 }
322 Component::Prefix(_) => return Err(VfError::client(0, crate::ERR_INVAL)),
323 }
324 match self.create_dir(¤t) {
325 Ok(()) => {}
326 Err(error) if error.err_no() == crate::ERR_EXIST => {
327 if !self.metadata(¤t)?.is_dir() {
328 return Err(VfError::client(0, crate::ERR_NOTDIR)
329 .with_context("create_dir_all", ¤t));
330 }
331 }
332 Err(error) => return Err(error),
333 }
334 }
335 Ok(())
336 }
337}
338
339impl<F: NamespaceFileSystem + MetadataFileSystem> FsClient<F> {
340 fn removal_metadata(&self, path: &Path, operation: &'static str) -> VfResult<Metadata> {
341 let mut filesystem = self.lock()?;
342 let follow = !filesystem.capabilities().contains(Capabilities::LSTAT);
343 filesystem
344 .metadata_path(path, follow)
345 .map_err(|error| error.with_context(operation, path))
346 }
347
348 pub fn remove_file(&self, path: impl AsRef<Path>) -> VfResult<()> {
349 let path = path.as_ref();
350 if self.removal_metadata(path, "remove_file")?.is_dir() {
351 return Err(VfError::client(0, crate::ERR_ISDIR).with_context("remove_file", path));
352 }
353 self.lock()?.remove_one(path, false)
354 }
355
356 pub fn remove_dir(&self, path: impl AsRef<Path>) -> VfResult<()> {
357 let path = path.as_ref();
358 if !self.removal_metadata(path, "remove_dir")?.is_dir() {
359 return Err(VfError::client(0, crate::ERR_NOTDIR).with_context("remove_dir", path));
360 }
361 self.lock()?.remove_one(path, false)
362 }
363
364 pub fn remove_dir_all(&self, path: impl AsRef<Path>) -> VfResult<()> {
365 let path = path.as_ref();
366 if !self.removal_metadata(path, "remove_dir_all")?.is_dir() {
367 return Err(VfError::client(0, crate::ERR_NOTDIR).with_context("remove_dir_all", path));
368 }
369 self.lock()?.remove_one(path, true)
370 }
371
372 pub fn remove_dir_contents(&self, path: impl AsRef<Path>) -> VfResult<()> {
374 let path = path.as_ref();
375 if !self.removal_metadata(path, "remove_dir_contents")?.is_dir() {
376 return Err(
377 VfError::client(0, crate::ERR_NOTDIR).with_context("remove_dir_contents", path)
378 );
379 }
380 self.lock()?.remove_dir_contents(path)
381 }
382
383 pub fn rename(&self, from: impl AsRef<Path>, to: impl AsRef<Path>) -> VfResult<()> {
384 self.lock()?.rename_one(from.as_ref(), to.as_ref())
385 }
386}
387
388impl<F: VecFs> FsClient<F> {
389 pub fn read_dirs<P: AsRef<Path>>(&self, paths: &[P]) -> VfResult<Vec<DirectoryListing>> {
392 self.read_dirs_with_options(paths, AttrMask::stat(), ReadDirOptions::default())
393 }
394
395 pub fn walk(&self, root: impl AsRef<Path>) -> VfResult<Vec<DirectoryListing>> {
398 self.walk_with_options(root, AttrMask::stat(), crate::WalkOptions::default())
399 }
400
401 pub fn ensure_empty_dir(&self, path: impl AsRef<Path>) -> VfResult<()> {
404 self.lock()?.ensure_empty_dir(path.as_ref())
405 }
406
407 pub fn open_dir(&self, path: impl AsRef<Path>) -> VfResult<VfDir> {
411 self.lock()?.open_dir(path.as_ref())
412 }
413
414 pub fn remove_dir_contents_handle(&self, dir: &VfDir) -> VfResult<()> {
416 self.lock()?.rm_dir_contents(dir)
417 }
418
419 pub fn close_dir(&self, dir: &VfDir) -> VfResult<()> {
421 self.lock()?.close_dir(dir)
422 }
423}
424
425impl<F: LinkFileSystem> FsClient<F> {
426 pub fn symlink(&self, target: impl AsRef<Path>, link: impl AsRef<Path>) -> VfResult<()> {
427 self.lock()?.symlink_one(target.as_ref(), link.as_ref())
428 }
429
430 pub fn hard_link(&self, source: impl AsRef<Path>, link: impl AsRef<Path>) -> VfResult<()> {
431 self.lock()?.hard_link_one(source.as_ref(), link.as_ref())
432 }
433
434 pub fn read_link(&self, path: impl AsRef<Path>) -> VfResult<PathBuf> {
435 self.lock()?.read_link_one(path.as_ref())
436 }
437}
438
439impl<F: CopyFileSystem> FsClient<F> {
440 pub fn copy(&self, source: impl AsRef<Path>, destination: impl AsRef<Path>) -> VfResult<()> {
441 self.lock()?.copy_one(source.as_ref(), destination.as_ref())
442 }
443}
444
445impl<F: VecFs> FsClient<F> {
446 pub fn symlink_metadata_with_fields(
449 &self,
450 path: impl AsRef<Path>,
451 fields: AttrMask,
452 ) -> VfResult<Metadata> {
453 let path = path.as_ref();
454 let mut attrs = crate::VfAttrs {
455 file: VfFile::from_os_path(path),
456 masks: fields | AttrMask::MODE | AttrMask::SIZE,
457 ..crate::VfAttrs::default()
458 };
459 self.lock()?
460 .lgetattrsv(std::slice::from_mut(&mut attrs))
461 .map_err(|error| error.with_context("symlink_metadata", path))?;
462 Ok(attrs.into())
463 }
464
465 pub fn read_dirs_with_options<P: AsRef<Path>>(
472 &self,
473 paths: &[P],
474 fields: AttrMask,
475 options: ReadDirOptions,
476 ) -> VfResult<Vec<DirectoryListing>> {
477 if paths.is_empty() {
478 return Ok(Vec::new());
479 }
480 let paths: Vec<&Path> = paths.iter().map(AsRef::as_ref).collect();
481 let mut positions: HashMap<PathBuf, Vec<usize>> = HashMap::with_capacity(paths.len());
482 let mut unique_paths = Vec::with_capacity(paths.len());
483 for (index, path) in paths.iter().enumerate() {
484 if let Some(indices) = positions.get_mut(*path) {
485 indices.push(index);
486 } else {
487 positions.insert(path.to_path_buf(), vec![index]);
488 unique_paths.push(*path);
489 }
490 }
491 let mut listings: Vec<DirectoryListing> = paths
492 .iter()
493 .map(|path| DirectoryListing {
494 path: path.to_path_buf(),
495 entries: Vec::new(),
496 })
497 .collect();
498 let mut entry_count = 0usize;
499 let mut path_bytes = 0usize;
500 const DIRECTORY_COHORT: usize = 32;
503 for cohort in unique_paths.chunks(DIRECTORY_COHORT) {
504 let mut callback_error = None;
505 let requested = options
506 .entry_limit()
507 .saturating_sub(entry_count)
508 .saturating_add(1);
509 let result = self.lock()?.listdirv(
510 cohort,
511 fields | AttrMask::MODE | AttrMask::SIZE,
512 requested,
513 false,
514 &mut |attributes, directory| {
515 let Some(indices) = positions.get(directory) else {
516 callback_error = Some(VfError::transport(
517 None,
518 format!(
519 "read_dirs backend returned an unexpected directory: {}",
520 directory.display()
521 ),
522 ));
523 return false;
524 };
525 if !cohort.contains(&directory) {
526 callback_error = Some(VfError::transport(
527 None,
528 "read_dirs backend returned a directory outside the active cohort",
529 ));
530 return false;
531 }
532 let Some(path) = attributes.file.path() else {
533 callback_error = Some(VfError::transport(
534 None,
535 "read_dirs backend returned an entry without a path",
536 ));
537 return false;
538 };
539 if path.parent() != Some(directory) {
540 callback_error = Some(VfError::transport(
541 None,
542 format!(
543 "read_dirs backend returned an entry outside {}",
544 directory.display()
545 ),
546 ));
547 return false;
548 }
549 for &index in indices {
550 entry_count += 1;
551 path_bytes = path_bytes.saturating_add(path.as_os_str().len());
552 if entry_count > options.entry_limit()
553 || path_bytes > options.path_byte_limit()
554 {
555 callback_error = Some(
556 VfError::failure(index, libc::EFBIG as u32)
557 .with_context("read_dirs", directory),
558 );
559 return false;
560 }
561 listings[index]
562 .entries
563 .push(DirEntry::new(path.to_path_buf(), attributes.clone().into()));
564 }
565 true
566 },
567 );
568 if let Some(error) = callback_error {
569 return Err(error);
570 }
571 result.map_err(|error| {
572 error.map_index(|index| {
573 cohort
574 .get(index)
575 .and_then(|path| positions.get(*path))
576 .and_then(|indices| indices.first())
577 .copied()
578 .unwrap_or(index)
579 })
580 })?;
581 }
582 Ok(listings)
583 }
584
585 pub fn walk_with_options(
589 &self,
590 root: impl AsRef<Path>,
591 fields: AttrMask,
592 options: crate::WalkOptions,
593 ) -> VfResult<Vec<DirectoryListing>> {
594 let root = root.as_ref();
595 let tree = self.lock()?.walk_with_options(
596 root,
597 fields | AttrMask::MODE | AttrMask::SIZE,
598 options,
599 &mut |_, _| {},
600 )?;
601 tree.into_iter()
602 .map(|directory| {
603 let entries = directory
604 .entries
605 .into_iter()
606 .enumerate()
607 .map(|(index, attributes)| {
608 let path = attributes
609 .file
610 .path()
611 .ok_or_else(|| {
612 VfError::transport(
613 None,
614 format!("walk backend returned entry {index} without a path"),
615 )
616 })?
617 .to_path_buf();
618 Ok(DirEntry::new(path, attributes.into()))
619 })
620 .collect::<VfResult<Vec<_>>>()?;
621 Ok(DirectoryListing {
622 path: directory.path,
623 entries,
624 })
625 })
626 .collect()
627 }
628
629 pub fn copy_files<P: AsRef<Path>, Q: AsRef<Path>>(&self, pairs: &[(P, Q)]) -> VfResult<()> {
632 let extents: Vec<_> = pairs
633 .iter()
634 .map(|(from, to)| {
635 crate::ExtentPair::from_os_paths(from.as_ref(), 0, to.as_ref(), 0, None)
636 })
637 .collect();
638 self.lock()?.copyv(&extents).map_err(|error| {
639 error
640 .index_opt()
641 .and_then(|index| pairs.get(index))
642 .map_or(error.clone(), |(_, to)| {
643 error.with_context("copy_files", to.as_ref())
644 })
645 })
646 }
647
648 pub fn remove_paths<P: AsRef<Path>>(&self, paths: &[P], recursive: bool) -> VfResult<()> {
651 let paths: Vec<&Path> = paths.iter().map(AsRef::as_ref).collect();
652 self.lock()?.rm(&paths, recursive).map_err(|error| {
653 error
654 .index_opt()
655 .and_then(|index| paths.get(index))
656 .map_or(error.clone(), |path| {
657 error.with_context("remove_paths", path)
658 })
659 })
660 }
661
662 pub fn symlink_metadatav(&self, paths: &[&Path]) -> VfResult<Vec<Metadata>> {
665 let mut attrs: Vec<_> = paths
666 .iter()
667 .map(|path| crate::VfAttrs {
668 file: VfFile::from_os_path(path),
669 masks: AttrMask::MODE,
670 ..crate::VfAttrs::default()
671 })
672 .collect();
673 self.lock()?.lgetattrsv(&mut attrs).map_err(|error| {
674 error
675 .index_opt()
676 .and_then(|index| paths.get(index))
677 .map_or(error.clone(), |path| {
678 error.with_context("symlink_metadatav", path)
679 })
680 })?;
681 Ok(attrs.into_iter().map(Metadata::from).collect())
682 }
683}
684
685impl<F: VectorFileSystem> FsClient<F> {
686 pub fn openv(&self, requests: &[OpenRequest]) -> VfResult<Vec<FsFile<F>>> {
692 let mut filesystem = self.lock()?;
693 let files = filesystem.open_many(requests).map_err(|error| {
694 error
695 .index_opt()
696 .and_then(|index| requests.get(index))
697 .map_or(error.clone(), |request| {
698 error.with_context("openv", &request.path)
699 })
700 })?;
701 if files.len() != requests.len() {
702 let error = wrong_result_count("openv", requests.len(), files.len());
703 for file in &files {
704 let _ = filesystem.close_one(file);
705 }
706 return Err(error);
707 }
708 drop(filesystem);
709 Ok(files
710 .into_iter()
711 .zip(requests)
712 .map(|(file, request)| FsFile {
713 inner: Arc::clone(&self.inner),
714 file: Some(file),
715 path: request.path.clone(),
716 })
717 .collect())
718 }
719
720 pub fn try_closev(&self, files: &mut [FsFile<F>]) -> VfResult<()> {
724 for (index, file) in files.iter().enumerate() {
725 self.validate_owner(file, index)?;
726 }
727 let descriptors: Vec<VfFile> = files
728 .iter()
729 .map(|file| file.raw().cloned())
730 .collect::<VfResult<_>>()?;
731 self.lock()?.close_many(&descriptors).map_err(|error| {
732 error
733 .index_opt()
734 .and_then(|index| files.get(index))
735 .map_or(error.clone(), |file| {
736 error.with_context("closev", file.path())
737 })
738 })?;
739 for file in files.iter_mut() {
740 file.file = None;
741 }
742 Ok(())
743 }
744
745 pub fn closev(&self, mut files: Vec<FsFile<F>>) -> VfResult<()> {
751 self.try_closev(&mut files)
752 }
753
754 pub fn readv(&self, requests: &[FsRead<'_, F>]) -> VfResult<Vec<ReadResult>> {
758 self.readv_with_limit(requests, DEFAULT_READV_MAX_TOTAL_BYTES)
759 }
760
761 pub fn readv_with_limit(
763 &self,
764 requests: &[FsRead<'_, F>],
765 max_total_bytes: usize,
766 ) -> VfResult<Vec<ReadResult>> {
767 let mut requested = 0usize;
768 for (index, request) in requests.iter().enumerate() {
769 requested = requested
770 .checked_add(request.length)
771 .ok_or_else(|| VfError::client(index, libc::EFBIG as u32))?;
772 if requested > max_total_bytes {
773 return Err(VfError::client(index, libc::EFBIG as u32)
774 .with_context("readv", request.file.path()));
775 }
776 }
777 let reads = self.read_ops(requests)?;
778 let results = self.lock()?.read_many(&reads).map_err(|error| {
779 error
780 .index_opt()
781 .and_then(|index| requests.get(index))
782 .map_or(error.clone(), |request| {
783 error.with_context("readv", request.file.path())
784 })
785 })?;
786 if results.len() != requests.len() {
787 return Err(wrong_result_count("readv", requests.len(), results.len()));
788 }
789 validate_read_results("readv", &reads, &results).map_err(|error| {
790 error
791 .index_opt()
792 .and_then(|index| requests.get(index))
793 .map_or(error.clone(), |request| {
794 error.with_context("readv", request.file.path())
795 })
796 })?;
797 Ok(results)
798 }
799
800 fn read_ops(&self, requests: &[FsRead<'_, F>]) -> VfResult<Vec<ReadOp>> {
801 let mut reads = Vec::with_capacity(requests.len());
802 for (index, request) in requests.iter().enumerate() {
803 self.validate_owner(request.file, index)?;
804 reads.push(ReadOp::new(
805 request.file.raw()?.clone(),
806 request.offset,
807 request.length,
808 ));
809 }
810 Ok(reads)
811 }
812
813 pub fn readv_into(&self, requests: &mut [FsReadInto<'_, F>]) -> VfResult<Vec<usize>> {
814 let mut reads = Vec::with_capacity(requests.len());
815 for (index, request) in requests.iter().enumerate() {
816 self.validate_owner(request.file, index)?;
817 reads.push(ReadOp::new(
818 request.file.raw()?.clone(),
819 request.offset,
820 request.buffer.len(),
821 ));
822 }
823 let results = {
824 let mut buffers: Vec<&mut [u8]> = requests
825 .iter_mut()
826 .map(|request| &mut *request.buffer)
827 .collect();
828 self.lock()?.read_many_into(&reads, &mut buffers)
829 }
830 .map_err(|error| {
831 error
832 .index_opt()
833 .and_then(|index| requests.get(index))
834 .map_or(error.clone(), |request| {
835 error.with_context("readv_into", request.file.path())
836 })
837 })?;
838 validate_read_into_results("readv_into", &reads, &results).map_err(|error| {
839 error
840 .index_opt()
841 .and_then(|index| requests.get(index))
842 .map_or(error.clone(), |request| {
843 error.with_context("readv_into", request.file.path())
844 })
845 })?;
846 Ok(results.into_iter().map(|result| result.read).collect())
847 }
848
849 pub fn writev(&self, requests: &[FsWrite<'_, F>]) -> VfResult<Vec<WriteResult>> {
850 let writes = self.write_ops(requests)?;
851 let results = self.lock()?.write_many(&writes).map_err(|error| {
852 error
853 .index_opt()
854 .and_then(|index| requests.get(index))
855 .map_or(error.clone(), |request| {
856 error.with_context("writev", request.file.path())
857 })
858 })?;
859 if results.len() != requests.len() {
860 return Err(wrong_result_count("writev", requests.len(), results.len()));
861 }
862 validate_write_results("writev", &writes, &results).map_err(|error| {
863 error
864 .index_opt()
865 .and_then(|index| requests.get(index))
866 .map_or(error.clone(), |request| {
867 error.with_context("writev", request.file.path())
868 })
869 })?;
870 Ok(results)
871 }
872
873 pub fn write_allv(&self, requests: &[FsWrite<'_, F>]) -> VfResult<Vec<WriteResult>> {
879 let mut prior_by_path: HashMap<&Path, Vec<(usize, u64, u64)>> = HashMap::new();
882 let mut blocked_by = vec![Vec::new(); requests.len()];
883 for (index, request) in requests.iter().enumerate() {
884 self.validate_owner(request.file, index)?;
885 request.file.raw()?;
886 let VfOffset::At(offset) = request.offset else {
887 return Err(VfError::client(index, crate::ERR_INVAL)
888 .with_context("write_allv", request.file.path()));
889 };
890 let end = offset
891 .checked_add(request.data.len() as u64)
892 .ok_or_else(|| {
893 VfError::client(index, libc::EOVERFLOW as u32)
894 .with_context("write_allv", request.file.path())
895 })?;
896 let prior = prior_by_path.entry(request.file.path()).or_default();
900 for &(earlier, start, earlier_end) in prior.iter() {
901 if offset < earlier_end && start < end {
902 blocked_by[index].push(earlier);
903 }
904 }
905 prior.push((index, offset, end));
906 }
907 let mut totals = vec![0usize; requests.len()];
908 let mut stable = vec![true; requests.len()];
909 let mut pending: Vec<usize> = (0..requests.len())
910 .filter(|&index| !requests[index].data.is_empty())
911 .collect();
912 while !pending.is_empty() {
913 let wave_indices: Vec<usize> = pending
914 .iter()
915 .copied()
916 .filter(|&index| {
917 blocked_by[index]
918 .iter()
919 .all(|&earlier| totals[earlier] == requests[earlier].data.len())
920 })
921 .collect();
922 let wave: Vec<_> = wave_indices
923 .iter()
924 .map(|&index| {
925 let request = &requests[index];
926 let VfOffset::At(offset) = request.offset else {
927 return Err(VfError::client(index, crate::ERR_INVAL)
928 .with_context("write_allv", request.file.path()));
929 };
930 let offset = offset.checked_add(totals[index] as u64).ok_or_else(|| {
931 VfError::client(index, libc::EOVERFLOW as u32)
932 .with_context("write_allv", request.file.path())
933 })?;
934 Ok(FsWrite {
935 file: request.file,
936 offset: VfOffset::At(offset),
937 data: &request.data[totals[index]..],
938 })
939 })
940 .collect::<VfResult<_>>()?;
941 let results = self.writev(&wave).map_err(|error| {
942 error.map_index(|index| wave_indices.get(index).copied().unwrap_or(index))
943 })?;
944 for (&index, result) in wave_indices.iter().zip(results) {
945 if result.written == 0 {
946 return Err(VfError::client(index, crate::ERR_IO)
947 .with_context("write_allv", requests[index].file.path()));
948 }
949 totals[index] += result.written;
950 stable[index] &= result.stable;
951 }
952 pending.retain(|&index| totals[index] < requests[index].data.len());
953 }
954 requests
955 .iter()
956 .enumerate()
957 .map(|(index, request)| {
958 let VfOffset::At(offset) = request.offset else {
959 return Err(VfError::client(index, crate::ERR_INVAL)
960 .with_context("write_allv", request.file.path()));
961 };
962 Ok(WriteResult {
963 file: request.file.raw()?.clone(),
964 offset,
965 written: totals[index],
966 stable: stable[index],
967 })
968 })
969 .collect()
970 }
971
972 fn write_ops<'a>(&self, requests: &'a [FsWrite<'a, F>]) -> VfResult<Vec<WriteOpRef<'a>>> {
973 let mut writes = Vec::with_capacity(requests.len());
974 for (index, request) in requests.iter().enumerate() {
975 self.validate_owner(request.file, index)?;
976 writes.push(WriteOpRef::new(
977 request.file.raw()?,
978 request.offset,
979 request.data,
980 ));
981 }
982 Ok(writes)
983 }
984
985 fn validate_owner(&self, file: &FsFile<F>, index: usize) -> VfResult<()> {
986 if Arc::ptr_eq(&self.inner, &file.inner) {
987 Ok(())
988 } else {
989 Err(VfError::client(index, crate::ERR_INVAL))
990 }
991 }
992}
993
994impl<F: VectorFileSystem + VecFs> FsClient<F> {
995 pub fn read_files<P: AsRef<Path>>(&self, paths: &[P]) -> VfResult<Vec<Vec<u8>>> {
1002 self.read_files_with_options(paths, ReadAllOptions::default())
1003 }
1004
1005 pub fn read_files_with_options<P: AsRef<Path>>(
1007 &self,
1008 paths: &[P],
1009 options: ReadAllOptions,
1010 ) -> VfResult<Vec<Vec<u8>>> {
1011 let files: Vec<_> = paths
1012 .iter()
1013 .map(|path| VfFile::from_os_path(path.as_ref()))
1014 .collect();
1015 let result = self
1016 .lock()?
1017 .read_allv_with_options(&files, options)
1018 .map_err(|error| {
1019 error
1020 .index_opt()
1021 .and_then(|index| paths.get(index))
1022 .map_or(error.clone(), |path| {
1023 error.with_context("read_files", path.as_ref())
1024 })
1025 });
1026 result.and_then(|buffers| {
1027 if buffers.len() != paths.len() {
1028 Err(wrong_result_count("read_files", paths.len(), buffers.len()))
1029 } else {
1030 Ok(buffers)
1031 }
1032 })
1033 }
1034
1035 pub fn write_files<P: AsRef<Path>, B: AsRef<[u8]>>(&self, entries: &[(P, B)]) -> VfResult<()> {
1042 let mut seen = HashSet::with_capacity(entries.len());
1043 for (index, (path, _)) in entries.iter().enumerate() {
1044 if !seen.insert(path.as_ref()) {
1045 return Err(VfError::client(index, crate::ERR_INVAL)
1046 .with_context("write_files", path.as_ref()));
1047 }
1048 }
1049 let requests: Vec<_> = entries
1050 .iter()
1051 .map(|(path, _)| {
1052 OpenRequest::new(
1053 path.as_ref(),
1054 OpenFlags::WRITE | OpenFlags::CREATE | OpenFlags::TRUNCATE,
1055 )
1056 })
1057 .collect();
1058 let files = self.openv(&requests)?;
1059 let writes: Vec<_> = files
1060 .iter()
1061 .zip(entries)
1062 .map(|(file, (_, data))| file.write_request_at(0, data.as_ref()))
1063 .collect();
1064 let result = self.write_allv(&writes);
1065 drop(writes);
1066 let close_result = self.closev(files);
1067 result?;
1068 close_result
1069 }
1070}
1071
1072pub struct OpenOptions<'a, F: FileSystem> {
1074 client: &'a FsClient<F>,
1075 flags: OpenFlags,
1076 mode: u32,
1077}
1078
1079impl<F: FileSystem> Clone for OpenOptions<'_, F> {
1080 fn clone(&self) -> Self {
1081 Self {
1082 client: self.client,
1083 flags: self.flags,
1084 mode: self.mode,
1085 }
1086 }
1087}
1088
1089impl<F: FileSystem> fmt::Debug for OpenOptions<'_, F> {
1090 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
1091 formatter
1092 .debug_struct("OpenOptions")
1093 .field("flags", &self.flags)
1094 .field("mode", &format_args!("{:#o}", self.mode))
1095 .finish()
1096 }
1097}
1098
1099impl<'a, F: FileSystem> OpenOptions<'a, F> {
1100 fn new(client: &'a FsClient<F>) -> Self {
1101 Self {
1102 client,
1103 flags: OpenFlags::empty(),
1104 mode: 0o666,
1105 }
1106 }
1107
1108 pub fn read(&mut self, enabled: bool) -> &mut Self {
1109 self.flags.set(OpenFlags::READ, enabled);
1110 self
1111 }
1112
1113 pub fn write(&mut self, enabled: bool) -> &mut Self {
1114 self.flags.set(OpenFlags::WRITE, enabled);
1115 self
1116 }
1117
1118 pub fn append(&mut self, enabled: bool) -> &mut Self {
1119 self.flags.set(OpenFlags::APPEND, enabled);
1120 self
1121 }
1122
1123 pub fn truncate(&mut self, enabled: bool) -> &mut Self {
1124 self.flags.set(OpenFlags::TRUNCATE, enabled);
1125 self
1126 }
1127
1128 pub fn create(&mut self, enabled: bool) -> &mut Self {
1129 self.flags.set(OpenFlags::CREATE, enabled);
1130 self
1131 }
1132
1133 pub fn create_new(&mut self, enabled: bool) -> &mut Self {
1134 self.flags.set(OpenFlags::CREATE_NEW, enabled);
1135 self
1136 }
1137
1138 pub fn mode(&mut self, mode: u32) -> &mut Self {
1139 self.mode = mode;
1140 self
1141 }
1142
1143 pub fn open(&self, path: impl AsRef<Path>) -> VfResult<FsFile<F>> {
1144 self.client
1145 .open_with(OpenRequest::new(path.as_ref(), self.flags).mode(self.mode))
1146 }
1147}
1148
1149impl<F: VectorFileSystem> OpenOptions<'_, F> {
1150 pub fn openv<P: AsRef<Path>>(&self, paths: &[P]) -> VfResult<Vec<FsFile<F>>> {
1151 let requests: Vec<OpenRequest> = paths
1152 .iter()
1153 .map(|path| OpenRequest::new(path.as_ref(), self.flags).mode(self.mode))
1154 .collect();
1155 self.client.openv(&requests)
1156 }
1157}
1158
1159pub struct SetMetadata<'a, F: MetadataFileSystem> {
1161 client: &'a FsClient<F>,
1162 path: PathBuf,
1163 update: MetadataUpdate,
1164 follow: bool,
1165}
1166
1167impl<F: MetadataFileSystem> Clone for SetMetadata<'_, F> {
1168 fn clone(&self) -> Self {
1169 Self {
1170 client: self.client,
1171 path: self.path.clone(),
1172 update: self.update.clone(),
1173 follow: self.follow,
1174 }
1175 }
1176}
1177
1178impl<F: MetadataFileSystem> fmt::Debug for SetMetadata<'_, F> {
1179 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
1180 formatter
1181 .debug_struct("SetMetadata")
1182 .field("path", &self.path)
1183 .field("update", &self.update)
1184 .field("follow", &self.follow)
1185 .finish_non_exhaustive()
1186 }
1187}
1188
1189impl<F: MetadataFileSystem> SetMetadata<'_, F> {
1190 pub fn permissions(&mut self, permissions: Permissions) -> &mut Self {
1191 self.update.permissions = Some(permissions);
1192 self
1193 }
1194
1195 pub fn len(&mut self, len: u64) -> &mut Self {
1196 self.update.len = Some(len);
1197 self
1198 }
1199
1200 pub fn accessed(&mut self, accessed: SystemTime) -> &mut Self {
1201 self.update.accessed = Some(accessed);
1202 self
1203 }
1204
1205 pub fn modified(&mut self, modified: SystemTime) -> &mut Self {
1206 self.update.modified = Some(modified);
1207 self
1208 }
1209
1210 pub fn follow_symlinks(&mut self, follow: bool) -> &mut Self {
1211 self.follow = follow;
1212 self
1213 }
1214
1215 pub fn apply(&self) -> VfResult<()> {
1216 self.client
1217 .lock()?
1218 .set_metadata_path(&self.path, self.update.clone(), self.follow)
1219 }
1220}
1221
1222pub struct FsFile<F: FileSystem> {
1228 inner: Arc<Mutex<F>>,
1229 file: Option<VfFile>,
1230 path: PathBuf,
1231}
1232
1233impl<F: FileSystem> fmt::Debug for FsFile<F> {
1234 fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
1235 formatter
1236 .debug_struct("FsFile")
1237 .field("path", &self.path)
1238 .field("is_open", &self.file.is_some())
1239 .finish()
1240 }
1241}
1242
1243impl<F: FileSystem> FsFile<F> {
1244 fn raw(&self) -> VfResult<&VfFile> {
1245 self.file
1246 .as_ref()
1247 .ok_or_else(|| VfError::client(0, crate::ERR_EBADF))
1248 }
1249
1250 pub fn read_at(&self, buffer: &mut [u8], offset: u64) -> VfResult<usize> {
1252 self.read_into(buffer, VfOffset::At(offset))
1253 }
1254
1255 pub fn write_at(&self, buffer: &[u8], offset: u64) -> VfResult<usize> {
1257 self.write_from(buffer, VfOffset::At(offset))
1258 }
1259
1260 pub fn read_request_at(&self, offset: u64, length: usize) -> FsRead<'_, F> {
1261 FsRead {
1262 file: self,
1263 offset: VfOffset::At(offset),
1264 length,
1265 }
1266 }
1267
1268 pub fn path(&self) -> &Path {
1269 &self.path
1270 }
1271
1272 pub fn metadata(&self) -> VfResult<Metadata> {
1274 let attributes = AttrMask::MODE
1275 | AttrMask::SIZE
1276 | AttrMask::NLINK
1277 | AttrMask::FILEID
1278 | AttrMask::UID
1279 | AttrMask::GID
1280 | AttrMask::ATIME
1281 | AttrMask::MTIME
1282 | AttrMask::CTIME
1283 | AttrMask::CHANGE;
1284 self.inner
1285 .lock()
1286 .map_err(|_| poisoned())?
1287 .metadata(MetadataQuery::new(self.raw()?.clone(), attributes))
1288 .map(Metadata::from)
1289 .map_err(|error| error.with_context("metadata", &self.path))
1290 }
1291
1292 pub fn set_len(&self, len: u64) -> VfResult<()> {
1294 let mut update = SetAttributes::new(self.raw()?.clone());
1295 update.size = Some(len);
1296 self.inner
1297 .lock()
1298 .map_err(|_| poisoned())?
1299 .set_attributes(update)
1300 .map_err(|error| error.with_context("set_len", &self.path))
1301 }
1302
1303 pub fn set_permissions(&self, permissions: Permissions) -> VfResult<()> {
1305 let mut update = SetAttributes::new(self.raw()?.clone());
1306 update.mode = Some(permissions.mode());
1307 self.inner
1308 .lock()
1309 .map_err(|_| poisoned())?
1310 .set_attributes(update)
1311 .map_err(|error| error.with_context("set_permissions", &self.path))
1312 }
1313
1314 pub fn write_request_at<'a>(&'a self, offset: u64, data: &'a [u8]) -> FsWrite<'a, F> {
1315 FsWrite {
1316 file: self,
1317 offset: VfOffset::At(offset),
1318 data,
1319 }
1320 }
1321
1322 pub fn read_request_at_into<'a>(
1323 &'a self,
1324 offset: u64,
1325 buffer: &'a mut [u8],
1326 ) -> FsReadInto<'a, F> {
1327 FsReadInto {
1328 file: self,
1329 offset: VfOffset::At(offset),
1330 buffer,
1331 }
1332 }
1333
1334 pub fn read_native(&mut self, buffer: &mut [u8]) -> VfResult<usize> {
1335 self.read_into(buffer, VfOffset::Cur)
1336 }
1337
1338 pub fn write_native(&mut self, buffer: &[u8]) -> VfResult<usize> {
1339 self.write_from(buffer, VfOffset::Cur)
1340 }
1341
1342 fn read_into(&self, buffer: &mut [u8], offset: VfOffset) -> VfResult<usize> {
1343 if buffer.is_empty() {
1344 return Ok(0);
1345 }
1346 let request = ReadOp::new(self.raw()?.clone(), offset, buffer.len());
1347 let result = self
1348 .inner
1349 .lock()
1350 .map_err(|_| poisoned())?
1351 .read_one_into(&request, buffer)
1352 .map_err(|error| error.with_context("read", &self.path))?;
1353 validate_read_into_results(
1354 "read",
1355 std::slice::from_ref(&request),
1356 std::slice::from_ref(&result),
1357 )
1358 .map_err(|error| error.with_context("read", &self.path))?;
1359 Ok(result.read)
1360 }
1361
1362 fn write_from(&self, buffer: &[u8], offset: VfOffset) -> VfResult<usize> {
1363 if buffer.is_empty() {
1364 return Ok(0);
1365 }
1366 let file = self.raw()?.clone();
1367 let result = self
1368 .inner
1369 .lock()
1370 .map_err(|_| poisoned())?
1371 .write_one(WriteOpRef::new(&file, offset, buffer))
1372 .map_err(|error| error.with_context("write", &self.path))?;
1373 if result.written > buffer.len() {
1374 return Err(VfError::client(0, crate::ERR_IO).with_context("write", &self.path));
1375 }
1376 Ok(result.written)
1377 }
1378
1379 pub fn sync_data(&self) -> VfResult<()> {
1380 self.inner
1381 .lock()
1382 .map_err(|_| poisoned())?
1383 .sync_data(self.raw()?)
1384 .map_err(|error| error.with_context("sync_data", &self.path))
1385 }
1386
1387 pub fn sync_all(&self) -> VfResult<()> {
1388 self.inner
1389 .lock()
1390 .map_err(|_| poisoned())?
1391 .sync_all(self.raw()?)
1392 .map_err(|error| error.with_context("sync_all", &self.path))
1393 }
1394
1395 pub fn try_close(&mut self) -> VfResult<()> {
1398 let Some(file) = self.file.as_ref() else {
1399 return Ok(());
1400 };
1401 self.inner
1402 .lock()
1403 .map_err(|_| poisoned())?
1404 .close_one(file)
1405 .map_err(|error| error.with_context("close", &self.path))?;
1406 self.file = None;
1407 Ok(())
1408 }
1409
1410 pub fn close(mut self) -> VfResult<()> {
1413 self.try_close()
1414 }
1415
1416 pub fn seek_native(&mut self, position: IoSeekFrom) -> VfResult<u64> {
1418 self.inner
1419 .lock()
1420 .map_err(|_| poisoned())?
1421 .seek_one(self.raw()?, position)
1422 .map_err(|error| error.with_context("seek", &self.path))
1423 }
1424}
1425
1426impl<F: FileSystem> Read for FsFile<F> {
1427 fn read(&mut self, buffer: &mut [u8]) -> io::Result<usize> {
1428 self.read_native(buffer).map_err(io_error)
1429 }
1430}
1431
1432impl<F: FileSystem> Write for FsFile<F> {
1433 fn write(&mut self, buffer: &[u8]) -> io::Result<usize> {
1434 self.write_native(buffer).map_err(io_error)
1435 }
1436
1437 fn flush(&mut self) -> io::Result<()> {
1438 self.sync_data().map_err(io_error)
1439 }
1440}
1441
1442impl<F: FileSystem> Seek for FsFile<F> {
1443 fn seek(&mut self, position: IoSeekFrom) -> io::Result<u64> {
1444 self.seek_native(position).map_err(io_error)
1445 }
1446}
1447
1448pub struct FsRead<'a, F: FileSystem> {
1450 file: &'a FsFile<F>,
1451 offset: VfOffset,
1452 length: usize,
1453}
1454
1455pub struct FsWrite<'a, F: FileSystem> {
1457 file: &'a FsFile<F>,
1458 offset: VfOffset,
1459 data: &'a [u8],
1460}
1461
1462pub struct FsReadInto<'a, F: FileSystem> {
1464 file: &'a FsFile<F>,
1465 offset: VfOffset,
1466 buffer: &'a mut [u8],
1467}
1468
1469impl<F: FileSystem> Drop for FsFile<F> {
1470 fn drop(&mut self) {
1471 if let Some(file) = self.file.take()
1472 && let Ok(mut filesystem) = self.inner.lock()
1473 {
1474 let _ = filesystem.close_one(&file);
1475 }
1476 }
1477}