Fix streaming hangs and enable streaming on Julia 1.10 by s-celles · Pull Request #127 · JuliaIO/gRPCClient.jl

s-celles · 2026-07-30T04:51:55Z

Closes #68.

The problem

src/Streaming.jl is compiled out below Julia 1.12 because streaming calls hang there. I reproduced it against the Go test server: a bidirectional stream moving 1000 messages with -t 4 hangs roughly one run in three on Julia 1.10.

It is not a Julia version problem. The same wedge, plus a second one at end of stream, also reproduces on 1.12 (libcurl 8.15.0) given enough runs — about 1 in 150 — so main has a live bug today, it is just rarer there.

Root cause

read_callback pauses libcurl's send direction whenever the request buffer runs dry, which for a streaming request is every time the caller has nothing queued. Each curl_easy_pause(CURLPAUSE_CONT) the request pump issues to resume it is one-shot.

libcurl can drop a transfer on the floor around one of those un-pauses: it stops servicing the transfer's socket entirely. In one captured hang the client's receive queue held 150 KB of unread response bytes — exactly the messages the caller was still blocked waiting for — while the socket watcher called curl_multi_socket_action(sock, CURL_CSELECT_IN) roughly 190,000 times per second, each returning CURLM_OK and consuming nothing. Probing the wedged handle established that only another un-pause recovers it; driving the multi handle, by socket event or by timeout, does not.

The end-of-stream path had the mirror-image bug, and unlike the first one it is reachable on every version. It un-paused and only then closed req.request_c, down in the finally. A read_callback landing in that window still saw an open request stream, so it paused again instead of returning 0, and the transfer sat paused with nothing left to un-pause it: all data exchanged, both peers done, call hung until its deadline.

The fix

An in-flight request-streaming call re-issues the un-pause every 50 ms while libcurl holds none of its request data, rather than relying on a single one landing. curl_easy_pause returns immediately when the transfer is not paused, and an un-pause with an empty request buffer just draws one read_callback that pauses straight back, so a healthy call pays only for the call itself. The timer is armed only for request-streaming calls and closed in cleanup_request alongside the deadline watchdog.
The end-of-stream path closes req.request_c before the un-pause, so the callback that un-pause draws can only take its end-of-stream branch.
Streaming is enabled on all supported versions: the @static if VERSION >= v"1.12" gate is gone from src/gRPCClient.jl and from the streaming testsets, and the docs no longer describe streaming as 1.12 only.

Notes for review

test/runtests.jl is mostly a de-indent. git diff -w shows the real change is three hunks: the removed guard, and the Don't Stick User Tasks edit below.
Don't Stick User Tasks is now broken = VERSION < v"1.12". It is a unary test that only happened to sit inside the guarded block. The stickiness comes from arming the deadline watchdog with Timer(cb, delay), whose callback loop is an @async; on Julia ≤ 1.11 scheduling a sticky task marks the scheduling task sticky too (@async tasks executes simultaneously with parent task if launched with @spawn JuliaLang/julia#41324, fixed by the 1.12 scheduler). Confirmed directly: on 1.10 Timer(cb, delay) stickies its caller and Timer(delay) does not. Nothing this package can do short of dropping the watchdog.
One change was tried and deliberately dropped. socket_callback tears down and rebuilds the FDWatcher on every flag change, and since FDWatcher is a handle onto one refcounted libuv poll per fd, that discards readiness libuv had recorded but the watcher task had not consumed. Creating the replacement before releasing the old one looked correct and reduced churn, but a control run against unmodified main proved it introduced end-of-stream hangs on 1.12, so it is not in this PR. Worth a separate look.
No version bump in Project.toml — left to you.

Verification

Against the Go test server on Julia 1.10.11 (libcurl 8.4.0) and 1.12.6 (libcurl 8.15.0):

1000 bidirectional stress runs (N=1000), across both versions and both request-channel close orderings — zero hangs. Every one of these soaks reproduced a hang before the fix.
Full suite 12 times per version, plus -t 1, -t 2 and -t 8 on 1.10 — all pass.
Runic clean.

Assisted by AI.

codecov-commenter · 2026-07-30T04:53:13Z

⚠️ Please install the to ensure uploads and comments are reliably processed by Codecov.

