use sentlist in ringTx implementation and back to square 1

tcp/txqueue.go

@@ -2,11 +2,15 @@ package tcp
 
 import (
 	"errors"
-	"fmt"
+	"slices"
 
 	"github.com/soypat/lneto/internal"
 )
 
+var (
+	errPacketQueueFull = errors.New("packet queue full")
+)
+
 const (
 	// this must be at least 2 for buffer to work.
 	minBufferSize = 2
@@ -19,8 +23,7 @@ const (
 type ringTx struct {
 	// rawbuf contains the ring buffer of ordered bytes. It should be the size of the window.
 	rawbuf []byte
-	// packets contains
-	packets []ringidx
+	slist  sentlist
 	// unsentOff is the offset of start of unsent data in rawbuf.
 	unsentoff int
 	// unsentend is the offset of end of unsent data in rawbuf. If zero then unsent buffer is empty.
@@ -57,16 +60,11 @@ func (rtx *ringTx) Reset(buf []byte, maxqueuedPackets int, iss Value) error {
 	} else if len(buf) < minBufferSize || len(buf) < maxqueuedPackets {
 		return errors.New("invalid buffer size")
 	}
-	if cap(rtx.packets) < maxqueuedPackets {
-		rtx.packets = make([]ringidx, maxqueuedPackets)
-	}
+
 	*rtx = ringTx{
-		rawbuf:  buf,
-		packets: rtx.packets[:maxqueuedPackets],
-	}
-	for i := range rtx.packets {
-		rtx.packets[i].markRcvd()
+		rawbuf: buf,
 	}
+	rtx.slist.Reset(maxqueuedPackets, iss)
 	rtx.iss = iss
 	return nil
 }
@@ -78,7 +76,7 @@ func (rtx *ringTx) ResetOrReuse(buf []byte, maxQueuedPackets int, ack Value) err
 		buf = rtx.rawbuf
 	}
 	if maxQueuedPackets == 0 {
-		maxQueuedPackets = len(rtx.packets)
+		maxQueuedPackets = cap(rtx.slist.pkts)
 	}
 	return rtx.Reset(buf, maxQueuedPackets, ack)
 }
