package main import ( "sync" "sync/atomic" "time" ) // How long a sent frame may remain unreceived before it is charged as lost. // The mask outlives a measure by the same window, so the judge's last // masked-era charges still land inside it. const lossDeadline = 100 * time.Millisecond type frontier struct { t time.Time seq uint64 } // Per-stream loss accounting with no tolerance of reordering: an arrival // charges everything between the previous arrival and itself as lost, and // judge charges whatever was sent more than lossDeadline ago without // arriving. next is the first sequence number not yet accounted for, by // arrival or by charge; anything arriving below it is late. type lossWindow struct { mu sync.Mutex next uint64 frontiers []frontier lost atomic.Uint64 late atomic.Uint64 // The sending half of this stream. Both halves are in this process and the // receiving socket ignores its own outgoing frames, so this window is fed by // that sender alone and by nothing else on the wire. sent *atomic.Uint64 } // Takes the sending halves rather than a count, so a window cannot be built // without the frontier it judges sequence numbers against. func newLossWindows(tx []*txStats) []lossWindow { w := make([]lossWindow, len(tx)) for i := range w { w[i].sent = &tx[i].sent } return w } func (w *lossWindow) observe(seq uint64) { w.mu.Lock() if seq < w.next { w.mu.Unlock() w.late.Add(1) return } lost := seq - w.next w.next = seq + 1 w.mu.Unlock() w.lost.Add(lost) } // Charges everything the sender had put on the wire by one deadline ago that // has not arrived since, so a silent wire is charged within lossDeadline of // going silent rather than when traffic resumes. func (w *lossWindow) judge(now time.Time) { w.mu.Lock() var bound uint64 keep := 0 for i, f := range w.frontiers { if now.Sub(f.t) < lossDeadline { break } bound = f.seq keep = i + 1 } w.frontiers = append(w.frontiers[:0], w.frontiers[keep:]...) w.frontiers = append(w.frontiers, frontier{t: now, seq: w.sent.Load()}) var lost uint64 if bound > w.next { lost = bound - w.next w.next = bound } w.mu.Unlock() w.lost.Add(lost) }