อ่านจาก offset ที่ขอ แล้วจอดคำขอไว้แทนที่จะตอบว่าง

ผู้อ่านรายหนึ่งหายไปหนึ่งสัปดาห์แล้วกลับมา มันขอ record ที่ offset 7

broker ไม่ต้องไปเปิดตารางว่าผู้อ่านรายนี้เคยอ่านถึงไหน เพราะมันไม่มีตารางนั้น ตัวเลข 7 มาพร้อมคำขอ และนั่นคือสถานะทั้งหมดที่มีอยู่

ผลที่ตามมามีสองข้อ ข้อแรกคือการย้อนกลับไปอ่านใหม่ไม่มีราคา ข้อที่สองแพงกว่านั้น เมื่อผู้อ่านตามทันปลาย log แล้ว broker ไม่มีอะไรจะตอบ และมันเลือกที่จะไม่ตอบว่าง

📦 code ของบทนี้

บล็อก code ในบทนี้ถูกคัดมาจาก scripts/log-broker/src/fetch.rs เป็นหลัก และแตะอีกสอง file ที่ไม่มีบทเป็นเจ้าของ คือ src/partition.rs กับ src/pool.rs บทนี้เป็นบทเดียวของคอร์สที่ยก src/pool.rs มาแสดงได้

ผลรันทุกบรรทัดมาจาก scripts/log-broker/expected/04-fetch.txt ซึ่งเป็นผลของคำสั่งเดียว คือ klog fetch-demo --dir data/orders-0 --fetch-offset 7 --min-records 2 --max-wait-ms 500 --steps 4

ทั้งสองฝั่งถูกเทียบทีละ byte ด้วยเทสต์ ถ้าแก้ในบทเรียนอย่างเดียวโดยไม่แก้ที่ต้นทาง เทสต์จะแดงทันที

โดเมนยังเป็นของเดิม คือ topic orders และผู้อ่านสามรายที่บทที่ 1 แนะนำไว้ คือ kitchen courier กับ analytics

คำขอ fetch พก offset ของตัวเองมาทุกครั้ง

หัวข้อนี้แสดงว่า broker ไม่ถือสถานะของผู้อ่านไว้เลย และดูจาก field ที่คำขอจริงบรรทุกมา

คำขอ fetch ของ Kafka บรรทุกของสามอย่างที่บทนี้สนใจ

MaxWaitMs — "The maximum time in milliseconds to wait for the response."
MinBytes — "The minimum bytes to accumulate in the response."
FetchOffset ต่อหนึ่ง partition ชนิด int64 และคำอธิบายของมันสั้นมาก คือ "The message offset."

สองตัวแรกอยู่ระดับคำขอ ส่วนตัวที่สามอยู่ระดับ partition คำขอเดียวจึงขอหลาย partition พร้อมกันได้ โดยแต่ละ partition พกตำแหน่งของตัวเองมา

ไม่มี field ไหนในคำขอที่บอกว่า “ฉันคือใคร” สำหรับจุดประสงค์ของการอ่าน broker จึงไม่ต้องจำอะไรระหว่างคำขอสองครั้งเลย

นี่คือความต่างที่แท้จริงระหว่าง log กับคิว คิวถือสถานะของผู้อ่านไว้ที่ตัวมันเอง ส่วน log ผลักสถานะนั้นออกไปให้ผู้อ่านถือ แล้วรับมันกลับมาเป็นจำนวนเต็มในทุกคำขอ

บทที่ 5 จะพูดถึงที่ที่ผู้อ่านฝากตัวเลขนั้นไว้ได้ เมื่อมันไม่อยากถือเองข้ามการรีสตาร์ต แต่ตัวเลขที่ฝากไว้กับตัวเลขที่พกมากับคำขอ ยังเป็นคนละตัวอยู่ดี

เพดานของการอ่าน มี field เดียวเป็นคนแยก

หัวข้อนี้บอกว่าผู้อ่านคนละชนิดเห็น log ได้ไม่เท่ากัน และค่าที่ตัดสินคืออะไร

fetch isolation คือกติกาว่าคำขอหนึ่งอ่านได้ถึง offset ไหน Kafka ตัดสินด้วยการแตกสามทางที่เดียวใน FetchIsolation.of ซึ่งเขียนเป็น if else if else

ถ้าคำขอไม่ได้มาจาก consumer เพดานคือ LOG_END ถ้ามาจาก consumer ที่ขอ READ_COMMITTED เพดานคือ TXN_COMMITTED นอกนั้นคือ HIGH_WATERMARK

และ field ที่แยก follower ออกจาก consumer คือ ReplicaId ตัวเดียว คำอธิบายของมันใน file ของ protocol เขียนว่า "The broker ID of the follower, of -1 if this request is from a consumer."

kaen-log สร้างสองในสามทางนั้น ทางที่สามมีแต่ชื่อ บทที่ 7 อธิบายกลไกของมันไว้ แต่ไม่มีบทไหนสร้าง

/// เพดานที่สามของ Kafka คือ last stable offset · บทที่ 7 อธิบายกลไกของมัน แต่ไม่มีบทไหนสร้าง
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum FetchIsolation {
    LogEnd,
    HighWatermark,
}

/// field เดียวที่แยก follower fetch ออกจาก consumer fetch คือ `replica_id`
/// · `-1` แปลว่ามาจาก consumer
pub fn isolation_of(replica_id: i32) -> FetchIsolation {
    if replica_id >= 0 { FetchIsolation::LogEnd } else { FetchIsolation::HighWatermark }
}

fn ceiling(log: &Log, iso: FetchIsolation) -> u64 {
    match iso {
        FetchIsolation::LogEnd => log.log_end_offset(),
        FetchIsolation::HighWatermark => log.high_watermark(),
    }
}

ceiling เป็นที่เดียวที่ค่าของ enum กลายเป็นตัวเลข ทางอ่านที่เหลือรับตัวเลขนั้นมาใช้อย่างเดียว ไม่มีที่ไหนตัดสินเพดานซ้ำอีก

⚠ ตอนนี้สองเพดานให้ผลเท่ากัน และนั่นไม่ใช่ข้ออ้างของบทนี้

