mirror of
https://github.com/luxfi/fasthttp.git
synced 2026-09-24 00:24:04 +00:00
Two defects, both in server.go, both fatal: TCPKeepalive is fasthttp own on/off flag (bool); the interval field is TCPKeepalivePeriod. Assigning 3*time.Minute to the flag is a type error, so the package failed to build for every consumer — luxfi/adx among them. HandleRequest computed `duration := time.Since(start)` under a comment reading "Increment counters atomically" and then did nothing with it. An unused local is also a compile error, so even fixing the first defect left this one. The three metrics fields (requestsInFlight, requestsTotal, requestDuration) were declared and never written anywhere. Rather than delete `duration` to silence the compiler, this wires the counters the comment described and adds Stats() to read them — write-only counters would be no better than not keeping any. Builds clean, go vet clean. Co-authored-by: Hanzo Dev <[email protected]>
202 lines
5.2 KiB
Go
202 lines
5.2 KiB
Go
// Package fasthttp provides optimized HTTP server functionality
|
|
// Inspired by VictoriaMetrics' fast HTTP handling
|
|
package fasthttp
|
|
|
|
import (
|
|
"context"
|
|
"net"
|
|
"net/http"
|
|
"sync"
|
|
"sync/atomic"
|
|
"time"
|
|
|
|
"github.com/valyala/fasthttp"
|
|
"github.com/valyala/fasthttp/fasthttpadaptor"
|
|
)
|
|
|
|
// Server wraps fasthttp.Server with additional optimizations
|
|
type Server struct {
|
|
fasthttp.Server
|
|
|
|
// Metrics tracking
|
|
requestsInFlight int64
|
|
requestsTotal int64
|
|
requestDuration time.Duration
|
|
|
|
// Connection pooling
|
|
connPool sync.Pool
|
|
|
|
// Buffer pools for zero-allocation handling
|
|
readBufPool sync.Pool
|
|
writeBufPool sync.Pool
|
|
}
|
|
|
|
// NewServer creates a new optimized HTTP server
|
|
func NewServer(handler http.Handler) *Server {
|
|
s := &Server{
|
|
Server: fasthttp.Server{
|
|
Handler: fasthttpadaptor.NewFastHTTPHandler(handler),
|
|
Name: "LuxFastHTTP",
|
|
},
|
|
connPool: sync.Pool{
|
|
New: func() interface{} {
|
|
return &net.TCPConn{}
|
|
},
|
|
},
|
|
readBufPool: sync.Pool{
|
|
New: func() interface{} {
|
|
return make([]byte, 4096)
|
|
},
|
|
},
|
|
writeBufPool: sync.Pool{
|
|
New: func() interface{} {
|
|
return make([]byte, 4096)
|
|
},
|
|
},
|
|
}
|
|
|
|
// Configure fasthttp optimizations. TCPKeepalive is the on/off switch; the
|
|
// interval is TCPKeepalivePeriod. Assigning a Duration to the flag is what
|
|
// kept this package from compiling.
|
|
s.Server.TCPKeepalive = true
|
|
s.Server.TCPKeepalivePeriod = 3 * time.Minute
|
|
s.Server.ReadTimeout = 10 * time.Second
|
|
s.Server.WriteTimeout = 10 * time.Second
|
|
s.Server.IdleTimeout = 30 * time.Second
|
|
s.Server.MaxConnsPerIP = 1000
|
|
s.Server.MaxRequestsPerConn = 1000
|
|
|
|
return s
|
|
}
|
|
|
|
// ListenAndServe starts the server with optimizations
|
|
func (s *Server) ListenAndServe(addr string) error {
|
|
ln, err := net.Listen("tcp", addr)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
// Set TCP optimizations
|
|
if tcpLn, ok := ln.(*net.TCPListener); ok {
|
|
if err := setTCPOptions(tcpLn); err != nil {
|
|
// Log error but continue
|
|
}
|
|
}
|
|
|
|
return s.Server.Serve(ln)
|
|
}
|
|
|
|
// setTCPOptions applies TCP-level optimizations
|
|
func setTCPOptions(ln *net.TCPListener) error {
|
|
// These would be platform-specific optimizations
|
|
// Similar to VictoriaMetrics' network optimizations
|
|
return nil
|
|
}
|
|
|
|
// HandleRequest provides optimized request handling
|
|
func (s *Server) HandleRequest(ctx *fasthttp.RequestCtx) {
|
|
start := time.Now()
|
|
atomic.AddInt64(&s.requestsInFlight, 1)
|
|
|
|
// Use pooled buffers
|
|
ctx.SetUserValue("start_time", start)
|
|
|
|
// Call the standard handler through adapter
|
|
s.Server.Handler(ctx)
|
|
|
|
// Update metrics. These three fields were declared and never written — the
|
|
// comment that used to stand here was the only implementation, which left
|
|
// `duration` unused and the package uncompilable.
|
|
atomic.AddInt64(&s.requestsInFlight, -1)
|
|
atomic.AddInt64(&s.requestsTotal, 1)
|
|
atomic.StoreInt64((*int64)(&s.requestDuration), int64(time.Since(start)))
|
|
}
|
|
|
|
// Stats reports the counters HandleRequest maintains: requests currently being
|
|
// served, requests completed since start, and the most recent request duration.
|
|
// Without a reader the counters would be write-only, which is no better than
|
|
// not keeping them.
|
|
func (s *Server) Stats() (inFlight, total int64, last time.Duration) {
|
|
return atomic.LoadInt64(&s.requestsInFlight),
|
|
atomic.LoadInt64(&s.requestsTotal),
|
|
time.Duration(atomic.LoadInt64((*int64)(&s.requestDuration)))
|
|
}
|
|
|
|
// WithContext wraps the server with context support
|
|
func (s *Server) WithContext(ctx context.Context) *Server {
|
|
// Add context cancellation support
|
|
go func() {
|
|
<-ctx.Done()
|
|
s.Server.Shutdown()
|
|
}()
|
|
|
|
return s
|
|
}
|
|
|
|
// OptimizedHandler wraps an http.Handler with fasthttp optimizations
|
|
func OptimizedHandler(handler http.Handler) fasthttp.RequestHandler {
|
|
return func(ctx *fasthttp.RequestCtx) {
|
|
// Apply VictoriaMetrics-style optimizations:
|
|
// 1. Zero-allocation request parsing
|
|
// 2. Optimized header handling
|
|
// 3. Fast path for common routes
|
|
|
|
// Use the adapter but with our optimizations
|
|
fasthttpadaptor.NewFastHTTPHandler(handler)(ctx)
|
|
}
|
|
}
|
|
|
|
// NewRouter creates an optimized router
|
|
func NewRouter() *Router {
|
|
return &Router{
|
|
routes: make(map[string]RouteHandler),
|
|
}
|
|
}
|
|
|
|
// Router provides optimized routing
|
|
type Router struct {
|
|
mu sync.RWMutex
|
|
routes map[string]RouteHandler
|
|
}
|
|
|
|
// RouteHandler defines a route handler type
|
|
type RouteHandler func(ctx *fasthttp.RequestCtx)
|
|
|
|
// AddRoute adds a route to the router
|
|
func (r *Router) AddRoute(method, path string, handler RouteHandler) {
|
|
r.mu.Lock()
|
|
defer r.mu.Unlock()
|
|
|
|
key := method + " " + path
|
|
r.routes[key] = handler
|
|
}
|
|
|
|
// HandleRequest handles an incoming request
|
|
func (r *Router) HandleRequest(ctx *fasthttp.RequestCtx) {
|
|
path := string(ctx.Path())
|
|
method := string(ctx.Method())
|
|
|
|
r.mu.RLock()
|
|
handler, ok := r.routes[method+" "+path]
|
|
r.mu.RUnlock()
|
|
|
|
if ok {
|
|
handler(ctx)
|
|
return
|
|
}
|
|
|
|
// Fallback to not found
|
|
ctx.Error("Not Found", fasthttp.StatusNotFound)
|
|
}
|
|
|
|
// FastHTTPMiddleware provides middleware support
|
|
type FastHTTPMiddleware func(handler RouteHandler) RouteHandler
|
|
|
|
// ChainMiddleware chains multiple middleware functions
|
|
func ChainMiddleware(handler RouteHandler, middleware ...FastHTTPMiddleware) RouteHandler {
|
|
for i := len(middleware) - 1; i >= 0; i-- {
|
|
handler = middleware[i](handler)
|
|
}
|
|
return handler
|
|
}
|