From 6ba06acdcb78f5efd437e8774f61662de556dead Mon Sep 17 00:00:00 2001 From: Pat Whittingslow Date: Tue, 9 Jun 2026 10:20:40 -0300 Subject: [PATCH] lneto rework: Require explicit BackoffStrategy in all APIs (#116) * make backoff explicit API * fix tests to use explicit backoff * fix examples with explicit tcp --- dhcp/dhcpv6/stringers.go | 161 +++++++++++++++++- errors.go | 3 +- examples/berkeley-listener/berkeley_server.go | 13 ++ examples/httpserver/main.go | 13 ++ examples/min-working-example/main-mwe.go | 13 ++ examples/xcurl/main.go | 12 ++ internal/backoff.go | 35 ---- internet/pcap/stringers.go | 5 +- internet/stackbasic_test.go | 9 + internet/tcplistener_test.go | 2 + ipv6/icmpv6/stringers.go | 2 - stringers.go | 34 ++-- tcp/conn.go | 9 +- tcp/conn_test.go | 8 + udp/conn.go | 9 +- udp/conn_test.go | 7 + udp/packetconn.go | 10 +- x/xnet/stack-blocking.go | 10 +- x/xnet/stack-go.go | 5 + x/xnet/stack-retrying.go | 3 + x/xnet/stack6_test.go | 2 + x/xnet/tcppool.go | 8 +- x/xnet/xnet_concurrent_test.go | 8 +- x/xnet/xnet_fuzz_test.go | 4 + x/xnet/xnet_listener_test.go | 7 + x/xnet/xnet_subnettable_test.go | 3 + x/xnet/xnet_test.go | 6 + x/xnet/xnet_udppktconn_test.go | 7 + 28 files changed, 324 insertions(+), 84 deletions(-) delete mode 100644 internal/backoff.go diff --git a/dhcp/dhcpv6/stringers.go b/dhcp/dhcpv6/stringers.go index 8e65a9d..c7c29bc 100644 --- a/dhcp/dhcpv6/stringers.go +++ b/dhcp/dhcpv6/stringers.go @@ -1,4 +1,161 @@ +// Code generated by "stringer -type=MsgType,ClientState,OptCode,StatusCode,DUIDType -linecomment -output stringers.go"; DO NOT EDIT. + package dhcpv6 -// Code generated by "stringer -type=MsgType,ClientState,OptCode,StatusCode,DUIDType -linecomment -output stringers.go"; DO NOT EDIT. -// Run go generate ./dhcp/dhcpv6/ to regenerate. +import "strconv" + +func _() { + // An "invalid array index" compiler error signifies that the constant values have changed. + // Re-run the stringer command to generate them again. + var x [1]struct{} + _ = x[MsgSolicit-1] + _ = x[MsgAdvertise-2] + _ = x[MsgRequest-3] + _ = x[MsgConfirm-4] + _ = x[MsgRenew-5] + _ = x[MsgRebind-6] + _ = x[MsgReply-7] + _ = x[MsgRelease-8] + _ = x[MsgDecline-9] + _ = x[MsgReconfigure-10] + _ = x[MsgInformRequest-11] + _ = x[MsgRelayForw-12] + _ = x[MsgRelayRepl-13] +} + +const _MsgType_name = "solicitadvertiserequestconfirmrenewrebindreplyreleasedeclinereconfigureinform-requestrelay-forwrelay-repl" + +var _MsgType_index = [...]uint8{0, 7, 16, 23, 30, 35, 41, 46, 53, 60, 71, 85, 95, 105} + +func (i MsgType) String() string { + i -= 1 + if i >= MsgType(len(_MsgType_index)-1) { + return "MsgType(" + strconv.FormatInt(int64(i+1), 10) + ")" + } + return _MsgType_name[_MsgType_index[i]:_MsgType_index[i+1]] +} +func _() { + // An "invalid array index" compiler error signifies that the constant values have changed. + // Re-run the stringer command to generate them again. + var x [1]struct{} + _ = x[StateInit-1] + _ = x[StateSoliciting-2] + _ = x[StateRequesting-3] + _ = x[StateBound-4] + _ = x[StateRenewing-5] + _ = x[StateRebinding-6] +} + +const _ClientState_name = "initsolicitingrequestingboundrenewingrebinding" + +var _ClientState_index = [...]uint8{0, 4, 14, 24, 29, 37, 46} + +func (i ClientState) String() string { + i -= 1 + if i >= ClientState(len(_ClientState_index)-1) { + return "ClientState(" + strconv.FormatInt(int64(i+1), 10) + ")" + } + return _ClientState_name[_ClientState_index[i]:_ClientState_index[i+1]] +} +func _() { + // An "invalid array index" compiler error signifies that the constant values have changed. + // Re-run the stringer command to generate them again. + var x [1]struct{} + _ = x[OptClientID-1] + _ = x[OptServerID-2] + _ = x[OptIANA-3] + _ = x[OptIATA-4] + _ = x[OptIAAddr-5] + _ = x[OptORO-6] + _ = x[OptPreference-7] + _ = x[OptElapsedTime-8] + _ = x[OptRelayMsg-9] + _ = x[OptAuth-11] + _ = x[OptUnicast-12] + _ = x[OptStatusCode-13] + _ = x[OptRapidCommit-14] + _ = x[OptUserClass-15] + _ = x[OptVendorClass-16] + _ = x[OptVendorOpts-17] + _ = x[OptInterfaceID-18] + _ = x[OptReconfMsg-19] + _ = x[OptReconfAccept-20] + _ = x[OptDNSServers-23] + _ = x[OptDomainList-24] + _ = x[OptIAPD-25] + _ = x[OptIAPrefix-26] + _ = x[OptNTPServer-56] +} + +const ( + _OptCode_name_0 = "client-idserver-idia-naia-taiaaddroropreferenceelapsed-timerelay-msg" + _OptCode_name_1 = "authunicaststatus-coderapid-commituser-classvendor-classvendor-optsinterface-idreconf-msgreconf-accept" + _OptCode_name_2 = "dns-serversdomain-listia-pdiaprefix" + _OptCode_name_3 = "ntp-server" +) + +var ( + _OptCode_index_0 = [...]uint8{0, 9, 18, 23, 28, 34, 37, 47, 59, 68} + _OptCode_index_1 = [...]uint8{0, 4, 11, 22, 34, 44, 56, 67, 79, 89, 102} + _OptCode_index_2 = [...]uint8{0, 11, 22, 27, 35} +) + +func (i OptCode) String() string { + switch { + case 1 <= i && i <= 9: + i -= 1 + return _OptCode_name_0[_OptCode_index_0[i]:_OptCode_index_0[i+1]] + case 11 <= i && i <= 20: + i -= 11 + return _OptCode_name_1[_OptCode_index_1[i]:_OptCode_index_1[i+1]] + case 23 <= i && i <= 26: + i -= 23 + return _OptCode_name_2[_OptCode_index_2[i]:_OptCode_index_2[i+1]] + case i == 56: + return _OptCode_name_3 + default: + return "OptCode(" + strconv.FormatInt(int64(i), 10) + ")" + } +} +func _() { + // An "invalid array index" compiler error signifies that the constant values have changed. + // Re-run the stringer command to generate them again. + var x [1]struct{} + _ = x[StatusSuccess-0] + _ = x[StatusUnspecFail-1] + _ = x[StatusNoAddrsAvail-2] + _ = x[StatusNoBinding-3] + _ = x[StatusNotOnLink-4] + _ = x[StatusUseMulticast-5] +} + +const _StatusCode_name = "successunspec-failno-addrs-availno-bindingnot-on-linkuse-multicast" + +var _StatusCode_index = [...]uint8{0, 7, 18, 32, 42, 53, 66} + +func (i StatusCode) String() string { + if i >= StatusCode(len(_StatusCode_index)-1) { + return "StatusCode(" + strconv.FormatInt(int64(i), 10) + ")" + } + return _StatusCode_name[_StatusCode_index[i]:_StatusCode_index[i+1]] +} +func _() { + // An "invalid array index" compiler error signifies that the constant values have changed. + // Re-run the stringer command to generate them again. + var x [1]struct{} + _ = x[DUIDTypeLLT-1] + _ = x[DUIDTypeEN-2] + _ = x[DUIDTypeLL-3] +} + +const _DUIDType_name = "duid-lltduid-enduid-ll" + +var _DUIDType_index = [...]uint8{0, 8, 15, 22} + +func (i DUIDType) String() string { + i -= 1 + if i >= DUIDType(len(_DUIDType_index)-1) { + return "DUIDType(" + strconv.FormatInt(int64(i+1), 10) + ")" + } + return _DUIDType_name[_DUIDType_index[i]:_DUIDType_index[i+1]] +} diff --git a/errors.go b/errors.go index 81e8d41..0637e9f 100644 --- a/errors.go +++ b/errors.go @@ -17,12 +17,13 @@ const ( ErrMismatch // mismatch ErrMismatchLen // mismatched length ErrInvalidConfig // invalid configuration + _ // little stitious ErrInvalidField // invalid field ErrInvalidLengthField // invalid length field ErrExhausted // resource exhausted ErrAlreadyRegistered // protocol already registered ErrTruncatedFrame // truncated frame - + ErrMissingHALConfig // missing HAL configuration // Below are potentially good future error additions // based on one or two encountered use cases, example use case included. /* diff --git a/examples/berkeley-listener/berkeley_server.go b/examples/berkeley-listener/berkeley_server.go index f68d00a..a723f5b 100644 --- a/examples/berkeley-listener/berkeley_server.go +++ b/examples/berkeley-listener/berkeley_server.go @@ -210,6 +210,7 @@ func run() error { RxBufSize: mtu, EstablishedTimeout: 5 * time.Second, ClosingTimeout: 5 * time.Second, + NewBackoff: func() lneto.BackoffStrategy { return tcpBackoff }, }, }) @@ -408,3 +409,15 @@ func stackBackoff(consecutiveBackoffs uint) time.Duration { } return 10 * time.Millisecond } + +func tcpBackoff(consecutiveBackoffs uint) time.Duration { + const ( + minWait = uint32(time.Microsecond) + maxWait = 5 * uint32(time.Millisecond) + maxShift = 22 + _overflowCheck = minWait << maxShift + ) + shifted := minWait << min(consecutiveBackoffs, maxShift) + wait := min(shifted, maxWait) + return time.Duration(wait) +} diff --git a/examples/httpserver/main.go b/examples/httpserver/main.go index d3527e9..83dd463 100644 --- a/examples/httpserver/main.go +++ b/examples/httpserver/main.go @@ -232,6 +232,7 @@ func run() (err error) { RxBuf: make([]byte, mtu), TxBuf: make([]byte, mtu), TxPacketQueueSize: 3, + RWBackoff: tcpBackoff, }) err = stack.ListenTCP4(&conn, svPort) if err != nil { @@ -363,3 +364,15 @@ func stackBackoff(consecutiveBackoffs uint) time.Duration { } return 10 * time.Millisecond } + +func tcpBackoff(consecutiveBackoffs uint) time.Duration { + const ( + minWait = uint32(time.Microsecond) + maxWait = 5 * uint32(time.Millisecond) + maxShift = 22 + _overflowCheck = minWait << maxShift + ) + shifted := minWait << min(consecutiveBackoffs, maxShift) + wait := min(shifted, maxWait) + return time.Duration(wait) +} diff --git a/examples/min-working-example/main-mwe.go b/examples/min-working-example/main-mwe.go index 098f28d..ea8fb19 100644 --- a/examples/min-working-example/main-mwe.go +++ b/examples/min-working-example/main-mwe.go @@ -102,6 +102,7 @@ func run(ctx context.Context, stack *xnet.StackAsync) error { NanoTime: nanotime, EstablishedTimeout: tcpEstablishedTimeout, ClosingTimeout: tcpCloseTimeout, + NewBackoff: func() lneto.BackoffStrategy { return tcpBackoff }, }, }) @@ -178,3 +179,15 @@ func stackBackoff(consecutiveBackoffs uint) time.Duration { } return 10 * time.Millisecond } + +func tcpBackoff(consecutiveBackoffs uint) time.Duration { + const ( + minWait = uint32(time.Microsecond) + maxWait = 5 * uint32(time.Millisecond) + maxShift = 22 + _overflowCheck = minWait << maxShift + ) + shifted := minWait << min(consecutiveBackoffs, maxShift) + wait := min(shifted, maxWait) + return time.Duration(wait) +} diff --git a/examples/xcurl/main.go b/examples/xcurl/main.go index b3f721c..5859003 100644 --- a/examples/xcurl/main.go +++ b/examples/xcurl/main.go @@ -281,6 +281,7 @@ func run() (err error) { RxBuf: make([]byte, mtu), TxBuf: make([]byte, mtu), TxPacketQueueSize: 3, + RWBackoff: tcpBackoff, }) timeHTTPCreate := timer("create HTTP GET request") @@ -393,3 +394,14 @@ func stackBackoff(consecutiveBackoffs uint) time.Duration { } return 10 * time.Millisecond } +func tcpBackoff(consecutiveBackoffs uint) time.Duration { + const ( + minWait = uint32(time.Microsecond) + maxWait = 5 * uint32(time.Millisecond) + maxShift = 22 + _overflowCheck = minWait << maxShift + ) + shifted := minWait << min(consecutiveBackoffs, maxShift) + wait := min(shifted, maxWait) + return time.Duration(wait) +} diff --git a/internal/backoff.go b/internal/backoff.go deleted file mode 100644 index 9af77e5..0000000 --- a/internal/backoff.go +++ /dev/null @@ -1,35 +0,0 @@ -package internal - -import ( - "time" -) - -// BackoffConnRW implements exponential backoff suitable for TCP connection -// read/write polling. It starts at 1us and caps at 5ms, doubling on each consecutive backoff. -func BackoffConnRW(consecutiveBackoffs uint) { - const ( - minWait = uint32(time.Microsecond) - maxWait = 5 * uint32(time.Millisecond) - maxShift = 22 - _overflowCheck = minWait << maxShift - ) - shifted := minWait << min(consecutiveBackoffs, maxShift) - wait := min(shifted, maxWait) - time.Sleep(time.Duration(wait)) -} - -// BackoffStackProto implements exponential backoff suitable for stack-level -// protocol processing polling. It starts at 1us and caps at 100ms, doubling on each consecutive backoff. -func BackoffStackProto(consecutiveBackoffs uint) { - const ( - minWait = uint32(time.Microsecond) - maxWait = 100 * uint32(time.Millisecond) - - // Statically calculated numbers below. - maxShift = 22 - _overflowCheck = minWait << maxShift - ) - shifted := minWait << min(consecutiveBackoffs, maxShift) - wait := min(shifted, maxWait) - time.Sleep(time.Duration(wait)) -} diff --git a/internet/pcap/stringers.go b/internet/pcap/stringers.go index d81f419..ef2dae0 100644 --- a/internet/pcap/stringers.go +++ b/internet/pcap/stringers.go @@ -25,11 +25,12 @@ func _() { _ = x[FieldClassBinaryText-14] _ = x[FieldClassOperation-15] _ = x[FieldClassTimestamp-16] + _ = x[FieldClassDNSName-17] } -const _FieldClass_name = "undefinedsourcedestinationprotocolversiontypesizeflagsidentificationchecksumoptionspayloadtextaddressbinary-textoptimestamp" +const _FieldClass_name = "undefinedsourcedestinationprotocolversiontypesizeflagsidentificationchecksumoptionspayloadtextaddressbinary-textoptimestampdns name" -var _FieldClass_index = [...]uint8{0, 9, 15, 26, 34, 41, 45, 49, 54, 68, 76, 83, 90, 94, 101, 112, 114, 123} +var _FieldClass_index = [...]uint8{0, 9, 15, 26, 34, 41, 45, 49, 54, 68, 76, 83, 90, 94, 101, 112, 114, 123, 131} func (i FieldClass) String() string { if i >= FieldClass(len(_FieldClass_index)-1) { diff --git a/internet/stackbasic_test.go b/internet/stackbasic_test.go index 67d529e..4958779 100644 --- a/internet/stackbasic_test.go +++ b/internet/stackbasic_test.go @@ -4,6 +4,7 @@ import ( "math/rand" "net/netip" "testing" + "time" "github.com/soypat/lneto" "github.com/soypat/lneto/tcp" @@ -146,6 +147,7 @@ func setupClientServer6(t *testing.T, rng *rand.Rand, client, server *StackIPv6, RxBuf: make([]byte, bufsize), TxBuf: make([]byte, bufsize), TxPacketQueueSize: 3, + RWBackoff: backoffYield, }) if err != nil { t.Fatal(err) @@ -154,6 +156,7 @@ func setupClientServer6(t *testing.T, rng *rand.Rand, client, server *StackIPv6, RxBuf: make([]byte, bufsize), TxBuf: make([]byte, bufsize), TxPacketQueueSize: 3, + RWBackoff: backoffYield, }) if err != nil { t.Fatal(err) @@ -187,6 +190,7 @@ func setupClientServer(t *testing.T, rng *rand.Rand, client, server *StackIPv4, TxBuf: make([]byte, bufsize), TxPacketQueueSize: 3, Logger: nil, + RWBackoff: backoffYield, }) if err != nil { t.Fatal(err) @@ -196,6 +200,7 @@ func setupClientServer(t *testing.T, rng *rand.Rand, client, server *StackIPv4, TxBuf: make([]byte, bufsize), TxPacketQueueSize: 3, Logger: nil, + RWBackoff: backoffYield, }) if err != nil { t.Fatal(err) @@ -219,3 +224,7 @@ func setupClientServer(t *testing.T, rng *rand.Rand, client, server *StackIPv4, t.Fatal(err) } } + +func backoffYield(consecutiveBackoffs uint) time.Duration { + return lneto.BackoffFlagGosched +} diff --git a/internet/tcplistener_test.go b/internet/tcplistener_test.go index 01bd63d..7031ae4 100644 --- a/internet/tcplistener_test.go +++ b/internet/tcplistener_test.go @@ -616,6 +616,7 @@ func setupClient(t *testing.T, client *StackIPv4, conn *tcp.Conn, serverAddr net RxBuf: make([]byte, bufsize), TxBuf: make([]byte, bufsize), TxPacketQueueSize: 3, + RWBackoff: backoffYield, }) if err != nil { t.Fatal(err) @@ -649,6 +650,7 @@ func newMockTCPPool(n, queuesize, bufsize int) *mockTCPPool { RxBuf: make([]byte, bufsize), TxBuf: make([]byte, bufsize), TxPacketQueueSize: queuesize, + RWBackoff: backoffYield, }) if err != nil { panic(err) diff --git a/ipv6/icmpv6/stringers.go b/ipv6/icmpv6/stringers.go index f331802..a184f29 100644 --- a/ipv6/icmpv6/stringers.go +++ b/ipv6/icmpv6/stringers.go @@ -48,7 +48,6 @@ func (i Type) String() string { return "Type(" + strconv.FormatInt(int64(i), 10) + ")" } } - func _() { // An "invalid array index" compiler error signifies that the constant values have changed. // Re-run the stringer command to generate them again. @@ -72,7 +71,6 @@ func (i CodeDestinationUnreachable) String() string { } return _CodeDestinationUnreachable_name[_CodeDestinationUnreachable_index[i]:_CodeDestinationUnreachable_index[i+1]] } - func _() { // An "invalid array index" compiler error signifies that the constant values have changed. // Re-run the stringer command to generate them again. diff --git a/stringers.go b/stringers.go index ba79f1e..58e5b74 100644 --- a/stringers.go +++ b/stringers.go @@ -206,21 +206,33 @@ func _() { _ = x[ErrMismatch-10] _ = x[ErrMismatchLen-11] _ = x[ErrInvalidConfig-12] - _ = x[ErrInvalidField-13] - _ = x[ErrInvalidLengthField-14] - _ = x[ErrExhausted-15] - _ = x[ErrAlreadyRegistered-16] - _ = x[ErrTruncatedFrame-17] + _ = x[ErrInvalidField-14] + _ = x[ErrInvalidLengthField-15] + _ = x[ErrExhausted-16] + _ = x[ErrAlreadyRegistered-17] + _ = x[ErrTruncatedFrame-18] + _ = x[ErrMissingHALConfig-19] } -const _errGeneric_name = "lneto-bug(use build tag \"debugheaplog\")packet droppedincorrect checksumzero source(port/addr)zero destination(port/addr)short bufferbuffer fullinvalid addressunsupportedmismatchmismatched lengthinvalid configurationinvalid fieldinvalid length fieldresource exhaustedprotocol already registeredtruncated frame" +const ( + _errGeneric_name_0 = "lneto-bug(use build tag \"debugheaplog\")packet droppedincorrect checksumzero source(port/addr)zero destination(port/addr)short bufferbuffer fullinvalid addressunsupportedmismatchmismatched lengthinvalid configuration" + _errGeneric_name_1 = "invalid fieldinvalid length fieldresource exhaustedprotocol already registeredtruncated framemissing HAL configuration" +) -var _errGeneric_index = [...]uint16{0, 39, 53, 71, 93, 120, 132, 143, 158, 169, 177, 194, 215, 228, 248, 266, 293, 308} +var ( + _errGeneric_index_0 = [...]uint8{0, 39, 53, 71, 93, 120, 132, 143, 158, 169, 177, 194, 215} + _errGeneric_index_1 = [...]uint8{0, 13, 33, 51, 78, 93, 118} +) func (i errGeneric) String() string { - i -= 1 - if i >= errGeneric(len(_errGeneric_index)-1) { - return "errGeneric(" + strconv.FormatInt(int64(i+1), 10) + ")" + switch { + case 1 <= i && i <= 12: + i -= 1 + return _errGeneric_name_0[_errGeneric_index_0[i]:_errGeneric_index_0[i+1]] + case 14 <= i && i <= 19: + i -= 14 + return _errGeneric_name_1[_errGeneric_index_1[i]:_errGeneric_index_1[i+1]] + default: + return "errGeneric(" + strconv.FormatInt(int64(i), 10) + ")" } - return _errGeneric_name[_errGeneric_index[i]:_errGeneric_index[i+1]] } diff --git a/tcp/conn.go b/tcp/conn.go index b51927d..c77ec11 100644 --- a/tcp/conn.go +++ b/tcp/conn.go @@ -62,6 +62,9 @@ type ConnConfig struct { } func (conn *Conn) Configure(config ConnConfig) (err error) { + if config.RWBackoff == nil { + return lneto.ErrMissingHALConfig + } conn.mu.Lock() defer conn.mu.Unlock() err = conn.h.SetBuffers(config.TxBuf, config.RxBuf, config.TxPacketQueueSize) @@ -483,9 +486,5 @@ func (conn *Conn) ConnectionID() *uint64 { } func (conn *Conn) backoff(consecutiveBackoffs uint) { - if conn._backoff != nil { - conn._backoff.Do(consecutiveBackoffs) - } else { - internal.BackoffConnRW(consecutiveBackoffs) - } + conn._backoff.Do(consecutiveBackoffs) } diff --git a/tcp/conn_test.go b/tcp/conn_test.go index 3620d04..b032998 100644 --- a/tcp/conn_test.go +++ b/tcp/conn_test.go @@ -5,6 +5,8 @@ import ( "net/netip" "testing" "time" + + "github.com/soypat/lneto" ) func newConfiguredConn(t *testing.T) *Conn { @@ -14,6 +16,7 @@ func newConfiguredConn(t *testing.T) *Conn { RxBuf: make([]byte, 512), TxBuf: make([]byte, 512), TxPacketQueueSize: 4, + RWBackoff: backoffYield, }) if err != nil { t.Fatal(err) @@ -121,6 +124,7 @@ func TestConn_Configure_Twice(t *testing.T) { RxBuf: make([]byte, 1024), TxBuf: make([]byte, 1024), TxPacketQueueSize: 8, + RWBackoff: backoffYield, }) if err != nil { t.Fatalf("reconfigure should succeed: %v", err) @@ -160,3 +164,7 @@ func TestConn_ImplementsNetConn(t *testing.T) { } = conn _ = net.ErrClosed } + +func backoffYield(consecutiveBackoffs uint) time.Duration { + return lneto.BackoffFlagGosched +} diff --git a/udp/conn.go b/udp/conn.go index a86908f..77f7368 100644 --- a/udp/conn.go +++ b/udp/conn.go @@ -48,6 +48,9 @@ type ConnConfig struct { // Configure initializes the connection with the given buffer and queue configuration. // Must be called before [Conn.Open]. Calling Configure on an active connection aborts it. func (conn *Conn) Configure(cfg ConnConfig) error { + if cfg.RWBackoff == nil { + return lneto.ErrMissingHALConfig + } conn.mu.Lock() defer conn.mu.Unlock() conn.abort() @@ -302,11 +305,7 @@ func (conn *Conn) FreeInput() int { } func (conn *Conn) backoff(consecutiveBackoffs uint) { - if conn._backoff != nil { - conn._backoff.Do(consecutiveBackoffs) - } else { - internal.BackoffConnRW(consecutiveBackoffs) - } + conn._backoff.Do(consecutiveBackoffs) } func (conn *Conn) lockPipeConnID() (uint64, error) { diff --git a/udp/conn_test.go b/udp/conn_test.go index b259852..916b7c0 100644 --- a/udp/conn_test.go +++ b/udp/conn_test.go @@ -4,7 +4,9 @@ import ( "encoding/binary" "net/netip" "testing" + "time" + "github.com/soypat/lneto" "github.com/soypat/lneto/internal" ) @@ -26,6 +28,7 @@ func newTestConn(t *testing.T) *Conn { TxBuf: make([]byte, 256), RxQueueSize: 4, TxQueueSize: 4, + RWBackoff: backoffYield, }) if err != nil { t.Fatal(err) @@ -239,3 +242,7 @@ func TestConn_FrameOffset(t *testing.T) { t.Fatalf("got %q, want %q", string(buf[:n]), "hi") } } + +func backoffYield(backoffs uint) time.Duration { + return lneto.BackoffFlagGosched +} diff --git a/udp/packetconn.go b/udp/packetconn.go index 6fe7d44..9a6ad1d 100644 --- a/udp/packetconn.go +++ b/udp/packetconn.go @@ -8,7 +8,6 @@ import ( "time" "github.com/soypat/lneto" - "github.com/soypat/lneto/internal" ) // lnetopacketconn is the lneto interpretation of @@ -53,6 +52,9 @@ type PacketConnConfig struct { // Configure initializes the PacketConn with the given buffer and queue configuration. // Must be called before [PacketConn.Open]. Calling Configure on an active connection aborts it. func (pc *PacketConn) Configure(cfg PacketConnConfig) error { + if cfg.RWBackoff == nil { + return lneto.ErrMissingHALConfig + } pc.mu.Lock() defer pc.mu.Unlock() pc.abort() @@ -243,11 +245,7 @@ func (pc *PacketConn) deadlineExceeded(deadline *time.Time) bool { } func (pc *PacketConn) backoff(n uint) { - if pc._backoff != nil { - pc._backoff.Do(n) - } else { - internal.BackoffConnRW(n) - } + pc._backoff.Do(n) } func (pc *PacketConn) lockConnID() (uint64, error) { diff --git a/x/xnet/stack-blocking.go b/x/xnet/stack-blocking.go index f57d169..ee54a9a 100644 --- a/x/xnet/stack-blocking.go +++ b/x/xnet/stack-blocking.go @@ -8,7 +8,6 @@ import ( "github.com/soypat/lneto" "github.com/soypat/lneto/dhcp/dhcpv4" - "github.com/soypat/lneto/internal" "github.com/soypat/lneto/tcp" ) @@ -21,6 +20,9 @@ var ( ) func (s *StackAsync) StackBlocking(stackProtoBackoff lneto.BackoffStrategy) StackBlocking { + if stackProtoBackoff == nil { + panic("nil backoff to StackBlocking") + } return StackBlocking{ async: s, _backoff: stackProtoBackoff, @@ -206,9 +208,5 @@ func (s StackBlocking) backoff(consecutiveBackoffs uint) { } func backoff(bo lneto.BackoffStrategy, consecutiveBackoffs uint) { - if bo != nil { - bo.Do(consecutiveBackoffs) - } else { - internal.BackoffStackProto(consecutiveBackoffs) - } + bo.Do(consecutiveBackoffs) } diff --git a/x/xnet/stack-go.go b/x/xnet/stack-go.go index 569fb89..3c43d1a 100644 --- a/x/xnet/stack-go.go +++ b/x/xnet/stack-go.go @@ -25,6 +25,9 @@ type StackGoConfig struct { } func (s *StackAsync) StackGo(stackProtoBackoff lneto.BackoffStrategy, cfg StackGoConfig) StackGo { + if stackProtoBackoff == nil || cfg.ListenerPoolConfig.NewBackoff == nil { + panic("nil backoff to StackGo") + } return s.StackBlocking(stackProtoBackoff).StackGo(cfg) } @@ -142,9 +145,11 @@ func (s StackGo) SocketNetip(ctx context.Context, network string, family, sotype var conn tcp.Conn // DIAL TCP: active connection a.k.a TCP Client branch. err = conn.Configure(tcp.ConnConfig{ + // TODO(pato): Eventually add UDP configuration. we use TCP for now for simplicity's sake. TxBuf: make([]byte, s.plcfg.TxBufSize), RxBuf: make([]byte, s.plcfg.RxBufSize), TxPacketQueueSize: s.plcfg.QueueSize, + RWBackoff: s.plcfg.NewBackoff(), }) if err != nil { return nil, err diff --git a/x/xnet/stack-retrying.go b/x/xnet/stack-retrying.go index 451841b..dc1bdaa 100644 --- a/x/xnet/stack-retrying.go +++ b/x/xnet/stack-retrying.go @@ -10,6 +10,9 @@ import ( ) func (s *StackAsync) StackRetrying(stackProtoBackoff lneto.BackoffStrategy) StackRetrying { + if stackProtoBackoff == nil { + panic("nil backoff to StackRetrying") + } return StackRetrying{ block: s.StackBlocking(stackProtoBackoff), } diff --git a/x/xnet/stack6_test.go b/x/xnet/stack6_test.go index 182a38d..69c0469 100644 --- a/x/xnet/stack6_test.go +++ b/x/xnet/stack6_test.go @@ -70,6 +70,7 @@ func newUDPConn6(t testing.TB) *udp.Conn { TxBuf: make([]byte, bufSize), RxQueueSize: 4, TxQueueSize: 4, + RWBackoff: backoffYield, }); err != nil { t.Fatal("UDP Configure:", err) } @@ -85,6 +86,7 @@ func newTCPConn6(t testing.TB) *tcp.Conn { RxBuf: make([]byte, bufSize), TxBuf: make([]byte, bufSize), TxPacketQueueSize: 4, + RWBackoff: backoffYield, }); err != nil { t.Fatal("TCP Configure:", err) } diff --git a/x/xnet/tcppool.go b/x/xnet/tcppool.go index e794254..0013a3d 100644 --- a/x/xnet/tcppool.go +++ b/x/xnet/tcppool.go @@ -53,7 +53,7 @@ type TCPPoolConfig struct { ClosingTimeout time.Duration // NewUserData is used to create user data used for each individual TCP connection and returned on GetTCP. NewUserData func() any - // NewBackoff returns the backoff to use for every newly configured TCP connection. + // NewBackoff returns the backoff to use for every newly configured TCP connection. Must be non-nil. // This should always return a static(non-method) function unless you know what you are doing. NewBackoff func() lneto.BackoffStrategy } @@ -61,6 +61,8 @@ type TCPPoolConfig struct { func NewTCPPool(cfg TCPPoolConfig) (*TCPPool, error) { if cfg.EstablishedTimeout <= 0 || cfg.ClosingTimeout <= 0 { return nil, lneto.ErrInvalidConfig + } else if cfg.NewBackoff == nil { + return nil, lneto.ErrMissingHALConfig } n := int(cfg.PoolSize) pool := &TCPPool{ @@ -84,9 +86,7 @@ func NewTCPPool(cfg TCPPoolConfig) (*TCPPool, error) { TxBuf: bufSpace[txOff : txOff+cfg.TxBufSize], TxPacketQueueSize: cfg.QueueSize, Logger: cfg.ConnLogger, - } - if cfg.NewBackoff != nil { - conncfg.RWBackoff = cfg.NewBackoff() + RWBackoff: cfg.NewBackoff(), } err := pool.conns[i].Configure(conncfg) if err != nil { diff --git a/x/xnet/xnet_concurrent_test.go b/x/xnet/xnet_concurrent_test.go index 630fea2..8169b72 100644 --- a/x/xnet/xnet_concurrent_test.go +++ b/x/xnet/xnet_concurrent_test.go @@ -53,6 +53,7 @@ func TestTCPListener_ConcurrentEcho(t *testing.T) { RxBufSize: 512, EstablishedTimeout: 5 * time.Second, ClosingTimeout: 5 * time.Second, + NewBackoff: func() lneto.BackoffStrategy { return backoffYield }, }) if err != nil { t.Fatal(err) @@ -96,6 +97,7 @@ func TestTCPListener_ConcurrentEcho(t *testing.T) { RxBuf: connBufs[bufOff : bufOff+tcpBufSize], TxBuf: connBufs[bufOff+tcpBufSize : bufOff+2*tcpBufSize], TxPacketQueueSize: 4, + RWBackoff: backoffYield, }) if err != nil { t.Fatalf("client %d conn configure: %v", i, err) @@ -327,7 +329,7 @@ func testCloseTransmitsPending(tst *tester, s1, s2 *StackAsync, c1, c2 *tcp.Conn RxBuf: nil, TxBuf: make([]byte, tx1Buf), TxPacketQueueSize: queueSize, - RWBackoff: backoffGosched, + RWBackoff: backoffYield, Logger: logger, }) if err != nil { @@ -431,7 +433,3 @@ func testCloseTransmitsPending(tst *tester, s1, s2 *StackAsync, c1, c2 *tcp.Conn } } - -func backoffGosched(consecutiveBackoffs uint) (sleep time.Duration) { - return lneto.BackoffFlagGosched -} diff --git a/x/xnet/xnet_fuzz_test.go b/x/xnet/xnet_fuzz_test.go index 7fb4a8b..43bce85 100644 --- a/x/xnet/xnet_fuzz_test.go +++ b/x/xnet/xnet_fuzz_test.go @@ -318,6 +318,7 @@ func testStackSeeded(t *testing.T, seed1, seed2 int64) { RxBuf: make([]byte, bufsize), TxBuf: make([]byte, bufsize), TxPacketQueueSize: 1 + int(uint16(seed1)%10), + RWBackoff: backoffYield, }) if err != nil { t.Fatal(err) @@ -326,6 +327,7 @@ func testStackSeeded(t *testing.T, seed1, seed2 int64) { RxBuf: make([]byte, bufsize), TxBuf: make([]byte, bufsize), TxPacketQueueSize: 1 + int(uint16(seed2)%10), + RWBackoff: backoffYield, }) if err != nil { t.Fatal(err) @@ -335,6 +337,7 @@ func testStackSeeded(t *testing.T, seed1, seed2 int64) { TxBuf: make([]byte, bufsize), RxQueueSize: int(1 + uint16(seed1>>32)%10), TxQueueSize: int(1 + uint16(seed1>>32)%10), + RWBackoff: backoffYield, }) if err != nil { t.Fatal(err) @@ -344,6 +347,7 @@ func testStackSeeded(t *testing.T, seed1, seed2 int64) { TxBuf: make([]byte, bufsize), RxQueueSize: int(1 + uint16(seed2>>32)%10), TxQueueSize: int(1 + uint16(seed2>>32)%10), + RWBackoff: backoffYield, }) if err != nil { t.Fatal(err) diff --git a/x/xnet/xnet_listener_test.go b/x/xnet/xnet_listener_test.go index 3f7b783..894bfbd 100644 --- a/x/xnet/xnet_listener_test.go +++ b/x/xnet/xnet_listener_test.go @@ -5,6 +5,7 @@ import ( "testing" "time" + "github.com/soypat/lneto" "github.com/soypat/lneto/ethernet" "github.com/soypat/lneto/tcp" ) @@ -49,6 +50,7 @@ func TestStackAsyncListener_SingleConnection(t *testing.T) { RxBuf: make([]byte, MTU), TxBuf: make([]byte, MTU), TxPacketQueueSize: 4, + RWBackoff: backoffYield, }) if err != nil { t.Fatal(err) @@ -62,6 +64,7 @@ func TestStackAsyncListener_SingleConnection(t *testing.T) { RxBufSize: MTU, EstablishedTimeout: 10e9, ClosingTimeout: 10e9, + NewBackoff: func() lneto.BackoffStrategy { return backoffYield }, }) if err != nil { t.Fatal(err) @@ -150,6 +153,7 @@ func TestStackAsyncListener_MultiSequentialConn(t *testing.T) { RxBufSize: bufsize, EstablishedTimeout: 10e9, ClosingTimeout: 10e9, + NewBackoff: func() lneto.BackoffStrategy { return backoffYield }, }) if err != nil { t.Fatal(err) @@ -188,6 +192,7 @@ func TestStackAsyncListener_MultiSequentialConn(t *testing.T) { RxBuf: make([]byte, bufsize), TxBuf: make([]byte, bufsize), TxPacketQueueSize: 4, + RWBackoff: backoffYield, }) if err != nil { t.Fatal(err) @@ -239,6 +244,7 @@ func TestListener_Close(t *testing.T) { RxBufSize: 512, EstablishedTimeout: 10e9, ClosingTimeout: 10e9, + NewBackoff: func() lneto.BackoffStrategy { return backoffYield }, }) if err != nil { t.Fatal(err) @@ -278,6 +284,7 @@ func TestListener_ResetAfterClose(t *testing.T) { RxBufSize: 512, EstablishedTimeout: 10e9, ClosingTimeout: 10e9, + NewBackoff: func() lneto.BackoffStrategy { return backoffYield }, }) if err != nil { t.Fatal(err) diff --git a/x/xnet/xnet_subnettable_test.go b/x/xnet/xnet_subnettable_test.go index 2c486ca..f65e931 100644 --- a/x/xnet/xnet_subnettable_test.go +++ b/x/xnet/xnet_subnettable_test.go @@ -5,6 +5,7 @@ import ( "net/netip" "testing" + "github.com/soypat/lneto" "github.com/soypat/lneto/ethernet" "github.com/soypat/lneto/ipv4" "github.com/soypat/lneto/tcp" @@ -73,6 +74,7 @@ func TestStackAsync_ListenerSynAckAddressedToClient(t *testing.T) { RxBufSize: mtu, EstablishedTimeout: 10e9, ClosingTimeout: 10e9, + NewBackoff: func() lneto.BackoffStrategy { return backoffYield }, }) if err != nil { t.Fatal(err) @@ -104,6 +106,7 @@ func TestStackAsync_ListenerSynAckAddressedToClient(t *testing.T) { if err = clConn.Configure(tcp.ConnConfig{ RxBuf: make([]byte, mtu), TxBuf: make([]byte, mtu), TxPacketQueueSize: 4, + RWBackoff: backoffYield, }); err != nil { t.Fatal(err) } diff --git a/x/xnet/xnet_test.go b/x/xnet/xnet_test.go index be3e2a2..8d38e2e 100644 --- a/x/xnet/xnet_test.go +++ b/x/xnet/xnet_test.go @@ -208,6 +208,7 @@ func newTCPStacks(t testing.TB, randSeed int64, mtu int) (s1, s2 *StackAsync, c1 RxBuf: buf[:mtu], TxBuf: buf[mtu : mtu*2], TxPacketQueueSize: 4, + RWBackoff: backoffYield, }) if err != nil { t.Fatal(err) @@ -216,6 +217,7 @@ func newTCPStacks(t testing.TB, randSeed int64, mtu int) (s1, s2 *StackAsync, c1 RxBuf: buf[2*mtu : 3*mtu], TxBuf: buf[3*mtu : 4*mtu], TxPacketQueueSize: 4, + RWBackoff: backoffYield, }) if err != nil { t.Fatal(err) @@ -1045,3 +1047,7 @@ func getTCPFrame(etherFrame []byte) (tcp.Frame, bool) { } return tfrm, true } + +func backoffYield(consecutiveBackoffs uint) time.Duration { + return lneto.BackoffFlagGosched +} diff --git a/x/xnet/xnet_udppktconn_test.go b/x/xnet/xnet_udppktconn_test.go index 979aec6..5871857 100644 --- a/x/xnet/xnet_udppktconn_test.go +++ b/x/xnet/xnet_udppktconn_test.go @@ -60,6 +60,7 @@ func TestStackAsyncRegisterListenerUDP_ReceiveData(t *testing.T) { if err := pc.Configure(udp.PacketConnConfig{ RxBuf: make([]byte, testUDPBufSize), TxBuf: make([]byte, testUDPBufSize), RxQueueSize: testUDPQueueSize, TxQueueSize: testUDPQueueSize, + RWBackoff: backoffYield, }); err != nil { t.Fatal("pc Configure:", err) } @@ -74,6 +75,7 @@ func TestStackAsyncRegisterListenerUDP_ReceiveData(t *testing.T) { if err := conn.Configure(udp.ConnConfig{ RxBuf: make([]byte, testUDPBufSize), TxBuf: make([]byte, testUDPBufSize), RxQueueSize: testUDPQueueSize, TxQueueSize: testUDPQueueSize, + RWBackoff: backoffYield, }); err != nil { t.Fatal("conn Configure:", err) } @@ -119,6 +121,7 @@ func TestStackAsyncRegisterListenerUDP_ReplyToClient(t *testing.T) { if err := pc.Configure(udp.PacketConnConfig{ RxBuf: make([]byte, testUDPBufSize), TxBuf: make([]byte, testUDPBufSize), RxQueueSize: testUDPQueueSize, TxQueueSize: testUDPQueueSize, + RWBackoff: backoffYield, }); err != nil { t.Fatal("pc Configure:", err) } @@ -133,6 +136,7 @@ func TestStackAsyncRegisterListenerUDP_ReplyToClient(t *testing.T) { if err := conn.Configure(udp.ConnConfig{ RxBuf: make([]byte, testUDPBufSize), TxBuf: make([]byte, testUDPBufSize), RxQueueSize: testUDPQueueSize, TxQueueSize: testUDPQueueSize, + RWBackoff: backoffYield, }); err != nil { t.Fatal("conn Configure:", err) } @@ -228,6 +232,7 @@ func TestStackAsyncRegisterListenerUDP_MultiSource(t *testing.T) { if err := pc.Configure(udp.PacketConnConfig{ RxBuf: make([]byte, testUDPBufSize), TxBuf: make([]byte, testUDPBufSize), RxQueueSize: testUDPQueueSize, TxQueueSize: testUDPQueueSize, + RWBackoff: backoffYield, }); err != nil { t.Fatal("pc Configure:", err) } @@ -243,6 +248,7 @@ func TestStackAsyncRegisterListenerUDP_MultiSource(t *testing.T) { if err := conn1.Configure(udp.ConnConfig{ RxBuf: make([]byte, testUDPBufSize), TxBuf: make([]byte, testUDPBufSize), RxQueueSize: testUDPQueueSize, TxQueueSize: testUDPQueueSize, + RWBackoff: backoffYield, }); err != nil { t.Fatal("conn1 Configure:", err) } @@ -260,6 +266,7 @@ func TestStackAsyncRegisterListenerUDP_MultiSource(t *testing.T) { if err := conn2.Configure(udp.ConnConfig{ RxBuf: make([]byte, testUDPBufSize), TxBuf: make([]byte, testUDPBufSize), RxQueueSize: testUDPQueueSize, TxQueueSize: testUDPQueueSize, + RWBackoff: backoffYield, }); err != nil { t.Fatal("conn2 Configure:", err) }