From 794af5590f6c9465055c63c3e94193c4b8e73066 Mon Sep 17 00:00:00 2001 From: Jacob Quinn Date: Wed, 5 Aug 2026 21:37:53 -0600 Subject: [PATCH 1/2] fix(server): use interactive thread pool --- CHANGELOG.md | 4 ++ README.md | 9 ++++ docs/src/guides/server.md | 37 +++++++++++++++++ src/HTTP.jl | 4 ++ src/http2_server.jl | 5 ++- src/http_handlers.jl | 3 +- src/http_server.jl | 15 ++++++- src/http_sse.jl | 2 +- src/http_websockets.jl | 10 ++++- test/http2_server_tests.jl | 5 +++ test/http_handlers_tests.jl | 4 ++ test/http_server_http1_tests.jl | 64 +++++++++++++++++++++++++++++ test/http_websocket_server_tests.jl | 4 ++ 13 files changed, 159 insertions(+), 7 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 82c9ea9ad..51e4a6551 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - Added `HTTP.peeraddr(::HTTP.Stream)`, returning the remote (client) `SocketAddr` of a server stream for both plain-TCP and TLS connections and both HTTP/1 and HTTP/2. This is the supported way to obtain the client IP (for rate limiting, audit logging, and per-client policy) without reaching into transport internals, and restores the capability `Sockets.getpeername(::HTTP.Stream)` provided in HTTP.jl 1.x. ### Fixed +- Restored HTTP and WebSocket server task scheduling to Julia's `:interactive` + thread pool so default-pool compute work cannot starve server and health-check + tasks when an interactive thread is configured. ([#1342]) - Percent-decode `userinfo` before building the `Basic` auth header (RFC 3986); fixes wrong credentials for request URLs and proxies containing percent-encoded characters. ## [v2.0.0] - 2026-04-27 @@ -822,3 +825,4 @@ See changes for 0.9.15: this release is equivalent to 0.9.15 with [#752] reverte [#1119]: https://github.com/JuliaWeb/HTTP.jl/issues/1119 [#1126]: https://github.com/JuliaWeb/HTTP.jl/issues/1126 [#1127]: https://github.com/JuliaWeb/HTTP.jl/issues/1127 +[#1342]: https://github.com/JuliaWeb/HTTP.jl/issues/1342 diff --git a/README.md b/README.md index 7fc7dda85..f266273f1 100644 --- a/README.md +++ b/README.md @@ -119,6 +119,14 @@ Each callback receives an `HTTP.SSEEvent` with the parsed `data`, `event`, Use `HTTP.serve!` for request/response handlers: +> HTTP.jl schedules server tasks on Julia's `:interactive` thread pool so they +> can remain responsive when the default pool is busy. Start production servers +> with at least one interactive thread, for example +> `julia --threads=4,1 server.jl`. Without an interactive thread, Julia falls +> back to the default pool and non-yielding compute tasks can delay HTTP work, +> including health checks. See the +> [server guide][server-guide-url] for configuration and handler guidance. + ```julia using HTTP @@ -174,3 +182,4 @@ HTTP.WebSockets.forceclose(server) [issues-url]: https://github.com/JuliaWeb/HTTP.jl/issues [migration-guide-url]: https://juliaweb.github.io/HTTP.jl/dev/guides/migration-1x/ +[server-guide-url]: https://juliaweb.github.io/HTTP.jl/dev/guides/server/ diff --git a/docs/src/guides/server.md b/docs/src/guides/server.md index 77032a1cd..cd7c2c0d0 100644 --- a/docs/src/guides/server.md +++ b/docs/src/guides/server.md @@ -8,6 +8,43 @@ CurrentModule = HTTP handlers. The right choice depends on how much control you need over read/write sequencing. +## Interactive Thread Pool + +HTTP.jl schedules its server tasks on Julia's `:interactive` thread pool. This +includes listener, connection, request-handler, HTTP/2 stream, server-side SSE, +and WebSocket server tasks. Keeping this work separate from the `:default` pool +allows the server to accept and handle requests, including health checks, while +the default pool runs compute-intensive tasks that may not yield. + +Configure at least one interactive thread when starting a production server. +For example, this command creates four default worker threads and one +interactive thread: + +```sh +julia --threads=4,1 --project=. server.jl +``` + +The equivalent environment setting is `JULIA_NUM_THREADS=4,1`. Check the live +configuration with `Threads.nthreads(:interactive)`, which should return at +least `1`. + +If no interactive thread exists, Julia runs tasks requested for `:interactive` +on the default pool. The server still starts, but it loses isolation from +default-pool work. Non-yielding compute tasks can then delay all HTTP work and +make health checks appear unresponsive. + +Interactive tasks should remain responsive. Do not run long, non-yielding +compute kernels directly in a server handler. Move that work to the default +pool and wait for it from the handler so the interactive task can yield: + +```julia +result = fetch(Threads.@spawn :default expensive_work()) +``` + +A non-yielding handler can still monopolize the interactive pool. The separate +pool protects HTTP work from compute tasks assigned to `:default`; it cannot +make non-yielding handler code cooperative. + ## Request Handlers Use `HTTP.serve!` or `HTTP.serve` when your application naturally maps diff --git a/src/HTTP.jl b/src/HTTP.jl index af743234d..59d5d6099 100644 --- a/src/HTTP.jl +++ b/src/HTTP.jl @@ -26,6 +26,10 @@ using URIs const VERSION = v"2.0.0" +macro _spawn_interactive(ex) + return esc(:(errormonitor(Threads.@spawn :interactive $ex))) +end + export WebSockets export escape diff --git a/src/http2_server.jl b/src/http2_server.jl index a19d732f0..e5e54250d 100644 --- a/src/http2_server.jl +++ b/src/http2_server.jl @@ -1593,7 +1593,10 @@ function _dispatch_h2_stream!( _fail_h2_server_stream!(server, tracked, conn, write_lock, states_lock, states, send_state, state, _H2_ERROR_PROTOCOL) return nothing end - Threads.@spawn _handle_h2_stream!(server, tracked, conn, write_lock, send_state, states_lock, states, state.stream_id, state, decoded_headers::Vector{HeaderField}) + @_spawn_interactive _handle_h2_stream!( + server, tracked, conn, write_lock, send_state, states_lock, states, + state.stream_id, state, decoded_headers::Vector{HeaderField}, + ) return nothing end diff --git a/src/http_handlers.jl b/src/http_handlers.jl index 0cd061aa1..1e48480ab 100644 --- a/src/http_handlers.jl +++ b/src/http_handlers.jl @@ -33,6 +33,7 @@ import ..canceled import ..body_close! import ..get_request_context import .._request_with_context +import ..@_spawn_interactive import ..@try_ignore """ @@ -429,7 +430,7 @@ function (middleware::_HandlerTimeoutMiddleware)(req::Request) derived_ctx = _timeout_child_context(get_request_context(req), middleware.timeout_ns) timed_req = _request_with_context(req, derived_ctx) result = Channel{Tuple{Bool,Any}}(1) - Threads.@spawn begin + @_spawn_interactive begin try put!(result, (true, middleware.handler(timed_req))) catch err diff --git a/src/http_server.jl b/src/http_server.jl index 2288d3680..2a00b2fbf 100644 --- a/src/http_server.jl +++ b/src/http_server.jl @@ -63,6 +63,11 @@ The handle owns the listener, background task, active-connection set, and timeout configuration. Keep it around for lifecycle operations such as [`port`](@ref), `wait(server)`, `close(server)`, or [`forceclose`](@ref). +HTTP.jl schedules server tasks on Julia's `:interactive` thread pool. Start +Julia with at least one interactive thread, such as `--threads=4,1`, to keep +server and health-check work isolated from non-yielding tasks on the default +pool. Without an interactive thread, Julia falls back to the default pool. + Timeout fields are stored in nanoseconds. Use the convenience `listen!` and `serve!` keywords to configure request-read, header-read, response-write, and idle deadlines without constructing a `Server` manually. @@ -1238,7 +1243,7 @@ function _serve_listener!(server::Server, listener::Union{TCP.Listener,TLS.Liste end tracked = _ServerConn(conn, ReentrantLock(), nothing, _ConnState.NEW, floor(Int64, time())) _track_conn!(server, tracked) - Threads.@spawn _serve_conn!(server, tracked) + @_spawn_interactive _serve_conn!(server, tracked) end return nothing end @@ -1262,7 +1267,7 @@ function _start_server_task!(f::F, server::Server)::Server where {F} state == _ServerState.CLOSED && throw(ProtocolError("closed servers cannot be restarted")) state == _ServerState.RUNNING && throw(ProtocolError("server is already running")) ready = Threads.Event(true) - task = Threads.@spawn begin + task = @_spawn_interactive begin try f(ready) catch @@ -1355,6 +1360,9 @@ request and writing the response. Timeout keywords ending in `_ns` are nanoseconds; `read_timeout`, `read_header_timeout`, `write_timeout`, and `idle_timeout` accept seconds. The older `readtimeout` keyword is accepted as a seconds-valued migration alias for `read_timeout`. + +Server tasks use Julia's `:interactive` thread pool. Configure at least one +interactive thread; see [`Server`](@ref) and the [Server Guide](@ref). """ function listen!( handler::F, host::AbstractString="127.0.0.1", port_num::Integer=8080; @@ -1519,6 +1527,9 @@ Timeout keywords ending in `_ns` are nanoseconds; the older `readtimeout` keyword is accepted as a seconds-valued migration alias for `read_timeout`. Ordinary request handlers buffer request bodies before dispatch; `max_body_bytes` caps that buffering, and `0` restores the legacy unbounded behavior. + +Server tasks use Julia's `:interactive` thread pool. Configure at least one +interactive thread; see [`Server`](@ref) and the [Server Guide](@ref). """ function serve!( handler::F, diff --git a/src/http_sse.jl b/src/http_sse.jl index 328366889..beeaf3a8a 100644 --- a/src/http_sse.jl +++ b/src/http_sse.jl @@ -239,7 +239,7 @@ end function sse_stream(response::Response, f::Function; max_len::Integer=_DEFAULT_SSE_STREAM_MAX_LEN)::SSEStream stream = sse_stream(response; max_len=max_len) - Threads.@spawn begin + @_spawn_interactive begin try f(stream) catch err diff --git a/src/http_websockets.jl b/src/http_websockets.jl index b9517ffdc..1f1bbb7b3 100644 --- a/src/http_websockets.jl +++ b/src/http_websockets.jl @@ -92,6 +92,7 @@ import .._is_transport_timeout import .._wrap_transport_timeout import ..Stream import .._clear_deadlines! +import ..@_spawn_interactive include("http_websocket_pmce.jl") include("http_websocket_codec.jl") @@ -1744,7 +1745,7 @@ function serve!(server::Server, listener, ready::Threads.Event)::Server err isa EOFError && return server rethrow(err) end - Threads.@spawn _serve_ws_conn!(server, conn) + @_spawn_interactive _serve_ws_conn!(server, conn) end return server end @@ -1846,6 +1847,11 @@ read the actual address afterwards with [`server_addr`](@ref). Pass ([RFC 7692](https://www.rfc-editor.org/rfc/rfc7692)); it is negotiated per connection and clients must also opt in. `maxframesize` defaults to 16 MiB and bounds incoming frame/message buffering. + +WebSocket server tasks use Julia's `:interactive` thread pool. Start Julia with +at least one interactive thread, such as `--threads=4,1`, to isolate server work +from non-yielding tasks on the default pool. Without an interactive thread, +Julia falls back to the default pool. """ function listen!( handler::Function, @@ -1874,7 +1880,7 @@ function listen!( compress=compress, ) ready = Threads.Event(true) - server.serve_task = Threads.@spawn begin + server.serve_task = @_spawn_interactive begin try _listen_ws(server, ready) catch diff --git a/test/http2_server_tests.jl b/test/http2_server_tests.jl index 076e2fefa..4de371200 100644 --- a/test/http2_server_tests.jl +++ b/test/http2_server_tests.jl @@ -203,7 +203,9 @@ end end @testset "HTTP/2 server request handling" begin + handler_pools = Channel{Symbol}(2) server = HT.serve!("127.0.0.1", 0; listenany = true) do request + put!(handler_pools, Threads.threadpool()) payload = collect(codeunits("h2:" * request.target)) return HT.Response(200, HT.BytesBody(payload); content_length = length(payload), proto_major = 2, proto_minor = 0) end @@ -218,6 +220,9 @@ end @test res2.status == 200 @test String(_read_all_h2_server(res1.body)) == "h2:/one" @test String(_read_all_h2_server(res2.body)) == "h2:/two" + expected_pool = Threads.nthreads(:interactive) > 0 ? :interactive : :default + @test take!(handler_pools) == expected_pool + @test take!(handler_pools) == expected_pool finally close(conn) HT.forceclose(server) diff --git a/test/http_handlers_tests.jl b/test/http_handlers_tests.jl index e01ac0e7b..e206fab7f 100644 --- a/test/http_handlers_tests.jl +++ b/test/http_handlers_tests.jl @@ -145,13 +145,17 @@ end end @testset "HTTP handlers request timeout middleware" begin + handler_pool = Channel{Symbol}(1) fast = HT.Handlers.handlertimeout(5.0)(req -> begin _ = req + put!(handler_pool, Threads.threadpool()) return _response_with_text("ok") end) fast_resp = fast(HT.Request("GET", "/")) @test fast_resp.status == 200 @test String(_read_all_handler_bytes(fast_resp.body)) == "ok" + expected_pool = Threads.nthreads(:interactive) > 0 ? :interactive : :default + @test take!(handler_pool) == expected_pool slow = HT.Handlers.handlertimeout(0.02; status = 504, body = "custom timeout")(req -> begin _ = req diff --git a/test/http_server_http1_tests.jl b/test/http_server_http1_tests.jl index f223c6f9f..2e6ea3dec 100644 --- a/test/http_server_http1_tests.jl +++ b/test/http_server_http1_tests.jl @@ -8,6 +8,18 @@ const NC = Reseau.TCP const ND = Reseau.HostResolvers const IOP = Reseau.IOPoll +mutable struct _DefaultPoolBlockState + @atomic started::Int + @atomic stop::Bool +end + +function _block_default_pool!(state::_DefaultPoolBlockState, deadline_ns::Int64)::Nothing + @atomic state.started += 1 + while !(@atomic :acquire state.stop) && time_ns() < deadline_ns + end + return nothing +end + function _read_all_server_bytes(body::HT.AbstractBody)::Vector{UInt8} out = UInt8[] buf = Vector{UInt8}(undef, 32) @@ -246,6 +258,54 @@ function _read_until_close(conn::NC.Conn; timeout_s::Float64 = 1.0, wait_for_fir end end +@testset "HTTP server remains responsive on the interactive pool" begin + if Threads.nthreads(:interactive) == 0 + @test true + else + state = _DefaultPoolBlockState(0, false) + blockers = Task[] + server = HT.serve!("127.0.0.1", 0; listenany = true) do request + _ = request + @atomic :release state.stop = true + return HT.Response(200, string(Threads.threadpool())) + end + address = HT.server_addr(server) + try + warm = HT.get( + "http://$(address)/warm"; + headers = ["Connection" => "close"], + proxy = HT.ProxyConfig(), + ) + @test String(warm.body) == "interactive" + @atomic :release state.stop = false + worker_count = Threads.nthreads(:default) + deadline_ns = Int64(time_ns()) + Int64(4_000_000_000) + for _ in 1:worker_count + push!(blockers, errormonitor(Threads.@spawn :default _block_default_pool!(state, deadline_ns))) + end + started = timedwait( + () -> (@atomic :acquire state.started) == worker_count, + 2.0; + pollint=0.001, + ) + @test started == :ok + elapsed = @elapsed response = HT.get( + "http://$(address)/health"; + headers = ["Connection" => "close"], + proxy = HT.ProxyConfig(), + ) + @test response.status == 200 + @test String(response.body) == "interactive" + @test elapsed < 2.0 + finally + @atomic :release state.stop = true + wait.(blockers) + HT.forceclose(server) + wait(server) + end + end +end + @testset "HTTP server SSE helper" begin response = HT.sse_stream(200) @test response.body isa HT.SSEStream @@ -257,9 +317,11 @@ end end @testset "HTTP server SSE roundtrip" begin + producer_pool = Channel{Symbol}(1) server = HT.serve!("127.0.0.1", 0; listenany = true) do request _ = request response = HT.sse_stream(200) do stream + put!(producer_pool, Threads.threadpool()) write(stream, HT.SSEEvent("first")) write(stream, HT.SSEEvent("second"; event = "update", id = "2", retry = 2500)) write(stream, HT.SSEEvent("multi\nline\ndata")) @@ -283,6 +345,8 @@ end @test events[2].id == "2" @test events[2].retry == 2500 @test events[3].data == "multi\nline\ndata" + expected_pool = Threads.nthreads(:interactive) > 0 ? :interactive : :default + @test take!(producer_pool) == expected_pool finally _run_with_timeout(() -> HT.forceclose(server); label = "server forceclose") _run_with_timeout(() -> wait(server); label = "server task completion") diff --git a/test/http_websocket_server_tests.jl b/test/http_websocket_server_tests.jl index e6f8543dc..4c14d6b8d 100644 --- a/test/http_websocket_server_tests.jl +++ b/test/http_websocket_server_tests.jl @@ -124,7 +124,9 @@ end end @testset "HTTP.WebSockets server listen! over ws" begin + handler_pool = Channel{Symbol}(1) server = W.listen!("127.0.0.1", 0) do ws + put!(handler_pool, Threads.threadpool()) msg = W.receive(ws) W.send(ws, msg) end @@ -134,6 +136,8 @@ end try W.send(ws, "hello") @test W.receive(ws) == "hello" + expected_pool = Threads.nthreads(:interactive) > 0 ? :interactive : :default + @test take!(handler_pool) == expected_pool finally close(ws) end From d86edb61011c2e762ade2f9927b7828a6e7fd0fb Mon Sep 17 00:00:00 2001 From: Jacob Quinn Date: Wed, 5 Aug 2026 23:44:44 -0600 Subject: [PATCH 2/2] fix(server): keep interactive spawn trim-safe Preserve the existing task error behavior while selecting the interactive pool. Wrapping server tasks in errormonitor pulls Base error-display I/O into JuliaC strict trim compilation on Julia 1.13 and produces unresolved dynamic calls. --- src/HTTP.jl | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/HTTP.jl b/src/HTTP.jl index 59d5d6099..f2aebf7cc 100644 --- a/src/HTTP.jl +++ b/src/HTTP.jl @@ -27,7 +27,7 @@ using URIs const VERSION = v"2.0.0" macro _spawn_interactive(ex) - return esc(:(errormonitor(Threads.@spawn :interactive $ex))) + return esc(:(Threads.@spawn :interactive $ex)) end export WebSockets