adding gun adapter
This commit is contained in:
Alexander Strizhakov
parent
962eb8d4ac
commit
514c899275
63 changed files with 3615 additions and 389 deletions
415
lib/pleroma/pool/connections.ex
Normal file
415
lib/pleroma/pool/connections.ex
Normal file
|
|
@ -0,0 +1,415 @@
|
|||
# Pleroma: A lightweight social networking server
|
||||
# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
|
||||
# SPDX-License-Identifier: AGPL-3.0-only
|
||||
|
||||
defmodule Pleroma.Pool.Connections do
|
||||
use GenServer
|
||||
|
||||
require Logger
|
||||
|
||||
@type domain :: String.t()
|
||||
@type conn :: Pleroma.Gun.Conn.t()
|
||||
|
||||
@type t :: %__MODULE__{
|
||||
conns: %{domain() => conn()},
|
||||
opts: keyword()
|
||||
}
|
||||
|
||||
defstruct conns: %{}, opts: []
|
||||
|
||||
alias Pleroma.Gun.API
|
||||
alias Pleroma.Gun.Conn
|
||||
|
||||
@spec start_link({atom(), keyword()}) :: {:ok, pid()}
|
||||
def start_link({name, opts}) do
|
||||
GenServer.start_link(__MODULE__, opts, name: name)
|
||||
end
|
||||
|
||||
@impl true
|
||||
def init(opts), do: {:ok, %__MODULE__{conns: %{}, opts: opts}}
|
||||
|
||||
@spec checkin(String.t() | URI.t(), atom()) :: pid() | nil
|
||||
def checkin(url, name)
|
||||
def checkin(url, name) when is_binary(url), do: checkin(URI.parse(url), name)
|
||||
|
||||
def checkin(%URI{} = uri, name) do
|
||||
timeout = Pleroma.Config.get([:connections_pool, :receive_connection_timeout], 250)
|
||||
|
||||
GenServer.call(
|
||||
name,
|
||||
{:checkin, uri},
|
||||
timeout
|
||||
)
|
||||
end
|
||||
|
||||
@spec open_conn(String.t() | URI.t(), atom(), keyword()) :: :ok
|
||||
def open_conn(url, name, opts \\ [])
|
||||
def open_conn(url, name, opts) when is_binary(url), do: open_conn(URI.parse(url), name, opts)
|
||||
|
||||
def open_conn(%URI{} = uri, name, opts) do
|
||||
pool_opts = Pleroma.Config.get([:connections_pool], [])
|
||||
|
||||
opts =
|
||||
opts
|
||||
|> Enum.into(%{})
|
||||
|> Map.put_new(:receive, false)
|
||||
|> Map.put_new(:retry, pool_opts[:retry] || 5)
|
||||
|> Map.put_new(:retry_timeout, pool_opts[:retry_timeout] || 100)
|
||||
|> Map.put_new(:await_up_timeout, pool_opts[:await_up_timeout] || 5_000)
|
||||
|
||||
GenServer.cast(name, {:open_conn, %{opts: opts, uri: uri}})
|
||||
end
|
||||
|
||||
@spec alive?(atom()) :: boolean()
|
||||
def alive?(name) do
|
||||
pid = Process.whereis(name)
|
||||
if pid, do: Process.alive?(pid), else: false
|
||||
end
|
||||
|
||||
@spec get_state(atom()) :: t()
|
||||
def get_state(name) do
|
||||
GenServer.call(name, :state)
|
||||
end
|
||||
|
||||
@spec checkout(pid(), pid(), atom()) :: :ok
|
||||
def checkout(conn, pid, name) do
|
||||
GenServer.cast(name, {:checkout, conn, pid})
|
||||
end
|
||||
|
||||
@impl true
|
||||
def handle_cast({:open_conn, %{opts: opts, uri: uri}}, state) do
|
||||
Logger.debug("opening new #{compose_uri(uri)}")
|
||||
max_connections = state.opts[:max_connections]
|
||||
|
||||
key = compose_key(uri)
|
||||
|
||||
if Enum.count(state.conns) < max_connections do
|
||||
open_conn(key, uri, state, opts)
|
||||
else
|
||||
try_to_open_conn(key, uri, state, opts)
|
||||
end
|
||||
end
|
||||
|
||||
@impl true
|
||||
def handle_cast({:checkout, conn_pid, pid}, state) do
|
||||
Logger.debug("checkout #{inspect(conn_pid)}")
|
||||
|
||||
state =
|
||||
with true <- Process.alive?(conn_pid),
|
||||
{key, conn} <- find_conn(state.conns, conn_pid),
|
||||
used_by <- List.keydelete(conn.used_by, pid, 0) do
|
||||
conn_state =
|
||||
if used_by == [] do
|
||||
:idle
|
||||
else
|
||||
conn.conn_state
|
||||
end
|
||||
|
||||
put_in(state.conns[key], %{conn | conn_state: conn_state, used_by: used_by})
|
||||
else
|
||||
false ->
|
||||
Logger.warn("checkout for closed conn #{inspect(conn_pid)}")
|
||||
state
|
||||
|
||||
nil ->
|
||||
Logger.info("checkout for alive conn #{inspect(conn_pid)}, but is not in state")
|
||||
state
|
||||
end
|
||||
|
||||
{:noreply, state}
|
||||
end
|
||||
|
||||
@impl true
|
||||
def handle_call({:checkin, uri}, from, state) do
|
||||
Logger.debug("checkin #{compose_uri(uri)}")
|
||||
key = compose_key(uri)
|
||||
|
||||
case state.conns[key] do
|
||||
%{conn: conn, gun_state: gun_state} = current_conn when gun_state == :up ->
|
||||
Logger.debug("reusing conn #{compose_uri(uri)}")
|
||||
|
||||
with time <- :os.system_time(:second),
|
||||
last_reference <- time - current_conn.last_reference,
|
||||
current_crf <- crf(last_reference, 100, current_conn.crf),
|
||||
state <-
|
||||
put_in(state.conns[key], %{
|
||||
current_conn
|
||||
| last_reference: time,
|
||||
crf: current_crf,
|
||||
conn_state: :active,
|
||||
used_by: [from | current_conn.used_by]
|
||||
}) do
|
||||
{:reply, conn, state}
|
||||
end
|
||||
|
||||
%{gun_state: gun_state} when gun_state == :down ->
|
||||
{:reply, nil, state}
|
||||
|
||||
nil ->
|
||||
{:reply, nil, state}
|
||||
end
|
||||
end
|
||||
|
||||
@impl true
|
||||
def handle_call(:state, _from, state), do: {:reply, state, state}
|
||||
|
||||
@impl true
|
||||
def handle_info({:gun_up, conn_pid, _protocol}, state) do
|
||||
state =
|
||||
with true <- Process.alive?(conn_pid),
|
||||
conn_key when is_binary(conn_key) <- compose_key_gun_info(conn_pid),
|
||||
{key, conn} <- find_conn(state.conns, conn_pid, conn_key),
|
||||
time <- :os.system_time(:second),
|
||||
last_reference <- time - conn.last_reference,
|
||||
current_crf <- crf(last_reference, 100, conn.crf) do
|
||||
put_in(state.conns[key], %{
|
||||
conn
|
||||
| gun_state: :up,
|
||||
last_reference: time,
|
||||
crf: current_crf,
|
||||
conn_state: :active,
|
||||
retries: 0
|
||||
})
|
||||
else
|
||||
:error_gun_info ->
|
||||
Logger.warn(":gun.info caused error")
|
||||
state
|
||||
|
||||
false ->
|
||||
Logger.warn(":gun_up message for closed conn #{inspect(conn_pid)}")
|
||||
state
|
||||
|
||||
nil ->
|
||||
Logger.warn(
|
||||
":gun_up message for alive conn #{inspect(conn_pid)}, but deleted from state"
|
||||
)
|
||||
|
||||
:ok = API.close(conn_pid)
|
||||
|
||||
state
|
||||
end
|
||||
|
||||
{:noreply, state}
|
||||
end
|
||||
|
||||
@impl true
|
||||
def handle_info({:gun_down, conn_pid, _protocol, _reason, _killed}, state) do
|
||||
# we can't get info on this pid, because pid is dead
|
||||
state =
|
||||
with true <- Process.alive?(conn_pid),
|
||||
{key, conn} <- find_conn(state.conns, conn_pid) do
|
||||
if conn.retries == 5 do
|
||||
Logger.debug("closing conn if retries is eq 5 #{inspect(conn_pid)}")
|
||||
:ok = API.close(conn.conn)
|
||||
|
||||
put_in(
|
||||
state.conns,
|
||||
Map.delete(state.conns, key)
|
||||
)
|
||||
else
|
||||
put_in(state.conns[key], %{
|
||||
conn
|
||||
| gun_state: :down,
|
||||
retries: conn.retries + 1
|
||||
})
|
||||
end
|
||||
else
|
||||
false ->
|
||||
# gun can send gun_down for closed conn, maybe connection is not closed yet
|
||||
Logger.warn(":gun_down message for closed conn #{inspect(conn_pid)}")
|
||||
state
|
||||
|
||||
nil ->
|
||||
Logger.warn(
|
||||
":gun_down message for alive conn #{inspect(conn_pid)}, but deleted from state"
|
||||
)
|
||||
|
||||
:ok = API.close(conn_pid)
|
||||
|
||||
state
|
||||
end
|
||||
|
||||
{:noreply, state}
|
||||
end
|
||||
|
||||
defp compose_key(%URI{scheme: scheme, host: host, port: port}), do: "#{scheme}:#{host}:#{port}"
|
||||
|
||||
defp compose_key_gun_info(pid) do
|
||||
try do
|
||||
# sometimes :gun.info can raise MatchError, which lead to pool terminate
|
||||
%{origin_host: origin_host, origin_scheme: scheme, origin_port: port} = API.info(pid)
|
||||
|
||||
host =
|
||||
case :inet.ntoa(origin_host) do
|
||||
{:error, :einval} -> origin_host
|
||||
ip -> ip
|
||||
end
|
||||
|
||||
"#{scheme}:#{host}:#{port}"
|
||||
rescue
|
||||
_ -> :error_gun_info
|
||||
end
|
||||
end
|
||||
|
||||
defp find_conn(conns, conn_pid) do
|
||||
Enum.find(conns, fn {_key, conn} ->
|
||||
conn.conn == conn_pid
|
||||
end)
|
||||
end
|
||||
|
||||
defp find_conn(conns, conn_pid, conn_key) do
|
||||
Enum.find(conns, fn {key, conn} ->
|
||||
key == conn_key and conn.conn == conn_pid
|
||||
end)
|
||||
end
|
||||
|
||||
defp open_conn(key, uri, state, %{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} <- API.open(proxy_host, proxy_port, open_opts),
|
||||
{:ok, _} <- API.await_up(conn),
|
||||
stream <- API.connect(conn, connect_opts),
|
||||
{:response, :fin, 200, _} <- API.await(conn, stream),
|
||||
state <-
|
||||
put_in(state.conns[key], %Conn{
|
||||
conn: conn,
|
||||
gun_state: :up,
|
||||
conn_state: :active,
|
||||
last_reference: :os.system_time(:second)
|
||||
}) do
|
||||
{:noreply, state}
|
||||
else
|
||||
error ->
|
||||
Logger.warn(
|
||||
"Received error on opening connection with http proxy #{uri.scheme}://#{
|
||||
compose_uri(uri)
|
||||
}: #{inspect(error)}"
|
||||
)
|
||||
|
||||
{:noreply, state}
|
||||
end
|
||||
end
|
||||
|
||||
defp open_conn(key, uri, state, %{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} <- API.open(proxy_host, proxy_port, opts),
|
||||
{:ok, _} <- API.await_up(conn),
|
||||
state <-
|
||||
put_in(state.conns[key], %Conn{
|
||||
conn: conn,
|
||||
gun_state: :up,
|
||||
conn_state: :active,
|
||||
last_reference: :os.system_time(:second)
|
||||
}) do
|
||||
{:noreply, state}
|
||||
else
|
||||
error ->
|
||||
Logger.warn(
|
||||
"Received error on opening connection with socks proxy #{uri.scheme}://#{
|
||||
compose_uri(uri)
|
||||
}: #{inspect(error)}"
|
||||
)
|
||||
|
||||
{:noreply, state}
|
||||
end
|
||||
end
|
||||
|
||||
defp open_conn(key, %URI{host: host, port: port} = uri, state, opts) do
|
||||
Logger.debug("opening conn #{compose_uri(uri)}")
|
||||
{_type, host} = Pleroma.HTTP.Adapter.domain_or_ip(host)
|
||||
|
||||
with {:ok, conn} <- API.open(host, port, opts),
|
||||
{:ok, _} <- API.await_up(conn),
|
||||
state <-
|
||||
put_in(state.conns[key], %Conn{
|
||||
conn: conn,
|
||||
gun_state: :up,
|
||||
conn_state: :active,
|
||||
last_reference: :os.system_time(:second)
|
||||
}) do
|
||||
Logger.debug("new conn opened #{compose_uri(uri)}")
|
||||
Logger.debug("replying to the call #{compose_uri(uri)}")
|
||||
{:noreply, state}
|
||||
else
|
||||
error ->
|
||||
Logger.warn(
|
||||
"Received error on opening connection #{uri.scheme}://#{compose_uri(uri)}: #{
|
||||
inspect(error)
|
||||
}"
|
||||
)
|
||||
|
||||
{:noreply, state}
|
||||
end
|
||||
end
|
||||
|
||||
defp destination_opts(%URI{host: host, port: port}) do
|
||||
{_type, host} = Pleroma.HTTP.Adapter.domain_or_ip(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
|
||||
|
||||
@spec get_unused_conns(map()) :: [{domain(), conn()}]
|
||||
def get_unused_conns(conns) do
|
||||
conns
|
||||
|> Enum.filter(fn {_k, v} ->
|
||||
v.conn_state == :idle and v.used_by == []
|
||||
end)
|
||||
|> Enum.sort(fn {_x_k, x}, {_y_k, y} ->
|
||||
x.crf <= y.crf and x.last_reference <= y.last_reference
|
||||
end)
|
||||
end
|
||||
|
||||
defp try_to_open_conn(key, uri, state, opts) do
|
||||
Logger.debug("try to open conn #{compose_uri(uri)}")
|
||||
|
||||
with [{close_key, least_used} | _conns] <- get_unused_conns(state.conns),
|
||||
:ok <- API.close(least_used.conn),
|
||||
state <-
|
||||
put_in(
|
||||
state.conns,
|
||||
Map.delete(state.conns, close_key)
|
||||
) do
|
||||
Logger.debug(
|
||||
"least used conn found and closed #{inspect(least_used.conn)} #{compose_uri(uri)}"
|
||||
)
|
||||
|
||||
open_conn(key, uri, state, opts)
|
||||
else
|
||||
[] -> {:noreply, state}
|
||||
end
|
||||
end
|
||||
|
||||
def crf(current, steps, crf) do
|
||||
1 + :math.pow(0.5, current / steps) * crf
|
||||
end
|
||||
|
||||
def compose_uri(%URI{} = uri), do: "#{uri.host}#{uri.path}"
|
||||
end
|
||||
22
lib/pleroma/pool/pool.ex
Normal file
22
lib/pleroma/pool/pool.ex
Normal file
|
|
@ -0,0 +1,22 @@
|
|||
# Pleroma: A lightweight social networking server
|
||||
# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
|
||||
# SPDX-License-Identifier: AGPL-3.0-only
|
||||
|
||||
defmodule Pleroma.Pool do
|
||||
def child_spec(opts) do
|
||||
poolboy_opts =
|
||||
opts
|
||||
|> Keyword.put(:worker_module, Pleroma.Pool.Request)
|
||||
|> Keyword.put(:name, {:local, opts[:name]})
|
||||
|> Keyword.put(:size, opts[:size])
|
||||
|> Keyword.put(:max_overflow, opts[:max_overflow])
|
||||
|
||||
%{
|
||||
id: opts[:id] || {__MODULE__, make_ref()},
|
||||
start: {:poolboy, :start_link, [poolboy_opts, [name: opts[:name]]]},
|
||||
restart: :permanent,
|
||||
shutdown: 5000,
|
||||
type: :worker
|
||||
}
|
||||
end
|
||||
end
|
||||
72
lib/pleroma/pool/request.ex
Normal file
72
lib/pleroma/pool/request.ex
Normal file
|
|
@ -0,0 +1,72 @@
|
|||
# Pleroma: A lightweight social networking server
|
||||
# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
|
||||
# SPDX-License-Identifier: AGPL-3.0-only
|
||||
|
||||
defmodule Pleroma.Pool.Request do
|
||||
use GenServer
|
||||
|
||||
require Logger
|
||||
|
||||
def start_link(args) do
|
||||
GenServer.start_link(__MODULE__, args)
|
||||
end
|
||||
|
||||
@impl true
|
||||
def init(_), do: {:ok, []}
|
||||
|
||||
@spec execute(pid() | atom(), Tesla.Client.t(), keyword(), pos_integer()) ::
|
||||
{:ok, Tesla.Env.t()} | {:error, any()}
|
||||
def execute(pid, client, request, timeout) do
|
||||
GenServer.call(pid, {:execute, client, request}, timeout)
|
||||
end
|
||||
|
||||
@impl true
|
||||
def handle_call({:execute, client, request}, _from, state) do
|
||||
response = Pleroma.HTTP.request_try(client, request)
|
||||
|
||||
{:reply, response, state}
|
||||
end
|
||||
|
||||
@impl true
|
||||
def handle_info({:gun_data, _conn, stream, _, _}, state) do
|
||||
# in some cases if we reuse conn and got {:error, :body_too_large}
|
||||
# gun continues to send messages to this process,
|
||||
# so we flush messages for this request
|
||||
:ok = :gun.flush(stream)
|
||||
|
||||
{:noreply, state}
|
||||
end
|
||||
|
||||
@impl true
|
||||
def handle_info({:gun_up, _conn, _protocol}, state) do
|
||||
{:noreply, state}
|
||||
end
|
||||
|
||||
@impl true
|
||||
def handle_info({:gun_down, _conn, _protocol, _reason, _killed}, state) do
|
||||
# don't flush messages here, because gun can reconnect
|
||||
{:noreply, state}
|
||||
end
|
||||
|
||||
@impl true
|
||||
def handle_info({:gun_error, _conn, stream, _error}, state) do
|
||||
:ok = :gun.flush(stream)
|
||||
{:noreply, state}
|
||||
end
|
||||
|
||||
@impl true
|
||||
def handle_info({:gun_push, _conn, _stream, _new_stream, _method, _uri, _headers}, state) do
|
||||
{:noreply, state}
|
||||
end
|
||||
|
||||
@impl true
|
||||
def handle_info({:gun_response, _conn, _stream, _, _status, _headers}, state) do
|
||||
{:noreply, state}
|
||||
end
|
||||
|
||||
@impl true
|
||||
def handle_info(msg, state) do
|
||||
Logger.warn("Received unexpected message #{inspect(__MODULE__)} #{inspect(msg)}")
|
||||
{:noreply, state}
|
||||
end
|
||||
end
|
||||
36
lib/pleroma/pool/supervisor.ex
Normal file
36
lib/pleroma/pool/supervisor.ex
Normal file
|
|
@ -0,0 +1,36 @@
|
|||
# Pleroma: A lightweight social networking server
|
||||
# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
|
||||
# SPDX-License-Identifier: AGPL-3.0-only
|
||||
|
||||
defmodule Pleroma.Pool.Supervisor do
|
||||
use Supervisor
|
||||
|
||||
alias Pleroma.Pool
|
||||
|
||||
def start_link(args) do
|
||||
Supervisor.start_link(__MODULE__, args, name: __MODULE__)
|
||||
end
|
||||
|
||||
def init(_) do
|
||||
children =
|
||||
[
|
||||
%{
|
||||
id: Pool.Connections,
|
||||
start:
|
||||
{Pool.Connections, :start_link,
|
||||
[{:gun_connections, Pleroma.Config.get([:connections_pool])}]}
|
||||
}
|
||||
] ++ pools()
|
||||
|
||||
Supervisor.init(children, strategy: :one_for_one)
|
||||
end
|
||||
|
||||
defp pools do
|
||||
for {pool_name, pool_opts} <- Pleroma.Config.get([:pools]) do
|
||||
pool_opts
|
||||
|> Keyword.put(:id, {Pool, pool_name})
|
||||
|> Keyword.put(:name, pool_name)
|
||||
|> Pool.child_spec()
|
||||
end
|
||||
end
|
||||
end
|
||||
Loading…
Add table
Add a link
Reference in a new issue