demo พิมพ์ทั้งสองทางออกมาให้ดู แล้วบอกเองว่ามันยังแยกกันไม่ออก

kaen-log ยังมี replica เดียว high watermark จึงเดินตาม log end ทุกก้าว เพดานสองตัวจึงชี้ไปที่เลขเดียวกันเสมอ

บทที่ 8 คือบทที่ทำให้มันต่างกันจริง จนกว่าจะถึงตอนนั้น HighWatermark คือ LogEnd ที่ใส่ชื่ออื่น และบทนี้ไม่อ้างมากกว่านั้น

== field เดียวที่แยก follower ออกจาก consumer ==
replica_id -1  → HighWatermark   (consumer อ่านได้ถึง high watermark เท่านั้น)
replica_id  2  → LogEnd   (follower อ่านได้ถึง log end)
follower อ่านจาก offset 0 ได้ record 3 ตัว
จนกว่าบทที่ 8 จะมาถึง high watermark เดินตาม log end ทุกก้าว
สองเพดานนี้จึงยังให้ผลเท่ากัน · HighWatermark ตอนนี้คือ LogEnd ที่ใส่ชื่ออื่น

การอ่านถือ lock แค่ตอนวางแผน ไม่ใช่ตลอดทาง

หัวข้อนี้กางว่า read_range ถือ lock ช่วงไหน และปล่อยก่อนอะไร

ทางอ่านของ kaen-log ต่อยอดจากการอ่านสองจังหวะที่บทที่ 2 วางไว้ คือหา entry ที่ใกล้ที่สุดในสารบัญ แล้วสแกนต่อจากจุดนั้น บทนี้ไม่สอนซ้ำ แต่รับผลของมันมาใช้

สิ่งที่บทนี้เพิ่มเข้าไปคือคำถามว่า ระหว่างที่อ่าน byte จาก disk อยู่นั้น ใครถูกกันออกไปบ้าง

คำตอบที่ผู้อ่านชอบสรุปเองคือ “ไม่ต้องถือ lock เลย” ซึ่งไม่จริง read_range ถือ lock ของ Log จริง แต่ถือแค่ช่วงสั้นๆ ช่วงเดียว

/// คืน byte ดิบของ batch ที่ครอบช่วง `[from, ceiling)` · **คืนทั้ง batch เสมอ**
/// batch ก้อนแรกอาจเริ่มที่ offset ต่ำกว่า `from` และผู้อ่านเป็นคนทิ้งส่วนหน้าเอง
/// เหมือนที่ Kafka ทำทุกประการ
pub fn read_range(p: &Partition, from: u64, ceiling: u64, max_bytes: usize) -> io::Result<Vec<u8>> {
    if from >= ceiling {
        return Ok(Vec::new());
    }
    // ถือ lock ของ log แค่ตอนวางแผน แล้วสำเนา descriptor ออกมา
    let plan = {
        let log = p.log.lock().expect("log ถูก poison");
        plan_read(&log, from, ceiling, max_bytes)?
    };
    // อ่าน byte จริงหลังปล่อย lock แล้ว · `read_at` ไม่ขยับหัวอ่านของ file
    // จึงเรียกพร้อมกันได้จากหลาย thread บน descriptor เดียวกัน
    let mut out = Vec::new();
    for (pos, len) in plan.spans {
        let mut buf = vec![0u8; u64_to_len(u64::from(len))?];
        plan.file.read_exact_at(&mut buf, pos)?;
        out.extend_from_slice(&buf);
    }
    Ok(out)
}

กลไกอยู่ที่บล็อกกลาง ใต้ lock มันทำสองอย่างแล้วจบ คือคิดว่าจะอ่านช่วงไหนบ้าง และสำเนา descriptor ของ file ออกมาด้วย try_clone

พอออกจากบล็อก guard ตายไปพร้อมกับ scope และ lock ถูกปล่อยคืน read_at ทุกครั้งที่ตามมาจึงรันตอนไม่ถือ lock

read_at เป็นตัวที่ทำให้ทรงนี้ใช้ได้ มันรับตำแหน่งมาเป็น argument แทนที่จะพึ่งหัวอ่านของ file descriptor สำเนาเดียวจึงถูกอ่านพร้อมกันจากหลาย thread ได้โดยไม่ต้องกันใคร

อีกข้อที่บล็อกนี้ตัดสินไว้คือ คืนทั้ง batch เสมอ batch ก้อนแรกอาจเริ่มที่ offset ต่ำกว่าที่ขอ และผู้อ่านเป็นคนทิ้งส่วนหน้าเอง ซึ่งเป็นสิ่งที่ Kafka ทำเหมือนกันทุกประการ

ผลของกติกา “หนึ่งคำขอเสิร์ฟจากท่อนเดียว” มองเห็นได้ตรงๆ ในผลรัน ตอนตั้งต้น demo เขียนไป 7 record แล้วท่อนถูก roll จริงสามท่อน

== ตั้งต้น ==
เขียน 7 record ผ่าน pool ที่มี worker 2 ตัว · หนึ่ง batch ต่อหนึ่ง record
log end offset    7
high watermark    7
จำนวนท่อน          3
base offset ของแต่ละท่อน  [0, 3, 6]
offset 0 อยู่บนท่อนที่กำลังเขียนอยู่: ไม่ใช่

คำขอที่ขอจาก offset 0 จึงได้ record กลับไปแค่สามตัว ทั้งที่ log end offset อยู่ที่ 7 เพราะท่อนแรกจบที่ offset 2 และคำขอนี้ไม่ข้ามไปท่อนถัดไป

จอดคำขอไว้ แทนที่จะตอบว่าง

หัวข้อนี้อธิบายว่าทำไม broker ถึงไม่ตอบว่าง และมันจอดคำขอไว้ด้วยอะไร

เอกสารออกแบบของ Kafka ตั้งปัญหาไว้ก่อน แล้วค่อยเฉลย ว่า "The deficiency of a naive pull-based system is that if the broker has no data the consumer may end up polling in a tight loop, effectively busy-waiting for data to arrive."

ทางแก้ในประโยคถัดมาคือ "To avoid this we have parameters in our pull request that allow the consumer request to block in a 'long poll' waiting until data arrives"

