# Multithreaded compute farm

**URL:** <https://discourse.julialang.org/t/multithreaded-compute-farm/65047>\
**Category:** Performance\
**Tags:** review, multithreading\
**Created:** [July 21, 2021, 4:11pm UTC](https://discourse.julialang.org/t/multithreaded-compute-farm/65047 "2021-07-21T16:11:55Z")\
**Posts on this page:** 1\
**Showing post:** 3

<div class="post-metadata">

**Author:** ![tkf](https://sea2.discourse-cdn.com/julialang/user_avatar/discourse.julialang.org/tkf/32/17635_2.png) [@tkf](https://discourse.julialang.org/u/tkf)\
**Post date:** [July 22, 2021, 11:05pm UTC](https://discourse.julialang.org/t/multithreaded-compute-farm/65047/3 "2021-07-22T23:05:34Z")

</div>

I’m curious where you picked up the term “compute farm” from. In the [parallel skeleton](https://link.springer.com/referenceworkentry/10.1007%2F978-0-387-09766-4_24) literature, _(task) farm_ skeleton refers to something different (FYI, my [Concurrency patterns for controlled parallelisms](https://discourse.julialang.org/t/tutorial-concurrency-patterns-for-controlled-parallelisms/62651) tutorial includes a simple implementation of task farm in Julia.) I think what you are referring to is close to the _feedback_ skeleton (although I don’t know how standard this nomenclature is):

> ![image](https://global.discourse-cdn.com/julialang/original/3X/1/3/1340a8a061464e365578d8473655d1d4ef6c84dc.png)
> 
> — Castro, David, Kevin Hammond, and Susmit Sarkar. 2016. “Farms, Pipes, Streams and Reforestation: Reasoning about Structured Parallel Processes Using Types and Hylomorphisms.” ACM SIGPLAN Notices 51 (9): 4–17. [Farms, pipes, streams and reforestation: reasoning about structured parallel processes using types and hylomorphisms | ACM SIGPLAN Notices](https://doi.org/10.1145/3022670.2951920).

I agree with @jpsamaroo’s comment on the bug in using atomic operation on `busy`. To avoid a bug like this, it’d be better to use a pre-existing abstraction. This is what C++ calls [`latch`](https://en.cppreference.com/w/cpp/thread/latch) and [`CountDownLatch`](https://docs.oracle.com/javase/7/docs/api/java/util/concurrent/CountDownLatch.html) (although I’ve seen [some dispute](https://news.ycombinator.com/item?id=27217877) on the naming). Anyway, I think you can wrap it into a abstraction called latch or barrier, which makes testing and reviewing easier. FYI, my [SyncBarriers.jl](https://github.com/tkf/SyncBarriers.jl) library has a barrier implementations that can be used for this. A _fuzzy barrier_ is useful for this purpose since (presumably) the workers don’t have to wait for other workers.

Having said that, I also agree with:

> [@jpsamaroo](#):
>
> > [@nabla](#):
> >
> > `sleep(y) # give some time to the queues to start`
> 
> If you have to do this, it’s usually an indication that you’re doing something wrong

One solution may be to use a simpler solution like below (untested), which doesn’t require the barrier dance at all:

```julia
function parallel_feedback(f, inits, T = eltype(inits); ntasks=Threads.nthreads())
    worklist = Channel{T}() do worklist
        for x in inits
            put!(worklist, x)
        end
    end
    @sync for _ in 1:ntasks
        Threads.@spawn try
            f(worklist)
        finally
            close(worklist)
        end
    end
end

```

It can be used as

```julia
parallel_feedback(rand(10)) do worklist
    for x in worklist
        for k in 0:20 # simulate Poisson distribution
            t -= λ^k * exp(-λ) / factorial(k)
            t < 0 && break
            y = rand()
            put!(worklist, y)
            # Note: omitting actual output handling
        end
    end
end

```

This has an extra benefit that you don’t need to allocate the temporary `next` vector (cutting GC interaction is important for scaling up threaded program ATM). It also makes the work available right away for other workers (i.e., the loop inside `f` does not have to complete for “submitting” the work). Furthermore, it is now fairly easy to swap the `worklist` implementation with a list of _work-stealing deque_, which is very appealing here since the single worker case degenerate into the usual (single-threaded) worklist-based recursion. This makes this construct robustly perform well when nested inside other parallel programs.

~~Also, if you want to bound the channel size, you can create a simple wrapper type like

```julia
struct BoundedChannel{T}
    ch::Channel{T}
end

function Base.put!(ch::BoundedChannel, x)
    Base.n_avail(ch.ch)<size || error("BoundedChannel: channel is full")
    put!(ch.ch, x)
end

Base.iterate(ch::BoundedChannel, _ = nothing) = iterate(ch.ch)

```

~~

~~and use it as `f(BoundedChannel(worklist))` instead of `f(worklist)`.  
~~ (Edit: `n_avail` is an internal function)

Some more random comments:

- `ChannelFarm{T}(f::Function, ...)`: every object in Julia is potentially callable. I don’t think bounding `::Function` is a good idea. See [Add mergewith[!](combine, dicts...) by tkf · Pull Request #34296 · JuliaLang/julia · GitHub](https://github.com/JuliaLang/julia/pull/34296) for some discussion.
- `ChannelFarm{T}(...; nthreads=nthreads())`: The argument `nthreads` does not specify the number of threads. It specifies the number of worker tasks. I’d call it `ntasks` (or maybe `nworkers`).

---

_[View the full topic](https://discourse.julialang.org/t/multithreaded-compute-farm/65047)._