Implementing a Worker Pool in 4 Acts

A Worker Pool Application Benjamin Tan Wei Hao @bentanweihao Erlang
User Conference 2017 Implementing or: How I finally grokked OTP 9th June 2017 in 4 acts

WHAT IS A WORKER POOL ?

WHY BUILD A WORKER POOL ?

is a lightweight, generic pooling library for Erlang with a
focus on simplicity, performance, and rock-solid disaster recovery. Poolboy < 400 lines!

1 2 Check-out Check-in

iex(1)> {:ok, pid} = ChuckFetcher.start_link(:ok) iex(2)> ChuckFetcher.fetch(pid) "Chuck Norris can
instantiate an abstract class.” Creating a Process:

iex(1)> {:ok, pid} = ChuckFetcher.start_link(:ok) iex(2)> ChuckFetcher.fetch(pid) "Chuck Norris can
instantiate an abstract class.” iex(1)> pid = Pooly.checkout("ChuckNorris") #PID<0.180.0> iex(2)> ChuckFetcher.fetch(pid) "Chuck Norris can unit test entire applications with a single assert." iex(3)> Pooly.checkin("ChuckNorris", pid) :ok Creating a Process: Checking Out & In a Process:

Version 1 & 2 Version 3 & 4

Version 1 Type of Pool Single Multiple Creation of Workers
Fixed Dynamic Consumer Recovery No Yes Worker Recovery No Yes Queueing for busy workers No Yes

REALLY NEEDED?

-module(supervisor). -behaviour(gen_server). -export([start_link/2, start_link/3, start_child/2, restart_child/2, delete_child/2, terminate_child/2, which_children/1, count_children/1,
check_childspecs/1, get_childspec/2]). -export([init/1, handle_call/3, handle_cast/2, handle_info/2, terminate/2, code_change/3, format_status/2]). -export([try_again_restart/2]).

Logic goes here?

What GOES INTO THE SERVER STATE ?

defmodule Pooly.WorkerSupervisor do use Supervisor def start_link({_,_,_} = mfa) do
Supervisor.start_link(__MODULE__, mfa) end def init({m,f,a}) do worker_opts = [restart: :permanent, shutdown: 5000, function: f] children = [worker(m, a, worker_opts)] opts = [strategy: :simple_one_for_one, max_restarts: 5, max_seconds: 5] supervise(children, opts) end end

Supervisor.start_link(__MODULE__, mfa) end def init({m,f,a}) do worker_opts = [restart: :permanent, shutdown: 5000, function: f] children = [worker(m, a, worker_opts)] opts = [strategy: :simple_one_for_one, max_restarts: 5, max_seconds: 5] supervise(children, opts) end end def start_link({_,_,_} = mfa) do Supervisor.start_link(__MODULE__, mfa) end def init({m,f,a}) do end

Supervisor.start_link(__MODULE__, mfa) end def init({m,f,a}) do worker_opts = [restart: :permanent, shutdown: 5000, function: f] children = [worker(m, a, worker_opts)] opts = [strategy: :simple_one_for_one, max_restarts: 5, max_seconds: 5] supervise(children, opts) end end worker_opts = [restart: :permanent, shutdown: 5000, function: f]

Supervisor.start_link(__MODULE__, mfa) end def init({m,f,a}) do worker_opts = [restart: :permanent, shutdown: 5000, function: f] children = [worker(m, a, worker_opts)] opts = [strategy: :simple_one_for_one, max_restarts: 5, max_seconds: 5] supervise(children, opts) end end children = [worker(m, a, worker_opts)] opts = [strategy: :simple_one_for_one, max_restarts: 5, max_seconds: 5]

Pool Server: OF THE OPERATION The

defmodule Pooly.Server do use GenServer import Supervisor.Spec defmodule State do
defstruct sup: nil, size: nil, mfa: nil end def start_link(sup, pool_config) do GenServer.start_link(__MODULE__, [sup, pool_config], name: __MODULE__) end def init([sup, pool_config]) when is_pid(sup) do init(pool_config, %{sup: sup}) end def init([{:mfa, mfa}|rest], state) do: init(rest, %{state | mfa: mfa}) def init([{:size, s}|rest], state), do: init(rest, %{state | size: s}) def init([_|rest], state), do: init(rest, state) def init([], state) do send(self, :start_worker_supervisor) {:ok, state} end end

defstruct sup: nil, size: nil, mfa: nil end def start_link(sup, pool_config) do GenServer.start_link(__MODULE__, [sup, pool_config], name: __MODULE__) end def init([sup, pool_config]) when is_pid(sup) do init(pool_config, %{sup: sup}) end def init([{:mfa, mfa}|rest], state) do: init(rest, %{state | mfa: mfa}) def init([{:size, s}|rest], state), do: init(rest, %{state | size: s}) def init([_|rest], state), do: init(rest, state) def init([], state) do send(self, :start_worker_supervisor) {:ok, state} end end def start_link(sup, pool_config) do GenServer.start_link(__MODULE__, [sup, pool_config], name: __MODULE__) end

defstruct sup: nil, size: nil, mfa: nil end def start_link(sup, pool_config) do GenServer.start_link(__MODULE__, [sup, pool_config], name: __MODULE__) end def init([sup, pool_config]) when is_pid(sup) do init(pool_config, %{sup: sup}) end def init([{:mfa, mfa}|rest], state) do: init(rest, %{state | mfa: mfa}) def init([{:size, s}|rest], state), do: init(rest, %{state | size: s}) def init([_|rest], state), do: init(rest, state) def init([], state) do send(self, :start_worker_supervisor) {:ok, state} end end def start_link(sup, pool_config) do GenServer.start_link(__MODULE__, [sup, pool_config], name: __MODULE__) end def init([sup, pool_config]) when is_pid(sup) do init(pool_config, %{sup: sup}) end

defstruct sup: nil, size: nil, mfa: nil end def start_link(sup, pool_config) do GenServer.start_link(__MODULE__, [sup, pool_config], name: __MODULE__) end def init([sup, pool_config]) when is_pid(sup) do init(pool_config, %{sup: sup}) end def init([{:mfa, mfa}|rest], state) do: init(rest, %{state | mfa: mfa}) def init([{:size, s}|rest], state), do: init(rest, %{state | size: s}) def init([_|rest], state), do: init(rest, state) def init([], state) do send(self, :start_worker_supervisor) {:ok, state} end end def init([{:mfa, mfa}|rest], state) do: init(rest, %{state | mfa: mfa}) def init([{:size, s}|rest], state), do: init(rest, %{state | size: s}) def init([_|rest], state), do: init(rest, state)

