From 5e5f21a2b8642da865d5fdee1af8b30b58eaff80 Mon Sep 17 00:00:00 2001 From: Patricio Whittingslow Date: Wed, 5 Nov 2025 02:35:53 -0300 Subject: [PATCH] rethinking ring buffer semantics and breaking everything in the process --- internal/ring.go | 58 +++++++--- internal/ring_test.go | 144 +++++++++++++++---------- internal/zoneprinter.go | 123 +++++++++++++++++++++ tcp/txqueue.go | 164 +++------------------------- tcp/txqueue_test.go | 234 +++++++++++++++++++++++++++------------- x/xnet/xnet_test.go | 3 +- 6 files changed, 432 insertions(+), 294 deletions(-) create mode 100644 internal/zoneprinter.go diff --git a/internal/ring.go b/internal/ring.go index 439620d..9f50338 100644 --- a/internal/ring.go +++ b/internal/ring.go @@ -8,7 +8,10 @@ import ( "unsafe" ) -var errRingBufferFull = errors.New("lneto/ring: buffer full") +var ( + errRingBufferFull = errors.New("lneto/ring: buffer full") + errRingNoData = errors.New("lneto/ring: empty write") +) // Ring implements basic Ring buffer functionality. type Ring struct { @@ -18,26 +21,42 @@ type Ring struct { // There is no readable data when End==0. Buf []byte // Start of readable data which indexes into Buf. - // If Off==End and End!=0 the buffer is full and data begins at Off. + // If Off==End and End!=0 the buffer is full and data begins at Off. Off end { - free = limitOffset - end - } else { - free = len(r.Buf) - end + limitOffset + + // Write start position. + var writeAt = r.End + if writeAt == 0 { + // Write start is End except when empty, in which case we writeAt at Off. + writeAt = r.Off + if limitOffset >= writeAt { + return limitOffset - writeAt // Contiguous case. + } + return r.Size() - writeAt + limitOffset // Wrap case. } - return free + + // normal (non-empty): write at End up to limitOffset, or Off, whichever comes first. + if writeAt <= limitOffset && writeAt <= r.Off { + return min(r.Off, limitOffset) - writeAt + } else if writeAt <= limitOffset { + return limitOffset - writeAt + } else if writeAt <= r.Off { + return r.Off - writeAt + } + return r.Size() - writeAt + min(limitOffset, r.Off) } // WriteLimited performs a write that does not write over the ring buffer's @@ -64,9 +83,10 @@ func (r *Ring) WriteString(s string) (int, error) { // Write appends data to the ring buffer that can then be read back in order with [Ring.Read] methods. // An error is returned if length of data too large for buffer. Write is guaranteed to start at buffer index [Ring.Off]. func (r *Ring) Write(b []byte) (int, error) { - free := r.Free() - if len(b) > free { + if r.isFull() { return 0, errRingBufferFull + } else if len(b) == 0 { + return 0, errRingNoData } midFree := r.midFree() if midFree > 0 { @@ -74,6 +94,9 @@ func (r *Ring) Write(b []byte) (int, error) { // | used | mfree | used | n := copy(r.Buf[r.End:r.Off], b) r.End += n + if r.End <= 0 { + panic("zero end after write") // TODO: remove panics after validation. + } return n, nil } else if r.End == 0 { // To ensure Write begins on r.Off. @@ -89,6 +112,9 @@ func (r *Ring) Write(b []byte) (int, error) { r.End = n2 n += n2 } + if r.End <= 0 { + panic("zero end after write") + } return n, nil } @@ -203,6 +229,10 @@ func (r *Ring) midFree() int { return r.Off - r.End } +func (r *Ring) isFull() bool { + return r.End != 0 && (r.End == r.Off || (r.End == len(r.Buf) && r.Off == 0)) +} + // onReadEnd does some cleanup of [ring.off] and [ring.end] fields if possible for contiguous read performance benefits. func (r *Ring) onReadEnd(totalRead int) { if totalRead <= 0 { diff --git a/internal/ring_test.go b/internal/ring_test.go index a5ae339..53b0940 100644 --- a/internal/ring_test.go +++ b/internal/ring_test.go @@ -21,8 +21,8 @@ func TestRing(t *testing.T) { const data = "hello" // Set random data and write some more and read it back. for i := 0; i < 32; i++ { - nfirst := rng.Intn(bufSize) / 2 - nsecond := rng.Intn(bufSize) / 2 + nfirst := max(1, rng.Intn(bufSize)/2) + nsecond := max(1, rng.Intn(bufSize)/2) if nfirst+nsecond > bufSize { nfirst = bufSize - nsecond } @@ -222,9 +222,11 @@ func TestRingNonEmpty(t *testing.T) { if checkWrite { testRingSanity(t, r) free := r.Size() - buf - n, err := r.Write(data[:free]) - if n != free || err != nil { - t.Errorf("want %d to fill buffer, got n=%d err=%v", free, n, err) + if free != 0 { + n, err := r.Write(data[:free]) + if n != free || err != nil { + t.Errorf("want %d to fill buffer, got n=%d err=%v", free, n, err) + } } } if checkRead { @@ -336,57 +338,60 @@ func TestRingOverwrite(t *testing.T) { } } -func TestRingWriteLimited(t *testing.T) { - rng := rand.New(rand.NewSource(2)) - const bufSize = 8 - r := &Ring{ - Buf: make([]byte, bufSize), - } - testRingSanity(t, r) - var data [bufSize]byte - var wdata [bufSize]byte - for i := 0; i < 10000; i++ { - for i := range r.Buf { - r.Buf[i] = 0 - } - buffered, _ := rng.Read(data[:rng.Intn(bufSize-2)+1]) - off := rng.Intn(bufSize) - setRingData(t, r, off, data[:buffered]) - if r.Buffered() != buffered { - t.Fatalf("failed to set buffered amount of data") - } else if r.Off != off { - t.Fatal("bad offset") - } - free := r.Free() +// func TestRingWriteLimited(t *testing.T) { +// rng := rand.New(rand.NewSource(2)) +// const bufSize = 8 +// r := &Ring{ +// Buf: make([]byte, bufSize), +// } +// testRingSanity(t, r) +// var data [bufSize]byte +// var wdata [bufSize]byte +// for i := 0; i < 10000; i++ { +// for i := range r.Buf { +// r.Buf[i] = 0 +// } +// buffered, _ := rng.Read(data[:rng.Intn(bufSize-2)+1]) +// off := rng.Intn(bufSize) +// setRingData(t, r, off, data[:buffered]) +// if r.Buffered() != buffered { +// t.Fatalf("failed to set buffered amount of data") +// } else if r.Off != off { +// t.Fatal("bad offset") +// } +// limOff := rng.Intn(bufSize) + 1 +// freeLim := r.FreeLimited(limOff) +// free := r.Free() - toWrite := rng.Intn(free-1) + 1 - rng.Read(wdata[:toWrite]) - limOff := rng.Intn(bufSize) + 1 - var wantN int - isContiguous := limOff > r.End - if isContiguous { - wantN = min(toWrite, limOff-r.End) - } else { - wantN = min(toWrite, len(r.Buf)-r.End+limOff) - } - overwrite := toWrite > wantN +// toWrite := rng.Intn(freeLim+1) + 1 +// rng.Read(wdata[:toWrite]) - n, err := r.WriteLimited(wdata[:toWrite], limOff) - if !overwrite && err != nil { - t.Errorf("limited write: %s", err) - } else if !overwrite && n != wantN { - t.Errorf("nwant=%d ngot=%d off=%d lim=%d towrite=%d buffered=%d/%d wantremain=%d gotremain=%d endOff=%d", wantN, n, off, limOff, toWrite, buffered, r.Size(), free-wantN, free-n, r.Off) - } else if overwrite && (err == nil || n != 0) { - t.Errorf("expected full buffer error and no data written on limit overwrite, got %d", n) - } - for i := r.End % r.Size(); i != limOff && i != r.Off; i = (i + 1) % r.Size() { - if r.Buf[i] != 0 { - t.Fatalf("OVERWRITE pos=%d end=%d lim=%d", i, r.End, limOff) - } - } - testRingSanity(t, r) - } -} +// var wantN int +// isContiguous := limOff > r.End +// if isContiguous { +// wantN = min(toWrite, limOff-r.End) +// } else { +// wantN = min(toWrite, len(r.Buf)-r.End+limOff) +// } +// overwrite := toWrite > wantN + +// n, err := r.WriteLimited(wdata[:toWrite], limOff) +// if !overwrite && err != nil { +// t.Errorf("limited write: %s", err) +// t.Errorf("size=%d off=%d end=%d lim=%d nwrite=%d", r.Size(), r.Off, r.End, limOff, toWrite) +// } else if !overwrite && n != wantN { +// t.Errorf("nwant=%d ngot=%d off=%d lim=%d towrite=%d buffered=%d/%d wantremain=%d gotremain=%d endOff=%d", wantN, n, off, limOff, toWrite, buffered, r.Size(), free-wantN, free-n, r.Off) +// } else if overwrite && (err == nil || n != 0) { +// t.Errorf("expected full buffer error and no data written on limit overwrite, got %d", n) +// } +// for i := r.End % r.Size(); i != limOff && i != r.Off; i = (i + 1) % r.Size() { +// if r.Buf[i] != 0 { +// t.Fatalf("OVERWRITE pos=%d end=%d lim=%d", i, r.End, limOff) +// } +// } +// testRingSanity(t, r) +// } +// } func TestRing_findcrash(t *testing.T) { const maxsize = 33 @@ -437,6 +442,37 @@ func TestRing_findcrash(t *testing.T) { } } +func TestRingFreeLimited(t *testing.T) { + const n = 16 + rng := rand.New(rand.NewSource(1)) + buffer := make([]byte, n) + r := Ring{Buf: buffer} + for itest := 0; itest < 100000; itest++ { + r.Off = rng.Intn(n) + r.End = rng.Intn(n + 1) + limit := rng.Intn(n) + wantFreeLimited := 0 + writeStart := r.End + if writeStart == 0 { + // Empty buffer case. + writeStart = r.Off + for i := writeStart % n; i != limit; i = (i + 1) % n { + wantFreeLimited++ + } + } else { + // Buffer with contents. Beware limit and own offset. + for i := writeStart % n; i != limit && i != r.Off; i = (i + 1) % n { + wantFreeLimited++ + } + } + + gotFreeLimited := r.FreeLimited(limit) + if gotFreeLimited != wantFreeLimited { + t.Fatalf("[%d] len=%d off=%d end=%d limit=%d GOT=%d, WANT=%d", itest, n, r.Off, r.End, limit, gotFreeLimited, wantFreeLimited) + } + } +} + func testRing1_loopback(t *testing.T, rng *rand.Rand, ringbuf, data, auxbuf []byte) bool { if len(data) > len(ringbuf) || len(data) > len(auxbuf) { panic("invalid ringbuf or data") diff --git a/internal/zoneprinter.go b/internal/zoneprinter.go new file mode 100644 index 0000000..3f2f915 --- /dev/null +++ b/internal/zoneprinter.go @@ -0,0 +1,123 @@ +package internal + +import ( + "errors" + "fmt" + "strconv" +) + +type ZonePrinter struct { + zbuf []BufferZone + aux []byte +} + +type BufferZone struct { + Name string + Start, End int +} + +func (bp *ZonePrinter) AppendPrintZones(dst []byte, bufSize int, zones ...BufferZone) ([]byte, error) { + n := bufSize + if n == 0 { + return dst, errors.New("empty buffer") + } + if len(bp.zbuf) != 0 { + panic("race condition detected: zbuf not zero lengthed- ZonePrinter not to be used concurrently") + } + // Zeroed zone buffer. Will be used to detect collisions. + // TODO(soypat): Really wasteful, there's an obvious way to just have len(zones)+1 zbuf, but I've already spent too much time here. This also makes collision detection really easy so whatever. + bp.zbuf = append(bp.zbuf[:0], make([]BufferZone, n)...) + labels := bp.zbuf + + // helper to paint an interval on the ring + paint := func(z BufferZone) error { + if z.End == 0 { // 0 == empty + return nil + } + if z.Start < z.End { + for i := z.Start; i < z.End; i++ { + if labels[i].Name != "" { + return fmt.Errorf("paint collision: %q/%q @%d size=%d", labels[i].Name, z.Name, i, bufSize) + } + labels[i].Name = z.Name + } + } else { // wrap + for i := z.Start; i < n; i++ { + if labels[i].Name != "" { + return fmt.Errorf("wrap(start) paint collision: %q/%q @%d size=%d", labels[i].Name, z.Name, i, bufSize) + } + labels[i].Name = z.Name + } + for i := 0; i < z.End; i++ { + if labels[i].Name != "" { + return fmt.Errorf("wrap(end) paint collision: %q/%q @%d size=%d", labels[i].Name, z.Name, i, bufSize) + } + labels[i].Name = z.Name + } + } + return nil + } + + // paint + for _, zone := range zones { + err := paint(zone) + if err != nil { + return dst, err + } + } + + // unpainted sections are free. + for i := range labels { + if labels[i].Name == "" { + labels[i].Name = "free" + } + } + // Compress to segments. + icurrent := 0 + for i := 1; i < n; i++ { + current := labels[icurrent] + next := labels[i] + if current.Name != next.Name { + labels[icurrent].End = i + icurrent++ + labels[icurrent].Start = i + labels[icurrent].Name = next.Name + } + } + segs := labels[:icurrent+1] + segs[icurrent].End = n + + var l1 []byte = dst + var l2 []byte = bp.aux[:0] + for _, s := range segs { + l2start := len(l2) + if s.Name == "free" { + l2 = append(l2, "| "...) + } else { + l2 = append(l2, "|--"...) + } + l2 = append(l2, s.Name...) + l2 = append(l2, '(') + l2 = strconv.AppendInt(l2, int64(s.End-s.Start), 10) + if s.Name == "free" { + l2 = append(l2, ") "...) + } else { + l2 = append(l2, ")--"...) + } + fraglen := len(l2) - l2start + + l1start := len(l1) + l1 = strconv.AppendInt(l1, int64(s.Start), 10) + + paddingNeeded := fraglen - (len(l1) - l1start) + for range paddingNeeded { + l1 = append(l1, ' ') + } + } + l1 = strconv.AppendInt(l1, int64(bufSize), 10) + l1 = append(l1, '\n') + l2 = append(l2, '|') + l1 = append(l1, l2...) + bp.zbuf = bp.zbuf[:0] // zero out, used to detect concurrent usage of ZonePrinter. + return l1, nil +} diff --git a/tcp/txqueue.go b/tcp/txqueue.go index 46e175a..f466dd9 100644 --- a/tcp/txqueue.go +++ b/tcp/txqueue.go @@ -1,11 +1,8 @@ package tcp import ( - "bytes" "errors" - "fmt" "slices" - "strconv" "github.com/soypat/lneto/internal" ) @@ -26,7 +23,6 @@ const ( type ringTx struct { // rawbuf contains the ring buffer of ordered bytes. It should be the size of the window. rawbuf []byte - 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. @@ -35,6 +31,7 @@ type ringTx struct { sentoff int // sentend is the offset of end of sent data in rawbuf. If zero then sent buffer is empty. sentend int + slist sentlist // seq Value // always empty ring. emptyRing ringidx @@ -112,7 +109,7 @@ func (rtx *ringTx) Write(b []byte) (n int, err error) { if err != nil { return 0, err } - rtx.unsentend = rtx.addEnd(rtx.unsentend, n) + rtx.unsentend = r.End return n, err } @@ -123,36 +120,35 @@ func (rtx *ringTx) MakePacket(b []byte, currentSeq Value) (int, error) { if free == 0 { return 0, errPacketQueueFull } - endSeq, ok := rtx.endSeq() + endSeq, ok := rtx.sentEndSeq() 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() - oldSentOff := r.Off - n, err := r.Read(b) + unsent, _ := rtx.unsentRing() + oldSentOff := unsent.Off + n, err := unsent.Read(b) if err != nil { - return n, err + return 0, err } // 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) + newUnsentOff := unsent.Off pkt := rtx.slist.AddPacket(n, oldSentOff, rtx.Size()) if pkt.off != oldSentOff || pkt.end != addEnd(pkt.off, n, rtx.Size()) { panic("invalid generated packet") } rtx.unsentoff = newUnsentOff rtx.sentend = newUnsentOff - if newUnsentOff == rtx.unsentend { - rtx.unsentend = 0 // Mark unsent as being empty. - } + rtx.unsentend = unsent.End return n, nil } // RecvSegment processes an incoming segment and updates the sent packet queue func (rtx *ringTx) RecvACK(ack Value) error { - err := rtx.slist.RecvAck(ack, rtx.Size()) + size := rtx.Size() + err := rtx.slist.RecvAck(ack, size) if err != nil { return err } @@ -203,7 +199,7 @@ func (rtx *ringTx) consolidateBufs() { } } -func (rtx *ringTx) endSeq() (Value, bool) { +func (rtx *ringTx) sentEndSeq() (Value, bool) { newest := rtx.slist.Newest() if newest == nil { return 0, false @@ -302,7 +298,9 @@ func (sl *sentlist) RecvAck(ack Value, bufsize int) error { newest := sl.Newest() if newest == nil { return errors.New("no packet to ack") - } else if newest.endSeq().LessThan(ack) { + } + endseq := newest.endSeq() + if endseq.LessThan(ack) { return errors.New("ack of unsent packet") } // Mark fully acked. @@ -366,135 +364,3 @@ func addOff(a, b int, size int) int { } return result } - -// prints out buffer zones with indices: -// -// 0 32 42 47 -// |---free(32)---|---usnt(10)---|---free(5)---| -func (rtx *ringTx) appendString(b []byte) []byte { - size := rtx.Size() - type zone struct { - name string - start, end int - } - zcontains := func(off int, z *zone) bool { - if z.end == 0 { - return false // Empty - } else if z.end < z.start { - // zone wraps. - } - return off >= z.start && off < z.end - } - - zs := zone{name: "sent", start: rtx.sentoff, end: rtx.sentend} - zu := zone{name: "usnt", start: rtx.unsentoff, end: rtx.unsentend} - bufStart := zs.start - if bufStart == 0 { - bufStart = zu.start - } - bufEnd := zu.end - if bufEnd == 0 { - bufEnd = zs.end - } - zf := zone{name: "free", start: bufEnd, end: bufStart} - getZone := func(off int) *zone { - if zcontains(0, &zs) { - return &zs - } else if zcontains(0, &zu) { - return &zu - } else { - return &zf - } - } - - zones := []*zone{getZone(0)} - for i := 1; i < size; i++ { - z := getZone(i) - if z != zones[len(zones)-1] { - zones = append(zones, z) - } - } - - var wrapZone *zone - for i := range zones { - wraps := zones[i].end != 0 && zones[i].end < zones[i].start - if wraps { - if wrapZone != nil { - panic("illegal to have more than one wrap zone") - } - wrapZone = zones[i] - } - } - // ---- your simple approach starts here ---- - var currentZone *zone - if wrapZone != nil { - currentZone = wrapZone - } else { - currentZone = zones[0] - } - - var lastPrintedZone *zone - var l1, l2 bytes.Buffer - changes := 0 - zoneLen := func(z *zone, sz int) int { - if z.end == 0 { - return 0 - } - if z.end < z.start { - return (sz - z.start) + z.end - } - return z.end - z.start - } - for ib := 0; ib < size; ib++ { - // see if current zone still contains this index - currentContainsIdx := currentZone != nil && zcontains(ib, currentZone) - if !currentContainsIdx { - // find which zone contains this index - for _, z := range zones { - if zcontains(ib, z) { - currentZone = z - currentContainsIdx = true - break - } - } - } - // if still same zone, keep going - if currentZone == lastPrintedZone { - continue - } - - // zone changed - changes++ - if changes > 4 { - panic("found too many zone changes") - } - lastPrintedZone = currentZone - - // build the bottom line segment - seg := "|---" + currentZone.name + "(" + strconv.Itoa(zoneLen(currentZone, size)) + ")---" - l2.WriteString(seg) - - // write the start index aligned to seg width - n, _ := fmt.Fprintf(&l1, "%d", currentZone.start) - for i := 0; i < len(seg)-n; i++ { - l1.WriteByte(' ') - } - } - - // close last zone: print its end index and closing bar - l2.WriteByte('|') - - // if the last zone "ends" at 0 because of wrap, use sz - endIdx := lastPrintedZone.end - if endIdx == 0 { - endIdx = size - } - fmt.Fprintf(&l1, "%d\n", endIdx) - - // write second line under the first - l2.WriteTo(&l1) - l1.WriteByte('\n') - - b = append(b, l1.Bytes()...) - return b -} diff --git a/tcp/txqueue_test.go b/tcp/txqueue_test.go index 311aba8..ea93bf8 100644 --- a/tcp/txqueue_test.go +++ b/tcp/txqueue_test.go @@ -2,12 +2,124 @@ package tcp import ( "bytes" - "fmt" "math/rand" "slices" "testing" + "unsafe" + + "github.com/soypat/lneto/internal" ) +func TestRingTx_op(t *testing.T) { + const maxBuf = 32 + const maxpkt = 3 + const Nops = 3 + type op uint8 + const ( + opWrite op = iota + opSend + opAck + opmax + ) + + rng := rand.New(rand.NewSource(666)) + randop := func() op { return op(rng.Intn(int(opmax))) } + var buf, auxbuf [maxBuf]byte + dataWritten := make([]byte, 0, maxBuf*10) + dataSent := make([]byte, 0, maxBuf*10) + var rtx ringTx + for itest := 0; itest < 3; itest++ { + bufsize := rng.Intn(maxBuf/2) + maxBuf/2 + iss := Value(0) + npackets := rng.Intn(maxpkt-1) + 1 + err := rtx.Reset(buf[:bufsize], npackets, iss) + if err != nil { + t.Fatal(err) + } + // Prepare state for keeping track of test. + currentAcked := iss + currentSeq := iss + nsent := 0 + nunsent := 0 + nacked := 0 + dataWritten = dataWritten[:0] + dataSent = dataSent[:0] + for iop := 0; iop < Nops; iop++ { + free := bufsize - nsent - nunsent + availPkt := rtx.slist.Free() + op := randop() + var oplen int + var opname string + var opWriteData []byte + switch op { + case opWrite: + opname, oplen = "write", rng.Intn(free+1)+1 + opWriteData = auxbuf[:oplen] + rng.Read(opWriteData) + clear(auxbuf[oplen:]) + case opSend: + opname, oplen = "send", rng.Intn(nunsent+1)+1 + case opAck: + opname, oplen = "ack", rng.Intn(nsent+1)+1 + } + if itest < 2 { + continue // Debugging. + } + _ = opname + t.Logf("\n%s\nitest=%d iop=%d op=%s len=%d", rtx.mustAppendString(nil), itest, iop, opname, oplen) + switch op { + case opWrite: + // oplen=number of + nwgot, err := rtx.Write(opWriteData) + wantErr := oplen > free + if err != nil && oplen <= free { + t.Fatal(itest, iop, err) + } else if err == nil { + if wantErr { + panic("wanted write error") + } + nunsent += nwgot + dataWritten = append(dataWritten, opWriteData[:nwgot]...) + } else { + t.Logf("opwrite: %s", err) + } + case opSend: + // oplen=num bytes sent. + nsgot, err := rtx.MakePacket(auxbuf[:oplen], currentSeq) + megafail := nsgot > nunsent + if err != nil && oplen <= nunsent && availPkt > 0 { + t.Fatal(itest, iop, err) + } else if err == nil { + if megafail { + panic("megafail") + } + nunsent -= nsgot + nsent += nsgot + dataSent = append(dataSent, auxbuf[:nsgot]...) + currentSeq += Value(nsgot) + } else { + t.Logf("opsend: %s", err) + } + case opAck: + // oplen=acklength. + tryAck := currentAcked + Value(oplen) + err = rtx.RecvACK(tryAck) + if err != nil && oplen <= nsent { + t.Fatal(itest, iop, err) + } else if err == nil { + nsent -= oplen + nacked += oplen + currentAcked = tryAck + } else { + t.Logf("opack: %s", err) + } + default: + panic("unknown op") + } + } + } +} + func TestSentlist_multi(t *testing.T) { const bufsize = 10 var sl sentlist @@ -37,7 +149,7 @@ func TestSentlist_multi(t *testing.T) { _ = p3 } -func TestSentlist(t *testing.T) { +func TestSentlist_simple(t *testing.T) { var sl sentlist sl.Reset(3, 0) // Test full ack. @@ -75,17 +187,18 @@ func TestSentlist(t *testing.T) { } func TestTxQueue_multipacket(t *testing.T) { - const mtu = 256 + const mtu = 32 const iss = 1 const maxPkts = 3 - const maxWrites = 20 + const maxWrites = 6 const maxWriteSize = mtu / maxWrites var rtx ringTx internalbuff := make([]byte, mtu) rng := rand.New(rand.NewSource(3)) var wbuf, rbuf [mtu]byte for itest := 0; itest < 32; itest++ { - // rng.Seed(int64(itest)) + t.Log(itest) + rng.Seed(int64(itest)) println(itest) err := rtx.Reset(internalbuff, maxPkts, iss) if err != nil { @@ -133,6 +246,7 @@ func TestTxQueue_multipacket(t *testing.T) { } acked := 0 for acked < roff { + t.Log(acked) maxToack := min(roff-acked, maxWriteSize) toack := rng.Intn(maxToack) + 1 // t.Log("\n", rtx.string()) @@ -323,7 +437,7 @@ func testQueueSanity(t *testing.T, rtx *ringTx) { sz := rtx.Size() gotSz := free + sent + unsent if gotSz != sz { - t.Error("\n" + string(rtx.appendString(nil))) + 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() @@ -371,7 +485,7 @@ func testQueueSanity(t *testing.T, rtx *ringTx) { return } - endseq, ok := rtx.endSeq() + endseq, ok := rtx.sentEndSeq() lastEndSeq := Add(last.seq, last.size) if last.seq.LessThan(first.seq) { @@ -384,73 +498,8 @@ func testQueueSanity(t *testing.T, rtx *ringTx) { } func (rx *ringTx) string() string { - sz := rx.Size() - unsent, _ := rx.unsentRing() - sent, _ := rx.sentRing() - all := rx.sentAndUnsentBuffer() - if all.End == 0 || // Empty buffer, set offset so that free zone occupies whole buffer. - all.Off == 0 { // Buffer offset starts at zero which would set Free.End to 0 making it empty, patch that. - all.Off = sz - } - type zone struct { - name string - start, end int - } - zcontains := func(off int, z *zone) bool { - if z.end == 0 { - return false // Empty - } else if z.end < z.start { - return off < z.end || off >= z.start - } - return off >= z.start && off < z.end - } - var zones = []zone{ - {name: "free", start: all.End, end: all.Off}, - {name: "usnt", start: unsent.Off, end: unsent.End}, - {name: "sent", start: sent.Off, end: sent.End}, - } - var wrapZone *zone - for i := range zones { - wraps := zones[i].end != 0 && zones[i].end < zones[i].start - if wraps { - if wrapZone != nil { - panic("illegal to have more than one wrap zone") - } - wrapZone = &zones[i] - } - } - var currentZone *zone = wrapZone - var lastPrintedZone *zone - var l1, l2 bytes.Buffer - changes := 0 - for ib := 0; ib < sz; ib++ { - currentContainsIdx := currentZone != nil && zcontains(ib, currentZone) - for iz := 0; !currentContainsIdx && iz < len(zones); iz++ { - z := &zones[iz] - if zcontains(ib, z) { - currentZone = z - } - } - if currentZone == lastPrintedZone { - continue - } - changes++ - if changes > 4 { - panic("found too many zone changes") - } - lastPrintedZone = currentZone - // Change of zone. - top := "|-----" + currentZone.name + "-----" - l2.WriteString(top) - n, _ := fmt.Fprintf(&l1, "%d", currentZone.start) - for i := 0; i < len(top)-n; i++ { - l1.WriteByte(' ') - } - } - l2.WriteByte('|') - fmt.Fprintf(&l1, "%d\n", currentZone.end) - l2.WriteTo(&l1) - return l1.String() + s := rx.appendString(nil) + return unsafe.String(&s[0], len(s)) } func removeEmptyMsgs(msgs [][]byte) [][]byte { @@ -526,7 +575,7 @@ func operateOnRing(t *testing.T, rtx *ringTx, write, readPacket, aux []byte, new if lastSeqOK { lastSeq = last.seq } - endSeq, endSeqOK := rtx.endSeq() + endSeq, endSeqOK := rtx.sentEndSeq() if !lastSeqOK || lastSeq != newPacketSeq { t.Fatalf("expected last seq to be %d, got %d (or lastSeqOK=%v)", newPacketSeq, lastSeq, lastSeqOK) } else if !endSeqOK || endSeq != Add(newPacketSeq, Size(n)) { @@ -590,3 +639,38 @@ func operateOnRing(t *testing.T, rtx *ringTx, write, readPacket, aux []byte, new } testQueueSanity(t, rtx) } + +// prints out buffer zones with indices: +// +// 0 32 42 47 +// |---free(32)---|---usnt(10)---|---free(5)---| +func (rtx *ringTx) appendString(b []byte) []byte { + var zprinter internal.ZonePrinter + result, err := zprinter.AppendPrintZones(b, rtx.Size(), rtx.zones()...) + if err != nil { + result = append(result, err.Error()...) + } + return result +} + +func (rtx *ringTx) mustAppendString(b []byte) []byte { + var zprinter internal.ZonePrinter + result, err := zprinter.AppendPrintZones(b, rtx.Size(), rtx.zones()...) + if err != nil { + panic(err) + } + return result +} + +func (rtx *ringTx) zones() []internal.BufferZone { + return []internal.BufferZone{ + { + Name: "sent", + Start: rtx.sentoff, End: rtx.sentend, + }, + { + Name: "usnt", + Start: rtx.unsentoff, End: rtx.unsentend, + }, + } +} diff --git a/x/xnet/xnet_test.go b/x/xnet/xnet_test.go index ffabb64..086bf99 100644 --- a/x/xnet/xnet_test.go +++ b/x/xnet/xnet_test.go @@ -20,10 +20,9 @@ const ( func TestStackAsyncTCP_multipacket(t *testing.T) { const seed = 1234 - const MTU = 1500 + const MTU = 512 const svPort = 8080 const maxPktLen = 30 - const maxNPkt = 32 client, sv, clconn, svconn := newTCPStacks(t, seed, MTU) tst := tester{ t: t, buf: make([]byte, MTU),