git.squeep.com Git - akkoma/blob - lib/pleroma/gun/connection_pool/worker.ex

   1 defmodule Pleroma.Gun.ConnectionPool.Worker do
   2   alias Pleroma.Gun
   3   use GenServer, restart: :temporary
   4
   5   @registry Pleroma.Gun.ConnectionPool
   6
   7   def start_link([key | _] = opts) do
   8     GenServer.start_link(__MODULE__, opts, name: {:via, Registry, {@registry, key}})
   9   end
  10
  11   @impl true
  12   def init([key, uri, opts, client_pid]) do
  13     with {:ok, conn_pid} <- Gun.Conn.open(uri, opts),
  14          Process.link(conn_pid) do
  15       time = :erlang.monotonic_time(:millisecond)
  16
  17       {_, _} =
  18         Registry.update_value(@registry, key, fn _ ->
  19           {conn_pid, [client_pid], 1, time}
  20         end)
  21
  22       send(client_pid, {:conn_pid, conn_pid})
  23
  24       {:ok,
  25        %{key: key, timer: nil, client_monitors: %{client_pid => Process.monitor(client_pid)}},
  26        :hibernate}
  27     else
  28       err -> {:stop, err}
  29     end
  30   end
  31
  32   @impl true
  33   def handle_cast({:add_client, client_pid, send_pid_back}, %{key: key} = state) do
  34     time = :erlang.monotonic_time(:millisecond)
  35
  36     {{conn_pid, _, _, _}, _} =
  37       Registry.update_value(@registry, key, fn {conn_pid, used_by, crf, last_reference} ->
  38         {conn_pid, [client_pid | used_by], crf(time - last_reference, crf), time}
  39       end)
  40
  41     if send_pid_back, do: send(client_pid, {:conn_pid, conn_pid})
  42
  43     state =
  44       if state.timer != nil do
  45         Process.cancel_timer(state[:timer])
  46         %{state | timer: nil}
  47       else
  48         state
  49       end
  50
  51     ref = Process.monitor(client_pid)
  52
  53     state = put_in(state.client_monitors[client_pid], ref)
  54     {:noreply, state, :hibernate}
  55   end
  56
  57   @impl true
  58   def handle_cast({:remove_client, client_pid}, %{key: key} = state) do
  59     {{_conn_pid, used_by, _crf, _last_reference}, _} =
  60       Registry.update_value(@registry, key, fn {conn_pid, used_by, crf, last_reference} ->
  61         {conn_pid, List.delete(used_by, client_pid), crf, last_reference}
  62       end)
  63
  64     {ref, state} = pop_in(state.client_monitors[client_pid])
  65     Process.demonitor(ref)
  66
  67     timer =
  68       if used_by == [] do
  69         max_idle = Pleroma.Config.get([:connections_pool, :max_idle_time], 30_000)
  70         Process.send_after(self(), :idle_close, max_idle)
  71       else
  72         nil
  73       end
  74
  75     {:noreply, %{state | timer: timer}, :hibernate}
  76   end
  77
  78   @impl true
  79   def handle_info(:idle_close, state) do
  80     # Gun monitors the owner process, and will close the connection automatically
  81     # when it's terminated
  82     {:stop, :normal, state}
  83   end
  84
  85   # Gracefully shutdown if the connection got closed without any streams left
  86   @impl true
  87   def handle_info({:gun_down, _pid, _protocol, _reason, []}, state) do
  88     {:stop, :normal, state}
  89   end
  90
  91   # Otherwise, shutdown with an error
  92   @impl true
  93   def handle_info({:gun_down, _pid, _protocol, _reason, _killed_streams} = down_message, state) do
  94     {:stop, {:error, down_message}, state}
  95   end
  96
  97   @impl true
  98   def handle_info({:DOWN, _ref, :process, pid, reason}, state) do
  99     # Sometimes the client is dead before we demonitor it in :remove_client, so the message
 100     # arrives anyway
 101
 102     case state.client_monitors[pid] do
 103       nil ->
 104         {:noreply, state, :hibernate}
 105
 106       _ref ->
 107         :telemetry.execute(
 108           [:pleroma, :connection_pool, :client_death],
 109           %{client_pid: pid, reason: reason},
 110           %{key: state.key}
 111         )
 112
 113         handle_cast({:remove_client, pid}, state)
 114     end
 115   end
 116
 117   # LRFU policy: https://citeseerx.ist.psu.edu/viewdoc/summary?doi=10.1.1.55.1478
 118   defp crf(time_delta, prev_crf) do
 119     1 + :math.pow(0.5, 0.0001 * time_delta) * prev_crf
 120   end
 121 end