defstruct sup: nil, size: nil, mfa: nil end def start_link(sup, pool_config) do GenServer.start_link(__MODULE__, [sup, pool_config], name: __MODULE__) end def init([sup, pool_config]) when is_pid(sup) do init(pool_config, %{sup: sup}) end def init([{:mfa, mfa}|rest], state) do: init(rest, %{state | mfa: mfa}) def init([{:size, s}|rest], state), do: init(rest, %{state | size: s}) def init([_|rest], state), do: init(rest, state) def init([], state) do send(self, :start_worker_supervisor) {:ok, state} end end def init([], state) do

defstruct sup: nil, size: nil, mfa: nil end def start_link(sup, pool_config) do GenServer.start_link(__MODULE__, [sup, pool_config], name: __MODULE__) end def init([sup, pool_config]) when is_pid(sup) do init(pool_config, %{sup: sup}) end def init([{:mfa, mfa}|rest], state) do: init(rest, %{state | mfa: mfa}) def init([{:size, s}|rest], state), do: init(rest, %{state | size: s}) def init([_|rest], state), do: init(rest, state) def init([], state) do send(self, :start_worker_supervisor) {:ok, state} end end def init([], state) do send(self, :start_worker_supervisor) {:ok, state} end

defmodule Pooly.Server do defstruct State do sup: nil, worker_sup: nil,
size: nil, workers: nil, mfa: nil end def handle_info(:start_worker_supervisor, state) do %{sup: sup, mfa: mfa, size: size} = state {:ok, worker_sup} = Supervisor.start_child(sup, supervisor_spec(mfa)) workers = prepopulate(size, worker_sup) {:noreply, %{state | worker_sup: worker_sup, workers: workers}} end defp prepopulate(size, sup) when size > 1 do 1..size |> Enum.map(fn _ -> new_worker(sup) end) end defp prepopulate(_size, _sup), do: [] defp new_worker(sup) do {:ok, worker} = Supervisor.start_child(sup, [[]]) worker end defp supervisor_spec(mfa) do supervisor(Pooly.WorkerSupervisor, [mfa], [restart: :temporary]) end end

size: nil, workers: nil, mfa: nil end def handle_info(:start_worker_supervisor, state) do %{sup: sup, mfa: mfa, size: size} = state {:ok, worker_sup} = Supervisor.start_child(sup, supervisor_spec(mfa)) workers = prepopulate(size, worker_sup) {:noreply, %{state | worker_sup: worker_sup, workers: workers}} end defp prepopulate(size, sup) when size > 1 do 1..size |> Enum.map(fn _ -> new_worker(sup) end) end defp prepopulate(_size, _sup), do: [] defp new_worker(sup) do {:ok, worker} = Supervisor.start_child(sup, [[]]) worker end defp supervisor_spec(mfa) do supervisor(Pooly.WorkerSupervisor, [mfa], [restart: :temporary]) end end sup: nil %{sup: sup = state def handle_info(:start_worker_supervisor, state) do

size: nil, workers: nil, mfa: nil end def handle_info(:start_worker_supervisor, state) do %{sup: sup, mfa: mfa, size: size} = state {:ok, worker_sup} = Supervisor.start_child(sup, supervisor_spec(mfa)) workers = prepopulate(size, worker_sup) {:noreply, %{state | worker_sup: worker_sup, workers: workers}} end defp prepopulate(size, sup) when size > 1 do 1..size |> Enum.map(fn _ -> new_worker(sup) end) end defp prepopulate(_size, _sup), do: [] defp new_worker(sup) do {:ok, worker} = Supervisor.start_child(sup, [[]]) worker end defp supervisor_spec(mfa) do supervisor(Pooly.WorkerSupervisor, [mfa], [restart: :temporary]) end end sup: nil %{sup: sup = state state {:ok, worker_sup} = Supervisor.start_child(sup, supervisor_spec(mfa)) mfa: mfa defp supervisor_spec(mfa) do supervisor(Pooly.WorkerSupervisor, [mfa], [restart: :temporary]) end

size: nil, workers: nil, mfa: nil end def handle_info(:start_worker_supervisor, state) do %{sup: sup, mfa: mfa, size: size} = state {:ok, worker_sup} = Supervisor.start_child(sup, supervisor_spec(mfa)) workers = prepopulate(size, worker_sup) {:noreply, %{state | worker_sup: worker_sup, workers: workers}} end defp prepopulate(size, sup) when size > 1 do 1..size |> Enum.map(fn _ -> new_worker(sup) end) end defp prepopulate(_size, _sup), do: [] defp new_worker(sup) do {:ok, worker} = Supervisor.start_child(sup, [[]]) worker end defp supervisor_spec(mfa) do supervisor(Pooly.WorkerSupervisor, [mfa], [restart: :temporary]) end end sup: nil state {:ok, worker_sup} = Supervisor.start_child(sup, supervisor_spec(mfa)) defp supervisor_spec(mfa) do supervisor(Pooly.WorkerSupervisor, [mfa], [restart: :temporary]) end workers = prepopulate(size, worker_sup) %{sup: sup, mfa: mfa, size: size} = state

size: nil, workers: nil, mfa: nil end def handle_info(:start_worker_supervisor, state) do %{sup: sup, mfa: mfa, size: size} = state {:ok, worker_sup} = Supervisor.start_child(sup, supervisor_spec(mfa)) workers = prepopulate(size, worker_sup) {:noreply, %{state | worker_sup: worker_sup, workers: workers}} end defp prepopulate(size, sup) when size > 1 do 1..size |> Enum.map(fn _ -> new_worker(sup) end) end defp prepopulate(_size, _sup), do: [] defp new_worker(sup) do {:ok, worker} = Supervisor.start_child(sup, [[]]) worker end defp supervisor_spec(mfa) do supervisor(Pooly.WorkerSupervisor, [mfa], [restart: :temporary]) end end workers = prepopulate(size, worker_sup) defp prepopulate(size, sup) when size > 1 do 1..size |> Enum.map(fn _ -> new_worker(sup) end) end defp prepopulate(_size, _sup), do: [] defp new_worker(sup) do {:ok, worker} = Supervisor.start_child(sup, [[]]) worker end

size: nil, workers: nil, mfa: nil end def handle_info(:start_worker_supervisor, state) do %{sup: sup, mfa: mfa, size: size} = state {:ok, worker_sup} = Supervisor.start_child(sup, supervisor_spec(mfa)) workers = prepopulate(size, worker_sup) {:noreply, %{state | worker_sup: worker_sup, workers: workers}} end defp prepopulate(size, sup) when size > 1 do 1..size |> Enum.map(fn _ -> new_worker(sup) end) end defp prepopulate(_size, _sup), do: [] defp new_worker(sup) do {:ok, worker} = Supervisor.start_child(sup, [[]]) worker end defp supervisor_spec(mfa) do supervisor(Pooly.WorkerSupervisor, [mfa], [restart: :temporary]) end end defp supervisor_spec(mfa) do supervisor(Pooly.WorkerSupervisor, [mfa], [restart: :temporary]) end

Worker Check-Out defmodule Pooly.Server do def handle_call(:checkout, {from_pid, _ref}, state)
do %{workers: workers, monitors: monitors} = state case workers do [worker|rest] -> ref = Process.monitor(from_pid) true = :ets.insert(monitors, {worker, ref}) {:reply, worker, %{state | workers: rest}} [] -> {:reply, :noproc, state} end end end

Worker Check-IN defmodule Pooly.Server do def handle_cast({:checkin, worker}, state) do
%{workers: workers, monitors: monitors} = state case :ets.lookup(monitors, worker) do [{pid, ref}] -> true = Process.demonitor(ref) true = :ets.delete(monitors, pid) {:noreply, %{state | workers: [pid|workers]}} [] -> {:noreply, state} end end end

defmodule Pooly.Supervisor do use Supervisor def start_link(pool_config) do Supervisor.start_link(__MODULE__, pool_config)
end def init(pool_config) do children = [ worker(Pooly.Server, [self, pool_config]) ] opts = [strategy: :one_for_all] supervise(children, opts) end end TOp-LEVEL SUPERVISOR

defmodule Pooly.Supervisor do use Supervisor def start_link(pool_config) do Supervisor.start_link(__MODULE__, pool_config)
end def init(pool_config) do children = [ worker(Pooly.Server, [self, pool_config]) ] opts = [strategy: :one_for_all] supervise(children, opts) end end TOp-LEVEL SUPERVISOR def init(pool_config) do children = [ worker(Pooly.Server, [self, pool_config]) ] opts = [strategy: :one_for_all] supervise(children, opts) end

What happens when CONSUMER PROCESS DIES ?

def handle_info({:DOWN, ref, _, _, _}, state) do %{monitors: monitors,
workers: workers} = state case :ets.match(monitors, {:"$1", ref}) do [[pid]] -> true = :ets.delete(monitors, pid) new_state = %{state | workers: [pid|workers]} {:noreply, new_state} [[]] -> {:noreply, state} end end defmodule Pooly.Server do end

workers: workers} = state case :ets.match(monitors, {:"$1", ref}) do [[pid]] -> true = :ets.delete(monitors, pid) new_state = %{state | workers: [pid|workers]} {:noreply, new_state} [[]] -> {:noreply, state} end end defmodule Pooly.Server do end def handle_info({:DOWN, ref, _, _, _}, state) do end

workers: workers} = state case :ets.match(monitors, {:"$1", ref}) do [[pid]] -> true = :ets.delete(monitors, pid) new_state = %{state | workers: [pid|workers]} {:noreply, new_state} [[]] -> {:noreply, state} end end case :ets.match(monitors, {:"$1", ref}) do [[pid]] -> true = :ets.delete(monitors, pid) defmodule Pooly.Server do end def handle_info({:DOWN, ref, _, _, _}, state) do

