capstone — ยก kaen-kvstore ขึ้น Tokio: 1 task ต่อคอนเนกชัน พร้อม graceful shutdown

เจ็ดบทที่ผ่านมาเราแยกชิ้นส่วนของ async Rust ออกดูทีละชิ้น: Future ที่ขี้เกียจกับ executor 15 บรรทัดที่เขียนเอง (บท 1) · Tokio runtime ที่เป็น scheduler + I/O driver + timer (บท 2) · spawn กับ JoinHandle (บท 3) · async TCP กับ wire แบบนำหน้าด้วยความยาว (บท 4) · select! กับการยกเลิกด้วยการ drop (บท 5) · state ที่ share กับ channel (บท 6) · และกับดักคลาสสิกอย่าง function colouring กับการ block runtime (บท 7)

บทนี้คือบทที่ ประกอบมันกลับเข้าด้วยกันเป็น server ตัวเดียวที่รันจริง — kaen-kvstore ตัวเดิมจาก #22 แต่คราวนี้คอนเนกชันไม่ต้องยึด worker จาก ThreadPool ขนาดคงที่ 4 ตัวไว้ทั้งเส้นอีกแล้ว เป็น 1 task ต่อคอนเนกชัน และปิดงานด้วย graceful shutdown ที่ครบสามส่วนจริงๆ ไม่ใช่แค่ process::exit(0) แล้วภาวนา

📦 kaen-kvstore

บทนี้ ต่อยอด repo kaen-kvstore จาก #22 (code ตัวอย่างกำลังจัดทำ) เป็นบทปิด — เรา ไม่ได้เขียน store ใหม่ engine ชั้นล่าง (append-only log, hash index, write→fsync→ack, tombstone/compaction) ยังเป็นของ #22 ทุกบรรทัด สิ่งที่บทนี้เปลี่ยนคือ ชั้นรับคอนเนกชัน เท่านั้น: จาก ThreadPool ที่เขียนเองในบท 5 ของ #22 กลายเป็น tokio::spawn เพียงแต่ทุก call บนสายกลายเป็น .await ส่วนเรื่อง protocol ต้องพูดให้ตรง — เราเก็บ framing ของ #22 ไว้ทั้งดุ้น (u32 little-endian นำหน้าความยาว ตามด้วย payload ตรงกันทุก byte) แต่ทำ command layer ให้เป็น text เรียบง่าย เพื่อให้บทนี้โฟกัสที่ async ไม่ใช่ที่ opcode (รายละเอียดและราคาที่จ่ายอยู่ในหัวข้อถัดไป) code ทั้งก้อนในบทนี้ compile และรันได้จริงในแซนด์บ็อกซ์ ไม่มีอะไรเป็นไดอะแกรมอย่างเดียว

toolchain + feature scope ของบทนี้

ทุก snippet pin ที่ rustc 1.97.1 (8bab26f4f 2026-07-14) · edition = “2024” · tokio 1.53.1 และนี่คือ [dependencies] ทั้งหมดของ capstone:

[dependencies]
tokio = { version = "1.53.1", features = ["rt-multi-thread", "net", "io-util", "macros", "sync", "signal"] }

