🎓 A1Learner
🔍
0 of 46 lessons completed

Elixir Project: Job Scheduler

A job scheduler is a process that runs named jobs at planned steps. Each tick moves the step forward by one and returns the jobs due at that step.

Ticks run jobs in planned step order

A backup is planned for step 1 and a report for step 2. Each tick returns only the job due at that step.

defmodule Scheduler do
  use GenServer
  def start_link(opts), do: GenServer.start_link(__MODULE__, %{step: 0, jobs: []}, opts)
  def schedule(pid, name, at_step), do: GenServer.call(pid, {:schedule, name, at_step})
  def tick(pid), do: GenServer.call(pid, :tick)
  def step(pid), do: GenServer.call(pid, :step)
  def init(state), do: {:ok, state}
  def handle_call({:schedule, name, at_step}, _from, state), do: {:reply, :ok, %{state | jobs: state.jobs ++ [{name, at_step, nil}]}}
  def handle_call(:tick, _from, state) do
    next_step = state.step + 1
    due = for {name, at_step, _interval} <- state.jobs, at_step == next_step, do: name
    next_jobs = Enum.flat_map(state.jobs, fn job ->
      case job do
        {_name, at_step, nil} when at_step == next_step -> []
        {name, at_step, interval} when at_step == next_step -> [{name, at_step + interval, interval}]
        other -> [other]
      end
    end)
    {:reply, due, %{state | step: next_step, jobs: next_jobs}}
  end
  def handle_call(:step, _from, state), do: {:reply, state.step, state}
end
{:ok, sched} = Scheduler.start_link([])
Scheduler.schedule(sched, "report", 2)
Scheduler.schedule(sched, "backup", 1)
IO.inspect(Scheduler.tick(sched))
IO.inspect(Scheduler.tick(sched))
IO.inspect(Scheduler.step(sched))
GenServer.stop(sched)

A job due later waits through early ticks

A report is planned for step 3. Two ticks return nothing, and the report stays pending.

defmodule Scheduler do
  use GenServer
  def start_link(opts), do: GenServer.start_link(__MODULE__, %{step: 0, jobs: []}, opts)
  def schedule(pid, name, at_step), do: GenServer.call(pid, {:schedule, name, at_step})
  def tick(pid), do: GenServer.call(pid, :tick)
  def pending(pid), do: GenServer.call(pid, :pending)
  def step(pid), do: GenServer.call(pid, :step)
  def init(state), do: {:ok, state}
  def handle_call({:schedule, name, at_step}, _from, state), do: {:reply, :ok, %{state | jobs: state.jobs ++ [{name, at_step, nil}]}}
  def handle_call(:tick, _from, state) do
    next_step = state.step + 1
    due = for {name, at_step, _interval} <- state.jobs, at_step == next_step, do: name
    next_jobs = Enum.flat_map(state.jobs, fn job ->
      case job do
        {_name, at_step, nil} when at_step == next_step -> []
        {name, at_step, interval} when at_step == next_step -> [{name, at_step + interval, interval}]
        other -> [other]
      end
    end)
    {:reply, due, %{state | step: next_step, jobs: next_jobs}}
  end
  def handle_call(:pending, _from, state), do: {:reply, Enum.map(state.jobs, fn {name, _at, _interval} -> name end), state}
  def handle_call(:step, _from, state), do: {:reply, state.step, state}
end
{:ok, sched} = Scheduler.start_link([])
Scheduler.schedule(sched, "report", 3)
IO.inspect(Scheduler.tick(sched))
IO.inspect(Scheduler.tick(sched))
IO.inspect(Scheduler.pending(sched))
IO.inspect(Scheduler.step(sched))
GenServer.stop(sched)

A recurring job plans itself again

A heartbeat runs every two steps, starting at step 1. It runs on steps 1 and 3, and stays pending after step 3.

