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(opts) do
   8     GenServer.start_link(__MODULE__, opts)
   9   end
  10
  11   @impl true
  12   def init([uri, key, opts, client_pid]) do
  13     time = :os.system_time(:second)
  14     # Register before opening connection to prevent race conditions
  15     with {:ok, _owner} <- Registry.register(@registry, key, {nil, [client_pid], 1, time}),
  16          {:ok, conn_pid} <- Gun.Conn.open(uri, opts),
  17          Process.link(conn_pid) do
  18       {_, _} =
  19         Registry.update_value(@registry, key, fn {_, used_by, crf, last_reference} ->
  20           {conn_pid, used_by, crf, last_reference}
  21         end)
  22
  23       send(client_pid, {:conn_pid, conn_pid})
  24       {:ok, %{key: key, timer: nil}, :hibernate}
  25     else
  26       err -> {:stop, err}
  27     end
  28   end
  29
  30   @impl true
  31   def handle_cast({:add_client, client_pid, send_pid_back}, %{key: key} = state) do
  32     time = :os.system_time(:second)
  33
  34     {{conn_pid, _, _, _}, _} =
  35       Registry.update_value(@registry, key, fn {conn_pid, used_by, crf, last_reference} ->
  36         {conn_pid, [client_pid | used_by], crf(time - last_reference, crf), time}
  37       end)
  38
  39     if send_pid_back, do: send(client_pid, {:conn_pid, conn_pid})
  40
  41     state =
  42       if state.timer != nil do
  43         Process.cancel_timer(state[:timer])
  44         %{state | timer: nil}
  45       else
  46         state
  47       end
  48
  49     {:noreply, state, :hibernate}
  50   end
  51
  52   @impl true
  53   def handle_cast({:remove_client, client_pid}, %{key: key} = state) do
  54     {{_conn_pid, used_by, _crf, _last_reference}, _} =
  55       Registry.update_value(@registry, key, fn {conn_pid, used_by, crf, last_reference} ->
  56         {conn_pid, List.delete(used_by, client_pid), crf, last_reference}
  57       end)
  58
  59     timer =
  60       if used_by == [] do
  61         max_idle = Pleroma.Config.get([:connections_pool, :max_idle_time], 30_000)
  62         Process.send_after(self(), :idle_close, max_idle)
  63       else
  64         nil
  65       end
  66
  67     {:noreply, %{state | timer: timer}, :hibernate}
  68   end
  69
  70   @impl true
  71   def handle_info(:idle_close, state) do
  72     # Gun monitors the owner process, and will close the connection automatically
  73     # when it's terminated
  74     {:stop, :normal, state}
  75   end
  76
  77   # Gracefully shutdown if the connection got closed without any streams left
  78   @impl true
  79   def handle_info({:gun_down, _pid, _protocol, _reason, []}, state) do
  80     {:stop, :normal, state}
  81   end
  82
  83   # Otherwise, shutdown with an error
  84   @impl true
  85   def handle_info({:gun_down, _pid, _protocol, _reason, _killed_streams} = down_message, state) do
  86     {:stop, {:error, down_message}, state}
  87   end
  88
  89   # LRFU policy: https://citeseerx.ist.psu.edu/viewdoc/summary?doi=10.1.1.55.1478
  90   defp crf(time_delta, prev_crf) do
  91     1 + :math.pow(0.5, time_delta / 100) * prev_crf
  92   end
  93 end