Merge pull request #1 from abmBispo/schedulers-improvement

Schedulers improvement

Author: abmBispo

lib/elixir_queue/job.ex

@@ -1,3 +1,3 @@
 defmodule ElixirQueue.Job do
-  defstruct [:mod, :func, :args]
+  defstruct [:mod, :func, :args, :retry_attempts]
 end

lib/elixir_queue/queue.ex

@@ -13,15 +13,28 @@ defmodule ElixirQueue.Queue do
   def init(_opts), do: {:ok, {}}
 
   @doc ~S"""
-  Get next job to be processed
+  Enqueues job to be processed
   ## Examples
       iex> ElixirQueue.Queue.clear()
       iex> ElixirQueue.Queue.perform_later(Enum, :reverse, [[1,2,3,4,5]])
       :ok
   """
   @spec perform_later(atom, atom, list(any)) :: :ok
   def perform_later(mod, func, args \\ []) do
-    job = %Job{mod: mod, func: func, args: args}
+    job = %Job{mod: mod, func: func, args: args, retry_attempts: 0}
+    GenServer.call(Queue, {:perform_later, job})
+  end
+
+  @doc ~S"""
+  Enqueues job to be processed
+  ## Examples
+      iex> ElixirQueue.Queue.clear()
+      iex> job = %ElixirQueue.Job{mod: Enum, func: :reverse, args: [[1,2,3,4,5]]}
+      iex> ElixirQueue.Queue.perform_later(job)
+      :ok
+  """
+  @spec perform_later(ElixirQueue.Job.t()) :: :ok
+  def perform_later(job = %Job{}) do
     GenServer.call(Queue, {:perform_later, job})
   end
 
@@ -41,6 +54,7 @@ defmodule ElixirQueue.Queue do
   @spec fetch :: {:ok, any} | {:error, :empty}
   def fetch, do: GenServer.call(Queue, :fetch)
 
+  @spec clear :: :ok
   def clear, do: GenServer.call(Queue, :clear)
 
   @impl true

lib/elixir_queue/worker.ex

@@ -1,44 +1,45 @@
 defmodule ElixirQueue.Worker do
-  use GenServer, restart: :temporary
+  use Agent, restart: :temporary
   require Logger
 
-  alias ElixirQueue.Job
+  alias ElixirQueue.{
+    Job,
+    WorkerPool
+  }
 
+  @spec start_link(any) :: {:error, any} | {:ok, pid}
   @doc """
   Starts a new worker.
   """
-  @spec start_link(any) :: {:error, any} | {:ok, pid}
-  def start_link(opts), do: GenServer.start_link(__MODULE__, opts)
-
-  @impl true
-  @spec init(any) :: {:ok, Job.t()}
-  def init(_opts), do: {:ok, %Job{}}
-
-  @impl true
-  def handle_call({:start, job}, _from, _state),
-    do: {:reply, :ok, job}
-
-  def handle_call({:perform, %Job{mod: mod, func: func, args: args}}, _from, state),
-    do: {:reply, apply(mod, func, args), state}
+  def start_link(_opts) do
+    Agent.start_link(fn -> %Job{} end)
+  end
 
-  def handle_call(:halt, _from, _state),
-    do: {:reply, :ok, %Job{}}
+  @spec get(pid()) :: Job.t()
+  def get(worker), do: Agent.get(worker, fn state -> state end)
 
-  def handle_call(:idle?, _from, state),
-    do: {:reply, state == %Job{}, state}
+  @spec idle?(pid()) :: boolean
+  def idle?(worker), do: Agent.get(worker, fn state -> state end) == %Job{}
 
-  @spec perform(pid(), Job.t()) :: any()
-  def perform(worker, job) do
-    GenServer.call(worker, {:start, job})
-    result = GenServer.call(worker, {:perform, job})
-    GenServer.call(worker, :halt)
+  @spec perform(pid(), ElixirQueue.Job.t()) :: any()
+  def perform(worker, job = %Job{mod: mod, func: func, args: args}) do
+    start_job(worker, job)
+    result = apply(mod, func, args)
+    end_job(worker)
 
     unless Mix.env() == :test,
       do: Logger.info("JOB DONE SUCCESSFULLY #{inspect(job)} ====> RESULT: #{inspect(result)}")
 