Codecov Report

❌ Patch coverage is 90.00000% with 1 line in your changes missing coverage. Please review.
✅ Project coverage is 92.21%. Comparing base (440c5fc) to head (6c32b4c).

Files with missing lines	Patch %	Lines
src/Curl.jl	88.88%	1 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main     #127      +/-   ##
==========================================
+ Coverage   92.09%   92.21%   +0.12%     
==========================================
  Files           7        7              
  Lines         645      655      +10     
==========================================
+ Hits          594      604      +10     
  Misses         51       51

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

`src/Streaming.jl` was compiled out below Julia 1.12 because streaming calls hung there. The hang reproduces on a bidirectional stream that moves ~1000 messages with `-t 4`, roughly one run in three on Julia 1.10, and it is not a Julia version problem: the same wedge, plus a second one at end of stream, also reproduces on 1.12 given enough runs. Both come from the same place. read_callback pauses libcurl's send direction whenever the request buffer runs dry, and each `curl_easy_pause(CURLPAUSE_CONT)` the request pump issues to resume it is one-shot. libcurl can drop a transfer on the floor around one of those un-pauses: it stops servicing the transfer's socket entirely, so the peer's response bytes pile up unread in the receive queue - 150KB of them in one captured hang, exactly the responses the caller was still waiting for - while `curl_multi_socket_action` keeps reporting CURLM_OK without consuming any. Nothing but another un-pause recovers it; driving the multi handle, by socket event or by timeout, does not. So a caller that waits on responses before sending more deadlocks until its deadline. An in-flight request-streaming call now re-issues the un-pause every 50ms while libcurl holds none of its request data, instead of relying on a single one landing. `curl_easy_pause` returns immediately when the transfer is not paused, and an un-pause with an empty request buffer just draws one read_callback that pauses straight back, so a healthy call pays only for the call itself. The end-of-stream path had the mirror-image bug, and it was reachable on every version: it un-paused and only then closed `req.request_c` in its `finally`, so a read_callback landing in between still saw an open stream, paused again, and left the transfer paused with nobody to un-pause it - all data exchanged, both peers done, call hung until the deadline. It now closes the stream first, so the callback the un-pause draws can only take its end-of-stream branch. Streaming is enabled on all supported versions, the version guard around the streaming testsets is gone, and the docs no longer describe streaming as 1.12 only. "Don't Stick User Tasks" is marked broken below 1.12: it is a unary test that only sat in the guarded block, and the stickiness comes from arming the deadline watchdog with `Timer(cb, delay)`, whose `@async` marks the scheduling task sticky on those versions (JuliaLang/julia#41324). Verified against the Go test server on 1.10.11 (libcurl 8.4.0) and 1.12.6 (libcurl 8.15.0): 1000 bidirectional stress runs across both versions and both close orderings with no hang, the full suite 12 times per version plus `-t 1`, `-t 2` and `-t 8` on 1.10, and Runic clean. Assisted by AI.

csvance · 2026-07-30T13:09:05Z

Thank you for getting to the bottom of this issue @s-celles! I believe I understand it now and I'm surprised I managed to overlook this aspect of it. Going to try and work through this today and test everything out.

csvance · 2026-07-30T14:04:46Z

@s-celles I was able to reproduce on 1.12. Also, I believe a single critical section could be used in-place of polling in order to avoid the race. Not sure what this does to lock contention, but I would think its a reasonable tradeoff to guarantee the hang won't happen. I guess another potential problem is if the close calls yield to another task and this tasks is not scheduled again for a significant amount of time.

            if !req.completed
                # Wait for any request data to be flushed by curl
                wait(req.curl_done_reading)

                # Mutually exclusive with read_callback
                lock(req.lock) do
                    req.completed && return

                    close(req.request_c)

                    # Trigger a "return 0" in read_callback so curl ends the current request
                    req.send_paused && curl_easy_pause(req.easy, CURLPAUSE_CONT)
                end
            end

Here is a script I used to reproduce the issue:

