lneto/http/httphi/router.go

package httphi

import (
	"errors"
	"io"
	"log/slog"
	"math"
	"sync"
	"sync/atomic"
	"time"

	"github.com/soypat/lneto"
	"github.com/soypat/lneto/internal"
)

//go:generate stringer -type Method -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")
	errBadRequestProto = errors.New("httphi: unsupported HTTP version in request line")
	errBusyExchanges   = errors.New("httphi: exchanges still serving, cannot reuse their buffers")
	errRouterTornDown  = errors.New("httphi: router torn down, configure it before serving")

	errNotFormEncoded            = errors.New("httphi: request body is not application/x-www-form-urlencoded")
	errNotMultipart              = errors.New("httphi: request body is not multipart/form-data")
	errUnsupportedTransferCoding = errors.New("httphi: transfer coding not decoded, read the body directly")
)

type conn = io.ReadWriteCloser

// Router serves HTTP connections handed to it with [Router.Handle], routing
// each request to a handler found through its [Mux]. It plays the part of
// http.Server minus the listening: accepting connections is the caller's job,
// which is what lets the same router run over a TCP stack, a socket or a test
// pipe.
//
// A Router owns the exchanges and goroutines that serve connections and sizes
// both at [Router.Configure] time, so serving load costs no allocation and
// bounded memory. Connections arriving with nothing left to serve them are
// refused rather than queued, see [Router.Handle].
//
// Methods are safe for concurrent use. The zero value is not usable: configure
// it first.
type Router struct {
	mu              sync.Mutex
	gen             atomic.Uint32
	numGoro         int
	reqBuf          int
	respBuf         int
	reqNumHeaderCap int
	maxPathValues   int
	normalizeKeys   bool
	pendingConns    chan job
	mux             Mux

	globbuf  []byte
	globpath []PathValue
	exchs    []Exchange
	freeList *Exchange

	log *slog.Logger
}

// job is a connection waiting on an exchange for a worker goroutine to serve it.
type job struct {
	exch *Exchange
}

// RouterConfig configures a [Router]. See [Router.Configure].
//
// Each field below opens with Required, Conditional or Optional followed by the
// constraint in brackets, that being what [RouterConfig.Validate] rejects on.
type RouterConfig struct {
	// Required [-1 or >0] number of goroutines to spawn on [Router.Configure],
	// -1 meaning allocate them freely per connection instead.
	FixedNumGoroutines int
	// Required [>=32, sum with ResponseHeaderMinBufferSize <=65535] buffer for the
	// request header: request-target (URI), protocol and key/value pairs.
	RequestHeaderBufferSize int
	// Required [>=2, <=65535] buffer for response headers. Reuses unused request
	// memory so it is not a strict limit, and the status line does not count
	// towards it. Once consumed [Exchange.StageHeader] appends no more fields.
	ResponseHeaderMinBufferSize int
	// Required [>0] request header key/value pairs to parse before failing with
	// [StatusRequestHeaderFieldsTooLarge].
	RequestNumHeaderKVCap int
	// Optional [any] normalization of response header field keys as they are
	// staged, i.e: "content-type" becomes "Content-Type".
	NormalizeOutgoingKeys bool
	// Optional [nil disables] sink for failed exchanges.
	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()
	switch {
	case !workerMode && cfg.FixedNumGoroutines != -1,
		cfg.RequestNumHeaderKVCap <= 0,
		cfg.RequestHeaderBufferSize < minRequestHeaderBuffer,
		cfg.ResponseHeaderMinBufferSize < minResponseHeaderBuffer,
		cfg.ResponseHeaderMinBufferSize > maxExchangeBuffer,
		cfg.RequestHeaderBufferSize > maxExchangeBuffer-cfg.ResponseHeaderMinBufferSize:
		return lneto.ErrInvalidConfig
	}
	if workerMode {
		// Buffer sizes are bounded by maxExchangeBuffer above so the sum cannot
		// overflow; the products below allocate and can.
		exchBuf := cfg.RequestHeaderBufferSize + cfg.ResponseHeaderMinBufferSize
		if cfg.FixedNumGoroutines > math.MaxInt/exchBuf ||
			cfg.FixedNumGoroutines > math.MaxInt/cfg.RequestNumHeaderKVCap {
			return lneto.ErrInvalidConfig
		}
	}
	return nil
}

func (cfg RouterConfig) workerMode() bool {
	return cfg.FixedNumGoroutines > 0
}

// Shutdown stops the router once in-flight exchanges finish; until
// [Router.Configure] is called again, connections are refused with a
// non-nil error. Configure calls it before installing a new generation.
func (r *Router) Shutdown() {
	r.mu.Lock()
	defer r.mu.Unlock()
	r.shutdownLocked()
}

