mirror of
https://github.com/soypat/lneto.git
synced 2026-08-11 18:33:46 +00:00
add http/httphi
This commit is contained in:
Patricio Whittingslow
@@ -0,0 +1,486 @@
|
||||
package httphi
|
||||
|
||||
import (
|
||||
"io"
|
||||
"log/slog"
|
||||
"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
|
||||
|
||||
type conn = io.ReadWriteCloser
|
||||
|
||||
// Router hosts concurrent safe data.
|
||||
type Router struct {
|
||||
mu sync.Mutex
|
||||
gen atomic.Uint32
|
||||
numGoro int
|
||||
reqBuf int
|
||||
respBuf int
|
||||
pendingConns chan job
|
||||
mux Mux
|
||||
|
||||
globbuf []byte
|
||||
exchs []Exchange
|
||||
freeList *Exchange
|
||||
|
||||
backoff lneto.BackoffStrategy
|
||||
log *slog.Logger
|
||||
}
|
||||
|
||||
type job struct {
|
||||
exch *Exchange
|
||||
}
|
||||
|
||||
type Mux interface {
|
||||
Handler(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.
|
||||
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.gen.Add(1)
|
||||
if r.pendingConns != nil {
|
||||
close(r.pendingConns)
|
||||
}
|
||||
}
|
||||
|
||||
func (r *Router) Configure(cfg RouterConfig) error {
|
||||
if err := cfg.Validate(); err != nil {
|
||||
return err
|
||||
}
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
r.TeardownGoroutines()
|
||||
gen := r.gen.Load()
|
||||
numgoro := cfg.FixedNumGoroutines
|
||||
workerMode := cfg.workerMode()
|
||||
r.reqBuf = cfg.RequestBufferSize
|
||||
r.respBuf = cfg.ResponseMinBufferSize
|
||||
r.mux = cfg.Mux
|
||||
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 {
|
||||
// Previously existing goroutine manager, wait a bit for it to close.
|
||||
time.Sleep(5 * time.Millisecond)
|
||||
}
|
||||
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
|
||||
}
|
||||
|
||||
func (r *Router) Handle(conn conn) error {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
exch := r.getExch(conn)
|
||||
if exch == nil {
|
||||
return lneto.ErrExhausted
|
||||
}
|
||||
|
||||
if r.numGoro == 0 {
|
||||
go r.goroHandle(exch, r.backoff, r.mux)
|
||||
return nil
|
||||
}
|
||||
select {
|
||||
case r.pendingConns <- job{exch: exch}:
|
||||
default:
|
||||
exch.used.Store(false) // release.
|
||||
}
|
||||
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)
|
||||
reqhdr := &exch.reqHdr
|
||||
reqhdr.Reset(nil)
|
||||
var consecutiveBackoffs uint
|
||||
for {
|
||||
n, err := reqhdr.ReadFromLimited(exch.rw, reqhdr.BufferFree())
|
||||
if err != nil {
|
||||
r.error("goroHandle:ReadFromLimited", slog.String("err", err.Error()))
|
||||
exch.rw.Close()
|
||||
return
|
||||
} else if n == 0 {
|
||||
backoff(consecutiveBackoffs)
|
||||
consecutiveBackoffs++
|
||||
continue
|
||||
}
|
||||
consecutiveBackoffs = 0
|
||||
const asRequest = false
|
||||
needMore, err := reqhdr.TryParse(asRequest)
|
||||
if !needMore && err == nil {
|
||||
// Done!
|
||||
break
|
||||
} else if err != nil {
|
||||
r.error("goroHandle:TryParse", slog.String("err", err.Error()))
|
||||
exch.rw.Close()
|
||||
return
|
||||
}
|
||||
}
|
||||
r.info("goroHandle:headerParsedSuccess")
|
||||
// Setup Exchange fields necessary for correct functioning.
|
||||
parsed := reqhdr.BufferParsed()
|
||||
exch.respRemains = reqhdr.BufferReceived() - parsed
|
||||
exch.respHeaderOff = uint16(parsed)
|
||||
exch.respHeaderLen = 0
|
||||
// Mux URI.
|
||||
uri := reqhdr.RequestURI()
|
||||
meth := reqhdr.Method()
|
||||
handler := mux.Handler(MethodFromBytes(meth), uri)
|
||||
if handler != nil {
|
||||
handler(exch)
|
||||
}
|
||||
// Reuse request space as response header start.
|
||||
|
||||
// TODO write response from exchange here.
|
||||
exch.rw.Close()
|
||||
|
||||
}
|
||||
|
||||
func (r *Router) freeExch(exch *Exchange) {
|
||||
const freelistMaxDepth = 5
|
||||
r.mu.Lock()
|
||||
if r.freeList == nil {
|
||||
r.freeList = exch
|
||||
} else {
|
||||
node := r.freeList
|
||||
depth := 0
|
||||
for depth < freelistMaxDepth && node.nextFree != nil {
|
||||
node = node.nextFree
|
||||
depth++
|
||||
}
|
||||
node.nextFree = exch
|
||||
}
|
||||
exch.Release()
|
||||
r.mu.Unlock()
|
||||
}
|
||||
|
||||
func (r *Router) getExch(conn conn) (exch *Exchange) {
|
||||
r.mu.Lock()
|
||||
defer r.mu.Unlock()
|
||||
if r.freeList != nil {
|
||||
if r.freeList.Acquire(conn) {
|
||||
exch = r.freeList
|
||||
}
|
||||
r.freeList = r.freeList.nextFree
|
||||
}
|
||||
if exch != nil {
|
||||
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, false)
|
||||
|
||||
return exch
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (r *Router) allocBuffers(exch *Exchange) {
|
||||
|
||||
}
|
||||
|
||||
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...)
|
||||
}
|
||||
|
||||
type Exchange struct {
|
||||
used atomic.Bool
|
||||
respTopBuf [32]byte
|
||||
respTopWritten uint8
|
||||
|
||||
rawbuf []byte
|
||||
respHeaderOff uint16
|
||||
respHeaderLen uint16
|
||||
reqHdr httpraw.Header
|
||||
rw conn
|
||||
|
||||
respRemains int
|
||||
headerWritten bool
|
||||
normalizeKeys bool
|
||||
nextFree *Exchange
|
||||
}
|
||||
|
||||
func (exch *Exchange) Configure(rawbuf []byte, requestLim int, normalizeKeys bool) {
|
||||
respSize := len(rawbuf) - requestLim
|
||||
if respSize < 0 {
|
||||
panic("request lim larger than buffer")
|
||||
}
|
||||
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.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() {
|
||||
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
|
||||
if len(key)+len(value)+2 > free {
|
||||
return false
|
||||
}
|
||||
n := copy(exch.rawbuf[off:], key)
|
||||
if exch.normalizeKeys {
|
||||
httpraw.NormalizeHeaderKey(exch.rawbuf[off : off+n])
|
||||
}
|
||||
exch.rawbuf[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
|
||||
}
|
||||
|
||||
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)
|
||||
}
|
||||
|
||||
func (exch *Exchange) WriteHeader(code int) {
|
||||
if !exch.headerWritten {
|
||||
exch.StageWriteStatus(code)
|
||||
exch.FlushHeader()
|
||||
}
|
||||
}
|
||||
func (exch *Exchange) FlushHeader() (int, error) {
|
||||
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 {
|
||||
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)
|
||||
return ng + ng2, err
|
||||
}
|
||||
|
||||
func (exch *Exchange) Write(buf []byte) (int, error) {
|
||||
if !exch.headerWritten {
|
||||
exch.FlushHeader()
|
||||
}
|
||||
if len(buf) == 0 {
|
||||
return 0, nil
|
||||
}
|
||||
return exch.rw.Write(buf)
|
||||
}
|
||||
|
||||
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()
|
||||
}
|
||||
|
||||
func (exch *Exchange) RequestMethod() []byte {
|
||||
return exch.RequestHeaderRaw().Method()
|
||||
}
|
||||
|
||||
func (exch *Exchange) RequestConnectionClose() bool {
|
||||
return exch.RequestHeaderRaw().ConnectionClose()
|
||||
}
|
||||
|
||||
func (exch *Exchange) ReadBody(dst []byte) (n int, _ error) {
|
||||
if exch.respRemains > 0 {
|
||||
_, err := exch.reqHdr.Body()
|
||||
if err != nil {
|
||||
return 0, 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:]
|
||||
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
|
||||
}
|
||||
|
||||
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))
|
||||
}
|
||||
Reference in New Issue
Block a user