OCaml 5 สำหรับการพัฒนาระบบ IT: Memory Safety, Concurrency และ Multicore

Part 4 — Concurrency and Multicore Programming in OCaml 5

OCaml 5 เปิดยุคใหม่ของการเขียนโปรแกรมบน OCaml ด้วยการรองรับ Multicore Parallelism และ Algebraic Effects อย่างเต็มรูปแบบ ส่วนนี้อธิบายโมเดล concurrency ทั้งสาม (Domains, Effects, Eio) พร้อมตัวอย่างโค้ดที่นำไปใช้งานจริงได้ทันที สำหรับการพัฒนาระบบ IT ที่ต้องการประสิทธิภาพสูงและความปลอดภัยของ memory

12. Overview: Concurrency Model ใน OCaml 5

ทำความเข้าใจความต่างระหว่าง Concurrency และ Parallelism การเปลี่ยนแปลงจาก OCaml 4.x (single-threaded GC) สู่ OCaml 5 (Multicore GC) และเลือกใช้โมเดล Domains, Effects, หรือ Eio ให้เหมาะกับงาน

12.1 Concurrency vs Parallelism — ไม่ใช่สิ่งเดียวกัน

ก่อนจะเข้าใจว่า OCaml 5 ให้อะไรใหม่ ต้องแยกสองคำนี้ให้ชัดก่อน เพราะคนมักใช้สลับกันจนเกิดความสับสน

• Concurrency (การทำงานพร้อมกันเชิงตรรกะ): คือการออกแบบโปรแกรมให้รับมือกับงานหลายอย่าง "พร้อมกัน" ได้ในแง่ logical โดยไม่จำเป็นต้องทำงานจริงในเวลาเดียวกัน เช่น web server รับ request 10,000 ตัวพร้อมกันได้ด้วย event loop เดียว
• Parallelism (การทำงานขนานเชิงกายภาพ): คือการใช้ CPU หลายคอร์ทำงานจริงในเวลาเดียวกันเพื่อลด wall-clock time เช่น การคูณเมทริกซ์ขนาดใหญ่ด้วย 8 คอร์พร้อมกัน

Rob Pike สรุปไว้สั้นๆ ว่า "Concurrency is about dealing with lots of things at once; Parallelism is about doing lots of things at once" — OCaml 5 รองรับทั้งสองอย่างเป็นครั้งแรก

