multithreading - GenServer上的Elixir非阻塞线程？

Question

我正在尝试完成一个简单的任务，但遇到了巨大的困难。

请假设我有一个GenServer，它的回调之一如下:

  @impl true
  def handle_call(:state, _, state) do
    # Something that would require 10 seconds
    newState = do_job()
    {:reply, newState, newState}
  end

如果我是对的，那么从客户端调用 GenServer.call(:server, :state)将阻塞服务器10秒钟，然后将新状态返回给客户端。

哎呀我希望服务器处理此任务而不会被阻塞。我尝试使用任务，但是 Task.await/2和 Task.yield/2阻止了服务器。

我希望服务器不阻塞，在那10秒钟之后，在客户端上接收结果。这怎么可能？

Answer 1

If I am right, invoking GenServer.call(:server, :state) from a client side would block the server for 10 seconds, and then the new state would be returned to the client.

是的。 Elixir会执行您要执行的操作，并在此行中:

newState = do_job()

您正在告诉elixir将 do_job()的返回值分配给变量 newState。 Elixir 可以执行该分配的唯一方法是获取 go_job() ....的返回值，这将需要10秒钟。

I want the server not to block, and after those 10 seconds, receive the result on the client terminal.

一种方法是让GenServer对新进程进行 spawn()编码以执行10秒功能，并将客户端的pid传递给新进程。当新进程从10秒函数获得返回值时，新进程可以使用客户端pid对客户端消息进行 send()编码。

这意味着客户端将需要调用 handle_call()而不是 handle_cast()，因为服务器的 handle_cast()的实现没有包含客户端pid的 from参数变量。另一方面， handle_call()确实在 from参数变量中接收到客户端pid，因此服务器可以将客户端pid传递给生成的进程。请注意， spawn()立即返回，这意味着 handle_call()可以立即返回并返回类似 :working_on_it的回复。

下一个问题是:客户端将如何知道GenServer产生的新进程何时完成了10秒功能的执行？客户端不知道服务器上某些无关的进程何时完成执行，因此客户端需要等待接收，直到消息从生成的进程到达为止。而且，如果客户端正在检查其邮箱中的邮件，了解发件人是有帮助的，这意味着 handle_call()也应将生成的进程的pid返回给客户端。客户端的另一种选择是在执行其他工作之间频繁地轮询其邮箱。为此，客户端可以在 after clause中定义具有短超时的接收，然后在 after clause中调用一个函数来完成一些客户端工作，然后递归调用包含接收的函数，以便该函数再次检查邮箱。

现在 Task呢？根据 Task docs:

If you are using async tasks, you must await a reply...

那么，如果您必须等待，那么异步任务有什么好处呢？答:如果一个进程至少需要执行两个长期运行的函数，则该进程可以使用 Task.async()来同时运行所有函数，而不是执行一个函数并等待直到完成，然后再执行另一个函数并等待直到完成，然后再执行另一个，依此类推。

但是，Task还定义了一个 start()函数:

start(mod, fun, args)

Starts a task.

This is only used when the task is used for side-effects (i.e. no interest in the returned result) and it should not be linked to the current process.

听起来 Task.start()完成了我在第一种方法中描述的内容。您需要定义 fun，以便它将运行10秒功能，然后在10秒功能完成执行(=副作用)后将消息发送回客户端。

下面是一个生成了长期运行功能的GenServer的简单示例，该服务器允许服务器在长期运行功能执行时保持对其他客户端请求的响应:

a.exs:

defmodule Gen1.Server do
  use GenServer

  @impl true
  def init(init_state) do
    {:ok, init_state}
  end

  def long_func({pid, _ref}) do
    Process.sleep 10_000
    result = :dog
    send(pid, {self(), result})
  end

  @impl true
  def handle_call(:go_long, from, state) do
    long_pid = spawn(Gen1.Server, :long_func, [from])
    {:reply, long_pid, state}
  end
  def handle_call(:other, _from, state) do
    {:reply, :other_stuff, state}
  end

end

一个iex session 将是客户端:

~/elixir_programs$ iex a.exs
Erlang/OTP 20 [erts-9.3] [source] [64-bit] [smp:4:4] [ds:4:4:10] [async-threads:10] [hipe] [kernel-poll:false]
Interactive Elixir (1.6.6) - press Ctrl+C to exit (type h() ENTER for help)

iex(1)> {:ok, server_pid} = GenServer.start_link(Gen1.Server, [])
{:ok, #PID<0.93.0>}

iex(2)> long_pid = GenServer.call(server_pid, :go_long, 15_000)
#PID<0.100.0>

iex(3)> GenServer.call(server_pid, :other)                       
:other_stuff

iex(4)> receive do                                             
...(4)> {^long_pid, reply} -> reply                            
...(4)> end                                                    
:dog

iex(7)> 

像 long_pid这样的变量将匹配任何内容。要使 long_pid仅匹配其当前值，请指定 ^long_pid( ^被称为pin运算符)。

GenServer还允许您阻止客户端对 handle_call()的调用，同时允许服务器继续执行。如果客户端在从服务器获取一些需要的数据之前无法继续运行，但是您希望服务器对其他客户端保持响应，则这很有用。这是一个例子:

defmodule Gen1.Server do
  use GenServer

  @impl true
  def init(init_state) do
    {:ok, init_state}
  end

  @impl true
  def handle_call(:go_long, from, state) do
    spawn(Gen1.Server, :long_func, [from])
    {:noreply, state}  #The server doesn't send anything to the client, 
                       #so the client's call of handle_call() blocks until 
                       #somebody calls GenServer.reply().
  end

  def long_func(from) do
    Process.sleep 10_000
    result = :dog
    GenServer.reply(from, result) 
  end

end

在IEX中:

iex(1)> {:ok, server_pid} = GenServer.start_link(Gen1.Server, [])
{:ok, #PID<0.93.0>}

iex(2)> result = GenServer.call(server_pid, :go_long, 15_000)
...hangs for 10 seconds...   
:dog

iex(3)>