6 feature นี้คือรายการที่ ใช้จริง ทุกตัว: rt-multi-thread (scheduler หลาย thread ที่ #[tokio::main] เปล่าๆ ต้องการ) · net (TcpListener/TcpStream) · io-util (AsyncReadExt/AsyncWriteExt) · macros (#[tokio::main], #[tokio::test], select!) · sync (broadcast, mpsc, Semaphore) · signal (ctrl_c)

สองตัวแรกในรายการนั้นคือคู่ที่พลาดกันบ่อยที่สุด และมันพลาดกันคนละทาง: ถ้าลืม rt-multi-thread แล้วเขียน #[tokio::main] เปล่าๆ จะได้ error: The default runtime flavor is `multi_thread`, but the `rt-multi-thread` feature is disabled. เพราะ macros แจกแต่ตัวมาโคร ไม่ได้พา scheduler มาให้เลยสักตัว (flavor current_thread ต้องการแค่ "rt" ส่วน default ต้องการ "rt-multi-thread") — 🔁 บท 2 กับบท 5 เจอกับดักคู่นี้มาแล้วทั้งสองทาง

ยังคงห้าม features = ["full"] ด้วย scope นี้ cargo ตอบกลับมาว่า:

Locking 16 packages to latest Rust 1.97.1 compatible versions

ตัวเลข 16 คือจำนวน entry ใน Cargo.lock (ไม่นับ crate ของเราเอง) ซึ่งรวม wasi, windows-link, windows-sys ที่ lock ไว้เผื่อ platform อื่นและ ไม่ถูก compile บน Linux เลยสักตัว ถ้านับเฉพาะ crate ที่ compile จริง บน x86_64-unknown-linux-gnu ตัวเลขคือ 7 → 13 เทียบกับ 7 crate ของบท 1 ที่มีแค่ ["rt", "macros"] ส่วนที่งอกมาคือหกตัวพอดี: mio 1.2.2, socket2 0.6.5, libc 0.2.189, bytes 1.12.1, errno 0.3.14 และ signal-hook-registry 1.4.8 ซึ่งมาจาก net กับ signal โดยตรง

code ของ capstone ทั้งก้อน compile เขียวและผ่าน cargo clippy --target x86_64-unknown-linux-gnu -- -D warnings แบบ zero-warnings ยกเว้น snippet ที่ติดป้าย ❌ ซึ่งตั้งใจให้ compile ไม่ผ่าน — นั่นคือประเด็นของมัน ทุกบรรทัด output ในบทนี้คือข้อความจริงที่พิมพ์ออกมา

จาก worker-per-connection ไป task-per-connection

kaen-kvstore ของ #22 รับงานแบบนี้: listener.incoming() ได้ socket มาหนึ่งตัว โยนเข้า ThreadPool::new(4) ที่เราเขียนเอง แล้ว worker ตัวที่รับงานไปจะ หมุนวนอ่าน-ตอบจนคอนเนกชันปิด model นี้อ่านง่ายมาก และ #22 ก็เลือกมันอย่างจงใจแล้ว (บท 5 ของ #22 ปฏิเสธการ spawn thread ต่อหนึ่งคอนเนกชันดื้อๆ ไว้ตั้งแต่ต้น) ปัญหาจึงไม่ใช่ “thread เยอะเกิน” — ปัญหาคือ จำนวน worker เป็นเพดานของจำนวนคอนเนกชันพร้อมกัน คอนเนกชันที่ 5 ต้องรอจนกว่าจะมีคนวางสาย และตลอดเวลาที่ worker นั่ง block รอ byte ที่ยังมาไม่ถึง มันก็ยังกิน stack ก้อนนั้นเต็มๆ ทั้งที่ไม่ได้ทำงานอะไรเลย

ถ้าคุณเคยดูประวัติฝั่ง .NET จะรู้สึกคุ้นทันที — ยุค TcpListener.AcceptTcpClient() + new Thread(HandleClient) ก็ตันแบบเดียวกันเป๊ะ ทางออกของ .NET คือย้ายไปเป็น await AcceptTcpClientAsync() แล้วโยนงานเข้า Task ซึ่งลงจอดที่ ThreadPool ไม่กี่ thread ทางออกของ Rust คือสิ่งเดียวกัน แค่เรียกว่า tokio::spawn และ task ของ Tokio ไม่มี stack เป็นของตัวเอง — มันมีขนาดเท่ากับ state machine ที่ compiler สร้างจาก async fn ของคุณ ซึ่งเป็น struct ก้อนเดียวที่ scheduler ถือไว้

โครงของ accept loop จึงกลายเป็นแบบนี้ (นี่คือรูปที่จะเติมให้ครบตลอดทั้งบท):

loop {
    let (sock, _) = listener.accept().await?;
    let db = Arc::clone(&db);
    tokio::spawn(async move { handle_conn(sock, db).await; });
}

หนึ่งบรรทัดต่อคอนเนกชัน ไม่มี thread pool ที่ต้องเขียนเอง ไม่มี Worker ที่ต้อง join ตอนปิด — แต่ ความสะดวกนี้เองคือที่มาของปัญหาทั้งบท เพราะ tokio::spawn แล้วปล่อยทิ้ง แปลว่าเราไม่มีอะไรถือไว้เลยว่ามีกี่คอนเนกชันที่ยังทำงานอยู่ ตอน shutdown เราจึงไม่รู้ว่าจะรอใคร

framing เดิมของ #22 แต่ทุก call กลายเป็น await

framing ไม่เปลี่ยนแม้แต่ byte เดียว: u32 little-endian บอกความยาว ตามด้วย payload เท่านั้น ที่เปลี่ยนคือฝั่ง I/O — Read/Write ของ std กลายเป็น AsyncReadExt/AsyncWriteExt ของ Tokio และทุก method กลายเป็น .await และ2 function นี้ไม่ใช่ของใหม่ด้วยซ้ำ — มันคือ read_frame_or_eof/write_frame ที่ 🔁 บท 4 เขียนและ test ไว้แล้ว ยกมาทั้งดุ้นรวมทั้ง bound R: AsyncReadExt + Unpin:

use std::collections::HashMap;
use std::future::Future;
use std::sync::{Arc, Mutex};
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::net::{TcpListener, TcpStream};
use tokio::sync::{Semaphore, broadcast, mpsc};

type Db = Arc<Mutex<HashMap<String, Vec<u8>>>>;

const MAX_CONNS: usize = 64;

async fn read_frame_or_eof<R: AsyncReadExt + Unpin>(
    r: &mut R,
) -> std::io::Result<Option<Vec<u8>>> {
    let mut hdr = [0u8; 4];
    let mut filled = 0;
    while filled < 4 {
        match r.read(&mut hdr[filled..]).await? {
            0 if filled == 0 => return Ok(None),
            0 => {
                return Err(std::io::Error::new(
                    std::io::ErrorKind::UnexpectedEof,
                    "header truncated",
                ));
            }
            n => filled += n,
        }
    }
    let len = u32::from_le_bytes(hdr) as usize;
    let mut buf = vec![0u8; len];
    r.read_exact(&mut buf).await?;
    Ok(Some(buf))
}

async fn write_frame<W: AsyncWriteExt + Unpin>(w: &mut W, payload: &[u8]) -> std::io::Result<()> {
    w.write_u32_le(payload.len() as u32).await?;
    w.write_all(payload).await?;
    w.flush().await?;
    Ok(())
}

เหตุผลที่เราไม่ลัดด้วย read_u32_le() แล้ว map UnexpectedEof เป็น Ok(None) คือสิ่งที่บท 4 พิสูจน์ไปแล้ว: ทางลัดนั้น แยก “ปิดสายอย่างสุภาพ” กับ “หัว frame ขาดกลางคัน” ไม่ออก — peer ที่ตายหลังส่งหัว frame ไป 2 จาก 4 byte จะถูกรายงานว่าบอกลาเรียบร้อย loop while filled < 4 ที่อ่านหัว frame ด้วยมือจึงจำเป็น: Ok(None) เกิดได้ เฉพาะตอน filled == 0 เท่านั้น คือยังไม่ได้อ่านอะไรเลยสัก byte ส่วน byte ที่ 1–3 แล้วสายขาดคือ Err(UnexpectedEof, "header truncated") test ตัวที่สองท้ายบทตรึงสองเคสนี้ไว้ทั้งคู่ จำจุดนี้ไว้ เดี๋ยวมันจะกลับมาตอนพูดเรื่อง cancellation-safety

เทียบกับ C# ตรงๆ: loop อ่านหัว frame + read_exact คือ ReadAsync ที่วนจนครบ บวก BinaryPrimitives.ReadUInt32LittleEndian; write_all + flush คือ WriteAsync + FlushAsync สัญชาตญาณโอนมาได้ทั้งดุ้น

state ที่ share: `std::sync::Mutex` ไม่ใช่ tokio Mutex

นี่คือจุดที่คนอ่านคู่มือ Tokio ผ่านๆ แล้วเลือกผิดบ่อยที่สุด เห็นคำว่า async ก็คว้า tokio::sync::Mutex มาใช้ทันที ทั้งที่คู่มือของ Tokio เองเขียนไว้ตรงข้าม: std mutex ใช้จาก async ได้และมักจะเป็นตัวเลือกที่ถูกต้อง ตราบใดที่ contention ต่ำและ ไม่ถือ guard ข้าม .await

KV map ของเราเข้าเงื่อนไขนั้นเต็มๆ — insert กับ get บน HashMap ในหน่วยความจำใช้เวลาระดับไมโครวินาที ไม่มี I/O อยู่ระหว่างกลาง การไปใช้ async mutex ที่ต้องผ่าน scheduler จะช้ากว่าโดยเปล่าประโยชน์

กุญแจของ discipline นี้คือ เอา lock ไปไว้ใน function non-async แล้วให้ guard ตายไปพร้อมกับ frame ของ function นั้น ไม่มีทางที่มันจะรอดออกมาเจอ .await ได้เลย:

fn apply(db: &Db, req: &[u8]) -> Vec<u8> {
    let text = match std::str::from_utf8(req) {
        Ok(t) => t,
        Err(_) => return b"ERR bad-utf8".to_vec(),
    };
    let mut parts = text.splitn(3, ' ');
    let mut map = db.lock().unwrap();
    match (parts.next(), parts.next(), parts.next()) {
        (Some("SET"), Some(k), Some(v)) => {
            map.insert(k.to_string(), v.as_bytes().to_vec());
            b"OK".to_vec()
        }
        (Some("GET"), Some(k), None) => match map.get(k) {
            Some(v) => v.clone(),
            None => b"NOTFOUND".to_vec(),
        },
        _ => b"ERR bad-request".to_vec(),
    }
} // guard drop ที่นี่ — ก่อน caller จะ await อะไรได้

สังเกตว่า apply คืน Vec<u8> ที่ คัดลอกค่าออกมาแล้ว ไม่ได้คืน reference เข้าไปใน map นั่นคือราคาที่จ่ายเพื่อให้ guard ตายทันเวลา — และเป็นราคาที่ถูกมากเมื่อเทียบกับสิ่งที่จะเกิดถ้าเราไม่จ่าย

❌ version ดิบ: ถือ guard ข้าม `.await`

ลองเขียนแบบที่ อ่านแล้วดูสมเหตุสมผล — lock ไว้ในตัว handler เลย จะได้ไม่ต้อง clone ค่าออกมา file เต็มๆ ของ ❌ version ดิบ เป็นแบบนี้:

use std::collections::HashMap;
use std::sync::{Arc, Mutex};

type Db = Arc<Mutex<HashMap<String, Vec<u8>>>>;

async fn flush_to_disk() {}

async fn handle_conn(db: Db) {
    let mut map = db.lock().unwrap();
    map.insert("k".into(), b"v".to_vec());
    flush_to_disk().await;
    println!("{}", map.len());
}

#[tokio::main]
async fn main() {
    let db: Db = Arc::new(Mutex::new(HashMap::new()));
    tokio::spawn(handle_conn(db));
}

cargo build --target x86_64-unknown-linux-gnu ตอบกลับมาแบบนี้:

error: future cannot be sent between threads safely
   --> src/main.rs:18:18
    |
 18 |     tokio::spawn(handle_conn(db));
    |                  ^^^^^^^^^^^^^^^ future returned by `handle_conn` is not `Send`
    |
    = help: within `impl Future<Output = ()>`, the trait `Send` is not implemented for `std::sync::MutexGuard<'_, HashMap<String, Vec<u8>>>`
note: future is not `Send` as this value is used across an await
   --> src/main.rs:11:21
    |
  9 |     let mut map = db.lock().unwrap();
    |         ------- has type `std::sync::MutexGuard<'_, HashMap<String, Vec<u8>>>` which is not `Send`
 10 |     map.insert("k".into(), b"v".to_vec());
 11 |     flush_to_disk().await;
    |                     ^^^^^ await occurs here, with `mut map` maybe used later
note: required by a bound in `tokio::spawn`
   --> /home/nook/.cargo/registry/src/index.crates.io-1949cf8c6b5b557f/tokio-1.53.1/src/task/spawn.rs:176:21
    |
174 |     pub fn spawn<F>(future: F) -> JoinHandle<F::Output>
    |            ----- required by a bound in this function
175 |     where
176 |         F: Future + Send + 'static,
    |                     ^^^^ required by this bound in `spawn`

อ่านลูกโซ่ของเหตุผลจากล่างขึ้นบน: tokio::spawn ต้องการ F: Future + Send + 'static เพราะ scheduler แบบหลาย thread อาจย้าย task ไปทำต่อบน thread อื่นกลางคัน · future ตัวนี้ไม่ Send เพราะมีค่าที่ ยังมีชีวิตข้าม .await · ค่านั้นคือ std::sync::MutexGuard ซึ่งจงใจไม่ implement Send มาตั้งแต่ต้น

นี่คือ บรรทัดเดียวกันกับ CS1996 ของ C# ที่ห้าม await ใน block lock — แต่ Rust ไปไกลกว่าตรงที่ไม่ต้องมีกฎเฉพาะกิจ มันตกมาจากระบบ type เอง MutexGuard ไม่ Send, future ที่อุ้มมันไว้ข้าม .await จึงไม่ Send, และ spawn ก็ปฏิเสธ

`drop(guard)` ไม่พอ — ต้องปิดด้วย scope

ปฏิกิริยาแรกของทุกคนคือใส่ drop(map); คั่นไว้ก่อน .await ลองดูจริงๆ file เดิมทุกอย่าง แค่แทรก drop(map); เป็นบรรทัดที่ 11 (และตัด println! ที่ใช้ map ทิ้ง):

async fn handle_conn(db: Db) {
    let mut map = db.lock().unwrap();
    map.insert("k".into(), b"v".to_vec());
    drop(map);
    flush_to_disk().await;
}

ยัง ไม่ผ่าน (ตัดเหลือบรรทัดที่มีน้ำหนัก):

error: future cannot be sent between threads safely
   --> src/main.rs:18:18
    |
 18 |     tokio::spawn(handle_conn(db));
    |                  ^^^^^^^^^^^^^^^ future returned by `handle_conn` is not `Send`
    |
    = help: within `impl Future<Output = ()>`, the trait `Send` is not implemented for `std::sync::MutexGuard<'_, HashMap<String, Vec<u8>>>`
note: future is not `Send` as this value is used across an await
   --> src/main.rs:12:21
    |
  9 |     let mut map = db.lock().unwrap();
    |         ------- has type `std::sync::MutexGuard<'_, HashMap<String, Vec<u8>>>` which is not `Send`
...
 12 |     flush_to_disk().await;
    |                     ^^^^^ await occurs here, with `mut map` maybe used later

เหตุผลอยู่ในวลี maybe used later — การวิเคราะห์ที่ตัดสินว่าอะไรต้องอยู่ในตัว state machine ทำงานที่ระดับ ขอบเขตของตัวแปร (scope) ไม่ใช่ระดับ “บรรทัดสุดท้ายที่มันถูกใช้” ตัวแปร map ยัง อยู่ในขอบเขต ตอนที่ .await เกิด compiler จึงจองที่ให้มันในโครงสร้าง future ไปแล้ว ต่อให้ค่าข้างในถูก drop ไปก่อนก็ตาม

ทางที่ผ่านมีสองทาง และทั้งสองทางคือ ปิด scope จริงๆ (ทาง A ข้างล่างคือ file เดิมทั้ง file — use สองบรรทัด, type Db, flush_to_disk ที่โชว์ไปแล้วข้างบน — เปลี่ยนแค่ handle_conn กับ main):

// ทาง A — block ครอบ: guard ตายที่วงเล็บปิด
async fn handle_conn(db: Db) {
    let len = {
        let mut map = db.lock().unwrap();
        map.insert("k".into(), b"v".to_vec());
        map.len()
    }; // guard drop ที่วงเล็บปิดนี้
    flush_to_disk().await;
    println!("GREEN scope-block ok, len={len}");
}

#[tokio::main]
async fn main() {
    let db: Db = Arc::new(Mutex::new(HashMap::new()));
    tokio::spawn(handle_conn(db)).await.unwrap(); // .await ตัว JoinHandle ไม่งั้น main จบก่อน task
}

GREEN scope-block ok, len=1

.await ต่อท้าย tokio::spawn(...) ในบรรทัดสุดท้ายไม่ใช่ของประดับ: ถ้า main คืนค่าไปเฉยๆ runtime จะถูกปิดพร้อมกับ process และ task ที่เพิ่ง spawn ไปอาจยังไม่ทันได้พิมพ์อะไรเลย — บรรทัด GREEN scope-block ok, len=1 จะกลายเป็นการแข่งกับการปิด runtime แทนที่จะเป็นผลลัพธ์ที่แน่นอน (รันสามรอบติดกันได้บรรทัดเดิมทุกรอบเพราะเรา join แล้ว)

ทาง B คือสิ่งที่ capstone ใช้จริง: ยกทั้งก้อนออกไปเป็น function non-async อย่าง apply() ข้างบน ซึ่งได้ผลเดียวกันแต่แข็งแรงกว่า เพราะ function fn ธรรมดา เขียน .await ลงไปไม่ได้เลยตั้งแต่แรก — คนที่มาแก้ code ต่อจากคุณอีกหกเดือนข้างหน้าจึงพลาดไม่ได้ ไม่ใช่แค่ “ยังไม่พลาด”

ประโยคเดียวที่ต้องจำจากบทนี้

drop(guard) ไม่ใช่คำตอบ — scope ต่างหาก เอา lock ไปไว้ใน function non-async แล้วปัญหาทั้งชั้นนี้หายไปโดยโครงสร้าง ไม่ใช่โดยวินัย

accept loop + graceful shutdown ครบสามส่วน

ทีนี้มาถึงหัวใจของบท graceful shutdown ที่ทำจริงมีสามส่วน และขาดส่วนไหนไปก็พังคนละแบบ:

ตรวจจับ — รู้ว่าถูกสั่งให้ปิด (tokio::signal::ctrl_c())
แพร่สัญญาณ — บอกทุก task ที่กำลังทำงานอยู่ (broadcast ที่ทุกคอนเนกชัน subscribe)
รอ drain — รอจน in-flight connection จบจริงๆ ก่อน process จะตาย (mpsc ที่ recv() คืน None เมื่อ sender หมด)

ส่วนที่ 3 คือส่วนที่คนลืมบ่อยที่สุด และเป็นส่วนที่ทำให้คำว่า graceful มีความหมาย ถ้ามีแค่ 1 กับ 2 คุณก็แค่ บอก ทุกคนว่าจะปิด แล้วปิดทับหน้าเขาทันที

flowchart TD
    A[ctrl_c หรือ SIGTERM] --> B[select ใน loop accept แตกออกจาก loop]
    B --> C[broadcast send แพร่สัญญาณไปทุก task]
    C --> D1[task คอนเนกชัน 1 ตอบ BYE แล้วจบ]
    C --> D2[task คอนเนกชัน 2 ตอบ BYE แล้วจบ]
    B --> E[drop drain_tx ตัวหลักทิ้ง]
    D1 --> F[sender ที่ task ถือ ถูก drop ตาม]
    D2 --> F
    E --> F
    F --> G[drain_rx recv คืน None เพราะ sender หมดเกลี้ยง]
    G --> H[ถึงตรงนี้แปลว่าไม่มี in-flight connection เหลือ ปิด process ได้]

คำบรรยายภาพ: กลไก graceful shutdown สามส่วนของ capstone — สัญญาณจาก ctrl_c ทำให้ loop accept แตกออก จากนั้น broadcast แพร่สัญญาณให้ทุก task คอนเนกชันรู้ตัวและตอบ BYE ก่อนจบ พร้อมกันนั้นเราทิ้ง sender ตัวหลักของช่อง drain ทิ้ง เมื่อ task สุดท้ายจบและ sender สำเนาสุดท้ายถูก drop ตาม ตัวรับของช่อง drain จะคืน None ซึ่งเป็นหลักฐานว่าไม่มีคอนเนกชันไหนค้างอยู่แล้ว จึงปิด process ได้อย่างปลอดภัย

กลไก drain ในข้อ 3 คือทริกที่สวยที่สุดในบทนี้ และมันไม่ได้ส่งข้อมูลอะไรเลยสัก byte — เราสร้าง mpsc::channel::<()>(1) แล้วยัด Sender สำเนาหนึ่งเข้าไปในทุก task ที่ spawn โดย ไม่เคยเรียก send เลย ตัว task ถือมันไว้เฉยๆ พอ task จบ Sender ก็ถูก drop ไปด้วย เมื่อฝั่งหลักทิ้ง drain_tx ตัวต้นแบบและ task ตัวสุดท้ายจบลง sender ก็หมดเกลี้ยง drain_rx.recv().await จึงคืน None — นี่คือ Task.WhenAll ที่ไม่ต้องเก็บ List<Task> ไว้เลย

async fn run_server(
    listener: TcpListener,
    shutdown: impl Future<Output = ()>,
) -> std::io::Result<Db> {
    tokio::pin!(shutdown);

    let db: Db = Arc::new(Mutex::new(HashMap::new()));
    let (shutdown_tx, _) = broadcast::channel::<()>(1);
    let (drain_tx, mut drain_rx) = mpsc::channel::<()>(1);
    let limit = Arc::new(Semaphore::new(MAX_CONNS));

    loop {
        tokio::select! {
            res = listener.accept() => {
                let (sock, _) = res?;
                let permit = Arc::clone(&limit).acquire_owned().await.unwrap();
                let db = Arc::clone(&db);
                let rx = shutdown_tx.subscribe();
                let d = drain_tx.clone();
                tokio::spawn(async move {
                    let _permit = permit;
                    let _drain = d;
                    handle_conn(sock, db, rx).await;
                });
            }
            _ = &mut shutdown => {
                let _ = shutdown_tx.send(());
                break;
            }
        }
    }

    drop(drain_tx);
    let _ = drain_rx.recv().await; // None เมื่อ sender ฝั่ง task หมด = drain ครบ
    Ok(db)
}

รายละเอียดที่ไม่ใช่ของประดับ:

tokio::pin!(shutdown) — future ของสัญญาณปิดต้องอยู่รอดข้ามการวน loop เราจึงตรึงมันไว้ก่อนเข้า loop แล้วใช้ &mut shutdown ในแต่ละรอบ ถ้าเขียน shutdown เปล่าๆ ใน loop มันจะถูก move ในรอบแรก — เป็น E0382 แบบเดียวกับที่ลืม .as_mut() ในบท 1 เป๊ะ
drop(drain_tx) ก่อน recv() — ถ้าลืมบรรทัดนี้ โปรแกรมจะค้างตลอดกาล เพราะยังมี sender ตัวหนึ่งมีชีวิตอยู่ในมือเราเอง recv() จึงไม่มีวันคืน None
let _ = shutdown_tx.send(()); — broadcast::send คืน Err เมื่อ ไม่มีใคร subscribe อยู่เลย ซึ่งกรณีปิด server ที่ไม่มีคอนเนกชันค้างเป็นเรื่องปกติ ไม่ใช่ error
run_server รับ shutdown เป็น impl Future แทนที่จะเรียก ctrl_c() ข้างในตรงๆ — นี่คือสิ่งเดียวที่ทำให้ test ในหัวข้อถัดไปมีอยู่ได้

ฝั่งคอนเนกชัน: `select!` ระหว่างอ่าน frame กับสัญญาณปิด

async fn handle_conn(mut sock: TcpStream, db: Db, mut shutdown: broadcast::Receiver<()>) {
    loop {
        tokio::select! {
            frame = read_frame_or_eof(&mut sock) => match frame {
                Ok(Some(req)) => {
                    let resp = apply(&db, &req);
                    if write_frame(&mut sock, &resp).await.is_err() {
                        return;
                    }
                }
                _ => return,
            },
            _ = shutdown.recv() => {
                let _ = write_frame(&mut sock, b"BYE").await;
                return;
            }
        }
    }
}

apply(&db, &req) เรียกแบบ sync ตรงกลางของ async fn ได้อย่างสบายใจ เพราะมันเป็น fn ธรรมดาที่ guard ตายก่อนคืนค่าเสมอ — นี่คือทาง B ของหัวข้อก่อนหน้าใช้งานจริง

cancellation-safety: ทำไม branch นี้ถึงปลอดภัย (และเมื่อไหร่ที่มันจะไม่)

บท 5 วางกฎไว้แล้วว่า select! ที่วนใน loop จะ drop future ของ branch ที่แพ้ทิ้งทุกรอบ ฉะนั้นทุกครั้งที่เขียน select! ใน loop คำถามที่ต้องตอบให้ได้คือ: ถ้า future ตัวนี้ถูก drop กลางคัน ข้อมูลหายไหม

read_frame_or_eof ของเรา ไม่ cancel-safe ชัดเจน — ลองไล่ดู: มันอ่าน 4 byte ของความยาวสำเร็จ แล้วเข้า read_exact เพื่ออ่าน payload ถ้าโดน drop ตอนอ่าน payload ไปได้ครึ่งทาง byte ที่ดูดออกจาก socket ไปแล้วจะหายไปกับ buffer ที่ถูก drop ทิ้ง รอบถัดไปที่เรียกใหม่ มันจะไปอ่านกลาง frame เก่าแล้วตีความ payload เป็น “ความยาว” — สาย corrupt ทันทีโดยไม่มีใครสังเกต

แล้วทำไม code ข้างบนถึงยังถูก? เพราะ branch ที่ชนะแล้วเราไม่กลับเข้า loop อีกเลย สังเกตว่า branch shutdown.recv() ลงท้ายด้วย return ไม่ใช่ continue เมื่อสัญญาณปิดมาถึง เราไม่ได้ “พัก read_frame_or_eof ไว้ก่อนแล้วค่อยกลับมาอ่านต่อ” — เรา เลิกอ่านถาวร frame ที่อ่านค้างจึงไม่มีใครสนใจอีก เพราะคอนเนกชันนี้กำลังจะปิดอยู่แล้ว

นี่คือรูปทั่วไปของกฎ: future ที่ไม่ cancel-safe วางใน select! ได้ ก็ต่อเมื่อ branch ที่ทำให้มันถูก cancel เป็น branch ที่จบเรื่อง ถ้าคุณเขียน code ที่ดูคล้ายกันมากแต่ลงท้ายด้วย continue — เช่น เอา heartbeat timer มาแข่งกับ read_frame_or_eof แล้ววนต่อ — คุณจะได้ protocol corruption แบบสุ่มที่ reproduce ยากที่สุดในชีวิต ทางแก้มาตรฐานคือย้าย buffer ที่อ่านค้างออกไปไว้ นอก future (เป็น field ของ struct connection) เพื่อให้การอ่านรอบใหม่กลับมาต่อจากเดิมได้ Tokio เองก็เดินเส้นนี้: เอกสารของ AsyncBufReadExt::read_line ระบุตรงๆ ว่ามัน ไม่ cancel safe — ถูก cancel กลางคันแล้วข้อมูลที่อ่านไปแล้วหายเลย — แต่ next_line() ของ stream ที่ได้จาก lines() cancel safe เพราะสถานะที่อ่านค้างไปเก็บอยู่ในตัว Lines ไม่ใช่ในตัว future ที่ถูก drop 🔁 ทบทวนรายละเอียดที่ บท 5 — select! กับการยกเลิก

รันจริง end-to-end

เหลือสองชิ้นที่ต้องเติมให้ครบ: client จับคู่ กับตัว main ที่ประกอบทุกอย่าง client เป็นแค่ function เดียวที่ใช้ write_frame/read_frame_or_eof ตัวเดียวกับ server (wire เดียวกัน code เดียวกัน):

async fn request(addr: &str, req: &str) -> std::io::Result<Vec<u8>> {
    let mut sock = TcpStream::connect(addr).await?;
    write_frame(&mut sock, req.as_bytes()).await?;
    let resp = read_frame_or_eof(&mut sock).await?.unwrap_or_default();
    Ok(resp)
}

และงาน sync ก้อนสุดท้ายที่ยกมาจาก #22 ตรงๆ คือการเขียน snapshot ลง disk — สังเกตว่ามันเป็น fn ธรรมดา ไม่มี async ปนเลยสักบรรทัด และมี sync_all() (คือ fsync) อยู่ข้างใน:

fn write_snapshot_sync(path: &str, db: &HashMap<String, Vec<u8>>) -> std::io::Result<usize> {
    use std::io::Write;
    let mut body = Vec::new();
    for (k, v) in db {
        body.extend_from_slice(k.as_bytes());
        body.push(b'=');
        body.extend_from_slice(v);
        body.push(b'\n');
    }
    let mut f = std::fs::File::create(path)?;
    f.write_all(&body)?;
    f.sync_all()?; // fsync — syscall แบบ blocking ของแท้
    Ok(body.len())
}

เหลือชิ้นสุดท้ายคือตัวสัญญาณปิด — จุดนี้ต้องพูดให้ตรง การกด Ctrl-C จริงเป็น interactive เราจึงรันมันในแซนด์บ็อกซ์ไม่ได้ วิธีที่ซื่อสัตย์ที่สุดคือไม่แกล้งทำเป็นว่ารันได้ แต่เขียนสัญญาณให้ select! ระหว่าง ctrl_c() ตัวจริง กับตัวกระตุ้นจำลองผ่าน mpsc — ctrl_c() จึงถูก compile และตรวจ signature จริงใต้ feature signal ทุกครั้ง ส่วนสิ่งที่ รัน ในแซนด์บ็อกซ์คือฝั่งจำลอง นี่คือ จุดเดียวในทั้งคอร์สที่รันจริงในแซนด์บ็อกซ์ไม่ได้ 100% และหน้าตาของมันบวกกับ main ที่ประกอบทุกอย่างเข้าด้วยกันคือแบบนี้:

async fn shutdown_signal(mut simulated: mpsc::Receiver<()>) {
    tokio::select! {
        r = tokio::signal::ctrl_c() => {
            if let Err(e) = r {
                eprintln!("ctrl_c listener failed: {e}");
            }
        }
        _ = simulated.recv() => {}
    }
}

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

    let (sim_tx, sim_rx) = mpsc::channel::<()>(1);
    tokio::spawn(async move {
        let _ = request(&addr, "SET foo bar").await;
        let got = request(&addr, "GET foo").await.unwrap_or_default();
        println!("SELFTEST_GET_RESULT={}", String::from_utf8_lossy(&got));
        let _ = sim_tx.send(()).await;
    });

    let db = run_server(listener, shutdown_signal(sim_rx)).await?;
    println!("GRACEFUL_SHUTDOWN_OK");

    let snapshot = { db.lock().unwrap().clone() };
    let n = tokio::task::spawn_blocking(move || write_snapshot_sync("kaen.snapshot", &snapshot))
        .await
        .unwrap()?;
    println!("SNAPSHOT_BYTES={n}");
    Ok(())
}

