Worked review 2: worker pool processing files

ProcessAll อ่านไฟล์เป็นชุดด้วย worker goroutine กลุ่มเล็กๆ แล้วคืนขนาดไฟล์ มัน compile ผ่าน, go vet ไม่บ่นอะไรเลย, และมันรันได้ — แต่มันไม่เคยประมวลผลไฟล์ครบทุกตัว บางทีก็รั่ว goroutine และ (ภายใต้ -race) มี data race จริงที่ demo เล็กๆ ด้านล่างนี้บังเอิญไม่ crash มันมี bug จริง 5 ตัว

The program

Open in Playground

package main

import (
  "fmt"
  "os"
  "runtime"
  "time"
)

type Result struct {
  Path string
  Size int
}

func processFile(path string, delay time.Duration) (Result, error) {
  time.Sleep(delay) // simulates variable I/O time per file
  data, err := os.ReadFile(path)
  if err != nil {
    return Result{}, err
  }
  return Result{Path: path, Size: len(data)}, nil
}

func ProcessAll(paths []string, delays map[string]time.Duration, numWorkers int) []Result {
  jobs := make(chan string)
  var results []Result

  for i := 0; i < numWorkers; i++ {
    go func() {
      for p := range jobs {
        r, err := processFile(p, delays[p])
        if err == nil {
          results = append(results, r)
        }
      }
    }()
  }

  for _, p := range paths {
    jobs <- p
  }

  time.Sleep(150 * time.Millisecond) // "wait" for workers to finish
  return results
}

func main() {
  dir, _ := os.MkdirTemp("", "wr2")
  defer func() { _ = os.RemoveAll(dir) }()

  names := []string{"a.txt", "bb.txt", "ccc.txt", "slow.txt"}
  delays := map[string]time.Duration{}
  var paths []string
  for i, name := range names {
    p := dir + "/" + name
    _ = os.WriteFile(p, make([]byte, i+1), 0o644)
    paths = append(paths, p)
    delays[p] = 20 * time.Millisecond
  }
  delays[dir+"/slow.txt"] = 400 * time.Millisecond // slower than ProcessAll's fixed wait
  paths = append(paths, dir+"/missing.txt")

  before := runtime.NumGoroutine()
  results := ProcessAll(paths, delays, 2)
  after := runtime.NumGoroutine()

  fmt.Println("processed", len(results), "of", len(paths), "files")
  for _, r := range results {
    fmt.Printf("  %s: %d bytes\n", r.Path[len(dir)+1:], r.Size)
  }
  fmt.Println("goroutines before:", before, "goroutines still running after:", after)
}

หา bug

missing.txt ไม่มีอยู่จริง และ slow.txt ก็ถูกทำให้ช้าโดยตั้งใจ — แต่ output แสดงว่าไฟล์ที่หายไปมีมากกว่า 2 ตัวจาก 5 ตัว และจำนวน goroutine ก็ไม่กลับไปเท่าจุดเริ่มต้น ลองระบุ bug อย่างน้อย 3 ใน 5 ตัวก่อนเปิดเฉลย

เฉลย 5 bug

ProcessAll “รอ” worker ด้วย time.Sleep(150 * time.Millisecond) แบบตายตัว แทนที่จะใช้ synchronization primitive จริง ไม่มี sync.WaitGroup, ไม่มี done channel — แค่การเดาว่างานควรใช้เวลานานแค่ไหน slow.txt ถูกตั้งใจให้มี delay 400ms ซึ่งนานกว่าที่เดาไว้ 150ms มาก จึงหายไปจาก results เงียบๆ ทุกครั้งที่รัน — ไม่ถูกรายงานเป็น error แค่หายไปเฉยๆ เหมือนไม่เคยมีอยู่
channel jobs ไม่เคยถูกปิด หลังจากส่งทุก path แล้ว ไม่มีอะไรบอก worker goroutine ให้หยุด แต่ละตัวค้างอยู่ใน for p := range jobs ตลอดกาล รอ work ที่จะไม่มีวันมาถึง output จริง: goroutines before: 1 goroutines still running after: 3 — worker goroutine ทั้ง 2 ตัวไม่เคย exit เลย
results เป็น []Result ธรรมดาที่ถูก append จากหลาย goroutine โดยไม่มี mutex นี่คือ data race จริง — ดูเหมือนไม่มีอะไรใน run เล็กๆ นี้ (go run แบบไม่มี -race มักจะรันจบโดยไม่ crash) แต่เป็นจริงภายใต้ go run -race:
```
WARNING: DATA RACE
Write at 0x00c0001ca000 by goroutine 8:
  main.ProcessAll.func1()
Previous read at 0x00c0001ca000 by goroutine 7:
  main.ProcessAll.func1()
```
error จากการอ่าน missing.txt ถูกทิ้งเงียบๆ if err == nil { results = append(...) } แปลว่าความล้มเหลวใดๆ ก็หายไปเฉยๆ — caller บอกไม่ได้ว่า “ไฟล์ fail” ต่างจาก “ไฟล์ไม่เคยถูก schedule” ยังไง
ลำดับของ result ไม่เคยถูกรับประกันเลย แต่ไม่มีตรงไหนรองรับเรื่องนี้ การรันโปรแกรมเดียวกันนี้ 3 ครั้งติดกันได้ 2 of 5, 3 of 5, และ 3 of 5 — subset ต่างกันและลำดับต่างกันทุกครั้ง มาจาก timing ของการ schedule goroutine ล้วนๆ โค้ดที่สมมติว่า results[0] ตรงกับ paths[0] จะผิดเงียบๆ

