lneto/tcp/txqueue_test.go

package tcp

import (
	"bytes"
	"math/rand"
	"slices"
	"testing"
	"unsafe"

	"github.com/soypat/lneto/internal"
)

func TestRingTx_op(t *testing.T) {
	const maxBuf = 32
	const maxpkt = 3
	const Nops = 6
	const log = false
	type op uint8
	const (
		opWrite op = iota
		opSend
		opAck
		opmax
	)
	var buf, auxbuf [maxBuf]byte
	dataWritten := make([]byte, 0, maxBuf*10)
	dataSent := make([]byte, 0, maxBuf*10)
	var rtx ringTx
	rng := rand.New(rand.NewSource(0))
	for iseed := range int64(1000) {
		rng.Seed(iseed + rng.Int63())
		for itest := range 32 {
			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 := range Nops {
				free := bufsize - nsent - nunsent
				availPkt := rtx.slist.Free()
				op := op(rng.Intn(int(opmax)))
				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)
				case opSend:
					opname, oplen = "send", rng.Intn(nunsent+1)+1
				case opAck:
					opname, oplen = "ack", rng.Intn(nsent+1)+1
				}
				if log {
					t.Logf("\n%s\nseed=%d itest=%d iop=%d op=%s len=%d npkt=%d", rtx.mustAppendString(nil), iseed, itest, iop, opname, oplen, len(rtx.slist.pkts))
				}
				switch op {
				case opWrite:
					// oplen=number of bytes to write into unsent buffer.
					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 if log {
						t.Logf("opwrite: %s", err)
					}
					clear(opWriteData)
				case opSend:
					// oplen=num bytes to send in this operation.
					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 if log {
						t.Logf("opsend: %s", err)
					}
					clear(auxbuf[:oplen])
				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 if log {
						t.Logf("opack: %s", err)
					}
				default:
					panic("unknown op")
				}
				wantFree := bufsize - nsent - nunsent
				gotFree := rtx.Free()
				if wantFree != gotFree {
					t.Fatalf("free mismatch got=%d want=%d  size=%d sent=%d nunsent=%d", gotFree, wantFree, bufsize, nsent, nunsent)
				}
				minlen := min(len(dataSent), len(dataWritten))
				matched := bytes.Equal(dataSent[:minlen], dataWritten[:minlen])
				if !matched {
					t.Fatal("mismatched data written\n", dataSent, "\n", dataWritten)
				}
				testQueueSanity(t, &rtx)
			}
		}
	}
}

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, 0)
	p2 := sl.AddPacket(5, p1.end, bufsize, p1.endSeq())
	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, 0)
	p2 = sl.AddPacket(3, p1.end, bufsize, p1.endSeq())
	p3 := sl.AddPacket(4, p2.end, bufsize, p2.endSeq())
	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_simple(t *testing.T) {
	var sl sentlist
	sl.Reset(3, 0)
	// Test full ack.
	const bufsize = 16
	const pkt = 10
	sl.AddPacket(pkt, 0, bufsize, 0)
	if sl.Oldest() == nil || sl.Newest() != sl.Oldest() {
		t.Error("expected same oldest/newest non-nil packet")
	}
	ack := Value(pkt)
	sl.RecvAck(ack, bufsize)
	oldest := sl.Oldest()
	if oldest != nil {
		t.Fatal("expected packet to be fully read")
	}

	// Test partial ack.
	sl.AddPacket(pkt, 0, bufsize, sl.ssn)
	for range Value(pkt - 1) {
		ack++
		sl.RecvAck(ack, bufsize)
		oldest = sl.Oldest()
		if oldest == nil {
			t.Fatal("partially acked packet removed")
		} else if oldest.seq != ack {
			t.Errorf("want pkt.seq=%d got %d", ack, oldest.seq)
		}
	}
	ack++
	sl.RecvAck(ack, bufsize)
	oldest = sl.Oldest()
	if oldest != nil {
		t.Fatal("expected packet to be fully read")
	}
}