workers: workers} = state case :ets.match(monitors, {:"$1", ref}) do [[pid]] -> true = :ets.delete(monitors, pid) new_state = %{state | workers: [pid|workers]} {:noreply, new_state} [[]] -> {:noreply, state} end end case :ets.match(monitors, {:"$1", ref}) do [[pid]] -> true = :ets.delete(monitors, pid) new_state = %{state | workers: [pid|workers]} {:noreply, new_state} defmodule Pooly.Server do end

workers: workers} = state case :ets.match(monitors, {:"$1", ref}) do [[pid]] -> true = :ets.delete(monitors, pid) new_state = %{state | workers: [pid|workers]} {:noreply, new_state} [[]] -> {:noreply, state} end end case :ets.match(monitors, {:"$1", ref}) do [[]] -> {:noreply, state} end defmodule Pooly.Server do end

What happens when WORKER PROCESS DIES ?

defmodule Pooly.Server do def init([sup, pool_config]) when is_pid(sup) do Process.flag(:trap_exit,
true) monitors = :ets.new(:monitors, [:private]) init(pool_config, %State{sup: sup, monitors: monitors}) end end

defmodule Pooly.Server do def init([sup, pool_config]) when is_pid(sup) do Process.flag(:trap_exit,
true) monitors = :ets.new(:monitors, [:private]) init(pool_config, %State{sup: sup, monitors: monitors}) end end Process.flag(:trap_exit, true)

def handle_info({:EXIT, pid, _reason}, state) do %{monitors: monitors, workers: workers,
worker_sup: worker_sup} = state case :ets.lookup(monitors, pid) do [{pid, ref}] -> true = Process.demonitor(ref) true = :ets.delete(monitors, pid) new_state = %{state | workers: [new_worker(worker_sup)|workers]} {:noreply, new_state} [] -> {:noreply, state} end end case :ets.lookup(monitors, pid) do [{pid, ref}] -> [] -> end def handle_info({:EXIT, pid, _reason}, state) do end

worker_sup: worker_sup} = state case :ets.lookup(monitors, pid) do [{pid, ref}] -> true = Process.demonitor(ref) true = :ets.delete(monitors, pid) new_state = %{state | workers: [new_worker(worker_sup)|workers]} {:noreply, new_state} [] -> {:noreply, state} end end case :ets.lookup(monitors, pid) do end [] -> {:noreply, state}

worker_sup: worker_sup} = state case :ets.lookup(monitors, pid) do [{pid, ref}] -> true = Process.demonitor(ref) true = :ets.delete(monitors, pid) new_state = %{state | workers: [new_worker(worker_sup)|workers]} {:noreply, new_state} [] -> {:noreply, state} end end case :ets.lookup(monitors, pid) do [{pid, ref}] -> true = Process.demonitor(ref) true = :ets.delete(monitors, pid)

