Golang Concurrency

1. log.Fatal 盡量用在main 或 init

log.Fatal 調用了 os.Exit(), 會無條件終止程式; defer不會被調用到

2. Keep yourself busy or do the work yourself.

啟 goroutine 的人一定要在調用者

3. Never start a goroutine without knowning when it will stop.

你必需知道goroutine 何時被停止

goroutine leak

Bad: goroutine leak 因為 ch 是在 leak function 內. 當 leak執行完後, 裡面的 go func() 還在等 ch 的值, 此 goroutine 被 block.

// leak is a buggy function, 
// It launches a goroutine that block receiving from a channel. 
// Nothing will ever be sent on that channel and the channel is never closed so that goroutine will be blocked forever.
func leak() {
  ch := make(chan int)

  go func() {
    val := <-ch
    fmt.Prinln("We received a value:", val)
  }()
}

超時控制

Bad:

// search simulates a function that finds a record based on a search term.
// It takes 200 ms to perform this work.
func search(term string) (string, error) {
  time.Sleep(200 * time.Millisecond)
  return "some value", nil
}

// process is the work for the program.
// It finds a record then prints it.
func process(term string) error {
  record, err := search(term)
  if err != nil {
    return err
  }

  fmt.Println("Received: ", record)
  return nil
}

Good:

type result struct {
  record string
  err error
}

// process is the work for the program.
// It finds a record then prints it.
func process(term string) error {
  // Create a context that will be canceled in 100 ms
  ctx, cancel := context.WithTimeout(context.Backgroud(), 100*time.Millisecond)
  defer cancel()

  // Make a channel for the goroutine to report its result.
  ch := make(chan result)

  // Launch a goroutine to find the record. Create a result 
  // from the returned values to send through the channel.
  go func() {
    record, err := search(term)
    ch <- result{record, err}
  }

  // Block waiting to either receive from the goroutine's
  // channel or for the context to be canceled.
  select {
    case <-ctx.Done():
      return errors.New("search canceled")
    case result := <-ch
      if result.err != nil{
        return result.err
      }
      fmt.Println("Received: ", result.record)
      return nil
  }
}

超時控制 2

Bad 1: 無法保證創建的 goroutine 生命週期管理, 會導致在服務關閉時, 有些事件丟失

// Tracker knows how to track events for the application
type Tracker struct{}

// Event records an event to a database or stream.
func (t *Tracker) Event(data string) {
  time.Sleep(time.Millisecond) // Simulate network write latency.
  log.Println(data)
}

type App struct {
  track Tracker
}

func (a *App) Handle(w http.ResponseWriter, r * http.Request) {
  // Do some actual work.

  // Respond to the client
  w.WriteHeader(http.StatusCreated)

  // Fire and Hope.
  // BUG: We are not managing this goroutine
  go a.track.Event("this event")
}

Bad 2: 使用 sync.WaitGroup 來追蹤每一個創建的 goroutine 大量創建 goroutine 代價高

// Tracker knows how to track events for the application
type Tracker struct {
  wg sync.WaitGroup
}

// Event stars tracking an event. It runs asynchronously to
// not block the caller. Be sure to call the Shutdown function
// before the program exits so all tracked events finish.
func (t *Tracker) Event(data string) {
  // Increment counter so Shutdown knows to wait for this event.
  t.wg.Add(1)

  // Track event in a goroutine so caller is not blocked.
  go func(){
    // Decrement counter to tell Shutdown this goroutine finished.
    defer t.wg.Done()

    time.Sleep(time.Millisecond) // Simulate network write latency.
    log.Println(data)
  }()
}

func (t *Tracker) Shutdown() {
  t.wg.Wait()
}

func main() {
  // Start a server.
  // Details not shown...
  var a App

  // Shut the server down.
  // Details not shown...

  // Wait for all event goroutines to finish
  a.track.Shutdown()
}

Good 1: // M Sender one Receiver : 由 Receiver 發起一個close到額外的一個 channel // Sender: tr.Event() ; Receiver tr.Run()

package main

import (
    "context"
    "fmt"
    "time"
)

type Tracker struct {
    ch   chan string
    stop chan struct{}
}

// Tracker knows how to track events for the application.
func (t *Tracker) Event(ctx context.Context, data string) error {
    select {
    case t.ch <- data:
        return nil
    case <-ctx.Done():
        return ctx.Err()
    }
}

func (t *Tracker) Run() {
    for data := range t.ch {
        time.Sleep(1 * time.Second)
        fmt.Println(data)
    }
    t.stop <- struct{}{}
}

