use std::ffi::{OsStr, OsString};
use std::path::PathBuf;
use std::sync::Arc;
use async_recursion::async_recursion;
use crate::error::OperationError;
use crate::progress::Progress;
use crate::safedir::{Dir, Handle};
use crate::walk::{self, EntryKind, LeafPermit, PermitKind};
pub trait WalkSummary: Default + std::ops::Add<Output = Self> + Send + Sized + 'static {}
impl<T> WalkSummary for T where T: Default + std::ops::Add<Output = T> + Send + Sized + 'static {}
#[derive(Clone)]
pub struct EntryCx {
pub parent: Arc<Dir>,
pub name: OsString,
pub rel_path: PathBuf,
pub filter_path: PathBuf,
pub real_path: PathBuf,
pub dry_run: bool,
pub prog_track: &'static Progress,
}
impl EntryCx {
#[must_use]
pub fn child(&self, child_dir: Arc<Dir>, child_name: &OsStr) -> EntryCx {
EntryCx {
parent: child_dir,
name: child_name.to_owned(),
rel_path: self.rel_path.join(child_name),
filter_path: self.filter_path.join(child_name),
real_path: self.real_path.join(child_name),
dry_run: self.dry_run,
prog_track: self.prog_track,
}
}
}
pub enum DirAction<Sum, Ctx, State> {
Skip(Sum),
Descend {
dir: Arc<Dir>,
child_ctx: Ctx,
state: State,
},
}
#[derive(Debug, Default)]
pub struct ProcessedChildren {
names: Vec<OsString>,
}
impl ProcessedChildren {
#[must_use]
pub fn names(&self) -> &[OsString] {
&self.names
}
#[must_use]
pub fn into_names(self) -> Vec<OsString> {
self.names
}
}
pub type DirPreResult<V> = Result<
DirAction<
<V as WalkVisitor>::Summary,
<V as WalkVisitor>::DirContext,
<V as WalkVisitor>::DirState,
>,
OperationError<<V as WalkVisitor>::Summary>,
>;
pub trait WalkVisitor: Send + Sync + 'static {
type Summary: WalkSummary;
type DirContext: Clone + Send + Sync + 'static;
type DirState: Send;
fn root_dir_context(&self) -> Self::DirContext;
fn permit_kind(&self) -> PermitKind;
fn want_permit(&self, hint: Option<EntryKind>) -> bool;
fn fail_early(&self) -> bool;
fn filter(&self) -> Option<&crate::filter::FilterSettings>;
fn on_skip(
&self,
_cx: &EntryCx,
_kind: EntryKind,
_skip_result: &crate::filter::FilterResult,
) -> Self::Summary {
Self::Summary::default()
}
fn visit_leaf(
&self,
cx: &EntryCx,
parent_ctx: &Self::DirContext,
handle: Handle,
kind: EntryKind,
permit: Option<LeafPermit>,
) -> impl std::future::Future<Output = Result<Self::Summary, OperationError<Self::Summary>>> + Send;
fn dir_pre(
&self,
cx: &EntryCx,
parent_ctx: &Self::DirContext,
handle: &Handle,
) -> impl std::future::Future<Output = DirPreResult<Self>> + Send;
fn dir_post(
&self,
cx: &EntryCx,
state: Self::DirState,
processed: &ProcessedChildren,
child_result: Result<Self::Summary, OperationError<Self::Summary>>,
) -> impl std::future::Future<Output = Result<Self::Summary, OperationError<Self::Summary>>> + Send;
}
#[async_recursion]
pub async fn process_entry<V>(
visitor: Arc<V>,
cx: EntryCx,
parent_ctx: V::DirContext,
permit: Option<LeafPermit>,
) -> Result<V::Summary, OperationError<V::Summary>>
where
V: WalkVisitor,
{
let _ops_guard = cx.prog_track.ops.guard();
let handle = match cx.parent.child(&cx.name).await {
Ok(handle) => handle,
Err(err) => {
let err = anyhow::Error::new(err)
.context(format!("failed reading metadata from {:?}", &cx.real_path));
return Err(OperationError::new(err, Default::default()));
}
};
let kind = handle.kind();
if kind != EntryKind::Dir {
return visitor
.visit_leaf(&cx, &parent_ctx, handle, kind, permit)
.await;
}
drop(permit);
match visitor.dir_pre(&cx, &parent_ctx, &handle).await? {
DirAction::Skip(summary) => Ok(summary),
DirAction::Descend {
dir,
child_ctx,
state,
} => {
match walk_dir_contents(Arc::clone(&visitor), dir, &cx, &child_ctx).await {
Ok((child_summary, processed)) => {
visitor
.dir_post(&cx, state, &processed, Ok(child_summary))
.await
}
Err(walk_err) => {
if visitor.fail_early() {
Err(walk_err)
} else {
visitor
.dir_post(&cx, state, &ProcessedChildren::default(), Err(walk_err))
.await
}
}
}
}
}
}
#[async_recursion]
pub async fn walk_dir_contents<V>(
visitor: Arc<V>,
dir: Arc<Dir>,
parent_cx: &EntryCx,
dir_ctx: &V::DirContext,
) -> Result<(V::Summary, ProcessedChildren), OperationError<V::Summary>>
where
V: WalkVisitor,
{
let entries = match dir.read_entries().await {
Ok(entries) => entries,
Err(err) => {
let err = anyhow::Error::new(err)
.context(format!("cannot read directory {:?}", &parent_cx.real_path));
return Err(OperationError::new(err, Default::default()));
}
};
let mut skipped_summary = V::Summary::default();
let mut processed = ProcessedChildren::default();
let mut join_set = tokio::task::JoinSet::new();
for (entry_name, hint) in entries {
let entry_is_dir =
walk::filter_is_dir(visitor.filter(), &dir, &entry_name, hint, false).await;
let child_cx = parent_cx.child(Arc::clone(&dir), &entry_name);
if let Some(skip_result) =
walk::should_skip_entry_ref(visitor.filter(), &child_cx.filter_path, entry_is_dir)
{
let entry_kind = hint.unwrap_or(if entry_is_dir {
EntryKind::Dir
} else {
EntryKind::File
});
tracing::debug!("skipping {:?} due to filter", &child_cx.real_path);
entry_kind.inc_skipped(parent_cx.prog_track);
skipped_summary =
skipped_summary + visitor.on_skip(&child_cx, entry_kind, &skip_result);
continue;
}
let permit =
walk::preacquire_leaf_permit(visitor.permit_kind(), hint, |h| visitor.want_permit(h))
.await;
let task_visitor = Arc::clone(&visitor);
let task_ctx = dir_ctx.clone();
processed.names.push(entry_name);
join_set
.spawn(async move { process_entry(task_visitor, child_cx, task_ctx, permit).await });
}
let folded =
join_and_fold::<V::Summary>(join_set, visitor.fail_early(), skipped_summary).await?;
Ok((folded, processed))
}
pub async fn join_and_fold<S>(
mut join_set: tokio::task::JoinSet<Result<S, OperationError<S>>>,
fail_early: bool,
base: S,
) -> Result<S, OperationError<S>>
where
S: WalkSummary,
{
let mut summary = base;
let errors = crate::error_collector::ErrorCollector::default();
while let Some(res) = join_set.join_next().await {
match res {
Ok(Ok(child_summary)) => summary = summary + child_summary,
Ok(Err(error)) => {
tracing::error!("walk child failed with: {:#}", &error);
summary = summary + error.summary;
if fail_early {
return Err(OperationError::new(error.source, summary));
}
errors.push(error.source);
}
Err(join_error) => {
if fail_early {
return Err(OperationError::new(join_error.into(), summary));
}
errors.push(join_error.into());
}
}
}
if let Some(error) = errors.into_error() {
return Err(OperationError::new(error, summary));
}
Ok(summary)
}
#[cfg(test)]
mod tests {
use super::*;
use crate::filter::FilterSettings;
use crate::progress::Progress;
use std::sync::atomic::{AtomicUsize, Ordering};
use std::time::Duration;
static PROGRESS: std::sync::LazyLock<Progress> = std::sync::LazyLock::new(Progress::new);
#[derive(Clone, Copy, Default, Debug, PartialEq, Eq)]
struct CountSummary {
files: usize,
dirs: usize,
symlinks: usize,
}
impl std::ops::Add for CountSummary {
type Output = Self;
fn add(self, other: Self) -> Self {
Self {
files: self.files + other.files,
dirs: self.dirs + other.dirs,
symlinks: self.symlinks + other.symlinks,
}
}
}
struct CountingVisitor {
leaves_seen: Arc<AtomicUsize>,
}
impl WalkVisitor for CountingVisitor {
type Summary = CountSummary;
type DirContext = ();
type DirState = ();
fn root_dir_context(&self) {}
fn permit_kind(&self) -> PermitKind {
PermitKind::PendingMeta
}
fn want_permit(&self, hint: Option<EntryKind>) -> bool {
hint.is_some_and(|k| k != EntryKind::Dir)
}
fn fail_early(&self) -> bool {
false
}
fn filter(&self) -> Option<&FilterSettings> {
None
}
async fn visit_leaf(
&self,
_cx: &EntryCx,
_parent_ctx: &(),
_handle: Handle,
kind: EntryKind,
permit: Option<LeafPermit>,
) -> Result<CountSummary, OperationError<CountSummary>> {
self.leaves_seen.fetch_add(1, Ordering::SeqCst);
drop(permit);
Ok(match kind {
EntryKind::Symlink => CountSummary {
symlinks: 1,
..Default::default()
},
_ => CountSummary {
files: 1,
..Default::default()
},
})
}
async fn dir_pre(
&self,
cx: &EntryCx,
_parent_ctx: &(),
_handle: &Handle,
) -> Result<DirAction<CountSummary, (), ()>, OperationError<CountSummary>> {
let dir = cx.parent.open_dir(&cx.name).await.map_err(|err| {
OperationError::new(
anyhow::Error::new(err)
.context(format!("cannot open directory {:?}", &cx.real_path)),
Default::default(),
)
})?;
Ok(DirAction::Descend {
dir: Arc::new(dir),
child_ctx: (),
state: (),
})
}
async fn dir_post(
&self,
_cx: &EntryCx,
_state: (),
_processed: &ProcessedChildren,
child_result: Result<CountSummary, OperationError<CountSummary>>,
) -> Result<CountSummary, OperationError<CountSummary>> {
let child_summary = child_result?;
Ok(child_summary
+ CountSummary {
dirs: 1,
..Default::default()
})
}
}
fn root_cx(parent: Arc<Dir>, name: &OsStr, real_path: PathBuf) -> EntryCx {
EntryCx {
parent,
name: name.to_owned(),
rel_path: PathBuf::new(),
filter_path: PathBuf::new(),
real_path,
dry_run: false,
prog_track: &PROGRESS,
}
}
#[tokio::test]
async fn counts_entries_in_a_tree() -> anyhow::Result<()> {
let tmp = crate::testutils::setup_test_dir().await?;
let foo = tmp.join("foo");
let root = Arc::new(Dir::open_root_dir(&foo, false, congestion::Side::Source).await?);
let leaves_seen = Arc::new(AtomicUsize::new(0));
let visitor = Arc::new(CountingVisitor {
leaves_seen: Arc::clone(&leaves_seen),
});
let cx = root_cx(Arc::clone(&root), std::ffi::OsStr::new("foo"), foo.clone());
let (summary, processed) = walk_dir_contents(visitor, root, &cx, &()).await?;
assert_eq!(
summary,
CountSummary {
files: 5,
dirs: 2,
symlinks: 2,
},
"the walk must count every entry once, by kind"
);
assert_eq!(processed.names().len(), 3, "foo has three direct children");
assert_eq!(
leaves_seen.load(Ordering::SeqCst),
7,
"every non-directory leaf (5 files + 2 symlinks) was visited"
);
Ok(())
}
mod max_open_files_tests {
use super::*;
#[tokio::test]
async fn hinted_leaf_that_is_dir_drops_permit_before_recursion() -> anyhow::Result<()> {
let root = crate::testutils::create_temp_dir().await?;
let dir_path = root.join("d");
tokio::fs::create_dir(&dir_path).await?;
tokio::fs::write(dir_path.join("c"), b"x").await?;
throttle::set_max_open_files(1);
let parent = Arc::new(
Dir::open_parent_dir(&root, congestion::Side::Source)
.await?
.into_tree(),
);
let name = std::ffi::OsStr::new("d");
let handle = parent.child(name).await?;
assert_eq!(
handle.kind(),
EntryKind::Dir,
"fixture `d` must be a directory"
);
drop(handle);
let leaves_seen = Arc::new(AtomicUsize::new(0));
let visitor = Arc::new(CountingVisitor {
leaves_seen: Arc::clone(&leaves_seen),
});
let cx = root_cx(Arc::clone(&parent), name, dir_path.clone());
let permit = walk::preacquire_leaf_permit(
PermitKind::PendingMeta,
Some(EntryKind::File),
|_| true,
)
.await;
assert!(permit.is_some(), "the pre-acquire must take the one permit");
let result = tokio::time::timeout(
Duration::from_secs(20),
process_entry(visitor, cx, (), permit),
)
.await;
throttle::set_max_open_files(0);
let summary = result
.map_err(|_| {
anyhow::anyhow!(
"process_entry hung — leaf permit held across directory recursion (deadlock)"
)
})?
.map_err(|e| e.source)?;
assert_eq!(
summary,
CountSummary {
files: 1,
dirs: 1,
symlinks: 0,
},
"the directory and its one child file are both counted"
);
assert_eq!(
leaves_seen.load(Ordering::SeqCst),
1,
"the child file was visited (its permit was acquired after the drop)"
);
Ok(())
}
}
}