initial_timeout: 30,
max_retries: 5
-config :pleroma, Pleroma.Jobs,
- federator_incoming: [max_jobs: 50],
- federator_outgoing: [max_jobs: 50],
- mailer: [max_jobs: 10]
+config :pleroma_job_queue, :queues,
+ federator_incoming: 50,
+ federator_outgoing: 50,
+ mailer: 10
config :pleroma, :fetch_initial_posts,
enabled: false,
config :web_push_encryption, :http_client, Pleroma.Web.WebPushHttpClientMock
-config :pleroma, Pleroma.Jobs, testing: [max_jobs: 2]
+config :pleroma_job_queue, disabled: true
try do
import_config "test.secret.exs"
curl "http://localhost:4000/api/pleroma/admin/invite_token?admin_token=somerandomtoken"
```
-## Pleroma.Jobs
+## :pleroma_job_queue
-A list of job queues and their settings.
-
-Job queue settings:
-
-* `max_jobs`: The maximum amount of parallel jobs running at the same time.
+[Pleroma Job Queue][https://git.pleroma.social/pleroma/pleroma_job_queue] configuration: a list of queues with maximum concurrent jobs.
Example:
-```exs
-config :pleroma, Pleroma.Jobs,
- federator_incoming: [max_jobs: 50],
- federator_outgoing: [max_jobs: 50]
+```elixir
+config :pleroma_job_queue, :queues,
+ federator_incoming: 50,
+ federator_outgoing: 50
```
This config contains two queues: `federator_incoming` and `federator_outgoing`. Both have the `max_jobs` set to `50`.
-
## Pleroma.Web.Federator.RetryQueue
* `enabled`: If set to `true`, failed federation jobs will be retried
worker(Pleroma.Web.Federator.RetryQueue, []),
worker(Pleroma.Stats, []),
worker(Pleroma.Web.Push, []),
- worker(Pleroma.Jobs, []),
worker(Task, [&Pleroma.Web.Federator.init/0], restart: :temporary)
] ++
streamer_child() ++
use Swoosh.Mailer, otp_app: :pleroma
def deliver_async(email, config \\ []) do
- Pleroma.Jobs.enqueue(:mailer, __MODULE__, [:deliver_async, email, config])
+ PleromaJobQueue.enqueue(:mailer, __MODULE__, [:deliver_async, email, config])
end
def perform(:deliver_async, email, config), do: deliver(email, config)
+++ /dev/null
-# Pleroma: A lightweight social networking server
-# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
-# SPDX-License-Identifier: AGPL-3.0-only
-
-defmodule Pleroma.Jobs do
- @moduledoc """
- A basic job queue
- """
- use GenServer
-
- require Logger
-
- def init(args) do
- {:ok, args}
- end
-
- def start_link do
- queues =
- Pleroma.Config.get(Pleroma.Jobs)
- |> Enum.map(fn {name, _} -> create_queue(name) end)
- |> Enum.into(%{})
-
- state = %{
- queues: queues,
- refs: %{}
- }
-
- GenServer.start_link(__MODULE__, state, name: __MODULE__)
- end
-
- def create_queue(name) do
- {name, {:sets.new(), []}}
- end
-
- @doc """
- Enqueues a job.
-
- Returns `:ok`.
-
- ## Arguments
-
- - `queue_name` - a queue name(must be specified in the config).
- - `mod` - a worker module (must have `perform` function).
- - `args` - a list of arguments for the `perform` function of the worker module.
- - `priority` - a job priority (`0` by default).
-
- ## Examples
-
- Enqueue `Module.perform/0` with `priority=1`:
-
- iex> Pleroma.Jobs.enqueue(:example_queue, Module, [])
- :ok
-
- Enqueue `Module.perform(:job_name)` with `priority=5`:
-
- iex> Pleroma.Jobs.enqueue(:example_queue, Module, [:job_name], 5)
- :ok
-
- Enqueue `Module.perform(:another_job, data)` with `priority=1`:
-
- iex> data = "foobar"
- iex> Pleroma.Jobs.enqueue(:example_queue, Module, [:another_job, data])
- :ok
-
- Enqueue `Module.perform(:foobar_job, :foo, :bar, 42)` with `priority=1`:
-
- iex> Pleroma.Jobs.enqueue(:example_queue, Module, [:foobar_job, :foo, :bar, 42])
- :ok
-
- """
-
- def enqueue(queue_name, mod, args, priority \\ 1)
-
- if Mix.env() == :test do
- def enqueue(_queue_name, mod, args, _priority) do
- apply(mod, :perform, args)
- end
- else
- @spec enqueue(atom(), atom(), [any()], integer()) :: :ok
- def enqueue(queue_name, mod, args, priority) do
- GenServer.cast(__MODULE__, {:enqueue, queue_name, mod, args, priority})
- end
- end
-
- def handle_cast({:enqueue, queue_name, mod, args, priority}, state) do
- {running_jobs, queue} = state[:queues][queue_name]
-
- queue = enqueue_sorted(queue, {mod, args}, priority)
-
- state =
- state
- |> update_queue(queue_name, {running_jobs, queue})
- |> maybe_start_job(queue_name, running_jobs, queue)
-
- {:noreply, state}
- end
-
- def handle_info({:DOWN, ref, :process, _pid, _reason}, state) do
- queue_name = state.refs[ref]
-
- {running_jobs, queue} = state[:queues][queue_name]
-
- running_jobs = :sets.del_element(ref, running_jobs)
-
- state =
- state
- |> remove_ref(ref)
- |> update_queue(queue_name, {running_jobs, queue})
- |> maybe_start_job(queue_name, running_jobs, queue)
-
- {:noreply, state}
- end
-
- def maybe_start_job(state, queue_name, running_jobs, queue) do
- if :sets.size(running_jobs) < Pleroma.Config.get([__MODULE__, queue_name, :max_jobs]) &&
- queue != [] do
- {{mod, args}, queue} = queue_pop(queue)
- {:ok, pid} = Task.start(fn -> apply(mod, :perform, args) end)
- mref = Process.monitor(pid)
-
- state
- |> add_ref(queue_name, mref)
- |> update_queue(queue_name, {:sets.add_element(mref, running_jobs), queue})
- else
- state
- end
- end
-
- def enqueue_sorted(queue, element, priority) do
- [%{item: element, priority: priority} | queue]
- |> Enum.sort_by(fn %{priority: priority} -> priority end)
- end
-
- def queue_pop([%{item: element} | queue]) do
- {element, queue}
- end
-
- defp add_ref(state, queue_name, ref) do
- refs = Map.put(state[:refs], ref, queue_name)
- Map.put(state, :refs, refs)
- end
-
- defp remove_ref(state, ref) do
- refs = Map.delete(state[:refs], ref)
- Map.put(state, :refs, refs)
- end
-
- defp update_queue(state, queue_name, data) do
- queues = Map.put(state[:queues], queue_name, data)
- Map.put(state, :queues, queues)
- end
-end
defmodule Pleroma.Web.Federator do
alias Pleroma.Activity
- alias Pleroma.Jobs
alias Pleroma.User
alias Pleroma.Web.ActivityPub.ActivityPub
alias Pleroma.Web.ActivityPub.Relay
# Client API
def incoming_doc(doc) do
- Jobs.enqueue(:federator_incoming, __MODULE__, [:incoming_doc, doc])
+ PleromaJobQueue.enqueue(:federator_incoming, __MODULE__, [:incoming_doc, doc])
end
def incoming_ap_doc(params) do
- Jobs.enqueue(:federator_incoming, __MODULE__, [:incoming_ap_doc, params])
+ PleromaJobQueue.enqueue(:federator_incoming, __MODULE__, [:incoming_ap_doc, params])
end
def publish(activity, priority \\ 1) do
- Jobs.enqueue(:federator_outgoing, __MODULE__, [:publish, activity], priority)
+ PleromaJobQueue.enqueue(:federator_outgoing, __MODULE__, [:publish, activity], priority)
end
def publish_single_ap(params) do
- Jobs.enqueue(:federator_outgoing, __MODULE__, [:publish_single_ap, params])
+ PleromaJobQueue.enqueue(:federator_outgoing, __MODULE__, [:publish_single_ap, params])
end
def publish_single_websub(websub) do
- Jobs.enqueue(:federator_outgoing, __MODULE__, [:publish_single_websub, websub])
+ PleromaJobQueue.enqueue(:federator_outgoing, __MODULE__, [:publish_single_websub, websub])
end
def verify_websub(websub) do
- Jobs.enqueue(:federator_outgoing, __MODULE__, [:verify_websub, websub])
+ PleromaJobQueue.enqueue(:federator_outgoing, __MODULE__, [:verify_websub, websub])
end
def request_subscription(sub) do
- Jobs.enqueue(:federator_outgoing, __MODULE__, [:request_subscription, sub])
+ PleromaJobQueue.enqueue(:federator_outgoing, __MODULE__, [:request_subscription, sub])
end
def refresh_subscriptions do
- Jobs.enqueue(:federator_outgoing, __MODULE__, [:refresh_subscriptions])
+ PleromaJobQueue.enqueue(:federator_outgoing, __MODULE__, [:refresh_subscriptions])
end
def publish_single_salmon(params) do
- Jobs.enqueue(:federator_outgoing, __MODULE__, [:publish_single_salmon, params])
+ PleromaJobQueue.enqueue(:federator_outgoing, __MODULE__, [:publish_single_salmon, params])
end
# Job Worker Callbacks
{:timex, "~> 3.5"},
{:auto_linker,
git: "https://git.pleroma.social/pleroma/auto_linker.git",
- ref: "94193ca5f97c1f9fdf3d1469653e2d46fac34bcd"}
+ ref: "94193ca5f97c1f9fdf3d1469653e2d46fac34bcd"},
+ {:pleroma_job_queue, "~> 0.2.0"}
]
end
"phoenix_ecto": {:hex, :phoenix_ecto, "4.0.0", "c43117a136e7399ea04ecaac73f8f23ee0ffe3e07acfcb8062fe5f4c9f0f6531", [:mix], [{:ecto, "~> 3.0", [hex: :ecto, repo: "hexpm", optional: false]}, {:phoenix_html, "~> 2.9", [hex: :phoenix_html, repo: "hexpm", optional: true]}, {:plug, "~> 1.0", [hex: :plug, repo: "hexpm", optional: false]}], "hexpm"},
"phoenix_html": {:hex, :phoenix_html, "2.13.1", "fa8f034b5328e2dfa0e4131b5569379003f34bc1fafdaa84985b0b9d2f12e68b", [:mix], [{:plug, "~> 1.5", [hex: :plug, repo: "hexpm", optional: false]}], "hexpm"},
"phoenix_pubsub": {:hex, :phoenix_pubsub, "1.1.1", "6668d787e602981f24f17a5fbb69cc98f8ab085114ebfac6cc36e10a90c8e93c", [:mix], [], "hexpm"},
+ "pleroma_job_queue": {:hex, :pleroma_job_queue, "0.2.0", "879e660aa1cebe8dc6f0aaaa6aa48b4875e89cd961d4a585fd128e0773b31a18", [:mix], [], "hexpm"},
"plug": {:hex, :plug, "1.7.2", "d7b7db7fbd755e8283b6c0a50be71ec0a3d67d9213d74422d9372effc8e87fd1", [:mix], [{:mime, "~> 1.0", [hex: :mime, repo: "hexpm", optional: false]}, {:plug_crypto, "~> 1.0", [hex: :plug_crypto, repo: "hexpm", optional: false]}], "hexpm"},
"plug_cowboy": {:hex, :plug_cowboy, "2.0.1", "d798f8ee5acc86b7d42dbe4450b8b0dadf665ce588236eb0a751a132417a980e", [:mix], [{:cowboy, "~> 2.5", [hex: :cowboy, repo: "hexpm", optional: false]}, {:plug, "~> 1.7", [hex: :plug, repo: "hexpm", optional: false]}], "hexpm"},
"plug_crypto": {:hex, :plug_crypto, "1.0.0", "18e49317d3fa343f24620ed22795ec29d4a5e602d52d1513ccea0b07d8ea7d4d", [:mix], [], "hexpm"},
+++ /dev/null
-# Pleroma: A lightweight social networking server
-# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
-# SPDX-License-Identifier: AGPL-3.0-only
-
-defmodule Pleroma.JobsTest do
- use ExUnit.Case, async: true
-
- alias Jobs.WorkerMock
- alias Pleroma.Jobs
-
- setup do
- state = %{
- queues: Enum.into([Jobs.create_queue(:testing)], %{}),
- refs: %{}
- }
-
- [state: state]
- end
-
- test "creates queue" do
- queue = Jobs.create_queue(:foobar)
-
- assert {:foobar, set} = queue
- assert :set == elem(set, 0) |> elem(0)
- end
-
- test "enqueues an element according to priority" do
- queue = [%{item: 1, priority: 2}]
-
- new_queue = Jobs.enqueue_sorted(queue, 2, 1)
- assert new_queue == [%{item: 2, priority: 1}, %{item: 1, priority: 2}]
-
- new_queue = Jobs.enqueue_sorted(queue, 2, 3)
- assert new_queue == [%{item: 1, priority: 2}, %{item: 2, priority: 3}]
- end
-
- test "pop first item" do
- queue = [%{item: 2, priority: 1}, %{item: 1, priority: 2}]
-
- assert {2, [%{item: 1, priority: 2}]} = Jobs.queue_pop(queue)
- end
-
- test "enqueue a job", %{state: state} do
- assert {:noreply, new_state} =
- Jobs.handle_cast({:enqueue, :testing, WorkerMock, [:test_job, :foo, :bar], 3}, state)
-
- assert %{queues: %{testing: {running_jobs, []}}, refs: _} = new_state
- assert :sets.size(running_jobs) == 1
- assert [ref] = :sets.to_list(running_jobs)
- assert %{refs: %{^ref => :testing}} = new_state
- end
-
- test "max jobs setting", %{state: state} do
- max_jobs = Pleroma.Config.get([Jobs, :testing, :max_jobs])
-
- {:noreply, state} =
- Enum.reduce(1..(max_jobs + 1), {:noreply, state}, fn _, {:noreply, state} ->
- Jobs.handle_cast({:enqueue, :testing, WorkerMock, [:test_job, :foo, :bar], 3}, state)
- end)
-
- assert %{
- queues: %{
- testing:
- {running_jobs, [%{item: {WorkerMock, [:test_job, :foo, :bar]}, priority: 3}]}
- }
- } = state
-
- assert :sets.size(running_jobs) == max_jobs
- end
-
- test "remove job after it finished", %{state: state} do
- {:noreply, new_state} =
- Jobs.handle_cast({:enqueue, :testing, WorkerMock, [:test_job, :foo, :bar], 3}, state)
-
- %{queues: %{testing: {running_jobs, []}}} = new_state
- [ref] = :sets.to_list(running_jobs)
-
- assert {:noreply, %{queues: %{testing: {running_jobs, []}}, refs: %{}}} =
- Jobs.handle_info({:DOWN, ref, :process, nil, nil}, new_state)
-
- assert :sets.size(running_jobs) == 0
- end
-end
+++ /dev/null
-# Pleroma: A lightweight social networking server
-# Copyright © 2017-2019 Pleroma Authors <https://pleroma.social/>
-# SPDX-License-Identifier: AGPL-3.0-only
-
-defmodule Pleroma.Jobs.WorkerMock do
- require Logger
-
- def perform(:test_job, arg, arg2) do
- Logger.debug({:perform, :test_job, arg, arg2})
- end
-
- def perform(:test_job, payload) do
- Logger.debug({:perform, :test_job, payload})
- end
-
- def test_job(payload) do
- Pleroma.Jobs.enqueue(:testing, __MODULE__, [:test_job, payload])
- end
-end
projects / akkoma / commitdiff