cargo run --target x86_64-unknown-linux-gnu ได้ผลตามนี้:

SELFTEST_GET_RESULT=bar
GRACEFUL_SHUTDOWN_OK
SNAPSHOT_BYTES=8

สามบรรทัดนี้คือหลักฐานครบชุด: SET foo bar แล้ว GET foo ได้ bar กลับมาจริงผ่าน socket จริง · GRACEFUL_SHUTDOWN_OK พิมพ์ หลัง drain_rx.recv() คืน None แปลว่า in-flight connection จบครบก่อนแล้ว · และ 8 byte ของ snapshot คือ foo=bar\n ที่ถูก fsync ลง disk บน blocking pool

`#[tokio::test]`: round-trip + drain แล้วตรึงขอบ frame ไว้อีกตัว

เพราะ run_server รับสัญญาณปิดเป็น parameter เราจึง test ทั้งวงจรได้ใน test เดียวโดยไม่ต้องมี signal จริงเลย bind("127.0.0.1:0") ให้ OS เลือก port ว่างให้ — test จึงรันขนานกันได้ไม่ชนกัน ส่วน test ตัวที่สองมีไว้ตรึงเส้นแบ่งที่ read_frame_or_eof ทั้ง function มีอยู่เพื่อมัน: ปิดสายตรงขอบ frame = Ok(None) ส่วนหัว frame ขาดกลางคัน = Err ถ้าวันหนึ่งมีคนมา “ทำให้สั้นลง” ด้วย read_u32_le() test ตัวนี้จะแดงทันที:

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn round_trip_then_graceful_drain() {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap().to_string();
        let (tx, rx) = mpsc::channel::<()>(1);

        let server = tokio::spawn(run_server(listener, shutdown_signal(rx)));

        assert_eq!(request(&addr, "SET foo bar").await.unwrap(), b"OK");
        assert_eq!(request(&addr, "GET foo").await.unwrap(), b"bar");
        assert_eq!(request(&addr, "GET nope").await.unwrap(), b"NOTFOUND");

        tx.send(()).await.unwrap();
        let db = server.await.unwrap().unwrap();
        assert_eq!(db.lock().unwrap().get("foo").unwrap(), b"bar");
    }

    #[tokio::test]
    async fn clean_close_is_none_truncated_header_is_err() {
        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
        let addr = listener.local_addr().unwrap().to_string();

        tokio::spawn(async move {
            // สายแรก: ปิดตรงขอบ frame พอดี ยังไม่ส่งอะไรเลยสัก byte
            let (sock, _) = listener.accept().await.unwrap();
            drop(sock);
            // สายที่สอง: ส่งหัว frame ไป 2 จาก 4 byte แล้วตาย
            let (mut sock, _) = listener.accept().await.unwrap();
            sock.write_all(&[1, 0]).await.unwrap();
            drop(sock);
        });

        let mut clean = TcpStream::connect(&addr).await.unwrap();
        assert!(read_frame_or_eof(&mut clean).await.unwrap().is_none());

        let mut truncated = TcpStream::connect(&addr).await.unwrap();
        let err = read_frame_or_eof(&mut truncated).await.unwrap_err();
        assert_eq!(err.kind(), std::io::ErrorKind::UnexpectedEof);
        assert_eq!(err.to_string(), "header truncated");
    }
}