@@ -118,106 +116,50 @@ func (rtx *ringTx) Write(b []byte) (n int, err error) {
 // MakePacket reads from the unsent data ring buffer and generates a new packet segment.
 // It fails if the sent packet queue is full.
 func (rtx *ringTx) MakePacket(b []byte, currentSeq Value) (int, error) {
-	nxtpkt := rtx.nextPkt()
-	if nxtpkt < 0 {
-		return 0, errors.New("queue full")
+	free := rtx.slist.Free()
+	if free == 0 {
+		return 0, errPacketQueueFull
 	}
 	endSeq, ok := rtx.endSeq()
 	if ok && currentSeq.LessThan(endSeq) {
 		return 0, errors.New("sequence number less than last sequence number")
 	}
+	// Reading unsent ring consumes unsent and converts it to "sent".
 	r, _ := rtx.unsentRing()
-	start := r.Off
+	oldSentOff := r.Off
 	n, err := r.Read(b)
 	if err != nil {
 		return n, err
 	}
-	pkt := &rtx.packets[nxtpkt]
-
-	off := rtx.addEnd(rtx.unsentoff, n)
-	rtx.unsentoff = off
-	rtx.sentend = off
-	if off == rtx.unsentend {
-		rtx.unsentend = 0 // Mark unsent as being empty.
+	// unsentOff increases, sentEnd matches this value.
+	// Start of buffer will be SENT, end of buffer will be UNSENT(or empty).
+	// Packet generated has offset at old unsentOff.
+	newUnsentOff := rtx.addEnd(rtx.unsentoff, n)
+	pkt := rtx.slist.AddPacket(n, oldSentOff, rtx.Size())
+	if pkt.off != oldSentOff || pkt.end != addEnd(pkt.off, n, rtx.Size()) {
+		panic("invalid generated packet")
 	}
-	*pkt = ringidx{
-		off:  start,
-		end:  off,
-		seq:  currentSeq,
-		size: Size(n),
+	rtx.unsentoff = newUnsentOff
+	rtx.sentend = newUnsentOff
+	if newUnsentOff == rtx.unsentend {
+		rtx.unsentend = 0 // Mark unsent as being empty.
 	}
 	return n, nil
 }
 
 // RecvSegment processes an incoming segment and updates the sent packet queue
 func (rtx *ringTx) RecvACK(ack Value) error {
-	first := rtx.firstPkt()
-	if first < 0 {
-		return errors.New("no packets to ack")
-	}
-	pkt0 := rtx.pkt(first)
-	if ack.LessThanEq(pkt0.seq) {
-		return fmt.Errorf("incoming ack %d older than first packet seq %d", ack, pkt0.seq)
-	}
-	// lastAckedPkt stores last fully acked packet.
-	var lastAckedPkt *ringidx
-	var partialPkt *ringidx
-	for i := 0; i < len(rtx.packets); i++ {
-		pkt := &rtx.packets[i]
-		if !pkt.sent() || ack.LessThanEq(pkt.seq) {
-			continue
-		}
-		endseq := pkt.endSeq()
-		isFullyAcked := endseq.LessThanEq(ack)
-		isPartialAcked := ack.InRange(pkt.seq, endseq)
-		isLast := lastAckedPkt == nil || lastAckedPkt.seq.LessThanEq(pkt.seq)
-		isBeforeLast := lastAckedPkt != nil && !isLast
-		if isFullyAcked == isPartialAcked { // is either or.
-			panic("unreachable")
-		}
-		if isLast && isFullyAcked {
-			if lastAckedPkt != nil {
-				lastAckedPkt.markRcvd()
-			}
-			lastAckedPkt = pkt
-		} else if isBeforeLast {
-			if isPartialAcked {
-				panic("unreachable")
-			}
-			pkt.markRcvd()
-		} else if !isPartialAcked {
-			panic("unreachable")
-		} else {
-			// Is partial acked.
-			if partialPkt != nil {
-				panic("unreachable") // can't have two partially acked packets.
-			}
-			acked := int(ack - pkt.seq)
-			pring := rtx.ring(pkt.off, pkt.end)
-			buffered := pring.Buffered()
-			if acked > buffered {
-				panic("unreachable")
-			}
-			off := rtx.addOff(pkt.off, acked)
-			pkt.off = off
-			pkt.seq = ack
-			pkt.size = pkt.size - Size(acked)
-			rtx.sentoff = off
-			partialPkt = pkt
-		}
-	}
-	if partialPkt != nil {
-		return nil
-	}
-	if lastAckedPkt != nil {
-		rtx.sentoff = lastAckedPkt.end
-		lastAckedPkt.markRcvd()
-		if rtx.sentoff == rtx.sentend {
-			// All data acked.
-			rtx.sentend = 0
-			rtx.consolidateBufs()
-		}
+	rtx.slist.RecvAck(ack, rtx.Size())
+	oldest := rtx.slist.Oldest()
+	newest := rtx.slist.Newest()
+	if oldest == nil {
+		// All sent data received, discard.
+		rtx.sentend = 0
+	} else {
+		rtx.sentoff = oldest.off
+		rtx.sentend = newest.end
 	}
+	rtx.consolidateBufs()
 	return nil
 }
 
@@ -245,55 +187,6 @@ func (rtx *ringTx) ring(off, end int) internal.Ring {
 // Result of addEnd will never be 0 unless arguments are (0,0).
 func (rtx *ringTx) addEnd(a, b int) int { return addEnd(a, b, len(rtx.rawbuf)) }
 
-func (rtx *ringTx) addOff(a, b int) int { return addOff(a, b, len(rtx.rawbuf)) }
-
-func (rtx *ringTx) pkt(i int) *ringidx {
-	if i == -1 {
-		return &rtx.emptyRing
-	} else if i < 0 || i >= len(rtx.packets) {
-		panic("invalid packet index")
-	}
-	return &rtx.packets[i]
-}
-
-func (rtx *ringTx) firstPkt() int {
-	var seq Value
-	idx := -1
-	for i := 0; i < len(rtx.packets); i++ {
-		pkt := &rtx.packets[i]
-		if pkt.sent() && (idx == -1 || pkt.seq.LessThan(seq)) {
-			seq = pkt.seq
-			idx = i
-		}
-	}
-	return idx
-}
-
-func (rtx *ringTx) lastPkt() int {
-	var seq Value
-	idx := -1
-	for i := 0; i < len(rtx.packets); i++ {
-		pkt := &rtx.packets[i]
-		if pkt.sent() && (idx == -1 || seq.LessThan(pkt.seq)) {
-			seq = pkt.seq
-			idx = i
-		}
-	}
-	return idx
-}
-
-func (rtx *ringTx) nextPkt() int {
-	idx := -1
-	for i := 0; i < len(rtx.packets); i++ {
-		pkt := &rtx.packets[i]
-		if !pkt.sent() {
-			idx = i
-			break
-		}
-	}
-	return idx
-}
-
 func (rtx *ringTx) consolidateBufs() {
 	unsentEmpty := rtx.unsentend == 0
 	sentEmpty := rtx.sentend == 0
@@ -305,26 +198,11 @@ func (rtx *ringTx) consolidateBufs() {
 }
 
 func (rtx *ringTx) endSeq() (Value, bool) {
-	pkt := rtx.lastPkt()
-	if pkt < 0 {
+	newest := rtx.slist.Newest()
+	if newest == nil {
 		return 0, false
 	}
-	last := rtx.pkt(pkt)
-	return last.endSeq(), true
-}
-func (rtx *ringTx) lastSeq() (Value, bool) {
-	pkt := rtx.lastPkt()
-	if pkt < 0 {
-		return 0, false
-	}
-	return rtx.pkt(pkt).seq, true
-}
-func (rtx *ringTx) firstSeq() (Value, bool) {
-	pkt := rtx.firstPkt()
-	if pkt < 0 {
-		return 0, false
-	}
-	return rtx.pkt(pkt).seq, true
+	return newest.endSeq(), true
 }
 
 // lims returns the limits of free|sent|unsent buffers.
@@ -356,11 +234,19 @@ func (pkt *ringidx) endSeq() Value {
 
 // sentlist stores information about sent TCP packets
 type sentlist struct {
+	// ssn is an auxiliary sequence counter.
+	// If there are no packets then ssn is reset to be the end sequence number of the last acked packet such that
+	// the next packet added has their
+	ssn Value
 	// pkts is an ordered list of packets. First packet is 'oldest' packet, last packet is the most recently sent.
-	iss  Value
 	pkts []ringidx
 }
 
+func (sl *sentlist) Reset(pktQueueSize int, iss Value) {
+	sl.pkts = slices.Grow(sl.pkts[:0], pktQueueSize)
+	sl.ssn = iss
+}
+
 func (sl sentlist) Newest() *ringidx {
 	if len(sl.pkts) == 0 {
 		return nil
@@ -376,7 +262,7 @@ func (sl sentlist) Oldest() *ringidx {
 }
 
 func (sl *sentlist) EndSeq() Value {
-	seq := sl.iss
+	seq := sl.ssn
 	lastPkt := sl.Newest()
 	if lastPkt != nil {
 		seq = lastPkt.endSeq()
@@ -384,23 +270,26 @@ func (sl *sentlist) EndSeq() Value {
 	return seq
 }
 
-func (sl *sentlist) AddPacket(datalen int, bufsize int) {
-	free := cap(sl.pkts) - len(sl.pkts)
+func (sl *sentlist) Free() int {
+	return cap(sl.pkts) - len(sl.pkts)
+}
+
+func (sl *sentlist) AddPacket(datalen, off, bufsize int) *ringidx {
+	free := sl.Free()
 	if free == 0 {
 		panic("pkt buffer full")
 	}
 	lastPkt := sl.Newest()
-	lastEnd := 0
-	if lastPkt != nil {
-		lastEnd = lastPkt.end
+	if lastPkt != nil && off != lastPkt.end {
+		panic("new sent packet offset must match last sent packet end")
 	}
-	pkt := ringidx{
-		off:  lastEnd,
-		end:  addEnd(lastEnd, datalen, bufsize),
+	sl.pkts = append(sl.pkts, ringidx{
+		off:  off,
+		end:  addEnd(off, datalen, bufsize),
 		seq:  sl.EndSeq(),
 		size: Size(datalen),
-	}
-	sl.pkts = append(sl.pkts, pkt)
+	})
+	return &sl.pkts[len(sl.pkts)-1]
 }
 
 func (sl *sentlist) RecvAck(ack Value, bufsize int) {
@@ -410,7 +299,7 @@ func (sl *sentlist) RecvAck(ack Value, bufsize int) {
 		endseq := pkt.endSeq()
 		isFullyAcked := endseq.LessThanEq(ack)
 		if isFullyAcked {
-			sl.iss = endseq
+			sl.ssn = endseq
 			pkt.markRcvd()
 		} else {
 			break
@@ -432,6 +321,9 @@ func (sl *sentlist) RecvAck(ack Value, bufsize int) {
 }
 
 func (sl *sentlist) removeRecvd() {
+	if !sl.Oldest().isRecvd() {
+		return // No packets to remove.
+	}
 	off := 0
 	for i := 0; i < len(sl.pkts); i++ {
 		if sl.pkts[i].isRecvd() {

tcp/txqueue_test.go

@@ -8,14 +8,42 @@ import (
 	"testing"
 )
 
-func TestSentlist(t *testing.T) {
-	sl := sentlist{
-		pkts: make([]ringidx, 0, 3),
+func TestSentlist_multi(t *testing.T) {
+	const bufsize = 10
+	var sl sentlist
+	sl.Reset(3, 0)
+
+	// Test multi packet x2.
+	p1 := sl.AddPacket(5, 0, bufsize)
+	p2 := sl.AddPacket(5, p1.end, bufsize)
+	sl.RecvAck(Value(p2.size+p1.size), bufsize)
+	if sl.Oldest() != nil {
+		t.Fatal("expected full ack")
 	}
+	// multi packet x3.
+	sl.Reset(3, 0)
+	p1 = sl.AddPacket(3, 0, bufsize)
+	p2 = sl.AddPacket(3, p1.end, bufsize)
+	p3 := sl.AddPacket(4, p2.end, bufsize)
+	sl.RecvAck(2, bufsize)
+	oldest := sl.Oldest()
+	if oldest != p1 {
+		t.Error("oldest should be partial acked")
+	} else if oldest.size != 1 {
+		t.Error("bad size")
+	} else if oldest.off != 2 {
+		t.Error("bad offset")
+	}
+	_ = p3
+}
+
+func TestSentlist(t *testing.T) {
+	var sl sentlist
+	sl.Reset(3, 0)
 	// Test full ack.
 	const bufsize = 16
 	const pkt = 10
-	sl.AddPacket(pkt, bufsize)
+	sl.AddPacket(pkt, 0, bufsize)
 	if sl.Oldest() == nil || sl.Newest() != sl.Oldest() {
 		t.Error("expected same oldest/newest non-nil packet")
 	}
@@ -27,7 +55,7 @@ func TestSentlist(t *testing.T) {
 	}
 
 	// Test partial ack.
-	sl.AddPacket(pkt, bufsize)
+	sl.AddPacket(pkt, 0, bufsize)
 	for i := Value(0); i < pkt-1; i++ {
 		ack++
 		sl.RecvAck(ack, bufsize)
@@ -274,6 +302,10 @@ func testQueueSanity(t *testing.T, rtx *ringTx) {
 	alreadyFailed := t.Failed()
 	if !alreadyFailed {
 		defer func() {
+			a := recover()
+			if a != nil {
+				t.Log("panic", a)
+			}
 			if t.Failed() {
 				t.Helper()
 				t.Log("sanity failed with:\n" + rtx.string())
@@ -291,7 +323,7 @@ func testQueueSanity(t *testing.T, rtx *ringTx) {
 	sz := rtx.Size()
 	gotSz := free + sent + unsent
 	if gotSz != sz {
-		t.Fatal("\n" + rtx.string())
+		t.Error("\n" + rtx.string())
 		t.Fatalf("want size=%d, got size=%d (free+sent+unsent=%d+%d+%d)", sz, gotSz, free, sent, unsent)
 	}
 	rsent, _ := rtx.sentRing()
@@ -329,21 +361,21 @@ func testQueueSanity(t *testing.T, rtx *ringTx) {
 	}
 
 	// Check sanenness of last/first packets.
-	last := rtx.lastPkt()
-	first := rtx.firstPkt()
-	if first < 0 && last >= 0 || last < 0 && first >= 0 {
+	last := rtx.slist.Newest()
+	first := rtx.slist.Oldest()
+	if first == nil && last != nil || last == nil && first != nil {
 		t.Fatalf("found first/last(%d,%d) but did not find last/first", first, last)
 	}
 	// Check sent data or return if no sent data available.
 	if sent == 0 {
 		return
 	}
-	lastPkt := rtx.pkt(last)
+
 	endseq, ok := rtx.endSeq()
-	firstPkt := rtx.pkt(first)
-	lastEndSeq := Add(lastPkt.seq, lastPkt.size)
-	if lastPkt.seq.LessThan(firstPkt.seq) {
-		t.Fatalf("first packet not previous to last packet seq, wanted %d<%d", firstPkt.seq, lastPkt.seq)
+
+	lastEndSeq := Add(last.seq, last.size)
+	if last.seq.LessThan(first.seq) {
+		t.Fatalf("first packet not previous to last packet seq, wanted %d<%d", first.seq, last.seq)
 	} else if !ok {
 		t.Fatal("unexpected end sequence not found")
 	} else if lastEndSeq != endseq {
@@ -433,7 +465,12 @@ func operateOnRing(t *testing.T, rtx *ringTx, write, readPacket, aux []byte, new
 	// Prepare aux with data expected from read after write.
 	runsent, _ := rtx.unsentRing()
 	unsent := runsent.Buffered()
-	startSeq, startSeqOK := rtx.firstSeq()
+	oldest := rtx.slist.Oldest()
+	var startSeq Value
+	startSeqOk := oldest != nil
+	if startSeqOk {
+		startSeq = oldest.seq
+	}
 	wantWritten := min(free, len(write))
 	wantBufRead := aux[:min(unsent+wantWritten, len(readPacket))]
 
@@ -483,7 +520,12 @@ func operateOnRing(t *testing.T, rtx *ringTx, write, readPacket, aux []byte, new
 		} else if n != wantRead {
 			t.Errorf("want read %d, got %d", wantRead, n)
 		}
-		lastSeq, lastSeqOK := rtx.lastSeq()
+		last := rtx.slist.Newest()
+		var lastSeq Value
+		lastSeqOK := last != nil
+		if lastSeqOK {
+			lastSeq = last.seq
+		}
 		endSeq, endSeqOK := rtx.endSeq()
 		if !lastSeqOK || lastSeq != newPacketSeq {
 			t.Fatalf("expected last seq to be %d, got %d (or lastSeqOK=%v)", newPacketSeq, lastSeq, lastSeqOK)
@@ -498,9 +540,14 @@ func operateOnRing(t *testing.T, rtx *ringTx, write, readPacket, aux []byte, new
 			t.Errorf("want data written to be %d calculated from BufferedSent diff, got %d", n, gotCalcRead)
 		}
 	}
+	oldest2 := rtx.slist.Oldest()
+	var startSeq2 Value
+	sseqOk := oldest != nil
+	if sseqOk {
+		startSeq2 = oldest2.seq
+	}
 
-	startSeq2, sseqOK := rtx.firstSeq()
-	if sseqOK == startSeqOK && startSeq2 != startSeq {
+	if sseqOk == startSeqOk && startSeq2 != startSeq {
 		t.Fatalf("expected FIRST seq to not change during writes")
 	}
 
@@ -508,7 +555,12 @@ func operateOnRing(t *testing.T, rtx *ringTx, write, readPacket, aux []byte, new
 		testQueueSanity(t, rtx)
 		// preAcked := rtx.BufferedSent()
 		rcvAck := *argRecvAck
-		seq, ok := rtx.firstSeq()
+		oldest := rtx.slist.Oldest()
+		var seq Value
+		ok := oldest != nil
+		if ok {
+			seq = oldest.seq
+		}
 		if !ok {
 			t.Fatal("no first packet found")
 		}
@@ -519,7 +571,12 @@ func operateOnRing(t *testing.T, rtx *ringTx, write, readPacket, aux []byte, new
 			t.Errorf("expected correct acking %d < %d <= %d: %s", startSeq, rcvAck, seq, err)
 		}
 		bufSent := rtx.BufferedSent()
-		gotFirstSeq, ok := rtx.firstSeq()
+		var gotFirstSeq Value
+		oldest = rtx.slist.Oldest()
+		ok = oldest != nil
+		if ok {
+			gotFirstSeq = oldest.seq
+		}
 		if !ok && bufSent != 0 {
 			t.Fatalf("no first packet found after acking")
 		}