The corrected program

Open in Playground

package main

import (
  "fmt"
  "os"
  "runtime"
  "sort"
  "sync"
  "time"
)

type Result struct {
  Path string
  Size int
  Err  error
}

func processFile(path string, delay time.Duration) (Result, error) {
  time.Sleep(delay) // simulates variable I/O time per file
  data, err := os.ReadFile(path)
  if err != nil {
    return Result{}, err
  }
  return Result{Path: path, Size: len(data)}, nil
}

func ProcessAll(paths []string, delays map[string]time.Duration, numWorkers int) []Result {
  jobs := make(chan string)
  resultsCh := make(chan Result)
  var wg sync.WaitGroup // fixed: real completion signal instead of a fixed Sleep

  for i := 0; i < numWorkers; i++ {
    wg.Add(1)
    go func() {
      defer wg.Done()
      for p := range jobs {
        r, err := processFile(p, delays[p])
        if err != nil {
          r = Result{Path: p, Err: err} // fixed: report the failure, don't drop it
        }
        resultsCh <- r
      }
    }()
  }

  go func() {
    for _, p := range paths {
      jobs <- p
    }
    close(jobs) // fixed: workers can now see the channel drain and return
  }()
  go func() {
    wg.Wait()
    close(resultsCh)
  }()

  var results []Result
  for r := range resultsCh { // fixed: single collector goroutine — no shared-slice race
    results = append(results, r)
  }
  sort.Slice(results, func(i, j int) bool { return results[i].Path < results[j].Path }) // fixed: deterministic order
  return results
}

func main() {
  dir, _ := os.MkdirTemp("", "wr2fixed")
  defer func() { _ = os.RemoveAll(dir) }()

  names := []string{"a.txt", "bb.txt", "ccc.txt", "slow.txt"}
  delays := map[string]time.Duration{}
  var paths []string
  for i, name := range names {
    p := dir + "/" + name
    _ = os.WriteFile(p, make([]byte, i+1), 0o644)
    paths = append(paths, p)
    delays[p] = 20 * time.Millisecond
  }
  delays[dir+"/slow.txt"] = 400 * time.Millisecond
  paths = append(paths, dir+"/missing.txt")

  before := runtime.NumGoroutine()
  results := ProcessAll(paths, delays, 2)
  after := runtime.NumGoroutine()

  fmt.Println("processed", len(results), "of", len(paths), "files")
  for _, r := range results {
    name := r.Path[len(dir)+1:]
    if r.Err != nil {
      fmt.Printf("  %s: error: %v\n", name, r.Err)
      continue
    }
    fmt.Printf("  %s: %d bytes\n", name, r.Size)
  }
  fmt.Println("goroutines before:", before, "goroutines still running after:", after)
}

รันเวอร์ชันที่แก้แล้ว มันรายงาน processed 5 of 5 files ทุกครั้ง เรียงลำดับเดียวกันทุกครั้ง missing.txt แสดงเป็น error: open .../missing.txt: no such file or directory แทนที่จะหายไปเงียบๆ และ goroutines before: 1 goroutines still running after: 1 — pool ปิดตัวจริงๆ go run -race บนเวอร์ชันนี้ไม่รายงาน race เลย และ golangci-lint run รายงาน 0 issue