running 2 tests
test tests::round_trip_then_graceful_drain ... ok
test tests::clean_close_is_none_truncated_header_is_err ... ok

test result: ok. 2 passed; 0 failed; 0 ignored; 0 measured; 0 filtered out; finished in 0.01s

บรรทัด server.await.unwrap().unwrap() คือหัวใจของ test นี้: มันจะคืนค่าก็ต่อเมื่อ run_server เดินผ่าน drain เรียบร้อยแล้ว ถ้า drain ค้าง (เช่นลืม drop(drain_tx)) test นี้จะแขวนจนกว่า harness จะฆ่าทิ้ง — เป็นตาข่ายที่จับ bug ที่ debug ยากที่สุดของบทได้ตรงๆ

production hardening: กับดักที่ compiler เงียบสนิท

หัวข้อก่อนหน้าทั้งหมดมี compiler ยืนคุ้มกันอยู่ ทีนี้มาถึงส่วนที่ ไม่มีใครคุ้มกันเลย

กับดักที่ 1 — block runtime แล้วไม่มีใครบอก

ย้อนกลับไปดูข้อความ error ของ guard-across-await ที่ยาวเป็นหน้า นั่นคือ Rust ตอนที่มันดังที่สุด ทีนี้ดูโปรแกรมนี้ — task A เรียก std::thread::sleep ซึ่งเป็น blocking syscall เต็มตัว ส่วน task B ไม่ทำอะไรเลยนอกจากพิมพ์เวลา:

