diff --git a/.gitignore b/.gitignore index c5d9618..473fc44 100644 --- a/.gitignore +++ b/.gitignore @@ -15,6 +15,7 @@ vendor/ *.so *.dylib *.hex +**__debug_bin* # `__debug_bin` Debug binary generated in VSCode when using the built-in debugger. *bin diff --git a/internal/ring.go b/internal/ring.go index 819aae2..439620d 100644 --- a/internal/ring.go +++ b/internal/ring.go @@ -28,10 +28,14 @@ type Ring struct { // SizeLimited returns the amount of bytes that can be written up to the // argument offset limitOffset. See [Ring.WriteLimited] func (r *Ring) FreeLimited(limitOffset int) (free int) { - if limitOffset > r.End { - free = limitOffset - r.End + end := r.End + if r.End == 0 { + end = r.Off + } + if limitOffset > end { + free = limitOffset - end } else { - free = len(r.Buf) - r.End + limitOffset + free = len(r.Buf) - end + limitOffset } return free } @@ -57,7 +61,8 @@ func (r *Ring) WriteString(s string) (int, error) { return r.Write(unsafe.Slice(unsafe.StringData(s), len(s))) } -// 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 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 { @@ -70,6 +75,10 @@ func (r *Ring) Write(b []byte) (int, error) { n := copy(r.Buf[r.End:r.Off], b) r.End += n return n, nil + } else if r.End == 0 { + // To ensure Write begins on r.Off. + // Specialised for when user controls Off manually instead of by internal calls to [Ring.onReadEnd] or calls to [Ring.Reset]. + r.End = r.Off } // start off end len(buf) // | sfree | used | efree | @@ -86,14 +95,14 @@ func (r *Ring) Write(b []byte) (int, error) { // ReadDiscard is a performance auxiliary method that performs a dummy read or no-op read // for advancing the read pointer n bytes without actually copying data. // This method panics if amount of bytes is more than buffered (see [Ring.Buffered]). -func (r *Ring) ReadDiscard(n int) { - if n < 0 { - panic("negative discard amount") +func (r *Ring) ReadDiscard(n int) error { + if n <= 0 { + return errors.New("invalid discard amount") } buffered := r.Buffered() switch { case n > buffered: - panic("discard exceeds length") + return errors.New("discard exceeds length") case n == buffered: r.Reset() case n+r.Off > len(r.Buf): @@ -101,6 +110,7 @@ func (r *Ring) ReadDiscard(n int) { default: r.Off += n } + return nil } // ReadAt reads data at an offset from start of readable data but does not advance read pointer. [io.EOF] returned when no data available. @@ -171,7 +181,7 @@ func (r *Ring) Buffered() int { // Free returns amount of bytes that can be read into ring buffer before reaching maximum capacity given by [ring.Size]. Always less than [ring.Size]. func (r *Ring) Free() int { - if r.Off == 0 || r.End == 0 { + if r.End == 0 || r.Off == 0 { return len(r.Buf) - r.End } if r.Off < r.End { @@ -201,11 +211,15 @@ func (r *Ring) onReadEnd(totalRead int) { newOff := r.addOff(r.Off, totalRead) if newOff == r.End { r.Reset() + } else if newOff == len(r.Buf) { + r.Off = 0 // Optimization case. } else { r.Off = newOff } } +// addOff sums a and b to return an index within 1..[Ring.Size] supposing a and b are each less-equal than [Ring.Size]. +// Result will never be 0 unless both a and b are 0. func (r *Ring) addOff(a, b int) int { result := a + b if result > len(r.Buf) { diff --git a/internal/ring_test.go b/internal/ring_test.go index da3849a..a5ae339 100644 --- a/internal/ring_test.go +++ b/internal/ring_test.go @@ -175,6 +175,7 @@ func TestRingEmpty(t *testing.T) { if buf2 != 0 { t.Fatalf("want 0 bytes buffered(second call), got buf=%d->%d for off=%d->%d, end=0->%d size=%d", buf, buf2, off, r.Off, r.End, r.Size()) } + testRingSanity(t, r) for _, read := range readCalls { n, err := read(data) if err != io.EOF { @@ -182,8 +183,8 @@ func TestRingEmpty(t *testing.T) { } else if n != 0 { t.Fatalf("expected no bytes read, got %d", n) } + testRingSanity(t, r) } - testRingSanity(t, r) } }) } @@ -243,6 +244,42 @@ func TestRingNonEmpty(t *testing.T) { } } +func TestRing_OffWrite(t *testing.T) { + const bufSize = 8 + var rawbuf, auxbuf, readback [bufSize]byte + r := &Ring{Buf: rawbuf[:]} + for n := 1; n < bufSize+1; n++ { + for off := 0; off < bufSize+1; off++ { + r.Off = off // Start write at off. + r.End = 0 // Reset to use no data. + for i := 0; i < n; i++ { + rawbuf[(off+i)%len(rawbuf)] = 0 + auxbuf[i] = byte(i) + 1 + } + ngot, err := r.Write(auxbuf[:n]) + if err != nil { + t.Fatal(err) + } else if ngot != n { + t.Fatal(n, ngot) + } + for i := 0; i < n; i++ { + offz := (off + i) % len(rawbuf) + if rawbuf[offz] != auxbuf[i] { + t.Fatalf("mismatch pos=%d off=%d %q!=%q", i, offz, rawbuf[offz], auxbuf[i]) + } + } + ngot, err = r.Read(readback[:]) + if err != nil { + t.Fatal(err) + } else if ngot != n { + t.Fatal(n, ngot) + } else if !bytes.Equal(readback[:n], auxbuf[:n]) { + t.Fatalf("want readback %q, got %q", auxbuf[:n], readback[:n]) + } + } + } +} + func TestRing_TwoWrite(t *testing.T) { const bufSize = 8 rng := rand.New(rand.NewSource(1)) @@ -441,24 +478,6 @@ func testRing1_loopback(t *testing.T, rng *rand.Rand, ringbuf, data, auxbuf []by return !t.Failed() } -func fragmentReadInto(r io.Reader, buf []byte) (n int, _ error) { - maxSize := len(buf) / 4 - for { - ntop := min(n+rand.Intn(maxSize)+1, len(buf)) - ngot, err := r.Read(buf[n:ntop]) - n += ngot - if err != nil { - if err == io.EOF { - return n, nil - } - return n, err - } - if n == len(buf) { - return n, nil - } - } -} - func setRingData(t *testing.T, r *Ring, offset int, data []byte) { t.Helper() sz := r.Size() @@ -520,7 +539,7 @@ func testRingSanity(t *testing.T, r *Ring) { func canonRing(r *Ring) { if r.Buffered() == 0 { - // r.End = r.addOff(r.Off, 1) - // r.onReadEnd(1) + r.End = r.addOff(r.Off, 1) + r.onReadEnd(1) } } diff --git a/tcp/txqueue.go b/tcp/txqueue.go index 83612ee..44268d1 100644 --- a/tcp/txqueue.go +++ b/tcp/txqueue.go @@ -17,9 +17,9 @@ type ringTx struct { rawbuf []byte // packets contains packets []ringidx - // unsentOff is the offset of start of unsent data into rawbuf. + // unsentOff is the offset of start of unsent data in rawbuf. unsentoff int - // unsentend is the offset of end of unsent data in rawbuf. + // unsentend is the offset of end of unsent data in rawbuf. If zero then unsent buffer is empty. unsentend int seq Value // always empty ring. @@ -71,32 +71,36 @@ func (rx *ringTx) ResetOrReuse(buf []byte, maxQueuedPackets int, ack Value) erro return rx.Reset(buf, maxQueuedPackets, ack) } +// Size returns the total storage space of the transmission buffer. +func (tx *ringTx) Size() int { return len(tx.rawbuf) } + +// Free returns the total available space for Write calls. +func (tx *ringTx) Free() int { + freeStart, freeEnd, _ := tx.lims() + r := tx.ring(freeEnd, freeStart) + return r.Free() +} + // Buffered returns the amount of unsent bytes. func (tx *ringTx) Buffered() int { - r := tx.unsentRing() + r, _ := tx.unsentRing() return r.Buffered() } // BufferedSent returns the total amount of bytes sent but not acked. func (tx *ringTx) BufferedSent() int { - r := tx.sentRing() + r, _ := tx.sentRing() return r.Buffered() } // Write writes data to the underlying unsent data ring buffer. func (tx *ringTx) Write(b []byte) (n int, err error) { - first := tx.pkt(tx.firstPkt()) - r := tx.unsentRing() - if !first.sent() { - // No packets in queue case. - n, err = r.Write(b) - } else { - n, err = r.WriteLimited(b, first.off) - } + r, lim := tx.unsentRing() + n, err = r.WriteLimited(b, lim) if err != nil { return 0, err } - tx.unsentend = tx.addOff(tx.unsentend, n) + tx.unsentend = tx.addEnd(tx.unsentend, n) return n, err } @@ -107,7 +111,7 @@ func (tx *ringTx) MakePacket(b []byte) (int, Value, error) { if tx.nextPkt() < 0 { return 0, 0, errors.New("queue full") } - r := tx.unsentRing() + r, _ := tx.unsentRing() start := r.Off n, err := r.Read(b) if err != nil { @@ -115,12 +119,14 @@ func (tx *ringTx) MakePacket(b []byte) (int, Value, error) { } plen := Value(n) seq := tx.seq - tx.unsentoff = tx.addOff(tx.unsentoff, n) + tx.unsentoff = tx.addEnd(tx.unsentoff, n) + if tx.unsentoff == tx.unsentend { + tx.unsentend = 0 // Mark unsent as being empty. + } tx.seq += plen - pkt := &tx.packets[nxtpkt] pkt.off = start - pkt.end = tx.addOff(start, n) + pkt.end = tx.addEnd(start, n) pkt.seq = seq + plen return n, seq, nil } @@ -133,37 +139,43 @@ func (tx *ringTx) RecvACK(ack Value) error { pkt.markRcvd() } } + return nil } -func (tx *ringTx) unsentRing() internal.Ring { - off := tx.unsentoff - if off == tx.unsentend && off != 0 { - off-- +func (tx *ringTx) unsentRing() (internal.Ring, int) { + freeStart, freeEnd, sentEnd := tx.lims() + if tx.unsentend == 0 { + freeStart = 0 // Unsent is empty. } - return tx.ring(off, tx.unsentend) + return tx.ring(sentEnd, freeStart), freeEnd } -func (tx *ringTx) sentRing() internal.Ring { - first := tx.pkt(tx.firstPkt()) - if !first.sent() { - return internal.Ring{} - } - last := tx.pkt(tx.lastPkt()) - return tx.ring(first.off, last.end) +func (tx *ringTx) sentRing() (internal.Ring, int) { + freeStart, freeEnd, sentEnd := tx.lims() + return tx.ring(freeEnd, sentEnd), freeStart } func (tx *ringTx) ring(off, end int) internal.Ring { return internal.Ring{Buf: tx.rawbuf, Off: off, End: end} } -// addOff adds two integers together and wraps the value around the ring's buffer size. -func (tx *ringTx) addOff(a, b int) int { - off := a + b - if off >= len(tx.rawbuf) { - off -= len(tx.rawbuf) +// 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 { + result := a + b + if result > len(tx.rawbuf) { + result -= len(tx.rawbuf) } - return off + return result +} + +func (tx *ringTx) addOff(a, b int) int { + result := a + b + if result >= len(tx.rawbuf) { + result -= len(tx.rawbuf) + } + return result } func (tx *ringTx) pkt(i int) *ringidx { @@ -180,14 +192,9 @@ func (tx *ringTx) firstPkt() int { idx := -1 for i := 0; i < len(tx.packets); i++ { pkt := &tx.packets[i] - if pkt.sent() { - if idx == -1 { - seq = pkt.seq - } - if seq.LessThan(pkt.seq) { - seq = pkt.seq - idx = i - } + if pkt.sent() && (idx == -1 || seq.LessThan(pkt.seq)) { + seq = pkt.seq + idx = i } } return idx @@ -198,14 +205,9 @@ func (tx *ringTx) lastPkt() int { idx := -1 for i := 0; i < len(tx.packets); i++ { pkt := &tx.packets[i] - if pkt.sent() { - if idx == -1 { - seq = pkt.seq - } - if pkt.seq.LessThan(seq) { - seq = pkt.seq - idx = i - } + if pkt.sent() && (idx == -1 || pkt.seq.LessThan(seq)) { + seq = pkt.seq + idx = i } } return idx @@ -223,6 +225,31 @@ func (tx *ringTx) nextPkt() int { return idx } +// lims returns the limits of free|sent|unsent buffers. +// Example: +// +// | acked(free) | sent | unsent | free | +// 0 freeEnd=first.off last.end==unsent.off freeStart=unsent.end Size() +func (tx *ringTx) lims() (freeStart, freeEnd, sentEndorUnsentStart int) { + freeStart = tx.unsentend + if freeStart == 0 { + freeStart = tx.unsentoff + } + first := tx.pkt(tx.firstPkt()) + if first.sent() { + freeEnd = first.off + sentEndorUnsentStart = tx.unsentoff + } else if tx.unsentend != 0 { + // sent section empty and unsent not empty. + freeEnd = tx.unsentoff + sentEndorUnsentStart = tx.unsentoff + } else { + freeEnd = tx.unsentoff + sentEndorUnsentStart = tx.unsentoff + } + return freeStart, freeEnd, sentEndorUnsentStart +} + func (pkt *ringidx) sent() bool { return pkt.end != 0 || pkt.off != 0 } diff --git a/tcp/txqueue_test.go b/tcp/txqueue_test.go index 9daf248..0905a86 100644 --- a/tcp/txqueue_test.go +++ b/tcp/txqueue_test.go @@ -11,21 +11,37 @@ func TestTxQueue(t *testing.T) { rng := rand.New(rand.NewSource(1)) var rtx ringTx - t.Run("SequentialMessages", func(t *testing.T) { - 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, Value(rng.Int())) + increasingComplexityTests := []struct { + name string + test func(*testing.T) + }{ + 0: { + name: "SequentialMessages", + test: func(t *testing.T) { + 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) + } + }, + }, + 1: { + name: "N-Messages", + test: func(t *testing.T) { + 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) + } + }, + }, + } + for i, test := range increasingComplexityTests { + t.Run(test.name, test.test) + if t.Failed() { + t.Fatalf("subtest %d/%d %q failed, not running more complex tests until fixed", i+1, len(increasingComplexityTests), test.name) } - }) - t.Run("N-Messages", func(t *testing.T) { - for i := 0; i < 10; i++ { - rng.Read(msgBuf[:]) - msgs := bytes.SplitAfter(msgBuf[:], []byte{0}) - - testTxQueue_NMessages(t, &rtx, msgs, buf[:], aux[:], len(msgs), Value(rng.Int())) - } - }) + } } func testTxQueue_NMessages(t *testing.T, rtx *ringTx, msgs [][]byte, buf, aux []byte, maxPkt int, startAck Value) { @@ -53,10 +69,12 @@ func testTxQueue_NMessages(t *testing.T, rtx *ringTx, msgs [][]byte, buf, aux [] } else if n != len(msg) { t.Fatalf("want %d written, got %d", len(msg), n) } + testQueueSanity(t, rtx) unsent := rtx.Buffered() if unsent != n { t.Fatalf("want unset %d, got %d", n, unsent) } + testQueueSanity(t, rtx) n, seq, err := rtx.MakePacket(aux[sent : sent+len(msg)]) if err != nil { t.Fatal(err) @@ -65,10 +83,12 @@ func testTxQueue_NMessages(t *testing.T, rtx *ringTx, msgs [][]byte, buf, aux [] } else if n != len(msg) { t.Fatalf("want full message %d sent, got %d", len(msg), n) } + testQueueSanity(t, rtx) gotSent := rtx.BufferedSent() if gotSent != sent+n { t.Fatalf("want sent %d, got %d", sent+n, gotSent) } + testQueueSanity(t, rtx) packets = append(packets, aux[sent:sent+n]) prevSeq += Value(n) sent += n @@ -91,14 +111,17 @@ func testTxQueue_SequentialMessages(t *testing.T, rtx *ringTx, msgs [][]byte, bu } else if n != len(msg) { t.Fatalf("want %d written, got %d", len(msg), n) } + testQueueSanity(t, rtx) unsent := rtx.Buffered() if len(msg) != unsent { - t.Fatalf("want %d unsent buffered, got %d", unsent, len(msg)) + t.Fatalf("want %d unsent buffered, got %d", len(msg), unsent) } + testQueueSanity(t, rtx) sent := rtx.BufferedSent() if sent != 0 { t.Fatalf("want 0 bytes sent, got %d", sent) } + testQueueSanity(t, rtx) n, seq, err := rtx.MakePacket(aux[:]) data := aux[:n] if err != nil { @@ -110,14 +133,43 @@ func testTxQueue_SequentialMessages(t *testing.T, rtx *ringTx, msgs [][]byte, bu } else if seq != prevSeq { t.Fatalf("want seq %d, got %d", prevSeq, seq) } + testQueueSanity(t, rtx) sent = rtx.BufferedSent() if sent != len(msg) { t.Fatalf("want %d sent, got %d", len(msg), sent) } + testQueueSanity(t, rtx) prevSeq += Value(n) err = rtx.RecvACK(prevSeq) if err != nil { t.Fatal(err) } + testQueueSanity(t, rtx) + } +} + +func testQueueSanity(t *testing.T, rtx *ringTx) { + t.Helper() + if rtx.emptyRing != (ringidx{}) { + t.Fatalf("empty ring not empty") + } + free := rtx.Free() + sent := rtx.BufferedSent() + unsent := rtx.Buffered() + sz := rtx.Size() + gotSz := free + sent + unsent + if gotSz != sz { + t.Fatalf("want size=%d, got size=%d (free+sent+unsent=%d+%d+%d)", sz, gotSz, free, sent, unsent) + } + freeStart, freeEnd, sentEnd := rtx.lims() + gotFreeEnd := rtx.addOff(freeStart, free) + gotSentEnd := rtx.addOff(freeEnd, sent) + gotUnsentEnd := rtx.addOff(sentEnd, unsent) + if free != 0 && gotFreeEnd != freeEnd { + t.Fatalf("want freeEnd=%d, got %d", freeEnd, gotFreeEnd) + } else if sent != 0 && gotSentEnd != sentEnd { + t.Fatalf("want sentEnd=%d, got %d", sentEnd, gotSentEnd) + } else if unsent != 0 && gotUnsentEnd != freeStart { + t.Fatalf("want unsentEnd=%d, got %d (freeStart)", freeStart, gotUnsentEnd) } }