long poll จึงไม่ใช่ feature ของ client มันคือการที่ฝั่ง broker ยอมค้างคำขอไว้ในมือ แทนที่จะรีบตอบว่างแล้วให้ผู้อ่านวนกลับมาถามใหม่

ที่พักของคำขอที่ถูกค้างไว้แบบนั้น Kafka เรียกว่า purgatory ของ kaen-log มีขนาดเล็กที่สุดเท่าที่ยังทำงานได้ คือ Mutex<u64> กับ Condvar หนึ่งคู่ต่อหนึ่ง partition

/// ที่จอดของคำขอที่รออยู่ · **หนึ่ง `Condvar` ต่อหนึ่ง partition**
/// สิ่งที่มันถือได้มีแค่ `u64` ตัวเดียวคือ log end offset
pub struct PartitionWaiters {
    pub lock: Mutex<u64>,
    pub cv: Condvar,
}

สอง field นี้กำหนดขอบเขตของทั้งบท ทุกอย่างที่ที่จอดตัวนี้ตอบได้ ต้องคิดออกมาจากจำนวนเต็มตัวเดียว

serve_fetch เป็น function ที่จอดคำขอ และผลของมันมีสามช่อง

pub struct FetchOutcome {
    /// เข้าไปรอใน `wait_timeout` อย่างน้อยหนึ่งครั้งหรือเปล่า
    pub parked: bool,
    /// ออกมาเพราะเส้นตายถึง ไม่ใช่เพราะได้ของครบ
    pub timed_out: bool,
    pub bytes: Vec<u8>,
}

parked กับ timed_out เป็นคนละคำถามกัน ข้อแรกถามว่าเคยเข้าไปรอไหม ข้อที่สองถามว่าออกมาเพราะหมดเวลาหรือเพราะได้ของ

เส้นตายถูกตรึงก่อนเข้า loop แล้วนับถอยหลังจากมัน

หัวข้อนี้กางรูปของการรอออกมาดู แล้วบอกตามตรงว่าการกางมันออกมาเป็นการเลือกของบทเรียน

serve_fetch ตรึงเส้นตายไว้ก่อนแตะ Condvar แล้วเข้า loop ทุกรอบมันตรวจเงื่อนไขก่อน ถ้าครบก็ออกทันทีโดยไม่รอเลย

    let deadline = Instant::now() + max_wait;
    let mut parked = false;
    let mut timed_out = false;
    let mut end = p.waiters.lock.lock().expect("waiters ถูก poison");
    loop {
        if records_between(fetch_offset, *end) >= min_records {
            break;
        }

การตรวจก่อนรอไม่ใช่รายละเอียด มันคือสิ่งที่ทำให้คำขอที่มีของอยู่แล้วไม่ต้องแตะ Condvar เลยสักครั้ง และเป็นเหตุผลที่ parked ยังเป็น false ได้แม้จะเข้ามาถึง loop แล้ว

ท่อนที่เหลือของ loop คือการนับถอยหลัง

        let remaining = deadline.saturating_duration_since(Instant::now());
        if remaining.is_zero() {
            timed_out = true;
            break;
        }
        parked = true;
        let (guard, _timeout) =
            p.waiters.cv.wait_timeout(end, remaining).expect("waiters ถูก poison");
        end = guard;
    }

remaining ถูกคิดใหม่ทุกรอบจากเส้นตายเดิมตัวเดิม ไม่ใช่จาก max_wait ทั้งก้อน เวลารวมของ loop จึงถูกจำกัดด้วยเส้นตายนั้น ไม่ว่าจะถูกปลุกกี่ครั้ง

การตรวจเงื่อนไขซ้ำทุกครั้งที่ตื่นก็จำเป็นเหมือนกัน เอกสารของ Condvar เขียนไว้เองว่า "This function is susceptible to spurious wakeups." และตามด้วยว่า predicate "must always be checked each time this function returns"

Instant เป็นชนิดข้อมูลเดียวในเรื่องเวลาที่ทั้ง crate ยอมให้ใช้ และใช้ได้ที่ src/fetch.rs file เดียว ที่เหลือรับนาฬิกาเข้ามาเป็น now_ms จากผู้เรียกทั้งหมด

เหตุผลของข้อยกเว้นข้อนี้อยู่ที่เอกสารของ wait_timeout เอง มันบอกว่าเวลาที่รอ "is measured with a monotonic clock, and not affected by the changes made to the system time" เส้นตายที่วัดด้วยนาฬิกาแบบนั้นจึงไม่ขยับตามการปรับเวลาของเครื่อง

⚠ เหตุผลที่ฟังขึ้น แต่ตรวจแล้วไม่จริง

คำอธิบายที่ได้ยินบ่อยสำหรับการเขียน loop เองคือ wait_timeout_while ตั้งเวลาใหม่เต็มจำนวน ทุกครั้งที่ถูกปลุกเก้อ เวลารวมของมันจึงไม่ถูกจำกัดด้วย dur

ข้ออ้างนั้นไม่จริง เอกสารของ wait_timeout_while เขียนด้วยประโยคเดียวกับ wait_timeout ว่า thread "will be blocked for roughly no longer than dur"

และ source ของ std รุ่น 1.97.1 ซึ่งเป็นรุ่นที่ crate นี้ตรึงไว้ ก็ทำเลขคณิตชุดเดียวกับ serve_fetch เป๊ะ คือจับ Instant::now() ไว้ก่อนเข้า loop แล้วคิด dur.checked_sub(start.elapsed()) ใหม่ทุกรอบ ถ้าลบไม่ได้แล้วก็คืนค่าออกมาพร้อมสถานะหมดเวลา

เหตุผลผิดที่พาไปสู่ข้อสรุปถูก คือของที่ตรวจยากที่สุดในบทเรียนทั้งใบ

แล้วเหตุผลจริงคืออะไร คำตอบสั้นและไม่มีอะไรซ่อนอยู่ wait_timeout_while ทำงานตรงนี้ได้เหมือนกัน และเป็นการเรียกที่สั้นกว่า ถ้า kaen-log เรียกมัน การรอก็ยังถูกจำกัดด้วยเส้นตายเหมือนเดิม