worker_sup: worker_sup} = state case :ets.lookup(monitors, pid) do [{pid, ref}] -> true = Process.demonitor(ref) true = :ets.delete(monitors, pid) new_state = %{state | workers: [new_worker(worker_sup)|workers]} {:noreply, new_state} [] -> {:noreply, state} end end new_state = %{state | workers: [new_worker(worker_sup)|workers]} {:noreply, new_state} case :ets.lookup(monitors, pid) do end

ERROR RECOVERY FTW!!!

Attempt #1 What’s WRONG WITH THIS?

Attempt #1 THIS.

Attempt #2 What’s WRONG WITH THIS?

Attempt #2 THIS.

Attempt #3

DB_POOL EACH POOL GETS A NAME Redis_Pool

defmodule Pooly do use Application def start(_type, _args) do pools_config
= [ [name: "Pool1", mfa: {SampleWorker, :start_link, []}, size: 2], [name: "Pool2", mfa: {SampleWorker, :start_link, []}, size: 3], [name: "Pool3", mfa: {SampleWorker, :start_link, []}, size: 4], ] start_pools(pools_config) end end

Adding the top most top level supervisor

Adding the top level supervisor defmodule Pooly.Supervisor do use Supervisor
def start_link(pools_config) do Supervisor.start_link(__MODULE__, pools_config, name: __MODULE__) end def init(pools_config) do children = [ supervisor(Pooly.PoolsSupervisor, []), worker(Pooly.Server, [pools_config]) ] opts = [strategy: :one_for_all] supervise(children, opts) end end

def start_link(pools_config) do Supervisor.start_link(__MODULE__, pools_config, name: __MODULE__) end def init(pools_config) do children = [ supervisor(Pooly.PoolsSupervisor, []), worker(Pooly.Server, [pools_config]) ] opts = [strategy: :one_for_all] supervise(children, opts) end end name: __MODULE__

def start_link(pools_config) do Supervisor.start_link(__MODULE__, pools_config, name: __MODULE__) end def init(pools_config) do children = [ supervisor(Pooly.PoolsSupervisor, []), worker(Pooly.Server, [pools_config]) ] opts = [strategy: :one_for_all] supervise(children, opts) end end children = [ supervisor(Pooly.PoolsSupervisor, []), worker(Pooly.Server, [pools_config]) ]

def start_link(pools_config) do Supervisor.start_link(__MODULE__, pools_config, name: __MODULE__) end def init(pools_config) do children = [ supervisor(Pooly.PoolsSupervisor, []), worker(Pooly.Server, [pools_config]) ] opts = [strategy: :one_for_all] supervise(children, opts) end end opts = [strategy: :one_for_all] children = [ supervisor(Pooly.PoolsSupervisor, []), worker(Pooly.Server, [pools_config]) ]

Adding the POOLS supervisor

defmodule Pooly.PoolsSupervisor do use Supervisor def start_link do Supervisor.start_link(__MODULE__, [],
name: __MODULE__) end def init(_) do opts = [strategy: :one_for_one] supervise([], opts) end end Adding the POOLS supervisor

name: __MODULE__) end def init(_) do opts = [strategy: :one_for_one] supervise([], opts) end end Adding the POOLS supervisor name: __MODULE__

name: __MODULE__) end def init(_) do opts = [strategy: :one_for_one] supervise([], opts) end end Adding the POOLS supervisor supervise([], opts) Empty child spec!

name: __MODULE__) end def init(_) do opts = [strategy: :one_for_one] supervise([], opts) end end Adding the POOLS supervisor supervise([], opts) opts = [strategy: :one_for_one]

Making Pooly.Server Dumber

defmodule Pooly.Server do use GenServer import Supervisor.Spec def start_link(pools_config) do
GenServer.start_link(__MODULE__, pools_config, name: __MODULE__) end def checkout(pool_name) do GenServer.call(:"#{pool_name}Server", :checkout) end def checkin(pool_name, worker_pid) do GenServer.cast(:"#{pool_name}Server", {:checkin, worker_pid}) end def init(pools_config) do pools_config |> Enum.each(fn(pool_config) -> send(self, {:start_pool, pool_config}) end) {:ok, pools_config} end def handle_info({:start_pool, pool_config}, state) do {:ok, _pool_sup} = Supervisor.start_child(Pooly.PoolsSupervisor, supervisor_spec(pool_config)) {:noreply, state} end defp supervisor_spec(pool_config) do opts = [id: :"#{pool_config[:name]}Supervisor"] supervisor(Pooly.PoolSupervisor, [pool_config], opts) end end

GenServer.start_link(__MODULE__, pools_config, name: __MODULE__) end def checkout(pool_name) do GenServer.call(:"#{pool_name}Server", :checkout) end def checkin(pool_name, worker_pid) do GenServer.cast(:"#{pool_name}Server", {:checkin, worker_pid}) end def init(pools_config) do pools_config |> Enum.each(fn(pool_config) -> send(self, {:start_pool, pool_config}) end) {:ok, pools_config} end def handle_info({:start_pool, pool_config}, state) do {:ok, _pool_sup} = Supervisor.start_child(Pooly.PoolsSupervisor, supervisor_spec(pool_config)) {:noreply, state} end defp supervisor_spec(pool_config) do opts = [id: :"#{pool_config[:name]}Supervisor"] supervisor(Pooly.PoolSupervisor, [pool_config], opts) end end def checkout(pool_name) do GenServer.call(:"#{pool_name}Server", :checkout) end def checkin(pool_name, worker_pid) do GenServer.cast(:"#{pool_name}Server", {:checkin, worker_pid}) end

GenServer.start_link(__MODULE__, pools_config, name: __MODULE__) end def checkout(pool_name) do GenServer.call(:"#{pool_name}Server", :checkout) end def checkin(pool_name, worker_pid) do GenServer.cast(:"#{pool_name}Server", {:checkin, worker_pid}) end def init(pools_config) do pools_config |> Enum.each(fn(pool_config) -> send(self, {:start_pool, pool_config}) end) {:ok, pools_config} end def handle_info({:start_pool, pool_config}, state) do {:ok, _pool_sup} = Supervisor.start_child(Pooly.PoolsSupervisor, supervisor_spec(pool_config)) {:noreply, state} end defp supervisor_spec(pool_config) do opts = [id: :"#{pool_config[:name]}Supervisor"] supervisor(Pooly.PoolSupervisor, [pool_config], opts) end end def init(pools_config) do pools_config |> Enum.each(fn(pool_config) -> send(self, {:start_pool, pool_config}) end) {:ok, pools_config} end

