git.squeep.com Git - akkoma/blob - lib/pleroma/pool/connections.ex

   1 # Pleroma: A lightweight social networking server
   2 # Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
   3 # SPDX-License-Identifier: AGPL-3.0-only
   4
   5 defmodule Pleroma.Pool.Connections do
   6   use GenServer
   7
   8   alias Pleroma.Config
   9
  10   require Logger
  11
  12   @type domain :: String.t()
  13   @type conn :: Pleroma.Gun.Conn.t()
  14
  15   @type t :: %__MODULE__{
  16           conns: %{domain() => conn()},
  17           opts: keyword()
  18         }
  19
  20   defstruct conns: %{}, opts: []
  21
  22   alias Pleroma.Gun.API
  23
  24   @spec start_link({atom(), keyword()}) :: {:ok, pid()}
  25   def start_link({name, opts}) do
  26     GenServer.start_link(__MODULE__, opts, name: name)
  27   end
  28
  29   @impl true
  30   def init(opts), do: {:ok, %__MODULE__{conns: %{}, opts: opts}}
  31
  32   @spec checkin(String.t() | URI.t(), atom()) :: pid() | nil
  33   def checkin(url, name)
  34   def checkin(url, name) when is_binary(url), do: checkin(URI.parse(url), name)
  35
  36   def checkin(%URI{} = uri, name) do
  37     timeout = Config.get([:connections_pool, :receive_connection_timeout], 250)
  38
  39     GenServer.call(
  40       name,
  41       {:checkin, uri},
  42       timeout
  43     )
  44   end
  45
  46   @spec alive?(atom()) :: boolean()
  47   def alive?(name) do
  48     pid = Process.whereis(name)
  49     if pid, do: Process.alive?(pid), else: false
  50   end
  51
  52   @spec get_state(atom()) :: t()
  53   def get_state(name) do
  54     GenServer.call(name, :state)
  55   end
  56
  57   @spec count(atom()) :: pos_integer()
  58   def count(name) do
  59     GenServer.call(name, :count)
  60   end
  61
  62   @spec get_unused_conns(atom()) :: [{domain(), conn()}]
  63   def get_unused_conns(name) do
  64     GenServer.call(name, :unused_conns)
  65   end
  66
  67   @spec checkout(pid(), pid(), atom()) :: :ok
  68   def checkout(conn, pid, name) do
  69     GenServer.cast(name, {:checkout, conn, pid})
  70   end
  71
  72   @spec add_conn(atom(), String.t(), Pleroma.Gun.Conn.t()) :: :ok
  73   def add_conn(name, key, conn) do
  74     GenServer.cast(name, {:add_conn, key, conn})
  75   end
  76
  77   @spec remove_conn(atom(), String.t()) :: :ok
  78   def remove_conn(name, key) do
  79     GenServer.cast(name, {:remove_conn, key})
  80   end
  81
  82   @impl true
  83   def handle_cast({:add_conn, key, conn}, state) do
  84     state = put_in(state.conns[key], conn)
  85
  86     Process.monitor(conn.conn)
  87     {:noreply, state}
  88   end
  89
  90   @impl true
  91   def handle_cast({:checkout, conn_pid, pid}, state) do
  92     Logger.debug("checkout #{inspect(conn_pid)}")
  93
  94     state =
  95       with true <- Process.alive?(conn_pid),
  96            {key, conn} <- find_conn(state.conns, conn_pid),
  97            used_by <- List.keydelete(conn.used_by, pid, 0) do
  98         conn_state =
  99           if used_by == [] do
 100             :idle
 101           else
 102             conn.conn_state
 103           end
 104
 105         put_in(state.conns[key], %{conn | conn_state: conn_state, used_by: used_by})
 106       else
 107         false ->
 108           Logger.debug("checkout for closed conn #{inspect(conn_pid)}")
 109           state
 110
 111         nil ->
 112           Logger.debug("checkout for alive conn #{inspect(conn_pid)}, but is not in state")
 113           state
 114       end
 115
 116     {:noreply, state}
 117   end
 118
 119   @impl true
 120   def handle_cast({:remove_conn, key}, state) do
 121     state = put_in(state.conns, Map.delete(state.conns, key))
 122     {:noreply, state}
 123   end
 124
 125   @impl true
 126   def handle_call({:checkin, uri}, from, state) do
 127     key = "#{uri.scheme}:#{uri.host}:#{uri.port}"
 128     Logger.debug("checkin #{key}")
 129
 130     case state.conns[key] do
 131       %{conn: conn, gun_state: :up} = current_conn ->
 132         Logger.debug("reusing conn #{key}")
 133
 134         with time <- :os.system_time(:second),
 135              last_reference <- time - current_conn.last_reference,
 136              current_crf <- crf(last_reference, 100, current_conn.crf),
 137              state <-
 138                put_in(state.conns[key], %{
 139                  current_conn
 140                  | last_reference: time,
 141                    crf: current_crf,
 142                    conn_state: :active,
 143                    used_by: [from | current_conn.used_by]
 144                }) do
 145           {:reply, conn, state}
 146         end
 147
 148       %{gun_state: :down} ->
 149         {:reply, nil, state}
 150
 151       nil ->
 152         {:reply, nil, state}
 153     end
 154   end
 155
 156   @impl true
 157   def handle_call(:state, _from, state), do: {:reply, state, state}
 158
 159   @impl true
 160   def handle_call(:count, _from, state) do
 161     {:reply, Enum.count(state.conns), state}
 162   end
 163
 164   @impl true
 165   def handle_call(:unused_conns, _from, state) do
 166     unused_conns =
 167       state.conns
 168       |> Enum.filter(fn {_k, v} ->
 169         v.conn_state == :idle and v.used_by == []
 170       end)
 171       |> Enum.sort(fn {_x_k, x}, {_y_k, y} ->
 172         x.crf <= y.crf and x.last_reference <= y.last_reference
 173       end)
 174
 175     {:reply, unused_conns, state}
 176   end
 177
 178   @impl true
 179   def handle_info({:gun_up, conn_pid, _protocol}, state) do
 180     state =
 181       with conn_key when is_binary(conn_key) <- compose_key_gun_info(conn_pid),
 182            {key, conn} <- find_conn(state.conns, conn_pid, conn_key),
 183            {true, key} <- {Process.alive?(conn_pid), key} do
 184         time = :os.system_time(:second)
 185         last_reference = time - conn.last_reference
 186         current_crf = crf(last_reference, 100, conn.crf)
 187         put_in(state.conns[key], %{
 188           conn
 189           | gun_state: :up,
 190             last_reference: time,
 191             crf: current_crf,
 192             conn_state: :active,
 193             retries: 0
 194         })
 195       else
 196         :error_gun_info ->
 197           Logger.debug(":gun.info caused error")
 198           state
 199
 200         {false, key} ->
 201           Logger.debug(":gun_up message for closed conn #{inspect(conn_pid)}")
 202
 203           put_in(
 204             state.conns,
 205             Map.delete(state.conns, key)
 206           )
 207
 208         nil ->
 209           Logger.debug(":gun_up message for conn which is not found in state")
 210
 211           :ok = API.close(conn_pid)
 212
 213           state
 214       end
 215
 216     {:noreply, state}
 217   end
 218
 219   @impl true
 220   def handle_info({:gun_down, conn_pid, _protocol, _reason, _killed}, state) do
 221     retries = Config.get([:connections_pool, :retry], 0)
 222     # we can't get info on this pid, because pid is dead
 223     state =
 224       with {key, conn} <- find_conn(state.conns, conn_pid),
 225            {true, key} <- {Process.alive?(conn_pid), key} do
 226         if conn.retries == retries do
 227           Logger.debug("closing conn if retries is eq  #{inspect(conn_pid)}")
 228           :ok = API.close(conn.conn)
 229
 230           put_in(
 231             state.conns,
 232             Map.delete(state.conns, key)
 233           )
 234         else
 235           put_in(state.conns[key], %{
 236             conn
 237             | gun_state: :down,
 238               retries: conn.retries + 1
 239           })
 240         end
 241       else
 242         {false, key} ->
 243           # gun can send gun_down for closed conn, maybe connection is not closed yet
 244           Logger.debug(":gun_down message for closed conn #{inspect(conn_pid)}")
 245
 246           put_in(
 247             state.conns,
 248             Map.delete(state.conns, key)
 249           )
 250
 251         nil ->
 252           Logger.debug(":gun_down message for conn which is not found in state")
 253
 254           :ok = API.close(conn_pid)
 255
 256           state
 257       end
 258
 259     {:noreply, state}
 260   end
 261
 262   @impl true
 263   def handle_info({:DOWN, _ref, :process, conn_pid, reason}, state) do
 264     Logger.debug("received DOWM message for #{inspect(conn_pid)} reason -> #{inspect(reason)}")
 265
 266     state =
 267       with {key, conn} <- find_conn(state.conns, conn_pid) do
 268         Enum.each(conn.used_by, fn {pid, _ref} ->
 269           Process.exit(pid, reason)
 270         end)
 271
 272         put_in(
 273           state.conns,
 274           Map.delete(state.conns, key)
 275         )
 276       else
 277         nil ->
 278           Logger.debug(":DOWN message for conn which is not found in state")
 279
 280           state
 281       end
 282
 283     {:noreply, state}
 284   end
 285
 286   defp compose_key_gun_info(pid) do
 287     try do
 288       # sometimes :gun.info can raise MatchError, which lead to pool terminate
 289       %{origin_host: origin_host, origin_scheme: scheme, origin_port: port} = API.info(pid)
 290
 291       host =
 292         case :inet.ntoa(origin_host) do
 293           {:error, :einval} -> origin_host
 294           ip -> ip
 295         end
 296
 297       "#{scheme}:#{host}:#{port}"
 298     rescue
 299       _ -> :error_gun_info
 300     end
 301   end
 302
 303   defp find_conn(conns, conn_pid) do
 304     Enum.find(conns, fn {_key, conn} ->
 305       conn.conn == conn_pid
 306     end)
 307   end
 308
 309   defp find_conn(conns, conn_pid, conn_key) do
 310     Enum.find(conns, fn {key, conn} ->
 311       key == conn_key and conn.conn == conn_pid
 312     end)
 313   end
 314
 315   def crf(current, steps, crf) do
 316     1 + :math.pow(0.5, current / steps) * crf
 317   end
 318
 319   def compose_uri_log(%URI{scheme: scheme, host: host, path: path}) do
 320     "#{scheme}://#{host}#{path}"
 321   end
 322 end