diff --git a/arp/handler_test.go b/arp/arp_test.go similarity index 100% rename from arp/handler_test.go rename to arp/arp_test.go diff --git a/tcp/txqueue.go b/tcp/txqueue.go index 3f47ddb..d7841bd 100644 --- a/tcp/txqueue.go +++ b/tcp/txqueue.go @@ -2,11 +2,15 @@ package tcp import ( "errors" - "fmt" + "slices" "github.com/soypat/lneto/internal" ) +var ( + errPacketQueueFull = errors.New("packet queue full") +) + const ( // this must be at least 2 for buffer to work. minBufferSize = 2 @@ -19,8 +23,7 @@ const ( type ringTx struct { // rawbuf contains the ring buffer of ordered bytes. It should be the size of the window. rawbuf []byte - // packets contains - packets []ringidx + 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. @@ -57,16 +60,11 @@ func (rtx *ringTx) Reset(buf []byte, maxqueuedPackets int, iss Value) error { } else if len(buf) < minBufferSize || len(buf) < maxqueuedPackets { return errors.New("invalid buffer size") } - if cap(rtx.packets) < maxqueuedPackets { - rtx.packets = make([]ringidx, maxqueuedPackets) - } + *rtx = ringTx{ - rawbuf: buf, - packets: rtx.packets[:maxqueuedPackets], - } - for i := range rtx.packets { - rtx.packets[i].markRcvd() + rawbuf: buf, } + rtx.slist.Reset(maxqueuedPackets, iss) rtx.iss = iss return nil } @@ -78,7 +76,7 @@ func (rtx *ringTx) ResetOrReuse(buf []byte, maxQueuedPackets int, ack Value) err buf = rtx.rawbuf } if maxQueuedPackets == 0 { - maxQueuedPackets = len(rtx.packets) + maxQueuedPackets = cap(rtx.slist.pkts) } return rtx.Reset(buf, maxQueuedPackets, ack) } @@ -118,106 +116,50 @@ func (rtx *ringTx) Write(b []byte) (n int, err error) { // MakePacket reads from the unsent data ring buffer and generates a new packet segment. // It fails if the sent packet queue is full. func (rtx *ringTx) MakePacket(b []byte, currentSeq Value) (int, error) { - nxtpkt := rtx.nextPkt() - if nxtpkt < 0 { - return 0, errors.New("queue full") + free := rtx.slist.Free() + if free == 0 { + return 0, errPacketQueueFull } endSeq, ok := rtx.endSeq() 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() - start := r.Off + oldSentOff := r.Off n, err := r.Read(b) if err != nil { return n, err } - pkt := &rtx.packets[nxtpkt] - - off := rtx.addEnd(rtx.unsentoff, n) - rtx.unsentoff = off - rtx.sentend = off - if off == rtx.unsentend { - rtx.unsentend = 0 // Mark unsent as being empty. + // 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) + pkt := rtx.slist.AddPacket(n, oldSentOff, rtx.Size()) + if pkt.off != oldSentOff || pkt.end != addEnd(pkt.off, n, rtx.Size()) { + panic("invalid generated packet") } - *pkt = ringidx{ - off: start, - end: off, - seq: currentSeq, - size: Size(n), + rtx.unsentoff = newUnsentOff + rtx.sentend = newUnsentOff + if newUnsentOff == rtx.unsentend { + rtx.unsentend = 0 // Mark unsent as being empty. } return n, nil } // RecvSegment processes an incoming segment and updates the sent packet queue func (rtx *ringTx) RecvACK(ack Value) error { - first := rtx.firstPkt() - if first < 0 { - return errors.New("no packets to ack") - } - pkt0 := rtx.pkt(first) - if ack.LessThanEq(pkt0.seq) { - return fmt.Errorf("incoming ack %d older than first packet seq %d", ack, pkt0.seq) - } - // lastAckedPkt stores last fully acked packet. - var lastAckedPkt *ringidx - var partialPkt *ringidx - for i := 0; i < len(rtx.packets); i++ { - pkt := &rtx.packets[i] - if !pkt.sent() || ack.LessThanEq(pkt.seq) { - continue - } - endseq := pkt.endSeq() - 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") - } else { - // Is partial acked. - if partialPkt != nil { - panic("unreachable") // can't have two partially acked packets. - } - acked := int(ack - pkt.seq) - pring := rtx.ring(pkt.off, pkt.end) - buffered := pring.Buffered() - if acked > buffered { - panic("unreachable") - } - off := rtx.addOff(pkt.off, acked) - pkt.off = off - pkt.seq = ack - pkt.size = pkt.size - Size(acked) - rtx.sentoff = off - partialPkt = pkt - } - } - if partialPkt != nil { - return nil - } - if lastAckedPkt != nil { - rtx.sentoff = lastAckedPkt.end - lastAckedPkt.markRcvd() - if rtx.sentoff == rtx.sentend { - // All data acked. - rtx.sentend = 0 - rtx.consolidateBufs() - } + rtx.slist.RecvAck(ack, rtx.Size()) + oldest := rtx.slist.Oldest() + newest := rtx.slist.Newest() + if oldest == nil { + // All sent data received, discard. + rtx.sentend = 0 + } else { + rtx.sentoff = oldest.off + rtx.sentend = newest.end } + rtx.consolidateBufs() return nil } @@ -245,55 +187,6 @@ func (rtx *ringTx) ring(off, end int) internal.Ring { // Result of addEnd will never be 0 unless arguments are (0,0). func (rtx *ringTx) addEnd(a, b int) int { return addEnd(a, b, len(rtx.rawbuf)) } -func (rtx *ringTx) addOff(a, b int) int { return addOff(a, b, len(rtx.rawbuf)) } - -func (rtx *ringTx) pkt(i int) *ringidx { - if i == -1 { - return &rtx.emptyRing - } else if i < 0 || i >= len(rtx.packets) { - panic("invalid packet index") - } - return &rtx.packets[i] -} - -func (rtx *ringTx) firstPkt() int { - var seq Value - idx := -1 - 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 - } - } - return idx -} - -func (rtx *ringTx) lastPkt() int { - var seq Value - idx := -1 - 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 - } - } - return idx -} - -func (rtx *ringTx) nextPkt() int { - idx := -1 - for i := 0; i < len(rtx.packets); i++ { - pkt := &rtx.packets[i] - if !pkt.sent() { - idx = i - break - } - } - return idx -} - func (rtx *ringTx) consolidateBufs() { unsentEmpty := rtx.unsentend == 0 sentEmpty := rtx.sentend == 0 @@ -305,26 +198,11 @@ func (rtx *ringTx) consolidateBufs() { } func (rtx *ringTx) endSeq() (Value, bool) { - pkt := rtx.lastPkt() - if pkt < 0 { + newest := rtx.slist.Newest() + if newest == nil { return 0, false } - last := rtx.pkt(pkt) - return last.endSeq(), true -} -func (rtx *ringTx) lastSeq() (Value, bool) { - pkt := rtx.lastPkt() - if pkt < 0 { - return 0, false - } - return rtx.pkt(pkt).seq, true -} -func (rtx *ringTx) firstSeq() (Value, bool) { - pkt := rtx.firstPkt() - if pkt < 0 { - return 0, false - } - return rtx.pkt(pkt).seq, true + return newest.endSeq(), true } // lims returns the limits of free|sent|unsent buffers. @@ -356,11 +234,19 @@ func (pkt *ringidx) endSeq() Value { // sentlist stores information about sent TCP packets type sentlist struct { + // ssn is an auxiliary sequence counter. + // If there are no packets then ssn is reset to be the end sequence number of the last acked packet such that + // the next packet added has their + ssn Value // pkts is an ordered list of packets. First packet is 'oldest' packet, last packet is the most recently sent. - iss Value pkts []ringidx } +func (sl *sentlist) Reset(pktQueueSize int, iss Value) { + sl.pkts = slices.Grow(sl.pkts[:0], pktQueueSize) + sl.ssn = iss +} + func (sl sentlist) Newest() *ringidx { if len(sl.pkts) == 0 { return nil @@ -376,7 +262,7 @@ func (sl sentlist) Oldest() *ringidx { } func (sl *sentlist) EndSeq() Value { - seq := sl.iss + seq := sl.ssn lastPkt := sl.Newest() if lastPkt != nil { seq = lastPkt.endSeq() @@ -384,23 +270,26 @@ func (sl *sentlist) EndSeq() Value { return seq } -func (sl *sentlist) AddPacket(datalen int, bufsize int) { - free := cap(sl.pkts) - len(sl.pkts) +func (sl *sentlist) Free() int { + return cap(sl.pkts) - len(sl.pkts) +} + +func (sl *sentlist) AddPacket(datalen, off, bufsize int) *ringidx { + free := sl.Free() if free == 0 { panic("pkt buffer full") } lastPkt := sl.Newest() - lastEnd := 0 - if lastPkt != nil { - lastEnd = lastPkt.end + if lastPkt != nil && off != lastPkt.end { + panic("new sent packet offset must match last sent packet end") } - pkt := ringidx{ - off: lastEnd, - end: addEnd(lastEnd, datalen, bufsize), + sl.pkts = append(sl.pkts, ringidx{ + off: off, + end: addEnd(off, datalen, bufsize), seq: sl.EndSeq(), size: Size(datalen), - } - sl.pkts = append(sl.pkts, pkt) + }) + return &sl.pkts[len(sl.pkts)-1] } func (sl *sentlist) RecvAck(ack Value, bufsize int) { @@ -410,7 +299,7 @@ func (sl *sentlist) RecvAck(ack Value, bufsize int) { endseq := pkt.endSeq() isFullyAcked := endseq.LessThanEq(ack) if isFullyAcked { - sl.iss = endseq + sl.ssn = endseq pkt.markRcvd() } else { break @@ -432,6 +321,9 @@ func (sl *sentlist) RecvAck(ack Value, bufsize int) { } func (sl *sentlist) removeRecvd() { + if !sl.Oldest().isRecvd() { + return // No packets to remove. + } off := 0 for i := 0; i < len(sl.pkts); i++ { if sl.pkts[i].isRecvd() { diff --git a/tcp/txqueue_test.go b/tcp/txqueue_test.go index 38993ff..b41a454 100644 --- a/tcp/txqueue_test.go +++ b/tcp/txqueue_test.go @@ -8,14 +8,42 @@ import ( "testing" ) -func TestSentlist(t *testing.T) { - sl := sentlist{ - pkts: make([]ringidx, 0, 3), +func TestSentlist_multi(t *testing.T) { + const bufsize = 10 + var sl sentlist + sl.Reset(3, 0) + + // Test multi packet x2. + p1 := sl.AddPacket(5, 0, bufsize) + p2 := sl.AddPacket(5, p1.end, bufsize) + sl.RecvAck(Value(p2.size+p1.size), bufsize) + if sl.Oldest() != nil { + t.Fatal("expected full ack") } + // multi packet x3. + sl.Reset(3, 0) + p1 = sl.AddPacket(3, 0, bufsize) + p2 = sl.AddPacket(3, p1.end, bufsize) + p3 := sl.AddPacket(4, p2.end, bufsize) + sl.RecvAck(2, bufsize) + oldest := sl.Oldest() + if oldest != p1 { + t.Error("oldest should be partial acked") + } else if oldest.size != 1 { + t.Error("bad size") + } else if oldest.off != 2 { + t.Error("bad offset") + } + _ = p3 +} + +func TestSentlist(t *testing.T) { + var sl sentlist + sl.Reset(3, 0) // Test full ack. const bufsize = 16 const pkt = 10 - sl.AddPacket(pkt, bufsize) + sl.AddPacket(pkt, 0, bufsize) if sl.Oldest() == nil || sl.Newest() != sl.Oldest() { t.Error("expected same oldest/newest non-nil packet") } @@ -27,7 +55,7 @@ func TestSentlist(t *testing.T) { } // Test partial ack. - sl.AddPacket(pkt, bufsize) + sl.AddPacket(pkt, 0, bufsize) for i := Value(0); i < pkt-1; i++ { ack++ sl.RecvAck(ack, bufsize) @@ -274,6 +302,10 @@ func testQueueSanity(t *testing.T, rtx *ringTx) { alreadyFailed := t.Failed() if !alreadyFailed { defer func() { + a := recover() + if a != nil { + t.Log("panic", a) + } if t.Failed() { t.Helper() t.Log("sanity failed with:\n" + rtx.string()) @@ -291,7 +323,7 @@ func testQueueSanity(t *testing.T, rtx *ringTx) { sz := rtx.Size() gotSz := free + sent + unsent if gotSz != sz { - t.Fatal("\n" + rtx.string()) + 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() @@ -329,21 +361,21 @@ func testQueueSanity(t *testing.T, rtx *ringTx) { } // Check sanenness of last/first packets. - last := rtx.lastPkt() - first := rtx.firstPkt() - if first < 0 && last >= 0 || last < 0 && first >= 0 { + last := rtx.slist.Newest() + first := rtx.slist.Oldest() + if first == nil && last != nil || last == nil && first != nil { t.Fatalf("found first/last(%d,%d) but did not find last/first", first, last) } // Check sent data or return if no sent data available. if sent == 0 { return } - lastPkt := rtx.pkt(last) + endseq, ok := rtx.endSeq() - firstPkt := rtx.pkt(first) - lastEndSeq := Add(lastPkt.seq, lastPkt.size) - if lastPkt.seq.LessThan(firstPkt.seq) { - t.Fatalf("first packet not previous to last packet seq, wanted %d<%d", firstPkt.seq, lastPkt.seq) + + lastEndSeq := Add(last.seq, last.size) + if last.seq.LessThan(first.seq) { + t.Fatalf("first packet not previous to last packet seq, wanted %d<%d", first.seq, last.seq) } else if !ok { t.Fatal("unexpected end sequence not found") } else if lastEndSeq != endseq { @@ -433,7 +465,12 @@ func operateOnRing(t *testing.T, rtx *ringTx, write, readPacket, aux []byte, new // Prepare aux with data expected from read after write. runsent, _ := rtx.unsentRing() unsent := runsent.Buffered() - startSeq, startSeqOK := rtx.firstSeq() + oldest := rtx.slist.Oldest() + var startSeq Value + startSeqOk := oldest != nil + if startSeqOk { + startSeq = oldest.seq + } wantWritten := min(free, len(write)) wantBufRead := aux[:min(unsent+wantWritten, len(readPacket))] @@ -483,7 +520,12 @@ func operateOnRing(t *testing.T, rtx *ringTx, write, readPacket, aux []byte, new } else if n != wantRead { t.Errorf("want read %d, got %d", wantRead, n) } - lastSeq, lastSeqOK := rtx.lastSeq() + last := rtx.slist.Newest() + var lastSeq Value + lastSeqOK := last != nil + if lastSeqOK { + lastSeq = last.seq + } endSeq, endSeqOK := rtx.endSeq() if !lastSeqOK || lastSeq != newPacketSeq { t.Fatalf("expected last seq to be %d, got %d (or lastSeqOK=%v)", newPacketSeq, lastSeq, lastSeqOK) @@ -498,9 +540,14 @@ func operateOnRing(t *testing.T, rtx *ringTx, write, readPacket, aux []byte, new t.Errorf("want data written to be %d calculated from BufferedSent diff, got %d", n, gotCalcRead) } } + oldest2 := rtx.slist.Oldest() + var startSeq2 Value + sseqOk := oldest != nil + if sseqOk { + startSeq2 = oldest2.seq + } - startSeq2, sseqOK := rtx.firstSeq() - if sseqOK == startSeqOK && startSeq2 != startSeq { + if sseqOk == startSeqOk && startSeq2 != startSeq { t.Fatalf("expected FIRST seq to not change during writes") } @@ -508,7 +555,12 @@ func operateOnRing(t *testing.T, rtx *ringTx, write, readPacket, aux []byte, new testQueueSanity(t, rtx) // preAcked := rtx.BufferedSent() rcvAck := *argRecvAck - seq, ok := rtx.firstSeq() + oldest := rtx.slist.Oldest() + var seq Value + ok := oldest != nil + if ok { + seq = oldest.seq + } if !ok { t.Fatal("no first packet found") } @@ -519,7 +571,12 @@ func operateOnRing(t *testing.T, rtx *ringTx, write, readPacket, aux []byte, new t.Errorf("expected correct acking %d < %d <= %d: %s", startSeq, rcvAck, seq, err) } bufSent := rtx.BufferedSent() - gotFirstSeq, ok := rtx.firstSeq() + var gotFirstSeq Value + oldest = rtx.slist.Oldest() + ok = oldest != nil + if ok { + gotFirstSeq = oldest.seq + } if !ok && bufSent != 0 { t.Fatalf("no first packet found after acking") }