DebTik/internal/bwtest/udp_control.go

package bwtest

import (
	"context"
	"fmt"
	"net"
	"sync"
	"time"

	"github.com/vishvananda/netlink"
)

const udpFeedbackVersion = "receiver-v1"
const udpProbeType uint8 = 4

// UDP without a requested limit probes capacity using receiver feedback instead
// of enqueuing an unbounded amount of traffic ahead of the reverse direction.
type udpPacer struct {
	mu                                sync.Mutex
	limiter                           *RateLimiter
	automatic                         bool
	rate                              int64
	started, lastFeedback, lastChange time.Time
	lastLoss                          time.Time
	delivered                         int64
	adjustPackets, adjustLost         int64
}

func newUDPPacer(limit int64) *udpPacer {
	now := time.Now()
	automatic := limit <= 0
	if automatic {
		limit = 2_000_000
	}
	return &udpPacer{limiter: NewRateLimiter(limit), automatic: automatic, rate: limit, started: now, lastFeedback: now, lastChange: now}
}
func (p *udpPacer) wait(ctx context.Context, n int) bool {
	// Take only a small token-accounting lock; feedback and other streams can
	// update the rate while this stream waits for its next datagram.
	for {
		p.mu.Lock()
		if p.automatic && time.Since(p.lastFeedback) > 2*time.Second && time.Since(p.lastChange) > time.Second {
			p.setRate(max(int64(128000), p.rate/2))
			p.lastChange = time.Now()
		}
		rl := p.limiter
		now := time.Now()
		rl.tokens += now.Sub(rl.lastCheckTime).Seconds() * float64(p.rate) / 8
		rl.lastCheckTime = now
		cap := max(float64(n), float64(p.rate)/8*.02)
		if rl.tokens > cap {
			rl.tokens = cap
		}
		if rl.tokens >= float64(n) {
			rl.tokens -= float64(n)
			p.mu.Unlock()
			return ctx.Err() == nil
		}
		delay := time.Duration((float64(n) - rl.tokens) / (float64(p.rate) / 8) * float64(time.Second))
		p.mu.Unlock()
		if delay > 20*time.Millisecond {
			delay = 20 * time.Millisecond
		}
		timer := time.NewTimer(delay)
		select {
		case <-ctx.Done():
			timer.Stop()
			return false
		case <-timer.C:
		}
	}
}
func (p *udpPacer) setRate(rate int64) { p.rate = rate; p.limiter.bps = rate }
func (p *udpPacer) observe(bytes, packets, lost int64, now time.Time) {
	p.mu.Lock()
	defer p.mu.Unlock()
	dt := now.Sub(p.lastFeedback).Seconds()
	delta := bytes - p.delivered
	if dt <= 0 || delta < 0 {
		return
	}
	p.lastFeedback = now
	p.delivered = bytes
	if !p.automatic || now.Sub(p.started) < time.Second || now.Sub(p.lastChange) < 450*time.Millisecond {
		return
	}
	// Loss counts can decrease when delayed packets arrive. Compare cumulative
	// counters since the previous decision and never interpret reordering as
	// negative loss. Older peers without packet counts still use byte feedback.
	packetDelta := max(int64(0), packets-p.adjustPackets)
	lossDelta := max(int64(0), lost-p.adjustLost)
	p.adjustPackets, p.adjustLost = packets, lost
	deliveredRate := int64(float64(delta) * 8 / dt)
	switch {
	case packetDelta+lossDelta >= 10 && lossDelta*200 >= packetDelta+lossDelta:
		p.setRate(max(int64(128000), min(p.rate*8/10, deliveredRate*9/10)))
		p.lastLoss = now
	case deliveredRate < p.rate*7/10:
		p.setRate(max(int64(128000), deliveredRate*11/10))
	case deliveredRate >= p.rate*85/100 && (p.lastLoss.IsZero() || now.Sub(p.lastLoss) >= 2*time.Second):
		p.setRate(min(int64(10_000_000_000), p.rate*115/100))
	}
	p.lastChange = now
}
func udpPayloadSize(requested int, ip net.IP) (int, error) {
	routes, err := netlink.RouteGet(ip)
	if err != nil {
		return 0, fmt.Errorf("bandwidth target route: %w", err)
	}
	if len(routes) == 0 {
		return 0, fmt.Errorf("no route to bandwidth target")
	}
	link, err := netlink.LinkByIndex(routes[0].LinkIndex)
	if err != nil {
		return 0, err
	}
	mtu := link.Attrs().MTU
	if routes[0].MTU > 0 {
		mtu = min(mtu, routes[0].MTU)
	}
	overhead := 28
	if ip.To4() == nil {
		overhead = 48
	}
	size := min(requested, mtu-overhead)
	if size < 64 {
		return 0, fmt.Errorf("target MTU is too small for bandwidth datagrams")
	}
	return size, nil
}
func udpProbe(hash uint64) []byte {
	b := make([]byte, HeaderSize)
	EncodeHeader(b, PacketHeader{Magic: MagicBytes, Version: ProtoVer, PacketType: udpProbeType, SessionHash: hash})
	return b
}

// Each UDP stream has its own sequence space. Late packets within the bounded
// reorder window undo inferred losses, and duplicate datagrams are not counted.
type udpSequenceState struct {
	max  uint64
	seen [1024]uint64
}
type udpTracker struct {
	streams map[uint8]*udpSequenceState
	lost    int64
}

func (t *udpTracker) accept(stream uint8, seq uint64) (bool, int64) {
	if seq == 0 {
		return false, t.lost
	}
	if t.streams == nil {
		t.streams = make(map[uint8]*udpSequenceState)
	}
	s := t.streams[stream]
	if s == nil {
		s = &udpSequenceState{}
		t.streams[stream] = s
	}
	if seq > s.max {
		t.lost += int64(seq - s.max - 1)
		s.max = seq
	} else {
		if s.max-seq >= 1024 || s.seen[seq%1024] == seq {
			return false, t.lost
		}
		if t.lost > 0 {
			t.lost--
		}
	}
	s.seen[seq%1024] = seq
	return true, t.lost
}