[rust] mutex系のパフォーマンスを調べてみる

前回の記事でasync/awaitの書き方を複数パターンを考えてみましたが今度はmutex系のパターンを試してみます。

ロックして排他的に読み書きできる機能は幾つかクレートが提供されていますがtokio、標準ライブラリ、parking_lot提供のものを試してみます。

環境はvCPU=6、worker=5000です。

std::sync::Mutex

グローバル変数で直接mutex系を使用することは以前はできなかったようですが、ver1.63より可能になったということで試してみます。これはおそらくLinux系のみで他プラットフォームは依然OnceCell等と組み合わせが必要かと思います。

use anyhow::Result; // 1.0.71
use clap::Parser; // 4.3.11

use std::sync::{Arc, Mutex};

use tokio::fs; // 1.32.0
use tokio::task::JoinSet;

#[derive(Parser, Debug)]
#[command(author, version, about, long_about = None)]
struct Args {
    /// target directory
    #[arg(short, long, default_value_t = String::from(".") )]
    dir: String,
}

static LIST_FROM_THD: std::sync::Mutex<Vec<String>> = Mutex::new(Vec::<String>::new());

#[tokio::main]
async fn main() -> Result<()> {
    let args = Args::parse();

    let dir_string = args.dir.to_string();
    println!("{}", dir_string);

    let mut dir_list;

    // Get lock and push dir
    {
        let mut lock = LIST_FROM_THD.lock().unwrap();
        lock.push(dir_string);
    }

    // Like a do-while
    while {
        // Get lock and take array as snapshot
        {
            let mut lock = LIST_FROM_THD.lock().unwrap();
            dir_list = lock.to_vec();
            lock.clear();
        }

        !dir_list.is_empty()
    } {
        let mut thds = JoinSet::new();
        let mut dir_list = dir_list.iter();
        while let Some(item) = dir_list.next() {
            thds.spawn(get_dirs(item.to_string()));
        }

        while let Some(thd) = thds.join_next().await {
            let _ = thd;
        }
    }

    Ok(())
}

async fn get_dirs(dir: String) -> Result<()> {
    let mut entries = fs::read_dir(dir).await?;

    // Folder list
    let dirs = Arc::new(Mutex::new(Vec::new()));

    let mut thds = JoinSet::new();

    while let Some(entry) = entries.next_entry().await? {
        let metadata = entry.metadata().await?;
        let path = entry.path();

        thds.spawn(
            (|path: std::path::PathBuf, dirs: std::sync::Arc<Mutex<Vec<String>>>| async move {
                let path = path.display().to_string();
                if metadata.is_dir() {
                    println!("{}", path);

                    // Get lock and add array
                    {
                        let mut lock = dirs.lock().unwrap();
                        lock.push(path);
                    }
                }
            })(path.clone(), dirs.clone()),
        );

        thds.spawn(async move {
            if let Ok(symlink) = fs::read_link(&path).await {
                if path.is_dir() {
                    println!("{}@ -> {}", path.display(), symlink.display());
                }
            }
        });
    }

    while let Some(thd) = thds.join_next().await {
        let _ = thd;
    }

    // Get lock and add array
    {
        let mut lock = LIST_FROM_THD.lock().unwrap();
        let mut dirs = dirs.lock().unwrap();
        lock.append(&mut dirs);
    }

    Ok(())
}

std::sync::RwLock

tokio::sync::OnceCellとRwLockの組み合わせです。

Vecを使ったのでwrite lockのみ使用しています。

use anyhow::Result; // 1.0.71
use clap::Parser; // 4.3.11

use std::sync::Arc;
use std::sync::RwLock;

use tokio::fs; // 1.32.0
use tokio::sync::OnceCell;
use tokio::task::JoinSet;

#[derive(Parser, Debug)]
#[command(author, version, about, long_about = None)]
struct Args {
    /// target directory
    #[arg(short, long, default_value_t = String::from(".") )]
    dir: String,
}

static LIST_FROM_THD: tokio::sync::OnceCell<RwLock<Vec<String>>> = OnceCell::const_new();

