(see also: https://godoc.org/github.com/creachadair/msync#Collector)
Many Go APIs allow multiple concurrent writers to a single channel, for example, to enqueue work for a task-processing API. When we do this, there is a tricky question of how to safely close the channel. Because this is tricky, many APIs do not bother, which complicates cleanup for the reader side of the channel. Dave Cheney has a good survey of channel corners that I recommend to anyone who is working in this space.
The basic problem is that writing to a closed channel triggers a panic. If some writer can detect that it is the "last", it can close the channel itself -- however that is often not practical unless all the writers are fully under the control of the API itself.
I have found a useful tactic for dealing with this situation.
As a first step, we can try to signal to the writers when they should not attempt any further writes. A typical way to do this is to close a sentinel channel, and then check for it in the writer:
// Setup: Sentinel channel, closed when no more writes are desired.
done := make(chan struct{})
// Writer:
func writeStuff(msg message) error {
select {
case <-done:
return writerClosed
case ch <- msg:
return nil
}
}This alone is not sufficient, though: To understand why, consider how we shut down. At some point, we will close the sentinel channel, and the idea is that once that's done, there can be no more writers. We might say:
// Closer:
close(done)
close(ch)This leaves open a data race: After calling close(done) to signal the writers, one or more writers may have already passed the <-done check before we get around to closing the write channel ch. This leads to a panic from writing on a closed channel (or, if we are lucky, the race detector finds it in our tests).
So then, we need some way to wait for active writers to "drain" after closing the sentinel. No writers who enter after the sentinel close will cause a problem. The usual way to deal with a group of active goroutines is with a sync.WaitGroup:
// Writer: Activity check, attempt 1. This does not quite work.
func writeAdmissionBad(msg message) error {
active.Add(1)
defer active.Done()
select {
case <-done:
return writerClosed
case ch <- msg:
return nil
}
}The reason this doens't work is a little subtle. Here's the other side:
// Closer:
close(done)
active.Wait()
close(ch)The problem is that after we have closed done, and while we are pending in active.Wait, new goroutines may enter and increment the waitgroup. This is a data race, since the contract of sync.WaitGroup requires that all Add calls complete before any Wait. Moving the Add call after the done check doesn't help -- it just puts us back in the original situation, where a writer who entered before done was closed is still active.
What does work, however, is a shared (reader/writer) lock:
var active sync.RWMutex
// Writer: Activity check, attempt 2.
func writeAdmissionBetter(msg message) error {
active.RLock() // N.B. shared
defer active.RUnlock()
select {
case <-done:
return writerClosed
case ch <- msg:
return nil
}
}
// The other side (elsewhere):
close(done)
active.Lock() // N.B. exclusive
defer active.Unlock()
close(ch)This works because it is always safe for the closer to acquire the lock, and once it has done so, we can be sure there are no writers in the critical section (because it is exclusive). Moreover, it is low-cost for concurrent writers, since they can safely share concurrent access to the write method, and the only time they will contend with an exclusive lock is at the point of shutdown, which happens once at the end of the channel's lifecycle.
One general concern with this tactic is that it might be possible for arriving writers to "starve" the close. However, in Go this is not possible, since the standard library RWMutex guarantees that if a goroutine is blocked on an exclusive acquire, no further readers will be admitted until after that critical section.