erlang - 使用 Elixir Genstage 的运行时动态计算图

我希望能够在运行时动态更改计算管道，但似乎 GenStage 要求在编译时通过 subscribe_to: [...] 机制定义计算图。有没有办法创建动态计算图？例如在下面，我想在运行时在我的管道图中的“减去 7”和“减去 4”顶点之间切换。

这可以使用 GenStage 吗？我可能会有非常复杂的管道，所以我需要一个解决方案，以复杂的方式扩展到改变图形，而不是临时解决方案，例如在这种情况下，参数化要减去的整数。我希望能够添加或删除整个子树，在子树之间切换，并将节点添加到图中，包括将它们拼接到任何子树(包括主树)的中间。
请进一步引用编辑
这是最初的生产者:

defmodule GenstageTest.Producer do
  use GenStage

  def start_link(initial \\ 1) do
    GenStage.start_link(__MODULE__, initial, name: __MODULE__)
  end

  def init(counter), do: {:producer, counter}

  def handle_demand(demand, state) do
    events = Enum.to_list(state..(state + demand - 1))
    {:noreply, events, state + demand}
  end
end

这是 producer_consumers 之一:

defmodule GenstageTest.PcTimesFive do
  use GenStage

  def start_link do
    GenStage.start_link(__MODULE__, :state_doesnt_matter, name: __MODULE__)
  end

  def init(state) do
    {:producer_consumer, state, subscribe_to: [GenstageTest.PcAddOne]}
  end

  def handle_events(events, _from, state) do
    numbers =
      events
      |> Enum.map(&(&1 * 5))
    {:noreply, numbers, state}
  end
end

这是最终消费者:

defmodule GenstageTest.Consumer do
  use GenStage

  def start_link do
    GenStage.start_link(__MODULE__, :state_doesnt_matter)
  end

  def init(state) do
    {:consumer, state, subscribe_to: [GenstageTest.PcDivTwo]}
  end

  def handle_events(events, _from, state) do
    for event <- events do
      IO.inspect({self(), event, state})
    end

    # As a consumer we never emit events
    {:noreply, [], state}
  end
end

一世
它全部以 Elixir School Genstage tutorial 为模型。
所有模块和 mix.exs 可以是 found on github 。
在@AquarHEAD L 部分回答后 3 天后编辑。
我已经设法让运行时订阅工作。下面分别是一些修改后的生产者、生产者消费者和消费者:
制作人:

defmodule GenstageTest.Producer do
  use GenStage

  def start_link(initial \\ 1) do
    GenStage.start_link(__MODULE__, initial, name: __MODULE__)
  end

  def init(counter), do: {:producer, counter}

  def handle_demand(demand, state) do
    events = Enum.to_list(state..(state + demand - 1))
    {:noreply, events, state + demand}
  end

  def handle_info({:doprint}, state) do
    IO.puts "yep"
    {:noreply, [], state}
  end

  def handle_info({:cancel, sublink}, state) do
    GenStage.cancel sublink, []
    {:noreply, [], state}
  end

end

生产者_消费者:

defmodule GenstageTest.PcAddOne do
  use GenStage

  def start_link do
    GenStage.start_link(__MODULE__, :state_doesnt_matter, name: __MODULE__)
  end

  def init(state) do
    {:producer_consumer, state}
  end

  def handle_events(events, _from, state) do
    numbers =
      events
      |> Enum.map(&(&1 + 1))
    {:noreply, numbers, state}
  end
end

消费者:

defmodule GenstageTest.Consumer do
  use GenStage

  def start_link do
    GenStage.start_link(__MODULE__, :state_doesnt_matter)
  end

  def init(state) do
    {:consumer, state}
  end

  def handle_events(events, _from, state) do
    for event <- events do
      IO.inspect event
      #File.write("/home/tbrowne/scratch/output.txt",
      #  Kernel.inspect(event) <> " ", [:append])
      :timer.sleep(100)
    end

    # As a consumer we never emit events
    {:noreply, [], state}
  end
end

现在一旦这些都在 lib 目录中可用(记得将 {:gen_stage, "~> 0.11"} 添加到你的 mix.exs deps 中)，或者复制并粘贴到 IEX 中，那么以下内容将完美运行:

{:ok, p} = GenstageTest.Producer.start_link(0)
{:ok, a1} = GenstageTest.PcAddOne.start_link()
{:ok, c} = GenstageTest.Consumer.start_link()
{:ok, link1} = GenStage.sync_subscribe(a1, to: p, min_demand: 0, max_demand: 1, cancel: :transient)
{:ok, link2} = GenStage.sync_subscribe(c, to: a1, min_demand: 0, max_demand: 1, cancel: :transient)

现在的问题是，我仍然不知道如何取消订阅。有一个 cancel function ，还有一个 stop function 。例如 GenStage.stop(c) 似乎什么都不做，而我在 GenStage.cancel/3 的各种尝试只会给出错误。
总结一下，我现在需要的是能够停止某些阶段并用其他阶段替换它们。取消订阅的语法是什么，从哪里调用？文档中没有很好地解释，因为没有具体的例子。

最佳答案

您绝对可以在运行时更改管道 checkout the first example in GenStage documentation 、 you can also use the :manual mode to fine control the demand 。还有 API to cancel subscription 。我认为这些足以动态管理 GenStage 管道。

关于erlang - 使用 Elixir Genstage 的运行时动态计算图，我们在Stack Overflow上找到一个类似的问题：https://stackoverflow.com/questions/52968929/