1
Fork 0
mirror of https://github.com/jech/galene.git synced 2024-12-23 07:45:46 +01:00
galene/packetcache/packetcache.go
2020-10-12 14:43:28 +02:00

546 lines
12 KiB
Go

// Package packetcache implement a packet cache that maintains a history
// of recently seen packets, the last keyframe, and a number of statistics
// that are needed for sending receiver reports.
package packetcache
import (
"math/bits"
"sync"
)
// The maximum size of packets stored in the cache. Chosen to be
// a multiple of 8.
const BufSize = 1504
// The maximum number of packets that constitute a keyframe.
const maxFrame = 1024
// entry represents a cached packet.
type entry struct {
seqno uint16
lengthAndMarker uint16 // 1 bit of marker, 15 bits of length
timestamp uint32
buf [BufSize]byte
}
func (e *entry) length() uint16 {
return e.lengthAndMarker & 0x7FFF
}
func (e *entry) marker() bool {
return (e.lengthAndMarker & 0x8000) != 0
}
// bitmap keeps track of recent loss history
type bitmap struct {
valid bool
first uint16
bitmap uint32
}
// frame is used for storing the last keyframe
type frame struct {
timestamp uint32
complete bool
entries []entry
}
type Cache struct {
mu sync.Mutex
//stats
last uint16
cycle uint16
lastValid bool
expected uint32
lost uint32
totalLost uint32
// bitmap
bitmap bitmap
// buffered keyframe
keyframe frame
// the actual cache
tail uint16
entries []entry
}
// New creates a cache with the given capacity.
func New(capacity int) *Cache {
if capacity > int(^uint16(0)) {
return nil
}
return &Cache{
entries: make([]entry, capacity),
}
}
// compare performs comparison modulo 2^16.
func compare(s1, s2 uint16) int {
if s1 == s2 {
return 0
}
if ((s2 - s1) & 0x8000) != 0 {
return 1
}
return -1
}
// seqnoInvalid returns true if seqno is unreasonably far in the past
func seqnoInvalid(seqno, reference uint16) bool {
if compare(reference, seqno) < 0 {
return false
}
if reference-seqno > 0x100 {
return true
}
return false
}
// set sets a bit in the bitmap, shifting if necessary
func (bitmap *bitmap) set(seqno uint16) {
if !bitmap.valid || seqnoInvalid(seqno, bitmap.first) {
bitmap.first = seqno
bitmap.bitmap = 1
bitmap.valid = true
return
}
if compare(bitmap.first, seqno) > 0 {
return
}
if seqno-bitmap.first >= 32 {
shift := seqno - bitmap.first - 31
bitmap.bitmap >>= shift
bitmap.first += shift
}
if (bitmap.bitmap & 1) == 1 {
ones := bits.TrailingZeros32(^bitmap.bitmap)
bitmap.bitmap >>= ones
bitmap.first += uint16(ones)
}
bitmap.bitmap |= (1 << uint16(seqno-bitmap.first))
return
}
// BitmapGet shifts up to 17 bits out of the bitmap. It returns a boolean
// indicating if any were 0, the index of the first 0 bit, and a bitmap
// indicating any 0 bits after the first one.
func (cache *Cache) BitmapGet(next uint16) (bool, uint16, uint16) {
cache.mu.Lock()
defer cache.mu.Unlock()
return cache.bitmap.get(next)
}
func (bitmap *bitmap) get(next uint16) (bool, uint16, uint16) {
first := bitmap.first
if compare(first, next) >= 0 {
return false, first, 0
}
count := next - first
if count > 17 {
count = 17
}
bm := (^bitmap.bitmap) & ^((^uint32(0)) << count)
bitmap.bitmap >>= count
bitmap.first += count
if bm == 0 {
return false, first, 0
}
if (bm & 1) == 0 {
count := bits.TrailingZeros32(bm)
bm >>= count
first += uint16(count)
}
return true, first, uint16(bm >> 1)
}
// insert inserts a packet into a frame.
func (frame *frame) insert(seqno uint16, timestamp uint32, marker bool, data []byte) bool {
n := len(frame.entries)
i := 0
if n == 0 || seqno > frame.entries[n-1].seqno {
// fast path
i = n
} else {
for i < n {
if frame.entries[i].seqno >= seqno {
break
}
i++
}
if i < n && frame.entries[i].seqno == seqno {
// duplicate
return false
}
}
if n >= maxFrame {
// overflow
return false
}
lam := uint16(len(data))
if marker {
lam |= 0x8000
}
e := entry{
seqno: seqno,
lengthAndMarker: lam,
timestamp: timestamp,
}
copy(e.buf[:], data)
if i >= n {
frame.entries = append(frame.entries, e)
return true
}
frame.entries = append(frame.entries, entry{})
copy(frame.entries[i+1:], frame.entries[i:])
frame.entries[i] = e
return true
}
// store checks whether a packet is part of the current keyframe and, if
// so, inserts it.
func (frame *frame) store(seqno uint16, timestamp uint32, first bool, marker bool, data []byte) bool {
if first {
if frame.timestamp != timestamp {
frame.timestamp = timestamp
frame.complete = false
frame.entries = frame.entries[:0]
}
} else if len(frame.entries) > 0 {
if frame.timestamp != timestamp {
delta := seqno - frame.entries[0].seqno
if (delta&0x8000) == 0 && delta > 0x4000 {
frame.complete = false
frame.entries = frame.entries[:0]
}
return false
}
} else {
return false
}
done := frame.insert(seqno, timestamp, marker, data)
if done && !frame.complete {
marker := false
fst := frame.entries[0].seqno
for i := 1; i < len(frame.entries); i++ {
if frame.entries[i].seqno != fst+uint16(i) {
return done
}
if frame.entries[i].marker() {
marker = true
}
}
if marker {
frame.complete = true
}
}
return done
}
// Store stores a packet in the cache. It returns the first seqno in the
// bitmap, and the index at which the packet was stored.
func (cache *Cache) Store(seqno uint16, timestamp uint32, keyframe bool, marker bool, buf []byte) (uint16, uint16) {
cache.mu.Lock()
defer cache.mu.Unlock()
if !cache.lastValid || seqnoInvalid(seqno, cache.last) {
cache.last = seqno
cache.lastValid = true
cache.expected++
} else {
if compare(cache.last, seqno) <= 0 {
cache.expected += uint32(seqno - cache.last)
cache.lost += uint32(seqno - cache.last - 1)
if seqno < cache.last {
cache.cycle++
}
cache.last = seqno
} else {
if cache.lost > 0 {
cache.lost--
}
}
}
cache.bitmap.set(seqno)
done := cache.keyframe.store(seqno, timestamp, keyframe, marker, buf)
if done && !cache.keyframe.complete {
completeKeyframe(cache)
}
i := cache.tail
cache.entries[i].seqno = seqno
copy(cache.entries[i].buf[:], buf)
lam := uint16(len(buf))
if marker {
lam |= 0x8000
}
cache.entries[i].lengthAndMarker = lam
cache.entries[i].timestamp = timestamp
cache.tail = (i + 1) % uint16(len(cache.entries))
return cache.bitmap.first, i
}
// completeKeyFrame attempts to complete the current keyframe.
func completeKeyframe(cache *Cache) {
l := len(cache.keyframe.entries)
if l == 0 {
return
}
first := cache.keyframe.entries[0].seqno
last := cache.keyframe.entries[l-1].seqno
count := (last - first) // may wrap around
if count > 0x4000 {
// this shouldn't happen
return
}
var buf []byte
if count > 1 {
if buf == nil {
buf = make([]byte, BufSize)
}
for i := uint16(1); i < count; i++ {
n, ts, marker := get(first+i, cache.entries, buf)
if n > 0 {
cache.keyframe.store(
first+i, ts, false, marker, buf,
)
}
}
}
if !cache.keyframe.complete {
// Try to find packets after the last one.
for {
l := len(cache.keyframe.entries)
if cache.keyframe.entries[l-1].marker() {
break
}
if buf == nil {
buf = make([]byte, BufSize)
}
seqno := cache.keyframe.entries[l-1].seqno + 1
n, ts, marker := get(seqno, cache.entries, buf)
if n <= 0 {
break
}
done := cache.keyframe.store(
seqno, ts, false, marker, buf,
)
if !done || marker {
break
}
}
}
}
// Expect records that we expect n packets. It is used for loss statistics.
func (cache *Cache) Expect(n int) {
if n <= 0 {
return
}
cache.mu.Lock()
defer cache.mu.Unlock()
cache.expected += uint32(n)
}
// get retrieves a packet from a slice of entries.
func get(seqno uint16, entries []entry, result []byte) (uint16, uint32, bool) {
for i := range entries {
if entries[i].lengthAndMarker == 0 || entries[i].seqno != seqno {
continue
}
var n uint16
if len(result) > 0 {
n = uint16(copy(
result[:entries[i].length()],
entries[i].buf[:]))
} else {
n = entries[i].length()
}
return n, entries[i].timestamp, entries[i].marker()
}
return 0, 0, false
}
// Get retrieves a packet from the cache, returns the number of bytes
// copied. If result is of length 0, returns the size of the packet.
func (cache *Cache) Get(seqno uint16, result []byte) uint16 {
cache.mu.Lock()
defer cache.mu.Unlock()
n, _, _ := get(seqno, cache.keyframe.entries, result)
if n > 0 {
return n
}
n, _, _ = get(seqno, cache.entries, result)
if n > 0 {
return n
}
return 0
}
func (cache *Cache) Last() (bool, uint16, uint32) {
cache.mu.Lock()
defer cache.mu.Unlock()
if !cache.lastValid {
return false, 0, 0
}
len, ts, _ := get(cache.last, cache.entries, nil)
if len == 0 {
return false, 0, 0
}
return true, cache.last, ts
}
// GetAt retrieves a packet from the cache assuming it is at the given index.
func (cache *Cache) GetAt(seqno uint16, index uint16, result []byte) uint16 {
cache.mu.Lock()
defer cache.mu.Unlock()
if int(index) > len(cache.entries) {
return 0
}
if cache.entries[index].seqno != seqno {
return 0
}
return uint16(copy(
result[:cache.entries[index].length()],
cache.entries[index].buf[:]),
)
}
// Keyframe returns the last buffered keyframe. It returns the frame's
// timestamp and a boolean indicating if the frame is complete.
func (cache *Cache) Keyframe() (uint32, bool, []uint16) {
cache.mu.Lock()
defer cache.mu.Unlock()
if len(cache.keyframe.entries) == 0 {
return 0, false, nil
}
seqnos := make([]uint16, len(cache.keyframe.entries))
for i := range cache.keyframe.entries {
seqnos[i] = cache.keyframe.entries[i].seqno
}
return cache.keyframe.timestamp, cache.keyframe.complete, seqnos
}
func (cache *Cache) KeyframeSeqno() (bool, uint16, uint32) {
cache.mu.Lock()
defer cache.mu.Unlock()
if len(cache.keyframe.entries) == 0 {
return false, 0, 0
}
return true, cache.keyframe.entries[0].seqno, cache.keyframe.timestamp
}
func (cache *Cache) resize(capacity int) {
if len(cache.entries) == capacity {
return
}
entries := make([]entry, capacity)
if capacity > len(cache.entries) {
copy(entries, cache.entries[:cache.tail])
copy(entries[int(cache.tail)+capacity-len(cache.entries):],
cache.entries[cache.tail:])
} else if capacity > int(cache.tail) {
copy(entries, cache.entries[:cache.tail])
copy(entries[cache.tail:],
cache.entries[int(cache.tail)+
len(cache.entries)-capacity:])
} else {
// too bad, invalidate all indices
copy(entries,
cache.entries[int(cache.tail)-capacity:cache.tail])
cache.tail = 0
}
cache.entries = entries
}
// Resize resizes the cache to the given capacity. This might invalidate
// indices of recently stored packets.
func (cache *Cache) Resize(capacity int) {
cache.mu.Lock()
defer cache.mu.Unlock()
cache.resize(capacity)
}
// ResizeCond is like Resize, but avoids invalidating recent indices.
func (cache *Cache) ResizeCond(capacity int) bool {
cache.mu.Lock()
defer cache.mu.Unlock()
current := len(cache.entries)
if current >= capacity*3/4 && current < capacity*2 {
return false
}
if capacity < current {
if int(cache.tail) > capacity {
// this would invalidate too many indices
return false
}
}
cache.resize(capacity)
return true
}
// GetStats returns statistics about received packets. If reset is true,
// the statistics are reset.
func (cache *Cache) GetStats(reset bool) (uint32, uint32, uint32, uint32) {
cache.mu.Lock()
defer cache.mu.Unlock()
expected := cache.expected
lost := cache.lost
totalLost := cache.totalLost + cache.lost
eseqno := uint32(cache.cycle)<<16 | uint32(cache.last)
if reset {
cache.expected = 0
cache.totalLost += cache.lost
cache.lost = 0
}
return expected, lost, totalLost, eseqno
}
// ToBitmap takes a non-empty sorted list of seqnos, and computes a bitmap
// covering a prefix of the list. It returns the part of the list that
// couldn't be covered.
func ToBitmap(seqnos []uint16) (first uint16, bitmap uint16, remain []uint16) {
first = seqnos[0]
bitmap = uint16(0)
remain = seqnos[1:]
for len(remain) > 0 {
delta := remain[0] - first - 1
if delta >= 16 {
break
}
bitmap = bitmap | (1 << delta)
remain = remain[1:]
}
return
}