summaryrefslogtreecommitdiff
path: root/lib/pleroma/gun
diff options
context:
space:
mode:
Diffstat (limited to 'lib/pleroma/gun')
-rw-r--r--lib/pleroma/gun/api.ex46
-rw-r--r--lib/pleroma/gun/conn.ex135
-rw-r--r--lib/pleroma/gun/connection_pool.ex82
-rw-r--r--lib/pleroma/gun/connection_pool/reclaimer.ex85
-rw-r--r--lib/pleroma/gun/connection_pool/worker.ex133
-rw-r--r--lib/pleroma/gun/connection_pool/worker_supervisor.ex45
-rw-r--r--lib/pleroma/gun/gun.ex31
7 files changed, 557 insertions, 0 deletions
diff --git a/lib/pleroma/gun/api.ex b/lib/pleroma/gun/api.ex
new file mode 100644
index 000000000..09be74392
--- /dev/null
+++ b/lib/pleroma/gun/api.ex
@@ -0,0 +1,46 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Gun.API do
+ @behaviour Pleroma.Gun
+
+ alias Pleroma.Gun
+
+ @gun_keys [
+ :connect_timeout,
+ :http_opts,
+ :http2_opts,
+ :protocols,
+ :retry,
+ :retry_timeout,
+ :trace,
+ :transport,
+ :tls_opts,
+ :tcp_opts,
+ :socks_opts,
+ :ws_opts,
+ :supervise
+ ]
+
+ @impl Gun
+ def open(host, port, opts \\ %{}), do: :gun.open(host, port, Map.take(opts, @gun_keys))
+
+ @impl Gun
+ defdelegate info(pid), to: :gun
+
+ @impl Gun
+ defdelegate close(pid), to: :gun
+
+ @impl Gun
+ defdelegate await_up(pid, timeout \\ 5_000), to: :gun
+
+ @impl Gun
+ defdelegate connect(pid, opts), to: :gun
+
+ @impl Gun
+ defdelegate await(pid, ref), to: :gun
+
+ @impl Gun
+ defdelegate set_owner(pid, owner), to: :gun
+end
diff --git a/lib/pleroma/gun/conn.ex b/lib/pleroma/gun/conn.ex
new file mode 100644
index 000000000..a3f75a4bb
--- /dev/null
+++ b/lib/pleroma/gun/conn.ex
@@ -0,0 +1,135 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Gun.Conn do
+ alias Pleroma.Gun
+
+ require Logger
+
+ def open(%URI{} = uri, opts) do
+ pool_opts = Pleroma.Config.get([:connections_pool], [])
+
+ opts =
+ opts
+ |> Enum.into(%{})
+ |> Map.put_new(:await_up_timeout, pool_opts[:await_up_timeout] || 5_000)
+ |> Map.put_new(:supervise, false)
+ |> maybe_add_tls_opts(uri)
+
+ do_open(uri, opts)
+ end
+
+ defp maybe_add_tls_opts(opts, %URI{scheme: "http"}), do: opts
+
+ defp maybe_add_tls_opts(opts, %URI{scheme: "https"}) do
+ tls_opts = [
+ verify: :verify_peer,
+ cacertfile: CAStore.file_path(),
+ depth: 20,
+ reuse_sessions: false,
+ log_level: :warning,
+ customize_hostname_check: [match_fun: :public_key.pkix_verify_hostname_match_fun(:https)]
+ ]
+
+ tls_opts =
+ if Keyword.keyword?(opts[:tls_opts]) do
+ Keyword.merge(tls_opts, opts[:tls_opts])
+ else
+ tls_opts
+ end
+
+ Map.put(opts, :tls_opts, tls_opts)
+ end
+
+ defp do_open(uri, %{proxy: {proxy_host, proxy_port}} = opts) do
+ connect_opts =
+ uri
+ |> destination_opts()
+ |> add_http2_opts(uri.scheme, Map.get(opts, :tls_opts, []))
+
+ with open_opts <- Map.delete(opts, :tls_opts),
+ {:ok, conn} <- Gun.open(proxy_host, proxy_port, open_opts),
+ {:ok, _} <- Gun.await_up(conn, opts[:await_up_timeout]),
+ stream <- Gun.connect(conn, connect_opts),
+ {:response, :fin, 200, _} <- Gun.await(conn, stream) do
+ {:ok, conn}
+ else
+ error ->
+ Logger.warn(
+ "Opening proxied connection to #{compose_uri_log(uri)} failed with error #{
+ inspect(error)
+ }"
+ )
+
+ error
+ end
+ end
+
+ defp do_open(uri, %{proxy: {proxy_type, proxy_host, proxy_port}} = opts) do
+ version =
+ proxy_type
+ |> to_string()
+ |> String.last()
+ |> case do
+ "4" -> 4
+ _ -> 5
+ end
+
+ socks_opts =
+ uri
+ |> destination_opts()
+ |> add_http2_opts(uri.scheme, Map.get(opts, :tls_opts, []))
+ |> Map.put(:version, version)
+
+ opts =
+ opts
+ |> Map.put(:protocols, [:socks])
+ |> Map.put(:socks_opts, socks_opts)
+
+ with {:ok, conn} <- Gun.open(proxy_host, proxy_port, opts),
+ {:ok, _} <- Gun.await_up(conn, opts[:await_up_timeout]) do
+ {:ok, conn}
+ else
+ error ->
+ Logger.warn(
+ "Opening socks proxied connection to #{compose_uri_log(uri)} failed with error #{
+ inspect(error)
+ }"
+ )
+
+ error
+ end
+ end
+
+ defp do_open(%URI{host: host, port: port} = uri, opts) do
+ host = Pleroma.HTTP.AdapterHelper.parse_host(host)
+
+ with {:ok, conn} <- Gun.open(host, port, opts),
+ {:ok, _} <- Gun.await_up(conn, opts[:await_up_timeout]) do
+ {:ok, conn}
+ else
+ error ->
+ Logger.warn(
+ "Opening connection to #{compose_uri_log(uri)} failed with error #{inspect(error)}"
+ )
+
+ error
+ end
+ end
+
+ defp destination_opts(%URI{host: host, port: port}) do
+ host = Pleroma.HTTP.AdapterHelper.parse_host(host)
+ %{host: host, port: port}
+ end
+
+ defp add_http2_opts(opts, "https", tls_opts) do
+ Map.merge(opts, %{protocols: [:http2], transport: :tls, tls_opts: tls_opts})
+ end
+
+ defp add_http2_opts(opts, _, _), do: opts
+
+ def compose_uri_log(%URI{scheme: scheme, host: host, path: path}) do
+ "#{scheme}://#{host}#{path}"
+ end
+end
diff --git a/lib/pleroma/gun/connection_pool.ex b/lib/pleroma/gun/connection_pool.ex
new file mode 100644
index 000000000..f34602b73
--- /dev/null
+++ b/lib/pleroma/gun/connection_pool.ex
@@ -0,0 +1,82 @@
+defmodule Pleroma.Gun.ConnectionPool do
+ @registry __MODULE__
+
+ alias Pleroma.Gun.ConnectionPool.WorkerSupervisor
+
+ def children do
+ [
+ {Registry, keys: :unique, name: @registry},
+ Pleroma.Gun.ConnectionPool.WorkerSupervisor
+ ]
+ end
+
+ @spec get_conn(URI.t(), keyword()) :: {:ok, pid()} | {:error, term()}
+ def get_conn(uri, opts) do
+ key = "#{uri.scheme}:#{uri.host}:#{uri.port}"
+
+ case Registry.lookup(@registry, key) do
+ # The key has already been registered, but connection is not up yet
+ [{worker_pid, nil}] ->
+ get_gun_pid_from_worker(worker_pid, true)
+
+ [{worker_pid, {gun_pid, _used_by, _crf, _last_reference}}] ->
+ GenServer.call(worker_pid, :add_client)
+ {:ok, gun_pid}
+
+ [] ->
+ # :gun.set_owner fails in :connected state for whatevever reason,
+ # so we open the connection in the process directly and send it's pid back
+ # We trust gun to handle timeouts by itself
+ case WorkerSupervisor.start_worker([key, uri, opts, self()]) do
+ {:ok, worker_pid} ->
+ get_gun_pid_from_worker(worker_pid, false)
+
+ {:error, {:already_started, worker_pid}} ->
+ get_gun_pid_from_worker(worker_pid, true)
+
+ err ->
+ err
+ end
+ end
+ end
+
+ defp get_gun_pid_from_worker(worker_pid, register) do
+ # GenServer.call will block the process for timeout length if
+ # the server crashes on startup (which will happen if gun fails to connect)
+ # so instead we use cast + monitor
+
+ ref = Process.monitor(worker_pid)
+ if register, do: GenServer.cast(worker_pid, {:add_client, self()})
+
+ receive do
+ {:conn_pid, pid} ->
+ Process.demonitor(ref)
+ {:ok, pid}
+
+ {:DOWN, ^ref, :process, ^worker_pid, reason} ->
+ case reason do
+ {:shutdown, {:error, _} = error} -> error
+ {:shutdown, error} -> {:error, error}
+ _ -> {:error, reason}
+ end
+ end
+ end
+
+ @spec release_conn(pid()) :: :ok
+ def release_conn(conn_pid) do
+ # :ets.fun2ms(fn {_, {worker_pid, {gun_pid, _, _, _}}} when gun_pid == conn_pid ->
+ # worker_pid end)
+ query_result =
+ Registry.select(@registry, [
+ {{:_, :"$1", {:"$2", :_, :_, :_}}, [{:==, :"$2", conn_pid}], [:"$1"]}
+ ])
+
+ case query_result do
+ [worker_pid] ->
+ GenServer.call(worker_pid, :remove_client)
+
+ [] ->
+ :ok
+ end
+ end
+end
diff --git a/lib/pleroma/gun/connection_pool/reclaimer.ex b/lib/pleroma/gun/connection_pool/reclaimer.ex
new file mode 100644
index 000000000..cea800882
--- /dev/null
+++ b/lib/pleroma/gun/connection_pool/reclaimer.ex
@@ -0,0 +1,85 @@
+defmodule Pleroma.Gun.ConnectionPool.Reclaimer do
+ use GenServer, restart: :temporary
+
+ @registry Pleroma.Gun.ConnectionPool
+
+ def start_monitor do
+ pid =
+ case :gen_server.start(__MODULE__, [], name: {:via, Registry, {@registry, "reclaimer"}}) do
+ {:ok, pid} ->
+ pid
+
+ {:error, {:already_registered, pid}} ->
+ pid
+ end
+
+ {pid, Process.monitor(pid)}
+ end
+
+ @impl true
+ def init(_) do
+ {:ok, nil, {:continue, :reclaim}}
+ end
+
+ @impl true
+ def handle_continue(:reclaim, _) do
+ max_connections = Pleroma.Config.get([:connections_pool, :max_connections])
+
+ reclaim_max =
+ [:connections_pool, :reclaim_multiplier]
+ |> Pleroma.Config.get()
+ |> Kernel.*(max_connections)
+ |> round
+ |> max(1)
+
+ :telemetry.execute([:pleroma, :connection_pool, :reclaim, :start], %{}, %{
+ max_connections: max_connections,
+ reclaim_max: reclaim_max
+ })
+
+ # :ets.fun2ms(
+ # fn {_, {worker_pid, {_, used_by, crf, last_reference}}} when used_by == [] ->
+ # {worker_pid, crf, last_reference} end)
+ unused_conns =
+ Registry.select(
+ @registry,
+ [
+ {{:_, :"$1", {:_, :"$2", :"$3", :"$4"}}, [{:==, :"$2", []}], [{{:"$1", :"$3", :"$4"}}]}
+ ]
+ )
+
+ case unused_conns do
+ [] ->
+ :telemetry.execute(
+ [:pleroma, :connection_pool, :reclaim, :stop],
+ %{reclaimed_count: 0},
+ %{
+ max_connections: max_connections
+ }
+ )
+
+ {:stop, :no_unused_conns, nil}
+
+ unused_conns ->
+ reclaimed =
+ unused_conns
+ |> Enum.sort(fn {_pid1, crf1, last_reference1}, {_pid2, crf2, last_reference2} ->
+ crf1 <= crf2 and last_reference1 <= last_reference2
+ end)
+ |> Enum.take(reclaim_max)
+
+ reclaimed
+ |> Enum.each(fn {pid, _, _} ->
+ DynamicSupervisor.terminate_child(Pleroma.Gun.ConnectionPool.WorkerSupervisor, pid)
+ end)
+
+ :telemetry.execute(
+ [:pleroma, :connection_pool, :reclaim, :stop],
+ %{reclaimed_count: Enum.count(reclaimed)},
+ %{max_connections: max_connections}
+ )
+
+ {:stop, :normal, nil}
+ end
+ end
+end
diff --git a/lib/pleroma/gun/connection_pool/worker.ex b/lib/pleroma/gun/connection_pool/worker.ex
new file mode 100644
index 000000000..fec9d0efa
--- /dev/null
+++ b/lib/pleroma/gun/connection_pool/worker.ex
@@ -0,0 +1,133 @@
+defmodule Pleroma.Gun.ConnectionPool.Worker do
+ alias Pleroma.Gun
+ use GenServer, restart: :temporary
+
+ @registry Pleroma.Gun.ConnectionPool
+
+ def start_link([key | _] = opts) do
+ GenServer.start_link(__MODULE__, opts, name: {:via, Registry, {@registry, key}})
+ end
+
+ @impl true
+ def init([_key, _uri, _opts, _client_pid] = opts) do
+ {:ok, nil, {:continue, {:connect, opts}}}
+ end
+
+ @impl true
+ def handle_continue({:connect, [key, uri, opts, client_pid]}, _) do
+ with {:ok, conn_pid} <- Gun.Conn.open(uri, opts),
+ Process.link(conn_pid) do
+ time = :erlang.monotonic_time(:millisecond)
+
+ {_, _} =
+ Registry.update_value(@registry, key, fn _ ->
+ {conn_pid, [client_pid], 1, time}
+ end)
+
+ send(client_pid, {:conn_pid, conn_pid})
+
+ {:noreply,
+ %{key: key, timer: nil, client_monitors: %{client_pid => Process.monitor(client_pid)}},
+ :hibernate}
+ else
+ err ->
+ {:stop, {:shutdown, err}, nil}
+ end
+ end
+
+ @impl true
+ def handle_cast({:add_client, client_pid}, state) do
+ case handle_call(:add_client, {client_pid, nil}, state) do
+ {:reply, conn_pid, state, :hibernate} ->
+ send(client_pid, {:conn_pid, conn_pid})
+ {:noreply, state, :hibernate}
+ end
+ end
+
+ @impl true
+ def handle_cast({:remove_client, client_pid}, state) do
+ case handle_call(:remove_client, {client_pid, nil}, state) do
+ {:reply, _, state, :hibernate} ->
+ {:noreply, state, :hibernate}
+ end
+ end
+
+ @impl true
+ def handle_call(:add_client, {client_pid, _}, %{key: key} = state) do
+ time = :erlang.monotonic_time(:millisecond)
+
+ {{conn_pid, _, _, _}, _} =
+ Registry.update_value(@registry, key, fn {conn_pid, used_by, crf, last_reference} ->
+ {conn_pid, [client_pid | used_by], crf(time - last_reference, crf), time}
+ end)
+
+ state =
+ if state.timer != nil do
+ Process.cancel_timer(state[:timer])
+ %{state | timer: nil}
+ else
+ state
+ end
+
+ ref = Process.monitor(client_pid)
+
+ state = put_in(state.client_monitors[client_pid], ref)
+ {:reply, conn_pid, state, :hibernate}
+ end
+
+ @impl true
+ def handle_call(:remove_client, {client_pid, _}, %{key: key} = state) do
+ {{_conn_pid, used_by, _crf, _last_reference}, _} =
+ Registry.update_value(@registry, key, fn {conn_pid, used_by, crf, last_reference} ->
+ {conn_pid, List.delete(used_by, client_pid), crf, last_reference}
+ end)
+
+ {ref, state} = pop_in(state.client_monitors[client_pid])
+ Process.demonitor(ref)
+
+ timer =
+ if used_by == [] do
+ max_idle = Pleroma.Config.get([:connections_pool, :max_idle_time], 30_000)
+ Process.send_after(self(), :idle_close, max_idle)
+ else
+ nil
+ end
+
+ {:reply, :ok, %{state | timer: timer}, :hibernate}
+ end
+
+ @impl true
+ def handle_info(:idle_close, state) do
+ # Gun monitors the owner process, and will close the connection automatically
+ # when it's terminated
+ {:stop, :normal, state}
+ end
+
+ # Gracefully shutdown if the connection got closed without any streams left
+ @impl true
+ def handle_info({:gun_down, _pid, _protocol, _reason, []}, state) do
+ {:stop, :normal, state}
+ end
+
+ # Otherwise, shutdown with an error
+ @impl true
+ def handle_info({:gun_down, _pid, _protocol, _reason, _killed_streams} = down_message, state) do
+ {:stop, {:error, down_message}, state}
+ end
+
+ @impl true
+ def handle_info({:DOWN, _ref, :process, pid, reason}, state) do
+ :telemetry.execute(
+ [:pleroma, :connection_pool, :client_death],
+ %{client_pid: pid, reason: reason},
+ %{key: state.key}
+ )
+
+ handle_cast({:remove_client, pid}, state)
+ end
+
+ # LRFU policy: https://citeseerx.ist.psu.edu/viewdoc/summary?doi=10.1.1.55.1478
+ defp crf(time_delta, prev_crf) do
+ 1 + :math.pow(0.5, 0.0001 * time_delta) * prev_crf
+ end
+end
diff --git a/lib/pleroma/gun/connection_pool/worker_supervisor.ex b/lib/pleroma/gun/connection_pool/worker_supervisor.ex
new file mode 100644
index 000000000..39615c956
--- /dev/null
+++ b/lib/pleroma/gun/connection_pool/worker_supervisor.ex
@@ -0,0 +1,45 @@
+defmodule Pleroma.Gun.ConnectionPool.WorkerSupervisor do
+ @moduledoc "Supervisor for pool workers. Does not do anything except enforce max connection limit"
+
+ use DynamicSupervisor
+
+ def start_link(opts) do
+ DynamicSupervisor.start_link(__MODULE__, opts, name: __MODULE__)
+ end
+
+ def init(_opts) do
+ DynamicSupervisor.init(
+ strategy: :one_for_one,
+ max_children: Pleroma.Config.get([:connections_pool, :max_connections])
+ )
+ end
+
+ def start_worker(opts, retry \\ false) do
+ case DynamicSupervisor.start_child(__MODULE__, {Pleroma.Gun.ConnectionPool.Worker, opts}) do
+ {:error, :max_children} ->
+ if retry or free_pool() == :error do
+ :telemetry.execute([:pleroma, :connection_pool, :provision_failure], %{opts: opts})
+ {:error, :pool_full}
+ else
+ start_worker(opts, true)
+ end
+
+ res ->
+ res
+ end
+ end
+
+ defp free_pool do
+ wait_for_reclaimer_finish(Pleroma.Gun.ConnectionPool.Reclaimer.start_monitor())
+ end
+
+ defp wait_for_reclaimer_finish({pid, mon}) do
+ receive do
+ {:DOWN, ^mon, :process, ^pid, :no_unused_conns} ->
+ :error
+
+ {:DOWN, ^mon, :process, ^pid, :normal} ->
+ :ok
+ end
+ end
+end
diff --git a/lib/pleroma/gun/gun.ex b/lib/pleroma/gun/gun.ex
new file mode 100644
index 000000000..4043e4880
--- /dev/null
+++ b/lib/pleroma/gun/gun.ex
@@ -0,0 +1,31 @@
+# Pleroma: A lightweight social networking server
+# Copyright © 2017-2020 Pleroma Authors <https://pleroma.social/>
+# SPDX-License-Identifier: AGPL-3.0-only
+
+defmodule Pleroma.Gun do
+ @callback open(charlist(), pos_integer(), map()) :: {:ok, pid()}
+ @callback info(pid()) :: map()
+ @callback close(pid()) :: :ok
+ @callback await_up(pid, pos_integer()) :: {:ok, atom()} | {:error, atom()}
+ @callback connect(pid(), map()) :: reference()
+ @callback await(pid(), reference()) :: {:response, :fin, 200, []}
+ @callback set_owner(pid(), pid()) :: :ok
+
+ @api Pleroma.Config.get([Pleroma.Gun], Pleroma.Gun.API)
+
+ defp api, do: @api
+
+ def open(host, port, opts), do: api().open(host, port, opts)
+
+ def info(pid), do: api().info(pid)
+
+ def close(pid), do: api().close(pid)
+
+ def await_up(pid, timeout \\ 5_000), do: api().await_up(pid, timeout)
+
+ def connect(pid, opts), do: api().connect(pid, opts)
+
+ def await(pid, ref), do: api().await(pid, ref)
+
+ def set_owner(pid, owner), do: api().set_owner(pid, owner)
+end
generated by cgit v1.2.3 (git 2.44.0) at 2024-04-28 15:54:40 +0000