Constant deadlocks with any async usage

[csvance]

Constantly getting random deadlocks with this error message when I use the non blocking client in an async context:

┌ Error: curl_multi_socket_action: 8
└ @ Downloads.Curl ~/.virtualenvs/full-solution/julia_env/pyjuliapkg/install/share/julia/stdlib/v1.11/Downloads/src/Curl/utils.jl:57

Here is how I am doing async:

@sync
    B = Int64(N ÷ max_batch_size)
    if mod(N, max_batch_size) != 0
        B+= 1
    end
    
    channel = Channel(B)
    
    i = 1
    while i <= N
        j = min(N, (i + max_batch_size) - 1)
        input__0 = NetworkInput("INPUT__0", Float32.(C[i:j, :, :]))
        let i = i, j = j
            TritonInferAsync(client, model, [input__0], (r) -> put!(channel, (r, [i, j])))
        end
        i = i + max_batch_size
    end
    
    for b in 1:B
        response, ij = take!(channel)
        let response=response; ij=ij
            ...
        end
    end
end

julia> versioninfo()
Julia Version 1.11.5
Commit 760b2e5b739 (2025-04-14 06:53 UTC)
Build Info:
  Official https://julialang.org/ release
Platform Info:
  OS: Linux (x86_64-linux-gnu)
  CPU: 128 × AMD EPYC 7773X 64-Core Processor
  WORD_SIZE: 64
  LLVM: libLLVM-16.0.6 (ORCJIT, znver3)
Threads: 1 default, 0 interactive, 1 GC (on 128 virtual cores)

[csvance]

I got the blocking client to work with async, but it often gives me the same repeated "Error: curl_multi_socket_action: 8"

@sync begin
    i = 1
    while i <= N
        j = min(N, (i + max_batch_size) - 1)
        let i = i, j = j
            @async begin
                input__0 = NetworkInput("INPUT__0", Float32.(C[i:j, :, :]))
                response, status = TritonInferSync(client, model, [input__0])
                gRPCCheck(status)
                ...
            end
        end
        i = i + max_batch_size
    end
end

It works perfectly with no concurrency, but its a ~20% performance hit for my application (neural net model inference w/ nVidia Triton Inference Server)

I wonder where the code is that is yielding callback threads to the event loop.

[csvance]

Using a new connection for every gRPC call instead of multiplexing works with async but isn't as fast as pure async obviously. The problem definitely seems to be related to multiplexing / concurrent re-use of a single connection.

For me its sufficient to maintain a connection pool, limiting the active number of tasks to the size of the pool. When a worker finishes it can release a connection back to the next worker trying to get it (maybe using channels or something).

Would be great to be able to take advantage of multiplexing in gRPC in Julia though! I may ask over at Downloads.jl to see if they have any idea.

[csvance]

So I tried to implement a connection pool re-using connections, but it still triggers a deadlock somehow. Here is what finally worked reliably and quickly for me, essentially using one connection for every request, but maintaining the connection pool using async to avoid any delay.

function TritonClient(url::String)
    return GRPCInferenceServiceBlockingClient(url)
end

function TritonClientPool(url::String, size=32)
    channel = Channel{GRPCInferenceServiceBlockingClient}(size)
    for i in 1:size
        put!(channel, TritonClient(url))
    end

    return channel
end

function TritonInfer(client_pool::Channel{GRPCInferenceServiceBlockingClient}, model_name::String, inputs::Vector{ModelInferRequest_InferInputTensor})

    request = ModelInferRequest(
        model_name=model_name,
        inputs=inputs
    );

    client = take!(client_pool)

    response = nothing
    try
        response, status = ModelInfer(client, request);
        gRPCCheck(status)
    finally
        @async put!(client_pool, TritonClient(client.channel.baseurl))
    end

    return response
end

[csvance]

I'm going to dive deep into this over the weekend. Julia is SO close to working amazingly for my project.

[csvance]

