Skip to main content

pagers_core/
crawl.rs

1use std::io::{self, BufRead};
2use std::path::{Path, PathBuf};
3use std::sync::atomic::Ordering;
4
5use ignore::WalkBuilder;
6
7use crate::mincore::PageMap;
8use crate::mode::DisplayMode;
9use crate::ops::{FileRange, Op, Stats};
10use crate::par::{InodeSet, SeenInodes as _};
11
12#[cfg(feature = "rayon")]
13pub use crate::par::Threads;
14
15#[derive(Debug, Clone, PartialEq)]
16#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))]
17pub struct CrawlConfig {
18    pub follow_symlinks: bool,
19    pub single_filesystem: bool,
20    pub count_hardlinks: bool,
21    pub ignore_patterns: Vec<String>,
22    pub filter_patterns: Vec<String>,
23    pub max_file_size: Option<u64>,
24    pub batch: Option<PathBuf>,
25    pub nul_delim: bool,
26    #[cfg(feature = "rayon")]
27    pub threads: Threads,
28}
29
30pub fn crawl_and_process<O: Op, PM: PageMap + Send + Sync, D: DisplayMode<PM>>(
31    paths: &[PathBuf],
32    crawl_config: &CrawlConfig,
33    op: &O,
34    range: &FileRange,
35    stats: &Stats,
36    display: &D,
37) -> crate::Result<Vec<O::Output>> {
38    tracing::info!("starting {} on {} path(s)", O::LABEL, paths.len());
39    let seen_inodes = InodeSet::default();
40
41    #[cfg(feature = "rayon")]
42    {
43        use rayon::prelude::*;
44
45        let pool = rayon::ThreadPoolBuilder::new()
46            .num_threads(crawl_config.threads.num_threads())
47            .build()?;
48
49        let buf = std::thread::available_parallelism().map_or(16, |n| n.get() * 4);
50        let (tx, rx) = std::sync::mpsc::sync_channel::<PathBuf>(buf);
51
52        let outputs = pool.install(|| {
53            rayon::scope(|s| {
54                s.spawn({
55                    let tx = tx;
56                    move |_| {
57                        collect_paths(paths, crawl_config, &seen_inodes, stats, |p| {
58                            let _ = tx.send(p);
59                        });
60                    }
61                });
62
63                rx.into_iter()
64                    .par_bridge()
65                    .filter_map(|path| display.process_one::<O>(op, &path, range, stats))
66                    .collect::<Vec<_>>()
67            })
68        });
69
70        display.finish();
71        op.finish()?;
72        tracing::info!(
73            "done: {} files, {} pages",
74            stats.total_files.load(Ordering::Relaxed),
75            stats.total_pages.load(Ordering::Relaxed),
76        );
77        Ok(outputs)
78    }
79
80    #[cfg(not(feature = "rayon"))]
81    {
82        let mut file_paths = Vec::new();
83        collect_paths(paths, crawl_config, &seen_inodes, stats, |p| {
84            file_paths.push(p);
85        });
86        tracing::info!("discovered {} files", file_paths.len());
87        let outputs = file_paths
88            .iter()
89            .filter_map(|path| display.process_one::<O>(op, path, range, stats))
90            .collect();
91        display.finish();
92        op.finish()?;
93        tracing::info!(
94            "done: {} files, {} pages",
95            stats.total_files.load(Ordering::Relaxed),
96            stats.total_pages.load(Ordering::Relaxed),
97        );
98        Ok(outputs)
99    }
100}
101
102fn collect_paths(
103    paths: &[PathBuf],
104    crawl_config: &CrawlConfig,
105    seen_inodes: &InodeSet,
106    stats: &Stats,
107    mut emit: impl FnMut(PathBuf),
108) {
109    let mut all_paths: Vec<PathBuf> = paths.to_vec();
110
111    if let Some(batch_path) = &crawl_config.batch {
112        match read_batch_paths(batch_path, crawl_config.nul_delim) {
113            Ok(batch_paths) => all_paths.extend(batch_paths),
114            Err(e) => tracing::warn!("batch file: {e}"),
115        }
116    }
117
118    let needs_meta = crawl_config.max_file_size.is_some() || !crawl_config.count_hardlinks;
119
120    for path in &all_paths {
121        if path.is_dir() {
122            tracing::info!("crawling directory {}", path.display());
123            stats.total_dirs.fetch_add(1, Ordering::Relaxed);
124            walk_dir_entries(path, crawl_config, needs_meta, seen_inodes, &mut emit);
125        } else if path.is_file() {
126            emit(path.clone());
127        } else {
128            tracing::warn!("skipping {}: not a file or directory", path.display());
129        }
130    }
131}
132
133fn walk_dir_entries(
134    root: &Path,
135    config: &CrawlConfig,
136    needs_meta: bool,
137    seen_inodes: &InodeSet,
138    mut emit: impl FnMut(PathBuf),
139) {
140    let mut builder = WalkBuilder::new(root);
141    builder
142        .follow_links(config.follow_symlinks)
143        .same_file_system(config.single_filesystem)
144        .hidden(false)
145        .git_ignore(false)
146        .git_global(false)
147        .git_exclude(false);
148
149    if !config.ignore_patterns.is_empty() || !config.filter_patterns.is_empty() {
150        let mut overrides = ignore::overrides::OverrideBuilder::new(root);
151        for pat in &config.ignore_patterns {
152            let _ = overrides.add(&format!("!{pat}"));
153        }
154        for pat in &config.filter_patterns {
155            let _ = overrides.add(pat);
156        }
157        if let Ok(ov) = overrides.build() {
158            builder.overrides(ov);
159        }
160    }
161
162    for entry in builder.build() {
163        let Ok(entry) = entry.inspect_err(|e| tracing::warn!("{e}")) else {
164            continue;
165        };
166
167        let Some(ft) = entry.file_type() else {
168            continue;
169        };
170
171        if !ft.is_file() {
172            continue;
173        }
174
175        let entry_path = entry.path();
176        let meta = if needs_meta {
177            fs_err::metadata(entry_path).ok()
178        } else {
179            None
180        };
181
182        if let Some(max_size) = config.max_file_size
183            && let Some(ref m) = meta
184            && m.len() > max_size
185        {
186            continue;
187        }
188
189        if !config.count_hardlinks {
190            #[cfg(unix)]
191            {
192                use std::os::unix::fs::MetadataExt;
193                if let Some(ref m) = meta
194                    && m.nlink() > 1
195                    && seen_inodes.already_seen((m.dev(), m.ino()))
196                {
197                    continue;
198                }
199            }
200        }
201
202        emit(entry_path.to_path_buf());
203    }
204}
205
206pub fn read_batch_paths(path: &Path, nul_delim: bool) -> io::Result<Vec<PathBuf>> {
207    use std::os::unix::ffi::OsStrExt;
208
209    let reader: Box<dyn BufRead> = if path == Path::new("-") {
210        Box::new(io::stdin().lock())
211    } else {
212        Box::new(io::BufReader::new(fs_err::File::open(path)?))
213    };
214
215    let delim = if nul_delim { b'\0' } else { b'\n' };
216    reader
217        .split(delim)
218        .filter_map(|r| match r {
219            Ok(buf) if !buf.is_empty() => {
220                Some(Ok(PathBuf::from(std::ffi::OsStr::from_bytes(&buf))))
221            }
222            Ok(_) => None,
223            Err(e) => Some(Err(e)),
224        })
225        .collect()
226}