#[tokio::main]
async fn main() -> Result<()> {
    let args = Args::parse();

    let dir_string = args.dir.to_string();
    println!("{}", dir_string);

    let _ = LIST_FROM_THD.set(RwLock::new(Vec::<String>::new()));
    let mut dir_list;

    // Get lock and push dir
    {
        let mut lock = LIST_FROM_THD
            .get_or_init(|| async move { RwLock::new(Vec::<String>::new()) })
            .await
            .write()
            .unwrap();
        lock.push(dir_string);
    }

    // Like a do-while
    while {
        // Get lock and take array as snapshot
        {
            let mut lock = LIST_FROM_THD
                .get_or_init(|| async move { RwLock::new(Vec::<String>::new()) })
                .await
                .write()
                .unwrap();
            dir_list = lock.to_vec();
            lock.clear();
        }

        !dir_list.is_empty()
    } {
        let mut thds = JoinSet::new();
        let mut dir_list = dir_list.iter();
        while let Some(item) = dir_list.next() {
            thds.spawn(get_dirs(item.to_string()));
        }

        while let Some(thd) = thds.join_next().await {
            let _ = thd;
        }
    }

    Ok(())
}

async fn get_dirs(dir: String) -> Result<()> {
    let mut entries = fs::read_dir(dir).await?;

    // Folder list
    let dirs = Arc::new(RwLock::new(Vec::new()));

    let mut thds = JoinSet::new();

    while let Some(entry) = entries.next_entry().await? {
        let metadata = entry.metadata().await?;
        let path = entry.path();

        thds.spawn(
            (|path: std::path::PathBuf, dirs: std::sync::Arc<RwLock<Vec<String>>>| async move {
                let path = path.display().to_string();
                if metadata.is_dir() {
                    println!("{}", path);

                    // Get lock and add array
                    {
                        let mut lock = dirs.write().ok().unwrap();
                        lock.push(path);
                    }
                }
            })(path.clone(), dirs.clone()),
        );

        thds.spawn(async move {
            if let Ok(symlink) = fs::read_link(&path).await {
                if path.is_dir() {
                    println!("{}@ -> {}", path.display(), symlink.display());
                }
            }
        });
    }

    while let Some(thd) = thds.join_next().await {
        let _ = thd;
    }

    // Get lock and add array
    {
        let mut lock = LIST_FROM_THD
            .get_or_init(|| async move { RwLock::new(Vec::<String>::new()) })
            .await
            .write()
            .unwrap();
        let mut dirs = dirs.write().ok().unwrap();
        lock.append(&mut dirs);
    }

    Ok(())
}

tokio::sync::Mutex

tokio::sync::OnceCellとMutexの組み合わせです。どちらも非同期対応しています。その為メソッドチェーンが長めです。

use anyhow::Result; // 1.0.71
use clap::Parser; // 4.3.11

use std::sync::Arc;

use tokio::fs; // 1.32.0
use tokio::sync::{Mutex, OnceCell};
use tokio::task::JoinSet;

#[derive(Parser, Debug)]
#[command(author, version, about, long_about = None)]
struct Args {
    /// target directory
    #[arg(short, long, default_value_t = String::from(".") )]
    dir: String,
}

static LIST_FROM_THD: tokio::sync::OnceCell<tokio::sync::Mutex<Vec<String>>> =
    OnceCell::const_new();

#[tokio::main]
async fn main() -> Result<()> {
    let args = Args::parse();

    let dir_string = args.dir.to_string();
    println!("{}", dir_string);

    let _ = LIST_FROM_THD.set(Mutex::new(Vec::<String>::new()));
    let mut dir_list;

    // Get lock and push dir
    {
        let mut lock = LIST_FROM_THD
            .get_or_init(|| async move { Mutex::new(Vec::<String>::new()) })
            .await
            .lock()
            .await;
        lock.push(dir_string);
    }

    // Like a do-while
    while {
        // Get lock and take array as snapshot
        {
            let mut lock = LIST_FROM_THD
                .get_or_init(|| async move { Mutex::new(Vec::<String>::new()) })
                .await
                .lock()
                .await;
            dir_list = lock.to_vec();
            lock.clear();
        }

        !dir_list.is_empty()
    } {
        let mut thds = JoinSet::new();
        let mut dir_list = dir_list.iter();
        while let Some(item) = dir_list.next() {
            thds.spawn(get_dirs(item.to_string()));
        }

        while let Some(thd) = thds.join_next().await {
            let _ = thd;
        }
    }

    Ok(())
}