GenServer.start_link(__MODULE__, pools_config, name: __MODULE__) end def checkout(pool_name) do GenServer.call(:"#{pool_name}Server", :checkout) end def checkin(pool_name, worker_pid) do GenServer.cast(:"#{pool_name}Server", {:checkin, worker_pid}) end def init(pools_config) do pools_config |> Enum.each(fn(pool_config) -> send(self, {:start_pool, pool_config}) end) {:ok, pools_config} end def handle_info({:start_pool, pool_config}, state) do {:ok, _pool_sup} = Supervisor.start_child(Pooly.PoolsSupervisor, supervisor_spec(pool_config)) {:noreply, state} end defp supervisor_spec(pool_config) do opts = [id: :"#{pool_config[:name]}Supervisor"] supervisor(Pooly.PoolSupervisor, [pool_config], opts) end end send(self, {:start_pool, pool_config}) def handle_info({:start_pool, pool_config}, state) do {:ok, _pool_sup} = Supervisor.start_child(Pooly.PoolsSupervisor, supervisor_spec(pool_config)) {:noreply, state} end

GenServer.start_link(__MODULE__, pools_config, name: __MODULE__) end def checkout(pool_name) do GenServer.call(:"#{pool_name}Server", :checkout) end def checkin(pool_name, worker_pid) do GenServer.cast(:"#{pool_name}Server", {:checkin, worker_pid}) end def init(pools_config) do pools_config |> Enum.each(fn(pool_config) -> send(self, {:start_pool, pool_config}) end) {:ok, pools_config} end def handle_info({:start_pool, pool_config}, state) do {:ok, _pool_sup} = Supervisor.start_child(Pooly.PoolsSupervisor, supervisor_spec(pool_config)) {:noreply, state} end defp supervisor_spec(pool_config) do opts = [id: :"#{pool_config[:name]}Supervisor"] supervisor(Pooly.PoolSupervisor, [pool_config], opts) end end send(self, {:start_pool, pool_config}) def handle_info({:start_pool, pool_config}, state) do {:ok, _pool_sup} = Supervisor.start_child(Pooly.PoolsSupervisor, supervisor_spec(pool_config)) {:noreply, state} end defp supervisor_spec(pool_config) do opts = [id: :"#{pool_config[:name]}Supervisor"] supervisor(Pooly.PoolSupervisor, [pool_config], opts) end Unique spec ID!

Adding the Pool Supervisor

defmodule Pooly.PoolSupervisor do use Supervisor def start_link(pool_config) do Supervisor.start_link(__MODULE__, pool_config,
name: :"#{pool_config[:name]}Supervisor") end def init(pool_config) do children = [ worker(Pooly.PoolServer, [self, pool_config]) ] supervise(children, strategy: :one_for_all) end end

name: :"#{pool_config[:name]}Supervisor") end def init(pool_config) do children = [ worker(Pooly.PoolServer, [self, pool_config]) ] supervise(children, strategy: :one_for_all) end end def init(pool_config) do children = [ worker(Pooly.PoolServer, [self, pool_config]) ] supervise(children, strategy: :one_for_all) end

name: :"#{pool_config[:name]}Supervisor") end def init(pool_config) do children = [ worker(Pooly.PoolServer, [self, pool_config]) ] supervise(children, strategy: :one_for_all) end end name: :"#{pool_config[:name]}Supervisor"

Implementing the pool server defmodule Pooly.PoolServer do defmodule State do
defstruct pool_sup: nil, worker_sup: nil, monitors: nil, size: nil, workers: nil, name: nil, mfa: nil end end

defstruct pool_sup: nil, worker_sup: nil, monitors: nil, size: nil, workers: nil, name: nil, mfa: nil end end pool_sup: nil, worker_sup: nil

defstruct pool_sup: nil, worker_sup: nil, monitors: nil, size: nil, workers: nil, name: nil, mfa: nil end end monitors: nil, size: nil, workers: nil, name: nil, mfa: nil

Worker Supervisor for the Pool

defmodule Pooly.WorkerSupervisor do use Supervisor def start_link(pool_server, {_,_,_} = mfa)
do Supervisor.start_link(__MODULE__, [pool_server, mfa]) end def init([pool_server, {m,f,a}]) do Process.link(pool_server) worker_opts = [restart: :temporary, shutdown: 5000, function: f] children = [worker(m, a, worker_opts)] opts = [strategy: :simple_one_for_one, max_restarts: 5, max_seconds: 5] supervise(children, opts) end end

do Supervisor.start_link(__MODULE__, [pool_server, mfa]) end def init([pool_server, {m,f,a}]) do Process.link(pool_server) worker_opts = [restart: :temporary, shutdown: 5000, function: f] children = [worker(m, a, worker_opts)] opts = [strategy: :simple_one_for_one, max_restarts: 5, max_seconds: 5] supervise(children, opts) end end def start_link(pool_server, {_,_,_} = mfa) do Supervisor.start_link(__MODULE__, [pool_server, mfa]) end

do Supervisor.start_link(__MODULE__, [pool_server, mfa]) end def init([pool_server, {m,f,a}]) do Process.link(pool_server) worker_opts = [restart: :temporary, shutdown: 5000, function: f] children = [worker(m, a, worker_opts)] opts = [strategy: :simple_one_for_one, max_restarts: 5, max_seconds: 5] supervise(children, opts) end end def start_link(pool_server, {_,_,_} = mfa) do Supervisor.start_link(__MODULE__, [pool_server, mfa]) end def init([pool_server, {m,f,a}]) do Process.link(pool_server) end

Handling a crash when worker supervisor goes down. defmodule Pooly.PoolServer
do def handle_info({:EXIT, worker_sup, reason}, state) do %{worker_sup: ^worker_sup} = state {:stop, reason, state} end end

do def handle_info({:EXIT, worker_sup, reason}, state) do %{worker_sup: ^worker_sup} = state {:stop, reason, state} end end reason {:stop, reason, state}

do def handle_info({:EXIT, worker_sup, reason}, state) do %{worker_sup: ^worker_sup} = state {:stop, reason, state} end end worker_sup %{worker_sup: ^worker_sup} = state

