Judge loss by arrival order and a deadline, one tx ring per stream

loss.go

@@ -1,22 +1,30 @@
 package main
 
 import (
-	"math/bits"
 	"sync"
 	"sync/atomic"
+	"time"
 )
 
-const (
-	lossSlots = 1 << 16
-	lossWords = lossSlots / 64
-)
+// 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
-	base      uint64
-	inited    bool
-	maskedEra bool
-	bits      []uint64
+	next      uint64
+	frontiers []frontier
 	lost      atomic.Uint64
 	late      atomic.Uint64
 
@@ -27,97 +35,49 @@ type lossWindow struct {
 }
 
 // Takes the sending halves rather than a count, so a window cannot be built
-// without the bound it judges sequence numbers against.
+// without the frontier it judges sequence numbers against.
 func newLossWindows(tx []*txStats) []lossWindow {
 	w := make([]lossWindow, len(tx))
 	for i := range w {
-		w[i].bits = make([]uint64, lossWords)
 		w[i].sent = &tx[i].sent
 	}
 	return w
 }
 
-// A sequence number is only judged once it falls out of the window, so a frame
-// that has arrived but not yet been drained is never miscounted as lost.
-//
-// Reports whether the sequence number could have come off the wire at all. One
-// above what the sender has reached is a damaged header rather than a gap, and
-// is refused before it can touch the base. Believing one would drag the window
-// up past every real sequence, write off the span in between as lost, and leave
-// every frame after it arriving below the base and counted late for the rest of
-// the run. Nothing that arrives later is evidence enough to undo that, which is
-// why the sender's own frontier is the guard rather than a limit on how far a
-// single step may move.
-// masked slides the window without charging. A gap's charge lands at eviction,
-// which can fall after the mask lifts, so this source judges each charge
-// against the mask itself rather than leaving a deferred judgment to the
-// sampler's gate.
-func (w *lossWindow) observe(seq uint64, masked bool) bool {
-	if seq >= w.sent.Load() {
-		return false
-	}
+func (w *lossWindow) observe(seq uint64) {
 	w.mu.Lock()
-	if w.maskedEra && !masked {
-		// Leaving a measure: the window was full when it began, so every hole
-		// still inside went missing while the mask was up. Marking them
-		// delivered keeps the write-off complete however the eviction timing
-		// falls, instead of charging the gap's in-window tail later.
-		for i := range w.bits {
-			w.bits[i] = ^uint64(0)
-		}
-	}
-	w.maskedEra = masked
-	if !w.inited {
-		// Start half a window below the first sequence seen, so anything the
-		// sender put on the wire before it lands inside the window rather than
-		// below the base.
-		if seq > lossSlots/2 {
-			w.base = seq - lossSlots/2
-		}
-		w.inited = true
-	}
-	if seq < w.base {
+	if seq < w.next {
 		w.mu.Unlock()
-		if !masked {
-			w.late.Add(1)
-		}
-		return true
-	}
-	if seq >= w.base+lossSlots {
-		w.evict(seq-lossSlots+1, masked)
-	}
-	idx := seq & (lossSlots - 1)
-	w.bits[idx>>6] |= 1 << (idx & 63)
-	w.mu.Unlock()
-	return true
-}
-
-func (w *lossWindow) evict(newBase uint64, masked bool) {
-	span := newBase - w.base
-	if span >= lossSlots {
-		var missing uint64
-		for i := range w.bits {
-			missing += uint64(64 - bits.OnesCount64(w.bits[i]))
-			w.bits[i] = 0
-		}
-		if !masked {
-			w.lost.Add(missing + span - lossSlots)
-		}
-		w.base = newBase
+		w.late.Add(1)
 		return
 	}
-	var missing uint64
-	for s := w.base; s < newBase; s++ {
-		idx := s & (lossSlots - 1)
-		word, bit := idx>>6, uint64(1)<<(idx&63)
-		if w.bits[word]&bit == 0 {
-			missing++
-			continue
-		}
-		w.bits[word] &^= bit
-	}
-	if !masked {
-		w.lost.Add(missing)
-	}
-	w.base = newBase
+	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)
 }