【Rust】特定のディレクトリから検索文字列を探す

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

【Rust】ディレクトリのパス一覧をStringでVectorに入れる

時間かかったー

fn main() -> Result<(), Box<dyn std::error::Error>>{    

    let mut entries: Vec<String> = Vec::new();
    let dir = Path::new("./src");
    if dir.is_dir(){
        for entry in fs::read_dir(dir)? {
            let e = entry?;
            let p = e.path().file_name().unwrap().to_string_lossy().into_owned();
            entries.push(p);
        }
    }
    println!("{:?}", entries);
    Ok(())
}

Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.31s
Running `target/debug/parallel queue`
[“pipeline.rs”, “unix_domain_socket.rs”, “password_cracking.rs”, “shared_ipc.rs”, “main.rs”, “child_processes.rs”, “multithreading.rs”, “queue.rs”, “thread_pool.rs”, “password_cracking_parallel.rs”, “unixstream.rs”, “pipe.rs”]

【Rust】コマンドライン引数を取得する

use std::env;

fn main() {    
    let args: Vec<String> = env::args().collect();
    println!("{:?}", args[1]);
}

$ cargo run queue
Compiling parallel v0.1.0 (/home/vagrant/dev/rust/parallel)
Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.15s
Running `target/debug/parallel queue`
“queue”

【Python】ファイルから文字列を検索

短いコードなんだけと、割と高度なことをやっている…

import os
import time
import glob
import typing as T

def search_file(file_location: str, search_string: str) -> bool:
    with open(file_location, "r", encoding="utf8") as file:
        return search_string in file.read()
    
def search_files_sequentially(file_locations: T.List[str], search_string: str) -> None:
    for file_name in file_locations:
        result = search_file(file_name, search_string)
        if result:
            print(f"Found word in file: `{file_name}`")

if __name__ == "__main__":
    file_locations = list(
        glob.glob(f"{os.path.abspath(os.getcwd())}/*.py"))
    search_string = input("what word are you trying to find?: ")

    start_time = time.perf_counter()
    search_files_sequentially(file_locations, search_string)
    process_time = time.perf_counter() - start_time
    print(f"PROCESS TIME: {process_time}")

$ python3 find_files_sequential.py
what word are you trying to find?: queue
Found word in file: `/home/vagrant/dev/rust/parallel/python/thread_pool.py`
Found word in file: `/home/vagrant/dev/rust/parallel/python/pipeline.py`
Found word in file: `/home/vagrant/dev/rust/parallel/python/message_queue.py`
PROCESS TIME: 0.004676494048908353

OK, これをRustで書きたい…

【Rust】複数スレッドを同時にループ処理で走らせる【並列処理】

見た目はかなり悪いが、、、thread::spawnで同時にスレッドを走らせることができる。
各スレッドでは全てループ処理で待ち受けておいて、目的を達成したらbreakする。

static Washer_queue: Mutex<VecDeque<u32>> = Mutex::new(VecDeque::new());
static Dryer_queue: Mutex<VecDeque<u32>> = Mutex::new(VecDeque::new());
static Folder_queue: Mutex<VecDeque<u32>> = Mutex::new(VecDeque::new());
static Done_queue: Mutex<VecDeque<u32>> = Mutex::new(VecDeque::new());

fn assemble_laundry(n: u32) {
    for i in 1..(n+1) {
        Washer_queue.lock().unwrap().push_back(i);
    }
}

fn washer() {
    let w = Washer_queue.lock().unwrap().pop_front();
    if w != None {
        println!("washing {:?}...", w.unwrap());
        thread::sleep(Duration::from_millis(300));
        Dryer_queue.lock().unwrap().push_back(w.unwrap());
    }
}

fn dryer() {
    let d = Dryer_queue.lock().unwrap().pop_front();
    if d != None {
        println!("Drying {:?}...", d.unwrap());
        thread::sleep(Duration::from_millis(200));
        Folder_queue.lock().unwrap().push_back(d.unwrap());
    }
}

fn folder() {
    let f = Folder_queue.lock().unwrap().pop_front();
    if f != None {
        println!("Folding {:?}...", f.unwrap());
        thread::sleep(Duration::from_millis(100));
        Done_queue.lock().unwrap().push_back(f.unwrap());
    }
}

fn main() {    
    assemble_laundry(4);
    println!("{:?}", Washer_queue);
    let wash_handle = thread::spawn(|| {
        loop {
            if Washer_queue.lock().unwrap().len() == 0 {
                break;
            }
            washer();
        }
    });
    let dry_handle = thread::spawn(|| {
        loop {
            if Done_queue.lock().unwrap().len() == 4 {
                break;
            }
            dryer();
        }
    });
    let fold_handle = thread::spawn(|| {
        loop {
            if Done_queue.lock().unwrap().len() == 4{
                break;
            }
            folder();
        }
    });
    wash_handle.join().unwrap();
    dry_handle.join().unwrap();
    fold_handle.join().unwrap();
    println!("Washer {:?}", Washer_queue);
    println!("Dryer {:?}", Dryer_queue);
    println!("Folder {:?}", Folder_queue);
    println!("All work finished");
}

Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.20s
Running `target/debug/parallel`
Mutex { data: [1, 2, 3, 4], poisoned: false, .. }
washing 1…
washing 2…
Drying 1…
Folding 1…
washing 3…
Drying 2…
Folding 2…
washing 4…
Drying 3…
Folding 3…
Drying 4…
Folding 4…
Mutex { data: [], poisoned: false, .. }
Mutex { data: [], poisoned: false, .. }
Mutex { data: [], poisoned: false, .. }
All work finished

