Use cache for packet level processing.
* Use cache for packet level processing.
Do not overload ops queue for packet level operations.
* byte -> uint8
@@ -25,6 +25,17 @@ const (
ReportDelta = 1e9
)
+type twccMetadata struct {
+ sn uint16
+ arrivalTime int64
+ marker bool
+}
+
+type audioLevelMetadata struct {
+ level uint8
+ duration uint32
+}
+
type pendingPacket struct {
arrivalTime int64
packet []byte
@@ -69,6 +80,9 @@ type Buffer struct {
latestTSForAudioLevelInitialized bool
latestTSForAudioLevel uint32
+ twccPending []twccMetadata
+ audioLevelPending []audioLevelMetadata
+
lastPacketRead int
pliThrottle int64
@@ -396,11 +410,14 @@ func (b *Buffer) processHeaderExtensions(p *rtp.Packet, arrivalTime int64) {
// for bandwidth estimation
if b.twcc {
if ext := p.GetExtension(b.twccExt); len(ext) > 1 {
- sn := binary.BigEndian.Uint16(ext[0:2])
- marker := p.Marker
- b.callbacksQueue.Enqueue(func() {
- b.feedbackTWCC(sn, arrivalTime, marker)
+ b.twccPending = append(b.twccPending, twccMetadata{
+ sn: binary.BigEndian.Uint16(ext[0:2]),
+ arrivalTime: arrivalTime,
+ marker: p.Marker,
})
+ if len(b.twccPending) == 1 {
+ b.callbacksQueue.Enqueue(b.processTWCCPending)
+ }
}
}
@@ -415,9 +432,13 @@ func (b *Buffer) processHeaderExtensions(p *rtp.Packet, arrivalTime int64) {
if (p.Timestamp - b.latestTSForAudioLevel) < (1 << 31) {
duration := (int64(p.Timestamp) - int64(b.latestTSForAudioLevel)) * 1e3 / int64(b.clockRate)
if duration > 0 {
- b.callbacksQueue.Enqueue(func() {
- b.onAudioLevel(ext.Level, uint32(duration))
+ b.audioLevelPending = append(b.audioLevelPending, audioLevelMetadata{
+ level: ext.Level,
+ duration: uint32(duration),
})
+ if len(b.audioLevelPending) == 1 {
+ b.callbacksQueue.Enqueue(b.processAudioLevelPending)
+ }
}
b.latestTSForAudioLevel = p.Timestamp
@@ -631,3 +652,29 @@ func (b *Buffer) GetDeltaStats() *StreamStatsWithLayers {
Layers: layers,
}
}
+
+func (b *Buffer) processTWCCPending() {
+ b.Lock()
+ pending := b.twccPending
+ b.twccPending = nil
+ b.Unlock()
+
+ if b.feedbackTWCC != nil {
+ for _, p := range pending {
+ b.feedbackTWCC(p.sn, p.arrivalTime, p.marker)
+ }
+ }
+}
+
+func (b *Buffer) processAudioLevelPending() {
+ b.Lock()
+ pending := b.audioLevelPending
+ b.audioLevelPending = nil
+ b.Unlock()
+
+ if b.onAudioLevel != nil {
+ for _, p := range pending {
+ b.onAudioLevel(p.level, p.duration)
+ }
+ }
+}
