Cancelling Goroutines

• Upstream stages close their outbound channels when all the send operation are done.
• Pass a read-only done channel to goroutine.
• Close the channel, to send broadcast signal to all goroutines.
• On receiving the signal on done channel, Goroutines needs to abandon their work and terminate.
• We use select to make send/receive operation on channel pre-emptible

  select{
    case out<-n:
    case <-done:
        return
  }
  

func generator(nums ...int) <-chan int {
    out := make(chan int)

    go func() {
        for _, n := range nums {
            out <- n
        }
        close(out) // <-- Here
    }()
    return out
}

• Downstream stages keep receiving valuse from inbound channel until the channel is closed.

  func square(in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        for n := range in { //<-- Here
            out <- n * n
        }
        close(out)
    }()
    return out
  }
  

• All goroutines exit once all values have been successfully send downstream. ```go func merge(cs ...<-chan int) <-chan int { // Implement fan-in out := make(chan int) var wg sync.WaitGroup

  // merge a list of channels to a single channel output := func(c <-chan int) {

    defer wg.Done()
    for n := range c {
        out <- n
    }
  

  }

  wg.Add(len(cs)) for _, c := range cs {

    go output(c)
  

  }

  go func() { // <-- Here

    wg.Wait()
    close(out)
  

  }()

  return out }

func main() { in := generator(2, 3)

// fan out square stage to run two instances.
c1 := square(in)
c2 := square(in)

//  fan in the results of square stages.
for n := range merge(c1, c2) { // <-- Here
    fmt.Println(n)
}

}

* Real pipeline - Receiver Stages may only need a subset of values to make progress(進展;演進).
* A stage can exit early because an inbound value represents an error in an earlier stage.
* Receiver should not have to wait for the remaining values to arrive.
* we want earlier stages to stop producing values that later stages don't need.

## Goroutine Leak
```go
func main() {
    in := generator(2, 3)

    c1 := square(in)
    c2 := square(in)

    out := merge(c1, c2)
    fmt.Println(<-out)
}

• Main goroutine just receives one value.
• Abandones the inbound channel from merge.
• Merge goroutines will be blocked on channel send operation.
• Square and generator goroutine will also be blocked on send
• This leads to Goroutine Leak

Exercises

https://go.dev/play/p/6E0rblqFsUI

// generator() -> square() ->
//                                                        -> merge -> print
//             -> square() ->
package main

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

func generator(stopCh <-chan struct{}, nums ...int) <-chan int {
    out := make(chan int)

    go func() {
        for _, n := range nums {
            select {
            case out <- n:
            case <-stopCh:
                return
            }
        }
        close(out)
    }()
    return out
}

func square(stopCh <-chan struct{}, in <-chan int) <-chan int {
    out := make(chan int)
    go func() {
        for n := range in {
            select {
            case out <- n * n:
            case <-stopCh:
                return
            }
        }
        close(out)
    }()
    return out
}

func merge(stopCh <-chan struct{}, cs ...<-chan int) <-chan int {
    out := make(chan int)
    var wg sync.WaitGroup

    output := func(c <-chan int) {
        for n := range c {
            select {
            case out <- n:
            case <-stopCh:
                return
            }
        }
        wg.Done()
    }

    wg.Add(len(cs))
    for _, c := range cs {
        go output(c)
    }

    go func() {
        wg.Wait()
        close(out)
    }()
    return out
}

func main() {
    // 一個接收, 多個發送. 由接收端發送一個close 到一個額外的channel(stopCh)
    stopCh := make(chan struct{})
    defer close(stopCh)

    in := generator(stopCh, 2, 3)

    c1 := square(stopCh, in)
    c2 := square(stopCh, in)

    out := merge(stopCh, c1, c2)
    g := runtime.NumGoroutine()
    fmt.Println(<-out)

    time.Sleep(1 * time.Second)
    g = runtime.NumGoroutine()
    fmt.Printf("number of goroutines active = %d \n", g)
}

// guidelines for pipeline construction

// stages close their outbound channels when all the send operations are done.
// stages keep receiving values from inbound channels until those channels are closed or the senders are unblocked.