diff --git a/README.md b/README.md index bb3e05d..e69aad4 100644 --- a/README.md +++ b/README.md @@ -21,6 +21,7 @@ Bluetooth is in progress, along with more processors. - TCP and UDP Berkeley sockets API - Raw Ethernet frame send/receive - MQTT client support (uses [`natiu-mqtt`](https://github.com/soypat/natiu-mqtt)) +- HTTP concurrency throttling for memory-constrained devices (`httputil` package) - QEMU simulation target for ESP32-C3 ## How to use @@ -38,6 +39,7 @@ import ( "tinygo.org/x/drivers/netdev" nl "tinygo.org/x/drivers/netlink" + "tinygo.org/x/espradio/httputil" link "tinygo.org/x/espradio/netlink" ) @@ -69,7 +71,9 @@ func main() { h, _ := link.Addr() host := h.String() println("HTTP server listening on http://" + host + port) - err = http.ListenAndServe(host+port, nil) + + // Throttle to 2 concurrent requests to avoid OOM on ESP32. + err = http.ListenAndServe(host+port, httputil.ThrottleHandler(2, nil)) for err != nil { println("error:", err.Error()) time.Sleep(5 * time.Second) @@ -361,6 +365,22 @@ AP: rems RSSI -79 Starts the ESP32 radio. +## httputil + +The `httputil` package provides HTTP utilities for memory-constrained devices. The main utility is `ThrottleHandler`, which limits concurrent HTTP request processing to a fixed number of slots. When all slots are in use, excess requests are immediately rejected with `503 Service Unavailable` instead of being queued — this prevents unbounded goroutine/memory growth on ESP32-class hardware. + +```go +import "tinygo.org/x/espradio/httputil" + +mux := http.NewServeMux() +mux.HandleFunc("/", handler) + +// Allow at most 2 concurrent requests; excess get 503. +err := http.ListenAndServe(":80", httputil.ThrottleHandler(2, mux)) +``` + +All responses include `Connection: close` to discourage HTTP/1.1 keep-alive pipelining which consumes memory for idle connections. + ## Architecture ```mermaid diff --git a/examples/apwebserver/apwebserver.go b/examples/apwebserver/apwebserver.go index 2b377aa..f8dc3df 100644 --- a/examples/apwebserver/apwebserver.go +++ b/examples/apwebserver/apwebserver.go @@ -16,6 +16,7 @@ import ( "tinygo.org/x/drivers/netdev" "tinygo.org/x/espradio" + "tinygo.org/x/espradio/httputil" link "tinygo.org/x/espradio/netlink" ) @@ -65,7 +66,7 @@ func main() { http.Handle("/on", logRequest(LED_ON)) println("HTTP server listening on http://" + apIP + port) - err = http.ListenAndServe(apIP+port, nil) + err = http.ListenAndServe(apIP+port, httputil.ThrottleHandler(lnk.TCPPoolSize(), nil)) if err != nil { failure("http.ListenAndServe: " + err.Error()) } diff --git a/examples/hello/main.go b/examples/hello/main.go index 07df5fa..873b00f 100644 --- a/examples/hello/main.go +++ b/examples/hello/main.go @@ -7,6 +7,7 @@ import ( "tinygo.org/x/drivers/netdev" nl "tinygo.org/x/drivers/netlink" + "tinygo.org/x/espradio/httputil" link "tinygo.org/x/espradio/netlink" ) @@ -38,7 +39,7 @@ func main() { h, _ := link.Addr() host := h.String() println("HTTP server listening on http://" + host + port) - err = http.ListenAndServe(host+port, nil) + err = http.ListenAndServe(host+port, httputil.ThrottleHandler(link.TCPPoolSize(), nil)) if err != nil { failure("HTTP server error: " + err.Error()) } diff --git a/examples/webserver/webserver.go b/examples/webserver/webserver.go index 8962445..23b9b6b 100644 --- a/examples/webserver/webserver.go +++ b/examples/webserver/webserver.go @@ -13,6 +13,7 @@ import ( "tinygo.org/x/drivers/netdev" nl "tinygo.org/x/drivers/netlink" + "tinygo.org/x/espradio/httputil" link "tinygo.org/x/espradio/netlink" ) @@ -54,7 +55,7 @@ func main() { h, _ := link.Addr() host := h.String() println("HTTP server listening on http://" + host + port) - err = http.ListenAndServe(host+port, nil) + err = http.ListenAndServe(host+port, httputil.ThrottleHandler(link.TCPPoolSize(), nil)) if err != nil { failure("http.ListenAndServe: " + err.Error()) } diff --git a/httputil/httputil.go b/httputil/httputil.go new file mode 100644 index 0000000..15fc965 --- /dev/null +++ b/httputil/httputil.go @@ -0,0 +1,48 @@ +// Package httputil provides HTTP utilities for memory-constrained espradio devices. +// +// The primary utility is [ThrottleHandler], a concurrency limiter designed for +// ESP32-class microcontrollers where RAM is scarce and unbounded goroutine +// creation from incoming HTTP requests can exhaust the heap. It uses a +// non-blocking semaphore so excess requests are shed immediately (503) rather +// than queued, keeping memory usage deterministic. +// +// Typical usage on an ESP32 with ~200 KB usable RAM: +// +// mux := http.NewServeMux() +// mux.HandleFunc("/", handler) +// err := http.ListenAndServe(":80", httputil.ThrottleHandler(2, mux)) +package httputil + +import ( + "net/http" +) + +// ThrottleHandler returns an http.Handler that limits concurrent request +// processing to maxConn. When all slots are occupied, excess requests receive +// 503 Service Unavailable immediately with minimal memory allocation. +// +// All responses include a "Connection: close" header to discourage HTTP/1.1 +// keep-alive pipelining on memory-constrained devices. +// +// If handler is nil, [http.DefaultServeMux] is used. If maxConn <= 0 it is +// clamped to 1. +func ThrottleHandler(maxConn int, handler http.Handler) http.Handler { + if handler == nil { + handler = http.DefaultServeMux + } + if maxConn <= 0 { + maxConn = 1 + } + sem := make(chan struct{}, maxConn) + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + select { + case sem <- struct{}{}: + defer func() { <-sem }() + w.Header().Set("Connection", "close") + handler.ServeHTTP(w, r) + default: + w.Header().Set("Connection", "close") + http.Error(w, "busy", http.StatusServiceUnavailable) + } + }) +} diff --git a/httputil/httputil_test.go b/httputil/httputil_test.go new file mode 100644 index 0000000..64e65c7 --- /dev/null +++ b/httputil/httputil_test.go @@ -0,0 +1,163 @@ +package httputil + +import ( + "net/http" + "net/http/httptest" + "sync" + "sync/atomic" + "testing" +) + +func TestThrottleHandler_LimitsConcurrency(t *testing.T) { + const maxConn = 2 + var active int64 + var maxSeen int64 + + gate := make(chan struct{}) + inner := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + cur := atomic.AddInt64(&active, 1) + defer atomic.AddInt64(&active, -1) + // track peak concurrency + for { + old := atomic.LoadInt64(&maxSeen) + if cur <= old || atomic.CompareAndSwapInt64(&maxSeen, old, cur) { + break + } + } + <-gate // block until test releases + w.WriteHeader(http.StatusOK) + }) + + h := ThrottleHandler(maxConn, inner) + + // Launch maxConn requests that will block in handler. + var wg sync.WaitGroup + for i := 0; i < maxConn; i++ { + wg.Add(1) + go func() { + defer wg.Done() + req := httptest.NewRequest(http.MethodGet, "/", nil) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + }() + } + + // Wait until both slots occupied. + for atomic.LoadInt64(&active) < int64(maxConn) { + // spin-wait; in practice converges instantly + } + + // Extra request should get 503 (non-blocking). + req := httptest.NewRequest(http.MethodGet, "/", nil) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + if rec.Code != http.StatusServiceUnavailable { + t.Errorf("expected 503, got %d", rec.Code) + } + + // Release blocked handlers. + close(gate) + wg.Wait() + + if atomic.LoadInt64(&maxSeen) > int64(maxConn) { + t.Errorf("peak concurrency %d exceeded maxConn %d", maxSeen, maxConn) + } +} + +func TestThrottleHandler_SetsConnectionClose(t *testing.T) { + inner := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + w.WriteHeader(http.StatusOK) + }) + h := ThrottleHandler(1, inner) + + req := httptest.NewRequest(http.MethodGet, "/", nil) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + + if got := rec.Header().Get("Connection"); got != "close" { + t.Errorf("Connection header = %q, want %q", got, "close") + } +} + +func TestThrottleHandler_503SetsConnectionClose(t *testing.T) { + gate := make(chan struct{}) + inner := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + <-gate + }) + h := ThrottleHandler(1, inner) + + // Occupy the single slot. + go func() { + req := httptest.NewRequest(http.MethodGet, "/", nil) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + }() + + // Wait for slot to be taken (handler blocks on gate). + // Send another request synchronously — it should get 503. + // We need the first goroutine to actually enter the handler. + // Use a small sync mechanism: + occupied := make(chan struct{}) + innerWithSignal := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + close(occupied) + <-gate + }) + h2 := ThrottleHandler(1, innerWithSignal) + + go func() { + req := httptest.NewRequest(http.MethodGet, "/first", nil) + rec := httptest.NewRecorder() + h2.ServeHTTP(rec, req) + }() + <-occupied + + req := httptest.NewRequest(http.MethodGet, "/second", nil) + rec := httptest.NewRecorder() + h2.ServeHTTP(rec, req) + + if rec.Code != http.StatusServiceUnavailable { + t.Fatalf("expected 503, got %d", rec.Code) + } + if got := rec.Header().Get("Connection"); got != "close" { + t.Errorf("503 response Connection header = %q, want %q", got, "close") + } + close(gate) +} + +func TestThrottleHandler_NilHandlerUsesDefaultMux(t *testing.T) { + // Register a handler on DefaultServeMux for this test. + pattern := "/httputil-test-nil-handler" + http.HandleFunc(pattern, func(w http.ResponseWriter, r *http.Request) { + w.Write([]byte("ok")) + }) + + h := ThrottleHandler(1, nil) + req := httptest.NewRequest(http.MethodGet, pattern, nil) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + + if rec.Code != http.StatusOK { + t.Errorf("expected 200, got %d", rec.Code) + } +} + +func TestThrottleHandler_MaxConnClampedToOne(t *testing.T) { + var called int64 + inner := http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + atomic.AddInt64(&called, 1) + w.WriteHeader(http.StatusOK) + }) + + // maxConn=0 should be treated as 1. + h := ThrottleHandler(0, inner) + req := httptest.NewRequest(http.MethodGet, "/", nil) + rec := httptest.NewRecorder() + h.ServeHTTP(rec, req) + + if rec.Code != http.StatusOK { + t.Errorf("expected 200, got %d", rec.Code) + } + if atomic.LoadInt64(&called) != 1 { + t.Error("handler was not called") + } +} diff --git a/netlink/ap.go b/netlink/ap.go index bbfa73e..065894e 100644 --- a/netlink/ap.go +++ b/netlink/ap.go @@ -145,17 +145,19 @@ func (n *Esplink) NetConnectAP(params APConnectParams) error { println("Esplink NetConnectAP: stack started, addr:", params.StaticAddr.String()) } n.stackloop.Do(func() { + poolCfg := xnet.TCPPoolConfig{ + PoolSize: 2, + QueueSize: 4, + TxBufSize: 4096, + RxBufSize: 1024, + EstablishedTimeout: 10 * time.Second, + ClosingTimeout: 5 * time.Second, + NewBackoff: func() lneto.BackoffStrategy { return pollBackoff }, + } gostack := n.netstack.LnetoStack().StackGo(pollBackoff, xnet.StackGoConfig{ - ListenerPoolConfig: xnet.TCPPoolConfig{ - PoolSize: 2, - QueueSize: 4, - TxBufSize: 4096, - RxBufSize: 1024, - EstablishedTimeout: 10 * time.Second, - ClosingTimeout: 5 * time.Second, - NewBackoff: func() lneto.BackoffStrategy { return pollBackoff }, - }, + ListenerPoolConfig: poolCfg, }) + n.tcpPoolSize = int(poolCfg.PoolSize) n.berkeley = *xnet.NewBerkeleyStack(gostack.Socket) go handleStack(espstack) }) diff --git a/netlink/netlink.go b/netlink/netlink.go index 1fb4d2f..42b3569 100644 --- a/netlink/netlink.go +++ b/netlink/netlink.go @@ -26,14 +26,22 @@ type Esplink struct { params *nl.ConnectParams notifyCb func(nl.Event) - netstack *espradio.Stack - berkeley xnet.StackBerkeley - stackloop sync.Once + netstack *espradio.Stack + berkeley xnet.StackBerkeley + stackloop sync.Once + tcpPoolSize int // ArenaPoolSize overrides the default arena pool size (bytes). Zero uses target default. ArenaPoolSize int } +// TCPPoolSize returns the maximum number of concurrent TCP connections +// supported by the listener pool. Use this value with espradio.ThrottleHandler +// to limit in-flight HTTP requests and prevent out-of-memory crashes. +func (n *Esplink) TCPPoolSize() int { + return n.tcpPoolSize +} + func (n *Esplink) rstack() xnet.StackRetrying { return n.netstack.LnetoStack().StackRetrying(pollBackoff) } @@ -122,18 +130,19 @@ func (n *Esplink) NetConnect(params *nl.ConnectParams) error { } n.stackloop.Do(func() { // Start stack goroutine once. + poolCfg := xnet.TCPPoolConfig{ + PoolSize: 2, + QueueSize: 4, + TxBufSize: 4096, + RxBufSize: 1024, + EstablishedTimeout: 2 * time.Second, + ClosingTimeout: 2 * time.Second, + NewBackoff: func() lneto.BackoffStrategy { return pollBackoff }, + } gostack := n.netstack.LnetoStack().StackGo(pollBackoff, xnet.StackGoConfig{ - ListenerPoolConfig: xnet.TCPPoolConfig{ - PoolSize: 2, - QueueSize: 4, - TxBufSize: 4096, - RxBufSize: 1024, - EstablishedTimeout: 2 * time.Second, - ClosingTimeout: 2 * time.Second, - NewBackoff: func() lneto.BackoffStrategy { return pollBackoff }, - }, + ListenerPoolConfig: poolCfg, }) - + n.tcpPoolSize = int(poolCfg.PoolSize) n.berkeley = *xnet.NewBerkeleyStack(gostack.Socket) go handleStack(espstack) })