kaen-log กาง loop ออกมาเขียนเอง เพราะบทนี้ต้องให้ผู้อ่านเห็นว่าการจอดคำขอประกอบด้วยอะไรบ้าง เงื่อนไข เส้นตาย และการถูกปลุก ทั้งสามอย่างอยู่ในสายตาเดียวกันเมื่อ loop ถูกกางออกมา ส่วนการเรียกตัวเดียวจะรวบทั้งสามอย่างไปไว้หลังชื่อ function

นี่เป็นการเลือกของการสอน ไม่ใช่ข้อกำหนดของความถูกต้อง และบทนี้เขียนมันออกมาตรงๆ แบบนี้ เพราะข้ออ้างเชิงกลไกที่ยกมาแทนคำว่า “เพื่อให้เห็น” คือข้ออ้างที่ไม่มีใครตรวจแล้วอยู่ยาว

ทางที่ไม่เคยจอดเลย

หัวข้อนี้แสดงเส้นทางที่ตัดสินก่อนทุกอย่าง และเหตุผลที่มันตัดสินด้วยท่อน

คำขอที่อ่านของเก่าไม่ควรจอด เพราะของที่มันขออยู่บน disk ครบแล้ว การรอจึงไม่ได้อะไรเพิ่ม Kafka จัดการข้อนี้ด้วย Case F ในบรรดาแปดเงื่อนไขที่ทำให้คำขอที่จอดอยู่ถูกปลุก

หัวข้อของ Case F ใน DelayedFetch.scala เขียนว่า "The fetch offset locates not on the last segment of the log" และเงื่อนไขที่แท้จริงคือ fetchOffset.onOlderSegment(endOffset)

สังเกตว่ามันตัดสินที่ระดับท่อน ไม่ใช่ที่ระยะห่างของ offset kaen-log ลอกทรงนี้มาทั้งดุ้น