#!/usr/bin/env julia
#
# Self-contained reproducer for the streaming wedge described in
# https://github.com/JuliaIO/gRPCClient.jl/pull/127
#
# Run with the Go test server on localhost:8001 and several threads:
#
#   cd test/go && go build -o grpc_test_server && ./grpc_test_server &
#   julia --project -t 8 repro_wedge.jl [mode] [iterations]
#
# modes:
#   eos       burst-send N, drain N responses, close the request channel, time the await.
#             Targets the end-of-stream ordering bug: the pump un-pauses libcurl and only
#             then closes req.request_c, so a read_callback landing in between re-pauses a
#             transfer nobody will un-pause again.
#   pingpong  strict put!/take! alternation, so the request buffer runs dry on every
#             message and the caller is always blocked on a response before it sends more.
#             Targets the mid-stream wedge.
#
# env:
#   POKE=1    run a background task that drives the multi handle in a loop. This does not
#             change any package state, it only widens the window in which libcurl can
#             invoke read_callback, so a race that is otherwise 1-in-hundreds becomes
#             frequent. Fault injection, not a fix and not part of normal operation.
#   STALL=5   seconds without progress before a round is declared wedged
#
# On a wedge the script prints the request state, then probes it: first by driving the
# multi handle (socket action on CURL_SOCKET_TIMEOUT), then by re-issuing
# curl_easy_pause(CURLPAUSE_CONT). Which probe releases the call identifies the failure.

using gRPCClient
using ProtoBuf
using Printf

import gRPCClient:
    gRPCRequest,
    grpc_global_handle,
    grpc_cancel,
    check_multi_info,
    curl_easy_pause,
    CURLPAUSE_CONT,
    curl_multi_socket_action,
    CURL_SOCKET_TIMEOUT

const REPO = get(ENV, "GRPCCLIENT_REPO", "/home/csvance/Git/gRPCClient.jl")
include(joinpath(REPO, "test", "gen", "test", "test_pb.jl"))

const HOST = get(ENV, "GRPC_TEST_SERVER_HOST", "localhost")
const PORT = parse(UInt16, get(ENV, "GRPC_TEST_SERVER_PORT", "8001"))
const STALL = parse(Float64, get(ENV, "STALL", "5"))
const POKE = get(ENV, "POKE", "0") == "1"

# ---------------------------------------------------------------- diagnostics

function state(req::gRPCRequest)
    g = grpc_global_handle()
    nwatch = lock(g.watchers_lock) do
        length(g.watchers)
    end
    return @sprintf(
        "completed=%-5s ex=%-24s request_ptr=%-6d request.size=%-6d done_reading=%-5s request_c_open=%-5s watchers=%d inflight=%d",
        req.completed,
        req.ex === nothing ? "nothing" : string(nameof(typeof(req.ex))),
        req.request_ptr,
        req.request.size,
        req.curl_done_reading.set,
        isopen(req.request_c),
        nwatch,
        length(g.requests),
    )
end

# Drive the multi handle exactly the way the package's own libcurl timeout Timer does.
function drive_multi()
    g = grpc_global_handle()
    return lock(g.lock) do
        if g.running && g.multi != Ptr{Cvoid}(0)
            curl_multi_socket_action(g.multi, CURL_SOCKET_TIMEOUT, 0, Ref{Cint}())
            check_multi_info(g)
        end
    end
end

# Re-issue the one-shot un-pause the request pump uses.
function reissue_unpause(req::gRPCRequest)
    g = grpc_global_handle()
    return lock(g.lock) do
        req.completed && return :completed
        curl_easy_pause(req.easy, CURLPAUSE_CONT)
        return :issued
    end
end

# Wait up to `secs` for `counter` to move off `from`.
function moved(counter::Threads.Atomic{Int}, from::Int, secs::Float64)
    t0 = time()
    while time() - t0 < secs
        counter[] != from && return true
        sleep(0.01)
    end
    return false
end

function probe(req::gRPCRequest, progress::Threads.Atomic{Int}, label::String)
    println("\n=== WEDGED ($label) ===")
    println("  before: ", state(req))

    at = progress[]
    drive_multi()
    if moved(progress, at, 1.0)
        println("  RECOVERED by driving the multi handle (socket action on CURL_SOCKET_TIMEOUT)")
        return :multi
    end
    println("  driving the multi handle changed nothing: ", state(req))

    at = progress[]
    r = reissue_unpause(req)
    if moved(progress, at, 1.0)
        println("  RECOVERED by re-issuing curl_easy_pause(CURLPAUSE_CONT) ($r)")
        return :unpause
    end
    println("  re-issuing the un-pause changed nothing either ($r): ", state(req))
    return :neither
