lneto/tcp/txqueue.go

package tcp

import (
	"errors"
	"fmt"

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

const (
	// this must be at least 2 for buffer to work.
	minBufferSize = 2
)

// ringTx is a ring buffer with retransmission queue functionality added.
//
//	|   acked(free)  |          sent         |          unsent          |             free       |
//	0       freeEnd=first.off       last.end==unsent.off        freeStart=unsent.end         Size()
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
	// 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.
	unsentend int
	// sentoff is the offset of start of sent data in rawbuf.
	sentoff int
	// sentend is the offset of end of sent data in rawbuf. If zero then sent buffer is empty.
	sentend int
	// seq     Value
	// always empty ring.
	emptyRing ringidx
}

// ringidx represents packet data inside RingTx
type ringidx struct {
	// off is data start offset of packet data inside buf. Follows [internal.Ring] semantics.
	off int
	// end is the ringed data end offset, non-inclusive. Follows [internal.Ring] semantics.
	end int
	// seq is the sequence number of the first byte in the packet.
	seq Value
	// size is the size of the packet in bytes.
	size Size
	// time is a measure of the instant of time message was sent at.
}

// Reset resets the RingTx's internal state to use buf as the main ring buffer and creates or reuses
// the packet ring buffer.
func (rtx *ringTx) Reset(buf []byte, maxqueuedPackets int, seq Value) error {
	buf = buf[:len(buf):len(buf)] // safely omit capacity section.
	if maxqueuedPackets <= 0 {
		return errors.New("queued packets <=0")
	} 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()
	}
	return nil
}

// ResetOrReuse is identical to a call to [ringTx.Reset] with the additional detail that
// the zero value of buf (nil) and maxQueuedPackets (0) will selectively reuse existing data buffer and/or packet index buffer.
func (rtx *ringTx) ResetOrReuse(buf []byte, maxQueuedPackets int, ack Value) error {
	if buf == nil {
		buf = rtx.rawbuf
	}
	if maxQueuedPackets == 0 {
		maxQueuedPackets = len(rtx.packets)
	}
	return rtx.Reset(buf, maxQueuedPackets, ack)
}

// Size returns the total storage space of the transmission buffer.
func (rtx *ringTx) Size() int { return len(rtx.rawbuf) }

// Free returns the total available space for Write calls.
func (rtx *ringTx) Free() int {
	r := rtx.sentAndUnsentBuffer()
	return r.Free()
}

// Buffered returns the amount of written but unsent bytes.
func (rtx *ringTx) Buffered() int {
	r, _ := rtx.unsentRing()
	return r.Buffered()
}

// BufferedSent returns the total amount of bytes sent but not acked.
func (rtx *ringTx) BufferedSent() int {
	r, _ := rtx.sentRing()
	return r.Buffered()
}

// Write writes data to the underlying unsent data ring buffer.
func (rtx *ringTx) Write(b []byte) (n int, err error) {
	r, lim := rtx.unsentRing()
	n, err = r.WriteLimited(b, lim)
	if err != nil {
		return 0, err
	}
	rtx.unsentend = rtx.addEnd(rtx.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 (rtx *ringTx) MakePacket(b []byte, currentSeq Value) (int, error) {
	nxtpkt := rtx.nextPkt()
	if nxtpkt < 0 {
		return 0, errors.New("queue full")
	}
	endSeq, ok := rtx.endSeq()
	if ok && currentSeq.LessThan(endSeq) {
		return 0, errors.New("sequence number less than last sequence number")
	}
	r, _ := rtx.unsentRing()
	start := 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.
	}
	*pkt = ringidx{
		off:  start,
		end:  off,
		seq:  currentSeq,
		size: Size(n),
	}
	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)
		// return errors.New("old packet")
	}
	// lastAckedPkt stores last fully acked packet.
	var lastAckedPkt *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)
		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")
		}
		if isPartialAcked {
			if lastAckedPkt != nil && lastAckedPkt.seq.LessThan(pkt.seq) {
				panic("unreachable")
			}
			acked := int(ack - pkt.seq)
			pring := rtx.ring(pkt.off, pkt.end)
			buffered := pring.Buffered()
			if acked > buffered || acked < minBufferSize {
				panic("unreachable")
			}
			off := rtx.addOff(pkt.off, acked)
			pkt.off = off
			pkt.seq = ack
			pkt.size = pkt.size - Size(acked)
			rtx.sentoff = off
		}
	}
	if lastAckedPkt != nil {
		rtx.sentoff = lastAckedPkt.end
		lastAckedPkt.markRcvd()
		if rtx.sentoff == rtx.sentend {
			// All data acked.
			rtx.sentend = 0
			rtx.consolidateBufs()
		}
	}
	return nil
}

func (rtx *ringTx) sentAndUnsentBuffer() internal.Ring {
	end := rtx.unsentend
	if end == 0 {
		end = rtx.sentend
	}
	return internal.Ring{Buf: rtx.rawbuf, Off: rtx.sentoff, End: end}
}

func (rtx *ringTx) unsentRing() (internal.Ring, int) {
	return rtx.ring(rtx.unsentoff, rtx.unsentend), rtx.sentoff
}

func (rtx *ringTx) sentRing() (internal.Ring, int) {
	return rtx.ring(rtx.sentoff, rtx.sentend), rtx.unsentoff // unsentoff should match with sentend, so no writes can be performed to sentring.
}

func (rtx *ringTx) ring(off, end int) internal.Ring {
	return internal.Ring{Buf: rtx.rawbuf, Off: off, End: end}
}

// addEnd adds two integers together and wraps the value around the ring's buffer size.
// Result of addEnd will never be 0 unless arguments are (0,0).
func (rtx *ringTx) addEnd(a, b int) int {
	result := a + b
	if result > len(rtx.rawbuf) {
		result -= len(rtx.rawbuf)
	}
	return result
}

func (rtx *ringTx) addOff(a, b int) int {
	result := a + b
	if result >= len(rtx.rawbuf) {
		result -= len(rtx.rawbuf)
	}
	return result
}

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
	if unsentEmpty && sentEmpty {
		// reset start of buffers.
		rtx.sentoff = 0
		rtx.unsentoff = 0
	}
}

func (rtx *ringTx) endSeq() (Value, bool) {
	pkt := rtx.lastPkt()
	if pkt < 0 {
		return 0, false
	}
	last := rtx.pkt(pkt)
	return Add(last.seq, last.size), 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
}

// lims returns the limits of free|sent|unsent buffers.
// Example:
//
//	|   acked(free)  |          sent         |          unsent          |             free       |
//	0       freeEnd=first.off       last.end==unsent.off        freeStart=unsent.end         Size()
func (tx *ringTx) lims() (unsentStart, unsentEnd, sentStart, sentEnd int) {
	return tx.unsentoff, tx.unsentend, tx.sentoff, tx.sentend
}

func (pkt *ringidx) sent() bool {
	return pkt.end != 0 || pkt.off != 0
}

func (pkt *ringidx) markRcvd() {
	*pkt = ringidx{}
	// pkt.end = 0
	// pkt.off = 0
}