massive documentation push and code reordering in files

This commit is contained in:
Patricio Whittingslow
2026-07-26 02:59:13 -03:00
parent 2ac4380442
commit 3b3fb1328a
18 changed files with 886 additions and 770 deletions
@@ -288,7 +288,7 @@ func handleConnNet(conn net.Conn) error {
}
}
method := string(hdr.Method())
uri := string(hdr.RequestURI())
uri := string(hdr.RequestTarget())
fmt.Printf("< %s %s\n", method, uri)
var resp httpraw.Header
@@ -370,7 +370,7 @@ func mockClient(stack *xnet.StackAsync, port uint16, subnet netip.Prefix) {
var hdr httpraw.Header
hdr.SetMethod("GET")
hdr.SetRequestURI("/")
hdr.SetRequestTarget("/")
hdr.SetProtocol("HTTP/1.1")
hdr.Set("Host", string(ipv4.AppendFormatAddr(nil, stack.Addr4())))
hdr.Set("User-Agent", "lneto-mock")
+2 -2
View File
@@ -97,8 +97,8 @@ func homepage(exch *httphi.Exchange) {
n += len(strconv.AppendUint(page[n:n], visits.Add(1), 10))
n += copy(page[n:], htmlTail)
exch.SetHeader("Content-Type", "text/html")
exch.SetHeaderInt("Content-Length", int64(n), 10)
exch.StageHeader("Content-Type", "text/html")
exch.StageHeaderInt("Content-Length", int64(n), 10)
exch.WriteHeader(int(httphi.StatusOK))
exch.Write(page[:n])
}
+1 -1
View File
@@ -27,7 +27,7 @@ func run() error {
// Prepare GET request.
var hdr httpraw.Header
hdr.SetMethod("GET")
hdr.SetRequestURI("/")
hdr.SetRequestTarget("/")
hdr.SetProtocol("HTTP/1.1")
req, err := hdr.AppendRequest(nil)
if err != nil {
+1 -1
View File
@@ -282,7 +282,7 @@ func handleConnection(conn *tcp.Conn) error {
}
method := string(hdr.Method())
uri := string(hdr.RequestURI())
uri := string(hdr.RequestTarget())
fmt.Printf("< %s %s\n", method, uri)
// Build response body.
+1 -1
View File
@@ -307,7 +307,7 @@ func run() (err error) {
timeHTTPCreate := timer("create HTTP GET request")
var hdr httpraw.Header
hdr.SetMethod("GET")
hdr.SetRequestURI("/")
hdr.SetRequestTarget("/")
hdr.SetProtocol("HTTP/1.1")
hdr.Set("Host", flagHostToResolve)
hdr.Set("User-Agent", "lneto")
+2 -2
View File
@@ -61,8 +61,8 @@ func BenchmarkHandle(b *testing.B) {
name: "GETWithHeadersAndQuery",
request: "GET /?abc=123 HTTP/1.1\r\nHost: tinygo.org\r\nUser-Agent: bench\r\nAccept: */*\r\nConnection: close\r\n\r\n",
handler: func(ex *Exchange) {
ex.SetHeader("Content-Type", "text/plain")
ex.SetHeaderInt("Content-Length", int64(len(benchBody)), 10)
ex.StageHeader("Content-Type", "text/plain")
ex.StageHeaderInt("Content-Length", int64(len(benchBody)), 10)
data, present := ex.AppendQuery(buf[:0], "abc", true)
if !present || !internal.BytesEqual(data, expect) {
panic("invalid result")
+380
View File
@@ -0,0 +1,380 @@
package httphi
import (
"slices"
"strconv"
"sync/atomic"
"github.com/soypat/lneto/http/httpraw"
"github.com/soypat/lneto/internal"
)
// maxStatusLine bounds the response status line: "HTTP/1.1 " + 3 digit code +
// " " + longest [StatusText] + CRLF.
const maxStatusLine = len("HTTP/1.1 ") + 3 + 1 + len("Network Authentication Required") + 2
// Exchange is a single request-response cycle over a connection, playing the
// part of both http.Request and http.ResponseWriter: Request* methods read the
// request, [Exchange.StageHeader] and [Exchange.Write] produce the response.
// A [Router] owns a fixed pool of them, which is what bounds its memory.
//
// Request and response share one buffer, the response header being written over
// 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
respTopBuf [maxStatusLine]byte
respTopWritten uint8
rawbuf []byte
respHeaderOff uint16
respHeaderLen uint16
reqHdr httpraw.Header
hijacked bool
rw conn
respRemains int
respErr error // Sticky: response is unrecoverable once a write fails.
headerWritten bool
normalizeKeys bool
nextFree *Exchange
readErr error
}
// HijackRaw is a low-level implementation of http.Hijacker interface.
// A Hijack method is not exposed due to heap allocation implications and correctness concerns.
// Below is what an actual implementation may look like:
//
// func (exch *Exchange) Hijack() (net.Conn, *bufio.ReadWriter, error) {
// conn, ok := exch.rw.(net.Conn)
// if !ok {
// return nil, nil, errors.New("net.Conn not implemented")
// }
// _, data, err := exch.HijackRaw(nil)
// if err != nil {
// return nil, nil, err
// }
// var rd *bufio.ReadWriter
// if len(data) > 0 {
// rd = &bufio.ReadWriter{Reader: bufio.NewReader(bytes.NewReader(data))}
// }
// return conn, rd, nil
// }
func (exch *Exchange) HijackRaw(dstBody []byte) (conn, []byte, error) {
data, err := exch.remainingSurplusBody()
if err != nil {
return nil, nil, err
}
exch.hijacked = true
dstBody = append(dstBody, data...)
return exch.rw, dstBody, nil
}
// Configure sets the memory the exchange works with for the rest of its life:
// rawbuf holds the request header, the response header and any surplus body,
// of which the first requestLim bytes are reserved for the request header.
// Panics if requestLim exceeds the buffer. Set normalizeKeys to normalize
// outgoing header keys, i.e: "content-type" to "Content-Type".
func (exch *Exchange) Configure(rawbuf []byte, requestLim int, normalizeKeys bool) {
respSize := len(rawbuf) - requestLim
if respSize < 0 {
panic("request lim larger than buffer")
}
exch.rawbuf = rawbuf
exch.reqHdr.Reset(rawbuf[:0:requestLim])
exch.normalizeKeys = normalizeKeys
}
// Acquire claims the exchange for conn and resets it to serve a new request,
// 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) {
return false
}
exch.readErr = nil
exch.respErr = nil
exch.hijacked = false
exch.respTopWritten = 0
exch.respHeaderOff = 0
exch.respHeaderLen = 0
exch.respRemains = 0
exch.rw = conn
exch.headerWritten = false
exch.nextFree = nil
exch.reqHdr.Reset(nil)
return true
}
// Release closes the exchange's connection and frees the exchange for a future
// [Exchange.Acquire]. The connection is left open if the handler took ownership
// of it with [Exchange.HijackRaw].
func (exch *Exchange) Release() {
if !exch.hijacked {
exch.rw.Close()
}
exch.rw = nil
exch.used.Store(false)
}
// StageHeader stages a response header field, written on the first
// [Exchange.FlushHeader], [Exchange.WriteHeader] or [Exchange.Write].
// 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]
// appends past the last field to close the header block.
if len(key)+len(value)+len(":\r\n")+len("\r\n") > free {
return false
}
n := copy(exch.rawbuf[off:], key)
if exch.normalizeKeys {
httpraw.NormalizeHeaderKey(exch.rawbuf[off : off+n])
}
exch.rawbuf[off+n] = ':'
n++
n += copy(exch.rawbuf[off+n:], value)
exch.rawbuf[off+n] = '\r'
exch.rawbuf[off+n+1] = '\n'
n += 2
exch.respHeaderLen += uint16(n)
return true
}
// StageHeaderInt is [Exchange.StageHeader] with an integer value, i.e: Content-Length.
// It formats the value directly into the response buffer without allocating.
// 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) StageHeaderInt(key string, value int64, base int) (enoughMemory bool) {
if exch.headerWritten || base < 10 || base > 36 {
return false
}
off := int(exch.respHeaderOff) + int(exch.respHeaderLen)
free := len(exch.rawbuf) - off
if len(key)+internal.IntLen(value, base)+len(":\r\n")+len("\r\n") > free {
return false
}
n := copy(exch.rawbuf[off:], key)
if exch.normalizeKeys {
httpraw.NormalizeHeaderKey(exch.rawbuf[off : off+n])
}
exch.rawbuf[off+n] = ':'
n++
n += len(strconv.AppendInt(exch.rawbuf[off+n:off+n], value, base))
exch.rawbuf[off+n] = '\r'
exch.rawbuf[off+n+1] = '\n'
n += 2
exch.respHeaderLen += uint16(n)
return true
}
// StageWriteStatus 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.
func (exch *Exchange) StageWriteStatus(code int) {
if code >= 1000 || exch.headerWritten {
return
} else if code == 200 {
// Common case.
exch.respTopWritten = uint8(copy(exch.respTopBuf[:], "HTTP/1.1 200 OK\r\n"))
return
}
n := copy(exch.respTopBuf[:], "HTTP/1.1 ")
n += len(strconv.AppendInt(exch.respTopBuf[n:n], int64(code), 10))
text := StatusText(code)
exch.respTopBuf[n] = ' '
n++
n += copy(exch.respTopBuf[n:], text)
exch.respTopBuf[n] = '\r'
exch.respTopBuf[n+1] = '\n'
exch.respTopWritten = uint8(n + 2)
}
// 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) {
if !exch.headerWritten {
exch.StageWriteStatus(code)
exch.FlushHeader()
}
}
// FlushHeader writes the status line and staged header fields to the connection
// and returns the bytes written, defaulting to a 200 status if none was staged.
// Does nothing if the header was already written.
func (exch *Exchange) FlushHeader() (int, error) {
if exch.respErr != nil {
return 0, exch.respErr
} else if exch.headerWritten {
return 0, nil
}
if exch.respTopWritten == 0 {
exch.StageWriteStatus(200)
}
exch.headerWritten = true
ng, err := exch.rw.Write(exch.respTopBuf[:exch.respTopWritten])
if err != nil {
exch.respErr = err
return ng, err
}
off := int(exch.respHeaderOff)
headers := exch.rawbuf[off : off+int(exch.respHeaderLen)+2]
headers[len(headers)-1] = '\n'
headers[len(headers)-2] = '\r'
ng2, err := exch.rw.Write(headers)
exch.respErr = err
return ng + ng2, err
}
// Write writes response body bytes, flushing the header first if the handler
// has not written it yet. Once a write to the connection fails the response is
// unrecoverable and every later write returns that same error, so a body never
// reaches the wire without its header.
func (exch *Exchange) Write(buf []byte) (int, error) {
if exch.respErr != nil {
return 0, exch.respErr
} else if !exch.headerWritten {
_, err := exch.FlushHeader()
if err != nil {
return 0, err // Body must not reach the wire without its header.
}
}
if len(buf) == 0 {
return 0, nil
}
n, err := exch.rw.Write(buf)
exch.respErr = err
return n, err
}
// ReadBody reads the request body into dst, starting with the bytes that
// arrived in the same read as the header and continuing from the connection.
// The exchange does not know the body's length: use Content-Length or the
// transfer encoding to know when to stop reading.
func (exch *Exchange) ReadBody(dst []byte) (n int, _ error) {
if exch.respRemains > 0 {
toRead, err := exch.remainingSurplusBody()
if err != nil {
return 0, err
}
n = copy(dst, toRead)
exch.respRemains -= n
if len(dst) == n {
return n, nil
}
dst = dst[n:]
}
nr, err := exch.rw.Read(dst)
return nr + n, err
}
func (exch *Exchange) remainingSurplusBody() ([]byte, error) {
_, err := exch.reqHdr.Body()
if err != nil {
return nil, err // Returns mangled buffer error if request header has been misused.
}
surplus := exch.rawbuf[exch.reqHdr.BufferParsed():exch.reqHdr.BufferReceived()]
toRead := surplus[len(surplus)-exch.respRemains:]
return toRead, nil
}
// RequestHeaderRaw returns the parsed request header for access beyond the
// Request* methods, such as [httpraw.Header.ForEach]. Valid until the exchange
// is released, and writing to it corrupts the response.
func (exch *Exchange) RequestHeaderRaw() *httpraw.Header {
return &exch.reqHdr
}
// RequestParseCookie parses the request's key header field into dst, i.e:
// "Cookie". The caller owns dst and its buffer, so it may be reused between
// requests.
func (exch *Exchange) RequestParseCookie(dst *httpraw.Cookie, key string) error {
value := exch.RequestHeader(key)
return dst.ParseBytes(value)
}
// RequestHeader returns the value of the first request header field matching
// key, or nil if absent. Key matching is case sensitive.
func (exch *Exchange) RequestHeader(key string) []byte {
header := exch.RequestHeaderRaw()
return header.Get(key)
}
// RequestTarget returns the request-target (URI) of the request line, i.e:
// "/search?q=go". See [httpraw.Header.RequestTarget].
func (exch *Exchange) RequestTarget() []byte {
return exch.RequestHeaderRaw().RequestTarget()
}
// RequestPath returns the request-target (URI) up to the query string. This is
// what the [Mux] matches on, i.e: "/search" for a request to "/search?q=go".
func (exch *Exchange) RequestPath() []byte {
return exch.RequestHeaderRaw().RequestPath()
}
// RequestQuery returns the request's query string as it appears on the wire.
// Iterate it with [httpraw.NextQueryPair]. See [httpraw.Header.RequestQuery].
func (exch *Exchange) RequestQuery() []byte {
return exch.RequestHeaderRaw().RequestQuery()
}
// AppendQuery appends the value of the first query parameter matching key to
// dst and reports whether the parameter was present. A parameter with no value
// ("?debug") and one with an empty value ("?debug=") are both present with
// nothing appended.
//
// Keys are matched decoded, so key "a b" finds "a%20b" and "a+b". Values are
// appended raw unless decoded is set, in which case percent escapes and '+' are
// decoded. A parameter whose value fails to decode is reported absent, and a
// parameter whose key fails to decode is skipped.
//
// dst doubles as scratch space for decoding candidate keys, so AppendQuery only
// allocates when dst lacks the capacity to hold the longest key it inspects.
func (exch *Exchange) AppendQuery(dst []byte, key string, decoded bool) (valueAppended []byte, present bool) {
const plusAsSpace = true // Query strings are form encoded, unlike paths.
base := len(dst)
rawkey, rawval, rest := httpraw.NextQueryPair(exch.RequestQuery())
for ; rawkey != nil; rawkey, rawval, rest = httpraw.NextQueryPair(rest) {
if b2s(rawkey) != key {
// Key may be encoded: decode it over dst's free space and compare.
// A decoded key cannot appear raw, so this cannot alias a real key.
dst = slices.Grow(dst, len(rawkey))
scratch := dst[base : base+len(rawkey)]
n, err := httpraw.CopyDecodedPercentURL(scratch, rawkey, plusAsSpace)
if err != nil || b2s(scratch[:n]) != key {
continue // Malformed or different key, keep looking.
}
}
if len(rawval) == 0 {
return dst[:base], true // Flag or empty value, nothing to append.
}
dst = slices.Grow(dst, len(rawval))
if !decoded {
return append(dst[:base], rawval...), true
}
n, err := httpraw.CopyDecodedPercentURL(dst[base:base+len(rawval)], rawval, plusAsSpace)
if err != nil {
return dst[:base], false // Do not hand back half a decode.
}
return dst[:base+n], true
}
return dst[:base], false
}
// RequestMethod returns the request line's method, i.e: "GET". See
// [MethodFromBytes] to compare it against a [Method].
func (exch *Exchange) RequestMethod() []byte {
return exch.RequestHeaderRaw().Method()
}
// RequestConnectionClose returns true if the client asked for the connection to
// be closed after this exchange with a "Connection: close" header field.
func (exch *Exchange) RequestConnectionClose() bool {
return exch.RequestHeaderRaw().ConnectionClose()
}
+8 -7
View File
@@ -5,10 +5,11 @@ import (
"errors"
"strings"
"github.com/soypat/lneto/http/httpraw"
"testing"
"time"
"github.com/soypat/lneto/http/httpraw"
"github.com/soypat/lneto"
)
@@ -122,7 +123,7 @@ func TestExchangeSetHeader(t *testing.T) {
conn := newConn("")
exch := newExchange(t, conn, 128, test.normalize)
for _, kv := range test.set {
if !exch.SetHeader(kv[0], kv[1]) {
if !exch.StageHeader(kv[0], kv[1]) {
t.Fatalf("SetHeader(%q,%q) reported insufficient memory", kv[0], kv[1])
}
}
@@ -144,7 +145,7 @@ func TestExchangeSetHeaderOOM(t *testing.T) {
const bufferSize = 32
conn := newConn("")
exch := newExchange(t, conn, bufferSize, false)
if exch.SetHeader("X-Big", strings.Repeat("v", 4*bufferSize)) {
if exch.StageHeader("X-Big", strings.Repeat("v", 4*bufferSize)) {
t.Fatal("want insufficient memory reported for oversized header value")
}
exch.WriteHeader(200)
@@ -195,7 +196,7 @@ func TestHandleRequestFields(t *testing.T) {
route, _, _ := strings.Cut(test.wantURI, "?") // Mux matches on path.
sm.Handle(route, func(ex *Exchange) {
gotMethod = string(ex.RequestMethod())
gotURI = string(ex.RequestURI())
gotURI = string(ex.RequestTarget())
gotHost = string(ex.RequestHeader("Host"))
ex.WriteHeader(200)
})
@@ -299,7 +300,7 @@ func TestExchangeSetHeaderExactFit(t *testing.T) {
if !exch.Acquire(conn) {
t.Fatal("fresh exchange failed to acquire connection")
}
set := exch.SetHeader(key, value)
set := exch.StageHeader(key, value)
exch.WriteHeader(200)
want := "HTTP/1.1 200 OK\r\n"
@@ -437,7 +438,7 @@ func TestExchangeSetHeaderInt(t *testing.T) {
} {
conn := newConn("")
exch := newExchange(t, conn, 256, false)
exch.SetHeaderInt("N", test.value, test.base)
exch.StageHeaderInt("N", test.value, test.base)
exch.WriteHeader(200)
got, _ := strings.CutPrefix(conn.ViewWritten(), "HTTP/1.1 200 OK\r\n")
if got != test.want {
@@ -450,7 +451,7 @@ func TestExchangeSetHeaderInt(t *testing.T) {
func TestExchangeSetHeaderIntNoAlloc(t *testing.T) {
exch := newExchange(t, newConn(""), 256, false)
allocs := testing.AllocsPerRun(100, func() {
exch.SetHeaderInt("Content-Length", 1234567890, 10)
exch.StageHeaderInt("Content-Length", 1234567890, 10)
})
if allocs != 0 {
t.Errorf("SetHeaderInt allocated %v times, want 0", allocs)
-709
View File
@@ -1,709 +0,0 @@
package httphi
import (
"errors"
"io"
"log/slog"
"slices"
"strconv"
"sync"
"sync/atomic"
"time"
"unsafe"
"github.com/soypat/lneto"
"github.com/soypat/lneto/http/httpraw"
"github.com/soypat/lneto/internal"
)
//go:generate stringer -type Method,status -linecomment -output stringers.go
// reconfigureWait bounds how long [Router.Configure] waits for the previous
// generation to stop serving before reusing its exchange buffers.
const reconfigureWait = 10 * time.Millisecond
var (
errNoRequestProto = errors.New("httphi: request line with no HTTP version")
errBusyExchanges = errors.New("httphi: exchanges still serving, cannot reuse their buffers")
)
type conn = io.ReadWriteCloser
// Router hosts concurrent safe data.
type Router struct {
mu sync.Mutex
gen atomic.Uint32
numGoro int
reqBuf int
respBuf int
normalizeKeys bool
pendingConns chan job
mux Mux
globbuf []byte
exchs []Exchange
freeList *Exchange
backoff lneto.BackoffStrategy
log *slog.Logger
}
type job struct {
exch *Exchange
}
type RouterConfig struct {
// FixedNumGoroutines must be set to either -1 (freely allocate new goroutines) or to the number of goroutines
// to spawn on [Router.Configure] being called.
FixedNumGoroutines int
// RequestBufferSize determines the buffer allocated
// for processing requests.
RequestBufferSize int
// ResponseMinBufferSize determines buffer allocated for processing responses.
// Response buffer will reuse unused request memory so this is not a strict limit.
ResponseMinBufferSize int
NormalizeOutgoingKeys bool
MaxAwaitingConns int
Backoff lneto.BackoffStrategy
Mux Mux
Logger *slog.Logger
}
func (cfg RouterConfig) Validate() error {
workerMode := cfg.workerMode()
if workerMode && cfg.MaxAwaitingConns == 0 ||
cfg.Mux == nil ||
!workerMode && cfg.FixedNumGoroutines != -1 {
return lneto.ErrInvalidConfig
} else if cfg.Backoff == nil {
return lneto.ErrMissingHALConfig
}
return nil
}
func (cfg RouterConfig) workerMode() bool {
return cfg.FixedNumGoroutines > 0
}
// Teardown stops fixed goroutines.
func (r *Router) TeardownGoroutines() {
r.mu.Lock()
defer r.mu.Unlock()
r.teardownGoroutinesLocked()
}
// teardownGoroutinesLocked closes the job queue fixed goroutines feed from.
// Requires r.mu held so that a concurrent [Router.Handle] cannot be enqueueing
// on the channel being closed.
func (r *Router) teardownGoroutinesLocked() {
r.gen.Add(1)
if r.pendingConns != nil {
close(r.pendingConns)
r.pendingConns = nil
}
}
func (r *Router) Configure(cfg RouterConfig) error {
if err := cfg.Validate(); err != nil {
return err
}
r.mu.Lock()
defer r.mu.Unlock()
r.teardownGoroutinesLocked()
gen := r.gen.Load()
numgoro := cfg.FixedNumGoroutines
workerMode := cfg.workerMode()
r.reqBuf = cfg.RequestBufferSize
r.respBuf = cfg.ResponseMinBufferSize
r.mux = cfg.Mux
r.log = cfg.Logger
r.normalizeKeys = cfg.NormalizeOutgoingKeys
if !workerMode {
r.backoff = cfg.Backoff
r.numGoro = 0
r.pendingConns = nil
return nil
}
if workerMode {
jobqueue := make(chan job, cfg.MaxAwaitingConns)
if gen > 1 {
// Exchange buffers below are reused: the previous generation must be
// done serving before they may be handed to the new one.
err := r.awaitIdleExchangesLocked(reconfigureWait)
if err != nil {
return err
}
}
r.freeList = nil // Freelist entries point into the buffers reused below.
internal.SliceReuse(&r.exchs, numgoro)
r.exchs = r.exchs[:numgoro]
rawBuflen := cfg.RequestBufferSize + cfg.ResponseMinBufferSize
internal.SliceReuse(&r.globbuf, numgoro*rawBuflen)
for i := range numgoro {
// TODO exchange buffer alloc
goff := i * rawBuflen
r.exchs[i].Configure(r.globbuf[goff:goff+rawBuflen], cfg.RequestBufferSize, cfg.NormalizeOutgoingKeys)
go r.goroWorker(gen, jobqueue, cfg.Backoff, cfg.Mux)
}
r.pendingConns = jobqueue
r.numGoro = numgoro
}
return nil
}
// awaitIdleExchangesLocked waits up to maxWait for exchanges of the previous
// generation to finish serving so their buffers may be reused. Requires r.mu
// held; the lock is released while waiting since [Router.freeExch] needs it to
// free the exchanges being waited on.
func (r *Router) awaitIdleExchangesLocked(maxWait time.Duration) error {
const pollInterval = time.Millisecond
for waited := time.Duration(0); ; waited += pollInterval {
busy := false
for i := range r.exchs {
if r.exchs[i].used.Load() {
busy = true
break
}
}
if !busy {
return nil
} else if waited >= maxWait {
return errBusyExchanges
}
r.mu.Unlock()
time.Sleep(pollInterval)
r.mu.Lock()
}
}
// Handle is a extremely low-level HTTP handling method used internally in [Router].
// 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 {
reqhdr := &exch.reqHdr
reqhdr.Reset(nil)
var consecutiveBackoffs uint
for {
n, err := reqhdr.ReadFromLimited(exch.rw, reqhdr.BufferFree())
if err != nil {
exch.readErr = err
return err
} else if n == 0 {
backoff.Do(consecutiveBackoffs)
consecutiveBackoffs++
continue
}
consecutiveBackoffs = 0
const asRequest = false
needMore, err := reqhdr.TryParse(asRequest)
if needMore {
continue // Request header split across reads, accumulate the rest.
} else if err != nil {
return err
}
break // Done!
}
// Setup Exchange fields necessary for correct functioning.
parsed := reqhdr.BufferParsed()
exch.respRemains = reqhdr.BufferReceived() - parsed
exch.respHeaderOff = uint16(parsed)
exch.respHeaderLen = 0
if len(reqhdr.Protocol()) == 0 {
// Request line with no HTTP version is a HTTP/0.9 simple-request, which
// httpraw tolerates. It is not a valid HTTP/1.1 request-line, RFC 9112 3.
exch.WriteHeader(int(StatusBadRequest))
return errNoRequestProto
}
// Mux on the request path: the query string is the handler's business.
path := reqhdr.RequestPath()
meth := reqhdr.Method()
handler := mux.LookupHandler(MethodFromBytes(meth), path)
if handler != nil {
handler(exch)
exch.FlushHeader()
} else {
exch.WriteHeader(404)
}
// TODO write response from exchange here.
return nil
}
func (r *Router) Handle(conn conn) error {
// Exchange acquisition and the configuration it is served with must be read
// under the same lock: [Router.Configure] may run concurrently.
r.mu.Lock()
exch := r.getExchLocked(conn)
numGoro, backoff, mux := r.numGoro, r.backoff, r.mux
if exch == nil {
r.mu.Unlock()
return lneto.ErrExhausted
} else if numGoro == 0 {
r.mu.Unlock()
go r.goroHandle(exch, backoff, mux)
return nil
}
// Enqueue under the lock: [Router.Configure] closes pendingConns while
// holding it, so an unlocked send could land on a closed channel. The send
// never blocks, so holding the lock cannot stall a worker.
var enqueued bool
select {
case r.pendingConns <- job{exch: exch}:
enqueued = true
default:
// pendingConns cannot store another Conn, we drop and return error.
exch.used.Store(false) // release.
}
r.mu.Unlock()
if enqueued {
return nil
}
return lneto.ErrPacketDrop
}
func (r *Router) goroWorker(gen uint32, queue chan job, backoff lneto.BackoffStrategy, mux Mux) {
for job := range queue {
exch := job.exch
if gen != r.gen.Load() {
return
} else if exch == nil {
panic("httplo: unreachable nil job")
}
r.goroHandle(exch, backoff, mux)
}
}
func (r *Router) goroHandle(exch *Exchange, backoff lneto.BackoffStrategy, mux Mux) {
defer r.freeExch(exch)
err := Handle(exch, mux, backoff)
if err != nil {
if exch.readErr != nil {
r.error("goroHandle:ReadFromLimited", slog.String("err", err.Error()))
} else {
r.error("goroHandle:TryParse?", slog.String("err", err.Error()))
}
}
}
func (r *Router) freeExch(exch *Exchange) {
const freelistMaxDepth = 5
r.mu.Lock()
depth := 0
for node := r.freeList; node != nil && depth < freelistMaxDepth; node = node.nextFree {
depth++
}
if depth < freelistMaxDepth {
// Push at head: appending at the tail would drop every node past the
// depth limit instead of dropping the exchange we cannot store.
exch.nextFree = r.freeList
r.freeList = exch
} else {
exch.nextFree = nil // Freelist full, exchange is dropped.
}
exch.Release()
r.mu.Unlock()
}
// getExchLocked returns an exchange acquired on conn. Requires r.mu held.
func (r *Router) getExchLocked(conn conn) (exch *Exchange) {
if r.freeList != nil {
// Successor must be read before Acquire: Acquire clears nextFree, so
// popping afterwards would truncate the freelist to the popped node.
next := r.freeList.nextFree
if r.freeList.Acquire(conn) {
exch = r.freeList
r.freeList = next
return exch
}
}
for i := range r.exchs {
if r.exchs[i].Acquire(conn) {
return &r.exchs[i]
}
}
if r.numGoro == 0 {
exch := new(Exchange)
exch.Configure(make([]byte, r.respBuf+r.reqBuf), r.reqBuf, r.normalizeKeys)
exch.Acquire(conn) // Fresh exchange, CAS cannot fail.
return exch
}
return nil
}
func (r *Router) error(msg string, attrs ...slog.Attr) {
internal.LogAttrs(r.log, slog.LevelError, msg, attrs...)
}
func (r *Router) info(msg string, attrs ...slog.Attr) {
internal.LogAttrs(r.log, slog.LevelInfo, msg, attrs...)
}
// maxStatusLine bounds the response status line: "HTTP/1.1 " + 3 digit code +
// " " + longest [StatusText] + CRLF.
const maxStatusLine = len("HTTP/1.1 ") + 3 + 1 + len("Network Authentication Required") + 2
type Exchange struct {
used atomic.Bool
respTopBuf [maxStatusLine]byte
respTopWritten uint8
rawbuf []byte
respHeaderOff uint16
respHeaderLen uint16
reqHdr httpraw.Header
hijacked bool
rw conn
respRemains int
respErr error // Sticky: response is unrecoverable once a write fails.
headerWritten bool
normalizeKeys bool
nextFree *Exchange
readErr error
}
// HijackRaw is a low-level implementation of http.Hijacker interface.
// A Hijack method is not exposed due to heap allocation implications and correctness concerns.
// Below is what an actual implementation may look like:
//
// func (exch *Exchange) Hijack() (net.Conn, *bufio.ReadWriter, error) {
// conn, ok := exch.rw.(net.Conn)
// if !ok {
// return nil, nil, errors.New("net.Conn not implemented")
// }
// _, data, err := exch.HijackRaw(nil)
// if err != nil {
// return nil, nil, err
// }
// var rd *bufio.ReadWriter
// if len(data) > 0 {
// rd = &bufio.ReadWriter{Reader: bufio.NewReader(bytes.NewReader(data))}
// }
// return conn, rd, nil
// }
func (exch *Exchange) HijackRaw(dstBody []byte) (conn, []byte, error) {
data, err := exch.remainingSurplusBody()
if err != nil {
return nil, nil, err
}
exch.hijacked = true
dstBody = append(dstBody, data...)
return exch.rw, dstBody, nil
}
func (exch *Exchange) Configure(rawbuf []byte, requestLim int, normalizeKeys bool) {
respSize := len(rawbuf) - requestLim
if respSize < 0 {
panic("request lim larger than buffer")
}
exch.rawbuf = rawbuf
exch.reqHdr.Reset(rawbuf[:0:requestLim])
exch.normalizeKeys = normalizeKeys
}
func (exch *Exchange) Acquire(conn conn) bool {
if !exch.used.CompareAndSwap(false, true) {
return false
}
exch.readErr = nil
exch.respErr = nil
exch.hijacked = false
exch.respTopWritten = 0
exch.respHeaderOff = 0
exch.respHeaderLen = 0
exch.respRemains = 0
exch.rw = conn
exch.headerWritten = false
exch.nextFree = nil
exch.reqHdr.Reset(nil)
return true
}
func (exch *Exchange) Release() {
if !exch.hijacked {
exch.rw.Close()
}
exch.rw = nil
exch.used.Store(false)
}
func (exch *Exchange) SetHeader(key, value string) (enoughMemory bool) {
off := int(exch.respHeaderOff) + int(exch.respHeaderLen)
free := len(exch.rawbuf) - off
// Field costs key+':'+value+CRLF, plus the CRLF [Exchange.FlushHeader]
// appends past the last field to close the header block.
if len(key)+len(value)+len(":\r\n")+len("\r\n") > free {
return false
}
n := copy(exch.rawbuf[off:], key)
if exch.normalizeKeys {
httpraw.NormalizeHeaderKey(exch.rawbuf[off : off+n])
}
exch.rawbuf[off+n] = ':'
n++
n += copy(exch.rawbuf[off+n:], value)
exch.rawbuf[off+n] = '\r'
exch.rawbuf[off+n+1] = '\n'
n += 2
exch.respHeaderLen += uint16(n)
return true
}
// SetHeaderInt is [Exchange.SetHeader] with an integer value, i.e: Content-Length.
// It formats the value directly into the response buffer without allocating.
// 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) SetHeaderInt(key string, value int64, base int) (enoughMemory bool) {
if base < 10 || base > 36 {
return false
}
off := int(exch.respHeaderOff) + int(exch.respHeaderLen)
free := len(exch.rawbuf) - off
if len(key)+internal.IntLen(value, base)+len(":\r\n")+len("\r\n") > free {
return false
}
n := copy(exch.rawbuf[off:], key)
if exch.normalizeKeys {
httpraw.NormalizeHeaderKey(exch.rawbuf[off : off+n])
}
exch.rawbuf[off+n] = ':'
n++
n += len(strconv.AppendInt(exch.rawbuf[off+n:off+n], value, base))
exch.rawbuf[off+n] = '\r'
exch.rawbuf[off+n+1] = '\n'
n += 2
exch.respHeaderLen += uint16(n)
return true
}
func (exch *Exchange) StageWriteStatus(code int) {
if code >= 1000 || exch.headerWritten {
return
} else if code == 200 {
// Common case.
exch.respTopWritten = uint8(copy(exch.respTopBuf[:], "HTTP/1.1 200 OK\r\n"))
return
}
n := copy(exch.respTopBuf[:], "HTTP/1.1 ")
n += len(strconv.AppendInt(exch.respTopBuf[n:n], int64(code), 10))
text := StatusText(code)
exch.respTopBuf[n] = ' '
n++
n += copy(exch.respTopBuf[n:], text)
exch.respTopBuf[n] = '\r'
exch.respTopBuf[n+1] = '\n'
exch.respTopWritten = uint8(n + 2)
}
func (exch *Exchange) WriteHeader(code int) {
if !exch.headerWritten {
exch.StageWriteStatus(code)
exch.FlushHeader()
}
}
func (exch *Exchange) FlushHeader() (int, error) {
if exch.respErr != nil {
return 0, exch.respErr
} else if exch.headerWritten {
return 0, nil
}
if exch.respTopWritten == 0 {
exch.StageWriteStatus(200)
}
exch.headerWritten = true
ng, err := exch.rw.Write(exch.respTopBuf[:exch.respTopWritten])
if err != nil {
exch.respErr = err
return ng, err
}
off := int(exch.respHeaderOff)
headers := exch.rawbuf[off : off+int(exch.respHeaderLen)+2]
headers[len(headers)-1] = '\n'
headers[len(headers)-2] = '\r'
ng2, err := exch.rw.Write(headers)
exch.respErr = err
return ng + ng2, err
}
func (exch *Exchange) Write(buf []byte) (int, error) {
if exch.respErr != nil {
return 0, exch.respErr
} else if !exch.headerWritten {
_, err := exch.FlushHeader()
if err != nil {
return 0, err // Body must not reach the wire without its header.
}
}
if len(buf) == 0 {
return 0, nil
}
n, err := exch.rw.Write(buf)
exch.respErr = err
return n, err
}
func (exch *Exchange) ReadBody(dst []byte) (n int, _ error) {
if exch.respRemains > 0 {
toRead, err := exch.remainingSurplusBody()
if err != nil {
return 0, err
}
n = copy(dst, toRead)
exch.respRemains -= n
if len(dst) == n {
return n, nil
}
dst = dst[n:]
}
nr, err := exch.rw.Read(dst)
return nr + n, err
}
func (exch *Exchange) remainingSurplusBody() ([]byte, error) {
_, err := exch.reqHdr.Body()
if err != nil {
return nil, err // Returns mangled buffer error if request header has been misused.
}
surplus := exch.rawbuf[exch.reqHdr.BufferParsed():exch.reqHdr.BufferReceived()]
toRead := surplus[len(surplus)-exch.respRemains:]
return toRead, nil
}
func (exch *Exchange) RequestHeaderRaw() *httpraw.Header {
return &exch.reqHdr
}
func (exch *Exchange) RequestParseCookie(dst *httpraw.Cookie, key string) error {
value := exch.RequestHeader(key)
return dst.ParseBytes(value)
}
func (exch *Exchange) RequestHeader(key string) []byte {
header := exch.RequestHeaderRaw()
return header.Get(key)
}
func (exch *Exchange) RequestURI() []byte {
return exch.RequestHeaderRaw().RequestURI()
}
// RequestPath returns the request URI up to the query string. This is what the
// [Mux] matches on, i.e: "/search" for a request to "/search?q=go".
func (exch *Exchange) RequestPath() []byte {
return exch.RequestHeaderRaw().RequestPath()
}
// RequestQuery returns the request's query string as it appears on the wire.
// Iterate it with [httpraw.NextQueryPair]. See [httpraw.Header.RequestQuery].
func (exch *Exchange) RequestQuery() []byte {
return exch.RequestHeaderRaw().RequestQuery()
}
// AppendQuery appends the value of the first query parameter matching key to
// dst and reports whether the parameter was present. A parameter with no value
// ("?debug") and one with an empty value ("?debug=") are both present with
// nothing appended.
//
// Keys are matched decoded, so key "a b" finds "a%20b" and "a+b". Values are
// appended raw unless decoded is set, in which case percent escapes and '+' are
// decoded. A parameter whose value fails to decode is reported absent, and a
// parameter whose key fails to decode is skipped.
//
// dst doubles as scratch space for decoding candidate keys, so AppendQuery only
// allocates when dst lacks the capacity to hold the longest key it inspects.
func (exch *Exchange) AppendQuery(dst []byte, key string, decoded bool) (valueAppended []byte, present bool) {
const plusAsSpace = true // Query strings are form encoded, unlike paths.
base := len(dst)
rawkey, rawval, rest := httpraw.NextQueryPair(exch.RequestQuery())
for ; rawkey != nil; rawkey, rawval, rest = httpraw.NextQueryPair(rest) {
if b2s(rawkey) != key {
// Key may be encoded: decode it over dst's free space and compare.
// A decoded key cannot appear raw, so this cannot alias a real key.
dst = slices.Grow(dst, len(rawkey))
scratch := dst[base : base+len(rawkey)]
n, err := httpraw.CopyDecodedPercentURL(scratch, rawkey, plusAsSpace)
if err != nil || b2s(scratch[:n]) != key {
continue // Malformed or different key, keep looking.
}
}
if len(rawval) == 0 {
return dst[:base], true // Flag or empty value, nothing to append.
}
dst = slices.Grow(dst, len(rawval))
if !decoded {
return append(dst[:base], rawval...), true
}
n, err := httpraw.CopyDecodedPercentURL(dst[base:base+len(rawval)], rawval, plusAsSpace)
if err != nil {
return dst[:base], false // Do not hand back half a decode.
}
return dst[:base+n], true
}
return dst[:base], false
}
func (exch *Exchange) RequestMethod() []byte {
return exch.RequestHeaderRaw().Method()
}
func (exch *Exchange) RequestConnectionClose() bool {
return exch.RequestHeaderRaw().ConnectionClose()
}
type HandlerFunc func(ex *Exchange)
type Method uint8
const (
MethUndefined Method = iota // undefined
MethGet // GET
// lol.
MethHead // HEAD
MethPost // POST
MethPut // PUT
// RFC 5789
MethPatch // PATCH
MethDelete // DELETE
MethConnect // CONNECT
MethOptions // OPTIONS
MethTrace // TRACE
MethUnknown // unknown
)
func MethodFromBytes(meth []byte) (res Method) {
if len(meth) == 0 {
return MethUndefined
}
switch unsafe.String(&meth[0], len(meth)) {
case "GET":
res = MethGet
case "HEAD":
res = MethHead
case "POST":
res = MethPost
case "PUT":
res = MethPut
case "PATCH":
res = MethPatch
case "DELETE":
res = MethDelete
case "CONNECT":
res = MethConnect
case "OPTIONS":
res = MethOptions
case "TRACE":
res = MethTrace
default:
res = MethUnknown
}
return res
}
// b2s converts byte slice to a string without memory allocation.
// See https://groups.google.com/forum/#!msg/Golang-Nuts/ENgbUzYvCuU/90yGx7GUAgAJ .
func b2s(b []byte) string {
return unsafe.String(unsafe.SliceData(b), len(b))
}
+135 -6
View File
@@ -2,50 +2,179 @@ package httphi
import (
"strings"
"unsafe"
"github.com/soypat/lneto"
"github.com/soypat/lneto/internal"
)
// Handle is a extremely low-level HTTP handling method used internally in [Router].
// 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 {
reqhdr := &exch.reqHdr
reqhdr.Reset(nil)
var consecutiveBackoffs uint
for {
n, err := reqhdr.ReadFromLimited(exch.rw, reqhdr.BufferFree())
if err != nil {
exch.readErr = err
return err
} else if n == 0 {
backoff.Do(consecutiveBackoffs)
consecutiveBackoffs++
continue
}
consecutiveBackoffs = 0
const asRequest = false
needMore, err := reqhdr.TryParse(asRequest)
if needMore {
continue // Request header split across reads, accumulate the rest.
} else if err != nil {
return err
}
break // Done!
}
// Setup Exchange fields necessary for correct functioning.
parsed := reqhdr.BufferParsed()
exch.respRemains = reqhdr.BufferReceived() - parsed
exch.respHeaderOff = uint16(parsed)
exch.respHeaderLen = 0
if len(reqhdr.Protocol()) == 0 {
// Request line with no HTTP version is a HTTP/0.9 simple-request, which
// httpraw tolerates. It is not a valid HTTP/1.1 request-line, RFC 9112 3.
exch.WriteHeader(int(StatusBadRequest))
return errNoRequestProto
}
// Mux on the request path: the query string is the handler's business.
path := reqhdr.RequestPath()
meth := reqhdr.Method()
handler := mux.LookupHandler(MethodFromBytes(meth), path)
if handler != nil {
handler(exch)
exch.FlushHeader()
} else {
exch.WriteHeader(404)
}
// TODO write response from exchange here.
return nil
}
// HandlerFunc serves a single request, playing the part of http.Handler.
// The exchange is only valid for the duration of the call: it is released to
// the router's pool on return, so a handler must not retain it nor any slice it
// handed out.
type HandlerFunc func(ex *Exchange)
// Mux resolves a request to the handler that serves it. [Handle] calls
// LookupHandler with the request-target's path, not the whole target, and
// replies 404 when it returns nil.
type Mux interface {
LookupHandler(get Method, uri []byte) HandlerFunc
}
// MuxSlice is a [Mux] backed by a slice of registered endpoints, matched by
// exact path. Lookup is linear in the number of registrations.
type MuxSlice struct {
// TODO: binary search worth it?
_handlers []struct {
method Method
uri string
path string
handler HandlerFunc
}
}
// Reset discards all registered handlers, reusing the backing array and growing
// it to fit capacity registrations.
func (sm *MuxSlice) Reset(capacity int) {
internal.SliceReuse(&sm._handlers, capacity)
}
func (sm *MuxSlice) LookupHandler(method Method, uri []byte) HandlerFunc {
// LookupHandler returns the handler registered for request path, or nil if none matches.
// The first registration matching both method and uri wins.
func (sm *MuxSlice) LookupHandler(method Method, path []byte) HandlerFunc {
for _, endpoint := range sm._handlers {
if endpoint.method != MethUndefined && endpoint.method != method {
continue
}
// Method matches.
if b2s(uri) == endpoint.uri {
if b2s(path) == endpoint.path {
return endpoint.handler
}
}
return nil
}
func (sm *MuxSlice) Handle(reg string, handler HandlerFunc) {
// Handle registers handler for reg, either a bare path matching any method or a
// method and path separated by a space, i.e: "/health" or "GET /health".
// Handle does not check for duplicate registrations: the first one added wins.
func (sm *MuxSlice) Handle(optMethodAndPath string, handler HandlerFunc) {
v := internal.SliceReclaim(&sm._handlers)
method := MethUndefined
methodOrURL, url, methodFound := strings.Cut(reg, " ")
methodOrURL, url, methodFound := strings.Cut(optMethodAndPath, " ")
if methodFound {
method = MethodFromBytes([]byte(methodOrURL))
} else {
url = methodOrURL
}
v.method = method
v.uri = url
v.path = url
v.handler = handler
}
// Method is a HTTP request method, parsed by [MethodFromBytes].
type Method uint8
const (
MethUndefined Method = iota // undefined
MethGet // GET
// lol.
MethHead // HEAD
MethPost // POST
MethPut // PUT
// RFC 5789
MethPatch // PATCH
MethDelete // DELETE
MethConnect // CONNECT
MethOptions // OPTIONS
MethTrace // TRACE
MethUnknown // unknown
)
// MethodFromBytes returns the [Method] matching meth, [MethUndefined] if meth is
// empty and [MethUnknown] if it names a method this package does not know.
// Comparison is case sensitive: methods are uppercase, RFC 9110 9.1.
func MethodFromBytes(meth []byte) (res Method) {
if len(meth) == 0 {
return MethUndefined
}
switch unsafe.String(&meth[0], len(meth)) {
case "GET":
res = MethGet
case "HEAD":
res = MethHead
case "POST":
res = MethPost
case "PUT":
res = MethPut
case "PATCH":
res = MethPatch
case "DELETE":
res = MethDelete
case "CONNECT":
res = MethConnect
case "OPTIONS":
res = MethOptions
case "TRACE":
res = MethTrace
default:
res = MethUnknown
}
return res
}
// b2s converts byte slice to a string without memory allocation.
// See https://groups.google.com/forum/#!msg/Golang-Nuts/ENgbUzYvCuU/90yGx7GUAgAJ .
func b2s(b []byte) string {
return unsafe.String(unsafe.SliceData(b), len(b))
}
+304
View File
@@ -0,0 +1,304 @@
package httphi
import (
"errors"
"io"
"log/slog"
"sync"
"sync/atomic"
"time"
"github.com/soypat/lneto"
"github.com/soypat/lneto/internal"
)
//go:generate stringer -type Method,status -linecomment -output stringers.go
// reconfigureWait bounds how long [Router.Configure] waits for the previous
// generation to stop serving before reusing its exchange buffers.
const reconfigureWait = 10 * time.Millisecond
var (
errNoRequestProto = errors.New("httphi: request line with no HTTP version")
errBusyExchanges = errors.New("httphi: exchanges still serving, cannot reuse their buffers")
)
type conn = io.ReadWriteCloser
// Router hosts concurrent safe data.
type Router struct {
mu sync.Mutex
gen atomic.Uint32
numGoro int
reqBuf int
respBuf int
normalizeKeys bool
pendingConns chan job
mux Mux
globbuf []byte
exchs []Exchange
freeList *Exchange
backoff lneto.BackoffStrategy
log *slog.Logger
}
type job struct {
exch *Exchange
}
// RouterConfig configures a [Router]. See [Router.Configure].
type RouterConfig struct {
// FixedNumGoroutines must be set to either -1 (freely allocate new goroutines) or to the number of goroutines
// to spawn on [Router.Configure] being called.
FixedNumGoroutines int
// RequestBufferSize determines the buffer allocated
// for processing requests.
RequestBufferSize int
// ResponseMinBufferSize determines buffer allocated for processing responses.
// Response buffer will reuse unused request memory so this is not a strict limit.
ResponseMinBufferSize int
NormalizeOutgoingKeys bool
MaxAwaitingConns int
Backoff lneto.BackoffStrategy
Mux Mux
Logger *slog.Logger
}
// 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 ||
!workerMode && cfg.FixedNumGoroutines != -1 {
return lneto.ErrInvalidConfig
} else if cfg.Backoff == nil {
return lneto.ErrMissingHALConfig
}
return nil
}
func (cfg RouterConfig) workerMode() bool {
return cfg.FixedNumGoroutines > 0
}
// Teardown stops fixed goroutines.
func (r *Router) TeardownGoroutines() {
r.mu.Lock()
defer r.mu.Unlock()
r.teardownGoroutinesLocked()
}
// teardownGoroutinesLocked closes the job queue fixed goroutines feed from.
// Requires r.mu held so that a concurrent [Router.Handle] cannot be enqueueing
// on the channel being closed.
func (r *Router) teardownGoroutinesLocked() {
r.gen.Add(1)
if r.pendingConns != nil {
close(r.pendingConns)
r.pendingConns = nil
}
}
// Configure prepares the router to serve connections, tearing down the previous
// generation of goroutines and exchanges first. In worker mode it spawns
// [RouterConfig.FixedNumGoroutines] goroutines and allocates their exchange
// buffers up front, so the router's memory use does not grow with load.
//
// 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 {
if err := cfg.Validate(); err != nil {
return err
}
r.mu.Lock()
defer r.mu.Unlock()
r.teardownGoroutinesLocked()
gen := r.gen.Load()
numgoro := cfg.FixedNumGoroutines
workerMode := cfg.workerMode()
r.reqBuf = cfg.RequestBufferSize
r.respBuf = cfg.ResponseMinBufferSize
r.mux = cfg.Mux
r.log = cfg.Logger
r.normalizeKeys = cfg.NormalizeOutgoingKeys
if !workerMode {
r.backoff = cfg.Backoff
r.numGoro = 0
r.pendingConns = nil
return nil
}
if workerMode {
jobqueue := make(chan job, cfg.MaxAwaitingConns)
if gen > 1 {
// Exchange buffers below are reused: the previous generation must be
// done serving before they may be handed to the new one.
err := r.awaitIdleExchangesLocked(reconfigureWait)
if err != nil {
return err
}
}
r.freeList = nil // Freelist entries point into the buffers reused below.
internal.SliceReuse(&r.exchs, numgoro)
r.exchs = r.exchs[:numgoro]
rawBuflen := cfg.RequestBufferSize + cfg.ResponseMinBufferSize
internal.SliceReuse(&r.globbuf, numgoro*rawBuflen)
for i := range numgoro {
// TODO exchange buffer alloc
goff := i * rawBuflen
r.exchs[i].Configure(r.globbuf[goff:goff+rawBuflen], cfg.RequestBufferSize, cfg.NormalizeOutgoingKeys)
go r.goroWorker(gen, jobqueue, cfg.Backoff, cfg.Mux)
}
r.pendingConns = jobqueue
r.numGoro = numgoro
}
return nil
}
// awaitIdleExchangesLocked waits up to maxWait for exchanges of the previous
// generation to finish serving so their buffers may be reused. Requires r.mu
// held; the lock is released while waiting since [Router.freeExch] needs it to
// free the exchanges being waited on.
func (r *Router) awaitIdleExchangesLocked(maxWait time.Duration) error {
const pollInterval = time.Millisecond
for waited := time.Duration(0); ; waited += pollInterval {
busy := false
for i := range r.exchs {
if r.exchs[i].used.Load() {
busy = true
break
}
}
if !busy {
return nil
} else if waited >= maxWait {
return errBusyExchanges
}
r.mu.Unlock()
time.Sleep(pollInterval)
r.mu.Lock()
}
}
// Handle takes ownership of conn and serves one exchange on it, closing it when
// done. It does not block on the exchange: the connection is handed to a
// goroutine and Handle returns immediately.
//
// Handle returns [lneto.ErrExhausted] when no exchange is free and
// [lneto.ErrPacketDrop] when the job queue is full. Both leave conn untouched
// and unclosed for the caller to dispose of: dropping is how a router with
// fixed memory applies backpressure.
func (r *Router) Handle(conn conn) error {
// Exchange acquisition and the configuration it is served with must be read
// under the same lock: [Router.Configure] may run concurrently.
r.mu.Lock()
exch := r.getExchLocked(conn)
numGoro, backoff, mux := r.numGoro, r.backoff, r.mux
if exch == nil {
r.mu.Unlock()
return lneto.ErrExhausted
} else if numGoro == 0 {
r.mu.Unlock()
go r.goroHandle(exch, backoff, mux)
return nil
}
// Enqueue under the lock: [Router.Configure] closes pendingConns while
// holding it, so an unlocked send could land on a closed channel. The send
// never blocks, so holding the lock cannot stall a worker.
var enqueued bool
select {
case r.pendingConns <- job{exch: exch}:
enqueued = true
default:
// pendingConns cannot store another Conn, we drop and return error.
exch.used.Store(false) // release.
}
r.mu.Unlock()
if enqueued {
return nil
}
return lneto.ErrPacketDrop
}
func (r *Router) goroWorker(gen uint32, queue chan job, backoff lneto.BackoffStrategy, mux Mux) {
for job := range queue {
exch := job.exch
if gen != r.gen.Load() {
return
} else if exch == nil {
panic("httplo: unreachable nil job")
}
r.goroHandle(exch, backoff, mux)
}
}
func (r *Router) goroHandle(exch *Exchange, backoff lneto.BackoffStrategy, mux Mux) {
defer r.freeExch(exch)
err := Handle(exch, mux, backoff)
if err != nil {
if exch.readErr != nil {
r.error("goroHandle:ReadFromLimited", slog.String("err", err.Error()))
} else {
r.error("goroHandle:TryParse?", slog.String("err", err.Error()))
}
}
}
func (r *Router) freeExch(exch *Exchange) {
const freelistMaxDepth = 5
r.mu.Lock()
depth := 0
for node := r.freeList; node != nil && depth < freelistMaxDepth; node = node.nextFree {
depth++
}
if depth < freelistMaxDepth {
// Push at head: appending at the tail would drop every node past the
// depth limit instead of dropping the exchange we cannot store.
exch.nextFree = r.freeList
r.freeList = exch
} else {
exch.nextFree = nil // Freelist full, exchange is dropped.
}
exch.Release()
r.mu.Unlock()
}
// getExchLocked returns an exchange acquired on conn. Requires r.mu held.
func (r *Router) getExchLocked(conn conn) (exch *Exchange) {
if r.freeList != nil {
// Successor must be read before Acquire: Acquire clears nextFree, so
// popping afterwards would truncate the freelist to the popped node.
next := r.freeList.nextFree
if r.freeList.Acquire(conn) {
exch = r.freeList
r.freeList = next
return exch
}
}
for i := range r.exchs {
if r.exchs[i].Acquire(conn) {
return &r.exchs[i]
}
}
if r.numGoro == 0 {
exch := new(Exchange)
exch.Configure(make([]byte, r.respBuf+r.reqBuf), r.reqBuf, r.normalizeKeys)
exch.Acquire(conn) // Fresh exchange, CAS cannot fail.
return exch
}
return nil
}
func (r *Router) error(msg string, attrs ...slog.Attr) {
internal.LogAttrs(r.log, slog.LevelError, msg, attrs...)
}
func (r *Router) info(msg string, attrs ...slog.Attr) {
internal.LogAttrs(r.log, slog.LevelInfo, msg, attrs...)
}
+1 -1
View File
@@ -202,7 +202,7 @@ func TestRouterRequestVisibleToHandler(t *testing.T) {
var gotMethod, gotURI, gotHost string
sm.Handle("GET /index.html", func(ex *Exchange) {
gotMethod = string(ex.RequestMethod())
gotURI = string(ex.RequestURI())
gotURI = string(ex.RequestTarget())
gotHost = string(ex.RequestHeader("Host"))
ex.WriteHeader(200)
})
+4
View File
@@ -74,6 +74,8 @@ func (c *Cookie) Parse() error {
return nil
}
// ForEach iterates over the cookie's key-value pairs, stopping on the first
// error returned by cb and returning it.
func (c *Cookie) ForEach(cb func(key, value []byte) error) error {
nc := len(c.kvs)
for i := range nc {
@@ -100,6 +102,8 @@ func (c *Cookie) Get(key string) []byte {
return nil
}
// HasKeyOrSingleValue returns true if the cookie contains a pair with the given
// key or a valueless attribute with the given text, i.e: "Secure" or "HttpOnly".
func (c *Cookie) HasKeyOrSingleValue(keyOrSingleValue string) bool {
nc := len(c.kvs)
for i := range nc {
+31 -24
View File
@@ -15,6 +15,9 @@ const (
strClose = "close"
)
// Flags is a bitset of signals gathered while parsing or building a header,
// such as a status code having been set or the peer requesting connection
// close. See [Header.Flags].
type Flags uint16
const (
@@ -27,9 +30,9 @@ const (
flagReaderEOF
// set if [Header.SetStatus] or [Header.SetStatusInt] has been called.
FlagStatusSet
FlagReaderError
)
// HasAny returns true if any of the argument flags are set.
func (f Flags) HasAny(checkThese Flags) bool {
return f&checkThese != 0
}
@@ -45,9 +48,9 @@ type Header struct {
hbuf headerBuf
// Request fields.
method headerSlice
requestURI headerSlice
proto headerSlice
method headerSlice
requestTarget headerSlice
proto headerSlice
// Response fields.
statusCode headerSlice
@@ -108,7 +111,7 @@ func (h *Header) TryParse(asResponse bool) (needMoreData bool, err error) {
} else if h.flags.HasAny(flagMangledBuffer) {
return false, errMangledBuffer
}
if asResponse && h.statusCode.len == 0 || !asResponse && h.requestURI.start == 0 {
if asResponse && h.statusCode.len == 0 || !asResponse && h.requestTarget.start == 0 {
err = h.parseFirstLine(asResponse)
if err != nil {
return err == errNeedMore, err
@@ -351,37 +354,38 @@ func (h *Header) SetMethod(method string) {
h.method = h.reuseOrAppend(h.method, method)
}
// SetRequestURI sets RequestURI for the first HTTP request line.
func (h *Header) SetRequestURI(requestURI string) {
h.requestURI = h.reuseOrAppend(h.requestURI, requestURI)
// SetRequestTarget sets request-target (URI) for the first HTTP request line.
func (h *Header) SetRequestTarget(requestTarget string) {
h.requestTarget = h.reuseOrAppend(h.requestTarget, requestTarget)
}
// RequestURI returns RequestURI from the first HTTP request line.
func (h *Header) RequestURI() []byte {
return h.getNonEmptyValue(h.requestURI)
// RequestTarget returns a view of the request-target (URI) of the first HTTP request line.
// Called Request-URI in the obsolete RFC 2616, renamed request-target by RFC 9112.
func (h *Header) RequestTarget() []byte {
return h.getNonEmptyValue(h.requestTarget)
}
// RequestPath returns the request URI up to the query string, i.e: "/search"
// for "/search?q=go". Returns the whole URI if it contains no query string.
// RequestPath returns the request-target (URI) up to the query string, i.e: "/search"
// for "/search?q=go". Returns the whole target if it contains no query string.
func (h *Header) RequestPath() []byte {
uri := h.RequestURI()
query := bytes.IndexByte(uri, '?')
target := h.RequestTarget()
query := bytes.IndexByte(target, '?')
if query < 0 {
return uri
return target
}
return uri[:query]
return target[:query]
}
// RequestQuery returns the request URI's query string as it appears on the
// RequestQuery returns the request-target (URI) query string as it appears on the
// wire, percent-encoded and with '+' undecoded, i.e: "q=go" for "/search?q=go".
// Returns nil if the URI has no query string. Iterate it with [NextQueryPair].
// Returns nil if the target has no query string. Iterate it with [NextQueryPair].
func (h *Header) RequestQuery() []byte {
uri := h.RequestURI()
start := bytes.IndexByte(uri, '?')
target := h.RequestTarget()
start := bytes.IndexByte(target, '?')
if start < 0 {
return nil
}
return uri[start+1:]
return target[start+1:]
}
// NextQueryPair splits the leading key-value pair off a query string and returns
@@ -461,7 +465,7 @@ func (h *Header) AppendRequest(dst []byte) ([]byte, error) {
proto := h.Protocol()
if h.flags.HasAny(flagOOMReached) {
return dst, errOOM
} else if h.requestURI.len == 0 || h.method.len == 0 {
} else if h.requestTarget.len == 0 || h.method.len == 0 {
return dst, errNeedMethodURI
} else if len(proto) == 0 {
return dst, errNoProto
@@ -473,7 +477,7 @@ func (h *Header) AppendRequest(dst []byte) ([]byte, error) {
} else {
dst = append(dst, method...)
}
uri := h.RequestURI()
uri := h.RequestTarget()
dst = append(dst, ' ')
dst = append(dst, uri...)
@@ -531,6 +535,9 @@ func (h *Header) AppendHeaders(dst []byte) []byte {
return dst
}
// String returns the header's wire representation, as a request if it has a
// request line and as a response otherwise. Returns the error text if neither
// can be built. Allocates, so it is meant for debugging and logging only.
func (h *Header) String() string {
buf, err := h.AppendRequest(nil)
if err != nil {
+5 -5
View File
@@ -50,8 +50,8 @@ func TestHeaderParseRequest(t *testing.T) {
if string(hdr.Method()) != wantMethod {
t.Errorf("want method %s, got %q", wantMethod, hdr.Method())
}
if !bytes.Equal(hdr.RequestURI(), []byte(wantURI)) {
t.Errorf("want request URI %q, got %q", wantURI, hdr.RequestURI())
if !bytes.Equal(hdr.RequestTarget(), []byte(wantURI)) {
t.Errorf("want request URI %q, got %q", wantURI, hdr.RequestTarget())
}
contentLength, _ := strconv.Atoi(string(hdr.Get("Content-Length")))
if contentLength != len(wantMessage) {
@@ -356,7 +356,7 @@ func TestHeaderSetOverwrite(t *testing.T) {
var h Header
h.Reset(nil)
h.SetMethod("GET")
h.SetRequestURI("/")
h.SetRequestTarget("/")
h.SetProtocol("HTTP/1.1")
h.Set("Host", "first.example.com")
@@ -485,7 +485,7 @@ func TestHeader_AddFullBufferNoPanic(t *testing.T) {
h.Reset(buf)
h.EnableBufferGrowth(false)
h.SetMethod("GET")
h.SetRequestURI("/")
h.SetRequestTarget("/")
h.SetProtocol("HTTP/1.1")
defer func() {
@@ -531,7 +531,7 @@ func TestHeader_SetIntOverwrite(t *testing.T) {
var h Header
h.Reset(nil)
h.SetMethod("GET")
h.SetRequestURI("/")
h.SetRequestTarget("/")
h.SetProtocol("HTTP/1.1")
h.SetInt("Content-Length", 100, 10)
+1 -1
View File
@@ -106,7 +106,7 @@ func (h *Header) parseFirstLine(asResponse bool) (err error) {
if asResponse {
h.statusCode, h.statusText, h.flags, err = h.hbuf.parseFirstLineResponse(h.flags)
} else {
h.method, h.requestURI, h.proto, h.flags, err = h.hbuf.parseFirstLineRequest(h.flags)
h.method, h.requestTarget, h.proto, h.flags, err = h.hbuf.parseFirstLineRequest(h.flags)
}
return err
}
+7 -7
View File
@@ -52,8 +52,8 @@ func TestTryParse_IncrementalRequest(t *testing.T) {
if string(hdr.Method()) != "GET" {
t.Errorf("method = %q; want GET", hdr.Method())
}
if string(hdr.RequestURI()) != "/index.html" {
t.Errorf("URI = %q; want /index.html", hdr.RequestURI())
if string(hdr.RequestTarget()) != "/index.html" {
t.Errorf("URI = %q; want /index.html", hdr.RequestTarget())
}
// Verify headers via ForEach.
@@ -432,7 +432,7 @@ func TestHeader_RequestRoundTrip(t *testing.T) {
hdr.Reset(make([]byte, 0, 256))
hdr.SetProtocol("HTTP/1.1")
hdr.SetMethod("POST")
hdr.SetRequestURI("/api/data")
hdr.SetRequestTarget("/api/data")
hdr.Add("Host", "example.com")
hdr.Add("Content-Type", "application/json")
@@ -454,8 +454,8 @@ func TestHeader_RequestRoundTrip(t *testing.T) {
if string(hdr2.Method()) != "POST" {
t.Errorf("re-parsed method = %q; want POST", hdr2.Method())
}
if string(hdr2.RequestURI()) != "/api/data" {
t.Errorf("re-parsed URI = %q; want /api/data", hdr2.RequestURI())
if string(hdr2.RequestTarget()) != "/api/data" {
t.Errorf("re-parsed URI = %q; want /api/data", hdr2.RequestTarget())
}
if string(hdr2.Get("Host")) != "example.com" {
t.Errorf("re-parsed Host = %q; want example.com", hdr2.Get("Host"))
@@ -504,8 +504,8 @@ func TestParseRequest_NoProtocol(t *testing.T) {
if string(hdr.Method()) != "GET" {
t.Errorf("method = %q; want GET", hdr.Method())
}
if string(hdr.RequestURI()) != "/simple" {
t.Errorf("URI = %q; want /simple", hdr.RequestURI())
if string(hdr.RequestTarget()) != "/simple" {
t.Errorf("URI = %q; want /simple", hdr.RequestTarget())
}
if hdr.Protocol() != nil {
t.Errorf("protocol should be nil for version-less request, got %q", hdr.Protocol())
+1 -1
View File
@@ -32,7 +32,7 @@ func FuzzStackPacketHTTP(f *testing.F) {
}
hdr.SetMethod("GET")
hdr.SetProtocol("HTTP/1.1")
hdr.SetRequestURI("/")
hdr.SetRequestTarget("/")
data := hdr.AppendHeaders(nil)
pktnum := 0