provenant/scanner/process/
orchestrator.rs

1// SPDX-FileCopyrightText: Provenant contributors
2// SPDX-License-Identifier: Apache-2.0
3
4use super::pipeline::process_file;
5use super::special_cases::process_directory;
6use super::spill::{FileInfoSpillStore, MemoryMode, retain_or_spill_chunk};
7use crate::license_detection::LicenseDetectionEngine;
8use crate::models::FileInfo;
9use crate::parsers::with_parser_scan_root;
10use crate::progress::ScanProgress;
11use crate::scanner::collect::CollectedPaths;
12use crate::scanner::{LicenseScanOptions, ProcessResult, TextDetectionOptions};
13use anyhow::Result;
14use rayon::prelude::*;
15use std::sync::Arc;
16
17pub fn process_collected(
18    collected: &CollectedPaths,
19    progress: Arc<ScanProgress>,
20    license_engine: Option<Arc<LicenseDetectionEngine>>,
21    license_options: LicenseScanOptions,
22    text_options: &TextDetectionOptions,
23) -> ProcessResult {
24    let mut all_files: Vec<FileInfo> = collected
25        .files
26        .par_iter()
27        .map(|(path, metadata)| {
28            let file_entry = with_parser_scan_root(collected.scan_root(), || {
29                process_file(
30                    path,
31                    metadata,
32                    progress.as_ref(),
33                    license_engine.clone(),
34                    license_options,
35                    text_options,
36                )
37            });
38            progress.file_completed(path, metadata.len(), &file_entry.scan_diagnostics);
39            file_entry
40        })
41        .collect();
42
43    for (path, metadata) in &collected.directories {
44        all_files.push(process_directory(
45            path,
46            metadata,
47            text_options.collect_info,
48            license_engine.is_some(),
49        ));
50    }
51
52    ProcessResult {
53        files: all_files,
54        excluded_count: collected.excluded_count,
55    }
56}
57
58pub fn process_collected_sequential(
59    collected: &CollectedPaths,
60    progress: Arc<ScanProgress>,
61    license_engine: Option<Arc<LicenseDetectionEngine>>,
62    license_options: LicenseScanOptions,
63    text_options: &TextDetectionOptions,
64) -> ProcessResult {
65    let mut all_files: Vec<FileInfo> =
66        Vec::with_capacity(collected.files.len() + collected.directories.len());
67
68    for (path, metadata) in &collected.files {
69        let file_entry = with_parser_scan_root(collected.scan_root(), || {
70            process_file(
71                path,
72                metadata,
73                progress.as_ref(),
74                license_engine.clone(),
75                license_options,
76                text_options,
77            )
78        });
79        progress.file_completed(path, metadata.len(), &file_entry.scan_diagnostics);
80        all_files.push(file_entry);
81    }
82
83    for (path, metadata) in &collected.directories {
84        all_files.push(process_directory(
85            path,
86            metadata,
87            text_options.collect_info,
88            license_engine.is_some(),
89        ));
90    }
91
92    ProcessResult {
93        files: all_files,
94        excluded_count: collected.excluded_count,
95    }
96}
97
98pub fn process_collected_with_memory_limit(
99    collected: &CollectedPaths,
100    progress: Arc<ScanProgress>,
101    license_engine: Option<Arc<LicenseDetectionEngine>>,
102    license_options: LicenseScanOptions,
103    text_options: &TextDetectionOptions,
104    max_in_memory: MemoryMode,
105) -> Result<ProcessResult> {
106    let Some((memory_limit, chunk_size)) = memory_limit_settings(max_in_memory) else {
107        return Ok(process_collected(
108            collected,
109            progress,
110            license_engine,
111            license_options,
112            text_options,
113        ));
114    };
115
116    let mut retained_files = Vec::new();
117    let mut spill_store: Option<FileInfoSpillStore> = None;
118
119    for chunk in collected.files.chunks(chunk_size) {
120        let processed_chunk: Vec<FileInfo> = chunk
121            .par_iter()
122            .map(|(path, metadata)| {
123                let file_entry = with_parser_scan_root(collected.scan_root(), || {
124                    process_file(
125                        path,
126                        metadata,
127                        progress.as_ref(),
128                        license_engine.clone(),
129                        license_options,
130                        text_options,
131                    )
132                });
133                progress.file_completed(path, metadata.len(), &file_entry.scan_diagnostics);
134                file_entry
135            })
136            .collect();
137
138        retain_or_spill_chunk(
139            processed_chunk,
140            &mut retained_files,
141            &mut spill_store,
142            memory_limit,
143        )?;
144    }
145
146    for (path, metadata) in &collected.directories {
147        let entry = process_directory(
148            path,
149            metadata,
150            text_options.collect_info,
151            license_engine.is_some(),
152        );
153        retain_or_spill_chunk(
154            vec![entry],
155            &mut retained_files,
156            &mut spill_store,
157            memory_limit,
158        )?;
159    }
160
161    if let Some(spill_store) = spill_store {
162        retained_files.extend(spill_store.load_all()?);
163    }
164
165    Ok(ProcessResult {
166        files: retained_files,
167        excluded_count: collected.excluded_count,
168    })
169}
170
171pub fn process_collected_with_memory_limit_sequential(
172    collected: &CollectedPaths,
173    progress: Arc<ScanProgress>,
174    license_engine: Option<Arc<LicenseDetectionEngine>>,
175    license_options: LicenseScanOptions,
176    text_options: &TextDetectionOptions,
177    max_in_memory: MemoryMode,
178) -> Result<ProcessResult> {
179    let Some((memory_limit, chunk_size)) = memory_limit_settings(max_in_memory) else {
180        return Ok(process_collected_sequential(
181            collected,
182            progress,
183            license_engine,
184            license_options,
185            text_options,
186        ));
187    };
188
189    let mut retained_files = Vec::new();
190    let mut spill_store: Option<FileInfoSpillStore> = None;
191
192    for chunk in collected.files.chunks(chunk_size) {
193        let mut processed_chunk: Vec<FileInfo> = Vec::with_capacity(chunk.len());
194        for (path, metadata) in chunk {
195            let file_entry = with_parser_scan_root(collected.scan_root(), || {
196                process_file(
197                    path,
198                    metadata,
199                    progress.as_ref(),
200                    license_engine.clone(),
201                    license_options,
202                    text_options,
203                )
204            });
205            progress.file_completed(path, metadata.len(), &file_entry.scan_diagnostics);
206            processed_chunk.push(file_entry);
207        }
208
209        retain_or_spill_chunk(
210            processed_chunk,
211            &mut retained_files,
212            &mut spill_store,
213            memory_limit,
214        )?;
215    }
216
217    for (path, metadata) in &collected.directories {
218        let entry = process_directory(
219            path,
220            metadata,
221            text_options.collect_info,
222            license_engine.is_some(),
223        );
224        retain_or_spill_chunk(
225            vec![entry],
226            &mut retained_files,
227            &mut spill_store,
228            memory_limit,
229        )?;
230    }
231
232    if let Some(spill_store) = spill_store {
233        retained_files.extend(spill_store.load_all()?);
234    }
235
236    Ok(ProcessResult {
237        files: retained_files,
238        excluded_count: collected.excluded_count,
239    })
240}
241
242fn memory_limit_settings(max_in_memory: MemoryMode) -> Option<(usize, usize)> {
243    match max_in_memory {
244        MemoryMode::CollectFirst => None,
245        MemoryMode::StreamUnlimited => Some((0, 256)),
246        MemoryMode::Limit(n) => Some((n, n.max(1))),
247    }
248}
provenant/scanner/process/orchestrator.rs

provenant/scanner/process/
orchestrator.rs
