Packages
fnord
0.4.32
0.9.40
0.9.39
0.9.38
0.9.37
0.9.36
0.9.35
0.9.34
0.9.33
0.9.32
0.9.31
0.9.30
0.9.29
0.9.28
0.9.27
0.9.26
0.9.25
0.9.24
0.9.23
0.9.22
0.9.21
0.9.20
0.9.19
0.9.18
0.9.17
0.9.16
0.9.15
0.9.14
0.9.13
0.9.12
0.9.11
0.9.10
0.9.9
0.9.8
0.9.7
0.9.6
0.9.5
0.9.4
0.9.3
0.9.2
0.9.1
0.9.0
0.8.99
0.8.98
0.8.97
0.8.96
0.8.95
0.8.94
0.8.93
0.8.92
0.8.91
0.8.90
0.8.89
0.8.88
0.8.87
0.8.86
0.8.85
0.8.84
0.8.83
0.8.82
0.8.81
0.8.80
0.8.79
0.8.78
0.8.77
0.8.76
0.8.75
0.8.74
0.8.73
0.8.72
0.8.71
0.8.70
0.8.69
0.8.68
0.8.67
0.8.66
0.8.65
0.8.64
0.8.63
0.8.62
0.8.61
0.8.60
0.8.59
0.8.58
0.8.57
0.8.56
0.8.55
0.8.54
0.8.53
0.8.52
0.8.51
0.8.50
0.8.49
0.8.48
0.8.47
0.8.46
0.8.45
0.8.44
0.8.43
0.8.42
0.8.41
0.8.40
0.8.39
0.8.38
0.8.37
0.8.36
0.8.35
0.8.34
0.8.33
0.8.32
0.8.31
0.8.30
0.8.29
0.8.27
0.8.26
0.8.25
0.8.24
0.8.23
0.8.22
0.8.21
0.8.20
0.8.19
0.8.18
0.8.17
0.8.16
0.8.15
0.8.14
0.8.13
0.8.12
0.8.11
0.8.1
0.8.0
0.7.24
0.7.23
0.7.22
0.7.21
0.7.20
0.7.19
0.7.18
0.7.17
0.7.16
0.7.15
0.7.14
0.7.13
0.7.12
0.7.11
0.7.10
0.7.9
0.7.8
0.7.7
0.7.6
0.7.5
0.7.3
0.7.2
0.7.1
0.7.0
0.6.9
0.6.8
0.6.7
0.6.6
0.6.5
0.6.4
0.6.3
0.6.1
0.6.0
0.5.9
0.5.8
0.5.7
0.5.6
0.5.5
0.5.4
0.5.3
0.5.2
0.5.1
0.5.0
0.4.44
0.4.43
0.4.42
0.4.41
0.4.40
0.4.39
0.4.38
0.4.37
0.4.36
0.4.35
0.4.34
0.4.33
0.4.32
0.4.30
0.4.29
0.4.28
0.4.27
0.4.26
0.4.25
0.4.24
0.4.23
0.4.22
0.4.21
0.4.20
0.4.19
0.4.18
0.4.17
0.4.16
0.4.15
0.4.14
0.4.13
0.4.12
0.4.11
0.4.10
0.4.9
0.4.8
0.4.7
0.4.6
0.4.5
0.4.4
0.4.3
0.4.2
0.4.1
0.4.0
0.3.0
0.2.0
0.1.0
AI code archaeology
Current section
Files
Jump to
Current section
Files
lib/queue.ex
defmodule Queue do
use GenServer
@moduledoc """
A module that implements a process pool using a GenServer.
"""
# -----------------------------------------------------------------------------
# Client API
# -----------------------------------------------------------------------------
@doc """
Starts the Queue with the given callback function and maximum number of workers.
Returns {:ok, pid}
"""
def start_link(callback) do
max_workers = Application.get_env(:fnord, :concurrency)
GenServer.start_link(__MODULE__, {callback, max_workers})
end
@doc """
Adds a job to the queue and returns a Task.
"""
def queue(pid, args) do
task =
Task.async(fn ->
receive do
{:result, result} -> result
end
end)
GenServer.cast(pid, {:queue_job, args, task.pid})
task
end
@doc """
Shuts down the queue, preventing new jobs from being added.
"""
def shutdown(pid) do
GenServer.cast(pid, :shutdown)
end
@doc """
Waits until the queue is empty and all workers have exited.
"""
def join(pid) do
GenServer.call(pid, :join, :infinity)
end
@doc """
Queues an Enum of jobs, executes them, and returns the results in order.
"""
def map(enum, pid) do
enum
|> Enum.map(&queue(pid, &1))
|> Enum.map(&Task.await(&1, :infinity))
end
# -----------------------------------------------------------------------------
# Server Callbacks
# -----------------------------------------------------------------------------
def init({callback, max_workers}) do
state = %{
jobs: :queue.new(),
max_workers: max_workers,
callback: callback,
workers: [],
shutdown: false,
waiting: []
}
state = start_workers(state)
{:ok, state}
end
def handle_cast({:queue_job, args, task_pid}, state) do
if state.shutdown do
send(task_pid, {:result, {:error, :queue_shutdown}})
{:noreply, state}
else
jobs = :queue.in({args, task_pid}, state.jobs)
{:noreply, %{state | jobs: jobs}}
end
end
def handle_cast(:shutdown, state) do
{:noreply, %{state | shutdown: true}}
end
def handle_cast({:worker_exited, pid}, state) do
workers = List.delete(state.workers, pid)
state = %{state | workers: workers}
check_if_done(state)
end
def handle_call({:request_job, _worker_pid}, _from, state) do
if :queue.is_empty(state.jobs) do
if state.shutdown do
{:reply, :shutdown, state}
else
{:reply, :no_job, state}
end
else
{{:value, {args, task_pid}}, jobs} = :queue.out(state.jobs)
state = %{state | jobs: jobs}
{:reply, {:job, {args, task_pid}}, state}
end
end
def handle_call(:join, from, state) do
if :queue.is_empty(state.jobs) and state.shutdown and length(state.workers) == 0 do
{:reply, :ok, state}
else
waiting = [from | state.waiting]
{:noreply, %{state | waiting: waiting}}
end
end
# -----------------------------------------------------------------------------
# Helper Functions
# -----------------------------------------------------------------------------
defp start_workers(state) do
workers =
Enum.map(1..state.max_workers, fn _ ->
{:ok, pid} = Queue.Worker.start_link(self(), state.callback)
pid
end)
%{state | workers: workers}
end
defp check_if_done(state) do
if state.shutdown and :queue.is_empty(state.jobs) and length(state.workers) == 0 do
Enum.each(state.waiting, fn from -> GenServer.reply(from, :ok) end)
{:noreply, %{state | waiting: []}}
else
{:noreply, state}
end
end
end
defmodule Queue.Worker do
@moduledoc false
def start_link(queue_pid, callback) do
pid = spawn_link(__MODULE__, :loop, [queue_pid, callback])
{:ok, pid}
end
def loop(queue_pid, callback) do
job = GenServer.call(queue_pid, {:request_job, self()})
case job do
{:job, {args, task_pid}} ->
result = callback.(args)
send(task_pid, {:result, result})
loop(queue_pid, callback)
:no_job ->
:timer.sleep(100)
loop(queue_pid, callback)
:shutdown ->
GenServer.cast(queue_pid, {:worker_exited, self()})
:ok
end
end
end