func TestTxQueue_multipacket(t *testing.T) {
	const mtu = 32
	const iss = 1
	const maxPkts = 3
	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 := range 32 {
		rng.Seed(int64(itest))
		err := rtx.Reset(internalbuff, maxPkts, iss)
		if err != nil {
			t.Fatal(err)
		}
		numWrites := rng.Intn(maxWrites) + 1
		total := 0
		woff := 0
		for range numWrites {
			wlen := rng.Intn(maxWriteSize) + 1
			towrite := wbuf[woff : woff+wlen]
			rng.Read(towrite)
			n, err := rtx.Write(towrite)
			testQueueSanity(t, &rtx)
			woff += n
			if err != nil {
				t.Fatal(err)
			} else if n != wlen {
				t.Fatal("expected wlen==n", wlen, n)
			}
			total += wlen
		}
		npkt := rng.Intn(maxPkts) + 1
		roff := 0
		seq := Value(iss)
		for range npkt {
			maxToPacket := min(total-roff, maxWriteSize)
			pktlen := rng.Intn(maxToPacket) + 1
			pkt := rbuf[roff : roff+pktlen]
			expectPkt := wbuf[roff : roff+pktlen]
			ngot, err := rtx.MakePacket(pkt, seq)
			testQueueSanity(t, &rtx)
			roff += ngot
			seq += Value(ngot)
			if err != nil {
				t.Fatal(err)
			} else if pktlen != ngot && roff != total {
				t.Fatal(err)
			} else if !bytes.Equal(expectPkt, pkt) {
				t.Fatal("mismatched data written", expectPkt, pkt)
			}
			if roff == total {
				break // made packet from all data.
			}
		}
		acked := 0
		for acked < roff {
			maxToack := min(roff-acked, maxWriteSize)
			toack := rng.Intn(maxToack) + 1
			// t.Log("\n", rtx.string())
			err = rtx.RecvACK(iss + Value(acked+toack))
			testQueueSanity(t, &rtx)
			if err != nil {
				t.Fatal(err)
			}
			acked += toack
		}
	}
}