OVERFLOW OVERFLOW OVERFLOW IMPLEMENTING

defmodule Pooly do def start(_type, _args) do pools_config = [
[name: "ChuckNorris", mfa: {ChuckFetcher, :start_link, []}, size: 2, max_overflow: 3 ], [name: "StarWars", mfa: {SwapiFetcher, :start_link, []}, size: 4, max_overflow: 3 ] ] start_pools(pools_config) end end

defmodule Pooly.PoolServer do defmodule State do defstruct pool_sup: nil, worker_sup:
nil, monitors: nil, size: nil, workers: nil, name: nil, mfa: nil, overflow: nil, max_overflow: nil end end end

defmodule Pooly.PoolServer do defmodule State do defstruct pool_sup: nil, worker_sup:
nil, monitors: nil, size: nil, workers: nil, name: nil, mfa: nil, overflow: nil, max_overflow: nil end end end overflow: nil, max_overflow: nil

defmodule Pooly.PoolServer do def handle_call(:checkout, {from_pid, _ref} = from, state)
do %{worker_sup: worker_sup, workers: workers, monitors: monitors, overflow: overflow, max_overflow: max_overflow} = state case workers do [worker|rest] -> # ... {:reply, worker, %{state | workers: rest}} [] when max_overflow > 0 and overflow < max_overflow -> {worker, ref} = new_worker(worker_sup, from_pid) true = :ets.insert(monitors, {worker, ref}) {:reply, worker, %{state | overflow: overflow+1}} [] -> {:reply, :full, state}; end end end Overflow: Handling Worker Checkouts

do %{worker_sup: worker_sup, workers: workers, monitors: monitors, overflow: overflow, max_overflow: max_overflow} = state case workers do [worker|rest] -> # ... {:reply, worker, %{state | workers: rest}} [] when max_overflow > 0 and overflow < max_overflow -> {worker, ref} = new_worker(worker_sup, from_pid) true = :ets.insert(monitors, {worker, ref}) {:reply, worker, %{state | overflow: overflow+1}} [] -> {:reply, :full, state}; end end end Overflow: Handling Worker Checkouts defmodule Pooly.PoolServer do def handle_call(:checkout, {from_pid, _ref} = from, state) do end case workers do end [] when max_overflow > 0 and overflow < max_overflow ->

do %{worker_sup: worker_sup, workers: workers, monitors: monitors, overflow: overflow, max_overflow: max_overflow} = state case workers do [worker|rest] -> # ... {:reply, worker, %{state | workers: rest}} [] when max_overflow > 0 and overflow < max_overflow -> {worker, ref} = new_worker(worker_sup, from_pid) true = :ets.insert(monitors, {worker, ref}) {:reply, worker, %{state | overflow: overflow+1}} [] -> {:reply, :full, state}; end end end Overflow: Handling Worker Checkouts defmodule Pooly.PoolServer do def handle_call(:checkout, {from_pid, _ref} = from, state) do end case workers do end [] when max_overflow > 0 and overflow < max_overflow -> {worker, ref} = new_worker(worker_sup, from_pid) true = :ets.insert(monitors, {worker, ref})

do %{worker_sup: worker_sup, workers: workers, monitors: monitors, overflow: overflow, max_overflow: max_overflow} = state case workers do [worker|rest] -> # ... {:reply, worker, %{state | workers: rest}} [] when max_overflow > 0 and overflow < max_overflow -> {worker, ref} = new_worker(worker_sup, from_pid) true = :ets.insert(monitors, {worker, ref}) {:reply, worker, %{state | overflow: overflow+1}} [] -> {:reply, :full, state}; end end end Overflow: Handling Worker Checkouts defmodule Pooly.PoolServer do def handle_call(:checkout, {from_pid, _ref} = from, state) do end case workers do end [] when max_overflow > 0 and overflow < max_overflow -> {worker, ref} = new_worker(worker_sup, from_pid) true = :ets.insert(monitors, {worker, ref}) {:reply, worker, %{state | overflow: overflow+1}}

{:noreply, %{state | workers: [pid|workers]}} Worker dismissal: Unlink + Terminate
Child Overflow: Handling Worker CHECKINS Previously: Now:

defp handle_checkin(pid, state) do %{worker_sup: worker_sup, workers: workers, monitors: monitors,
waiting: waiting, overflow: overflow} = state if overflow > 0 do :ok = dismiss_worker(worker_sup, pid) %{state | waiting: empty, overflow: overflow-1} else %{state | waiting: empty, workers: [pid|workers], overflow: 0} end end defp dismiss_worker(sup, pid) do true = Process.unlink(pid) Supervisor.terminate_child(sup, pid) end

waiting: waiting, overflow: overflow} = state if overflow > 0 do :ok = dismiss_worker(worker_sup, pid) %{state | waiting: empty, overflow: overflow-1} else %{state | waiting: empty, workers: [pid|workers], overflow: 0} end end defp dismiss_worker(sup, pid) do true = Process.unlink(pid) Supervisor.terminate_child(sup, pid) end if overflow > 0 do :ok = dismiss_worker(worker_sup, pid) %{state | waiting: empty, overflow: overflow-1} defp dismiss_worker(sup, pid) do true = Process.unlink(pid) Supervisor.terminate_child(sup, pid) end

waiting: waiting, overflow: overflow} = state if overflow > 0 do :ok = dismiss_worker(worker_sup, pid) %{state | waiting: empty, overflow: overflow-1} else %{state | waiting: empty, workers: [pid|workers], overflow: 0} end end defp dismiss_worker(sup, pid) do true = Process.unlink(pid) Supervisor.terminate_child(sup, pid) end else %{state | waiting: empty, workers: [pid|workers], overflow: 0}

Handling Worker S

HANDLING WORKER EXITS defmodule Pooly.PoolServer do defp handle_worker_exit(pid, state) do
%{worker_sup: worker_sup, workers: workers, monitors: monitors, overflow: overflow} = state if overflow > 0 do %{state | overflow: overflow-1} else %{state | workers: [new_worker(worker_sup)|workers]} end end end defp handle_checkin(pid, state) do if overflow > 0 do :ok = dismiss_worker(worker_sup, pid) %{state | waiting: empty, overflow: overflow-1} else %{state | waiting: empty, workers: [pid|workers], overflow: 0} end end

