Run HTTP server tasks on the interactive thread pool

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

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/

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

src/HTTP.jl

@@ -26,6 +26,10 @@ using URIs
 
 const VERSION = v"2.0.0"
 
+macro _spawn_interactive(ex)
+    return esc(:(Threads.@spawn :interactive $ex))
+end
+
 export WebSockets
 export escape
 

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
 

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

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,

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

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

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)

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

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")

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