Category: Rust

use threadpool::ThreadPool;
use std::sync::mpsc;

pub struct Data { n: i32 }

impl Data {
    fn incr(&mut self) { self.n += 1; }
}

fn main() {
    
    let n_workers = 4;
    let pool = ThreadPool::new(n_workers);
    let (tx, rx) = mpsc::channel();
    let v = vec![Data{n:0}, Data{n:1}, Data{n:2}];
    let n_jobs = v.len();
    for mut data in v {
        let tx = tx.clone();
        pool.execute(move || {
            data.incr();
            tx.send(data).expect("channel will be there waiting for the pool");
        });
    }
    let sum: i32 = rx.iter().take(n_jobs).map(|data| data.n).sum();
    println!("sum= {}", sum);
}

sum= 6

thread poolはデフォルトの選択肢？

use threadpool::ThreadPool;
use std::sync::mpsc;
use std::thread;


fn main() {
    
    let n_workers = 4;
    let pool = ThreadPool::new(n_workers);
    let (tx, rx) = mpsc::channel();
    for i in 0..20 {
        let tx = tx.clone();
        pool.execute(move || {
            // println!("{}", thread::current().name().unwrap());
            tx.send(i).expect("channel will be there waiting for the pool");        
        });
    }
    for i in 0..20 {
        let j = rx.recv().unwrap();
        println!("{}", j);
    }
}

let (mut tx, mut rx) = channel();でチャンネルを開設したとき、tx, rxの型が何かに苦戦しました。結局、pipe_channel::Sender、pipe_channel::Receiverになるのね。相互にやり取りするのではなく、一方的にデータを送るときなどに有効活用できそうです。

use std::thread;
use pipe_channel::*;

fn writer(mut tx: pipe_channel::Sender<String>) {
    let name = "Writer".to_string();
    let handle = thread::spawn(move || {
        println!("{}: Sending rubber duck...", name);
        tx.send("Rubber duck".to_string()).unwrap();
    });
    handle.join().unwrap();
}

fn reader(mut rx: pipe_channel::Receiver<String>) {
    let name = "Reader".to_string();
    let handle = thread::spawn(move || {
        println!("{}: Reading...", name);
        println!("{}: received ({})", name, rx.recv().unwrap());
    });
    handle.join().unwrap();
}

fn main() {
    let (mut tx, mut rx) = channel();
    writer(tx);
    reader(rx);
}

fn print_typename<T>(_: T) {
    println!("{}", std::any::type_name::<T>());
}

Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.29s
Running `target/debug/parallel`
Writer: Sending rubber duck…
Reader: Reading…
Reader: received (Rubber duck)

shared_memory = [-1] * SIZE は、[-1, -1, -1, -1, -1] になり、producerで[0, 1, 2, 3, 4]に変わる。ここがややこしい。

import time
from threading import Thread, current_thread

SIZE = 5
shared_memory = [-1] * SIZE

class Producer(Thread):
    def run(self) -> None:
        self.name = "Producer"
        global shared_memory
        for i in range(SIZE):
            print(f"{current_thread().name}: Writing {int(i)}")
            shared_memory[i - 1] = i

class Consumer(Thread):
    def run(self) -> None:
        self.name = "Consumer"
        global shared_memory
        for i in range(SIZE):
            while True:
                line = shared_memory[i]
                if line == -1:
                    print(f"{current_thread().name}: Data not available\n"
                            f"Sleeping for 1 second before retrying")
                    time.sleep(1)
                    continue
                print(f"{current_thread().name}: Read: {int(line)}")
                break

def main() -> None:
    threads = [
        Consumer(),
        Producer(),
    ]

    for thread in threads:
        thread.start()

    for thread in threads:
        thread.join()

if __name__ == "__main__":
    main()
    

use nix::unistd::{fork, getpid, getppid, ForkResult};
use std::thread;
use std::time::Duration;