%{worker_sup: worker_sup, workers: workers, monitors: monitors, overflow: overflow} = state if overflow > 0 do %{state | overflow: overflow-1} else %{state | workers: [new_worker(worker_sup)|workers]} end end end if overflow > 0 do %{state | overflow: overflow-1}

%{worker_sup: worker_sup, workers: workers, monitors: monitors, overflow: overflow} = state if overflow > 0 do %{state | overflow: overflow-1} else %{state | workers: [new_worker(worker_sup)|workers]} end end end else %{state | workers: [new_worker(worker_sup)|workers]}

HANDLING WORKER EXITS defmodule Pooly.PoolServer do def handle_info({:EXIT, pid, _reason},
state) do %{monitors: monitors, workers: workers, worker_sup: worker_sup} = state case :ets.lookup(monitors, pid) do [{pid, ref}] -> # ... new_state = handle_worker_exit(pid, state) {:noreply, new_state} _ -> {:noreply, state} end end end

What IF A CONSUMER PROCESS BLOCK ? IS WILLING TO

defmodule Pooly.PoolServer do defmodule State do defstruct ..., waiting: nil,
overflow: nil, max_overflow: nil end def init([pool_sup, pool_config]) when is_pid(pool_sup) do Process.flag(:trap_exit, true) monitors = :ets.new(:monitors, [:private]) waiting = :queue.new state = %State{pool_sup: pool_sup, monitors: monitors, waiting: waiting, overflow: 0} init(pool_config, state) end end

overflow: nil, max_overflow: nil end def init([pool_sup, pool_config]) when is_pid(pool_sup) do Process.flag(:trap_exit, true) monitors = :ets.new(:monitors, [:private]) waiting = :queue.new state = %State{pool_sup: pool_sup, monitors: monitors, waiting: waiting, overflow: 0} init(pool_config, state) end end defstruct ..., waiting: nil, overflow: nil, max_overflow: nil waiting = :queue.new

overflow: nil, max_overflow: nil end def init([pool_sup, pool_config]) when is_pid(pool_sup) do Process.flag(:trap_exit, true) monitors = :ets.new(:monitors, [:private]) waiting = :queue.new state = %State{pool_sup: pool_sup, monitors: monitors, waiting: waiting, overflow: 0} init(pool_config, state) end end state = %State{pool_sup: pool_sup, monitors: monitors, waiting: waiting, overflow: 0} init(pool_config, state) waiting = :queue.new defstruct ..., waiting: nil, overflow: nil, max_overflow: nil

QUEUING: CHECKING OUT def handle_call({:checkout, block}, {from_pid, _ref} = from,
state) do %{worker_sup: worker_sup, workers: workers, monitors: monitors, waiting: waiting, overflow: overflow, max_overflow: max_overflow} = state case workers do [worker|rest] -> # .. [] when max_overflow > 0 and overflow < max_overflow -> # … [] when block == true -> ref = Process.monitor(from_pid) waiting = :queue.in({from, ref}, waiting) {:noreply, %{state | waiting: waiting}, :infinity} [] -> {:reply, :full, state} end end

QUEUING: CHECKING OUT %{worker_sup: worker_sup, workers: workers, monitors: monitors, waiting:
waiting, overflow: overflow, max_overflow: max_overflow} = state def handle_call({:checkout, block}, {from_pid, _ref} = from, state) do %{worker_sup: worker_sup, workers: workers, monitors: monitors, waiting: waiting, overflow: overflow, max_overflow: max_overflow} = state case workers do [worker|rest] -> # .. [] when max_overflow > 0 and overflow < max_overflow -> # … [] when block == true -> ref = Process.monitor(from_pid) waiting = :queue.in({from, ref}, waiting) {:noreply, %{state | waiting: waiting}, :infinity} [] -> {:reply, :full, state} end end

QUEUING: CHECKING OUT {from_pid, _ref} = from [] when block
== true -> ref = Process.monitor(from_pid) waiting = :queue.in({from, ref}, waiting) {:noreply, %{state | waiting: waiting}, :infinity} case workers do end def handle_call({:checkout, block}, {from_pid, _ref} = from, state) do %{worker_sup: worker_sup, workers: workers, monitors: monitors, waiting: waiting, overflow: overflow, max_overflow: max_overflow} = state case workers do [worker|rest] -> # .. [] when max_overflow > 0 and overflow < max_overflow -> # … [] when block == true -> ref = Process.monitor(from_pid) waiting = :queue.in({from, ref}, waiting) {:noreply, %{state | waiting: waiting}, :infinity} [] -> {:reply, :full, state} end end

def handle_checkin(pid, state) do %{worker_sup: worker_sup, workers: workers, monitors: monitors,
waiting: waiting, overflow: overflow} = state case :queue.out(waiting) do {{:value, {from, ref}}, left} -> true = :ets.insert(monitors, {pid, ref}) GenServer.reply(from, pid) %{state | waiting: left} {:empty, empty} when overflow > 0 -> :ok = dismiss_worker(worker_sup, pid) %{state | waiting: empty, overflow: overflow-1} {:empty, empty} -> %{state | waiting: empty, workers: [pid|workers], overflow: 0} end end QUEUING: CHECKING IN

waiting: waiting, overflow: overflow} = state case :queue.out(waiting) do {{:value, {from, ref}}, left} -> true = :ets.insert(monitors, {pid, ref}) GenServer.reply(from, pid) %{state | waiting: left} {:empty, empty} when overflow > 0 -> :ok = dismiss_worker(worker_sup, pid) %{state | waiting: empty, overflow: overflow-1} {:empty, empty} -> %{state | waiting: empty, workers: [pid|workers], overflow: 0} end end QUEUING: CHECKING IN case :queue.out(waiting) do end

waiting: waiting, overflow: overflow} = state case :queue.out(waiting) do {{:value, {from, ref}}, left} -> true = :ets.insert(monitors, {pid, ref}) GenServer.reply(from, pid) %{state | waiting: left} {:empty, empty} when overflow > 0 -> :ok = dismiss_worker(worker_sup, pid) %{state | waiting: empty, overflow: overflow-1} {:empty, empty} -> %{state | waiting: empty, workers: [pid|workers], overflow: 0} end end QUEUING: CHECKING IN case :queue.out(waiting) do end {{:value, {from, ref}}, left} -> true = :ets.insert(monitors, {pid, ref})

