One of the bigger problems in designing something like a pipeline or producer/consumer model is there's somewhat of a black hole when it comes to the state of any given goroutine at any given time.

Consider the following loop, wherein a producer channel creates an arbitrary set of consumer channels and expects each to do one and only one thing:

package main

import (
  "fmt"
  "time"
)

const CONSUMERS = 5

func main() {

  Producer := make(chan (chan int))

  for i := 0; i < CONSUMERS; i++ {
    go func() {
      time.Sleep(1000 * time.Microsecond)
      conChan := make(chan int)

      go func() {
        for {
          select {
          case _,ok := <-conChan:
            if ok  {
              Producer <- conChan
            }else {
              return
            }
          default:
          }
        }
      }()

      conChan <- 1
      close(conChan)
    }()
  }

Given a random amount of consumers to produce, we attach a channel to each and pass a message upstream to the Producer via that consumer's channel. We send just a single message (which we could handle with a buffered channel), but we simply close the channel after.

Whether in a multithreaded application, a distributed application, or a highly concurrent application, an essential attribute of a producer-consumer model is the ability for data to move across a queue/channel in a steady, reliable fashion. This requires some modicum of mutual knowledge to be shared between both the producer and consumers.

Unlike environments that are distributed (or multicore), we do possess some inherent awareness of the status on both ends of that arrangement. We'll next look at a listening loop for producer messages:

  for {
    select {
    case consumer, ok := <-Producer:
      if ok == false {
        fmt.Println("Goroutine closed?")
        close(Producer)
      } else {
        log.Println(consumer)
        // consumer <- 1
      }
      fmt.Println("Got message from secondary channel")
    default:
    }
  }
}

The primary issue is that one of the Producer channel doesn't know much about any given Consumer, including when it's actively running. If we uncommented the // consumer <- 1 line, we'll get a panic, because we're attempting to send a message on a closed channel.

As a message is passed across a secondary goroutine's channel, upstream to the channel of the Producer, we get an appropriate reception, but cannot detect when the downstream goroutine is closed.

Knowing when a goroutine has terminated is in many cases inconsequential, but consider an application that spawns new goroutines when a certain number of tasks are complete, effectively breaking a task into mini tasks. Perhaps each chunk is dependent on the total completion of the last chunk, and a broadcaster must know the status of the current goroutines before moving on.

func main() {

  var channel chan int

    channel <- 1

  for {
    select {
      case <- channel:
        
      default:
    }
  }

}

  var done chan int
  defer close(done)
  defer log.Println("End of script")
  go func() {
    time.Sleep(time.Second * 5)
    done <- 1
  }()

  for {
    select {
      case <- done:
        log.Println("Got transmission")
        return
      default:
    }
  }