use std::time::{Duration, Instant};

#[tokio::main(flavor = "current_thread")]
async fn main() {
    let t0 = Instant::now();
    let a = tokio::spawn(async move {
        std::thread::sleep(Duration::from_millis(300)); // blocking syscall ใน async task
        println!("A done  at {:>4}ms", t0.elapsed().as_millis());
    });
    let b = tokio::spawn(async move {
        println!("B done  at {:>4}ms", t0.elapsed().as_millis());
    });
    let _ = tokio::join!(a, b);
}

cargo clippy --target x86_64-unknown-linux-gnu -- -D warnings ผ่านเงียบๆ ไม่มี warning สักบรรทัด แล้วรันได้ผลนี้:

A done  at  306ms
B done  at  306ms

task B ไม่มีงานให้ทำเลยแม้แต่นิดเดียว แต่มันได้พิมพ์ตอน 306 ms เพราะมันติดอยู่หลัง A ที่ยึด thread ของ runtime ไว้ทั้งดุ้น ใน server จริงนี่คือทุกคอนเนกชันที่ต่อคิวอยู่บน thread นั้นค้างพร้อมกันหมด และ scheduler ก็ทำอะไรไม่ได้ เพราะ Rust ยึดคืน task ที่กำลังรันอยู่ไม่ได้ — มันได้คืน worker เฉพาะตอนที่ task ยอม .await เท่านั้น

