How to Gracefully Close Channels

Several days ago, I wrote an article which explains the channel rules in Go. That article got many votes on reddit and HN, but there are also some criticisms on Go channel design details.

I collected some criticisms on the following designs and rules of Go channels:

1. no easy and universal ways to check whether or not a channel is closed without modifying the status of the channel.
2. closing a closed channel will panic, so it is dangerous to close a channel if the closers don't know whether or not the channel is closed.
3. sending values to a closed channel will panic, so it is dangerous to send values to a channel if the senders don't know whether or not the channel is closed.

There is indeed a simple method to check whether or not a channel is closed if you can make sure no values were (and will be) ever sent to the channel. The method has been shown in the last article. Here, for a better coherence, the method is listed in the following example again.

package main

import "fmt"

type T int

func IsClosed(ch <-chan T) bool {
	select {
	case <-ch:
		return true
	default:
	}

	return false
}

func main() {
	c := make(chan T)
	fmt.Println(IsClosed(c)) // false
	close(c)
	fmt.Println(IsClosed(c)) // true
}

As above mentioned, this is not a universal way to check whether a channel is closed.

If you would close a channel from the receiver side or in one of the multiple senders of the channel anyway, then you can use the recover mechanism to prevent the possible panic from crashing your program. Here is an example (assume the channel element type is T).

func SafeClose(ch chan T) (justClosed bool) {
	defer func() {
		if recover() != nil {
			// The return result can be altered
			// in a defer function call.
			justClosed = false
		}
	}()

	// assume ch != nil here.
	close(ch)   // panic if ch is closed
	return true // <=> justClosed = true; return
}

This solution obviously breaks the channel closing principle.

The same idea can be used for sending values to a potential closed channel.

func SafeSend(ch chan T, value T) (closed bool) {
	defer func() {
		if recover() != nil {
			closed = true
		}
	}()

	ch <- value  // panic if ch is closed
	return false // <=> closed = false; return
}

Not only does the rude solution break the channel closing principle, and data races might happen in the process.

Many people prefer using sync.Once to close channels:

type MyChannel struct {
	C    chan T
	once sync.Once
}

func NewMyChannel() *MyChannel {
	return &MyChannel{C: make(chan T)}
}

func (mc *MyChannel) SafeClose() {
	mc.once.Do(func() {
		close(mc.C)
	})
}

Surely, we can also use sync.Mutex to avoid closing a channel multiple times:

type MyChannel struct {
	C      chan T
	closed bool
	mutex  sync.Mutex
}

func NewMyChannel() *MyChannel {
	return &MyChannel{C: make(chan T)}
}

func (mc *MyChannel) SafeClose() {
	mc.mutex.Lock()
	defer mc.mutex.Unlock()
	if !mc.closed {
		close(mc.C)
		mc.closed = true
	}
}

func (mc *MyChannel) IsClosed() bool {
	mc.mutex.Lock()
	defer mc.mutex.Unlock()
	return mc.closed
}

These ways may be polite, but they may not avoid data races. Currently, Go specification doesn't guarantee that there are no data races happening when a channel close and a channel send operations are executed concurrently. If a SafeClose function is called concurrently with a channel send operation to the same channel, data races might happen (though such data races generally don't do any harm).

This is the simplest situation, just let the sender close the data channel when it doesn't want to send more.

package main

import (
	"time"
	"math/rand"
	"sync"
	"log"
)

func main() {
	rand.Seed(time.Now().UnixNano()) // needed before Go 1.20
	log.SetFlags(0)

	// ...
	const Max = 100000
	const NumReceivers = 100

	wgReceivers := sync.WaitGroup{}
	wgReceivers.Add(NumReceivers)

	// ...
	dataCh := make(chan int)

	// the sender
	go func() {
		for {
			if value := rand.Intn(Max); value == 0 {
				// The only sender can close the
				// channel at any time safely.
				close(dataCh)
				return
			} else {
				dataCh <- value
			}
		}
	}()

	// receivers
	for i := 0; i < NumReceivers; i++ {
		go func() {
			defer wgReceivers.Done()

			// Receive values until dataCh is
			// closed and the value buffer queue
			// of dataCh becomes empty.
			for value := range dataCh {
				log.Println(value)
			}
		}()
	}

	wgReceivers.Wait()
}

