1 # Pleroma: A lightweight social networking server
2 # Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
3 # SPDX-License-Identifier: AGPL-3.0-only
5 defmodule Pleroma.Pool.Connections do
13 @type domain :: String.t()
14 @type conn :: Pleroma.Gun.Conn.t()
16 @type t :: %__MODULE__{
17 conns: %{domain() => conn()},
21 defstruct conns: %{}, opts: []
23 @spec start_link({atom(), keyword()}) :: {:ok, pid()}
24 def start_link({name, opts}) do
25 GenServer.start_link(__MODULE__, opts, name: name)
29 def init(opts), do: {:ok, %__MODULE__{conns: %{}, opts: opts}}
31 @spec checkin(String.t() | URI.t(), atom()) :: pid() | nil
32 def checkin(url, name)
33 def checkin(url, name) when is_binary(url), do: checkin(URI.parse(url), name)
35 def checkin(%URI{} = uri, name) do
36 timeout = Config.get([:connections_pool, :checkin_timeout], 250)
38 GenServer.call(name, {:checkin, uri}, timeout)
41 @spec alive?(atom()) :: boolean()
43 if pid = Process.whereis(name) do
50 @spec get_state(atom()) :: t()
51 def get_state(name) do
52 GenServer.call(name, :state)
55 @spec count(atom()) :: pos_integer()
57 GenServer.call(name, :count)
60 @spec get_unused_conns(atom()) :: [{domain(), conn()}]
61 def get_unused_conns(name) do
62 GenServer.call(name, :unused_conns)
65 @spec checkout(pid(), pid(), atom()) :: :ok
66 def checkout(conn, pid, name) do
67 GenServer.cast(name, {:checkout, conn, pid})
70 @spec add_conn(atom(), String.t(), Pleroma.Gun.Conn.t()) :: :ok
71 def add_conn(name, key, conn) do
72 GenServer.cast(name, {:add_conn, key, conn})
75 @spec remove_conn(atom(), String.t()) :: :ok
76 def remove_conn(name, key) do
77 GenServer.cast(name, {:remove_conn, key})
81 def handle_cast({:add_conn, key, conn}, state) do
82 state = put_in(state.conns[key], conn)
84 Process.monitor(conn.conn)
89 def handle_cast({:checkout, conn_pid, pid}, state) do
91 with true <- Process.alive?(conn_pid),
92 {key, conn} <- find_conn(state.conns, conn_pid),
93 used_by <- List.keydelete(conn.used_by, pid, 0) do
94 conn_state = if used_by == [], do: :idle, else: conn.conn_state
96 put_in(state.conns[key], %{conn | conn_state: conn_state, used_by: used_by})
99 Logger.debug("checkout for closed conn #{inspect(conn_pid)}")
103 Logger.debug("checkout for alive conn #{inspect(conn_pid)}, but is not in state")
111 def handle_cast({:remove_conn, key}, state) do
112 state = put_in(state.conns, Map.delete(state.conns, key))
117 def handle_call({:checkin, uri}, from, state) do
118 key = "#{uri.scheme}:#{uri.host}:#{uri.port}"
120 case state.conns[key] do
121 %{conn: pid, gun_state: :up} = conn ->
122 time = :os.system_time(:second)
123 last_reference = time - conn.last_reference
124 crf = crf(last_reference, 100, conn.crf)
127 put_in(state.conns[key], %{
129 | last_reference: time,
132 used_by: [from | conn.used_by]
137 %{gun_state: :down} ->
146 def handle_call(:state, _from, state), do: {:reply, state, state}
149 def handle_call(:count, _from, state) do
150 {:reply, Enum.count(state.conns), state}
154 def handle_call(:unused_conns, _from, state) do
157 |> Enum.filter(&filter_conns/1)
158 |> Enum.sort(&sort_conns/2)
160 {:reply, unused_conns, state}
163 defp filter_conns({_, %{conn_state: :idle, used_by: []}}), do: true
164 defp filter_conns(_), do: false
166 defp sort_conns({_, c1}, {_, c2}) do
167 c1.crf <= c2.crf and c1.last_reference <= c2.last_reference
171 def handle_info({:gun_up, conn_pid, _protocol}, state) do
172 # TODO: temp fix for gun MatchError https://github.com/ninenines/gun/issues/222
176 %{origin_host: host, origin_scheme: scheme, origin_port: port} = Gun.info(conn_pid)
179 case :inet.ntoa(host) do
180 {:error, :einval} -> host
184 key = "#{scheme}:#{host}:#{port}"
185 find_conn(state.conns, conn_pid, key)
187 MatcheError -> find_conn(state.conns, conn_pid)
191 with {true, key} <- {Process.alive?(conn_pid), key} do
192 put_in(state.conns[key], %{
202 Map.delete(state.conns, key)
206 :ok = Gun.close(conn_pid)
215 def handle_info({:gun_down, conn_pid, _protocol, _reason, _killed}, state) do
216 retries = Config.get([:connections_pool, :retry], 1)
217 # we can't get info on this pid, because pid is dead
219 with {key, conn} <- find_conn(state.conns, conn_pid),
220 {true, key} <- {Process.alive?(conn_pid), key} do
221 if conn.retries == retries do
222 :ok = Gun.close(conn.conn)
226 Map.delete(state.conns, key)
229 put_in(state.conns[key], %{
232 retries: conn.retries + 1
239 Map.delete(state.conns, key)
243 Logger.debug(":gun_down for conn which isn't found in state")
252 def handle_info({:DOWN, _ref, :process, conn_pid, reason}, state) do
253 Logger.debug("received DOWM message for #{inspect(conn_pid)} reason -> #{inspect(reason)}")
256 with {key, conn} <- find_conn(state.conns, conn_pid) do
257 Enum.each(conn.used_by, fn {pid, _ref} ->
258 Process.exit(pid, reason)
263 Map.delete(state.conns, key)
267 Logger.debug(":DOWN for conn which isn't found in state")
275 defp find_conn(conns, conn_pid) do
276 Enum.find(conns, fn {_key, conn} ->
277 conn.conn == conn_pid
281 defp find_conn(conns, conn_pid, conn_key) do
282 Enum.find(conns, fn {key, conn} ->
283 key == conn_key and conn.conn == conn_pid
287 def crf(current, steps, crf) do
288 1 + :math.pow(0.5, current / steps) * crf