我正在实现多个参与者同时并行传播八卦的Gossip Algorithm。当每个演员听了 10 次八卦时，系统将停止。
现在，我有一个场景，在向接收者发送八卦之前，我正在检查接收者演员的收听计数。如果收听计数已经是 10，则八卦不会发送给接收者演员。我正在使用同步调用来获取侦听计数。
def get_message(server, msg) do
GenServer.call(server, {:get_message, msg})
end
def handle_call({:get_message, msg}, _from, state) do
listen_count = hd(state)
{:reply, listen_count, state}
end
该程序在启动时运行良好，但一段时间后Genserver.call停止并出现如下超时错误。经过一些调试，我意识到Genserver.call进入休眠状态，无法启动相应的handle_call方法。使用同步调用时是否会出现此行为？由于所有参与者都是独立的，因此Genserver.call方法不应该独立运行而不等待彼此的响应。
02:28:05.634 [error] GenServer #PID<0.81.0> terminating
** (stop) exited in: GenServer.call(#PID<0.79.0>, {:get_message, []}, 5000)
** (EXIT) time out
(elixir) lib/gen_server.ex:774: GenServer.call/3
编辑：以下代码可以在 iex shell 中运行时重现错误。
defmodule RumourActor do
use GenServer
def start_link(opts) do
{:ok, pid} = GenServer.start_link(__MODULE__,opts)
{pid}
end
def set_message(server, msg, recipient) do
GenServer.cast(server, {:set_message, msg, server, recipient})
end
def get_message(server, msg) do
GenServer.call(server, :get_message)
end
def init(opts) do
state=opts
{:ok,state}
end
def handle_cast({:set_message, msg, server, recipient},state) do
:timer.sleep(5000)
c = RumourActor.get_message(recipient, [])
IO.inspect c
{:noreply,state}
end
def handle_call(:get_message, _from, state) do
count = tl(state)
{:reply, count, state}
end
end
打开 iex 外壳并加载上面的模块。使用以下方法启动两个进程：
a = RumourActor.start_link(["", 3])
b = RumourActor.start_link(["", 5])
通过调用 Dogbert 在注释中提到的死锁条件来产生错误。运行没有太大的时差。
cb = RumourActor.set_message(elem(a,0), [], elem(b,0))
ca = RumourActor.set_message(elem(b,0), [], elem(a,0))
等待 5 秒钟。将出现错误。
八卦协议是一种处理异步、未知、未配置(随机)网络的方法，这些网络可能会遭受间歇性中断和分区，并且不存在领导者或默认结构。(请注意，这种情况在现实世界中有些不寻常，并且总是以某种方式对系统施加带外控制。
考虑到这一点，让我们将其更改为异步系统(使用cast)，以便我们遵循聊天八卦风格交流概念的精神。
我们需要消息摘要来计算给定消息的接收次数，已收到并且已经超过幻数的消息摘要(因此，如果很晚，我们不会重新发送)，以及系统中注册的进程列表，以便我们知道我们正在向谁广播：
(下面的例子是在 Erlang 中，因为我自从停止使用它以来就绊倒了 Elixir 语法......
-module(rumor).
-record(s,
{peers = [] :: [pid()],
digest = #{} :: #{message_id(), non_neg_integer()},
dead = sets:new() :: sets:set(message_id())}).
-type message_id() :: zuuid:uuid().
在这里，我使用的是UUID，但它可能是任何东西。对于测试用例来说，Erlang 引用是可以的，但是由于八卦在 Erlang 集群中没有用，并且引用在原始系统之外是不安全的，我只是跳到假设这是针对网络系统的。
我们将需要一个接口函数，允许我们告诉进程将新消息注入系统。我们还需要一个接口函数，一旦消息已经在系统中，它就会在两个进程之间发送消息。然后，我们将需要一个内部函数，该函数将消息广播到所有已知(订阅)的对等方。啊，这意味着我们需要一个问候界面，以便对等进程可以相互通知它们的存在。
我们还希望有一种方法，让一个过程告诉自己随着时间的推移继续广播。设置重传间隔的时间实际上并不是一个简单的决定 - 它与网络拓扑，延迟，可变性等有关(实际上，您实际上可能会偶尔ping对等体并根据延迟开发一些启发式方法，丢弃似乎无响应的对等体，等等 - 但我们不会在这里陷入这种疯狂)。在这里，我只是将其设置为 1 秒，因为对于观察系统的人类来说，这是一个易于解释的间隔。
请注意，以下所有内容都是异步的。
接口。。。
insert(Pid, Message) ->
gen_server:cast(Pid, {insert, Message}).
relay(Pid, ID, Message) ->
gen_server:cast(Pid, {relay, ID, Message}).
greet(Pid) ->
gen_server:cast(Pid, {greet, self()}).
make_introduction(Pid, PeerPid) ->
gen_server:cast(Pid, {make_introduction, PeerPid}).
最后一个函数将成为我们作为系统测试人员的方式，使其中一个进程在某些目标 Pid 上调用greet/1，以便他们开始构建对等网络。在现实世界中，通常会发生一些略有不同的事情。
在接收演员表的gen_server回调中，我们将得到：
handle_cast({insert, Message}, State) ->
NewState = do_insert(Message, State);
{noreply, NewState};
handle_cast({relay, ID, Message}, State) ->
NewState = do_relay(ID, Message, State),
{noreply, NewState};
handle_cast({greet, Peer}, State) ->
NewState = do_greet(Peer, State),
{noreply, NewState};
handle_cast({make_introduction, Peer}, State) ->
NewState = do_make_introduction(Peer, State),
{noreply, NewState}.
很简单的东西。
上面我提到我们需要一种方法让这个东西告诉自己在延迟后重新发送。为此，我们将在使用erlang:send_after/3延迟后向"redo_relay"发送一条裸消息，因此我们需要一个 handle_info/2 来处理它：
handle_info({redo_relay, ID, Message}, State) ->
NewState = do_relay(ID, Message, State),
{noreply, NewState}.
消息位的实现是有趣的部分，但这些都不是非常棘手。请原谅下面的do_relay/3- 它可能更简洁，但我是在浏览器中写的，所以......
do_insert(Message, State = #s{peers = Peers, digest = Digest}) ->
MessageID = zuuid:v1(),
NewDigest = maps:put(MessageID, 1, Digest),
ok = broadcast(Message, Peers),
ok = schedule_resend(MessageID, Message),
State#s{digest = NewDigest}.
do_relay(ID,
Message,
State = #s{peers = Peers, digest = Digest, dead = Dead}) ->
case maps:find(ID, Digest) of
{ok, Count} when Count >= 10 ->
NewDigest = maps:remove(ID, Digest),
NewDead = sets:add_element(ID, Dead),
ok = broadcast(Message, Peers),
State#s{digest = NewDigest, dead = NewDead};
{ok, Count} ->
NewDigest = maps:put(ID, Count + 1),
ok = broadcast(ID, Message, Peers),
ok = schedule_resend(ID, Message),
State#s{digest = NewDigest};
error ->
case set:is_element(ID, Dead) of
true ->
State;
false ->
NewDigest = maps:put(ID, 1),
ok = broadcast(Message, Peers),
ok = schedule_resend(ID, Message),
State#s{digest = NewDigest}
end
end.
broadcast(ID, Message, Peers) ->
Forward = fun(P) -> relay(P, ID, Message),
lists:foreach(Forward, Peers).
schedule_resend(ID, Message) ->
_ = erlang:send_after(1000, self(), {redo_relay, ID, Message}),
ok.
现在我们需要社交位...
do_greet(Peer, State = #s{peers = Peers}) ->
case lists:member(Peer, Peers) of
false -> State#s{peers = [Peer | Peers]};
true -> State
end.
do_make_introduction(Peer, State = #s{peers = Peers}) ->
ok = greet(Peer),
do_greet(Peer, State).
那么，那里所有可怕的未打字的东西都做了什么呢？
它避免了任何陷入僵局的可能性。死锁在对等系统中如此致命的原因是，只要你有两个相同的进程(或参与者，或其他什么)同步通信，你就创造了一个潜在死锁的教科书案例。
每当A有一个同步消息朝向B，而B有一个同步消息朝向A，同时你现在有一个死锁。无法在不创建潜在死锁的情况下创建同步调用彼此的相同进程。在大规模并发系统中，几乎肯定会发生任何事情，所以你迟早会遇到这种情况。
八卦旨在异步是有原因的：它是一种草率、不可靠、低效的方式来处理草率、不可靠、低效的网络拓扑。试图进行调用而不是强制转换不仅违背了八卦式消息中继的目的，而且还将您推入不可能的僵局领域，从而将协议的性质从异步更改为同步。
Genser.call的默认超时为 5000 毫秒。所以可能发生的情况是，参与者的消息队列充满了数百万条消息，当它到达call时，调用参与者已经超时。
您可以使用try...catch处理超时：
try do
c = RumourActor.get_message(recipient, [])
catch
:exit, reason ->
# handle timeout
现在，被调用的参与者最终将到达call消息并做出响应，这将作为对第一个进程的意外消息。您需要使用handle_info来处理此问题。因此，一种方法是忽略catch块中的错误并从handle_info发送谣言。
此外，如果有许多进程等待超时 5 秒，然后继续前进，这将显著降低性能。可以故意减少超时并在handle_info中处理回复。这将减少到使用cast和处理来自其他进程的回复。
您的阻止呼叫需要分为两个非阻止呼叫。因此，如果 A 正在向 B 发出阻止调用，而不是等待回复，A 可以要求 B 在给定地址(A 的地址)上发送其状态并继续前进。 然后 A 将单独处理该消息，并在必要时回复。
A.fun1():
body of A before blocking call
result = blockingcall()
do things based on result
需要分为：
A.send():
body of A before blocking call
nonblockingcall(A.receive) #A.receive is where B should send results
do other things
A.receive(result):
do things based on result