improvements and fixes for http requests

- fix for gun worker termination in some circumstances
- pool for http clients (ex_aws, tzdata)
- default pool timeouts for gun
- gun retries on gun_down messages
- s3 upload timeout if streaming enabled
This commit is contained in:
Alexander Strizhakov 2020-08-19 10:22:59 +03:00
parent d48755791d
commit c17d83cd73
No known key found for this signature in database
GPG Key ID: 022896A53AEF1381
7 changed files with 70 additions and 36 deletions

View File

@ -740,19 +740,23 @@
config :pleroma, :pools, config :pleroma, :pools,
federation: [ federation: [
size: 50, size: 50,
max_waiting: 10 max_waiting: 10,
timeout: 10_000
], ],
media: [ media: [
size: 50, size: 50,
max_waiting: 10 max_waiting: 10,
timeout: 10_000
], ],
upload: [ upload: [
size: 25, size: 25,
max_waiting: 5 max_waiting: 5,
timeout: 15_000
], ],
default: [ default: [
size: 10, size: 10,
max_waiting: 2 max_waiting: 2,
timeout: 5_000
] ]
config :pleroma, :hackney_pools, config :pleroma, :hackney_pools,

View File

@ -83,6 +83,11 @@ def handle_call(:remove_client, {client_pid, _}, %{key: key} = state) do
end) end)
{ref, state} = pop_in(state.client_monitors[client_pid]) {ref, state} = pop_in(state.client_monitors[client_pid])
# DOWN message can receive right after `remove_client` call and cause worker to terminate
state =
if is_nil(ref) do
state
else
Process.demonitor(ref) Process.demonitor(ref)
timer = timer =
@ -93,7 +98,10 @@ def handle_call(:remove_client, {client_pid, _}, %{key: key} = state) do
nil nil
end end
{:reply, :ok, %{state | timer: timer}, :hibernate} %{state | timer: timer}
end
{:reply, :ok, state, :hibernate}
end end
@impl true @impl true
@ -103,16 +111,21 @@ def handle_info(:idle_close, state) do
{:stop, :normal, state} {:stop, :normal, state}
end end
@impl true
def handle_info({:gun_up, _pid, _protocol}, state) do
{:noreply, state, :hibernate}
end
# Gracefully shutdown if the connection got closed without any streams left # Gracefully shutdown if the connection got closed without any streams left
@impl true @impl true
def handle_info({:gun_down, _pid, _protocol, _reason, []}, state) do def handle_info({:gun_down, _pid, _protocol, _reason, []}, state) do
{:stop, :normal, state} {:stop, :normal, state}
end end
# Otherwise, shutdown with an error # Otherwise, wait for retry
@impl true @impl true
def handle_info({:gun_down, _pid, _protocol, _reason, _killed_streams} = down_message, state) do def handle_info({:gun_down, _pid, _protocol, _reason, _killed_streams}, state) do
{:stop, {:error, down_message}, state} {:noreply, state, :hibernate}
end end
@impl true @impl true

View File

