(5) Multithreaded server using Threadpool

README.md

@@ -36,4 +36,15 @@ Pada tahap ini, kita menambahkan rute `/sleep` yang akan melakukan simulasi pros
 Dari hasil percobaan membuka dua *browser window*, ketika kita mengakses `/sleep` di window pertama dan segera mengakses `/` di window kedua, window kedua tertahan (*loading* terus) dan baru memuat halaman setelah window pertama selesai.
 
 **Kenapa ini terjadi?**
-Ini terjadi karena web server yang kita bangun saat ini bersifat **Single-Threaded**. Artinya, server hanya memproses satu *request* pada satu waktu secara berurutan. Ketika rute `/sleep` diakses, satu-satunya *thread* yang dimiliki program akan delay selama 10 detik. Selama masa itu, *thread* tersebut *blocked* dan tidak bisa menerima atau memproses koneksi baru yang masuk (seperti *request* ke `/` dari window kedua). Window kedua harus menunggu dalam antrean sistem operasi (TCP queue) sampai *thread* utama selesai memproses rute `/sleep`. Ini menunjukkan bahwa server *single-thread* sangat tidak optimal dan tidak tangguh untuk menangani banyak koneksi (karena *bottleneck*).
\ No newline at end of file
+Ini terjadi karena web server yang kita bangun saat ini bersifat **Single-Threaded**. Artinya, server hanya memproses satu *request* pada satu waktu secara berurutan. Ketika rute `/sleep` diakses, satu-satunya *thread* yang dimiliki program akan delay selama 10 detik. Selama masa itu, *thread* tersebut *blocked* dan tidak bisa menerima atau memproses koneksi baru yang masuk (seperti *request* ke `/` dari window kedua). Window kedua harus menunggu dalam antrean sistem operasi (TCP queue) sampai *thread* utama selesai memproses rute `/sleep`. Ini menunjukkan bahwa server *single-thread* sangat tidak optimal dan tidak tangguh untuk menangani banyak koneksi (karena *bottleneck*).
+
+# Commit 5 Reflection notes
+
+Pada tahap ini, kita menyelesaikan masalah *single-thread bottleneck* dengan mengimplementasikan **ThreadPool**.
+
+**Bagaimana ThreadPool Bekerja:**
+1. Saat program berjalan, `ThreadPool::new(4)` akan langsung membuat 4 *thread* pekerja (`Worker`) yang berjalan di latar belakang.
+2. Kita menggunakan struktur komunikasi **Channel** (`mpsc::channel`) di Rust. `ThreadPool` bertindak sebagai pengirim pesan (*Sender*), dan setiap `Worker` berbagi akses sebagai penerima pesan (*Receiver*) melalui `Arc<Mutex<Receiver>>` agar aman dari *race condition* antar *thread*.
+3. Setiap kali ada koneksi masuk dari browser, fungsi `main` akan memanggil `pool.execute(...)` dan mengirimkan tugas (closure/pekerjaan yang harus dieksekusi) ke dalam *channel*.
+4. *Worker* yang sedang menganggur akan menerima tugas tersebut dari *channel* dan langsung mengeksekusinya. 
+5. Dengan arsitektur ini, jika ada pengguna yang mengakses rute lambat seperti `/sleep`, hanya satu *Worker* yang akan tertidur. Tiga *Worker* lainnya tetap *standby* dan bisa langsung melayani koneksi baru yang masuk secara bersamaan (karena itu tes membuka 2 window secara bersamaan sekarang berhasil tanpa harus antre).
\ No newline at end of file

src/lib.rs

+use std::{ sync::{ mpsc, Arc, Mutex }, thread };
+
+pub struct ThreadPool {
+    workers: Vec<Worker>,
+    sender: mpsc::Sender<Job>,
+}
+
+type Job = Box<dyn FnOnce() + Send + 'static>;
+
+impl ThreadPool {
+    /// Create a new ThreadPool.
+    ///
+    /// The size is the number of threads in the pool.
+    ///
+    /// # Panics
+    ///
+    /// The `new` function will panic if the size is zero.
+    pub fn new(size: usize) -> ThreadPool {
+        assert!(size > 0);
+
+        let (sender, receiver) = mpsc::channel();
+        let receiver = Arc::new(Mutex::new(receiver));
+
+        let mut workers = Vec::with_capacity(size);
+
+        for id in 0..size {
+            workers.push(Worker::new(id, Arc::clone(&receiver)));
+        }
+
+        ThreadPool { workers, sender }
+    }
+
+    pub fn execute<F>(&self, f: F) where F: FnOnce() + Send + 'static {
+        let job = Box::new(f);
+        self.sender.send(job).unwrap();
+    }
+}
+
+struct Worker {
+    id: usize,
+    thread: thread::JoinHandle<()>,
+}
+
+impl Worker {
+    fn new(id: usize, receiver: Arc<Mutex<mpsc::Receiver<Job>>>) -> Worker {
+        let thread = thread::spawn(move || {
+            loop {
+                let job = receiver.lock().unwrap().recv().unwrap();
+                println!("Worker {id} got a job; executing.");
+                job();
+            }
+        });
+
+        Worker { id, thread }
+    }
+}

src/main.rs

+use hello::ThreadPool; // ambil ThreadPool dari src/lib.rs (template code orang btw)
 use std::{
     fs,
     io::{ prelude::*, BufReader },
@@ -8,11 +9,15 @@ use std::{
 
 fn main() {
     let listener = TcpListener::bind("127.0.0.1:7878").unwrap();
+    let pool = ThreadPool::new(4); // set 4 thread
 
     for stream in listener.incoming() {
         let stream = stream.unwrap();
 
-        handle_connection(stream);
+        // oper task ke ThreadPool
+        pool.execute(|| {
+            handle_connection(stream);
+        });
     }
 }