如何以及何时在 Julia 中使用 @async 和 @sync

Question

我已阅读 documentation 中的 @async 和 @sync 宏，但仍然无法弄清楚如何以及何时使用它们，也无法在 Internet 上的其他地方找到许多资源或示例。

我的近期目标是找到一种方法，让多个工作人员并行工作，然后等到他们全部完成后再继续编写我的代码。这篇文章：Waiting for a task to be completed on remote processor in Julia 包含了一种成功的方法来实现这一点。我原以为应该可以使用@async 和@sync 宏，但我最初未能做到这一点让我想知道我是否正确理解了如何以及何时使用这些宏。

Answer 1

根据?@async 下的文档，“@async 将表达式包装在任务中。”这意味着对于其范围内的任何内容，Julia 将开始运行此任务，然后继续执行脚本中接下来的任何内容，而无需等待任务完成。因此，例如，如果没有宏，您将获得：

julia> @time sleep(2)
  2.005766 seconds (13 allocations: 624 bytes)

但是使用宏，你会得到：

julia> @time @async sleep(2)
  0.000021 seconds (7 allocations: 657 bytes)
Task (waiting) @0x0000000112a65ba0

julia> 

因此，Julia 允许脚本继续执行（并完全执行 @time 宏），而无需等待任务（在本例中，休眠两秒钟）完成。

相比之下，@sync 宏将“等到 @async、@spawn、@spawnat 和 @parallel 的所有动态封闭使用完成。” （根据?@sync 下的文档）。因此，我们看到：

julia> @time @sync @async sleep(2)
  2.002899 seconds (47 allocations: 2.986 KB)
Task (done) @0x0000000112bd2e00

那么，在这个简单的示例中，将@async 和@sync 的单个实例包含在一起是没有意义的。但是，@sync 可能有用的地方是您将@async 应用于您希望允许一次全部启动而无需等待每个操作完成的多个操作。

例如，假设我们有多个工作人员，我们希望让每个工作人员同时处理一项任务，然后从这些任务中获取结果。最初的（但不正确的）尝试可能是：

using Distributed
cell(N) = Vector{Any}(undef, N)

addprocs(2)
@time begin
    a = cell(nworkers())
    for (idx, pid) in enumerate(workers())
        a[idx] = remotecall_fetch(sleep, pid, 2)
    end
end
## 4.011576 seconds (177 allocations: 9.734 KB)

这里的问题是循环等待每个remotecall_fetch() 操作完成，即每个进程在继续开始下一个remotecall_fetch() 操作之前完成其工作（在这种情况下休眠2 秒）。就实际情况而言，我们在这里没有获得并行性的好处，因为我们的进程不会同时进行它们的工作（即休眠）。

不过，我们可以通过结合使用 @async 和 @sync 宏来纠正此问题：

@time begin
    a = cell(nworkers())
    @sync for (idx, pid) in enumerate(workers())
        @async a[idx] = remotecall_fetch(sleep, pid, 2)
    end
end
## 2.009416 seconds (274 allocations: 25.592 KB)

现在，如果我们将循环的每一步都算作一个单独的操作，我们会看到前面有两个单独的操作，前面有 @async 宏。宏允许这些中的每一个启动，并且代码在每个完成之前继续（在这种情况下是循环的下一步）。但是，使用 @sync 宏（其范围包含整个循环）意味着在 @async 之前的所有操作都完成之前，我们不会允许脚本继续通过该循环。

通过进一步调整上面的例子，看看它在某些修改下是如何变化的，可以更清楚地了解这些宏的操作。例如，假设我们只有 @async 而没有 @sync：

@time begin
    a = cell(nworkers())
    for (idx, pid) in enumerate(workers())
        println("sending work to $pid")
        @async a[idx] = remotecall_fetch(sleep, pid, 2)
    end
end
## 0.001429 seconds (27 allocations: 2.234 KB)

在这里，@async 宏允许我们在每个 remotecall_fetch() 操作完成执行之前继续循环。但是，无论好坏，我们没有 @sync 宏来阻止代码继续通过此循环，直到所有 remotecall_fetch() 操作完成。

