From ae203c27e3556ec962e330d5b7c54fdd29ad6ab0 Mon Sep 17 00:00:00 2001 From: David Zhao Date: Sat, 6 Feb 2021 23:02:21 -0800 Subject: [PATCH] send resync to ensure keyframes have been requested --- go.mod | 8 +- go.sum | 10 +- pkg/config/config.go | 3 +- pkg/rtc/mediaengine.go | 88 +++++++++++++ pkg/rtc/mediatrack.go | 132 ++++++++++--------- pkg/rtc/participant.go | 79 ++++++----- pkg/rtc/subscribedtrack.go | 52 ++++++++ pkg/rtc/types/interfaces.go | 35 ++--- pkg/rtc/types/typesfakes/fake_participant.go | 6 +- 9 files changed, 270 insertions(+), 143 deletions(-) create mode 100644 pkg/rtc/mediaengine.go create mode 100644 pkg/rtc/subscribedtrack.go diff --git a/go.mod b/go.mod index bf43879ca..423b6d34d 100644 --- a/go.mod +++ b/go.mod @@ -4,7 +4,6 @@ go 1.15 require ( github.com/bep/debounce v1.2.0 - github.com/gammazero/workerpool v1.1.1 github.com/go-redis/redis/v8 v8.4.8 github.com/golang/protobuf v1.4.3 github.com/google/wire v0.4.0 @@ -15,12 +14,11 @@ require ( github.com/magefile/mage v1.10.0 github.com/maxbrunsfeld/counterfeiter/v6 v6.3.0 github.com/mitchellh/go-homedir v1.1.0 - github.com/pion/ion-log v1.0.0 - github.com/pion/ion-sfu v1.8.2 + github.com/pion/ion-sfu v1.8.3 github.com/pion/rtcp v1.2.6 github.com/pion/rtp v1.6.2 + github.com/pion/sdp/v3 v3.0.4 github.com/pion/stun v0.3.5 - github.com/pion/transport v0.12.2 github.com/pion/webrtc/v3 v3.0.5 github.com/pkg/errors v0.9.1 github.com/stretchr/testify v1.7.0 @@ -36,4 +34,4 @@ require ( gopkg.in/yaml.v3 v3.0.0-20200615113413-eeeca48fe776 ) -replace github.com/pion/ion-sfu => github.com/davidzhao/ion-sfu v1.8.3-0.20210205014052-b5e378606cec +replace github.com/pion/ion-sfu => ../ion-sfu diff --git a/go.sum b/go.sum index 7dd20a85e..478d9bd09 100644 --- a/go.sum +++ b/go.sum @@ -71,10 +71,8 @@ github.com/creack/pty v1.1.7/go.mod h1:lj5s0c3V2DBrqTV7llrYr5NG6My20zk30Fl46Y7Do github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= -github.com/davidzhao/ion-sfu v1.8.3-0.20210205010743-13f2b5e71196 h1:lYSupwHjpNUtXO3GEHjhX9S1F3efREDs4FpTAjf+Frc= -github.com/davidzhao/ion-sfu v1.8.3-0.20210205010743-13f2b5e71196/go.mod h1:MKDh4JSLpBA/hFeYAYALb6nYpe6Ruknd3pmxPMbdpM8= -github.com/davidzhao/ion-sfu v1.8.3-0.20210205014052-b5e378606cec h1:HjyyD0TXuCVheKu7HyWdye3jQ2/+dE/f2q7I+u3X2oA= -github.com/davidzhao/ion-sfu v1.8.3-0.20210205014052-b5e378606cec/go.mod h1:MKDh4JSLpBA/hFeYAYALb6nYpe6Ruknd3pmxPMbdpM8= +github.com/davidzhao/ion-sfu v1.8.3-0.20210205220645-c9ec7ad1db86 h1:POGjEQ8fyEQn96DMXMG7xhvWxHjNRMQ5aoiRzQOzOdY= +github.com/davidzhao/ion-sfu v1.8.3-0.20210205220645-c9ec7ad1db86/go.mod h1:MKDh4JSLpBA/hFeYAYALb6nYpe6Ruknd3pmxPMbdpM8= github.com/desertbit/timer v0.0.0-20180107155436-c41aec40b27f/go.mod h1:xH/i4TFMt8koVQZ6WFms69WAsDWr2XsYL3Hkl7jkoLE= github.com/dgrijalva/jwt-go v3.2.0+incompatible h1:7qlOGliEKZXTDg6OTjfoBKDXWrumCAMpl/TFQ4/5kLM= github.com/dgrijalva/jwt-go v3.2.0+incompatible/go.mod h1:E3ru+11k8xSBh+hMPgOLZmtrrCbhqsmaPHjLKYnJCaQ= @@ -331,10 +329,6 @@ github.com/pion/interceptor v0.0.9 h1:fk5hTdyLO3KURQsf/+RjMpEm4NE3yeTY9Kh97b5Bvw github.com/pion/interceptor v0.0.9/go.mod h1:dHgEP5dtxOTf21MObuBAjJeAayPxLUAZjerGH8Xr07c= github.com/pion/ion-log v1.0.0 h1:2lJLImCmfCWCR38hLWsjQfBWe6NFz/htbqiYHwvOP/Q= github.com/pion/ion-log v1.0.0/go.mod h1:jwcla9KoB9bB/4FxYDSRJPcPYSLp5XiUUMnOLaqwl4E= -github.com/pion/ion-sfu v1.8.1 h1:N3xaDZ2Om2pCzPLlusPmV7BJYVedE5VEx8um6d0Crns= -github.com/pion/ion-sfu v1.8.1/go.mod h1:b6FI2WFTBsifw9EhxvfDks5/9SGYmVvm00IogZEgaik= -github.com/pion/ion-sfu v1.8.2 h1:YIkelxhX5EZgWkmj6XY+0AVKi/se8W1bkOobEe2qjYc= -github.com/pion/ion-sfu v1.8.2/go.mod h1:MKDh4JSLpBA/hFeYAYALb6nYpe6Ruknd3pmxPMbdpM8= github.com/pion/logging v0.2.2 h1:M9+AIj/+pxNsDfAT64+MAVgJO0rsyLnoJKCqf//DoeY= github.com/pion/logging v0.2.2/go.mod h1:k0/tDVsRCX2Mb2ZEmTqNa7CWsQPc+YYCB7Q+5pahoms= github.com/pion/mdns v0.0.4 h1:O4vvVqr4DGX63vzmO6Fw9vpy3lfztVWHGCQfyw0ZLSY= diff --git a/pkg/config/config.go b/pkg/config/config.go index b14900538..7a531d287 100644 --- a/pkg/config/config.go +++ b/pkg/config/config.go @@ -25,7 +25,7 @@ type RTCConfig struct { StunServers []string `yaml:"stun_servers"` UseExternalIP bool `yaml:"use_external_ip"` - MaxBitrate uint64 `yaml:"max_bandwidth"` + MaxBitrate uint64 `yaml:"max_bitrate"` MaxBufferTime int `yaml:"max_buffer_time"` } @@ -44,6 +44,7 @@ func NewConfig(confString string) (*Config, error) { StunServers: []string{ "stun.l.google.com:19302", }, + MaxBitrate: 2 * 1024 * 1024, // 3 mbps }, Redis: RedisConfig{}, Keys: map[string]string{}, diff --git a/pkg/rtc/mediaengine.go b/pkg/rtc/mediaengine.go new file mode 100644 index 000000000..5647824b8 --- /dev/null +++ b/pkg/rtc/mediaengine.go @@ -0,0 +1,88 @@ +package rtc + +import ( + "github.com/pion/sdp/v3" + "github.com/pion/webrtc/v3" +) + +const ( + repairedRTP = "urn:ietf:params:rtp-hdrext:sdes:repaired-rtp-stream-id" + frameMarking = "urn:ietf:params:rtp-hdrext:framemarking" +) + +func createMediaEngine() (*webrtc.MediaEngine, error) { + me := &webrtc.MediaEngine{} + if err := me.RegisterCodec(webrtc.RTPCodecParameters{ + RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeOpus, ClockRate: 48000, Channels: 2, SDPFmtpLine: "minptime=10;useinbandfec=1", RTCPFeedback: nil}, + PayloadType: 111, + }, webrtc.RTPCodecTypeAudio); err != nil { + return nil, err + } + + videoRTCPFeedback := []webrtc.RTCPFeedback{ + {webrtc.TypeRTCPFBGoogREMB, ""}, + //{webrtc.TypeRTCPFBCCM, "fir"}, + {webrtc.TypeRTCPFBNACK, ""}, + {webrtc.TypeRTCPFBNACK, "pli"}} + for _, codec := range []webrtc.RTPCodecParameters{ + { + RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeVP8, ClockRate: 90000, RTCPFeedback: videoRTCPFeedback}, + PayloadType: 96, + }, + { + RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeVP9, ClockRate: 90000, SDPFmtpLine: "profile-id=0", RTCPFeedback: videoRTCPFeedback}, + PayloadType: 98, + }, + { + RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeVP9, ClockRate: 90000, SDPFmtpLine: "profile-id=1", RTCPFeedback: videoRTCPFeedback}, + PayloadType: 100, + }, + { + RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeH264, ClockRate: 90000, SDPFmtpLine: "level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=42001f", RTCPFeedback: videoRTCPFeedback}, + PayloadType: 102, + }, + { + RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeH264, ClockRate: 90000, SDPFmtpLine: "level-asymmetry-allowed=1;packetization-mode=0;profile-level-id=42001f", RTCPFeedback: videoRTCPFeedback}, + PayloadType: 127, + }, + { + RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeH264, ClockRate: 90000, SDPFmtpLine: "level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=42e01f", RTCPFeedback: videoRTCPFeedback}, + PayloadType: 125, + }, + { + RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeH264, ClockRate: 90000, SDPFmtpLine: "level-asymmetry-allowed=1;packetization-mode=0;profile-level-id=42e01f", RTCPFeedback: videoRTCPFeedback}, + PayloadType: 108, + }, + { + RTPCodecCapability: webrtc.RTPCodecCapability{MimeType: webrtc.MimeTypeH264, ClockRate: 90000, SDPFmtpLine: "level-asymmetry-allowed=1;packetization-mode=1;profile-level-id=640032", RTCPFeedback: videoRTCPFeedback}, + PayloadType: 123, + }, + } { + if err := me.RegisterCodec(codec, webrtc.RTPCodecTypeVideo); err != nil { + return nil, err + } + } + + for _, extension := range []string{ + sdp.SDESMidURI, + sdp.SDESRTPStreamIDURI, + repairedRTP, + //sdp.TransportCCURI, + frameMarking, + } { + if err := me.RegisterHeaderExtension(webrtc.RTPHeaderExtensionCapability{URI: extension}, webrtc.RTPCodecTypeVideo); err != nil { + return nil, err + } + } + for _, extension := range []string{ + sdp.SDESMidURI, + sdp.SDESRTPStreamIDURI, + sdp.AudioLevelURI, + } { + if err := me.RegisterHeaderExtension(webrtc.RTPHeaderExtensionCapability{URI: extension}, webrtc.RTPCodecTypeAudio); err != nil { + return nil, err + } + } + + return me, nil +} diff --git a/pkg/rtc/mediatrack.go b/pkg/rtc/mediatrack.go index b685af786..cca0445b6 100644 --- a/pkg/rtc/mediatrack.go +++ b/pkg/rtc/mediatrack.go @@ -21,7 +21,6 @@ var ( maxPLIFrequency = 1 * time.Second feedbackTypes = []webrtc.RTCPFeedback{ {webrtc.TypeRTCPFBGoogREMB, ""}, - {webrtc.TypeRTCPFBTransportCC, ""}, {webrtc.TypeRTCPFBNACK, ""}, {webrtc.TypeRTCPFBNACK, "pli"}} ) @@ -44,26 +43,26 @@ type MediaTrack struct { // channel to send RTCP packets to the source rtcpCh chan []rtcp.Packet lock sync.RWMutex - // map of target participantId -> DownTrack - downtracks map[string]*sfu.DownTrack - twcc *twcc.Responder - receiver sfu.Receiver + // map of target participantId -> *SubscribedTrack + subscribedTracks map[string]*SubscribedTrack + twcc *twcc.Responder + receiver sfu.Receiver //lastNack int64 lastPLI time.Time } func NewMediaTrack(trackId string, pId string, rtcpCh chan []rtcp.Packet, conf ReceiverConfig, track *webrtc.TrackRemote) *MediaTrack { t := &MediaTrack{ - id: trackId, - participantId: pId, - ssrc: track.SSRC(), - streamID: track.StreamID(), - kind: ToProtoTrackKind(track.Kind()), - codec: track.Codec(), - conf: conf, - rtcpCh: rtcpCh, - lock: sync.RWMutex{}, - downtracks: make(map[string]*sfu.DownTrack), + id: trackId, + participantId: pId, + ssrc: track.SSRC(), + streamID: track.StreamID(), + kind: ToProtoTrackKind(track.Kind()), + codec: track.Codec(), + conf: conf, + rtcpCh: rtcpCh, + lock: sync.RWMutex{}, + subscribedTracks: make(map[string]*SubscribedTrack), } return t @@ -89,13 +88,13 @@ func (t *MediaTrack) IsMuted() bool { } func (t *MediaTrack) SetMuted(muted bool) { - if !t.muted.TrySet(muted) { - return - } - // mute all of the downtracks + t.muted.TrySet(muted) + + // mute all of the subscribedtracks t.lock.RLock() - for _, dt := range t.downtracks { - dt.Mute(muted) + for id, st := range t.subscribedTracks { + logger.Debugw("setting muted", "dstParticipant", id, "muted", muted, "track", t.ID()) + st.SetPublisherMuted(muted) } t.lock.RUnlock() } @@ -106,20 +105,20 @@ func (t *MediaTrack) OnClose(f func()) { // subscribes participant to current remoteTrack // creates and add necessary forwarders and starts them -func (t *MediaTrack) AddSubscriber(participant types.Participant) error { - t.lock.RLock() - existingDt := t.downtracks[participant.ID()] - t.lock.RUnlock() +func (t *MediaTrack) AddSubscriber(sub types.Participant) error { + t.lock.Lock() + defer t.lock.Unlock() + existingSt := t.subscribedTracks[sub.ID()] // don't subscribe to the same track multiple times - if existingDt != nil { + if existingSt != nil { logger.Warnw("participant already subscribed to track", - "participant", participant.Identity(), - "track", existingDt.ID()) + "sub", sub.Identity(), + "track", t.ID()) return nil } - codec := t.codec + codec := t.receiver.Codec() // using DownTrack from ion-sfu downTrack, err := sfu.NewDownTrack(webrtc.RTPCodecCapability{ @@ -128,12 +127,13 @@ func (t *MediaTrack) AddSubscriber(participant types.Participant) error { Channels: codec.Channels, SDPFmtpLine: codec.SDPFmtpLine, RTCPFeedback: feedbackTypes, - }, t.receiver, t.participantId) + }, t.receiver, sub.ID()) if err != nil { return err } + subTrack := NewSubscribedTrack(downTrack) - transceiver, err := participant.PeerConnection().AddTransceiverFromTrack(downTrack, webrtc.RTPTransceiverInit{ + transceiver, err := sub.PeerConnection().AddTransceiverFromTrack(downTrack, webrtc.RTPTransceiverInit{ Direction: webrtc.RTPTransceiverDirectionSendrecv, }) if err != nil { @@ -143,21 +143,21 @@ func (t *MediaTrack) AddSubscriber(participant types.Participant) error { downTrack.SetTransceiver(transceiver) // when outtrack is bound, start loop to send reports downTrack.OnBind(func() { - downTrack.Mute(t.muted.Get()) - t.sendDownTrackBindingReports(participant.ID(), participant.RTCPChan()) + subTrack.SetPublisherMuted(t.IsMuted()) + t.sendDownTrackBindingReports(sub.ID(), sub.RTCPChan()) }) downTrack.OnCloseHandler(func() { t.lock.Lock() - delete(t.downtracks, participant.ID()) + delete(t.subscribedTracks, sub.ID()) t.lock.Unlock() - // ignore if the subscribing participant is not connected - if participant.PeerConnection().ConnectionState() == webrtc.PeerConnectionStateClosed { + // ignore if the subscribing sub is not connected + if sub.PeerConnection().ConnectionState() == webrtc.PeerConnectionStateClosed { return } - // if the source has been terminated, we'll need to terminate all of the downtracks - // however, if the dest participant has disconnected, then we can skip + // if the source has been terminated, we'll need to terminate all of the subscribedtracks + // however, if the dest sub has disconnected, then we can skip sender := transceiver.Sender() if sender == nil { return @@ -165,27 +165,25 @@ func (t *MediaTrack) AddSubscriber(participant types.Participant) error { logger.Debugw("removing peerconnection track", "track", t.id, "participantId", t.participantId, - "destParticipant", participant.Identity()) - if err := participant.PeerConnection().RemoveTrack(sender); err != nil { + "destParticipant", sub.Identity()) + if err := sub.PeerConnection().RemoveTrack(sender); err != nil { if err == webrtc.ErrConnectionClosed { - // participant closing, can skip removing downtracks + // sub closing, can skip removing subscribedtracks return } if _, ok := err.(*rtcerr.InvalidStateError); !ok { logger.Warnw("could not remove remoteTrack from forwarder", - "participant", participant.Identity(), + "sub", sub.Identity(), "err", err) } } - participant.RemoveDownTrack(t.streamID, downTrack) + sub.RemoveSubscribedTrack(t.participantId, subTrack) }) - t.lock.Lock() - t.downtracks[participant.ID()] = downTrack - t.lock.Unlock() + t.subscribedTracks[sub.ID()] = subTrack - participant.AddDownTrack(t.streamID, downTrack) + sub.AddSubscribedTrack(t.participantId, subTrack) t.receiver.AddDownTrack(downTrack, true) return nil @@ -203,14 +201,14 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track *webrtc.Tra if t.Kind() == livekit.TrackType_AUDIO { // TODO: audio level stuff } else if t.Kind() == livekit.TrackType_VIDEO { - if t.twcc == nil { - t.twcc = twcc.NewTransportWideCCResponder(uint32(track.SSRC())) - t.twcc.OnFeedback(func(p rtcp.RawPacket) { - t.rtcpCh <- []rtcp.Packet{&p} - }) - } + //if t.twcc == nil { + // t.twcc = twcc.NewTransportWideCCResponder(uint32(track.SSRC())) + // t.twcc.OnFeedback(func(p rtcp.RawPacket) { + // t.rtcpCh <- []rtcp.Packet{&p} + // }) + //} buff.OnTransportWideCC(func(sn uint16, timeNS int64, marker bool) { - t.twcc.Push(sn, timeNS, marker) + //t.twcc.Push(sn, timeNS, marker) }) } @@ -231,21 +229,21 @@ func (t *MediaTrack) AddReceiver(receiver *webrtc.RTPReceiver, track *webrtc.Tra } }) + t.lock.Lock() + defer t.lock.Unlock() if t.receiver == nil { // pack ID to identify all publishedTracks packedId := PackTrackId(t.participantId, track.ID()) - t.lock.Lock() t.receiver = NewWrappedReceiver(sfu.NewWebRTCReceiver(receiver, track, t.participantId), packedId) - t.lock.Unlock() t.receiver.SetRTCPCh(t.rtcpCh) t.receiver.OnCloseHandler(func() { + t.lock.Lock() + defer t.lock.Unlock() // source track closed if t.Kind() == livekit.TrackType_AUDIO { // TODO: remove audio level observer } - t.lock.Lock() t.receiver = nil - t.lock.Unlock() }) } t.receiver.AddUpTrack(track, buff) @@ -262,8 +260,8 @@ func (t *MediaTrack) RemoveSubscriber(participantId string) { t.lock.RLock() defer t.lock.RUnlock() - if dt := t.downtracks[participantId]; dt != nil { - go dt.Close() + if subTrack := t.subscribedTracks[participantId]; subTrack != nil { + go subTrack.DownTrack().Close() } } @@ -271,20 +269,24 @@ func (t *MediaTrack) RemoveAllSubscribers() { logger.Debugw("removing all subscribers", "track", t.id) t.lock.RLock() defer t.lock.RUnlock() - for _, dt := range t.downtracks { - go dt.Close() + for _, subTrack := range t.subscribedTracks { + go subTrack.DownTrack().Close() } - t.downtracks = make(map[string]*sfu.DownTrack) + t.subscribedTracks = make(map[string]*SubscribedTrack) } func (t *MediaTrack) sendDownTrackBindingReports(participantId string, rtcpCh chan []rtcp.Packet) { var sd []rtcp.SourceDescriptionChunk t.lock.RLock() - dt := t.downtracks[participantId] + subTrack := t.subscribedTracks[participantId] t.lock.RUnlock() - chunks := dt.CreateSourceDescriptionChunks() + if subTrack == nil { + return + } + + chunks := subTrack.DownTrack().CreateSourceDescriptionChunks() if chunks == nil { return } diff --git a/pkg/rtc/participant.go b/pkg/rtc/participant.go index 77b6cf554..b12ac767a 100644 --- a/pkg/rtc/participant.go +++ b/pkg/rtc/participant.go @@ -7,7 +7,6 @@ import ( "time" "github.com/bep/debounce" - "github.com/pion/ion-sfu/pkg/sfu" "github.com/pion/rtcp" "github.com/pion/webrtc/v3" "github.com/pkg/errors" @@ -33,16 +32,17 @@ const ( ) type ParticipantImpl struct { - id string - peerConn types.PeerConnection - responseSink routing.MessageSink - receiverConfig ReceiverConfig - isClosed utils.AtomicFlag - mediaEngine *webrtc.MediaEngine - identity string - state atomic.Value // livekit.ParticipantInfo_State - rtcpCh chan []rtcp.Packet - subscribedTracks map[string][]*sfu.DownTrack + id string + peerConn types.PeerConnection + responseSink routing.MessageSink + receiverConfig ReceiverConfig + isClosed utils.AtomicFlag + mediaEngine *webrtc.MediaEngine + identity string + state atomic.Value // livekit.ParticipantInfo_State + rtcpCh chan []rtcp.Packet + // tracks the current participant is subscribed to, map of otherParticipantId => []DownTrack + subscribedTracks map[string][]types.SubscribedTrack // publishedTracks that participant is publishing publishedTracks map[string]types.PublishedTrack // client intended to publish, yet to be reconciled @@ -64,8 +64,10 @@ type ParticipantImpl struct { } func NewPeerConnection(conf *WebRTCConfig) (*webrtc.PeerConnection, error) { - me := &webrtc.MediaEngine{} - me.RegisterDefaultCodecs() + me, err := createMediaEngine() + if err != nil { + return nil, err + } se := conf.SettingEngine se.BufferFactory = bufferFactory.GetOrNew @@ -76,8 +78,6 @@ func NewPeerConnection(conf *WebRTCConfig) (*webrtc.PeerConnection, error) { func NewParticipant(identity string, pc types.PeerConnection, rs routing.MessageSink, receiverConfig ReceiverConfig) (*ParticipantImpl, error) { // TODO: check to ensure params are valid, id and identity can't be empty - me := &webrtc.MediaEngine{} - me.RegisterDefaultCodecs() participant := &ParticipantImpl{ id: utils.NewGuid(utils.ParticipantPrefix), @@ -86,12 +86,11 @@ func NewParticipant(identity string, pc types.PeerConnection, rs routing.Message responseSink: rs, receiverConfig: receiverConfig, rtcpCh: make(chan []rtcp.Packet, 50), - subscribedTracks: make(map[string][]*sfu.DownTrack), + subscribedTracks: make(map[string][]types.SubscribedTrack), lock: sync.RWMutex{}, negotiationCond: sync.NewCond(&sync.Mutex{}), publishedTracks: make(map[string]types.PublishedTrack, 0), pendingTracks: make(map[string]*livekit.TrackInfo), - mediaEngine: me, debouncedNegotiate: debounce.New(negotiationFrequency), } participant.state.Store(livekit.ParticipantInfo_JOINING) @@ -249,7 +248,7 @@ func (p *ParticipantImpl) Answer(sdp webrtc.SessionDescription) (answer webrtc.S logger.Debugw("sending answer to client", "participant", p.Identity(), - //"sdp", sdp.SDP, + //"sdp", sdp.SDP, ) err = p.responseSink.WriteMessage(&livekit.SignalResponse{ Message: &livekit.SignalResponse_Answer{ @@ -300,7 +299,7 @@ func (p *ParticipantImpl) HandleAnswer(sdp webrtc.SessionDescription) error { } logger.Debugw("setting participant answer", "participant", p.Identity(), - //"sdp", sdp.SDP, + //"sdp", sdp.SDP, ) if err := p.peerConn.SetRemoteDescription(sdp); err != nil { return errors.Wrap(err, "could not set remote description") @@ -456,23 +455,22 @@ func (p *ParticipantImpl) PeerConnection() types.PeerConnection { return p.peerConn } -func (p *ParticipantImpl) AddDownTrack(streamId string, dt *sfu.DownTrack) { +// add a track to the participant's subscribed list +func (p *ParticipantImpl) AddSubscribedTrack(pubId string, subTrack types.SubscribedTrack) { + logger.Debugw("added subscribedTrack", "srcParticipant", pubId, + "participant", p.Identity()) p.lock.Lock() - p.subscribedTracks[streamId] = append(p.subscribedTracks[streamId], dt) + p.subscribedTracks[pubId] = append(p.subscribedTracks[pubId], subTrack) p.lock.Unlock() } -func (p *ParticipantImpl) RemoveDownTrack(streamId string, dt *sfu.DownTrack) { +// remove a track to the participant's subscribed list +func (p *ParticipantImpl) RemoveSubscribedTrack(pubId string, subTrack types.SubscribedTrack) { + logger.Debugw("removed subscribedTrack", "srcParticipant", pubId, + "participant", p.Identity()) p.lock.Lock() defer p.lock.Unlock() - tracks := p.subscribedTracks[streamId] - newTracks := make([]*sfu.DownTrack, 0, len(tracks)) - for _, track := range tracks { - if track != dt { - newTracks = append(newTracks, track) - } - } - p.subscribedTracks[streamId] = newTracks + p.subscribedTracks[pubId] = funk.Without(p.subscribedTracks[pubId], subTrack).([]types.SubscribedTrack) } func (p *ParticipantImpl) scheduleNegotiate() { @@ -563,6 +561,7 @@ func (p *ParticipantImpl) onMediaTrack(track *webrtc.TrackRemote, rtpReceiver *w var mt *MediaTrack var newTrack bool if trk, ok := ptrack.(*MediaTrack); ok { + logger.Debugw("using existing mediatrack, simulcast", "rid", track.RID()) mt = trk } else { mt = NewMediaTrack(ti.Sid, p.id, p.rtcpCh, p.receiverConfig, track) @@ -575,6 +574,16 @@ func (p *ParticipantImpl) onMediaTrack(track *webrtc.TrackRemote, rtpReceiver *w if newTrack { p.handleTrackPublished(mt) } + + // TODO: video tracks the current participant is subscribed to tends to freeze when the participant adds tracks + // to get around this, we'll trigger a resync on all tracks it's subscribed to + p.lock.RLock() + defer p.lock.RUnlock() + for _, tracks := range p.subscribedTracks { + for _, subTrack := range tracks { + subTrack.Resync() + } + } } func (p *ParticipantImpl) onDataChannel(dc *webrtc.DataChannel) { @@ -655,14 +664,14 @@ func (p *ParticipantImpl) downTracksRTCPWorker() { var pkts []rtcp.Packet var sd []rtcp.SourceDescriptionChunk p.lock.RLock() - for _, dts := range p.subscribedTracks { - for _, dt := range dts { - sr := dt.CreateSenderReport() - chunks := dt.CreateSourceDescriptionChunks() + for _, tracks := range p.subscribedTracks { + for _, subTrack := range tracks { + sr := subTrack.DownTrack().CreateSenderReport() + chunks := subTrack.DownTrack().CreateSourceDescriptionChunks() if sr == nil || chunks == nil { continue } - pkts = append(pkts, dt.CreateSenderReport()) + pkts = append(pkts, sr) sd = append(sd, chunks...) } } diff --git a/pkg/rtc/subscribedtrack.go b/pkg/rtc/subscribedtrack.go new file mode 100644 index 000000000..c5eb1dbc9 --- /dev/null +++ b/pkg/rtc/subscribedtrack.go @@ -0,0 +1,52 @@ +package rtc + +import ( + "github.com/pion/ion-sfu/pkg/sfu" + + "github.com/livekit/livekit-server/pkg/utils" +) + +type SubscribedTrack struct { + dt *sfu.DownTrack + subMuted utils.AtomicFlag + pubMuted utils.AtomicFlag +} + +func NewSubscribedTrack(dt *sfu.DownTrack) *SubscribedTrack { + return &SubscribedTrack{ + dt: dt, + } +} + +func (t *SubscribedTrack) DownTrack() *sfu.DownTrack { + return t.dt +} + +// has subscriber indicated it wants to mute this track +func (t *SubscribedTrack) IsMuted() bool { + return t.subMuted.Get() +} + +// set subscriber mute preference +func (t *SubscribedTrack) SetMuted(muted bool) { + t.subMuted.TrySet(muted) + t.updateDownTrackMute() +} + +func (t *SubscribedTrack) SetPublisherMuted(muted bool) { + t.pubMuted.TrySet(muted) + t.updateDownTrackMute() +} + +// cause downtrack to resync and request a keyframe +// this is sort of a hack to prevent a subscriber from having tracks freeze on the client due to missing +// keyframes +func (t *SubscribedTrack) Resync() { + t.dt.Mute(true) + t.updateDownTrackMute() +} + +func (t *SubscribedTrack) updateDownTrackMute() { + muted := t.subMuted.Get() || t.pubMuted.Get() + t.dt.Mute(muted) +} diff --git a/pkg/rtc/types/interfaces.go b/pkg/rtc/types/interfaces.go index 6d6ca31d0..e06b0e8b3 100644 --- a/pkg/rtc/types/interfaces.go +++ b/pkg/rtc/types/interfaces.go @@ -3,10 +3,8 @@ package types import ( "time" - "github.com/pion/ion-sfu/pkg/buffer" "github.com/pion/ion-sfu/pkg/sfu" "github.com/pion/rtcp" - "github.com/pion/rtp" "github.com/pion/webrtc/v3" "github.com/livekit/livekit-server/pkg/routing" @@ -82,8 +80,8 @@ type Participant interface { OnClose(func(Participant)) // package methods - AddDownTrack(streamId string, dt *sfu.DownTrack) - RemoveDownTrack(streamId string, dt *sfu.DownTrack) + AddSubscribedTrack(participantId string, st SubscribedTrack) + RemoveSubscribedTrack(participantId string, st SubscribedTrack) PeerConnection() PeerConnection } @@ -105,28 +103,13 @@ type PublishedTrack interface { OnClose(func()) } -//counterfeiter:generate . Receiver -type Receiver interface { - RTPChan() <-chan buffer.ExtPacket - GetBufferedPacket(pktBuf []byte, sn uint16, snOffset uint16) (rtp.Packet, error) -} - -// DownTrack publishes data to a target participant -// using this interface to make testing more practical -//counterfeiter:generate . DownTrack -type DownTrack interface { - ID() string - WriteRTP(p rtp.Packet) error - IsBound() bool - Close() - OnCloseHandler(fn func()) - OnBind(fn func()) - SSRC() uint32 - LastSSRC() uint32 - SnOffset() uint16 - TsOffset() uint32 - GetNACKSeqNo(seqNo []uint16) []uint16 - CreateSourceDescriptionChunks() []rtcp.SourceDescriptionChunk +//counterfeiter:generate . SubscribedTrack +type SubscribedTrack interface { + DownTrack() *sfu.DownTrack + IsMuted() bool + SetMuted(muted bool) + SetPublisherMuted(muted bool) + Resync() } // interface for properties of webrtc.TrackRemote diff --git a/pkg/rtc/types/typesfakes/fake_participant.go b/pkg/rtc/types/typesfakes/fake_participant.go index d569d6154..34d1ad8e7 100644 --- a/pkg/rtc/types/typesfakes/fake_participant.go +++ b/pkg/rtc/types/typesfakes/fake_participant.go @@ -244,14 +244,14 @@ type FakeParticipant struct { invocationsMutex sync.RWMutex } -func (fake *FakeParticipant) AddDownTrack(arg1 string, arg2 *sfu.DownTrack) { +func (fake *FakeParticipant) AddSubscribedTrack(arg1 string, arg2 *sfu.DownTrack) { fake.addDownTrackMutex.Lock() fake.addDownTrackArgsForCall = append(fake.addDownTrackArgsForCall, struct { arg1 string arg2 *sfu.DownTrack }{arg1, arg2}) stub := fake.AddDownTrackStub - fake.recordInvocation("AddDownTrack", []interface{}{arg1, arg2}) + fake.recordInvocation("AddSubscribedTrack", []interface{}{arg1, arg2}) fake.addDownTrackMutex.Unlock() if stub != nil { fake.AddDownTrackStub(arg1, arg2) @@ -1120,7 +1120,7 @@ func (fake *FakeParticipant) RemoveDownTrack(arg1 string, arg2 *sfu.DownTrack) { arg2 *sfu.DownTrack }{arg1, arg2}) stub := fake.RemoveDownTrackStub - fake.recordInvocation("RemoveDownTrack", []interface{}{arg1, arg2}) + fake.recordInvocation("RemoveSubscribedTrack", []interface{}{arg1, arg2}) fake.removeDownTrackMutex.Unlock() if stub != nil { fake.RemoveDownTrackStub(arg1, arg2)