/// จอดคำขอไว้จนกว่าจะมี record ครบ `min_records` หรือจนกว่าจะหมดเวลา
pub fn serve_fetch(
    p: &Partition,
    fetch_offset: u64,
    min_records: usize,
    max_wait: Duration,
    iso: FetchIsolation,
) -> io::Result<FetchOutcome> {
    // Case F มาก่อนทุกอย่าง · คำขอที่อ่านของเก่าไม่เคยจอด
    // เป็นการตรวจระดับ **ท่อน** ไม่ใช่ระดับระยะห่างของ offset — เหมือน Kafka
    if !p.on_active_segment(fetch_offset) {
        let top = { ceiling(&p.log.lock().expect("log ถูก poison"), iso) };
        let bytes = read_range(p, fetch_offset, top, MAX_FETCH_BYTES)?;
        return Ok(FetchOutcome { parked: false, timed_out: false, bytes });
    }

การ return ตรงนั้นเกิดขึ้นก่อนบรรทัดที่แตะ PartitionWaiters แม้แต่บรรทัดเดียว คำขอทางนี้จึงไม่เคยรู้จัก Condvar เลย

== รอบที่ 1 · Case F — คำขอที่อ่านของเก่าไม่เคยจอด ==
fetch จาก offset 0
  จอด            ไม่ใช่
  หมดเวลา        ไม่ใช่
  byte ที่ได้      411
  record ที่ได้    3
  ครอบ offset    0 ถึง 2
ไม่แตะ Condvar เลยสักครั้ง · ตัดสินที่ระดับท่อน ไม่ใช่ระยะห่างของ offset
และหนึ่งคำขอถูกเสิร์ฟจากท่อนเดียว จึงได้แค่ record ที่อยู่ในท่อนแรก

รอบนี้ขอ min_records ไว้ที่ 2 และได้กลับมา 3 ตัว ผลรันรอบเดียวจึงยังไม่พิสูจน์ Case F เพราะเงื่อนไขจำนวน record ก็ครบพอดีอยู่แล้ว

ตัวที่พิสูจน์อยู่ในเทสต์ มันตั้ง min_records ไว้ที่ 99 ซึ่งสูงเกินจริงจนไม่มีทางครบ แล้วยืนยันว่าคำขอยังไม่จอด

        assert!(p.segment_count() > 1, "ตั้งใจให้เกิดการ roll จริง");
        assert!(!p.on_active_segment(0));
        let outcome =
            serve_fetch(&p, 0, 99, Duration::from_secs(60), FetchIsolation::HighWatermark).unwrap();
        assert!(!outcome.parked, "อ่านของเก่าต้องไม่จอด แม้ min_records จะสูงเกินจริง");

สองบรรทัดบนสุดคือสิ่งที่ทำให้เทสต์ตัวนี้ไม่โกหก ถ้าท่อนไม่ roll จริง เทสต์จะล้มที่บรรทัดแรก แทนที่จะเขียวโดยเดินคนละทางกับที่ชื่อของมันอ้าง

demo เองก็มีด่านแบบเดียวกัน ถ้าตั้งต้นแล้วท่อนไม่ roll มันคืน error ออกมาแทนที่จะพิมพ์ข้ออ้าง เรื่อง Case F ทั้งที่ไม่เคยเดินเข้าไปในสาขานั้น

คนปลุกคือทาง produce

หัวข้อนี้ชี้ว่าใครเป็นคนขยับตัวเลขในที่จอด และปลุกคนที่รออยู่

คำขอที่จอดอยู่รอค่าเดียวคือ log end offset ตัวใหม่ และคนเดียวที่ขยับมันได้คือทาง produce ของบทที่ 3

    /// ประกาศ log end ตัวใหม่แล้วปลุกทุกคนที่จอดรออยู่
    /// · นี่คือ `DelayedFetch.tryComplete` ของ Kafka ในรูปที่เล็กที่สุดที่ยังจริง
    pub fn publish_log_end(&self, new_end: u64) {
        let mut end = self.waiters.lock.lock().expect("waiters ถูก poison");
        *end = new_end;
        self.waiters.cv.notify_all();
    }

สาม statement นี้คือ tryComplete ทั้งดุ้นของ kaen-log ขอ lock ของที่จอด เขียนตัวเลขใหม่ แล้วปลุกทุกคน

notify_all ไม่ใช่ notify_one เพราะผู้อ่านหลายรายรอที่ offset คนละที่ได้ การปลุกรายเดียวอาจปลุกโดนรายที่เงื่อนไขยังไม่ครบ แล้วรายที่ครบแล้วนอนต่อ

sequenceDiagram
    participant K as kitchen · ตามทันปลาย log
    participant B as broker · kaen-log
    participant P as producer
    participant A as analytics · อ่านจากต้น log
    K->>B: fetch จาก offset 10 · ขออย่างน้อย 2 record
    Note over B: ยังไม่มีของให้ตอบ จึงจอดไว้บน Condvar<br/>เส้นตายถูกตรึงไว้ตั้งแต่ก่อนเข้า loop
    A->>B: fetch จาก offset 0
    B-->>A: ตอบทันที เพราะ offset 0 อยู่บนท่อนเก่า จึงไม่เคยจอด
    P->>B: produce เข้ามาบน worker ของ pool
    Note over B: publish_log_end ขยับตัวเลข แล้ว notify_all
    B-->>K: ปลุกแล้วตอบ ครอบ offset 10 ถึง 11

คำบรรยายภาพ: เส้นทางของ analytics ไม่เคยแตะ Condvar เลย เพราะมันอ่านของเก่า ส่วนเส้นทางของ kitchen จอดรอค่าที่ producer เป็นคนขยับ · การปลุกกับการตอบเป็นคนละก้าว เพราะ guard ถูกปล่อยคืนก่อนการอ่านจะเริ่ม

ผลรันรอบที่ 4 เดินเส้นทางนี้จริง ฝั่งอ่านรออยู่บน thread ของตัวเอง ส่วน produce วิ่งบน pool

== รอบที่ 4 · produce มาถึงระหว่างที่ fetch รออยู่ ==
fetch จอดรออยู่ที่ offset 10 · produce วิ่งบน pool คนละ thread
record ที่ได้กลับมา 2
  ครอบ offset    10 ถึง 11
log end offset หลังรอบนี้ 12
รอบนี้ไม่พิมพ์ว่า `จอด` หรือเปล่า เพราะคำตอบขึ้นกับว่า thread ไหนถึงก่อน
ส่วน record ที่ได้กลับมาไม่ขึ้นกับลำดับ · `expected/` บันทึกได้เฉพาะอย่างหลัง

สองบรรทัดสุดท้ายคือส่วนที่สำคัญที่สุดของทั้งรอบ รอบนี้เป็นรอบเดียวที่ parked กับ timed_out มีโอกาสตอบไม่ตรงกัน และมันคือรอบเดียวที่ผลรันไม่ยอมพิมพ์ parked ออกมา

เหตุผลคือค่านั้นขึ้นกับว่า thread ไหนถึงก่อน ผลที่ขึ้นกับการสลับ thread ไม่ใช่ข้อเท็จจริงที่บันทึกซ้ำได้ ส่วน record ที่ได้กลับมาเป็นข้อเท็จจริงแบบนั้น จึงบันทึกไว้ได้

คำขอที่หมดเวลาแล้วตอบว่าง ไม่ใช่ความผิดพลาด

หัวข้อนี้แสดงรอบที่จอดจริงแล้วหมดเวลาจริง แล้วบอกว่าคำตอบว่างแปลว่าอะไร

รอบที่ 2 ของ demo ไม่มี producer เลย คำขอจึงจอดจนเส้นตายถึง แล้วออกมามือเปล่า

== รอบที่ 2 · จอดแล้วหมดเวลา — ไม่มี producer ในรอบนี้ ==
fetch จาก offset 7
  จอด            ใช่
  หมดเวลา        ใช่
  byte ที่ได้      0
  record ที่ได้    0
  ครอบ offset    (ไม่มี)
จอดจริงและหมดเวลาจริง เพราะรอบนี้ไม่มีใครเขียนอะไรเลย
คำตอบว่างเปล่าเป็นคำตอบที่ถูกต้อง ไม่ใช่ความผิดพลาด

demo มีด่านของตัวเองก่อนรอบนี้ ถ้ามี record พออยู่แล้วตั้งแต่ต้น มันคืน error ออกมา เพราะคำขอจะไม่จอด และข้ออ้างของรอบนี้จะว่างเปล่าทั้งที่ผลรันดูสวยงาม

รอบที่ 3 เป็นด้านตรงข้าม producer มาก่อนที่คำขอจะมาถึง เงื่อนไขจึงครบตั้งแต่ตรวจครั้งแรก

== รอบที่ 3 · producer มาก่อน — ไม่ต้องจอด ==
fetch จาก offset 7
  จอด            ไม่ใช่
  หมดเวลา        ไม่ใช่
  byte ที่ได้      289
  record ที่ได้    3
  ครอบ offset    7 ถึง 9
log end offset หลังรอบนี้ 10

สามรอบนี้กับรอบที่ 4 รวมกันจึงครอบทางเดินทั้งหมดที่ serve_fetch มี คือไม่แตะ Condvar เลย จอดแล้วหมดเวลา เงื่อนไขครบตั้งแต่ตรวจครั้งแรก และจอดแล้วถูกปลุกจนเงื่อนไขครบ

`drop(end)` เขียนแยกบรรทัด เพราะกับดักนี้เคยกัดมาแล้ว

หัวข้อนี้ชี้บรรทัดเดียวที่กันไม่ให้บทนี้ทำผิดในสิ่งที่คอร์สที่ 22 สอนไว้แล้ว

คอร์ส kaen-kvstore สอนกับดักการถือ guard คร่อม I/O ที่ block ไว้แล้ว คอร์สนี้ไม่สอนซ้ำ แต่ต้องแสดงว่าเลี่ยงมันตรงไหน

จุดที่เผลอง่ายที่สุดคือจุดนี้พอดี เพราะตอนออกจาก loop guard ของที่จอดยังอยู่ในมือ และการอ่าน disk อยู่บรรทัดถัดไป

    // *** บรรทัดที่บทนี้มีไว้เพื่อสอน ***
    // ห้ามถือ guard คร่อมการอ่าน disk · #22 บทที่ 5 สอนกับดักนี้ไปแล้ว และ function
    // ที่จะเผลอทำก็คือ function ที่บทนี้ตั้งชื่อตาม · เขียนแยกบรรทัดไว้เลย
    // เพื่อไม่ให้ใครลบทิ้งเพราะคิดว่าไม่จำเป็น
    drop(end);

    let top = { ceiling(&p.log.lock().expect("log ถูก poison"), iso) };
    let bytes = read_range(p, fetch_offset, top, MAX_FETCH_BYTES)?;
    Ok(FetchOutcome { parked, timed_out, bytes })
}

