tcp:txqueue: rewrite RecvAck logic

tcp/txqueue.go

@@ -2,6 +2,7 @@ package tcp
 
 import (
 	"errors"
+	"fmt"
 
 	"github.com/soypat/lneto/internal"
 )
@@ -118,6 +119,10 @@ func (rtx *ringTx) MakePacket(b []byte, currentSeq Value) (int, error) {
 	if nxtpkt < 0 {
 		return 0, errors.New("queue full")
 	}
+	endSeq, ok := rtx.endSeq()
+	if ok && currentSeq.LessThan(endSeq) {
+		return 0, errors.New("sequence number less than last sequence number")
+	}
 	r, _ := rtx.unsentRing()
 	start := r.Off
 	n, err := r.Read(b)
@@ -149,26 +154,62 @@ func (rtx *ringTx) RecvACK(ack Value) error {
 	}
 	pkt0 := rtx.pkt(first)
 	if ack.LessThanEq(pkt0.seq) {
-		return errors.New("old packet")
+		return fmt.Errorf("incoming ack %d older than first packet seq %d", ack, pkt0.seq)
+		// return errors.New("old packet")
 	}
-
-	hiSeq := Add(pkt0.seq, pkt0.size)
+	// lastAckedPkt stores last fully acked packet.
+	var lastAckedPkt *ringidx
 	for i := 0; i < len(rtx.packets); i++ {
 		pkt := &rtx.packets[i]
-		pktendSeq := Add(pkt.seq, pkt.size)
-		if pkt.sent() && pktendSeq.LessThanEq(ack) {
-			if hiSeq.LessThanEq(pktendSeq) {
-				rtx.sentoff = pkt.end
-				hiSeq = pktendSeq
+		if !pkt.sent() || ack.LessThanEq(pkt.seq) {
+			continue
+		}
+		endseq := Add(pkt.seq, pkt.size)
+		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")
+		}
+		if isPartialAcked {
+			if lastAckedPkt != nil && lastAckedPkt.seq.LessThan(pkt.seq) {
+				panic("unreachable")
+			}
+			acked := int(ack - pkt.seq)
+			pring := rtx.ring(pkt.off, pkt.end)
+			buffered := pring.Buffered()
+			if acked > buffered || acked <= minBufferSize {
+				panic("unreachable")
+			}
+			off := rtx.addOff(pkt.off, acked)
+			pkt.off = off
+			pkt.seq = ack
+			pkt.size = pkt.size - Size(acked)
+			rtx.sentoff = off
 		}
 	}
-	firstAcked := !rtx.pkt(first).sent()
-	if firstAcked && rtx.sentoff == rtx.sentend {
-		// All data acked.
-		rtx.sentend = 0
-		rtx.consolidateBufs()
+	if lastAckedPkt != nil {
+		rtx.sentoff = lastAckedPkt.end
+		lastAckedPkt.markRcvd()
+		if rtx.sentoff == rtx.sentend {
+			// All data acked.
+			rtx.sentend = 0
+			rtx.consolidateBufs()
+		}
 	}
 	return nil
 }

tcp/txqueue_test.go

@@ -9,7 +9,8 @@ import (
 )
 
 func TestTxQueue(t *testing.T) {
-	var msgBuf, ringBuf, readBuf, aux [1024]byte
+	const bufsize = 1024
+	var msgBuf, ringBuf, readBuf, aux [bufsize]byte
 	rng := rand.New(rand.NewSource(1))
 
 	var rtx ringTx
@@ -28,7 +29,10 @@ func TestTxQueue(t *testing.T) {
 					rng.Read(msgBuf[:])
 					msgs := removeEmptyMsgs(bytes.SplitAfter(msgBuf[:], []byte{0}))
 					currentAck := Value(startAck)
-					rtx.Reset(ringBuf[:], rng.Intn(4)+1, startAck)
+					err := rtx.Reset(ringBuf[:], rng.Intn(4)+1, startAck)
+					if err != nil {
+						t.Fatal(err)
+					}
 					for imsg, msg := range msgs {
 						// Write and create packet from single messages.
 						seq := currentAck
@@ -38,17 +42,6 @@ func TestTxQueue(t *testing.T) {
 						if buffered != 0 {
 							t.Fatalf("msg%d: want no buffered data after transaction, got %d", imsg, buffered)
 						}
-						// newSeq, ok := rtx.firstSeq()
-						// if !ok {
-						// 	t.Fatal("no first packet found")
-						// }
-						// 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)
-						// }
 					}
 				}
 			},
@@ -61,7 +54,10 @@ func TestTxQueue(t *testing.T) {
 					rng.Read(msgBuf[:])
 					msgs := removeEmptyMsgs(bytes.SplitAfter(msgBuf[:], []byte{0}))
 					currentAck := Value(startAck)
-					rtx.Reset(ringBuf[:], rng.Intn(4)+1, startAck)
+					err := rtx.Reset(ringBuf[:], rng.Intn(4)+1, startAck)
+					if err != nil {
+						t.Fatal(err)
+					}
 					expectBuffered := 0
 					for _, msg := range msgs {
 						// Send all messages.
@@ -95,6 +91,61 @@ func TestTxQueue(t *testing.T) {
 				}
 			},
 		},
+		2: {
+			name: "ParialAcks",
+			test: func(t *testing.T) {
+				const startAck = 0
+				const packets = 100
+				const maxPacketSize = bufsize / 4
+				var datalens [][]byte
+				for i := 0; i < 10; i++ {
+					rng.Read(msgBuf[:])
+					err := rtx.Reset(ringBuf[:], packets, startAck)
+					if err != nil {
+						t.Fatal(err)
+					}
+					operateOnRing(t, &rtx, msgBuf[:], nil, aux[:], 0, nil)
+					// Send all bytes over wire.
+					currentSeq := Value(startAck)
+					datalens = datalens[:0]
+					for rtx.Buffered() != 0 {
+						nbytes := rng.Intn(maxPacketSize-minBufferSize) + minBufferSize
+						n, err := rtx.MakePacket(readBuf[:nbytes], currentSeq)
+						if err != nil {
+							t.Fatal(err)
+						} else if n == 0 {
+							t.Fatal("got zero length")
+						}
+						// Reuse memory in slice of byte buffers.
+						if len(datalens) == cap(datalens) {
+							datalens = append(datalens, append([]byte{}, readBuf[:n]...))
+						} else {
+							datalens = datalens[:len(datalens)+1]
+							datalens[len(datalens)-1] = append(datalens[len(datalens)-1][:0], readBuf[:n]...)
+						}
+						currentSeq += Value(n)
+					}
+					currentAck := Value(startAck)
+					for idata, data := range datalens {
+						plen := len(data)
+						partialLen0 := plen - (rng.Intn(plen)/2 + minBufferSize)
+						// partialLen1 := plen - partialLen0
+						// sent := rtx.BufferedSent()
+						ack1 := currentAck + Value(partialLen0)
+						ack2 := currentAck + Value(plen)
+						err = rtx.RecvACK(ack1)
+						if err != nil {
+							t.Fatalf("data%d acking first partial %d..%d(..%d): %s", idata, currentAck, ack1, ack2, err)
+						}
+						err = rtx.RecvACK(ack2)
+						if err != nil {
+							t.Fatalf("data%d acking second partial (%d..)%d..%d: %s", idata, currentAck, ack1, ack2, err)
+						}
+						currentAck = ack2
+					}
+				}
+			},
+		},
 	}
 	for i, test := range increasingComplexityTests {
 		t.Run(test.name, test.test)