several bugfixes, add internal.IntLen, round up http-linux example with new router API

examples/http-linux/conn-linux.go

@@ -5,6 +5,7 @@ package main
 import (
 	"net/netip"
 	"syscall"
+	"time"
 )
 
 // Conn wraps an accepted TCP connection from a raw Linux socket file descriptor.
@@ -47,6 +48,25 @@ func (c *Conn) Close() error {
 	return syscall.Close(c.fd)
 }
 
+// SetReadTimeout limits how long a single Read waits for data before failing
+// with [syscall.EAGAIN]. Zero blocks indefinitely. This is what stops a peer
+// that opens a connection and then stalls from holding a server worker: the
+// connection, not the HTTP handler, owns the idle policy.
+func (c *Conn) SetReadTimeout(timeout time.Duration) error {
+	return setSockTimeout(c.fd, syscall.SO_RCVTIMEO, timeout)
+}
+
+// SetWriteTimeout limits how long a single Write waits for the send buffer to
+// drain before failing with [syscall.EAGAIN]. Zero blocks indefinitely.
+func (c *Conn) SetWriteTimeout(timeout time.Duration) error {
+	return setSockTimeout(c.fd, syscall.SO_SNDTIMEO, timeout)
+}
+
+func setSockTimeout(fd, option int, timeout time.Duration) error {
+	tv := syscall.NsecToTimeval(int64(timeout))
+	return syscall.SetsockoptTimeval(fd, syscall.SOL_SOCKET, option, &tv)
+}
+
 // RemoteAddr returns the peer address of the connection.
 func (c *Conn) RemoteAddr() netip.AddrPort { return c.remote }
 

examples/http-linux/main-httplinux.go

@@ -3,15 +3,22 @@
 package main
 
 import (
+	"log/slog"
 	"os"
 	"strconv"
 	"sync/atomic"
 	"time"
 
-	"github.com/soypat/lneto/http/httpraw"
+	"github.com/soypat/lneto/http/httphi"
 )
 