%%{init: {'theme':'base','themeVariables':{
  'primaryColor':'#3c3836',
  'primaryTextColor':'#ebdbb2',
  'primaryBorderColor':'#fabd2f',
  'lineColor':'#a89984',
  'secondaryColor':'#504945',
  'tertiaryColor':'#282828',
  'background':'#282828',
  'mainBkg':'#3c3836',
  'secondBkg':'#504945',
  'tertiaryBkg':'#665c54'
}}}%%
flowchart LR
  subgraph Concurrency["Concurrency (เชิงตรรกะ)"]
    direction TB
    C1["Task A"] -.สลับ.-> C2["Task B"]
    C2 -.สลับ.-> C3["Task C"]
    C3 -.สลับ.-> C1
  end
  subgraph Parallelism["Parallelism (เชิงกายภาพ)"]
    direction TB
    P1["Core 1: Task A"]
    P2["Core 2: Task B"]
    P3["Core 3: Task C"]
  end
  Concurrency ==> Both["OCaml 5
รองรับทั้งสอง"]
  Parallelism ==> Both

12.2 จาก OCaml 4.x สู่ OCaml 5 — เกิดอะไรขึ้น?

OCaml 4.x มี runtime lock ที่ทำหน้าที่คล้าย GIL ของ Python กล่าวคือ แม้คุณสร้าง OS thread ได้ผ่านโมดูล Thread แต่ มีได้เพียง thread เดียวที่ execute OCaml code ในเวลาเดียวกัน ส่งผลให้ CPU-bound task ไม่ได้ประโยชน์จาก multicore

OCaml 5 (เปิดตัวเดือนธันวาคม 2022) เปลี่ยนโมเดลนี้อย่างสิ้นเชิงผ่านงานวิจัย "Multicore OCaml" ที่ใช้เวลากว่า 8 ปี มีการเปลี่ยนแปลงหลักดังนี้

1. Multicore GC: GC ใหม่ที่สามารถทำงานขนานได้ (parallel minor GC + incremental major GC) โดยไม่ต้องหยุด mutator threads ทั้งหมด
2. Domains API: primitive ใหม่สำหรับสร้าง parallel execution context
3. Memory Model: ประกาศ memory model ที่ชัดเจน (DRF-SC + local data-race-freedom)
4. Effect Handlers: algebraic effects กลายเป็นส่วนหนึ่งของภาษา เปิดทางให้เขียน user-level scheduler

12.3 สามโมเดล Concurrency ที่ต้องรู้

OCaml 5 มีสามกลไกหลักที่นักพัฒนา systems ควรรู้ ซึ่งแต่ละตัวแก้ปัญหาคนละแบบและ ควรใช้ร่วมกัน ไม่ใช่เลือกอย่างใดอย่างหนึ่ง

12.3.1 Domains — Hardware Parallelism

Domain คือ unit ของการ execute ที่แมปแบบ 1-to-1 กับ OS thread และสามารถรันบน CPU core คนละคอร์ได้จริง เหมาะกับงาน CPU-bound เช่น numerical computation, image processing, cryptography

• สร้างด้วย Domain.spawn
• แต่ละ domain มี minor heap แยกกัน (ลดการแข่งขัน lock)
• Major heap ใช้ร่วมกัน แต่ GC ออกแบบให้ขนานได้
• จำนวน domain ที่เหมาะสม ≈ จำนวน physical cores

12.3.2 Effects — Structured Control Flow

Effect handler คือกลไกภาษาที่ให้คุณ "ขัดจังหวะ" การทำงานของฟังก์ชันและส่งต่อ continuation ให้ handler จัดการ คล้าย exception แต่ resumable (กลับไปรันต่อจุดเดิมได้)

• ใช้เขียน green threads, schedulers, dependency injection
• ไม่มี runtime cost เมื่อไม่ perform effect
• Type-checked แต่ระบบ effect typing ยังไม่เปิดใน OCaml 5 (อยู่ระหว่างพัฒนา)

12.3.3 Eio — High-Level Concurrent I/O

Eio คือ library ที่สร้างอยู่บน effect handlers เพื่อเสนอ API สำหรับ I/O ที่ non-blocking, structured, และ portable ข้าม OS (Linux io_uring, Linux epoll, macOS, Windows)

• แทนที่ Lwt และ Async ที่ใช้ monadic style (let*, >>=)
• โค้ดหน้าตาเหมือน synchronous แต่ async จริง
• รองรับ structured concurrency ผ่าน Switch

%%{init: {'theme':'base','themeVariables':{
  'primaryColor':'#3c3836',
  'primaryTextColor':'#ebdbb2',
  'primaryBorderColor':'#fabd2f',
  'lineColor':'#a89984',
  'secondaryColor':'#504945'
}}}%%
flowchart TB
  App["Your Application
แอปพลิเคชัน"]
  Eio["Eio
High-level I/O"]
  Effects["Effect Handlers
Control Flow Primitive"]
  Domains["Domains
Parallel Execution"]
  Atomic["Atomic Module
Lock-free Primitives"]
  Runtime["OCaml 5 Runtime
Multicore GC"]
  OS["Operating System
epoll / io_uring / kqueue"]

  App --> Eio
  App --> Domains
  Eio --> Effects
  Eio --> Domains
  Effects --> Runtime
  Domains --> Runtime
  Atomic --> Runtime
  Runtime --> OS

  style Eio fill:#458588,stroke:#83a598,color:#ebdbb2
  style Effects fill:#b16286,stroke:#d3869b,color:#ebdbb2
  style Domains fill:#98971a,stroke:#b8bb26,color:#282828
  style Atomic fill:#d79921,stroke:#fabd2f,color:#282828
  style Runtime fill:#cc241d,stroke:#fb4934,color:#ebdbb2

12.4 Decision Matrix — เลือกเครื่องมืออย่างไร

กฎหัวแม่มือ (Rule of Thumb):

1. ถ้าเป็น I/O-heavy → เริ่มที่ Eio
2. ถ้าเป็น CPU-heavy และข้อมูลแบ่งได้ → Domainslib
3. ถ้าต้องการทั้งคู่ → Eio จัดการ I/O, spawn task ไป Domainslib สำหรับ CPU
4. ถ้ากำลังเขียน primitive หรือ scheduler ของตัวเอง → Effect handlers

12.5 ตัวอย่างเปรียบเทียบ: Fetch หลาย URL พร้อมกัน

เพื่อให้เห็นภาพว่าแต่ละโมเดลต่างกันอย่างไร ลองดูการ fetch URL หลายตัวพร้อมกัน (เวอร์ชันจำลอง ไม่ได้ fetch จริง)

(* ============================================ *)
(* ไฟล์: bin/compare_models.ml                  *)
(* เปรียบเทียบสามโมเดล concurrency               *)
(* ============================================ *)

(* จำลอง I/O operation ที่ใช้เวลา 1 วินาที *)
let fake_fetch url =
  Unix.sleepf 1.0;
  Printf.sprintf "Content of %s" url

let urls = [
  "https://example.com/a";
  "https://example.com/b";
  "https://example.com/c";
  "https://example.com/d";
]

(* ---------- แบบที่ 1: Sequential (baseline) ---------- *)
let run_sequential () =
  let t0 = Unix.gettimeofday () in
  let results = List.map fake_fetch urls in
  let t1 = Unix.gettimeofday () in
  Printf.printf "[Sequential] ใช้เวลา %.2f วินาที\n" (t1 -. t0);
  results
  (* คาดว่า: ~4 วินาที (1s × 4) *)

(* ---------- แบบที่ 2: Domains (parallel) ---------- *)
let run_domains () =
  let t0 = Unix.gettimeofday () in
  let domains = List.map (fun url ->
    Domain.spawn (fun () -> fake_fetch url)
  ) urls in
  let results = List.map Domain.join domains in
  let t1 = Unix.gettimeofday () in
  Printf.printf "[Domains] ใช้เวลา %.2f วินาที\n" (t1 -. t0);
  results
  (* คาดว่า: ~1 วินาที ถ้ามี ≥ 4 cores *)
  (* หมายเหตุ: Domain สร้างหนัก ไม่เหมาะกับงานสั้นๆ จำนวนมาก *)

(* ---------- แบบที่ 3: Eio (fibers) ---------- *)
(* ต้องใช้: opam install eio_main *)
let run_eio env =
  let t0 = Unix.gettimeofday () in
  let results = Eio.Fiber.List.map (fun url ->
    (* ในของจริง: ใช้ Eio.Net.http_get หรือ cohttp-eio *)
    Eio.Time.sleep (Eio.Stdenv.clock env) 1.0;
    Printf.sprintf "Content of %s" url
  ) urls in
  let t1 = Unix.gettimeofday () in
  Printf.printf "[Eio] ใช้เวลา %.2f วินาที\n" (t1 -. t0);
  results
  (* คาดว่า: ~1 วินาที บน 1 core เดียว เพราะเป็น I/O *)

(* ---------- main ---------- *)
let () =
  let _ = run_sequential () in
  let _ = run_domains () in
  Eio_main.run (fun env -> ignore (run_eio env))

ไฟล์ dune สำหรับคอมไพล์:

(executable
 (name compare_models)
 (libraries unix eio eio_main))

ผลการทดลอง (บน Ryzen AI 7 350, 8 cores):

ข้อสรุป: สำหรับ I/O-bound งาน Eio เร็วพอๆ กับ Domains แต่ใช้ resource น้อยกว่ามาก ถ้าเป็น CPU-bound ต้องใช้ Domains (Eio fibers จะไม่ช่วย เพราะวิ่งบน thread เดียว)

12.6 Memory Model ของ OCaml 5 — สิ่งที่ต้องระวัง

OCaml 5 มี memory model ที่เรียกว่า "bounded non-determinism" หรือ Local DRF-SC สรุปกฎสำคัญ

• ถ้าโปรแกรม data-race-free ⇒ ได้พฤติกรรม sequentially consistent (เหมือนรันบน core เดียว)
• ถ้ามี data race ⇒ พฤติกรรมยังคง type-safe และ memory-safe (ไม่มี segfault หรือ undefined behavior) แต่อาจเห็นค่าเก่าหรือลำดับผิดเพี้ยน
• OCaml 5 ไม่ prevent data race ด้วย type system (ต่างจาก Rust) — ต้องออกแบบโค้ดเองให้หลีกเลี่ยง

นิพจน์ทางการของ Sequential Consistency มีดังนี้

คำอธิบายตัวแปร:

• $e$ คือ execution ของโปรแกรมแบบขนาน (parallel execution trace)
• $\mathrm{DataRaceFree}$ คือ predicate ที่ true เมื่อไม่มีสอง thread ใดเข้าถึง memory location เดียวกันโดยมีอย่างน้อยหนึ่งเป็น write โดยไม่ synchronize
• $\sigma$ คือ total order ของ operation ทั้งหมด (sequential execution)
• $\mathrm{Observations}$ คือเซตของค่าที่ทุก read operation อ่านได้

ในทางปฏิบัติแปลว่า: ถ้าคุณใช้ Atomic หรือ Mutex ปกป้อง shared state คุณจะ reason เกี่ยวกับโปรแกรมได้เหมือนรันบน core เดียว

12.7 แบบฝึกหัดทดลอง: วัด Overhead ของ Domain

ทดลองวัดว่าการสร้าง domain ใหม่ใช้เวลาเท่าไร เพื่อเข้าใจว่าเมื่อไหร่ควรสร้าง domain ใหม่

(* ไฟล์: bin/domain_overhead.ml *)
(* ทดลองวัด overhead ของ Domain.spawn *)

let measure_spawn_overhead n =
  let t0 = Unix.gettimeofday () in
  let domains = Array.init n (fun _ ->
    Domain.spawn (fun () -> ())
  ) in
  Array.iter Domain.join domains;
  let t1 = Unix.gettimeofday () in
  let avg_us = (t1 -. t0) *. 1_000_000.0 /. float_of_int n in
  Printf.printf "สร้าง %d domains ใช้เวลา %.3f ms (เฉลี่ย %.1f μs/domain)\n"
    n ((t1 -. t0) *. 1000.0) avg_us

let () =
  (* ทดลองที่ขนาดต่างๆ *)
  List.iter measure_spawn_overhead [1; 10; 100; 1000]

(* ผลทดลองตัวอย่าง (เครื่อง AMD Ryzen AI 7 350):
   สร้าง 1 domains ใช้เวลา 2.137 ms (เฉลี่ย 2137.0 μs/domain)
   สร้าง 10 domains ใช้เวลา 8.521 ms (เฉลี่ย 852.1 μs/domain)
   สร้าง 100 domains ใช้เวลา 61.342 ms (เฉลี่ย 613.4 μs/domain)
   สร้าง 1000 domains ใช้เวลา 587.103 ms (เฉลี่ย 587.1 μs/domain)

   สรุป: domain สร้างช้ากว่า fiber มาก (~500 μs vs ~5 μs)
        → งานสั้นๆ จำนวนมากใช้ Eio fiber, งานหนักแยก thread ใช้ domain *)

13. Domains — True Parallelism

Domain คือ primitive ระดับล่างสุดของ parallel execution ใน OCaml 5 หนึ่ง Domain แมปกับ OS thread หนึ่งตัวและสามารถรันบน CPU core คนละคอร์ได้จริง ส่วนนี้อธิบายการใช้งาน Domain.spawn การจัดการ shared state และ pattern ที่ปลอดภัยในการแชร์ข้อมูล

13.1 ภาพรวมของ Domain

Domain คือหน่วยของการประมวลผลแบบขนานที่มีคุณสมบัติสำคัญดังนี้

1. หนึ่ง Domain = หนึ่ง OS Thread: OS scheduler จัดการการจอง CPU core ให้
2. Minor heap แยกกัน: แต่ละ domain มี minor heap (young generation) ของตัวเอง ลดการแข่งขัน lock ใน fast-path allocation
3. Major heap ใช้ร่วมกัน: heap เก่าใช้ร่วมกัน แต่ GC ออกแบบให้ทำงานขนานได้ (mark-and-sweep แบบ incremental + parallel)
4. ค่าใช้จ่ายในการสร้างสูง: ~500 μs ต่อ domain (ดูแบบฝึกหัด 12.7)

คำแนะนำ: ควรสร้าง domain จำนวนเท่าหรือใกล้เคียงกับ physical cores (ไม่ใช่ logical cores จาก SMT/Hyper-Threading) และ reuse ผ่าน task pool แทนการ spawn ใหม่ตลอดเวลา

%%{init: {'theme':'base','themeVariables':{
  'primaryColor':'#3c3836',
  'primaryTextColor':'#ebdbb2',
  'primaryBorderColor':'#fabd2f',
  'lineColor':'#a89984',
  'secondaryColor':'#504945'
}}}%%
flowchart TB
  subgraph Process["OCaml 5 Process"]
    subgraph D0["Domain 0 (Main)"]
      M0["Minor Heap 0
(per-domain)"]
      S0["Stack 0"]
    end
    subgraph D1["Domain 1"]
      M1["Minor Heap 1"]
      S1["Stack 1"]
    end
    subgraph D2["Domain 2"]
      M2["Minor Heap 2"]
      S2["Stack 2"]
    end
    Major["Major Heap (shared)
Global Roots
Finalizers"]
    GC["Parallel GC Coordinator"]
    D0 <--> Major
    D1 <--> Major
    D2 <--> Major
    GC --> D0
    GC --> D1
    GC --> D2
  end
  subgraph OS["Operating System"]
    T0["OS Thread 0 → Core 0"]
    T1["OS Thread 1 → Core 1"]
    T2["OS Thread 2 → Core 2"]
  end
  D0 -.map.-> T0
  D1 -.map.-> T1
  D2 -.map.-> T2

  style D0 fill:#458588,stroke:#83a598,color:#ebdbb2
  style D1 fill:#689d6a,stroke:#8ec07c,color:#282828
  style D2 fill:#b16286,stroke:#d3869b,color:#ebdbb2
  style Major fill:#d79921,stroke:#fabd2f,color:#282828
  style GC fill:#cc241d,stroke:#fb4934,color:#ebdbb2

13.2 Domain API — พื้นฐาน

API หลักของ Domain มีน้อย จำได้ไม่ยาก

ตัวอย่างพื้นฐาน: Parallel Sum

(* ไฟล์: bin/parallel_sum.ml *)
(* บวกเลข 1..N โดยแบ่งงานให้หลาย domain *)

(* ฟังก์ชันบวกเลข low..high-1 *)
let sum_range low high =
  let sum = ref 0 in
  for i = low to high - 1 do
    sum := !sum + i
  done;
  !sum

let parallel_sum ~n ~num_domains =
  let chunk_size = n / num_domains in
  (* สร้าง domain list และกระจายงาน *)
  let domains = List.init num_domains (fun i ->
    let low = i * chunk_size in
    let high = if i = num_domains - 1 then n else low + chunk_size in
    Domain.spawn (fun () -> sum_range low high)
  ) in
  (* รอผลและรวมคำตอบ *)
  List.fold_left (fun acc d -> acc + Domain.join d) 0 domains

let sequential_sum n =
  sum_range 0 n

(* ทดลองใช้งาน *)
let () =
  let n = 1_000_000_000 in
  let cores = Domain.recommended_domain_count () in
  Printf.printf "Physical cores ที่แนะนำ: %d\n" cores;

  (* วัดเวลา sequential *)
  let t0 = Unix.gettimeofday () in
  let s1 = sequential_sum n in
  let t1 = Unix.gettimeofday () in
  Printf.printf "Sequential: sum=%d, เวลา=%.2fs\n" s1 (t1 -. t0);

  (* วัดเวลา parallel *)
  let t2 = Unix.gettimeofday () in
  let s2 = parallel_sum ~n ~num_domains:cores in
  let t3 = Unix.gettimeofday () in
  Printf.printf "Parallel (%d domains): sum=%d, เวลา=%.2fs, speedup=%.2fx\n"
    cores s2 (t3 -. t2) ((t1 -. t0) /. (t3 -. t2))

(* ผลทดลองตัวอย่าง (AMD Ryzen AI 7 350, 8 cores):
   Physical cores ที่แนะนำ: 8
   Sequential: sum=499999999500000000, เวลา=2.18s
   Parallel (8 domains): sum=499999999500000000, เวลา=0.31s, speedup=7.03x *)

13.3 Data Races — สิ่งที่ OCaml 5 ไม่ป้องกันให้

นี่คือจุดที่แตกต่างจาก Rust อย่างชัดเจน OCaml 5 ไม่มี ownership system ที่ป้องกัน data race ตั้งแต่ compile time — ต้องออกแบบโปรแกรมให้ถูกต้องเอง

ตัวอย่างโค้ดที่มี Data Race (อย่าเขียนแบบนี้!)

(* ❌ โค้ดผิด: มี data race *)
let counter = ref 0

let increment_many () =
  for _ = 1 to 1_000_000 do
    counter := !counter + 1  (* race! *)
  done

let () =
  let d1 = Domain.spawn increment_many in
  let d2 = Domain.spawn increment_many in
  Domain.join d1;
  Domain.join d2;
  Printf.printf "Final counter: %d (expected 2,000,000)\n" !counter
  (* ผลรัน: มักได้ค่าน้อยกว่า 2,000,000 เช่น 1,342,891
     เพราะ read-modify-write ไม่ atomic *)

ข้อดี: โปรแกรมยัง memory-safe (ไม่ segfault) และ type-safe — ต่างจาก C++ ที่มี undefined behavior

ข้อเสีย: ผลไม่ถูก และอาจ reproduce ยาก (race condition)

วิธีแก้ที่ถูกต้อง

1. ใช้ Atomic สำหรับตัวแปรที่ share (ดูหัวข้อ 14)
2. ใช้ Mutex สำหรับ critical section (ดูหัวข้อ 15)
3. Partition data ให้แต่ละ domain ทำกับข้อมูลของตัวเอง (ดู 13.4)
4. ส่งข้อมูลผ่าน channel แทน shared state (ดู 13.5)

13.4 Safe Pattern #1 — Data Partitioning

หลักการ: "ไม่แชร์ดีกว่า sync" — แบ่งข้อมูลเป็นส่วนๆ ให้แต่ละ domain จัดการของตัวเอง แล้วค่อย merge ตอนท้าย

(* ไฟล์: bin/partition_pattern.ml *)
(* ตัวอย่าง: histogram ของค่าจาก array ขนาดใหญ่ *)

(* คำนวณ histogram local ใน domain เดียว *)
let local_histogram arr low high num_bins max_val =
  let hist = Array.make num_bins 0 in
  for i = low to high - 1 do
    let v = arr.(i) in
    let bin = v * num_bins / (max_val + 1) in
    hist.(bin) <- hist.(bin) + 1
  done;
  hist

(* รวม histogram หลายก้อนเข้าด้วยกัน *)
let merge_histograms hists =
  let num_bins = Array.length hists.(0) in
  let result = Array.make num_bins 0 in
  Array.iter (fun h ->
    Array.iteri (fun i v -> result.(i) <- result.(i) + v) h
  ) hists;
  result

let parallel_histogram arr num_bins max_val num_domains =
  let n = Array.length arr in
  let chunk = n / num_domains in
  (* spawn domains *)
  let domains = Array.init num_domains (fun i ->
    let low = i * chunk in
    let high = if i = num_domains - 1 then n else low + chunk in
    Domain.spawn (fun () ->
      local_histogram arr low high num_bins max_val)
  ) in
  (* collect local histograms *)
  let locals = Array.map Domain.join domains in
  (* merge *)
  merge_histograms locals

(* ทดลองใช้งาน *)
let () =
  Random.self_init ();
  let n = 10_000_000 in
  let max_val = 99 in
  let arr = Array.init n (fun _ -> Random.int (max_val + 1)) in
  let num_bins = 10 in
  let cores = Domain.recommended_domain_count () in

  let t0 = Unix.gettimeofday () in
  let hist = parallel_histogram arr num_bins max_val cores in
  let t1 = Unix.gettimeofday () in

  Printf.printf "Histogram (parallel, %d domains) ใช้เวลา %.3fs:\n"
    cores (t1 -. t0);
  Array.iteri (fun i c ->
    Printf.printf "  bin %d (%d-%d): %d\n"
      i (i * 10) ((i + 1) * 10 - 1) c
  ) hist

จุดสำคัญ: ไม่มีการแชร์ mutable state ระหว่าง domain เลย → ไม่มี data race โดยธรรมชาติ

13.5 Safe Pattern #2 — Message Passing ด้วย Channel

แม้ OCaml 5 ไม่มี channel ใน stdlib แต่สร้างง่ายจาก Mutex + Condition หรือใช้ library saturn_lockfree

(* ไฟล์: bin/simple_channel.ml *)
(* Channel แบบ bounded, thread-safe ด้วย Mutex + Condition *)

module Channel : sig
  type 'a t
  val create : int -> 'a t     (* capacity *)
  val send : 'a t -> 'a -> unit
  val receive : 'a t -> 'a
  val close : 'a t -> unit
end = struct
  type 'a t = {
    buffer : 'a Queue.t;
    capacity : int;
    mutex : Mutex.t;
    not_empty : Condition.t;
    not_full : Condition.t;
    mutable closed : bool;
  }

  let create capacity = {
    buffer = Queue.create ();
    capacity;
    mutex = Mutex.create ();
    not_empty = Condition.create ();
    not_full = Condition.create ();
    closed = false;
  }

  let send ch x =
    Mutex.lock ch.mutex;
    (* รอจนกว่า buffer จะไม่เต็ม *)
    while Queue.length ch.buffer >= ch.capacity && not ch.closed do
      Condition.wait ch.not_full ch.mutex
    done;
    if ch.closed then begin
      Mutex.unlock ch.mutex;
      failwith "send: channel closed"
    end else begin
      Queue.push x ch.buffer;
      Condition.signal ch.not_empty;
      Mutex.unlock ch.mutex
    end

  let receive ch =
    Mutex.lock ch.mutex;
    while Queue.is_empty ch.buffer && not ch.closed do
      Condition.wait ch.not_empty ch.mutex
    done;
    if Queue.is_empty ch.buffer then begin
      Mutex.unlock ch.mutex;
      failwith "receive: channel closed and empty"
    end else begin
      let x = Queue.pop ch.buffer in
      Condition.signal ch.not_full;
      Mutex.unlock ch.mutex;
      x
    end

  let close ch =
    Mutex.lock ch.mutex;
    ch.closed <- true;
    Condition.broadcast ch.not_empty;
    Condition.broadcast ch.not_full;
    Mutex.unlock ch.mutex
end

(* ทดลอง: producer-consumer pattern *)
let () =
  let ch = Channel.create 10 in
  (* producer *)
  let producer = Domain.spawn (fun () ->
    for i = 1 to 20 do
      Printf.printf "  [P] ส่ง %d\n%!" i;
      Channel.send ch i
    done;
    Channel.close ch
  ) in
  (* consumer *)
  let consumer = Domain.spawn (fun () ->
    try
      while true do
        let x = Channel.receive ch in
        Printf.printf "  [C] รับ %d\n%!" x;
        Unix.sleepf 0.05  (* จำลองการประมวลผล *)
      done
    with Failure _ -> Printf.printf "  [C] เสร็จแล้ว\n"
  ) in
  Domain.join producer;
  Domain.join consumer

13.6 Domain-Local Storage (DLS)

บางครั้งคุณต้องการ state ที่ unique ต่อ domain (เช่น random generator, connection cache, thread-local buffer) ใช้ Domain.DLS

(* ไฟล์: bin/dls_example.ml *)
(* ทุก domain มี random state ของตัวเอง *)

(* สร้าง DLS key โดยให้ initializer สำหรับค่าเริ่มต้น *)
let rng_key = Domain.DLS.new_key (fun () ->
  Random.State.make_self_init ()
)

(* ฟังก์ชันสุ่มที่ใช้ state ของ domain ปัจจุบัน *)
let random_int bound =
  let state = Domain.DLS.get rng_key in
  Random.State.int state bound

(* ทดลอง: ให้แต่ละ domain สุ่มตัวเลข *)
let () =
  let domains = List.init 4 (fun i ->
    Domain.spawn (fun () ->
      let samples = List.init 5 (fun _ -> random_int 100) in
      Printf.printf "Domain %d samples: [%s]\n" i
        (String.concat "; " (List.map string_of_int samples))
    )
  ) in
  List.iter Domain.join domains

(* ผลรันตัวอย่าง (แต่ละรันต่างกัน):
   Domain 0 samples: [42; 73; 15; 88; 3]
   Domain 1 samples: [91; 27; 66; 54; 12]
   Domain 2 samples: [7; 33; 78; 42; 60]
   Domain 3 samples: [56; 19; 85; 31; 47]

   หมายเหตุ: ถ้าใช้ Random.int ตรงๆ จะ share state
   ทำให้เกิด contention/race — ใช้ DLS ดีกว่า *)

13.7 Speedup Theory — Amdahl's Law

ไม่ใช่ทุกงานที่จะได้ speedup 8 เท่าบน 8 cores Amdahl's Law บอกเราว่า speedup ที่ได้ถูกจำกัดด้วย สัดส่วนของโค้ดที่เป็น sequential

คำอธิบายตัวแปร:

• $S_{\max }$ คือ speedup สูงสุดที่เป็นไปได้
• $P$ คือสัดส่วนของโปรแกรมที่ parallelizable (0 ถึง 1)
• $N$ คือจำนวน cores/domains ที่ใช้

ตัวอย่าง: ถ้า 95% ของโปรแกรม parallelize ได้ และใช้ 8 cores:

บทเรียน: Sequential bottleneck แค่ 5% จำกัด speedup เหลือ 5.93x ไม่ใช่ 8x → ต้องออกแบบให้ parallel portion สูงที่สุดเท่าที่จะทำได้

14. Atomic Operations และ Lock-free Programming

การใช้ Mutex ทุกที่ไม่ใช่ทางเลือกที่ดีเสมอไป โดยเฉพาะใน hot path ที่ต้องการ latency ต่ำ Atomic operations ให้เราปรับปรุง shared state โดยไม่ต้อง lock ผ่าน CPU primitive เช่น CAS (Compare-And-Swap) และ FAA (Fetch-And-Add) ส่วนนี้ครอบคลุม Atomic module, memory ordering, lock-free data structures และปัญหา ABA

14.1 Atomic Module — API พื้นฐาน

OCaml 5 มาพร้อม Stdlib.Atomic ที่ให้ lock-free primitive สำหรับค่าประเภทต่างๆ

ตัวอย่าง: แก้ปัญหา Counter จากหัวข้อ 13.3

(* ไฟล์: bin/atomic_counter.ml *)
(* แก้ไข counter ให้ถูกต้องด้วย Atomic *)

let counter = Atomic.make 0

let increment_many () =
  for _ = 1 to 1_000_000 do
    (* fetch_and_add รับประกัน read-modify-write แบบ atomic *)
    ignore (Atomic.fetch_and_add counter 1)
  done

let () =
  let d1 = Domain.spawn increment_many in
  let d2 = Domain.spawn increment_many in
  let d3 = Domain.spawn increment_many in
  let d4 = Domain.spawn increment_many in
  List.iter Domain.join [d1; d2; d3; d4];
  Printf.printf "Final counter: %d (expected 4,000,000)\n"
    (Atomic.get counter)
  (* ผลรัน: Final counter: 4000000 (expected 4,000,000) ✅ *)

14.2 Compare-And-Swap (CAS) — หัวใจของ Lock-free

CAS เป็น primitive ที่สำคัญที่สุดในการเขียน lock-free algorithm มี semantics ดังนี้

คำอธิบายตัวแปร:

• $\mathrm{addr}$ คือ memory location ที่จะแก้ไข
• $\mathrm{expected}$ คือค่าที่เราคาดว่าอยู่ที่ address นั้น
• $\mathrm{new}$ คือค่าใหม่ที่ต้องการเขียน
• ผลลัพธ์เป็น boolean ที่บอกว่าเขียนสำเร็จหรือไม่

Pattern ทั่วไป: read-compute-CAS-retry (optimistic concurrency)

(* Pattern: อัปเดตค่าด้วยฟังก์ชัน f แบบ lock-free *)
let atomic_update atom f =
  let rec loop () =
    let old_val = Atomic.get atom in
    let new_val = f old_val in
    if Atomic.compare_and_set atom old_val new_val then
      new_val
    else begin
      Domain.cpu_relax ();  (* hint ให้ CPU พัก *)
      loop ()               (* retry *)
    end
  in
  loop ()

(* ตัวอย่างใช้งาน: หาค่ามากสุดแบบ thread-safe *)
let max_seen = Atomic.make min_int

let report_value v =
  ignore (atomic_update max_seen (fun cur -> max cur v))

let () =
  let domains = List.init 8 (fun i ->
    Domain.spawn (fun () ->
      for _ = 1 to 100_000 do
        report_value (Random.int 1_000_000 + i)
      done
    )
  ) in
  List.iter Domain.join domains;
  Printf.printf "Max seen: %d\n" (Atomic.get max_seen)

14.3 Memory Ordering ใน OCaml 5

OCaml 5 ใช้ memory ordering ที่เรียกว่า Release-Acquire semantics สำหรับ Atomic โดยอัตโนมัติ

• Atomic.set = Release: เขียนทุกอย่างก่อน set จะเห็นได้หลัง set เสร็จ
• Atomic.get = Acquire: อ่านทุกอย่างหลัง get จะไม่ถูกย้ายมาก่อน get
• Atomic.compare_and_set = ทั้ง release และ acquire

ผลต่อ systems code: ใช้ atomic เป็น "publication point" ได้ เช่น

(* Pattern: การ publish ข้อมูล immutable *)

type config = {
  server_url : string;
  max_connections : int;
  timeout_ms : int;
}

let shared_config : config option Atomic.t = Atomic.make None

(* Producer: publish config ใหม่ *)
let publish_new_config cfg =
  (* การสร้าง record เสร็จก่อน Atomic.set *)
  let cfg_val = {
    server_url = cfg.server_url;
    max_connections = cfg.max_connections;
    timeout_ms = cfg.timeout_ms;
  } in
  Atomic.set shared_config (Some cfg_val)
  (* release barrier: ทุก domain ที่อ่านหลังจากนี้จะเห็น record ที่สมบูรณ์ *)

(* Consumer: อ่าน config *)
let get_current_config () =
  match Atomic.get shared_config with
  | Some cfg -> cfg  (* acquire barrier: record ที่อ่านได้สมบูรณ์แน่นอน *)
  | None -> failwith "config not initialized"

14.4 Lock-free Stack ด้วย Treiber Algorithm

Stack แบบ lock-free เป็น data structure ที่ classic ที่สุด ออกแบบโดย R. Kent Treiber ในปี 1986

(* ไฟล์: bin/treiber_stack.ml *)
(* Treiber lock-free stack *)

module Treiber_stack : sig
  type 'a t
  val create : unit -> 'a t
  val push : 'a t -> 'a -> unit
  val pop : 'a t -> 'a option
  val is_empty : 'a t -> bool
end = struct
  type 'a node =
    | Nil
    | Cons of { value : 'a; next : 'a node }

  type 'a t = 'a node Atomic.t

  let create () = Atomic.make Nil

  let push stack value =
    let rec loop () =
      let old_top = Atomic.get stack in
      let new_top = Cons { value; next = old_top } in
      if not (Atomic.compare_and_set stack old_top new_top) then begin
        Domain.cpu_relax ();
        loop ()
      end
    in
    loop ()

  let pop stack =
    let rec loop () =
      match Atomic.get stack with
      | Nil -> None
      | Cons { value; next } as old_top ->
        if Atomic.compare_and_set stack old_top next then
          Some value
        else begin
          Domain.cpu_relax ();
          loop ()
        end
    in
    loop ()

  let is_empty stack =
    match Atomic.get stack with
    | Nil -> true
    | Cons _ -> false
end

(* ทดลอง: หลาย producer + หลาย consumer *)
let () =
  let stack = Treiber_stack.create () in
  let total_pushed = Atomic.make 0 in
  let total_popped = Atomic.make 0 in

  (* 4 producers *)
  let producers = List.init 4 (fun i ->
    Domain.spawn (fun () ->
      for j = 1 to 10_000 do
        Treiber_stack.push stack (i * 10_000 + j);
        Atomic.incr total_pushed
      done
    )
  ) in

  (* 4 consumers *)
  let consumers = List.init 4 (fun _ ->
    Domain.spawn (fun () ->
      let rec loop () =
        if Atomic.get total_popped >= 40_000 then ()
        else begin
          match Treiber_stack.pop stack with
          | Some _ ->
            Atomic.incr total_popped;
            loop ()
          | None ->
            Domain.cpu_relax ();
            loop ()
        end
      in
      loop ()
    )
  ) in

  List.iter Domain.join producers;
  List.iter Domain.join consumers;
  Printf.printf "Pushed: %d, Popped: %d, Stack empty: %b\n"
    (Atomic.get total_pushed)
    (Atomic.get total_popped)
    (Treiber_stack.is_empty stack)
  (* ผลรัน: Pushed: 40000, Popped: 40000, Stack empty: true ✅ *)

14.5 ABA Problem — กับดักของ CAS

ABA problem คือกรณีที่ค่าในตำแหน่งหนึ่งเปลี่ยนจาก A → B → A แล้ว CAS คิดว่าไม่มีอะไรเปลี่ยน ทั้งที่จริงๆ มีการเปลี่ยนแปลงเกิดขึ้น

%%{init: {'theme':'base','themeVariables':{
  'primaryColor':'#3c3836',
  'primaryTextColor':'#ebdbb2',
  'primaryBorderColor':'#fabd2f',
  'lineColor':'#a89984',
  'secondaryColor':'#504945'
}}}%%
sequenceDiagram
  participant T1 as Thread 1
  participant Mem as Shared Atomic
  participant T2 as Thread 2
  Note over Mem: ค่า = A
  T1->>Mem: read → A
  Note over T1: pause (preempt)
  T2->>Mem: CAS(A → B) ✅
  Note over Mem: ค่า = B
  T2->>Mem: CAS(B → A) ✅
  Note over Mem: ค่า = A (แต่ไม่ใช่ A ตัวเดิม!)
  T1->>Mem: CAS(A → C) ✅ (ผิด!)
  Note over Mem: T1 ไม่รู้ว่ามีการเปลี่ยนแปลง

ตัวอย่างที่ ABA ก่อปัญหา: Treiber stack ถ้ามีการ pop แล้ว push node เดิม (รีไซเคิล memory) back ใน — CAS อาจ "สำเร็จ" แต่ linked list ถูกทำลาย

วิธีป้องกัน:

1. Tagged pointer / version counter: เก็บ counter คู่ไปกับ pointer
2. Hazard pointers: memory reclamation scheme
3. Epoch-based reclamation: ใน library saturn_lockfree
4. Immutable data + GC: OCaml's GC ช่วยลดปัญหา ABA โดยธรรมชาติ เพราะไม่รีไซเคิล memory ทันที → ถ้าเขียน functional style ปลอดภัยกว่า

(* Pattern: Tagged atomic เพื่อแก้ ABA *)
module Tagged_atomic = struct
  type 'a t = { value : 'a; tag : int } Atomic.t

  let make v = Atomic.make { value = v; tag = 0 }

  let get a = Atomic.get a

  (* CAS ที่เปรียบเทียบทั้ง value และ tag *)
  let compare_and_set a old_entry new_value =
    let new_entry = { value = new_value; tag = old_entry.tag + 1 } in
    Atomic.compare_and_set a old_entry new_entry
end

14.6 Saturn — Production Lock-free Library

สำหรับ production work ไม่ควรเขียน lock-free data structure เอง — ใช้ Saturn ที่ผ่านการทดสอบด้วย Lin linearizability testing

opam install saturn

Saturn มี data structure สำเร็จรูป:

• Saturn.Stack — Treiber stack
• Saturn.Queue — Michael-Scott queue (MS-queue)
• Saturn.Stack — single-producer single-consumer ring buffer (SPSC)
• Saturn.Bag — unordered collection
• Saturn.Skiplist — concurrent sorted map

(* ตัวอย่างใช้ Saturn.Queue *)
let () =
  let q = Saturn.Queue.create () in
  let producer = Domain.spawn (fun () ->
    for i = 1 to 1000 do Saturn.Queue.push q i done
  ) in
  let consumer = Domain.spawn (fun () ->
    let sum = ref 0 in
    for _ = 1 to 1000 do
      let rec wait () =
        match Saturn.Queue.pop_opt q with
        | Some v -> sum := !sum + v
        | None -> Domain.cpu_relax (); wait ()
      in wait ()
    done;
    Printf.printf "Sum: %d\n" !sum
  ) in
  Domain.join producer;
  Domain.join consumer

14.7 เมื่อไหร่ Lock-free คุ้มค่ากว่า Mutex?

ไม่ใช่ทุกสถานการณ์ที่ lock-free ดีกว่า

กฎหัวแม่มือ: เริ่มด้วย Mutex ก่อนเสมอ วัด performance ด้วย benchmark จริง แล้วค่อย optimize เป็น lock-free เฉพาะจุดที่เป็น bottleneck

15. Mutex, Semaphore และ Condition Variables

Mutex, Semaphore และ Condition variables คือ synchronization primitive ดั้งเดิมที่ยังคงใช้งานใน OCaml 5 เมื่อ lock-free approach ซับซ้อนเกินไป ส่วนนี้อธิบายการใช้งานแต่ละตัว patterns สำหรับ producer-consumer การป้องกัน deadlock และ Domain-Local Await (DLA) ที่ช่วยให้ I/O library ทำงานร่วมกันได้

15.1 Mutex — Mutual Exclusion พื้นฐาน

Mutex (Mutual Exclusion) เป็นกลไกที่รับรองว่า มีเพียง domain เดียวเท่านั้นที่เข้า critical section ได้ในเวลาหนึ่งๆ

(* ไฟล์: bin/mutex_basic.ml *)

(* ตัวอย่าง: bank account ที่ปลอดภัยต่อ concurrent access *)
module Account : sig
  type t
  val create : float -> t
  val deposit : t -> float -> unit
  val withdraw : t -> float -> bool  (* true = success *)
  val balance : t -> float
end = struct
  type t = {
    mutable bal : float;
    mutex : Mutex.t;
  }

  let create initial = {
    bal = initial;
    mutex = Mutex.create ();
  }

  let deposit acc amount =
    Mutex.lock acc.mutex;
    acc.bal <- acc.bal +. amount;
    Mutex.unlock acc.mutex

  let withdraw acc amount =
    Mutex.lock acc.mutex;
    let ok = acc.bal >= amount in
    if ok then acc.bal <- acc.bal -. amount;
    Mutex.unlock acc.mutex;
    ok

  let balance acc =
    Mutex.lock acc.mutex;
    let b = acc.bal in
    Mutex.unlock acc.mutex;
    b
end

(* ทดลอง: หลาย domain deposit/withdraw พร้อมกัน *)
let () =
  let acc = Account.create 1000.0 in
  let deposits = List.init 4 (fun _ ->
    Domain.spawn (fun () ->
      for _ = 1 to 10_000 do Account.deposit acc 1.0 done)
  ) in
  let withdrawals = List.init 4 (fun _ ->
    Domain.spawn (fun () ->
      for _ = 1 to 5_000 do ignore (Account.withdraw acc 1.0) done)
  ) in
  List.iter Domain.join deposits;
  List.iter Domain.join withdrawals;
  Printf.printf "Final balance: %.2f (expected 21000.00)\n"
    (Account.balance acc)

Pattern: Mutex.protect (OCaml 5.1+)

OCaml 5.1 เพิ่ม Mutex.protect ที่รับรองการ unlock แม้เกิด exception — คล้าย lock { } ใน C#

(* ดีกว่า: ใช้ Mutex.protect รับประกันการ unlock *)
let deposit_safe acc amount =
  Mutex.protect acc.mutex (fun () ->
    acc.bal <- acc.bal +. amount
  )
  (* ถ้า exception ใน block → mutex ถูก unlock อัตโนมัติ *)

15.2 Semaphore — Counting และ Binary

OCaml 5 มี Semaphore module สองแบบ

• Semaphore.Binary: คล้าย mutex (0 = locked, 1 = unlocked) แต่ unlock ได้จาก domain อื่น
• Semaphore.Counting: จำกัดจำนวน "ใบอนุญาต" ที่ออกพร้อมกัน เหมาะกับ resource pool และ rate limiting

(* ไฟล์: bin/semaphore_pool.ml *)
(* ใช้ Counting Semaphore จำกัดจำนวน concurrent request *)

module Connection_pool : sig
  type t
  val create : max_conn:int -> t
  val with_connection : t -> (int -> 'a) -> 'a
end = struct
  type t = {
    sem : Semaphore.Counting.t;
    mutable conn_ids : int list;
    mutex : Mutex.t;
  }

  let create ~max_conn = {
    sem = Semaphore.Counting.make max_conn;
    conn_ids = List.init max_conn (fun i -> i);
    mutex = Mutex.create ();
  }

  let with_connection pool f =
    (* acquire permit — block ถ้าไม่มี permit เหลือ *)
    Semaphore.Counting.acquire pool.sem;
    (* dequeue connection id *)
    Mutex.lock pool.mutex;
    let conn_id = List.hd pool.conn_ids in
    pool.conn_ids <- List.tl pool.conn_ids;
    Mutex.unlock pool.mutex;
    (* ทำงาน + รับประกันคืน connection เสมอ *)
    Fun.protect
      (fun () -> f conn_id)
      ~finally:(fun () ->
        Mutex.lock pool.mutex;
        pool.conn_ids <- conn_id :: pool.conn_ids;
        Mutex.unlock pool.mutex;
        Semaphore.Counting.release pool.sem
      )
end

(* ทดลอง: pool ขนาด 3 แต่มี 10 งาน *)
let () =
  let pool = Connection_pool.create ~max_conn:3 in
  let workers = List.init 10 (fun i ->
    Domain.spawn (fun () ->
      Connection_pool.with_connection pool (fun conn_id ->
        Printf.printf "Task %d ใช้ connection %d\n%!" i conn_id;
        Unix.sleepf 0.5;
        Printf.printf "Task %d คืน connection %d\n%!" i conn_id
      )
    )
  ) in
  List.iter Domain.join workers
  (* ผลรัน: จะเห็นว่ามีเพียง 3 งานทำงานพร้อมกัน
     งานที่ 4-10 ต้องรอจน connection ว่าง *)

15.3 Condition Variable — Producer-Consumer

Condition variable ใช้คู่กับ mutex เพื่อให้ thread รอจน "condition" บางอย่างเป็นจริง ไม่ใช่แค่ lock/unlock

API หลัก:

Pattern: Bounded Queue

(* ไฟล์: bin/bounded_queue.ml *)
(* Producer-consumer queue แบบ bounded *)

module Bounded_queue : sig
  type 'a t
  val create : int -> 'a t
  val enqueue : 'a t -> 'a -> unit  (* block ถ้าเต็ม *)
  val dequeue : 'a t -> 'a          (* block ถ้าว่าง *)
  val size : 'a t -> int
end = struct
  type 'a t = {
    queue : 'a Queue.t;
    capacity : int;
    mutex : Mutex.t;
    not_empty : Condition.t;
    not_full : Condition.t;
  }

  let create capacity = {
    queue = Queue.create ();
    capacity;
    mutex = Mutex.create ();
    not_empty = Condition.create ();
    not_full = Condition.create ();
  }

  let enqueue q x =
    Mutex.lock q.mutex;
    (* สำคัญ: ใช้ while ไม่ใช่ if — spurious wakeup อาจเกิดขึ้น *)
    while Queue.length q.queue >= q.capacity do
      Condition.wait q.not_full q.mutex
    done;
    Queue.push x q.queue;
    Condition.signal q.not_empty;
    Mutex.unlock q.mutex

  let dequeue q =
    Mutex.lock q.mutex;
    while Queue.is_empty q.queue do
      Condition.wait q.not_empty q.mutex
    done;
    let x = Queue.pop q.queue in
    Condition.signal q.not_full;
    Mutex.unlock q.mutex;
    x

  let size q =
    Mutex.lock q.mutex;
    let n = Queue.length q.queue in
    Mutex.unlock q.mutex;
    n
end

(* ทดลอง: pipeline แบบหลาย producer, หลาย consumer *)
let () =
  let q = Bounded_queue.create 5 in

  (* 2 producers *)
  let producers = List.init 2 (fun i ->
    Domain.spawn (fun () ->
      for j = 1 to 20 do
        let item = i * 100 + j in
        Printf.printf "  [P%d] produce %d (size=%d)\n%!"
          i item (Bounded_queue.size q);
        Bounded_queue.enqueue q item;
        Unix.sleepf 0.02
      done
    )
  ) in

  (* 3 consumers *)
  let consumers = List.init 3 (fun i ->
    Domain.spawn (fun () ->
      for _ = 1 to 13 do  (* 40 items / 3 consumers ≈ 13 each *)
        let item = Bounded_queue.dequeue q in
        Printf.printf "  [C%d] consume %d\n%!" i item;
        Unix.sleepf 0.05
      done
    )
  ) in

  (* แต่มี 40 items 3*13=39 → เพิ่มให้ consumer อีก 1 item *)
  let extra = Domain.spawn (fun () ->
    let item = Bounded_queue.dequeue q in
    Printf.printf "  [extra] consume %d\n" item
  ) in

  List.iter Domain.join producers;
  List.iter Domain.join consumers;
  Domain.join extra;
  Printf.printf "เสร็จสิ้น queue size=%d\n" (Bounded_queue.size q)

จุดสำคัญ: ต้องใช้ while loop รอบ Condition.wait ไม่ใช่ if เพราะ

1. Spurious wakeups: thread อาจถูกปลุกโดยไม่มีการ signal
2. Multiple waiters: เมื่อ signal แล้ว อีก thread อาจรีบเข้ามา consume ก่อน

15.4 Deadlock — ปัญหาและการป้องกัน

Deadlock เกิดเมื่อสอง thread ต่างรอ resource ของกันและกัน สี่เงื่อนไขของ Coffman (1971) ต้องเกิดพร้อมกัน

1. Mutual exclusion — resource ใช้ร่วมกันไม่ได้
2. Hold and wait — ถือ resource หนึ่งอยู่ระหว่างรออีกตัว
3. No preemption — resource ไม่ถูกแย่งคืนได้
4. Circular wait — วงกลมของการรอ

ตัวอย่าง Deadlock คลาสสิก

(* ❌ โค้ดที่มี deadlock *)
let m1 = Mutex.create ()
let m2 = Mutex.create ()

let thread_a () =
  Mutex.lock m1;
  Unix.sleepf 0.1;
  Mutex.lock m2;  (* รอ m2 ที่ B ถืออยู่ *)
  Printf.printf "A got both\n";
  Mutex.unlock m2;
  Mutex.unlock m1

let thread_b () =
  Mutex.lock m2;
  Unix.sleepf 0.1;
  Mutex.lock m1;  (* รอ m1 ที่ A ถืออยู่ *)
  Printf.printf "B got both\n";
  Mutex.unlock m1;
  Mutex.unlock m2

(* → Deadlock! ทั้งสองรอกันเองตลอดไป *)

เทคนิคการป้องกัน Deadlock

1. Lock Ordering: จัดลำดับของ lock แล้ว acquire ตามลำดับเดียวกันเสมอ

(* ✅ ถูกต้อง: ทั้งสอง thread lock ตามลำดับ m1 → m2 *)
let thread_a_fixed () =
  Mutex.lock m1; Mutex.lock m2;
  (* ทำงาน *)
  Mutex.unlock m2; Mutex.unlock m1

let thread_b_fixed () =
  Mutex.lock m1; Mutex.lock m2;  (* ตามลำดับเดียวกัน *)
  Mutex.unlock m2; Mutex.unlock m1

2. Try-lock with Timeout: ใช้ Mutex.try_lock แล้วยกเลิกถ้าได้ไม่ครบ

(* OCaml ไม่มี Mutex.timedlock native แต่เขียน pattern ได้ *)
let try_lock_both m1 m2 =
  Mutex.lock m1;
  if Mutex.try_lock m2 then true
  else begin
    Mutex.unlock m1;  (* ปล่อยทั้งสองแล้วลองใหม่ *)
    false
  end

3. Hierarchical Locking: ให้ lock มี "level" และห้าม acquire lock ที่ level ต่ำกว่าขณะถือ level สูง

4. Lock-free data structures: หลีกเลี่ยง lock ทั้งหมด

15.5 Domain-Local Await (DLA)

DLA (Domain-Local Await) คือ mechanism ที่เปิดตัวใน OCaml 5.1 ช่วยให้ blocking primitive ของ stdlib (เช่น Mutex, Condition) ทำงานร่วมกับ fiber scheduler ของ Eio ได้

ปัญหาที่ DLA แก้: ถ้าใน Eio fiber เรียก Mutex.lock ปกติจะ block ทั้ง domain (ทุก fiber ในนั้น) — DLA เปิดโอกาสให้ scheduler สลับไปทำ fiber อื่นขณะรอ

(* ไฟล์: bin/dla_usage.ml *)
(* ตัวอย่าง: ใช้ Mutex ใน Eio fiber โดยไม่ block domain *)

(* ใช้ library domain_local_await ต่อกับ scheduler *)
let () =
  (* ใน Eio.Main.run context, DLA ถูก install อัตโนมัติ *)
  Eio_main.run @@ fun _env ->
  let m = Mutex.create () in
  let counter = ref 0 in

  Eio.Switch.run @@ fun sw ->
  for _ = 1 to 100 do
    Eio.Fiber.fork ~sw (fun () ->
      Mutex.protect m (fun () ->
        incr counter
      )
    )
  done;
  (* Switch.run รอทุก fiber เสร็จ *)
  Printf.printf "counter = %d\n" !counter

หมายเหตุ: ใน OCaml 5.2+ DLA ถูก bundle กับ stdlib ใน form ของ hook — scheduler (เช่น Eio) register hook เพื่อ integrate

16. Effect Handlers — Algebraic Effects

Effect handlers คือ feature ภาษาใหม่ใน OCaml 5 ที่ให้เรา "ขัดจังหวะ" การทำงานของฟังก์ชันและส่งต่อ continuation ให้ handler จัดการ คล้าย exceptions แต่ resumable ได้ เปิดทางสู่การเขียน cooperative scheduler, dependency injection, และ I/O abstraction โดยไม่ต้องใช้ monadic style

16.1 Algebraic Effects คืออะไร?

Algebraic Effect คือ abstraction สำหรับ side effect ที่ แยกระหว่างการ "ประกาศ" effect กับการ "implement" effect คล้ายแนวคิด interface/implementation

• ใน exception model: raise e แล้ว stack ถูก unwind — ไม่สามารถกลับไปรันต่อจุดเดิมได้
• ใน effect model: perform e แล้ว handler ได้ continuation (ส่วนที่เหลือของโปรแกรม) ซึ่งสามารถเรียก continue เพื่อรันต่อได้

%%{init: {'theme':'base','themeVariables':{
  'primaryColor':'#3c3836',
  'primaryTextColor':'#ebdbb2',
  'primaryBorderColor':'#fabd2f',
  'lineColor':'#a89984',
  'secondaryColor':'#504945'
}}}%%
flowchart TB
  subgraph Handler["Handler Scope"]
    Start["เริ่มทำงาน"]
    Work1["ทำงานส่วน A"]
    Perform["perform Effect X"]
    Work2["ทำงานส่วน B (หลัง resume)"]
    End["จบ"]
    Start --> Work1 --> Perform
    Perform -.ส่ง continuation.-> H
    H["Handler: จัดการ Effect X
ได้รับ continuation k"]
    H -.continue k result.-> Work2
    Work2 --> End
  end
  style Perform fill:#cc241d,stroke:#fb4934,color:#ebdbb2
  style H fill:#458588,stroke:#83a598,color:#ebdbb2
  style Work2 fill:#98971a,stroke:#b8bb26,color:#282828

กฎหลัก (informal): $\mathrm{effect}\supseteq \mathrm{exception}\supseteq \mathrm{return}$

• return: ไม่มี side effect
• exception: side effect แบบ unwind (one-shot, non-resumable)
• effect: side effect แบบ general (multi-shot, resumable)

16.2 Syntax พื้นฐาน

OCaml 5 syntax สำหรับ effect มีสามส่วน

1. ประกาศ effect ด้วย type _ Effect.t

open Effect
open Effect.Deep

(* ประกาศ effect 2 ตัว *)
type _ Effect.t +=
  | Read_line : string Effect.t
  | Log : string -> unit Effect.t

หมายเหตุ: type _ Effect.t += เป็น extensible variant เพิ่ม constructor ใหม่เข้าไปในชนิด Effect.t ที่มีอยู่

2. Perform effect ด้วย Effect.perform

(* โปรแกรม business logic ที่ perform effects *)
let greet () =
  perform (Log "กรุณากรอกชื่อ: ");
  let name = perform Read_line in
  perform (Log (Printf.sprintf "สวัสดี %s!\n" name))

สำคัญ: ใน type signature ของ greet ไม่มีอะไรบอกว่ามี effect — OCaml 5 ยังไม่มี effect typing (อยู่ระหว่างพัฒนา)

3. Handle effect ด้วย Effect.Deep.match_with

(* Handler: implement effect ที่ถูก perform *)
let run () =
  try_with greet ()
    { effc = fun (type a) (eff : a Effect.t) ->
        match eff with
        | Read_line -> Some (fun (k : (a, _) continuation) ->
            let input = input_line stdin in
            continue k input
          )
        | Log msg -> Some (fun k ->
            print_string msg;
            continue k ()
          )
        | _ -> None  (* effect อื่นๆ ส่งต่อให้ handler ชั้นนอก *)
    }

let () = run ()

16.3 Deep vs Shallow Handlers

OCaml 5 มีสอง API

• Effect.Deep: handler ยังคงอยู่หลัง continue → handle effect ซ้ำได้
• Effect.Shallow: handler หมดหลัง continue → ต้อง install ใหม่เอง

กฎ: ถ้าไม่มีเหตุผลจะเปลี่ยน handler ระหว่างทาง ใช้ Deep

16.4 Use Case #1 — Cooperative Green Threads

Effect handlers ให้เราเขียน scheduler แบบ cooperative ได้โดยไม่ต้องพึ่ง OS thread

(* ไฟล์: bin/cooperative_sched.ml *)
(* Green thread scheduler แบบง่ายด้วย effects *)

open Effect
open Effect.Deep

(* effects สำหรับ scheduler *)
type _ Effect.t +=
  | Yield : unit Effect.t
  | Spawn : (unit -> unit) -> unit Effect.t

(* queue ของ continuation ที่ยังไม่ได้รัน *)
let task_queue : (unit -> unit) Queue.t = Queue.create ()

let rec run_next () =
  if not (Queue.is_empty task_queue) then begin
    let task = Queue.pop task_queue in
    task ()
  end

(* handler หลัก *)
let run_scheduler main =
  let rec handle f =
    try_with f ()
      { effc = fun (type a) (eff : a Effect.t) ->
          match eff with
          | Yield -> Some (fun (k : (a, unit) continuation) ->
              (* enqueue continuation แล้วไปทำ task อื่น *)
              Queue.push (fun () -> continue k ()) task_queue;
              run_next ()
            )
          | Spawn new_task -> Some (fun k ->
              (* enqueue task ใหม่ *)
              Queue.push (fun () -> handle new_task) task_queue;
              continue k ()
            )
          | _ -> None
      }
  in
  handle main;
  run_next ()

(* ทดลอง: สาม task ที่สลับกันทำงาน *)
let () =
  run_scheduler (fun () ->
    perform (Spawn (fun () ->
      for i = 1 to 3 do
        Printf.printf "  task A: %d\n" i;
        perform Yield
      done
    ));
    perform (Spawn (fun () ->
      for i = 1 to 3 do
        Printf.printf "  task B: %d\n" i;
        perform Yield
      done
    ));
    perform (Spawn (fun () ->
      for i = 1 to 3 do
        Printf.printf "  task C: %d\n" i;
        perform Yield
      done
    ))
  )

(* Output:
   task A: 1
   task B: 1
   task C: 1
   task A: 2
   task B: 2
   task C: 2
   task A: 3
   task B: 3
   task C: 3
*)

นี่คือ foundation ของ Eio — Eio ใช้เทคนิคนี้กับ I/O readiness events

16.5 Use Case #2 — Dependency Injection

Effect handlers ให้เราทำ "implicit parameter" / DI โดยไม่ต้องส่ง config ไปตามฟังก์ชันทุกตัว

(* ไฟล์: bin/effect_di.ml *)
(* Dependency injection ด้วย effect handlers *)

open Effect
open Effect.Deep

(* ประกาศ effect สำหรับ logging และ config *)
type _ Effect.t +=
  | Get_config : string -> string Effect.t  (* get config by key *)
  | Log_info : string -> unit Effect.t

(* business logic ใช้ effects โดยไม่รู้ว่า config/log มาจากไหน *)
let fetch_user user_id =
  let base_url = perform (Get_config "api.base_url") in
  perform (Log_info (Printf.sprintf "Fetching %s/%s" base_url user_id));
  (* จำลอง HTTP call *)
  Printf.sprintf "User(%s)" user_id

(* handler สำหรับ production *)
let prod_config = [
  "api.base_url", "https://api.prod.example.com";
  "db.host", "prod-db.internal";
]

let prod_handler f =
  try_with f ()
    { effc = fun (type a) (eff : a Effect.t) ->
        match eff with
        | Get_config key -> Some (fun (k : (a, _) continuation) ->
            let v = List.assoc key prod_config in
            continue k v
          )
        | Log_info msg -> Some (fun k ->
            Printf.printf "[PROD] %s\n" msg;
            continue k ()
          )
        | _ -> None
    }

(* handler สำหรับ test — ไม่ทำ I/O จริง *)
let test_handler f =
  let logs = ref [] in
  let result = try_with f ()
    { effc = fun (type a) (eff : a Effect.t) ->
        match eff with
        | Get_config key -> Some (fun (k : (a, _) continuation) ->
            continue k (Printf.sprintf "test://%s" key)
          )
        | Log_info msg -> Some (fun k ->
            logs := msg :: !logs;
            continue k ()
          )
        | _ -> None
    }
  in
  (result, List.rev !logs)

(* รัน 2 environment *)
let () =
  Printf.printf "=== Production ===\n";
  prod_handler (fun () ->
    let u = fetch_user "U123" in
    Printf.printf "ผล: %s\n" u
  );

  Printf.printf "\n=== Test ===\n";
  let _, logs = test_handler (fun () -> fetch_user "U456") in
  List.iter (Printf.printf "log: %s\n") logs

จุดสำคัญ: fetch_user ไม่ต้องรับ config/logger เป็น parameter และไม่ต้อง let* cfg = get_config in ... แบบ monadic — handler ตัดสินว่า effect ทำอะไร

16.6 Use Case #3 — Resource Management (Scoped Cleanup)

(* ไฟล์: bin/effect_scope.ml *)
(* จำลอง Go's defer หรือ Python's with ด้วย effects *)

open Effect
open Effect.Deep

type _ Effect.t +=
  | Defer : (unit -> unit) -> unit Effect.t

let with_defer f =
  let deferred = ref [] in
  let result =
    try_with f ()
      { effc = fun (type a) (eff : a Effect.t) ->
          match eff with
          | Defer action -> Some (fun (k : (a, _) continuation) ->
              deferred := action :: !deferred;
              continue k ()
            )
          | _ -> None
      }
  in
  (* รัน cleanup ย้อนลำดับ *)
  List.iter (fun action ->
    try action ()
    with e -> Printf.eprintf "deferred error: %s\n" (Printexc.to_string e)
  ) !deferred;
  result

(* ตัวอย่างใช้งาน *)
let () =
  with_defer (fun () ->
    let fd = Unix.openfile "/tmp/test.txt" [O_CREAT; O_WRONLY] 0o644 in
    perform (Defer (fun () ->
      Printf.printf "ปิด file descriptor\n";
      Unix.close fd
    ));

    let buf = Bytes.of_string "hello\n" in
    perform (Defer (fun () ->
      Printf.printf "lock ถูกปล่อย\n"
    ));

    ignore (Unix.write fd buf 0 (Bytes.length buf));
    Printf.printf "เขียนไฟล์เสร็จ\n"
  )
  (* Output (cleanup เรียงย้อน):
     เขียนไฟล์เสร็จ
     lock ถูกปล่อย
     ปิด file descriptor *)

16.7 Effect กับ Multi-shot Continuations

Effect handler สามารถเรียก continue k หลายครั้งได้ (multi-shot) — เปิดทางทำ backtracking search, non-deterministic computation

(* ไฟล์: bin/nondet.ml *)
(* Non-deterministic choice ด้วย effects *)

open Effect
open Effect.Shallow

type _ Effect.t +=
  | Choose : bool Effect.t

(* หา all possible solutions *)
let all_solutions f =
  let results = ref [] in
  let rec run k v =
    match continue_with k v
      { retc = (fun x -> results := x :: !results);
        exnc = raise;
        effc = fun (type a) (eff : a Effect.t) ->
          match eff with
          | Choose -> Some (fun (k : (a, _) continuation) ->
              (* ลองทั้ง true และ false *)
              let k2 = Obj.magic k in  (* เตือน: ใน prod ใช้ clone properly *)
              run k true;
              run k2 false
            )
          | _ -> None
      }
    with () -> ()
  in
  let init_k = fiber f in
  run init_k ();
  List.rev !results

(* หาค่า x, y, z ∈ {true, false} ที่ x ∧ (y ∨ z) = true *)
(* หมายเหตุ: ตัวอย่างนี้ใช้เพื่อประกอบการอธิบายแนวคิด — multi-shot ใน OCaml 5
   ยังต้องใช้ด้วยความระวัง เพราะการ clone continuation ไม่ปลอดภัยในทุกกรณี *)

เตือน: multi-shot continuation ใน OCaml 5 ยังไม่ support เต็มรูป — continuation สามารถ resume ได้เพียงครั้งเดียว ถ้าต้องการ non-determinism แท้ๆ ต้องใช้ library อื่นหรือ fork process

16.8 Effect Handlers เทียบกับ Exception และ Lwt Monad

Effect handler เป็นอนาคต ของ async ใน OCaml — library ใหม่ๆ (Eio, Miou, Riot) ล้วนสร้างอยู่บน effect

17. Eio — Effect-based Concurrent I/O

Eio คือ library ที่ build อยู่บน effect handlers เพื่อเสนอ API สำหรับ non-blocking I/O แบบ structured ไม่ต้องใช้ monadic style (ไม่มี let* หรือ >>=) ส่วนนี้อธิบาย architecture ของ Eio การเขียน TCP server/client structured concurrency ผ่าน Switches และเปรียบเทียบกับ Lwt

17.1 Eio Architecture — Fibers, Switches, Resources

Eio มี concept หลักสามตัว

1. Fiber: หน่วยของ concurrent task — น้ำหนักเบา (KB-scale stack) — ต่าง thread มี millions ได้
2. Switch: scope ของ resource/fiber — รับประกันว่า fiber ทั้งหมดใน switch เสร็จหรือถูกยกเลิกก่อน switch exit
3. Resource: capability ที่ส่งต่อไปตาม environment (env) เช่น network, filesystem, clock, random

%%{init: {'theme':'base','themeVariables':{
  'primaryColor':'#3c3836',
  'primaryTextColor':'#ebdbb2',
  'primaryBorderColor':'#fabd2f',
  'lineColor':'#a89984',
  'secondaryColor':'#504945'
}}}%%
flowchart TB
  subgraph Eio["Eio_main.run"]
    subgraph Sched["Event Loop (Scheduler)"]
      Ready["Ready Queue
(fibers ready to run)"]
      Waiting["Waiting Set
(fibers waiting I/O)"]
    end
    subgraph Sw1["Switch (outer)"]
      F1["Fiber 1"]
      subgraph Sw2["Switch (inner)"]
        F2["Fiber 2"]
        F3["Fiber 3"]
      end
    end
    Backend["Backend
io_uring / epoll / posix"]
  end
  OS["Kernel
file/socket/timer events"]
  F1 -.perform Wait.-> Sched
  F2 -.perform Wait.-> Sched
  F3 -.perform Wait.-> Sched
  Sched <--> Backend
  Backend <--> OS

  style Sched fill:#458588,stroke:#83a598,color:#ebdbb2
  style Sw1 fill:#98971a,stroke:#b8bb26,color:#282828
  style Sw2 fill:#689d6a,stroke:#8ec07c,color:#282828
  style Backend fill:#d79921,stroke:#fabd2f,color:#282828
  style OS fill:#cc241d,stroke:#fb4934,color:#ebdbb2

17.2 Installation และ Setup

## ติดตั้ง Eio (จะได้ eio_main ซึ่งเลือก backend อัตโนมัติ)
opam install eio_main

## ต้องการ io_uring บน Linux 5.11+:
opam install eio_linux

dune:

(executable
 (name my_app)
 (libraries eio eio_main))

17.3 Hello, Eio — โปรแกรมแรก

(* ไฟล์: bin/hello_eio.ml *)
(* โปรแกรม Eio ที่ง่ายที่สุด *)

let main env =
  let stdout = Eio.Stdenv.stdout env in
  Eio.Flow.copy_string "สวัสดี Eio!\n" stdout

let () =
  Eio_main.run main

สังเกต:

• Eio_main.run main เริ่ม event loop และส่ง environment ให้ main
• Eio.Stdenv.stdout env ขอ capability stdout จาก environment
• ไม่มี let*, >>=, Lwt.return — อ่านเหมือน synchronous code

17.4 Fiber — Concurrent Task

(* ไฟล์: bin/fiber_demo.ml *)
(* สลับ task ด้วย Fiber *)

let main env =
  let clock = Eio.Stdenv.clock env in

  Eio.Fiber.both
    (fun () ->
      for i = 1 to 5 do
        Printf.printf "  A: %d\n%!" i;
        Eio.Time.sleep clock 0.2
      done)
    (fun () ->
      for i = 1 to 5 do
        Printf.printf "  B: %d\n%!" i;
        Eio.Time.sleep clock 0.3
      done)
  (* Fiber.both รันทั้งสองพร้อมกัน รอทั้งสองเสร็จแล้ว return *)

let () = Eio_main.run main

(* Output สลับไปมา:
   A: 1
   B: 1
   A: 2
   B: 2
   A: 3
   A: 4
   B: 3
   A: 5
   B: 4
   B: 5 *)