This is a situation a little more complicated than the above one. We can't let the receiver close the data channel to stop data transferring, for doing this will break the channel closing principle. But we can let the receiver close an additional signal channel to notify senders to stop sending values.

package main

import (
	"time"
	"math/rand"
	"sync"
	"log"
)

func main() {
	rand.Seed(time.Now().UnixNano()) // needed before Go 1.20
	log.SetFlags(0)

	// ...
	const Max = 100000
	const NumSenders = 1000

	wgReceivers := sync.WaitGroup{}
	wgReceivers.Add(1)

	// ...
	dataCh := make(chan int)
	stopCh := make(chan struct{})
		// stopCh is an additional signal channel.
		// Its sender is the receiver of channel
		// dataCh, and its receivers are the
		// senders of channel dataCh.

	// senders
	for i := 0; i < NumSenders; i++ {
		go func() {
			for {
				// The try-receive operation is to try
				// to exit the goroutine as early as
				// possible. For this specified example,
				// it is not essential.
				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 for this
				// example, so the first select block
				// above can be omitted.
				select {
				case <- stopCh:
					return
				case dataCh <- rand.Intn(Max):
				}
			}
		}()
	}

	// the receiver
	go func() {
		defer wgReceivers.Done()

		for value := range dataCh {
			if value == Max-1 {
				// The receiver of channel dataCh is
				// also the sender of stopCh. It is
				// safe to close the stop channel here.
				close(stopCh)
				return
			}

			log.Println(value)
		}
	}()

	// ...
	wgReceivers.Wait()
}

As mentioned in the comments, for the additional signal channel, its sender is the receiver of the data channel. The additional signal channel is closed by its only sender, which holds the channel closing principle.

In this example, the channel dataCh is never closed. Yes, channels don't have to be closed. A channel will be eventually garbage collected if no goroutines reference it any more, whether it is closed or not. So the gracefulness of closing a channel here is not to close the channel.

This is a the most complicated situation. We can't let any of the receivers and the senders close the data channel. And we can't let any of the receivers close an additional signal channel to notify all senders and receivers to exit the game. Doing either will break the channel closing principle. However, we can introduce a moderator role to close the additional signal channel. One trick in the following example is how to use a try-send operation to notify the moderator to close the additional signal channel.

package main

import (
	"time"
	"math/rand"
	"sync"
	"log"
	"strconv"
)

