funcSafeSend(ch chan T, value T)(closed bool) { deferfunc() { ifrecover() != nil { // The return result can be altered // in a defer function call. closed = true } }() ch <- value // panic if ch is closed returnfalse// <=> closed = false; return }
funcmain() { rand.Seed(time.Now().UnixNano()) log.SetFlags(0) // ... const MaxRandomNumber = 100000 const NumReceivers = 100 wgReceivers := sync.WaitGroup{} wgReceivers.Add(NumReceivers) // ... dataCh := make(chanint, 100) // the sender gofunc() { for { if value := rand.Intn(MaxRandomNumber); value == 0 { // The only sender can close the channel safely. close(dataCh) return } else { dataCh <- value } } }() // receivers for i := 0; i < NumReceivers; i++ { gofunc() { defer wgReceivers.Done() // Receive values until dataCh is closed and // the value buffer queue of dataCh is empty. for value := range dataCh { log.Println(value) } }() } wgReceivers.Wait() }
funcmain() { rand.Seed(time.Now().UnixNano()) log.SetFlags(0) // ... const MaxRandomNumber = 100000 const NumSenders = 1000 wgReceivers := sync.WaitGroup{} wgReceivers.Add(1) // ... dataCh := make(chanint, 100) stopCh := make(chanstruct{}) // stopCh is an additional signal channel. // Its sender is the receiver of channel dataCh. // Its reveivers are the senders of channel dataCh. // senders for i := 0; i < NumSenders; i++ { gofunc() { for { // The first select here is to try to exit the goroutine // as early as possible. In fact, it is not essential // for this example, so it can be omitted. select { case <- stopCh: return default: } // Even if stopCh is closed, the first branch in the // second select may be still not selected for some // loops if the send to dataCh is also unblocked. // But this is acceptable, so the first select // can be omitted. select { case <- stopCh: return case dataCh <- rand.Intn(MaxRandomNumber): } } }() } // the receiver gofunc() { defer wgReceivers.Done() for value := range dataCh { if value == MaxRandomNumber-1 { // The receiver of the dataCh channel is // also the sender of the stopCh cahnnel. // It is safe to close the stop channel here. close(stopCh) return } log.Println(value) } }() // ... wgReceivers.Wait() }
funcmain() { rand.Seed(time.Now().UnixNano()) log.SetFlags(0) // ... const MaxRandomNumber = 100000 const NumReceivers = 10 const NumSenders = 1000 wgReceivers := sync.WaitGroup{} wgReceivers.Add(NumReceivers) // ... dataCh := make(chanint, 100) stopCh := make(chanstruct{}) // stopCh is an additional signal channel. // Its sender is the moderator goroutine shown below. // Its reveivers are all senders and receivers of dataCh. toStop := make(chanstring, 1) // The channel toStop is used to notify the moderator // to close the additional signal channel (stopCh). // Its senders are any senders and receivers of dataCh. // Its reveiver is the moderator goroutine shown below. var stoppedBy string // moderator gofunc() { stoppedBy = <- toStop close(stopCh) }() // senders for i := 0; i < NumSenders; i++ { gofunc(id string) { for { value := rand.Intn(MaxRandomNumber) if value == 0 { // Here, a trick is used to notify the moderator // to close the additional signal channel. select { case toStop <- "sender#" + id: default: } return } // The first select here is to try to exit the goroutine // as early as possible. This select blocks with one // receive operation case and one default branches will // be optimized as a try-receive operation by the // official Go compiler. select { case <- stopCh: return default: } // Even if stopCh is closed, the first branch in the // second select may be still not selected for some // loops (and for ever in theory) if the send to // dataCh is also unblocked. // This is why the first select block is needed. select { case <- stopCh: return case dataCh <- value: } } }(strconv.Itoa(i)) } // receivers for i := 0; i < NumReceivers; i++ { gofunc(id string) { defer wgReceivers.Done() for { // Same as the sender goroutine, the first select here // is to try to exit the goroutine as early as possible. select { case <- stopCh: return default: } // Even if stopCh is closed, the first branch in the // second select may be still not selected for some // loops (and for ever in theory) if the receive from // dataCh is also unblocked. // This is why the first select block is needed. select { case <- stopCh: return case value := <-dataCh: if value == MaxRandomNumber-1 { // The same trick is used to notify // the moderator to close the // additional signal channel. select { case toStop <- "receiver#" + id: default: } return } log.Println(value) } } }(strconv.Itoa(i)) } // ... wgReceivers.Wait() log.Println("stopped by", stoppedBy) }
if value == 0 { // Here, a trick is used to notify the moderator // to close the additional signal channel. select { case toStop <- "sender#" + id: default: } return }
case value := <-dataCh: if value == MaxRandomNumber-1 { // The same trick is used to notify // the moderator to close the // additional signal channel. select { case toStop <- "receiver#" + id: default: } return }