fly49

fly49

How to properly stop Producer -> ConsumerSupervisor?

I am quite new to Elixir and trying to understand GenStage flow.

I created a producer that emits a finite list of numbers

defmodule Producer do
  use GenStage

  def start_link(_) do
    GenStage.start_link(Producer, Enum.to_list(1..10), name: Producer)
  end

  def init(list) do
    {:producer, list}
  end

  def handle_demand(demand, state) when demand > 0 do
    case Enum.split(state, demand) do
      {[],[]} ->
        GenStage.async_info(self(), :stop)
        {:noreply, [], []}

      {pulled, remaining} ->
        {:noreply, pulled, remaining}
    end
  end

  def handle_info(:stop, state) do
    {:stop, :normal, state}
  end
end

and ConsumerSupervisor that starts new children for each number

defmodule ConSupervisor do
  use ConsumerSupervisor

  @opts [
    strategy: :one_for_one,
    subscribe_to: [{Producer, max_demand: 2, min_demand: 0}],
    max_restarts: 20
  ]

  def start_link(arg) do
    ConsumerSupervisor.start_link(__MODULE__, arg)
  end

  def init(_arg) do
    children = [%{id: Consumer, start: {Consumer, :start_link, []}, restart: :transient}]

    ConsumerSupervisor.init(children, @opts)
  end
end

Children are just tasks with 30% rate of success

defmodule Consumer do
  def start_link(event) do
    Task.start_link(fn ->
      Process.sleep(1000)

      if :rand.uniform(3) == 3 do
        IO.puts("task failed with #{event}")
        raise("ooops")
      end

      num = Kernel.inspect(event)
      IO.puts("Inserted #{num}")
    end)
  end
end

I want to terminate Producer the way that it can be assured that all Tasks are completed. For now when it terminates it emits cancel for ConsumerSupervisor that discard all Tasks retrying to finish successfully.

[info] GenStage consumer #PID<0.433.0> is stopping after receiving cancel from producer #PID<0.431.0> with reason: :normal

I feel like it should be tricky and require some consecutive cast/call messaging, but maybe there is more straightforward way to do this?

Marked As Solved

NeutronStein

NeutronStein

As @Ljzn mentioned you can check active workers with count_children and stop if there are none active.

  def handle_info(:stop, state) do
    %{active: active} = ConsumerSupervisor.count_children(consumer_pid_or_registered_name)
    if active > 0 do
      Process.send_after(self(), :stop, 5000)
      {:noreply, [], state}
    else
      {:stop, :normal, state}
    end
  end

Also Liked

Ljzn

Ljzn

I’m not very familiar with genState. Maybe you can use ConsumerSupervisor.count_children/1 repeatedly to check is all tasks completed.

Where Next?

Popular in Questions Top

vertexbuffer
Hello, can anybody help here..? I have a list of players and I what to delete an element, but every for loop the list is reverting to ori...
New
Tee
can someone please explain to me how Enum.reduce works with maps
New
qwerescape
Is there a way to get the call stack or stack trace at any point in the code? Not from exceptions, but an expression that returns how the...
New
fireproofsocks
Forgive me if this is obvious, but how does one delete a database record WITHOUT selecting it first? Ecto.Repo — Ecto v3.14.0 has exampl...
New
Lily
In templates/appointment/index.html.eex: &lt;%= for appointment &lt;- @appointments do %&gt; &lt;tr&gt; &lt;td&gt;&lt;%= appoi...
New
nobody
Hi! In PHP: $_SERVER[‘SERVER_ADDR’] - in Elixir? Searched the docs for ip address and the web, no good results. Thanks!
New
fayddelight
I tried installing elixir 1.11.2 erlang 23.3.4 via asdf in my zsh shell. Enabled the versions locally and globally. When I list them ...
New
dblack
I’ve got an issue with an app and I’ve no idea of how to troubleshoot it. I’m hoping someone here might have seen something similar. I p...
New
PeterCarter
There are pre-rolled solutions for other frameworks that do work. However, Phoenix does not seem to have these. Have people had good expe...
New
vonH
In asking this question I am more interested about the expressiveness of the language itself and less concerned about the availability of...
New

Other popular topics Top

siddhant3030
Hi, I have to write a raw query for one of my project. But till now I have used ecto queries and don’t have much experience writing raw ...
New
lastday4you
I wanted to check elixir version in phoenix because i found that my elixir is 1.5 but when i use Enum.chunk_by it said the function is un...
New
marius95
Hello everyone, I try to use an Javascript Event Handler in my root.html.leex file. Therefore I created a function in the app.js file: ...
New
Nvim
Anybody knows a comprehensive comparison of Django and Phoenix, thanks for the help. Where are they similar? Where do they differ the m...
New
johnnyicon
Hi all, I’ve just started learning Elixir and Phoenix Framework, so please pardon my n00bness at this stage. I’m trying to use Postgres...
New
gausby
I asked this very same question on twitter and got some interesting feedback, but I thought it would be a good question to ask here as we...
1207 39467 209
New
vonH
When I run the Plug and I recompile I wind up having to use Ctrl C to quit iex and start again. Witht the help of rlwrap I can use the cu...
New
dblack
I’ve got an issue with an app and I’ve no idea of how to troubleshoot it. I’m hoping someone here might have seen something similar. I p...
New
klo
Got a question about when to concat vs. prepending items to list then reversing to achieve appending. So i know lists boil down to [1 | ...
New
jononomo
For some reason my phoenix channels are working for me in my local dev environment, but as soon as I deploy via Docker, I get a 403 error...
New

Latest on Elixir Forum

Elixir Forum

We're in Beta

About us Mission Statement