Repository: general-CbIC/poolex (elixir)

Issue:
Split interface into `run/3` and `run!/3`
- `run/3` will return `{:error, reason :: atom()}` on errors
- `run!/3` will raise on errors

Pull request:
Add `run/3` interface with error handling
## New interface

New `run/3` interface will return `{:error, reason :: any()}` instead of raising errors or throwing exit code.
`run!/3` will raise errors as earlier.

## Enhancements

- Use [DynamicSupervisor](https://hexdocs.pm/elixir/1.14.3/DynamicSupervisor.html) to manage workers
- Fix bug on workers releasing (now we check if it can be released)

Closes #14 

Patch:
diff --git a/README.md b/README.md
index ac55e3f..dc11409 100644
--- a/README.md
+++ b/README.md
@@ -152,11 +152,11 @@ process #PID<0.223.0> calculating square root of 1
 process #PID<0.225.0> calculating square root of 3
 process #PID<0.224.0> calculating square root of 2
 process #PID<0.226.0> calculating square root of 4
-1.0
-1.4142135623730951
-1.7320508075688772
-2.0
-2.23606797749979
-2.449489742783178
+{:ok, 1.0}
+{:ok, 1.4142135623730951}
+{:ok, 1.7320508075688772}
+{:ok, 2.0}
+{:ok, 2.23606797749979}
+{:ok, 2.449489742783178}
 ...
 ```
diff --git a/lib/poolex.ex b/lib/poolex.ex
index adf41ed..79d65e9 100644
--- a/lib/poolex.ex
+++ b/lib/poolex.ex
@@ -16,24 +16,36 @@ defmodule Poolex do
           | {:worker_args, list(any())}
           | {:workers_count, pos_integer()}
 
+  @spec start(pool_id(), list(poolex_option())) :: GenServer.on_start()
+  def start(pool_id, opts) do
+    GenServer.start(__MODULE__, {pool_id, opts}, name: pool_id)
+  end
+
   @spec start_link(pool_id(), list(poolex_option())) :: GenServer.on_start()
   def start_link(pool_id, opts) do
     GenServer.start_link(__MODULE__, {pool_id, opts}, name: pool_id)
   end
 
   @type run_option() :: {:timeout, timeout()}
-  @spec run(pool_id(), (worker :: pid() -> any()), list(run_option())) :: any()
+  @spec run(pool_id(), (worker :: pid() -> any()), list(run_option())) ::
+          {:ok, any()} | :all_workers_are_busy | {:runtime_error, any()}
   def run(pool_id, fun, options \\ []) do
+    {:ok, run!(pool_id, fun, options)}
+  catch
+    :exit, {:timeout, _meta} -> :all_workers_are_busy
+    :exit, reason -> {:runtime_error, reason}
+  end
+
+  @spec run!(pool_id(), (worker :: pid() -> any()), list(run_option())) :: any()
+  def run!(pool_id, fun, options \\ []) do
     timeout = Keyword.get(options, :timeout, @default_wait_timeout)
 
-    case GenServer.call(pool_id, :get_idle_worker, timeout) do
-      {:ok, pid} ->
-        result = fun.(pid)
-        GenServer.cast(pool_id, {:release_busy_worker, pid})
-        result
+    {:ok, pid} = GenServer.call(pool_id, :get_idle_worker, timeout)
 
-      error ->
-        error
+    try do
+      fun.(pid)
+    after
+      GenServer.cast(pool_id, {:release_busy_worker, pid})
     end
   end
 
@@ -50,10 +62,11 @@ defmodule Poolex do
     worker_args = Keyword.get(opts, :worker_args, [])
 
     {:ok, monitor_id} = Monitoring.init(pool_id)
+    {:ok, supervisor} = Poolex.Supervisor.start_link()
 
     worker_pids =
       Enum.map(1..workers_count, fn _ ->
-        {:ok, worker_pid} = start_worker(worker_module, worker_start_fun, worker_args)
+        {:ok, worker_pid} = start_worker(worker_module, worker_start_fun, worker_args, supervisor)
         Monitoring.add(monitor_id, worker_pid, :worker)
 
         worker_pid
@@ -65,15 +78,19 @@ defmodule Poolex do
       worker_args: worker_args,
       idle_workers_count: workers_count,
       idle_workers_pids: worker_pids,
-      monitor_id: monitor_id
+      monitor_id: monitor_id,
+      supervisor: supervisor
     }
 
     {:ok, state}
   end
 
-  @spec start_worker(module(), atom(), list(any())) :: {:ok, pid()}
-  defp start_worker(worker_module, worker_start_fun, worker_args) do
-    apply(worker_module, worker_start_fun, worker_args)
+  @spec start_worker(module(), atom(), list(any()), pid()) :: {:ok, pid()}
+  defp start_worker(worker_module, worker_start_fun, worker_args, supervisor) do
+    DynamicSupervisor.start_child(supervisor, %{
+      id: make_ref(),
+      start: {worker_module, worker_start_fun, worker_args}
+    })
   end
 
   def handle_call(
@@ -124,15 +141,18 @@ defmodule Poolex do
           waiting_callers: {[], []}
         } = state
       ) do
-    state = %State{
-      state
-      | busy_workers_count: busy_workers_count - 1,
-        busy_workers_pids: List.delete(busy_workers_pids, worker_pid),
-        idle_workers_count: idle_workers_count + 1,
-        idle_workers_pids: [worker_pid | idle_workers_pids]
-    }
-
-    {:noreply, state}
+    if Enum.member?(busy_workers_pids, worker_pid) do
+      {:noreply,
+       %State{
+         state
+         | busy_workers_count: busy_workers_count - 1,
+           busy_workers_pids: List.delete(busy_workers_pids, worker_pid),
+           idle_workers_count: idle_workers_count + 1,
+           idle_workers_pids: [worker_pid | idle_workers_pids]
+       }}
+    else
+      {:noreply, state}
+    end
   end
 
   def handle_cast(
@@ -155,12 +175,13 @@ defmodule Poolex do
           worker_module: worker_module,
           worker_start_fun: worker_start_fun,
           worker_args: worker_args,
-          waiting_callers: waiting_callers
+          waiting_callers: waiting_callers,
+          supervisor: supervisor
         } = state
       ) do
     case Monitoring.remove(monitor_id, monitoring_reference) do
       :worker ->
-        {:ok, new_worker} = start_worker(worker_module, worker_start_fun, worker_args)
+        {:ok, new_worker} = start_worker(worker_module, worker_start_fun, worker_args, supervisor)
         Monitoring.add(monitor_id, new_worker, :worker)
 
         idle_workers_pids = [new_worker | List.delete(idle_workers_pids, dead_process_pid)]
diff --git a/lib/poolex/state.ex b/lib/poolex/state.ex
index b5c7560..747b47e 100644
--- a/lib/poolex/state.ex
+++ b/lib/poolex/state.ex
@@ -8,7 +8,8 @@ defmodule Poolex.State do
             worker_start_fun: nil,
             worker_args: [],
             waiting_callers: :queue.new(),
-            monitor_id: nil
+            monitor_id: nil,
+            supervisor: nil
 
   @type t() :: %__MODULE__{
           busy_workers_count: non_neg_integer(),
@@ -19,6 +20,7 @@ defmodule Poolex.State do
           worker_start_fun: atom(),
           worker_args: list(any()),
           waiting_callers: :queue.queue(),
-          monitor_id: atom() | reference()
+          monitor_id: atom() | reference(),
+          supervisor: pid()
         }
 end
diff --git a/lib/poolex/supervisor.ex b/lib/poolex/supervisor.ex
new file mode 100644
index 0000000..0db0a90
--- /dev/null
+++ b/lib/poolex/supervisor.ex
@@ -0,0 +1,12 @@
+defmodule Poolex.Supervisor do
+  @moduledoc false
+  use DynamicSupervisor
+
+  def start_link do
+    DynamicSupervisor.start_link(__MODULE__, nil)
+  end
+
+  def init(_arg) do
+    DynamicSupervisor.init(strategy: :one_for_one)
+  end
+end


Test patch:
diff --git a/test/poolex_test.exs b/test/poolex_test.exs
index 4d397ba..c0e6b14 100644
--- a/test/poolex_test.exs
+++ b/test/poolex_test.exs
@@ -1,19 +1,21 @@
 defmodule PoolexTest do
   use ExUnit.Case
 
-  @pool_name :test_pool
+  setup do
+    [pool_name: pool_name()]
+  end
 
   describe "state" do
-    test "valid after initialization" do
+    test "valid after initialization", %{pool_name: pool_name} do
       initial_fun = fn -> 0 end
 
-      Poolex.start_link(@pool_name,
+      Poolex.start_link(pool_name,
         worker_module: Agent,
         worker_args: [initial_fun],
         workers_count: 5
       )
 
-      state = Poolex.get_state(@pool_name)
+      state = Poolex.get_state(pool_name)
 
       assert state.__struct__ == Poolex.State
       assert state.busy_workers_count == 0
@@ -25,10 +27,10 @@ defmodule PoolexTest do
       assert state.waiting_callers == :queue.new()
     end
 
-    test "valid after holding some workers" do
+    test "valid after holding some workers", %{pool_name: pool_name} do
       initial_fun = fn -> 0 end
 
-      Poolex.start_link(@pool_name,
+      Poolex.start_link(pool_name,
         worker_module: Agent,
         worker_args: [initial_fun],
         workers_count: 5
@@ -37,7 +39,7 @@ defmodule PoolexTest do
       test_process = self()
 
       spawn(fn ->
-        Poolex.run(@pool_name, fn _pid ->
+        Poolex.run(pool_name, fn _pid ->
           Process.send(test_process, nil, [])
           :timer.sleep(:timer.seconds(5))
         end)
@@ -47,7 +49,7 @@ defmodule PoolexTest do
         _message -> nil
       end
 
-      state = Poolex.get_state(@pool_name)
+      state = Poolex.get_state(pool_name)
 
       assert state.__struct__ == Poolex.State
       assert state.busy_workers_count == 1
@@ -61,44 +63,38 @@ defmodule PoolexTest do
   end
 
   describe "run/2" do
-    test "updates agent's state" do
-      Poolex.start_link(@pool_name,
+    test "updates agent's state", %{pool_name: pool_name} do
+      Poolex.start_link(pool_name,
         worker_module: Agent,
         worker_args: [fn -> 0 end],
         workers_count: 1
       )
 
-      Poolex.run(@pool_name, fn pid -> Agent.update(pid, fn _state -> 1 end) end)
+      Poolex.run(pool_name, fn pid -> Agent.update(pid, fn _state -> 1 end) end)
 
-      [agent_pid] = Poolex.get_state(@pool_name).idle_workers_pids
+      [agent_pid] = Poolex.get_state(pool_name).idle_workers_pids
 
       assert 1 == Agent.get(agent_pid, fn state -> state end)
     end
 
-    test "get result from custom worker" do
-      Poolex.start_link(@pool_name,
-        worker_module: SomeWorker,
-        worker_args: [],
-        workers_count: 2
-      )
+    test "get result from custom worker", %{pool_name: pool_name} do
+      Poolex.start_link(pool_name, worker_module: SomeWorker, workers_count: 2)
+
+      result = Poolex.run(pool_name, fn pid -> GenServer.call(pid, :do_some_work) end)
+      assert result == {:ok, :some_result}
 
-      result = Poolex.run(@pool_name, fn pid -> GenServer.call(pid, :do_some_work) end)
+      result = Poolex.run!(pool_name, fn pid -> GenServer.call(pid, :do_some_work) end)
       assert result == :some_result
     end
 
-    test "test waiting queue" do
-      Poolex.start_link(
-        @pool_name,
-        worker_module: SomeWorker,
-        worker_args: [],
-        workers_count: 5
-      )
+    test "test waiting queue", %{pool_name: pool_name} do
+      Poolex.start_link(pool_name, worker_module: SomeWorker, workers_count: 5)
 
       result =
         1..20
         |> Enum.map(fn _ ->
           Task.async(fn ->
-            Poolex.run(@pool_name, fn pid ->
+            Poolex.run!(pool_name, fn pid ->
               GenServer.call(pid, {:do_some_work_with_delay, 100})
             end)
           end)
@@ -111,27 +107,27 @@ defmodule PoolexTest do
   end
 
   describe "restarting terminated processes" do
-    test "works on idle workers" do
-      Poolex.start_link(@pool_name,
+    test "works on idle workers", %{pool_name: pool_name} do
+      Poolex.start_link(pool_name,
         worker_module: Agent,
         worker_args: [fn -> 0 end],
         workers_count: 1
       )
 
-      [agent_pid] = Poolex.get_state(@pool_name).idle_workers_pids
+      [agent_pid] = Poolex.get_state(pool_name).idle_workers_pids
 
       Process.exit(agent_pid, :kill)
 
       # To be sure that DOWN message will be handed
       :timer.sleep(1)
 
-      [new_agent_pid] = Poolex.get_state(@pool_name).idle_workers_pids
+      [new_agent_pid] = Poolex.get_state(pool_name).idle_workers_pids
 
       assert agent_pid != new_agent_pid
     end
 
-    test "works on busy workers" do
-      Poolex.start_link(@pool_name,
+    test "works on busy workers", %{pool_name: pool_name} do
+      Poolex.start_link(pool_name,
         worker_module: Agent,
         worker_args: [fn -> 0 end],
         workers_count: 1
@@ -140,7 +136,7 @@ defmodule PoolexTest do
       test_process = self()
 
       spawn(fn ->
-        Poolex.run(@pool_name, fn _pid ->
+        Poolex.run(pool_name, fn _pid ->
           Process.send(test_process, nil, [])
           :timer.sleep(:timer.seconds(5))
         end)
@@ -150,30 +146,25 @@ defmodule PoolexTest do
         _message -> nil
       end
 
-      [agent_pid] = Poolex.get_state(@pool_name).busy_workers_pids
+      [agent_pid] = Poolex.get_state(pool_name).busy_workers_pids
 
       Process.exit(agent_pid, :kill)
 
       # To be sure that DOWN message will be handed
       :timer.sleep(1)
 
-      [new_agent_pid] = Poolex.get_state(@pool_name).idle_workers_pids
+      [new_agent_pid] = Poolex.get_state(pool_name).idle_workers_pids
 
       assert agent_pid != new_agent_pid
     end
 
-    test "works on callers" do
-      Poolex.start_link(
-        @pool_name,
-        worker_module: SomeWorker,
-        worker_args: [],
-        workers_count: 1
-      )
+    test "works on callers", %{pool_name: pool_name} do
+      Poolex.start_link(pool_name, worker_module: SomeWorker, workers_count: 1)
 
       1..10
       |> Enum.each(fn _ ->
         spawn(fn ->
-          Poolex.run(@pool_name, fn pid ->
+          Poolex.run(pool_name, fn pid ->
             GenServer.call(pid, {:do_some_work_with_delay, :timer.seconds(4)})
           end)
         end)
@@ -181,14 +172,14 @@ defmodule PoolexTest do
 
       waiting_caller =
         spawn(fn ->
-          Poolex.run(@pool_name, fn pid ->
+          Poolex.run(pool_name, fn pid ->
             GenServer.call(pid, {:do_some_work_with_delay, :timer.seconds(3)})
           end)
         end)
 
       :timer.sleep(10)
 
-      state = Poolex.get_state(@pool_name)
+      state = Poolex.get_state(pool_name)
       assert :queue.len(state.waiting_callers) == 10
 
       assert Enum.find(:queue.to_list(state.waiting_callers), fn {pid, _} ->
@@ -198,26 +189,34 @@ defmodule PoolexTest do
       Process.exit(waiting_caller, :kill)
       :timer.sleep(10)
 
-      state = Poolex.get_state(@pool_name)
+      state = Poolex.get_state(pool_name)
       assert :queue.len(state.waiting_callers) == 9
 
       refute Enum.find(:queue.to_list(state.waiting_callers), fn {pid, _} ->
                pid == waiting_caller
              end)
     end
+
+    test "runtime errors", %{pool_name: pool_name} do
+      Poolex.start(pool_name, worker_module: SomeWorker, workers_count: 1)
+      Poolex.run(pool_name, fn pid -> GenServer.call(pid, :do_raise) end)
+
+      :timer.sleep(10)
+
+      state = Poolex.get_state(pool_name)
+
+      assert state.busy_workers_count == 0
+      assert state.idle_workers_count == 1
+      assert state.idle_workers_pids |> hd() |> Process.alive?()
+    end
   end
 
   describe "timeouts" do
-    test "when caller waits too long" do
-      Poolex.start_link(
-        @pool_name,
-        worker_module: SomeWorker,
-        worker_args: [],
-        workers_count: 1
-      )
+    test "when caller waits too long", %{pool_name: pool_name} do
+      Poolex.start_link(pool_name, worker_module: SomeWorker, workers_count: 1)
 
       spawn(fn ->
-        Poolex.run(@pool_name, fn pid ->
+        Poolex.run(pool_name, fn pid ->
           GenServer.call(pid, {:do_some_work_with_delay, :timer.seconds(4)})
         end)
       end)
@@ -227,7 +226,7 @@ defmodule PoolexTest do
       waiting_caller =
         spawn(fn ->
           Poolex.run(
-            @pool_name,
+            pool_name,
             fn pid ->
               GenServer.call(pid, {:do_some_work_with_delay, :timer.seconds(4)})
             end,
@@ -237,7 +236,7 @@ defmodule PoolexTest do
 
       :timer.sleep(10)
 
-      state = Poolex.get_state(@pool_name)
+      state = Poolex.get_state(pool_name)
       assert :queue.len(state.waiting_callers) == 1
 
       assert Enum.find(:queue.to_list(state.waiting_callers), fn {pid, _} ->
@@ -245,8 +244,58 @@ defmodule PoolexTest do
              end)
 
       :timer.sleep(100)
-      state = Poolex.get_state(@pool_name)
+      state = Poolex.get_state(pool_name)
       assert :queue.len(state.waiting_callers) == 0
     end
+
+    test "run/3 returns :all_workers_are_busy on timeout", %{pool_name: pool_name} do
+      Poolex.start_link(pool_name, worker_module: SomeWorker, workers_count: 1)
+
+      spawn(fn ->
+        Poolex.run(
+          pool_name,
+          fn pid -> GenServer.call(pid, {:do_some_work_with_delay, :timer.seconds(4)}) end
+        )
+      end)
+
+      :timer.sleep(10)
+
+      result =
+        Poolex.run(
+          pool_name,
+          fn pid -> GenServer.call(pid, {:do_some_work_with_delay, :timer.seconds(4)}) end,
+          timeout: 100
+        )
+
+      assert result == :all_workers_are_busy
+    end
+
+    test "run!/3 exits on timeout", %{pool_name: pool_name} do
+      Poolex.start_link(pool_name, worker_module: SomeWorker, workers_count: 1)
+
+      spawn(fn ->
+        Poolex.run(
+          pool_name,
+          fn pid -> GenServer.call(pid, {:do_some_work_with_delay, :timer.seconds(4)}) end
+        )
+      end)
+
+      :timer.sleep(10)
+
+      assert catch_exit(
+               Poolex.run!(
+                 pool_name,
+                 fn pid -> GenServer.call(pid, {:do_some_work_with_delay, :timer.seconds(4)}) end,
+                 timeout: 100
+               )
+             ) == {:timeout, {GenServer, :call, [pool_name, :get_idle_worker, 100]}}
+    end
+  end
+
+  defp pool_name do
+    1..10
+    |> Enum.map(fn _ -> Enum.random(?a..?z) end)
+    |> to_string()
+    |> String.to_atom()
   end
 end
diff --git a/test/support/some_worker.ex b/test/support/some_worker.ex
index e0c49ec..7184e5c 100644
--- a/test/support/some_worker.ex
+++ b/test/support/some_worker.ex
@@ -19,4 +19,10 @@ defmodule SomeWorker do
 
     {:reply, :some_result, state}
   end
+
+  def handle_call(:do_raise, _from, state) do
+    raise RuntimeError
+
+    {:reply, nil, state}
+  end
 end