Category: Rust

TcpListener
https://docs.rs/tokio/latest/tokio/net/struct.TcpListener.html

TcpListnerは元々Asyncにラップされているから、シングルスレッドでは使えない。

use tokio::{
    io::{AsyncReadExt, AsyncWriteExt},
    net::TcpListener,
};

#[tokio::main]
async fn main() -> std::io::Result<()> {
    let addr = "0.0.0.0:8080";
    let listener = TcpListener::bind(addr).await?;

    loop {
        match listener.accept().await {
            Ok((mut socket, _)) => {
                let mut buf = Vec::with_capacity(4096);
                socket.read_buf(&mut buf).await?;

                let msg = String::from_utf8(buf).expect("failed to convert str");
                println!("{msg}");

                socket.write(msg.as_bytes()).await?;
            }
            Err(err) => {
                println!("{err:?}");
            }   
        };
    }
}

Compiling parallel v0.1.0 (/home/vagrant/dev/rust/parallel)
Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.74s
Running `target/debug/parallel`
GET / HTTP/1.1
Host: 192.168.33.10:8080
User-Agent: curl/7.81.0
Accept: */*

ncコマンドで接続できる
$ nc 192.168.33.10 8080
hello
hello

スレッド自体はProducerとConsumerで同時に走ってるんだけど、なんか全然違うなぁ…

pub static SIZE: u32 = 5;
static slot: Mutex<Vec<i32>> = Mutex::new(Vec::new());

fn work() {
    for i in 0..10 {
        let producer_handle = thread::spawn(move || {
            if slot.lock().unwrap().len() < SIZE.try_into().unwrap() {
                slot.lock().unwrap().push(1);
                println!("Producer {} work: {:?}", i, slot);
                thread::sleep(Duration::from_millis(500));
            } else {
                for j in 0..5 {
                    if slot.lock().unwrap()[j] == 0 {
                        slot.lock().unwrap()[j] = 1;
                        println!("Producer {} work: {:?}", i, slot);
                        thread::sleep(Duration::from_millis(500));
                    } 
                }
            }
            
        });

        let consumer_handle = thread::spawn(move || {
            if slot.lock().unwrap().len() > 0 {
                if slot.lock().unwrap()[0] == 1 {
                    slot.lock().unwrap()[0] = 0;
                    thread::sleep(Duration::from_millis(700));
                    println!("Consumer {} work: {:?}", i, slot);
                } else if slot.lock().unwrap()[0] == 1 {
                    slot.lock().unwrap()[1] = 0;
                    thread::sleep(Duration::from_millis(700));
                    println!("Consumer {} work: {:?}", i, slot);
                }
            }
        });
        
        producer_handle.join().unwrap();
        consumer_handle.join().unwrap();
    }
}

fn main() {
    work();
    println!("{:?}", slot);
}

Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.31s
Running `target/debug/parallel`
Producer 0 work: Mutex { data: [1], poisoned: false, .. }
Producer 1 work: Mutex { data: [0, 1], poisoned: false, .. }
Consumer 1 work: Mutex { data: [0, 1], poisoned: false, .. }
Producer 2 work: Mutex { data: [0, 1, 1], poisoned: false, .. }
Producer 3 work: Mutex { data: [0, 1, 1, 1], poisoned: false, .. }
Producer 4 work: Mutex { data: [0, 1, 1, 1, 1], poisoned: false, .. }
Producer 5 work: Mutex { data: [0, 1, 1, 1, 1], poisoned: false, .. }
Consumer 5 work: Mutex { data: [0, 1, 1, 1, 1], poisoned: false, .. }
Producer 6 work: Mutex { data: [1, 1, 1, 1, 1], poisoned: false, .. }
Producer 7 work: Mutex { data: [1, 1, 1, 1, 1], poisoned: false, .. }
Consumer 7 work: Mutex { data: [1, 1, 1, 1, 1], poisoned: false, .. }
Producer 8 work: Mutex { data: [1, 1, 1, 1, 1], poisoned: false, .. }
Consumer 8 work: Mutex { data: [1, 1, 1, 1, 1], poisoned: false, .. }
Producer 9 work: Mutex { data: [1, 1, 1, 1, 1], poisoned: false, .. }
Consumer 9 work: Mutex { data: [1, 1, 1, 1, 1], poisoned: false, .. }

シングルスレッド
mutexのデータにアクセスするには、lockメソッドを使用してロックを獲得する

use std::sync::Mutex;

fn main() {
    let m = Mutex::new(5);

    {
        let mut num = m.lock().unwrap();
        *num = 6;
    }

    println!("m= {:?}", m);
}

スレッド間でsemaforeを利用しようとするとエラーになる。

use std::sync::{Mutex, Arc};
use std::thread;
use rand::Rng;
use std_semaphore::Semaphore;
use std::time::Duration;

static TOTAL_SPOTS: u32 = 3;

fn main() {
    let parked_cars = Arc::new(Mutex::new(0));
    let mut handles = vec![];

    let sem = Semaphore::new(TOTAL_SPOTS.try_into().unwrap());

    for _ in 0..10 {

        let counter = Arc::clone(&parked_cars);

        let handle = thread::spawn(move || {

            let mut rng = rand::thread_rng(); 
            // enter
            sem.acquire();
            let mut num = counter.lock().unwrap();
            *num += 1;
            thread::sleep(Duration::from_millis(rng.gen_range(1..10)));

            // exit
            sem.release();
        });
        handles.push(handle);
    }

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

    println!("Result: {}", *parked_cars.lock().unwrap());
}

15 | for _ in 0..10 {
| ————– inside of this loop
…
19 | let handle = thread::spawn(move || {
| ^^^^^^^ value moved into closure here, in previous iteration of loop
…
23 | sem.acquire();
| — use occurs due to use in closure

中々奥が深い..

use std::fs;
use std::path::Path;
use std::fs::File;
use std::io::{self, BufRead, BufReader};

fn search_file(filename: String, search_string: String) -> Result<bool, Box<dyn std::error::Error>> {
    for line in BufReader::new(File::open(filename.clone())?).lines() {
        if line.unwrap().contains(&search_string) {
            return Ok(true)
        }
    }
    Ok(false)
}

fn get_files(dirname: &str) -> io::Result<Vec<String>> {
    let mut entries: Vec<String> = Vec::new();
    let dir = Path::new(dirname);
    if dir.is_dir(){
        for entry in fs::read_dir(dirname)? {
            let e = entry?;
            let p = e.path().file_name().unwrap().to_string_lossy().into_owned();
            entries.push(p);
        }
    }
    Ok(entries)
}

fn search_files_sequentially(file_locations: String, search_string: String) {
    let entries: Vec<String> = get_files(&file_locations).unwrap();
    for entry in entries {
        let pass = format!("{}{}", file_locations, entry);
        if search_file(pass, search_string.clone()).unwrap() {
            println!("Found word in file: {}", entry);
        }
    }
}

fn main(){    
    search_files_sequentially("./src/".to_string(), "password".to_string());
}

Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.22s
Running `target/debug/parallel queue`
Found word in file: password_cracking.rs
Found word in file: main.rs
Found word in file: password_cracking_parallel.rs