尽管如此，每个remotecall_fetch() 操作仍在并行运行，即使我们继续进行。我们可以看到，因为如果我们等待两秒钟，那么包含结果的数组 a 将包含：

sleep(2)
julia> a
2-element Array{Any,1}:
 nothing
 nothing

（“nothing”元素是sleep函数成功获取结果，不返回任何值）

我们还可以看到，两个remotecall_fetch() 操作基本上同时开始，因为它们之前的打印命令也快速连续执行（这些命令的输出未在此处显示）。将此与下一个示例进行对比，在该示例中，打印命令的执行延迟为 2 秒：

如果我们将@async 宏放在整个循环上（而不是只是它的内部步骤），那么我们的脚本将再次立即继续，而无需等待remotecall_fetch() 操作完成。然而，现在我们只允许脚本作为一个整体继续通过循环。我们不允许循环的每个单独步骤在前一个步骤完成之前开始。因此，与上例不同，在脚本执行循环后两秒后，结果数组仍有一个元素为 #undef，表示第二个 remotecall_fetch() 操作仍未完成。

@time begin
    a = cell(nworkers())
    @async for (idx, pid) in enumerate(workers())
        println("sending work to $pid")
        a[idx] = remotecall_fetch(sleep, pid, 2)
    end
end
# 0.001279 seconds (328 allocations: 21.354 KB)
# Task (waiting) @0x0000000115ec9120
## This also allows us to continue to

sleep(2)

a
2-element Array{Any,1}:
    nothing
 #undef    

而且，毫不奇怪，如果我们将 @sync 和 @async 并排放置，我们会发现每个 remotecall_fetch() 按顺序（而不是同时）运行，但我们不会继续代码，直到每个已完成。换句话说，我相信，这基本上相当于如果我们没有宏，就像sleep(2) 的行为基本上与@sync @async sleep(2) 相同

@time begin
    a = cell(nworkers())
    @sync @async for (idx, pid) in enumerate(workers())
        a[idx] = remotecall_fetch(sleep, pid, 2)
    end
end
# 4.019500 seconds (4.20 k allocations: 216.964 KB)
# Task (done) @0x0000000115e52a10

还要注意，在@async 宏的范围内可以进行更复杂的操作。 documentation 给出了一个示例，其中包含@async 范围内的整个循环。

更新：回想一下，同步宏的帮助声明它将“等到 @async、@spawn、@spawnat 和 @parallel 的所有动态封闭使用完成。”就什么算“完成”而言，重要的是您如何在@sync 和@async 宏的范围内定义任务。考虑以下示例，它与上面给出的示例之一略有不同：

@time begin
    a = cell(nworkers())
    @sync for (idx, pid) in enumerate(workers())
        @async a[idx] = remotecall(sleep, pid, 2)
    end
end
## 0.172479 seconds (93.42 k allocations: 3.900 MB)

julia> a
2-element Array{Any,1}:
 RemoteRef{Channel{Any}}(2,1,3)
 RemoteRef{Channel{Any}}(3,1,4)

前面的示例大约需要 2 秒的时间来执行，这表明这两个任务是并行运行的，并且脚本在继续之前等待每个任务完成其功能的执行。然而，这个例子的时间评估要低得多。原因是对于@sync 的目的，remotecall() 操作一旦向工作人员发送了要完成的工作，就已经“完成”了。 （请注意，此处的结果数组 a 仅包含 RemoteRef 对象类型，这仅表明特定进程正在发生某些事情，理论上可以在将来的某个时间获取）。相比之下，remotecall_fetch() 操作只有在从工作人员那里获得任务已完成的消息时才“完成”。

因此，如果您正在寻找方法以确保在继续执行脚本之前已完成与工作人员的某些操作（例如，在这篇文章中讨论：Waiting for a task to be completed on remote processor in Julia），有必要仔细考虑什么是重要的“完成”以及如何衡量并在脚本中实施。