1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
|
use std::ops::Deref;
use std::path::Path;
use std::sync::Arc;
use anyhow::Error;
use anyhow::Result;
use tokio_stream::Stream;
/// Helper type to limit access to the filesystem in massive-parallel cases where we would run into
/// system error 24 ("too many open files").
#[derive(Clone)]
pub struct FileSystemAccessor {
sem: Arc<tokio::sync::Semaphore>,
}
impl FileSystemAccessor {
pub fn new(limit: usize) -> Self {
FileSystemAccessor { sem: Arc::new(tokio::sync::Semaphore::new(limit)) }
}
#[inline]
pub fn available_permits(&self) -> usize {
self.sem.available_permits()
}
pub async fn read_dir(&self, path: impl AsRef<Path>) -> Result<impl Stream<Item = Result<DirEntry>>> {
log::trace!("Accessor::read_dir({}): Acquiring Semaphore: {} - 1", path.as_ref().display(), self.available_permits());
let perm = self.sem.acquire().await?;
let sema = self.sem.clone();
let mut read_dir = tokio::fs::read_dir(&path).await?;
let stream = async_stream::stream! {
loop {
log::trace!("Accessor::read_dir(_): Acquiring Semaphore: {} - 1", sema.available_permits());
let permit = sema.clone().acquire_owned().await?;
if let Some(entry) = read_dir.next_entry().await? {
let dentry = DirEntry { permit, entry };
yield Ok(dentry) as Result<_>;
} else {
break;
}
}
};
log::trace!("Accessor::read_dir({}): Dropping Semaphore: {} + 1", path.as_ref().display(), self.available_permits());
drop(perm);
Ok(stream)
}
pub async fn exists(&self, p: impl AsRef<Path>) -> Result<bool> {
log::trace!("Accessor::exists({}): Acquiring Semaphore: {} - 1", p.as_ref().display(), self.available_permits());
let perm = self.sem.acquire().await?;
let r = p.as_ref().exists();
log::trace!("Accessor::exists({}): Dropping Semaphore: {} + 1", p.as_ref().display(), self.available_permits());
drop(perm);
Ok(r)
}
pub async fn is_file(&self, p: impl AsRef<Path>) -> Result<bool> {
log::trace!("Accessor::is_file({}): Acquiring Semaphore: {} - 1", p.as_ref().display(), self.available_permits());
let perm = self.sem.acquire().await?;
let r = p.as_ref().is_file();
log::trace!("Accessor::is_file({}): Dropping Semaphore: {} + 1", p.as_ref().display(), self.available_permits());
drop(perm);
Ok(r)
}
pub async fn read_to_string(&self, p: impl AsRef<Path>) -> Result<String> {
log::trace!("Accessor::read_to_string({}): Acquiring Semaphore: {} - 1", p.as_ref().display(), self.available_permits());
let perm = self.sem.acquire().await?;
let r = tokio::fs::read_to_string(&p).await.map_err(Error::from);
log::trace!("Accessor::read_to_string({}): Dropping Semaphore: {} + 1", p.as_ref().display(), self.available_permits());
drop(perm);
r
}
}
#[derive(Debug)]
pub struct DirEntry {
permit: tokio::sync::OwnedSemaphorePermit,
entry: tokio::fs::DirEntry,
}
impl Deref for DirEntry {
type Target = tokio::fs::DirEntry;
fn deref(&self) -> &Self::Target {
&self.entry
}
}
|