async fn get_dirs(dir: String) -> Result<()> {
    let mut entries = fs::read_dir(dir).await?;

    // Folder list
    let dirs = Arc::new(Mutex::new(Vec::new()));

    let mut thds = JoinSet::new();

    while let Some(entry) = entries.next_entry().await? {
        let metadata = entry.metadata().await?;
        let path = entry.path();

        thds.spawn(
            (|path: std::path::PathBuf, dirs: std::sync::Arc<Mutex<Vec<String>>>| async move {
                let path = path.display().to_string();
                if metadata.is_dir() {
                    println!("{}", path);

                    // Get lock and add array
                    {
                        let mut lock = dirs.lock().await;
                        lock.push(path);
                    }
                }
            })(path.clone(), dirs.clone()),
        );

        thds.spawn(async move {
            if let Ok(symlink) = fs::read_link(&path).await {
                if path.is_dir() {
                    println!("{}@ -> {}", path.display(), symlink.display());
                }
            }
        });
    }

    while let Some(thd) = thds.join_next().await {
        let _ = thd;
    }

    // Get lock and add array
    {
        let mut lock = LIST_FROM_THD
            .get_or_init(|| async move { Mutex::new(Vec::<String>::new()) })
            .await
            .lock()
            .await;
        let mut dirs = dirs.lock().await;
        lock.append(&mut dirs);
    }

    Ok(())
}

tokio::sync::RwLock

tokio::sync::OnceCellとRwLockの組み合わせです。こちらも非同期対応しています。同じくwrite lockのみ使用しています。

use anyhow::Result; // 1.0.71
use clap::Parser; // 4.3.11

use std::sync::Arc;

use tokio::fs; // 1.32.0
use tokio::sync::{OnceCell, RwLock};
use tokio::task::JoinSet;

#[derive(Parser, Debug)]
#[command(author, version, about, long_about = None)]
struct Args {
    /// target directory
    #[arg(short, long, default_value_t = String::from(".") )]
    dir: String,
}

static LIST_FROM_THD: tokio::sync::OnceCell<tokio::sync::RwLock<Vec<String>>> =
    OnceCell::const_new();

// #[tokio::main]
#[tokio::main(flavor = "multi_thread", worker_threads = 5000)]
async fn main() -> Result<()> {
    let args = Args::parse();

    let dir_string = args.dir.to_string();
    println!("{}", dir_string);

    let _ = LIST_FROM_THD.set(RwLock::new(Vec::<String>::new()));
    let mut dir_list;

    // Get lock and push dir
    {
        let mut lock = LIST_FROM_THD
            .get_or_init(|| async move { RwLock::new(Vec::<String>::new()) })
            .await
            .write()
            .await;
        lock.push(dir_string);
    }

    // Like a do-while
    while {
        // Get lock and take array as snapshot
        {
            let mut lock = LIST_FROM_THD
                .get_or_init(|| async move { RwLock::new(Vec::<String>::new()) })
                .await
                .write()
                .await;
            dir_list = lock.to_vec();
            lock.clear();
        }

        !dir_list.is_empty()
    } {
        let mut thds = JoinSet::new();
        let mut dir_list = dir_list.iter();
        while let Some(item) = dir_list.next() {
            thds.spawn(get_dirs(item.to_string()));
        }

        while let Some(thd) = thds.join_next().await {
            let _ = thd;
        }
    }

    Ok(())
}

