Implementing a Worker Pool in 4 Acts

This talk is a walkthrough on how to build a worker pool. The talk is broken up into four parts, starting with the simplest implementation leading up to the final product. Through this talk, the audience will see OTP principles demonstrated in actual code, and how GenServers and Supervisors fit together to provide concurrency and robustness.

Transcript

1. 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
4. WHAT IS A WORKER POOL ?
5. WHY BUILD A WORKER POOL ?
6. is a lightweight, generic pooling library for Erlang with a focus on simplicity, performance, and rock-solid disaster recovery. Poolboy < 400 lines!
7. 1 2 Check-out Check-in
8. 1 2 Check-out Check-in
9. iex(1)> {:ok, pid} = ChuckFetcher.start_link(:ok) iex(2)> ChuckFetcher.fetch(pid) "Chuck Norris can instantiate an abstract class.” Creating a Process:
10. iex(1)> {:ok, pid} = ChuckFetcher.start_link(:ok) iex(2)> ChuckFetcher.fetch(pid) "Chuck Norris can instantiate an abstract class.” Creating a Process:
11. 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:
12. 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:
13. 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:
14. Version 1 & 2 Version 3 & 4
15. 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
21. REALLY NEEDED?
22. -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]).
23. Logic goes here?
25. What GOES INTO THE SERVER STATE ?
27. 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
28. 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 def start_link({_,_,_} = mfa) do Supervisor.start_link(__MODULE__, mfa) end def init({m,f,a}) do end
29. 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 worker_opts = [restart: :permanent, shutdown: 5000, function: f]
30. 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 children = [worker(m, a, worker_opts)] opts = [strategy: :simple_one_for_one, max_restarts: 5, max_seconds: 5]
31. Pool Server: OF THE OPERATION The
32. 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
33. 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 def start_link(sup, pool_config) do GenServer.start_link(__MODULE__, [sup, pool_config], name: __MODULE__) end
34. 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 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
35. 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 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)
36. 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 def init([], state) do
37. 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 def init([], state) do send(self, :start_worker_supervisor) {:ok, state} end
38. 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
39. 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 sup: nil %{sup: sup = state def handle_info(:start_worker_supervisor, state) do
40. 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 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
41. 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 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
42. 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 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
43. 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 defp supervisor_spec(mfa) do supervisor(Pooly.WorkerSupervisor, [mfa], [restart: :temporary]) end
44. 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 defp supervisor_spec(mfa) do supervisor(Pooly.WorkerSupervisor, [mfa], [restart: :temporary]) end
45. 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
46. 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
47. 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
48. 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
49. 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
50. Version 2 Type of Pool Single Multiple Creation of Workers Fixed Dynamic Consumer Recovery No Yes Worker Recovery No Yes Queueing for busy workers No Yes
51. What happens when CONSUMER PROCESS DIES ?
52. 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
53. 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 def handle_info({:DOWN, ref, _, _, _}, state) do end
54. 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 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
55. 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 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
56. 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 case :ets.match(monitors, {:"$1", ref}) do [[]] -> {:noreply, state} end defmodule Pooly.Server do end
57. What happens when WORKER PROCESS DIES ?
58. 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
59. 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)
60. 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
61. 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 end [] -> {:noreply, state}
62. 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}] -> true = Process.demonitor(ref) true = :ets.delete(monitors, pid)
63. 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 new_state = %{state | workers: [new_worker(worker_sup)|workers]} {:noreply, new_state} case :ets.lookup(monitors, pid) do end
64. ERROR RECOVERY FTW!!!
65. Version 2 Type of Pool Single Multiple Creation of Workers Fixed Dynamic Consumer Recovery No Yes Worker Recovery No Yes Queueing for busy workers No Yes
66. Version 3 Type of Pool Single Multiple Creation of Workers Fixed Dynamic Consumer Recovery No Yes Worker Recovery No Yes Queueing for busy workers No Yes
67. Attempt #1 What’s WRONG WITH THIS?
68. Attempt #1 THIS.
69. Attempt #2 What’s WRONG WITH THIS?
70. Attempt #2 THIS.
71. Attempt #3
72. DB_POOL EACH POOL GETS A NAME Redis_Pool
73. 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
74. Adding the top most top level supervisor
75. 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
76. 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 name: __MODULE__
77. 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 children = [ supervisor(Pooly.PoolsSupervisor, []), worker(Pooly.Server, [pools_config]) ]
78. 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 opts = [strategy: :one_for_all] children = [ supervisor(Pooly.PoolsSupervisor, []), worker(Pooly.Server, [pools_config]) ]
79. Adding the POOLS supervisor
80. 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
81. 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__
82. 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 supervise([], opts) Empty child spec!
83. 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 supervise([], opts) opts = [strategy: :one_for_one]
84. Making Pooly.Server Dumber
85. 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
86. 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 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
87. 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 def init(pools_config) do pools_config |> Enum.each(fn(pool_config) -> send(self, {:start_pool, pool_config}) end) {:ok, pools_config} end
88. 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 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
89. 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 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!
90. Adding the Pool Supervisor
91. 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
92. 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 def init(pool_config) do children = [ worker(Pooly.PoolServer, [self, pool_config]) ] supervise(children, strategy: :one_for_all) end
93. 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"
94. 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
95. 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 pool_sup: nil, worker_sup: nil
96. 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 monitors: nil, size: nil, workers: nil, name: nil, mfa: nil
97. Worker Supervisor for the Pool
98. 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
99. 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 def start_link(pool_server, {_,_,_} = mfa) do Supervisor.start_link(__MODULE__, [pool_server, mfa]) end
100. 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 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
101. 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
102. 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 reason {:stop, reason, state}
103. 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 worker_sup %{worker_sup: ^worker_sup} = state
104. Version 3 Type of Pool Single Multiple Creation of Workers Fixed Dynamic Consumer Recovery No Yes Worker Recovery No Yes Queueing for busy workers No Yes
105. Version 4 Type of Pool Single Multiple Creation of Workers Fixed Dynamic Consumer Recovery No Yes Worker Recovery No Yes Queueing for busy workers No Yes
106. OVERFLOW OVERFLOW OVERFLOW IMPLEMENTING
107. 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
108. 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
109. 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
110. 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
111. 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 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 ->
112. 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 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})
113. 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 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}}
114. {:noreply, %{state | workers: [pid|workers]}} Worker dismissal: Unlink + Terminate Child Overflow: Handling Worker CHECKINS Previously: Now:
115. 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
116. 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 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
117. 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 else %{state | waiting: empty, workers: [pid|workers], overflow: 0}
118. Handling Worker S
119. 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
120. 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 if overflow > 0 do %{state | overflow: overflow-1}
121. 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 else %{state | workers: [new_worker(worker_sup)|workers]}
122. 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
123. What IF A CONSUMER PROCESS BLOCK ? IS WILLING TO
124. 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
125. 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 defstruct ..., waiting: nil, overflow: nil, max_overflow: nil waiting = :queue.new
126. 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 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
127. 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
128. 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
129. 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
130. 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
131. 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 case :queue.out(waiting) do end
132. 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 case :queue.out(waiting) do end {{:value, {from, ref}}, left} -> true = :ets.insert(monitors, {pid, ref})
133. 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 case :queue.out(waiting) do end {{:value, {from, ref}}, left} -> GenServer.reply(from, pid) %{state | waiting: left}
134. 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 case :queue.out(waiting) do end {:empty, empty} when overflow > 0 -> :ok = dismiss_worker(worker_sup, pid) %{state | waiting: empty, overflow: overflow-1}
135. 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 case :queue.out(waiting) do end {:empty, empty} -> %{state | waiting: empty, workers: [pid|workers], overflow: 0}
136. ENSURE THAT EVERY CHECK-OUT IS FOLLOWED BY A CHECK-IN ? How Do You
137. 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
138. def transaction(pool_name, fun, timeout) do worker = checkout(pool_name, true, timeout) try do fun.(worker) after checkin(pool_name, worker) end end pool_name TRANSACTIONS
139. def transaction(pool_name, fun, timeout) do worker = checkout(pool_name, true, timeout) try do fun.(worker) after checkin(pool_name, worker) end end fun TRANSACTIONS
140. def transaction(pool_name, fun, timeout) do worker = checkout(pool_name, true, timeout) try do fun.(worker) after checkin(pool_name, worker) end end timeout TRANSACTIONS
141. def transaction(pool_name, fun, timeout) do worker = checkout(pool_name, true, timeout) try do fun.(worker) after checkin(pool_name, worker) end end worker = checkout(pool_name, true, timeout) TRANSACTIONS
142. def transaction(pool_name, fun, timeout) do worker = checkout(pool_name, true, timeout) try do fun.(worker) after checkin(pool_name, worker) end end fun.(worker) TRANSACTIONS
143. def transaction(pool_name, fun, timeout) do worker = checkout(pool_name, true, timeout) try do fun.(worker) after checkin(pool_name, worker) end end try do fun.(worker) after checkin(pool_name, worker) end TRANSACTIONS
144. 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))
145. 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)) Pooly.transaction("ChuckNorris", fn(worker_pid) -> end, 5_000)
146. 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)) Pooly.transaction("ChuckNorris", fn(worker_pid) -> ChuckFetcher.fetch(worker_pid) end, 5_000)
147. 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)) Task.async(fn -> Pooly.transaction("ChuckNorris", fn(worker_pid) -> ChuckFetcher.fetch(worker_pid) end, 5_000) end)
148. 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))
149. DEMO TIME!
150. Demo: Killing Workers
151. Demo: Killing Workers Supervisor
152. Demo: Killing Workers Supervisor
153. Demo: Killing Pool Supervisor
154. Demo: Killing Pools Supervisor
155. Demo: Killing Pools Server
156. Demo: Killing The Top Level Supervisor
158. Resources https://github.com/ benjamintanweihao/the-little- elixir-otp-guidebook-code https://github.com/devinus/poolboy THE REAL THING™
159. 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 modification for non-commercial applications. Resources http://bit.ly/genservercheatsheet http://bit.ly/supcheatsheet