もうちょっとうまい書き方をしたいが、やりたいこと自体はできている。。

【Rust】rustでqueueのpush, popをやりたい

VecDequeを使うと、vectorのpush, popが抽象化されている。

use std::collections::VecDeque;

fn main() {

    let mut washload: VecDeque<u32> = VecDeque::new();
    washload.push_back(1);
    washload.push_back(2);
    washload.push_back(3);
    println!("{:?}", washload);

    let n = washload.pop_front();
    println!("{:?}", n.unwrap());
}

Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.17s
Running `target/debug/parallel`
[1, 2, 3]
1

なるほど、確かに使いやすい。

【並列処理】各スレッドでQueueを使ったパイプライン処理

whileのループでqueueから待ち受けるという処理を3つのスレッドで同時に実行している。うむ、中々複雑になってきた。

import time
from queue import Queue
from threading import Thread

Washload = str

class Washer(Thread):
    def __init__(self, in_queue: Queue[Washload],
        out_queue: Queue[Washload]):
        super().__init__()
        self.in_queue = in_queue
        self.out_queue = out_queue

    def run(self) -> None:
        while True:
            washload = self.in_queue.get()
            print(f"Washer: washing{washload}...")
            time.sleep(4)
            self.out_queue.put(f'{washload}')
            self.in_queue.task_done()

class Dryer(Thread):
    def __init__(self, in_queue: Queue[Washload],
        out_queue: Queue[Washload]):
        super().__init__()
        self.in_queue = in_queue
        self.out_queue = out_queue

    def run(self) -> None:
        while True:
            washload = self.in_queue.get()
            print(f"Dryer: dying {washload} ...")
            time.sleep(2)
            self.out_queue.put(f'{washload}')
            self.in_queue.task_done()

class Folder(Thread):
    def __init__(self, in_queue: Queue[Washload]):
        super().__init__()
        self.in_queue = in_queue

    def run(self) -> None:
        while True:
            washload = self.in_queue.get()
            print(f"Folder: folding {washload}...")
            time.sleep(1)
            print(f"Folder: {washload} done!")
            self.in_queue.task_done()

class Pipeline:
    def assemble_laundry_for_washing(self) -> Queue[Washload]:
        washload_count = 4
        washloads_in: Queue[Washload] = Queue(washload_count)
        for washload_num in range(washload_count):
            washloads_in.put(f'Washload #{washload_num}')
        return washloads_in

    def run_concurrently(self) -> None:
        to_be_washed = self.assemble_laundry_for_washing()
        to_be_dried: Queue[Washload] = Queue()
        to_be_folded: Queue[Washload] = Queue()

        Washer(to_be_washed, to_be_dried).start()
        Dryer(to_be_dried, to_be_folded).start()
        Folder(to_be_folded).start()

        to_be_washed.join()
        to_be_dried.join()
        to_be_folded.join()

        print("All done!")

if __name__ == "__main__":
    pipeline = Pipeline()
    pipeline.run_concurrently()

$ python3 pipeline.py
Washer: washingWashload #0…
Washer: washingWashload #1…
Dryer: dying Washload #0 …
Folder: folding Washload #0…
Folder: Washload #0 done!
Washer: washingWashload #2…
Dryer: dying Washload #1 …
Folder: folding Washload #1…
Folder: Washload #1 done!
Washer: washingWashload #3…
Dryer: dying Washload #2 …
Folder: folding Washload #2…
Folder: Washload #2 done!
Dryer: dying Washload #3 …
Folder: folding Washload #3…
Folder: Washload #3 done!
All done!

【並列処理】プリエンプティブマルチタスク

タイムシェアリング方式により、1つのコアでもスレッドが同時に実行されているような印象を与える。
現在のOSは殆どがプリエンプティブマルチタスクの機能がある。

import typing as T
from threading import Thread, Event, Timer, Event

from pacman import get_user_input, compute_game_world, render_next_screen

processor_free = Event()
processor_free_set()
TIME_SLICE = 0.5

class Task(Thread):
    def __init__(self, func: T.Callable[..., None]):
        super().__init__()
        self.func = func

    def run(self) -> None:
        while True:
            processor_free.wait()
            processor_free.clear()
            self.func()

class InterruptService(Timer):
    def __init__(self):
        super().__init__(TIME_SLICE, lambda: None)

    def run(self):
        while not self.finished.wait(self.interval):
            print("Tick!")
            processor_free.set()

