Elixir - GenServer with initial state of several other GenServer packages
I am facing a deadlock issue with my watch tree starting automatically. The initial state of one of the GenServers is the child workers of the other manager in the tree. Here's the code:
Supervisor and employees:
defmodule Parallel.Worker.Supervisor do
import Supervisor.Spec
def start_link do
# Start up a worker for each core
schedulers = :erlang.system_info(:schedulers)
children = Enum.map(1..schedulers,
&(worker(Parallel.Worker.Server, [], id: "Parallel.Worker#{&1}")))
opts = [strategy: :one_for_one, name: Parallel.Worker.Supervisor]
Supervisor.start_link(children, opts)
end
def workers do
Process.whereis(Parallel.Supervisor)
|> Supervisor.which_children
|> Enum.reduce [], fn
{_name, pid, :worker, _module}, acc -> [{make_ref, pid} | acc]
_, acc -> acc
end
end
end
GenServer with the state of these working pids:
defmodule Parallel.Process.Server do
use GenServer
def start_link do
GenServer.start_link(__MODULE__, workers: [Parallel.Worker.Supervisor.workers])
end
end
As you can see in the last line, I call "Parallel.Worker.Supervisor.workers", which seems to block waiting for the tree to initialize, which will not complete until this method returns. How can you monitor the PID as the initial state of the GenServer?
UPDATE:
I didn't want to use pulboy (although it's a good suggestion to look at the source) to help me find out more. I am not trying to do anything in particular, my worker just processes the passed function with its arguments. Heres a working GenServer:
defmodule Parallel.Worker do
use GenServer
require Logger
def start_link(state) do
GenServer.start_link(__MODULE__, state, [])
end
def init(state) do
{:ok, state}
end
# Using cast to be async as the job could take longer than the default 5 seconds,
# Don't want client blocked on waiting for job to complete
def handle_cast({:execute, fun, args, return_pid, job_ref}, state) do
Logger.debug fn()-> "#{inspect self}: Recevied job with args: #{inspect args} for job #{inspect job_ref} to return to #{inspect return_pid}" end
send(return_pid, {job_ref, apply(fun, args), self})
{:noreply, state}
end
end
source to share
I am assuming you want to create some sort of pool here? As mentioned in the comments, you should learn pulboy. If, for the sake of practice, you want to implement this yourself, it's still worth looking at the pooling code for inspiration.
Basically, the pool pool is managed by the pool manager, gen_server, which maintains a collection of known workers. This pool manager process internally starts the manager simple_one_for_one
, which is then used to start and monitor workers.
The pool manager process, during initialization, starts the manager first . It then invokes the controlled workflowsprepopulate/1
to run . This function will dynamically create N workers via supervisor: start_child / 2 and the pool manager can keep a list of worker pids internally.
This ensures that the pool manager process does not have to talk to the parent supervisor during initialization (which is what causes your deadlock). Instead, the manager creates the children themselves. Relying on an internal leader still ensures that workers are in the oversight tree.
There are still some small-format details needed to ensure proper functioning. The Poolboy process (pool manager) will capture outputs, communicate with monitors, and monitor workers when they are checked out. This ensures proper detection of worker failures. I suggest reading the code for further analysis.
My opinion is that this might be an interesting exercise to better understand OTP. However, if you are doing this for production you are probably better off using poolboy directly.
source to share