Abort mux srflx gather on close

Author: JoTurk

agent_test.go

@@ -376,6 +376,134 @@ func TestAgentCloseAbortsBlockedUDPMuxWrite(t *testing.T) {
 	require.NoError(t, <-runErr)
 }
 
+func TestAgentCloseAbortsBlockedUDPMuxSrflxGatherWrite(t *testing.T) {
+	defer test.CheckRoutines(t)()
+	defer test.TimeOut(2 * time.Second).Stop()
+
+	udpConn := newDeadlineBlockingPacketConn()
+	udpMux := NewUniversalUDPMuxDefault(UniversalUDPMuxParams{UDPConn: udpConn})
+	defer func() {
+		_ = udpMux.Close()
+	}()
+
+	agent, err := NewAgent(&AgentConfig{
+		NetworkTypes:   []NetworkType{NetworkTypeUDP4},
+		CandidateTypes: []CandidateType{CandidateTypeServerReflexive},
+		Urls: []*stun.URI{{
+			Scheme: stun.SchemeTypeSTUN,
+			Host:   "192.0.2.2",
+			Port:   3478,
+		}},
+		UDPMuxSrflx: udpMux,
+	})
+	require.NoError(t, err)
+
+	require.NoError(t, agent.OnCandidate(func(Candidate) {}))
+	require.NoError(t, agent.GatherCandidates())
+
+	select {
+	case <-udpConn.writeStarted:
+	case <-time.After(time.Second):
+		require.FailNow(t, "timed out waiting for UDP mux srflx gather write to block")
+	}
+
+	closeErr := make(chan error, 1)
+	go func() {
+		closeErr <- agent.Close()
+	}()
+
+	select {
+	case err := <-closeErr:
+		require.NoError(t, err)
+	case <-time.After(time.Second):
+		require.FailNow(t, "agent close did not abort blocked UDP mux srflx gather write")
+	}
+}
+
+func TestAgentCloseDoesNotAbortOtherAgentUDPMuxSrflxGatherWrite(t *testing.T) { //nolint:cyclop
+	defer test.CheckRoutines(t)()
+	defer test.TimeOut(2 * time.Second).Stop()
+
+	udpConn := newDeadlineBlockingPacketConn()
+	udpMux := NewUniversalUDPMuxDefault(UniversalUDPMuxParams{UDPConn: udpConn})
+	defer func() {
+		_ = udpMux.Close()
+	}()
+
+	newSrflxAgent := func(t *testing.T) *Agent {
+		t.Helper()
+
+		agent, err := NewAgent(&AgentConfig{
+			NetworkTypes:   []NetworkType{NetworkTypeUDP4},
+			CandidateTypes: []CandidateType{CandidateTypeServerReflexive},
+			Urls: []*stun.URI{{
+				Scheme: stun.SchemeTypeSTUN,
+				Host:   "192.0.2.2",
+				Port:   3478,
+			}},
+			UDPMuxSrflx: udpMux,
+		})
+		require.NoError(t, err)
+
+		return agent
+	}
+
+	agent1 := newSrflxAgent(t)
+	defer func() {
+		_ = agent1.Close()
+	}()
+
+	agent2 := newSrflxAgent(t)
+	defer func() {
+		_ = agent2.Close()
+	}()
+
+	require.NoError(t, agent2.OnCandidate(func(Candidate) {}))
+	require.NoError(t, agent2.GatherCandidates())
+
+	select {
+	case <-udpConn.writeStarted:
+	case <-time.After(time.Second):
+		require.FailNow(t, "timed out waiting for second agent UDP mux srflx gather write to block")
+	}
+
+	closeErr := make(chan error, 1)
+	go func() {
+		closeErr <- agent1.Close()
+	}()
+
+	select {
+	case err := <-closeErr:
+		require.NoError(t, err)
+	case <-time.After(time.Second):
+		require.FailNow(t, "first agent close blocked")
+	}
+
+	select {
+	case <-udpConn.writeDeadlineSet:
+		require.FailNow(t, "first agent close aborted another agent's UDP mux srflx gather write")
+	case <-time.After(200 * time.Millisecond):
+	}
+
+	secondCloseErr := make(chan error, 1)
+	go func() {
+		secondCloseErr <- agent2.Close()
+	}()
+
+	select {
+	case <-udpConn.writeDeadlineSet:
+	case <-time.After(time.Second):
+		require.FailNow(t, "second agent close did not abort its own UDP mux srflx gather write")
+	}
+
+	select {
+	case err := <-secondCloseErr:
+		require.NoError(t, err)
+	case <-time.After(time.Second):
+		require.FailNow(t, "second agent close blocked")
+	}
+}
+
 func TestAgentCloseClearsSharedUDPMuxAbortDeadlineForOtherAgent(t *testing.T) { //nolint:cyclop
 	defer test.CheckRoutines(t)()
 	defer test.TimeOut(2 * time.Second).Stop()

gather.go

@@ -760,7 +760,7 @@ func (a *Agent) gatherCandidatesSrflxUDPMux(ctx context.Context, urls []*stun.UR
 						return
 					}
 
-					xorAddr, err := a.udpMuxSrflx.GetXORMappedAddr(serverAddr, a.stunGatherTimeout)
+					xorAddr, err := getXORMappedAddr(ctx, a.udpMuxSrflx, serverAddr, a.stunGatherTimeout)
 					if err != nil {
 						a.log.Warnf("Failed get server reflexive address %s %s: %v", network, url, err)
 
@@ -809,6 +809,27 @@ func (a *Agent) gatherCandidatesSrflxUDPMux(ctx context.Context, urls []*stun.UR
 	}
 }
 
+type contextXORMappedAddrGetter interface {
+	GetXORMappedAddrContext(context.Context, net.Addr, time.Duration) (*stun.XORMappedAddress, error)
+}
+
+func getXORMappedAddr(
+	ctx context.Context,
+	mux UniversalUDPMux,
+	serverAddr net.Addr,
+	deadline time.Duration,
+) (*stun.XORMappedAddress, error) {
+	if muxWithContext, ok := mux.(contextXORMappedAddrGetter); ok {
+		return muxWithContext.GetXORMappedAddrContext(ctx, serverAddr, deadline)
+	}
+
+	if err := ctx.Err(); err != nil {
+		return nil, err
+	}
+
+	return mux.GetXORMappedAddr(serverAddr, deadline)
+}
+
 //nolint:cyclop,gocognit
 func (a *Agent) gatherCandidatesSrflx(ctx context.Context, urls []*stun.URI, networkTypes []NetworkType) {
 	var wg sync.WaitGroup

gather_test.go

@@ -1522,7 +1522,7 @@ func TestGatherCandidatesRelayRespectsInterfaceFilter(t *testing.T) {
 			},
 		}),
 		WithInterfaceFilter(func(iface string) bool {
-			return iface == "eth0"
+			return iface == "eth0" //nolint:goconst
 		}),
 		WithIncludeLoopback(),
 	)

