From aa8da93f6b61be1ab6c54dc23c60500abb9e8326 Mon Sep 17 00:00:00 2001 From: soypat Date: Sat, 8 Feb 2025 16:06:14 -0300 Subject: [PATCH] fix tests finally --- tcp/txqueue.go | 161 ++++++++++++++++++++------------------ tcp/txqueue_test.go | 183 ++++++++++++++++++++++++++++++++++++++++---- 2 files changed, 256 insertions(+), 88 deletions(-) diff --git a/tcp/txqueue.go b/tcp/txqueue.go index f83721f..fd4295a 100644 --- a/tcp/txqueue.go +++ b/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 { - rx.packets = make([]ringidx, maxqueuedPackets) + if cap(rtx.packets) < 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 { - rx.packets[i].markRcvd() + for i := range rtx.packets { + 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 { - r := tx.sentAndUnsentBuffer() +func (rtx *ringTx) Free() int { + r := rtx.sentAndUnsentBuffer() return r.Free() } -// Buffered returns the amount of unsent bytes. -func (tx *ringTx) Buffered() int { - r, _ := tx.unsentRing() +// Buffered returns the amount of written but unsent bytes. +func (rtx *ringTx) Buffered() int { + r, _ := rtx.unsentRing() return r.Buffered() } // BufferedSent returns the total amount of bytes sent but not acked. -func (tx *ringTx) BufferedSent() int { - r, _ := tx.sentRing() +func (rtx *ringTx) BufferedSent() int { + 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) { - r, lim := tx.unsentRing() +func (rtx *ringTx) Write(b []byte) (n int, err error) { + 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) { - nxtpkt := tx.nextPkt() - if tx.nextPkt() < 0 { +func (rtx *ringTx) MakePacket(b []byte) (int, Value, error) { + nxtpkt := rtx.nextPkt() + 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) - tx.unsentoff = off - tx.sentend = off - if off == tx.unsentend { - tx.unsentend = 0 // Mark unsent as being empty. + off := rtx.addEnd(rtx.unsentoff, n) + rtx.unsentoff = off + rtx.sentend = off + if off == rtx.unsentend { + 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 { - if ack.LessThan(tx.seq) { +func (rtx *ringTx) RecvACK(ack Value) error { + 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 - for i := 0; i < len(tx.packets); i++ { - pkt := &tx.packets[i] + hiSeq := rtx.pkt(first).seq + for i := 0; i < len(rtx.packets); i++ { + pkt := &rtx.packets[i] if pkt.sent() && pkt.seq.LessThanEq(ack) { - if hiSeq.LessThan(pkt.seq) { - tx.sentoff = pkt.end + if hiSeq.LessThanEq(pkt.seq) { + rtx.sentoff = pkt.end hiSeq = pkt.seq } pkt.markRcvd() } } - firstAcked := !tx.pkt(first).sent() - if firstAcked && tx.sentoff == tx.sentend { + firstAcked := !rtx.pkt(first).sent() + 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 { - end := tx.unsentend +func (rtx *ringTx) sentAndUnsentBuffer() internal.Ring { + 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) { - return tx.ring(tx.unsentoff, tx.unsentend), tx.sentoff +func (rtx *ringTx) unsentRing() (internal.Ring, int) { + return rtx.ring(rtx.unsentoff, rtx.unsentend), rtx.sentoff } -func (tx *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. +func (rtx *ringTx) sentRing() (internal.Ring, int) { + 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 { - return internal.Ring{Buf: tx.rawbuf, Off: off, End: end} +func (rtx *ringTx) ring(off, end int) internal.Ring { + 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) { - result -= len(tx.rawbuf) + if result > len(rtx.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) { - result -= len(tx.rawbuf) + if result >= len(rtx.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 - } else if i < 0 || i >= len(tx.packets) { + return &rtx.emptyRing + } 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++ { - pkt := &tx.packets[i] + 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 @@ -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++ { - pkt := &tx.packets[i] + 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 @@ -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++ { - pkt := &tx.packets[i] + for i := 0; i < len(rtx.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: // diff --git a/tcp/txqueue_test.go b/tcp/txqueue_test.go index 13c7048..e5b471c 100644 --- a/tcp/txqueue_test.go +++ b/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}) - testTxQueue_SequentialMessages(t, &rtx, msgs, buf[:], aux[:], rng.Intn(4)+1, 0) + msgs := removeEmptyMsgs(bytes.SplitAfter(msgBuf[:], []byte{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[:], ¤tAck) + 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}) - testTxQueue_NMessages(t, &rtx, msgs, buf[:], aux[:], len(msgs), 0) + msgs := removeEmptyMsgs(bytes.SplitAfter(msgBuf[:], []byte{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[:], ¤tAck) } }, }, @@ -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() { - if t.Failed() { - t.Log("\n" + rtx.string()) - } - }() + alreadyFailed := t.Failed() + if !alreadyFailed { + 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 +}