From aaeb3c933c4da9bd0d7b444718a504b9b2c1ae68 Mon Sep 17 00:00:00 2001 From: cnderrauber Date: Mon, 28 Nov 2022 13:04:56 +0800 Subject: [PATCH] Fix rtcp lost for downtrack used incorrect buffer factory (#1195) * Fix rtcp lost for downtrack used incorrect buffer factory In buffer factory change(#1173), every pariticipant has its own buffer factory, can't use publisher's bufferfactory to create DownTrack * clean code --- pkg/rtc/mediatrack.go | 1 - pkg/rtc/mediatrackreceiver.go | 2 - pkg/rtc/mediatracksubscriptions.go | 4 +- pkg/rtc/participant.go | 5 ++ pkg/rtc/types/interfaces.go | 2 + .../typesfakes/fake_local_participant.go | 66 +++++++++++++++++++ 6 files changed, 74 insertions(+), 6 deletions(-) diff --git a/pkg/rtc/mediatrack.go b/pkg/rtc/mediatrack.go index 5af45dd8d..e2a0b6229 100644 --- a/pkg/rtc/mediatrack.go +++ b/pkg/rtc/mediatrack.go @@ -68,7 +68,6 @@ func NewMediaTrack(params MediaTrackParams) *MediaTrack { ParticipantID: params.ParticipantID, ParticipantIdentity: params.ParticipantIdentity, ParticipantVersion: params.ParticipantVersion, - BufferFactory: params.BufferFactory, ReceiverConfig: params.ReceiverConfig, SubscriberConfig: params.SubscriberConfig, Telemetry: params.Telemetry, diff --git a/pkg/rtc/mediatrackreceiver.go b/pkg/rtc/mediatrackreceiver.go index fcbe6e4ed..824a2a5e7 100644 --- a/pkg/rtc/mediatrackreceiver.go +++ b/pkg/rtc/mediatrackreceiver.go @@ -69,7 +69,6 @@ type MediaTrackReceiverParams struct { ParticipantID livekit.ParticipantID ParticipantIdentity livekit.ParticipantIdentity ParticipantVersion uint32 - BufferFactory *buffer.Factory ReceiverConfig ReceiverConfig SubscriberConfig DirectionConfig Telemetry telemetry.TelemetryService @@ -110,7 +109,6 @@ func NewMediaTrackReceiver(params MediaTrackReceiverParams) *MediaTrackReceiver t.MediaTrackSubscriptions = NewMediaTrackSubscriptions(MediaTrackSubscriptionsParams{ MediaTrack: params.MediaTrack, IsRelayed: params.IsRelayed, - BufferFactory: params.BufferFactory, ReceiverConfig: params.ReceiverConfig, SubscriberConfig: params.SubscriberConfig, Telemetry: params.Telemetry, diff --git a/pkg/rtc/mediatracksubscriptions.go b/pkg/rtc/mediatracksubscriptions.go index b4e26d020..bb189a9db 100644 --- a/pkg/rtc/mediatracksubscriptions.go +++ b/pkg/rtc/mediatracksubscriptions.go @@ -15,7 +15,6 @@ import ( "github.com/livekit/livekit-server/pkg/rtc/types" "github.com/livekit/livekit-server/pkg/sfu" - "github.com/livekit/livekit-server/pkg/sfu/buffer" "github.com/livekit/livekit-server/pkg/telemetry" ) @@ -40,7 +39,6 @@ type MediaTrackSubscriptionsParams struct { MediaTrack types.MediaTrack IsRelayed bool - BufferFactory *buffer.Factory ReceiverConfig ReceiverConfig SubscriberConfig DirectionConfig @@ -110,7 +108,7 @@ func (t *MediaTrackSubscriptions) AddSubscriber(sub types.LocalParticipant, wr * downTrack, err := sfu.NewDownTrack( codecs, wr, - t.params.BufferFactory, + sub.GetBufferFactory(), subscriberID, t.params.ReceiverConfig.PacketBufferSize, LoggerWithTrack(sub.GetLogger(), trackID, t.params.IsRelayed), diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 14b200219..bbd1113bd 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -21,6 +21,7 @@ import ( "github.com/livekit/livekit-server/pkg/rtc/supervisor" "github.com/livekit/livekit-server/pkg/rtc/types" "github.com/livekit/livekit-server/pkg/sfu" + "github.com/livekit/livekit-server/pkg/sfu/buffer" "github.com/livekit/livekit-server/pkg/sfu/connectionquality" "github.com/livekit/livekit-server/pkg/telemetry" "github.com/livekit/mediatransportutil/pkg/twcc" @@ -250,6 +251,10 @@ func (p *ParticipantImpl) GetICEConnectionType() types.ICEConnectionType { return p.TransportManager.GetICEConnectionType() } +func (p *ParticipantImpl) GetBufferFactory() *buffer.Factory { + return p.params.Config.BufferFactory +} + // SetMetadata attaches metadata to the participant func (p *ParticipantImpl) SetMetadata(metadata string) { p.lock.Lock() diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index e8974a5bd..8a74d4513 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -13,6 +13,7 @@ import ( "github.com/livekit/livekit-server/pkg/routing" "github.com/livekit/livekit-server/pkg/sfu" + "github.com/livekit/livekit-server/pkg/sfu/buffer" ) //go:generate go run github.com/maxbrunsfeld/counterfeiter/v6 -generate @@ -244,6 +245,7 @@ type LocalParticipant interface { SubscriberAsPrimary() bool GetClientConfiguration() *livekit.ClientConfiguration GetICEConnectionType() ICEConnectionType + GetBufferFactory() *buffer.Factory SetResponseSink(sink routing.MessageSink) CloseSignalConnection() diff --git a/pkg/rtc/types/typesfakes/fake_local_participant.go b/pkg/rtc/types/typesfakes/fake_local_participant.go index bc4a43634..8370ca176 100644 --- a/pkg/rtc/types/typesfakes/fake_local_participant.go +++ b/pkg/rtc/types/typesfakes/fake_local_participant.go @@ -8,6 +8,7 @@ import ( "github.com/livekit/livekit-server/pkg/routing" "github.com/livekit/livekit-server/pkg/rtc/types" "github.com/livekit/livekit-server/pkg/sfu" + "github.com/livekit/livekit-server/pkg/sfu/buffer" "github.com/livekit/protocol/auth" "github.com/livekit/protocol/livekit" "github.com/livekit/protocol/logger" @@ -215,6 +216,16 @@ type FakeLocalParticipant struct { result1 float64 result2 bool } + GetBufferFactoryStub func() *buffer.Factory + getBufferFactoryMutex sync.RWMutex + getBufferFactoryArgsForCall []struct { + } + getBufferFactoryReturns struct { + result1 *buffer.Factory + } + getBufferFactoryReturnsOnCall map[int]struct { + result1 *buffer.Factory + } GetCachedDownTrackStub func(livekit.TrackID) (*webrtc.RTPTransceiver, sfu.DownTrackState) getCachedDownTrackMutex sync.RWMutex getCachedDownTrackArgsForCall []struct { @@ -1781,6 +1792,59 @@ func (fake *FakeLocalParticipant) GetAudioLevelReturnsOnCall(i int, result1 floa }{result1, result2} } +func (fake *FakeLocalParticipant) GetBufferFactory() *buffer.Factory { + fake.getBufferFactoryMutex.Lock() + ret, specificReturn := fake.getBufferFactoryReturnsOnCall[len(fake.getBufferFactoryArgsForCall)] + fake.getBufferFactoryArgsForCall = append(fake.getBufferFactoryArgsForCall, struct { + }{}) + stub := fake.GetBufferFactoryStub + fakeReturns := fake.getBufferFactoryReturns + fake.recordInvocation("GetBufferFactory", []interface{}{}) + fake.getBufferFactoryMutex.Unlock() + if stub != nil { + return stub() + } + if specificReturn { + return ret.result1 + } + return fakeReturns.result1 +} + +func (fake *FakeLocalParticipant) GetBufferFactoryCallCount() int { + fake.getBufferFactoryMutex.RLock() + defer fake.getBufferFactoryMutex.RUnlock() + return len(fake.getBufferFactoryArgsForCall) +} + +func (fake *FakeLocalParticipant) GetBufferFactoryCalls(stub func() *buffer.Factory) { + fake.getBufferFactoryMutex.Lock() + defer fake.getBufferFactoryMutex.Unlock() + fake.GetBufferFactoryStub = stub +} + +func (fake *FakeLocalParticipant) GetBufferFactoryReturns(result1 *buffer.Factory) { + fake.getBufferFactoryMutex.Lock() + defer fake.getBufferFactoryMutex.Unlock() + fake.GetBufferFactoryStub = nil + fake.getBufferFactoryReturns = struct { + result1 *buffer.Factory + }{result1} +} + +func (fake *FakeLocalParticipant) GetBufferFactoryReturnsOnCall(i int, result1 *buffer.Factory) { + fake.getBufferFactoryMutex.Lock() + defer fake.getBufferFactoryMutex.Unlock() + fake.GetBufferFactoryStub = nil + if fake.getBufferFactoryReturnsOnCall == nil { + fake.getBufferFactoryReturnsOnCall = make(map[int]struct { + result1 *buffer.Factory + }) + } + fake.getBufferFactoryReturnsOnCall[i] = struct { + result1 *buffer.Factory + }{result1} +} + func (fake *FakeLocalParticipant) GetCachedDownTrack(arg1 livekit.TrackID) (*webrtc.RTPTransceiver, sfu.DownTrackState) { fake.getCachedDownTrackMutex.Lock() ret, specificReturn := fake.getCachedDownTrackReturnsOnCall[len(fake.getCachedDownTrackArgsForCall)] @@ -4902,6 +4966,8 @@ func (fake *FakeLocalParticipant) Invocations() map[string][][]interface{} { defer fake.getAdaptiveStreamMutex.RUnlock() fake.getAudioLevelMutex.RLock() defer fake.getAudioLevelMutex.RUnlock() + fake.getBufferFactoryMutex.RLock() + defer fake.getBufferFactoryMutex.RUnlock() fake.getCachedDownTrackMutex.RLock() defer fake.getCachedDownTrackMutex.RUnlock() fake.getClientConfigurationMutex.RLock()