From 51ca0a95b59ecc0f4c62d01ba5a676a6437e4748 Mon Sep 17 00:00:00 2001 From: soypat Date: Sat, 18 Jan 2025 09:30:36 -0300 Subject: [PATCH] rephrase tcp.Value methods; add Ring.FreeLimited; work on TxQueue --- internal/ring.go | 20 ++++++---- tcp/control.go | 12 +++--- tcp/txqueue.go | 89 +++++++++++++++++++++++++++++++++++---------- tcp/txqueue_test.go | 1 + tcp/valuesize.go | 12 +++--- 5 files changed, 95 insertions(+), 39 deletions(-) diff --git a/internal/ring.go b/internal/ring.go index c04a8f8..a99a7ed 100644 --- a/internal/ring.go +++ b/internal/ring.go @@ -23,8 +23,19 @@ type Ring struct { End int } +// 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 + } else { + free = len(r.Buf) - r.End + limitOffset + } + return free +} + // WriteLimited performs a write that does not write over the ring buffer's -// limitOffset index, which points to a position to r.Buf. +// limitOffset index, which points to a position to r.Buf. Up to [Ring.FreeLimited] bytes can be written. func (r *Ring) WriteLimited(b []byte, limitOffset int) (int, error) { if limitOffset > len(r.Buf) { panic("bad limit offset") @@ -32,12 +43,7 @@ func (r *Ring) WriteLimited(b []byte, limitOffset int) (int, error) { if len(b) > len(r.Buf) { return 0, io.ErrShortBuffer } - var limit int - if limitOffset > r.End { - limit = limitOffset - r.End - } else { - limit = len(r.Buf) - r.End + limitOffset - } + limit := r.FreeLimited(limitOffset) if len(b) > limit { return 0, errRingBufferFull } diff --git a/tcp/control.go b/tcp/control.go index 0668af9..4c6bf8b 100644 --- a/tcp/control.go +++ b/tcp/control.go @@ -352,7 +352,7 @@ func (tcb *ControlBlock) validateOutgoingSegment(seg Segment) (err error) { seglast := seg.Last() // Extra check for when send Window is zero and no data is being sent. zeroWindowOK := tcb.snd.WND == 0 && seg.DATALEN == 0 && seg.SEQ == tcb.snd.NXT - outOfWindow := checkSeq && !InWindow(seg.SEQ, tcb.snd.NXT, tcb.snd.WND) && + outOfWindow := checkSeq && !seg.SEQ.InWindow(tcb.snd.NXT, tcb.snd.WND) && !zeroWindowOK switch { case tcb.state == StateClosed: @@ -375,7 +375,7 @@ func (tcb *ControlBlock) validateOutgoingSegment(seg Segment) (err error) { case checkSeq && tcb.snd.WND == 0 && seg.DATALEN > 0 && seg.SEQ == tcb.snd.NXT: err = errZeroWindow - case checkSeq && !InWindow(seglast, tcb.snd.NXT, tcb.snd.WND) && !zeroWindowOK: + case checkSeq && !seglast.InWindow(tcb.snd.NXT, tcb.snd.WND) && !zeroWindowOK: err = errLastNotInWindow } return err @@ -388,8 +388,8 @@ func (tcb *ControlBlock) validateIncomingSegment(seg Segment) (err error) { checkSEQ := !flags.HasAny(FlagSYN) established := tcb.state == StateEstablished preestablished := tcb.state.IsPreestablished() - acksOld := hasAck && !LessThan(tcb.snd.UNA, seg.ACK) - acksUnsentData := hasAck && !LessThanEq(seg.ACK, tcb.snd.NXT) + acksOld := hasAck && !tcb.snd.UNA.LessThan(seg.ACK) + acksUnsentData := hasAck && !seg.ACK.LessThanEq(tcb.snd.NXT) ctlOrDataSegment := established && (seg.DATALEN > 0 || flags.HasAny(FlagFIN|FlagRST)) zeroWindowOK := tcb.rcv.WND == 0 && seg.DATALEN == 0 && seg.SEQ == tcb.rcv.NXT // See section 3.4 of RFC 9293 for more on these checks. @@ -402,10 +402,10 @@ func (tcb *ControlBlock) validateIncomingSegment(seg Segment) (err error) { case checkSEQ && tcb.rcv.WND == 0 && seg.DATALEN > 0 && seg.SEQ == tcb.rcv.NXT: err = errZeroWindow - case checkSEQ && !InWindow(seg.SEQ, tcb.rcv.NXT, tcb.rcv.WND) && !zeroWindowOK: + case checkSEQ && !seg.SEQ.InWindow(tcb.rcv.NXT, tcb.rcv.WND) && !zeroWindowOK: err = errSeqNotInWindow - case checkSEQ && !InWindow(seg.Last(), tcb.rcv.NXT, tcb.rcv.WND) && !zeroWindowOK: + case checkSEQ && !seg.Last().InWindow(tcb.rcv.NXT, tcb.rcv.WND) && !zeroWindowOK: err = errLastNotInWindow case checkSEQ && seg.SEQ != tcb.rcv.NXT: diff --git a/tcp/txqueue.go b/tcp/txqueue.go index 4543345..4e3b611 100644 --- a/tcp/txqueue.go +++ b/tcp/txqueue.go @@ -23,9 +23,9 @@ type ringTx struct { rawbuf []byte // packets contains packets []ringidx - // firstPkt is the index of the oldest packet in the packets field. - firstPkt int - lastPkt int + // _firstPkt is the index of the oldest packet in the packets field. + _firstPkt int + _lastPkt int // unsentOff is the offset of start of unsent data into rawbuf. unsentoff int // unsentend is the offset of end of unsent data in rawbuf. @@ -58,21 +58,27 @@ func (tx *ringTx) BufferedSent() int { } // Write writes data to the underlying unsent data ring buffer. -func (tx *ringTx) Write(b []byte) (int, error) { - first := tx.packets[tx.firstPkt] +func (tx *ringTx) Write(b []byte) (n int, err error) { + first := tx.packets[tx._firstPkt] r := tx.unsentRing() if first.off < 0 { // No packets in queue case. - return r.Write(b) + n, err = r.Write(b) + } else { + n, err = r.WriteLimited(b, first.off) } - return r.WriteLimited(b, first.off) + if err != nil { + return 0, err + } + tx.unsentend = tx.addOff(tx.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, error) { - nxtpkt := (tx.lastPkt + 1) % len(tx.packets) - if tx.firstPkt == nxtpkt { + nxtpkt := (tx._lastPkt + 1) % len(tx.packets) + if tx._firstPkt == nxtpkt { return 0, errors.New("packet queue full") } @@ -82,20 +88,20 @@ func (tx *ringTx) MakePacket(b []byte) (int, error) { if err != nil { return n, err } - last := &tx.packets[tx.lastPkt] - rlast := tx.packetRing(tx.lastPkt) + last := &tx.packets[tx._lastPkt] + rlast := tx.packetRing(tx._lastPkt) tx.packets[nxtpkt].off = start - tx.packets[nxtpkt].end = r.Off + tx.packets[nxtpkt].end = tx.addOff(start, n) tx.packets[nxtpkt].seq = last.seq + Value(rlast.Buffered()) - tx.lastPkt = nxtpkt - tx.unsentoff = r.Off + tx._lastPkt = nxtpkt + tx.unsentoff = tx.addOff(tx.unsentoff, n) return n, nil } // IsQueueFull returns true if the sent packet queue is full in which // case a call to ReadPacket is guaranteed to fail. func (tx *ringTx) IsQueueFull() bool { - return tx.firstPkt == (tx.lastPkt+1)%len(tx.packets) + return tx._firstPkt == (tx._lastPkt+1)%len(tx.packets) } func (tx *ringTx) packetRing(i int) internal.Ring { @@ -108,16 +114,16 @@ func (tx *ringTx) packetRing(i int) internal.Ring { // RecvSegment processes an incoming segment and updates the sent packet queue func (tx *ringTx) RecvACK(ack Value) error { - i := tx.firstPkt + i := tx._firstPkt for { pkt := &tx.packets[i] if ack >= pkt.seq { // Packet was received by remote. Mark it as acked. pkt.off = -1 - tx.firstPkt++ + tx._firstPkt++ continue } - if i == tx.lastPkt { + if i == tx._lastPkt { break } i = (i + 1) % len(tx.packets) @@ -129,15 +135,58 @@ func (tx *ringTx) unsentRing() internal.Ring { return tx.ring(tx.unsentoff, tx.unsentend) } +func (tx *ringTx) freeRing() (internal.Ring, int) { + return tx.ring(tx.unsentoff, tx.unsentend), 0 +} + +func (tx *ringTx) a() { + +} + func (tx *ringTx) sentRing() internal.Ring { - first := tx.packets[tx.firstPkt] + first := tx.packets[tx._firstPkt] if first.off < 0 { return tx.ring(0, 0) } - last := tx.packets[tx.lastPkt] + last := tx.packets[tx._lastPkt] return tx.ring(first.off, last.end) } 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) + } + return off +} + +func (tx *ringTx) firstPkt() int { + seq := tx.packets[0].seq + idx := -1 + for i := 0; i < len(tx.packets); i++ { + pkt := &tx.packets[i] + if (pkt.end != 0 || pkt.off != 0) && seq.LessThanEq(pkt.seq) { + seq = pkt.seq + idx = i + } + } + return idx +} + +func (tx *ringTx) lastPkt() int { + seq := tx.packets[0].seq + idx := -1 + for i := 0; i < len(tx.packets); i++ { + pkt := &tx.packets[i] + if (pkt.end != 0 || pkt.off != 0) && pkt.seq.LessThanEq(seq) { + seq = pkt.seq + idx = i + } + } + return idx +} diff --git a/tcp/txqueue_test.go b/tcp/txqueue_test.go index 00b3d72..4715d8a 100644 --- a/tcp/txqueue_test.go +++ b/tcp/txqueue_test.go @@ -23,6 +23,7 @@ func TestTxQueueWrite(t *testing.T) { } else if n != len(buf) { t.Fatalf("want %d written, got %d", len(buf), n) } + n, err = rtx.MakePacket(data[:]) if err != nil { t.Fatalf("making packet %d: %s", i, err) diff --git a/tcp/valuesize.go b/tcp/valuesize.go index 38deda3..2051720 100644 --- a/tcp/valuesize.go +++ b/tcp/valuesize.go @@ -24,24 +24,24 @@ type Value uint32 type Size uint32 // LessThan checks if v is before w (modulo 32) i.e., v < w. -func LessThan(v, w Value) bool { +func (v Value) LessThan(w Value) bool { return int32(v-w) < 0 } // LessThanEq returns true if v==w or v is before (modulo 32) i.e., v < w. -func LessThanEq(v, w Value) bool { - return v == w || LessThan(v, w) +func (v Value) LessThanEq(w Value) bool { + return v == w || v.LessThan(w) } // InRange checks if v is in the range [a,b) (modulo 32), i.e., a <= v < b. -func InRange(v, a, b Value) bool { +func (v Value) InRange(a, b Value) bool { return v-a < b-a } // InWindow checks if v is in the window that starts at 'first' and spans 'size' // sequence numbers (modulo 32). -func InWindow(v, first Value, size Size) bool { - return InRange(v, first, Add(first, size)) +func (v Value) InWindow(first Value, size Size) bool { + return v.InRange(first, Add(first, size)) } // Add calculates the sequence number following the [v, v+s) window.