diff --git a/tcp/txqueue.go b/tcp/txqueue.go index 755c067..19d52a6 100644 --- a/tcp/txqueue.go +++ b/tcp/txqueue.go @@ -161,12 +161,13 @@ func (rtx *ringTx) RecvACK(ack Value) error { } // 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 := Add(pkt.seq, pkt.size) + endseq := pkt.endSeq() isFullyAcked := endseq.LessThanEq(ack) isPartialAcked := ack.InRange(pkt.seq, endseq) isLast := lastAckedPkt == nil || lastAckedPkt.seq.LessThanEq(pkt.seq) @@ -186,10 +187,10 @@ func (rtx *ringTx) RecvACK(ack Value) error { pkt.markRcvd() } else if !isPartialAcked { panic("unreachable") - } - if isPartialAcked { - if lastAckedPkt != nil && lastAckedPkt.seq.LessThan(pkt.seq) { - 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) @@ -202,8 +203,12 @@ func (rtx *ringTx) RecvACK(ack Value) error { 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() @@ -317,7 +322,7 @@ func (rtx *ringTx) endSeq() (Value, bool) { return 0, false } last := rtx.pkt(pkt) - return Add(last.seq, last.size), true + return last.endSeq(), true } func (rtx *ringTx) lastSeq() (Value, bool) { pkt := rtx.lastPkt() @@ -352,3 +357,7 @@ func (pkt *ringidx) markRcvd() { // pkt.end = 0 // pkt.off = 0 } + +func (pkt *ringidx) endSeq() Value { + return Add(pkt.seq, pkt.size) +} diff --git a/tcp/txqueue_test.go b/tcp/txqueue_test.go index d463a9b..1e517bb 100644 --- a/tcp/txqueue_test.go +++ b/tcp/txqueue_test.go @@ -16,9 +16,11 @@ func TestTxQueue_multipacket(t *testing.T) { const maxWriteSize = mtu / maxWrites var rtx ringTx internalbuff := make([]byte, mtu) - rng := rand.New(rand.NewSource(1)) + rng := rand.New(rand.NewSource(3)) var wbuf, rbuf [mtu]byte for itest := 0; itest < 32; itest++ { + // rng.Seed(int64(itest)) + println(itest) err := rtx.Reset(internalbuff, maxPkts, iss) if err != nil { t.Fatal(err) @@ -67,7 +69,7 @@ func TestTxQueue_multipacket(t *testing.T) { for acked < roff { maxToack := min(roff-acked, maxWriteSize) toack := rng.Intn(maxToack) + 1 - t.Log("\n", rtx.string()) + // t.Log("\n", rtx.string()) err = rtx.RecvACK(iss + Value(acked+toack)) testQueueSanity(t, &rtx) if err != nil { @@ -347,7 +349,7 @@ func (rx *ringTx) string() string { wrapZone = &zones[i] } } - var currentZone *zone + var currentZone *zone = wrapZone var lastPrintedZone *zone var l1, l2 bytes.Buffer changes := 0