func TestTxQueue(t *testing.T) {
	const bufsize = 1024
	var msgBuf, ringBuf, readBuf, aux [bufsize]byte
	rng := rand.New(rand.NewSource(1))
	panicked := true
	var rtx ringTx
	defer func() {
		if panicked {
			t.Error("panicked, rtx:\n", rtx.string())
		}
		testQueueSanity(t, &rtx)
	}()
	increasingComplexityTests := []struct {
		name string
		test func(*testing.T)
	}{
		0: {
			name: "SequentialMessages",
			test: func(t *testing.T) {
				const startAck = 0
				for range 10 {
					rng.Read(msgBuf[:])
					msgs := removeEmptyMsgs(bytes.SplitAfter(msgBuf[:], []byte{0}))
					currentAck := Value(startAck)
					err := rtx.Reset(ringBuf[:], rng.Intn(4)+1, startAck)
					if err != nil {
						t.Fatal(err)
					}
					for imsg, msg := range msgs {
						// Write and create packet from single messages.
						seq := currentAck
						currentAck = Add(currentAck, Size(len(msg)))
						operateOnRing(t, &rtx, msg, readBuf[:], aux[:], seq, &currentAck)
						buffered := rtx.BufferedUnsent()
						if buffered != 0 {
							t.Fatalf("msg%d: want no buffered data after transaction, got %d", imsg, buffered)
						}
					}
				}
			},
		},
		1: {
			name: "N-Messages",
			test: func(t *testing.T) {
				const startAck = 0
				for range 10 {
					rng.Read(msgBuf[:])
					msgs := removeEmptyMsgs(bytes.SplitAfter(msgBuf[:], []byte{0}))
					currentAck := Value(startAck)
					err := rtx.Reset(ringBuf[:], rng.Intn(4)+1, startAck)
					if err != nil {
						t.Fatal(err)
					}
					expectBuffered := 0
					for _, msg := range msgs {
						// Send all messages.
						seq := currentAck
						operateOnRing(t, &rtx, msg, nil, aux[:], seq, nil)
						if t.Failed() {
							return
						}
						expectBuffered += len(msg)
						buffered := rtx.BufferedUnsent()
						if buffered != expectBuffered {
							t.Fatalf("expected seq to not change during writes")
						}
						currentAck = Add(currentAck, Size(len(msg)))
					}
					sent := rtx.BufferedSent()
					unsent := rtx.BufferedUnsent()
					wantUnsent := int(Add(currentAck, -startAck))
					if unsent != wantUnsent {
						t.Fatalf("want %d data buffered, got %d", wantUnsent, unsent)
					} else if sent != 0 {
						t.Fatalf("want no data sent, got %d", sent)
					}
					operateOnRing(t, &rtx, nil, readBuf[:], aux[:], 0, &currentAck)
					unsent = rtx.BufferedUnsent()
					if unsent != 0 {
						t.Fatalf("expected all data to be sent after ack of most recent packet, %d", unsent)
					} else if rtx.BufferedSent() != 0 {
						t.Fatal("unexpected buffer not completely acked")
					}
				}
			},
		},
		2: {
			name: "PartialAcks",
			test: func(t *testing.T) {
				const startAck = 0
				const packets = 100
				const maxPacketSize = bufsize / 4
				var datalens [][]byte
				for range 10 {
					rng.Read(msgBuf[:])
					err := rtx.Reset(ringBuf[:], packets, startAck)
					if err != nil {
						t.Fatal(err)
					}
					operateOnRing(t, &rtx, msgBuf[:], nil, aux[:], 0, nil)
					// Send all bytes over wire.
					currentSeq := Value(startAck)
					datalens = datalens[:0]
					for rtx.BufferedUnsent() != 0 {
						nbytes := rng.Intn(maxPacketSize-minBufferSize) + minBufferSize
						n, err := rtx.MakePacket(readBuf[:nbytes], currentSeq)
						if err != nil {
							t.Fatal(err)
						} else if n == 0 {
							t.Fatal("got zero length")
						}
						// Reuse memory in slice of byte buffers.
						if len(datalens) == cap(datalens) {
							datalens = append(datalens, append([]byte{}, readBuf[:n]...))
						} else {
							datalens = datalens[:len(datalens)+1]
							datalens[len(datalens)-1] = append(datalens[len(datalens)-1][:0], readBuf[:n]...)
						}
						currentSeq += Value(n)
					}
					currentAck := Value(startAck)
					for idata, data := range datalens {
						plen := len(data)
						partialLen0 := plen - (rng.Intn(plen)/2 + minBufferSize)
						// partialLen1 := plen - partialLen0
						// sent := rtx.BufferedSent()
						ack1 := currentAck + Value(partialLen0)
						ack2 := currentAck + Value(plen)
						err = rtx.RecvACK(ack1)
						if err != nil {
							t.Fatalf("data%d acking first partial %d..%d(..%d): %s", idata, currentAck, ack1, ack2, err)
						}
						err = rtx.RecvACK(ack2)
						if err != nil {
							t.Fatalf("data%d acking second partial (%d..)%d..%d: %s", idata, currentAck, ack1, ack2, err)
						}
						currentAck = ack2
					}
				}
			},
		},
	}
	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)
		}
	}
	panicked = false
}

