Priority queue fixes.

This commit is contained in:
Lain Iwakura 2017-12-06 16:51:11 +01:00
parent 66c3813ea6
commit e31a5ff4af
2 changed files with 12 additions and 7 deletions

View File

@ -29,7 +29,12 @@ def generate_id(type) do
Enqueues an activity for federation if it's local Enqueues an activity for federation if it's local
""" """
def maybe_federate(%Activity{local: true} = activity) do def maybe_federate(%Activity{local: true} = activity) do
Pleroma.Web.Federator.enqueue(:publish, activity) priority = case activity.data["type"] do
"Delete" -> 10
"Create" -> 1
_ -> 5
end
Pleroma.Web.Federator.enqueue(:publish, activity, priority)
:ok :ok
end end
def maybe_federate(_), do: :ok def maybe_federate(_), do: :ok

View File

@ -15,7 +15,7 @@ def start_link do
enqueue(:refresh_subscriptions, nil) enqueue(:refresh_subscriptions, nil)
end) end)
GenServer.start_link(__MODULE__, %{ GenServer.start_link(__MODULE__, %{
in: {:sets.new(), [], in: {:sets.new(), []},
out: {:sets.new(), []} out: {:sets.new(), []}
}, name: __MODULE__) }, name: __MODULE__)
end end
@ -79,17 +79,17 @@ def handle(type, _) do
{:error, "Don't know what do do with this"} {:error, "Don't know what do do with this"}
end end
def enqueue(type, payload) do def enqueue(type, payload, priority \\ 1) do
if Mix.env == :test do if Mix.env == :test do
handle(type, payload) handle(type, payload)
else else
GenServer.cast(__MODULE__, {:enqueue, type, payload}) GenServer.cast(__MODULE__, {:enqueue, type, payload, priority})
end end
end end
def maybe_start_job(running_jobs, queue) do def maybe_start_job(running_jobs, queue) do
if (:sets.size(running_jobs) < @max_jobs) && queue != [] do if (:sets.size(running_jobs) < @max_jobs) && queue != [] do
{{:value, {type, payload}}, queue} = queue_pop(queue) {{type, payload}, queue} = queue_pop(queue)
{:ok, pid} = Task.start(fn -> handle(type, payload) end) {:ok, pid} = Task.start(fn -> handle(type, payload) end)
mref = Process.monitor(pid) mref = Process.monitor(pid)
{:sets.add_element(mref, running_jobs), queue} {:sets.add_element(mref, running_jobs), queue}
@ -98,14 +98,14 @@ def maybe_start_job(running_jobs, queue) do
end end
end end
def handle_cast({:enqueue, type, payload}, state) when type in [:incoming_doc] do def handle_cast({:enqueue, type, payload, priority}, state) when type in [:incoming_doc] do
%{in: {i_running_jobs, i_queue}, out: {o_running_jobs, o_queue}} = state %{in: {i_running_jobs, i_queue}, out: {o_running_jobs, o_queue}} = state
i_queue = enqueue_sorted(i_queue, {type, payload}, 1) i_queue = enqueue_sorted(i_queue, {type, payload}, 1)
{i_running_jobs, i_queue} = maybe_start_job(i_running_jobs, i_queue) {i_running_jobs, i_queue} = maybe_start_job(i_running_jobs, i_queue)
{:noreply, %{in: {i_running_jobs, i_queue}, out: {o_running_jobs, o_queue}}} {:noreply, %{in: {i_running_jobs, i_queue}, out: {o_running_jobs, o_queue}}}
end end
def handle_cast({:enqueue, type, payload}, state) do def handle_cast({:enqueue, type, payload, priority}, state) do
%{in: {i_running_jobs, i_queue}, out: {o_running_jobs, o_queue}} = state %{in: {i_running_jobs, i_queue}, out: {o_running_jobs, o_queue}} = state
o_queue = enqueue_sorted(o_queue, {type, payload}, 1) o_queue = enqueue_sorted(o_queue, {type, payload}, 1)
{o_running_jobs, o_queue} = maybe_start_job(o_running_jobs, o_queue) {o_running_jobs, o_queue} = maybe_start_job(o_running_jobs, o_queue)