1 # Pleroma: A lightweight social networking server
2 # Copyright © 2017-2018 Pleroma Authors <https://pleroma.social/>
3 # SPDX-License-Identifier: AGPL-3.0-only
5 defmodule Pleroma.Web.Federator.RetryQueue do
17 queue_table = :ets.new(:pleroma_retry_queue, [:bag, :protected])
19 {:ok, %{args | queue_table: queue_table, running_jobs: :sets.new()}}
24 if Mix.env() == :test, do: true, else: Pleroma.Config.get([:retry_queue, :enabled], false)
27 Logger.info("Starting retry queue")
32 %{delivered: 0, dropped: 0, queue_table: nil, running_jobs: nil},
39 Logger.info("Retry queue disabled")
44 def enqueue(data, transport, retries \\ 0) do
45 GenServer.cast(__MODULE__, {:maybe_enqueue, data, transport, retries + 1})
49 GenServer.call(__MODULE__, :get_stats)
53 GenServer.call(__MODULE__, :reset_stats)
56 def get_retry_params(retries) do
57 if retries > @max_retries do
58 {:drop, "Max retries reached"}
60 {:retry, growth_function(retries)}
64 def get_retry_timer_interval() do
65 Pleroma.Config.get([:retry_queue, :interval], 1000)
68 defp ets_count_expires(table, current_time) do
74 [{:"=<", :"$1", {:const, current_time}}],
81 defp ets_pop_n_expired(table, current_time, desired) do
82 {popped, _continuation} =
88 [{:"=<", :"$1", {:const, current_time}}],
96 |> List.foldl(true, fn e, acc ->
97 :ets.delete_object(table, e)
104 def maybe_start_job(running_jobs, queue_table) do
105 # we don't want to hit the ets or the DateTime more times than we have to
106 # could optimize slightly further by not using the count, and instead grabbing
107 # up to N objects early...
108 current_time = DateTime.to_unix(DateTime.utc_now())
109 n_running_jobs = :sets.size(running_jobs)
111 if n_running_jobs < @max_jobs do
112 n_ready_jobs = ets_count_expires(queue_table, current_time)
114 if n_ready_jobs > 0 do
115 # figure out how many we could start
116 available_job_slots = @max_jobs - n_running_jobs
117 start_n_jobs(running_jobs, queue_table, current_time, available_job_slots)
126 defp start_n_jobs(running_jobs, _queue_table, _current_time, 0) do
130 defp start_n_jobs(running_jobs, queue_table, current_time, available_job_slots)
131 when available_job_slots > 0 do
132 candidates = ets_pop_n_expired(queue_table, current_time, available_job_slots)
135 |> List.foldl(running_jobs, fn {_, e}, rj ->
136 {:ok, pid} = Task.start(fn -> worker(e) end)
137 mref = Process.monitor(pid)
138 :sets.add_element(mref, rj)
142 def worker({:send, data, transport, retries}) do
143 case transport.publish_one(data) do
145 GenServer.cast(__MODULE__, :inc_delivered)
149 enqueue(data, transport, retries)
154 def handle_call(:get_stats, _from, %{delivered: delivery_count, dropped: drop_count} = state) do
155 {:reply, %{delivered: delivery_count, dropped: drop_count}, state}
158 def handle_call(:reset_stats, _from, %{delivered: delivery_count, dropped: drop_count} = state) do
159 {:reply, %{delivered: delivery_count, dropped: drop_count},
160 %{state | delivered: 0, dropped: 0}}
163 def handle_cast(:reset_stats, state) do
164 {:noreply, %{state | delivered: 0, dropped: 0}}
168 {:maybe_enqueue, data, transport, retries},
169 %{dropped: drop_count, queue_table: queue_table, running_jobs: running_jobs} = state
171 case get_retry_params(retries) do
173 :ets.insert(queue_table, {timeout, {:send, data, transport, retries}})
174 running_jobs = maybe_start_job(running_jobs, queue_table)
175 {:noreply, %{state | running_jobs: running_jobs}}
178 Logger.debug(message)
179 {:noreply, %{state | dropped: drop_count + 1}}
183 def handle_cast(:kickoff_timer, state) do
184 retry_interval = get_retry_timer_interval()
185 Process.send_after(__MODULE__, :retry_timer_run, retry_interval)
189 def handle_cast(:inc_delivered, %{delivered: delivery_count} = state) do
190 {:noreply, %{state | delivered: delivery_count + 1}}
193 def handle_cast(:inc_dropped, %{dropped: drop_count} = state) do
194 {:noreply, %{state | dropped: drop_count + 1}}
197 def handle_info({:send, data, transport, retries}, %{delivered: delivery_count} = state) do
198 case transport.publish_one(data) do
200 {:noreply, %{state | delivered: delivery_count + 1}}
203 enqueue(data, transport, retries)
210 %{queue_table: queue_table, running_jobs: running_jobs} = state
212 maybe_kickoff_timer()
213 running_jobs = maybe_start_job(running_jobs, queue_table)
214 {:noreply, %{state | running_jobs: running_jobs}}
217 def handle_info({:DOWN, ref, :process, _pid, _reason}, state) do
218 %{running_jobs: running_jobs, queue_table: queue_table} = state
219 running_jobs = :sets.del_element(ref, running_jobs)
220 running_jobs = maybe_start_job(running_jobs, queue_table)
221 {:noreply, %{state | running_jobs: running_jobs}}
224 def handle_info(unknown, state) do
225 Logger.debug("RetryQueue: don't know what to do with #{inspect(unknown)}, ignoring")
229 if Mix.env() == :test do
230 defp growth_function(_retries) do
231 _shutit = @initial_timeout
232 DateTime.to_unix(DateTime.utc_now()) - 1
235 defp growth_function(retries) do
236 round(@initial_timeout * :math.pow(retries, 3)) + DateTime.to_unix(DateTime.utc_now())
240 defp maybe_kickoff_timer() do
241 GenServer.cast(__MODULE__, :kickoff_timer)