defmodule Scheduler do
  use GenServer
  def start_link(opts), do: GenServer.start_link(__MODULE__, %{step: 0, jobs: []}, opts)
  def schedule_every(pid, name, first_step, interval), do: GenServer.call(pid, {:schedule_every, name, first_step, interval})
  def tick(pid), do: GenServer.call(pid, :tick)
  def pending(pid), do: GenServer.call(pid, :pending)
  def init(state), do: {:ok, state}
  def handle_call({:schedule_every, name, first_step, interval}, _from, state), do: {:reply, :ok, %{state | jobs: state.jobs ++ [{name, first_step, interval}]}}
  def handle_call(:tick, _from, state) do
    next_step = state.step + 1
    due = for {name, at_step, _interval} <- state.jobs, at_step == next_step, do: name
    next_jobs = Enum.flat_map(state.jobs, fn job ->
      case job do
        {_name, at_step, nil} when at_step == next_step -> []
        {name, at_step, interval} when at_step == next_step -> [{name, at_step + interval, interval}]
        other -> [other]
      end
    end)
    {:reply, due, %{state | step: next_step, jobs: next_jobs}}
  end
  def handle_call(:pending, _from, state), do: {:reply, Enum.map(state.jobs, fn {name, _at, _interval} -> name end), state}
end
{:ok, sched} = Scheduler.start_link([])
Scheduler.schedule_every(sched, "heartbeat", 1, 2)
IO.inspect(Scheduler.tick(sched))
IO.inspect(Scheduler.tick(sched))
IO.inspect(Scheduler.tick(sched))
IO.inspect(Scheduler.pending(sched))
GenServer.stop(sched)

Example

defmodule Scheduler do
  use GenServer
  def start_link(opts), do: GenServer.start_link(__MODULE__, %{step: 0, jobs: []}, opts)
  def schedule(pid, name, at_step), do: GenServer.call(pid, {:schedule, name, at_step})
  def schedule_every(pid, name, first_step, interval), do: GenServer.call(pid, {:schedule_every, name, first_step, interval})
  def tick(pid), do: GenServer.call(pid, :tick)
  def pending(pid), do: GenServer.call(pid, :pending)
  def step(pid), do: GenServer.call(pid, :step)
  def init(state), do: {:ok, state}
  def handle_call({:schedule, name, at_step}, _from, state), do: {:reply, :ok, %{state | jobs: state.jobs ++ [{name, at_step, nil}]}}
  def handle_call({:schedule_every, name, first_step, interval}, _from, state), do: {:reply, :ok, %{state | jobs: state.jobs ++ [{name, first_step, interval}]}}
  def handle_call(:tick, _from, state) do
    next_step = state.step + 1
    due = for {name, at_step, _interval} <- state.jobs, at_step == next_step, do: name
    next_jobs = Enum.flat_map(state.jobs, fn job ->
      case job do
        {_name, at_step, nil} when at_step == next_step -> []
        {name, at_step, interval} when at_step == next_step -> [{name, at_step + interval, interval}]
        other -> [other]
      end
    end)
    {:reply, due, %{state | step: next_step, jobs: next_jobs}}
  end
  def handle_call(:pending, _from, state), do: {:reply, Enum.map(state.jobs, fn {name, _at, _interval} -> name end), state}
  def handle_call(:step, _from, state), do: {:reply, state.step, state}
end
{:ok, sched} = Scheduler.start_link([])
Scheduler.schedule(sched, "backup", 1)
Scheduler.schedule(sched, "report", 2)
Scheduler.schedule_every(sched, "heartbeat", 1, 2)
IO.inspect(Scheduler.tick(sched))
IO.inspect(Scheduler.tick(sched))
IO.inspect(Scheduler.tick(sched))
IO.inspect(Scheduler.pending(sched))
IO.inspect(Scheduler.step(sched))
GenServer.stop(sched)

Key points

  • Job: a named piece of work planned for a step.
  • Step: the scheduler clock, counted in ticks.
  • Tick: one advance of the step that returns the due jobs.
  • Due: a job whose planned step matches the current step.
  • Pending: a job planned for a later step.
  • Recurring: a job that plans itself again after it runs.

In this example, a backup and a heartbeat run on step 1, a report runs on step 2, and the heartbeat runs again on step 3 while staying pending for its next run.

Note

a job due later never runs on an earlier tick.

Try it Yourself

Edit the code, then press Run.

Running is simulated

Exercise

A job is planned for step 3 and the scheduler is at step 2. What does the next tick return?