Fix pc.GracefulClose concurrency > 2

peerconnection.go

@@ -56,8 +56,9 @@ type PeerConnection struct {
 	idpLoginURL *string
 
 	isClosed                                *atomicBool
-	isGracefulClosed                        *atomicBool
-	isGracefulClosedDone                    chan struct{}
+	isGracefullyClosingOrClosed             bool
+	isCloseDone                             chan struct{}
+	isGracefulCloseDone                     chan struct{}
 	isNegotiationNeeded                     *atomicBool
 	updateNegotiationNeededFlagOnEmptyChain *atomicBool
 
@@ -119,8 +120,8 @@ func (api *API) NewPeerConnection(configuration Configuration) (*PeerConnection,
 			ICECandidatePoolSize: 0,
 		},
 		isClosed:                                &atomicBool{},
-		isGracefulClosed:                        &atomicBool{},
-		isGracefulClosedDone:                    make(chan struct{}),
+		isCloseDone:                             make(chan struct{}),
+		isGracefulCloseDone:                     make(chan struct{}),
 		isNegotiationNeeded:                     &atomicBool{},
 		updateNegotiationNeededFlagOnEmptyChain: &atomicBool{},
 		lastOffer:                               "",
@@ -2111,22 +2112,41 @@ func (pc *PeerConnection) GracefulClose() error {
 func (pc *PeerConnection) close(shouldGracefullyClose bool) error {
 	// https://www.w3.org/TR/webrtc/#dom-rtcpeerconnection-close (step #1)
 	// https://www.w3.org/TR/webrtc/#dom-rtcpeerconnection-close (step #2)
-	alreadyGracefullyClosed := shouldGracefullyClose && pc.isGracefulClosed.swap(true)
-	if pc.isClosed.swap(true) {
-		if alreadyGracefullyClosed {
-			// similar but distinct condition where we may be waiting for some
-			// other graceful close to finish. Incorrectly using isClosed may
-			// leak a goroutine.
-			<-pc.isGracefulClosedDone
-		}
-		return nil
+
+	pc.mu.Lock()
+	// A lock in this critical section is needed because pc.isClosed and
+	// pc.isGracefullyClosingOrClosed are related to each other in that we
+	// want to make graceful and normal closure one time operations to avoid
+	// any double closure errors from cropping up.
+	isAlreadyClosingOrClosed := pc.isClosed.swap(true)
+	isAlreadyGracefullyClosingOrClosed := pc.isGracefullyClosingOrClosed
+	if shouldGracefullyClose && !isAlreadyGracefullyClosingOrClosed {
+		pc.isGracefullyClosingOrClosed = true
 	}
-	if shouldGracefullyClose && !alreadyGracefullyClosed {
-		defer close(pc.isGracefulClosedDone)
+	pc.mu.Unlock()
+
+	if isAlreadyClosingOrClosed {
+		if !shouldGracefullyClose {
+			return nil
+		}
+		// even if we're already closing, it may not be graceful:
+		// if we are not closing the graceful channel, we just wait for the close
+		// to happen and return.
+		if isAlreadyGracefullyClosingOrClosed {
+			<-pc.isGracefulCloseDone
+			return nil
+		}
+		// Otherwise we need to go through the graceful flow once the current normal closer
+		// is done since there are extra steps to take with a graceful close.
+		<-pc.isCloseDone
 	}
 
-	// https://www.w3.org/TR/webrtc/#dom-rtcpeerconnection-close (step #3)
-	pc.signalingState.Set(SignalingStateClosed)
+	if !isAlreadyClosingOrClosed {
+		defer close(pc.isCloseDone)
+	}
+	if shouldGracefullyClose && !isAlreadyGracefullyClosingOrClosed {
+		defer close(pc.isGracefulCloseDone)
+	}
 
 	// Try closing everything and collect the errors
 	// Shutdown strategy:
@@ -2136,6 +2156,35 @@ func (pc *PeerConnection) close(shouldGracefullyClose bool) error {
 	//    continue the chain the Mux has to be closed.
 	closeErrs := make([]error, 4)
 
+	doGracefulCloseOps := func() []error {
+		if !shouldGracefullyClose {
+			return nil
+		}
+
+		var gracefulCloseErrors []error
+		if pc.iceTransport != nil {
+			// note that it isn't canon to stop gracefully
+			gracefulCloseErrors = append(gracefulCloseErrors, pc.iceTransport.GracefulStop())
+		}
+
+		pc.ops.GracefulClose()
+
+		// note that it isn't canon to stop gracefully
+		pc.sctpTransport.lock.Lock()
+		for _, d := range pc.sctpTransport.dataChannels {
+			gracefulCloseErrors = append(gracefulCloseErrors, d.GracefulClose())
+		}
+		pc.sctpTransport.lock.Unlock()
+		return gracefulCloseErrors
+	}
+
+	if isAlreadyClosingOrClosed {
+		return util.FlattenErrs(doGracefulCloseOps())
+	}
+
+	// https://www.w3.org/TR/webrtc/#dom-rtcpeerconnection-close (step #3)
+	pc.signalingState.Set(SignalingStateClosed)
+
 	closeErrs = append(closeErrs, pc.api.interceptor.Close())
 
 	// https://www.w3.org/TR/webrtc/#dom-rtcpeerconnection-close (step #4)
@@ -2167,27 +2216,16 @@ func (pc *PeerConnection) close(shouldGracefullyClose bool) error {
 
 	// https://www.w3.org/TR/webrtc/#dom-rtcpeerconnection-close (step #8, #9, #10)
 	if pc.iceTransport != nil {
-		if shouldGracefullyClose {
-			// note that it isn't canon to stop gracefully
-			closeErrs = append(closeErrs, pc.iceTransport.GracefulStop())
-		} else {
+		// we will stop gracefully in doGracefulCloseOps
+		if !shouldGracefullyClose {
 			closeErrs = append(closeErrs, pc.iceTransport.Stop())
 		}
 	}
 
 	// https://www.w3.org/TR/webrtc/#dom-rtcpeerconnection-close (step #11)
 	pc.updateConnectionState(pc.ICEConnectionState(), pc.dtlsTransport.State())
 
-	if shouldGracefullyClose {
-		pc.ops.GracefulClose()
-
-		// note that it isn't canon to stop gracefully
-		pc.sctpTransport.lock.Lock()
-		for _, d := range pc.sctpTransport.dataChannels {
-			closeErrs = append(closeErrs, d.GracefulClose())
-		}
-		pc.sctpTransport.lock.Unlock()
-	}
+	closeErrs = append(closeErrs, doGracefulCloseOps()...)
 
 	return util.FlattenErrs(closeErrs)
 }

peerconnection_close_test.go

@@ -180,7 +180,7 @@ func TestPeerConnection_Close_DuringICE(t *testing.T) {
 	}
 }
 
-func TestPeerConnection_CloseWithIncomingMessages(t *testing.T) {
+func TestPeerConnection_GracefulCloseWithIncomingMessages(t *testing.T) {
 	// Limit runtime in case of deadlocks
 	lim := test.TimeOut(time.Second * 20)
 	defer lim.Stop()
@@ -287,3 +287,59 @@ func TestPeerConnection_GracefulCloseWhileOpening(t *testing.T) {
 		t.Fatal(err)
 	}
 }
+
+func TestPeerConnection_GracefulCloseConcurrent(t *testing.T) {
+	// Limit runtime in case of deadlocks
+	lim := test.TimeOut(time.Second * 5)
+	defer lim.Stop()
+
+	report := test.CheckRoutinesStrict(t)
+	defer report()
+
+	pc, err := NewPeerConnection(Configuration{})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	go func() {
+		if err := pc.GracefulClose(); err != nil {
+			t.Fatal(err)
+		}
+	}()
+	go func() {
+		if err := pc.GracefulClose(); err != nil {
+			t.Fatal(err)
+		}
+	}()
+	if err := pc.GracefulClose(); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func TestPeerConnection_GracefulAndNormalCloseConcurrent(t *testing.T) {
+	// Limit runtime in case of deadlocks
+	lim := test.TimeOut(time.Second * 5)
+	defer lim.Stop()
+
+	report := test.CheckRoutinesStrict(t)
+	defer report()
+
+	pc, err := NewPeerConnection(Configuration{})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	go func() {
+		if err := pc.GracefulClose(); err != nil {
+			t.Fatal(err)
+		}
+	}()
+	go func() {
+		if err := pc.GracefulClose(); err != nil {
+			t.Fatal(err)
+		}
+	}()
+	if err := pc.Close(); err != nil {
+		t.Fatal(err)
+	}
+}