# Pipeline of async tasks with channels

**URL:** <https://discourse.julialang.org/t/pipeline-of-async-tasks-with-channels/127810>\
**Category:** General Usage\
**Created:** [April 7, 2025, 4:59pm UTC](https://discourse.julialang.org/t/pipeline-of-async-tasks-with-channels/127810 "2025-04-07T16:59:14Z")\
**Posts on this page:** 4\
**Page:** 1

<div class="post-metadata">

**Author:** ![rfourquet](https://sea2.discourse-cdn.com/julialang/user_avatar/discourse.julialang.org/rfourquet/32/3610_2.png) [@rfourquet](https://discourse.julialang.org/u/rfourquet)\
**Post date:** [April 7, 2025, 4:59pm UTC](https://discourse.julialang.org/t/pipeline-of-async-tasks-with-channels/127810/1 "2025-04-07T16:59:14Z")

</div>

I want to do some work on a list, and each step of the “work” depends on the previous one. Some steps can be IO-bound while some other can be CPU-bound. I want for example to organize as follows:

- do step 1 in parallel with at most 5 threads; put results in a channel `items1`
- do step 2 in parallel with at most 12 threads; put results in a channel `items2`

With channels, the following snippet is what works for me. But it’s verbose; for example I use channels to simulate thread pools with a specific number of tasks. I wonder whether there is something cleaner. I tried with ThreadPools.jl and OhMyThreads.jl without success, but I don’t know these packages well.

With channels:

```julia
function work_channels(nn)
    work1 = Channel{Nothing}(5)
    for _=1:5; put!(work1, nothing); end
    work2 = Channel{Nothing}(12)
    for _=1:12; put!(work2, nothing); end

    items1 = Channel(Inf)
    items2 = Channel(Inf)

    @sync begin
        @async begin
            @sync for i = 1:nn
                take!(work1)
                Threads.@spawn begin
                    sleep(0.2) # fake work
                    put!(items1, -i)
                    put!(work1, nothing)
                end
            end
            close(work1)
            close(items1)
        end

        @async begin
            @sync for x in items1
                take!(work2)
                Threads.@spawn begin
                    sleep(0.1) # fake work
                    put!(items2, x^2)
                    put!(work2, nothing)
                end
            end
            close(work2)
            close(items2)
        end
    end
    items2
end

```

With ThreadPools.jl, I do not manage to have step 2 start before step 1 is fully done (as `@info` reveals):

```julia
using ThreadPools: twith, QueuePool, tforeach

function work_queuepools(nn)
    items1 = Channel(Inf)
    items2 = Channel(Inf)

    @sync begin
        Threads.@spawn begin
            twith(QueuePool(2, 5)) do pool
                tforeach(pool, 1:nn) do i
                    sleep(0.2)
                    @info "put 1 $(-i)"
                    put!(items1, -i)
                end
            end
            close(items1)
        end

        Threads.@spawn begin
            twith(QueuePool(2, 12)) do pool
                tforeach(pool, items1) do x
                    sleep(0.1)
                    @info "put 2 $(x^2)"
                    put!(items2, x^2)
                end
            end
            close(items2)
        end
    end
    items2
end

```

With OhMyThreads.jl, I don’t seem to be able to pass a `Channel` as the iterator to functions like `tforeach`.

---

<div class="post-metadata">

**Author:** ![Sukera](https://avatars.discourse-cdn.com/v4/letter/s/ce7236/32.png) [@Sukera](https://discourse.julialang.org/u/Sukera)\
**Post date:** [April 7, 2025, 5:54pm UTC](https://discourse.julialang.org/t/pipeline-of-async-tasks-with-channels/127810/2 "2025-04-07T17:54:43Z")

</div>

This is _so very close_ to being really convenient to do with just Base ☹ If we’d have the ability to add custom sized threadpools dynamically, you could use `Threads.@threads :greedy for`…

Alternatively, you could take a look at how `:greedy` is implemented and add a user supplied upper limit on the spawned tasks to `:greedy`?

---

<div class="post-metadata">

**Author:** ![sgaure](https://sea2.discourse-cdn.com/julialang/user_avatar/discourse.julialang.org/sgaure/32/14779_2.png) [@sgaure](https://discourse.julialang.org/u/sgaure)\
**Post date:** [April 7, 2025, 6:08pm UTC](https://discourse.julialang.org/t/pipeline-of-async-tasks-with-channels/127810/3 "2025-04-07T18:08:22Z")

</div>

It’s doable with `Threads.@threads` and `Base.Semaphore`:

```julia

function wchan(nn)

    items1 = Channel(Inf)
    items2 = Channel(Inf)

    # step 1 should have max 5 concurrent
    sem1 = Base.Semaphore(5)
    step1 = @async begin
        Threads.@threads for i in 1:nn
            Base.acquire(sem1)
            sleep(0.2)
            @ccall printf("step1: %d\n"::Cstring; i::Cint)::Cint
            put!(items1, i)
            Base.release(sem1)
        end
        close(items1)
    end

    sem2 = Base.Semaphore(12)
    step2 = @async begin
        Threads.@threads for i in 1:nn
            Base.acquire(sem2)
            x = take!(items1)
            sleep(0.2)
            @ccall printf("step2: %d\n"::Cstring; x::Cint)::Cint
            put!(items2, x^2)
            Base.release(sem2)
        end
        close(items2)
    end
    wait.((step1, step2))

    items2
end

nn = 50
ch = wchan(nn)
result = sort([take!(ch) for _ in 1:nn])

```

I use `@ccall printf` since it’s immediate, not waiting for some interactive thread to write.

---

<div class="post-metadata">

**Author:** ![bertschi](https://sea2.discourse-cdn.com/julialang/user_avatar/discourse.julialang.org/bertschi/32/33462_2.png) [@bertschi](https://discourse.julialang.org/u/bertschi)\
**Post date:** [April 7, 2025, 7:53pm UTC](https://discourse.julialang.org/t/pipeline-of-async-tasks-with-channels/127810/4 "2025-04-07T19:53:37Z")

</div>

Does transducers do what you want?

```julia
using Transducers: Map, append_unordered!

function work_channels(inch)
    items1 = Channel()
    items2 = Channel()
    @async try
        append_unordered!(items1, Map(i -> begin sleep(0.1); @info "i = $i"; i - 1 end), inch;
                          ntasks = 5)
    finally close(items1) end
           
    @async try
        append_unordered!(items2, Map(x -> begin sleep(0.2); @info "x = $x"; x^2 end), items1;
                          ntasks = 12)
    finally close(items2) end
    items2
end

Channel(Map(identity), 1:nn) |> work_channels |> collect

```

Otherwise, you might need/want to implement some helper functions mimicking `map, filter` and friends for channels. For inspiration check [clojure.core.async](https://clojuredocs.org/clojure.core.async) …