ถ้าไม่มีบรรทัดนั้น guard จะมีชีวิตจนจบ function และ lock ของที่จอดจะถูกถือไว้ตลอดการอ่าน disk

ผลไม่ใช่แค่ช้า มันคือการกันไม่ให้ publish_log_end เข้ามาขยับตัวเลขได้เลย ทาง produce ที่เป็นคนปลุกจึงถูกกันออกไปด้วยตัวคนที่รอมันอยู่

สอง lock คนละตัว และบรรทัดนี้ปล่อยแค่ตัวเดียว

drop(end) ปล่อย lock ของ PartitionWaiters เท่านั้น

บรรทัดถัดมาขอ lock ของ Log ใหม่เพื่อหาเพดาน แล้ว read_range ก็ขอมันอีกครั้งเพื่อวางแผนอ่าน lock ตัวนั้นถูกถือสั้นๆ สองครั้ง ไม่ใช่ไม่ถูกถือเลย

ข้อสรุปว่า “การอ่านไม่ถือ lock อะไรเลย” จึงผิด ข้อที่ถูกคือ ไม่มี lock ตัวไหนถูกถือคร่อม read_at

คำขอที่จอดอยู่ ห้ามกิน worker ของ pool

หัวข้อนี้อธิบาย deadlock ที่เกิดจากการเอา pool ของคอร์สที่ 22 มาใช้กับทาง fetch ตรงๆ

pool ของ worker ที่คอร์สที่ 22 สร้างไว้ถูกยกมาใช้ใน kaen-log ทั้งดุ้น แต่ใช้กับทาง produce เท่านั้น

เหตุผลเขียนไว้ที่หัว file ของมันเอง

//! pool ของ worker แบบมีเพดาน — **ใช้ซ้ำจากคอร์ส #22 บทที่ 5 ไม่ได้คิดขึ้นใหม่**
//!
//! kaen-log ให้ pool นี้รับใช้ **ทาง produce เท่านั้น** · คำขอ fetch ที่จอดรอ
//! วิ่งบน thread ของตัวเอง · เหตุผลอยู่ที่ `src/fetch.rs`: fetch ที่จอดรออยู่บน
//! worker ของ pool ที่มี N ตัว จะกิน worker จนหมดทั้ง N ตัว แล้ว produce ที่เป็น
//! คนเดียวที่ปลุกมันได้จะไม่เหลือ worker ให้รัน · broker ค้างทั้งตัวโดยไม่มี lock
//! ตัวไหนถูกถือไว้เลย
//!
//! **crate นี้ไม่มี code ที่แตะ network สักบรรทัด** · `std::net` ไม่ถูกอ้างที่ไหนเลย
//! ข้ออ้าง "thread ต่อหนึ่งการเชื่อมต่อ" จึงเป็นข้ออ้างที่วัดจากที่นี่ไม่ได้ · สิ่งที่
//! demo ของบทที่ 4 แสดงได้จริงคือ fetch ที่จอดอยู่กิน thread ของตัวเองหนึ่งตัว
//! ขณะที่ produce วิ่งบน pool

ประโยคสุดท้ายคือส่วนที่ทำให้ข้อนี้อันตราย ไม่มี lock ตัวไหนถูกถือไว้เลย เครื่องมือที่ตรวจ deadlock จากการวน lock จึงไม่มีอะไรให้เห็น

รูปของมันคือวงจรระหว่างทรัพยากรกับเงื่อนไข ไม่ใช่วงจรระหว่าง lock สองตัว

worker ทุกตัวถูกยึดโดยคำขอที่รอเงื่อนไขอยู่
เงื่อนไขนั้นครบได้จากทาง produce ทางเดียว
ทาง produce ต้องการ worker เพื่อจะรัน และไม่เหลือ worker แล้ว

ผลรันปิดท้ายด้วยการพูดข้อนี้ออกมาตรงๆ พร้อมจำนวน worker ที่ demo ใช้จริง

และ fetch ที่จอดอยู่ กิน thread ของตัวเองหนึ่งตัว ไม่ใช่ worker ของ pool
pool มี worker 2 ตัว · ถ้า fetch ที่จอดกิน worker ครบทุกตัว
producer ที่เป็นคนเดียวที่ปลุกมันได้จะไม่เหลือ worker ให้รัน

ทางออกของ kaen-log ตรงไปตรงมา คือคำขอ fetch ที่จอดอยู่ไม่วิ่งบน pool มันวิ่งบน thread ของตัวเอง demo ของบทนี้แสดงข้อนั้นออกมาตรงๆ fetch ที่จอดรออยู่กิน thread ของตัวเองหนึ่งตัว ขณะที่ produce เดินบน worker ของ pool แล้วเข้าไปปลุกมัน

⚠ ขอบเขตของข้ออ้างนี้ · crate นี้ไม่มี network เลย

kaen-log ไม่มี code ที่แตะ network สักบรรทัด std::net ไม่ถูกอ้างที่ไหนใน src/ เลย

บทนี้จึงพูดว่า “thread ต่อหนึ่งการเชื่อมต่อ” ไม่ได้ เพราะไม่มีการเชื่อมต่อให้นับตั้งแต่ต้น สิ่งที่วัดได้จริงจาก demo คือ fetch ที่จอดอยู่บน thread ของตัวเอง ส่วน produce วิ่งบน pool และนั่นคือขอบเขตทั้งหมดของข้ออ้าง

ของจริงที่ต่อ socket เข้ามาด้วยจะต้องตัดสินอีกชั้นว่า thread ที่จอดนั้นผูกกับอะไร คอร์สนี้ไม่ได้สร้างชั้นนั้น จึงไม่พูดถึงมัน