def arcade_machine() -> None:
    get_user_input_task = Task(get_user_input)
    compute_game_world_task = Task(compute_game_world)
    render_next_screen_task = Task(render_next_screen)

    InterruptService().start()
    get_user_input_task.start()
    compute_game_world_task.start()
    render_next_screen_task.start()

if __name__ == "__main__":
    arcade_machine()

【並列処理】マルチタスクを学ぶ

アプリケーションの処理は主にCPUバウンド(CPUの計算処理)とI/Oバウンド(ディスクからの読み込み、入出力取得等)の2種類に分類される。
CPUバウンドの負荷は並列化によりパフォーマンスが改善される可能性がある。

スレッドごとに別々のタスクをループで実行

import typing as T
from threading import Thread, Event

from pacman import get_user_input, compute_game_world, render_next_screen

processor_free = Event()
#processor_free_set()

class Task(Thread):
    def __init__(self, func: T.Callable[..., None]):
        super().__init__()
        self.func = func

    def run(self) -> None:
        while True:
            processor_free.wait()
            processor_free.clear()
            self.func()

def arcade_machine() -> None:
    get_user_input_task = Task(get_user_input)
    compute_game_world_task = Task(compute_game_world)
    render_next_screen_task = Task(render_next_screen)

    get_user_input_task.start()
    compute_game_world_task.start()
    render_next_screen_task.start()

if __name__ == "__main__":
    arcade_machine()

ソースコードで見ると、構造がわかりやすい。

【Rust】rustでthread Poolによるパスワードクラッキング【並列処理】

use sha2::{Digest, Sha256};
use std::time;
use std::collections::HashMap;
use threadpool::ThreadPool;

fn main() {

    let hash = "2e9352c704043c75fa1c2a424fce7bef0569ec08af453e841101596d911d26e3".to_string();
    let length = 4;
    crack_password_parallel(hash, length);
}

fn crack_password_parallel(crypto_hash: String, length: u32) {
    let num_cores = num_cpus::get() as u32;
    let chunks = get_chunks(num_cores, length);
    let pool = ThreadPool::new(num_cores as usize);
    println!("{:?}", chunks);
    
    for (chunk_start, chunk_end) in chunks {
        let hash = crypto_hash.clone();
        pool.execute(move|| {
            println!("{}:{}", chunk_start, chunk_end);
            let combinations = get_chunk_combinations(length, chunk_start, chunk_end);
            for combination in combinations {
                if check_password(&hash, combination.clone()) {
                    println!("PASSWORD CRACKED:{}", combination);
                    break;
                }
            }
        });
    }
    pool.join();

}

fn get_chunk_combinations(length: u32, min_number: u32, max_number: u32) -> Vec<String> {
    let mut combinations: Vec<String> = Vec::new();
    for i in min_number..max_number {
        let str_num: String = i.to_string();
        let zeros: String = "0".repeat((length - str_num.chars().count() as u32).try_into().unwrap());
        combinations.push(format!("{}{}", zeros, str_num));
    }
    return combinations;
}

fn get_chunks(num_ranges: u32, length: u32) -> HashMap<u32, u32>{
    let max_number = 10_i32.pow(length) as u32;

    let mut chunk_starts = Vec::new();
    for i in 0..num_ranges {
        chunk_starts.push(max_number / num_ranges * i )
    }

    let mut chunk_ends = Vec::new();
    for i in &chunk_starts[1..] {
        chunk_ends.push(i - 1);
    }
    chunk_ends.push(max_number);

    let mut chunks:HashMap<u32, u32> = HashMap::new();
    for i in 0..chunk_starts.len() {
        chunks.insert(chunk_starts[i], chunk_ends[i]);
    }
    return chunks
}

fn get_combinations(length: u32) -> Vec<String> {
    let mut combinations: Vec<String> = Vec::new();
    let min_number = 0;
    let max_number = 10_i32.pow(length);

    for i in min_number..max_number {
        let str_num: String = i.to_string();
        let zeros: String = "0".repeat((length - str_num.chars().count() as u32).try_into().unwrap());
        combinations.push(format!("{}{}", zeros, str_num));
    }
    return combinations;
}

fn get_crypto_hash(password: String) -> String {
    let sha = Sha256::digest(password);
    hex::encode(sha).to_string()
}

fn check_password(expected_crypto_hash: &String, possible_password: String) -> bool {
    let actual_crypto_hash = get_crypto_hash(possible_password);
    return *expected_crypto_hash == actual_crypto_hash
}

fn crack_password(crypto_hash: String, length: u32) {
    println!("Processing number combinations sequentially");
    let start_time = time::Instant::now();
    let combinations: Vec<String> = get_combinations(length);
    for combination in combinations {
        if check_password(&crypto_hash.clone(), combination.clone()) {
            println!("PASSWORD CRACKED:{}", combination);
            break;
        }
    }
    println!("PROCESS TIME: {:?}", start_time.elapsed());
}

Finished `dev` profile [unoptimized + debuginfo] target(s) in 0.32s
Running `target/debug/parallel`
{5000: 10000, 0: 4999}
5000:10000
0:4999
PASSWORD CRACKED:5231

これは中々素晴らしいね^^