soypat/lneto
1
0
Fork 0
mirror of https://github.com/soypat/lneto.git synced 2026-08-26 17:39:29 +00:00
Code Issues Packages Projects Releases 1 Wiki Activity

done for today :)

Browse Source
This commit is contained in:
soypat
2025-01-26 19:02:16 -03:00
parent 94c59ee84e
commit 6a30268f83
5 changed files with 209 additions and 96 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tcp/txqueue.go
+78 -51
View File
@@ -17,9 +17,9 @@ type ringTx struct {
rawbuf []byte
// packets contains
packets []ringidx
// unsentOff is the offset of start of unsent data into rawbuf.
// unsentOff is the offset of start of unsent data in rawbuf.
unsentoff int
// unsentend is the offset of end of unsent data in rawbuf.
// unsentend is the offset of end of unsent data in rawbuf. If zero then unsent buffer is empty.
unsentend int
seq Value
// always empty ring.
@@ -71,32 +71,36 @@ func (rx *ringTx) ResetOrReuse(buf []byte, maxQueuedPackets int, ack Value) erro
return rx.Reset(buf, maxQueuedPackets, ack)
}
// Size returns the total storage space of the transmission buffer.
func (tx *ringTx) Size() int { return len(tx.rawbuf) }
// Free returns the total available space for Write calls.
func (tx *ringTx) Free() int {
freeStart, freeEnd, _ := tx.lims()
r := tx.ring(freeEnd, freeStart)
return r.Free()
}
// Buffered returns the amount of unsent bytes.
func (tx *ringTx) Buffered() int {
r := tx.unsentRing()
r, _ := tx.unsentRing()
return r.Buffered()
}
// BufferedSent returns the total amount of bytes sent but not acked.
func (tx *ringTx) BufferedSent() int {
r := tx.sentRing()
r, _ := tx.sentRing()
return r.Buffered()
}
// Write writes data to the underlying unsent data ring buffer.
func (tx *ringTx) Write(b []byte) (n int, err error) {
first := tx.pkt(tx.firstPkt())
r := tx.unsentRing()
if !first.sent() {
// No packets in queue case.
n, err = r.Write(b)
} else {
n, err = r.WriteLimited(b, first.off)
}
r, lim := tx.unsentRing()
n, err = r.WriteLimited(b, lim)
if err != nil {
return 0, err
}
tx.unsentend = tx.addOff(tx.unsentend, n)
tx.unsentend = tx.addEnd(tx.unsentend, n)
return n, err
}
@@ -107,7 +111,7 @@ func (tx *ringTx) MakePacket(b []byte) (int, Value, error) {
if tx.nextPkt() < 0 {
return 0, 0, errors.New("queue full")
}
r := tx.unsentRing()
r, _ := tx.unsentRing()
start := r.Off
n, err := r.Read(b)
if err != nil {
@@ -115,12 +119,14 @@ func (tx *ringTx) MakePacket(b []byte) (int, Value, error) {
}
plen := Value(n)
seq := tx.seq
tx.unsentoff = tx.addOff(tx.unsentoff, n)
tx.unsentoff = tx.addEnd(tx.unsentoff, n)
if tx.unsentoff == tx.unsentend {
tx.unsentend = 0 // Mark unsent as being empty.
}
tx.seq += plen
pkt := &tx.packets[nxtpkt]
pkt.off = start
pkt.end = tx.addOff(start, n)
pkt.end = tx.addEnd(start, n)
pkt.seq = seq + plen
return n, seq, nil
}
@@ -133,37 +139,43 @@ func (tx *ringTx) RecvACK(ack Value) error {
pkt.markRcvd()
}
}
return nil
}
func (tx *ringTx) unsentRing() internal.Ring {
off := tx.unsentoff
if off == tx.unsentend && off != 0 {
off--
func (tx *ringTx) unsentRing() (internal.Ring, int) {
freeStart, freeEnd, sentEnd := tx.lims()
if tx.unsentend == 0 {
freeStart = 0 // Unsent is empty.
}
return tx.ring(off, tx.unsentend)
return tx.ring(sentEnd, freeStart), freeEnd
}
func (tx *ringTx) sentRing() internal.Ring {
first := tx.pkt(tx.firstPkt())
if !first.sent() {
return internal.Ring{}
}
last := tx.pkt(tx.lastPkt())
return tx.ring(first.off, last.end)
func (tx *ringTx) sentRing() (internal.Ring, int) {
freeStart, freeEnd, sentEnd := tx.lims()
return tx.ring(freeEnd, sentEnd), freeStart
}
func (tx *ringTx) ring(off, end int) internal.Ring {
return internal.Ring{Buf: tx.rawbuf, Off: off, End: end}
}
// addOff adds two integers together and wraps the value around the ring's buffer size.
func (tx *ringTx) addOff(a, b int) int {
off := a + b
if off >= len(tx.rawbuf) {
off -= len(tx.rawbuf)
// 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 (tx *ringTx) addEnd(a, b int) int {
result := a + b
if result > len(tx.rawbuf) {
result -= len(tx.rawbuf)
}
return off
return result
}
func (tx *ringTx) addOff(a, b int) int {
result := a + b
if result >= len(tx.rawbuf) {
result -= len(tx.rawbuf)
}
return result
}
func (tx *ringTx) pkt(i int) *ringidx {
@@ -180,14 +192,9 @@ func (tx *ringTx) firstPkt() int {
idx := -1
for i := 0; i < len(tx.packets); i++ {
pkt := &tx.packets[i]
if pkt.sent() {
if idx == -1 {
seq = pkt.seq
}
if seq.LessThan(pkt.seq) {
seq = pkt.seq
idx = i
}
if pkt.sent() && (idx == -1 || seq.LessThan(pkt.seq)) {
seq = pkt.seq
idx = i
}
}
return idx
@@ -198,14 +205,9 @@ func (tx *ringTx) lastPkt() int {
idx := -1
for i := 0; i < len(tx.packets); i++ {
pkt := &tx.packets[i]
if pkt.sent() {
if idx == -1 {
seq = pkt.seq
}
if pkt.seq.LessThan(seq) {
seq = pkt.seq
idx = i
}
if pkt.sent() && (idx == -1 || pkt.seq.LessThan(seq)) {
seq = pkt.seq
idx = i
}
}
return idx
@@ -223,6 +225,31 @@ func (tx *ringTx) nextPkt() int {
return idx
}
// 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() (freeStart, freeEnd, sentEndorUnsentStart int) {
freeStart = tx.unsentend
if freeStart == 0 {
freeStart = tx.unsentoff
}
first := tx.pkt(tx.firstPkt())
if first.sent() {
freeEnd = first.off
sentEndorUnsentStart = tx.unsentoff
} else if tx.unsentend != 0 {
// sent section empty and unsent not empty.
freeEnd = tx.unsentoff
sentEndorUnsentStart = tx.unsentoff
} else {
freeEnd = tx.unsentoff
sentEndorUnsentStart = tx.unsentoff
}
return freeStart, freeEnd, sentEndorUnsentStart
}
func (pkt *ringidx) sent() bool {
return pkt.end != 0 || pkt.off != 0
}
tcp/txqueue_test.go
+67 -15
View File
@@ -11,21 +11,37 @@ func TestTxQueue(t *testing.T) {
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()))
increasingComplexityTests := []struct {
name string
test func(*testing.T)
}{
0: {
name: "SequentialMessages",
test: 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, 0)
}
},
},
1: {
name: "N-Messages",
test: 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), 0)
}
},
},
}
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)
}
})
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) {
@@ -53,10 +69,12 @@ func testTxQueue_NMessages(t *testing.T, rtx *ringTx, msgs [][]byte, buf, aux []
} else if n != len(msg) {
t.Fatalf("want %d written, got %d", len(msg), n)
}
testQueueSanity(t, rtx)
unsent := rtx.Buffered()
if unsent != n {
t.Fatalf("want unset %d, got %d", n, unsent)
}
testQueueSanity(t, rtx)
n, seq, err := rtx.MakePacket(aux[sent : sent+len(msg)])
if err != nil {
t.Fatal(err)
@@ -65,10 +83,12 @@ func testTxQueue_NMessages(t *testing.T, rtx *ringTx, msgs [][]byte, buf, aux []
} else if n != len(msg) {
t.Fatalf("want full message %d sent, got %d", len(msg), n)
}
testQueueSanity(t, rtx)
gotSent := rtx.BufferedSent()
if gotSent != sent+n {
t.Fatalf("want sent %d, got %d", sent+n, gotSent)
}
testQueueSanity(t, rtx)
packets = append(packets, aux[sent:sent+n])
prevSeq += Value(n)
sent += n
@@ -91,14 +111,17 @@ func testTxQueue_SequentialMessages(t *testing.T, rtx *ringTx, msgs [][]byte, bu
} else if n != len(msg) {
t.Fatalf("want %d written, got %d", len(msg), n)
}
testQueueSanity(t, rtx)
unsent := rtx.Buffered()
if len(msg) != unsent {
t.Fatalf("want %d unsent buffered, got %d", unsent, len(msg))
t.Fatalf("want %d unsent buffered, got %d", len(msg), unsent)
}
testQueueSanity(t, rtx)
sent := rtx.BufferedSent()
if sent != 0 {
t.Fatalf("want 0 bytes sent, got %d", sent)
}
testQueueSanity(t, rtx)
n, seq, err := rtx.MakePacket(aux[:])
data := aux[:n]
if err != nil {
@@ -110,14 +133,43 @@ func testTxQueue_SequentialMessages(t *testing.T, rtx *ringTx, msgs [][]byte, bu
} else if seq != prevSeq {
t.Fatalf("want seq %d, got %d", prevSeq, seq)
}
testQueueSanity(t, rtx)
sent = rtx.BufferedSent()
if sent != len(msg) {
t.Fatalf("want %d sent, got %d", len(msg), sent)
}
testQueueSanity(t, rtx)
prevSeq += Value(n)
err = rtx.RecvACK(prevSeq)
if err != nil {
t.Fatal(err)
}
testQueueSanity(t, rtx)
}
}
func testQueueSanity(t *testing.T, rtx *ringTx) {
t.Helper()
if rtx.emptyRing != (ringidx{}) {
t.Fatalf("empty ring not empty")
}
free := rtx.Free()
sent := rtx.BufferedSent()
unsent := rtx.Buffered()
sz := rtx.Size()
gotSz := free + sent + unsent
if gotSz != sz {
t.Fatalf("want size=%d, got size=%d (free+sent+unsent=%d+%d+%d)", sz, gotSz, free, sent, unsent)
}
freeStart, freeEnd, sentEnd := rtx.lims()
gotFreeEnd := rtx.addOff(freeStart, free)
gotSentEnd := rtx.addOff(freeEnd, sent)
gotUnsentEnd := rtx.addOff(sentEnd, unsent)
if free != 0 && gotFreeEnd != freeEnd {
t.Fatalf("want freeEnd=%d, got %d", freeEnd, gotFreeEnd)
} else if sent != 0 && gotSentEnd != sentEnd {
t.Fatalf("want sentEnd=%d, got %d", sentEnd, gotSentEnd)
} else if unsent != 0 && gotUnsentEnd != freeStart {
t.Fatalf("want unsentEnd=%d, got %d (freeStart)", freeStart, gotUnsentEnd)
}
}