diff --git a/grpc/lib/grpc/client/application.ex b/grpc/lib/grpc/client/application.ex index b131a8e9..9981c938 100644 --- a/grpc/lib/grpc/client/application.ex +++ b/grpc/lib/grpc/client/application.ex @@ -3,9 +3,16 @@ defmodule GRPC.Client.Application do use Application def start(_type, _args) do + supervisor_config = Application.get_env(:grpc, __MODULE__, []) + children = [ {Registry, [keys: :unique, name: GRPC.Client.Registry]}, - {DynamicSupervisor, [name: GRPC.Client.Supervisor]} + {DynamicSupervisor, + [ + name: GRPC.Client.Supervisor, + hibernate_after: Keyword.get(supervisor_config, :hibernate_after, 15_000), + spawn_opt: [fullsweep_after: Keyword.get(supervisor_config, :fullsweep_after, 20)] + ]} ] opts = [strategy: :one_for_one, name: GRPC.Supervisor] diff --git a/grpc/lib/grpc/client/connection.ex b/grpc/lib/grpc/client/connection.ex index 1414cf24..0f4dd7b1 100644 --- a/grpc/lib/grpc/client/connection.ex +++ b/grpc/lib/grpc/client/connection.ex @@ -52,6 +52,20 @@ defmodule GRPC.Client.Connection do first establishment attempt finishes and returns `{:error, reason}` (tearing the process down) if that attempt fails. + ## Supervisor memory + + Connection start arguments (for example large default headers) can linger in + the `GRPC.Client.Supervisor` process heap after children exit. That supervisor + therefore hibernates when idle and runs fullsweep GCs more often. Tune via + config: + + config :grpc, GRPC.Client.Application, + hibernate_after: 15_000, + fullsweep_after: 20 + + * `:hibernate_after` – idle ms before the supervisor hibernates (default: `15_000`) + * `:fullsweep_after` – minor GCs between fullsweeps on the supervisor (default: `20`) + ## Target syntax The `target` argument to `connect/2` accepts URI-like strings that are resolved @@ -283,7 +297,16 @@ defmodule GRPC.Client.Connection do {:ok, pid} -> case ready_status(name, validated.resolver_target, timeout) do :ok -> - finalize_connection(name) + case finalize_connection(name) do + {:ok, %Channel{}} = result -> + result + + {:error, _reason} = error -> + # A connection that reported ready but cannot hand out a + # channel is unusable. Keeping it in the supervisor could leak memory. + :ok = DynamicSupervisor.terminate_child(GRPC.Client.Supervisor, pid) + error + end {:error, reason} -> _ = DynamicSupervisor.terminate_child(GRPC.Client.Supervisor, pid) diff --git a/grpc/test/grpc/client/connection_test.exs b/grpc/test/grpc/client/connection_test.exs index 925a136d..6360ff86 100644 --- a/grpc/test/grpc/client/connection_test.exs +++ b/grpc/test/grpc/client/connection_test.exs @@ -16,12 +16,31 @@ defmodule GRPC.Client.ConnectionTest do end end + defmodule NoPickResolver do + def resolve(_target) do + {:ok, + %{ + addresses: [%{address: "127.0.0.1", port: 50051}], + service_config: %{} + }} + end + + def init(_target, _opts) do + test_pid = Application.fetch_env!(:grpc, __MODULE__) + lb_table = Enum.find(:ets.all(), &(:ets.info(&1, :owner) == self())) + true = :ets.delete(lb_table) + send(test_pid, {:connection_started, self()}) + {:ok, nil} + end + end + setup do %{ ref: make_ref(), ip: "127.0.0.1", target: "ipv4:127.0.0.1:50051", - adapter: GRPC.Test.ClientAdapter + adapter: GRPC.Test.ClientAdapter, + supervisor_children: DynamicSupervisor.count_children(GRPC.Client.Supervisor).active } end @@ -40,6 +59,33 @@ defmodule GRPC.Client.ConnectionTest do {:ok, channel} = Connection.connect(target, adapter: adapter, name: ref) assert {:ok, ^channel} = Connection.pick_channel(%Channel{ref: ref}) + + Connection.disconnect(channel) + end + end + + describe "connect/2 - failed finalize" do + test "terminates a newly started child when no channel can be picked", %{ + ref: ref, + adapter: adapter, + supervisor_children: supervisor_children + } do + Application.put_env(:grpc, NoPickResolver, self()) + on_exit(fn -> Application.delete_env(:grpc, NoPickResolver) end) + + assert {:error, :no_connection} = + Connection.connect("test://connection", + adapter: adapter, + name: ref, + resolver: NoPickResolver + ) + + assert_receive {:connection_started, pid}, 500 + monitor_ref = Process.monitor(pid) + assert_receive {:DOWN, ^monitor_ref, :process, ^pid, _reason}, 500 + + assert DynamicSupervisor.count_children(GRPC.Client.Supervisor).active == + supervisor_children end end @@ -92,16 +138,20 @@ defmodule GRPC.Client.ConnectionTest do test "GenServer process is no longer alive after disconnect", %{ ref: ref, target: target, - adapter: adapter + adapter: adapter, + supervisor_children: supervisor_children } do {:ok, channel} = Connection.connect(target, adapter: adapter, name: ref) pid = whereis_name(ref) + ref_mon = Process.monitor(pid) {:ok, _} = Connection.disconnect(channel) - ref_mon = Process.monitor(pid) assert_receive {:DOWN, ^ref_mon, :process, ^pid, _reason}, 500 + + assert DynamicSupervisor.count_children(GRPC.Client.Supervisor).active == + supervisor_children end test "pick_channel returns {:error, :no_connection} after disconnect (persistent_term entry is erased)", @@ -114,23 +164,6 @@ defmodule GRPC.Client.ConnectionTest do end end - describe "terminate/2 - persistent_term cleanup on process kill" do - test "persistent_term entry is erased when process is killed without disconnect", %{ - ref: ref, - target: target, - adapter: adapter - } do - {:ok, channel} = Connection.connect(target, adapter: adapter, name: ref) - - pid = whereis_name(ref) - ref_mon = Process.monitor(pid) - GenServer.stop(pid, :shutdown) - assert_receive {:DOWN, ^ref_mon, :process, ^pid, :shutdown}, 500 - - assert {:error, :no_connection} = Connection.pick_channel(channel) - end - end - describe "LB ETS table lifecycle" do test "disconnect/1 exits the GenServer and the ETS table is freed", %{ ref: ref, @@ -215,6 +248,35 @@ defmodule GRPC.Client.ConnectionTest do end describe "resource leaks over repeated connect/disconnect" do + test "50 binary-heavy cycles leave supervisor memory bounded after garbage collection", %{ + target: target, + adapter: adapter, + supervisor_children: supervisor_children + } do + supervisor = Process.whereis(GRPC.Client.Supervisor) + :erlang.garbage_collect(supervisor) + {:memory, before_memory} = Process.info(supervisor, :memory) + + for cycle <- 1..50 do + ref = make_ref() + header = {"x-test-data", :binary.copy(<>, 100_000)} + + {:ok, channel} = + Connection.connect(target, adapter: adapter, name: ref, headers: [header]) + + {:ok, _} = Connection.disconnect(channel) + end + + :erlang.garbage_collect(supervisor) + {:memory, after_memory} = Process.info(supervisor, :memory) + + assert %{active: ^supervisor_children} = + DynamicSupervisor.count_children(GRPC.Client.Supervisor) + + assert after_memory <= before_memory + 100_000, + "supervisor memory grew: before=#{before_memory} after=#{after_memory}" + end + test "500 cycles leave persistent_term clean and no per-LB tables leak", %{ target: target, adapter: adapter