1 # Pleroma: A lightweight social networking server
2 # Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
3 # SPDX-License-Identifier: AGPL-3.0-only
5 defmodule Pleroma.Pool.Connections do
12 @type domain :: String.t()
13 @type conn :: Pleroma.Gun.Conn.t()
15 @type t :: %__MODULE__{
16 conns: %{domain() => conn()},
20 defstruct conns: %{}, opts: []
24 @spec start_link({atom(), keyword()}) :: {:ok, pid()}
25 def start_link({name, opts}) do
26 GenServer.start_link(__MODULE__, opts, name: name)
30 def init(opts), do: {:ok, %__MODULE__{conns: %{}, opts: opts}}
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)
36 def checkin(%URI{} = uri, name) do
37 timeout = Config.get([:connections_pool, :checkin_timeout], 250)
39 GenServer.call(name, {:checkin, uri}, timeout)
42 @spec alive?(atom()) :: boolean()
44 if pid = Process.whereis(name) do
51 @spec get_state(atom()) :: t()
52 def get_state(name) do
53 GenServer.call(name, :state)
56 @spec count(atom()) :: pos_integer()
58 GenServer.call(name, :count)
61 @spec get_unused_conns(atom()) :: [{domain(), conn()}]
62 def get_unused_conns(name) do
63 GenServer.call(name, :unused_conns)
66 @spec checkout(pid(), pid(), atom()) :: :ok
67 def checkout(conn, pid, name) do
68 GenServer.cast(name, {:checkout, conn, pid})
71 @spec add_conn(atom(), String.t(), Pleroma.Gun.Conn.t()) :: :ok
72 def add_conn(name, key, conn) do
73 GenServer.cast(name, {:add_conn, key, conn})
76 @spec remove_conn(atom(), String.t()) :: :ok
77 def remove_conn(name, key) do
78 GenServer.cast(name, {:remove_conn, key})
82 def handle_cast({:add_conn, key, conn}, state) do
83 state = put_in(state.conns[key], conn)
85 Process.monitor(conn.conn)
90 def handle_cast({:checkout, conn_pid, pid}, state) do
91 Logger.debug("checkout #{inspect(conn_pid)}")
94 with true <- Process.alive?(conn_pid),
95 {key, conn} <- find_conn(state.conns, conn_pid),
96 used_by <- List.keydelete(conn.used_by, pid, 0) do
104 put_in(state.conns[key], %{conn | conn_state: conn_state, used_by: used_by})
107 Logger.debug("checkout for closed conn #{inspect(conn_pid)}")
111 Logger.debug("checkout for alive conn #{inspect(conn_pid)}, but is not in state")
119 def handle_cast({:remove_conn, key}, state) do
120 state = put_in(state.conns, Map.delete(state.conns, key))
125 def handle_call({:checkin, uri}, from, state) do
126 key = "#{uri.scheme}:#{uri.host}:#{uri.port}"
127 Logger.debug("checkin #{key}")
129 case state.conns[key] do
130 %{conn: conn, gun_state: :up} = current_conn ->
131 Logger.debug("reusing conn #{key}")
133 time = :os.system_time(:second)
134 last_reference = time - current_conn.last_reference
135 current_crf = crf(last_reference, 100, current_conn.crf)
138 put_in(state.conns[key], %{
140 | last_reference: time,
143 used_by: [from | current_conn.used_by]
146 {:reply, conn, state}
148 %{gun_state: :down} ->
157 def handle_call(:state, _from, state), do: {:reply, state, state}
160 def handle_call(:count, _from, state) do
161 {:reply, Enum.count(state.conns), state}
165 def handle_call(:unused_conns, _from, state) do
168 |> Enum.filter(fn {_k, v} ->
169 v.conn_state == :idle and v.used_by == []
171 |> Enum.sort(fn {_x_k, x}, {_y_k, y} ->
172 x.crf <= y.crf and x.last_reference <= y.last_reference
175 {:reply, unused_conns, state}
179 def handle_info({:gun_up, conn_pid, _protocol}, state) do
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)
188 put_in(state.conns[key], %{
191 last_reference: time,
198 Logger.debug(":gun.info caused error")
202 Logger.debug(":gun_up message for closed conn #{inspect(conn_pid)}")
206 Map.delete(state.conns, key)
210 Logger.debug(":gun_up message for conn which is not found in state")
212 :ok = API.close(conn_pid)
221 def handle_info({:gun_down, conn_pid, _protocol, _reason, _killed}, state) do
222 retries = Config.get([:connections_pool, :retry], 0)
223 # we can't get info on this pid, because pid is dead
225 with {key, conn} <- find_conn(state.conns, conn_pid),
226 {true, key} <- {Process.alive?(conn_pid), key} do
227 if conn.retries == retries do
228 Logger.debug("closing conn if retries is eq #{inspect(conn_pid)}")
229 :ok = API.close(conn.conn)
233 Map.delete(state.conns, key)
236 put_in(state.conns[key], %{
239 retries: conn.retries + 1
244 # gun can send gun_down for closed conn, maybe connection is not closed yet
245 Logger.debug(":gun_down message for closed conn #{inspect(conn_pid)}")
249 Map.delete(state.conns, key)
253 Logger.debug(":gun_down message for conn which is not found in state")
255 :ok = API.close(conn_pid)
264 def handle_info({:DOWN, _ref, :process, conn_pid, reason}, state) do
265 Logger.debug("received DOWM message for #{inspect(conn_pid)} reason -> #{inspect(reason)}")
268 with {key, conn} <- find_conn(state.conns, conn_pid) do
269 Enum.each(conn.used_by, fn {pid, _ref} ->
270 Process.exit(pid, reason)
275 Map.delete(state.conns, key)
279 Logger.debug(":DOWN message for conn which is not found in state")
287 defp compose_key_gun_info(pid) do
289 # sometimes :gun.info can raise MatchError, which lead to pool terminate
290 %{origin_host: origin_host, origin_scheme: scheme, origin_port: port} = API.info(pid)
293 case :inet.ntoa(origin_host) do
294 {:error, :einval} -> origin_host
298 "#{scheme}:#{host}:#{port}"
304 defp find_conn(conns, conn_pid) do
305 Enum.find(conns, fn {_key, conn} ->
306 conn.conn == conn_pid
310 defp find_conn(conns, conn_pid, conn_key) do
311 Enum.find(conns, fn {key, conn} ->
312 key == conn_key and conn.conn == conn_pid
316 def crf(current, steps, crf) do
317 1 + :math.pow(0.5, current / steps) * crf
320 def compose_uri_log(%URI{scheme: scheme, host: host, path: path}) do
321 "#{scheme}://#{host}#{path}"