7 Commits

Author SHA1 Message Date
Patricio Whittingslow 0860a6fe2e fix CI 2026-08-24 16:55:10 -03:00
Patricio Whittingslow 52a3926428 implement tcp.Policy and refactor rto to use it 2026-08-24 16:47:06 -03:00
Patricio Whittingslow 936790a5d0 begin prepping policy refactor manually 2026-08-24 15:59:17 -03:00
Yoshio HANAWA 263b1ecf11 fix(dns): decode up to four answer records (#186)
* fix(dns): decode up to four answer records

A single DNS question can return multiple answer records. Use an answer
decode limit of four in Client.StartResolve and add a regression test
covering a single-question response with multiple A records.

* fix(dns): add MaxResponseAnswers to ResolveConfig

This allows callers to explicitly declare the maximum number of answer
records to retain, removing the hardcoded limit in Client.StartResolve.

xnet.StackAsync is updated to set this limit to match the length of its
lookup-result buffer. This maintains the zero-allocation design while
fixing the issue where multiple A records were ignored.
2026-08-18 11:55:33 -07:00
Marvin 马维 Drees ab91d08f41 fix: elimite netdev test failure (#175)
Signed-off-by: Marvin Drees <marvin.drees@9elements.com>
2026-08-03 22:15:13 -07:00
Pat Whittingslow d05cd14018 httphi router refactor (#176)
* httphi: RouterConfig refactor to enable DefaultRouterConfig

* RequestHeader not case sensitive anymore

* httphi: use DefaultRouterConfig in examples

* httphi: remove gated stage complexity

Misusing Stage methods by calling them once header has been written is totally harmless as far as I can tell. We simplify the codebase on this occasion by removing the headerWritten check for all stage methods

* httphi: improve APIs

* httphi: remove status
2026-08-03 15:33:43 -07:00
Joel Wetzell 4517010070 use net.UDPAddrFromPort now available in tinygo (#95) 2026-07-31 15:14:10 -03:00
32 changed files with 1824 additions and 919 deletions
+8 -1
View File
@@ -24,6 +24,9 @@ type ResolveConfig struct {
Questions []Question
Additional []Resource
EnableRecursion bool
// MaxResponseAnswers limits how many answer records are decoded from the
// DNS response. If zero it defaults to the number of Questions.
MaxResponseAnswers uint16
}
func (sudp *Client) Protocol() uint64 { return uint64(lneto.IPProtoUDP) }
@@ -37,8 +40,12 @@ func (c *Client) StartResolve(localPort, txid uint16, cfg ResolveConfig) error {
if nd > math.MaxUint16 {
return lneto.ErrInvalidConfig
}
maxAns := cfg.MaxResponseAnswers
if maxAns == 0 {
maxAns = uint16(nd)
}
c.reset(localPort, txid, CQueryPending, cfg.EnableRecursion)
c.msg.LimitResourceDecoding(uint16(nd), uint16(nd), 0, 0)
c.msg.LimitResourceDecoding(uint16(nd), maxAns, 0, 0)
c.msg.AddQuestions(cfg.Questions)
c.msg.AddAdditionals(cfg.Additional)
return nil
+88 -64
View File
@@ -243,76 +243,100 @@ func TestClient_ReceivesDNSResponse(t *testing.T) {
const hostname = "example.com"
const txid = uint16(12345)
const clientPort = uint16(54321)
wantIP := [4]byte{93, 184, 216, 34}
// Build a DNS response message.
name := MustNewName(hostname)
responseMsg := Message{
Questions: []Question{{
Name: name,
Type: TypeA,
Class: ClassINET,
}},
Answers: []Resource{
NewResource(name, TypeA, ClassINET, 300, wantIP[:]),
},
const maxAnswers = 4
allIPs := [5][4]byte{
{192, 0, 2, 1},
{192, 0, 2, 2},
{192, 0, 2, 3},
{192, 0, 2, 4},
{192, 0, 2, 5},
}
// Response flags: QR=1 (response), RD=1, RA=1.
responseFlags := HeaderFlags(1<<15 | 1<<8 | 1<<7)
var buf [512]byte
dnsPayload, err := responseMsg.AppendTo(buf[:0], txid, responseFlags)
if err != nil {
t.Fatal("failed to build DNS response:", err)
tests := []struct {
name string
responseIPs [][4]byte
wantAnswers int
}{
{name: "single_answer", responseIPs: allIPs[:1], wantAnswers: 1},
{name: "multiple_answers", responseIPs: allIPs[:4], wantAnswers: 4},
{name: "answer_limit", responseIPs: allIPs[:5], wantAnswers: maxAnswers},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
name := MustNewName(hostname)
responseMsg := Message{
Questions: []Question{{
Name: name,
Type: TypeA,
Class: ClassINET,
}},
Answers: make([]Resource, len(tt.responseIPs)),
}
for i := range tt.responseIPs {
responseMsg.Answers[i] = NewResource(name, TypeA, ClassINET, 300, tt.responseIPs[i][:])
}
// Set up the DNS client.
var client Client
client.StartResolve(clientPort, txid, ResolveConfig{
Questions: []Question{{
Name: MustNewName(hostname),
Type: TypeA,
Class: ClassINET,
}},
EnableRecursion: true,
})
// Response flags: QR=1 (response), RD=1, RA=1.
responseFlags := HeaderFlags(1<<15 | 1<<8 | 1<<7)
var responseBuf [512]byte
dnsPayload, err := responseMsg.AppendTo(responseBuf[:0], txid, responseFlags)
if err != nil {
t.Fatal("failed to build DNS response:", err)
}
// Simulate sending by calling Encapsulate (changes state to AwaitResponse).
var dummy [512]byte
client.Encapsulate(dummy[:], 0, 0)
var client Client
err = client.StartResolve(clientPort, txid, ResolveConfig{
Questions: []Question{{
Name: name,
Type: TypeA,
Class: ClassINET,
}},
EnableRecursion: true,
MaxResponseAnswers: maxAnswers,
})
if err != nil {
t.Fatal("failed to start DNS resolve:", err)
}
// Call Demux with DNS payload.
err = client.Demux(dnsPayload, 0)
if err != nil {
t.Fatal("Client Demux error:", err)
}
// Encapsulate the query to move the client into the outstanding state.
var queryBuf [512]byte
_, err = client.Encapsulate(queryBuf[:], 0, 0)
if err != nil {
t.Fatal("failed to encapsulate DNS query:", err)
}
if err := client.Demux(dnsPayload, 0); err != nil {
t.Fatal("failed to demux DNS response:", err)
}
// Check the client received the answer.
var addrs [4]netip.Addr
answers, err := client.ResponseAnswerLookup(addrs[:], hostname)
if answers != 1 {
t.Fatalf("expected 1 answer, got %d", answers)
}
addr := addrs[0]
if !addr.Is4() {
t.Fatalf("expected 4 bytes in answer, got %d", addr.BitLen()/8)
}
if addr.As4() != wantIP {
t.Errorf("expected IP %v, got %v", wantIP, addr.String())
}
var addrs [maxAnswers]netip.Addr
answers, err := client.ResponseAnswerLookup(addrs[:], hostname)
if err != nil {
t.Fatal("failed to look up DNS response answers:", err)
}
if answers != uint16(tt.wantAnswers) {
t.Fatalf("expected %d answers, got %d", tt.wantAnswers, answers)
}
for i := 0; i < tt.wantAnswers; i++ {
addr := addrs[i]
if !addr.Is4() {
t.Errorf("answer %d: expected IPv4 address, got %v", i, addr)
continue
}
if addr.As4() != tt.responseIPs[i] {
t.Errorf("answer %d: expected IP %v, got %v", i, tt.responseIPs[i], addr)
}
}
// Test MessageCopyTo as well.
var lookup Message
lookup.LimitResourceDecoding(1, 1, 0, 0)
done, err := client.ResponseCopyTo(&lookup)
if err != nil {
t.Fatal("MessageCopyTo error:", err)
}
if !done {
t.Fatal("expected done=true")
}
if len(lookup.Answers) != 1 {
t.Fatalf("MessageCopyTo: expected 1 answer, got %d", len(lookup.Answers))
var lookup Message
done, err := client.ResponseCopyTo(&lookup)
if err != nil {
t.Fatal("failed to copy DNS response:", err)
}
if !done {
t.Fatal("expected done=true")
}
if len(lookup.Answers) != tt.wantAnswers {
t.Fatalf("expected %d copied answers, got %d", tt.wantAnswers, len(lookup.Answers))
}
})
}
}
+7 -17
View File
@@ -18,14 +18,10 @@ import (
)
const (
kB = 1 << 10
listenPort = 8080
bufferSizes = 2 * kB
// A browser sends around twenty header fields; a request carrying more
// than this is answered 431 rather than parsed into memory it was not
// given. Each field costs 8 bytes of table.
numHeaderFields = 32
readTimeout = 2 * time.Second
kB = 1 << 10
listenPort = 8080
memoryPerConn = 4 * kB
readTimeout = 2 * time.Second
)
// Credentials the endpoints check. They are in the source on purpose: this is a
@@ -84,14 +80,8 @@ func run() error {
server.Handle("/echo", server.echo) // No method: any method matches.
var router httphi.Router
err = router.Configure(httphi.RouterConfig{
FixedNumGoroutines: *flagThreads,
RequestHeaderBufferSize: bufferSizes,
RequestNumHeaderKVCap: numHeaderFields,
ResponseHeaderMinBufferSize: bufferSizes,
Mux: &server.mux,
Logger: slog.Default(),
})
cfg := httphi.DefaultRouterConfig(*flagThreads, memoryPerConn, server.mux.MaxPathValues())
err = router.Configure(&server.mux, cfg)
if err != nil {
return err
}
@@ -435,7 +425,7 @@ func (sv *Server) upload(exch *httphi.Exchange) {
func (sv *Server) echo(exch *httphi.Exchange) {
s := sv.acquireScratch()
defer sv.releaseScratch(s)
body := append(s.out[:0], exch.RequestMethodRaw()...)
body := append(s.out[:0], exch.RequestMethodBytes()...)
body = append(body, ' ')
body = append(body, exch.RequestTarget()...)
body = append(body, '\n')
+5 -15
View File
@@ -38,12 +38,7 @@ var indexhtml string
// Router memory. The router allocates all of it on Configure and never again,
// so these are the whole cost of serving HTTP over the stack.
const (
// A browser sends around 700 bytes of header on a landing page request.
requestHeaderBuffer = 1024
// Response headers reuse whatever the request left unused on top of this,
// and the status line does not count towards it.
responseHeaderBuffer = 256
numHeaderFields = 16
httpConnMemoryUse = 4 * 1024
// One exchange is allocated per worker, and a worker holds its exchange for
// the whole request, so this is what bounds requests served at once.
numWorkers = 2
@@ -252,14 +247,9 @@ func run() (err error) {
server.handle("GET /stats", server.stats)
var router httphi.Router
err = router.Configure(httphi.RouterConfig{
FixedNumGoroutines: numWorkers,
RequestHeaderBufferSize: requestHeaderBuffer,
ResponseHeaderMinBufferSize: responseHeaderBuffer,
RequestNumHeaderKVCap: numHeaderFields,
Mux: &server.mux,
Logger: slog.Default(),
})
cfg := httphi.DefaultRouterConfig(numWorkers, httpConnMemoryUse, server.mux.MaxPathValues())
cfg.Logger = slog.Default()
err = router.Configure(&server.mux, cfg)
if err != nil {
return fmt.Errorf("configuring HTTP router: %w", err)
}
@@ -322,7 +312,7 @@ type httpServer struct {
func (sv *httpServer) handle(pattern string, handler httphi.HandlerFunc) {
sv.mux.Handle(pattern, func(exch *httphi.Exchange) {
sv.served.Add(1)
fmt.Printf("< %s %s\n", exch.RequestMethodRaw(), exch.RequestTarget())
fmt.Printf("< %s %s\n", exch.RequestMethodBytes(), exch.RequestTarget())
handler(exch)
})
}
+7 -17
View File
@@ -14,15 +14,11 @@ import (
)
const (
kB = 1 << 10
listenPort = 8080
bufferSizes = 2 * kB
// A browser sends around twenty header fields; a request carrying more
// than this is answered 431 rather than parsed into memory it was not
// given. Each field costs 8 bytes of table.
numHeaderFields = 32
numGoroutines = 4
readTimeout = 2 * time.Second
kB = 1 << 10
listenPort = 8080
connMemoryUse = 4 * kB
numGoroutines = 4
readTimeout = 2 * time.Second
)
func main() {
@@ -45,14 +41,8 @@ func run() error {
server.Handle("GET /", server.homepage)
var router httphi.Router
err = router.Configure(httphi.RouterConfig{
FixedNumGoroutines: numGoroutines,
RequestHeaderBufferSize: bufferSizes,
RequestNumHeaderKVCap: numHeaderFields,
ResponseHeaderMinBufferSize: bufferSizes,
Mux: &server.mux,
Logger: slog.Default(),
})
cfg := httphi.DefaultRouterConfig(numGoroutines, connMemoryUse, server.mux.MaxPathValues())
err = router.Configure(&server.mux, cfg)
if err != nil {
return err
}
+2 -7
View File
@@ -22,13 +22,8 @@ mux.Handle("GET /", func(ex *httphi.Exchange) {
})
var router httphi.Router
err := router.Configure(httphi.RouterConfig{
FixedNumGoroutines: 4, // 4 workers, 4 exchanges, allocated here and never again.
RequestHeaderBufferSize: 1024,
ResponseHeaderMinBufferSize: 32, // Shares the request buffer.
RequestNumHeaderKVCap: 32,
Mux: &mux,
})
cfg := httphi.DefaultRouterConfig(numWorkers, memoryPerConn, mux.MaxPathValues())
err := router.Configure(&mux, cfg)
if err != nil {
log.Fatal(err)
}
+4 -13
View File
@@ -4,7 +4,6 @@ import (
"fmt"
"io"
"log"
"log/slog"
"net"
"os"
@@ -15,23 +14,15 @@ import (
// ExampleRouter_linux goes over how to setup a linux server using raw linux connections.
// See [ExampleMuxSlice_query_forms_multipart] on how to define handlers for common HTTP processing.
func ExampleRouter() {
// Chrome tends to send ~700 bytes on a typical landing page request.
const requestBuffer = 1024
const numHeaderKV = requestBuffer / 32 //
const numWorkers = 8
const memoryPerConn = 2048
var mux httphi.MuxSlice
mux.Handle("GET /", func(ex *httphi.Exchange) {
ex.WriteBody([]byte("hello world"))
})
var router httphi.Router
err := router.Configure(httphi.RouterConfig{
FixedNumGoroutines: -1, // Unbounded goroutines and allocations.
RequestHeaderBufferSize: requestBuffer,
ResponseHeaderMinBufferSize: 32, // Shared buffer with Request, not strictly necessary, especially if not sending headers.
RequestNumHeaderKVCap: numHeaderKV,
NormalizeOutgoingKeys: true,
Mux: &mux,
Logger: slog.Default(),
})
cfg := httphi.DefaultRouterConfig(numWorkers, memoryPerConn, mux.MaxPathValues())
err := router.Configure(&mux, cfg)
if err != nil {
log.Fatal(err)
}
+17 -30
View File
@@ -75,10 +75,8 @@ type ExchangeConfig struct {
// Optional [any] cap holding the request header to RequestBufferLim rather than
// growing it. A header outgrowing it is answered 431, see [httpraw.ErrBufferExhausted].
NoRequestBufferGrowth bool
// Conditional [>=the most wildcards any one registered pattern binds] number of
// path values bindable, read back with [Exchange.PathValue]. A pattern binding
// more never matches, see [SetPathValues]. Zero suits a mux of literal patterns.
MaxPathValues int
// Conditional [len >=[Mux.MaxPathValues]] written to during [Mux.LookupHandler] in [Handle].
PathValuesBuf []PathValue
}
// HijackRaw is a low-level implementation of http.Hijacker interface.
@@ -124,8 +122,7 @@ func (exch *Exchange) Configure(cfg ExchangeConfig) {
exch.reqHdr.Reset(cfg.RawBuf[:0:cfg.RequestBufferLim], cfg.NumHeaderKVCap)
exch.reqHdr.ConfigBufferGrowth(!cfg.NoRequestBufferGrowth)
exch.normalizeKeys = cfg.NormalizeOutgoingKeys
internal.SliceReuse(&exch.pathValues, cfg.MaxPathValues)
exch.pathValues = exch.pathValues[:cfg.MaxPathValues]
exch.pathValues = cfg.PathValuesBuf
}
// Acquire claims the exchange for conn and resets it to serve a new request,
@@ -170,27 +167,20 @@ func (exch *Exchange) Release() {
// Does not return the buffer used for the response first line so can be safely
// written to and used without modifying the staged response first line.
//
// Staging headers will write to this buffer so use mindfully.
// To access only the request header buffer portion use [httpraw.HeaderV1.BufferRaw] limited
// to [httpraw.HeaderV1.BufferParsed] as returned by [Exchange.requestHeaderRaw].
// Writing to this section will not change the contents read by [Exchange.ReadBody].
// Writing to this aforementioned section will not change the contents read by [Exchange.ReadBody].
//
// In [Router] context, the size of this buffer is influenced directly by [RouterConfig] HeaderBufferSize fields.
func (exch *Exchange) UnsafeRawBuffer() []byte { return exch.rawbuf }
// RequestHeaderV1Raw returns the parsed request header for access beyond the
// Request* methods, such as [httpraw.HeaderV1.ForEach]. Valid until the exchange
// is released, and writing to it corrupts the response.
// RequestHeaderV1Raw returns the internal [Exchange] data structure used for HTTP/1.x requests.
func (exch *Exchange) RequestHeaderV1Raw() *httpraw.HeaderV1 { return &exch.reqHdr }
// StageHeader stages a response header field, written on the first
// [Exchange.FlushHeader], [Exchange.WriteHeader] or [Exchange.WriteBody].
// Returns false and drops the field if the response buffer cannot fit it.
// Has no effect once the header has been written.
func (exch *Exchange) StageHeader(key, value string) (enoughMemory bool) {
if exch.headerWritten {
return false
}
off := int(exch.respHeaderOff) + int(exch.respHeaderLen)
free := len(exch.rawbuf) - off
// Field costs key+':'+value+CRLF, plus the CRLF [Exchange.FlushHeader]
@@ -230,7 +220,7 @@ func (exch *Exchange) StageHeaderInt(key string, value int64) (enoughMemory bool
// base must be in the range 10..36; lower bases are dropped, no HTTP header
// field value is written below base 10.
func (exch *Exchange) StageHeaderIntBase(key string, value int64, base int) (enoughMemory bool) {
if exch.headerWritten || base < 10 || base > 36 {
if base < 10 || base > 36 {
return false
}
off := int(exch.respHeaderOff) + int(exch.respHeaderLen)
@@ -255,9 +245,9 @@ func (exch *Exchange) StageHeaderIntBase(key string, value int64, base int) (eno
// StageStatus prepares the status line for the given code without writing
// it, i.e: "HTTP/1.1 404 Not Found". Codes with no [StatusText] get an empty
// reason phrase. Has no effect once the header has been written.
// reason phrase.
func (exch *Exchange) StageStatus(code int) {
if code >= 1000 || exch.headerWritten {
if code >= 1000 {
return
} else if code == 200 {
// Common case.
@@ -278,11 +268,8 @@ func (exch *Exchange) StageStatus(code int) {
// WriteHeader sends the status line for code along with the staged header
// fields. Only the first call reaches the wire, as in http.ResponseWriter.
func (exch *Exchange) WriteHeader(code int) (n int, err error) {
if !exch.headerWritten {
exch.StageStatus(code)
n, err = exch.FlushHeader()
}
return n, err
exch.StageStatus(code)
return exch.FlushHeader()
}
// Respond writes a complete response in one call: Content-Type, a Content-Length
@@ -497,7 +484,7 @@ func (exch *Exchange) RequestParseCookie(dst *httpraw.Cookie, key string) error
func (exch *Exchange) RequestContentType() []byte {
// Folded: field names are case insensitive and HTTP/2 mandates lowercase, so
// a proxy translating h2 to h1 sends "content-type", RFC 9110 5.1.
return exch.RequestHeaderV1Raw().GetFold("Content-Type")
return exch.RequestHeader("Content-Type")
}
// RequestContentLength returns the body length declared by the request's
@@ -719,10 +706,10 @@ func (exch *Exchange) ReadMultiparts(dst []MultipartSink, buf []byte, newSink fu
}
// RequestHeader returns the value of the first request header field matching
// key, or nil if absent. Key matching is case sensitive.
// key, or nil if absent. Matching is not case sensitive.
func (exch *Exchange) RequestHeader(key string) []byte {
header := exch.RequestHeaderV1Raw()
return header.Get(key)
return header.GetFold(key)
}
// RequestTarget returns the request-target (URI) of the request line, i.e:
@@ -738,7 +725,7 @@ func (exch *Exchange) RequestPath() []byte {
}
// RequestQuery returns the request's query string as it appears on the wire.
// Iterate it with [httpraw.NextQueryPair]. See [httpraw.HeaderV1.RequestQuery].
// Iterate it with [httpraw.NextQueryPair].
func (exch *Exchange) RequestQuery() []byte {
return exch.RequestHeaderV1Raw().RequestQuery()
}
@@ -828,11 +815,11 @@ func (exch *Exchange) PathValueAppend(dst []byte, key string, decoded bool) ([]b
// RequestMethod returns the request's [Method] enum.
func (exch *Exchange) RequestMethod() Method {
return MethodFromBytes(exch.RequestMethodRaw())
return MethodFromBytes(exch.RequestMethodBytes())
}
// RequestMethod returns the request line's method as a []byte view, i.e: "GET".
func (exch *Exchange) RequestMethodRaw() []byte {
// RequestMethodBytes returns the request line's method as a []byte view, i.e: "GET".
func (exch *Exchange) RequestMethodBytes() []byte {
return exch.RequestHeaderV1Raw().Method()
}
+1 -1
View File
@@ -218,7 +218,7 @@ func TestHandleRequestFields(t *testing.T) {
var sm MuxSlice
route, _, _ := strings.Cut(test.wantURI, "?") // Mux matches on path.
sm.Handle(route, func(ex *Exchange) {
gotMethod = string(ex.RequestMethodRaw())
gotMethod = string(ex.RequestMethodBytes())
gotURI = string(ex.RequestTarget())
gotHost = string(ex.RequestHeader("Host"))
ex.WriteHeader(200)
+5 -4
View File
@@ -252,7 +252,7 @@ func (sm *MuxSlice) Reset(capacity int) {
// Every method this package does not name is [MethUnknown], so a request with an
// extension method matches a bare-path registration and any registration naming
// an extension method, whichever it names. Tell PROPFIND from MKCOL inside the
// handler with [Exchange.RequestMethodRaw].
// handler with [Exchange.RequestMethodBytes].
func (sm *MuxSlice) LookupHandler(method Method, path []byte, dstPathVals []PathValue) (matched string, _ HandlerFunc) {
best := -1
bestSpec := 0
@@ -423,9 +423,12 @@ func hasLowerASCII(s string) bool {
}
// Method is a HTTP request method, parsed by [MethodFrom].
// Method can only take standardized values and is set to [MethUnknown] for non-standard methods.
type Method uint8
const (
// MethUndefined returned by [MethodFrom] on an empty/missing method.
// Used by [MuxSlice] to denote an unset method kind for a request pattern.
MethUndefined Method = iota // undefined
MethGet // GET
// lol.
@@ -438,6 +441,7 @@ const (
MethConnect // CONNECT
MethOptions // OPTIONS
MethTrace // TRACE
// MethUnknown returned by [MethodFrom] on an non-standard method kind i.e: "get" and "FROBNICATE".
MethUnknown // unknown
)
@@ -475,9 +479,6 @@ func MethodFrom(meth string) (res Method) {
// MethodFromBytes is a [MethodFrom] wrapper with bytes argument instead of string.
func MethodFromBytes(meth []byte) (res Method) {
if len(meth) == 0 {
return MethUndefined
}
return MethodFrom(b2s(meth))
}
+4 -4
View File
@@ -183,7 +183,7 @@ func TestExchangePathValueClearedBetweenRequests(t *testing.T) {
exch := new(Exchange)
exch.Configure(ExchangeConfig{
RawBuf: make([]byte, 2048), RequestBufferLim: 1024,
NumHeaderKVCap: defaultNumHeaderKVCap, MaxPathValues: 4,
NumHeaderKVCap: defaultNumHeaderKVCap, PathValuesBuf: make([]PathValue, 4),
})
// First request binds id=42 off a wildcard pattern.
@@ -243,7 +243,7 @@ func TestMuxSliceNoStaleBindingsAcrossCandidates(t *testing.T) {
exch := new(Exchange)
exch.Configure(ExchangeConfig{
RawBuf: make([]byte, 2048), RequestBufferLim: 1024,
NumHeaderKVCap: defaultNumHeaderKVCap, MaxPathValues: sm.MaxPathValues(),
NumHeaderKVCap: defaultNumHeaderKVCap, PathValuesBuf: make([]PathValue, sm.MaxPathValues()),
})
conn := newConn("GET /a/1/b HTTP/1.1\r\nHost: h\r\n\r\n")
conn.Hangup()
@@ -292,7 +292,7 @@ func TestMuxSliceTrailingSlashPattern(t *testing.T) {
exch := new(Exchange)
exch.Configure(ExchangeConfig{
RawBuf: make([]byte, 2048), RequestBufferLim: 1024,
NumHeaderKVCap: defaultNumHeaderKVCap, MaxPathValues: sm.MaxPathValues(),
NumHeaderKVCap: defaultNumHeaderKVCap, PathValuesBuf: make([]PathValue, sm.MaxPathValues()),
})
conn := newConn("GET " + test.path + " HTTP/1.1\r\nHost: h\r\n\r\n")
conn.Hangup()
@@ -497,7 +497,7 @@ func TestMuxSliceZeroValueWildcardStillMatches(t *testing.T) {
exch := new(Exchange)
exch.Configure(ExchangeConfig{
RawBuf: make([]byte, 2048), RequestBufferLim: 1024,
NumHeaderKVCap: defaultNumHeaderKVCap, MaxPathValues: sm.MaxPathValues(),
NumHeaderKVCap: defaultNumHeaderKVCap, PathValuesBuf: make([]PathValue, sm.MaxPathValues()),
})
conn := newConn("GET " + test.path + " HTTP/1.1\r\nHost: h\r\n\r\n")
conn.Hangup()
+10 -27
View File
@@ -58,6 +58,7 @@ type Router struct {
mux Mux
globbuf []byte
globpath []PathValue
exchs []Exchange
freeList *Exchange
@@ -87,37 +88,19 @@ type RouterConfig struct {
// Required [>0] request header key/value pairs to parse before failing with
// [StatusRequestHeaderFieldsTooLarge].
RequestNumHeaderKVCap int
// Optional [any] normalization of response header field keys as they are
// staged, i.e: "content-type" becomes "Content-Type".
NormalizeOutgoingKeys bool
// Required [non-nil] resolver of each request's method and path to the handler
// serving it. Routes must be registered before Configure, see [Mux.MaxPathValues].
Mux Mux
// Optional [nil disables] sink for failed exchanges.
Logger *slog.Logger
}
const (
// minRequestHeaderBuffer is the smallest request buffer [httpraw.Header]
// accepts with buffer growth disabled, which is how exchanges are configured.
minRequestHeaderBuffer = 32
// minResponseHeaderBuffer is the room [Exchange.FlushHeader] needs for the
// CRLF closing the header block, written even when no field was staged.
minResponseHeaderBuffer = len("\r\n")
// maxExchangeBuffer bounds an exchange's whole buffer: [Exchange] indexes it
// with uint16 offsets, so a larger one would be addressed truncated.
maxExchangeBuffer = math.MaxUint16
)
// Validate returns a non-nil error if the configuration cannot be used to
// configure a [Router].
func (cfg RouterConfig) Validate() error {
workerMode := cfg.workerMode()
switch {
case cfg.Mux == nil,
!workerMode && cfg.FixedNumGoroutines != -1,
case !workerMode && cfg.FixedNumGoroutines != -1,
cfg.RequestNumHeaderKVCap <= 0,
cfg.RequestHeaderBufferSize < minRequestHeaderBuffer,
cfg.ResponseHeaderMinBufferSize < minResponseHeaderBuffer,
@@ -167,7 +150,7 @@ func (r *Router) shutdownLocked() {
// Configure may be called on a serving router, but since the exchange buffers
// are reused it waits for connections in flight to finish and fails with a
// non-nil error rather than reconfigure buffers still being served from.
func (r *Router) Configure(cfg RouterConfig) error {
func (r *Router) Configure(mux Mux, cfg RouterConfig) error {
if err := cfg.Validate(); err != nil {
return err
}
@@ -181,9 +164,9 @@ func (r *Router) Configure(cfg RouterConfig) error {
r.reqNumHeaderCap = cfg.RequestNumHeaderKVCap
r.reqBuf = cfg.RequestHeaderBufferSize
r.respBuf = cfg.ResponseHeaderMinBufferSize
r.mux = cfg.Mux
r.mux = mux
r.log = cfg.Logger
maxPathValues := cfg.Mux.MaxPathValues()
maxPathValues := mux.MaxPathValues()
if maxPathValues < 0 {
return errors.New("Mux paths must be registered before configuring Router")
}
@@ -211,20 +194,20 @@ func (r *Router) Configure(cfg RouterConfig) error {
r.exchs = r.exchs[:numgoro]
rawBuflen := cfg.RequestHeaderBufferSize + cfg.ResponseHeaderMinBufferSize
internal.SliceReuse(&r.globbuf, numgoro*rawBuflen)
internal.SliceReuse(&r.globpath, numgoro*maxPathValues)
for i := range numgoro {
// TODO exchange buffer alloc
goff := i * rawBuflen
// r.globbuf[goff:goff+rawBuflen], cfg.RequestHeaderBufferSize, cfg.RequestNumHeaderCap, cfg.NormalizeOutgoingKeys
poff := i * maxPathValues
r.exchs[i].Configure(ExchangeConfig{
RawBuf: r.globbuf[goff : goff+rawBuflen],
RequestBufferLim: cfg.RequestHeaderBufferSize,
NumHeaderKVCap: cfg.RequestNumHeaderKVCap,
NormalizeOutgoingKeys: cfg.NormalizeOutgoingKeys,
NoRequestBufferGrowth: true, // Hard memory limit.
MaxPathValues: maxPathValues,
PathValuesBuf: r.globpath[poff : poff+maxPathValues],
})
go r.goroWorker(gen, jobqueue, cfg.Mux)
go r.goroWorker(gen, jobqueue, mux)
}
r.pendingConns = jobqueue
r.numGoro = numgoro
@@ -388,7 +371,7 @@ func (r *Router) getExchLocked(conn conn) (exch *Exchange) {
NumHeaderKVCap: r.reqNumHeaderCap,
NormalizeOutgoingKeys: r.normalizeKeys,
NoRequestBufferGrowth: true,
MaxPathValues: r.maxPathValues,
PathValuesBuf: make([]PathValue, r.maxPathValues),
})
exch.Acquire(conn) // Fresh exchange, CAS cannot fail.
return exch
+6 -10
View File
@@ -150,9 +150,8 @@ func (r *rwconn) ViewWritten() string {
var _ Mux = (*MuxSlice)(nil)
func configSynchronousRouter(t *testing.T, router *Router, bufferSize int, mux Mux) {
err := router.Configure(RouterConfig{
err := router.Configure(mux, RouterConfig{
FixedNumGoroutines: -1,
Mux: mux,
RequestHeaderBufferSize: bufferSize,
RequestNumHeaderKVCap: 16,
ResponseHeaderMinBufferSize: bufferSize,
@@ -198,7 +197,7 @@ func TestRouterRequestVisibleToHandler(t *testing.T) {
var gotMethod, gotURI, gotHost string
var gotMethodEnum Method
sm.Handle("GET /index.html", func(ex *Exchange) {
gotMethod = string(ex.RequestMethodRaw())
gotMethod = string(ex.RequestMethodBytes())
gotMethodEnum = ex.RequestMethod()
gotURI = string(ex.RequestTarget())
gotHost = string(ex.RequestHeader("Host"))
@@ -388,9 +387,8 @@ func TestRouterHandleAfterTeardown(t *testing.T) {
router Router
)
sm.Handle("GET /", staticPage(t, "ok"))
err := router.Configure(RouterConfig{
err := router.Configure(&sm, RouterConfig{
FixedNumGoroutines: 2,
Mux: &sm,
RequestHeaderBufferSize: 512,
RequestNumHeaderKVCap: 16,
ResponseHeaderMinBufferSize: 512,
@@ -422,7 +420,6 @@ func TestRouterTeardownReleasesQueuedConns(t *testing.T) {
sm.Handle("GET /", staticPage(t, "ok"))
cfg := RouterConfig{
FixedNumGoroutines: numGoro,
Mux: &sm,
RequestHeaderBufferSize: 512,
RequestNumHeaderKVCap: 16,
ResponseHeaderMinBufferSize: 512,
@@ -436,7 +433,7 @@ func TestRouterTeardownReleasesQueuedConns(t *testing.T) {
// generation drops its connections, but it must not outlive it.
var err error
for range 100 {
if err = router.Configure(cfg); err == nil {
if err = router.Configure(&sm, cfg); err == nil {
break
}
time.Sleep(time.Millisecond)
@@ -468,12 +465,11 @@ func TestRouterConfigureDuringWorkerHandle(t *testing.T) {
sm.Handle("GET /", staticPage(t, "ok"))
cfg := RouterConfig{
FixedNumGoroutines: 2,
Mux: &sm,
RequestHeaderBufferSize: 512,
RequestNumHeaderKVCap: 16,
ResponseHeaderMinBufferSize: 512,
}
if err := router.Configure(cfg); err != nil {
if err := router.Configure(&sm, cfg); err != nil {
t.Fatal(err)
}
defer router.Shutdown()
@@ -495,7 +491,7 @@ func TestRouterConfigureDuringWorkerHandle(t *testing.T) {
for range 20 {
// errBusyExchanges is legitimate backpressure: the previous
// generation was still serving when the buffers were needed.
if err := router.Configure(cfg); err != nil && err != errBusyExchanges {
if err := router.Configure(&sm, cfg); err != nil && err != errBusyExchanges {
t.Error(err)
return
}
+122
View File
@@ -0,0 +1,122 @@
package httphi
import (
"math"
"unsafe"
"github.com/soypat/lneto/http/httpraw"
)
const (
// minRequestHeaderBuffer is the smallest request buffer [httpraw.Header]
// accepts with buffer growth disabled, which is how exchanges are configured.
minRequestHeaderBuffer = 32
// minResponseHeaderBuffer is the room [Exchange.FlushHeader] needs for the
// CRLF closing the header block, written even when no field was staged.
minResponseHeaderBuffer = len("\r\n")
// maxExchangeBuffer bounds an exchange's whole buffer: [Exchange] indexes it
// with uint16 offsets, so a larger one would be addressed truncated.
maxExchangeBuffer = math.MaxUint16
// sizeofExchange is the fixed cost of an exchange, which a Router pays per
// connection it can serve concurrently on top of the buffers it hands it.
// It dwarfs a small request buffer, so budgets must account for it.
sizeofExchange = int(unsafe.Sizeof(Exchange{}))
// sizeofPathValue is the per-wildcard cost of the path value table.
sizeofPathValue = int(unsafe.Sizeof(PathValue{}))
// sizeofJob is an exchange's slot in the queue connections wait on for a
// worker goroutine, sized to the goroutine count in worker mode.
sizeofJob = int(unsafe.Sizeof(job{}))
// bytesPerHeaderField is the request buffer [DefaultRouterConfig] budgets per
// parseable header field. Real fields run a little longer than this
// ("Accept-Encoding: gzip, deflate, br\r\n" is 35 bytes), so a request fills
// the buffer before it exhausts the field table, which is the cheaper of the
// two limits to hit: growing the table costs [httpraw.SizeKV] per field on top
// of the bytes the field already occupies.
bytesPerHeaderField = 32
// defaultResponseHeaderBuffer is the response header room
// [DefaultRouterConfig] reserves when the budget can afford it: enough for a
// Content-Type, a Content-Length and a Connection field with room to spare.
// It does not scale with the request buffer because what a response header
// costs depends on the fields a handler stages, not on the request's size.
defaultResponseHeaderBuffer = 128
)
// MemoryUsagePerConnection returns the heap bytes a [Router] configured with cfg
// reserves for each connection it can serve concurrently, maxPathValues being
// the [Mux.MaxPathValues] of the mux it is configured with. Goroutine stacks are
// not counted: those are the runtime's to size, not the router's.
//
// In worker mode this is exact and fixed, so a router's whole heap footprint is
// this times FixedNumGoroutines, plus the runtime's own header for the job
// queue. With FixedNumGoroutines -1 the router allocates one of these per
// connection in flight instead, so the total grows with peak concurrency.
//
// It is the inverse of [DefaultRouterConfig] and useful to check a hand written
// configuration against a memory budget.
func (cfg RouterConfig) MemoryUsagePerConnection(maxPathValues int) int {
if maxPathValues < 0 {
maxPathValues = 0 // Mux with no routes registered yet, see [Mux.MaxPathValues].
}
n := sizeofExchange + // Exchange itself, an element of the router's exchange store.
cfg.RequestHeaderBufferSize + cfg.ResponseHeaderMinBufferSize + // Its window into the raw buffer.
cfg.RequestNumHeaderKVCap*httpraw.SizeKV + // The request header's field table.
maxPathValues*sizeofPathValue // Its window into the path value store.
if cfg.workerMode() {
n += sizeofJob // Slot in the queue connections wait on for a worker.
}
return n
}
// DefaultRouterConfig is a general purpose configuration creator
// for small, medium, large, performant or embedded projects.
//
// Generalness is achieved with parameters that let the Configuration
// determine allocation buffer sizes based on typical usage for that
// number of goroutines and heap allocation on a per-connection basis.
func DefaultRouterConfig(numGoroutines, memoryPerConnectionBytes, maxPathValues int) RouterConfig {
// The budget is spent on the request header buffer first, since that is what
// decides which requests are answered at all, then on a field table sized to
// match it and a small response header reserve. A budget too small to fund the
// minimum viable exchange yields the minimum instead, so the returned config is
// always one [Router.Configure] accepts but may exceed a budget under roughly
// sizeofExchange + 200 bytes. Check it with MemoryUsagePerConnection when the
// bound has to hold.
if numGoroutines <= 0 {
numGoroutines = -1 // Unbounded mode, the only non-positive value Validate accepts.
}
// Everything the exchange costs before any buffer is sized: subtract it first
// so the buffers below divide up what is actually left to spend.
fixed := sizeofExchange + maxPathValues*sizeofPathValue
if numGoroutines > 0 {
fixed += sizeofJob
}
// A budget past what the buffers may grow to is only spendable up to the cap
// below, so clamp before the products: on a 32 bit target an unclamped
// multiply would overflow and wrap a generous budget into a tiny buffer.
const maxSpendable = (maxExchangeBuffer + defaultResponseHeaderBuffer) *
(bytesPerHeaderField + httpraw.SizeKV) / bytesPerHeaderField
avail := min(memoryPerConnectionBytes-fixed, maxSpendable)
// The response reserve is a floor rather than a share of the budget, but a
// budget this small cannot afford the full one without starving the request.
respBuf := min(defaultResponseHeaderBuffer, avail/4)
// Solve avail-respBuf = reqBuf + reqBuf/bytesPerHeaderField*httpraw.SizeKV for
// reqBuf, the field table growing with the buffer it parses.
reqBuf := (avail - respBuf) * bytesPerHeaderField / (bytesPerHeaderField + httpraw.SizeKV)
// Clamp to what [RouterConfig.Validate] accepts. Truncating division above
// keeps the result under budget; these floors are what can push it over.
respBuf = max(respBuf, minResponseHeaderBuffer)
reqBuf = max(reqBuf, minRequestHeaderBuffer)
reqBuf = min(reqBuf, maxExchangeBuffer-respBuf)
return RouterConfig{
FixedNumGoroutines: numGoroutines,
RequestHeaderBufferSize: reqBuf,
ResponseHeaderMinBufferSize: respBuf,
RequestNumHeaderKVCap: max(reqBuf/bytesPerHeaderField, 1),
NormalizeOutgoingKeys: false,
}
}
+231
View File
@@ -0,0 +1,231 @@
package httphi
import (
"testing"
"github.com/soypat/lneto/http/httpraw"
)
// budgetMux exposes a settable path value count so budget tests can sweep it
// without registering patterns that bind that many wildcards.
type budgetMux struct {
MuxSlice
maxPathValues int
}
func (m *budgetMux) MaxPathValues() int { return m.maxPathValues }
func newBudgetMux(maxPathValues int) *budgetMux {
mux := &budgetMux{maxPathValues: maxPathValues}
mux.Handle("GET /", func(*Exchange) {})
return mux
}
// TestDefaultRouterConfigHonorsBudget sweeps budgets and path value counts and
// checks the returned configuration both fits its budget and configures a
// router. The floor is documented: below it the minimum viable exchange comes
// back instead, which is the only case allowed to exceed the budget.
func TestDefaultRouterConfigHonorsBudget(t *testing.T) {
minCfg := RouterConfig{
FixedNumGoroutines: 1,
RequestHeaderBufferSize: minRequestHeaderBuffer,
ResponseHeaderMinBufferSize: minResponseHeaderBuffer,
RequestNumHeaderKVCap: 1,
}
for _, numGoro := range []int{-1, 1, 4} {
for _, maxPathValues := range []int{0, 1, 4, 32} {
floor := minCfg.MemoryUsagePerConnection(maxPathValues)
mux := newBudgetMux(maxPathValues)
for _, budget := range []int{0, 1, 64, 256, 512, 1024, 4096, 65536, 1 << 20} {
cfg := DefaultRouterConfig(numGoro, budget, maxPathValues)
if err := cfg.Validate(); err != nil {
t.Fatalf("goro=%d pathvals=%d budget=%d: %s", numGoro, maxPathValues, budget, err)
}
got := cfg.MemoryUsagePerConnection(maxPathValues)
if got > budget && budget >= floor {
t.Errorf("goro=%d pathvals=%d budget=%d: uses %d bytes, over budget",
numGoro, maxPathValues, budget, got)
}
var router Router
if err := router.Configure(mux, cfg); err != nil {
t.Fatalf("goro=%d pathvals=%d budget=%d: %s", numGoro, maxPathValues, budget, err)
}
router.Shutdown()
}
}
}
}
// TestDefaultRouterConfigSpendsBudget guards the other direction: a config that
// fits but leaves most of the budget unspent is as wrong as one that overruns,
// since the memory is reserved either way.
func TestDefaultRouterConfigSpendsBudget(t *testing.T) {
const maxPathValues = 2
for _, budget := range []int{1024, 2048, 4096, 16384} {
cfg := DefaultRouterConfig(4, budget, maxPathValues)
used := cfg.MemoryUsagePerConnection(maxPathValues)
if pct := used * 100 / budget; pct < 95 {
t.Errorf("budget=%d: spends only %d bytes (%d%%)", budget, used, pct)
}
}
}
// TestExchangeMemoryTerms pins each term of MemoryUsagePerConnection to the
// allocation it stands for, so a layout change downstream fails here rather than
// silently letting a router overrun its budget.
func TestExchangeMemoryTerms(t *testing.T) {
const maxPathValues = 4
cfg := RouterConfig{
FixedNumGoroutines: 2,
RequestHeaderBufferSize: 512,
ResponseHeaderMinBufferSize: 128,
RequestNumHeaderKVCap: 16,
}
want := sizeofExchange + 512 + 128 + 16*httpraw.SizeKV + maxPathValues*sizeofPathValue + sizeofJob
if got := cfg.MemoryUsagePerConnection(maxPathValues); got != want {
t.Errorf("worker mode: got %d want %d", got, want)
}
// Unbounded mode has no job queue to reserve a slot in.
cfg.FixedNumGoroutines = -1
if got := cfg.MemoryUsagePerConnection(maxPathValues); got != want-sizeofJob {
t.Errorf("unbounded mode: got %d want %d", got, want-sizeofJob)
}
// A mux with no routes registered reports -1, which must not subtract memory.
if got := cfg.MemoryUsagePerConnection(-1); got != cfg.MemoryUsagePerConnection(0) {
t.Errorf("unregistered mux: got %d want %d", got, cfg.MemoryUsagePerConnection(0))
}
}
// TestRouterSharesExchangeStores checks the invariant the memory accounting
// rests on: every exchange's buffer and path values are windows into the two
// stores the router allocates, non-overlapping and exactly the configured size.
// Measuring allocations would only observe this indirectly.
func TestRouterSharesExchangeStores(t *testing.T) {
const numGoro = 8
const maxPathValues = 3
mux := newBudgetMux(maxPathValues)
cfg := DefaultRouterConfig(numGoro, 1024, maxPathValues)
var router Router
if err := router.Configure(mux, cfg); err != nil {
t.Fatal(err)
}
defer router.Shutdown()
wantRaw := cfg.RequestHeaderBufferSize + cfg.ResponseHeaderMinBufferSize
if len(router.exchs) != numGoro {
t.Fatalf("got %d exchanges, want %d", len(router.exchs), numGoro)
}
if cap(router.globbuf) < numGoro*wantRaw {
t.Errorf("raw store holds %d bytes, want %d", cap(router.globbuf), numGoro*wantRaw)
}
if cap(router.globpath) < numGoro*maxPathValues {
t.Errorf("path store holds %d values, want %d", cap(router.globpath), numGoro*maxPathValues)
}
rawSeen := make(map[*byte]int, numGoro*wantRaw)
pathSeen := make(map[*PathValue]int, numGoro*maxPathValues)
for i := range router.exchs {
exch := &router.exchs[i]
if len(exch.rawbuf) != wantRaw {
t.Errorf("exchange %d: raw buffer is %d bytes, want %d", i, len(exch.rawbuf), wantRaw)
}
if len(exch.pathValues) != maxPathValues {
t.Errorf("exchange %d: %d path values, want %d", i, len(exch.pathValues), maxPathValues)
}
// Every byte must come from the shared store and belong to this exchange
// alone: an exchange allocating its own, or two sharing a window, would
// make the per-connection accounting a fiction.
for j := range exch.rawbuf {
p := &exch.rawbuf[j]
if owner, dup := rawSeen[p]; dup {
t.Fatalf("exchanges %d and %d share raw buffer byte %d", owner, i, j)
}
rawSeen[p] = i
}
for j := range exch.pathValues {
p := &exch.pathValues[j]
if owner, dup := pathSeen[p]; dup {
t.Fatalf("exchanges %d and %d share path value %d", owner, i, j)
}
pathSeen[p] = i
}
}
if len(rawSeen) != numGoro*wantRaw {
t.Errorf("exchanges cover %d raw bytes, want %d", len(rawSeen), numGoro*wantRaw)
}
}
// TestExchangeConfigureIsAllocationFree checks an exchange handed all of its
// memory allocates none of its own, which is what lets a router carve every
// exchange out of its two stores.
func TestExchangeConfigureIsAllocationFree(t *testing.T) {
cfg := ExchangeConfig{
RawBuf: make([]byte, 640),
RequestBufferLim: 512,
NumHeaderKVCap: 16,
NoRequestBufferGrowth: true,
PathValuesBuf: make([]PathValue, 4),
}
var exch Exchange
exch.Configure(cfg) // Field table allocates once, then settles.
allocs := testing.AllocsPerRun(100, func() {
exch.Configure(cfg)
})
if allocs != 0 {
t.Errorf("Exchange.Configure allocates %v times, want 0", allocs)
}
}
// TestMemoryUsagePerConnectionMatchesHeap checks the accounting against the heap
// a router actually takes, which is what makes the number worth budgeting
// against.
//
// It measures two budgets and compares the difference rather than either
// absolute figure. A router's heap carries costs the accounting does not claim
// and should not: size class rounding, the job queue's runtime header and the
// runtime's per-goroutine bookkeeping. Those are identical at both budgets, so
// subtracting cancels them and leaves only the buffers, whose growth is exactly
// what MemoryUsagePerConnection predicts. Goroutine stacks never enter into it,
// the runtime accounting them separately from the heap measured here.
func TestMemoryUsagePerConnectionMatchesHeap(t *testing.T) {
if testing.Short() {
t.Skip("measures heap over many Configure iterations")
}
const numGoro = 64
const maxPathValues = 3
mux := newBudgetMux(maxPathValues)
measure := func(budget int) (accounted, heap int) {
cfg := DefaultRouterConfig(numGoro, budget, maxPathValues)
res := testing.Benchmark(func(b *testing.B) {
b.ReportAllocs()
for range b.N {
var router Router
if err := router.Configure(mux, cfg); err != nil {
b.Fatal(err)
}
router.Shutdown()
}
})
return numGoro * cfg.MemoryUsagePerConnection(maxPathValues), int(res.AllocedBytesPerOp())
}
lowAcct, lowHeap := measure(2048)
highAcct, highHeap := measure(16384)
wantGrowth := highAcct - lowAcct
gotGrowth := highHeap - lowHeap
t.Logf("accounted %d->%d (+%d), heap %d->%d (+%d)",
lowAcct, highAcct, wantGrowth, lowHeap, highHeap, gotGrowth)
// What remains after cancelling is buffer growth, which the accounting covers
// term for term. Only size class rounding on the grown buffers is left over.
const tolerancePercent = 2
if diff := abs(gotGrowth - wantGrowth); diff*100 > wantGrowth*tolerancePercent {
t.Errorf("budget growth accounted %d bytes, heap grew %d (%+d)",
wantGrowth, gotGrowth, gotGrowth-wantGrowth)
}
}
func abs(x int) int {
if x < 0 {
return -x
}
return x
}
+1 -5
View File
@@ -3,7 +3,7 @@ package httphi
// StatusText returns a text for the HTTP status code. It returns the empty
// string if the code is unknown.
func StatusText(code int) string {
switch status(code) {
switch code {
case StatusContinue:
return "Continue"
case StatusSwitchingProtocols:
@@ -133,10 +133,6 @@ func StatusText(code int) string {
}
}
const ()
type status int
// HTTP status codes as registered with IANA.
// See: https://www.iana.org/assignments/http-status-codes/http-status-codes.xhtml
const (
+6
View File
@@ -462,6 +462,12 @@ type pairKV struct {
value view // value start >0 means value is present.
}
// SizeKV is the heap cost of a single key/value field slot, as reserved by the
// numHeaderCapacity argument to [HeaderV1.Reset] and by [Form.Reset]. Callers
// budgeting a fixed memory pool up front, such as a Router sizing its
// exchanges, multiply it by the pair capacity to account the field table.
const SizeKV = int(unsafe.Sizeof(pairKV{}))
// isValid is for stores parsed in place, where offset 0 is the first key so
// only length can signal presence. Empty keys are valid: see valueless cookies.
func (pair pairKV) isValid() bool {
+5 -15
View File
@@ -76,16 +76,10 @@ type ConnConfig struct {
// Logger sets the [Conn] logger.
// Lower level logging available at [Handler.SetLoggers] via [Conn.InternalHandler].
Logger *slog.Logger
// LossRecovery is the optional packet-loss recovery algorithm (RTO,
// congestion control, ...) for the connection. If set, Nanotime must also be
// set (else Configure returns an error). Leaving it nil disables loss
// recovery. See [LossRecovery].
LossRecovery LossRecovery
// Nanotime is the monotonic time source in nanoseconds (the func() int64
// convention used across lneto) that drives LossRecovery. It is required when
// LossRecovery is set and unused otherwise. The tcp package reads it only to
// stamp the loss-recovery hooks; it holds no clock itself.
Nanotime func() int64
// Policy is the optional transmit-steering algorithm (RTO, congestion
// control, ...) for the connection. nil disables it. A Policy needing time
// carries its own clock. See [Policy].
Policy Policy
}
// Configure should be called on any newly created connection before usage. See [ConnConfig].
@@ -93,10 +87,6 @@ func (conn *Conn) Configure(config ConnConfig) (err error) {
if config.RWBackoff == nil {
return lneto.ErrMissingHALConfig
}
if config.LossRecovery != nil && config.Nanotime == nil {
// The tcp package holds no clock: a loss-recovery algorithm cannot run without it.
return lneto.ErrInvalidConfig
}
conn.mu.Lock()
defer conn.mu.Unlock()
err = conn.h.SetBuffers(config.TxBuf, config.RxBuf, config.TxPacketQueueSize)
@@ -105,7 +95,7 @@ func (conn *Conn) Configure(config ConnConfig) (err error) {
}
conn._backoff = config.RWBackoff
conn.logger.log = config.Logger
conn.h.SetLossRecovery(config.LossRecovery, config.Nanotime)
conn.h.SetPolicy(config.Policy)
return nil
}
+27 -3
View File
@@ -95,6 +95,12 @@ func (tcb *ControlBlock) RecvWindow() Size { return tcb.rcv.WND }
// ISS returns the initial sequence number of the connection that was defined on a call to Open by user.
func (tcb *ControlBlock) ISS() Value { return tcb.snd.ISS }
// SendUNA returns snd.UNA, the oldest sequence number not yet acked by the remote.
func (tcb *ControlBlock) SendUNA() Value { return tcb.snd.UNA }
// SendNext returns snd.NXT, one past the highest sequence number sent.
func (tcb *ControlBlock) SendNext() Value { return tcb.snd.NXT }
// MaxInFlightData returns the maximum size of a segment that can be sent by taking into account
// the send window size and the unacked data. Returns 0 before StateSynRcvd.
func (tcb *ControlBlock) MaxInFlightData() Size {
@@ -257,14 +263,32 @@ func (tcb *ControlBlock) HasPendingRetransmit() bool {
return tcb._state.TxDataOpen() && tcb.dupack >= retransmitAfterDupacks && tcb.nRetransmit <= tcb.dupack-retransmitAfterDupacks
}
// RetransmitFrom rewinds snd.NXT back to newNxt so the next PendingSegment and
// Send calls retransmit unacknowledged data from that sequence number onwards.
// It must be paired with ringTx.RetransmitFrom to rewind the transmit buffer to
// the same point. Implements RFC 9293 §3.10.8 (RETRANSMISSION TIMEOUT).
//
// It reports false and changes nothing when newNxt falls outside the
// unacknowledged range [snd.UNA, snd.NXT] or the connection cannot send data, so
// a misbehaving [Policy] cannot corrupt the send sequence space.
func (tcb *ControlBlock) RetransmitFrom(newNxt Value) bool {
if !tcb._state.TxDataOpen() {
return false
} else if newNxt.LessThan(tcb.snd.UNA) || tcb.snd.NXT.LessThan(newNxt) {
return false
}
tcb.snd.NXT = newNxt
tcb.dupack = 0
tcb.nRetransmit = 0
return true
}
// RetransmitAll rewinds snd.NXT back to snd.UNA so the next PendingSegment and
// Send calls retransmit all unacknowledged data from the oldest sequence number
// (go-back-N). It must be paired with ringTx.RetransmitFromUNA to rewind the
// transmit buffer. Implements RFC 9293 §3.10.8 (RETRANSMISSION TIMEOUT).
func (tcb *ControlBlock) RetransmitAll() {
tcb.snd.NXT = tcb.snd.UNA
tcb.dupack = 0
tcb.nRetransmit = 0
tcb.RetransmitFrom(tcb.snd.UNA)
}
// PendingSegment calculates a suitable next segment to send from a payload length.
+69 -59
View File
@@ -31,14 +31,8 @@ type Handler struct {
optcodec OptionCodec
// reasm tracks out-of-order segments staged in bufRx's free region. Always
// enabled once buffers are set (see [Handler.SetBuffers]).
reasm reassembly
// loss is the optional packet-loss recovery algorithm (RTO, congestion
// control, ...) driven from the rx/tx hooks. nil disables loss recovery, in
// which case the connection behaves as if no timing existed. nanotime is the
// monotonic time source (nanoseconds) passed to those hooks; it is non-nil
// whenever loss is non-nil (enforced by [Conn.Configure]). See [LossRecovery].
loss LossRecovery
nanotime func() int64
reasm reassembly
policy Policy
closing bool
shutdownRx bool
@@ -79,27 +73,16 @@ func (h *Handler) SetBuffers(txbuf, rxbuf []byte, packets int) error {
return h.bufTx.ResetOrReuse(txbuf, packets, 0)
}
// SetLossRecovery installs the packet-loss recovery algorithm and the monotonic
// time source (nanoseconds, the func() int64 convention used across lneto) that
// drives it. The tcp package keeps no clock of its own; nanotime is read only to
// stamp the rx/tx hooks (see [LossRecovery]). Passing loss == nil disables loss
// recovery. It should be set before the connection is opened.
func (h *Handler) SetLossRecovery(loss LossRecovery, nanotime func() int64) {
h.loss = loss
h.nanotime = nanotime
// SetPolicy installs the transmit-steering algorithm. nil disables it.
// It should be set before the connection is opened. See [Policy].
func (h *Handler) SetPolicy(policy Policy) {
h.policy = policy
}
func (h *Handler) policyEnabled() bool { return h.policy != nil }
func (h *Handler) lossEnabled() bool { return h.loss != nil }
// NextDeadline returns the monotonic-nanosecond instant at which the connection
// must next be serviced by a transmit attempt (e.g. an RTO expiry), or 0 when
// there is no deadline or no loss recovery is configured. See [LossRecovery].
func (h *Handler) NextDeadline() int64 {
if h.loss == nil {
return 0
}
return h.loss.NextDeadline()
}
// ControlBlock returns the state machine underlying the Handler, mainly so a
// [Policy] can read the sequence spaces. Not for modification.
func (h *Handler) ControlBlock() *ControlBlock { return &h.scb }
// LocalPort returns the local port of the connection. Returns 0 if the connection is closed and uninitialized.
func (h *Handler) LocalPort() uint16 {
@@ -165,16 +148,15 @@ func (h *Handler) reset(localPort, remotePort uint16, iss Value) {
shutdownRx: false,
// Persist configuration across reopen:
validator: h.validator,
loss: h.loss,
nanotime: h.nanotime,
policy: h.policy,
logger: h.logger,
// persist memory across repoen:
bufTx: h.bufTx,
bufRx: h.bufRx,
reasm: h.reasm,
}
if h.lossEnabled() {
h.loss.Reset()
if h.policyEnabled() {
h.policy.Reset()
}
h.reasm.clear() // preserve metadata capacity across reopen, drop held segments.
h.bufTx.ResetOrReuse(nil, 0, iss)
@@ -212,9 +194,7 @@ func (h *Handler) Recv(incomingPacket []byte) error {
return nil
}
// Notify loss recovery of the received segment (RTT sampling, timer
// management) and let it drop the segment before processing if it asks to.
if h.lossEnabled() && !h.loss.PreRx(segIncoming, h.nanotime()).Keep {
if h.policyEnabled() && !h.policy.PreRx(h, tfrm) {
return nil
}
@@ -245,6 +225,9 @@ func (h *Handler) Recv(incomingPacket []byte) error {
if prevState != h.scb.State() {
h.info("tcp.Handler:rx-statechange", slog.Uint64("port", uint64(h.localPort)), slog.String("old", prevState.String()), slog.String("new", h.scb.State().String()), slog.String("rxflags", segIncoming.Flags.String()))
}
if h.policyEnabled() {
h.policy.PostRx(h, prevState, tfrm)
}
if segIncoming.DATALEN != 0 && h.shutdownRx && (h.scb.State() == StateFinWait1 || h.scb.State() == StateFinWait2) {
// soypat/lneto#50: the application is done in both directions — read side
// shut down (CloseRead) and our FIN sent (Close) — so inbound data has no
@@ -380,16 +363,28 @@ func (h *Handler) Send(b []byte) (int, error) {
if h.IsTxOver() {
return 0, net.ErrClosed
}
var now int64
if h.lossEnabled() {
now = h.nanotime()
if h.loss.PreTx(now).RetransmitAll {
// Go-back-N retransmission directed by loss recovery: rewind the
// send sequence and transmit buffer so unacknowledged data is resent
// from snd.UNA. Done before the early short-circuit below so an
// expired RTO retransmits even with no new data queued.
h.scb.RetransmitAll()
h.bufTx.RetransmitFromUNA()
tfrm, err := NewFrame(b)
if err != nil {
return 0, err
}
offset := uint8(5)
var holdNew bool
if h.policyEnabled() {
// Hand the Policy a defined frame: zeroed header at the minimum offset.
// It may append options and raise the offset, which is read back below.
tfrm.ClearHeader()
tfrm.SetOffsetAndFlags(offset, 0)
rtxFrom, doRtx, hold := h.policy.PreTx(h, tfrm)
holdNew = hold
if doRtx && h.scb.RetransmitFrom(rtxFrom) {
// Retransmission directed by the Policy: rewind the transmit buffer
// to match the send sequence so unacknowledged data is resent. Done
// before the early short-circuit below so an expired RTO
// retransmits even with no new data queued.
h.bufTx.RetransmitFrom(rtxFrom)
}
if o, _ := tfrm.OffsetAndFlags(); o > offset && int(o)*4 < len(b) {
offset = o
}
}
awaitingSyn := h.AwaitingSynSend()
@@ -405,29 +400,27 @@ func (h *Handler) Send(b []byte) (int, error) {
// Early nop short circuit.
return 0, nil
}
tfrm, err := NewFrame(b)
if err != nil {
return 0, err
}
if buffered == 0 && h.closing && (h.scb.State() != StateCloseWait || !h.scb.HasPending()) {
// If Close called and no more data to be sent, terminate connection.
// In CLOSE-WAIT: wait until the pending ACK is sent first, since scb.Close()
// overwrites pending with [FIN|ACK] (unlike ESTABLISHED which merges via bitmask).
h.closing = false
err = h.scb.Close()
err := h.scb.Close()
if err != nil {
h.logerr("tcp.Handler.Close", slog.String("err", errstr(err)), slog.String("state", h.State().String()))
h.Abort()
return 0, io.EOF
}
}
offset := uint8(5)
mss := uint16(len(b) - sizeHeaderTCP)
// optHead is where the Handler's own options begin: after the fixed header
// and after any options the Policy already wrote, so neither clobbers the other.
optHead := int(offset) * 4
mss := uint16(len(b) - optHead)
var segment Segment
if awaitingSyn || requeueControl && h.scb.State() == StateSynSent {
// Handling init syn segment.
segment = ClientSynSegment(h.bufTx.iss, Size(h.bufRx.Size()))
h.optcodec.PutOption16(b[sizeHeaderTCP:], OptMaxSegmentSize, mss)
h.optcodec.PutOption16(b[optHead:], OptMaxSegmentSize, mss)
offset++
if requeueControl {
h.info("tcp.Handler:requeue-syn", slog.Uint64("port", uint64(h.localPort)), slog.Uint64("rport", uint64(h.remotePort)))
@@ -439,7 +432,7 @@ func (h *Handler) Send(b []byte) (int, error) {
WND: Size(h.bufRx.Free()),
Flags: synack,
}
h.optcodec.PutOption16(b[sizeHeaderTCP:], OptMaxSegmentSize, mss)
h.optcodec.PutOption16(b[optHead:], OptMaxSegmentSize, mss)
offset++
h.info("tcp.Handler:requeue-synack", slog.Uint64("port", uint64(h.localPort)), slog.Uint64("rport", uint64(h.remotePort)))
} else if requeueControl {
@@ -447,17 +440,22 @@ func (h *Handler) Send(b []byte) (int, error) {
return 0, nil
} else {
var ok bool
maxPayload := len(b) - sizeHeaderTCP
maxPayload := len(b) - optHead
if holdNew && !h.nextSegmentIsRetransmit() {
// Policy is holding new data back (congestion window exhausted).
// A retransmission it directed in this same call still proceeds.
maxPayload = 0
}
segment, ok = h.scb.PendingSegment(maxPayload)
segment.WND = h.recvWindow()
if !ok {
// No pending control segment or data to send. Yield.
return 0, nil
} else if segment.Flags == synack {
h.optcodec.PutOption16(b[sizeHeaderTCP:], OptMaxSegmentSize, mss)
h.optcodec.PutOption16(b[optHead:], OptMaxSegmentSize, mss)
offset++
} else if segment.DATALEN > 0 {
n, err := h.bufTx.MakePacket(b[sizeHeaderTCP:sizeHeaderTCP+segment.DATALEN], segment.SEQ)
n, err := h.bufTx.MakePacket(b[optHead:optHead+int(segment.DATALEN)], segment.SEQ)
if err != nil {
return 0, err
}
@@ -474,15 +472,19 @@ func (h *Handler) Send(b []byte) (int, error) {
} else if prevState != h.scb.State() && h.logenabled(slog.LevelInfo) {
h.info("tcp.Handler:tx-statechange", slog.Uint64("port", uint64(h.localPort)), slog.String("oldState", prevState.String()), slog.String("newState", h.scb.State().String()), slog.String("txflags", segment.Flags.String()))
}
if h.lossEnabled() {
h.loss.PostTx(segment, now)
}
h.requeueControl = false
tfrm.SetSourcePort(h.localPort)
tfrm.SetDestinationPort(h.remotePort)
tfrm.SetSegment(segment, offset)
tfrm.SetUrgentPtr(0)
datalen := int(offset)*4 + int(segment.DATALEN)
if h.policyEnabled() {
// Frame trimmed to what is actually emitted so the Policy's Payload()
// is the segment data and nothing more.
if sent, err := NewFrame(b[:datalen]); err == nil {
h.policy.PostTx(h, sent)
}
}
closedSuccess := prevState == StateTimeWait && segment.Flags.HasAny(FlagACK)
if closedSuccess {
h.reset(0, 0, 0)
@@ -494,6 +496,14 @@ func (h *Handler) Send(b []byte) (int, error) {
return datalen, nil
}
// nextSegmentIsRetransmit reports whether the next data segment would resend
// already-transmitted bytes rather than open new sequence space. Used to let a
// retransmission through while a [Policy] holds new data back.
func (h *Handler) nextSegmentIsRetransmit() bool {
endSeq, hasSent := h.bufTx.sentEndSeq()
return hasSent && h.scb.snd.NXT.LessThan(endSeq)
}
// Write implements [io.Writer] by copying b to a internal buffer to be sent over the network on the next
// [Handler.Send] call that can send data to remote peer. Use [Handler.Free] to know the maximum length the argument slice can be before erroring.
func (h *Handler) Write(b []byte) (int, error) {
-83
View File
@@ -1,83 +0,0 @@
package tcp
// LossRecovery abstracts TCP packet-loss recovery: RTO, congestion control and
// any similar algorithm that observes segment traffic and steers the
// connection's transmit behaviour. As far as the tcp package is concerned these
// are all the same thing — packet-loss recovery algorithms — so they share one
// interface (see discussion #157).
//
// The tcp package stays free of any time source: the current monotonic time in
// nanoseconds (the func() int64 convention used across lneto) is passed in at
// each hook boundary. It originates from [ConnConfig.Nanotime] and satisfies the
// "WHEN was this segment rx/tx'd" requirement without a clock living inside the
// state machine, which also keeps implementations deterministic for testing
// (see issue #140).
//
// The interface is intentionally free of errors: an implementation handles or
// reports its own errors rather than propagating them into lneto internals.
//
// Introspection (smoothed RTT, current window, ...) is deliberately left off the
// interface; expose it on the concrete implementation the caller constructs and
// hands to [ConnConfig].
type LossRecovery interface {
// Reset returns the implementation to its initial, pre-connection state. It
// is invoked whenever the connection is (re)opened or aborted so a single
// LossRecovery value can be reused across the lifetime of connection reuse
// (see discussion #115).
Reset()
// NextDeadline returns the monotonic-nanosecond instant at which the
// connection must next be serviced by a transmit attempt — typically the RTO
// expiry. A return of 0 means there is no pending deadline. It replaces a
// poll/atomic-flag scheme with a deadline the caller's event loop can
// schedule against.
NextDeadline() int64
// PreRx is called for every segment received on the TCP port before the
// state machine processes it, with the monotonic time the segment arrived. It
// returns whether the segment should be kept (processed) or dropped.
PreRx(incoming Segment, now int64) RxDirective
// PreTx is called on entering the transmit path (Encapsulate), before a
// segment is built, with the current monotonic time. Its directive tells the
// connection whether to retransmit unacknowledged data, rewind the send
// pointer, or hold back new data.
PreTx(now int64) TxDirective
// PostTx is called on leaving the transmit path with the segment that was
// actually emitted and the monotonic time it was sent. This is where segment
// timing (for RTT sampling and the retransmission timer) is recorded.
PostTx(outgoing Segment, now int64)
}
// TxDirective is returned by [LossRecovery.PreTx] to steer the transmit path.
// The zero value directs the connection to proceed normally (send new data if
// available, no retransmission).
type TxDirective struct {
// RewindNXT is the number of sequence-space octets to rewind snd.NXT by
// before transmitting, for partial (e.g. selective) retransmission. Zero
// means no rewind. It is independent of Retransmit, which rewinds fully to
// snd.UNA.
// RewindNXT uint32
// RetransmitAll requests go-back-N retransmission: the connection rewinds
// snd.NXT to snd.UNA and resends unacknowledged data from the oldest
// sequence number.
RetransmitAll bool
// HoldNew pauses transmission of new data (for example when the congestion
// window is exhausted). Retransmissions already directed by this same
// directive still proceed.
// HoldNew bool
}
// RxDirective is returned by [LossRecovery.PreRx].
//
// NOTE: its shape is the minimum viable contract — it mirrors the original
// PreRx "keep" boolean from discussion #157 — and is the one element of the
// interface not yet fully settled there. It is a struct (rather than a bare
// bool) so fields can be added without breaking implementations.
type RxDirective struct {
// Keep reports whether the received segment should be handed to the state
// machine. A false value drops the segment before it is processed.
Keep bool
}
-262
View File
@@ -1,262 +0,0 @@
package tcp
import (
"math/rand"
"testing"
"github.com/soypat/lneto/ethernet"
)
// recordingLoss is a test LossRecovery that records every hook invocation and
// lets the test steer the directives returned to the Handler. It is the
// interface counterpart driven by the Handler under test.
type recordingLoss struct {
resets int
preRx []hookCall
preTx []int64
postTx []hookCall
deadline int64 // value NextDeadline reports back.
// Directives handed back to the Handler.
keep bool // PreRx result. Default true (see newRecordingLoss).
tx TxDirective // PreTx result.
}
type hookCall struct {
seg Segment
now int64
}
func newRecordingLoss() *recordingLoss { return &recordingLoss{keep: true} }
var _ LossRecovery = (*recordingLoss)(nil)
func (l *recordingLoss) Reset() { l.resets++ }
func (l *recordingLoss) NextDeadline() int64 { return l.deadline }
func (l *recordingLoss) PreRx(incoming Segment, now int64) RxDirective {
l.preRx = append(l.preRx, hookCall{seg: incoming, now: now})
return RxDirective{Keep: l.keep}
}
func (l *recordingLoss) PreTx(now int64) TxDirective {
l.preTx = append(l.preTx, now)
return l.tx
}
func (l *recordingLoss) PostTx(outgoing Segment, now int64) {
l.postTx = append(l.postTx, hookCall{seg: outgoing, now: now})
}
// TestLossRecovery_DisabledByDefault verifies the Handler runs normally with no
// loss recovery installed: NextDeadline reports no deadline and the transmit/
// receive paths never touch a nil LossRecovery.
func TestLossRecovery_DisabledByDefault(t *testing.T) {
const mtu = ethernet.MaxMTU
rng := rand.New(rand.NewSource(1))
client, server := newHandler(t, mtu, 3), newHandler(t, mtu, 3)
setupClientServer(t, rng, client, server)
if d := client.NextDeadline(); d != 0 {
t.Fatalf("NextDeadline with no loss recovery = %d, want 0", d)
}
var buf [mtu]byte
establish(t, client, server, buf[:]) // must not panic on nil loss recovery.
}
// TestLossRecovery_HooksInvoked verifies the Handler drives the full hook
// contract across a handshake: Reset on open, PreTx+PostTx on every transmit,
// PreRx on every receive, each stamped with the configured monotonic clock.
func TestLossRecovery_HooksInvoked(t *testing.T) {
const mtu = ethernet.MaxMTU
rng := rand.New(rand.NewSource(2))
client, server := newHandler(t, mtu, 3), newHandler(t, mtu, 3)
loss := newRecordingLoss()
const clockNow = 1_000_000
client.SetLossRecovery(loss, func() int64 { return clockNow })
setupClientServer(t, rng, client, server) // OpenActive → reset → Reset().
if loss.resets == 0 {
t.Fatal("Reset not called on open")
}
var buf [mtu]byte
establish(t, client, server, buf[:])
// Client emitted SYN and the final ACK: both paths must have hit PreTx/PostTx.
if len(loss.preTx) == 0 {
t.Fatal("PreTx never called on transmit")
}
if len(loss.postTx) == 0 {
t.Fatal("PostTx never called on transmit")
}
if len(loss.preTx) != len(loss.postTx) {
t.Fatalf("PreTx calls=%d, PostTx calls=%d, want equal", len(loss.preTx), len(loss.postTx))
}
// Client received the SYN-ACK: PreRx must have seen it.
if len(loss.preRx) == 0 {
t.Fatal("PreRx never called on receive")
}
// The Handler holds no clock: every hook must be stamped from the supplied
// nanotime source.
for i, c := range loss.postTx {
if c.now != clockNow {
t.Fatalf("PostTx[%d].now = %d, want clock %d", i, c.now, clockNow)
}
}
for i, now := range loss.preTx {
if now != clockNow {
t.Fatalf("PreTx[%d].now = %d, want clock %d", i, now, clockNow)
}
}
for i, c := range loss.preRx {
if c.now != clockNow {
t.Fatalf("PreRx[%d].now = %d, want clock %d", i, c.now, clockNow)
}
}
// PostTx receives the segment actually emitted: the first is the SYN.
if !loss.postTx[0].seg.Flags.HasAny(FlagSYN) {
t.Fatalf("first PostTx segment flags=%s, want SYN", loss.postTx[0].seg.Flags)
}
}
// TestLossRecovery_NextDeadlineDelegates verifies NextDeadline is forwarded to
// the installed LossRecovery unchanged.
func TestLossRecovery_NextDeadlineDelegates(t *testing.T) {
const mtu = ethernet.MaxMTU
rng := rand.New(rand.NewSource(3))
client, server := newHandler(t, mtu, 3), newHandler(t, mtu, 3)
loss := newRecordingLoss()
loss.deadline = 4242
client.SetLossRecovery(loss, func() int64 { return 1 })
setupClientServer(t, rng, client, server)
if d := client.NextDeadline(); d != 4242 {
t.Fatalf("NextDeadline = %d, want delegated 4242", d)
}
}
// TestLossRecovery_PreRxDropsSegment verifies a PreRx directive of Keep=false
// drops the segment before the state machine sees it: the payload is not
// buffered and connection state is untouched.
func TestLossRecovery_PreRxDropsSegment(t *testing.T) {
const mtu = ethernet.MaxMTU
rng := rand.New(rand.NewSource(4))
client, server := newHandler(t, mtu, 3), newHandler(t, mtu, 3)
loss := newRecordingLoss()
server.SetLossRecovery(loss, func() int64 { return 1 })
setupClientServer(t, rng, client, server)
var buf [mtu]byte
establish(t, client, server, buf[:]) // keep=true so handshake completes.
// Now start dropping everything the server receives.
loss.keep = false
preRxBefore := len(loss.preRx)
data := []byte("dropme")
if _, err := client.Write(data); err != nil {
t.Fatal("client write:", err)
}
clear(buf[:])
n, err := client.Send(buf[:])
if err != nil {
t.Fatal("client send:", err)
}
if err := server.Recv(buf[:n]); err != nil {
t.Fatalf("dropped segment must return nil, got %v", err)
}
if len(loss.preRx) != preRxBefore+1 {
t.Fatalf("PreRx calls=%d, want %d (segment must reach PreRx)", len(loss.preRx), preRxBefore+1)
}
if server.BufferedInput() != 0 {
t.Fatalf("dropped segment must not be buffered, got %d bytes", server.BufferedInput())
}
if server.State() != StateEstablished {
t.Fatalf("dropped segment must not change state, got %s", server.State())
}
}
// TestLossRecovery_PreTxRetransmitAll verifies a PreTx directive of
// RetransmitAll drives go-back-N: the Handler rewinds and re-emits already-sent,
// unacknowledged data from snd.UNA on the next transmit.
func TestLossRecovery_PreTxRetransmitAll(t *testing.T) {
const mtu = ethernet.MaxMTU
rng := rand.New(rand.NewSource(5))
client, server := newHandler(t, mtu, 3), newHandler(t, mtu, 3)
loss := newRecordingLoss()
client.SetLossRecovery(loss, func() int64 { return 1 })
setupClientServer(t, rng, client, server)
var buf [mtu]byte
establish(t, client, server, buf[:])
// Emit one data segment; server never ACKs, so it stays unacknowledged.
data := []byte("payload")
if _, err := client.Write(data); err != nil {
t.Fatal("client write:", err)
}
clear(buf[:])
n, err := client.Send(buf[:])
if err != nil {
t.Fatal("client send data:", err)
}
if n <= sizeHeaderTCP {
t.Fatal("expected data segment")
}
firstSeg := mustSegment(t, buf[:n], n-sizeHeaderTCP)
// Direct go-back-N on the next transmit.
loss.tx = TxDirective{RetransmitAll: true}
clear(buf[:])
n, err = client.Send(buf[:])
if err != nil {
t.Fatal("client send retransmit:", err)
}
if n <= sizeHeaderTCP {
t.Fatal("expected retransmitted data segment")
}
rtSeg := mustSegment(t, buf[:n], n-sizeHeaderTCP)
if rtSeg.SEQ != firstSeg.SEQ {
t.Fatalf("retransmit SEQ=%d, want original UNA SEQ=%d (go-back-N)", rtSeg.SEQ, firstSeg.SEQ)
}
if rtSeg.DATALEN != firstSeg.DATALEN {
t.Fatalf("retransmit DATALEN=%d, want %d", rtSeg.DATALEN, firstSeg.DATALEN)
}
}
// TestLossRecovery_ResetOnReopen verifies Reset fires on every (re)open and on
// Abort, so a single LossRecovery value can be reused across connection reuse.
func TestLossRecovery_ResetOnReopen(t *testing.T) {
const mtu = ethernet.MaxMTU
client := newHandler(t, mtu, 3)
loss := newRecordingLoss()
client.SetLossRecovery(loss, func() int64 { return 1 })
if err := client.OpenActive(1234, 5678, 0); err != nil {
t.Fatal("open 1:", err)
}
afterOpen := loss.resets
if afterOpen == 0 {
t.Fatal("Reset not called on first open")
}
client.Abort()
if loss.resets <= afterOpen {
t.Fatalf("Reset not called on Abort: resets=%d, want >%d", loss.resets, afterOpen)
}
afterAbort := loss.resets
if err := client.OpenActive(1234, 5678, 0); err != nil {
t.Fatal("open 2:", err)
}
if loss.resets <= afterAbort {
t.Fatalf("Reset not called on reopen: resets=%d, want >%d", loss.resets, afterAbort)
}
}
+24
View File
@@ -0,0 +1,24 @@
package tcp
// Policy observes segment traffic and steers transmit behaviour: RTO,
// congestion control and the like (discussion #157). The tcp package holds no
// clock, so a Policy needing time carries its own (issue #140).
// Introspection stays off the interface; put it on the concrete type.
type Policy interface {
// Reset returns the Policy to its pre-connection state. Called on every
// (re)open and Abort. Must preserve configuration such as a clock.
Reset()
// PreTx is called before writing to a frame.
// The outgoing frame options can be set by the Policy and will be respected if Frame offset >5.
// rtxFrom is ignored unless within [snd.UNA, snd.NXT]. Nothing is committed
// until PostTx: a transmit attempt may emit no segment at all.
PreTx(h *Handler, outgoingOpts Frame) (rtxFrom Value, retransmit, holdNew bool)
// PreRx is called by [Handler] on every incoming segment.
// PreRx can choose to drop segment if it returns keep=false.
PreRx(h *Handler, incoming Frame) (keep bool)
// PostRx is called by [Handler] after accepting an incoming segment.
// TODO: congestion control will also want the pre-Recv snd.UNA here.
PostRx(h *Handler, prevState State, accepted Frame)
// PostTx called on leaving the transmit path with the fully written frame.
PostTx(h *Handler, outgoing Frame)
}
+434
View File
@@ -0,0 +1,434 @@
package tcp
import (
"math/rand"
"testing"
"github.com/soypat/lneto/ethernet"
)
// recordingPolicy records every hook invocation and lets the test steer what is
// returned to the Handler. It is the [Policy] counterpart driven by the Handler
// under test.
type recordingPolicy struct {
resets int
preRx []Segment
preTx int
postRx []Segment
postTx []txRecord
// Values handed back to the Handler.
keep bool // PreRx result. Default true (see newRecordingPolicy).
rtxFrom Value
retransmit bool
holdNew bool
// writeOpts, when non-empty, is appended as TCP options by PreTx.
writeOpts []byte
}
// txRecord is what PostTx observed on the emitted frame.
type txRecord struct {
seg Segment
offset uint8
sport uint16
dport uint16
}
func newRecordingPolicy() *recordingPolicy { return &recordingPolicy{keep: true} }
var _ Policy = (*recordingPolicy)(nil)
func (p *recordingPolicy) Reset() { p.resets++ }
func (p *recordingPolicy) PreRx(h *Handler, incoming Frame) bool {
p.preRx = append(p.preRx, incoming.Segment(len(incoming.Payload())))
return p.keep
}
func (p *recordingPolicy) PostRx(h *Handler, prevState State, accepted Frame) {
p.postRx = append(p.postRx, accepted.Segment(len(accepted.Payload())))
}
func (p *recordingPolicy) PreTx(h *Handler, outgoingOpts Frame) (Value, bool, bool) {
p.preTx++
if len(p.writeOpts) > 0 {
// Raise the offset first: Options() is sized from it.
words := uint8(5 + (len(p.writeOpts)+3)/4)
outgoingOpts.SetOffsetAndFlags(words, 0)
copy(outgoingOpts.Options(), p.writeOpts)
}
return p.rtxFrom, p.retransmit, p.holdNew
}
func (p *recordingPolicy) PostTx(h *Handler, outgoing Frame) {
offset, _ := outgoing.OffsetAndFlags()
p.postTx = append(p.postTx, txRecord{
seg: outgoing.Segment(len(outgoing.Payload())),
offset: offset,
sport: outgoing.SourcePort(),
dport: outgoing.DestinationPort(),
})
}
// TestPolicy_DisabledByDefault verifies the Handler runs normally with no Policy
// installed: the transmit and receive paths never touch a nil Policy.
func TestPolicy_DisabledByDefault(t *testing.T) {
const mtu = ethernet.MaxMTU
rng := rand.New(rand.NewSource(1))
client, server := newHandler(t, mtu, 3), newHandler(t, mtu, 3)
setupClientServer(t, rng, client, server)
var buf [mtu]byte
establish(t, client, server, buf[:]) // must not panic on nil Policy.
}
// TestPolicy_HooksInvoked verifies the Handler drives the full hook contract
// across a handshake: Reset on open, PreTx+PostTx on transmit, PreRx+PostRx on
// receive.
func TestPolicy_HooksInvoked(t *testing.T) {
const mtu = ethernet.MaxMTU
rng := rand.New(rand.NewSource(2))
client, server := newHandler(t, mtu, 3), newHandler(t, mtu, 3)
pol := newRecordingPolicy()
client.SetPolicy(pol)
setupClientServer(t, rng, client, server) // OpenActive → reset → Reset().
if pol.resets == 0 {
t.Fatal("Reset not called on open")
}
var buf [mtu]byte
establish(t, client, server, buf[:])
if pol.preTx == 0 {
t.Fatal("PreTx never called on transmit")
}
if len(pol.postTx) == 0 {
t.Fatal("PostTx never called on transmit")
}
if pol.preTx < len(pol.postTx) {
t.Fatalf("PreTx calls=%d < PostTx calls=%d: PostTx must never fire without PreTx", pol.preTx, len(pol.postTx))
}
// Client received the SYN-ACK and accepted it.
if len(pol.preRx) == 0 {
t.Fatal("PreRx never called on receive")
}
if len(pol.postRx) == 0 {
t.Fatal("PostRx never called on accepted receive")
}
// PostTx receives the segment actually emitted: the first is the SYN.
if !pol.postTx[0].seg.Flags.HasAny(FlagSYN) {
t.Fatalf("first PostTx segment flags=%s, want SYN", pol.postTx[0].seg.Flags)
}
}
// TestPolicy_PostTxSeesWrittenFrame verifies PostTx observes the fully populated
// frame — ports, sequence numbers and payload length as emitted — and not the
// frame as it stood before the segment was written into it.
func TestPolicy_PostTxSeesWrittenFrame(t *testing.T) {
const mtu = ethernet.MaxMTU
rng := rand.New(rand.NewSource(6))
client, server := newHandler(t, mtu, 3), newHandler(t, mtu, 3)
pol := newRecordingPolicy()
client.SetPolicy(pol)
setupClientServer(t, rng, client, server)
var buf [mtu]byte
establish(t, client, server, buf[:])
data := []byte("payload")
if _, err := client.Write(data); err != nil {
t.Fatal("client write:", err)
}
clear(buf[:])
n, err := client.Send(buf[:])
if err != nil {
t.Fatal("client send:", err)
}
last := pol.postTx[len(pol.postTx)-1]
wantSeg := mustSegment(t, buf[:n], n-int(last.offset)*4)
if last.seg != wantSeg {
t.Fatalf("PostTx segment=%+v, want emitted %+v", last.seg, wantSeg)
}
if int(last.seg.DATALEN) != len(data) {
t.Fatalf("PostTx DATALEN=%d, want %d", last.seg.DATALEN, len(data))
}
if last.sport != client.LocalPort() || last.dport != client.RemotePort() {
t.Fatalf("PostTx ports=%d→%d, want %d→%d", last.sport, last.dport, client.LocalPort(), client.RemotePort())
}
}
// TestPolicy_NoPostTxWithoutSegment verifies a transmit attempt that emits
// nothing still runs PreTx but never PostTx, so a Policy cannot mistake a
// no-op Send for a segment on the wire.
func TestPolicy_NoPostTxWithoutSegment(t *testing.T) {
const mtu = ethernet.MaxMTU
rng := rand.New(rand.NewSource(7))
client, server := newHandler(t, mtu, 3), newHandler(t, mtu, 3)
pol := newRecordingPolicy()
client.SetPolicy(pol)
setupClientServer(t, rng, client, server)
var buf [mtu]byte
establish(t, client, server, buf[:])
preTxBefore, postTxBefore := pol.preTx, len(pol.postTx)
n, err := client.Send(buf[:]) // Nothing queued: no segment.
if err != nil {
t.Fatal("client send:", err)
}
if n != 0 {
t.Fatalf("expected no segment, got %d bytes", n)
}
if pol.preTx != preTxBefore+1 {
t.Fatalf("PreTx calls=%d, want %d: PreTx must run on every attempt", pol.preTx, preTxBefore+1)
}
if len(pol.postTx) != postTxBefore {
t.Fatalf("PostTx calls=%d, want %d: no segment was emitted", len(pol.postTx), postTxBefore)
}
}
// TestPolicy_PreTxOptions verifies options written by PreTx survive to the wire:
// the data offset accounts for them and the payload starts after them.
func TestPolicy_PreTxOptions(t *testing.T) {
const mtu = ethernet.MaxMTU
rng := rand.New(rand.NewSource(8))
client, server := newHandler(t, mtu, 3), newHandler(t, mtu, 3)
pol := newRecordingPolicy()
client.SetPolicy(pol)
setupClientServer(t, rng, client, server)
var buf [mtu]byte
establish(t, client, server, buf[:])
// One 4-byte option word: NOP,NOP,NOP,EOL.
opts := []byte{1, 1, 1, 0}
pol.writeOpts = opts
data := []byte("payload")
if _, err := client.Write(data); err != nil {
t.Fatal("client write:", err)
}
clear(buf[:])
n, err := client.Send(buf[:])
if err != nil {
t.Fatal("client send:", err)
}
frm, err := NewFrame(buf[:n])
if err != nil {
t.Fatal("frame:", err)
}
offset, _ := frm.OffsetAndFlags()
if offset != 6 {
t.Fatalf("data offset=%d, want 6 (header + one option word)", offset)
}
if got := frm.Options(); string(got) != string(opts) {
t.Fatalf("options=%v, want %v", got, opts)
}
if got := frm.Payload(); string(got) != string(data) {
t.Fatalf("payload=%q, want %q: options must not overlap data", got, data)
}
if n != int(offset)*4+len(data) {
t.Fatalf("frame length=%d, want %d", n, int(offset)*4+len(data))
}
}
// TestPolicy_PreRxDropsSegment verifies keep=false drops the segment before the
// state machine sees it: the payload is not buffered, connection state is
// untouched and PostRx never fires.
func TestPolicy_PreRxDropsSegment(t *testing.T) {
const mtu = ethernet.MaxMTU
rng := rand.New(rand.NewSource(4))
client, server := newHandler(t, mtu, 3), newHandler(t, mtu, 3)
pol := newRecordingPolicy()
server.SetPolicy(pol)
setupClientServer(t, rng, client, server)
var buf [mtu]byte
establish(t, client, server, buf[:]) // keep=true so handshake completes.
// Now start dropping everything the server receives.
pol.keep = false
preRxBefore, postRxBefore := len(pol.preRx), len(pol.postRx)
data := []byte("dropme")
if _, err := client.Write(data); err != nil {
t.Fatal("client write:", err)
}
clear(buf[:])
n, err := client.Send(buf[:])
if err != nil {
t.Fatal("client send:", err)
}
if err := server.Recv(buf[:n]); err != nil {
t.Fatalf("dropped segment must return nil, got %v", err)
}
if len(pol.preRx) != preRxBefore+1 {
t.Fatalf("PreRx calls=%d, want %d (segment must reach PreRx)", len(pol.preRx), preRxBefore+1)
}
if len(pol.postRx) != postRxBefore {
t.Fatalf("PostRx calls=%d, want %d: a dropped segment was never accepted", len(pol.postRx), postRxBefore)
}
if server.BufferedInput() != 0 {
t.Fatalf("dropped segment must not be buffered, got %d bytes", server.BufferedInput())
}
if server.State() != StateEstablished {
t.Fatalf("dropped segment must not change state, got %s", server.State())
}
}
// TestPolicy_PreTxRetransmit verifies a PreTx retransmit directive drives
// go-back-N: the Handler rewinds the send sequence and the transmit buffer
// together and re-emits already-sent, unacknowledged data from snd.UNA.
func TestPolicy_PreTxRetransmit(t *testing.T) {
const mtu = ethernet.MaxMTU
rng := rand.New(rand.NewSource(5))
client, server := newHandler(t, mtu, 3), newHandler(t, mtu, 3)
pol := newRecordingPolicy()
client.SetPolicy(pol)
setupClientServer(t, rng, client, server)
var buf [mtu]byte
establish(t, client, server, buf[:])
// Emit one data segment; server never ACKs, so it stays unacknowledged.
data := []byte("payload")
if _, err := client.Write(data); err != nil {
t.Fatal("client write:", err)
}
clear(buf[:])
n, err := client.Send(buf[:])
if err != nil {
t.Fatal("client send data:", err)
}
if n <= sizeHeaderTCP {
t.Fatal("expected data segment")
}
firstSeg := mustSegment(t, buf[:n], n-sizeHeaderTCP)
firstData := append([]byte(nil), buf[sizeHeaderTCP:n]...)
// Direct go-back-N on the next transmit.
pol.rtxFrom, pol.retransmit = client.ControlBlock().SendUNA(), true
clear(buf[:])
n, err = client.Send(buf[:])
if err != nil {
t.Fatal("client send retransmit:", err)
}
if n <= sizeHeaderTCP {
t.Fatal("expected retransmitted data segment")
}
rtSeg := mustSegment(t, buf[:n], n-sizeHeaderTCP)
if rtSeg.SEQ != firstSeg.SEQ {
t.Fatalf("retransmit SEQ=%d, want original UNA SEQ=%d (go-back-N)", rtSeg.SEQ, firstSeg.SEQ)
}
if rtSeg.DATALEN != firstSeg.DATALEN {
t.Fatalf("retransmit DATALEN=%d, want %d", rtSeg.DATALEN, firstSeg.DATALEN)
}
if got := buf[sizeHeaderTCP:n]; string(got) != string(firstData) {
t.Fatalf("retransmit payload=%q, want %q", got, firstData)
}
}
// TestPolicy_PreTxRetransmitOutOfRange verifies an out-of-range rtxFrom is
// refused, leaving the send sequence and transmit buffer untouched.
func TestPolicy_PreTxRetransmitOutOfRange(t *testing.T) {
const mtu = ethernet.MaxMTU
rng := rand.New(rand.NewSource(9))
client, server := newHandler(t, mtu, 3), newHandler(t, mtu, 3)
pol := newRecordingPolicy()
client.SetPolicy(pol)
setupClientServer(t, rng, client, server)
var buf [mtu]byte
establish(t, client, server, buf[:])
if _, err := client.Write([]byte("payload")); err != nil {
t.Fatal("client write:", err)
}
clear(buf[:])
if _, err := client.Send(buf[:]); err != nil {
t.Fatal("client send data:", err)
}
nxtBefore := client.ControlBlock().SendNext()
// Well beyond snd.NXT: must be refused.
pol.rtxFrom, pol.retransmit = nxtBefore+1000, true
clear(buf[:])
if _, err := client.Send(buf[:]); err != nil {
t.Fatal("client send:", err)
}
if got := client.ControlBlock().SendNext(); got != nxtBefore {
t.Fatalf("snd.NXT=%d, want unchanged %d: out-of-range rtxFrom must be refused", got, nxtBefore)
}
}
// TestPolicy_HoldNew verifies holdNew suppresses new data while leaving control
// segments free to go out.
func TestPolicy_HoldNew(t *testing.T) {
const mtu = ethernet.MaxMTU
rng := rand.New(rand.NewSource(10))
client, server := newHandler(t, mtu, 3), newHandler(t, mtu, 3)
pol := newRecordingPolicy()
client.SetPolicy(pol)
setupClientServer(t, rng, client, server)
var buf [mtu]byte
establish(t, client, server, buf[:])
pol.holdNew = true
if _, err := client.Write([]byte("payload")); err != nil {
t.Fatal("client write:", err)
}
clear(buf[:])
n, err := client.Send(buf[:])
if err != nil {
t.Fatal("client send:", err)
}
if n > sizeHeaderTCP {
t.Fatalf("holdNew must suppress new data, got %d payload bytes", n-sizeHeaderTCP)
}
// Releasing the hold lets the same data out.
pol.holdNew = false
clear(buf[:])
n, err = client.Send(buf[:])
if err != nil {
t.Fatal("client send after hold:", err)
}
if n <= sizeHeaderTCP {
t.Fatal("data must flow once holdNew is cleared")
}
}
// TestPolicy_ResetOnReopen verifies Reset fires on every (re)open and on Abort,
// so a single Policy value can be reused across connection reuse.
func TestPolicy_ResetOnReopen(t *testing.T) {
const mtu = ethernet.MaxMTU
client := newHandler(t, mtu, 3)
pol := newRecordingPolicy()
client.SetPolicy(pol)
if err := client.OpenActive(1234, 5678, 0); err != nil {
t.Fatal("open 1:", err)
}
afterOpen := pol.resets
if afterOpen == 0 {
t.Fatal("Reset not called on first open")
}
client.Abort()
if pol.resets <= afterOpen {
t.Fatalf("Reset not called on Abort: resets=%d, want >%d", pol.resets, afterOpen)
}
afterAbort := pol.resets
if err := client.OpenActive(1234, 5678, 0); err != nil {
t.Fatal("open 2:", err)
}
if pol.resets <= afterAbort {
t.Fatalf("Reset not called on reopen: resets=%d, want >%d", pol.resets, afterAbort)
}
}
+113 -47
View File
@@ -1,6 +1,11 @@
package tcp
package rto
import "time"
import (
"time"
"github.com/soypat/lneto"
"github.com/soypat/lneto/tcp"
)
// RFC 6298 retransmission-timeout (RTO) parameters. The algorithm keeps a
// single retransmission timer per connection (RFC 6298 §5): the timer is
@@ -30,61 +35,80 @@ const (
backoffMax = 12
)
// RTO implements the RFC 6298 round-trip-time estimator and the single
// retransmission timer as a [LossRecovery]. Construct it with new(RTO) and hand
// it to [ConnConfig.LossRecovery]; the connection calls [RTO.Reset] on open, so
// the zero value is ready to use.
// Timer implements the RFC 6298 round-trip-time estimator and the single
// retransmission timer as a [tcp.Policy]. Construct it with [NewTimer] and hand
// it to [tcp.ConnConfig.Policy].
//
// RTO is a pure, reactive state machine: it observes the segments a connection
// sends and receives (via the LossRecovery hooks) and the monotonic time handed
// in at each hook, and from those alone derives RTT estimates and retransmission
// decisions. It holds no clock and allocates nothing, which keeps it
// deterministic for unit testing (see issue #140).
// Timer is a pure, reactive state machine: it observes the segments a connection
// sends and receives (via the tcp.Policy hooks) and from those alone derives RTT
// estimates and retransmission decisions. The tcp package holds no clock, so the
// Timer carries its own; injecting it keeps the estimator deterministic for unit
// testing (see issue #140).
//
// RTO tracks its own shadow of the send sequence space purely from the segments
// it observes: [RTO.PostTx] advances the highest sequence sent and [RTO.PreRx]
// Timer tracks its own shadow of the send sequence space purely from the segments
// it observes: [Timer.PostTx] advances the highest sequence sent and [Timer.PreRx]
// advances the highest sequence acknowledged. This is what lets it manage the
// timer (RFC 6298 §5.2/§5.3) without reaching into the tcp state machine, and it
// is also how retransmissions are distinguished for Karn's algorithm — a segment
// whose sequence space is not beyond the shadow snd.NXT is a retransmission and
// is never RTT-sampled.
type RTO struct {
type Timer struct {
// nanotime is the monotonic time source in nanoseconds. Preserved by Reset.
nanotime func() int64
srtt time.Duration // smoothed round-trip time (SRTT).
rttvar time.Duration // round-trip-time variation (RTTVAR).
rto time.Duration // current retransmission timeout.
haveRTT bool // false until the first RTT sample is taken.
// Shadow of the send sequence space, derived from observed segments.
haveSeq bool // false until the first data segment is observed.
sndUNA Value // highest acknowledged sequence number seen on the wire.
sndNXT Value // one past the highest sequence number sent.
haveSeq bool // false until the first data segment is observed.
sndUNA tcp.Value // highest acknowledged sequence number seen on the wire.
sndNXT tcp.Value // one past the highest sequence number sent.
// RTT sampling state (Karn's algorithm, RFC 6298 §3): at most one segment is
// timed at a time and retransmitted segments are never sampled.
timing bool
timedSeq Value // ACK at or beyond this value completes the sample.
timedAt int64 // send time (monotonic ns) of the timed segment.
timedSeq tcp.Value // ACK at or beyond this value completes the sample.
timedAt int64 // send time (monotonic ns) of the timed segment.
// Retransmission timer state.
running bool
deadline int64 // time (monotonic ns) at which the timer expires.
backoff uint8 // consecutive timeouts, for exponential backoff.
// expirations counts timeouts since Reset. It exists so a policy sharing this
// timer can notice a timeout it did not itself drive: a congestion controller
// must collapse its window on one, and a policy that composes the timer as a
// peer never sees the timer's own directive.
expirations uint32
}
var _ LossRecovery = (*RTO)(nil)
var _ tcp.Policy = (*Timer)(nil)
// Reset returns the estimator to its pre-connection state with the initial RTO.
// It implements [LossRecovery] and is called when the connection opens or aborts
// so the estimator can be reused across connection reuse.
func (r *RTO) Reset() { *r = RTO{rto: rtoInitial} }
// Configure prepares the Timer for use with nanotime, the monotonic time source
// in nanoseconds (the func() int64 convention used across lneto). It must be
// called before the connection is opened.
func (r *Timer) Configure(nanotime func() int64) error {
if nanotime == nil {
return lneto.ErrMissingHALConfig // The estimator cannot run without a clock.
}
*r = Timer{rto: rtoInitial, nanotime: nanotime}
return nil
}
// Reset returns the estimator to its pre-connection state with the initial RTO,
// preserving the configured clock. It implements [tcp.Policy] and is called when
// the connection opens or aborts so the estimator survives connection reuse.
func (r *Timer) Reset() { *r = Timer{rto: rtoInitial, nanotime: r.nanotime} }
// SmoothedRTT returns the current smoothed round-trip time (SRTT), or zero
// before the first RTT measurement. It is concrete-type introspection and is
// intentionally not part of [LossRecovery].
func (r *RTO) SmoothedRTT() time.Duration { return r.srtt }
// intentionally not part of [tcp.Policy].
func (r *Timer) SmoothedRTT() time.Duration { return r.srtt }
// CurrentRTO returns the timeout currently in effect, clamped to [rtoMin, rtoMax].
func (r *RTO) CurrentRTO() time.Duration {
func (r *Timer) CurrentRTO() time.Duration {
rto := r.rto
if rto < rtoMin {
rto = rtoMin
@@ -95,23 +119,46 @@ func (r *RTO) CurrentRTO() time.Duration {
}
// Running reports whether the retransmission timer is currently armed.
func (r *RTO) Running() bool { return r.running }
func (r *Timer) Running() bool { return r.running }
// Expirations returns how many times the retransmission timer has expired since
// [Timer.Reset]. A policy that shares this timer rather than driving it watches
// this for a change to learn that a timeout happened, since it never sees the
// timer's own directive. It is concrete-type introspection and is intentionally
// not part of [tcp.Policy].
func (r *Timer) Expirations() uint32 { return r.expirations }
// NextDeadline returns the monotonic-nanosecond instant at which the timer
// expires, or 0 when it is not armed. It implements [LossRecovery].
func (r *RTO) NextDeadline() int64 {
// expires, or 0 when it is not armed. It is concrete-type introspection, not
// part of [tcp.Policy]: an event loop that wants to schedule against the RTO
// holds the Timer it configured and reads this.
func (r *Timer) NextDeadline() int64 {
if !r.running {
return 0
}
return r.deadline
}
// PreRx samples the RTT and manages the retransmission timer from a received
// segment (RFC 6298 §5.2/§5.3). It implements [LossRecovery] and always keeps
// the segment (the estimator never drops traffic).
func (r *RTO) PreRx(incoming Segment, now int64) RxDirective {
if !r.haveSeq || !incoming.Flags.HasAny(FlagACK) {
return RxDirective{Keep: true}
// PreRx keeps every segment: the estimator never drops traffic and records
// nothing before the connection has decided whether the segment counts. It
// implements [tcp.Policy].
func (r *Timer) PreRx(h *tcp.Handler, incoming tcp.Frame) bool {
return true
}
// PostRx samples the RTT and manages the retransmission timer from a segment the
// connection accepted (RFC 6298 §5.2/§5.3). It implements [tcp.Policy].
//
// Only accepted segments reach here. Acting on a refused one would let an
// acknowledgement the state machine rejected, for data never sent, collapse the
// backoff and take a bogus RTT sample.
func (r *Timer) PostRx(h *tcp.Handler, prevState tcp.State, accepted tcp.Frame) {
r.postRx(accepted.Segment(len(accepted.Payload())), r.nanotime())
}
func (r *Timer) postRx(incoming tcp.Segment, now int64) {
if !r.haveSeq || !incoming.Flags.HasAny(tcp.FlagACK) {
return
}
ack := incoming.ACK
if r.timing && !ack.LessThan(r.timedSeq) {
@@ -132,18 +179,22 @@ func (r *RTO) PreRx(incoming Segment, now int64) RxDirective {
r.running = true
r.deadline = now + int64(r.CurrentRTO())
}
return RxDirective{Keep: true}
}
// PreTx reports whether the retransmission timer has expired and, if so, applies
// the RFC 6298 §5.4–§5.6 timeout response — discard the outstanding RTT sample
// (Karn), back the RTO off exponentially and restart the timer — returning a
// directive that asks the connection to retransmit from snd.UNA (go-back-N). It
// implements [LossRecovery].
func (r *RTO) PreTx(now int64) TxDirective {
// (Karn), back the RTO off exponentially and restart the timer — and asks the
// connection to retransmit from snd.UNA (go-back-N). It writes no TCP options:
// retransmission timing needs none of its own. It implements [tcp.Policy].
func (r *Timer) PreTx(h *tcp.Handler, outgoingOpts tcp.Frame) (rtxFrom tcp.Value, retransmit, holdNew bool) {
return r.preTx(r.nanotime(), h.ControlBlock().SendUNA())
}
func (r *Timer) preTx(now int64, una tcp.Value) (rtxFrom tcp.Value, retransmit, holdNew bool) {
if !r.running || now < r.deadline || r.sndUNA == r.sndNXT {
return TxDirective{}
return 0, false, false
}
r.expirations++
r.timing = false // §5.4: do not sample a retransmitted segment.
if r.backoff < backoffMax {
r.backoff++
@@ -151,20 +202,24 @@ func (r *RTO) PreTx(now int64) TxDirective {
}
r.running = true
r.deadline = now + int64(r.CurrentRTO())
return TxDirective{RetransmitAll: true}
return una, true, false
}
// PostTx records an emitted segment: it advances the shadow send sequence,
// begins timing newly transmitted data (RFC 6298 §3) and arms the timer (§5.1).
// Segments that do not extend the send sequence are retransmissions and are
// never RTT-sampled (Karn's algorithm). Control-only segments (no data) are
// ignored. It implements [LossRecovery].
func (r *RTO) PostTx(outgoing Segment, now int64) {
// ignored. It implements [tcp.Policy].
func (r *Timer) PostTx(h *tcp.Handler, outgoing tcp.Frame) {
r.postTx(outgoing.Segment(len(outgoing.Payload())), r.nanotime())
}
func (r *Timer) postTx(outgoing tcp.Segment, now int64) {
if outgoing.DATALEN == 0 {
return // only data segments are timed / arm the RTO.
}
segStart := outgoing.SEQ
segEnd := segStart + Value(outgoing.LEN())
segEnd := segStart + tcp.Value(outgoing.LEN())
if !r.haveSeq {
r.haveSeq = true
r.sndUNA = segStart
@@ -189,9 +244,20 @@ func (r *RTO) PostTx(outgoing Segment, now int64) {
}
}
// ObserveRTT folds a round-trip measurement taken by other means into the
// estimator, for a policy that composes this timer and can measure the round trip
// more accurately than acknowledgement timing allows. The RFC 7323 timestamp echo
// is the case this exists for.
//
// Unlike the timer's own sampling this does not apply Karn's algorithm, because a
// sample derived from an echoed timestamp is unambiguous even when the segment
// carrying it was a retransmission (RFC 7323 §4.1). Non-positive samples are
// ignored.
func (r *Timer) ObserveRTT(rtt time.Duration) { r.updateRTT(rtt) }
// updateRTT folds a round-trip measurement into SRTT/RTTVAR/RTO using the
// integer-shift form of RFC 6298 §2.2/§2.3.
func (r *RTO) updateRTT(sample time.Duration) {
func (r *Timer) updateRTT(sample time.Duration) {
if sample <= 0 {
return
}
+311
View File
@@ -0,0 +1,311 @@
package rto
import (
"testing"
"time"
"github.com/soypat/lneto/tcp"
)
const rtoMs = int64(time.Millisecond)
// dataSeg builds a data segment of datalen octets starting at seq.
func dataSeg(seq uint32, datalen int) tcp.Segment {
return tcp.Segment{SEQ: tcp.Value(seq), DATALEN: tcp.Size(datalen), Flags: tcp.FlagPSH | tcp.FlagACK}
}
// ackSeg builds a bare ACK acknowledging up to ack.
func ackSeg(ack uint32) tcp.Segment {
return tcp.Segment{ACK: tcp.Value(ack), Flags: tcp.FlagACK}
}
func newRTO() *Timer {
var r Timer
if err := r.Configure(func() int64 { return 0 }); err != nil {
panic(err)
}
return &r
}
// frameOf renders a segment as the wire frame the [tcp.Policy] hooks receive.
func frameOf(t *testing.T, s tcp.Segment) tcp.Frame {
t.Helper()
frm, err := tcp.NewFrame(make([]byte, 20+int(s.DATALEN)))
if err != nil {
t.Fatal(err)
}
frm.SetSegment(s, 5)
return frm
}
func TestRTO_Configure(t *testing.T) {
var r Timer
if err := r.Configure(nil); err == nil {
t.Error("Configure must reject a nil clock")
}
if err := r.Configure(func() int64 { return 0 }); err != nil {
t.Fatal(err)
}
if r.nanotime == nil {
t.Fatal("clock not stored")
}
r.Reset()
if r.nanotime == nil {
t.Error("Reset must preserve the configured clock")
}
}
func TestRTO_Reset(t *testing.T) {
r := newRTO()
r.Reset()
if r.rto != rtoInitial {
t.Errorf("initial rto=%v, want %v", r.rto, rtoInitial)
}
if r.CurrentRTO() != rtoInitial {
t.Errorf("CurrentRTO=%v, want %v", r.CurrentRTO(), rtoInitial)
}
if r.haveRTT {
t.Error("haveRTT should be false before first sample")
}
if r.Running() || r.NextDeadline() != 0 {
t.Error("timer must be disarmed after Reset")
}
}
// TestRTO_ArmOnSendSampleOnAck sends data, verifies the timer arms, then acks it
// and verifies an RTT sample is taken and the timer stops once all data is acked.
func TestRTO_ArmOnSendSampleOnAck(t *testing.T) {
r := newRTO()
const iss = uint32(1000)
r.postTx(dataSeg(iss, 100), 0)
if !r.Running() {
t.Fatal("timer must arm after sending data")
}
if r.NextDeadline() != int64(rtoInitial) {
t.Errorf("deadline=%d, want %d", r.NextDeadline(), int64(rtoInitial))
}
// ACK arrives one RTT (40ms) later covering all sent data.
if !r.PreRx(nil, frameOf(t, ackSeg(iss+100))) {
t.Error("PreRx must keep the segment")
}
r.postRx(ackSeg(iss+100), 40*rtoMs)
if r.Running() {
t.Error("timer must stop once all data is acknowledged")
}
if r.SmoothedRTT() != 40*time.Millisecond {
t.Errorf("srtt=%v, want 40ms", r.SmoothedRTT())
}
}
// TestRTO_RetransmitOnTimeout verifies PreTx directs a go-back-N retransmit once
// the deadline passes with data outstanding, and backs the RTO off.
func TestRTO_RetransmitOnTimeout(t *testing.T) {
r := newRTO()
const iss = uint32(1000)
r.postTx(dataSeg(iss, 100), 0)
if _, rtx, _ := r.preTx(int64(rtoInitial)-1, tcp.Value(iss)); rtx {
t.Fatal("must not retransmit before the deadline")
}
from, rtx, hold := r.preTx(int64(rtoInitial), tcp.Value(iss))
if !rtx {
t.Fatal("RTO must fire at the deadline with data outstanding")
}
if hold {
t.Error("the estimator never holds new data back")
}
if from != tcp.Value(iss) {
t.Errorf("retransmit from %d, want snd.UNA=%d", from, iss)
}
if r.CurrentRTO() != 2*rtoInitial {
t.Errorf("rto=%v after one backoff, want %v", r.CurrentRTO(), 2*rtoInitial)
}
// The connection resends from snd.UNA; postTx sees a retransmission.
r.postTx(dataSeg(iss, 100), int64(rtoInitial))
if r.timing {
t.Error("retransmitted segment must not be RTT-sampled (Karn)")
}
}
// TestRTO_KarnNoSampleOnRetransmittedAck verifies that after a retransmission the
// ACK does not produce an RTT sample (Karn's algorithm).
func TestRTO_KarnNoSampleOnRetransmittedAck(t *testing.T) {
r := newRTO()
const iss = uint32(1000)
r.postTx(dataSeg(iss, 100), 0)
// Timeout and retransmit.
r.preTx(int64(rtoInitial), tcp.Value(iss))
r.postTx(dataSeg(iss, 100), int64(rtoInitial))
// ACK now arrives; no sample should be taken since timing was discarded.
r.postRx(ackSeg(iss+100), int64(rtoInitial)+10*rtoMs)
if r.haveRTT {
t.Error("no RTT sample should exist after a retransmission (Karn)")
}
}
// TestRTO_TimerRestartsWhilePartiallyAcked verifies the timer restarts (not
// stops) when an ACK advances UNA but data remains in flight (RFC 6298 §5.3).
func TestRTO_TimerRestartsWhilePartiallyAcked(t *testing.T) {
r := newRTO()
const iss = uint32(1000)
r.postTx(dataSeg(iss, 100), 0)
r.postTx(dataSeg(iss+100, 100), 0) // 200 octets outstanding, iss..iss+200.
r.postRx(ackSeg(iss+100), 40*rtoMs) // acks first 100 only.
if !r.Running() {
t.Fatal("timer must remain armed while data is still in flight")
}
if r.NextDeadline() != 40*rtoMs+int64(r.CurrentRTO()) {
t.Errorf("deadline=%d, want %d", r.NextDeadline(), 40*rtoMs+int64(r.CurrentRTO()))
}
}
// TestRTO_NoArmWithoutData verifies control-only segments neither arm the timer
// nor start an RTT sample.
func TestRTO_NoArmWithoutData(t *testing.T) {
r := newRTO()
r.postTx(tcp.Segment{SEQ: 1000, Flags: tcp.FlagACK}, 0) // pure ACK, DATALEN==0.
if r.Running() || r.timing {
t.Error("pure control segment must not arm the timer or start a sample")
}
}
// TestRTO_BackoffCollapsesOnValidSample verifies a valid RTT measurement
// collapses the exponential backoff counter (RFC 6298 §5.7).
func TestRTO_BackoffCollapsesOnValidSample(t *testing.T) {
r := newRTO()
const iss = uint32(1000)
r.postTx(dataSeg(iss, 100), 0)
r.preTx(int64(rtoInitial), tcp.Value(iss)) // one timeout: backoff=1.
r.postTx(dataSeg(iss, 100), int64(rtoInitial)) // retransmit (no sample).
if r.backoff != 1 {
t.Fatalf("backoff=%d, want 1 after a timeout", r.backoff)
}
// New data sent and freshly sampled, then acked.
r.postTx(dataSeg(iss+100, 100), int64(rtoInitial)+rtoMs)
r.postRx(ackSeg(iss+200), int64(rtoInitial)+30*rtoMs)
if r.backoff != 0 {
t.Errorf("backoff=%d, want 0 after a valid RTT sample", r.backoff)
}
}
// TestRTO_Clamped verifies CurrentRTO is clamped to [rtoMin, rtoMax].
func TestRTO_Clamped(t *testing.T) {
r := newRTO()
r.rto = time.Nanosecond
if got := r.CurrentRTO(); got != rtoMin {
t.Errorf("CurrentRTO=%v, want floor %v", got, rtoMin)
}
r.rto = time.Hour
if got := r.CurrentRTO(); got != rtoMax {
t.Errorf("CurrentRTO=%v, want ceiling %v", got, rtoMax)
}
}
// TestRTO_UpdateRTTFirstSample verifies the first-measurement initialization of
// SRTT/RTTVAR (RFC 6298 §2.2).
func TestRTO_UpdateRTTFirstSample(t *testing.T) {
r := newRTO()
r.updateRTT(100 * time.Millisecond)
if r.srtt != 100*time.Millisecond {
t.Errorf("srtt=%v, want 100ms", r.srtt)
}
if r.rttvar != 50*time.Millisecond {
t.Errorf("rttvar=%v, want 50ms", r.rttvar)
}
// RTO = SRTT + K*RTTVAR = 100 + 4*50 = 300ms.
if r.rto != 300*time.Millisecond {
t.Errorf("rto=%v, want 300ms", r.rto)
}
}
// TestRTO_PolicyHooksDeriveFromFrame exercises Timer through the [tcp.Policy]
// hooks, verifying it reads the segment out of the frame it is handed: sending
// data arms a deadline and a full ACK disarms it and yields the RTT sample.
func TestRTO_PolicyHooksDeriveFromFrame(t *testing.T) {
var clock int64
var r Timer
if err := r.Configure(func() int64 { return clock }); err != nil {
t.Fatal(err)
}
var pol tcp.Policy = &r
pol.Reset()
pol.PostTx(nil, frameOf(t, dataSeg(1000, 100)))
if r.NextDeadline() == 0 {
t.Fatal("expected an armed deadline after sending data")
}
clock = 10 * rtoMs
if !pol.PreRx(nil, frameOf(t, ackSeg(1100))) {
t.Error("PreRx must keep")
}
pol.PostRx(nil, tcp.StateEstablished, frameOf(t, ackSeg(1100)))
if r.NextDeadline() != 0 {
t.Error("expected disarmed timer after full ack")
}
if r.SmoothedRTT() != 10*time.Millisecond {
t.Errorf("srtt=%v, want 10ms sampled through the hooks", r.SmoothedRTT())
}
}
// TestRTO_PreRxNeverDrops verifies the estimator keeps every segment and records
// nothing at PreRx time. Dropping is not its business, and the connection has not
// yet judged the segment: an acknowledgement for data never sent would otherwise
// collapse the backoff and take a bogus round-trip sample. Only accepted segments
// reach PostRx, which the Handler guarantees.
func TestRTO_PreRxNeverDrops(t *testing.T) {
r := newRTO()
const iss = uint32(1000)
r.postTx(dataSeg(iss, 100), 0)
armed := r.NextDeadline()
if armed == 0 {
t.Fatal("timer must be armed after sending data")
}
// An acknowledgement far beyond anything sent, which the connection refuses.
if !r.PreRx(nil, frameOf(t, ackSeg(iss+100000))) {
t.Error("PreRx must keep: dropping is not the estimator's business")
}
if r.NextDeadline() != armed {
t.Errorf("deadline moved to %d at PreRx, want it left at %d", r.NextDeadline(), armed)
}
if r.SmoothedRTT() != 0 {
t.Errorf("took an RTT sample of %v at PreRx", r.SmoothedRTT())
}
if !r.Running() {
t.Error("timer disarmed at PreRx")
}
}
// TestRTO_RetransmitsZeroWindowProbe verifies the timer takes over the periodic
// probing of a closed send window. A zero-window probe is a single octet the peer
// cannot accept, so it goes unacknowledged; the timer must keep resending it, with
// exponential backoff, which is the persist-timer behaviour of RFC 9293 §3.8.6.1.
// The tcp package relies on this and refuses to probe without a policy installed.
func TestRTO_RetransmitsZeroWindowProbe(t *testing.T) {
r := newRTO()
const iss = uint32(5000)
probe := dataSeg(iss, 1) // The one-octet probe.
r.postTx(probe, 0)
now := int64(rtoInitial)
prevRTO := r.CurrentRTO()
for attempt := 1; attempt <= 4; attempt++ {
from, rtx, _ := r.preTx(now, tcp.Value(iss))
if !rtx {
t.Fatalf("attempt %d: timer did not fire; the probe would never be resent", attempt)
}
if from != tcp.Value(iss) {
t.Errorf("attempt %d: retransmit from %d, want the probe octet at %d", attempt, from, iss)
}
if got := r.CurrentRTO(); got <= prevRTO {
t.Errorf("attempt %d: rto %v did not back off past %v", attempt, got, prevRTO)
}
prevRTO = r.CurrentRTO()
// The peer still cannot accept the octet, so it stays unacknowledged.
r.postTx(probe, now)
now += int64(prevRTO)
}
}
-206
View File
@@ -1,206 +0,0 @@
package tcp
import (
"testing"
"time"
)
const rtoMs = int64(time.Millisecond)
// rtoDataSeg builds a data segment of datalen octets starting at seq.
func rtoDataSeg(seq uint32, datalen int) Segment {
return Segment{SEQ: Value(seq), DATALEN: Size(datalen), Flags: FlagPSH | FlagACK}
}
// rtoAckSeg builds a bare ACK acknowledging up to ack.
func rtoAckSeg(ack uint32) Segment {
return Segment{ACK: Value(ack), Flags: FlagACK}
}
func newRTO() *RTO {
var r RTO
r.Reset()
return &r
}
func TestRTO_Reset(t *testing.T) {
var r RTO
r.Reset()
if r.rto != rtoInitial {
t.Errorf("initial rto=%v, want %v", r.rto, rtoInitial)
}
if r.CurrentRTO() != rtoInitial {
t.Errorf("CurrentRTO=%v, want %v", r.CurrentRTO(), rtoInitial)
}
if r.haveRTT {
t.Error("haveRTT should be false before first sample")
}
if r.Running() || r.NextDeadline() != 0 {
t.Error("timer must be disarmed after Reset")
}
}
// TestRTO_ArmOnSendSampleOnAck sends data, verifies the timer arms, then acks it
// and verifies an RTT sample is taken and the timer stops once all data is acked.
func TestRTO_ArmOnSendSampleOnAck(t *testing.T) {
r := newRTO()
const iss = uint32(1000)
r.PostTx(rtoDataSeg(iss, 100), 0)
if !r.Running() {
t.Fatal("timer must arm after sending data")
}
if r.NextDeadline() != int64(rtoInitial) {
t.Errorf("deadline=%d, want %d", r.NextDeadline(), int64(rtoInitial))
}
// ACK arrives one RTT (40ms) later covering all sent data.
dir := r.PreRx(rtoAckSeg(iss+100), 40*rtoMs)
if !dir.Keep {
t.Error("PreRx must keep the segment")
}
if r.Running() {
t.Error("timer must stop once all data is acknowledged")
}
if r.SmoothedRTT() != 40*time.Millisecond {
t.Errorf("srtt=%v, want 40ms", r.SmoothedRTT())
}
}
// TestRTO_RetransmitOnTimeout verifies PreTx directs a go-back-N retransmit once
// the deadline passes with data outstanding, and backs the RTO off.
func TestRTO_RetransmitOnTimeout(t *testing.T) {
r := newRTO()
const iss = uint32(1000)
r.PostTx(rtoDataSeg(iss, 100), 0)
if r.PreTx(int64(rtoInitial) - 1).RetransmitAll {
t.Fatal("must not retransmit before the deadline")
}
dir := r.PreTx(int64(rtoInitial))
if !dir.RetransmitAll {
t.Fatal("RTO must fire at the deadline with data outstanding")
}
if r.CurrentRTO() != 2*rtoInitial {
t.Errorf("rto=%v after one backoff, want %v", r.CurrentRTO(), 2*rtoInitial)
}
// The connection resends from snd.UNA; PostTx sees a retransmission.
r.PostTx(rtoDataSeg(iss, 100), int64(rtoInitial))
if r.timing {
t.Error("retransmitted segment must not be RTT-sampled (Karn)")
}
}
// TestRTO_KarnNoSampleOnRetransmittedAck verifies that after a retransmission the
// ACK does not produce an RTT sample (Karn's algorithm).
func TestRTO_KarnNoSampleOnRetransmittedAck(t *testing.T) {
r := newRTO()
const iss = uint32(1000)
r.PostTx(rtoDataSeg(iss, 100), 0)
// Timeout and retransmit.
r.PreTx(int64(rtoInitial))
r.PostTx(rtoDataSeg(iss, 100), int64(rtoInitial))
// ACK now arrives; no sample should be taken since timing was discarded.
r.PreRx(rtoAckSeg(iss+100), int64(rtoInitial)+10*rtoMs)
if r.haveRTT {
t.Error("no RTT sample should exist after a retransmission (Karn)")
}
}
// TestRTO_TimerRestartsWhilePartiallyAcked verifies the timer restarts (not
// stops) when an ACK advances UNA but data remains in flight (RFC 6298 §5.3).
func TestRTO_TimerRestartsWhilePartiallyAcked(t *testing.T) {
r := newRTO()
const iss = uint32(1000)
r.PostTx(rtoDataSeg(iss, 100), 0)
r.PostTx(rtoDataSeg(iss+100, 100), 0) // 200 octets outstanding, iss..iss+200.
dir := r.PreRx(rtoAckSeg(iss+100), 40*rtoMs) // acks first 100 only.
if !r.Running() {
t.Fatal("timer must remain armed while data is still in flight")
}
if r.NextDeadline() != 40*rtoMs+int64(r.CurrentRTO()) {
t.Errorf("deadline=%d, want %d", r.NextDeadline(), 40*rtoMs+int64(r.CurrentRTO()))
}
if !dir.Keep {
t.Error("PreRx must keep the segment")
}
}
// TestRTO_NoArmWithoutData verifies control-only segments neither arm the timer
// nor start an RTT sample.
func TestRTO_NoArmWithoutData(t *testing.T) {
r := newRTO()
r.PostTx(Segment{SEQ: 1000, Flags: FlagACK}, 0) // pure ACK, DATALEN==0.
if r.Running() || r.timing {
t.Error("pure control segment must not arm the timer or start a sample")
}
}
// TestRTO_BackoffCollapsesOnValidSample verifies a valid RTT measurement
// collapses the exponential backoff counter (RFC 6298 §5.7).
func TestRTO_BackoffCollapsesOnValidSample(t *testing.T) {
r := newRTO()
const iss = uint32(1000)
r.PostTx(rtoDataSeg(iss, 100), 0)
r.PreTx(int64(rtoInitial)) // one timeout: backoff=1.
r.PostTx(rtoDataSeg(iss, 100), int64(rtoInitial)) // retransmit (no sample).
if r.backoff != 1 {
t.Fatalf("backoff=%d, want 1 after a timeout", r.backoff)
}
// New data sent and freshly sampled, then acked.
r.PostTx(rtoDataSeg(iss+100, 100), int64(rtoInitial)+rtoMs)
r.PreRx(rtoAckSeg(iss+200), int64(rtoInitial)+30*rtoMs)
if r.backoff != 0 {
t.Errorf("backoff=%d, want 0 after a valid RTT sample", r.backoff)
}
}
// TestRTO_Clamped verifies CurrentRTO is clamped to [rtoMin, rtoMax].
func TestRTO_Clamped(t *testing.T) {
var r RTO
r.Reset()
r.rto = time.Nanosecond
if got := r.CurrentRTO(); got != rtoMin {
t.Errorf("CurrentRTO=%v, want floor %v", got, rtoMin)
}
r.rto = time.Hour
if got := r.CurrentRTO(); got != rtoMax {
t.Errorf("CurrentRTO=%v, want ceiling %v", got, rtoMax)
}
}
// TestRTO_UpdateRTTFirstSample verifies the first-measurement initialization of
// SRTT/RTTVAR (RFC 6298 §2.2).
func TestRTO_UpdateRTTFirstSample(t *testing.T) {
var r RTO
r.Reset()
r.updateRTT(100 * time.Millisecond)
if r.srtt != 100*time.Millisecond {
t.Errorf("srtt=%v, want 100ms", r.srtt)
}
if r.rttvar != 50*time.Millisecond {
t.Errorf("rttvar=%v, want 50ms", r.rttvar)
}
// RTO = SRTT + K*RTTVAR = 100 + 4*50 = 300ms.
if r.rto != 300*time.Millisecond {
t.Errorf("rto=%v, want 300ms", r.rto)
}
}
// TestRTO_ImplementsLossRecovery exercises RTO through the [LossRecovery]
// interface: sending data arms a deadline and a full ACK disarms it.
func TestRTO_ImplementsLossRecovery(t *testing.T) {
var lr LossRecovery = newRTO()
lr.Reset()
lr.PostTx(rtoDataSeg(1000, 100), 0)
if lr.NextDeadline() == 0 {
t.Error("expected an armed deadline after sending data")
}
if !lr.PreRx(rtoAckSeg(1100), 10*rtoMs).Keep {
t.Error("PreRx must keep")
}
if lr.NextDeadline() != 0 {
t.Error("expected disarmed timer after full ack")
}
}
+62 -11
View File
@@ -227,18 +227,39 @@ func (rtx *ringTx) RetransmitFromUNA() {
if oldest == nil {
return // Nothing in the retransmission queue.
}
unaSeq := oldest.seq
if rtx.sentend != 0 {
// Merge sent region [sentoff, sentend) back into unsent.
rtx.unsentoff = rtx.sentoff
if rtx.unsentend == 0 {
rtx.unsentend = rtx.sentend
}
rtx.sentoff = 0
rtx.sentend = 0
rtx.RetransmitFrom(oldest.seq)
}
// RetransmitFrom rewinds the transmit queue so sent-but-unacked data at and
// after seq becomes unsent again; the next MakePacket calls re-send it. seq is
// snapped down to the start of the packet containing it — the retransmission
// queue tracks whole packets, so sub-packet rewind is not representable. It is
// a no-op when seq is not covered by any queued packet (nothing to resend).
//
// Callers must pair this with [ControlBlock.RetransmitFrom] using the same seq
// so the send sequence space and the transmit buffer rewind together.
func (rtx *ringTx) RetransmitFrom(seq Value) {
pkt := rtx.slist.packetContaining(seq)
if pkt == nil {
return // seq not in the retransmission queue.
}
// Clear packet metadata; sequence tracking restarts from UNA.
rtx.slist.Reset(cap(rtx.slist.pkts), unaSeq)
rewindOff, rewindSeq := pkt.off, pkt.seq
// The write position is unsentend, except when the unsent region is empty
// (unsentend==0) in which case data ends where the sent region ends. Capture
// it before reopening the unsent region over the rewound packets.
writeEnd := rtx.unsentend
if writeEnd == 0 {
writeEnd = rtx.sentend
}
if rewindOff == rtx.sentoff {
rtx.sentoff = 0 // Whole queue rewound: sent region becomes empty.
rtx.sentend = 0
} else {
rtx.sentend = rewindOff
}
rtx.unsentoff = rewindOff
rtx.unsentend = writeEnd
rtx.slist.truncateFrom(rewindSeq)
}
func (rtx *ringTx) consolidateBufs() {
@@ -331,6 +352,36 @@ func (sl *sentlist) Free() int {
return cap(sl.pkts) - len(sl.pkts)
}
// packetContaining returns the queued packet whose sequence range covers seq, or
// nil when no packet does. It is the floor lookup a retransmission rewind needs:
// seq lands inside a packet and the whole packet is resent.
func (sl *sentlist) packetContaining(seq Value) *ringidx {
for i := range sl.pkts {
pkt := &sl.pkts[i]
if pkt.seq.LessThanEq(seq) && seq.LessThan(pkt.endSeq()) {
return pkt
}
}
return nil
}
// truncateFrom drops the packet starting at seq and every packet sent after it,
// so their data can be re-queued as unsent. seq must be a packet start sequence
// (see [sentlist.packetContaining]). When no packet survives, the auxiliary
// sequence counter is rewound to seq so [sentlist.EndSeq] keeps reporting where
// the next packet begins.
func (sl *sentlist) truncateFrom(seq Value) {
for i := range sl.pkts {
if sl.pkts[i].seq == seq {
sl.pkts = sl.pkts[:i]
if i == 0 {
sl.ssn = seq
}
return
}
}
}
func (sl *sentlist) AddPacket(datalen, off, bufsize int, seq Value) *ringidx {
free := sl.Free()
if free == 0 {
+215
View File
@@ -0,0 +1,215 @@
package tcp
import (
"bytes"
"testing"
)
// newRetransmitQueue builds a queue holding npkt sent packets of pktlen octets
// each, starting at iss, plus any leftover unsent data. It returns the queue and
// the full byte stream that was written.
func newRetransmitQueue(t *testing.T, bufsize, maxPkts, npkt, pktlen, unsent int, iss Value) (*ringTx, []byte) {
t.Helper()
var rtx ringTx
if err := rtx.Reset(make([]byte, bufsize), maxPkts, iss); err != nil {
t.Fatal(err)
}
stream := make([]byte, npkt*pktlen+unsent)
for i := range stream {
stream[i] = byte(i + 1) // Non-zero so a stale ring shows up as a mismatch.
}
if n, err := rtx.Write(stream); err != nil || n != len(stream) {
t.Fatalf("write n=%d err=%v", n, err)
}
seq := iss
scratch := make([]byte, pktlen)
for i := range npkt {
n, err := rtx.MakePacket(scratch, seq)
if err != nil {
t.Fatalf("packet %d: %v", i, err)
}
if n != pktlen {
t.Fatalf("packet %d: n=%d, want %d", i, n, pktlen)
}
seq += Value(n)
}
testQueueSanity(t, &rtx)
return &rtx, stream
}
// mustRemake asserts the queue re-emits datalen octets at seq matching want.
func mustRemake(t *testing.T, rtx *ringTx, seq Value, want []byte) {
t.Helper()
got := make([]byte, len(want))
n, err := rtx.MakePacket(got, seq)
if err != nil {
t.Fatalf("MakePacket at seq %d: %v", seq, err)
}
if n != len(want) {
t.Fatalf("MakePacket at seq %d: n=%d, want %d", seq, n, len(want))
}
if !bytes.Equal(got, want) {
t.Fatalf("MakePacket at seq %d: got %v, want %v", seq, got, want)
}
}
// TestRingTx_RetransmitFromBoundary rewinds to the start of the second of three
// sent packets: the first stays sent, the rest become unsent and re-emit their
// original bytes.
func TestRingTx_RetransmitFromBoundary(t *testing.T) {
const iss, pktlen = Value(100), 4
rtx, stream := newRetransmitQueue(t, 64, 4, 3, pktlen, 0, iss)
sentBefore := rtx.BufferedSent()
rtx.RetransmitFrom(iss + pktlen) // Start of packet 2.
testQueueSanity(t, rtx)
if got := rtx.BufferedSent(); got != pktlen {
t.Fatalf("sent=%d, want %d (only packet 1 remains sent)", got, pktlen)
}
if got := rtx.BufferedUnsent(); got != sentBefore-pktlen {
t.Fatalf("unsent=%d, want %d", got, sentBefore-pktlen)
}
mustRemake(t, rtx, iss+pktlen, stream[pktlen:2*pktlen])
testQueueSanity(t, rtx)
mustRemake(t, rtx, iss+2*pktlen, stream[2*pktlen:3*pktlen])
testQueueSanity(t, rtx)
}
// TestRingTx_RetransmitFromMidPacket verifies a sequence inside a packet is
// snapped down to that packet's start: the queue tracks whole packets.
func TestRingTx_RetransmitFromMidPacket(t *testing.T) {
const iss, pktlen = Value(100), 4
rtx, stream := newRetransmitQueue(t, 64, 4, 3, pktlen, 0, iss)
rtx.RetransmitFrom(iss + pktlen + 2) // Two octets into packet 2.
testQueueSanity(t, rtx)
if got := rtx.BufferedSent(); got != pktlen {
t.Fatalf("sent=%d, want %d: rewind must floor to the packet start", got, pktlen)
}
mustRemake(t, rtx, iss+pktlen, stream[pktlen:2*pktlen])
}
// TestRingTx_RetransmitFromOldest rewinds the whole queue, which must match
// RetransmitFromUNA.
func TestRingTx_RetransmitFromOldest(t *testing.T) {
const iss, pktlen, npkt = Value(100), 4, 3
rtx, stream := newRetransmitQueue(t, 64, 4, npkt, pktlen, 0, iss)
rtx.RetransmitFrom(iss)
testQueueSanity(t, rtx)
viaUNA, _ := newRetransmitQueue(t, 64, 4, npkt, pktlen, 0, iss)
viaUNA.RetransmitFromUNA()
testQueueSanity(t, viaUNA)
if rtx.BufferedSent() != 0 {
t.Fatalf("sent=%d, want 0 after a full rewind", rtx.BufferedSent())
}
if rtx.BufferedUnsent() != npkt*pktlen {
t.Fatalf("unsent=%d, want %d", rtx.BufferedUnsent(), npkt*pktlen)
}
if rtx.BufferedSent() != viaUNA.BufferedSent() || rtx.BufferedUnsent() != viaUNA.BufferedUnsent() {
t.Fatal("RetransmitFrom(oldest) must match RetransmitFromUNA")
}
mustRemake(t, rtx, iss, stream[:pktlen])
}
// TestRingTx_RetransmitFromUnknownSeq verifies a sequence covered by no queued
// packet leaves the queue untouched.
func TestRingTx_RetransmitFromUnknownSeq(t *testing.T) {
const iss, pktlen, npkt = Value(100), 4, 3
rtx, _ := newRetransmitQueue(t, 64, 4, npkt, pktlen, 0, iss)
sent, unsent := rtx.BufferedSent(), rtx.BufferedUnsent()
rtx.RetransmitFrom(iss - 1) // Before the queue.
rtx.RetransmitFrom(iss + npkt*pktlen) // One past the last octet sent.
rtx.RetransmitFrom(iss + 1000) // Far beyond.
testQueueSanity(t, rtx)
if rtx.BufferedSent() != sent || rtx.BufferedUnsent() != unsent {
t.Fatalf("queue moved: sent %d→%d, unsent %d→%d", sent, rtx.BufferedSent(), unsent, rtx.BufferedUnsent())
}
}
// TestRingTx_RetransmitWithUnsentTail verifies a rewind reopens the unsent region
// over the rewound packets without losing the unsent tail behind them.
func TestRingTx_RetransmitWithUnsentTail(t *testing.T) {
const iss, pktlen, npkt, tail = Value(100), 4, 2, 5
rtx, stream := newRetransmitQueue(t, 64, 4, npkt, pktlen, tail, iss)
if got := rtx.BufferedUnsent(); got != tail {
t.Fatalf("unsent tail=%d, want %d", got, tail)
}
rtx.RetransmitFrom(iss + pktlen) // Rewind the second packet only.
testQueueSanity(t, rtx)
if got := rtx.BufferedUnsent(); got != pktlen+tail {
t.Fatalf("unsent=%d, want %d (rewound packet plus the tail)", got, pktlen+tail)
}
// The rewound packet re-emits first, then the tail follows in order.
mustRemake(t, rtx, iss+pktlen, stream[pktlen:2*pktlen])
testQueueSanity(t, rtx)
mustRemake(t, rtx, iss+2*pktlen, stream[2*pktlen:])
}
// TestRingTx_RetransmitAfterDrainedUnsent pins the write-position recovery: when
// every octet written has been packetized the unsent region is empty, so the
// rewind must reconstruct where data ends from the sent region.
func TestRingTx_RetransmitAfterDrainedUnsent(t *testing.T) {
const iss, pktlen, npkt = Value(100), 4, 3
rtx, stream := newRetransmitQueue(t, 64, 4, npkt, pktlen, 0, iss)
if got := rtx.BufferedUnsent(); got != 0 {
t.Fatalf("unsent=%d, want 0: all written data was packetized", got)
}
rtx.RetransmitFrom(iss + pktlen)
testQueueSanity(t, rtx)
if got := rtx.BufferedUnsent(); got != 2*pktlen {
t.Fatalf("unsent=%d, want %d: rewind lost the end of the data", got, 2*pktlen)
}
mustRemake(t, rtx, iss+pktlen, stream[pktlen:2*pktlen])
testQueueSanity(t, rtx)
mustRemake(t, rtx, iss+2*pktlen, stream[2*pktlen:3*pktlen])
}
// TestRingTx_RetransmitWrapped exercises a rewind on a queue whose regions wrap
// the end of the ring buffer.
func TestRingTx_RetransmitWrapped(t *testing.T) {
const bufsize, pktlen = 16, 4
const iss = Value(100)
var rtx ringTx
if err := rtx.Reset(make([]byte, bufsize), 4, iss); err != nil {
t.Fatal(err)
}
// Push the queue most of the way around the ring, acking as we go.
seq := iss
scratch := make([]byte, pktlen)
for round := range 3 {
chunk := make([]byte, pktlen)
for i := range chunk {
chunk[i] = byte(round*pktlen + i + 1)
}
if _, err := rtx.Write(chunk); err != nil {
t.Fatal(err)
}
if _, err := rtx.MakePacket(scratch, seq); err != nil {
t.Fatal(err)
}
seq += Value(pktlen)
if round < 2 {
if err := rtx.RecvACK(seq); err != nil {
t.Fatal(err)
}
}
testQueueSanity(t, &rtx)
}
// Two packets outstanding, straddling the wrap. Rewind the newest.
rewindSeq := seq - Value(pktlen)
want := append([]byte(nil), scratch...)
rtx.RetransmitFrom(rewindSeq)
testQueueSanity(t, &rtx)
mustRemake(t, &rtx, rewindSeq, want)
testQueueSanity(t, &rtx)
}
+35 -5
View File
@@ -107,9 +107,30 @@ func (bs *bufferSelect) numFree() (numFree int) {
}
// getRx returns the oldest published Rx frame, or nil if none is pending.
//
// The slot scan is not a consistent snapshot: goroPutRx may publish a frame
// into an already-scanned slot while this scan is in progress. Because the
// producer publishes frames in seq (arrival) order, any such straggler carries
// a lower seq than the candidate and must be delivered first to preserve
// arrival order. A confirming re-scan detects it; the loop retries until no
// older frame is observed, which terminates because the candidate seq strictly
// decreases and is bounded below by the true oldest pending frame.
func (bs *bufferSelect) getRx() []byte {
oldest := -1
var oldestSeq uint32
for {
oldest, oldestSeq := bs.scanOldest()
if oldest < 0 {
return nil
}
if !bs.hasPendingOlderThan(oldestSeq) {
return bs.bufs[oldest].buf[:bs.bufs[oldest].lenAcquire.Load()]
}
}
}
// scanOldest returns the index and seq of the pending Rx frame with the lowest
// arrival seq, or -1 if none is pending.
func (bs *bufferSelect) scanOldest() (oldest int, oldestSeq uint32) {
oldest = -1
for i := range bs.bufs {
n := bs.bufs[i].lenAcquire.Load()
if n > 0 && bs.bufs[i].isRx.Load() &&
@@ -118,10 +139,19 @@ func (bs *bufferSelect) getRx() []byte {
oldestSeq = bs.bufs[i].seq
}
}
if oldest < 0 {
return nil
return oldest, oldestSeq
}
// hasPendingOlderThan reports whether any pending Rx frame has a seq strictly
// less than seq, i.e. a frame that should be delivered before it.
func (bs *bufferSelect) hasPendingOlderThan(seq uint32) bool {
for i := range bs.bufs {
n := bs.bufs[i].lenAcquire.Load()
if n > 0 && bs.bufs[i].isRx.Load() && lessThan(bs.bufs[i].seq, seq) {
return true
}
}
return bs.bufs[oldest].buf[:bs.bufs[oldest].lenAcquire.Load()]
return false
}
func (bs *bufferSelect) release(buf []byte) {
+2 -1
View File
@@ -629,7 +629,8 @@ func (s *StackAsync) StartLookupIPType(host string, qtype dns.Type) error {
Additional: []dns.Resource{
s.ednsopt,
},
EnableRecursion: true,
EnableRecursion: true,
MaxResponseAnswers: uint16(len(s.addrbufnip)),
})
if err != nil {
return err
+3 -12
View File
@@ -162,11 +162,9 @@ func (s StackGo) SocketNetip(ctx context.Context, network string, family, sotype
return nil, err
}
uc := udpconn{
Conn: &conn,
// TODO: use udpaddr until UDPAddrFromAddrPort added to tinygo.
// https://github.com/tinygo-org/net/issues/45
localAddr: udpaddr(laddr),
raddr: udpaddr(raddr),
Conn: &conn,
localAddr: net.UDPAddrFromAddrPort(laddr),
raddr: net.UDPAddrFromAddrPort(raddr),
}
return uc, nil
case "tcp", "tcp4", "tcp6":
@@ -384,13 +382,6 @@ func (c udpconn) ReadFrom(b []byte) (int, net.Addr, error) {
func (c udpconn) WriteTo(b []byte, _ net.Addr) (int, error) {
return c.Conn.Write(b) // connected UDP: always writes to dialed remote
}
func udpaddr(addr netip.AddrPort) net.Addr {
return &net.UDPAddr{
IP: addr.Addr().AsSlice(),
Zone: addr.Addr().Zone(),
Port: int(addr.Port()),
}
}
// parseNetAddr converts a [net.Addr] to a [netip.AddrPort]. A nil or empty IP
// (e.g. ":22" from a listen address with no host) is treated as 0.0.0.0 so