apply @MDr164 various fixes

This commit is contained in:
Patricio Whittingslow
2026-07-28 20:29:06 -03:00
parent 743e09a4c1
commit de766c8a24
6 changed files with 114 additions and 35 deletions
+12 -11
View File
@@ -25,7 +25,7 @@ const maxStatusLine = len("HTTP/1.1 ") + 3 + 1 + len("Network Authentication Req
// the bytes that follow the parsed request header. Read the request body with
// [Exchange.ReadBody] before setting response headers.
type Exchange struct {
used atomic.Bool
acquired atomic.Bool
gen atomic.Uint32
respTopBuf [maxStatusLine]byte
respTopWritten uint8
@@ -105,7 +105,7 @@ func (exch *Exchange) Configure(cfg ExchangeConfig) {
// reusing the buffer set by [Exchange.Configure]. Returns false if the exchange
// is already serving, in which case conn is untouched.
func (exch *Exchange) Acquire(conn conn) bool {
if !exch.used.CompareAndSwap(false, true) {
if !exch.acquired.CompareAndSwap(false, true) {
return false
}
exch.matchedPattern = ""
@@ -133,7 +133,7 @@ func (exch *Exchange) Release() {
}
exch.rw = nil
exch.gen.Add(1)
exch.used.Store(false)
exch.acquired.Store(false)
}
// UnsafeRawBuffer returns the contiguous buffer owned by [Exchange] being used for the request and response.
@@ -278,9 +278,9 @@ type ExchangeRW struct {
}
// IsValid returns true while the handle still refers to the request it was
// taken from, i.e: false once the exchange was released or hijacked away.
// taken from, i.e: false once the exchange was released.
func (rw *ExchangeRW) IsValid() bool {
return rw.gen == rw.exch.gen.Load() && rw.exch.used.Load()
return rw.gen == rw.exch.gen.Load() && rw.exch.acquired.Load()
}
func (rw *ExchangeRW) validate() error {
@@ -407,7 +407,7 @@ func (exch *Exchange) RequestContentType() []byte {
// Content-Length field. An absent field is not a client error: such a request
// has no body at all, RFC 9112 6.3. Check for the error to answer 411 instead.
// See [httpraw.Header.ContentLength].
func (exch *Exchange) RequestContentLength() (int64, error) {
func (exch *Exchange) RequestContentLength() (int64, bool, error) {
return exch.RequestHeaderRaw().ContentLength()
}
@@ -432,12 +432,13 @@ func (exch *Exchange) RequestParseForm(dst *httpraw.Form, buf []byte) error {
// wire would parse chunk sizes as form data. httpraw does not decode them.
return errUnsupportedTransferCoding
}
length, err := exch.RequestContentLength()
if err != nil {
dst.Reset(buf[:0])
return dst.Parse() // No length is no body, RFC 9112 6.3.
length, present, err := exch.RequestContentLength()
if !present {
return nil // No length is no body, RFC 9112 6.3.
} else if err != nil {
return err
} else if length > int64(len(buf)) {
return lneto.ErrBufferFull // Refuse before reading, caller may answer 413.
return lneto.ErrShortBuffer // Refuse before reading, caller may answer 413.
}
buf = buf[:length]
for read := 0; read < len(buf); {
+1 -1
View File
@@ -707,7 +707,7 @@ func TestExchangeRequestParseForm(t *testing.T) {
name: "body larger than buffer",
request: "POST /f HTTP/1.1\r\nHost: h\r\n" + formType + "Content-Length: 11\r\n\r\na=1&b=2&c=3",
bufSize: 4,
wantErr: lneto.ErrBufferFull,
wantErr: lneto.ErrShortBuffer,
},
} {
t.Run(test.name, func(t *testing.T) {
+6 -2
View File
@@ -13,6 +13,9 @@ import (
// Requires exchange to be acquired and configured. Will panic if any argument is nil.
// Handle does not close the connection on any outcome: the caller owns it.
func Handle(exch *Exchange, mux Mux, backoff lneto.BackoffStrategy) error {
if !exch.acquired.Load() {
return lneto.ErrBadState
}
reqhdr := &exch.reqHdr
reqhdr.Reset(nil, 0) // Assume exchange has been configured and reuse memory.
var consecutiveBackoffs uint
@@ -56,11 +59,12 @@ func Handle(exch *Exchange, mux Mux, backoff lneto.BackoffStrategy) error {
if handler != nil {
exch.matchedPattern = matchedPattern
handler(exch)
exch.FlushHeader()
if !exch.hijacked {
exch.FlushHeader()
}
} else {
exch.WriteHeader(404)
}
// TODO write response from exchange here.
return nil
}
+34 -7
View File
@@ -4,6 +4,7 @@ import (
"errors"
"io"
"log/slog"
"math"
"sync"
"sync/atomic"
"time"
@@ -100,18 +101,44 @@ type RouterConfig struct {
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()
if workerMode && cfg.MaxAwaitingConns == 0 ||
cfg.Mux == nil ||
cfg.RequestNumHeaderKVCap <= 0 ||
!workerMode && cfg.FixedNumGoroutines != -1 {
switch {
case cfg.Mux == nil,
!workerMode && cfg.FixedNumGoroutines != -1,
workerMode && cfg.MaxAwaitingConns <= 0,
cfg.RequestNumHeaderKVCap <= 0,
cfg.RequestHeaderBufferSize < minRequestHeaderBuffer,
cfg.ResponseHeaderMinBufferSize < minResponseHeaderBuffer,
cfg.ResponseHeaderMinBufferSize > maxExchangeBuffer,
cfg.RequestHeaderBufferSize > maxExchangeBuffer-cfg.ResponseHeaderMinBufferSize:
return lneto.ErrInvalidConfig
} else if cfg.Backoff == nil {
case cfg.Backoff == nil:
return lneto.ErrMissingHALConfig
}
if workerMode {
// Buffer sizes are bounded by maxExchangeBuffer above so the sum cannot
// overflow; the products below allocate and can.
exchBuf := cfg.RequestHeaderBufferSize + cfg.ResponseHeaderMinBufferSize
if cfg.FixedNumGoroutines > math.MaxInt/exchBuf ||
cfg.FixedNumGoroutines > math.MaxInt/cfg.RequestNumHeaderKVCap {
return lneto.ErrInvalidConfig
}
}
return nil
}
@@ -215,7 +242,7 @@ func (r *Router) awaitIdleExchangesLocked(maxWait time.Duration) error {
for waited := time.Duration(0); ; waited += pollInterval {
busy := false
for i := range r.exchs {
if r.exchs[i].used.Load() {
if r.exchs[i].acquired.Load() {
busy = true
break
}
@@ -269,7 +296,7 @@ func (r *Router) Handle(conn io.ReadWriteCloser) error {
enqueued = true
default:
// pendingConns cannot store another Conn, we drop and return error.
exch.used.Store(false) // release.
exch.acquired.Store(false) // release.
}
r.mu.Unlock()
if enqueued {