Merge branch 'release/v0.7.6' into main

.tool-versions

@@ -1,2 +1,2 @@
 erlang 26.0.2
-elixir 1.15.1-otp-26
+elixir 1.15.4-otp-26

CHANGELOG.md

@@ -7,6 +7,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [0.7.6] - 2023-08-03
+
+### Fixed
+
+- Fixed a bug with workers stuck in busy status. Added caller monitoring. [PR](https://github.com/general-CbIC/poolex/pull/56)
+
 ## [0.7.5] - 2023-07-31
 
 ### Fixed

README.md

@@ -20,7 +20,7 @@ With `poolex` you can:
   <summary>Why `poolex` instead of `poolboy`?</summary>
 
 - `poolex` is written in Elixir. This library is much more convenient to use in Elixir projects.
-- `poolboy` is a great library, but not actively maintained at the moment :crying_cat_face:![Last poolboy commit](https://img.shields.io/github/last-commit/devinus/poolboy?style=flat)
+- `poolboy` is a great library, but not actively maintained :crying_cat_face:![Last poolboy commit](https://img.shields.io/github/last-commit/devinus/poolboy?style=flat)
 
 </details>
 
@@ -87,6 +87,6 @@ A detailed description of the available configuration or examples of use can be
 
 ## Contributions
 
-If you feel something can be improved, or have any questions about certain behaviors or pieces of implementation, please feel free to file an issue. Proposed changes should be taken to issues before any PRs to avoid wasting time on code that might not be merged upstream.
+If you feel something can be improved or have any questions about specific behaviors or pieces of implementation, please feel free to file an issue. Proposed changes should be taken to issues before any PRs to save time on code that might not be merged upstream.
 
-If you are ready to make changes to the project, then please read the [Contributing guide](docs/CONTRIBUTING.md) first.
+If you are ready to change the project, please read the [Contributing guide](docs/CONTRIBUTING.md) first.

lib/poolex.ex

@@ -199,9 +199,12 @@ defmodule Poolex do
 
     {:ok, pid} = GenServer.call(pool_id, :get_idle_worker, timeout)
 
+    monitor_process = monitor_caller(pool_id, self(), pid)
+
     try do
       fun.(pid)
     after
+      Process.exit(monitor_process, :kill)
       GenServer.cast(pool_id, {:release_busy_worker, pid})
     end
   end
@@ -342,7 +345,7 @@ defmodule Poolex do
 
         {:reply, {:ok, new_worker}, %State{state | overflow: state.overflow + 1}}
       else
-        Monitoring.add(state.monitor_id, from_pid, :caller)
+        Monitoring.add(state.monitor_id, from_pid, :waiting_caller)
 
         new_callers_state =
           WaitingCallers.add(state.waiting_callers_impl, state.waiting_callers_state, caller)
@@ -406,8 +409,8 @@ defmodule Poolex do
       :worker ->
         {:noreply, handle_down_worker(state, dead_process_pid)}
 
-      :caller ->
-        {:noreply, handle_down_caller(state, dead_process_pid)}
+      :waiting_caller ->
+        {:noreply, handle_down_waiting_caller(state, dead_process_pid)}
     end
   end
 
@@ -517,8 +520,8 @@ defmodule Poolex do
     end
   end
 
-  @spec handle_down_caller(State.t(), pid()) :: State.t()
-  defp handle_down_caller(%State{} = state, dead_process_pid) do
+  @spec handle_down_waiting_caller(State.t(), pid()) :: State.t()
+  defp handle_down_waiting_caller(%State{} = state, dead_process_pid) do
     new_waiting_callers_state =
       WaitingCallers.remove_by_pid(
         state.waiting_callers_impl,
@@ -536,4 +539,17 @@ defmodule Poolex do
 
     :ok
   end
+
+  # Monitor the `caller`. Release attached worker in case of caller's death.
+  @spec monitor_caller(pool_id(), caller :: pid(), worker :: pid()) :: monitor_process :: pid()
+  defp monitor_caller(pool_id, caller, worker) do
+    spawn(fn ->
+      reference = Process.monitor(caller)
+
+      receive do
+        {:DOWN, ^reference, :process, ^caller, _reason} ->
+          GenServer.cast(pool_id, {:release_busy_worker, worker})
+      end
+    end)
+  end
 end

lib/poolex/monitoring.ex

@@ -1,7 +1,7 @@
 defmodule Poolex.Monitoring do
   @moduledoc false
   @type monitor_id() :: atom() | reference()
-  @type kind_of_process() :: :worker | :caller
+  @type kind_of_process() :: :worker | :waiting_caller
 
   @spec init(Poolex.pool_id()) :: {:ok, monitor_id()}
   @doc false

mix.exs

@@ -15,7 +15,7 @@ defmodule Poolex.MixProject do
       package: package(),
       source_url: "https://github.com/general-CbIC/poolex",
       start_permanent: Mix.env() == :prod,
-      version: "0.7.5"
+      version: "0.7.6"
     ]
   end
 

test/poolex_test.exs

@@ -237,6 +237,60 @@ defmodule PoolexTest do
              end)
     end
 
+    test "release busy worker when caller dies" do
+      pool_name = start_pool(worker_module: SomeWorker, workers_count: 2)
+
+      caller =
+        spawn(fn ->
+          Poolex.run(pool_name, fn pid ->
+            GenServer.call(pid, {:do_some_work_with_delay, :timer.seconds(4)})
+          end)
+        end)
+
+      :timer.sleep(10)
+
+      debug_info = Poolex.get_debug_info(pool_name)
+
+      assert debug_info.busy_workers_count == 1
+      assert debug_info.idle_workers_count == 1
+
+      Process.exit(caller, :kill)
+
+      :timer.sleep(10)
+
+      debug_info = Poolex.get_debug_info(pool_name)
+
+      assert debug_info.busy_workers_count == 0
+      assert debug_info.idle_workers_count == 2
+    end
+
+    test "release busy worker when caller dies (overflow case)" do
+      pool_name = start_pool(worker_module: SomeWorker, workers_count: 0, max_overflow: 2)
+
+      caller =
+        spawn(fn ->
+          Poolex.run(pool_name, fn pid ->
+            GenServer.call(pid, {:do_some_work_with_delay, :timer.seconds(4)})
+          end)
+        end)
+
+      :timer.sleep(10)
+
+      debug_info = Poolex.get_debug_info(pool_name)
+
+      assert debug_info.busy_workers_count == 1
+      assert debug_info.idle_workers_count == 0
+
+      Process.exit(caller, :kill)
+
+      :timer.sleep(10)
+
+      debug_info = Poolex.get_debug_info(pool_name)
+
+      assert debug_info.busy_workers_count == 0
+      assert debug_info.idle_workers_count == 0
+    end
+
     test "runtime errors" do
       pool_name = start_pool(worker_module: SomeWorker, workers_count: 1)
       Poolex.run(pool_name, fn pid -> GenServer.call(pid, :do_raise) end)