async fn get_dirs(dir: String) -> Result<()> {
    let mut entries = fs::read_dir(dir).await?;

    // Folder list
    let dirs = Arc::new(RwLock::new(Vec::new()));

    let mut thds = JoinSet::new();

    while let Some(entry) = entries.next_entry().await? {
        let metadata = entry.metadata().await?;
        let path = entry.path();

        thds.spawn(
            (|path: std::path::PathBuf, dirs: std::sync::Arc<RwLock<Vec<String>>>| async move {
                let path = path.display().to_string();
                if metadata.is_dir() {
                    println!("{}", path);

                    // Get lock and add array
                    {
                        let mut lock = dirs.write().await;
                        lock.push(path);
                    }
                }
            })(path.clone(), dirs.clone()),
        );

        thds.spawn(async move {
            if let Ok(symlink) = fs::read_link(&path).await {
                if path.is_dir() {
                    println!("{}@ -> {}", path.display(), symlink.display());
                }
            }
        });
    }

    while let Some(thd) = thds.join_next().await {
        let _ = thd;
    }

    // Get lock and add array
    {
        let mut lock = LIST_FROM_THD
            .get_or_init(|| async move { RwLock::new(Vec::<String>::new()) })
            .await
            .write()
            .await;
        let mut dirs = dirs.write().await;
        lock.append(&mut dirs);
    }

    Ok(())
}

parking_lot::Mutex

tokio::sync::OnceCellとparking_lot::Mutexの組み合わせです。

parking_lot版はコンテキストスイッチが起きない場合に同じスレッドがロックを再取得しつづけないように考慮して公平性を保つ設計となっているようです。汚染されない事も利点の一つ用のようです。

use anyhow::Result; // 1.0.71
use clap::Parser; // 4.3.11

use std::sync::Arc;

use tokio::fs; // 1.32.0
use tokio::sync::OnceCell;
use tokio::task::JoinSet;

use parking_lot::Mutex;

#[derive(Parser, Debug)]
#[command(author, version, about, long_about = None)]
struct Args {
    /// target directory
    #[arg(short, long, default_value_t = String::from(".") )]
    dir: String,
}

static LIST_FROM_THD: tokio::sync::OnceCell<parking_lot::Mutex<Vec<String>>> =
    OnceCell::const_new();

#[tokio::main]
async fn main() -> Result<()> {
    let args = Args::parse();

    let dir_string = args.dir.to_string();
    println!("{}", dir_string);

    let _ = LIST_FROM_THD.set(Mutex::new(Vec::<String>::new()));
    let mut dir_list;

    // Get lock and push dir
    {
        let mut lock = LIST_FROM_THD
            .get_or_init(|| async move { Mutex::new(Vec::<String>::new()) })
            .await
            .lock();
        lock.push(dir_string);
    }

    // Like a do-while
    while {
        // Get lock and take array as snapshot
        {
            let mut lock = LIST_FROM_THD
                .get_or_init(|| async move { Mutex::new(Vec::<String>::new()) })
                .await
                .lock();
            dir_list = lock.to_vec();
            lock.clear();
        }

        !dir_list.is_empty()
    } {
        let mut thds = JoinSet::new();
        let mut dir_list = dir_list.iter();
        while let Some(item) = dir_list.next() {
            thds.spawn(get_dirs(item.to_string()));
        }

        while let Some(thd) = thds.join_next().await {
            let _ = thd;
        }
    }

    Ok(())
}

async fn get_dirs(dir: String) -> Result<()> {
    let mut entries = fs::read_dir(dir).await?;

    // Folder list
    let dirs = Arc::new(Mutex::new(Vec::new()));

    let mut thds = JoinSet::new();

    while let Some(entry) = entries.next_entry().await? {
        let metadata = entry.metadata().await?;
        let path = entry.path();

        thds.spawn(
            (|path: std::path::PathBuf, dirs: std::sync::Arc<Mutex<Vec<String>>>| async move {
                let path = path.display().to_string();
                if metadata.is_dir() {
                    println!("{}", path);

                    // Get lock and add array
                    {
                        let mut lock = dirs.lock();
                        lock.push(path);
                    }
                }
            })(path.clone(), dirs.clone()),
        );

        thds.spawn(async move {
            if let Ok(symlink) = fs::read_link(&path).await {
                if path.is_dir() {
                    println!("{}@ -> {}", path.display(), symlink.display());
                }
            }
        });
    }

    while let Some(thd) = thds.join_next().await {
        let _ = thd;
    }

    // Get lock and add array
    {
        let mut lock = LIST_FROM_THD
            .get_or_init(|| async move { Mutex::new(Vec::<String>::new()) })
            .await
            .lock();
        let mut dirs = dirs.lock();
        lock.append(&mut dirs);
    }

    Ok(())
}