func (t Tracker) Shutdown(ctx context.Context) {
    close(t.ch)
    select {
    case <-t.stop:
    case <-ctx.Done():
    }
}

func NewTracker() *Tracker {
    return &Tracker{
        ch: make(chan string, 10),
    }
}

func main() {
    tr := NewTracker()
    go tr.Run()
    _ = tr.Event(context.Background(), "test")
    _ = tr.Event(context.Background(), "test")
    _ = tr.Event(context.Background(), "test")
    ctx, cancel := context.WithDeadline(context.Background(), time.Now().Add(5*time.Second))
    defer cancel()
    tr.Shutdown(ctx)
}

// M Sender one Receiver : 由 Receiver 發起一個close到額外的一個 channel
// Sender: tr.Event() ; Receiver tr.Run()

Good 2:

package main

import (
    "fmt"
    "sync"
    "time"
)

type Tracker struct {
    ch   chan string
    stop chan struct{}
}

// Tracker knows how to track events for the application.
func (t *Tracker) Event(data string) error {
    select {
    case <-t.stop:
        fmt.Println("Event: t.stop 1")
        return nil
    default:
    }

    select {
    case t.ch <- data:
        // fmt.Println("send : ", data)
        return nil
    case <-t.stop:
        fmt.Println("Event: t.stop 2")
        return nil
    }
}

// one receiver
func (t *Tracker) Run() {
    for {
        select {
        case data := <-t.ch:
            time.Sleep(1000 * time.Millisecond)
            fmt.Println(data)
            if data == "test:15" {
                close(t.stop)
            }
        case <-t.stop:
            return
        }
    }
}

func NewTracker() *Tracker {
    return &Tracker{
        ch:   make(chan string, 10),
        stop: make(chan struct{}),
    }
}

func main() {
    tr := NewTracker()

    const NumSenders = 50
    const Timeout = 3

    // N senders
    for i := 0; i < NumSenders; i++ {
        go func(idx int) {
            msg := fmt.Sprintf("%s:%d", "test", idx)
            _ = tr.Event(msg)
        }(i)
    }

    go func() {
        // select {
        // case <-time.After(Timeout * time.Second):
        //     close(tr.stop)
        // }
        time.Sleep(Timeout * time.Second)
        close(tr.stop)
    }()

    // One receiver
    wg := sync.WaitGroup{}
    wg.Add(1)
    go func() {
        defer wg.Done()
        tr.Run()
    }()
    wg.Wait()

}

// M Sender one Receiver : 由 Receiver 發起一個close到額外的一個 channel
// Sender: tr.Event() ; Receiver tr.Run()

serve

Good:

func main() {
    done := make(chan error, 2)
    stop := make(chan struct{})

    go func() {
        done <- serveDebug(stop)
    }()
    go func() {
        done <- serveApp(stop)
    }()

    var stopped bool
    for i := 0 ; i < cap(done); i++ {
        if err := <-done; err != nil {
            fmt.Println("error : %v", err)
        }
        if !stopped {
            stopped = true
            close(stop)
        }
    }
}

func serve(addr string, handler http.Handler, stop <-chan strcut{}) error {
    s := http.Server{
        Addr: addr,
        Handler: handler,
    }

    go func() {
        <-stop // wait for stop signal
        s.Shutdown(context.Backgroud())
    }()

    return s.ListenAndServe()
}

https://github.com/da440dil/go-workgroup

4. Leave concurrency to the caller

Bad 1:

func ListDirectory(dir string)([]string, error)

根據目錄的大小, 這可能需要很長時間, 並且可能分配大量的內存來構建目錄列表名稱的slice

Bad 2:

func ListDirectory(dir string) chan string

當中途遇到了錯誤, 調用方無法區分是空目錄或者是讀取錯誤, 這兩種錯誤都會導致 channel關閉. 調用者必須繼續從通道讀取, 直到它關閉. 對於中大型目錄, 它可能在內存使用方面更為高效, 但這種方法並不比原始的slice 方法快.

Good: 參考 filepath: https://pkg.go.dev/path/filepath#WalkDir 如果函數啟動 goroutine , 則必須向調用方提供顯示停止該 goroutine 的方法. 通常將異步執行函數的決定權交給該函數的調用者通常更容易

func WalkDir(root string, fn fs.WalkDirFunc) error {
  info, err := os.Lstat(root)
  if err != nil {
    err = fn(root, nil, err)
  } else {
    err = walkDir(root, &statDirEntry{info}, fn)
  }
  if err == SkipDir {
    return nil
  }
  return err
}