Init.

.gitignore

@@ -1,3 +1,6 @@
+# Directories
+.idea
+
 # Binaries for programs and plugins
 *.exe
 *.exe~

buffer.go

@@ -0,0 +1,142 @@
+package jitter
+
+import (
+	"github.com/huandu/skiplist"
+	"github.com/samber/lo"
+	"math"
+
+	"sync"
+)
+
+type Buffer struct {
+	sync.Mutex
+
+	list *skiplist.SkipList
+
+	early *skiplist.SkipList
+	late  *skiplist.SkipList
+	loss  *skiplist.SkipList
+
+	current int64
+	latency int64
+
+	marked    bool
+	firstTime int64
+
+	tickInterval int64
+	minLatency   int64 // 200ms
+	maxLatency   int64 // 400ms
+	window       int64 // 2000ms
+}
+
+func NewBuffer(tickInterval, minLatency, maxLatency, window int64) *Buffer {
+	b := &Buffer{
+		list:         skiplist.New(skiplist.Int64),
+		early:        skiplist.New(skiplist.Int64),
+		late:         skiplist.New(skiplist.Int64),
+		loss:         skiplist.New(skiplist.Int64),
+		current:      0,
+		latency:      minLatency,
+		tickInterval: tickInterval,
+		minLatency:   minLatency,
+		maxLatency:   maxLatency,
+		window:       window,
+	}
+	return b
+}
+
+func (b *Buffer) Put(ts int64, data []byte) {
+	b.Lock()
+	defer b.Unlock()
+
+	if !b.marked {
+		b.firstTime = ts
+		b.marked = true
+	}
+
+	b.list.Set(ts, data)
+
+	delta := ts - b.targetTime()
+	if delta > 0 { // check
+		b.early.Set(ts, delta) // 일찍 온 시간을 기록
+	} else if delta > -b.maxLatency { // 늦게 온 것이면, 단 너무 늦으면 버림
+		b.late.Set(ts, -delta) // 늦은 시간을 기록
+	}
+
+}
+
+func (b *Buffer) Get() ([]byte, bool) {
+	b.Lock()
+	defer b.Unlock()
+
+	defer func() {
+		b.current += b.tickInterval
+	}()
+
+	b.adaptive()
+
+	targetTime := b.targetTime()
+
+	removeLessThan(b.list, targetTime)
+	removeLessThan(b.early, targetTime-b.window)
+	removeLessThan(b.late, targetTime-b.window)
+	removeLessThan(b.loss, targetTime-b.window)
+
+	front := b.list.Front()
+	if front != nil && front.Key() != nil && front.Key().(int64) == targetTime {
+		b.list.RemoveFront()
+		return front.Value.([]byte), true
+	} else {
+		// loss
+		b.loss.Set(targetTime, nil)
+		return nil, false
+	}
+}
+
+func (b *Buffer) adaptive() {
+	// late 가 너무 많다면 b.latency 를 늦춤
+	if b.late.Len() > int(b.window/b.tickInterval*2/100) { // 2% 보다 크면
+		candidate := b.latency + maxInList(b.late)
+		b.latency = lo.Min([]int64{candidate, b.maxLatency})
+		b.late.Init()
+		b.early.Init()
+	}
+
+	// loss 가 없이 안정적이라면
+	if b.loss.Len() == 0 && b.early.Len() > int(b.window/b.tickInterval*2/100) { // 2% 보다 크면
+		candidate := b.latency - minInList(b.early)
+		b.latency = lo.Max([]int64{candidate, b.minLatency})
+		b.late.Init()
+		b.early.Init()
+	}
+}
+
+func maxInList(list *skiplist.SkipList) int64 {
+	var res int64 = math.MinInt64
+	for el := list.Front(); el != nil; el = el.Next() {
+		res = lo.Max([]int64{res, el.Value.(int64)})
+	}
+	return res
+}
+
+func minInList(list *skiplist.SkipList) int64 {
+	var res int64 = math.MaxInt64
+	for el := list.Front(); el != nil; el = el.Next() {
+		res = lo.Min([]int64{res, el.Value.(int64)})
+	}
+	return res
+}
+
+func removeLessThan(list *skiplist.SkipList, ts int64) {
+	for {
+		front := list.Front()
+		if front == nil || front.Key() == nil || front.Key().(int64) >= ts {
+			break
+		}
+		list.RemoveFront()
+	}
+}
+
+func (b *Buffer) targetTime() int64 {
+	return b.firstTime + b.current - b.latency
+}

buffer_test.go

