5.4 运行时考虑事项

你已经学到了很多关于如何处理进程的知识。现在，是时候讨论一下 BEAM 并发的一些重要运行时属性了。

5.4.1 一个过程是顺序的

已经提到过，但这非常重要，所以我再强调一遍：单个进程是一个顺序程序——它按顺序逐个运行表达式。多个进程并发运行，因此它们可能会相互并行运行。但是，如果许多进程向单个进程发送消息，那么这个单个进程可能会成为瓶颈，这会显著影响系统的整体吞吐量。

让我们看一个例子。以下列表中的代码实现了一个慢回声服务器。

清单 5.13 过程瓶颈的演示 (process_bottleneck.ex)

defmodule Server do
  def start do
    spawn(fn -> loop() end)
  end
 
  def send_msg(server, message) do
    send(server, {self(), message})
 
    receive do
      {:response, response} -> response
    end
  end
 
  defp loop do
    receive do
      {caller, msg} ->
        Process.sleep(1000)               ?
        send(caller, {:response, msg})    ?
    end
 
    loop()
  end
end

? 模拟长时间处理

? 将消息回显

在接收到消息后，服务器将消息发送回调用者。在此之前，它会暂停一秒钟以模拟一个长时间运行的请求。

为了测试其在并发环境中的行为，启动服务器并启动五个并发客户端：

iex(1)> server = Server.start()
 
iex(2)> Enum.each(
          1..5,
          fn i ->
            spawn(fn ->                               ?
              IO.puts("Sending msg ##{i}")
              response = Server.send_msg(server, i)   ?
              IO.puts("Response: #{response}")
           end)
         end
        )

? 生成一个并发客户端

? 同步请求服务器

一旦您开始这个，您将看到以下行被打印：

Sending msg #1
Sending msg #2
Sending msg #3
Sending msg #4
Sending msg #5

到目前为止，一切顺利。五个进程已经启动并同时运行。但现在，问题开始了——响应慢慢返回，一个接一个，相隔一秒。

Response: 1        ?
Response: 2        ?
Response: 3        ?
Response: 4        ?
Response: 5        ?

? 一秒钟后

? 两秒钟后

三秒钟后

? 四秒钟后

五秒钟后

发生了什么？回声服务器每秒只能处理一个消息。因为所有其他进程都依赖于回声服务器，所以它们受到其吞吐量的限制。

你能对此做些什么？一旦你识别出瓶颈，就应该尝试在内部优化流程。通常，服务器进程的流程很简单。它逐个接收和处理消息。因此，目标是使服务器处理消息的速度至少与消息到达的速度相同。在这个例子中，服务器优化相当于去掉 Process.sleep/1 调用。

如果你无法快速处理消息，可以尝试将服务器拆分为多个进程，有效地并行化原始工作，并希望这样做能在多核系统上提升性能。不过，这应该是你的最后手段。并行化并不是解决结构不良算法的良方。

5.4.2 无限过程邮箱

理论上，进程邮箱的大小是无限的。然而在实践中，邮箱的大小受到可用内存的限制。因此，如果一个进程不断落后，意味着消息到达的速度快于进程处理的速度，邮箱将不断增长并越来越消耗内存。最终，一个单一的慢进程可能会通过消耗所有可用内存导致整个系统崩溃。

如果一个进程根本不处理某些消息，则会出现更微妙的同样问题。考虑以下服务器循环：

defp loop
  receive do
    {:message, msg} -> do_something(msg)
  end
 
  loop()
end

由此循环驱动的服务器仅处理以下形式的消息： {:message, something} 。所有其他消息将永远保留在进程邮箱中，无故占用内存空间。

过度增长的邮箱内容会显著影响性能。它给垃圾收集器带来了额外的压力，并可能导致在 receive 中模式匹配变慢。

为了避免这种情况，您应该引入一个匹配所有接收的条款，以处理意外类型的消息。通常，您会发出警告，表示某个进程已接收到未知消息，并且不再采取其他措施：

defp loop
  receive
    {:message, msg} -> do_something(msg)
    other -> warn_about_unknown_message(other)    ?
  end
 
  loop()
end

? 匹配所有条款

由于该过程处理所有类型的消息，因此其邮箱的无控制增长不太可能发生。

值得注意的是，BEAM 为您提供了在运行时分析进程的工具。特别是，您可以查询每个进程的邮箱大小，从而检测邮箱队列积累发生的进程。我们将在第 13 章讨论此功能。

5.4.3 无共享并发

如前所述，进程之间不共享内存。因此，向另一个进程发送消息会导致消息内容的深拷贝：

send(target_pid, data) \    ?

数据的内容是深度复制的。

不那么明显的是，在生成的元素中，变量闭包也会导致闭合变量的深拷贝

data = ...
 
spawn(fn ->
  ...
  some_fun(data)      ?
  ...
end)

结果是数据变量的深拷贝