udp_mux.go

@@ -4,6 +4,7 @@
 package ice
 
 import (
+	"context"
 	"errors"
 	"io"
 	"net"
@@ -50,7 +51,11 @@ type UDPMuxDefault struct {
 	// whether the UDP connection listens on an unspecified address
 	isUnspecified bool
 
-	// writeState packs write in-flight count.
+	// writeState coordinates context cancellation for WriteTo calls.
+	// Low bits count writes currently inside UDPConn.WriteTo. blocked means an
+	// abort arming the shared write deadline, so new writes wait.
+	// deadline means SetWriteDeadline(time.Now()) succeeded and the last
+	// in-flight writer must clear it before new writes can enter.
 	writeState atomic.Uint64
 }
 
@@ -269,13 +274,53 @@ func (m *UDPMuxDefault) Close() error {
 }
 
 func (m *UDPMuxDefault) writeTo(buf []byte, rAddr net.Addr) (n int, err error) {
-	m.startWrite()
+	return m.writeToContext(context.Background(), buf, rAddr)
+}
+
+func (m *UDPMuxDefault) writeToContext(ctx context.Context, buf []byte, rAddr net.Addr) (n int, err error) {
+	if err = m.startWriteContext(ctx); err != nil {
+		return 0, err
+	}
 
 	defer func() {
 		err = m.finishWrite(err)
 	}()
 
-	return m.params.UDPConn.WriteTo(buf, rAddr)
+	if err = ctx.Err(); err != nil {
+		return 0, err
+	}
+
+	if done := ctx.Done(); done != nil {
+		// net.PacketConn writes cannot be canceled directly. If ctx is
+		// canceled while WriteTo is blocked, abortWrite interrupts it by
+		// temporarily setting the shared socket write deadline to now.
+		stopAbort := make(chan struct{})
+		var stopped atomic.Bool
+		defer func() {
+			stopped.Store(true)
+			close(stopAbort)
+		}()
+		go func() {
+			select {
+			case <-done:
+				if !stopped.Load() {
+					if abortErr := m.abortWrite(); abortErr != nil {
+						m.params.Logger.Warnf("Failed to abort UDP write: %v", abortErr)
+					}
+				}
+			case <-stopAbort:
+			}
+		}()
+	}
+
+	n, err = m.params.UDPConn.WriteTo(buf, rAddr)
+	if err != nil {
+		if ctxErr := ctx.Err(); ctxErr != nil {
+			return n, ctxErr
+		}
+	}
+
+	return n, err
 }
 
 func (m *UDPMuxDefault) abortWrite() error {
@@ -289,6 +334,8 @@ func (m *UDPMuxDefault) abortWrite() error {
 			continue
 		}
 
+		// The deadline applies to the shared UDPConn, so blocked stays set
+		// until the final in-flight writer clears the deadline in finishWrite.
 		if err := m.params.UDPConn.SetWriteDeadline(time.Now()); err != nil {
 			m.clearWriteAbortState()
 
@@ -301,8 +348,12 @@ func (m *UDPMuxDefault) abortWrite() error {
 	}
 }
 
-func (m *UDPMuxDefault) startWrite() {
+func (m *UDPMuxDefault) startWriteContext(ctx context.Context) error {
 	for {
+		if err := ctx.Err(); err != nil {
+			return err
+		}
+
 		state := m.writeState.Load()
 		if state&udpMuxWriteBlockedBit != 0 {
 			runtime.Gosched()
@@ -311,7 +362,7 @@ func (m *UDPMuxDefault) startWrite() {
 		}
 
 		if m.writeState.CompareAndSwap(state, state+1) {
-			return
+			return nil
 		}
 	}
 }
@@ -357,6 +408,8 @@ func (m *UDPMuxDefault) clearWriteDeadlineAfterAbort(writeErr error) error {
 			return writeErr
 		}
 		if state&udpMuxWriteDeadlineBit == 0 {
+			// The last writer can race with abortWrite after blocked is set but
+			// before SetWriteDeadline returns.
 			runtime.Gosched()
 
 			continue