ทางแก้คือ spawn_blocking ซึ่งย้ายงาน blocking ไปที่ blocking pool แยกต่างหาก (default สูงสุด 512 thread) แล้วคืน worker ให้ scheduler ทันที:

let a = tokio::spawn(async move {
    tokio::task::spawn_blocking(move || std::thread::sleep(Duration::from_millis(300)))
        .await
        .unwrap();
    println!("A done  at {:>4}ms", t0.elapsed().as_millis());
});

B done  at    0ms
A done  at  302ms

0ms กับ 306ms — code 2 version นี้ต่างกันแค่ห่อด้วย spawn_blocking และทั้งคู่ compiler เงียบเท่ากัน นี่คือเหตุผลที่ capstone ห่อ write_snapshot_sync (ที่ข้างในมี File::sync_all() = fsync ของจริง) ไว้ใน spawn_blocking เสมอ — และเป็นเหตุผลที่ fsync กับ compaction ของ #22 ทุกจุดต้องเข้า blocking pool ตอนย้ายมาอยู่บน Tokio

เทียบกับ .NET: ThreadPool ของ CLR โตอัตโนมัติเมื่อเจอ thread starvation (ช้าหน่อยแต่มันโต) Tokio ไม่โต — worker มีเท่าที่ตั้งไว้ตอน build runtime ผลคือการ block worker ใน Rust เจ็บกว่า ใน C# มาก ทั้งที่คำเตือนน้อยกว่า

กับดักที่ 2 — spawn ไม่จำกัด

ย้อนกลับไปดู accept loop ร่างแรกของบทอีกครั้ง — นี่คือ ❌ version ดิบ ของหัวข้อนี้เต็มๆ:

loop {
    let (sock, _) = listener.accept().await?;
    let db = Arc::clone(&db);
    tokio::spawn(async move { handle_conn(sock, db).await; });   // ❌ ไม่มีเพดาน
}