end

# ---------------------------------------------------------------- workloads

# Burst N requests, drain N responses, then close the request channel and time the await.
# Everything but the close has already happened by then, so a slow await is the
# end-of-stream path and nothing else.
function round_eos(client, N::Int)
    request_c = Channel{TestRequest}(N)
    response_c = Channel{TestResponse}(N)

    req = grpc_async_request(client, request_c, response_c)
    progress = Threads.Atomic{Int}(0)

    for i in 1:N
        put!(request_c, TestRequest(1, UInt64[]))
    end
    for i in 1:N
        take!(response_c)
        Threads.atomic_add!(progress, 1)
    end

    # The window under test: the pump un-pauses libcurl and then closes req.request_c.
    close(request_c)

    awaited = Threads.Atomic{Int}(0)
    t = Threads.@spawn begin
        try
            grpc_async_await(req)
        catch
        end
        Threads.atomic_add!(awaited, 1)
    end

    t0 = time()
    while awaited[] == 0 && time() - t0 < STALL
        sleep(0.005)
    end

    if awaited[] == 0
        verdict = probe(req, awaited, "end of stream, await has not returned after $(STALL)s")
        grpc_cancel(req)
        wait(t)
        return (:wedged, verdict, time() - t0)
    end

    wait(t)
    return (:ok, :none, time() - t0)
end

# Strict alternation: one request, wait for its response, repeat. The request buffer is
# dry between every message and the caller is always blocked on a response, which is the
# state the PR describes as deadlocking.
function round_pingpong(client, N::Int)
    request_c = Channel{TestRequest}(1)
    response_c = Channel{TestResponse}(1)

    req = grpc_async_request(client, request_c, response_c)
    progress = Threads.Atomic{Int}(0)

    worker = Threads.@spawn begin
        try
            for i in 1:N
                put!(request_c, TestRequest(1, UInt64[]))
                Threads.atomic_add!(progress, 1)
                take!(response_c)
                Threads.atomic_add!(progress, 1)
            end
            close(request_c)
            grpc_async_await(req)
            Threads.atomic_add!(progress, 1)
        catch
            Threads.atomic_add!(progress, 1_000_000)
        end
    end

    target = 2 * N + 1
    last = -1
    stalled_at = time()
    while progress[] < target
        p = progress[]
        if p != last
            last = p
            stalled_at = time()
        elseif time() - stalled_at > STALL
            verdict = probe(req, progress, "ping-pong stalled at $(p)/$(target)")
            grpc_cancel(req)
            try
                wait(worker)
            catch
            end
            return (:wedged, verdict, p)
        end
        sleep(0.005)
    end

    try
        wait(worker)
    catch
    end
    return (:ok, :none, progress[])
end

# ---------------------------------------------------------------- driver

function main()
    mode = length(ARGS) >= 1 ? Symbol(ARGS[1]) : :eos
    iters = length(ARGS) >= 2 ? parse(Int, ARGS[2]) : 200
    N = length(ARGS) >= 3 ? parse(Int, ARGS[3]) : 200

    @printf(
        "julia %s, libcurl %s, threads=%d, mode=%s, iters=%d, N=%d, poke=%s, stall=%.1fs\n",
        VERSION, gRPCClient.CURL_VERSION, Threads.nthreads(), mode, iters, N, POKE, STALL
    )

    client = if mode === :eos || mode === :pingpong
        TestService_TestBidirectionalStreamRPC_Client(HOST, PORT; deadline = Inf)
    else
        error("unknown mode $mode")
    end

    poker = if POKE
        Threads.@spawn while true
            drive_multi()
            yield()
        end
    else
        nothing
    end

    wedged = 0
    verdicts = Dict{Symbol, Int}()
    t0 = time()

    for i in 1:iters
        status, verdict, extra = mode === :eos ? round_eos(client, N) : round_pingpong(client, N)
        if status === :wedged
            wedged += 1
            verdicts[verdict] = get(verdicts, verdict, 0) + 1
            @printf("iteration %d/%d WEDGED (recovery: %s)\n", i, iters, verdict)
        elseif i % 25 == 0
            @printf("iteration %d/%d ok (%.3fs elapsed, %d wedged so far)\n", i, iters, time() - t0, wedged)
        end
    end

    @printf("\n%d/%d rounds wedged in %.1fs", wedged, iters, time() - t0)
    isempty(verdicts) || print(", recovery: ", verdicts)
    println()

    poker === nothing || (@async Base.throwto(poker, InterruptException()))
    return wedged