-    {:ok, result}
+    {worker, job, result}
+  end
+
+  defp start_job(worker, job) do
+    Agent.update(worker, fn _ -> job end)
+    WorkerPool.backup_worker(worker, job)
   end
 
-  @spec idle?(pid()) :: any
-  def idle?(worker), do: GenServer.call(worker, :idle?)
+  defp end_job(worker) do
+    WorkerPool.clean_worker_backup(worker)
+    Agent.update(worker, fn _ -> %Job{} end)
+  end
 end

lib/elixir_queue/worker_pool.ex

@@ -5,7 +5,8 @@ defmodule ElixirQueue.WorkerPool do
   alias ElixirQueue.{
     WorkerSupervisor,
     WorkerPool,
-    Worker
+    Worker,
+    Queue
   }
 
   def start_link(opts) do
@@ -17,60 +18,65 @@ defmodule ElixirQueue.WorkerPool do
   @spec init(any) :: {:ok, %{failed_jobs: [], pids: [], successful_jobs: []}}
   def init(_opts) do
     pids =
-Fail proff      for _ <- 1..Application.fetch_env!(:elixir_queue, :workers) do
+      for _ <- 1..Application.fetch_env!(:elixir_queue, :workers) do
         {:ok, pid} = DynamicSupervisor.start_child(WorkerSupervisor, Worker)
-        ref = Process.monitor(pid)
-        {pid, ref}
+        Process.monitor(pid)
+        pid
       end
 
+    :ets.new(:worker_backup, [:set, :protected, :named_table])
+
     {:ok, %{pids: pids, successful_jobs: [], failed_jobs: []}}
   end
 
-  # Server side functions
   @impl true
-  def handle_call(:workers, _from, state), do: {:reply, Enum.map(state.pids, &elem(&1, 0)), state}
+  def handle_call(:workers, _from, state) do
+    {:reply, state.pids, state}
+  end
 
   def handle_call(:failed_jobs, _from, state), do: {:reply, state.failed_jobs, state}
 
   def handle_call(:successful_jobs, _from, state), do: {:reply, state.successful_jobs, state}
 
-  def handle_call({:add_successful_job, worker, job, result}, _from, state),
-    do: {
-      :reply,
-      :ok,
-      Map.put(state, :successful_jobs, [{worker, job, result} | state.successful_jobs])
-    }
+  def handle_call({:add_successful_job, worker, job, result}, _from, state) do
+    {:reply, :ok,
+     Map.put(state, :successful_jobs, [{worker, job, result} | state.successful_jobs])}
+  end
 
-  def handle_call({:add_failed_job, worker, job, err}, _from, state),
-    do: {:reply, :ok, Map.put(state, :failed_jobs, [{worker, job, err} | state.failed_jobs])}
+  def handle_call({:backup_worker, worker, job}, _from, state) do
+    :ets.insert(:worker_backup, {worker, job})
+    {:reply, :ok, state}
+  end
+
+  def handle_call({:clean_worker_backup, worker}, _from, state) do
+    :ets.delete(:worker_backup, worker)
+    {:reply, :ok, state}
+  end
 
   @impl true
-  def handle_info({:DOWN, _ref, :process, dead_worker, reason}, state) do
-    {:ok, worker} = DynamicSupervisor.start_child(WorkerSupervisor, Worker)
-    worker_reference = Process.monitor(worker)
+  def handle_info({:DOWN, _, :process, dead_worker, reason}, state) do
+    {:ok, pid} = DynamicSupervisor.start_child(WorkerSupervisor, Worker)
+    Process.monitor(pid)
     pids = Enum.filter(state.pids, &(&1 != dead_worker))
 
