From a238a1254b34e57f0d337fdb11f817650e187e96 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jos=C3=A9=20Valim?= Date: Tue, 10 Oct 2017 11:40:24 +0200 Subject: [PATCH 1/3] Checkin previous connections when setting ownership mode to shared This fixes a bug where processes started before the ownership mode is set to shared holds on to the previous connection, causing confusion. --- integration_test/ownership/manager_test.exs | 22 +++++++++++ lib/db_connection/ownership/manager.ex | 41 +++++++++++++++------ 2 files changed, 51 insertions(+), 12 deletions(-) diff --git a/integration_test/ownership/manager_test.exs b/integration_test/ownership/manager_test.exs index 1b37d352..eae3fbed 100644 --- a/integration_test/ownership/manager_test.exs +++ b/integration_test/ownership/manager_test.exs @@ -197,6 +197,28 @@ defmodule ManagerTest do Task.async(fn -> assert_checked_out pool end) |> Task.await end + test "shared mode checks in previous connections" do + {:ok, agent} = A.start_link([{:ok, :state}, {:ok, :state}]) + opts = [agent: agent, parent: self(), ownership_mode: :manual, pool_size: 2] + {:ok, pool} = P.start_link(opts) + parent = self() + + task = Task.async(fn -> + assert Ownership.ownership_checkout(pool, []) == :ok + send parent, :checked_out + receive do + :shared -> refute_checked_out pool + end + end) + + assert_receive :checked_out + assert Ownership.ownership_checkout(pool, []) == :ok + assert Ownership.ownership_mode(pool, {:shared, self()}, []) == :ok + assert Ownership.ownership_checkin(pool, []) == :ok + send task.pid, :shared + Task.await(task) + end + test "shared mode can be set back to manual" do {:ok, pool} = start_pool() parent = self() diff --git a/lib/db_connection/ownership/manager.ex b/lib/db_connection/ownership/manager.ex index 610c1a44..3bb1c012 100644 --- a/lib/db_connection/ownership/manager.ex +++ b/lib/db_connection/ownership/manager.ex @@ -115,7 +115,7 @@ defmodule DBConnection.Ownership.Manager do :not_found when mode == :manual -> {:reply, :not_found, state} :not_found when mode == :auto -> - {proxy, state} = checkout(state, caller, opts) + {proxy, state} = proxy_checkout(state, caller, opts) {:reply, {:init, proxy}, state} :not_found -> {:shared, shared} = mode @@ -125,15 +125,8 @@ defmodule DBConnection.Ownership.Manager do end def handle_call(:checkin, {caller, _}, state) do - case get_and_update_in(state.checkouts, &Map.pop(&1, caller, :not_found)) do - {{:owner, ref, proxy}, state} -> - Proxy.stop(proxy, caller) - {:reply, :ok, owner_down(state, ref)} - {{:allowed, _, _}, _} -> - {:reply, :not_owner, state} - {:not_found, _} -> - {:reply, :not_found, state} - end + {reply, state} = proxy_checkin(state, caller) + {:reply, reply, state} end def handle_call({:allow, caller, allow}, _from, %{checkouts: checkouts} = state) do @@ -155,7 +148,7 @@ defmodule DBConnection.Ownership.Manager do if kind = already_checked_out(checkouts, caller) do {:reply, {:already, kind}, state} else - {proxy, state} = checkout(state, caller, opts) + {proxy, state} = proxy_checkout(state, caller, opts) {:reply, {:init, proxy}, state} end end @@ -176,7 +169,7 @@ defmodule DBConnection.Ownership.Manager do end end - defp checkout(state, caller, opts) do + defp proxy_checkout(state, caller, opts) do %{pool: pool, owner_sup: owner_sup, checkouts: checkouts, owners: owners, ets: ets, log: log} = state {:ok, proxy} = ProxySupervisor.start_owner(owner_sup, caller, pool, opts) @@ -188,6 +181,29 @@ defmodule DBConnection.Ownership.Manager do {proxy, %{state | checkouts: checkouts, owners: owners}} end + defp proxy_checkin(state, caller) do + case get_and_update_in(state.checkouts, &Map.pop(&1, caller, :not_found)) do + {{:owner, ref, proxy}, state} -> + Proxy.stop(proxy, caller) + {:ok, state |> owner_down(ref) |> unshare(ref)} + {{:allowed, _, _}, _} -> + {:not_owner, state} + {:not_found, _} -> + {:not_found, state} + end + end + + defp proxy_checkin_all(state, pid) do + Enum.reduce(state.checkouts, state, fn {key, _}, state -> + if key == pid do + state + else + {_, state} = proxy_checkin(state, key) + state + end + end) + end + defp owner_allow(%{ets: ets, log: log} = state, allow, ref, proxy) do log && Logger.log(log, fn -> [inspect(allow), " allowed on proxy " | inspect(proxy)] end) state = put_in(state.checkouts[allow], {:allowed, ref, proxy}) @@ -217,6 +233,7 @@ defmodule DBConnection.Ownership.Manager do defp share_and_reply(%{checkouts: checkouts} = state, pid) do case Map.get(checkouts, pid, :not_found) do {:owner, ref, _} -> + state = proxy_checkin_all(state, pid) {:reply, :ok, %{state | mode: {:shared, pid}, mode_ref: ref}} {:allowed, _, _} -> {:reply, :not_owner, state} From aa30330f91752b6b79edd9f0238a731d62b26d66 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jos=C3=A9=20Valim?= Date: Tue, 10 Oct 2017 11:55:48 +0200 Subject: [PATCH 2/3] Always check in all connections on mode change --- integration_test/ownership/manager_test.exs | 41 ++++++++++++++------- lib/db_connection/ownership/manager.ex | 4 ++ 2 files changed, 32 insertions(+), 13 deletions(-) diff --git a/integration_test/ownership/manager_test.exs b/integration_test/ownership/manager_test.exs index eae3fbed..8f56778d 100644 --- a/integration_test/ownership/manager_test.exs +++ b/integration_test/ownership/manager_test.exs @@ -104,23 +104,25 @@ defmodule ManagerTest do test "owner's checkout automatically with caller option" do {:ok, pool} = start_pool() - :ok = Ownership.ownership_checkout(pool, []) parent = self() + assert Ownership.ownership_mode(pool, :manual, []) + :ok = Ownership.ownership_checkout(pool, []) Task.start_link fn -> assert_checked_out pool, [caller: parent] send parent, :checkin end assert_receive :checkin - assert Ownership.ownership_mode(pool, :auto, []) + assert Ownership.ownership_mode(pool, {:shared, parent}, []) Task.start_link fn -> assert_checked_out pool, [caller: parent] send parent, :checkin end assert_receive :checkin - assert Ownership.ownership_mode(pool, {:shared, parent}, []) + assert Ownership.ownership_mode(pool, :auto, []) + :ok = Ownership.ownership_checkout(pool, []) Task.start_link fn -> assert_checked_out pool, [caller: parent] send parent, :checkin @@ -130,6 +132,26 @@ defmodule ManagerTest do assert_checked_out pool, [caller: parent] end + test "setting manual mode checks in previous connections" do + {:ok, agent} = A.start_link([{:ok, :state}, {:ok, :state}]) + opts = [agent: agent, parent: self(), ownership_mode: :auto, pool_size: 2] + {:ok, pool} = P.start_link(opts) + parent = self() + assert Ownership.ownership_mode(pool, :auto, []) == :ok + + task = Task.async(fn -> + assert_checked_out pool + send parent, :checked_out + assert_receive :manual + refute_checked_out pool + end) + + assert_receive :checked_out + assert Ownership.ownership_mode(pool, :manual, []) == :ok + send task.pid, :manual + Task.await(task) + end + test "uses ETS when the pool is named (with pid access)" do {:ok, pool} = start_pool(name: :ownership_pid_access) parent = self() @@ -206,9 +228,8 @@ defmodule ManagerTest do task = Task.async(fn -> assert Ownership.ownership_checkout(pool, []) == :ok send parent, :checked_out - receive do - :shared -> refute_checked_out pool - end + assert_receive :shared + refute_checked_out pool end) assert_receive :checked_out @@ -223,7 +244,7 @@ defmodule ManagerTest do {:ok, pool} = start_pool() parent = self() - {:ok, pid} = Task.start fn -> + Task.start fn -> assert Ownership.ownership_checkout(pool, []) == :ok assert Ownership.ownership_mode(pool, {:shared, self()}, []) == :ok send parent, :shared @@ -233,12 +254,6 @@ defmodule ManagerTest do assert_receive :shared assert_checked_out pool assert Ownership.ownership_mode(pool, :manual, []) == :ok - assert_checked_out pool - - :erlang.trace(pool, true, [:receive]) - Process.exit(pid, :shutdown) - assert_receive {:trace, ^pool, :receive, {:DOWN, _, _, _, _}} - refute_checked_out pool assert Ownership.ownership_checkout(pool, []) == :ok end diff --git a/lib/db_connection/ownership/manager.ex b/lib/db_connection/ownership/manager.ex index 3bb1c012..650e4ea8 100644 --- a/lib/db_connection/ownership/manager.ex +++ b/lib/db_connection/ownership/manager.ex @@ -98,7 +98,11 @@ defmodule DBConnection.Ownership.Manager do def handle_call({:mode, {:shared, pid}}, _from, state) do share_and_reply(state, pid) end + def handle_call({:mode, mode}, _from, %{mode: mode} = state) do + {:reply, :ok, state} + end def handle_call({:mode, mode}, _from, state) do + state = proxy_checkin_all(state, nil) {:reply, :ok, %{state | mode: mode, mode_ref: nil}} end From a13cfe14b31e0d6652b817955cdf19ce8610d01d Mon Sep 17 00:00:00 2001 From: James Fish Date: Sat, 21 Oct 2017 21:46:16 -0700 Subject: [PATCH 3/3] Strictly revoke ownership of allowed processes on proxy checkin If a process is allowed on a connection/proxy that is checked in the process is marked revoked and requires explicit checkout/allow to regain an allowance on a process. Note that shared mode is as an explicit allow, and so delays the revoking until the shared mode ends. If shared mode is enter multiple times in a row the revoke is delayed each time. Also caller: pid will continue to work. In practice this means the only new time a new failure occurs is when a process is allowed on a connection/proxy that gets checked in, it can not gain a new connection in automatic mode without explicit checkout/allow. In the same situation in manual mode the error warns about revoking but an explicit checkout/allow was required. --- integration_test/ownership/manager_test.exs | 25 ++++--- lib/db_connection/ownership.ex | 41 +++++++++++ lib/db_connection/ownership/manager.ex | 77 ++++++++++++++++----- 3 files changed, 115 insertions(+), 28 deletions(-) diff --git a/integration_test/ownership/manager_test.exs b/integration_test/ownership/manager_test.exs index 8f56778d..fbbe504b 100644 --- a/integration_test/ownership/manager_test.exs +++ b/integration_test/ownership/manager_test.exs @@ -65,8 +65,8 @@ defmodule ManagerTest do parent = self() pid = spawn_link(fn() -> - assert_receive :refute_checkout - refute_checked_out pool + assert_receive :assert_revoked + assert_revoked pool send(parent, :no_checkout) end) @@ -82,7 +82,7 @@ defmodule ManagerTest do :ok = Ownership.ownership_checkout(pool, []) - send(pid, :refute_checkout) + send(pid, :assert_revoked) assert_receive :no_checkout end @@ -99,7 +99,7 @@ defmodule ManagerTest do end assert_receive :checkin - refute_checked_out pool + assert_revoked pool end test "owner's checkout automatically with caller option" do @@ -143,7 +143,7 @@ defmodule ManagerTest do assert_checked_out pool send parent, :checked_out assert_receive :manual - refute_checked_out pool + assert_revoked pool end) assert_receive :checked_out @@ -164,7 +164,7 @@ defmodule ManagerTest do assert_checked_out pool send parent, :allowed assert_receive :checked_in - refute_checked_out pool + assert_revoked pool end assert_receive :allowed @@ -186,7 +186,7 @@ defmodule ManagerTest do assert_checked_out pool send parent, :allowed assert_receive :checked_in - refute_checked_out pool + assert_revoked pool end assert_receive :allowed @@ -229,7 +229,7 @@ defmodule ManagerTest do assert Ownership.ownership_checkout(pool, []) == :ok send parent, :checked_out assert_receive :shared - refute_checked_out pool + assert_revoked pool end) assert_receive :checked_out @@ -254,8 +254,9 @@ defmodule ManagerTest do assert_receive :shared assert_checked_out pool assert Ownership.ownership_mode(pool, :manual, []) == :ok - refute_checked_out pool + assert_revoked pool assert Ownership.ownership_checkout(pool, []) == :ok + assert_checked_out pool end test "shared mode automatically rolls back to manual on owner crash" do @@ -298,4 +299,10 @@ defmodule ManagerTest do P.run(pool, fn _ -> :ok end) end end + + defp assert_revoked(pool) do + assert_raise DBConnection.OwnershipError, ~r/revoked ownership process/, fn -> + P.run(pool, fn _ -> :ok end) + end + end end diff --git a/lib/db_connection/ownership.ex b/lib/db_connection/ownership.ex index 8ece2cb7..14b5a367 100644 --- a/lib/db_connection/ownership.ex +++ b/lib/db_connection/ownership.ex @@ -155,6 +155,47 @@ defmodule DBConnection.Ownership do of the steps above or that the owner process has crashed. """ {:error, DBConnection.OwnershipError.exception(msg)} + :revoked -> + msg = """ + revoked ownership process for #{inspect self()}. + + When using ownership, you can revoke ownership in one of three ways: + + * Allowed processes are revoked on explicit ownership checkin + * Allowed processes are revoked on owner exit + * All owners and allowed processes are revoked on changing the pool mode + + Processes are revoked in these cases because they no longer have access + to the connection they depended on, and are assumed to be in an + inconsistent state. If a process is in an inconsistent state you may + want to restart it. + + Otherwise, you must explicitly manage connections in one + of the four ways: + + * By explicitly checking out a connection + * By explicitly allowing a spawned process + * By running the pool in shared mode + * By using :caller option with allowed process + + The first two options require every new process to explicitly + check a connection out or be allowed by calling checkout or + allow respectively. Once this occurs the process is no longer + has revoked ownership. + + The third option requires a {:shared, pid} mode to be set. + If using shared mode in tests, make sure your tests are not + async. Note that when shared mode ends for that pid the + ownership will become revoked again unless further, or repeat, + action is taken. + + The fourth option requires [caller: pid] to be used when + checking out a connection from the pool. The caller process + should already be allowed on a connection. Note that when + making subsequent checkouts the ownership will still be revoked, + unless further, or repeat, action is taken. + """ + {:error, DBConnection.OwnershipError.exception(msg)} end end diff --git a/lib/db_connection/ownership/manager.ex b/lib/db_connection/ownership/manager.ex index 650e4ea8..e4d6ab43 100644 --- a/lib/db_connection/ownership/manager.ex +++ b/lib/db_connection/ownership/manager.ex @@ -48,7 +48,7 @@ defmodule DBConnection.Ownership.Manager do end @spec lookup(GenServer.server, Keyword.t) :: - {:ok, pid} | {:init, pid} | :not_found + {:ok, pid} | {:init, pid} | :not_found | :revoked def lookup(manager, opts) when is_atom(manager) do client = self() case :ets.lookup(manager, client) do @@ -82,7 +82,7 @@ defmodule DBConnection.Ownership.Manager do mode = Keyword.get(pool_opts, :ownership_mode, :auto) log = Keyword.get(pool_opts, :ownership_log, nil) {:ok, %{pool: pool, owner_sup: owner_sup, checkouts: %{}, owners: %{}, - mode: mode, mode_ref: nil, ets: ets, log: log}} + revokes: %{}, mode: mode, mode_ref: nil, ets: ets, log: log}} end def handle_call({:mode, {:shared, pid}}, _from, %{mode: {:shared, current}} = state) do @@ -107,8 +107,9 @@ defmodule DBConnection.Ownership.Manager do end def handle_call({:lookup, opts}, {pid, _}, - %{checkouts: checkouts, mode: mode} = state) do + %{checkouts: checkouts, revokes: revokes, mode: mode} = state) do caller = Keyword.get(opts, :caller, pid) + revoked? = Map.has_key?(revokes, caller) case Map.get(checkouts, caller, :not_found) do {:owner, _, proxy} -> {:reply, {:ok, proxy}, state} @@ -116,6 +117,8 @@ defmodule DBConnection.Ownership.Manager do {:reply, {:ok, proxy}, state} :not_found when caller != pid -> {:reply, :not_found, state} + :not_found when revoked? -> + {:reply, :revoked, state} :not_found when mode == :manual -> {:reply, :not_found, state} :not_found when mode == :auto -> @@ -124,7 +127,7 @@ defmodule DBConnection.Ownership.Manager do :not_found -> {:shared, shared} = mode {:owner, ref, proxy} = Map.fetch!(checkouts, shared) - {:reply, {:ok, proxy}, owner_allow(state, caller, ref, proxy)} + {:reply, {:ok, proxy}, owner_allow(state, [caller], ref, proxy)} end end @@ -139,9 +142,9 @@ defmodule DBConnection.Ownership.Manager do else case Map.get(checkouts, caller, :not_found) do {:owner, ref, proxy} -> - {:reply, :ok, owner_allow(state, allow, ref, proxy)} + {:reply, :ok, owner_allow(state, [allow], ref, proxy)} {:allowed, ref, proxy} -> - {:reply, :ok, owner_allow(state, allow, ref, proxy)} + {:reply, :ok, owner_allow(state, [allow], ref, proxy)} :not_found -> {:reply, :not_found, state} end @@ -157,8 +160,13 @@ defmodule DBConnection.Ownership.Manager do end end - def handle_info({:DOWN, ref, _, _, _}, state) do - {:noreply, state |> owner_down(ref) |> unshare(ref)} + def handle_info({:DOWN, ref, _, pid, _}, state) do + case get_and_update_in(state.revokes, &Map.pop(&1, pid)) do + {^ref, state} -> + {:noreply, state} + {nil, state} -> + {:noreply, state |> owner_down(ref) |> unshare(ref)} + end end def handle_info(_msg, state) do @@ -178,6 +186,8 @@ defmodule DBConnection.Ownership.Manager do ets: ets, log: log} = state {:ok, proxy} = ProxySupervisor.start_owner(owner_sup, caller, pool, opts) log && Logger.log(log, fn -> [inspect(caller), " owns proxy " | inspect(proxy)] end) + {mon, state} = get_and_update_in(state.revokes, &Map.pop(&1, caller)) + mon && Process.demonitor(mon, [:flush]) ref = Process.monitor(proxy) checkouts = Map.put(checkouts, caller, {:owner, ref, proxy}) owners = Map.put(owners, ref, {proxy, caller, []}) @@ -202,20 +212,45 @@ defmodule DBConnection.Ownership.Manager do if key == pid do state else - {_, state} = proxy_checkin(state, key) - state + proxy_checkin_revoke(state, key) end end) end - defp owner_allow(%{ets: ets, log: log} = state, allow, ref, proxy) do - log && Logger.log(log, fn -> [inspect(allow), " allowed on proxy " | inspect(proxy)] end) - state = put_in(state.checkouts[allow], {:allowed, ref, proxy}) - state = update_in(state.owners[ref], fn {proxy, caller, allowed} -> - {proxy, caller, [allow|List.delete(allowed, allow)]} + defp proxy_checkin_revoke(state, pid) do + case proxy_checkin(state, pid) do + {:ok, state} -> + proxy_revoke(state, pid) + {_, state} -> + state + end + end + + defp proxy_revoke(state, pid) do + ref = Process.monitor(pid) + put_in(state.revokes[pid], ref) + end + + defp proxy_allow_all(%{checkouts: checkouts, revokes: revokes} = state, owner) do + {:owner, ref, proxy} = Map.fetch!(checkouts, owner) + owner_allow(state, Map.keys(revokes), ref, proxy) + end + + defp owner_allow(%{ets: ets, log: log} = state, entries, ref, proxy) do + log && Logger.log(log, fn -> + [Enum.map_join(entries, ", ", &inspect/1), " allowed on proxy " | + inspect(proxy)] + end) + Enum.reduce(entries, state, fn allow, state -> + {mon, state} = get_and_update_in(state.revokes, &Map.pop(&1, allow)) + mon && Process.demonitor(mon, [:flush]) + state = put_in(state.checkouts[allow], {:allowed, ref, proxy}) + state = update_in(state.owners[ref], fn {proxy, caller, allowed} -> + {proxy, caller, [allow|List.delete(allowed, allow)]} + end) + ets && :ets.insert(ets, {allow, proxy}) + state end) - ets && :ets.insert(ets, {allow, proxy}) - state end defp owner_down(%{ets: ets, log: log} = state, ref) do @@ -228,7 +263,8 @@ defmodule DBConnection.Ownership.Manager do inspect(proxy)] end) ets && Enum.each(entries, &:ets.delete(ets, &1)) - update_in(state.checkouts, &Map.drop(&1, entries)) + state = update_in(state.checkouts, &Map.drop(&1, entries)) + Enum.reduce(allowed, state, &proxy_revoke(&2, &1)) {nil, state} -> state end @@ -237,7 +273,10 @@ defmodule DBConnection.Ownership.Manager do defp share_and_reply(%{checkouts: checkouts} = state, pid) do case Map.get(checkouts, pid, :not_found) do {:owner, ref, _} -> - state = proxy_checkin_all(state, pid) + state = + state + |> proxy_checkin_all(pid) + |> proxy_allow_all(pid) {:reply, :ok, %{state | mode: {:shared, pid}, mode_ref: ref}} {:allowed, _, _} -> {:reply, :not_owner, state}