diff --git a/tcp/txqueue.go b/tcp/txqueue.go index 44fdccb..5691872 100644 --- a/tcp/txqueue.go +++ b/tcp/txqueue.go @@ -115,12 +115,13 @@ func (tx *ringTx) MakePacket(b []byte) (int, Value, error) { } plen := Value(n) seq := tx.seq - tx.packets[nxtpkt].off = start - tx.packets[nxtpkt].end = tx.addOff(start, n) - tx.packets[nxtpkt].seq = seq + plen - tx.unsentoff = tx.addOff(tx.unsentoff, n) tx.seq += plen + + pkt := &tx.packets[nxtpkt] + pkt.off = start + pkt.end = tx.addOff(start, n) + pkt.seq = seq + plen return n, seq, nil } @@ -175,26 +176,36 @@ func (tx *ringTx) pkt(i int) *ringidx { } func (tx *ringTx) firstPkt() int { - seq := tx.packets[0].seq + var seq Value idx := -1 for i := 0; i < len(tx.packets); i++ { pkt := &tx.packets[i] - if pkt.sent() && seq.LessThanEq(pkt.seq) { - seq = pkt.seq - idx = i + if pkt.sent() { + if idx == -1 { + seq = pkt.seq + } + if seq.LessThan(pkt.seq) { + seq = pkt.seq + idx = i + } } } return idx } func (tx *ringTx) lastPkt() int { - seq := tx.packets[0].seq + var seq Value idx := -1 for i := 0; i < len(tx.packets); i++ { pkt := &tx.packets[i] - if pkt.sent() && pkt.seq.LessThanEq(seq) { - seq = pkt.seq - idx = i + if pkt.sent() { + if idx == -1 { + seq = pkt.seq + } + if pkt.seq.LessThan(seq) { + seq = pkt.seq + idx = i + } } } return idx diff --git a/tcp/txqueue_test.go b/tcp/txqueue_test.go index 4ccae0f..9daf248 100644 --- a/tcp/txqueue_test.go +++ b/tcp/txqueue_test.go @@ -2,27 +2,89 @@ package tcp import ( "bytes" + "math/rand" "testing" ) -func TestTxQueue_SequentialMessages(t *testing.T) { - const ( - bufsize = 2 - maxPkt = 1 - msg = "hello world" - startAck = 0 // this is the initial sequence number. - ) - buf := make([]byte, bufsize) +func TestTxQueue(t *testing.T) { + var msgBuf, buf, aux [1024]byte + 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())) + } + }) + 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) { + if len(msgs) > maxPkt { + panic("need ring buffer to contain messages") + } + err := rtx.Reset(buf, maxPkt, startAck) + if err != nil { + t.Fatal(err) + } + + prevSeq := Value(startAck) + packets := make([][]byte, len(msgs)) + sent := 0 + for i := range aux { + aux[i] = 0 + } + for i, msg := range msgs { + if len(aux) < len(msg) { + panic("need aux to contain message") + } + n, err := rtx.Write(msg) + if err != nil { + t.Fatalf("writing packet %d: %s", i, err) + } else if n != len(msg) { + t.Fatalf("want %d written, got %d", len(msg), n) + } + unsent := rtx.Buffered() + if unsent != n { + t.Fatalf("want unset %d, got %d", n, unsent) + } + n, seq, err := rtx.MakePacket(aux[sent : sent+len(msg)]) + if err != nil { + t.Fatal(err) + } else if seq != prevSeq { + t.Fatalf("want seq %d, got %d", prevSeq, seq) + } else if n != len(msg) { + t.Fatalf("want full message %d sent, got %d", len(msg), n) + } + gotSent := rtx.BufferedSent() + if gotSent != sent+n { + t.Fatalf("want sent %d, got %d", sent+n, gotSent) + } + packets = append(packets, aux[sent:sent+n]) + prevSeq += Value(n) + sent += n + } +} + +func testTxQueue_SequentialMessages(t *testing.T, rtx *ringTx, msgs [][]byte, buf, aux []byte, maxPkt int, startAck Value) { err := rtx.Reset(buf, maxPkt, startAck) if err != nil { t.Fatal(err) } - // msgs := bytes.SplitAfter([]byte(msg), []byte("e")) - msgs := bytes.Split([]byte(msg), []byte("")) - var data [bufsize]byte prevSeq := Value(startAck) for i, msg := range msgs { + if len(aux) < len(msg) { + panic("need aux to contain message") + } n, err := rtx.Write(msg) if err != nil { t.Fatalf("writing packet %d: %s", i, err) @@ -37,12 +99,13 @@ func TestTxQueue_SequentialMessages(t *testing.T) { if sent != 0 { t.Fatalf("want 0 bytes sent, got %d", sent) } - n, seq, err := rtx.MakePacket(data[:]) + n, seq, err := rtx.MakePacket(aux[:]) + data := aux[:n] if err != nil { t.Fatalf("making packet %d: %s", i, err) } else if n != len(msg) { t.Fatalf("want %d packet read, got %d", len(msg), n) - } else if !bytes.Equal(msg, data[:n]) { + } else if !bytes.Equal(msg, aux[:n]) { t.Fatalf("want data %q, got data read %q", msg, data[:n]) } else if seq != prevSeq { t.Fatalf("want seq %d, got %d", prevSeq, seq)