@ -10,6 +10,7 @@ defmodule Pleroma.HTTP.AdapterHelper do
@type proxy_type() :: :socks4 | :socks5 @type proxy_type() :: :socks4 | :socks5
@type host() :: charlist() | :inet.ip_address() @type host() :: charlist() | :inet.ip_address()
@type pool() :: :federation | :upload | :media | :default
alias Pleroma.Config alias Pleroma.Config
alias Pleroma.HTTP.AdapterHelper alias Pleroma.HTTP.AdapterHelper
@ -44,14 +45,13 @@ def maybe_add_proxy(opts, proxy), do: Keyword.put_new(opts, :proxy, proxy)
@spec options(URI.t(), keyword()) :: keyword() @spec options(URI.t(), keyword()) :: keyword()
def options(%URI{} = uri, opts \\ []) do def options(%URI{} = uri, opts \\ []) do
@defaults @defaults
|> put_timeout()
|> Keyword.merge(opts) |> Keyword.merge(opts)
|> put_timeout()
|> adapter_helper().options(uri) |> adapter_helper().options(uri)
end end
# For Hackney, this is the time a connection can stay idle in the pool. @spec pool_timeout(pool()) :: non_neg_integer()
# For Gun, this is the timeout to receive a message from Gun. def pool_timeout(pool) do
defp put_timeout(opts) do
{config_key, default} = {config_key, default} =
if adapter() == Tesla.Adapter.Gun do if adapter() == Tesla.Adapter.Gun do
{:pools, Config.get([:pools, :default, :timeout], 5_000)} {:pools, Config.get([:pools, :default, :timeout], 5_000)}
@ -59,9 +59,13 @@ defp put_timeout(opts) do
{:hackney_pools, 10_000} {:hackney_pools, 10_000}
end end
timeout = Config.get([config_key, opts[:pool], :timeout], default) Config.get([config_key, pool, :timeout], default)
end
Keyword.merge(opts, timeout: timeout) # For Hackney, this is the time a connection can stay idle in the pool.
# For Gun, this is the timeout to receive a message from Gun.
defp put_timeout(opts) do
Keyword.put_new(opts, :timeout, pool_timeout(opts[:pool]))
end end
def get_conn(uri, opts), do: adapter_helper().get_conn(uri, opts) def get_conn(uri, opts), do: adapter_helper().get_conn(uri, opts)

View File

@ -5,6 +5,7 @@
defmodule Pleroma.HTTP.AdapterHelper.Gun do defmodule Pleroma.HTTP.AdapterHelper.Gun do
@behaviour Pleroma.HTTP.AdapterHelper @behaviour Pleroma.HTTP.AdapterHelper
alias Pleroma.Config
alias Pleroma.Gun.ConnectionPool alias Pleroma.Gun.ConnectionPool
alias Pleroma.HTTP.AdapterHelper alias Pleroma.HTTP.AdapterHelper
@ -14,7 +15,7 @@ defmodule Pleroma.HTTP.AdapterHelper.Gun do
connect_timeout: 5_000, connect_timeout: 5_000,
domain_lookup_timeout: 5_000, domain_lookup_timeout: 5_000,
tls_handshake_timeout: 5_000, tls_handshake_timeout: 5_000,
retry: 0, retry: 1,
retry_timeout: 1000, retry_timeout: 1000,
await_up_timeout: 5_000 await_up_timeout: 5_000
] ]
@ -22,10 +23,11 @@ defmodule Pleroma.HTTP.AdapterHelper.Gun do
@spec options(keyword(), URI.t()) :: keyword() @spec options(keyword(), URI.t()) :: keyword()
def options(incoming_opts \\ [], %URI{} = uri) do def options(incoming_opts \\ [], %URI{} = uri) do
proxy = proxy =
Pleroma.Config.get([:http, :proxy_url]) [:http, :proxy_url]
|> Config.get()
|> AdapterHelper.format_proxy() |> AdapterHelper.format_proxy()
config_opts = Pleroma.Config.get([:http, :adapter], []) config_opts = Config.get([:http, :adapter], [])
@defaults @defaults
|> Keyword.merge(config_opts) |> Keyword.merge(config_opts)
@ -37,8 +39,7 @@ def options(incoming_opts \\ [], %URI{} = uri) do
defp add_scheme_opts(opts, %{scheme: "http"}), do: opts defp add_scheme_opts(opts, %{scheme: "http"}), do: opts
defp add_scheme_opts(opts, %{scheme: "https"}) do defp add_scheme_opts(opts, %{scheme: "https"}) do
opts Keyword.put(opts, :certificates_verification, true)
|> Keyword.put(:certificates_verification, true)
end end
@spec get_conn(URI.t(), keyword()) :: {:ok, keyword()} | {:error, atom()} @spec get_conn(URI.t(), keyword()) :: {:ok, keyword()} | {:error, atom()}
@ -51,11 +52,11 @@ def get_conn(uri, opts) do
@prefix Pleroma.Gun.ConnectionPool @prefix Pleroma.Gun.ConnectionPool
def limiter_setup do def limiter_setup do
wait = Pleroma.Config.get([:connections_pool, :connection_acquisition_wait]) wait = Config.get([:connections_pool, :connection_acquisition_wait])
retries = Pleroma.Config.get([:connections_pool, :connection_acquisition_retries]) retries = Config.get([:connections_pool, :connection_acquisition_retries])
:pools :pools
|> Pleroma.Config.get([]) |> Config.get([])
|> Enum.each(fn {name, opts} -> |> Enum.each(fn {name, opts} ->
max_running = Keyword.get(opts, :size, 50) max_running = Keyword.get(opts, :size, 50)
max_waiting = Keyword.get(opts, :max_waiting, 10) max_waiting = Keyword.get(opts, :max_waiting, 10)
@ -69,7 +70,6 @@ def limiter_setup do
case result do case result do
:ok -> :ok :ok -> :ok
{:error, :existing} -> :ok {:error, :existing} -> :ok
e -> raise e
end end
end) end)