parking_lot::FairMutex

tokio::sync::OnceCellとparking_lot::FairMutexの組み合わせです。

parking_lot::Mutexとの違いはロック要求をキューに入れて公平性を保つ点のようです。

use anyhow::Result; // 1.0.71
use clap::Parser; // 4.3.11

use std::sync::Arc;

use tokio::fs; // 1.32.0
use tokio::sync::OnceCell;
use tokio::task::JoinSet;

use parking_lot::FairMutex; // 0.12.1

#[derive(Parser, Debug)]
#[command(author, version, about, long_about = None)]
struct Args {
    /// target directory
    #[arg(short, long, default_value_t = String::from(".") )]
    dir: String,
}

static LIST_FROM_THD: tokio::sync::OnceCell<parking_lot::FairMutex<Vec<String>>> =
    OnceCell::const_new();

#[tokio::main]
async fn main() -> Result<()> {
    let args = Args::parse();

    let dir_string = args.dir.to_string();
    println!("{}", dir_string);

    let _ = LIST_FROM_THD.set(FairMutex::new(Vec::<String>::new()));
    let mut dir_list;

    // Get lock and push dir
    {
        let mut lock = LIST_FROM_THD
            .get_or_init(|| async move { FairMutex::new(Vec::<String>::new()) })
            .await
            .lock();
        lock.push(dir_string);
    }

    // Like a do-while
    while {
        // Get lock and take array as snapshot
        {
            let mut lock = LIST_FROM_THD
                .get_or_init(|| async move { FairMutex::new(Vec::<String>::new()) })
                .await
                .lock();
            dir_list = lock.to_vec();
            lock.clear();
        }

        !dir_list.is_empty()
    } {
        let mut thds = JoinSet::new();
        let mut dir_list = dir_list.iter();
        while let Some(item) = dir_list.next() {
            thds.spawn(get_dirs(item.to_string()));
        }

        while let Some(thd) = thds.join_next().await {
            let _ = thd;
        }
    }

    Ok(())
}

async fn get_dirs(dir: String) -> Result<()> {
    let mut entries = fs::read_dir(dir).await?;

    // Folder list
    let dirs = Arc::new(FairMutex::new(Vec::new()));

    let mut thds = JoinSet::new();

    while let Some(entry) = entries.next_entry().await? {
        let metadata = entry.metadata().await?;
        let path = entry.path();

        thds.spawn(
            (|path: std::path::PathBuf, dirs: std::sync::Arc<FairMutex<Vec<String>>>| async move {
                let path = path.display().to_string();
                if metadata.is_dir() {
                    println!("{}", path);

                    // Get lock and add array
                    {
                        let mut lock = dirs.lock();
                        lock.push(path);
                    }
                }
            })(path.clone(), dirs.clone()),
        );

        thds.spawn(async move {
            if let Ok(symlink) = fs::read_link(&path).await {
                if path.is_dir() {
                    println!("{}@ -> {}", path.display(), symlink.display());
                }
            }
        });
    }

    while let Some(thd) = thds.join_next().await {
        let _ = thd;
    }

    // Get lock and add array
    {
        let mut lock = LIST_FROM_THD
            .get_or_init(|| async move { FairMutex::new(Vec::<String>::new()) })
            .await
            .lock();
        let mut dirs = dirs.lock();
        lock.append(&mut dirs);
    }

    Ok(())
}

parking_lot::RwLock

tokio::sync::OnceCellとparking_lot::RwLockの組み合わせです。

設計的にはparking_lot::Mutexと同等のデザインのようです。FairRwLockは無いです。