Stack trace:

Unhandled Task ERROR: "curl_multi_socket_action: 8"
Stacktrace:
 [1] macro expansion
   @ ~/.julia/juliaup/julia-1.11.5+0.x64.linux.gnu/share/julia/stdlib/v1.11/Downloads/src/Curl/utils.jl:57 [inlined]
 [2] (::Downloads.Curl.var"#39#44"{UInt8, Int32, FileWatching.FDWatcher, Downloads.Curl.Multi})()
   @ Downloads.Curl ~/.julia/juliaup/julia-1.11.5+0.x64.linux.gnu/share/julia/stdlib/v1.11/Downloads/src/Curl/Multi.jl:190
 [3] lock(f::Downloads.Curl.var"#39#44"{UInt8, Int32, FileWatching.FDWatcher, Downloads.Curl.Multi}, l::ReentrantLock)
   @ Base ./lock.jl:232
 [4] (::Downloads.Curl.var"#38#43"{Int32, FileWatching.FDWatcher, Downloads.Curl.Multi})()
   @ Downloads.Curl ~/.julia/juliaup/julia-1.11.5+0.x64.linux.gnu/share/julia/stdlib/v1.11/Downloads/src/Curl/Multi.jl:188

In socket_callback in Downloads/Curl/Multi.jl:

        if action in (CURL_POLL_IN, CURL_POLL_OUT, CURL_POLL_INOUT)
            readable = action in (CURL_POLL_IN,  CURL_POLL_INOUT)
            writable = action in (CURL_POLL_OUT, CURL_POLL_INOUT)
            watcher = FDWatcher(OS_HANDLE(sock), readable, writable)
            preserve_handle(watcher)
            watcher_p = pointer_from_objref(watcher)
            @check curl_multi_assign(multi.handle, sock, watcher_p)
            task = @async while watcher.readable || watcher.writable # isopen(watcher)
                events = try
                    wait(watcher)
                catch err
                    err isa EOFError && return
                    err isa Base.IOError || rethrow()
                    FileWatching.FDEvent()
                end
                flags = CURL_CSELECT_IN  * isreadable(events) +
                        CURL_CSELECT_OUT * iswritable(events) +
                        CURL_CSELECT_ERR * (events.disconnect || events.timedout)
                lock(multi.lock) do
                    watcher.readable || watcher.writable || return # !isopen
                    @check curl_multi_socket_action(multi.handle, sock, flags)
                    check_multi_info(multi)
                end
            end
            @isdefined(errormonitor) && errormonitor(task)
        else

The error code is

CURLM_RECURSIVE_API_CALL (8)

An API function was called from inside a callback.

Normally this should be running on an async task that gets spun off during the callback. But that is clearly not the case here.... or maybe the callback doesn't finish before the task starts?

Digging through libcurl source code now looking for a possible race condition.

[csvance]

So I think that since the curl callback is in a separate thread, there can be a race condition where the callback has not returned before the async task starts. Its still kindof hackish, but instead of doing task = @async using task = @task and calling schedule(task) in a try/finally block at the end of the function.

The other potential thing to watch out for is if Julia hijacks the thread and prevents it from returning. I think this is more likely what is happening, something within a callback is either silently failing or yielding to the event loop.

[csvance]

Looks like a race condition! Retrying curl_multi_socket_action yielding back to the event loop seems to completely eliminate the error! I'm going to run this overnight and hopefully it will still be going in the morning.

                lock(multi.lock) do
                    watcher.readable || watcher.writable || return # !isopen
                    while true
                      status = curl_multi_socket_action(multi.handle, sock, flags)
                      if status != 8
                        break
                      end
                      # Yield back to the event loop to let Julia return control back to curl, finish the callback, and we can break out of the loop
                      sleep(0.0)
                    end
                    check_multi_info(multi)
                end   

What I don't understand is why yielding back to the event loop for Julia to do more things would result in curl finishing its callback.