julia 多线程无法扩展以实现令人尴尬的并行作业

Question

以下代码计算从几组中获得 50 张独特卡片的平均抽奖次数。重要的是，这个问题不需要太多内存，并且在多线程模式下启动时不共享任何变量。当使用四个以上的线程启动以执行 400,000 次模拟时，与同时启动并执行 200,000 次模拟的两个进程相比，它始终需要多花一秒钟的时间。这一直困扰着我，我找不到任何解释。

这是epic_draw_multi_thread.jl 中的Julia 代码：

using Random
using Printf
import Base.Threads.@spawn

function pickone(dist)
    n = length(dist)
    i = 1
    r = rand()
    while r >= dist[i] && i<n 
        i+=1
    end
    return i
end  

function init_items(type_dist, unique_elements)
    return zeros(Int32, length(type_dist), maximum(unique_elements))
end

function draw(type_dist, unique_elements_dist)
    item_type = pickone(type_dist)
    item_number = pickone(unique_elements_dist[item_type])
    return item_type, item_number
end

function draw_unique(type_dist, unique_elements_dist, items, x)
    while sum(items .> 0) < x
        item_type, item_number = draw(type_dist, unique_elements_dist)
        items[item_type, item_number] += 1
    end
    return sum(items)
end

function average_for_unique(type_dist, unique_elements_dist, x, n, reset=true)
    println(@sprintf("Started average_for_unique on thread %d with n = %d", Threads.threadid(), n))
    items = init_items(type_dist, unique_elements)

    tot_draws = 0
    for i in 1:n
        tot_draws += draw_unique(type_dist, unique_elements_dist, items, x)
        if reset
            items .= 0
        else
            items[items.>1] -= 1
        end
    end

    println(@sprintf("Completed average_for_unique on thread %d with n = %d", Threads.threadid(), n))
    return tot_draws / n
end

function parallel_average_for_unique(type_dist, unique_elements_dist, x, n, reset=true)
    println("Started computing...")
    t = max(Threads.nthreads() - 1, 1)
    m = Int32(round(n / t))
    tasks = Array{Task}(undef, t)
    @sync for i in 1:t
        task = @spawn average_for_unique(type_dist, unique_elements_dist, x, m)
        tasks[i] = task
    end
    sum(fetch(t) for t in tasks) / t
end
    
type_dist = [0.3, 0.3, 0.2, 0.15, 0.05]
const cum_type_dist = cumsum(type_dist)

unique_elements = [21, 27, 32, 14, 10]
unique_elements_dist = [[1 / unique_elements[j] for i in 1:unique_elements[j]] for j in 1:length(unique_elements)]
const cum_unique_elements_dist = [cumsum(dist) for dist in unique_elements_dist]

str_n = ARGS[1]
n = parse(Int64, str_n)
avg = parallel_average_for_unique(cum_type_dist, cum_unique_elements_dist, 50, n)
print(avg)

这是在 shell 上发出的在两个线程上运行的命令以及输出和计时结果：

time julia --threads 3 epic_draw_multi_thread.jl 400000
Started computing...
Started average_for_unique on thread 3 with n = 200000
Started average_for_unique on thread 2 with n = 200000
Completed average_for_unique on thread 2 with n = 200000
Completed average_for_unique on thread 3 with n = 200000
70.44460749999999
real    0m14.347s
user    0m26.959s
sys     0m2.124s

这些是在 shell 上发出的命令，用于运行两个作业大小各一半的进程以及输出和计时结果：

time julia --threads 1 epic_draw_multi_thread.jl 200000 &
time julia --threads 1 epic_draw_multi_thread.jl 200000 &
Started computing...
Started computing...
Started average_for_unique on thread 1 with n = 200000
Started average_for_unique on thread 1 with n = 200000
Completed average_for_unique on thread 1 with n = 200000
Completed average_for_unique on thread 1 with n = 200000
70.434375
real    0m12.919s
user    0m12.688s
sys     0m0.300s
70.448695
real    0m12.996s
user    0m12.790s
sys     0m0.308s

无论我重复实验多少次，我总是让多线程模式变慢。笔记：

1. 我创建了并行代码来近似 PI 的值，但没有遇到同样的问题。但是，我在这段代码中没有看到任何可能导致线程之间发生任何冲突而导致缓慢的内容。
2. 当开始使用多个线程时，我使用线程数减一来执行绘制。否则，最后一个线程似乎挂了。此语句t = max(Threads.nthreads() - 1, 1)可以更改为t = Threads.nthreads()使用可用线程的确切数量。

2020 年 11 月 20 日编辑

实施了 Przemyslaw Szufel 的建议。这是新代码：

time julia --threads 3 epic_draw_multi_thread.jl 400000
Started computing...
Started average_for_unique on thread 3 with n = 200000
Started average_for_unique on thread 2 with n = 200000
Completed average_for_unique on thread 2 with n = 200000
Completed average_for_unique on thread 3 with n = 200000
70.44460749999999
real    0m14.347s
user    0m26.959s
sys     0m2.124s

更新的基准：

Threads          @btime     Linux Time       
1 (2 processes)  9.927 s    0m44.871s 
2 (1 process)   20.237 s    1m14.156s
3 (1 process)   14.302 s    1m2.114s

Answer 1

这里有两个问题：

1. 您没有正确衡量性能
2. 在线程中生成随机数时，您应该MersenneTwister为每个线程设置一个单独的随机状态以获得最佳性能（否则您的随机状态将在所有线程之间共享，并且需要进行同步）

目前您正在测量“Julia 启动时间”+“代码编译时间”+“运行时间”的时间。多线程代码的编译显然比单线程代码的编译时间更长。启动 Julia 本身也需要一两秒钟。

你在这里有两个选择。最简单的方法是使用BenchmarkTools @btime宏来测量代码内部的执行时间。另一种选择是将您的代码打包成一个包，然后通过PackageCompiler将其编译成一个 Julia 镜像。但是，您仍将测量“Julia 开始时间”+“Julia 执行时间”

随机数状态可以创建为：

mts = MersenneTwister.(1:Threads.nthreads());

然后使用如 rand(mts[Threads.threadid()])