ディレクトリとそのサブディレクトリを非同期で探索するにはどうすればよいですか?

3
Nick 2019-06-22 16:21.

ディレクトリとそのすべてのサブディレクトリを調べる必要があります。同期的に再帰を使用してディレクトリを簡単に探索できます。

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| | }
  | |_^

すべてのエラーを伝播しながら、ディレクトリとそのサブディレクトリを非同期に探索する正しい方法は何ですか?

2 answers

8
Shepmaster 2019-11-13 02:44.

私はロドリゴの既存の答えにいくつかの変更を加えます:

  1. Stream関数からaを返し、呼び出し元が特定のファイルエントリで必要なことを実行できるようにします。
  2. impl Stream代わりにを返しBox<dyn Stream>ます。これにより、実装の柔軟性が高まります。たとえば、効率の低い再帰型の代わりに内部スタックを使用するカスタム型を作成できます。
  3. io::Error関数から戻り、ユーザーがエラーに対処できるようにします。
  4. impl Into<PathBuf>より良いAPIを許可するには、を受け入れます。
  5. APIで具象型を使用する内部隠し実装関数を作成します。

先物0.3 /時尾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 /時尾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(()))
    });
}

参照:

  • 安定したRustで非同期Futureで計算された値を同期的に返すにはどうすればよいですか?
3
rodrigo 2019-06-23 00:17.

コードに2つのエラーがあります。

まず、impl 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)
}

2番目のエラーがまだあります:

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の完全なコードを参照してください。

Related questions

MORE COOL STUFF

ケイト・ブランシェットは3日間一緒に夫と一緒に寝て、25年経ってもまだ夫と結婚しています

ケイト・ブランシェットは3日間一緒に夫と一緒に寝て、25年経ってもまだ夫と結婚しています

ケイト・ブランシェットは、夫に会ったとき、典型的な交際のアドバイスに逆らいました。

マイケルシーンが非営利の俳優である理由

マイケルシーンが非営利の俳優である理由

マイケルシーンは非営利の俳優ですが、それは正確にはどういう意味ですか?

ホールマークスターのコリンエッグレスフィールドがRomaDramaLiveでスリル満点のファンと出会う![エクスクルーシブ]

ホールマークスターのコリンエッグレスフィールドがRomaDramaLiveでスリル満点のファンと出会う![エクスクルーシブ]

特徴的なスターのコリン・エッグレスフィールドは、RomaDrama Liveでのスリル満点のファンとの出会いについて料理しました!加えて、大会での彼のINSPIREプログラム。

「たどりつけば」をオンラインでストリーミングできない理由

「たどりつけば」をオンラインでストリーミングできない理由

ノーザンエクスポージャーが90年代の最も人気のある番組の1つになった理由を確認するには、Blu-rayまたはDVDプレーヤーをほこりで払う必要があります。

バイオニック読書はあなたをより速く読むことができますか?

バイオニック読書はあなたをより速く読むことができますか?

BionicReadingアプリの人気が爆発的に高まっています。しかし、それは本当にあなたを速読術にすることができますか?

ドミニカのボイリング湖:アクセスは簡単ではありませんが、ハイキングする価値があります

ドミニカのボイリング湖:アクセスは簡単ではありませんが、ハイキングする価値があります

ドミニカのボイリング湖は、世界で2番目に大きいボイリング湖です。そこにたどり着くまでのトレッキングは大変で長いですが、努力する価値は十分にあります。

私たちの水をきれいに保つのを助けるためにあなたの髪を寄付してください

私たちの水をきれいに保つのを助けるためにあなたの髪を寄付してください

サロンからのヘアトリミングや個人的な寄付は、油流出を吸収して環境を保護するのに役立つマットとして再利用できます。

ホワイトハウスの最も記憶に残る結婚式を見てください

ホワイトハウスの最も記憶に残る結婚式を見てください

過去200年以上の間にホワイトハウスで結婚したのはほんの数人です。彼らは誰でしたか、そしてそこで結婚式を獲得するために何が必要ですか?

驚くほど素晴らしいDropMixミュージックミキシングカードゲームは30ドルで驚くべき取引です

驚くほど素晴らしいDropMixミュージックミキシングカードゲームは30ドルで驚くべき取引です

DropMixはNFC対応のカードゲームで、基本的にはリミックスアーティストになります。現在、Amazonでは$ 30まで下がっており、これまでで最高の価格に匹敵します。ロックバンドで有名なHarmonixによって開発されたDropMixは、おそらく少し野心的すぎるように思われます。結局のところ、ほとんどの人は素晴らしいリズムを持っていませんが、ゲームは驚くほどうまく実行されます。

