harmos_storage/directory/
scan.rs

1/* ***
2 *
3 * Harmos
4 *
5 * SPDX-License-Identifier: Apache-2.0
6 *
7 * ***
8 */
9
10// Bounded Depth-First Enumeration With Retained Entry Ownership
11
12/* === Modules === */
13
14/* === Re-Exports === */
15
16/* === Imports === */
17
18// Standard Library Imports
19use std::fs::{self, ReadDir};
20use std::path::Path;
21use std::path::PathBuf;
22use std::sync::atomic::{AtomicBool, Ordering};
23use std::sync::{Arc, Mutex};
24// Third Party Imports
25use tokio::task::JoinHandle;
26// Local Imports
27use super::{DirectoryCompletion, DirectoryEntry, Selection, join_error, relative, retained_item, selection};
28use crate::capacity::{Held, Pool, Refusal, Reservation};
29use crate::manager::{ENTRY_OVERHEAD, Inner};
30use crate::{Error, ReadBatch, Result};
31use harmos_domain::model::storage::DirectorySelection;
32
33/* === Constants === */
34
35/* === Macros === */
36
37/* === Type Aliases === */
38
39type Scanned = (DirectoryEntry, Reservation);
40type Pending = JoinHandle<Result<Option<Scanned>>>;
41
42/* === Structs === */
43
44pub struct DirectoryReader {
45    context: crate::execution::Context,
46    pub(super) inner: Arc<Inner>,
47    state: Arc<Mutex<Scan>>,
48    pending: Option<Pending>,
49    cancelled: Arc<AtomicBool>,
50}
51
52struct Scan {
53    inner: Arc<Inner>,
54    held: Held,
55    selection: Arc<Selection>,
56    stack: Vec<Frame>,
57    visited: usize,
58    entries: u64,
59    complete: bool,
60    cancelled: Arc<AtomicBool>,
61}
62
63struct Frame {
64    directory: ReadDir,
65    _path: PathBuf,
66    _memory: Reservation,
67}
68
69/* === Enums === */
70
71/* === Traits === */
72
73/* === Implementations === */
74
75impl DirectoryReader {
76    pub(super) async fn new(inner: Arc<Inner>, value: DirectorySelection) -> Result<Self> {
77        let held = Held::default();
78        let selected = selection(&inner, value, &held).await?;
79        Self::selected(inner, selected, held).await
80    }
81
82    pub(super) async fn selected(inner: Arc<Inner>, selection: Arc<Selection>, held: Held) -> Result<Self> {
83        let cancelled = Arc::new(AtomicBool::new(false));
84        let scan = Scan {
85            inner: Arc::clone(&inner),
86            held: held.clone(),
87            selection,
88            stack: Vec::new(),
89            visited: 0,
90            entries: 0,
91            complete: false,
92            cancelled: Arc::clone(&cancelled),
93        };
94        let scan = {
95            let context = crate::execution::Context::current();
96            tokio::task::spawn_blocking(move || {
97                context.blocking(|| {
98                    let mut scan = scan;
99                    let selection = Arc::clone(&scan.selection);
100                    scan.enter(&selection.value.root)?;
101                    Ok::<_, Error>(scan)
102                })
103            })
104        }
105        .await
106        .map_err(join_error)??;
107        Ok(Self { context: crate::execution::Context::current(), inner, state: Arc::new(Mutex::new(scan)), pending: None, cancelled })
108    }
109
110    pub async fn next(&mut self) -> Result<Option<ReadBatch<DirectoryEntry>>> {
111        self.next_value().await?.map(|(value, memory)| retained_item(&self.inner, value, memory)).transpose()
112    }
113
114    pub fn completion(&self) -> Result<DirectoryCompletion> {
115        if self.pending.is_some() {
116            return Err(Error::NotReady { domain: "directory completion" });
117        }
118        let scan = self.state.lock().unwrap();
119        if !scan.complete {
120            return Err(Error::NotReady { domain: "directory completion" });
121        }
122        Ok(DirectoryCompletion { entries: scan.entries })
123    }
124
125    pub(super) async fn next_value(&mut self) -> Result<Option<Scanned>> {
126        let context = self.context.clone();
127        let span = context.sync(|| tracing::debug_span!("storage.directory.page", outcome = "cancelled"));
128        context
129            .result(span, async {
130                if self.pending.is_none() {
131                    let state = Arc::clone(&self.state);
132                    let context = self.context.clone();
133                    self.pending = Some(tokio::task::spawn_blocking(move || context.blocking(|| state.lock().unwrap().next())));
134                }
135                let result = self.pending.as_mut().unwrap().await;
136                self.pending = None;
137                result.map_err(join_error)?
138            })
139            .await
140    }
141}
142
143impl Drop for DirectoryReader {
144    fn drop(&mut self) {
145        self.cancelled.store(true, Ordering::Release);
146    }
147}
148
149impl Scan {
150    fn enter(&mut self, path: &Path) -> Result<()> {
151        let bytes = path.as_os_str().len().saturating_mul(6).saturating_add(std::mem::size_of::<Frame>() * 2 + ENTRY_OVERHEAD);
152        let memory = Reservation::try_new(&self.inner, Pool::Read, &self.held, bytes).map_err(Refusal::into_error)?;
153        let path = crate::local::validate_local_path(path)?;
154        let directory = fs::read_dir(&path).map_err(crate::local::io_error)?;
155        self.stack.push(Frame { directory, _path: path, _memory: memory });
156        Ok(())
157    }
158
159    fn next(&mut self) -> Result<Option<Scanned>> {
160        loop {
161            if self.cancelled.load(Ordering::Acquire) {
162                return Err(Error::Closing { domain: "directory scan" });
163            }
164            let Some(frame) = self.stack.last_mut() else {
165                self.complete = true;
166                return Ok(None);
167            };
168            let Some(entry) = frame.directory.next() else {
169                self.stack.pop();
170                continue;
171            };
172            let entry = entry.map_err(crate::local::io_error)?;
173            self.visited = self.visited.checked_add(1).ok_or(Error::ResourceExhausted { resource: "directory entries" })?;
174            if self.visited > self.inner.settings.filesystem.max_entries.get() {
175                return Err(Error::ResourceExhausted { resource: "directory entries" });
176            }
177            let name = entry.file_name();
178            let bytes = frame._path.capacity().saturating_add(name.capacity()).saturating_mul(3).saturating_add(ENTRY_OVERHEAD);
179            let memory = Reservation::try_new(&self.inner, Pool::Read, &self.held, bytes).map_err(Refusal::into_error)?;
180            let path = frame._path.join(name);
181            let metadata = match fs::symlink_metadata(&path) {
182                Ok(metadata) => metadata,
183                Err(error) if error.kind() == std::io::ErrorKind::NotFound => continue,
184                Err(error) => return Err(crate::local::io_error(error)),
185            };
186            if crate::local::linked(&metadata) {
187                continue;
188            }
189            if metadata.is_dir() && self.selection.value.recursive {
190                self.enter(&path)?;
191            }
192            let Some(path) = relative(&self.selection.value.root, &path, self.selection.value.recursive)? else { continue };
193            if !super::selection::selected(&path, &self.selection.value.rules) {
194                continue;
195            }
196            self.entries += 1;
197            let entry = DirectoryEntry { path, directory: metadata.is_dir(), length: if metadata.is_file() { metadata.len() } else { 0 } };
198            return Ok(Some((entry, memory)));
199        }
200    }
201}
202
203/* === Public Functions === */
204
205/* === Private Functions === */
206
207/* === Tests === */
harmos_storage/directory/scan.rs

harmos_storage/directory/
scan.rs
