package tcp import ( "errors" "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 packet. seq Value // 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 { 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], seq: seq, } 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) (int, Value, error) { nxtpkt := rtx.nextPkt() if rtx.nextPkt() < 0 { return 0, 0, errors.New("queue full") } r, _ := rtx.unsentRing() start := r.Off n, err := r.Read(b) if err != nil { return n, 0, 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.off = start pkt.end = off // Sequence number updates. oldseq := rtx.seq newseq := Add(oldseq, Size(n)) rtx.seq = newseq pkt.seq = newseq return n, oldseq, nil } // RecvSegment processes an incoming segment and updates the sent packet queue func (rtx *ringTx) RecvACK(ack Value) error { if ack.LessThan(rtx.seq) { return errors.New("old packet") } first := rtx.firstPkt() if first < 0 { return errors.New("no packets to ack") } hiSeq := rtx.pkt(first).seq for i := 0; i < len(rtx.packets); i++ { pkt := &rtx.packets[i] if pkt.sent() && pkt.seq.LessThanEq(ack) { if hiSeq.LessThanEq(pkt.seq) { rtx.sentoff = pkt.end hiSeq = pkt.seq } pkt.markRcvd() } } firstAcked := !rtx.pkt(first).sent() if firstAcked && 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 || seq.LessThan(pkt.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 || pkt.seq.LessThan(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) currentSeq() Value { return rtx.seq } // 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 }