diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 64b9d5b..a48e49d 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -14,8 +14,8 @@ jobs: strategy: fail-fast: false matrix: - otp: ['27.0', '27.1'] - python: ['3.12', '3.13'] + otp: ['27.0', '27.1', '28.0'] + python: ['3.12', '3.13', '3.14'] steps: - uses: actions/checkout@v4 @@ -66,7 +66,7 @@ jobs: - name: Set up Erlang uses: erlef/setup-beam@v1 with: - otp-version: '27.1' + otp-version: '28.0' rebar3-version: '3.24' - name: Build ex_doc @@ -82,7 +82,7 @@ jobs: - name: Set up Erlang uses: erlef/setup-beam@v1 with: - otp-version: '27.1' + otp-version: '28.0' rebar3-version: '3.24' - name: Compile diff --git a/CHANGELOG.md b/CHANGELOG.md index 97ed177..9c4f4a4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,32 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/), and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). +## [Unreleased] + +### Changed + +- **Shared Context Pool**: All mounts now share the default `py_context_router` pool + - Removed per-mount `workers` option (use global pool size instead) + - Better resource utilization across multiple mounted apps + - Simplified architecture with cached NIF refs + +- **ASGI Performance Optimizations**: + - Event loop pool for parallel ASGI task distribution + - Cached state proxies per mount_id (avoid allocation per request) + - Preloaded app modules via `py_import:ensure_imported` + - Lazy state proxy with ETS-backed callbacks + +### Performance + +- ASGI now outperforms WSGI by 11-16% across test scenarios: + - Simple requests (100 conc): ~70k req/s (+13%) + - High concurrency (500 conc): ~64k req/s (+16%) + - Sustained load (200 conc): ~71k req/s (+14%) + +### Removed + +- `workers` option from per-mount configuration (use shared pool) + ## [1.4.1] - 2026-02-25 ### Fixed diff --git a/benchmarks/asgi_large_response_app.py b/benchmarks/asgi_large_response_app.py new file mode 100644 index 0000000..2c62b16 --- /dev/null +++ b/benchmarks/asgi_large_response_app.py @@ -0,0 +1,27 @@ +# ASGI app that returns large responses to test streaming + +SMALL_BODY = b'x' * 1000 # 1KB - should buffer +LARGE_BODY = b'x' * 100000 # 100KB - should stream + +async def app(scope, receive, send): + """ASGI app with configurable response size.""" + if scope['type'] == 'http': + path = scope.get('path', '/') + + if path == '/large': + body = LARGE_BODY + else: + body = SMALL_BODY + + await send({ + 'type': 'http.response.start', + 'status': 200, + 'headers': [ + [b'content-type', b'text/plain'], + [b'content-length', str(len(body)).encode()], + ], + }) + await send({ + 'type': 'http.response.body', + 'body': body, + }) diff --git a/benchmarks/asgi_noop_app.py b/benchmarks/asgi_noop_app.py new file mode 100644 index 0000000..8d8c886 --- /dev/null +++ b/benchmarks/asgi_noop_app.py @@ -0,0 +1,17 @@ +# Minimal ASGI app - no processing + +async def app(scope, receive, send): + """Minimal ASGI app that returns Hello World.""" + if scope['type'] == 'http': + await send({ + 'type': 'http.response.start', + 'status': 200, + 'headers': [ + [b'content-type', b'text/plain'], + [b'content-length', b'13'], + ], + }) + await send({ + 'type': 'http.response.body', + 'body': b'Hello, World!', + }) diff --git a/benchmarks/bench_pooled_comparison.sh b/benchmarks/bench_pooled_comparison.sh new file mode 100755 index 0000000..14a8add --- /dev/null +++ b/benchmarks/bench_pooled_comparison.sh @@ -0,0 +1,213 @@ +#!/bin/bash +# Benchmark comparison: pooled vs non-pooled workers +# +# This script compares performance between: +# - Non-pooled: Traditional per-request Python invocation +# - Pooled: Persistent worker pool with channel-based dispatch + +set -e + +cd "$(dirname "$0")/.." + +# Check if ab is available +if ! command -v ab &> /dev/null; then + echo "Error: 'ab' (Apache Bench) not found." + echo " macOS: brew install httpd" + echo " Linux: apt-get install apache2-utils" + exit 1 +fi + +# Check if project is compiled +if [ ! -d "_build/default/lib" ]; then + echo "Compiling hornbeam..." + rebar3 compile +fi + +# Configuration +REQUESTS=10000 +CONCURRENCY=100 +PORT_NONPOOLED=8765 +PORT_POOLED=8766 +WORKERS=4 + +cleanup() { + echo "Cleaning up..." + kill $PID_NONPOOLED 2>/dev/null || true + kill $PID_POOLED 2>/dev/null || true + wait $PID_NONPOOLED 2>/dev/null || true + wait $PID_POOLED 2>/dev/null || true +} +trap cleanup EXIT + +echo "==============================================" +echo " Hornbeam Pooled vs Non-Pooled Benchmark" +echo "==============================================" +echo "" +echo "Configuration:" +echo " Requests: $REQUESTS" +echo " Concurrency: $CONCURRENCY" +echo " Workers (pooled): $WORKERS" +echo "" + +# Start non-pooled server (single-app mode) +echo "Starting non-pooled server on port $PORT_NONPOOLED..." +erl -pa _build/default/lib/*/ebin \ + -noshell \ + -eval " + application:ensure_all_started(hornbeam), + hornbeam:start(<<\"simple_app:application\">>, #{ + bind => <<\"127.0.0.1:$PORT_NONPOOLED\">>, + worker_class => wsgi, + pythonpath => [<<\"benchmarks\">>] + }). + " > /dev/null 2>&1 & +PID_NONPOOLED=$! + +# Start pooled server (multi-app mode with pool_enabled) +echo "Starting pooled server on port $PORT_POOLED..." +erl -pa _build/default/lib/*/ebin \ + -noshell \ + -eval " + application:ensure_all_started(hornbeam), + py:exec(<<\"import sys; sys.path.insert(0, 'benchmarks')\">>), + hornbeam:start(#{ + bind => <<\"127.0.0.1:$PORT_POOLED\">>, + mounts => [ + {<<\"/\">>, <<\"simple_app:application\">>, #{ + worker_class => wsgi, + workers => $WORKERS, + pool_enabled => true, + timeout => 30000 + }} + ] + }). + " > /dev/null 2>&1 & +PID_POOLED=$! + +# Wait for servers to be ready +echo "Waiting for servers to start..." +sleep 4 + +# Verify servers are running +for port in $PORT_NONPOOLED $PORT_POOLED; do + for i in {1..10}; do + if curl -s http://127.0.0.1:$port/ > /dev/null 2>&1; then + echo " Server on port $port is ready" + break + fi + if [ $i -eq 10 ]; then + echo " WARNING: Server on port $port may not be ready" + fi + sleep 0.5 + done +done + +echo "" + +# Warmup +echo "Warming up servers..." +ab -n 1000 -c 50 -k http://127.0.0.1:$PORT_NONPOOLED/ > /dev/null 2>&1 || true +ab -n 1000 -c 50 -k http://127.0.0.1:$PORT_POOLED/ > /dev/null 2>&1 || true +sleep 1 + +echo "" +echo "==============================================" +echo " Test 1: Simple Requests ($REQUESTS req, $CONCURRENCY concurrent)" +echo "==============================================" + +echo "" +echo "--- Non-Pooled ---" +RESULT_NP1=$(ab -n $REQUESTS -c $CONCURRENCY -k http://127.0.0.1:$PORT_NONPOOLED/ 2>&1) +RPS_NP1=$(echo "$RESULT_NP1" | grep "Requests per second" | awk '{print $4}') +LAT_NP1=$(echo "$RESULT_NP1" | grep "Time per request" | head -1 | awk '{print $4}') +FAIL_NP1=$(echo "$RESULT_NP1" | grep "Failed requests" | awk '{print $3}') +echo " Requests/sec: $RPS_NP1" +echo " Latency: ${LAT_NP1}ms" +echo " Failed: $FAIL_NP1" + +echo "" +echo "--- Pooled ---" +RESULT_P1=$(ab -n $REQUESTS -c $CONCURRENCY -k http://127.0.0.1:$PORT_POOLED/ 2>&1) +RPS_P1=$(echo "$RESULT_P1" | grep "Requests per second" | awk '{print $4}') +LAT_P1=$(echo "$RESULT_P1" | grep "Time per request" | head -1 | awk '{print $4}') +FAIL_P1=$(echo "$RESULT_P1" | grep "Failed requests" | awk '{print $3}') +echo " Requests/sec: $RPS_P1" +echo " Latency: ${LAT_P1}ms" +echo " Failed: $FAIL_P1" + +echo "" +echo "==============================================" +echo " Test 2: High Concurrency (5000 req, 500 concurrent)" +echo "==============================================" + +echo "" +echo "--- Non-Pooled ---" +RESULT_NP2=$(ab -n 5000 -c 500 -k http://127.0.0.1:$PORT_NONPOOLED/ 2>&1) +RPS_NP2=$(echo "$RESULT_NP2" | grep "Requests per second" | awk '{print $4}') +LAT_NP2=$(echo "$RESULT_NP2" | grep "Time per request" | head -1 | awk '{print $4}') +FAIL_NP2=$(echo "$RESULT_NP2" | grep "Failed requests" | awk '{print $3}') +echo " Requests/sec: $RPS_NP2" +echo " Latency: ${LAT_NP2}ms" +echo " Failed: $FAIL_NP2" + +echo "" +echo "--- Pooled ---" +RESULT_P2=$(ab -n 5000 -c 500 -k http://127.0.0.1:$PORT_POOLED/ 2>&1) +RPS_P2=$(echo "$RESULT_P2" | grep "Requests per second" | awk '{print $4}') +LAT_P2=$(echo "$RESULT_P2" | grep "Time per request" | head -1 | awk '{print $4}') +FAIL_P2=$(echo "$RESULT_P2" | grep "Failed requests" | awk '{print $3}') +echo " Requests/sec: $RPS_P2" +echo " Latency: ${LAT_P2}ms" +echo " Failed: $FAIL_P2" + +echo "" +echo "==============================================" +echo " Test 3: Sustained Load (20000 req, 200 concurrent)" +echo "==============================================" + +echo "" +echo "--- Non-Pooled ---" +RESULT_NP3=$(ab -n 20000 -c 200 -k http://127.0.0.1:$PORT_NONPOOLED/ 2>&1) +RPS_NP3=$(echo "$RESULT_NP3" | grep "Requests per second" | awk '{print $4}') +LAT_NP3=$(echo "$RESULT_NP3" | grep "Time per request" | head -1 | awk '{print $4}') +FAIL_NP3=$(echo "$RESULT_NP3" | grep "Failed requests" | awk '{print $3}') +echo " Requests/sec: $RPS_NP3" +echo " Latency: ${LAT_NP3}ms" +echo " Failed: $FAIL_NP3" + +echo "" +echo "--- Pooled ---" +RESULT_P3=$(ab -n 20000 -c 200 -k http://127.0.0.1:$PORT_POOLED/ 2>&1) +RPS_P3=$(echo "$RESULT_P3" | grep "Requests per second" | awk '{print $4}') +LAT_P3=$(echo "$RESULT_P3" | grep "Time per request" | head -1 | awk '{print $4}') +FAIL_P3=$(echo "$RESULT_P3" | grep "Failed requests" | awk '{print $3}') +echo " Requests/sec: $RPS_P3" +echo " Latency: ${LAT_P3}ms" +echo " Failed: $FAIL_P3" + +echo "" +echo "==============================================" +echo " Summary" +echo "==============================================" +echo "" +printf "%-25s %15s %15s %10s\n" "Test" "Non-Pooled" "Pooled" "Diff" +printf "%-25s %15s %15s %10s\n" "-------------------------" "---------------" "---------------" "----------" + +# Calculate differences (using awk for floating point) +if [ -n "$RPS_NP1" ] && [ -n "$RPS_P1" ]; then + DIFF1=$(echo "$RPS_P1 $RPS_NP1" | awk '{if($2>0) printf "%.1f%%", (($1-$2)/$2)*100; else print "N/A"}') + printf "%-25s %12s/s %12s/s %10s\n" "Simple (100 conc)" "$RPS_NP1" "$RPS_P1" "$DIFF1" +fi + +if [ -n "$RPS_NP2" ] && [ -n "$RPS_P2" ]; then + DIFF2=$(echo "$RPS_P2 $RPS_NP2" | awk '{if($2>0) printf "%.1f%%", (($1-$2)/$2)*100; else print "N/A"}') + printf "%-25s %12s/s %12s/s %10s\n" "High conc (500 conc)" "$RPS_NP2" "$RPS_P2" "$DIFF2" +fi + +if [ -n "$RPS_NP3" ] && [ -n "$RPS_P3" ]; then + DIFF3=$(echo "$RPS_P3 $RPS_NP3" | awk '{if($2>0) printf "%.1f%%", (($1-$2)/$2)*100; else print "N/A"}') + printf "%-25s %12s/s %12s/s %10s\n" "Sustained (200 conc)" "$RPS_NP3" "$RPS_P3" "$DIFF3" +fi + +echo "" +echo "Done!" diff --git a/benchmarks/bench_wsgi_vs_asgi.sh b/benchmarks/bench_wsgi_vs_asgi.sh new file mode 100755 index 0000000..15ea773 --- /dev/null +++ b/benchmarks/bench_wsgi_vs_asgi.sh @@ -0,0 +1,353 @@ +#!/bin/bash +# Benchmark comparison: WSGI vs WSGI (owngil) vs ASGI +# +# Compares performance between WSGI, WSGI with owngil, and ASGI worker classes +# +# For owngil mode (per-interpreter GIL), set PYTHON_CONFIG to point to +# a Python 3.14+ python-config script: +# export PYTHON_CONFIG=/path/to/python3.14-config +# ./benchmarks/bench_wsgi_vs_asgi.sh + +set -e + +cd "$(dirname "$0")/.." + +# Check if ab is available +if ! command -v ab &> /dev/null; then + echo "Error: 'ab' (Apache Bench) not found." + exit 1 +fi + +# Configuration +REQUESTS=10000 +CONCURRENCY=100 +PORT_WSGI=8765 +PORT_WSGI_OWNGIL=8767 +PORT_ASGI=8766 + +# Check for Python 3.12+ for owngil mode and rebuild if PYTHON_CONFIG is set +RUN_OWNGIL=false +if [ -n "$PYTHON_CONFIG" ]; then + if [ -x "$PYTHON_CONFIG" ]; then + PY_VERSION=$("$PYTHON_CONFIG" --version 2>&1 | grep -oE '[0-9]+\.[0-9]+' | head -1) + PY_MAJOR=$(echo "$PY_VERSION" | cut -d. -f1) + PY_MINOR=$(echo "$PY_VERSION" | cut -d. -f2) + if [ "$PY_MAJOR" -ge 3 ] && [ "$PY_MINOR" -ge 12 ]; then + RUN_OWNGIL=true + echo "Found Python $PY_VERSION for owngil mode" + echo "Rebuilding with PYTHON_CONFIG=$PYTHON_CONFIG..." + rm -rf _build/default/lib/erlang_python + PYTHON_CONFIG="$PYTHON_CONFIG" rebar3 compile + else + echo "Warning: PYTHON_CONFIG points to Python $PY_VERSION, owngil requires 3.12+" + echo "Compiling hornbeam..." + rebar3 compile + fi + else + echo "Warning: PYTHON_CONFIG=$PYTHON_CONFIG is not executable" + echo "Compiling hornbeam..." + rebar3 compile + fi +else + # Check if project is compiled + if [ ! -d "_build/default/lib" ]; then + echo "Compiling hornbeam..." + rebar3 compile + fi +fi + +cleanup() { + echo "Cleaning up..." + kill $PID_WSGI 2>/dev/null || true + [ "$RUN_OWNGIL" = true ] && kill $PID_WSGI_OWNGIL 2>/dev/null || true + kill $PID_ASGI 2>/dev/null || true + wait $PID_WSGI 2>/dev/null || true + [ "$RUN_OWNGIL" = true ] && wait $PID_WSGI_OWNGIL 2>/dev/null || true + wait $PID_ASGI 2>/dev/null || true +} +trap cleanup EXIT + +echo "==============================================" +echo " Hornbeam WSGI vs WSGI (owngil) vs ASGI" +echo "==============================================" +echo "" +echo "Configuration:" +echo " Requests: $REQUESTS" +echo " Concurrency: $CONCURRENCY" +echo " owngil mode: $RUN_OWNGIL" +echo "" + +# Start WSGI server +echo "Starting WSGI server on port $PORT_WSGI..." +erl -pa _build/default/lib/*/ebin \ + -noshell \ + -eval " + application:ensure_all_started(hornbeam), + hornbeam:start(<<\"simple_app:application\">>, #{ + bind => <<\"127.0.0.1:$PORT_WSGI\">>, + worker_class => wsgi, + pythonpath => [<<\"benchmarks\">>] + }). + " > /dev/null 2>&1 & +PID_WSGI=$! + +# Start WSGI server with owngil (per-interpreter GIL) if Python 3.12+ available +if [ "$RUN_OWNGIL" = true ]; then + echo "Starting WSGI (owngil) server on port $PORT_WSGI_OWNGIL..." + PYTHON_CONFIG="$PYTHON_CONFIG" erl -pa _build/default/lib/*/ebin \ + -noshell \ + -eval " + application:ensure_all_started(hornbeam), + hornbeam:start(<<\"simple_app:application\">>, #{ + bind => <<\"127.0.0.1:$PORT_WSGI_OWNGIL\">>, + worker_class => wsgi, + context_mode => owngil, + pythonpath => [<<\"benchmarks\">>] + }). + " > /dev/null 2>&1 & + PID_WSGI_OWNGIL=$! +fi + +# Start ASGI server +echo "Starting ASGI server on port $PORT_ASGI..." +erl -pa _build/default/lib/*/ebin \ + -noshell \ + -eval " + application:ensure_all_started(hornbeam), + hornbeam:start(<<\"simple_asgi_app:application\">>, #{ + bind => <<\"127.0.0.1:$PORT_ASGI\">>, + worker_class => asgi, + pythonpath => [<<\"benchmarks\">>] + }). + " > /dev/null 2>&1 & +PID_ASGI=$! + +# Wait for servers to be ready +echo "Waiting for servers to start..." +sleep 4 + +# Build list of ports to check +PORTS_TO_CHECK="$PORT_WSGI $PORT_ASGI" +[ "$RUN_OWNGIL" = true ] && PORTS_TO_CHECK="$PORT_WSGI $PORT_WSGI_OWNGIL $PORT_ASGI" + +# Verify servers are running +for port in $PORTS_TO_CHECK; do + for i in {1..10}; do + if curl -s http://127.0.0.1:$port/ > /dev/null 2>&1; then + echo " Server on port $port is ready" + break + fi + if [ $i -eq 10 ]; then + echo " WARNING: Server on port $port may not be ready" + fi + sleep 0.5 + done +done + +echo "" + +# Warmup +echo "Warming up servers..." +ab -n 1000 -c 50 -k http://127.0.0.1:$PORT_WSGI/ > /dev/null 2>&1 || true +[ "$RUN_OWNGIL" = true ] && ab -n 1000 -c 50 -k http://127.0.0.1:$PORT_WSGI_OWNGIL/ > /dev/null 2>&1 || true +ab -n 1000 -c 50 -k http://127.0.0.1:$PORT_ASGI/ > /dev/null 2>&1 || true +sleep 1 + +echo "" +echo "==============================================" +echo " Test 1: Simple Requests ($REQUESTS req, $CONCURRENCY concurrent)" +echo "==============================================" + +echo "" +echo "--- WSGI ---" +RESULT_W1=$(ab -n $REQUESTS -c $CONCURRENCY -k http://127.0.0.1:$PORT_WSGI/ 2>&1) +RPS_W1=$(echo "$RESULT_W1" | grep "Requests per second" | awk '{print $4}') +LAT_W1=$(echo "$RESULT_W1" | grep "Time per request" | head -1 | awk '{print $4}') +FAIL_W1=$(echo "$RESULT_W1" | grep "Failed requests" | awk '{print $3}') +echo " Requests/sec: $RPS_W1" +echo " Latency: ${LAT_W1}ms" +echo " Failed: $FAIL_W1" + +if [ "$RUN_OWNGIL" = true ]; then + echo "" + echo "--- WSGI (owngil) ---" + RESULT_O1=$(ab -n $REQUESTS -c $CONCURRENCY -k http://127.0.0.1:$PORT_WSGI_OWNGIL/ 2>&1) + RPS_O1=$(echo "$RESULT_O1" | grep "Requests per second" | awk '{print $4}') + LAT_O1=$(echo "$RESULT_O1" | grep "Time per request" | head -1 | awk '{print $4}') + FAIL_O1=$(echo "$RESULT_O1" | grep "Failed requests" | awk '{print $3}') + echo " Requests/sec: $RPS_O1" + echo " Latency: ${LAT_O1}ms" + echo " Failed: $FAIL_O1" +fi + +echo "" +echo "--- ASGI ---" +RESULT_A1=$(ab -n $REQUESTS -c $CONCURRENCY -k http://127.0.0.1:$PORT_ASGI/ 2>&1) +RPS_A1=$(echo "$RESULT_A1" | grep "Requests per second" | awk '{print $4}') +LAT_A1=$(echo "$RESULT_A1" | grep "Time per request" | head -1 | awk '{print $4}') +FAIL_A1=$(echo "$RESULT_A1" | grep "Failed requests" | awk '{print $3}') +echo " Requests/sec: $RPS_A1" +echo " Latency: ${LAT_A1}ms" +echo " Failed: $FAIL_A1" + +echo "" +echo "==============================================" +echo " Test 2: High Concurrency (5000 req, 500 concurrent)" +echo "==============================================" + +echo "" +echo "--- WSGI ---" +RESULT_W2=$(ab -n 5000 -c 500 -k http://127.0.0.1:$PORT_WSGI/ 2>&1) +RPS_W2=$(echo "$RESULT_W2" | grep "Requests per second" | awk '{print $4}') +LAT_W2=$(echo "$RESULT_W2" | grep "Time per request" | head -1 | awk '{print $4}') +FAIL_W2=$(echo "$RESULT_W2" | grep "Failed requests" | awk '{print $3}') +echo " Requests/sec: $RPS_W2" +echo " Latency: ${LAT_W2}ms" +echo " Failed: $FAIL_W2" + +if [ "$RUN_OWNGIL" = true ]; then + echo "" + echo "--- WSGI (owngil) ---" + RESULT_O2=$(ab -n 5000 -c 500 -k http://127.0.0.1:$PORT_WSGI_OWNGIL/ 2>&1) + RPS_O2=$(echo "$RESULT_O2" | grep "Requests per second" | awk '{print $4}') + LAT_O2=$(echo "$RESULT_O2" | grep "Time per request" | head -1 | awk '{print $4}') + FAIL_O2=$(echo "$RESULT_O2" | grep "Failed requests" | awk '{print $3}') + echo " Requests/sec: $RPS_O2" + echo " Latency: ${LAT_O2}ms" + echo " Failed: $FAIL_O2" +fi + +echo "" +echo "--- ASGI ---" +RESULT_A2=$(ab -n 5000 -c 500 -k http://127.0.0.1:$PORT_ASGI/ 2>&1) +RPS_A2=$(echo "$RESULT_A2" | grep "Requests per second" | awk '{print $4}') +LAT_A2=$(echo "$RESULT_A2" | grep "Time per request" | head -1 | awk '{print $4}') +FAIL_A2=$(echo "$RESULT_A2" | grep "Failed requests" | awk '{print $3}') +echo " Requests/sec: $RPS_A2" +echo " Latency: ${LAT_A2}ms" +echo " Failed: $FAIL_A2" + +echo "" +echo "==============================================" +echo " Test 3: Sustained Load (20000 req, 200 concurrent)" +echo "==============================================" + +echo "" +echo "--- WSGI ---" +RESULT_W3=$(ab -n 20000 -c 200 -k http://127.0.0.1:$PORT_WSGI/ 2>&1) +RPS_W3=$(echo "$RESULT_W3" | grep "Requests per second" | awk '{print $4}') +LAT_W3=$(echo "$RESULT_W3" | grep "Time per request" | head -1 | awk '{print $4}') +FAIL_W3=$(echo "$RESULT_W3" | grep "Failed requests" | awk '{print $3}') +echo " Requests/sec: $RPS_W3" +echo " Latency: ${LAT_W3}ms" +echo " Failed: $FAIL_W3" + +if [ "$RUN_OWNGIL" = true ]; then + echo "" + echo "--- WSGI (owngil) ---" + RESULT_O3=$(ab -n 20000 -c 200 -k http://127.0.0.1:$PORT_WSGI_OWNGIL/ 2>&1) + RPS_O3=$(echo "$RESULT_O3" | grep "Requests per second" | awk '{print $4}') + LAT_O3=$(echo "$RESULT_O3" | grep "Time per request" | head -1 | awk '{print $4}') + FAIL_O3=$(echo "$RESULT_O3" | grep "Failed requests" | awk '{print $3}') + echo " Requests/sec: $RPS_O3" + echo " Latency: ${LAT_O3}ms" + echo " Failed: $FAIL_O3" +fi + +echo "" +echo "--- ASGI ---" +RESULT_A3=$(ab -n 20000 -c 200 -k http://127.0.0.1:$PORT_ASGI/ 2>&1) +RPS_A3=$(echo "$RESULT_A3" | grep "Requests per second" | awk '{print $4}') +LAT_A3=$(echo "$RESULT_A3" | grep "Time per request" | head -1 | awk '{print $4}') +FAIL_A3=$(echo "$RESULT_A3" | grep "Failed requests" | awk '{print $3}') +echo " Requests/sec: $RPS_A3" +echo " Latency: ${LAT_A3}ms" +echo " Failed: $FAIL_A3" + +echo "" +echo "==============================================" +echo " Test 4: Large Response (1000 req, 50 concurrent)" +echo "==============================================" + +echo "" +echo "--- WSGI ---" +RESULT_W4=$(ab -n 1000 -c 50 -k http://127.0.0.1:$PORT_WSGI/large 2>&1) +RPS_W4=$(echo "$RESULT_W4" | grep "Requests per second" | awk '{print $4}') +LAT_W4=$(echo "$RESULT_W4" | grep "Time per request" | head -1 | awk '{print $4}') +FAIL_W4=$(echo "$RESULT_W4" | grep "Failed requests" | awk '{print $3}') +echo " Requests/sec: $RPS_W4" +echo " Latency: ${LAT_W4}ms" +echo " Failed: $FAIL_W4" + +if [ "$RUN_OWNGIL" = true ]; then + echo "" + echo "--- WSGI (owngil) ---" + RESULT_O4=$(ab -n 1000 -c 50 -k http://127.0.0.1:$PORT_WSGI_OWNGIL/large 2>&1) + RPS_O4=$(echo "$RESULT_O4" | grep "Requests per second" | awk '{print $4}') + LAT_O4=$(echo "$RESULT_O4" | grep "Time per request" | head -1 | awk '{print $4}') + FAIL_O4=$(echo "$RESULT_O4" | grep "Failed requests" | awk '{print $3}') + echo " Requests/sec: $RPS_O4" + echo " Latency: ${LAT_O4}ms" + echo " Failed: $FAIL_O4" +fi + +echo "" +echo "--- ASGI ---" +RESULT_A4=$(ab -n 1000 -c 50 -k http://127.0.0.1:$PORT_ASGI/large 2>&1) +RPS_A4=$(echo "$RESULT_A4" | grep "Requests per second" | awk '{print $4}') +LAT_A4=$(echo "$RESULT_A4" | grep "Time per request" | head -1 | awk '{print $4}') +FAIL_A4=$(echo "$RESULT_A4" | grep "Failed requests" | awk '{print $3}') +echo " Requests/sec: $RPS_A4" +echo " Latency: ${LAT_A4}ms" +echo " Failed: $FAIL_A4" + +echo "" +echo "==============================================" +echo " Summary" +echo "==============================================" +echo "" + +if [ "$RUN_OWNGIL" = true ]; then + printf "%-25s %12s %12s %12s\n" "Test" "WSGI" "WSGI(owngil)" "ASGI" + printf "%-25s %12s %12s %12s\n" "-------------------------" "------------" "------------" "------------" + + if [ -n "$RPS_W1" ] && [ -n "$RPS_O1" ] && [ -n "$RPS_A1" ]; then + printf "%-25s %9s/s %9s/s %9s/s\n" "Simple (100 conc)" "$RPS_W1" "$RPS_O1" "$RPS_A1" + fi + if [ -n "$RPS_W2" ] && [ -n "$RPS_O2" ] && [ -n "$RPS_A2" ]; then + printf "%-25s %9s/s %9s/s %9s/s\n" "High conc (500 conc)" "$RPS_W2" "$RPS_O2" "$RPS_A2" + fi + if [ -n "$RPS_W3" ] && [ -n "$RPS_O3" ] && [ -n "$RPS_A3" ]; then + printf "%-25s %9s/s %9s/s %9s/s\n" "Sustained (200 conc)" "$RPS_W3" "$RPS_O3" "$RPS_A3" + fi + if [ -n "$RPS_W4" ] && [ -n "$RPS_O4" ] && [ -n "$RPS_A4" ]; then + printf "%-25s %9s/s %9s/s %9s/s\n" "Large response (50 conc)" "$RPS_W4" "$RPS_O4" "$RPS_A4" + fi +else + printf "%-25s %15s %15s %10s\n" "Test" "WSGI" "ASGI" "Diff" + printf "%-25s %15s %15s %10s\n" "-------------------------" "---------------" "---------------" "----------" + + # Calculate differences + if [ -n "$RPS_W1" ] && [ -n "$RPS_A1" ]; then + DIFF1=$(echo "$RPS_A1 $RPS_W1" | awk '{if($2>0) printf "%.1f%%", (($1-$2)/$2)*100; else print "N/A"}') + printf "%-25s %12s/s %12s/s %10s\n" "Simple (100 conc)" "$RPS_W1" "$RPS_A1" "$DIFF1" + fi + + if [ -n "$RPS_W2" ] && [ -n "$RPS_A2" ]; then + DIFF2=$(echo "$RPS_A2 $RPS_W2" | awk '{if($2>0) printf "%.1f%%", (($1-$2)/$2)*100; else print "N/A"}') + printf "%-25s %12s/s %12s/s %10s\n" "High conc (500 conc)" "$RPS_W2" "$RPS_A2" "$DIFF2" + fi + + if [ -n "$RPS_W3" ] && [ -n "$RPS_A3" ]; then + DIFF3=$(echo "$RPS_A3 $RPS_W3" | awk '{if($2>0) printf "%.1f%%", (($1-$2)/$2)*100; else print "N/A"}') + printf "%-25s %12s/s %12s/s %10s\n" "Sustained (200 conc)" "$RPS_W3" "$RPS_A3" "$DIFF3" + fi + + if [ -n "$RPS_W4" ] && [ -n "$RPS_A4" ]; then + DIFF4=$(echo "$RPS_A4 $RPS_W4" | awk '{if($2>0) printf "%.1f%%", (($1-$2)/$2)*100; else print "N/A"}') + printf "%-25s %12s/s %12s/s %10s\n" "Large response (50 conc)" "$RPS_W4" "$RPS_A4" "$DIFF4" + fi +fi + +echo "" +echo "Done!" diff --git a/benchmarks/noop_asgi.py b/benchmarks/noop_asgi.py new file mode 100644 index 0000000..896fbec --- /dev/null +++ b/benchmarks/noop_asgi.py @@ -0,0 +1,12 @@ +# Minimal ASGI app - no processing +async def application(scope, receive, send): + if scope['type'] == 'http': + await send({ + 'type': 'http.response.start', + 'status': 200, + 'headers': [(b'content-type', b'text/plain'), (b'content-length', b'13')], + }) + await send({ + 'type': 'http.response.body', + 'body': b'Hello, World!', + }) diff --git a/benchmarks/simple_asgi_app.py b/benchmarks/simple_asgi_app.py index 7e6e8db..902d005 100644 --- a/benchmarks/simple_asgi_app.py +++ b/benchmarks/simple_asgi_app.py @@ -1,5 +1,4 @@ # Simple ASGI app for benchmarking -# Async version of the WSGI benchmark app async def application(scope, receive, send): diff --git a/config/sys.config b/config/sys.config index 8873334..01bfa52 100644 --- a/config/sys.config +++ b/config/sys.config @@ -7,9 +7,13 @@ {keepalive, 2}, {max_requests, 1000}, {preload_app, false}, - {pythonpath, [".", "examples"]} + {pythonpath, [".", "examples"]}, + %% WSGI body streaming settings + {wsgi_body_chunk_size, 65536}, %% 64KB chunks + {wsgi_streaming_threshold, 65536} %% Stream if > 64KB ]}, {erlang_python, [ - {num_workers, 4} + {num_workers, 4}, + {max_concurrent, 10000} ]} ]. diff --git a/docs/guides/multi-app.md b/docs/guides/multi-app.md index 24acc65..f2f8046 100644 --- a/docs/guides/multi-app.md +++ b/docs/guides/multi-app.md @@ -24,7 +24,7 @@ Each mount is a tuple of `{Prefix, AppSpec, Options}`: - **Prefix** - URL path prefix (must start with `/`) - **AppSpec** - Python module:callable (e.g., `"myapp:application"`) -- **Options** - Per-mount options (worker_class, workers, timeout) +- **Options** - Per-mount options (worker_class, timeout) ## Routing Behavior @@ -93,24 +93,21 @@ Each mount can have its own configuration: ```erlang hornbeam:start(#{ mounts => [ - %% High-performance async API with more workers + %% High-performance async API {"/api", "api:app", #{ worker_class => asgi, - workers => 8, timeout => 60000 }}, - %% Admin panel - fewer workers needed + %% Admin panel {"/admin", "admin:app", #{ worker_class => wsgi, - workers => 2, timeout => 30000 }}, %% Static frontend {"/", "frontend:app", #{ - worker_class => wsgi, - workers => 4 + worker_class => wsgi }} ], bind => "0.0.0.0:8000" @@ -122,9 +119,10 @@ hornbeam:start(#{ | Option | Type | Default | Description | |--------|------|---------|-------------| | `worker_class` | atom | `wsgi` | Protocol: `wsgi` or `asgi` | -| `workers` | integer | `4` | Number of Python workers | | `timeout` | integer | `30000` | Request timeout in ms | +> **Note:** All mounts share the global `py_context_router` pool. Configure pool size at the application level rather than per-mount. + ## Global Options Global options apply to all mounts: @@ -184,20 +182,17 @@ hornbeam:start(#{ mounts => [ %% FastAPI for real-time API {"/api/v2", "api_v2:app", #{ - worker_class => asgi, - workers => 8 + worker_class => asgi }}, %% Legacy Flask API {"/api/v1", "api_v1:app", #{ - worker_class => wsgi, - workers => 4 + worker_class => wsgi }}, %% Django admin {"/admin", "myproject.wsgi:application", #{ - worker_class => wsgi, - workers => 2 + worker_class => wsgi }}, %% React frontend (served by Flask) diff --git a/docs/reference/configuration.md b/docs/reference/configuration.md index 394ef73..a78c125 100644 --- a/docs/reference/configuration.md +++ b/docs/reference/configuration.md @@ -50,9 +50,10 @@ Each mount is a tuple: `{Prefix, AppSpec, Options}` | Option | Type | Default | Description | |--------|------|---------|-------------| | `worker_class` | atom | `wsgi` | Protocol: `wsgi` or `asgi` | -| `workers` | integer | `4` | Number of Python workers for this mount | | `timeout` | integer | `30000` | Request timeout in ms | +> **Note:** All mounts share the global `py_context_router` pool. Configure pool size at the application level. + ### Multi-App Example ```erlang @@ -60,12 +61,10 @@ hornbeam:start(#{ mounts => [ {"/api/v2", "api_v2:app", #{ worker_class => asgi, - workers => 8, timeout => 60000 }}, {"/api/v1", "api_v1:app", #{ - worker_class => wsgi, - workers => 4 + worker_class => wsgi }}, {"/", "frontend:app", #{worker_class => wsgi}} ], diff --git a/priv/hornbeam_asgi_runner.py b/priv/hornbeam_asgi_runner.py index 2a54f35..3b795ca 100644 --- a/priv/hornbeam_asgi_runner.py +++ b/priv/hornbeam_asgi_runner.py @@ -337,6 +337,12 @@ def reload_app(module_name: str, callable_name: str): return app +# Response object pool for reuse +_RESPONSE_POOL = [] +_RESPONSE_POOL_SIZE = 100 +_RESPONSE_POOL_LOCK = threading.Lock() + + class ASGIResponse: """Collects ASGI response messages. @@ -358,6 +364,16 @@ def __init__(self): self.trailers = [] self.early_hints = [] + def reset(self): + """Reset response for reuse from pool.""" + self.status = None + self.headers = [] + self.body_parts = [] + self.more_body = False + self.informational = [] + self.trailers = [] + self.early_hints = [] + async def send(self, message: dict) -> None: """ASGI send callable.""" msg_type = message['type'] if 'type' in message else '' @@ -410,6 +426,23 @@ def to_dict(self) -> dict: return result +def _get_response() -> ASGIResponse: + """Get an ASGIResponse from pool or create new.""" + with _RESPONSE_POOL_LOCK: + if _RESPONSE_POOL: + resp = _RESPONSE_POOL.pop() + resp.reset() + return resp + return ASGIResponse() + + +def _return_response(resp: ASGIResponse) -> None: + """Return an ASGIResponse to the pool.""" + with _RESPONSE_POOL_LOCK: + if len(_RESPONSE_POOL) < _RESPONSE_POOL_SIZE: + _RESPONSE_POOL.append(resp) + + async def _run_asgi_async(module_name: str, callable_name: str, scope: dict, body: bytes) -> dict: """Internal async runner for ASGI apps.""" @@ -561,220 +594,3 @@ def _run_asgi_sync(module_name: str, callable_name: str, result.get('headers', []), result.get('body', b'') ) - - -# Streaming support for real-time responses - -# Thread-safe streaming session storage -_streaming_sessions: Dict[str, 'StreamingASGIRunner'] = {} -_streaming_sessions_lock = threading.Lock() - - -class StreamingASGIRunner: - """Runner for streaming ASGI responses. - - This class supports: - - Server-Sent Events (SSE) - - Chunked transfer encoding - - Real-time response streaming - """ - - def __init__(self, module_name: str, callable_name: str, scope: dict): - self.module_name = module_name - self.callable_name = callable_name - self.scope = scope - self.app = None - self.status = None - self.headers = [] - self.body_queue: asyncio.Queue = None - self.finished = False - self.loop = None - self._response_started_event: asyncio.Event = None - self._error: Optional[Exception] = None - - def start(self, body: bytes, timeout_ms: int = 5000) -> dict: - """Start the streaming response. - - Args: - body: Request body bytes - timeout_ms: Max time to wait for response headers (default 5s) - - Returns the initial response headers. - """ - self.app = load_app(self.module_name, self.callable_name) - - # Inject lifespan state if available - if _get_lifespan_state is not None: - self.scope['state'] = _get_lifespan_state() - - self.loop = asyncio.new_event_loop() - asyncio.set_event_loop(self.loop) - self.body_queue = asyncio.Queue() - self._response_started_event = asyncio.Event() - - # Create receive/send callables - body_sent = False - - async def receive(): - nonlocal body_sent - if not body_sent: - body_sent = True - return { - 'type': 'http.request', - 'body': body, - 'more_body': False - } - return _DISCONNECT_MSG - - async def send(message): - msg_type = message.get('type', '') - if msg_type == 'http.response.start': - self.status = message.get('status', 200) - self.headers = message.get('headers', []) - self._response_started_event.set() - elif msg_type == 'http.response.body': - body_part = message.get('body', b'') - more_body = message.get('more_body', False) - await self.body_queue.put((body_part, more_body)) - if not more_body: - self.finished = True - - # Start app in background - async def run_app(): - try: - await self.app(self.scope, receive, send) - except Exception as e: - self._error = e - self._response_started_event.set() # Unblock waiter on error - await self.body_queue.put((b'', False)) - self.finished = True - - self.loop.create_task(run_app()) - - # Wait for response to start using event (not polling) - async def wait_for_start(): - await asyncio.wait_for( - self._response_started_event.wait(), - timeout=timeout_ms / 1000.0 - ) - - try: - self.loop.run_until_complete(wait_for_start()) - except asyncio.TimeoutError: - # Timeout waiting for response headers - self.finished = True - return {'status': 504, 'headers': [], 'error': 'timeout'} - - if self._error is not None: - return { - 'status': 500, - 'headers': [], - 'error': str(self._error) - } - - return { - 'status': self.status or 500, - 'headers': self.headers - } - - def next_chunk(self, timeout_ms: int = 30000) -> tuple: - """Get the next body chunk. - - Returns (chunk_bytes, more_body_bool) - """ - if self.finished or self.loop is None: - return (b'', False) - - try: - timeout_sec = timeout_ms / 1000.0 - future = asyncio.wait_for( - self.body_queue.get(), - timeout=timeout_sec - ) - chunk, more_body = self.loop.run_until_complete(future) - return (chunk, more_body) - except asyncio.TimeoutError: - return (b'', True) # Timeout, but may have more - except Exception: - return (b'', False) - - def close(self): - """Clean up resources.""" - if self.loop: - try: - self.loop.close() - except Exception: - pass - self.loop = None - - -def start_streaming(session_id: str, module_name: str, callable_name: str, - scope: dict, body: bytes, timeout_ms: int = 5000) -> dict: - """Start a streaming ASGI response session. - - Args: - session_id: Unique identifier for this streaming session - module_name: Python module containing the ASGI app - callable_name: Name of the ASGI callable - scope: ASGI scope dict - body: Request body bytes - timeout_ms: Max time to wait for response headers - - Returns initial response headers. - """ - runner = StreamingASGIRunner(module_name, callable_name, scope) - - with _streaming_sessions_lock: - _streaming_sessions[session_id] = runner - - if body.__class__ is str: - body = body.encode('utf-8') - elif body.__class__ is not bytes: - body = b'' - - return runner.start(body, timeout_ms) - - -def get_streaming_chunk(session_id: str, timeout_ms: int = 30000) -> dict: - """Get the next chunk from a streaming session. - - Returns {'chunk': bytes, 'more_body': bool} - """ - with _streaming_sessions_lock: - runner = _streaming_sessions.get(session_id) - - if runner is None: - return {'chunk': b'', 'more_body': False, 'error': 'session_not_found'} - - chunk, more_body = runner.next_chunk(timeout_ms) - return {'chunk': chunk, 'more_body': more_body} - - -def end_streaming(session_id: str) -> None: - """End a streaming session and clean up resources.""" - with _streaming_sessions_lock: - runner = _streaming_sessions.pop(session_id, None) - - if runner: - runner.close() - - -def cleanup_streaming_sessions() -> int: - """Clean up finished streaming sessions. - - Call this periodically to prevent memory leaks from abandoned sessions. - - Returns the number of sessions cleaned up. - """ - cleaned = 0 - with _streaming_sessions_lock: - finished_ids = [ - sid for sid, runner in _streaming_sessions.items() - if runner.finished or runner.loop is None - ] - for sid in finished_ids: - runner = _streaming_sessions.pop(sid, None) - if runner: - runner.close() - cleaned += 1 - return cleaned diff --git a/priv/hornbeam_asgi_worker.py b/priv/hornbeam_asgi_worker.py new file mode 100644 index 0000000..b02904a --- /dev/null +++ b/priv/hornbeam_asgi_worker.py @@ -0,0 +1,450 @@ +# Copyright 2026 Benoit Chesneau +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""ASGI handler using Cowboy loop handler with direct message passing. + +Design inspired by gunicorn's ASGI worker: +- BodyReceiver: Handles request body with Future-based waiting +- ASGIProtocol: Manages request lifecycle and response batching +- Response sent via erlang.send() directly to Cowboy handler +""" + +import asyncio +from typing import Callable, Optional + +try: + import erlang + from erlang import ByteChannel, ByteChannelClosed + HAS_ERLANG = True + _erlang_send = erlang.send +except ImportError: + HAS_ERLANG = False + erlang = None + _erlang_send = None + ByteChannel = None + ByteChannelClosed = Exception + +# Pre-allocated message constants (avoid dict allocation per request) +_EMPTY_BODY_MSG = {'type': 'http.request', 'body': b'', 'more_body': False} +_DISCONNECT_MSG = {'type': 'http.disconnect'} + + +class BodyReceiver: + """Body receiver with Future-based waiting (gunicorn pattern). + + Handles three body modes: + - empty: No body expected + - inline: Small body passed directly from Erlang + - channel: Large/streaming body via ByteChannel + + Uses asyncio.Future for efficient async waiting without polling. + """ + + __slots__ = ('_mode', '_data', '_channel', '_complete', '_disconnected', + '_waiter', '_chunks') + + def __init__(self, body_ref): + self._complete = False + self._disconnected = False + self._waiter = None + self._chunks = [] + + # Detect body mode from ref + if body_ref == b'empty' or body_ref == 'empty': + self._mode = 'empty' + self._data = None + self._channel = None + self._complete = True + elif isinstance(body_ref, tuple): + tag = body_ref[0] + if tag == b'body' or tag == 'body': + self._mode = 'inline' + data = body_ref[1] + # Ensure bytes (Erlang binary may decode as str) + if isinstance(data, str): + data = data.encode('latin-1') + self._data = data + self._channel = None + elif tag == b'channel' or tag == 'channel': + self._mode = 'channel' + self._data = None + self._channel = ByteChannel(body_ref[1]) + else: + # Unknown tuple, treat as channel ref + self._mode = 'channel' + self._data = None + self._channel = ByteChannel(body_ref) + else: + # Legacy: raw channel ref + self._mode = 'channel' + self._data = None + self._channel = ByteChannel(body_ref) + + def signal_disconnect(self): + """Signal client disconnection.""" + self._disconnected = True + self._wake_waiter() + + def _wake_waiter(self): + """Wake pending receive() call.""" + if self._waiter is not None and not self._waiter.done(): + self._waiter.set_result(None) + + async def receive(self) -> dict: + """ASGI receive callable with fast paths.""" + # Already disconnected + if self._disconnected: + return _DISCONNECT_MSG + + # Fast path: empty body + if self._mode == 'empty': + return _EMPTY_BODY_MSG + + # Fast path: inline body (complete body passed directly) + if self._mode == 'inline': + self._mode = 'done' + return {'type': 'http.request', 'body': self._data, 'more_body': False} + + # Body already consumed + if self._mode == 'done' or self._complete: + return _EMPTY_BODY_MSG + + # Fast path: chunks already buffered + if self._chunks: + return self._pop_chunk() + + # Channel mode: read from ByteChannel + return await self._receive_from_channel() + + def _pop_chunk(self) -> dict: + """Pop buffered chunk and return message.""" + chunk = self._chunks.pop(0) + more = bool(self._chunks) or not self._complete + if not more: + self._mode = 'done' + return {'type': 'http.request', 'body': chunk, 'more_body': more} + + async def _receive_from_channel(self) -> dict: + """Read body chunk from ByteChannel.""" + try: + chunk = await self._channel.async_receive_bytes() + if chunk: + return {'type': 'http.request', 'body': chunk, 'more_body': True} + # Empty chunk but channel still open - wait for more + return {'type': 'http.request', 'body': b'', 'more_body': True} + except ByteChannelClosed: + self._complete = True + self._mode = 'done' + return _EMPTY_BODY_MSG + + +class ASGIProtocol: + """ASGI protocol handler with response batching (gunicorn-inspired). + + Optimizations: + - Uses __slots__ to reduce memory and attribute access overhead + - Delegates body handling to BodyReceiver class + - Batches headers + body for simple responses (single message) + - Structured response state tracking + """ + + __slots__ = ('_caller_pid', '_app', '_scope', '_body_receiver', '_send_fn', + '_status', '_headers', '_response_started', '_response_finished') + + def __init__(self, caller_pid, app: Callable, scope: dict, body_receiver: BodyReceiver): + self._caller_pid = caller_pid + self._app = app + self._scope = scope + self._body_receiver = body_receiver + self._send_fn = _erlang_send + + # Response state (buffer headers for batching) + self._status = None + self._headers = None + self._response_started = False + self._response_finished = False + + async def receive(self) -> dict: + """ASGI receive - delegates to BodyReceiver.""" + return await self._body_receiver.receive() + + async def send(self, message: dict) -> None: + """ASGI send callable with simplified protocol. + + Uses 3 message types: + - start_response: headers + first chunk + - chunk: subsequent body chunks + - fin: end of response + """ + if self._response_finished: + raise RuntimeError("Response already completed") + + msg_type = message.get('type', '') + + if msg_type == 'http.response.start': + if self._response_started: + raise RuntimeError("http.response.start already sent") + + # Buffer headers for batching with first body + self._status = message.get('status', 200) + self._headers = message.get('headers', []) + self._response_started = True + + elif msg_type == 'http.response.body': + if not self._response_started: + raise RuntimeError("http.response.start must be sent first") + + body = message.get('body', b'') + if isinstance(body, str): + body = body.encode('utf-8') + + more_body = message.get('more_body', False) + + if self._headers is not None: + # First body - send start_response with headers + first chunk + self._send_fn(self._caller_pid, + (b'start_response', self._status, self._headers, body)) + self._headers = None + else: + # Subsequent chunk + if body: + self._send_fn(self._caller_pid, (b'chunk', body)) + + if not more_body: + # Send fin to terminate + self._send_fn(self._caller_pid, b'fin') + self._response_finished = True + + elif msg_type == 'http.response.informational': + status = message.get('status', 100) + headers = message.get('headers', []) + if status == 103: + self._send_fn(self._caller_pid, (b'early_hints', headers)) + + elif msg_type == 'http.disconnect': + self._response_finished = True + + async def run(self): + """Run the ASGI application.""" + try: + await self._app(self._scope, self.receive, self.send) + + # Ensure response is completed + if not self._response_finished: + if not self._response_started: + # No response started - send error response + self._send_fn(self._caller_pid, + (b'start_response', 500, [], b'')) + self._send_fn(self._caller_pid, b'fin') + elif self._headers is not None: + # Headers buffered but no body sent - send empty response + self._send_fn(self._caller_pid, + (b'start_response', self._status, self._headers, b'')) + self._send_fn(self._caller_pid, b'fin') + else: + # Streaming was started but not finished - send fin + self._send_fn(self._caller_pid, b'fin') + + except asyncio.CancelledError: + # Client disconnected - signal to body receiver + self._body_receiver.signal_disconnect() + raise + except Exception as e: + self._send_fn(self._caller_pid, (b'error', str(e).encode('utf-8'))) + + +# ============================================================================= +# Entry point +# ============================================================================= + +async def handle_asgi(caller_pid, app_module: str, app_callable: str, + scope: dict, req_body_ref): + """Handle ASGI request. + + Entry point called from hornbeam_asgi.erl. + Response is sent directly via erlang.send(). + + req_body_ref can be: + - 'empty' or b'empty': no request body + - (b'body', data): small body passed inline (< 64KB) + - (b'channel', channel_ref): large/streaming body via channel + """ + if not HAS_ERLANG: + return + + # Get app (cached) + app = _get_app(app_module, app_callable) + + # Use cached state proxy (shared per mount_id) + mount_id = scope.get('mount_id') + scope['state'] = _get_state_proxy(mount_id) + + # Create body receiver (handles body mode detection) + body_receiver = BodyReceiver(req_body_ref) + + # Create and run protocol + protocol = ASGIProtocol(caller_pid, app, scope, body_receiver) + await protocol.run() + + +# ============================================================================= +# Helpers +# ============================================================================= + +import sys + +# Cache apps by (module, callable) key - modules already imported by Erlang +_app_cache: dict = {} + +# Cache state proxies by mount_id - shared across requests for same mount +_state_cache: dict = {} + + +def _get_app(module_name: str, callable_name: str) -> Callable: + """Get ASGI application from cache or sys.modules. + + Module is already imported by Erlang via ensure_all_imported, + so we just look it up in sys.modules - no importlib needed. + """ + key = (module_name, callable_name) + app = _app_cache.get(key) + if app is not None: + return app + + # Module already imported by Erlang - just get from sys.modules + module = sys.modules.get(module_name) + if module is None: + # Fallback: import if somehow not in sys.modules + import importlib + module = importlib.import_module(module_name) + + app = getattr(module, callable_name) + _app_cache[key] = app + return app + + +def preload_app(app_module: str, app_callable: str) -> bytes: + """Preload ASGI application at startup.""" + _get_app(app_module, app_callable) + return b'ok' + + +def _get_state_proxy(mount_id): + """Get cached state proxy for mount_id, creating if needed.""" + if mount_id is None: + return _LazyStateProxy(None) + proxy = _state_cache.get(mount_id) + if proxy is None: + proxy = _LazyStateProxy(mount_id) + _state_cache[mount_id] = proxy + return proxy + + +class _LazyStateProxy(dict): + """Dict that lazily fetches from ETS and syncs mutations via callback. + + Optimizations: + - No erlang.whereis() on construction (callbacks are pre-registered) + - Lazy loading: state only fetched when accessed + - Direct ETS access via callbacks (no message passing for reads) + """ + + __slots__ = ('_mount_id', '_loaded') + + def __init__(self, mount_id=None): + super().__init__() + self._mount_id = mount_id + self._loaded = False + + def _ensure_loaded(self): + """Load full state on first access if needed.""" + if self._loaded: + return + if not HAS_ERLANG: + self._loaded = True + return + try: + if self._mount_id is not None: + state = erlang.call('lifespan_state_get', self._mount_id, None) + else: + state = erlang.call('lifespan_state_get') + if state and isinstance(state, dict): + super().update(state) + except Exception: + pass + self._loaded = True + + def __getitem__(self, key): + # Fast path: check local cache first + try: + return super().__getitem__(key) + except KeyError: + pass + # Fetch specific key from ETS + if HAS_ERLANG: + try: + if self._mount_id is not None: + value = erlang.call('lifespan_state_get', self._mount_id, key) + else: + value = erlang.call('lifespan_state_get', key) + if value is not None: + super().__setitem__(key, value) + return value + except Exception: + pass + raise KeyError(key) + + def __setitem__(self, key, value): + super().__setitem__(key, value) + # Sync to ETS via callback (no whereis needed) + if HAS_ERLANG: + try: + if self._mount_id is not None: + erlang.call('lifespan_state_set', self._mount_id, key, value) + else: + erlang.call('lifespan_state_set', key, value) + except Exception: + pass + + def __contains__(self, key): + if super().__contains__(key): + return True + self._ensure_loaded() + return super().__contains__(key) + + def keys(self): + self._ensure_loaded() + return super().keys() + + def values(self): + self._ensure_loaded() + return super().values() + + def items(self): + self._ensure_loaded() + return super().items() + + def get(self, key, default=None): + try: + return self[key] + except KeyError: + return default + + def __iter__(self): + self._ensure_loaded() + return super().__iter__() + + def __len__(self): + self._ensure_loaded() + return super().__len__() diff --git a/priv/hornbeam_lifespan_runner.py b/priv/hornbeam_lifespan_runner.py index 8e29fb4..7761dc0 100644 --- a/priv/hornbeam_lifespan_runner.py +++ b/priv/hornbeam_lifespan_runner.py @@ -48,7 +48,13 @@ def _install_erlang_loop() -> bool: _install_erlang_loop() -# Global lifespan state shared across requests +# Per-mount lifespan state for multi-app mode +_lifespan_states: Dict[str, Dict[str, Any]] = {} + +# Per-mount lifespan tracking for multi-app mode +_mount_lifespans: Dict[str, dict] = {} + +# Global lifespan state for single-app mode (backward compat) _lifespan_state: Dict[str, Any] = {} _lifespan_app = None _lifespan_task = None @@ -239,6 +245,214 @@ def shutdown(app_module: str, app_callable: str) -> dict: _cleanup() +def startup_mount(mount_id: str, app_module: str, app_callable: str, + timeout_ms: int = 30000) -> dict: + """Run lifespan startup protocol for a specific mount. + + This is used in multi-app mode to run lifespan per mounted app. + Each mount gets its own isolated state dict. + + Args: + mount_id: Unique identifier for this mount + app_module: Python module containing the ASGI app + app_callable: Name of the ASGI callable + timeout_ms: Timeout for startup in milliseconds (default: 30000) + + Returns: + Response dict with type and optional state + """ + global _lifespan_states, _mount_lifespans + + # erlang_python converts binaries to str in C - no decode needed + + # Load the app + try: + app = load_app(app_module, app_callable) + except Exception as e: + return {'type': 'lifespan.startup.failed', 'message': str(e)} + + # Create per-mount event loop and queues + loop = asyncio.new_event_loop() + receive_queue = asyncio.Queue() + send_queue = asyncio.Queue() + + # Initialize state dict for this mount + mount_state: Dict[str, Any] = {} + _lifespan_states[mount_id] = mount_state + + # Build lifespan scope with per-mount state + scope = { + 'type': 'lifespan', + 'asgi': { + 'version': '3.0', + 'spec_version': '2.4' + }, + 'state': mount_state + } + + # Create lifespan runner for this mount + async def run_mount_lifespan(): + async def receive(): + return await receive_queue.get() + async def send(message): + await send_queue.put(message) + try: + await app(scope, receive, send) + except Exception as e: + await send_queue.put({ + 'type': 'lifespan.startup.failed', + 'message': str(e) + }) + + # Start the lifespan task + task = loop.create_task(run_mount_lifespan()) + + # Store mount lifespan tracking + _mount_lifespans[mount_id] = { + 'app': app, + 'task': task, + 'loop': loop, + 'receive_queue': receive_queue, + 'send_queue': send_queue + } + + # Send startup event + loop.run_until_complete(receive_queue.put({'type': 'lifespan.startup'})) + + # Wait for response + async def wait_for_response(): + await asyncio.sleep(0.01) + if task.done(): + return None + total_timeout = timeout_ms / 1000.0 + check_interval = 0.5 + elapsed = 0.0 + while elapsed < total_timeout: + try: + response = await asyncio.wait_for( + send_queue.get(), + timeout=check_interval + ) + return response + except asyncio.TimeoutError: + if task.done(): + return None + elapsed += check_interval + raise asyncio.TimeoutError() + + try: + response = loop.run_until_complete(wait_for_response()) + if response is None: + _cleanup_mount(mount_id) + return {'type': 'lifespan.not_supported'} + except asyncio.TimeoutError: + return {'type': 'lifespan.startup.failed', + 'message': 'Startup timeout'} + except Exception as e: + return {'type': 'lifespan.startup.failed', 'message': str(e)} + + msg_type = response.get('type', '') + + if msg_type == 'lifespan.startup.complete': + # Store state for access by requests + _lifespan_states[mount_id] = scope.get('state', {}) + return { + 'type': 'lifespan.startup.complete', + 'state': _lifespan_states[mount_id] + } + elif msg_type == 'lifespan.startup.failed': + _cleanup_mount(mount_id) + return response + else: + _cleanup_mount(mount_id) + return {'type': 'lifespan.not_supported'} + + +def shutdown_mount(mount_id: str, app_module: str, app_callable: str) -> dict: + """Run lifespan shutdown protocol for a specific mount. + + Args: + mount_id: Unique identifier for this mount + app_module: Python module (for reference) + app_callable: ASGI callable name (for reference) + + Returns: + Response dict with shutdown status + """ + global _mount_lifespans, _lifespan_states + + # erlang_python converts binaries to str in C - no decode needed + + if mount_id not in _mount_lifespans: + return {'type': 'lifespan.shutdown.complete'} + + mount = _mount_lifespans[mount_id] + task = mount['task'] + loop = mount['loop'] + receive_queue = mount['receive_queue'] + send_queue = mount['send_queue'] + + try: + # Send shutdown event + loop.run_until_complete( + receive_queue.put({'type': 'lifespan.shutdown'}) + ) + + # Wait for response + try: + response = loop.run_until_complete( + asyncio.wait_for(send_queue.get(), timeout=10.0) + ) + except asyncio.TimeoutError: + response = {'type': 'lifespan.shutdown.complete'} + + # Wait for task to finish + try: + loop.run_until_complete( + asyncio.wait_for(task, timeout=5.0) + ) + except asyncio.TimeoutError: + task.cancel() + try: + loop.run_until_complete(task) + except asyncio.CancelledError: + pass + + return response + + except Exception as e: + return {'type': 'lifespan.shutdown.complete', + 'error': str(e)} + finally: + _cleanup_mount(mount_id) + + +def _cleanup_mount(mount_id: str): + """Clean up lifespan state for a specific mount.""" + global _mount_lifespans, _lifespan_states + + if mount_id in _mount_lifespans: + mount = _mount_lifespans[mount_id] + task = mount.get('task') + loop = mount.get('loop') + + if task and not task.done(): + task.cancel() + if loop: + try: + loop.run_until_complete(task) + except asyncio.CancelledError: + pass + + if loop: + loop.close() + + del _mount_lifespans[mount_id] + + if mount_id in _lifespan_states: + del _lifespan_states[mount_id] + + def _cleanup(): """Clean up lifespan state.""" global _lifespan_app, _lifespan_task, _receive_queue, _send_queue, _loop @@ -261,13 +475,24 @@ def _cleanup(): _loop = None -def get_state() -> dict: +def get_state(mount_id: Optional[str] = None) -> dict: """Get the lifespan state dict. This returns the actual dict (not a copy) so that modifications by request handlers persist across requests - as per ASGI spec. + + Args: + mount_id: Optional mount identifier for multi-app mode. + If None, returns single-app mode state. + + Returns: + The lifespan state dict for the specified mount (or global state). """ - return _lifespan_state + if mount_id is None: + return _lifespan_state + + # erlang_python converts binaries to str in C - no decode needed + return _lifespan_states.get(mount_id, {}) def set_state(key: str, value: Any) -> None: diff --git a/priv/hornbeam_wsgi_runner.py b/priv/hornbeam_wsgi_runner.py index 957f367..c8b8389 100644 --- a/priv/hornbeam_wsgi_runner.py +++ b/priv/hornbeam_wsgi_runner.py @@ -254,12 +254,50 @@ def reload_app(module_name, callable_name): # Cached WSGI version tuple _WSGI_VERSION = (1, 0) +# BytesIO pool for wsgi.input - reduces allocation overhead +_BYTESIO_POOL = [] +_BYTESIO_POOL_SIZE = 100 +_BYTESIO_POOL_LOCK = threading.Lock() + + +def _get_bytesio(data: bytes) -> io.BytesIO: + """Get a BytesIO from pool or create new one.""" + bio = None + with _BYTESIO_POOL_LOCK: + if _BYTESIO_POOL: + bio = _BYTESIO_POOL.pop() + if bio is not None: + bio.seek(0) + bio.truncate() + bio.write(data) + bio.seek(0) + return bio + return io.BytesIO(data) + + +def _return_bytesio(bio: io.BytesIO) -> None: + """Return a BytesIO to the pool for reuse.""" + with _BYTESIO_POOL_LOCK: + if len(_BYTESIO_POOL) < _BYTESIO_POOL_SIZE: + _BYTESIO_POOL.append(bio) + + +# Pre-computed environ template - shared base for all requests +_ENVIRON_TEMPLATE = { + 'wsgi.version': _WSGI_VERSION, + 'wsgi.multithread': True, + 'wsgi.multiprocess': True, + 'wsgi.run_once': False, + 'wsgi.file_wrapper': FileWrapper, + 'wsgi.input_terminated': True, +} + def create_environ(raw_environ): """Create a complete WSGI environ dict from raw environ. Ensures all required WSGI variables are present and properly typed. - Optimized for minimal overhead on hot path. + Optimized for minimal overhead on hot path using template and BytesIO pool. Args: raw_environ: Raw environ dict from Erlang @@ -273,47 +311,39 @@ def create_environ(raw_environ): def early_hints_callback(headers): early_hints_list.append(headers) - # Handle wsgi.input - most common case is bytes + # Handle wsgi.input - use pooled BytesIO for common bytes case wsgi_input = raw_environ.get('wsgi.input', b'') wsgi_input_type = type(wsgi_input) if wsgi_input_type is bytes: - wsgi_input_stream = io.BytesIO(wsgi_input) + wsgi_input_stream = _get_bytesio(wsgi_input) elif wsgi_input_type is str: - wsgi_input_stream = io.BytesIO(wsgi_input.encode('utf-8')) + wsgi_input_stream = _get_bytesio(wsgi_input.encode('utf-8')) elif hasattr(wsgi_input, 'read'): wsgi_input_stream = wsgi_input else: - wsgi_input_stream = io.BytesIO(b'') - - # Build environ with proper types - # Use direct access for speed on known keys - environ = { - # Required CGI variables with defaults - 'REQUEST_METHOD': raw_environ.get('REQUEST_METHOD', 'GET'), - 'SCRIPT_NAME': raw_environ.get('SCRIPT_NAME', ''), - 'PATH_INFO': raw_environ.get('PATH_INFO', '/'), - 'QUERY_STRING': raw_environ.get('QUERY_STRING', ''), - 'SERVER_NAME': raw_environ.get('SERVER_NAME', 'localhost'), - 'SERVER_PORT': str(raw_environ.get('SERVER_PORT', '80')), - 'SERVER_PROTOCOL': raw_environ.get('SERVER_PROTOCOL', 'HTTP/1.1'), - - # Required WSGI variables (use cached/shared where possible) - 'wsgi.version': _WSGI_VERSION, - 'wsgi.url_scheme': raw_environ.get('wsgi.url_scheme', 'http'), - 'wsgi.input': wsgi_input_stream, - 'wsgi.errors': _SHARED_ERRORS, - 'wsgi.multithread': True, - 'wsgi.multiprocess': True, - 'wsgi.run_once': False, - - # Recommended extensions - 'wsgi.file_wrapper': FileWrapper, - 'wsgi.input_terminated': True, - 'wsgi.early_hints': early_hints_callback, - - # Store early hints list reference - '_hornbeam.early_hints': early_hints_list, - } + wsgi_input_stream = _get_bytesio(b'') + + # Start from template copy (faster than inline dict creation) + environ = _ENVIRON_TEMPLATE.copy() + + # Update with request-specific required CGI variables + environ['REQUEST_METHOD'] = raw_environ.get('REQUEST_METHOD', 'GET') + environ['SCRIPT_NAME'] = raw_environ.get('SCRIPT_NAME', '') + environ['PATH_INFO'] = raw_environ.get('PATH_INFO', '/') + environ['QUERY_STRING'] = raw_environ.get('QUERY_STRING', '') + environ['SERVER_NAME'] = raw_environ.get('SERVER_NAME', 'localhost') + environ['SERVER_PORT'] = str(raw_environ.get('SERVER_PORT', '80')) + environ['SERVER_PROTOCOL'] = raw_environ.get('SERVER_PROTOCOL', 'HTTP/1.1') + + # Request-specific WSGI variables + environ['wsgi.url_scheme'] = raw_environ.get('wsgi.url_scheme', 'http') + environ['wsgi.input'] = wsgi_input_stream + environ['wsgi.errors'] = _SHARED_ERRORS + environ['wsgi.early_hints'] = early_hints_callback + + # Store references for cleanup and early hints + environ['_hornbeam.early_hints'] = early_hints_list + environ['_hornbeam.wsgi_input'] = wsgi_input_stream # Copy remaining keys (HTTP_*, CONTENT_TYPE, CONTENT_LENGTH, etc.) # Use pre-computed set for O(1) membership test @@ -383,6 +413,10 @@ def run_wsgi(module_name, callable_name, raw_environ): finally: if hasattr(result, 'close'): result.close() + # Return BytesIO to pool + wsgi_input = environ.get('_hornbeam.wsgi_input') + if wsgi_input is not None: + _return_bytesio(wsgi_input) body = b''.join(body_parts) @@ -427,3 +461,134 @@ def _run_wsgi_sync(module_name: str, callable_name: str, result.get('headers', []), result.get('body', b'') ) + + +def create_environ_from_tuple(req_tuple): + """Create WSGI environ from pre-parsed Erlang tuple - O(1) operations only. + + This is the fast path for WSGI requests. Erlang pre-parses all headers + into WSGI format so Python only does dict updates (no loops). + + Args: + req_tuple: Pre-parsed request tuple from hornbeam_request:build_wsgi_tuple/2 + (method, script_name, path_info, query_string, wsgi_headers, + content_type, content_length, body, server, client, scheme, + protocol, lifespan_state) + + Returns: + Complete WSGI environ dict + """ + (method, script_name, path_info, query_string, wsgi_headers, + content_type, content_length, body, server, client, scheme, + protocol, lifespan_state) = req_tuple + + # Create early hints callback (must be per-request) + early_hints_list = [] + + def early_hints_callback(headers): + early_hints_list.append(headers) + + # Get BytesIO from pool + wsgi_input = _get_bytesio(body if body.__class__ is bytes else b'') + + # Start with template copy (O(1) - shallow copy of small dict) + environ = _ENVIRON_TEMPLATE.copy() + + # Update with request-specific values (no loops!) + environ['REQUEST_METHOD'] = method + environ['SCRIPT_NAME'] = script_name if script_name else '' + environ['PATH_INFO'] = path_info + environ['QUERY_STRING'] = query_string + environ['SERVER_NAME'] = server[0] + environ['SERVER_PORT'] = str(server[1]) + environ['SERVER_PROTOCOL'] = protocol + environ['wsgi.url_scheme'] = scheme + environ['wsgi.input'] = wsgi_input + environ['wsgi.errors'] = _SHARED_ERRORS + environ['wsgi.early_hints'] = early_hints_callback + environ['REMOTE_ADDR'] = client[0] + environ['REMOTE_PORT'] = str(client[1]) + environ['_hornbeam.early_hints'] = early_hints_list + environ['_hornbeam.wsgi_input'] = wsgi_input # For pool return + environ['_hornbeam.lifespan_state'] = lifespan_state + + # Add pre-converted HTTP_* headers (already in correct format from Erlang) + environ.update(wsgi_headers) + + # Add content-type/length if present + if content_type is not None: + environ['CONTENT_TYPE'] = content_type + if content_length is not None: + environ['CONTENT_LENGTH'] = content_length + + return environ + + +def run_wsgi_fast(module_name, callable_name, req_tuple): + """Run WSGI app with pre-parsed request tuple - fastest path. + + Args: + module_name: Python module containing the WSGI app + callable_name: Name of the WSGI callable + req_tuple: Pre-parsed request tuple from Erlang + + Returns: + Dict with status, headers, body, and optional early_hints + """ + # Load the application + app = load_app(module_name, callable_name) + + # Create environ from pre-parsed tuple + environ = create_environ_from_tuple(req_tuple) + + # Create response handler + response = Response(environ) + + # Call the WSGI app + result = app(environ, response.start_response) + + # Collect body + body_parts = [] + + # Add any write() buffer content first + body_parts.extend(response._write_buffer) + + # Check if result is a FileWrapper (for optimized file serving) + is_file_wrapper = isinstance(result, FileWrapper) + + try: + if isinstance(result, (bytes, bytearray)): + body_parts.append(bytes(result)) + else: + for chunk in result: + if isinstance(chunk, (bytes, bytearray)): + body_parts.append(bytes(chunk)) + elif isinstance(chunk, str): + body_parts.append(chunk.encode('utf-8')) + finally: + if hasattr(result, 'close'): + result.close() + # Return BytesIO to pool + wsgi_input = environ.get('_hornbeam.wsgi_input') + if wsgi_input is not None: + _return_bytesio(wsgi_input) + + body = b''.join(body_parts) + + # Build response dict + result_dict = { + 'status': response.status or '500 Internal Server Error', + 'headers': response.headers, + 'body': body, + } + + # Include early hints if any were sent + early_hints = environ.get('_hornbeam.early_hints', []) + if early_hints: + result_dict['early_hints'] = early_hints + + # Include file wrapper info for potential sendfile optimization + if is_file_wrapper: + result_dict['file_wrapper'] = True + + return result_dict diff --git a/priv/hornbeam_wsgi_worker.py b/priv/hornbeam_wsgi_worker.py new file mode 100644 index 0000000..92496c1 --- /dev/null +++ b/priv/hornbeam_wsgi_worker.py @@ -0,0 +1,439 @@ +# Copyright 2026 Benoit Chesneau +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""WSGI worker using py_buffer for zero-copy request body streaming. + +This module provides a WSGI worker with: +- Single entry point for all requests +- py_buffer as wsgi.input (zero-copy shared memory) +- erlang.send for responses + +Architecture: +1. Erlang creates py_buffer and writes body data +2. Python uses buffer directly as wsgi.input (file-like interface) +3. Python sends responses via erlang.send() +""" + +import io +from typing import Callable, Tuple + +try: + import erlang + HAS_ERLANG = True +except ImportError: + HAS_ERLANG = False + erlang = None + + +# ============================================================================ +# Constants and shared instances +# ============================================================================ + +_WSGI_VERSION = (1, 0) +CHUNKS_PER_BATCH = 10 + + +class _WSGIErrorsWrapper: + """Minimal wsgi.errors wrapper that routes to logging.""" + + def write(self, msg): + if msg and msg.strip(): + try: + import logging + logging.getLogger('hornbeam.wsgi').error(msg.rstrip()) + except Exception: + pass + + def writelines(self, lines): + for line in lines: + self.write(line) + + def flush(self): + pass + + +class FileWrapper: + """Efficient file serving wrapper for WSGI.""" + + def __init__(self, filelike, blksize=8192): + self.filelike = filelike + self.blksize = blksize + if hasattr(filelike, 'close'): + self.close = filelike.close + + def __iter__(self): + return self + + def __next__(self): + data = self.filelike.read(self.blksize) + if data: + return data + raise StopIteration + + +_SHARED_ERRORS = _WSGIErrorsWrapper() + +_ENVIRON_TEMPLATE = { + 'wsgi.version': _WSGI_VERSION, + 'wsgi.multithread': True, + 'wsgi.multiprocess': True, + 'wsgi.run_once': False, + 'wsgi.file_wrapper': FileWrapper, + 'wsgi.input_terminated': True, +} + + +# ============================================================================ +# App loading and preloading +# ============================================================================ + +# Cached app reference - set by preload_app() for fast access +_preloaded_app: Callable = None +_preloaded_key: Tuple[str, str] = None + + +def preload_app(app_module: bytes, app_callable: bytes) -> bytes: + """Preload WSGI application at startup for zero-overhead access. + + Called from Erlang during context initialization. + """ + global _preloaded_app, _preloaded_key + + # erlang_python converts binaries to str in C + import importlib + module = importlib.import_module(app_module) + app = getattr(module, app_callable) + + _preloaded_app = app + _preloaded_key = (app_module, app_callable) + + return b'ok' + + +def _get_app(module_name: str, callable_name: str) -> Callable: + """Get WSGI application - uses preloaded app if available.""" + # Fast path: use preloaded app if it matches + if _preloaded_key == (module_name, callable_name): + return _preloaded_app + + # Fallback: import on demand + import importlib + module = importlib.import_module(module_name) + return getattr(module, callable_name) + + +# ============================================================================ +# Helpers +# ============================================================================ + +def _to_bytes(val) -> bytes: + """Convert value to bytes.""" + if isinstance(val, bytes): + return val + if isinstance(val, bytearray): + return bytes(val) + if isinstance(val, str): + return val.encode('utf-8') + return b'' + + +def _parse_status(status_str) -> int: + """Parse WSGI status string to integer.""" + try: + if isinstance(status_str, bytes): + status_str = status_str.decode('utf-8') + parts = status_str.split(' ', 1) + return int(parts[0]) + except (ValueError, IndexError, AttributeError): + return 500 + + +# ============================================================================ +# Response class +# ============================================================================ + +class _Response: + """WSGI response handler.""" + __slots__ = ('status', 'status_code', 'headers', '_write_buffer') + + def __init__(self): + self.status = None + self.status_code = 500 + self.headers = [] + self._write_buffer = [] + + def start_response(self, status, response_headers, exc_info=None): + if exc_info: + try: + if self.status is not None: + raise exc_info[1].with_traceback(exc_info[2]) + finally: + exc_info = None + elif self.status is not None: + raise RuntimeError("start_response already called") + + self.status = status + self.status_code = _parse_status(status) + self.headers = list(response_headers) + return self._write + + def _write(self, data): + if not self.status: + raise RuntimeError("write() called before start_response()") + self._write_buffer.append(data) + + +def _call_app(caller_pid, module_name: str, callable_name: str, environ: dict): + """Call WSGI app and iterate response. + + Args: + caller_pid: Erlang PID to send response to + module_name: Python module name + callable_name: WSGI callable name + environ: Prepared environ dict with wsgi.input + + Returns: + 'done' on success, or schedule_inline marker for continuation + """ + try: + # Get app (preloaded or import on demand) + app = _get_app(module_name, callable_name) + response = _Response() + result = app(environ, response.start_response) + + # Build state for response iteration + state = { + 'caller': caller_pid, + 'status': response.status_code, + 'headers': response.headers, + 'result': result, + 'result_iter': iter(result) if hasattr(result, '__iter__') else None, + 'write_buffer': list(response._write_buffer), + 'headers_sent': False, + } + + # Call iterate_response directly + return _iterate_response(state) + + except Exception as e: + try: + erlang.send(caller_pid, (b'error', str(e).encode('utf-8'))) + except Exception: + pass + return b'error' + + +# ============================================================================ +# Iterate response +# ============================================================================ + +def _iterate_response(state: dict): + """Send response chunks, yielding every CHUNKS_PER_BATCH. + + Args: + state: Dict containing caller, status, headers, result iterator, etc. + + Returns: + 'done' or schedule_inline marker for continuation + """ + caller = state['caller'] + result = state['result'] + result_iter = state['result_iter'] + write_buffer = state['write_buffer'] + headers_sent = state['headers_sent'] + + try: + # Handle single bytes response + if result_iter is None or isinstance(result, (bytes, bytearray)): + body = _to_bytes(result) if isinstance(result, (bytes, bytearray)) else b'' + if write_buffer: + body = b''.join(_to_bytes(p) for p in write_buffer) + body + erlang.send(caller, (b'response', state['status'], state['headers'], body)) + _cleanup_result(result) + return b'done' + + # Fast path: list with small number of items - collect and send as single response + if isinstance(result, list) and len(result) <= 2 and not write_buffer: + body = b''.join(_to_bytes(chunk) for chunk in result if chunk) + erlang.send(caller, (b'response', state['status'], state['headers'], body)) + _cleanup_result(result) + return b'done' + + # Send any buffered write() data first + if write_buffer and not headers_sent: + erlang.send(caller, (b'start_response', state['status'], state['headers'])) + state['headers_sent'] = True + for part in write_buffer: + erlang.send(caller, (b'chunk', _to_bytes(part))) + state['write_buffer'] = [] + + # Process chunks from iterator + chunks_processed = 0 + + while True: + try: + chunk = next(result_iter) + except StopIteration: + break + + if chunk: + chunk = _to_bytes(chunk) + + # Send headers on first chunk + if not state['headers_sent']: + erlang.send(caller, (b'start_response', state['status'], state['headers'])) + state['headers_sent'] = True + + erlang.send(caller, (b'chunk', chunk)) + chunks_processed += 1 + + # Yield after batch to release scheduler + if chunks_processed >= CHUNKS_PER_BATCH: + return erlang.schedule_inline( + 'hornbeam_wsgi_worker', '_iterate_response', + args=[state] + ) + + # Done iterating + if state['headers_sent']: + erlang.send(caller, b'done') + else: + # Empty response (no chunks produced) + erlang.send(caller, (b'response', state['status'], state['headers'], b'')) + + _cleanup_result(result) + return b'done' + + except Exception as e: + _cleanup_result(state.get('result')) + try: + erlang.send(caller, (b'error', str(e).encode('utf-8'))) + except Exception: + pass + return b'error' + + +def _cleanup_result(result): + """Close result iterator if it has a close method.""" + if result and hasattr(result, 'close'): + try: + result.close() + except Exception: + pass + + +# ============================================================================ +# Tuple fast path - O(1) environ creation +# ============================================================================ + +def handle_request_tuple(caller_pid, buffer, app_module: bytes, app_callable: bytes, req_tuple): + """Fast path entry point using pre-parsed tuple from Erlang. + + This avoids per-key iteration in Python by having Erlang pre-convert + all headers to WSGI format (HTTP_*). + + Args: + caller_pid: Erlang PID to send response to + buffer: py_buffer for request body, or 'empty' atom for bodyless requests + app_module: Python module containing WSGI app (bytes) + app_callable: Name of WSGI callable in module (bytes) + req_tuple: Pre-parsed request tuple from hornbeam_request:build_wsgi_tuple/2 + (method, script_name, path_info, query_string, wsgi_headers, + content_type, content_length, body, server, client, scheme, + protocol, lifespan_state) + + Returns: + 'done' on success, or schedule_inline marker for continuation + """ + if not HAS_ERLANG: + return b'error' + + try: + # Convert bytes to strings + # erlang_python converts binaries to str in C + module_name = app_module + callable_name = app_callable + + # Create environ from pre-parsed tuple (O(1) operations only) + environ = _create_environ_from_tuple(req_tuple, buffer) + + # Call app directly + return _call_app(caller_pid, module_name, callable_name, environ) + + except Exception as e: + try: + erlang.send(caller_pid, (b'error', str(e).encode('utf-8'))) + except Exception: + pass + return b'error' + + +def _create_environ_from_tuple(req_tuple, buffer): + """Create WSGI environ from pre-parsed Erlang tuple - O(1) operations only. + + Erlang pre-parses all headers into WSGI format so Python only does + dict updates (no loops over headers). + + Note: erlang_python converts Erlang binaries to Python str in C, + so no decode() calls are needed here. + + Args: + req_tuple: Pre-parsed request tuple from hornbeam_request:build_wsgi_tuple/2 + buffer: py_buffer for request body, or 'empty' atom for bodyless requests + + Returns: + Complete WSGI environ dict + """ + (method, script_name, path_info, query_string, wsgi_headers, + content_type, content_length, _body, server, client, scheme, + protocol, lifespan_state) = req_tuple + + # Start with template copy (O(1) - shallow copy of small dict) + environ = _ENVIRON_TEMPLATE.copy() + + # Update with request-specific values + # All values are already str (erlang_python converts binaries to str in C) + environ['REQUEST_METHOD'] = method + environ['SCRIPT_NAME'] = script_name if script_name else '' + environ['PATH_INFO'] = path_info + environ['QUERY_STRING'] = query_string + environ['SERVER_NAME'] = server[0] + environ['SERVER_PORT'] = str(server[1]) + environ['SERVER_PROTOCOL'] = protocol + environ['wsgi.url_scheme'] = scheme + environ['REMOTE_ADDR'] = client[0] + environ['wsgi.errors'] = _SHARED_ERRORS + + # Use buffer as wsgi.input, or empty BytesIO for bodyless requests + if buffer == b'empty' or (hasattr(buffer, '__class__') and buffer.__class__.__name__ == 'Atom'): + environ['wsgi.input'] = io.BytesIO() + else: + environ['wsgi.input'] = buffer + + # Add pre-converted HTTP_* headers (already str from erlang_python) + # Direct dict.update() - O(n) but no per-item function calls + if wsgi_headers: + environ.update(wsgi_headers) + + # Add content-type/length if present + if content_type is not None: + environ['CONTENT_TYPE'] = content_type + if content_length is not None: + environ['CONTENT_LENGTH'] = content_length + + # Store lifespan state + if lifespan_state: + environ['_hornbeam.lifespan_state'] = lifespan_state + + return environ diff --git a/rebar.config b/rebar.config index 0043d41..c7fa496 100644 --- a/rebar.config +++ b/rebar.config @@ -20,7 +20,7 @@ {deps, [ {cowboy, "2.12.0"}, - {erlang_python, "1.8.1"} + {erlang_python, "2.2.0"} ]}. {shell, [ @@ -39,7 +39,7 @@ {profiles, [ {test, [ {deps, [ - {hackney, "3.0.2"}, + {hackney, "3.2.1"}, {jsx, "3.1.0"} ]} ]} diff --git a/rebar.lock b/rebar.lock deleted file mode 100644 index ecd87bb..0000000 --- a/rebar.lock +++ /dev/null @@ -1,17 +0,0 @@ -{"1.2.0", -[{<<"cowboy">>,{pkg,<<"cowboy">>,<<"2.12.0">>},0}, - {<<"cowlib">>,{pkg,<<"cowlib">>,<<"2.13.0">>},1}, - {<<"erlang_python">>,{pkg,<<"erlang_python">>,<<"1.8.1">>},0}, - {<<"ranch">>,{pkg,<<"ranch">>,<<"1.8.0">>},1}]}. -[ -{pkg_hash,[ - {<<"cowboy">>, <<"F276D521A1FF88B2B9B4C54D0E753DA6C66DD7BE6C9FCA3D9418B561828A3731">>}, - {<<"cowlib">>, <<"DB8F7505D8332D98EF50A3EF34B34C1AFDDEC7506E4EE4DD4A3A266285D282CA">>}, - {<<"erlang_python">>, <<"4DAFC7AFD315F0D5D45792F722364D8721CED438E396B7D14F19518DC78D198B">>}, - {<<"ranch">>, <<"8C7A100A139FD57F17327B6413E4167AC559FBC04CA7448E9BE9057311597A1D">>}]}, -{pkg_hash_ext,[ - {<<"cowboy">>, <<"8A7ABE6D183372CEB21CAA2709BEC928AB2B72E18A3911AA1771639BEF82651E">>}, - {<<"cowlib">>, <<"E1E1284DC3FC030A64B1AD0D8382AE7E99DA46C3246B815318A4B848873800A4">>}, - {<<"erlang_python">>, <<"0F5893100A92285096519F8111D3A6D3F5DCD18668545ADF9F4144899D5997C2">>}, - {<<"ranch">>, <<"49FBCFD3682FAB1F5D109351B61257676DA1A2FDBE295904176D5E521A2DDFE5">>}]} -]. diff --git a/src/hornbeam.erl b/src/hornbeam.erl index f13bd75..3611b98 100644 --- a/src/hornbeam.erl +++ b/src/hornbeam.erl @@ -26,7 +26,7 @@ %%% %% Start with options %%% hornbeam:start("myapp:application", #{ %%% bind => "0.0.0.0:8000", -%%% workers => 4 +%%% num_contexts => 4 %%% }). %%% %%% %% Start ASGI app with lifespan @@ -38,7 +38,7 @@ %%% %% Multi-app mode - mount different apps at different prefixes %%% hornbeam:start(#{ %%% mounts => [ -%%% {"/api", "api:app", #{worker_class => asgi, workers => 4}}, +%%% {"/api", "api:app", #{worker_class => asgi, num_contexts => 4}}, %%% {"/admin", "admin:app", #{worker_class => wsgi}}, %%% {"/", "frontend:app", #{worker_class => wsgi}} %%% ], @@ -66,9 +66,10 @@ -type mount_spec() :: {Prefix :: string() | binary(), AppSpec :: app_spec(), Opts :: map()}. -type options() :: #{ bind => string() | binary(), - workers => pos_integer(), + num_contexts => pos_integer(), num_acceptors => pos_integer(), worker_class => wsgi | asgi, + context_mode => worker | owngil, timeout => pos_integer(), keepalive => pos_integer(), max_requests => pos_integer(), @@ -133,14 +134,14 @@ start(AppSpec) when is_list(AppSpec); is_binary(AppSpec) -> %% Options: %%