在将代码移入单独进程时，您应该注意这一点。深拷贝是内存中的操作，因此应该相对快速，偶尔发送大消息不应成为问题。但是，多个进程频繁发送大消息可能会影响系统性能。“小”和“大”的概念是主观的。简单数据，例如数字、原子或包含少量元素的元组，显然是小的。另一方面，包含一百万个复杂结构的列表则是大的。界限在两者之间，具体取决于您的特定情况。

在某些特殊情况下，数据是通过引用复制的。这发生在大于 64 字节的二进制数据（包括字符串）、硬编码常量（也称为字面量）以及通过 :persistent_term API 创建的术语（https://www.erlang.org/doc/man/persistent_term.xhtml）。

共享无状态并发确保进程之间的完全隔离：一个进程无法影响另一个进程的内部状态。这促进了系统的完整性和容错性。

此外，由于进程之间不共享内存，垃圾回收可以在进程级别进行。每个进程获得一小块初始堆内存（在 64 位 BEAM 上约为 2 KB）。当需要更多内存时，该进程会进行垃圾回收。因此，垃圾回收是并发和分布式的。与其进行一次大型的“停止整个系统”的回收，不如进行许多较小、通常更快的回收。这防止了不必要的长时间完全阻塞，并使整个系统保持更高的响应性。

5.4.4 调度器内部工作原理

通常，可以假设有 n 个调度程序运行 m 个进程，其中 m 通常远大于 n。这被称为 m:n 线程模型，它反映了您使用较少的操作系统线程运行大量逻辑微线程的事实，如图 5.3 所示。

每个 BEAM 调度器都是一个操作系统线程，负责管理 BEAM 进程的执行。默认情况下，BEAM 仅使用与可用逻辑处理器数量相同的调度器。您可以通过各种 Erlang 模拟器标志更改这些设置。

要提供 Erlang 标志，您可以使用以下语法：

$ iex --erl "put Erlang emulator flags here"

所有 Erlang 标志的列表可以在 https://erlang.org/doc/man/erl.xhtml 找到。

例如，要使用仅一个调度线程，您可以提供 +S 1 Erlang 标志：

$ iex --erl "+S 1"
Erlang/OTP 26 [erts-14.0] [source] [64-bit] [smp:1:1] [ds:1:1:10]

注意输出中的 smp:1:1 部分。这告诉我们只使用了一个调度线程。您还可以通过编程方式检查调度器的数量：

iex(1)> System.schedulers()
1

如果您在系统上运行其他外部服务，可以考虑减少 BEAM 调度程序线程的数量。这样做将为非 BEAM 服务留出更多计算资源。

在内部，每个调度器选择一个进程，运行一段时间，然后选择另一个进程。在调度器中，进程获得大约 2,000 次函数调用的小执行窗口，之后会被抢占。值得一提的是，在某些情况下，长时间运行的 CPU 密集型工作或更大的垃圾回收可能会在另一个线程上执行（称为脏调度器）。

如果一个进程正在进行网络 IO 或等待消息，它会将执行权交给调度器。当调用 Process.sleep 时也会发生同样的事情。因此，您不必关心进程中执行工作的性质。如果您想将一个函数的执行与其余代码分开，只需在一个单独的进程中运行该函数，无论该工作是 CPU 密集型还是 IO 密集型。

由于所有这些原因，上下文切换频繁进行。通常，一个进程在调度器中的时间少于一毫秒。这促进了基于 BEAM 的系统的响应能力。如果一个进程执行一个长时间的 CPU 密集型操作，例如计算π的值到十亿位小数，它不会阻塞整个调度器，其他进程也不应受到影响。

这可以很容易地证明。只需启动一个带有一个调度线程的 iex 会话：

$ iex --erl "+S 1"

生成一个运行无限 CPU 绑定循环的进程：

iex(1)> spawn(fn ->
          Stream.repeatedly(fn -> :rand.uniform() end)
          |> Stream.run()
        end)

此代码使用 Stream.repeatedly/1 创建一个懒惰的无限随机数流。该流通过 Stream.run/1 函数执行，这将有效地运行一个无限的 CPU 绑定循环。为了避免阻塞 iex shell 会话，工作在一个单独的进程中完成。

一旦您开始这个计算，您应该注意到 CPU 使用率达到 100%，这证明您现在正在运行一个密集的、长时间的 CPU 绑定工作。

尽管 BEAM 仅使用一个调度线程，但 iex 会话仍然是响应式的，您可以评估其他表达式。例如，让我们对前 1,000,000,000 个整数求和：

iex(2)> Enum.sum(1..1_000_000_000)
500000000500000000

我们能够在一个已经繁忙的调度线程上运行另一个任务，并且该任务几乎立即完成。这是频繁上下文切换的结果，它确保偶尔的长时间运行任务不会显著影响整个系统的响应能力。

摘要

• BEAM 进程是一个轻量级的并发执行单元。进程是完全隔离的，且不共享内存。
• 进程可以通过异步消息进行通信。同步发送和响应是在这个基本机制之上手动构建的。
• 服务器进程是一个长时间运行（可能永远运行）的进程，处理各种消息。服务器进程由无尽的递归驱动。
• 服务器进程可以使用无尽递归的参数维护自己的私有状态。