Kafka แก้ปัญหาคนละแบบ เอกสาร Network Layer เขียนว่า "The threading model is a single acceptor thread and N processor threads which handle a fixed number of connections each."

processor เหล่านั้นเป็น NIO ที่ไม่ block ส่วนคำขอที่ถูกจอดไม่ได้ถือ thread ไว้เลย มันเป็น object ในที่พัก std::net เขียนทรงนั้นไม่ได้ และคอร์สนี้เลือกไม่ไปหา runtime แบบ async มาช่วย

สามข้อที่ที่จอดของ kaen-log ไม่ใช่ของ Kafka

หัวข้อนี้ไล่ความต่างสามข้อที่ห้ามกลบ พร้อมเหตุผลเชิงกลไกของแต่ละข้อ

ข้อแรก หน่วยที่นับไม่เหมือนกัน ธงของ kaen-log ชื่อ --min-records และมันนับ record จริงๆ ส่วน MinBytes ของ Kafka นับ byte

ความต่างนี้ไม่ใช่เรื่องหน่วย แต่เป็นเรื่องของที่ที่จอดถืออยู่

/// จำนวน record ระหว่างสอง offset · **นับ record ไม่ใช่ byte**
///
/// Kafka เทียบด้วย `endOffset.positionDiff(fetchOffset)` ซึ่งเป็นผลต่างของ **byte**
/// และต้องใช้ `LogOffsetMetadata` (offset + ท่อน + ตำแหน่งในท่อนนั้น) จึงจะคิดได้
/// · ที่จอดของ kaen-log ถือ `u64` ตัวเดียว จึงคิดค่านั้นไม่ได้ และธงของมัน
/// ชื่อ `--min-records` ด้วยเหตุผลนี้ · ห้ามบทไหนเรียกมันว่า `min_bytes`
pub fn records_between(from: u64, to: u64) -> usize {
    u64_to_len(to.saturating_sub(from)).unwrap_or(usize::MAX)
}

Kafka คิดค่านั้นจาก endOffset.positionDiff(fetchOffset) ซึ่งต้องรู้ว่าปลาย log อยู่ท่อนไหน และอยู่ตำแหน่งที่เท่าไรในท่อนนั้น ที่จอดของ kaen-log ถือ u64 ตัวเดียว จึงคิดไม่ได้

ข้อที่สอง ขอบเขตของการเทียบไม่เหมือนกัน เกณฑ์ min_records ของ kaen-log เทียบทีละ partition เพราะมี Condvar หนึ่งตัวต่อหนึ่ง partition

ของ Kafka รวมทุก partition ในคำขอเดียวกันก่อนเทียบ DelayedFetch บวกค่าเข้า accumulatedSize ทีละ partition แล้วเทียบกับ params.minBytes ครั้งเดียวตอนท้าย

ข้อที่สาม ตัวจับเวลาไม่เหมือนกัน Condvar หนึ่งตัวรับคำขอที่จอดอยู่ได้เท่าที่ thread จะพอ ส่วน purgatory ของ Kafka เป็น hierarchical timing wheel

บทความที่อธิบายการออกแบบนั้นบอกเหตุผลไว้ว่า timing wheel มี "O(1) cost for insert/delete (start-timer/stop-timer) whereas priority queue based timers, such as java.util.concurrent.DelayQueue and java.util.Timer, have O(log n) insert/delete cost"

นั่นคือข้อแตกต่างเชิงโครงสร้าง ไม่ใช่ข้อแตกต่างเชิงความเร็วที่คอร์สนี้วัดเอง kaen-log ไม่มีตัวจับเวลาแยกต่างหากเลย เส้นตายของแต่ละคำขออยู่บน stack ของ thread ที่รอมันเอง

== สองข้อที่ kaen-log ต่างจาก Kafka และไม่ควรกลบ ==
1. min_records นับ record · min_bytes ของ Kafka นับ byte
   ที่จอดของ kaen-log ถือ u64 ตัวเดียว จึงคิดผลต่างเป็น byte ไม่ได้
2. เพดานนี้เป็นราย partition เพราะมี Condvar หนึ่งตัวต่อหนึ่ง partition
   ส่วน Kafka รวมทุก partition ในคำขอเดียวกันก่อนเทียบ

ผลรันพูดสองข้อแรก เพราะมันเป็นสองข้อที่ demo วัดได้จริงจาก code ของตัวเอง ข้อที่สามเป็นข้อเปรียบเทียบกับของที่ไม่ได้อยู่ในเครื่อง จึงอยู่ในบทเรียนแทน

เงื่อนไขปลุกของ Kafka มีแปดข้อ ของ kaen-log มีสอง

DelayedFetch ไล่เงื่อนไขไว้ตั้งแต่ Case A ถึง Case H เช่น broker ไม่ได้เป็น leader แล้ว replica ไม่อยู่บนเครื่องนี้แล้ว หรือพบ epoch ที่แตกกิ่งออกไป

kaen-log มีสองข้อ คือจำนวน record ครบตามที่ขอ และคำขออ่านของเก่าอยู่บนท่อนที่ไม่ใช่ท่อนสุดท้าย บวกเส้นตายที่ทำให้ออกมาได้เสมอ

ข้อที่หายไปทั้งหมดเป็นเงื่อนไขที่มีความหมายเมื่อมีหลาย replica เท่านั้น บทที่ 8 จะสร้างโลกใบนั้นขึ้นมา แล้วเงื่อนไขบางข้อในนั้นจะกลับมามีความหมาย

ของที่บทนี้ไม่ได้ทำ

หัวข้อนี้ประกาศขอบเขตของทาง fetch ไว้ล่วงหน้า

