Copilot commented on code in PR #890:
URL:
https://github.com/apache/arrow-rs-object-store/pull/890#discussion_r4459293773
##########
src/local.rs:
##########
@@ -900,79 +906,103 @@ impl LocalFileSystem {
/// Creates the parent directories of `path` or returns an error based on
`source` if no parent
///
-/// When `fsync` is true, every directory created here is fsynced, up to and
including the first
-/// pre-existing ancestor (whose entry list also changed), so the new
directory entries are durable.
+/// When `fsync` is true, every directory whose entries this changes is
fsynced, so the new
+/// directory entries are durable.
fn create_parent_dirs(path: &std::path::Path, source: io::Error, fsync: bool)
-> Result<()> {
let parent = path.parent().ok_or_else(|| {
let path = path.to_path_buf();
Error::UnableToCreateFile { path, source }
})?;
+ if fsync {
+ create_dir_all_durable(parent, |dir| std::fs::create_dir(dir),
fsync_dir)
+ } else {
+ std::fs::create_dir_all(parent).map_err(|source| {
+ let path = parent.into();
+ Error::UnableToCreateDir { source, path }.into()
+ })
+ }
+}
- // Record the deepest already-existing ancestor *before* creating any
directories, so that
- // afterwards we know exactly which directories are new and need to be
fsynced.
- let first_existing = fsync.then(|| {
- let mut dir = parent;
- while !dir.exists() {
- match dir.parent() {
- Some(p) => dir = p,
- None => break,
- }
+/// Creates `dir` and its missing ancestors with `create`, recording in
[`UNSYNCED`] every
+/// directory whose entries that changes, and fsyncs them with `sync`. Tests
pass closures to
+/// interleave other changes and to observe the sync order.
+fn create_dir_all_durable(
+ dir: &std::path::Path,
+ mut create: impl FnMut(&std::path::Path) -> io::Result<()>,
+ sync: impl FnMut(&std::path::Path) -> io::Result<()>,
+) -> Result<()> {
+ // A directory found missing here may exist by the time `create` reaches
it, made by a
+ // creator this process does not see (another process, a store without
`fsync`) that may not
+ // sync its entry. Recording the directories above the missing ones covers
that entry too.
+ let missing = dir.ancestors().take_while(|dir| !dir.exists());
+ let above_missing: Vec<_> = missing.filter_map(|dir|
dir.parent()).collect();
+ changing(above_missing, || create_dir_all_recording(dir, &mut
create)).map_err(|source| {
+ Error::UnableToCreateDir {
+ source,
+ path: dir.into(),
}
- dir.to_path_buf()
- });
-
- std::fs::create_dir_all(parent).map_err(|source| {
- let path = parent.into();
- Error::UnableToCreateDir { source, path }
})?;
-
- if let Some(first_existing) = first_existing {
- // Walk from `parent` up to `first_existing`, fsyncing each directory
whose entries changed.
- let mut dir = parent;
- loop {
- fsync_dir(dir).map_err(|source| Error::UnableToSyncFile {
- source,
- path: dir.into(),
- })?;
- if dir == first_existing {
- break;
- }
- dir = match dir.parent() {
- Some(p) => p,
- None => break,
- };
+ sync_unsynced_ancestors_with(dir, sync).map_err(|source| {
+ Error::UnableToSyncFile {
+ source,
+ path: dir.into(),
}
+ .into()
+ })
Review Comment:
When syncing fails, the error is attributed to `path: dir` (the requested
directory), but the actual failing sync may be for a specific ancestor
directory. This makes operational debugging harder (e.g., permission issues on
a particular parent). Consider plumbing the failing directory path out of
`sync_unsynced_ancestors_with` (e.g., return `(PathBuf, io::Error)` or a small
custom error type) so `UnableToSyncFile.path` can point at the directory that
couldn't be synced.
##########
src/local.rs:
##########
@@ -900,79 +906,103 @@ impl LocalFileSystem {
/// Creates the parent directories of `path` or returns an error based on
`source` if no parent
///
-/// When `fsync` is true, every directory created here is fsynced, up to and
including the first
-/// pre-existing ancestor (whose entry list also changed), so the new
directory entries are durable.
+/// When `fsync` is true, every directory whose entries this changes is
fsynced, so the new
+/// directory entries are durable.
fn create_parent_dirs(path: &std::path::Path, source: io::Error, fsync: bool)
-> Result<()> {
let parent = path.parent().ok_or_else(|| {
let path = path.to_path_buf();
Error::UnableToCreateFile { path, source }
})?;
+ if fsync {
+ create_dir_all_durable(parent, |dir| std::fs::create_dir(dir),
fsync_dir)
+ } else {
+ std::fs::create_dir_all(parent).map_err(|source| {
+ let path = parent.into();
+ Error::UnableToCreateDir { source, path }.into()
+ })
+ }
+}
- // Record the deepest already-existing ancestor *before* creating any
directories, so that
- // afterwards we know exactly which directories are new and need to be
fsynced.
- let first_existing = fsync.then(|| {
- let mut dir = parent;
- while !dir.exists() {
- match dir.parent() {
- Some(p) => dir = p,
- None => break,
- }
+/// Creates `dir` and its missing ancestors with `create`, recording in
[`UNSYNCED`] every
+/// directory whose entries that changes, and fsyncs them with `sync`. Tests
pass closures to
+/// interleave other changes and to observe the sync order.
+fn create_dir_all_durable(
+ dir: &std::path::Path,
+ mut create: impl FnMut(&std::path::Path) -> io::Result<()>,
+ sync: impl FnMut(&std::path::Path) -> io::Result<()>,
+) -> Result<()> {
+ // A directory found missing here may exist by the time `create` reaches
it, made by a
+ // creator this process does not see (another process, a store without
`fsync`) that may not
+ // sync its entry. Recording the directories above the missing ones covers
that entry too.
+ let missing = dir.ancestors().take_while(|dir| !dir.exists());
+ let above_missing: Vec<_> = missing.filter_map(|dir|
dir.parent()).collect();
+ changing(above_missing, || create_dir_all_recording(dir, &mut
create)).map_err(|source| {
+ Error::UnableToCreateDir {
+ source,
+ path: dir.into(),
}
- dir.to_path_buf()
- });
-
- std::fs::create_dir_all(parent).map_err(|source| {
- let path = parent.into();
- Error::UnableToCreateDir { source, path }
})?;
-
- if let Some(first_existing) = first_existing {
- // Walk from `parent` up to `first_existing`, fsyncing each directory
whose entries changed.
- let mut dir = parent;
- loop {
- fsync_dir(dir).map_err(|source| Error::UnableToSyncFile {
- source,
- path: dir.into(),
- })?;
- if dir == first_existing {
- break;
- }
- dir = match dir.parent() {
- Some(p) => p,
- None => break,
- };
+ sync_unsynced_ancestors_with(dir, sync).map_err(|source| {
+ Error::UnableToSyncFile {
+ source,
+ path: dir.into(),
}
+ .into()
+ })
+}
+
+/// Creates `dir` and its missing ancestors one at a time, as
[`std::fs::create_dir_all`] does,
+/// recording the parent of each as changing while `create` runs on it.
+fn create_dir_all_recording(
+ dir: &std::path::Path,
+ create: &mut impl FnMut(&std::path::Path) -> io::Result<()>,
+) -> io::Result<()> {
+ match changing(dir.parent(), || create(dir)) {
+ Err(e) if e.kind() == ErrorKind::NotFound => {}
+ Err(e) if !dir.is_dir() => return Err(e),
+ _ => return Ok(()),
+ }
+ match dir.parent() {
+ Some(p) => create_dir_all_recording(p, create)?,
+ None => return Err(io::Error::other("failed to create whole tree")),
+ }
+ match changing(dir.parent(), || create(dir)) {
+ Err(e) if !dir.is_dir() => Err(e),
+ _ => Ok(()),
}
Review Comment:
`create_dir_all_recording` is recursive on `dir.parent()`. For very deep
paths this can overflow the stack, whereas `std::fs::create_dir_all` is
iterative. Consider rewriting this to an iterative approach (collect missing
components, then create from shallowest to deepest), while keeping the
parent-recording behavior, to avoid a path-depth-dependent stack risk.
##########
src/local.rs:
##########
@@ -1032,6 +1062,131 @@ fn sync_and_close(file: File, path: &std::path::Path,
fsync: bool) -> Result<()>
Ok(())
}
+/// Directories whose entries changed, or are changing, without an fsync
since. Shared by every
+/// `LocalFileSystem` in the process: a durable write into a directory another
call is still
+/// making durable fsyncs it too, and a failed fsync leaves the directory to
the next write.
+///
+/// A map rather than a queue: changes to one directory collapse into one
entry, and the change
+/// counter tells a sync whether the directory changed again after the sync
started, so the entry
+/// is then kept.
+static UNSYNCED: Mutex<Unsynced> = Mutex::new(Unsynced {
+ dirs: BTreeMap::new(),
+ changes: 0,
+});
+
+struct Unsynced {
+ dirs: BTreeMap<PathBuf, Pending>,
+ /// The stamp of the last finished change.
+ changes: u64,
+}
+
+#[derive(Default)]
+struct Pending {
+ /// Changes begun but not finished; the entry stays while there is one.
+ in_flight: usize,
+ /// The stamp of the last finished change, until a sync covers it.
+ changed: Option<u64>,
+}
+
+/// Runs `change`, which changes the entries of `dirs`, with them recorded in
[`UNSYNCED`] from
+/// before it runs (so a write that sees the change also sees the record)
until a later sync
+/// succeeds. A failed `mkdir`, `rename` or `link` changes no directory, so an
error records
+/// nothing; a panic counts as a change.
+fn changing<'a, T>(
+ dirs: impl IntoIterator<Item = &'a std::path::Path>,
+ change: impl FnOnce() -> io::Result<T>,
+) -> io::Result<T> {
+ let dirs: Vec<_> = dirs.into_iter().collect();
+ {
+ let mut unsynced = UNSYNCED.lock();
+ for dir in &dirs {
+ unsynced
+ .dirs
+ .entry(dir.to_path_buf())
+ .or_default()
+ .in_flight += 1;
+ }
+ }
+ let mut guard = Changing {
+ dirs,
+ changed: true,
+ };
+ let result = change();
+ guard.changed = result.is_ok();
+ result
+}
+
+/// Finishes the changes [`changing`] began, also when `change` panics.
+struct Changing<'a> {
+ dirs: Vec<&'a std::path::Path>,
+ changed: bool,
+}
+
+impl Drop for Changing<'_> {
+ fn drop(&mut self) {
+ let mut unsynced = UNSYNCED.lock();
+ let Unsynced { dirs, changes } = &mut *unsynced;
+ for dir in &self.dirs {
+ // Always present: an entry is not removed while a change is in
flight.
+ let Some(pending) = dirs.get_mut(*dir) else {
+ continue;
+ };
+ pending.in_flight -= 1;
+ if self.changed {
+ *changes += 1;
+ pending.changed = Some(*changes);
+ } else if pending.in_flight == 0 && pending.changed.is_none() {
+ dirs.remove(*dir);
+ }
+ }
+ }
+}
+
+/// Fsyncs the directories recorded in [`UNSYNCED`] among `path` and its
ancestors, deepest first,
+/// so an operation returns only once every directory entry leading to what it
wrote is durable,
+/// whoever changed it. An entry leaves the set only when a sync that started
after its last
+/// change succeeds or finds the directory gone; a failed sync leaves it to
the next write.
+fn sync_unsynced_ancestors(path: &std::path::Path) -> io::Result<()> {
+ sync_unsynced_ancestors_with(path, fsync_dir)
+}
+
+/// [`sync_unsynced_ancestors`] with `sync` in place of `fsync_dir`, so tests
can observe the order.
+fn sync_unsynced_ancestors_with(
+ path: &std::path::Path,
+ mut sync: impl FnMut(&std::path::Path) -> io::Result<()>,
+) -> io::Result<()> {
+ let pending: Vec<_> = {
+ let unsynced = UNSYNCED.lock();
+ path.ancestors()
+ .filter_map(|dir| Some((dir, unsynced.dirs.get(dir)?.changed)))
+ .collect()
+ };
+ let mut synced = 0;
+ let mut result = Ok(());
+ for (dir, _) in &pending {
+ match sync(dir) {
+ Ok(()) => synced += 1,
+ Err(e) => {
+ // A directory that is gone has no entries left to make
durable.
+ if e.kind() == ErrorKind::NotFound {
+ synced += 1;
+ }
+ result = Err(e);
+ break;
+ }
+ }
Review Comment:
`ErrorKind::NotFound` is treated as 'covered' (increments `synced`) but
still sets `result = Err(e)` and breaks, which makes the durable operation fail
even though the comment/doc imply a missing directory should be treated as a
non-error case ('directory gone drops its entry'). Consider treating `NotFound`
as success (do not set `result` to `Err`, and do not break), so ancestor
syncing can continue and callers don't see a spurious failure.
##########
src/local.rs:
##########
@@ -1032,6 +1062,131 @@ fn sync_and_close(file: File, path: &std::path::Path,
fsync: bool) -> Result<()>
Ok(())
}
+/// Directories whose entries changed, or are changing, without an fsync
since. Shared by every
+/// `LocalFileSystem` in the process: a durable write into a directory another
call is still
+/// making durable fsyncs it too, and a failed fsync leaves the directory to
the next write.
+///
+/// A map rather than a queue: changes to one directory collapse into one
entry, and the change
+/// counter tells a sync whether the directory changed again after the sync
started, so the entry
+/// is then kept.
+static UNSYNCED: Mutex<Unsynced> = Mutex::new(Unsynced {
+ dirs: BTreeMap::new(),
+ changes: 0,
+});
+
+struct Unsynced {
+ dirs: BTreeMap<PathBuf, Pending>,
+ /// The stamp of the last finished change.
+ changes: u64,
+}
+
+#[derive(Default)]
+struct Pending {
+ /// Changes begun but not finished; the entry stays while there is one.
+ in_flight: usize,
+ /// The stamp of the last finished change, until a sync covers it.
+ changed: Option<u64>,
+}
+
+/// Runs `change`, which changes the entries of `dirs`, with them recorded in
[`UNSYNCED`] from
+/// before it runs (so a write that sees the change also sees the record)
until a later sync
+/// succeeds. A failed `mkdir`, `rename` or `link` changes no directory, so an
error records
+/// nothing; a panic counts as a change.
+fn changing<'a, T>(
+ dirs: impl IntoIterator<Item = &'a std::path::Path>,
+ change: impl FnOnce() -> io::Result<T>,
+) -> io::Result<T> {
+ let dirs: Vec<_> = dirs.into_iter().collect();
+ {
+ let mut unsynced = UNSYNCED.lock();
+ for dir in &dirs {
+ unsynced
+ .dirs
+ .entry(dir.to_path_buf())
+ .or_default()
+ .in_flight += 1;
Review Comment:
This allocates a new `PathBuf` for every `dir` on every call
(`to_path_buf()` happens even when the key is already present). Since this runs
on every durable operation, consider avoiding allocation on the common path by
first attempting `get_mut(dir)` (borrowed lookup) and only allocating/inserting
a `PathBuf` when missing.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
