Erlang课程并发练习：我的答案可以改进吗？

Question

我在erlang.org course做这个练习：

2) 编写一个启动 N 的函数 在一个环中处理，并发送一个 消息 M 次围绕所有 环中的过程。之后 消息已发送进程 应该优雅地终止。

这是我想出的：

-module(ring).
-export([start/2, node/2]).

node(NodeNumber, NumberOfNodes) ->
  NextNodeNumber = (NodeNumber + 1) rem NumberOfNodes,
  NextNodeName = node_name(NextNodeNumber),
  receive
    CircuitNumber ->
      io:format("Node ~p Circuit ~p~n", [NodeNumber, CircuitNumber]),
      LastNode = NodeNumber =:= NumberOfNodes - 1,
      NextCircuitNumber = case LastNode of
                           true ->
                             CircuitNumber - 1;
                           false ->
                             CircuitNumber
                         end,
      if
        NextCircuitNumber > 0 ->
          NextNodeName ! NextCircuitNumber;
        true ->
          ok
      end,
      if
        CircuitNumber > 1 ->
          node(NodeNumber, NumberOfNodes);
        true ->
          ok
      end
  end.

start(NumberOfNodes, NumberOfCircuits) ->
  lists:foreach(fun(NodeNumber) ->
                    register(node_name(NodeNumber),
                             spawn(ring, node, [NodeNumber, NumberOfNodes]))
                end,
                lists:seq(0, NumberOfNodes - 1)),
  node_name(0) ! NumberOfCircuits,
  ok.

node_name(NodeNumber) ->
  list_to_atom(lists:flatten(io_lib:format("node~w", [NodeNumber]))).

这是它的输出：

17> ring:start(3, 2).
Node 0 Circuit 2
ok
Node 1 Circuit 2
Node 2 Circuit 2
Node 0 Circuit 1
Node 1 Circuit 1
Node 2 Circuit 1

如果我真的了解 Erlang，我会采取不同的方式来改进这段代码吗？具体来说：

• 除了在最后两个 if 语句中指定一个无所事事的“真”子句之外，还有其他选择吗？

• 我真的是优雅地终止了吗？结束已注册的进程时是否需要任何特殊操作？

Answer 1

欢迎来到二郎！我希望你和我一样喜欢它。

除了在最后两个 if 语句中指定一个无所事事的“真”子句之外，还有其他选择吗？

你可以把这些关掉。我用这个运行了你的代码：

if NextCircuitNumber > 0 ->
  NextNodeName ! NextCircuitNumber
end,
if CircuitNumber > 1 ->
  node(NodeNumber, NumberOfNodes)
end

它对我有用。

我真的是优雅地终止了吗？结束已注册的进程时是否需要任何特殊操作？

是的，你是。您可以通过运行i(). 命令来验证这一点。这将显示进程列表，如果您的注册进程没有终止，您会看到很多已注册的进程，如node0、node1 等。您也将无法运行您的程序第二次，因为尝试注册已注册的名称会出错。

就您可以做的其他事情来改进代码而言，没有太多，因为您的代码基本上没问题。我可能会做的一件事是不要使用NextNodeName 变量。您可以直接向node_name(NextNodeNumber) 发送消息即可。

另外，您可能可以做更多的模式匹配来改进事情。例如，我在使用您的代码时所做的一项更改是通过传递最后一个节点的编号(NumberOfNodes - 1) 来生成进程，而不是传递NumberOfNodes。然后，我可以像这样在node/2 函数头中进行模式匹配

node(LastNode, LastNode) ->
    % Do things specific to the last node, like passing message back to node0
    % and decrementing the CircuitNumber
node(NodeNumber, LastNode) ->
    % Do things for every other node.

这让我可以在您的 node 函数中清理一些 case 和 if 逻辑，并使其更加整洁。

希望有帮助，祝你好运。

Answer 2

让我们看一下代码：

-module(ring).
-export([start/2, node/2]).

node 这个名字是我避免使用的，因为 Erlang 中的 node() 具有运行在某台机器上的 Erlang VM 的含义——通常几个节点在几台机器上运行。我宁愿称它为ring_proc 或类似的名称。

node(NodeNumber, NumberOfNodes) ->
   NextNodeNumber = (NodeNumber + 1) rem NumberOfNodes,
   NextNodeName = node_name(NextNodeNumber),

这是我们试图生成的，我们得到下一个节点的数字和下一个节点的名称。让我们看看node_name/1作为一个插曲：

node_name(NodeNumber) ->
   list_to_atom(lists:flatten(io_lib:format("node~w", [NodeNumber]))).

这个函数是个坏主意。您将需要一个需要是原子的本地名称，因此您创建了一个可以创建任意此类名称的函数。这里的警告是 atom table 不是垃圾收集和限制的，所以我们应该尽可能避免它。解决这个问题的诀窍是传递 pid 并反向构建环。然后最后的过程将打结：

mk_ring(N) ->
  Pid = spawn(fun() -> ring(none) end),
  mk_ring(N, Pid, Pid).

mk_ring(0, NextPid, Initiator) ->
   Initiator ! {set_next, NextPid},
   Initiator;
