公告：不再推荐使用线程局部状态

关于 threadid() 和 nthreads() 的常见误解

有问题的模式

部分原因是我们的并行和并发接口的演变历史^[1]，一些 Julia 程序员编写了错误的并行代码，这些代码可能存在竞争条件，从而导致结果错误。这种模式甚至在以前的官方 Julia 博客文章中被错误地推荐过。为了生态系统的稳定性和正确性，必须识别并修复这些用法。

此错误代码遵循的通用模板如下所示

using Base.Threads: nthreads, @threads, threadid

states = [some_initial_value for _ in 1:nthreads()]
@threads for x in some_data
    tid = threadid()
    old_val = states[tid]
    new_val = some_operator(old_val, f(x))
    states[tid] = new_val
end
do_something(states)

以上代码是错误的，因为 @threads 生成的任务允许在执行过程中让出给其他任务^[2]。这意味着在读取 old_val 和将 new_val 存储到存储区之间，任务可能会暂停，并且在同一线程上运行且具有相同 threadid() 的新任务可能会并发地写入 states[tid]，从而导致竞争条件，并导致工作丢失。

这实际上并不是多线程特有的问题，而是一个并发问题，即使只有一个线程也能证明这一点。例如

$ julia --threads=1

julia> f(i) = (sleep(0.001); i);

julia> let state = [0], N=100
           @sync for i ∈ 1:N
               Threads.@spawn begin
                   tid = Threads.threadid()  # Each task gets `tid = 1`.
                   old_var = state[tid]      # Each task reads the current value, which for
                                             # all is 0 (!) because...
                   new_var = old_var + f(i)  # ...the `sleep` in `f` causes all tasks to pause
                                             # *simultaneously* here (all loop iterations start,
                                             # but do not yet finish).
                   state[tid] = new_var      # After being released from the `sleep`, each task
                                             # sets `state[1]` to `i`.
               end
           end
           sum(state), sum(1:N)
       end
(100, 5050)

在上面的代码段中，我们故意使用 100 个单独的任务过度订阅了 CPU，以使错误更有可能出现，但即使没有生成很多任务，也可能出现此问题。

包或用户代码中任何使用 threadid() 的情况都应视为警告信号，表明代码依赖于实现细节，并且容易出现并发错误。

修复使用此模式的有问题的代码

快速修复：将 @threads 替换为 @threads :static

如果您碰巧使用 @threads，最简单（但从长远来看不推荐）的快速修复方法是将 @threads for ... 的用法替换为 @threads :static for ...。这样做的原因是 @threads 的 :static 调度程序不允许导致错误的异步任务迁移和让出。

但是，这不是一个好的长期解决方案，因为

1. 它依赖于不明显的隐式保护措施来防止原本错误的代码变得正确

2. @threads :static 不是协作的或可组合的，这意味着如果您的 @threads :static 循环内的代码也执行多线程，您的代码可能会降低到比单线程速度更慢，甚至死锁。

更好的修复：直接使用任务

如果您想要一个食谱，可以使用它将上述错误的食谱替换为仅使用 Base.Threads 模块编写的食谱，我们建议您远离 @threads，而是直接使用 @spawn 来创建和管理任务。原因是 @threads 没有任何内置机制来管理和合并来自不同线程的工作结果，而任务可以以安全的方式管理和返回自己的状态。

任务创建和返回自己的状态本质上比并行任务的生成器为生成的子任务设置要读取和写入的状态更安全。

替换上面显示的错误代码模式的代码可能如下所示

using Base.Threads: nthreads, @threads, @spawn
using Base.Iterators: partition

tasks_per_thread = 2 # customize this as needed. More tasks have more overhead, but better
                     # load balancing

chunk_size = max(1, length(some_data) ÷ (tasks_per_thread * nthreads()))
data_chunks = partition(some_data, chunk_size) # partition your data into chunks that
                                               # individual tasks will deal with
#See also ChunkSplitters.jl and SplittablesBase.jl for partitioning data

tasks = map(data_chunks) do chunk
    # Each chunk of your data gets its own spawned task that does its own local, sequential work
    # and then returns the result
    @spawn begin
        state = some_initial_value
        for x in chunk
            state = some_operator(state, f(x))
        end
        return state
    end
end
states = fetch.(tasks) # get all the values returned by the individual tasks. fetch is type
                       # unstable, so you may optionally want to assert a specific return type.

do_something(states)

这是对旧有错误模式的完全通用替换。但是，对于许多应用程序，这应该简化为 mapreduce 的并行版本

using Base.Threads: nthreads, @spawn
function tmapreduce(f, op, itr; tasks_per_thread::Int = 2, kwargs...)
    chunk_size = max(1, length(itr) ÷ (tasks_per_thread * nthreads()))
    tasks = map(Iterators.partition(itr, chunk_size)) do chunk
        @spawn mapreduce(f, op, chunk; kwargs...)
    end
    mapreduce(fetch, op, tasks; kwargs...)
end

在 tmapreduce(f, op, itr) 中，函数 f 应用于 itr 的每个元素，然后是关联^[3]的双参数函数 op。

希望以上 tmapreduce 可以在不久的将来添加到基础 Julia 中。然而，在此期间，编写自己的代码相对简单，如上所示。

另一种选择：使用正确处理此问题的包

我们鼓励大家查看各种多线程库，例如 Transducers.jl（或各种前端，如 ThreadsX.jl、FLoops.jl 和 Folds.jl）以及 MultiThreadedCaches.jl。

Transducers.jl 生态系统

Transducers.jl 从根本上讲是关于以结构化和原则性的方式表达数据遍历，通常的好处是它使并行计算更容易推理。

以上 tmapreduce(f, op, itr) 可以表示为

using Transducers
itr |> Map(f) |> foldxt(op; init=some_initial_value)

或

using Transducers
foldxt(op, Map(f), itr; init=some_initial_value)

列出的各种前端为 Transducers.jl 提供了不同的 API，有些人可能会发现这些 API 更易于使用。例如

using ThreadsX
ThreasdX.mapreduce(f, op, itr; init=some_initial_value)

或

using FLoops
@floop for x in itr
    @reduce out = op(some_initial_value, f(x))
end
out

MultiThreadedCaches.jl

另一方面，MultiThreadedCaches.jl 试图通过使用锁机制来阻止数据竞争，从而使 states[threadid()] 类的模式更安全。我们认为这不是一个理想的解决方法，但它可能使过渡到更安全的代码变得更容易，并且需要更少的重写，例如在包必须管理可能由用户并发写入的状态并且包无法控制用户如何构建其代码的情况下。

[1]

并发与并行

[2]

不要试图推理让出

[3]

关联性