มันรับทุกคอนเนกชันที่เข้ามาแล้ว spawn ทันทีโดยไม่มีเพดาน ในเว็บสาธารณะแปลว่าใครก็เปิดคอนเนกชันรัวจนหน่วยความจำหมดได้ ✅ ทางแก้คือ Semaphore ที่ acquire_owned() คืน permit ซึ่งย้ายเข้าไปอยู่ในตัว task แล้วปล่อยคืนอัตโนมัติตอน task จบ — นี่คือสามบรรทัดที่ทำให้ร่างแรกข้างบนกลายเป็น run_server ตัวจริงที่เราประกอบไปแล้ว:

let permit = Arc::clone(&limit).acquire_owned().await.unwrap();
// ...
tokio::spawn(async move {
    let _permit = permit;   // ถือไว้เฉย ๆ ปล่อยคืนตอน task จบ
    let _drain = d;
    handle_conn(sock, db, rx).await;
});

acquire_owned (ไม่ใช่ acquire) เป็นตัวที่ต้องใช้ เพราะ permit ต้องมีอายุยืนเท่ากับ task ที่ 'static ไม่ผูกกับ borrow ของ limit และนี่คือ backpressure แบบเดียวกับที่ 🔁 บท 6 — state ที่ share กับ channel สอนกับ bounded channel เพียงแต่ย้ายมาอยู่ที่ชั้นคอนเนกชัน

พูดให้ตรงข้อหนึ่ง: การ await permit ไว้ ในตัว branch ของ accept แปลว่าตอนที่คอนเนกชันเต็มเพดาน loop จะหยุดรับสายใหม่ชั่วคราว (kernel backlog รับไว้ให้แทน) ซึ่งเป็นสิ่งที่เราต้องการ — แต่มันก็หน่วงการรับรู้สัญญาณ shutdown ไปด้วยจนกว่าจะมี permit ว่าง ถ้าอยากปิดช่องนั้น ต้องเอา acquire_owned() ไปแข่งใน select! กับสัญญาณ shutdown อีกชั้น

แผนที่คำศัพท์จาก C#/.NET (สรุปรวมทั้งคอร์ส)

C# / .NET	Rust + Tokio	จุดที่ต่างจริง
`AcceptTcpClientAsync()` + `_ = HandleAsync(c)`	`listener.accept().await` + `tokio::spawn(...)`	ฝั่ง C# task เริ่มเดินทันทีที่เรียก ฝั่ง Rust ไม่มีอะไรเดินจนกว่า `spawn` จะส่งเข้า scheduler
`lock (obj) { await ... }` → CS1996	guard ข้าม `.await` → `future cannot be sent between threads safely`	C# มีกฎเฉพาะกิจ; Rust ตกมาจากระบบ type เพราะ `MutexGuard` ไม่ `Send`
`SemaphoreSlim.WaitAsync()`	`tokio::sync::Mutex::lock().await`	ใช้เมื่อ ต้อง ถือ lock ข้าม await จริงๆ เท่านั้น ไม่ใช่ default
`CancellationTokenSource` ที่ทุก task observe	`broadcast::Sender` + `Receiver` ต่อ task	Rust ไม่มี token กลาง แต่มีช่องทางแพร่ที่ทุก task subscribe เอง
`Task.WhenAll(connectionTasks)`	`mpsc` drain จน `recv()` คืน `None`	ไม่ต้องเก็บ list ของ task ไว้เลย ใช้การ drop `Sender` เป็นสัญญาณแทน
`Task.WhenAny(work, ct.AsTask())`	`tokio::select! { ... }`	`select!` drop future ที่แพ้ทิ้ง; `WhenAny` ปล่อยให้มันวิ่งต่อในเบื้องหลัง
`Console.CancelKeyPress` / `ApplicationStopping`	`tokio::signal::ctrl_c().await`	รูปเดียวกัน แต่ Rust ต้องเปิด feature `signal` เอง
`Task.Run(() => syncWork())` เพื่อ offload	`tokio::task::spawn_blocking(...)`	ThreadPool ของ CLR โตเองเมื่อ starve; worker ของ Tokio ไม่โต — block ที่นี่เจ็บกว่ามาก

honesty spine 4 เส้น — ปิดบัญชีทั้งคอร์ส

A — Future ขี้เกียจจริง: capstone นี้ไม่มีบรรทัดไหนที่ “รันเอง” เลย ทุก future เดินได้เพราะมีคน .await หรือ spawn มัน tokio::spawn(async move { ... }) ใน loop accept คือจุดเดียวที่งานของคอนเนกชันเริ่มมีชีวิต ถ้าลบ spawn ทิ้งแล้วเหลือ handle_conn(sock, db, rx); เป็นคำสั่งเปล่าๆ คอนเนกชันจะเงียบสนิท แต่ไม่ได้เงียบสนิทเสียทีเดียว — ตาข่าย #[must_use] ของบท 1 ยังทำงาน compiler ยิง warning: unused implementer of `Future` that must be used พร้อม note: futures do nothing unless you `.await` or poll them ให้ (ยืนยันแล้วด้วยการรันจริง) จุดที่ต้องระวังคือรูป let _ = handle_conn(sock, db, rx); ต่างหาก — let _ = ปิดปาก lint ตัวนี้สนิท compile ผ่านแบบ zero-warnings แล้วคอนเนกชันเงียบจริงๆ โดยไม่มีใครบอกอะไรเลย

B — std ไม่มี runtime มาให้: ทุกอย่างที่ทำให้บทนี้เป็นไปได้ — scheduler หลาย thread, reactor ที่คุย epoll, broadcast/mpsc, Semaphore, ตัวจับ signal, blocking pool — มาจาก crate เดียวชื่อ tokio ไม่มีอันไหนอยู่ใน std เลย และเรายังคงเลือกเปิดแค่ 6 feature ที่ใช้จริง ไม่ใช่ "full"

C — function colouring จริง: apply() เป็น fn ธรรมดา ซึ่งเป็น ข้อได้เปรียบ ในบทนี้ เพราะมันเขียน .await ลงไปไม่ได้เลย แต่ในทางกลับกัน มันก็เรียกอะไรที่เป็น async ไม่ได้เลยเช่นกัน ถ้าวันหนึ่ง apply ต้องอ่านจาก log ที่เป็น async ขึ้นมา คุณต้องเปลี่ยนสีมันทั้ง function แล้วปัญหา guard ก็จะกลับมาทันที

D — ที่แซนด์บ็อกซ์รันได้จริงแค่ไหน: ทั้ง 8 บทตรวจด้วย recipe เดียวคือ build/run/test/clippy บน target x86_64-unknown-linux-gnu — กล musl + rust-lld ที่ #22/#23 ใช้ได้เพราะเป็น zero-crate ปลดระวางไปตั้งแต่บทแรก (#21 เองมี clap/serde/thiserror อยู่แล้ว สิ่งที่ใหม่จริงในคอร์สนี้คือ crate dependency กลายเป็น แกน ของทุกบท และ tokio ลาก native dependency อย่าง mio/libc/socket2 เข้ามาด้วย ซึ่งต้องใช้ system linker) จุดเดียวในทั้งคอร์สที่รันจริงในแซนด์บ็อกซ์ไม่ได้คือการกด Ctrl-C เพราะมันเป็น interactive — เราจึงให้ ctrl_c() ตัวจริงถูก compile และตรวจ signature ทุกครั้งใต้ feature signal แล้วจำลองการกดด้วย mpsc สำหรับส่วนที่รันจริง เราไม่แกล้งทำเป็นว่ารันได้ และไม่แสดง output ที่ไม่ได้เกิดขึ้นจริงสักบรรทัด

