如何异步浏览目录及其子目录?
How to asynchronously explore a directory and its sub-directories?
我需要浏览一个目录及其所有子目录。我可以通过同步递归轻松探索目录:
use failure::Error;
use std::fs;
use std::path::Path;
fn main() -> Result<(), Error> {
visit(Path::new("."))
}
fn visit(path: &Path) -> Result<(), Error> {
for e in fs::read_dir(path)? {
let e = e?;
let path = e.path();
if path.is_dir() {
visit(&path)?;
} else if path.is_file() {
println!("File: {:?}", path);
}
}
Ok(())
}
当我尝试使用 tokio_fs
以异步方式执行相同操作时:
use failure::Error; // 0.1.6
use futures::Future; // 0.1.29
use std::path::PathBuf;
use tokio::{fs, prelude::*}; // 0.1.22
fn visit(path: PathBuf) -> impl Future<Item = (), Error = Error> {
let task = fs::read_dir(path)
.flatten_stream()
.for_each(|entry| {
println!("{:?}", entry.path());
let path = entry.path();
if path.is_dir() {
let task = visit(entry.path());
tokio::spawn(task.map_err(drop));
}
future::ok(())
})
.map_err(Error::from);
task
}
我收到以下错误:
error[E0391]: cycle detected when processing `visit::{{opaque}}#0`
--> src/lib.rs:6:28
|
6 | fn visit(path: PathBuf) -> impl Future<Item = (), Error = Error> {
| ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
|
note: ...which requires processing `visit`...
--> src/lib.rs:6:1
|
6 | fn visit(path: PathBuf) -> impl Future<Item = (), Error = Error> {
| ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
= note: ...which requires evaluating trait selection obligation `futures::future::map_err::MapErr<impl futures::future::Future, fn(failure::error::Error) {std::mem::drop::<failure::error::Error>}>: std::marker::Send`...
= note: ...which again requires processing `visit::{{opaque}}#0`, completing the cycle
note: cycle used when checking item types in top-level module
--> src/lib.rs:1:1
|
1 | / use failure::Error; // 0.1.6
2 | | use futures::Future; // 0.1.29
3 | | use std::path::PathBuf;
4 | | use tokio::{fs, prelude::*}; // 0.1.22
... |
20| | task
21| | }
| |_^
error[E0391]: cycle detected when processing `visit::{{opaque}}#0`
--> src/lib.rs:6:28
|
6 | fn visit(path: PathBuf) -> impl Future<Item = (), Error = Error> {
| ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
|
note: ...which requires processing `visit`...
--> src/lib.rs:6:1
|
6 | fn visit(path: PathBuf) -> impl Future<Item = (), Error = Error> {
| ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
= note: ...which again requires processing `visit::{{opaque}}#0`, completing the cycle
note: cycle used when checking item types in top-level module
--> src/lib.rs:1:1
|
1 | / use failure::Error; // 0.1.6
2 | | use futures::Future; // 0.1.29
3 | | use std::path::PathBuf;
4 | | use tokio::{fs, prelude::*}; // 0.1.22
... |
20| | task
21| | }
| |_^
在传播所有错误的同时异步探索目录及其子目录的正确方法是什么?
您的代码有两个错误:
首先,函数 returning impl Trait
当前不能递归,因为实际类型 returned 将取决于它自己。
要使您的示例正常工作,您需要 return 大小适中的类型。简单的候选是一个trait对象,即一个Box<dyn Future<...>>
:
fn visit(path: PathBuf) -> Box<dyn Future<Item = (), Error = Error>> {
// ...
let task = visit(entry.path());
tokio::spawn(task.map_err(drop));
// ...
Box::new(task)
}
还有你的第二个错误:
error[E0277]: `dyn futures::future::Future<Item = (), Error = failure::error::Error>` cannot be sent between threads safely
--> src/lib.rs:14:30
|
14 | tokio::spawn(task.map_err(drop));
| ^^^^^^^^^^^^^^^^^^ `dyn futures::future::Future<Item = (), Error = failure::error::Error>` cannot be sent between threads safely
|
::: /root/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-0.1.22/src/executor/mod.rs:131:52
|
131 | where F: Future<Item = (), Error = ()> + 'static + Send
| ---- required by this bound in `tokio::executor::spawn`
|
= help: the trait `std::marker::Send` is not implemented for `dyn futures::future::Future<Item = (), Error = failure::error::Error>`
= note: required because of the requirements on the impl of `std::marker::Send` for `std::ptr::Unique<dyn futures::future::Future<Item = (), Error = failure::error::Error>>`
= note: required because it appears within the type `std::boxed::Box<dyn futures::future::Future<Item = (), Error = failure::error::Error>>`
= note: required because it appears within the type `futures::future::map_err::MapErr<std::boxed::Box<dyn futures::future::Future<Item = (), Error = failure::error::Error>>, fn(failure::error::Error) {std::mem::drop::<failure::error::Error>}>`
这意味着您的特征对象不是 Send
,因此无法使用 tokio::spawn()
安排在另一个线程中执行。幸运的是,这很容易解决:只需将 + Send
添加到您的特征对象中:
fn visit(path: PathBuf) -> Box<dyn Future<Item = (), Error = Error> + Send> {
//...
}
请参阅 Playground 中的完整代码。
我会对做一些修改:
- Return 来自函数的
Stream
,允许调用者对给定的文件条目执行他们需要的操作。
- Return
impl Stream
而不是 Box<dyn Stream>
。这为实施提供了更大的灵活性。例如,可以创建使用内部堆栈而不是效率较低的递归类型的自定义类型。
- Return
io::Error
允许用户处理任何错误的功能。
- 接受
impl Into<PathBuf>
以允许更好的 API。
- 创建一个在其 API.
中使用具体类型的内部隐藏实现函数
期货 0.3 / Tokio 0.2
在这个版本中,我避免了深度递归调用,保留了本地访问路径堆栈 (to_visit
)。
use futures::{stream, Stream, StreamExt}; // 0.3.1
use std::{io, path::PathBuf};
use tokio::fs::{self, DirEntry}; // 0.2.4
fn visit(path: impl Into<PathBuf>) -> impl Stream<Item = io::Result<DirEntry>> + Send + 'static {
async fn one_level(path: PathBuf, to_visit: &mut Vec<PathBuf>) -> io::Result<Vec<DirEntry>> {
let mut dir = fs::read_dir(path).await?;
let mut files = Vec::new();
while let Some(child) = dir.next_entry().await? {
if child.metadata().await?.is_dir() {
to_visit.push(child.path());
} else {
files.push(child)
}
}
Ok(files)
}
stream::unfold(vec![path.into()], |mut to_visit| {
async {
let path = to_visit.pop()?;
let file_stream = match one_level(path, &mut to_visit).await {
Ok(files) => stream::iter(files).map(Ok).left_stream(),
Err(e) => stream::once(async { Err(e) }).right_stream(),
};
Some((file_stream, to_visit))
}
})
.flatten()
}
#[tokio::main]
async fn main() {
let root_path = std::env::args().nth(1).expect("One argument required");
let paths = visit(root_path);
paths
.for_each(|entry| {
async {
match entry {
Ok(entry) => println!("visiting {:?}", entry),
Err(e) => eprintln!("encountered an error: {}", e),
}
}
})
.await;
}
期货 0.1 / Tokio 0.1
use std::path::PathBuf;
use tokio::{fs, prelude::*}; // 0.1.22
use tokio_fs::DirEntry; // 1.0.6
fn visit(
path: impl Into<PathBuf>,
) -> impl Stream<Item = DirEntry, Error = std::io::Error> + Send + 'static {
fn visit_inner(
path: PathBuf,
) -> Box<dyn Stream<Item = DirEntry, Error = std::io::Error> + Send + 'static> {
Box::new({
fs::read_dir(path)
.flatten_stream()
.map(|entry| {
let path = entry.path();
if path.is_dir() {
// Optionally include `entry` if you want to
// include directories in the resulting
// stream.
visit_inner(path)
} else {
Box::new(stream::once(Ok(entry)))
}
})
.flatten()
})
}
visit_inner(path.into())
}
fn main() {
tokio::run({
let root_path = std::env::args().nth(1).expect("One argument required");
let paths = visit(root_path);
paths
.then(|entry| {
match entry {
Ok(entry) => println!("visiting {:?}", entry),
Err(e) => eprintln!("encountered an error: {}", e),
};
Ok(())
})
.for_each(|_| Ok(()))
});
}
另请参阅:
我需要浏览一个目录及其所有子目录。我可以通过同步递归轻松探索目录:
use failure::Error;
use std::fs;
use std::path::Path;
fn main() -> Result<(), Error> {
visit(Path::new("."))
}
fn visit(path: &Path) -> Result<(), Error> {
for e in fs::read_dir(path)? {
let e = e?;
let path = e.path();
if path.is_dir() {
visit(&path)?;
} else if path.is_file() {
println!("File: {:?}", path);
}
}
Ok(())
}
当我尝试使用 tokio_fs
以异步方式执行相同操作时:
use failure::Error; // 0.1.6
use futures::Future; // 0.1.29
use std::path::PathBuf;
use tokio::{fs, prelude::*}; // 0.1.22
fn visit(path: PathBuf) -> impl Future<Item = (), Error = Error> {
let task = fs::read_dir(path)
.flatten_stream()
.for_each(|entry| {
println!("{:?}", entry.path());
let path = entry.path();
if path.is_dir() {
let task = visit(entry.path());
tokio::spawn(task.map_err(drop));
}
future::ok(())
})
.map_err(Error::from);
task
}
我收到以下错误:
error[E0391]: cycle detected when processing `visit::{{opaque}}#0`
--> src/lib.rs:6:28
|
6 | fn visit(path: PathBuf) -> impl Future<Item = (), Error = Error> {
| ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
|
note: ...which requires processing `visit`...
--> src/lib.rs:6:1
|
6 | fn visit(path: PathBuf) -> impl Future<Item = (), Error = Error> {
| ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
= note: ...which requires evaluating trait selection obligation `futures::future::map_err::MapErr<impl futures::future::Future, fn(failure::error::Error) {std::mem::drop::<failure::error::Error>}>: std::marker::Send`...
= note: ...which again requires processing `visit::{{opaque}}#0`, completing the cycle
note: cycle used when checking item types in top-level module
--> src/lib.rs:1:1
|
1 | / use failure::Error; // 0.1.6
2 | | use futures::Future; // 0.1.29
3 | | use std::path::PathBuf;
4 | | use tokio::{fs, prelude::*}; // 0.1.22
... |
20| | task
21| | }
| |_^
error[E0391]: cycle detected when processing `visit::{{opaque}}#0`
--> src/lib.rs:6:28
|
6 | fn visit(path: PathBuf) -> impl Future<Item = (), Error = Error> {
| ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
|
note: ...which requires processing `visit`...
--> src/lib.rs:6:1
|
6 | fn visit(path: PathBuf) -> impl Future<Item = (), Error = Error> {
| ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
= note: ...which again requires processing `visit::{{opaque}}#0`, completing the cycle
note: cycle used when checking item types in top-level module
--> src/lib.rs:1:1
|
1 | / use failure::Error; // 0.1.6
2 | | use futures::Future; // 0.1.29
3 | | use std::path::PathBuf;
4 | | use tokio::{fs, prelude::*}; // 0.1.22
... |
20| | task
21| | }
| |_^
在传播所有错误的同时异步探索目录及其子目录的正确方法是什么?
您的代码有两个错误:
首先,函数 returning impl Trait
当前不能递归,因为实际类型 returned 将取决于它自己。
要使您的示例正常工作,您需要 return 大小适中的类型。简单的候选是一个trait对象,即一个Box<dyn Future<...>>
:
fn visit(path: PathBuf) -> Box<dyn Future<Item = (), Error = Error>> {
// ...
let task = visit(entry.path());
tokio::spawn(task.map_err(drop));
// ...
Box::new(task)
}
还有你的第二个错误:
error[E0277]: `dyn futures::future::Future<Item = (), Error = failure::error::Error>` cannot be sent between threads safely
--> src/lib.rs:14:30
|
14 | tokio::spawn(task.map_err(drop));
| ^^^^^^^^^^^^^^^^^^ `dyn futures::future::Future<Item = (), Error = failure::error::Error>` cannot be sent between threads safely
|
::: /root/.cargo/registry/src/github.com-1ecc6299db9ec823/tokio-0.1.22/src/executor/mod.rs:131:52
|
131 | where F: Future<Item = (), Error = ()> + 'static + Send
| ---- required by this bound in `tokio::executor::spawn`
|
= help: the trait `std::marker::Send` is not implemented for `dyn futures::future::Future<Item = (), Error = failure::error::Error>`
= note: required because of the requirements on the impl of `std::marker::Send` for `std::ptr::Unique<dyn futures::future::Future<Item = (), Error = failure::error::Error>>`
= note: required because it appears within the type `std::boxed::Box<dyn futures::future::Future<Item = (), Error = failure::error::Error>>`
= note: required because it appears within the type `futures::future::map_err::MapErr<std::boxed::Box<dyn futures::future::Future<Item = (), Error = failure::error::Error>>, fn(failure::error::Error) {std::mem::drop::<failure::error::Error>}>`
这意味着您的特征对象不是 Send
,因此无法使用 tokio::spawn()
安排在另一个线程中执行。幸运的是,这很容易解决:只需将 + Send
添加到您的特征对象中:
fn visit(path: PathBuf) -> Box<dyn Future<Item = (), Error = Error> + Send> {
//...
}
请参阅 Playground 中的完整代码。
我会对
- Return 来自函数的
Stream
,允许调用者对给定的文件条目执行他们需要的操作。 - Return
impl Stream
而不是Box<dyn Stream>
。这为实施提供了更大的灵活性。例如,可以创建使用内部堆栈而不是效率较低的递归类型的自定义类型。 - Return
io::Error
允许用户处理任何错误的功能。 - 接受
impl Into<PathBuf>
以允许更好的 API。 - 创建一个在其 API. 中使用具体类型的内部隐藏实现函数
期货 0.3 / Tokio 0.2
在这个版本中,我避免了深度递归调用,保留了本地访问路径堆栈 (to_visit
)。
use futures::{stream, Stream, StreamExt}; // 0.3.1
use std::{io, path::PathBuf};
use tokio::fs::{self, DirEntry}; // 0.2.4
fn visit(path: impl Into<PathBuf>) -> impl Stream<Item = io::Result<DirEntry>> + Send + 'static {
async fn one_level(path: PathBuf, to_visit: &mut Vec<PathBuf>) -> io::Result<Vec<DirEntry>> {
let mut dir = fs::read_dir(path).await?;
let mut files = Vec::new();
while let Some(child) = dir.next_entry().await? {
if child.metadata().await?.is_dir() {
to_visit.push(child.path());
} else {
files.push(child)
}
}
Ok(files)
}
stream::unfold(vec![path.into()], |mut to_visit| {
async {
let path = to_visit.pop()?;
let file_stream = match one_level(path, &mut to_visit).await {
Ok(files) => stream::iter(files).map(Ok).left_stream(),
Err(e) => stream::once(async { Err(e) }).right_stream(),
};
Some((file_stream, to_visit))
}
})
.flatten()
}
#[tokio::main]
async fn main() {
let root_path = std::env::args().nth(1).expect("One argument required");
let paths = visit(root_path);
paths
.for_each(|entry| {
async {
match entry {
Ok(entry) => println!("visiting {:?}", entry),
Err(e) => eprintln!("encountered an error: {}", e),
}
}
})
.await;
}
期货 0.1 / Tokio 0.1
use std::path::PathBuf;
use tokio::{fs, prelude::*}; // 0.1.22
use tokio_fs::DirEntry; // 1.0.6
fn visit(
path: impl Into<PathBuf>,
) -> impl Stream<Item = DirEntry, Error = std::io::Error> + Send + 'static {
fn visit_inner(
path: PathBuf,
) -> Box<dyn Stream<Item = DirEntry, Error = std::io::Error> + Send + 'static> {
Box::new({
fs::read_dir(path)
.flatten_stream()
.map(|entry| {
let path = entry.path();
if path.is_dir() {
// Optionally include `entry` if you want to
// include directories in the resulting
// stream.
visit_inner(path)
} else {
Box::new(stream::once(Ok(entry)))
}
})
.flatten()
})
}
visit_inner(path.into())
}
fn main() {
tokio::run({
let root_path = std::env::args().nth(1).expect("One argument required");
let paths = visit(root_path);
paths
.then(|entry| {
match entry {
Ok(entry) => println!("visiting {:?}", entry),
Err(e) => eprintln!("encountered an error: {}", e),
};
Ok(())
})
.for_each(|_| Ok(()))
});
}
另请参阅: