# How to achieve Non blocking IO with Julia

**URL:** <https://discourse.julialang.org/t/how-to-achieve-non-blocking-io-with-julia/117503>\
**Category:** General Usage\
**Tags:** package, multithreading, io, async\
**Created:** [July 26, 2024, 8:07am UTC](https://discourse.julialang.org/t/how-to-achieve-non-blocking-io-with-julia/117503 "2024-07-26T08:07:29Z")\
**Posts on this page:** 4\
**Page:** 1

<div class="post-metadata">

**Author:** ![Chamika\_Kasun](https://sea2.discourse-cdn.com/julialang/user_avatar/discourse.julialang.org/chamika_kasun/32/202926_2.png) [@Chamika\_Kasun](https://discourse.julialang.org/u/Chamika_Kasun)\
**Post date:** [July 26, 2024, 8:07am UTC](https://discourse.julialang.org/t/how-to-achieve-non-blocking-io-with-julia/117503/1 "2024-07-26T08:07:29Z")

</div>

I have a function that is executed through a Julia thread. In this function, there is a execute\_psql\_string\_await function that performs a PostgreSQL stored procedure call using the LibPQ library. Currently, while this query is being executed, my thread is blocked. However, I do not want this behavior. I want a non-blocking way to execute this query, meaning that I trigger the stored procedure and release the thread without waiting for the result. Any thoughts are welcome.

```julia

function sproc_caller(table_name::String, sql_string::String, request::Dict, request_id::String, sproc_name::String)
    try
        @info "Start sproc_caller table_name $(table_name) execute_psql_string, on thread $(Threads.threadid())"
        @time execute_psql_string_await(sql_string)
        @info "End sproc_caller table_name $(table_name) execute_psql_string, on thread $(Threads.threadid())"
    catch err
        @error "Error in Stored Procedure on thread $(Threads.threadid()), Error : $(err)" exception=(err, catch_backtrace())
end

```

---

<div class="post-metadata">

**Author:** ![abraemer](https://sea2.discourse-cdn.com/julialang/user_avatar/discourse.julialang.org/abraemer/32/51403_2.png) [@abraemer](https://discourse.julialang.org/u/abraemer)\
**Post date:** [July 26, 2024, 9:42am UTC](https://discourse.julialang.org/t/how-to-achieve-non-blocking-io-with-julia/117503/2 "2024-07-26T09:42:01Z")

</div>

> [@Chamika\_Kasun](#):
>
> Currently, while this query is being executed, my thread is blocked

Are you sure the thread is blocked? Julia distinguishes between `Task` and threads. Usually when a `Task` is waiting for some IO then the thread can switch to another `Task`. Of course if the wait is a spin lock/busy loop or similar then this does not work but this is then a fundamental limitation of whatever library you are using and there is no way around this. Usually there is a reason though why an API is designed that way.

Or do you just wish for your `Task` to progress instead of waiting on the return of `execute_psql_string_await`? In that case you can simply spawn that call in another `Task` such that is executed concurrently, i.e. do `Threads.@spawn execute_psql_string_await(sql_string)`. Another possibility could to look for a different method since the name `execute_psql_string_AWAIT` suggests that there might be a non-blocking variant perhaps called `execute_psql_string` or similar.

---

<div class="post-metadata">

**Author:** ![Chamika\_Kasun](https://sea2.discourse-cdn.com/julialang/user_avatar/discourse.julialang.org/chamika_kasun/32/202926_2.png) [@Chamika\_Kasun](https://discourse.julialang.org/u/Chamika_Kasun)\
**Post date:** [July 28, 2024, 11:23pm UTC](https://discourse.julialang.org/t/how-to-achieve-non-blocking-io-with-julia/117503/3 "2024-07-28T23:23:59Z")

</div>

> [@Chamika\_Kasun](#):
>
> `execute_psql_string_await(sql_string)`

Hi @abraemer

I’ve created a worker module, in which I’ve exposed `Workers.@async` macro to pass async to a pool of tasks; which effectively uses a thread available to execute any async task.

```julia
module Workers

const WORK_QUEUE = Channel{Task}(1000)

"Get the current length of the provided Channel `ch`, with the `WORK_QUEUE` as default."
get_queue_tasks(ch::Channel=WORK_QUEUE) = length(ch.data)

"Check if the provided Channel `ch` is at or over capacity, with the `WORK_QUEUE` as default."
is_queue_full(ch::Channel=WORK_QUEUE) = get_queue_tasks(ch) ≥ ch.sz_max

"Get the percentage fullness of Channel `ch` when compared to its maximum capacity, with the `WORK_QUEUE` as default."
function get_percent_queue(ch::Channel=WORK_QUEUE)
    is_queue_full(ch) && return 100.0
    x = get_queue_tasks(ch)
    x == 0 && return 0.0
    return round(x / ch.sz_max * 100.0, digits=2)
end

"""
    @async expr

Adds `expr` as a new task to `Workers.WORK_QUEUE`, wrapped in a try-catch
to display any error messages.

The tasks in `Workers.WORK_QUEUE` are run on any available threads except for
thread 1, which is reserved for the HTTP handler (unless only one thread is
in use).
"""
macro async(expr)
    esc(quote
        tsk = @task begin
            try
                $expr
            catch err
                @error "Async function failed: $(typeof(err))" exception=(err, catch_backtrace())
                return err
            end
        end
        tsk.sticky = VERSION >= v"1.7" # Prevent task from switching threads
        tsk.storage = current_task().storage
        put!(Workers.WORK_QUEUE, tsk)
        tsk
    end)
end

"""
    init()

If `Threads.nthreads()` is greater than 1, initialise async function on each
thread apart from thread 1, which is left for the HTTP handler.

If only 1 thread is available,
`Workers.@async` acts in the same way as `Base.@async`.
"""
function init()
    @info "DataExporterService is starting with $(Threads.nthreads()) threads."
    tids = Threads.nthreads() == 1 ? (1:1) : 2:Threads.nthreads()
    Threads.@threads :static for _ in 1:Threads.nthreads()
        if Threads.threadid() in tids
            Base.@async begin
                for task in WORK_QUEUE
                    # Force schedule task onto this worker thread
                    _ = ccall(:jl_set_task_tid, Cint, (Any, Cint), task, Threads.threadid()-1)
                    schedule(task)
                    wait(task)
                end
            end
        end
    end
    return
end

end # module

```

I start the julia application with 4 threads and I use main thread for the HTTP service, and I expect remaining 3 threads to be available for task execution, using the above pool, as and when I submit the tasks it starts executing;

Here’s how I execute those tasks:

```julia
for sproc in SPROC_REF
  sql_string = "CALL $(sproc[1]);"
  Workers.@async sproc_caller(sproc[2], sql_string, request, request_id, sproc_name, x_request_id=x_request_id)
end

```

And as you can see in the above `sproc_caller` function I’ve logs attached; My requirement is that; Once I submit that task to the worker pool; I want that task to be expected in the background and release the thread so that I can use that thread to submit more tasks to the worker pool. But here the issue is; `execute_psql_string_await` I have in the wait for the SQL command to return a results, I don’t want that; What I want is execute that and release the Julia thread for me to use for other tasks.

`execute_psql_string_await` is a wrapper function around LibPQ `async_execute` function as follows:

```julia
function execute_psql_string_await(pg_params::PostgreSQLConnectionParams, psql_string::AbstractString, wait::Bool, options=Dict{String, String}(), nowarn=false)
    !nowarn && @warn "Executing a PSQL query from a string"
    @debug psql_string
    with_postgresql(pg_params, options) do psql
        result = async_execute(psql, psql_string)
        if wait
            return fetch(result)
        else
            return result
        end
    end
end
execute_psql_string_await(psql_string::AbstractString) = execute_psql_string_await(global_psql_config(), psql_string, false)

```

Further to above; I also tried using `Workers.@async` with `execute_psql_string_await(sql_string)` function; Yet the thread is handing there until that query get executed. Not sure what other options I’ve for this.

Another issue with the options you suggested is that; I can’t directly use @spawn because it will use the main thread for other async tasks that I’m using; I need to keep the main thread free all the time as it should accept the incoming requests through the HTTP server.

---

<div class="post-metadata">

**Author:** ![abraemer](https://sea2.discourse-cdn.com/julialang/user_avatar/discourse.julialang.org/abraemer/32/51403_2.png) [@abraemer](https://discourse.julialang.org/u/abraemer)\
**Post date:** [July 29, 2024, 6:48am UTC](https://discourse.julialang.org/t/how-to-achieve-non-blocking-io-with-julia/117503/4 "2024-07-29T06:48:23Z")

</div>

So I don’t think you actually await the result of the call. The code you posted shows, that `fetch(result)` is called only if you pass `wait=true` which you don’t. Is is possible that you just see the overhead of using the `execute_psql_string_await`function? I found this issue which could be related:

> <https://github.com/JuliaDatabases/LibPQ.jl/issues/281>
>
> \<img width="489" alt="image" src="https://github.com/iamed2/LibPQ.jl/assets/8681…967/8e24fc7c-447f-43ba-b968-076cd961028a"\>
> 
> According to the Profile.jl, that's mostly completely spent at file watching (which is very sad and probably a FileWatching.jl issue)

Also I think you forgot to include the logs.

Anyways I think there might be a much easier way of handling this whole situation since the advent of [threadpools with 1.10](https://docs.julialang.org/en/v1/manual/multi-threading/#man-threadpools). Simply start your julia with `--threads 3,1` which gives you 3 `default` threads and 1 `interactive` thread with the main thread being in the `interactive` pool. Then you replace the calls to your `Workers.@async foo()` with `Threads.@spawn :default foo()` to spawn the task on the other threadpool. You don’t need any additional worker task or queue. This will all be handled by Julia’s scheduler. You can of course track how many `Task`s are currently schedule by e.g. using a global variable with an [atomic integer](https://docs.julialang.org/en/v1/manual/multi-threading/#Atomic-Operations) that you increment/decrement in the tasks. This should also solve any issues with `execute_psql_string_await` since even if this call should block, then Julia can simply schedule another Task on that thread.