ไม่มี quota และไม่มีการหน่วงผู้อ่านที่ดูดแรงเกินไป Kafka มีกลไกนั้น kaen-log ไม่มีเลย คำขอที่ขอมากที่สุดได้ไปมากที่สุดเสมอ
ไม่มี sendfile kaen-log อ่าน byte ขึ้นมาไว้ในหน่วยความจำก่อนแล้วค่อยส่ง ส่วน Kafka ส่งจาก page cache ลงสายได้โดยไม่ผ่าน buffer ของ process
ไม่มีการรวมหลาย partition ในคำขอเดียว คำขอหนึ่งครั้งของ kaen-log แตะ partition เดียว ซึ่งเป็นเหตุผลที่เกณฑ์ min_records ของมันเทียบทีละ partition ไปด้วย
ไม่มี TxnCommitted enum ของ kaen-log มีสองค่า และคอมเมนต์เหนือมันบอกไว้เองว่าทางที่สาม ถูกเอ่ยชื่อในบทที่ 7 แต่ไม่เคยถูกสร้าง

บทที่ 5 จะรับตัวเลขที่ผู้อ่านพกมากับคำขอนี้ ไปหาที่ฝากที่รอดจากการรีสตาร์ต และที่ฝากนั้นจะเป็น log อีกก้อนหนึ่ง ไม่ใช่ตารางในหน่วยความจำ

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

FetchRequest.json tag 4.0.0 (ตรวจแล้ว 2026-09-03) — คำอธิบายของ MaxWaitMs MinBytes และ FetchOffset ที่ยกมาทั้งสามประโยค และประโยคของ ReplicaId ที่บอกว่า -1 แปลว่ามาจาก consumer
FetchIsolation.java tag 4.0.0 (ตรวจแล้ว 2026-09-03) — การแตกสามทางด้วย if else if else ที่แม็ป replicaId กับ isolationLevel ไปเป็น LOG_END TXN_COMMITTED หรือ HIGH_WATERMARK
DelayedFetch.scala tag 4.0.0 (ตรวจแล้ว 2026-09-03) — รายการ Case A ถึง Case H ประโยคของ Case F ที่ยกมา เงื่อนไข fetchOffset.onOlderSegment(endOffset) และการบวก accumulatedSize ก่อนเทียบกับ params.minBytes ครั้งเดียวตอนท้าย
Apache Kafka 4.0 — Design (ตรวจแล้ว 2026-09-03) — ย่อหน้าที่ตั้งปัญหาเรื่องการวนถามในลูปแคบ แล้วเฉลยด้วยคำว่า long poll ทั้งสองประโยคที่ยกมาอยู่ติดกันในย่อหน้าเดียว
ConsumerConfig.java tag 4.0.0 (ตรวจแล้ว 2026-09-03) — ค่าปริยายฝั่ง client ที่บทนี้อ้างถึงในฐานะที่มาของธงใน demo DEFAULT_FETCH_MIN_BYTES เป็น 1 DEFAULT_FETCH_MAX_WAIT_MS เป็น 500 และ DEFAULT_MAX_PARTITION_FETCH_BYTES เขียนไว้เป็น 1 * 1024 * 1024
Apache Kafka 4.0 — Implementation › Network Layer (ตรวจแล้ว 2026-09-03) — ประโยคเรื่อง acceptor thread ตัวเดียวกับ processor thread N ตัว ที่ยกมาเทียบกับทรงของ kaen-log ซึ่งมี pool สำหรับ produce และ thread ต่อหนึ่ง fetch ที่จอดรอ · ห้ามเรียกทรงนี้ว่า thread ต่อหนึ่งการเชื่อมต่อ เพราะ crate นี้ไม่มีการเชื่อมต่อให้นับ ตามที่หัวข้อข้างบนวัดไว้แล้ว
Apache Kafka, Purgatory, and Hierarchical Timing Wheels (Yasuhiro Matsuda เผยแพร่ 2015-10-28 · ตรวจแล้ว 2026-09-03) — ประโยคเรื่อง O(1) ของ timing wheel เทียบกับ O(log n) ของตัวจับเวลาที่ใช้ priority queue ตัวเลข throughput ในบทความนั้นเป็นของเครื่องปี 2015 บทนี้จึงไม่ยกมา
Rust std — struct Condvar (ตรวจแล้ว 2026-09-03) — ประโยคเรื่อง spurious wakeup ประโยคที่บอกว่าต้องตรวจ predicate ทุกครั้ง ประโยคเรื่องนาฬิกา monotonic และประโยค "will be blocked for roughly no longer than dur" ที่เขียนไว้เหมือนกันทั้งใน wait_timeout และ wait_timeout_while
rust-lang/rust 1.97.1 — condvar.rs (ตรวจแล้ว 2026-09-03) — source ของ wait_timeout_while ที่จับ Instant::now() ไว้ก่อนเข้า loop แล้วคิด dur.checked_sub(start.elapsed()) ใหม่ทุกรอบ ซึ่งเป็นหลักฐานว่าเวลารวมของมันถูกจำกัด

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

ข้อ 1 / 3

ผู้อ่านรายหนึ่งหายไปนาน แล้วกลับมาขอ record ต่อจากที่เคยอ่านค้างไว้ broker หาสถานะของมันจากที่ไหน

อ่าน​จาก offset ที่​ขอ แล้ว​จอด​คำขอ​ไว้​แทนที่​จะ​ตอบ​ว่าง

คำขอ fetch พก offset ของ​ตัว​เอง​มา​ทุก​ครั้ง

เพดาน​ของ​การ​อ่าน มี field เดียว​เป็น​คน​แยก

การ​อ่าน​ถือ lock แค่​ตอน​วางแผน ไม่ใช่​ตลอด​ทาง

จอด​คำขอ​ไว้ แทนที่​จะ​ตอบ​ว่าง

เส้น​ตาย​ถูก​ตรึง​ก่อน​เข้า loop แล้ว​นับ​ถอย​หลัง​จาก​มัน

ทาง​ที่​ไม่​เคย​จอด​เลย

คน​ปลุก​คือ​ทาง produce

คำขอ​ที่​หมด​เวลา​แล้ว​ตอบ​ว่าง ไม่ใช่​ความ​ผิด​พลาด

drop(end) เขียน​แยก​บรรทัด เพราะ​กับ​ดัก​นี้​เคย​กัด​มา​แล้ว

คำขอ​ที่​จอด​อยู่ ห้าม​กิน worker ของ pool

สาม​ข้อ​ที่​ที่​จอด​ของ kaen-log ไม่ใช่​ของ Kafka

ของ​ที่​บท​นี้​ไม่​ได้​ทำ