メアリーJ.ブライジがついにハリウッドウォークオブフェイムスターを獲得

メアリーJ.ブライジがついにハリウッドウォークオブフェイムスターを獲得

写真:APメアリーJ.ブライジは、間もなくハリウッドウォークオブフェイムのスターを獲得します。これは、メアリーJよりもハリウッドウォークオブフェイムのほうが正直なところ恩恵です。

MeltdownとSpectreの脆弱性についてこれまでに知っていることはすべて、簡単な方法で説明されています

MeltdownとSpectreの脆弱性についてこれまでに知っていることはすべて、簡単な方法で説明されています

画像:グラズ工科大学/ NataschaEiblがデザインしたロゴ。MeltdownとSpectreは、攻撃者がシステムメモリに保存されているあらゆる種類の情報にアクセスできるようにする2つの脆弱性に付けられた名前です。

彼のニューヨークの家から追い出されようとしている97歳の第二次世界大戦の獣医。メリーエフィングクリスマス

彼のニューヨークの家から追い出されようとしている97歳の第二次世界大戦の獣医。メリーエフィングクリスマス

日本人に襲われたときに真珠湾にいた97歳の第二次世界大戦のベテランが、ニューヨークのブルックリンから追い出されています。

Zendaya Wishes Boyfriend Tom Holland Happy Birthday with Cuddly Photo: He 'Makes Me the Happiest'

Zendaya Wishes Boyfriend Tom Holland Happy Birthday with Cuddly Photo: He 'Makes Me the Happiest'

Zendaya shared a sweet photo in honor of boyfriend Tom Holland's 26th birthday Wednesday

小さな女性:脳卒中を患った後に病院から解放されたアトランタのジューシーな赤ちゃん:「まだ癒し」

小さな女性:脳卒中を患った後に病院から解放されたアトランタのジューシーな赤ちゃん:「まだ癒し」

シーレン「Ms.JuicyBaby」ピアソンは、先月脳卒中で入院した後、「もう一度たくさんのことをする方法を学ばなければならない」ため、言語療法を受けていることを明らかにしました。

エマストーンは彼女のクリフサイドマリブビーチハウスを420万ドルでリストアップしています—中を見てください!

エマストーンは彼女のクリフサイドマリブビーチハウスを420万ドルでリストアップしています—中を見てください!

オスカー受賞者の世紀半ばの家には、3つのベッドルーム、2つのバス、オーシャンフロントの景色があります。

ジーニー・メイ・ジェンキンスは、母乳育児の経験の中で、彼女は「本当に、本当に落ち込んでいる」と言います

ジーニー・メイ・ジェンキンスは、母乳育児の経験の中で、彼女は「本当に、本当に落ち込んでいる」と言います

ジーニー・メイ・ジェンキンスは、生後4か月の娘、モナコに母乳育児をしていると語った。

投資ノート:Bioscout AU$300万シード

投資ノート:Bioscout AU$300万シード

Bioscoutは、農家を運転席に置くという使命を負っています。Artesian(GrainInnovate)やUniseedと並んで、最新のシードラウンドでチームを支援できることをうれしく思います。問題真菌症による重大な作物の損失は、農民にとって試練であることが証明されています。

リトルマーケットリサーチ1| 2022年のクイックグリンプス遠隔医療市場

リトルマーケットリサーチ1| 2022年のクイックグリンプス遠隔医療市場

遠隔医療は、パンデミック後の時代では新しいものではなく、時代遅れの分野でもありません。しかし、業界を詳しく見ると、需要と供給の強力な持続可能性と、米国で絶え間ない革命となる強力な潜在的成長曲線を示しています。

スタートアップ資金調達環境:タイのスタートアップエコシステムの次は何ですか?

スタートアップ資金調達環境:タイのスタートアップエコシステムの次は何ですか?

2021年は、世界的なベンチャーキャピタル(VC)の資金調達にとって記録的な年でした。DealStreetAsiaによると、東南アジアも例外ではなく、この地域では年間で記録的な25の新しいユニコーンが採掘されました。

ムーアの法則を超えて

ムーアの法則を超えて

計算に対する私たちの欲求とムーアの法則が提供できるものとの間には、指数関数的に増大するギャップがあります。私たちの文明は計算に基づいています—建築と想像力の現在の限界を超える技術を見つけなければなりません。

Language