diff --git a/examples/berkeley-listener/berkeley_server.go b/examples/berkeley-listener/berkeley_server.go index f3cee5f..310dffb 100644 --- a/examples/berkeley-listener/berkeley_server.go +++ b/examples/berkeley-listener/berkeley_server.go @@ -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") diff --git a/examples/http-linux/main-httplinux.go b/examples/http-linux/main-httplinux.go index 6cc0659..c562cf9 100644 --- a/examples/http-linux/main-httplinux.go +++ b/examples/http-linux/main-httplinux.go @@ -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]) } diff --git a/examples/httpclient/main.go b/examples/httpclient/main.go index 66f0459..3c0cb34 100644 --- a/examples/httpclient/main.go +++ b/examples/httpclient/main.go @@ -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 { diff --git a/examples/httpserver/main.go b/examples/httpserver/main.go index 83dd463..7f55fff 100644 --- a/examples/httpserver/main.go +++ b/examples/httpserver/main.go @@ -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. diff --git a/examples/xcurl/main.go b/examples/xcurl/main.go index 85cd425..000d991 100644 --- a/examples/xcurl/main.go +++ b/examples/xcurl/main.go @@ -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") diff --git a/http/httphi/bench_test.go b/http/httphi/bench_test.go index e28ff5b..256bf83 100644 --- a/http/httphi/bench_test.go +++ b/http/httphi/bench_test.go @@ -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") diff --git a/http/httphi/exchange.go b/http/httphi/exchange.go new file mode 100644 index 0000000..622985a --- /dev/null +++ b/http/httphi/exchange.go @@ -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() +} diff --git a/http/httphi/exchange_test.go b/http/httphi/exchange_test.go index b69fca2..40c65b5 100644 --- a/http/httphi/exchange_test.go +++ b/http/httphi/exchange_test.go @@ -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) diff --git a/http/httphi/httphi.go b/http/httphi/httphi.go deleted file mode 100644 index 128e87a..0000000 --- a/http/httphi/httphi.go +++ /dev/null @@ -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)) -} diff --git a/http/httphi/mux.go b/http/httphi/mux.go index 9efec49..8ffd456 100644 --- a/http/httphi/mux.go +++ b/http/httphi/mux.go @@ -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)) +} diff --git a/http/httphi/router.go b/http/httphi/router.go new file mode 100644 index 0000000..2f5b94e --- /dev/null +++ b/http/httphi/router.go @@ -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...) +} diff --git a/http/httphi/router_test.go b/http/httphi/router_test.go index 17aa46b..3731d40 100644 --- a/http/httphi/router_test.go +++ b/http/httphi/router_test.go @@ -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) }) diff --git a/http/httpraw/cookie.go b/http/httpraw/cookie.go index ea8b530..6b63e8e 100644 --- a/http/httpraw/cookie.go +++ b/http/httpraw/cookie.go @@ -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 { diff --git a/http/httpraw/header.go b/http/httpraw/header.go index c32e984..3cc14ca 100644 --- a/http/httpraw/header.go +++ b/http/httpraw/header.go @@ -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 { diff --git a/http/httpraw/header_test.go b/http/httpraw/header_test.go index eaebb89..18756a1 100644 --- a/http/httpraw/header_test.go +++ b/http/httpraw/header_test.go @@ -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) diff --git a/http/httpraw/parse.go b/http/httpraw/parse.go index 846054f..cb10f01 100644 --- a/http/httpraw/parse.go +++ b/http/httpraw/parse.go @@ -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 } diff --git a/http/httpraw/parse_test.go b/http/httpraw/parse_test.go index c94616e..f0ef65c 100644 --- a/http/httpraw/parse_test.go +++ b/http/httpraw/parse_test.go @@ -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()) diff --git a/x/xnet/xnet_fuzz_test.go b/x/xnet/xnet_fuzz_test.go index 827f169..9db5503 100644 --- a/x/xnet/xnet_fuzz_test.go +++ b/x/xnet/xnet_fuzz_test.go @@ -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