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: []
23 alias Pleroma.Gun.Conn
25 @spec start_link({atom(), keyword()}) :: {:ok, pid()}
26 def start_link({name, opts}) do
27 GenServer.start_link(__MODULE__, opts, name: name)
31 def init(opts), do: {:ok, %__MODULE__{conns: %{}, opts: opts}}
33 @spec checkin(String.t() | URI.t(), atom()) :: pid() | nil
34 def checkin(url, name)
35 def checkin(url, name) when is_binary(url), do: checkin(URI.parse(url), name)
37 def checkin(%URI{} = uri, name) do
38 timeout = Config.get([:connections_pool, :receive_connection_timeout], 250)
47 @spec open_conn(String.t() | URI.t(), atom(), keyword()) :: :ok
48 def open_conn(url, name, opts \\ [])
49 def open_conn(url, name, opts) when is_binary(url), do: open_conn(URI.parse(url), name, opts)
51 def open_conn(%URI{} = uri, name, opts) do
52 pool_opts = Config.get([:connections_pool], [])
57 |> Map.put_new(:retry, pool_opts[:retry] || 0)
58 |> Map.put_new(:retry_timeout, pool_opts[:retry_timeout] || 100)
59 |> Map.put_new(:await_up_timeout, pool_opts[:await_up_timeout] || 5_000)
61 GenServer.cast(name, {:open_conn, %{opts: opts, uri: uri}})
64 @spec alive?(atom()) :: boolean()
66 pid = Process.whereis(name)
67 if pid, do: Process.alive?(pid), else: false
70 @spec get_state(atom()) :: t()
71 def get_state(name) do
72 GenServer.call(name, :state)
75 @spec checkout(pid(), pid(), atom()) :: :ok
76 def checkout(conn, pid, name) do
77 GenServer.cast(name, {:checkout, conn, pid})
81 def handle_cast({:open_conn, %{opts: opts, uri: uri}}, state) do
82 Logger.debug("opening new #{compose_uri(uri)}")
83 max_connections = state.opts[:max_connections]
85 key = compose_key(uri)
87 if Enum.count(state.conns) < max_connections do
88 open_conn(key, uri, state, opts)
90 try_to_open_conn(key, uri, state, opts)
95 def handle_cast({:checkout, conn_pid, pid}, state) do
96 Logger.debug("checkout #{inspect(conn_pid)}")
99 with true <- Process.alive?(conn_pid),
100 {key, conn} <- find_conn(state.conns, conn_pid),
101 used_by <- List.keydelete(conn.used_by, pid, 0) do
109 put_in(state.conns[key], %{conn | conn_state: conn_state, used_by: used_by})
112 Logger.debug("checkout for closed conn #{inspect(conn_pid)}")
116 Logger.debug("checkout for alive conn #{inspect(conn_pid)}, but is not in state")
124 def handle_call({:checkin, uri}, from, state) do
125 Logger.debug("checkin #{compose_uri(uri)}")
126 key = compose_key(uri)
128 case state.conns[key] do
129 %{conn: conn, gun_state: gun_state} = current_conn when gun_state == :up ->
130 Logger.debug("reusing conn #{compose_uri(uri)}")
132 with time <- :os.system_time(:second),
133 last_reference <- time - current_conn.last_reference,
134 current_crf <- crf(last_reference, 100, current_conn.crf),
136 put_in(state.conns[key], %{
138 | last_reference: time,
141 used_by: [from | current_conn.used_by]
143 {:reply, conn, state}
146 %{gun_state: gun_state} when gun_state == :down ->
155 def handle_call(:state, _from, state), do: {:reply, state, state}
158 def handle_info({:gun_up, conn_pid, _protocol}, state) do
160 with true <- Process.alive?(conn_pid),
161 conn_key when is_binary(conn_key) <- compose_key_gun_info(conn_pid),
162 {key, conn} <- find_conn(state.conns, conn_pid, conn_key),
163 time <- :os.system_time(:second),
164 last_reference <- time - conn.last_reference,
165 current_crf <- crf(last_reference, 100, conn.crf) do
166 put_in(state.conns[key], %{
169 last_reference: time,
176 Logger.debug(":gun.info caused error")
180 Logger.debug(":gun_up message for closed conn #{inspect(conn_pid)}")
185 ":gun_up message for alive conn #{inspect(conn_pid)}, but deleted from state"
188 :ok = API.close(conn_pid)
197 def handle_info({:gun_down, conn_pid, _protocol, _reason, _killed}, state) do
198 retries = Config.get([:connections_pool, :retry], 0)
199 # we can't get info on this pid, because pid is dead
201 with true <- Process.alive?(conn_pid),
202 {key, conn} <- find_conn(state.conns, conn_pid) do
203 if conn.retries == retries do
204 Logger.debug("closing conn if retries is eq #{inspect(conn_pid)}")
205 :ok = API.close(conn.conn)
209 Map.delete(state.conns, key)
212 put_in(state.conns[key], %{
215 retries: conn.retries + 1
220 # gun can send gun_down for closed conn, maybe connection is not closed yet
221 Logger.debug(":gun_down message for closed conn #{inspect(conn_pid)}")
226 ":gun_down message for alive conn #{inspect(conn_pid)}, but deleted from state"
229 :ok = API.close(conn_pid)
237 defp compose_key(%URI{scheme: scheme, host: host, port: port}), do: "#{scheme}:#{host}:#{port}"
239 defp compose_key_gun_info(pid) do
241 # sometimes :gun.info can raise MatchError, which lead to pool terminate
242 %{origin_host: origin_host, origin_scheme: scheme, origin_port: port} = API.info(pid)
245 case :inet.ntoa(origin_host) do
246 {:error, :einval} -> origin_host
250 "#{scheme}:#{host}:#{port}"
256 defp find_conn(conns, conn_pid) do
257 Enum.find(conns, fn {_key, conn} ->
258 conn.conn == conn_pid
262 defp find_conn(conns, conn_pid, conn_key) do
263 Enum.find(conns, fn {key, conn} ->
264 key == conn_key and conn.conn == conn_pid
268 defp open_conn(key, uri, state, %{proxy: {proxy_host, proxy_port}} = opts) do
271 |> destination_opts()
272 |> add_http2_opts(uri.scheme, Map.get(opts, :tls_opts, []))
274 with open_opts <- Map.delete(opts, :tls_opts),
275 {:ok, conn} <- API.open(proxy_host, proxy_port, open_opts),
276 {:ok, _} <- API.await_up(conn),
277 stream <- API.connect(conn, connect_opts),
278 {:response, :fin, 200, _} <- API.await(conn, stream),
280 put_in(state.conns[key], %Conn{
284 last_reference: :os.system_time(:second)
290 "Received error on opening connection with http proxy #{uri.scheme}://#{
292 }: #{inspect(error)}"
299 defp open_conn(key, uri, state, %{proxy: {proxy_type, proxy_host, proxy_port}} = opts) do
311 |> destination_opts()
312 |> add_http2_opts(uri.scheme, Map.get(opts, :tls_opts, []))
313 |> Map.put(:version, version)
317 |> Map.put(:protocols, [:socks])
318 |> Map.put(:socks_opts, socks_opts)
320 with {:ok, conn} <- API.open(proxy_host, proxy_port, opts),
321 {:ok, _} <- API.await_up(conn),
323 put_in(state.conns[key], %Conn{
327 last_reference: :os.system_time(:second)
333 "Received error on opening connection with socks proxy #{uri.scheme}://#{
335 }: #{inspect(error)}"
342 defp open_conn(key, %URI{host: host, port: port} = uri, state, opts) do
343 Logger.debug("opening conn #{compose_uri(uri)}")
344 {_type, host} = Pleroma.HTTP.Adapter.domain_or_ip(host)
346 with {:ok, conn} <- API.open(host, port, opts),
347 {:ok, _} <- API.await_up(conn),
349 put_in(state.conns[key], %Conn{
353 last_reference: :os.system_time(:second)
355 Logger.debug("new conn opened #{compose_uri(uri)}")
356 Logger.debug("replying to the call #{compose_uri(uri)}")
361 "Received error on opening connection #{uri.scheme}://#{compose_uri(uri)}: #{
370 defp destination_opts(%URI{host: host, port: port}) do
371 {_type, host} = Pleroma.HTTP.Adapter.domain_or_ip(host)
372 %{host: host, port: port}
375 defp add_http2_opts(opts, "https", tls_opts) do
376 Map.merge(opts, %{protocols: [:http2], transport: :tls, tls_opts: tls_opts})
379 defp add_http2_opts(opts, _, _), do: opts
381 @spec get_unused_conns(map()) :: [{domain(), conn()}]
382 def get_unused_conns(conns) do
384 |> Enum.filter(fn {_k, v} ->
385 v.conn_state == :idle and v.used_by == []
387 |> Enum.sort(fn {_x_k, x}, {_y_k, y} ->
388 x.crf <= y.crf and x.last_reference <= y.last_reference
392 defp try_to_open_conn(key, uri, state, opts) do
393 Logger.debug("try to open conn #{compose_uri(uri)}")
395 with [{close_key, least_used} | _conns] <- get_unused_conns(state.conns),
396 :ok <- API.close(least_used.conn),
400 Map.delete(state.conns, close_key)
403 "least used conn found and closed #{inspect(least_used.conn)} #{compose_uri(uri)}"
406 open_conn(key, uri, state, opts)
408 [] -> {:noreply, state}
412 def crf(current, steps, crf) do
413 1 + :math.pow(0.5, current / steps) * crf
416 def compose_uri(%URI{} = uri), do: "#{uri.host}#{uri.path}"