OhMyThreads.jl

When the workload per element is non-uniform (some elements take longer to compute than others), you can use different scheduling strategies to improve performance:

• :dynamic (DynamicScheduler): The default scheduler. It divides the workload into tasks that are dynamically distributed among threads. This provides load balancing but may increase allocations.
• :static (StaticScheduler): Does not provide load balancing. It divides the work upfront, which is faster in terms of overhead/allocations but can lead to thread idling if some tasks are much heavier than others.

You can fine-tune the granularity of the dynamic scheduler by increasing the ntasks parameter, which creates more tasks with smaller workloads.

using OhMyThreads

# Using dynamic scheduling with fine-grained tasks
compute_juliaset_parallel!(img; ntasks=N, scheduler=:dynamic)

# Using static scheduling (no load balancing)
compute_juliaset_parallel!(img; scheduler=:static)

False Sharing is a performance issue that occurs when multiple CPU cores operate on independent data elements that reside within the same cache line. Even if the data is logically independent (no race conditions), the hardware's cache coherency mechanism forces costly synchronization between cores, leading to significant slowdowns.

How to avoid False Sharing

1. Use Task-Local Accumulators: Instead of frequently updating a shared array (e.g., psums[c] += value), perform the computation using a local variable within the task and write the final result to the shared state only once at the end of the task.
2. Avoid Shared State Modification: Use patterns like map with @spawn to return values from tasks rather than modifying a pre-allocated shared buffer.
3. Use High-Level Abstractions: Use OhMyThreads.treduce to handle parallel reductions safely and efficiently without manual state management.

Another safe pattern is to put the input elements into a Channel and spawn a limited number of tasks (e.g., nthreads()) that each maintain their own long-lived task-local buffer.

Pattern:

1. Create a Channel containing the indices or elements of the input.
2. Use tmapreduce to spawn a fixed number of tasks.
3. Each task creates one local buffer and iterates over the Channel using map(chnl) do i ... end.

Note: This approach is non-deterministic in terms of output order, so you may need to sort results if order matters.

Optimization: For large inputs, use OhMyThreads.ChannelLike(range) instead of a standard Channel to avoid the overhead of copying elements into the channel.

using OhMyThreads: tmapreduce

function matmulsums_perthread_channel_flipped(As, Bs; ntasks = nthreads())
    N = size(first(As), 1)
    # Put input indices into a channel
    chnl = Channel{Int}(length(As); spawn = true) do chnl
        for i in 1:length(As)
            put!(chnl, i)
        end
    end

    # Spawn a limited number of tasks
    tmapreduce(vcat, 1:ntasks; chunking=false) do _ 
        local C = Matrix{Float64}(undef, N, N) # One buffer per task
        map(chnl) do i 
            A = As[i]
            B = Bs[i]
            mul!(C, A, B)
            sum(C)
        end
    end
end

If you must use threadid() to access per-thread buffers, you can mitigate the risk of task migration by opting out of dynamic scheduling. Setting scheduler = :static (or scheduler = StaticScheduler()) ensures tasks are statically assigned to threads and remain "sticky" to them.

Warning: This approach is still not recommended because:

1. It is fragile; removing the :static scheduler during refactoring will cause race conditions.
2. It is non-composable; calling other multithreaded functions inside a static region can lead to thread oversubscription.
3. It can be memory inefficient if maxthreadid() is significantly larger than the actual number of threads used.

using Base.Threads: threadid

function matmulsums_perthread_static(As, Bs)
    N = size(first(As), 1)
    # Use maxthreadid to avoid out-of-bounds
    num_to_store() = isdefined(Threads, :maxthreadid) ? Threads.maxthreadid() : Threads.nthreads()
    Cs = [Matrix{Float64}(undef, N, N) for _ in 1:num_to_store()]

    # scheduler = :static is REQUIRED for correctness here
    tmap(As, Bs; scheduler = :static) do A, B
        C = Cs[threadid()]
        mul!(C, A, B)
        sum(C)
    end
end

The recommended safe way to manage a pool of reusable buffers for parallel tasks is to use a Channel. This avoids the risks of task migration and threadid() indexing.

Pattern:

1. Pre-allocate a fixed number of buffers (e.g., nthreads()).
2. Put them into a Channel.
3. Inside the parallel loop (tmap), take! a buffer from the channel, use it, and then put! it back when finished.

This approach is highly efficient and works well with dynamic load balancing (when ntasks > nthreads()).

function matmulsums_perthread_channel(As, Bs; nbuffers = nthreads(), kwargs...)
    N = size(first(As), 1)
    chnl = Channel{Matrix{Float64}}(nbuffers)
    foreach(1:nbuffers) do _
        put!(chnl, Matrix{Float64}(undef, N, N))
    end

    tmap(As, Bs; kwargs...) do A, B
        C = take!(chnl) # Safely acquire buffer
        mul!(C, A, B)
        result = sum(C)
        put!(chnl, C)   # Safely return buffer
        result
    end
end