From 8a5f6f3f6d8d13bcdb15a75aedd47a991aac060b Mon Sep 17 00:00:00 2001 From: soypat Date: Sun, 16 Feb 2025 12:42:22 -0300 Subject: [PATCH] tcp:txqueue: rewrite RecvAck logic --- tcp/txqueue.go | 67 ++++++++++++++++++++++++++++++-------- tcp/txqueue_test.go | 79 +++++++++++++++++++++++++++++++++++++-------- 2 files changed, 119 insertions(+), 27 deletions(-) diff --git a/tcp/txqueue.go b/tcp/txqueue.go index e7d683d..33898e2 100644 --- a/tcp/txqueue.go +++ b/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 } diff --git a/tcp/txqueue_test.go b/tcp/txqueue_test.go index a963ecc..e3488c3 100644 --- a/tcp/txqueue_test.go +++ b/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)