use anyhow::Result; // 1.0.71
use clap::Parser; // 4.3.11

use std::sync::Arc;

use tokio::fs; // 1.32.0
use tokio::sync::OnceCell;
use tokio::task::JoinSet;

use parking_lot::RwLock; // 0.12.1

#[derive(Parser, Debug)]
#[command(author, version, about, long_about = None)]
struct Args {
    /// target directory
    #[arg(short, long, default_value_t = String::from(".") )]
    dir: String,
}

static LIST_FROM_THD: tokio::sync::OnceCell<parking_lot::RwLock<Vec<String>>> =
    OnceCell::const_new();

#[tokio::main]
async fn main() -> Result<()> {
    let args = Args::parse();

    let dir_string = args.dir.to_string();
    println!("{}", dir_string);

    let _ = LIST_FROM_THD.set(RwLock::new(Vec::<String>::new()));
    let mut dir_list;

    // Get lock and push dir
    {
        let mut lock = LIST_FROM_THD
            .get_or_init(|| async move { RwLock::new(Vec::<String>::new()) })
            .await
            .write();
        lock.push(dir_string);
    }

    // Like a do-while
    while {
        // Get lock and take array as snapshot
        {
            let mut lock = LIST_FROM_THD
                .get_or_init(|| async move { RwLock::new(Vec::<String>::new()) })
                .await
                .write();
            dir_list = lock.to_vec();
            lock.clear();
        }

        !dir_list.is_empty()
    } {
        let mut thds = JoinSet::new();
        let mut dir_list = dir_list.iter();
        while let Some(item) = dir_list.next() {
            thds.spawn(get_dirs(item.to_string()));
        }

        while let Some(thd) = thds.join_next().await {
            let _ = thd;
        }
    }

    Ok(())
}

async fn get_dirs(dir: String) -> Result<()> {
    let mut entries = fs::read_dir(dir).await?;

    // Folder list
    let dirs = Arc::new(RwLock::new(Vec::new()));

    let mut thds = JoinSet::new();

    while let Some(entry) = entries.next_entry().await? {
        let metadata = entry.metadata().await?;
        let path = entry.path();

        thds.spawn(
            (|path: std::path::PathBuf, dirs: std::sync::Arc<RwLock<Vec<String>>>| async move {
                let path = path.display().to_string();
                if metadata.is_dir() {
                    println!("{}", path);

                    // Get lock and add array
                    {
                        let mut lock = dirs.write();
                        lock.push(path);
                    }
                }
            })(path.clone(), dirs.clone()),
        );

        thds.spawn(async move {
            if let Ok(symlink) = fs::read_link(&path).await {
                if path.is_dir() {
                    println!("{}@ -> {}", path.display(), symlink.display());
                }
            }
        });
    }

    while let Some(thd) = thds.join_next().await {
        let _ = thd;
    }

    // Get lock and add array
    {
        let mut lock = LIST_FROM_THD
            .get_or_init(|| async move { RwLock::new(Vec::<String>::new()) })
            .await
            .write();
        let mut dirs = dirs.write();
        lock.append(&mut dirs);
    }

    Ok(())
}

計測結果

前回と同様にmultitimeで50回まわしてみます。比較としてデフォルト（#[tokio::main]のみ）設定の結果も付けました。

結果としては全体的に+5秒程度増えただけで大きなボトルネックになってはいないようです。あるいはロック解除待ち時間を非同期処理で効率的に行えているのかもしれません。

あまり差がないのでコンテキストスイッチ等他の要因が増分の可能性もあります。

何れにせよボトルネックにはなっていない、あるいはボトルネックになっていても非同期処理でカバーできているのだと思いますのでmutex系の処理速度に気を使う必要はなさそうです。

この中で使うとしたら汚染が無いなどメリットのあるparking_lotのものでしょうか。

しかしmutex系を使わなくても済みそうなので次回は無しバージョンのコードを書いてみたいと思います。

std::sync::Mutex

std::sync::RwLock

tokio::sync::Mutex

tokio::sync::RwLock

parking_lot::Mutex

parking_lot::FairMutex

parking_lot::RwLock