fix tests finally

tcp/txqueue.go

@@ -46,182 +46,183 @@ type ringidx struct {
 
 // Reset resets the RingTx's internal state to use buf as the main ring buffer and creates or reuses
 // the packet ring buffer.
-func (rx *ringTx) Reset(buf []byte, maxqueuedPackets int, seq Value) error {
+func (rtx *ringTx) Reset(buf []byte, maxqueuedPackets int, seq Value) error {
 	if maxqueuedPackets <= 0 {
 		return errors.New("queued packets <=0")
 	} else if len(buf) < minBufferSize || len(buf) < maxqueuedPackets {
 		return errors.New("invalid buffer size")
 	}
-	if cap(rx.packets) < maxqueuedPackets {
+	if cap(rtx.packets) < maxqueuedPackets {
-		rx.packets = make([]ringidx, maxqueuedPackets)
+		rtx.packets = make([]ringidx, maxqueuedPackets)
 	}
-	*rx = ringTx{
+	*rtx = ringTx{
 		rawbuf:  buf,
-		packets: rx.packets[:maxqueuedPackets],
+		packets: rtx.packets[:maxqueuedPackets],
 		seq:     seq,
 	}
-	for i := range rx.packets {
+	for i := range rtx.packets {
-		rx.packets[i].markRcvd()
+		rtx.packets[i].markRcvd()
 	}
 	return nil
 }
 
 // ResetOrReuse is identical to a call to [ringTx.Reset] with the additional detail that
 // the zero value of buf (nil) and maxQueuedPackets (0) will selectively reuse existing data buffer and/or packet index buffer.
-func (rx *ringTx) ResetOrReuse(buf []byte, maxQueuedPackets int, ack Value) error {
+func (rtx *ringTx) ResetOrReuse(buf []byte, maxQueuedPackets int, ack Value) error {
 	if buf == nil {
-		buf = rx.rawbuf
+		buf = rtx.rawbuf
 	}
 	if maxQueuedPackets == 0 {
-		maxQueuedPackets = len(rx.packets)
+		maxQueuedPackets = len(rtx.packets)
 	}
-	return rx.Reset(buf, maxQueuedPackets, ack)
+	return rtx.Reset(buf, maxQueuedPackets, ack)
 }
 
 // Size returns the total storage space of the transmission buffer.
-func (tx *ringTx) Size() int { return len(tx.rawbuf) }
+func (rtx *ringTx) Size() int { return len(rtx.rawbuf) }
 
 // Free returns the total available space for Write calls.
-func (tx *ringTx) Free() int {
+func (rtx *ringTx) Free() int {
-	r := tx.sentAndUnsentBuffer()
+	r := rtx.sentAndUnsentBuffer()
 	return r.Free()
 }
 
-// Buffered returns the amount of unsent bytes.
+// Buffered returns the amount of written but unsent bytes.
-func (tx *ringTx) Buffered() int {
+func (rtx *ringTx) Buffered() int {
-	r, _ := tx.unsentRing()
+	r, _ := rtx.unsentRing()
 	return r.Buffered()
 }
 
 // BufferedSent returns the total amount of bytes sent but not acked.
-func (tx *ringTx) BufferedSent() int {
+func (rtx *ringTx) BufferedSent() int {
-	r, _ := tx.sentRing()
+	r, _ := rtx.sentRing()
 	return r.Buffered()
 }
 
 // Write writes data to the underlying unsent data ring buffer.
-func (tx *ringTx) Write(b []byte) (n int, err error) {
+func (rtx *ringTx) Write(b []byte) (n int, err error) {
-	r, lim := tx.unsentRing()
+	r, lim := rtx.unsentRing()
 	n, err = r.WriteLimited(b, lim)
 	if err != nil {
 		return 0, err
 	}
-	tx.unsentend = tx.addEnd(tx.unsentend, n)
+	rtx.unsentend = rtx.addEnd(rtx.unsentend, n)
 	return n, err
 }
 
 // MakePacket reads from the unsent data ring buffer and generates a new packet segment.
 // It fails if the sent packet queue is full.
-func (tx *ringTx) MakePacket(b []byte) (int, Value, error) {
+func (rtx *ringTx) MakePacket(b []byte) (int, Value, error) {
-	nxtpkt := tx.nextPkt()
+	nxtpkt := rtx.nextPkt()
-	if tx.nextPkt() < 0 {
+	if rtx.nextPkt() < 0 {
 		return 0, 0, errors.New("queue full")
 	}
-	r, _ := tx.unsentRing()
+	r, _ := rtx.unsentRing()
 	start := r.Off
 	n, err := r.Read(b)
 	if err != nil {
 		return n, 0, err
 	}
-	pkt := &tx.packets[nxtpkt]
+	pkt := &rtx.packets[nxtpkt]
 
-	off := tx.addEnd(tx.unsentoff, n)
+	off := rtx.addEnd(rtx.unsentoff, n)
-	tx.unsentoff = off
+	rtx.unsentoff = off
-	tx.sentend = off
+	rtx.sentend = off
-	if off == tx.unsentend {
+	if off == rtx.unsentend {
-		tx.unsentend = 0 // Mark unsent as being empty.
+		rtx.unsentend = 0 // Mark unsent as being empty.
 	}
 	pkt.off = start
 	pkt.end = off
 
 	// Sequence number updates.
-	oldseq := tx.seq
+	oldseq := rtx.seq
 	newseq := Add(oldseq, Size(n))
-	tx.seq = newseq
+	rtx.seq = newseq
 	pkt.seq = newseq
 	return n, oldseq, nil
 }
 
 // RecvSegment processes an incoming segment and updates the sent packet queue
-func (tx *ringTx) RecvACK(ack Value) error {
+func (rtx *ringTx) RecvACK(ack Value) error {
-	if ack.LessThan(tx.seq) {
+	if ack.LessThan(rtx.seq) {
 		return errors.New("old packet")
 	}
-	first := tx.firstPkt()
+	first := rtx.firstPkt()
 	if first < 0 {
 		return errors.New("no packets to ack")
 	}
-	hiSeq := tx.pkt(first).seq
+	hiSeq := rtx.pkt(first).seq
-	for i := 0; i < len(tx.packets); i++ {
+	for i := 0; i < len(rtx.packets); i++ {
-		pkt := &tx.packets[i]
+		pkt := &rtx.packets[i]
 		if pkt.sent() && pkt.seq.LessThanEq(ack) {
-			if hiSeq.LessThan(pkt.seq) {
+			if hiSeq.LessThanEq(pkt.seq) {
-				tx.sentoff = pkt.end
+				rtx.sentoff = pkt.end
 				hiSeq = pkt.seq
 			}
 			pkt.markRcvd()
 		}
 	}
-	firstAcked := !tx.pkt(first).sent()
+	firstAcked := !rtx.pkt(first).sent()
-	if firstAcked && tx.sentoff == tx.sentend {
+	if firstAcked && rtx.sentoff == rtx.sentend {
 		// All data acked.
-		tx.sentend = 0
+		rtx.sentend = 0
+		rtx.consolidateBufs()
 	}
 	return nil
 }
 
-func (tx *ringTx) sentAndUnsentBuffer() internal.Ring {
+func (rtx *ringTx) sentAndUnsentBuffer() internal.Ring {
-	end := tx.unsentend
+	end := rtx.unsentend
 	if end == 0 {
-		end = tx.sentend
+		end = rtx.sentend
 	}
-	return internal.Ring{Buf: tx.rawbuf, Off: tx.sentoff, End: end}
+	return internal.Ring{Buf: rtx.rawbuf, Off: rtx.sentoff, End: end}
 }
 
-func (tx *ringTx) unsentRing() (internal.Ring, int) {
+func (rtx *ringTx) unsentRing() (internal.Ring, int) {
-	return tx.ring(tx.unsentoff, tx.unsentend), tx.sentoff
+	return rtx.ring(rtx.unsentoff, rtx.unsentend), rtx.sentoff
 }
 
-func (tx *ringTx) sentRing() (internal.Ring, int) {
+func (rtx *ringTx) sentRing() (internal.Ring, int) {
-	return tx.ring(tx.sentoff, tx.sentend), tx.unsentoff // unsentoff should match with sentend, so no writes can be performed to sentring.
+	return rtx.ring(rtx.sentoff, rtx.sentend), rtx.unsentoff // unsentoff should match with sentend, so no writes can be performed to sentring.
 }
 
-func (tx *ringTx) ring(off, end int) internal.Ring {
+func (rtx *ringTx) ring(off, end int) internal.Ring {
-	return internal.Ring{Buf: tx.rawbuf, Off: off, End: end}
+	return internal.Ring{Buf: rtx.rawbuf, Off: off, End: end}
 }
 
 // addEnd adds two integers together and wraps the value around the ring's buffer size.
 // Result of addEnd will never be 0 unless arguments are (0,0).
-func (tx *ringTx) addEnd(a, b int) int {
+func (rtx *ringTx) addEnd(a, b int) int {
 	result := a + b
-	if result > len(tx.rawbuf) {
+	if result > len(rtx.rawbuf) {
-		result -= len(tx.rawbuf)
+		result -= len(rtx.rawbuf)
 	}
 	return result
 }
 
-func (tx *ringTx) addOff(a, b int) int {
+func (rtx *ringTx) addOff(a, b int) int {
 	result := a + b
-	if result >= len(tx.rawbuf) {
+	if result >= len(rtx.rawbuf) {
-		result -= len(tx.rawbuf)
+		result -= len(rtx.rawbuf)
 	}
 	return result
 }
 
-func (tx *ringTx) pkt(i int) *ringidx {
+func (rtx *ringTx) pkt(i int) *ringidx {
 	if i == -1 {
-		return &tx.emptyRing
+		return &rtx.emptyRing
-	} else if i < 0 || i >= len(tx.packets) {
+	} else if i < 0 || i >= len(rtx.packets) {
 		panic("invalid packet index")
 	}
-	return &tx.packets[i]
+	return &rtx.packets[i]
 }
 
-func (tx *ringTx) firstPkt() int {
+func (rtx *ringTx) firstPkt() int {
 	var seq Value
 	idx := -1
-	for i := 0; i < len(tx.packets); i++ {
+	for i := 0; i < len(rtx.packets); i++ {
-		pkt := &tx.packets[i]
+		pkt := &rtx.packets[i]
 		if pkt.sent() && (idx == -1 || seq.LessThan(pkt.seq)) {
 			seq = pkt.seq
 			idx = i
@@ -230,11 +231,11 @@ func (tx *ringTx) firstPkt() int {
 	return idx
 }
 
-func (tx *ringTx) lastPkt() int {
+func (rtx *ringTx) lastPkt() int {
 	var seq Value
 	idx := -1
-	for i := 0; i < len(tx.packets); i++ {
+	for i := 0; i < len(rtx.packets); i++ {
-		pkt := &tx.packets[i]
+		pkt := &rtx.packets[i]
 		if pkt.sent() && (idx == -1 || pkt.seq.LessThan(seq)) {
 			seq = pkt.seq
 			idx = i
@@ -243,10 +244,10 @@ func (tx *ringTx) lastPkt() int {
 	return idx
 }
 
-func (tx *ringTx) nextPkt() int {
+func (rtx *ringTx) nextPkt() int {
 	idx := -1
-	for i := 0; i < len(tx.packets); i++ {
+	for i := 0; i < len(rtx.packets); i++ {
-		pkt := &tx.packets[i]
+		pkt := &rtx.packets[i]
 		if !pkt.sent() {
 			idx = i
 			break
@@ -255,6 +256,18 @@ func (tx *ringTx) nextPkt() int {
 	return idx
 }
 
+func (rtx *ringTx) consolidateBufs() {
+	unsentEmpty := rtx.unsentend == 0
+	sentEmpty := rtx.sentend == 0
+	if unsentEmpty && sentEmpty {
+		// reset start of buffers.
+		rtx.sentoff = 0
+		rtx.unsentoff = 0
+	}
+}
+
+func (rtx *ringTx) currentSeq() Value { return rtx.seq }
+
 // lims returns the limits of free|sent|unsent buffers.
 // Example:
 //

tcp/txqueue_test.go

@@ -4,11 +4,12 @@ import (
 	"bytes"
 	"fmt"
 	"math/rand"
+	"slices"
 	"testing"
 )
 
 func TestTxQueue(t *testing.T) {
-	var msgBuf, buf, aux [1024]byte
+	var msgBuf, ringBuf, readBuf, aux [1024]byte
 	rng := rand.New(rand.NewSource(1))
 
 	var rtx ringTx
@@ -22,20 +23,54 @@ func TestTxQueue(t *testing.T) {
 		0: {
 			name: "SequentialMessages",
 			test: func(t *testing.T) {
+				const startAck = 0
 				for i := 0; i < 10; i++ {
 					rng.Read(msgBuf[:])
-					msgs := bytes.SplitAfter(msgBuf[:], []byte{0})
+					msgs := removeEmptyMsgs(bytes.SplitAfter(msgBuf[:], []byte{0}))
-					testTxQueue_SequentialMessages(t, &rtx, msgs, buf[:], aux[:], rng.Intn(4)+1, 0)
+					currentAck := Value(startAck)
+					rtx.Reset(ringBuf[:], rng.Intn(4)+1, startAck)
+					for imsg, msg := range msgs {
+						// Write and create packet from single messages.
+						currentAck = Add(currentAck, Size(len(msg)))
+						operateOnRing(t, &rtx, msg, readBuf[:], aux[:], &currentAck)
+						buffered := rtx.Buffered()
+						if buffered != 0 {
+							t.Fatalf("msg%d: want no buffered data after transaction, got %d", imsg, buffered)
+						}
+						newSeq := rtx.currentSeq()
+						wantSeq := currentAck
+						if newSeq != wantSeq {
+							t.Fatalf("msg%d: want seq %d, got %d", imsg, wantSeq, newSeq)
+						}
+						if t.Failed() {
+							t.Fatalf("failed on msg %d", imsg)
+						}
+					}
 				}
 			},
 		},
 		1: {
 			name: "N-Messages",
 			test: func(t *testing.T) {
+				const startAck = 0
 				for i := 0; i < 10; i++ {
 					rng.Read(msgBuf[:])
-					msgs := bytes.SplitAfter(msgBuf[:], []byte{0})
+					msgs := removeEmptyMsgs(bytes.SplitAfter(msgBuf[:], []byte{0}))
-					testTxQueue_NMessages(t, &rtx, msgs, buf[:], aux[:], len(msgs), 0)
+					currentAck := Value(startAck)
+					rtx.Reset(ringBuf[:], rng.Intn(4)+1, startAck)
+					for _, msg := range msgs {
+						// Send all messages.
+						operateOnRing(t, &rtx, msg, nil, aux[:], nil)
+						if t.Failed() {
+							return
+						}
+						gotSeq := rtx.currentSeq()
+						if gotSeq != startAck {
+							t.Fatalf("expected seq to not change during writes")
+						}
+						currentAck = Add(currentAck, Size(len(msg)))
+					}
+					operateOnRing(t, &rtx, nil, readBuf[:], aux[:], &currentAck)
 				}
 			},
 		},
@@ -76,12 +111,12 @@ func testTxQueue_NMessages(t *testing.T, rtx *ringTx, msgs [][]byte, buf, aux []
 		testQueueSanity(t, rtx)
 		unsent := rtx.Buffered()
 		if unsent != n {
-			t.Fatalf("want unset %d, got %d", n, unsent)
+			t.Fatalf("want unsent %d, got %d", n, unsent)
 		}
 		testQueueSanity(t, rtx)
 		n, seq, err := rtx.MakePacket(aux[sent : sent+len(msg)])
 		if err != nil {
-			t.Fatal(err)
+			t.Fatal("MakePacket: ", err)
 		} else if seq != prevSeq {
 			t.Fatalf("want seq %d, got %d", prevSeq, seq)
 		} else if n != len(msg) {
@@ -106,6 +141,10 @@ func testTxQueue_SequentialMessages(t *testing.T, rtx *ringTx, msgs [][]byte, bu
 	}
 	prevSeq := Value(startAck)
 	for i, msg := range msgs {
+		if t.Failed() {
+			t.Errorf("%s failed on message %d", t.Name(), i)
+			return
+		}
 		if len(aux) < len(msg) {
 			panic("need aux to contain message")
 		}
@@ -148,17 +187,30 @@ func testTxQueue_SequentialMessages(t *testing.T, rtx *ringTx, msgs [][]byte, bu
 		if err != nil {
 			t.Fatal(err)
 		}
+		sent = rtx.BufferedSent()
+		unsent = rtx.Buffered()
+		if sent != 0 {
+			t.Errorf("message not marked as sent- expected no data left got %d", sent)
+		}
+		if unsent != 0 {
+			t.Errorf("huge bug, unexpected data loaded to unsent buffer")
+		}
 		testQueueSanity(t, rtx)
 	}
 }
 
 func testQueueSanity(t *testing.T, rtx *ringTx) {
 	// t.Helper()
-	defer func() {
+	alreadyFailed := t.Failed()
-		if t.Failed() {
+	if !alreadyFailed {
-			t.Log("\n" + rtx.string())
+		defer func() {
-		}
+			if t.Failed() {
-	}()
+				t.Helper()
+				t.Log("sanity failed with:\n" + rtx.string())
+			}
+		}()
+	}
+
 	if rtx.emptyRing != (ringidx{}) {
 		t.Fatalf("empty ring not empty")
 	}
@@ -184,7 +236,7 @@ func testQueueSanity(t *testing.T, rtx *ringTx) {
 		} else if rsent.End == 0 {
 			t.Fatalf("expected not empty sent buffer End to be !=0, got %d", rsent.End)
 		}
-		gotSentEnd := rtx.addOff(rsent.Off, sent)
+		gotSentEnd := rtx.addEnd(rsent.Off, sent)
 		if gotSentEnd != rsent.End {
 			t.Fatalf("calculated sent end mismatches lim sent end %d != %d", gotSentEnd, rsent.End)
 		}
@@ -195,7 +247,7 @@ func testQueueSanity(t *testing.T, rtx *ringTx) {
 		} else if runsent.End == 0 {
 			t.Fatalf("expected not empty unsent buffer End to be !=0, got %d", runsent.End)
 		}
-		gotUnsentEnd := rtx.addOff(runsent.Off, unsent)
+		gotUnsentEnd := rtx.addEnd(runsent.Off, unsent)
 		if gotUnsentEnd != runsent.End {
 			t.Fatalf("calculated unsent end mismatches lim unsent end %d != %d", gotUnsentEnd, runsent.End)
 		}
@@ -276,3 +328,106 @@ func (rx *ringTx) string() string {
 	l2.WriteTo(&l1)
 	return l1.String()
 }
+
+func removeEmptyMsgs(msgs [][]byte) [][]byte {
+	return slices.DeleteFunc(msgs, func(b []byte) bool { return len(b) == 0 })
+}
+
+func operateOnRing(t *testing.T, rtx *ringTx, write, readPacket, aux []byte, argRecvAck *Value) {
+	if len(aux) < rtx.Size() {
+		panic("too small auxiliary buffer")
+	}
+	free := rtx.Free()
+	// Prepare aux with data expected from read after write.
+	runsent, _ := rtx.unsentRing()
+	unsent := runsent.Buffered()
+
+	wantWritten := min(free, len(write))
+	wantBufRead := aux[:min(unsent+wantWritten, len(readPacket))]
+
+	if len(wantBufRead) > 0 {
+		testQueueSanity(t, rtx)
+		var n int
+		if runsent.Buffered() > 0 {
+			ngot, err := runsent.Read(wantBufRead)
+			wantRead := len(wantBufRead)
+			if err != nil {
+				panic(err)
+			} else if ngot < wantRead {
+				panic("expected read of at least length calculated above")
+			}
+			n = ngot
+		}
+		copy(wantBufRead[n:], write)
+	}
+
+	prevSeq := rtx.currentSeq()
+	if len(write) != 0 {
+		testQueueSanity(t, rtx)
+		preBuffered := rtx.Buffered()
+		n, err := rtx.Write(write)
+		if err != nil && wantWritten > 0 {
+			t.Errorf("error writing packet: %s", err)
+		} else if n != wantWritten {
+			t.Errorf("want %d written, got %d", wantWritten, n)
+		}
+		newBuffered := rtx.Buffered()
+		gotWritten := newBuffered - preBuffered
+		if gotWritten != wantWritten {
+			t.Errorf("expected %d data written, got %d", wantWritten, gotWritten)
+		}
+	}
+
+	if !t.Failed() && len(readPacket) != 0 {
+		testQueueSanity(t, rtx)
+		preSent := rtx.BufferedSent()
+		canRead := rtx.Buffered()
+		wantRead := min(canRead, len(readPacket))
+		if wantRead != len(wantBufRead) {
+			t.Fatalf("miscalculated expect read %d != %d", wantRead, len(wantBufRead))
+		}
+		n, seq, err := rtx.MakePacket(readPacket)
+		if err != nil && wantRead != 0 {
+			t.Errorf("error reading: %s", err)
+		} else if n != wantRead {
+			t.Errorf("want read %d, got %d", wantRead, n)
+		}
+		wantSeq := prevSeq
+		if seq != wantSeq {
+			t.Errorf("want new seq %d, got %d", wantSeq, seq)
+		}
+		if !bytes.Equal(readPacket[:n], wantBufRead) {
+			t.Error("data content packet read not match wanted packet")
+		}
+		gotCalcRead := rtx.BufferedSent() - preSent
+		if gotCalcRead != n {
+			t.Errorf("want data written to be %d calculated from BufferedSent diff, got %d", n, gotCalcRead)
+		}
+	}
+
+	if !t.Failed() && argRecvAck != nil {
+		testQueueSanity(t, rtx)
+		preAcked := rtx.BufferedSent()
+		rcvAck := *argRecvAck
+		seq := rtx.currentSeq()
+		startSeq := Add(seq, Size(-rtx.BufferedSent()))
+		acklInSentRange := startSeq.LessThan(rcvAck) && rcvAck.LessThanEq(seq)
+		err := rtx.RecvACK(rcvAck)
+		if err != nil && acklInSentRange {
+			t.Errorf("expected correct acking %d < %d <= %d: %s", startSeq, rcvAck, seq, err)
+		}
+		gotCalcAcked := preAcked - rtx.BufferedSent()
+		wantAcked := int(Sizeof(prevSeq, rtx.currentSeq()))
+		if gotCalcAcked != wantAcked {
+			t.Errorf("want acked %d, got %d", wantAcked, gotCalcAcked)
+		}
+	}
+	testQueueSanity(t, rtx)
+}
+
+func min(a, b int) int {
+	if a < b {
+		return a
+	}
+	return b
+}