@@ -0,0 +1,84 @@
+package jitter
+
+import (
+	"github.com/huandu/go-assert"
+	"testing"
+)
+
+func (b *Buffer) getDry() []byte {
+	if res, ok := b.Get(); ok {
+		return res
+	} else {
+		return nil
+	}
+}
+
+func Test_basic(t *testing.T) {
+	b := NewBuffer(20, 100, 400, 1000)
+
+	b.Put(1000, []byte{1})
+	b.Put(1020, []byte{2})
+	b.Put(1040, []byte{3})
+
+	assert.Equal(t, b.getDry(), nil)       // 900
+	assert.Equal(t, b.getDry(), nil)       // 920
+	assert.Equal(t, b.getDry(), nil)       // 940
+	assert.Equal(t, b.getDry(), nil)       // 960
+	assert.Equal(t, b.getDry(), nil)       // 980
+	assert.Equal(t, b.getDry(), []byte{1}) // 1000
+	assert.Equal(t, b.getDry(), []byte{2}) // 1020
+	assert.Equal(t, b.getDry(), []byte{3}) // 1040
+
+	assert.Equal(t, b.targetTime(), int64(1060))
+	assert.Equal(t, b.latency, int64(100))
+}
+
+func Test_basic2(t *testing.T) {
+	b := NewBuffer(20, 100, 400, 1000)
+
+	b.Put(1000, []byte{1})
+
+	assert.Equal(t, b.getDry(), nil)       // 900
+	assert.Equal(t, b.getDry(), nil)       // 920
+	assert.Equal(t, b.getDry(), nil)       // 940
+	assert.Equal(t, b.getDry(), nil)       // 960
+	assert.Equal(t, b.getDry(), nil)       // 980
+	assert.Equal(t, b.getDry(), []byte{1}) // 1000
+	assert.Equal(t, b.getDry(), nil)       // 1020
+	assert.Equal(t, b.getDry(), nil)       // 1040
+	assert.Equal(t, b.getDry(), nil)       // 1060
+	assert.Equal(t, b.getDry(), nil)       // 1080
+	assert.Equal(t, b.getDry(), nil)       // 1100
+	assert.Equal(t, b.getDry(), nil)       // 1120
+	assert.Equal(t, b.getDry(), nil)       // 1140
+	assert.Equal(t, b.getDry(), nil)       // 1160
+	assert.Equal(t, b.getDry(), nil)       // 1180
+
+	b.Put(1020, []byte{2}) // late 180
+	b.Put(1040, []byte{3}) // late 160
+
+	assert.Equal(t, b.targetTime(), int64(1200))
+	assert.Equal(t, b.latency, int64(100))
+
+	b.adaptive()
+
+	// 	assert.Equal(t, b.targetTime(), int64(1000))
+	assert.Equal(t, b.latency, int64(280))
+
+	//
+
+	i := 0
+	for ; i < 12; i++ {
+		b.Put(int64(1060+i*20), []byte{4 + byte(i)})
+	}
+
+	for ; i < 100; i++ {
+		b.Put(int64(1060+i*20), []byte{4 + byte(i)})
+		b.getDry()
+	}
+
+	b.adaptive()
+
+	// assert.Equal(t, b.targetTime(), int64(1200))
+	assert.Equal(t, b.latency, int64(100))
+}

go.mod

@@ -0,0 +1,35 @@
+module go-jitter
+
+go 1.19
+
+require (
+	github.com/huandu/go-assert v1.1.5
+	github.com/huandu/skiplist v1.2.0
+	github.com/pion/webrtc/v3 v3.1.59
+	github.com/samber/lo v1.38.1
+)
+
+require (
+	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/google/uuid v1.3.0 // indirect
+	github.com/pion/datachannel v1.5.5 // indirect
+	github.com/pion/dtls/v2 v2.2.6 // indirect
+	github.com/pion/ice/v2 v2.3.2 // indirect
+	github.com/pion/interceptor v0.1.12 // indirect
+	github.com/pion/logging v0.2.2 // indirect
+	github.com/pion/mdns v0.0.7 // indirect
+	github.com/pion/randutil v0.1.0 // indirect
+	github.com/pion/rtcp v1.2.10 // indirect
+	github.com/pion/rtp v1.7.13 // indirect
+	github.com/pion/sctp v1.8.6 // indirect
+	github.com/pion/sdp/v3 v3.0.6 // indirect
+	github.com/pion/srtp/v2 v2.0.12 // indirect
+	github.com/pion/stun v0.4.0 // indirect
+	github.com/pion/transport/v2 v2.0.2 // indirect
+	github.com/pion/turn/v2 v2.1.0 // indirect
+	github.com/pion/udp/v2 v2.0.1 // indirect
+	golang.org/x/crypto v0.6.0 // indirect
+	golang.org/x/exp v0.0.0-20220303212507-bbda1eaf7a17 // indirect
+	golang.org/x/net v0.7.0 // indirect
+	golang.org/x/sys v0.5.0 // indirect
+)