如何并行化排序？

Question

我想分类很多东西.

Julia的标准库排序是单线程的.如何利用我的多核机器更快地对事物进行排序？

Answer 1

这是使用(一种实验性)Base.Threads线程模块的解决方案.

使用pmap(等)用于分布式并行性的解决方案将是类似的.虽然我认为进程间通信开销会伤害你.

我们的想法是用块(每个线程一个)对它进行排序,因此每个线程可以完全独立,只需要处理它的块.

然后合并这些预先排序的块.

这是合并排序列表的一个众所周知的问题.另见其他问题.

并且不要忘记通过JULIA_NUM_THREADS在开始之前设置环境变量来设置自己的多线程.

这是我的代码:

using Base.Threads

function blockranges(nblocks, total_len)
    rem = total_len % nblocks
    main_len = div(total_len, nblocks)

    starts=Int[1]
    ends=Int[]
    for ii in 1:nblocks
        len = main_len
        if rem>0
            len+=1
            rem-=1
        end
        push!(ends, starts[end]+len-1)
        push!(starts, ends[end] + 1)
    end
    @assert ends[end] == total_len
    starts[1:end-1], ends
end

function threadedsort!(data::Vector)
    starts, ends = blockranges(nthreads(), length(data))

    # Sort each block
    @threads for (ss, ee) in collect(zip(starts, ends))
        @inbounds sort!(@view data[ss:ee])
    end


    # Go through each sorted block taking out the smallest item and putting it in the new array
    # This code could maybe be optimised. see https://stackoverflow.com/a/22057372/179081
    ret = similar(data) # main bit of allocation right here. avoiding it seems expensive.
    # Need to not overwrite data we haven't read yet
    @inbounds for ii in eachindex(ret)
        minblock_id = 1
        ret[ii]=data[starts[1]]
        @inbounds for blockid in 2:endof(starts) # findmin allocates a lot for some reason, so do the find by hand. (maybe use findmin! ?)
            ele = data[starts[blockid]]
            if ret[ii] > ele
                ret[ii] = ele
                minblock_id = blockid
            end
        end
        starts[minblock_id]+=1 # move the start point forward
        if starts[minblock_id] > ends[minblock_id]
            deleteat!(starts, minblock_id)
            deleteat!(ends, minblock_id)
        end
    end
    data.=ret  # copy back into orignal as we said we would do it inplace
    return data
end

我做了一些基准测试:

using Plots
function evaluate_timing(range)
    sizes = Int[]
    threadsort_times = Float64[]
    sort_times = Float64[]
        for sz in 2.^collect(range)
            data_orig = rand(Int, sz)
            push!(sizes, sz)

            data = copy(data_orig)
            push!(sort_times,       @elapsed sort!(data))

            data = copy(data_orig)
            push!(threadsort_times, @elapsed threadedsort!(data))

            @show (sz, sort_times[end], threadsort_times[end])
    end
    return sizes, threadsort_times, sort_times
end

sizes, threadsort_times, sort_times = evaluate_timing(0:28)
plot(sizes, [threadsort_times sort_times]; title="Sorting Time", ylabel="time(s)", xlabel="number of elements", label=["threadsort!" "sort!"])
plot(sizes, [threadsort_times sort_times]; title="Sorting Time", ylabel="time(s)", xlabel="number of elements", label=["threadsort!" "sort!"], xscale=:log10, yscale=:log10)

我的结果:使用8个线程.

我发现交叉点非常低,略高于1024.注意可以忽略最初的长时间 - 这是第一次运行时编译的JIT代码.

奇怪的是,这些结果在使用BenchmarkTools时无法重现.基准测试工具将停止计算初始时间.但是,当我使用上面的基准代码中的常规定时代码时,它们会非常一致地重现.我想它正在做一些杀死多线程的方法

非常感谢@xiaodai在我的分析代码中指出了一个错误