fn main() {
    thread::spawn(|| {
        for i in 1..10 {
            println!("hi number {} from the spawned thread!", i);
            thread::sleep(Duration::from_millis(1));
        }
    });

    for i in 1..5 {
        println!("hi number {} from the main thread!", i);
        thread::sleep(Duration::from_millis(1));
    }
}

Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.75s
Running `target/debug/parallel`
hi number 1 from the main thread!
hi number 1 from the spawned thread!
hi number 2 from the main thread!
hi number 2 from the spawned thread!
hi number 3 from the spawned thread!
hi number 3 from the main thread!
hi number 4 from the spawned thread!
hi number 4 from the main thread!
hi number 5 from the spawned thread!

joinを呼び出すことで、実行されたことを保証する

fn main() {
    let handle = thread::spawn(|| {
        for i in 1..5 {
            println!("hi number {} from the spawned thread!", i);
            thread::sleep(Duration::from_millis(1));
        }
    });

    for i in 1..5 {
        println!("hi number {} from the main thread!", i);
        thread::sleep(Duration::from_millis(1));
    }

    handle.join().unwrap();
}

所有権の移転

fn main() {
    let v = vec![1, 2, 3]; 
    let handle = thread::spawn(move|| {
        for i in 0..3 {
            println!("hi number {} from the spawned thread!", v[i]);
            thread::sleep(Duration::from_millis(1));
        }
    });

    for i in 1..3 {
        println!("hi number {} from the main thread!", i);
        thread::sleep(Duration::from_millis(1));
    }

    handle.join().unwrap();
}

なんか違うような気がするが…

use nix::unistd::{fork, getpid, getppid, ForkResult};
use std::thread;
use std::time::Duration;
use rayon::current_num_threads;

fn cpu_waster(i: u32) {
    let name = gettid::gettid().to_string();
    println!("{} doing {} work", name, i);
    thread::sleep(Duration::from_millis(5));
}

fn display_threads() {
    let str: String = "-".to_string();
    println!("{}", str.repeat(10));
    println!("Current process PID: {}", getpid());
    println!("Thread Count: {}", current_num_threads());
}

fn main() {
    let num_threads = 5;
    display_threads();

    println!("Starting {} CPU waters...", num_threads);
    let handle = thread::spawn(move|| {
        for i in 0..num_threads {
            cpu_waster(i);
        }
    });
    display_threads();

    handle.join().unwrap();
}

Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.23s
Running `target/debug/parallel`
———-
Current process PID: 582557
Thread Count: 2
Starting 5 CPU waters…
———-
Current process PID: 582557
Thread Count: 2
582600 doing 0 work
582600 doing 1 work
582600 doing 2 work
582600 doing 3 work
582600 doing 4 work

nixを使います。

nix = “0.16.1”

use nix::unistd::{getpid};

fn main() {
    println!("My id is {}", getpid());
}

My id is 577891

### 子プロセスの生成

use nix::unistd::{fork, getpid, getppid, ForkResult};
use nix::sys::wait::waitpid;
use std::process::exit;

fn main() {
    let num_children = 2;
    start_parent(num_children);
}

fn start_parent(num_children: u32) {
    println!("Parent: I am the parent process");
    println!("Parent PID is {}", getpid());
    for i in 0..num_children {
        run_child();
    }
}

fn run_child() {
    let child_pid = match fork() {
        Ok(ForkResult::Parent {child, ..}) => {
            println!("Main({}) forked child ({})", getpid(), child);
            child
        },
        Ok(ForkResult::Child) => {
            println!("Child({}) PPID is ({})", getpid(), getppid());
            exit(0);
        },
        Err(_) => panic!("fork failed"),
    };

    match waitpid(child_pid, None) {
        Ok(status) => println!(""),
        Err(_) => println!("waitpid() failed"),
    }
}

Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.20s
Running `target/debug/parallel`
Parent: I am the parent process
Parent PID is 579363
Main(579363) forked child (579390)
Child(579390) PPID is (579363)

Main(579363) forked child (579391)
Child(579391) PPID is (579363)

おおおおおおおおお、中々素晴らしい！