Multi-threaded evaluations over a Channel in julia

user_2788173 0
Viewed 308

I woud like to simultaneously use a Channel to produce values at which a function must be evaluated, and parallelise the evaluation of this function using Threads.@threads. In the absence of any multi-threading, my MWE is:

#####
# Produces input values
function producer(c::Channel)
    nb_inputs = rand(1:10)
    for n=1:nb_inputs
        println("Generating | ",n)
        put!(c,n)
    end
end
#####
# Expensive function
function f(n::Int64)
    println("Running | ",n)
    sleep(2.0)
end
#####
# Evaluates the function over all the input values
function test()
    for n in Channel(producer)
        f(n)
    end
end
#####
test()

Here, producer(c) returns input values; f(n) is an expensive function (whose evaluations are all independent from one another); and test() evaluates the function over all the input values returned by the producer. Now, I would like to run the for loop in test() in parallel over multiple threads. Naively, I tried

Threads.@threads for n in Channel(producer)

Yet, this fails because the number of elements of the Channel is initially unknown. How could I circumvent this issue?

julia channel multithreading
2 Answers

I think, from your MWE, that you do not know how many times you will call producer() until runtime? If so, you can index your loops on that value at runtime:

const nputs = [Threads.nthreads()]  # or whatever you need it to be > 0

# Produces input values
function producer(c::Channel)
    for n in 1:4
        println("Generating | ", n)
        put!(c,n)
    end
end
#####
# Expensive function
function f(n::Int64)
    println("Running | ", n)
    sleep(2.0)
end
#####
# Evaluates the function over all the input values
function test()
    @Threads.threads for n in 1:first(nputs)
        Channel(producer, first(nputs) * 4)
        f(n)
    end
end

#####
test()
user_42828470

Iterating over a Channel is thread safe: you can just use an outer @threads loop.

using Base.Threads

## ------------------------------------------------------------
# Produces input values
function producer(c::Channel)
    nb_inputs = rand(1:10)
    for n=1:nb_inputs
        println("Generating | ", n, " at: ", threadid())
        put!(c,n)
    end
end

## ------------------------------------------------------------
# Expensive function
function f(n::Int64)
    println("Running | ", n, " at: ", threadid())
    sleep(2.0)
end

## ------------------------------------------------------------
# Evaluates the function over all the input values
function test()
    Ch = Channel(producer)
    @threads for _ in 1:nthreads()
        for n in Ch
            f(n)
        end
    end
end

## ------------------------------------------------------------
test()

## ------------------------------------------------------------
# Or in julia > v1.6
function test()
    Ch = Channel(producer)
    Threads.foreach(Ch) do n
        f(n)
    end
end

## ------------------------------------------------------------
test()
user_104807580
Related