Rustで覗き見するLinuxのプロセスとスレッド

『LINUXプログラミングインターフェース』という本を興味本位で読んでいたのですが、プロセスとスレッドに関する記述を読んでいて（いつもRustで使っているスレッドもここでいうスレッドなのかな？）と思い、スレッドモニタ用のプログラムを作成しました。

Rustプログラムの内容に合わせてLinuxのプロセス・スレッドがどのような挙動をするのかを見ていきます。

標準ライブラリを使ったマルチスレッド

monitor.rsとthread.rsの2つを作成し、以下のようなコマンドを実行することでthread.rsのプロセスと、そのスレッドの状態を監視する仕組みを作りました。

cargo run --bin thread & PID=$! && \
cargo run --bin monitor $PID

※$!は直前にバックグラウンドで実行されたプロセスのIDを返します。

thread.rs

use std::thread;
use std::time::Duration;

fn main() {
    let mut handles = vec![];

    thread::sleep(Duration::from_secs(2)); // 2秒スレッド作成を待つ

    for _ in 0..5 {
        let handle = thread::spawn(|| {
            thread::sleep(Duration::from_secs(5)); // スレッドの持続時間
        });

        handles.push(handle);
        thread::sleep(Duration::from_secs(1)); // 1秒ごとにスレッドを生成
    }

    for handle in handles {
        handle.join().unwrap();
    }
    thread::sleep(Duration::from_secs(1)); // 1秒メインスレッド終了を待つ
}

monitor.rs

use std::process::Command;
use std::io::{self, Write};
use std::env;

fn get_tids(pid: u32) -> Vec<u32> {
    let output = Command::new("ps")
        .args(&["-T", "-p", &pid.to_string()])
        .output()
        .expect("Failed to execute ps command");

    let output_str = String::from_utf8_lossy(&output.stdout);
    let mut tids = vec![];

    for line in output_str.lines().skip(1) {
        if let Some(tid_str) = line.split_whitespace().nth(1) {
            if let Ok(tid) = tid_str.parse::<u32>() {
                tids.push(tid);
            }
        }
    }

    tids
}

fn main() {
    let pid = env::args()
        .nth(1)
        .expect("Please provide a PID")
        .parse::<u32>()
        .expect("Invalid PID");

    loop {
        let tids = get_tids(pid);
        print!("\x1B[2J\x1B[1;1H"); // 画面をクリア
        println!("PID: {}", pid);
        println!("TIDs: {:?}", tids);

        io::stdout().flush().unwrap();
        // 1秒待機
        std::thread::sleep(std::time::Duration::from_secs(1));

        if tids.len() == 0 {
            break;
        }
    }
}

実行結果は以下のようになりました。

PIDはProcess ID、TIDはThread IDの略です

実行されたプログラムのプロセス内部でスレッドが順次立ち上がっていくことがわかります。

ちなみに最初に立ち上がっているPIDと同じTIDはメインスレッドで、後から順次立ち上がっていくスレッドがthread::spawnで立ち上げているスレッドですね。

tokioクレートを使ったマルチスレッド

次にtokio::spawnを使った処理をモニタしてみましょう。

ソースコードについては、冒頭に提示したソースコードをtokioを使って書き直しただけのものなため割愛します。

実行してみると以下のような結果になります。（静止画ではないです）

プログラム実行直後から17のスレッドが立ち上がっており、プログラムの終了と同時にすべてのスレッドがなくなっています。

この17という数字はメインスレッド1つと、自分のプログラムを実行したPCの論理コア数（16）の合算値です。

tokioランタイムのスレッド数はデフォルトで論理コア数と同等になっており、#[tokio::main]アトリビュートを使った場合自動的にデフォルト設定になります。

スレッド数についてはtokio::runtime::Builderを使用することで変更可能です。

#[tokio::main(worker_threads = スレッド数)] でも変更可能です。（コメントで教えていただきました！）

use tokio::runtime::Builder;

fn main() {
    let num_threads = 4; // カスタムスレッド数を設定
    let rt = Builder::new_multi_thread()
        .worker_threads(num_threads)
        .enable_all()
        .build()
        .unwrap();

    rt.block_on(async {
        // 非同期処理
        println!("Running on Tokio runtime with 4 threads.");
    });
}

標準ライブラリを使った場合はosレベルでスケジューリングが行われ、スレッドを立ち上げるたびにスレッドが作成されていましたが、tokioを利用した場合、tokioランタイムによってタスクがスケジューリングされていきます。

このような手法をグリーンスレッドと呼ぶようです。

A task is a light weight, non-blocking unit of execution. A task is similar to an OS thread, but rather than being managed by the OS scheduler, they are managed by the Tokio runtime. Another name for this general pattern is green threads. If you are familiar with Go’s goroutines, Kotlin’s coroutines, or Erlang’s processes, you can think of Tokio’s tasks as something similar.
https://docs.rs/tokio/latest/tokio/task/index.html

タスクはTokioスケジューラによって管理され、実行すべきときに実行されるようになっているのだとか。

どのスレッドで実行されるかはTokioスケジューラ次第のようで、複数のタスクが同じスレッドで実行されることもあるようです。

標準ライブラリを使ったケースではスレッドとタスクは1:1の関係になっていましたが、必ずしもそのようなかたちになるわけではないようですね。

Tasks are the unit of execution managed by the scheduler. Spawning the task submits it to the Tokio scheduler, which then ensures that the task executes when it has work to do. The spawned task may be executed on the same thread as where it was spawned, or it may execute on a different runtime thread. The task can also be moved between threads after being spawned.
https://tokio.rs/tokio/tutorial/spawning

tokioでどのようにタスクが割り振られているか見てみる

use libc;
use tokio::task;
use tokio::time::{sleep, Duration};

#[tokio::main]
async fn main() {
    let pid = unsafe { libc::getpid() };
    println!("PID: {}", pid);

    let mut handles = vec![];

    for i in 0..5 {
        let handle = task::spawn(async move {
            let tid = unsafe { libc::syscall(libc::SYS_gettid) };
            println!("TID: {:?}", tid);
            sleep(Duration::from_secs(5)).await; // スレッドの持続時間
        });

        handles.push(handle);
        sleep(Duration::from_secs(1)).await; // 1秒ごとにスレッドを生成
    }

    for handle in handles {
        handle.await.unwrap();
    }

各タスクが自身のTIDをprintln!()するよう実装したコードです。

実行結果は以下のようになりました。

二度実行しましたが、どちらも5つのタスクを2つのスレッドで処理していました。

確かにドキュメントの記載通り、複数のタスクが同じスレッドで実行されることもあるようです。

標準ライブラリとtokioでのマルチスレッドの違いまとめ

	標準ライブラリ	tokio
スケジューリング	OSレベル	tokioランタイム
1OSスレッドあたりのタスク	1	多
実行単位	OSスレッド	グリーンスレッド

標準ライブラリのスレッドはOSによって管理され、各スレッドはOSスレッドに対応していました。

一方、tokioの場合はtokioランタイムによってタスクが管理されており、複数のタスクが一つのOSスレッド上で実行されることが確認できました。

Linuxのプロセスとスレッド

参考にした書籍の通りの結果が観測できたように思います。

例えば実行されたRustプログラムは一つのプロセスとして処理され、標準ライブラリを使ったケースではそのプロセス内部でスレッドが立ち上がっては消えていくのを確認できました。

---

ちなみに今回検証に使ったソースコードについては以下リポジトリで公開しました。興味ある方はよかったらどうぞです。

https://github.com/torohash/multi-thread-monitor