// shutdownLocked is [Router.Shutdown] but without locking requirement.
func (r *Router) shutdownLocked() {
	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(mux Mux, cfg RouterConfig) error {
	if err := cfg.Validate(); err != nil {
		return err
	}
	r.mu.Lock()
	defer r.mu.Unlock()
	r.shutdownLocked()
	gen := r.gen.Load()
	numgoro := cfg.FixedNumGoroutines
	workerMode := cfg.workerMode()

	r.reqNumHeaderCap = cfg.RequestNumHeaderKVCap
	r.reqBuf = cfg.RequestHeaderBufferSize
	r.respBuf = cfg.ResponseHeaderMinBufferSize
	r.mux = mux
	r.log = cfg.Logger
	maxPathValues := mux.MaxPathValues()
	if maxPathValues < 0 {
		return errors.New("Mux paths must be registered before configuring Router")
	}
	r.maxPathValues = maxPathValues
	r.normalizeKeys = cfg.NormalizeOutgoingKeys
	// Freelist entries were sized by the outgoing configuration: recycling one
	// would serve a request with buffer limits cfg never asked for.
	r.freeList = nil
	if !workerMode {
		r.numGoro = 0
		r.pendingConns = nil
		return nil
	}
	if workerMode {
		jobqueue := make(chan job, cfg.FixedNumGoroutines)
		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
			}
		}
		internal.SliceReuse(&r.exchs, numgoro)
		r.exchs = r.exchs[:numgoro]
		rawBuflen := cfg.RequestHeaderBufferSize + cfg.ResponseHeaderMinBufferSize
		internal.SliceReuse(&r.globbuf, numgoro*rawBuflen)
		internal.SliceReuse(&r.globpath, numgoro*maxPathValues)

		for i := range numgoro {
			goff := i * rawBuflen
			poff := i * maxPathValues
			r.exchs[i].Configure(ExchangeConfig{
				RawBuf:                r.globbuf[goff : goff+rawBuflen],
				RequestBufferLim:      cfg.RequestHeaderBufferSize,
				NumHeaderKVCap:        cfg.RequestNumHeaderKVCap,
				NormalizeOutgoingKeys: cfg.NormalizeOutgoingKeys,
				NoRequestBufferGrowth: true, // Hard memory limit.
				PathValuesBuf:         r.globpath[poff : poff+maxPathValues],
			})
			go r.goroWorker(gen, jobqueue, 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].acquired.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,
// [lneto.ErrPacketDrop] when the queue of connections awaiting a goroutine is
// full, and an error when the router's goroutines have been torn down. On every
// one of them conn is left untouched and unclosed for the caller to dispose of:
// refusing connections is how a router with fixed memory applies backpressure.
func (r *Router) Handle(conn io.ReadWriteCloser) 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()
	numGoro, mux := r.numGoro, r.mux
	gen := r.gen.Load() // Generation whose buffers the exchange below is sized by.
	if numGoro > 0 && r.pendingConns == nil {
		// Goroutines torn down: refuse before claiming an exchange.
		r.mu.Unlock()
		return errRouterTornDown
	}
	exch := r.getExchLocked(conn)
	if exch == nil {
		r.mu.Unlock()
		return lneto.ErrExhausted
	} else if numGoro == 0 {
		r.mu.Unlock()
		go r.goroHandle(gen, exch, 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.acquired.Store(false) // release.
	}
	r.mu.Unlock()
	if enqueued {
		return nil
	}
	return lneto.ErrPacketDrop
}

func (r *Router) goroWorker(gen uint32, queue chan job, mux Mux) {
	for job := range queue {
		exch := job.exch
		if exch == nil {
			panic("httphi: unreachable nil job")
		} else if gen != r.gen.Load() {
			// Not released with freeExch since generation torn down,
			// new buffer may have been allocated for Exchanges.
			exch.Release()
			continue
		}
		r.goroHandle(gen, exch, mux)
	}
}

func (r *Router) goroHandle(gen uint32, exch *Exchange, mux Mux) {
	defer r.freeExch(gen, exch)
	err := Handle(exch, mux, nil)
	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()))
		}
	}
}

// freeExch releases exch and offers it to the freelist for reuse. gen is the
// generation exch was acquired under: an exchange outliving its generation is
// dropped, its buffers being sized by a configuration the router no longer
// serves and possibly carved out of a globbuf it no longer owns.
func (r *Router) freeExch(gen uint32, exch *Exchange) {
	const freelistMaxDepth = 5
	r.mu.Lock()
	if gen != r.gen.Load() {
		exch.Release()
		r.mu.Unlock()
		return
	}
	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
		}
	}
	// Unbounded mode is stored as zero, see [Router.Configure].
	workerMode := r.numGoro > 0
	if workerMode {
		for i := range r.exchs {
			if r.exchs[i].Acquire(conn) {
				return &r.exchs[i]
			}
		}
	} else {
		// Unbounded growth mode when r.numGoro==0.
		exch := new(Exchange)
		exch.Configure(ExchangeConfig{
			RawBuf:                make([]byte, r.respBuf+r.reqBuf),
			RequestBufferLim:      r.reqBuf,
			NumHeaderKVCap:        r.reqNumHeaderCap,
			NormalizeOutgoingKeys: r.normalizeKeys,
			NoRequestBufferGrowth: true,
			PathValuesBuf:         make([]PathValue, r.maxPathValues),
		})
		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...)
}