func testQueueSanity(t *testing.T, rtx *ringTx) {
	// t.Helper()
	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())
			}
		}()
	}

	if rtx.emptyRing != (ringidx{}) {
		t.Fatalf("empty ring not empty")
	}

	free := rtx.Free()
	sent := rtx.BufferedSent()
	unsent := rtx.BufferedUnsent()
	sz := rtx.Size()
	gotSz := free + sent + unsent
	if gotSz != sz {
		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()
	sentEmpty := rsent.Buffered() == 0
	runsent, _ := rtx.unsentRing()
	unsentEmpty := runsent.Buffered() == 0
	all := rtx.sentAndUnsentBuffer()
	allEmpty := all.Buffered() == 0
	if !sentEmpty {
		if all.Off != rsent.Off {
			t.Fatalf("want entire buffer start %d to equal sent start %d", all.Off, rsent.Off)
		} else if rsent.End == 0 {
			t.Fatalf("expected not empty sent buffer End to be !=0, got %d", rsent.End)
		}
		gotSentEnd := rtx.addEnd(rsent.Off, sent)
		if gotSentEnd != rsent.End {
			t.Fatalf("calculated sent end mismatches lim sent end %d != %d", gotSentEnd, rsent.End)
		}
	}
	if !unsentEmpty {
		if all.End != runsent.End {
			t.Fatalf("want entire buffer end %d to equal unsent end %d", all.End, runsent.End)
		} else if runsent.End == 0 {
			t.Fatalf("expected not empty unsent buffer End to be !=0, got %d", runsent.End)
		}
		gotUnsentEnd := rtx.addEnd(runsent.Off, unsent)
		if gotUnsentEnd != runsent.End {
			t.Fatalf("calculated unsent end mismatches lim unsent end %d != %d", gotUnsentEnd, runsent.End)
		}
	}
	if allEmpty && (!sentEmpty || !unsentEmpty) {
		t.Fatalf("all buffer empty but sent|unsent(%v/%v) not empty", sentEmpty, unsentEmpty)
	} else if !allEmpty && sentEmpty && unsentEmpty {
		t.Fatal("all buffer not empty but sent&unsentempty")
	}

	// Check sanenness of last/first packets.
	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
	}

	endseq, ok := rtx.sentEndSeq()

	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 {
		t.Fatalf("last packet end sequence not match with got endSeq %d!=%d", lastEndSeq, endseq)
	}
}

func (rx *ringTx) string() string {
	s := rx.appendString(nil)
	return unsafe.String(&s[0], len(s))
}

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, newPacketSeq Value, 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()
	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))]

	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)
	}

	if len(write) != 0 {
		testQueueSanity(t, rtx)
		preBuffered := rtx.BufferedUnsent()
		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.BufferedUnsent()
		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.BufferedUnsent()
		wantRead := min(canRead, len(readPacket))
		if wantRead != len(wantBufRead) {
			t.Fatalf("miscalculated expect read %d != %d", wantRead, len(wantBufRead))
		}
		n, err := rtx.MakePacket(readPacket, newPacketSeq)
		if err != nil && wantRead != 0 {
			t.Errorf("error reading: %s", err)
		} else if n != wantRead {
			t.Errorf("want read %d, got %d", wantRead, n)
		}
		last := rtx.slist.Newest()
		var lastSeq Value
		lastSeqOK := last != nil
		if lastSeqOK {
			lastSeq = last.seq
		}
		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)) {
			t.Fatalf("expected end seq to be %d, got %d (or endSeqOK=%v)", Add(newPacketSeq, Size(n)), endSeq, endSeqOK)
		}
		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)
		}
	}
	oldest2 := rtx.slist.Oldest()
	var startSeq2 Value
	sseqOk := oldest != nil
	if sseqOk {
		startSeq2 = oldest2.seq
	}

	if sseqOk == startSeqOk && startSeq2 != startSeq {
		t.Fatalf("expected FIRST seq to not change during writes")
	}

	if !t.Failed() && argRecvAck != nil {
		testQueueSanity(t, rtx)
		// preAcked := rtx.BufferedSent()
		rcvAck := *argRecvAck
		oldest := rtx.slist.Oldest()
		var seq Value
		ok := oldest != nil
		if ok {
			seq = oldest.seq
		}
		if !ok {
			t.Fatal("no first packet found")
		}
		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)
		}
		bufSent := rtx.BufferedSent()
		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")
		}
		if ok && gotFirstSeq.LessThanEq(rcvAck) {
			t.Fatalf("expected first seq %d to be greater than ack %d", gotFirstSeq, rcvAck)
		}
		// wantAcked := int(Sizeof(prevSeq, gotFirstSeq))
		// if gotCalcAcked != wantAcked {
		// 	t.Errorf("want acked %d, got %d", wantAcked, gotCalcAcked)
		// }
	}
	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,
		},
	}
}