Project — Concurrent Batch Processing

နားလည်ထားရမယ့် အချက်

Lesson 21 ရဲ့ CLI content analyzer ကို file ရာနှင့်ချီအထိ scale ချင်ရင် sequential processing (file တစ်ခုပြီးမှ တစ်ခု) ဟာ multi-core CPU ရဲ့ core တွေအားလုံးကို အသုံးမချနိုင်ဘဲ time linearly ကြီးလာမှာဖြစ်ပါတယ်—ဒီ project က Lesson 18 ရဲ့ `std::thread`/`mpsc` channel ကို production-shaped parallel processing pipeline တစ်ခုအဖြစ် direct apply လုပ်တာပါ။ Design decision အဓိကတစ်ခုက file list ကို thread count (ဥပမာ CPU core number, `std::thread::available_parallelism()` ကနေ query လုပ်နိုင်) အလိုက် chunk ခွဲပြီး thread တစ်ခုစီကို chunk တစ်ခုစီ ownership move ပေးလိုက်တာပါ—thread တစ်ခုစီက independent ဖြစ်ပြီး shared mutable state (locks) လိုအပ်မှု လုံးဝမရှိပါဘူး—Lesson 18 ရဲ့ "shared state အစား message passing" philosophy ကို scale ကြီးကြီးထိ extend လုပ်တာပါ။ Thread တစ်ခုစီရဲ့ partial result (file count, total word count) ကို channel ကနေတစ်ဆင့် main thread ဆီ ပို့ပြီး main thread ကသာ aggregate (sum) လုပ်ပါတယ်—ဒါက race condition ဖြစ်နိုင်ချေ ကင်းစင်စေတဲ့ design ပါ။ Sequential version နှင့် concurrent version ကို timing (`std::time::Instant`) ဖြင့် compare လုပ်တဲ့အခါ—CPU core အနည်းအများ, file size, thread spawn overhead (thread ရေအရမ်းများရင် overhead ကိုယ်တိုင်က benefit ကို ကျော်လွန်နိုင်) အလိုက် speedup ရလဒ် ကွာနိုင်ပါတယ်—"concurrency ဟာ always faster" ဆိုတဲ့ assumption ကို data ဖြင့် စစ်ဆေးရမယ့် practical lesson တစ်ခုပါ။

လက်တွေ့ scenario နဲ့ ချိတ်ကြည့်မယ်

Lesson 21 ရဲ့ file list ကို `chunks(file_count / num_threads)` ဖြင့် ခွဲပြီး chunk တစ်ခုစီကို `thread::spawn(move || { ... })` ဆီ ownership move ပေးမယ်—thread တစ်ခုစီက chunk ထဲက file များကို sequential read+count လုပ်ပြီး `(files_processed, total_words)` tuple result ကို `tx.send(...)` ဖြင့် main thread ဆီ ပို့မယ်။ Main thread ကတော့ `rx.iter().take(num_threads)` ဖြင့် result အားလုံးကို collect ပြီး `.fold((0, 0), |acc, r| (acc.0 + r.0, acc.1 + r.1))` ဖြင့် final total ကို aggregate မယ်—individual thread ကို join() ခေါ်ဖို့ handle vector ကိုလည်း သိမ်းထားရမယ်။ Sequential baseline version (thread မသုံးဘဲ loop ချည်း) နှင့် concurrent version ကို `Instant::now()`/`elapsed()` ဖြင့် timing နှိုင်းယှဉ်ပြီး file count/thread count အမျိုးမျိုးနှင့် benchmark run ကြည့်မယ်။

• File list ကို thread count အလိုက် chunk ခွဲ, ownership move—shared lock မလို
• Thread partial result ကို channel ဖြင့် main thread ဆီ ပို့, main thread ကသာ aggregate
• Concurrency ဟာ always faster မဟုတ်—thread overhead ကို timing data ဖြင့် စစ်ဆေး

အတူတူ စမ်းရေးကြည့်မယ်

use std::sync::mpsc;
use std::thread;
use std::time::Instant;

fn word_count_for(files: &[&str]) -> (usize, usize) {
    let total: usize = files.iter().map(|f| f.split_whitespace().count()).sum();
    (files.len(), total)
}

fn main() {
    let files = vec!["rust ownership basics", "borrow checker rules", "traits and generics", "async runtime tokio"];
    let (tx, rx) = mpsc::channel();
    let start = Instant::now();

    for chunk in files.chunks(2) {
        let chunk: Vec<&str> = chunk.to_vec();
        let tx = tx.clone();
        thread::spawn(move || {
            tx.send(word_count_for(&chunk)).unwrap();
        });
    }
    drop(tx);

    let (files_done, words_done) = rx.iter().fold((0, 0), |acc, r| (acc.0 + r.0, acc.1 + r.1));
    println!("{files_done} files, {words_done} words in {:?}", start.elapsed());
}

"4 files, 12 words in <duration>" ကို print ထုတ်နိုင်မည်—thread များစွာက result ကို channel ကနေတစ်ဆင့် main thread ဆီ ပေါင်းစည်းပေးခဲ့ကြောင်း သက်သေပြမည်။

The Rust Programming Language — Using Message Passing to Transfer Data — Rust