summaryrefslogtreecommitdiff
path: root/lib/pleroma/gun/connection_pool/worker.ex
blob: c36332817d1c585a0f4776468f104bae639ffe05 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
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])
    # 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)

        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

        %{state | timer: timer}
      end

    {:reply, :ok, state, :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

  @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
  @impl true
  def handle_info({:gun_down, _pid, _protocol, _reason, []}, state) do
    {:stop, :normal, state}
  end

  # Otherwise, wait for retry
  @impl true
  def handle_info({:gun_down, _pid, _protocol, _reason, _killed_streams}, state) do
    {:noreply, state, :hibernate}
  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