View File

@ -11,6 +11,8 @@ defmodule Pleroma.HTTP.ExAws do
@impl true @impl true
def request(method, url, body \\ "", headers \\ [], http_opts \\ []) do def request(method, url, body \\ "", headers \\ [], http_opts \\ []) do
http_opts = Keyword.put(http_opts, :adapter, pool: :upload)
case HTTP.request(method, url, body, headers, http_opts) do case HTTP.request(method, url, body, headers, http_opts) do
{:ok, env} -> {:ok, env} ->
{:ok, %{status_code: env.status, headers: env.headers, body: env.body}} {:ok, %{status_code: env.status, headers: env.headers, body: env.body}}

View File

@ -11,6 +11,8 @@ defmodule Pleroma.HTTP.Tzdata do
@impl true @impl true
def get(url, headers, options) do def get(url, headers, options) do
options = Keyword.put(options, :adapter, pool: :upload)
with {:ok, %Tesla.Env{} = env} <- HTTP.get(url, headers, options) do with {:ok, %Tesla.Env{} = env} <- HTTP.get(url, headers, options) do
{:ok, {env.status, env.headers, env.body}} {:ok, {env.status, env.headers, env.body}}
end end
@ -18,6 +20,8 @@ def get(url, headers, options) do
@impl true @impl true
def head(url, headers, options) do def head(url, headers, options) do
options = Keyword.put(options, :adapter, pool: :upload)
with {:ok, %Tesla.Env{} = env} <- HTTP.head(url, headers, options) do with {:ok, %Tesla.Env{} = env} <- HTTP.head(url, headers, options) do
{:ok, {env.status, env.headers}} {:ok, {env.status, env.headers}}
end end

View File

@ -46,12 +46,19 @@ def put_file(%Pleroma.Upload{} = upload) do
op = op =
if streaming do if streaming do
op =
upload.tempfile upload.tempfile
|> ExAws.S3.Upload.stream_file() |> ExAws.S3.Upload.stream_file()
|> ExAws.S3.upload(bucket, s3_name, [ |> ExAws.S3.upload(bucket, s3_name, [
{:acl, :public_read}, {:acl, :public_read},
{:content_type, upload.content_type} {:content_type, upload.content_type}
]) ])
# set s3 upload timeout to respect :upload pool timeout
# timeout should be slightly larger, so s3 can retry upload on fail
timeout = Pleroma.HTTP.AdapterHelper.pool_timeout(:upload) + 1_000
opts = Keyword.put(op.opts, :timeout, timeout)
Map.put(op, :opts, opts)
else else
{:ok, file_data} = File.read(upload.tempfile) {:ok, file_data} = File.read(upload.tempfile)