diff --git a/pkg/rtc/config.go b/pkg/rtc/config.go index 50473078d..14aa9b73f 100644 --- a/pkg/rtc/config.go +++ b/pkg/rtc/config.go @@ -152,7 +152,7 @@ func checkUDPReadBuffer() (int, error) { return 0, err } defer func() { _ = conn.Close() }() - _ = conn.SetReadBuffer(minUDPBufferSize) + _ = conn.SetReadBuffer(defaultUDPBufferSize) fd, err := conn.File() if err != nil { return 0, nil diff --git a/pkg/rtc/transport.go b/pkg/rtc/transport.go index 5f9f6e169..12e8e980d 100644 --- a/pkg/rtc/transport.go +++ b/pkg/rtc/transport.go @@ -144,8 +144,10 @@ func (t *PCTransport) SetRemoteDescription(sd webrtc.SessionDescription) error { state := t.negotiationState t.negotiationState = negotiationStateNone if state == negotiationRetry { - // need to Negotiate again - t.Negotiate() + // need to Negotiate again, do it immediately + if err := t.CreateAndSendOffer(nil); err != nil { + logger.Errorw("could not negotiate", err) + } } return nil @@ -199,6 +201,9 @@ func (t *PCTransport) CreateAndSendOffer(options *webrtc.OfferOptions) error { t.negotiationState = negotiationRetry return nil } + } else if t.negotiationState == negotiationRetry { + // already set to retry, we can safely skip this attempt + return nil } offer, err := t.pc.CreateOffer(options) diff --git a/pkg/rtc/transport_test.go b/pkg/rtc/transport_test.go index 747109ec8..343de8521 100644 --- a/pkg/rtc/transport_test.go +++ b/pkg/rtc/transport_test.go @@ -1,6 +1,7 @@ package rtc import ( + "sync/atomic" "testing" "github.com/livekit/livekit-server/pkg/testutils" @@ -23,20 +24,8 @@ func TestMissingAnswerDuringICERestart(t *testing.T) { require.NoError(t, err) // exchange ICE - transportA.pc.OnICECandidate(func(candidate *webrtc.ICECandidate) { - if candidate == nil { - return - } - t.Logf("got ICE candidate from A: %v", candidate) - require.NoError(t, transportB.AddICECandidate(candidate.ToJSON())) - }) - transportB.pc.OnICECandidate(func(candidate *webrtc.ICECandidate) { - if candidate == nil { - return - } - t.Logf("got ICE candidate from B: %v", candidate) - require.NoError(t, transportA.AddICECandidate(candidate.ToJSON())) - }) + handleICEExchange(t, transportA, transportB) + // set offer/answer handleOffer := handleOfferFunc(t, transportA, transportB) transportA.OnOffer(handleOffer) @@ -71,6 +60,38 @@ func TestMissingAnswerDuringICERestart(t *testing.T) { }) } +func TestNegotiationTiming(t *testing.T) { + params := TransportParams{ + Target: livekit.SignalTarget_SUBSCRIBER, + Config: &WebRTCConfig{}, + Stats: nil, + } + transportA, err := NewPCTransport(params) + require.NoError(t, err) + _, err = transportA.pc.CreateDataChannel("test", nil) + require.NoError(t, err) + transportB, err := NewPCTransport(params) + require.NoError(t, err) + + handleICEExchange(t, transportA, transportB) + offer := atomic.Value{} + transportA.OnOffer(func(sd webrtc.SessionDescription) { + offer.Store(sd) + }) + + // initial offer + require.NoError(t, transportA.CreateAndSendOffer(nil)) + require.Equal(t, negotiationStateClient, transportA.negotiationState) + + // second try, should've flipped transport status to retry + require.NoError(t, transportA.CreateAndSendOffer(nil)) + require.Equal(t, negotiationRetry, transportA.negotiationState) + + // third try, should've stayed at retry + require.NoError(t, transportA.CreateAndSendOffer(nil)) + require.Equal(t, negotiationRetry, transportA.negotiationState) +} + func handleOfferFunc(t *testing.T, current, other *PCTransport) func(sd webrtc.SessionDescription) { return func(sd webrtc.SessionDescription) { t.Logf("handling offer") @@ -84,3 +105,20 @@ func handleOfferFunc(t *testing.T, current, other *PCTransport) func(sd webrtc.S require.NoError(t, current.SetRemoteDescription(answer)) } } + +func handleICEExchange(t *testing.T, a, b *PCTransport) { + a.pc.OnICECandidate(func(candidate *webrtc.ICECandidate) { + if candidate == nil { + return + } + t.Logf("got ICE candidate from A: %v", candidate) + require.NoError(t, b.AddICECandidate(candidate.ToJSON())) + }) + b.pc.OnICECandidate(func(candidate *webrtc.ICECandidate) { + if candidate == nil { + return + } + t.Logf("got ICE candidate from B: %v", candidate) + require.NoError(t, a.AddICECandidate(candidate.ToJSON())) + }) +}