-const listenPort = 8080
+const (
+	kB            = 1 << 10
+	listenPort    = 8080
+	bufferSizes   = 2 * kB
+	numGoroutines = 4
+	readTimeout   = 2 * time.Second
+)
 
 func main() {
 	if err := run(); err != nil {
@@ -27,18 +34,44 @@ func run() error {
 		return err
 	}
 	defer ln.Close()
-	println("listening on port", listenPort)
+	print("listening on http://localhost:", listenPort, "\n")
-	conn := new(Conn)
+
+	var mux httphi.MuxSlice
+	mux.Handle("GET /", homepage)
+
+	var router httphi.Router
+	err = router.Configure(httphi.RouterConfig{
+		FixedNumGoroutines:    numGoroutines,
+		RequestBufferSize:     bufferSizes,
+		ResponseMinBufferSize: bufferSizes,
+		MaxAwaitingConns:      256,
+		Backoff: func(consecutiveBackoffs uint) (sleepOrFlag time.Duration) {
+			return min(time.Second, time.Millisecond*time.Duration(consecutiveBackoffs))
+		},
+		Mux:    &mux,
+		Logger: slog.Default(),
+	})
+	if err != nil {
+		return err
+	}
+	defer router.TeardownGoroutines()
+
 	for {
+		conn := new(Conn)
 		err := ln.Accept(conn)
 		if err != nil {
 			return err
 		}
-		visits.Add(1)
+		// The connection owns the idle policy: a peer that opens a socket and
-		if err := handle(conn); err != nil {
+		// then stalls fails its read instead of holding a router goroutine.
-			println("handle:", conn.RemoteAddr().String(), err.Error())
+		conn.SetReadTimeout(readTimeout)
+		err = router.Handle(conn)
+		if err != nil {
+			// Every goroutine is busy and the queue is full. Dropping the
+			// connection is the backpressure: memory stays bounded.
+			slog.Warn("dropped connection", slog.String("remote", conn.RemoteAddr().String()), slog.String("err", err.Error()))
+			conn.Close()
 		}
-		conn.Close()
 	}
 }
 
@@ -50,58 +83,22 @@ const (
 	htmlTail = `!</blink></h1>` +
 		`<font color="#FF00FF">Sign my guestbook!</font>` +
 		`<br><hr>Best viewed in Netscape Navigator</center></body></html>`
+	// maxPage bounds the rendered page: both halves plus the visitor number.
+	maxPage = len(htmlHead) + 20 + len(htmlTail)
 )
 
-const maxHTTPHeader = 1024
+// visits counts served requests. Handlers run on the router's goroutines, so
+// every visitor gets their own number.
+var visits atomic.Uint64
 
-var (
+func homepage(exch *httphi.Exchange) {
-	hdr     httpraw.Header
+	var page [maxPage]byte
-	httpbuf [maxHTTPHeader]byte
+	n := copy(page[:], htmlHead)
-	htmlbuf [512]byte
+	n += len(strconv.AppendUint(page[n:n], visits.Add(1), 10))
-	visits  atomic.Uint64
+	n += copy(page[n:], htmlTail)
-)
 
-func handle(conn *Conn) error {
+	exch.SetHeader("Content-Type", "text/html")
-	hdr.Reset(httpbuf[:0])
+	exch.SetHeaderInt("Content-Length", int64(n), 10)
-	hdr.EnableBufferGrowth(false)    // Limit memory to buffer capacity.
+	exch.WriteHeader(int(httphi.StatusOK))
-	const incomingIsResponse = false // We get HTTP requests from clients.
+	exch.Write(page[:n])
-	deadline := time.Now().Add(50 * time.Millisecond)
-	for time.Until(deadline) > 0 {
-		if _, err := hdr.ReadFromLimited(conn, maxHTTPHeader); err != nil {
-			return err
-		}
-		needmoredata, err := hdr.TryParse(incomingIsResponse)
-		if err != nil {
-			return err
-		}
-		if needmoredata {
-			continue
-		}
-		break
-	}
-	println("\n\n================\n\n", hdr.String())
-	if time.Since(deadline) > 0 {
-		print("DEADLINE EXCEED: ", hdr.BufferParsed(), "/", hdr.BufferReceived(), " bytes parsed/read\n")
-		return nil
-	}
-	// Prepare tacky HTML response.
-	n := copy(htmlbuf[:], htmlHead)
-	n += len(strconv.AppendUint(htmlbuf[n:n], visits.Load(), 10))
-	n += copy(htmlbuf[n:], htmlTail)
-	contentLen := n
-	hdr.Reset(httpbuf[:0])
-	hdr.SetProtocol("HTTP/1.1")
-	hdr.SetStatus("200", "OK")
-	hdr.Set("Content-Type", "text/html")
-	hdr.SetInt("Content-Length", int64(contentLen), 10)
-	// Here we do some buffer juggling. We use remaining space
-	// of HTTP Header buffer to write the response that will be written over the wire.
-	respbuf := httpbuf[hdr.BufferUsed():]
-	header, err := hdr.AppendResponse(respbuf[:0])
-	if err != nil {
-		return err
-	}
-	conn.Write(header)
-	_, err = conn.Write(htmlbuf[:contentLen])
-	return err
 }

http/httphi/exchange_test.go

@@ -188,7 +188,7 @@ func TestHandleRequestFields(t *testing.T) {
 	} {
 		t.Run(test.name, func(t *testing.T) {
 			var gotMethod, gotURI, gotHost string
-			var sm sliceMux
+			var sm MuxSlice
 			sm.Handle(test.wantURI, func(ex *Exchange) {
 				gotMethod = string(ex.RequestMethod())
 				gotURI = string(ex.RequestURI())
@@ -223,7 +223,7 @@ func TestHandleMalformedRequest(t *testing.T) {
 	} {
 		t.Run(test.name, func(t *testing.T) {
 			var handled bool
-			var sm sliceMux
+			var sm MuxSlice
 			sm.Handle("/", func(ex *Exchange) { handled = true })
 			conn := newConn(test.request)
 			conn.Hangup()
@@ -240,7 +240,7 @@ func TestHandleMalformedRequest(t *testing.T) {
 
 // No registered handler must yield 404, not an empty response.
 func TestHandleNoHandler(t *testing.T) {
-	var sm sliceMux
+	var sm MuxSlice
 	sm.Handle("GET /", func(ex *Exchange) { t.Error("handler must not run") })
 	conn := serve(t, "GET /nowhere HTTP/1.1\r\nHost: h\r\n\r\n", &sm)
 	const want = "HTTP/1.1 404 Not Found\r\n\r\n"
@@ -251,7 +251,7 @@ func TestHandleNoHandler(t *testing.T) {
 
 // A handler that writes nothing must still produce a valid response.
 func TestHandleSilentHandler(t *testing.T) {
-	var sm sliceMux
+	var sm MuxSlice
 	sm.Handle("/", func(ex *Exchange) {})
 	conn := serve(t, "GET / HTTP/1.1\r\nHost: h\r\n\r\n", &sm)
 	const want = "HTTP/1.1 200 OK\r\n\r\n"
@@ -265,7 +265,7 @@ func TestExchangeReadBody(t *testing.T) {
 	const body = "message body"
 	var got string
 	var readErr error
-	var sm sliceMux
+	var sm MuxSlice
 	sm.Handle("POST /", func(ex *Exchange) {
 		dst := make([]byte, len(body))
 		n, err := ex.ReadBody(dst)
@@ -315,7 +315,7 @@ func TestExchangeSetHeaderExactFit(t *testing.T) {
 // Handle never closes the connection, on any outcome: the caller owns it so
 // that error policy and connection reuse stay the caller's decision.
 func TestHandleLeavesConnOpen(t *testing.T) {
-	var sm sliceMux
+	var sm MuxSlice
 	sm.Handle("GET /", staticPage(t, "ok"))
 	for _, request := range []string{
 		"GET / HTTP/1.1\r\nHost: h\r\n\r\n",         // Served.
@@ -337,7 +337,7 @@ func TestHandleLeavesConnOpen(t *testing.T) {
 // Ownership must not carry over: the next connection the exchange serves is
 // the router's again and must be closed on Release.
 func TestExchangeHijackOwnership(t *testing.T) {
-	var sm sliceMux
+	var sm MuxSlice
 	var hijackErr error
 	sm.Handle("GET /", func(ex *Exchange) {
 		_, _, hijackErr = ex.HijackRaw(nil)
@@ -370,7 +370,7 @@ func TestExchangeHijackOwnership(t *testing.T) {
 // read until the conn itself reports failure, so a stalled peer ends the
 // exchange through the conn's deadline instead of pinning the exchange.
 func TestHandleIdlePeerEndsOnConnDeadline(t *testing.T) {
-	var sm sliceMux
+	var sm MuxSlice
 	sm.Handle("/", func(ex *Exchange) { t.Error("handler must not run on partial request") })
 	conn := newConn("GET / HTTP") // Peer stalls mid request line, never hangs up.
 	conn.SetDeadline(time.Now().Add(10 * time.Millisecond))
@@ -415,3 +415,40 @@ func TestExchangeWriteHeaderFlushFails(t *testing.T) {
 		t.Errorf("want nothing on the wire, got %q", got)
 	}
 }
+
+func TestExchangeSetHeaderInt(t *testing.T) {
+	for _, test := range []struct {
+		value int64
+		base  int
+		want  string // Header block emitted after the status line.
+	}{
+		{value: 1234, base: 10, want: "N:1234\r\n\r\n"},
+		{value: 0, base: 10, want: "N:0\r\n\r\n"},
+		{value: -42, base: 10, want: "N:-42\r\n\r\n"},
+		{value: 255, base: 16, want: "N:ff\r\n\r\n"},
+		{value: 9223372036854775807, base: 10, want: "N:9223372036854775807\r\n\r\n"},
+		{value: -9223372036854775808, base: 10, want: "N:-9223372036854775808\r\n\r\n"},
+		{value: 1, base: 2, want: "\r\n"},  // Below base 10, dropped.
+		{value: 1, base: 37, want: "\r\n"}, // Above base 36, dropped.
+	} {
+		conn := newConn("")
+		exch := newExchange(t, conn, 256, false)
+		exch.SetHeaderInt("N", test.value, test.base)
+		exch.WriteHeader(200)
+		got, _ := strings.CutPrefix(conn.ViewWritten(), "HTTP/1.1 200 OK\r\n")
+		if got != test.want {
+			t.Errorf("value %d base %d: want %q, got %q", test.value, test.base, test.want, got)
+		}
+	}
+}
+
+// SetHeaderInt must format into the response buffer without allocating.
+func TestExchangeSetHeaderIntNoAlloc(t *testing.T) {
+	exch := newExchange(t, newConn(""), 256, false)
+	allocs := testing.AllocsPerRun(100, func() {
+		exch.SetHeaderInt("Content-Length", 1234567890, 10)
+	})
+	if allocs != 0 {
+		t.Errorf("SetHeaderInt allocated %v times, want 0", allocs)
+	}
+}

http/httphi/httphi.go

@@ -17,9 +17,9 @@ import (
 
 //go:generate stringer -type Method,status -linecomment -output stringers.go
 
-// defaultReconfigureWait is how long [Router.Configure] waits on a busy
+// reconfigureWait bounds how long [Router.Configure] waits for the previous
-// previous generation when [RouterConfig.MaxReconfigureWait] is unset.
+// generation to stop serving before reusing its exchange buffers.
-const defaultReconfigureWait = 100 * time.Millisecond
+const reconfigureWait = 10 * time.Millisecond
 
 var (
 	errNoRequestProto = errors.New("httphi: request line with no HTTP version")
@@ -51,10 +51,6 @@ type job struct {
 	exch *Exchange
 }
 
-type Mux interface {
-	LookupHandler(get Method, uri []byte) HandlerFunc
-}
-
 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.
@@ -121,6 +117,7 @@ func (r *Router) Configure(cfg RouterConfig) error {
 	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
@@ -133,7 +130,7 @@ func (r *Router) Configure(cfg RouterConfig) error {
 		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(10 * time.Millisecond)
+			err := r.awaitIdleExchangesLocked(reconfigureWait)
 			if err != nil {
 				return err
 			}
@@ -454,6 +451,33 @@ func (exch *Exchange) SetHeader(key, value string) (enoughMemory bool) {
 	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

http/httphi/mux.go

@@ -0,0 +1,51 @@
+package httphi
+
+import (
+	"strings"
+
+	"github.com/soypat/lneto/internal"
+)
+
+type Mux interface {
+	LookupHandler(get Method, uri []byte) HandlerFunc
+}
+
+type MuxSlice struct {
+	// TODO: binary search worth it?
+	_handlers []struct {
+		method  Method
+		uri     string
+		handler HandlerFunc
+	}
+}
+
+func (sm *MuxSlice) Reset(capacity int) {
+	internal.SliceReuse(&sm._handlers, capacity)
+}
+
+func (sm *MuxSlice) LookupHandler(method Method, uri []byte) HandlerFunc {
+	for _, endpoint := range sm._handlers {
+		if endpoint.method != MethUndefined && endpoint.method != method {
+			continue
+		}
+		// Method matches.
+		if b2s(uri) == endpoint.uri {
+			return endpoint.handler
+		}
+	}
+	return nil
+}
+
+func (sm *MuxSlice) Handle(reg string, handler HandlerFunc) {
+	v := internal.SliceReclaim(&sm._handlers)
+	method := MethUndefined
+	methodOrURL, url, methodFound := strings.Cut(reg, " ")
+	if methodFound {
+		method = MethodFromBytes([]byte(methodOrURL))
+	} else {
+		url = methodOrURL
+	}
+	v.method = method
+	v.uri = url
+	v.handler = handler
+}

http/httphi/router_test.go

@@ -11,7 +11,6 @@ import (
 	"time"
 
 	"github.com/soypat/lneto"
-	"github.com/soypat/lneto/internal"
 )
 
 // rwconn is a in-memory conn. The router handles connections on another
@@ -128,6 +127,7 @@ func (r *rwconn) AddReadable(b []byte) {
 	defer r.mu.Unlock()
 	r.readable.Write(b)
 }
+
 // SetDeadline makes reads and writes past t fail, as a conn with a read
 // deadline set would.
 func (r *rwconn) SetDeadline(t time.Time) {
@@ -149,41 +149,7 @@ func (r *rwconn) ViewWritten() string {
 	return r.written.String()
 }
 
-var _ Mux = (*sliceMux)(nil)
+var _ Mux = (*MuxSlice)(nil)
-
-type sliceMux struct {
-	_handlers []struct {
-		method  Method
-		uri     string
-		handler HandlerFunc
-	}
-}
-
-func (sm *sliceMux) LookupHandler(method Method, uri []byte) HandlerFunc {
-	for _, endpoint := range sm._handlers {
-		if endpoint.method != MethUndefined && endpoint.method != method {
-			continue
-		}
-		// Method matches.
-		if b2s(uri) == endpoint.uri {
-			return endpoint.handler
-		}
-	}
-	return nil
-}
-func (sm *sliceMux) Handle(reg string, handler HandlerFunc) {
-	v := internal.SliceReclaim(&sm._handlers)
-	method := MethUndefined
-	methodOrURL, url, methodFound := strings.Cut(reg, " ")
-	if methodFound {
-		method = MethodFromBytes([]byte(methodOrURL))
-	} else {
-		url = methodOrURL
-	}
-	v.method = method
-	v.uri = url
-	v.handler = handler
-}
 
 func configSynchronousRouter(t *testing.T, router *Router, bufferSize int, mux Mux) {
 	err := router.Configure(RouterConfig{
@@ -204,7 +170,7 @@ func TestRouterGet(t *testing.T) {
 	const bufferSize = 1024
 	const expectResponse = "its time"
 	var (
-		sm     sliceMux
+		sm     MuxSlice
 		router Router
 	)
 	sm.Handle("GET /", staticPage(t, expectResponse))
@@ -230,7 +196,7 @@ func TestRouterGet(t *testing.T) {
 func TestRouterRequestVisibleToHandler(t *testing.T) {
 	const bufferSize = 1024
 	var (
-		sm     sliceMux
+		sm     MuxSlice
 		router Router
 	)
 	var gotMethod, gotURI, gotHost string
@@ -276,7 +242,7 @@ func TestRouterMux(t *testing.T) {
 	} {
 		t.Run(test.name, func(t *testing.T) {
 			var (
-				sm     sliceMux
+				sm     MuxSlice
 				router Router
 			)
 			sm.Handle("GET /", staticPage(t, "root"))
@@ -309,7 +275,7 @@ func TestRouterSplitRequest(t *testing.T) {
 	const bufferSize = 1024
 	const expectResponse = "split ok"
 	var (
-		sm     sliceMux
+		sm     MuxSlice
 		router Router
 	)
 	sm.Handle("GET /", staticPage(t, expectResponse))
@@ -344,7 +310,7 @@ func staticPage(t *testing.T, page string) HandlerFunc {
 func TestRouterConfigureHandleRace(t *testing.T) {
 	const bufferSize = 1024
 	var (
-		sm     sliceMux
+		sm     MuxSlice
 		router Router
 	)
 	sm.Handle("GET /", staticPage(t, "ok"))
@@ -370,7 +336,7 @@ func TestRouterConfigureHandleRace(t *testing.T) {
 // sending a connection on. Connections may be dropped, but never panic.
 func TestRouterConfigureDuringWorkerHandle(t *testing.T) {
 	var (
-		sm     sliceMux
+		sm     MuxSlice
 		router Router
 	)
 	sm.Handle("GET /", staticPage(t, "ok"))

http/httpraw/parse.go

@@ -341,7 +341,7 @@ func (h *Header) appendHeader(key, value string) {
 // appendHeaderInt is appendHeader's integer counterpart: it appends key and the
 // formatted integer value as a new header field.
 func (h *Header) appendHeaderInt(key string, value int64, base int) {
-	n := intLen(value, base)
+	n := internal.IntLen(value, base)
 	if !h.reserve(len(key) + n) {
 		return // Drop and flag OOM; never panic.
 	}
@@ -382,7 +382,7 @@ func (h *Header) reserve(need int) bool {
 // reuseOrAppendInt writes value into tok's slot in place when it fits, avoiding
 // any buffer growth; otherwise it appends a fresh slot.
 func (h *Header) reuseOrAppendInt(tok headerSlice, value int64, base int) headerSlice {
-	n := intLen(value, base)
+	n := internal.IntLen(value, base)
 	if int(tok.len) >= n {
 		// Reuse: format directly over the existing slot. No free space needed
 		// since n <= tok.len and the slot already lives inside buf.
@@ -404,7 +404,7 @@ func (h *Header) appendInt(value int64, base, n int) headerSlice {
 }
 
 // mustAppendInt formats value into the buffer's free region and commits it.
-// The caller must have reserved at least intLen(value, base) free bytes.
+// The caller must have reserved at least internal.IntLen(value, base) free bytes.
 func (hb *headerBuf) mustAppendInt(value int64, base int) headerSlice {
 	L := len(hb.buf)
 	if L == 0 {
@@ -415,23 +415,6 @@ func (hb *headerBuf) mustAppendInt(value int64, base int) headerSlice {
 	return hb.slice(hb.buf[L : L+len(v)])
 }
 
-// intLen returns the number of bytes strconv.AppendInt would emit for value in
-// the given base (including a leading minus sign for negatives). Used to size
-// the buffer and to test whether a value fits an existing slot without writing.
-func intLen(value int64, base int) int {
-	n := 1
-	u := uint64(value)
-	if value < 0 {
-		n++    // Leading minus sign.
-		u = -u // Two's-complement magnitude; correct even for math.MinInt64.
-	}
-	for u >= uint64(base) {
-		u /= uint64(base)
-		n++
-	}
-	return n
-}
-
 func (hb *headerBuf) noKV() argsKV { return argsKV{} }
 
 func (hb *headerBuf) next(ss *scannerState) argsKV {

internal/strconv.go

@@ -0,0 +1,19 @@
+package internal
+
+// IntLen returns the number of bytes [strconv.AppendInt] emits for value in the
+// given base, including a leading minus sign for negatives. Lets callers size a
+// buffer, or test whether a value fits an existing slot, before writing a byte.
+// base must be in the range 2..36, as accepted by [strconv.AppendInt].
+func IntLen(value int64, base int) int {
+	n := 1
+	u := uint64(value)
+	if value < 0 {
+		n++    // Leading minus sign.
+		u = -u // Two's-complement magnitude; correct even for math.MinInt64.
+	}
+	for u >= uint64(base) {
+		u /= uint64(base)
+		n++
+	}
+	return n
+}