end

exit(main() == 0 ? 0 : 1)

csvance · 2026-07-31T18:14:36Z

@s-celles I think I figured out a way to handle this without polling / any blocking operations inside of the critical section. I'm testing out the changes now to make sure they do not cause any performance regressions. Do you mind If I take it the rest of the way and add you as a co-author on the PR that gets merged?

s-celles · 2026-07-31T18:51:24Z

Hi @csvance

Glad you find a clean way of fixing this. I'm currently busy on Giac.jl and LibPARI.jl so I couldn't answer.
I (and my AI) also did some more work on gRPCServer.jl (I know you have your own implementation! We will see how to manage that later but maybe you should have a look at it (there is now support of 3 HTTP/2 backends).
Yes you can add me as co-author but also Assisted-by: AI (because that's reality, I'm fine with AI disclosure but I don't think mentioning model is necessary except if we want all our repositories turning into AI-advertisement museum.

Sébastien

csvance · 2026-07-31T19:01:17Z

@s-celles on the gRPCServer part I think there might be more room to collaborate. While my package works at nearly a production level thanks mostly to HTTP.jl and Reseau.jl, I don't have a huge amount of time to spend on it. The trouble for me with the one you are working on is I'm not sure whether its to the point where things mostly work and rapid iteration has slowed down. Since AI makes software development so much faster these days, two people can easily be massively changing the same things and end up with lots of conflicts. When it's to the point where I can build on it, then I can also contribute small things / improvements in different places, like you are doing here with this bug report.

Maybe we could look at contributing my working HTTP.jl based backend to your package at some point. From there, I have a working backend regardless of what is going on with the other backends.

csvance · 2026-08-10T13:38:59Z

Closing since your changes were merged in and credited in 1.1.0.

s-celles force-pushed the fix-68-streaming-julia-1.10 branch from e075d38 to 6c32b4c Compare July 30, 2026 04:54

csvance mentioned this pull request Jul 31, 2026

Fix streaming pause handoff race + workaround Julia 1.10 streaming issues #129

Merged

csvance closed this Aug 10, 2026

Navigation Menu

Uh oh!

Fix streaming hangs and enable streaming on Julia 1.10 - #127

Fix streaming hangs and enable streaming on Julia 1.10#127
s-celles wants to merge 1 commit into
JuliaIO:mainfrom
s-celles:fix-68-streaming-julia-1.10

s-celles commented Jul 30, 2026 •

edited

Loading

Uh oh!

codecov-commenter commented Jul 30, 2026 •

edited

Loading

Uh oh!

csvance commented Jul 30, 2026

Uh oh!

csvance commented Jul 30, 2026 •

edited

Loading

Uh oh!

csvance commented Jul 31, 2026

Uh oh!

s-celles commented Jul 31, 2026

Uh oh!

csvance commented Jul 31, 2026 •

edited

Loading

Uh oh!

csvance commented Aug 10, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Uh oh!

Conversation

s-celles commented Jul 30, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

The problem

Root cause

The fix

Notes for review

Verification

Uh oh!

codecov-commenter commented Jul 30, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

csvance commented Jul 30, 2026

Uh oh!

csvance commented Jul 30, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

csvance commented Jul 31, 2026

Uh oh!

s-celles commented Jul 31, 2026

Uh oh!

csvance commented Jul 31, 2026 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

csvance commented Aug 10, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

s-celles commented Jul 30, 2026 •

edited

Loading

codecov-commenter commented Jul 30, 2026 •

edited

Loading

csvance commented Jul 30, 2026 •

edited

Loading

csvance commented Jul 31, 2026 •

edited

Loading