waiting: waiting, overflow: overflow} = state case :queue.out(waiting) do {{:value, {from, ref}}, left} -> true = :ets.insert(monitors, {pid, ref}) GenServer.reply(from, pid) %{state | waiting: left} {:empty, empty} when overflow > 0 -> :ok = dismiss_worker(worker_sup, pid) %{state | waiting: empty, overflow: overflow-1} {:empty, empty} -> %{state | waiting: empty, workers: [pid|workers], overflow: 0} end end QUEUING: CHECKING IN case :queue.out(waiting) do end {{:value, {from, ref}}, left} -> GenServer.reply(from, pid) %{state | waiting: left}

waiting: waiting, overflow: overflow} = state case :queue.out(waiting) do {{:value, {from, ref}}, left} -> true = :ets.insert(monitors, {pid, ref}) GenServer.reply(from, pid) %{state | waiting: left} {:empty, empty} when overflow > 0 -> :ok = dismiss_worker(worker_sup, pid) %{state | waiting: empty, overflow: overflow-1} {:empty, empty} -> %{state | waiting: empty, workers: [pid|workers], overflow: 0} end end QUEUING: CHECKING IN case :queue.out(waiting) do end {:empty, empty} when overflow > 0 -> :ok = dismiss_worker(worker_sup, pid) %{state | waiting: empty, overflow: overflow-1}

waiting: waiting, overflow: overflow} = state case :queue.out(waiting) do {{:value, {from, ref}}, left} -> true = :ets.insert(monitors, {pid, ref}) GenServer.reply(from, pid) %{state | waiting: left} {:empty, empty} when overflow > 0 -> :ok = dismiss_worker(worker_sup, pid) %{state | waiting: empty, overflow: overflow-1} {:empty, empty} -> %{state | waiting: empty, workers: [pid|workers], overflow: 0} end end QUEUING: CHECKING IN case :queue.out(waiting) do end {:empty, empty} -> %{state | waiting: empty, workers: [pid|workers], overflow: 0}

ENSURE THAT EVERY CHECK-OUT IS FOLLOWED BY A CHECK-IN ?
How Do You

def transaction(pool_name, fun, timeout) do worker = checkout(pool_name, true, timeout)
try do fun.(worker) after checkin(pool_name, worker) end end TRANSACTIONS

try do fun.(worker) after checkin(pool_name, worker) end end pool_name TRANSACTIONS

try do fun.(worker) after checkin(pool_name, worker) end end fun TRANSACTIONS

try do fun.(worker) after checkin(pool_name, worker) end end timeout TRANSACTIONS

try do fun.(worker) after checkin(pool_name, worker) end end worker = checkout(pool_name, true, timeout) TRANSACTIONS

try do fun.(worker) after checkin(pool_name, worker) end end fun.(worker) TRANSACTIONS

try do fun.(worker) after checkin(pool_name, worker) end end try do fun.(worker) after checkin(pool_name, worker) end TRANSACTIONS

TRANSACTIONS tasks = 1..5 |> Enum.map(fn(_) -> Task.async(fn -> Pooly.transaction("ChuckNorris",
fn(worker_pid) -> ChuckFetcher.fetch(worker_pid) end, 5_000) end) end) tasks |> Enum.map(&Task.await(&1, 5_000))

fn(worker_pid) -> ChuckFetcher.fetch(worker_pid) end, 5_000) end) end) tasks |> Enum.map(&Task.await(&1, 5_000)) Pooly.transaction("ChuckNorris", fn(worker_pid) -> end, 5_000)

fn(worker_pid) -> ChuckFetcher.fetch(worker_pid) end, 5_000) end) end) tasks |> Enum.map(&Task.await(&1, 5_000)) Pooly.transaction("ChuckNorris", fn(worker_pid) -> ChuckFetcher.fetch(worker_pid) end, 5_000)

fn(worker_pid) -> ChuckFetcher.fetch(worker_pid) end, 5_000) end) end) tasks |> Enum.map(&Task.await(&1, 5_000)) Task.async(fn -> Pooly.transaction("ChuckNorris", fn(worker_pid) -> ChuckFetcher.fetch(worker_pid) end, 5_000) end)

fn(worker_pid) -> ChuckFetcher.fetch(worker_pid) end, 5_000) end) end) tasks |> Enum.map(&Task.await(&1, 5_000))

DEMO TIME!

Demo: Killing Workers

Demo: Killing Workers Supervisor

Demo: Killing Pool Supervisor

Demo: Killing Pools Supervisor

Demo: Killing Pools Server

Demo: Killing The Top Level Supervisor

Resources https://github.com/ benjamintanweihao/the-little- elixir-otp-guidebook-code https://github.com/devinus/poolboy THE REAL THING™

THE CHEATSHEET GenServer! Initialization def start_link(opts \\ []) do! GenServer.start_link(__MODULE__,
:ok, opts)! end! CLIENT def init(:ok) do! state = init_state()! {:ok, state}! end! CALLBACK {:ok, state} ! {ok, state, 5_000} ! {:ok, state, :hibernate}! {:stop, reason*} ! :ignore! RETURN VALUES Synchronous Operation def sync_op(pid, args) do! GenServer.call(pid, {:sync_op, args})! end! CLIENT def handle_call({:sync_op, args}, from, state) do! new_state = f(state, args)! {:reply, new_state}! end! CALLBACK {:reply, reply, new_state}! {:reply, reply, new_state, 5_000}! {:reply, reply, new_state, :hibernate}! ! {:noreply, new_state}! {:noreply, new_state, 5_000}! {:noreply, new_state, :hibernate}! {:stop, reason*, reply, new_state}! {:stop, reason*, new_state}! RETURN VAL Asynchronous Operation def async_op(pid, args) do! GenServer.cast(pid, {:async_op, args})! end! CLIENT def handle_cast({:async_op, args}, state) do! new_state = f(state, args)! {:noreply, new_state}! end! CALLBACK {:noreply, new_state}! {:noreply, new_state, 5_000}! {:noreply, new_state, :hibernate}! {:stop, reason*, new_state} ! RETURN VAL Returns {:ok, pid}! Version 1.0! Copyright © Benjamin Tan Wei Hao. Free to use without modiﬁcation for non-commercial applications. Resources http://bit.ly/genservercheatsheet http://bit.ly/supcheatsheet

Implementing a Worker Pool in 4 Acts

Implementing a Worker Pool in 4 Acts

Transcript