ปิดคอร์ส: เส้นเรื่อง Systems pillar

คุณเดินครบสี่คอร์สของ Systems pillar แล้ว และมันเป็นเส้นเรื่องเดียวที่ต่อกันจริงๆ ไม่ใช่คอร์สสี่คอร์สที่บังเอิญใช้ภาษาเดียวกัน:

#21 rust-from-scratch — เรียนภาษา ownership/borrow/lifetime, trait, error handling, thread + channel จบด้วย CLI ที่ std ล้วน
#22 rust-kvstore — สร้างเอนจิน kaen-kvstore node เดียวที่รันจริง: wire แบบนำหน้าด้วยความยาว, append-only log + hash index, write→fsync→ack, tombstone/compaction และ thread pool ที่เขียนเอง ซึ่งบทนี้เพิ่งปลดระวางไป
#23 rust-distributed-systems — กระจายมัน เลื่อน node เดียวขึ้นเป็น cluster: logical clock, consistent hashing, quorum, CRDT, anti-entropy พร้อมเส้น scope ที่ซื่อสัตย์ว่า consensus พิสูจน์ในเครื่องเดียวไม่ได้
#24 rust-async-tokio — คอร์สนี้: ทำให้มัน concurrent อย่างที่ควรจะเป็น จาก Future ที่ขี้เกียจไปจนถึง server 1 task ต่อคอนเนกชัน ที่วิ่งบน worker ไม่กี่ตัว มีเพดานพร้อมกันที่ MAX_CONNS = 64 ตัว และปิดตัวเองได้อย่างสะอาดด้วย graceful shutdown สามส่วน (เราไม่อ้างตัวเลข throughput ใดๆ — ไม่ได้วัด จึงไม่พูด)

บทเรียนกลางของคอร์สนี้ ถ้าจะเหลือไว้ประโยคเดียว คือ compiler ของ Rust คุ้มกันคุณได้แค่ครึ่งเดียวในโลก async ครึ่งที่มันคุ้มกันนั้นทรงพลังจนน่าทึ่ง — ถือ MutexGuard ข้าม .await แล้วมันด่ายาวเป็นหน้าพร้อมชี้บรรทัดให้ครบ อีกครึ่งที่เหลือ — block worker ด้วย syscall, ปล่อย future ที่ไม่ cancel-safe ไว้ใน loop ที่ยังวนต่อ, spawn โดยไม่มีเพดาน, ลืม drain ก่อนปิด — มันเงียบสนิท ทุกกับดักในหัวข้อ production hardening ผ่าน clippy -D warnings ได้หมด นั่นแปลว่าความรู้เรื่อง runtime ไม่ใช่ของแถม แต่เป็นสิ่งที่คุณต้องถือไว้ในหัวเอง

ก้าวถัดไปของ Systems pillar ที่จ่อคิวอยู่คือ rust-interpreter (Systems V) — เขียน interpreter ของภาษาเล็กๆ ด้วย Rust ตั้งแต่ lexer ถึง tree-walking evaluator ซึ่งจะพาไปเจอ Rust อีกด้านที่คอร์สนี้ไม่ได้แตะเลย: enum กับ pattern matching ที่ลึกจริงๆ, ownership บน AST ที่เป็น recursive และ error reporting ที่มนุษย์อ่านรู้เรื่อง

🔗 อ้างอิงต้นทางของบทนี้

บทนี้อิงต้นทางที่ลงวันที่กำกับ อ่านต่อได้โดยตรง:

tokio — crates.io (เข้าถึง 2026-07-27) — รุ่นล่าสุด 1.53.1 ปล่อยราว 2026-07-20 ซึ่งเป็นรุ่นที่ทุก snippet ในคอร์สนี้ pin ไว้ และ MSRV ของมันคือ Rust 1.71
Tokio Tutorial — Shared state (เผยแพร่ 2024-01-01 · เข้าถึง 2026-07-27) — ที่มาของกฎที่บทนี้ยึด: std mutex ใช้จาก async ได้และมักเป็นตัวเลือกที่ถูกต้องเมื่อ contention ต่ำ ตราบใดที่ ไม่ถือ guard ข้าม .await เพราะ std::sync::MutexGuard ไม่ Send
Tokio Topics — Graceful Shutdown (เผยแพร่ 2024-01-01 · เข้าถึง 2026-07-27) — โครงสามส่วนที่ capstone นี้ใช้ทั้งดุ้น: ตรวจจับด้วย ctrl_c, แพร่สัญญาณผ่าน channel ที่ race ด้วย select! และรอ drain ด้วย mpsc ที่ recv() คืน None เมื่อ Sender หมด
tokio::signal::ctrl_c — docs.rs (เข้าถึง 2026-07-27) — signature ctrl_c() -> io::Result<()> ที่ complete เมื่อได้รับ Ctrl-C อยู่ใต้ feature signal
tokio::task::spawn_blocking — docs.rs (เข้าถึง 2026-07-27) — spawn_blocking<F, R> where F: FnOnce() -> R + Send + 'static รันบน blocking pool แยก (default สูงสุด 512 thread) คือที่ที่ fsync/compaction ของ #22 ต้องไปอยู่
ไทม์ไลน์อ้างอิงของทั้งคอร์ส (ไม่มี link เดียวรวม): เทรต Future stable ใน Rust 1.36.0 (2019-07-04) · async/.await stable ใน 1.39.0 (2019-11-07) · Tokio 1.0 ปล่อย 2020-12-23

เช็กความเข้าใจ — บทที่ 8

ข้อ 1 / 3

ใน version ดิบของบทนี้ เราแทรก drop(map); ไว้ก่อน .await แล้ว แต่ cargo build ก็ยังตอบ future cannot be sent between threads safely อยู่ดี ข้อใดอธิบายได้ถูกต้อง?

capstone — ยก kaen-kvstore ขึ้น Tokio: 1 task ต่อ​คอน​เนกชัน พร้อม graceful shutdown

จาก worker-per-connection ไป task-per-connection

framing เดิม​ของ #22 แต่​ทุก call กลาย​เป็น await

state ที่ share: std::sync::Mutex ไม่ใช่ tokio Mutex

❌ version ดิบ: ถือ guard ข้าม .await

drop(guard) ไม่​พอ — ต้อง​ปิด​ด้วย scope

accept loop + graceful shutdown ครบ​สาม​ส่วน

ฝั่ง​คอน​เนกชัน: select! ระหว่าง​อ่าน frame กับ​สัญญาณ​ปิด

cancellation-safety: ทำไม branch นี้​ถึง​ปลอดภัย (และ​เมื่อไหร่​ที่​มัน​จะ​ไม่)

รัน​จริง end-to-end

#[tokio::test]: round-trip + drain แล้ว​ตรึง​ขอบ frame ไว้​อีก​ตัว

production hardening: กับดัก​ที่ compiler เงียบ​สนิท

กับดัก​ที่ 1 — block runtime แล้ว​ไม่มี​ใคร​บอก

กับดัก​ที่ 2 — spawn ไม่​จำกัด

แผนที่​คำ​ศัพท์​จาก C#/.NET (สรุป​รวม​ทั้ง​คอร์ส)

ปิด​คอร์ส: เส้น​เรื่อง Systems pillar