func main() {
	rand.Seed(time.Now().UnixNano()) // needed before Go 1.20
	log.SetFlags(0)

	// ...
	const Max = 100000
	const NumReceivers = 10
	const NumSenders = 1000

	wgReceivers := sync.WaitGroup{}
	wgReceivers.Add(NumReceivers)

	// ...
	dataCh := make(chan int)
	stopCh := make(chan struct{})
		// stopCh is an additional signal channel.
		// Its sender is the moderator goroutine shown
		// below, and its receivers are all senders
		// and receivers of dataCh.
	toStop := make(chan string, 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, and its receiver is
		// the moderator goroutine shown below.
		// It must be a buffered channel.

	var stoppedBy string

	// moderator
	go func() {
		stoppedBy = <-toStop
		close(stopCh)
	}()

	// senders
	for i := 0; i < NumSenders; i++ {
		go func(id string) {
			for {
				value := rand.Intn(Max)
				if value == 0 {
					// Here, the try-send operation is
					// to notify the moderator to close
					// the additional signal channel.
					select {
					case toStop <- "sender#" + id:
					default:
					}
					return
				}

				// The try-receive operation here is to
				// try to exit the sender goroutine as
				// early as possible. Try-receive and
				// try-send select blocks are specially
				// optimized by the standard Go
				// compiler, so they are very efficient.
				select {
				case <- stopCh:
					return
				default:
				}

				// Even if stopCh is closed, the first
				// branch in this select block might be
				// still not selected for some loops
				// (and for ever in theory) if the send
				// to dataCh is also non-blocking. If
				// this is unacceptable, then the above
				// try-receive operation is essential.
				select {
				case <- stopCh:
					return
				case dataCh <- value:
				}
			}
		}(strconv.Itoa(i))
	}

	// receivers
	for i := 0; i < NumReceivers; i++ {
		go func(id string) {
			defer wgReceivers.Done()

			for {
				// Same as the sender goroutine, the
				// try-receive operation here is to
				// try to exit the receiver goroutine
				// as early as possible.
				select {
				case <- stopCh:
					return
				default:
				}

				// Even if stopCh is closed, the first
				// branch in this select block might be
				// still not selected for some loops
				// (and forever in theory) if the receive
				// from dataCh is also non-blocking. If
				// this is not acceptable, then the above
				// try-receive operation is essential.
				select {
				case <- stopCh:
					return
				case value := <-dataCh:
					if value == Max-1 {
						// Here, 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)
}

In this example, the channel closing principle is still held.

Please note that the buffer size (capacity) of channel toStop is one. This is to avoid the first notification is missed when it is sent before the moderator goroutine gets ready to receive notification from toStop.

We can also set the capacity of the toStop channel as the sum number of senders and receivers, then we don't need a try-send select block to notify the moderator.

...
toStop := make(chan string, NumReceivers + NumSenders)
...
			value := rand.Intn(Max)
			if value == 0 {
				toStop <- "sender#" + id
				return
			}
...
				if value == Max-1 {
					toStop <- "receiver#" + id
					return
				}
...

Sometimes, it is needed that the close signal must be made by a third-party goroutine. For such cases, we can use an extra signal channel to notify the sender to close the data channel. For example,

package main

import (
	"time"
	"math/rand"
	"sync"
	"log"
)

func main() {
	rand.Seed(time.Now().UnixNano()) // needed before Go 1.20
	log.SetFlags(0)

	// ...
	const Max = 100000
	const NumReceivers = 100
	const NumThirdParties = 15

	wgReceivers := sync.WaitGroup{}
	wgReceivers.Add(NumReceivers)

	// ...
	dataCh := make(chan int)
	closing := make(chan struct{}) // signal channel
	closed := make(chan struct{})
	
	// The stop function can be called
	// multiple times safely.
	stop := func() {
		select {
		case closing<-struct{}{}:
			<-closed
		case <-closed:
		}
	}
	
	// some third-party goroutines
	for i := 0; i < NumThirdParties; i++ {
		go func() {
			r := 1 + rand.Intn(3)
			time.Sleep(time.Duration(r) * time.Second)
			stop()
		}()
	}

	// the sender
	go func() {
		defer func() {
			close(closed)
			close(dataCh)
		}()

		for {
			select{
			case <-closing: return
			default:
			}

			select{
			case <-closing: return
			case dataCh <- rand.Intn(Max):
			}
		}
	}()

	// receivers
	for i := 0; i < NumReceivers; i++ {
		go func() {
			defer wgReceivers.Done()

			for value := range dataCh {
				log.Println(value)
			}
		}()
	}

	wgReceivers.Wait()
}

The idea used in the stop function is learned from a comment made by Roger Peppe.

In the solutions for the above N-sender situations, to hold the channel closing principle, we avoid closing the data channels. However, sometimes, it is required that the data channels must be closed in the end to let receivers know data sending is over. For such cases, we can translate a N-sender situation to a one-sender situation by using a middle channel. The middle channel has only one sender, so that we can close it instead of closing the original data channel.

package main

import (
	"time"
	"math/rand"
	"sync"
	"log"
	"strconv"
)

func main() {
	rand.Seed(time.Now().UnixNano()) // needed before Go 1.20
	log.SetFlags(0)

	// ...
	const Max = 1000000
	const NumReceivers = 10
	const NumSenders = 1000
	const NumThirdParties = 15

	wgReceivers := sync.WaitGroup{}
	wgReceivers.Add(NumReceivers)

	// ...
	dataCh := make(chan int)     // will be closed
	middleCh := make(chan int)   // will never be closed
	closing := make(chan string) // signal channel
	closed := make(chan struct{})

	var stoppedBy string

	// The stop function can be called
	// multiple times safely.
	stop := func(by string) {
		select {
		case closing <- by:
			<-closed
		case <-closed:
		}
	}
	
	// the middle layer
	go func() {
		exit := func(v int, needSend bool) {
			close(closed)
			if needSend {
				dataCh <- v
			}
			close(dataCh)
		}

		for {
			select {
			case stoppedBy = <-closing:
				exit(0, false)
				return
			case v := <- middleCh:
				select {
				case stoppedBy = <-closing:
					exit(v, true)
					return
				case dataCh <- v:
				}
			}
		}
	}()
	
	// some third-party goroutines
	for i := 0; i < NumThirdParties; i++ {
		go func(id string) {
			r := 1 + rand.Intn(3)
			time.Sleep(time.Duration(r) * time.Second)
			stop("3rd-party#" + id)
		}(strconv.Itoa(i))
	}

	// senders
	for i := 0; i < NumSenders; i++ {
		go func(id string) {
			for {
				value := rand.Intn(Max)
				if value == 0 {
					stop("sender#" + id)
					return
				}

				select {
				case <- closed:
					return
				default:
				}

				select {
				case <- closed:
					return
				case middleCh <- value:
				}
			}
		}(strconv.Itoa(i))
	}

	// receivers
	for range [NumReceivers]struct{}{} {
		go func() {
			defer wgReceivers.Done()

			for value := range dataCh {
				log.Println(value)
			}
		}()
	}

	// ...
	wgReceivers.Wait()
	log.Println("stopped by", stoppedBy)
}

• Go 101에 대해 - 이 책이 쓰여진 이유
• 감사의 말

• Go 소개 - Go를 배우는 가치
• Go 툴체인 - Go 프로그램을 컴파일하고 실행하는 방법

• Go 코드에 익숙해지기
  • 소스 코드 요소 소개
  • 키워드와 식별자
  • 기본 자료형과 기본 값 리터럴
  • 상수와 변수 - 무형성(untyped) 값과 자료형 추론 소개를 포함
  • 일반 연산자 - 더 많은 자료형 추론 규칙 소개를 포함
  • 함수 선언과 호출
  • 코드 패키지와 패키지 들여오기
  • 표현식, 구문과 단순 구문
  • 기본 흐름 제어
  • 고루틴, 지연된 함수 호출과 패닉/복구

• Go 자료형 체계
  • Go 자료형 체계 개요 - Go 프로그래밍 숙달을 위해 반드시 읽어봐야 하는
  • 포인터
  • 구조체
  • 변수 - Go 변수에 대한 더 깊은 이해
  • 배열, 슬라이스와 맵 - 1급 객체 컨테이너 자료형
  • 문자열
  • 함수 - 함수 자료형과 값, 가변 인자 함수
  • 채널 - Go에서 동시성 동기화를 하는 방법
  • 메서드
  • 인터페이스 - 리플렉션과 다형성을 하는 값 상자
  • 자료형 임베딩 - 자료형을 확장하는 방법
  • 자료형에 안전하지 않는 포인터
  • 제네릭 - 합성 자료형의 사용과 읽는 법
  • 리플렉션 - reflect 표준 패키지

• 특별 주제
  • 개행 규칙
  • 지연된 함수 호출 더 알아보기
  • 패닉/복구 사용 사례
  • 패닉/복구 메커니즘에 대한 고찰 - 함수 호출 종료 단계를 포함
  • 코드 블록과 식별자 스코프
  • 표현식 평가 순서
  • Go의 값 복사 비용
  • 경계 검사 제거(BCE)

• 동시성 프로그래밍
  • 동시성 동기화 개요
  • 채널 사용 사례
  • 채널을 깔끔하게 닫는 방법
  • 기타 동시성 동기화 기술 - sync 표준 패키지
  • 원자적 연산 - sync/atomic 표준 패키지
  • Go의 메모리 순서 보장
  • 흔히들 저지르는 동시성 프로그래밍 실수들

• 메모리 관련
  • 메모리 블록
  • 메모리 레이아웃
  • 메모리 누수 시나리오

• 일부 요약
  • 몇 가지 간단한 요약
  • Go의 nil
  • 값 변환, 할당과 비교 규칙
  • 구문/의미 예외
  • Go Details 101
  • Go FAQ 101
  • Go Tips 101

• 더 많은 토픽