mk_ring(N, NextPid, Initiator) ->
   Pid = spawn(fun() -> ring(NextPid) end),
   mk_ring(N-1, Pid, Initiator).

然后我们可以重写你的启动函数：

start(NumberOfNodes, NumberOfCircuits) ->
  RingStart = mk_ring(NumberOfNodes)
  RingStart ! {operate, NumberOfCircuits, self()},
  receive
    done ->
        RingStart ! stop
  end,
  ok.

Ring 代码类似于以下内容：

ring(NextPid) ->
  receive
    {set_next, Pid} ->
        ring(Pid);
    {operate, N, Who} ->
        ring_ping(N, NextPid),
        Who ! done,
        ring(NextPid);
    ping ->
        NextPid ! ping,
        ring(NextPid);
    stop ->
        NextPid ! stop,
        ok
  end.

然后在环周围发射 N 次：

ring_ping(0, _Next) -> ok;
ring_ping(N, Next) ->
  Next ! ping
  receive
    ping ->
      ring_ping(N-1, Next)
  end.

（顺便说一句，这段代码都没有经过测试，所以很可能是完全错误的）。

至于你的其余代码：

receive
  CircuitNumber ->
    io:format("Node ~p Circuit ~p~n", [NodeNumber, CircuitNumber]),

我会用一些原子标记CircuitNumber：{run, CN}。

  LastNode = NodeNumber =:= NumberOfNodes - 1,
  NextCircuitNumber = case LastNode of
                       true ->
                         CircuitNumber - 1;
                       false ->
                         CircuitNumber
                     end,

这可以通过 if 来完成：

  NextCN = if NodeNumber =:= NumberOfNodes - 1 -> CN -1;
              NodeNumber =/= NumberOfNodes - 1 -> CN
           end,

这里的下一部分：

  if
    NextCircuitNumber > 0 ->
      NextNodeName ! NextCircuitNumber;
    true ->
      ok
  end,
  if
    CircuitNumber > 1 ->
      node(NodeNumber, NumberOfNodes);
    true ->
      ok
  end

确实需要true 的情况，除非你从来没有打过它。如果if 中没有任何匹配项，该进程将崩溃。通常可以重新连接代码，使其不那么依赖于计数结构，就像我上面的代码提示一样。

---

使用此代码可以避免一些麻烦。当前代码的一个问题是，如果环中的某些东西崩溃了，它就会被破坏。我们可以使用spawn_link 而不是spawn 将环链接在一起，这样这样的错误会破坏整个环。此外，如果在环运行时发送消息，我们的ring_ping 函数将崩溃。这可以缓解，最简单的方法可能是改变环进程的状态，使其知道它当前正在运行并将ring_ping折叠成ring。最后，我们可能还应该链接初始生成，这样我们就不会得到一个活的但没有人参考的大环。也许我们可以注册初始进程，以便以后轻松抓住戒指。

start 函数在两个方面也很糟糕。首先，我们应该使用make_ref() 来标记一条唯一的消息并接收该标记，这样另一个进程就不会是险恶的，而只是在环工作时将done 发送到启动进程。我们可能还应该在环上添加一个 monitor，而它正在工作。否则我们将永远不会被告知，在我们等待done 消息（带有标签）时应该是环崩溃。顺便说一下，OTP 在其同步调用中都做到了。

---

最后，终于：不，您不必清理注册。

Answer 3

我的同事提出了一些很好的观点。我还想提一下，通过注册进程而不是实际创建环来避免问题的最初意图。这是一种可能的解决方案：

-module(ring).
-export([start/3]).
-record(message, {data, rounds, total_nodes, first_node}).

start(TotalNodes, Rounds, Data) ->
    FirstNode = spawn_link(fun() -> loop(1, 0) end),
    Message = #message{data=Data, rounds=Rounds, total_nodes=TotalNodes,
                       first_node=FirstNode},
    FirstNode ! Message, ok.

loop(Id, NextNode) when not is_pid(NextNode) ->
    receive
        M=#message{total_nodes=Total, first_node=First} when Id =:= Total ->
            First ! M,
            loop(Id, First);
        M=#message{} ->
            Next = spawn_link(fun() -> loop(Id+1, 0) end),
            Next ! M,
            loop(Id, Next)
    end;
loop(Id, NextNode) ->
    receive
        M=#message{rounds=0} ->
            io:format("node: ~w, stopping~n", [Id]),
            NextNode ! M;
        M=#message{data=D, rounds=R, total_nodes=Total} ->
            io:format("node: ~w, message: ~p~n", [Id, D]),
            if Id =:= Total -> NextNode ! M#message{rounds=R-1};
               Id =/= Total -> NextNode ! M
            end,
            loop(Id, NextNode)
    end.

此解决方案使用记录。如果您不熟悉它们，请阅读所有关于它们的信息here。

每个节点都由loop/2 函数定义。 loop/2 的第一个子句处理创建环（构建阶段），第二个子句处理打印消息（数据阶段）。请注意，所有子句都以对loop/2 的调用结束，但rounds=0 子句除外，这表明节点已完成其任务并且应该死亡。这就是优雅终止的意思。另请注意用于告诉节点它处于构建阶段的 hack - NextNode 不是 pid 而是整数。