From 6c039bc64f2b638728c5c447f7327fe368afe3ff Mon Sep 17 00:00:00 2001 From: David Zhao Date: Fri, 23 Oct 2020 23:01:35 -0700 Subject: [PATCH] receiver class, imported relevant SFU helpers --- go.mod | 3 +- go.sum | 8 +- pkg/rtc/config.go | 10 +- pkg/rtc/mediaengine.go | 16 +- pkg/rtc/peer.go | 21 +- pkg/rtc/receiver.go | 120 ++++++++++- pkg/rtc/track.go | 16 +- pkg/sfu/README.md | 5 + pkg/sfu/buffer.go | 442 +++++++++++++++++++++++++++++++++++++++++ pkg/sfu/errors.go | 16 ++ pkg/sfu/helpers.go | 134 +++++++++++++ pkg/sfu/packetqueue.go | 124 ++++++++++++ 12 files changed, 890 insertions(+), 25 deletions(-) create mode 100644 pkg/sfu/README.md create mode 100644 pkg/sfu/buffer.go create mode 100644 pkg/sfu/errors.go create mode 100644 pkg/sfu/helpers.go create mode 100644 pkg/sfu/packetqueue.go diff --git a/go.mod b/go.mod index 7ea6e1320..a7dad69d4 100644 --- a/go.mod +++ b/go.mod @@ -3,10 +3,12 @@ module github.com/livekit/livekit-server go 1.15 require ( + github.com/gammazero/deque v0.0.0-20201010052221-3932da5530cc github.com/golang/protobuf v1.4.3 github.com/google/uuid v1.1.2 github.com/google/wire v0.4.0 github.com/pion/ion-sfu v1.0.27 + github.com/pion/rtcp v1.2.4 github.com/pion/rtp v1.6.1 github.com/pion/sdp/v3 v3.0.2 github.com/pion/stun v0.3.5 @@ -15,6 +17,5 @@ require ( github.com/urfave/cli/v2 v2.2.0 github.com/urfave/negroni v1.0.0 go.uber.org/zap v1.16.0 - google.golang.org/protobuf v1.25.0 // indirect gopkg.in/yaml.v2 v2.3.0 ) diff --git a/go.sum b/go.sum index a4f58074c..682171127 100644 --- a/go.sum +++ b/go.sum @@ -171,6 +171,7 @@ github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE= github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= github.com/lucas-clemente/quic-go v0.18.0 h1:JhQDdqxdwdmGdKsKgXi1+coHRoGhvU6z0rNzOJqZ/4o= github.com/lucas-clemente/quic-go v0.18.0/go.mod h1:yXttHsSNxQi8AWijC/vLP+OJczXqzHSOcJrM5ITUlCg= +github.com/lucas-clemente/quic-go v0.18.1 h1:DMR7guC0NtVS8zNZR3IO7NARZvZygkSC56GGtC6cyys= github.com/lucas-clemente/quic-go v0.18.1/go.mod h1:yXttHsSNxQi8AWijC/vLP+OJczXqzHSOcJrM5ITUlCg= github.com/lucsky/cuid v1.0.2 h1:z4XlExeoderxoPj2/dxKOyPxe9RCOu7yNq9/XWxIUMQ= github.com/lucsky/cuid v1.0.2/go.mod h1:QaaJqckboimOmhRSJXSx/+IT+VTfxfPGSo/6mfgUfmE= @@ -182,6 +183,7 @@ github.com/marten-seemann/qtls v0.10.0 h1:ECsuYUKalRL240rRD4Ri33ISb7kAQ3qGDlrrl5 github.com/marten-seemann/qtls v0.10.0/go.mod h1:UvMd1oaYDACI99/oZUYLzMCkBXQVT0aGm99sJhbT8hs= github.com/marten-seemann/qtls-go1-15 v0.1.0 h1:i/YPXVxz8q9umso/5y474CNcHmTpA+5DH+mFPjx6PZg= github.com/marten-seemann/qtls-go1-15 v0.1.0/go.mod h1:GyFwywLKkRt+6mfU99csTEY1joMZz5vmB1WNZH3P81I= +github.com/marten-seemann/qtls-go1-15 v0.1.1 h1:LIH6K34bPVttyXnUWixk0bzH6/N07VxbSabxn5A5gZQ= github.com/marten-seemann/qtls-go1-15 v0.1.1/go.mod h1:GyFwywLKkRt+6mfU99csTEY1joMZz5vmB1WNZH3P81I= github.com/mattn/go-colorable v0.0.9/go.mod h1:9vuHe8Xs5qXnSaW/c/ABM9alt+Vo+STaOChaDxuIBZU= github.com/mattn/go-isatty v0.0.3/go.mod h1:M+lRXTBqGeGNdLjl/ufCoiOlB5xdOkqRJdNxMWT7Zi4= @@ -217,8 +219,6 @@ github.com/pion/datachannel v1.4.21 h1:3ZvhNyfmxsAqltQrApLPQMhSFNA+aT87RqyCq4OXm github.com/pion/datachannel v1.4.21/go.mod h1:oiNyP4gHx2DIwRzX/MFyH0Rz/Gz05OgBlayAI2hAWjg= github.com/pion/dtls/v2 v2.0.3 h1:3qQ0s4+TXD00rsllL8g8KQcxAs+Y/Z6oz618RXX6p14= github.com/pion/dtls/v2 v2.0.3/go.mod h1:TUjyL8bf8LH95h81Xj7kATmzMRt29F/4lxpIPj2Xe4Y= -github.com/pion/ice/v2 v2.0.7 h1:3Iv+EfthCdPbJCsPbhnL4xRzolR1oh9mJTuUsxyxPJI= -github.com/pion/ice/v2 v2.0.7/go.mod h1:0C4VBZxkO3A7Qj5ZEgZut8OptbJFSbuGNyPyJMftuHA= github.com/pion/ice/v2 v2.0.9 h1:oHbiN6Q9tgb8Gfu3I4cbr5mHRE1uqiuFABQ8CbWjIyk= github.com/pion/ice/v2 v2.0.9/go.mod h1:NK+o39ynb+N1YSj9fPgWs3vjVcrsWw0KCr/311MqVq8= github.com/pion/ion-sfu v1.0.27 h1:VR0LNI6Gno5rKUuhUoiESLGkkyd4u1QQtzg0SSUt5F8= @@ -254,8 +254,6 @@ github.com/pion/turn/v2 v2.0.5 h1:iwMHqDfPEDEOFzwWKT56eFmh6DYC6o/+xnLAEzgISbA= github.com/pion/turn/v2 v2.0.5/go.mod h1:APg43CFyt/14Uy7heYUOGWdkem/Wu4PhCO/bjyrTqMw= github.com/pion/udp v0.1.0 h1:uGxQsNyrqG3GLINv36Ff60covYmfrLoxzwnCsIYspXI= github.com/pion/udp v0.1.0/go.mod h1:BPELIjbwE9PRbd/zxI/KYBnbo7B6+oA6YuEaNE8lths= -github.com/pion/webrtc/v3 v3.0.0-beta.7 h1:wDzex7YkT3wErRGaW4yOs9ziCQeuVF04poWVTqQ8o/Q= -github.com/pion/webrtc/v3 v3.0.0-beta.7/go.mod h1:g2FHuaaKKuoZ9RlKIDAYvnO/rhocYG/d+2Ly1hRa85o= github.com/pion/webrtc/v3 v3.0.0-beta.9 h1:Vwep6pua7wJ5lSnR78KlSu0N3u9ZCBGfD65r5FwO35k= github.com/pion/webrtc/v3 v3.0.0-beta.9/go.mod h1:GlriYYHJ5KkNsCunm3oFDPql4TDTrrNoI9iSWWSnafA= github.com/pkg/errors v0.8.0/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= @@ -470,6 +468,7 @@ golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.1-0.20180807135948-17ff2d5776d2/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.2 h1:tW2bmiBqwgJj/UpqtC8EpXEZVYOwU0yG4iWbprSVAcs= golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk= +golang.org/x/text v0.3.3 h1:cokOdA+Jmi5PJGXLlLllQSgYigAEfHXJAERHVMaCc2k= golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= golang.org/x/time v0.0.0-20180412165947-fbb02b2291d2/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ= golang.org/x/time v0.0.0-20181108054448-85acf8d2951c/go.mod h1:tRJNPiyCQ0inRvYxbN9jk5I+vvW/OXSQhTDSoE431IQ= @@ -498,6 +497,7 @@ golang.org/x/tools v0.0.0-20191012152004-8de300cfc20a/go.mod h1:b+2E5dAYhXwXZwtn golang.org/x/tools v0.0.0-20191029041327-9cc4af7d6b2c/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= golang.org/x/tools v0.0.0-20191029190741-b9c20aec41a5 h1:hKsoRgsbwY1NafxrwTs+k64bikrLBkAgPir1TNCj3Zs= golang.org/x/tools v0.0.0-20191029190741-b9c20aec41a5/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= +golang.org/x/tools v0.0.0-20191112195655-aa38f8e97acc h1:NCy3Ohtk6Iny5V/reW2Ktypo4zIpWBdRJ1uFMjBxdg8= golang.org/x/tools v0.0.0-20191112195655-aa38f8e97acc/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= diff --git a/pkg/rtc/config.go b/pkg/rtc/config.go index b8a6be73c..263d115db 100644 --- a/pkg/rtc/config.go +++ b/pkg/rtc/config.go @@ -15,6 +15,10 @@ type WebRTCConfig struct { setting webrtc.SettingEngine feedbackTypes []webrtc.RTCPFeedback + receiver ReceiverConfig +} + +type ReceiverConfig struct { maxBandwidth uint64 maxBufferTime int } @@ -67,7 +71,9 @@ func NewWebRTCConfig(conf *config.RTCConfig, externalIP string) (*WebRTCConfig, configuration: c, setting: s, feedbackTypes: ft, - maxBandwidth: conf.MaxBandwidth, - maxBufferTime: conf.MaxBufferTime, + receiver: ReceiverConfig{ + maxBandwidth: conf.MaxBandwidth, + maxBufferTime: conf.MaxBufferTime, + }, }, nil } diff --git a/pkg/rtc/mediaengine.go b/pkg/rtc/mediaengine.go index a918d387b..b14350ff6 100644 --- a/pkg/rtc/mediaengine.go +++ b/pkg/rtc/mediaengine.go @@ -18,6 +18,7 @@ const ( type MediaEngine struct { webrtc.MediaEngine feedbackTypes []webrtc.RTCPFeedback + tCCExt int } // PopulateFromSDP finds all codecs in sd and adds them to m, using the dynamic @@ -27,16 +28,23 @@ type MediaEngine struct { // PopulateFromSDP allows an answerer to properly match the PayloadTypes from the offerer. // A MediaEngine populated by PopulateFromSDP should be used only for a single session. func (e *MediaEngine) PopulateFromSDP(sd webrtc.SessionDescription) error { - sdp := sdp.SessionDescription{} - if err := sdp.Unmarshal([]byte(sd.SDP)); err != nil { + s := sdp.SessionDescription{} + if err := s.Unmarshal([]byte(sd.SDP)); err != nil { return err } - for _, md := range sdp.MediaDescriptions { + for _, md := range s.MediaDescriptions { if md.MediaName.Media != mediaNameAudio && md.MediaName.Media != mediaNameVideo { continue } + for _, att := range md.Attributes { + if att.Key == sdp.AttrKeyExtMap && strings.HasSuffix(att.Value, sdp.TransportCCURI) { + e.tCCExt, _ = strconv.Atoi(att.Value[:1]) + break + } + } + for _, format := range md.MediaName.Formats { pt, err := strconv.Atoi(format) if err != nil { @@ -44,7 +52,7 @@ func (e *MediaEngine) PopulateFromSDP(sd webrtc.SessionDescription) error { } payloadType := uint8(pt) - payloadCodec, err := sdp.GetCodecForPayloadType(payloadType) + payloadCodec, err := s.GetCodecForPayloadType(payloadType) if err != nil { return fmt.Errorf("could not find codec for payload type %d", payloadType) } diff --git a/pkg/rtc/peer.go b/pkg/rtc/peer.go index 1b96c7f45..128c0eae9 100644 --- a/pkg/rtc/peer.go +++ b/pkg/rtc/peer.go @@ -7,19 +7,19 @@ import ( ) type WebRTCPeer struct { + id string conn *webrtc.PeerConnection ctx context.Context cancelFunc context.CancelFunc mediaEngine MediaEngine - maxBandwidth uint64 - maxBufferTime int + receiverConfig ReceiverConfig // callbacks & handlers onPeerTrack func(*PeerTrack) } -func NewWebRTCPeer(me MediaEngine, conf WebRTCConfig) (*WebRTCPeer, error) { +func NewWebRTCPeer(id string, me MediaEngine, conf WebRTCConfig) (*WebRTCPeer, error) { api := webrtc.NewAPI(webrtc.WithMediaEngine(me.MediaEngine), webrtc.WithSettingEngine(conf.setting)) pc, err := api.NewPeerConnection(conf.configuration) @@ -29,12 +29,12 @@ func NewWebRTCPeer(me MediaEngine, conf WebRTCConfig) (*WebRTCPeer, error) { ctx, cancel := context.WithCancel(context.Background()) peer := &WebRTCPeer{ - conn: pc, - ctx: ctx, - cancelFunc: cancel, - mediaEngine: me, - maxBandwidth: conf.maxBandwidth, - maxBufferTime: conf.maxBufferTime, + id: id, + conn: pc, + ctx: ctx, + cancelFunc: cancel, + mediaEngine: me, + receiverConfig: conf.receiver, } pc.OnTrack(peer.onTrack) @@ -44,5 +44,8 @@ func NewWebRTCPeer(me MediaEngine, conf WebRTCConfig) (*WebRTCPeer, error) { // when a new track is created, creates a PeerTrack and adds it to room func (p *WebRTCPeer) onTrack(track *webrtc.Track, receiver *webrtc.RTPReceiver) { + onPeerTrack := p.onPeerTrack + if onPeerTrack != nil { + } } diff --git a/pkg/rtc/receiver.go b/pkg/rtc/receiver.go index 450cdae15..18e57dd05 100644 --- a/pkg/rtc/receiver.go +++ b/pkg/rtc/receiver.go @@ -1,13 +1,129 @@ package rtc import ( - sfu "github.com/pion/ion-sfu/pkg" + "io" + + "github.com/pion/rtp" + + "github.com/livekit/livekit-server/pkg/sfu" + + "context" + + "github.com/pion/rtcp" "github.com/pion/webrtc/v3" ) +const ( + // TODO: could probably increase this depending on configuration/memory + maxChanSize = 1024 +) + // A receiver is responsible for pulling from a track type Receiver struct { - track *webrtc.Track + ctx context.Context + cancel context.CancelFunc rtpReceiver *webrtc.RTPReceiver + track *webrtc.Track buffer *sfu.Buffer + rtpChan chan *rtp.Packet + + onCloseHandler func(r *Receiver) +} + +func NewReceiver(ctx context.Context, rtpReceiver *webrtc.RTPReceiver, conf ReceiverConfig, me *MediaEngine) *Receiver { + ctx, cancel := context.WithCancel(ctx) + track := rtpReceiver.Track() + return &Receiver{ + ctx: ctx, + cancel: cancel, + rtpReceiver: rtpReceiver, + track: track, + rtpChan: make(chan *rtp.Packet, maxChanSize), + buffer: sfu.NewBuffer(track, sfu.BufferOptions{ + BufferTime: conf.maxBufferTime, + MaxBitRate: conf.maxBandwidth * 1000, + TCCExt: me.tCCExt, + }), + } +} + +// starts reading RTP and push to buffer +func (r *Receiver) Start() { + +} + +// close the stream +func (r *Receiver) Close() { + +} + +// reaturns channel to read rtp packets +func (r *Receiver) RTPChan() <-chan *rtp.Packet { + return r.rtpChan +} + +// Builds RTCP report from buffer +func (r *Receiver) BuildRTCP() (rtcp.ReceptionReport, []rtcp.Packet) { + return r.buffer.BuildRTCP() +} + +// WriteBufferedPacket writes buffered packet to track, return error if packet not found +func (r *Receiver) WriteBufferedPacket(sn uint16, track *webrtc.Track, snOffset uint16, tsOffset, ssrc uint32) error { + if r.buffer == nil || r.ctx.Err() != nil { + return nil + } + return r.buffer.WritePacket(sn, track, snOffset, tsOffset, ssrc) +} + +// rtpWorker reads RTP stream, fills buffer and channel +func (r *Receiver) rtpWorker() { + defer func() { + close(r.rtpChan) + if r.onCloseHandler != nil { + r.onCloseHandler(r) + } + }() + + for { + pkt, err := r.track.ReadRTP() + if err == io.EOF { + r.Close() + return + } + + if err != nil { + // log and continue + continue + } + + r.buffer.Push(pkt) + + select { + case <-r.ctx.Done(): + return + default: + r.rtpChan <- pkt + } + } +} + +// rtcpWorker reads RTCP messages from receiver, notifies buffer +func (r *Receiver) rtcpWorker() { + for { + pkts, err := r.rtpReceiver.ReadRTCP() + if err == io.ErrClosedPipe || r.ctx.Err() != nil { + return + } + if err != nil { + // TODO: log + continue + } + + for _, pkt := range pkts { + switch pkt := pkt.(type) { + case *rtcp.SenderReport: + r.buffer.SetSenderReportData(pkt.RTPTime, pkt.NTPTime) + } + } + } } diff --git a/pkg/rtc/track.go b/pkg/rtc/track.go index 0a596d5d1..28b2a8289 100644 --- a/pkg/rtc/track.go +++ b/pkg/rtc/track.go @@ -3,7 +3,6 @@ package rtc import ( "sync" - sfu "github.com/pion/ion-sfu/pkg" "github.com/pion/webrtc/v3" ) @@ -11,17 +10,28 @@ import ( * Peer track represents a track that needs to be forwarded */ type PeerTrack struct { - id string + id uint32 peerId string // source track track *webrtc.Track lock sync.RWMutex // map of target peerId -> forwarder forwarders map[string]Forwarder - receiver sfu.Receiver + receiver Receiver lastNack int64 } +func NewPeerTrack(peerId string, track *webrtc.Track, receiver Receiver) *PeerTrack { + return &PeerTrack{ + id: track.SSRC(), + peerId: peerId, + track: track, + lock: sync.RWMutex{}, + forwarders: make(map[string]Forwarder), + receiver: receiver, + } +} + // subscribes peer to current track // creates and add necessary forwarders and starts them func (t *PeerTrack) AddSubscriber(peer *WebRTCPeer) error { diff --git a/pkg/sfu/README.md b/pkg/sfu/README.md new file mode 100644 index 000000000..cc4b4542d --- /dev/null +++ b/pkg/sfu/README.md @@ -0,0 +1,5 @@ +This package is largely files from the wonderful ion-sfu project. + +https://github.com/pion/ion-sfu + +It's duplicated here since we needed to access a private method in the Buffer class \ No newline at end of file diff --git a/pkg/sfu/buffer.go b/pkg/sfu/buffer.go new file mode 100644 index 000000000..685488288 --- /dev/null +++ b/pkg/sfu/buffer.go @@ -0,0 +1,442 @@ +package sfu + +// file from ion-sfu project + +import ( + "math" + "sort" + "sync" + "time" + + "github.com/gammazero/deque" + "github.com/pion/rtcp" + "github.com/pion/rtp" + "github.com/pion/webrtc/v3" +) + +const ( + maxSN = 1 << 16 + // default buffer time by ms + defaultBufferTime = 1000 +) + +type rtpExtInfo struct { + ExtTSN uint32 + Timestamp int64 +} + +// Buffer contains all packets +type Buffer struct { + mu sync.RWMutex + + pktQueue queue + codecType webrtc.RTPCodecType + simulcast bool + clockRate uint32 + maxBitrate uint64 + + // supported feedbacks + remb bool + nack bool + tcc bool + + lastSRNTPTime uint64 + lastSRRTPTime uint32 + lastSRRecv int64 // Represents wall clock of the most recent sender report arrival + baseSN uint16 + cycles uint32 + lastExpected uint32 + lastReceived uint32 + lostRate float32 + ssrc uint32 + lastPacketTime int64 // Time the last RTP packet from this source was received + lastRtcpPacketTime int64 // Time the last RTCP packet was received. + lastRtcpSrTime int64 // Time the last RTCP SR was received. Required for DLSR computation. + packetCount uint32 // Number of packets received from this source. + lastTransit uint32 + maxSeqNo uint16 // The highest sequence number received in an RTP data packet + jitter float64 // An estimate of the statistical variance of the RTP data packet inter-arrival time. + totalByte uint64 + + // remb + rembSteps uint8 + + // transport-cc + tccExt uint8 + tccExtInfo []rtpExtInfo + tccCycles uint32 + tccLastExtSN uint32 + tccPktCtn uint8 + tccLastSn uint16 + lastExtInfo uint16 +} + +// BufferOptions provides configuration options for the buffer +type BufferOptions struct { + TCCExt int + BufferTime int + MaxBitRate uint64 +} + +// NewBuffer constructs a new Buffer +func NewBuffer(track *webrtc.Track, o BufferOptions) *Buffer { + b := &Buffer{ + ssrc: track.SSRC(), + clockRate: track.Codec().ClockRate, + codecType: track.Codec().Type, + maxBitrate: o.MaxBitRate, + simulcast: len(track.RID()) > 0, + rembSteps: 4, + } + if o.BufferTime <= 0 { + o.BufferTime = defaultBufferTime + } + b.pktQueue.duration = uint32(o.BufferTime) * b.clockRate / 1000 + b.pktQueue.ssrc = track.SSRC() + b.tccExt = uint8(o.TCCExt) + + for _, fb := range track.Codec().RTCPFeedback { + switch fb.Type { + case webrtc.TypeRTCPFBGoogREMB: + //log.Debugf("Setting feedback %s", webrtc.TypeRTCPFBGoogREMB) + b.remb = true + case webrtc.TypeRTCPFBTransportCC: + //log.Debugf("Setting feedback %s", webrtc.TypeRTCPFBTransportCC) + b.tccExtInfo = make([]rtpExtInfo, 1<<8) + b.tcc = true + case webrtc.TypeRTCPFBNACK: + //log.Debugf("Setting feedback %s", webrtc.TypeRTCPFBNACK) + b.nack = true + } + } + //log.Debugf("NewBuffer BufferOptions=%v", o) + return b +} + +// Push adds a RTP Packet, out of order, new packet may be arrived later +func (b *Buffer) Push(p *rtp.Packet) { + b.mu.Lock() + defer b.mu.Unlock() + b.totalByte += uint64(p.MarshalSize()) + if b.packetCount == 0 { + b.baseSN = p.SequenceNumber + b.maxSeqNo = p.SequenceNumber + b.pktQueue.headSN = p.SequenceNumber - 1 + } else if snDiff(b.maxSeqNo, p.SequenceNumber) <= 0 { + if p.SequenceNumber < b.maxSeqNo { + b.cycles += maxSN + } + b.maxSeqNo = p.SequenceNumber + } + b.packetCount++ + b.lastPacketTime = time.Now().UnixNano() + arrival := uint32(b.lastPacketTime / 1e6 * int64(b.clockRate/1e3)) + transit := arrival - p.Timestamp + if b.lastTransit != 0 { + d := int32(transit - b.lastTransit) + if d < 0 { + d = -d + } + b.jitter += (float64(d) - b.jitter) / 16 + } + b.lastTransit = transit + if b.codecType == webrtc.RTPCodecTypeVideo { + b.pktQueue.AddPacket(p, p.SequenceNumber == b.maxSeqNo) + } + + if b.tcc { + rtpTCC := rtp.TransportCCExtension{} + if err := rtpTCC.Unmarshal(p.GetExtension(b.tccExt)); err == nil { + if rtpTCC.TransportSequence < 0x0fff && (b.tccLastSn&0xffff) > 0xf000 { + b.tccCycles += maxSN + } + b.tccExtInfo = append(b.tccExtInfo, rtpExtInfo{ + ExtTSN: b.tccCycles | uint32(rtpTCC.TransportSequence), + Timestamp: b.lastPacketTime / 1e3, + }) + } + } +} + +func (b *Buffer) buildREMBPacket() *rtcp.ReceiverEstimatedMaximumBitrate { + br := b.maxBitrate + if b.rembSteps > 0 { + br /= uint64(b.rembSteps) + b.rembSteps-- + } + + return &rtcp.ReceiverEstimatedMaximumBitrate{ + SenderSSRC: b.ssrc, + Bitrate: br, + SSRCs: []uint32{b.ssrc}, + } +} + +func (b *Buffer) buildTransportCCPacket() *rtcp.TransportLayerCC { + if len(b.tccExtInfo) == 0 { + return nil + } + sort.Slice(b.tccExtInfo, func(i, j int) bool { + return b.tccExtInfo[i].ExtTSN < b.tccExtInfo[j].ExtTSN + }) + tccPkts := make([]rtpExtInfo, 0, int(float64(len(b.tccExtInfo))*1.2)) + for _, tccExtInfo := range b.tccExtInfo { + if tccExtInfo.ExtTSN < b.tccLastExtSN { + continue + } + if b.tccLastExtSN != 0 { + for j := b.tccLastExtSN + 1; j < tccExtInfo.ExtTSN; j++ { + tccPkts = append(tccPkts, rtpExtInfo{ExtTSN: j}) + } + } + b.tccLastExtSN = tccExtInfo.ExtTSN + tccPkts = append(tccPkts, tccExtInfo) + } + b.tccExtInfo = b.tccExtInfo[:0] + + rtcpTCC := &rtcp.TransportLayerCC{ + Header: rtcp.Header{ + Padding: true, + Count: rtcp.FormatTCC, + Type: rtcp.TypeTransportSpecificFeedback, + }, + MediaSSRC: b.ssrc, + BaseSequenceNumber: uint16(tccPkts[0].ExtTSN), + PacketStatusCount: uint16(len(tccPkts)), + FbPktCount: b.tccPktCtn, + } + b.tccPktCtn++ + + firstRecv := false + allSame := true + timestamp := int64(0) + deltaLen := 0 + lastStatus := rtcp.TypeTCCPacketReceivedWithoutDelta + maxStatus := rtcp.TypeTCCPacketNotReceived + + var statusList deque.Deque + + for _, stat := range tccPkts { + status := rtcp.TypeTCCPacketNotReceived + if stat.Timestamp != 0 { + var delta int64 + if !firstRecv { + firstRecv = true + timestamp = stat.Timestamp + rtcpTCC.ReferenceTime = uint32(stat.Timestamp / 64000) + } + + delta = (stat.Timestamp - timestamp) / 250 + if delta < 0 || delta > 255 { + status = rtcp.TypeTCCPacketReceivedLargeDelta + rDelta := int16(delta) + if int64(rDelta) != delta { + if rDelta > 0 { + rDelta = math.MaxInt16 + } else { + rDelta = math.MinInt16 + } + } + rtcpTCC.RecvDeltas = append(rtcpTCC.RecvDeltas, &rtcp.RecvDelta{ + Type: status, + Delta: int64(rDelta) * 250, + }) + deltaLen += 2 + } else { + status = rtcp.TypeTCCPacketReceivedSmallDelta + rtcpTCC.RecvDeltas = append(rtcpTCC.RecvDeltas, &rtcp.RecvDelta{ + Type: status, + Delta: delta * 250, + }) + deltaLen++ + } + timestamp = stat.Timestamp + } + + if allSame && lastStatus != rtcp.TypeTCCPacketReceivedWithoutDelta && status != lastStatus { + if statusList.Len() > 7 { + rtcpTCC.PacketChunks = append(rtcpTCC.PacketChunks, &rtcp.RunLengthChunk{ + PacketStatusSymbol: lastStatus, + RunLength: uint16(statusList.Len()), + }) + statusList.Clear() + lastStatus = rtcp.TypeTCCPacketReceivedWithoutDelta + maxStatus = rtcp.TypeTCCPacketNotReceived + allSame = true + } else { + allSame = false + } + } + statusList.PushBack(status) + if status > maxStatus { + maxStatus = status + } + lastStatus = status + + if !allSame { + if maxStatus == rtcp.TypeTCCPacketReceivedLargeDelta && statusList.Len() > 6 { + symbolList := make([]uint16, 7) + for i := 0; i < 7; i++ { + symbolList[i] = statusList.PopFront().(uint16) + } + rtcpTCC.PacketChunks = append(rtcpTCC.PacketChunks, &rtcp.StatusVectorChunk{ + SymbolSize: rtcp.TypeTCCSymbolSizeTwoBit, + SymbolList: symbolList, + }) + lastStatus = rtcp.TypeTCCPacketReceivedWithoutDelta + maxStatus = rtcp.TypeTCCPacketNotReceived + allSame = true + + for i := 0; i < statusList.Len(); i++ { + status = statusList.At(i).(uint16) + if status > maxStatus { + maxStatus = status + } + if allSame && lastStatus != rtcp.TypeTCCPacketReceivedWithoutDelta && status != lastStatus { + allSame = false + } + lastStatus = status + } + } else if statusList.Len() > 13 { + symbolList := make([]uint16, 14) + for i := 0; i < 14; i++ { + symbolList[i] = statusList.PopFront().(uint16) + } + rtcpTCC.PacketChunks = append(rtcpTCC.PacketChunks, &rtcp.StatusVectorChunk{ + SymbolSize: rtcp.TypeTCCSymbolSizeOneBit, + SymbolList: symbolList, + }) + lastStatus = rtcp.TypeTCCPacketReceivedWithoutDelta + maxStatus = rtcp.TypeTCCPacketNotReceived + allSame = true + } + } + } + + if statusList.Len() > 0 { + if allSame { + rtcpTCC.PacketChunks = append(rtcpTCC.PacketChunks, &rtcp.RunLengthChunk{ + PacketStatusSymbol: lastStatus, + RunLength: uint16(statusList.Len()), + }) + } else if maxStatus == rtcp.TypeTCCPacketReceivedLargeDelta { + symbolList := make([]uint16, statusList.Len()) + for i := 0; i < statusList.Len(); i++ { + symbolList[i] = statusList.PopFront().(uint16) + } + rtcpTCC.PacketChunks = append(rtcpTCC.PacketChunks, &rtcp.StatusVectorChunk{ + SymbolSize: rtcp.TypeTCCSymbolSizeTwoBit, + SymbolList: symbolList, + }) + } else { + symbolList := make([]uint16, statusList.Len()) + for i := 0; i < statusList.Len(); i++ { + symbolList[i] = statusList.PopFront().(uint16) + } + rtcpTCC.PacketChunks = append(rtcpTCC.PacketChunks, &rtcp.StatusVectorChunk{ + SymbolSize: rtcp.TypeTCCSymbolSizeOneBit, + SymbolList: symbolList, + }) + } + } + + pLen := uint16(20 + len(rtcpTCC.PacketChunks)*2 + deltaLen) + rtcpTCC.Header.Padding = pLen%4 != 0 + for pLen%4 != 0 { + pLen++ + } + rtcpTCC.Header.Length = (pLen / 4) - 1 + return rtcpTCC +} + +func (b *Buffer) buildReceptionReport() rtcp.ReceptionReport { + extMaxSeq := b.cycles | uint32(b.maxSeqNo) + expected := extMaxSeq - uint32(b.baseSN) + 1 + lost := expected - b.packetCount + if b.packetCount == 0 { + lost = 0 + } + expectedInterval := expected - b.lastExpected + b.lastExpected = expected + + receivedInterval := b.packetCount - b.lastReceived + b.lastReceived = b.packetCount + + lostInterval := expectedInterval - receivedInterval + + b.lostRate = float32(lostInterval) / float32(expectedInterval) + var fracLost uint8 + if expectedInterval != 0 && lostInterval > 0 { + fracLost = uint8((lostInterval << 8) / expectedInterval) + } + var dlsr uint32 + if b.lastSRRecv != 0 { + delayMS := uint32((time.Now().UnixNano() - b.lastSRRecv) / 1e6) + dlsr = (delayMS / 1e3) << 16 + dlsr |= (delayMS % 1e3) * 65536 / 1000 + } + + rr := rtcp.ReceptionReport{ + SSRC: b.ssrc, + FractionLost: fracLost, + TotalLost: lost, + LastSequenceNumber: extMaxSeq, + Jitter: uint32(b.jitter), + LastSenderReport: uint32(b.lastSRNTPTime >> 16), + Delay: dlsr, + } + return rr +} + +func (b *Buffer) SetSenderReportData(rtpTime uint32, ntpTime uint64) { + b.mu.Lock() + defer b.mu.Unlock() + b.lastSRRTPTime = rtpTime + b.lastSRNTPTime = ntpTime + b.lastSRRecv = time.Now().UnixNano() +} + +func (b *Buffer) BuildRTCP() (rtcp.ReceptionReport, []rtcp.Packet) { + b.mu.RLock() + defer b.mu.RUnlock() + var pkts []rtcp.Packet + var report rtcp.ReceptionReport + + report = b.buildReceptionReport() + + if b.remb { + pkts = append(pkts, b.buildREMBPacket()) + } + + if b.tcc { + if tccPkt := b.buildTransportCCPacket(); tccPkt != nil { + pkts = append(pkts, tccPkt) + } + } + + return report, pkts +} + +// WritePacket write buffer packet to requested track. and modify headers +func (b *Buffer) WritePacket(sn uint16, track *webrtc.Track, snOffset uint16, tsOffset, ssrc uint32) error { + b.mu.RLock() + defer b.mu.RUnlock() + if bufferPkt := b.pktQueue.GetPacket(sn); bufferPkt != nil { + bSsrc := bufferPkt.SSRC + bufferPkt.SequenceNumber -= snOffset + bufferPkt.Timestamp -= tsOffset + bufferPkt.SSRC = ssrc + err := track.WriteRTP(bufferPkt) + bufferPkt.Timestamp += tsOffset + bufferPkt.SequenceNumber += snOffset + bufferPkt.SSRC = bSsrc + return err + } + return errPacketNotFound +} + +func (b *Buffer) onLostHandler(fn func(nack *rtcp.TransportLayerNack)) { + if b.nack { + b.pktQueue.onLost = fn + } +} diff --git a/pkg/sfu/errors.go b/pkg/sfu/errors.go new file mode 100644 index 000000000..e7aac1e79 --- /dev/null +++ b/pkg/sfu/errors.go @@ -0,0 +1,16 @@ +package sfu + +import "errors" + +var ( + errPeerConnectionInitFailed = errors.New("pc init failed") + errPtNotSupported = errors.New("payload type not supported") + // router errors + errNoReceiverFound = errors.New("no receiver found") + // Helpers errors + errShortPacket = errors.New("packet is not large enough") + errNilPacket = errors.New("invalid nil packet") + // buffer errors + errPacketNotFound = errors.New("packet not found in cache") + errPacketTooOld = errors.New("packet not found in cache, too old") +) diff --git a/pkg/sfu/helpers.go b/pkg/sfu/helpers.go new file mode 100644 index 000000000..6f0614507 --- /dev/null +++ b/pkg/sfu/helpers.go @@ -0,0 +1,134 @@ +package sfu + +import ( + "encoding/binary" + "sync/atomic" +) + +const ntpEpoch = 2208988800 + +type atomicBool struct { + val int32 +} + +func (b *atomicBool) set(value bool) { // nolint: unparam + var i int32 + if value { + i = 1 + } + + atomic.StoreInt32(&(b.val), i) +} + +func (b *atomicBool) get() bool { + return atomic.LoadInt32(&(b.val)) != 0 +} + +// VP8Helper is a helper to get temporal data from VP8 packet header +/* + VP8Helper Payload Descriptor + 0 1 2 3 4 5 6 7 0 1 2 3 4 5 6 7 + +-+-+-+-+-+-+-+-+ +-+-+-+-+-+-+-+-+ + |X|R|N|S|R| PID | (REQUIRED) |X|R|N|S|R| PID | (REQUIRED) + +-+-+-+-+-+-+-+-+ +-+-+-+-+-+-+-+-+ + X: |I|L|T|K| RSV | (OPTIONAL) X: |I|L|T|K| RSV | (OPTIONAL) + +-+-+-+-+-+-+-+-+ +-+-+-+-+-+-+-+-+ + I: |M| PictureID | (OPTIONAL) I: |M| PictureID | (OPTIONAL) + +-+-+-+-+-+-+-+-+ +-+-+-+-+-+-+-+-+ + L: | TL0PICIDX | (OPTIONAL) | PictureID | + +-+-+-+-+-+-+-+-+ +-+-+-+-+-+-+-+-+ + T/K:|TID|Y| KEYIDX | (OPTIONAL) L: | TL0PICIDX | (OPTIONAL) + +-+-+-+-+-+-+-+-+ +-+-+-+-+-+-+-+-+ + T/K:|TID|Y| KEYIDX | (OPTIONAL) + +-+-+-+-+-+-+-+-+ +*/ +type VP8Helper struct { + TemporalSupported bool + // Optional Header + PictureID uint16 /* 8 or 16 bits, picture ID */ + picIDIdx uint8 + mBit bool + TL0PICIDX uint8 /* 8 bits temporal level zero index */ + tlzIdx uint8 + + // Optional Header If either of the T or K bits are set to 1, + // the TID/Y/KEYIDX extension field MUST be present. + TID uint8 /* 2 bits temporal layer idx*/ + // IsKeyFrame is a helper to detect if current packet is a keyframe + IsKeyFrame bool +} + +// Unmarshal parses the passed byte slice and stores the result in the VP8Helper this method is called upon +func (p *VP8Helper) Unmarshal(payload []byte) error { + if payload == nil { + return errNilPacket + } + + payloadLen := len(payload) + + if payloadLen < 4 { + return errShortPacket + } + + var idx uint8 + // Check for extended bit control + if payload[idx]&0x80 > 0 { + idx++ + // Check if T is present, if not, no temporal layer is available + p.TemporalSupported = payload[idx]&0x20 > 0 + L := payload[idx]&0x40 > 0 + // Check for PictureID + if payload[idx]&0x80 > 0 { + idx++ + p.picIDIdx = idx + pid := payload[idx] & 0x7f + // Check if m is 1, then Picture ID is 15 bits + if payload[idx]&0x80 > 0 { + idx++ + p.mBit = true + p.PictureID = binary.BigEndian.Uint16([]byte{pid, payload[idx]}) + } else { + p.PictureID = uint16(pid) + } + } + // Check if TL0PICIDX is present + if L { + idx++ + p.tlzIdx = idx + p.TL0PICIDX = payload[idx] + } + idx++ + // Set TID + p.TID = (payload[idx] & 0xc0) >> 6 + idx++ + if int(idx) >= payloadLen { + return errShortPacket + } + // Check is packet is a keyframe by looking at P bit in vp8 payload + p.IsKeyFrame = payload[idx]&0x01 == 0 + } + return nil +} + +func snDiff(sn1, sn2 uint16) int { + if sn1 == sn2 { + return 0 + } + if ((sn2 - sn1) & 0x8000) != 0 { + return 1 + } + return -1 +} + +func timeToNtp(ns int64) uint64 { + seconds := uint64(ns/1e9 + ntpEpoch) + fraction := uint64(((ns % 1e9) << 32) / 1e9) + return seconds<<32 | fraction +} + +// fromNtp converts a NTP timestamp into GO time +func fromNtp(seconds, fraction uint32) (tm int64) { + n := (int64(fraction) * 1e9) >> 32 + tm = (int64(seconds)-ntpEpoch)*1e9 + n + return +} diff --git a/pkg/sfu/packetqueue.go b/pkg/sfu/packetqueue.go new file mode 100644 index 000000000..4803cee27 --- /dev/null +++ b/pkg/sfu/packetqueue.go @@ -0,0 +1,124 @@ +package sfu + +import ( + //"github.com/pion/ion-sfu/pkg/log" + "github.com/pion/rtcp" + "github.com/pion/rtp" +) + +type queue struct { + pkts []*rtp.Packet + ssrc uint32 + head int + tail int + size int + headSN uint16 + counter int + duration uint32 + onLost func(nack *rtcp.TransportLayerNack) +} + +func (q *queue) AddPacket(pkt *rtp.Packet, latest bool) { + if !latest { + q.set(int(q.headSN-pkt.SequenceNumber), pkt) + return + } + diff := pkt.SequenceNumber - q.headSN + q.headSN = pkt.SequenceNumber + for i := uint16(1); i < diff; i++ { + q.push(nil) + q.counter++ + } + q.counter++ + q.push(pkt) + if q.counter >= 7 { + if n := q.nack(); n != nil && q.onLost != nil { + q.onLost(&rtcp.TransportLayerNack{ + MediaSSRC: q.ssrc, + Nacks: []rtcp.NackPair{*n}, + }) + } + q.clean() + q.counter -= 5 + } +} + +func (q *queue) GetPacket(sn uint16) *rtp.Packet { + return q.get(int(q.headSN - sn)) +} + +func (q *queue) push(pkt *rtp.Packet) { + q.resize() + q.head = (q.head - 1) & (len(q.pkts) - 1) + q.pkts[q.head] = pkt + q.size++ +} + +func (q *queue) shift() { + if q.size <= 0 { + return + } + q.tail = (q.tail - 1) & (len(q.pkts) - 1) + q.pkts[q.tail] = nil + q.size-- +} + +func (q *queue) last() *rtp.Packet { + return q.pkts[(q.tail-1)&(len(q.pkts)-1)] +} + +func (q *queue) get(i int) *rtp.Packet { + if i < 0 || i >= q.size { + return nil + } + return q.pkts[(q.head+i)&(len(q.pkts)-1)] +} + +func (q *queue) set(i int, pkt *rtp.Packet) { + if i < 0 || i >= q.size { + //log.Warnf("warn: %v:", errPacketTooOld) + return + } + q.pkts[(q.head+i)&(len(q.pkts)-1)] = pkt +} + +func (q *queue) resize() { + if len(q.pkts) == 0 { + q.pkts = make([]*rtp.Packet, 128) + return + } + if q.size == len(q.pkts) { + newBuf := make([]*rtp.Packet, q.size<<1) + if q.tail > q.head { + copy(newBuf, q.pkts[q.head:q.tail]) + } else { + n := copy(newBuf, q.pkts[q.head:]) + copy(newBuf[n:], q.pkts[:q.tail]) + } + q.head = 0 + q.tail = q.size + q.pkts = newBuf + } +} + +func (q *queue) nack() *rtcp.NackPair { + for i := 0; i < 5; i++ { + if q.get(q.counter-i-1) == nil { + blp := uint16(0) + for j := 1; j < q.counter-i; j++ { + if q.get(q.counter-i-j-1) == nil { + blp |= 1 << (j - 1) + } + } + return &rtcp.NackPair{PacketID: q.headSN - uint16(q.counter-i-1), LostPackets: rtcp.PacketBitmap(blp)} + } + } + return nil +} + +func (q *queue) clean() { + last := q.last() + for q.size > 120 && (last == nil || q.pkts[q.head].Timestamp-last.Timestamp > q.duration) { + q.shift() + } +}