-    reason |> IO.inspect(label: "reason")
+    {^dead_worker, backuped_job} =
+      :ets.lookup(:worker_backup, dead_worker)
+      |> List.first()
 
-    unless Mix.env() == :test,
-      do: Logger.error("Unexpected worker error:
-          Worker #{inspect(dead_worker)} received EXIT SIGNAL.
-          It have been replaced by #{inspect(worker)} worker.
-          All the job progress was lost and job failed.
-          By default job returned to the end of queue and will be performed again later.
-        ")
+    backuped_job = Map.put(backuped_job, :retry_attempts, backuped_job.retry_attempts + 1)
 
-    {:noreply, Map.put(state, :pids, [{worker, worker_reference} | pids])}
-  end
+    state =
+      state
+      |> Map.put(:pids, [pid | pids])
+      |> Map.put(:failed_jobs, [{dead_worker, backuped_job, reason} | state.failed_jobs])
 
-  def handle_info(_msg, state),
-    do: {:noreply, state}
+    if backuped_job.retry_attempts < Application.fetch_env!(:elixir_queue, :retries),
+      do: Queue.perform_later(backuped_job)
 
-  # Client side functions #
+    {:noreply, state}
+  end
 
-  @doc """
-  Returns _workers_ `PID`s kept in the state.
-  """
+  # Client side functions
   @spec workers :: list()
   def workers, do: GenServer.call(__MODULE__, :workers)
 
@@ -83,13 +89,17 @@ Fail proff      for _ <- 1..Application.fetch_env!(:elixir_queue, :workers) do
   @spec successful_jobs :: list()
   def successful_jobs, do: GenServer.call(__MODULE__, :successful_jobs)
 
-  @spec add_successful_job(pid(), ElixirQueue.Job.t(), any) :: :ok
-  def add_successful_job(worker, job, result),
+  @spec add_successful_job({pid(), ElixirQueue.Job.t(), any}) :: :ok
+  def add_successful_job({worker, job, result}),
     do: GenServer.call(__MODULE__, {:add_successful_job, worker, job, result})
 
-  @spec add_failed_job(pid(), ElixirQueue.Job.t(), any) :: :ok
-  def add_failed_job(worker, job, err),
-    do: GenServer.call(__MODULE__, {:add_failed_job, worker, job, err})
+  @spec backup_worker(pid(), ElixirQueue.Job.t()) :: true
+  def backup_worker(worker, job),
+    do: GenServer.call(__MODULE__, {:backup_worker, worker, job})
+
+  @spec clean_worker_backup(pid()) :: true
+  def clean_worker_backup(worker),
+    do: GenServer.call(__MODULE__, {:clean_worker_backup, worker})
 
   @spec idle_worker :: pid()
   def idle_worker do
@@ -102,10 +112,9 @@ Fail proff      for _ <- 1..Application.fetch_env!(:elixir_queue, :workers) do
   @spec perform(ElixirQueue.Job.t()) :: no_return()
   def perform(job) do
     worker = WorkerPool.idle_worker()
-
     Task.start(fn ->
-      {:ok, result} = Worker.perform(worker, job)
-      WorkerPool.add_successful_job(worker, job, result)
+      Worker.perform(worker, job)
+      |> WorkerPool.add_successful_job()
     end)
   end
 end

lib/fake_functions/fake.ex

@@ -25,7 +25,7 @@ defmodule ElixirQueue.Fake do
   @spec populate :: :ok
   def populate do
     for _ <- 0..999, do: Queue.perform_later(Fake, :task, [3])
-    :ok  
+    :ok
   end
 
   @spec spec :: %{

mix.exs

@@ -15,7 +15,10 @@ defmodule ElixirQueue.MixProject do
   def application do
     [
       extra_applications: [:logger],
-      env: [workers: System.schedulers_online()],
+      env: [
+        retries: 3,
+        workers: System.schedulers_online()
+      ],
       mod: {ElixirQueue.Application, []}
     ]
   end