mex.go
func (mexset *messageExchangeSet) stopExchanges(err error) {
	if mexset.log.Enabled(LogLevelDebug) {
		mexset.log.Debugf("stopping %v exchanges due to error: %v", mexset.count(), err)
	}

	mexset.Lock()
	shutdown, exchanges := mexset.copyExchanges()
	mexset.shutdown = true
	mexset.Unlock()

	if shutdown {
		mexset.log.Debugf("mexset has already been shutdown")
		return
	}

	for _, mex := range exchanges {
		// When there's a connection failure, we want to notify blocked callers that the
		// call will fail, but we don't want to shutdown the exchange as only the
		// arg reader/writer should shutdown the exchange. Otherwise, our guarantee
		// on sendChRefs that there's no references to sendCh is violated since
		// readers/writers could still have a reference to sendCh even though
		// we shutdown the exchange and called Done on sendChRefs.
		if mex.errChNotified.CAS(false, true) {
			mex.errCh.Notify(err)
		}
	}
}
func (mexset *messageExchangeSet) stopExchanges(err error) {
	if mexset.log.Enabled(LogLevelDebug) {
		mexset.log.Debugf("stopping %v exchanges due to error: %v", mexset.count(), err)
	}

	mexset.Lock()
	shutdown, exchanges := mexset.copyExchanges()
	mexset.shutdown = true
	mexset.Unlock()

	if shutdown {
		mexset.log.Debugf("mexset has already been shutdown")
		return
	}

	for _, mex := range exchanges {
		// When there's a connection failure, we want to notify blocked callers that the
		// call will fail, but we don't want to shutdown the exchange as only the
		// arg reader/writer should shutdown the exchange. Otherwise, our guarantee
		// on sendChRefs that there's no references to sendCh is violated since
		// readers/writers could still have a reference to sendCh even though
		// we shutdown the exchange and called Done on sendChRefs.
		if mex.errChNotified.CAS(false, true) {
			mex.errCh.Notify(err)
		}
	}
}
// stopExchanges stops all message exchanges to unblock all waiters on the mex.
// This should only be called on connection failures.