Specs

async_perform(pool(), function()) :: job() | not_found()

Performs an async job that can be awaited on for the result.

Returns the OddJob.Job.t() struct if the pool server exists at the time of calling, or {:error, :not_found} if it doesn't.

Functions similarly to Task.async/1 and Task.await/2 with some tweaks. Since the workers in the pool already exist when this function is called and may be busy performing other jobs, it is unknown at the time of calling when the job will be performed and by which worker process.

An OddJob.Async.Proxy is dynamically created and linked and monitored by the caller. The Proxy handles the transactions with the pool and is notified when a worker is assigned. The proxy links and monitors the worker immediately before performance of the job, forwarding any exits or crashes to the caller. The worker returns the results to the proxy, which then passes them to the caller with a {ref, result} message. The proxy immediately exits with reason :normal after the results are sent back to the caller.

See await/2 and await_many/2 for functions to retrieve the results, and the Task module for more information on the async/await patter and when you may want to use it.

Examples

iex> job = OddJob.async_perform(OddJob.Pool, fn -> :math.exp(100) end)
iex> OddJob.await(job)
2.6881171418161356e43

iex> OddJob.async_perform(:does_not_exist, fn -> "never called" end)
{:error, :not_found}

Specs

async_perform_many(pool(), list() | map(), function()) :: [job()] | not_found()

Sends a collection of async jobs to the pool.

Returns a list of OddJob.Job.t() structs if the pool server exists at the time of calling, or {:error, :not_found} if it doesn't.

Enumerates over the collection, using each member as the argument to an anonymous function with an arity of 1. See async_perform/2 and perform/2 for more information.

There's a limit to the number of jobs that can be started with this function that roughly equals the BEAM's process limit.

Examples

iex> jobs = OddJob.async_perform_many(OddJob.Pool, 1..5, fn x -> x ** 2 end)
iex> OddJob.await_many(jobs)
[1, 4, 9, 16, 25]

iex> OddJob.async_perform_many(:does_not_exist, ["never", "called"], fn x -> x end)
{:error, :not_found}

Specs

await(job(), timeout()) :: term()

Awaits on an async job reply and returns the results.

The current process and the worker process are linked during execution of the job by an OddJob.Async.Proxy. In case the worker process dies during execution, the current process will exit with the same reason as the worker.

A timeout, in milliseconds or :infinity, can be given with a default value of 5000. If the timeout is exceeded, then the current process will exit. If the worker process is linked to the current process which is the case when a job is started with async, then the worker process will also exit.

This function assumes the proxy's monitor is still active or the monitor's :DOWN message is in the message queue. If it has been demonitored, or the message already received, this function will wait for the duration of the timeout awaiting the message.

Examples

iex> OddJob.async_perform(OddJob.Pool, fn -> :math.log(2.6881171418161356e43) end)
...> |> OddJob.await()
100.0

Specs

await_many([job()], timeout()) :: [term()]

Awaits replies from multiple async jobs and returns them in a list.

This function receives a list of jobs and waits for their replies in the given time interval. It returns a list of the results, in the same order as the jobs supplied in the jobs input argument.

The current process and each worker process are linked during execution of the job by an OddJob.Async.Proxy. If any of the worker processes dies, the caller process will exit with the same reason as that worker.

A timeout, in milliseconds or :infinity, can be given with a default value of 5000. If the timeout is exceeded, then the caller process will exit. Any worker processes that are linked to the caller process (which is the case when a job is started with async_perform/2) will also exit.

This function assumes the proxies' monitors are still active or the monitor's :DOWN message is in the message queue. If any jobs have been demonitored, or the message already received, this function will wait for the duration of the timeout.

Examples

iex> job1 = OddJob.async_perform(OddJob.Pool, fn -> 2 ** 2 end)
iex> job2 = OddJob.async_perform(OddJob.Pool, fn -> 3 ** 2 end)
iex> [job1, job2] |> OddJob.await_many()
[4, 9]

Specs

cancel_timer(reference()) :: non_neg_integer() | false

Cancels a scheduled job.

timer_ref is the unique reference returned by perform_at/3 or perform_after/3. This function returns the number of milliseconds left in the timer when cancelled, or false if the timer already expired. If the return is false you can assume that the job has already been sent to the pool for execution.

NOTE: Cancelling the timer with this function ensures that the job is never executed and that the scheduler process is exited and not left "hanging". Using Process.cancel_timer/1 will also cancel execution, but may leave hanging processes. A hanging scheduler process will eventually timeout, but not until one second after the expiration of the original timer.

Examples

iex> ref = OddJob.perform_after(500, OddJob.Pool, fn -> :never end)
iex> time = OddJob.cancel_timer(ref)
iex> is_integer(time)
true

iex> ref = OddJob.perform_after(10, OddJob.Pool, fn -> :never end)
iex> Process.sleep(11)
iex> OddJob.cancel_timer(ref)
false

Specs

child_spec(options()) :: child_spec()

Returns a specification to start this module under a supervisor.

Examples

iex> children = [OddJob.child_spec(name: MyApp.Media)]
[%{id: MyApp.Media, start: {OddJob.Pool, :start_link, [[name: MyApp.Media]]}, type: :supervisor}]
iex> {:ok, _pid} = Supervisor.start_link(children, strategy: :one_for_one)
iex> (MyApp.Media |> OddJob.workers() |> length()) == System.schedulers_online()
true

Normally you would start an OddJob pool under a supervision tree with a child specification tuple and not call child_spec/1 directly.

children = [{OddJob, name: MyApp.Media}]
Supervisor.start_link(children, strategy: :one_for_one)

You must pass a keyword list of options to child_spec/1, or as the second element of the child specification tuple.

Options

• :name - An atom that names the pool. This argument is required.

• :pool_size - A positive integer, the number of concurrent workers in the pool. Defaults to 5 or your application's config value.

• :max_restarts - A positive integer, the number of worker restarts allowed in a given timeframe before all of the workers are restarted. Set a higher number if your jobs have a high rate of expected failure. Defaults to 5 or your application's config value.

• :max_seconds - A positive integer, the timeframe in seconds in which max_restarts applies. Defaults to 3 or your application's config value. See Supervisor for more info on restart intensity options.

See the Supervisor module for more about child specs.

Specs

perform(pool(), function()) :: :ok | not_found()

Performs a fire and forget job.

Casts the function to the job pool, and if a worker is available it is performed immediately. If no workers are available the job is added to the back of a FIFO queue to be performed as soon as a worker is available.

Returns :ok if the pool process exists at the time of calling, or {:error, :not_found} if it doesn't. The job is casted to the pool, so there can be no guarantee that the pool process will still exist if it crashes between the beginning of the function call and when the message is received.

Use this function for interaction with other processes or the outside world where you don't care about the return results or whether it succeeds.

Examples

iex> parent = self()
iex> :ok = OddJob.perform(OddJob.Pool, fn -> send(parent, :hello) end)
iex> receive do
...>   msg -> msg
...> end
:hello

iex> OddJob.perform(:does_not_exist, fn -> "never called" end)
{:error, :not_found}

Specs

perform_after(timer(), pool(), function()) :: reference() | not_found()

Sends a job to the pool after the given timer has elapsed.

timer is an integer that indicates the number of milliseconds that should elapse before the job is sent to the pool. The timed message is executed under a separate supervised process, so if the current process crashes the job will still be performed. A timer reference is returned, which can be read with Process.read_timer/1 or cancelled with OddJob.cancel_timer/1. Returns {:error, :not_found} if the pool server does not exist at the time this function is called.

Examples

timer_ref = OddJob.perform_after(5000, OddJob.Pool, fn -> deferred_job() end)
Process.read_timer(timer_ref)
#=> 2836 # time remaining before job is sent to the pool
OddJob.cancel_timer(timer_ref)
#=> 1175 # job has been cancelled

timer_ref = OddJob.perform_after(5000, OddJob.Pool, fn -> deferred_job() end)
Process.sleep(6000)
OddJob.cancel_timer(timer_ref)
#=> false # too much time has passed to cancel the job

iex> OddJob.perform_after(5000, :does_not_exist, fn -> "never called" end)
{:error, :not_found}

Specs

perform_at(date_time(), pool(), function()) :: reference() | not_found()

Sends a job to the pool at the given time.

time must be a DateTime struct for a time in the future. The timer is executed under a separate supervised process, so if the caller crashes the job will still be performed. A timer reference is returned, which can be read with Process.read_timer/1 or canceled with OddJob.cancel_timer/1. Returns {:error, :not_found} if the pool server does not exist at the time this function is called.

Examples

time = DateTime.utc_now() |> DateTime.add(600, :second)
OddJob.perform_at(time, OddJob.Pool, fn -> scheduled_job() end)

iex> time = DateTime.utc_now() |> DateTime.add(600, :second)
iex> OddJob.perform_at(time, :does_not_exist, fn -> "never called" end)
{:error, :not_found}

Specs

perform_many(pool(), list() | map(), function()) :: :ok | not_found()

Sends a collection of jobs to the pool.

Enumerates over the collection, using each member as the argument to an anonymous function witn an arity of 1. The collection of jobs is sent in bulk to the pool where they will be immediately performed by any available workers. If there are no available workers the jobs are added to the FIFO queue. Jobs are guaranteed to always process in the same order of the original collection, and insertion into the queue is atomic. No jobs sent from other processes can be inserted between any member of the collection.

Returns :ok if the pool process exists at the time of calling, or {:error, :not_found} if it doesn't. See perform/2 for more.

Examples

iex> caller = self()
iex> OddJob.perform_many(OddJob.Pool, 1..5, fn x -> send(caller, x) end)
iex> for x <- 1..5 do
...>   receive do
...>     ^x -> x
...>   end
...> end
[1, 2, 3, 4, 5]

iex> OddJob.perform_many(:does_not_exist, ["never", "called"], fn x -> x end)
{:error, :not_found}

Specs

perform_many_after(timer(), pool(), list() | map(), function()) ::
  [job()] | not_found()

Sends a collection of jobs to the pool after the given timer has elapsed.

Enumerates over the collection, using each member as the argument to an anonymous function with an arity of 1. Returns a single timer reference. The timer is watched by a single process that will send the entire batch of jobs to the pool when the timer expires. See perform_after/3 for more information about arguments and timers.

Consider using this function instead of perform_after/3 when scheduling a large batch of jobs. perform_after/3 starts a separate scheduler process per job, whereas this function starts a single scheduler process for the whole batch.

Returns {:error, :not_found} if the pool does not exist at the time the function is called.

Examples

timer_ref = OddJob.perform_many_after(5000, OddJob.Pool, 1..5, fn x -> x ** 2 end)
#=> #Reference<0.1431903625.286261254.39156>
# fewer than 5 seconds pass
OddJob.cancel_timer(timer_ref)
#=> 2554 # returns the time remaining. All jobs in the collection have been canceled

timer_ref = OddJob.perform_many_after(5000, OddJob.Pool, 1..5, fn x -> x **2 end)
#=> #Reference<0.1431903625.286261254.39178>
# more than 5 seconds pass
OddJob.cancel_timer(timer_ref)
#=> false # too much time has passed, all jobs have been queued

iex> OddJob.perform_many_after(5000, :does_not_exist, ["never", "called"], fn x -> x end)
{:error, :not_found}

Specs

perform_many_at(date_time(), pool(), list() | map(), function()) ::
  reference() | not_found()

Sends a collection of jobs to the pool at the given time.

Enumerates over the collection, using each member as the argument to an anonymous function with an arity of 1. Returns a single timer reference. The timer is watched by a single process that will send the entire batch of jobs to the pool when the timer expires. See perform_at/3 for more information about arguments and timers.

Consider using this function instead of perform_at/3 when scheduling a large batch of jobs. perform_at/3 starts a separate scheduler process per job, whereas this function starts a single scheduler process for the whole batch.

Returns {:error, :not_found} if the pool does not exist at the time the function is called.

Examples

time = DateTime.utc_now() |> DateTime.add(5, :second)
timer_ref = OddJob.perform_many_at(time, OddJob.Pool, 1..5, fn x -> x ** 2 end)
#=> #Reference<0.1431903625.286261254.39156>
# fewer than 5 seconds pass
OddJob.cancel_timer(timer_ref)
#=> 2554 # returns the time remaining. All jobs in the collection have been canceled

time = DateTime.utc_now() |> DateTime.add(5, :second)
timer_ref = OddJob.perform_many_at(time, OddJob.Pool, 1..5, fn x -> x **2 end)
#=> #Reference<0.1431903625.286261254.39178>
# more than 5 seconds pass
OddJob.cancel_timer(timer_ref)
#=> false # too much time has passed, all jobs have been queued

iex> time = DateTime.utc_now() |> DateTime.add(5, :second)
iex> OddJob.perform_many_at(time, :does_not_exist, ["never", "called"], fn x -> x end)
{:error, :not_found}

A macro for creating jobs with an expressive DSL.

perform_this/2 works like perform/2 except it accepts a do block instead of an anonymous function.

Examples

You must import or require OddJob to use macros:

import OddJob
alias MyApp.Job

perform_this Job do
  some_work()
  some_other_work()
end

perform_this Job, do: something_hard()

A macro for creating jobs with an expressive DSL.

perform_this/3 accepts a single configuration option as the second argument that will control execution of the job. The available options provide the functionality of async_perform/2, perform_at/3, and perform_after/3.

Options

• :async - Passing the atom :async as the second argument before the do block creates an async job that can be awaited on. See async_perform/2.

• at: time - Use this option to schedule the job for a specific time in the future. time must be a valid Time or DateTime struct. See perform_at/3.

• after: timer - Use this option to schedule the job to perform after the given timer has elapsed. timer must be in milliseconds. See perform_after/3.

Examples

import OddJob
alias MyApp.Job

time = ~T[03:00:00.000000]
perform_this Job, at: time do
  scheduled_work()
end

perform_this Job, after: 5000, do: something_important()

perform_this Job, :async do
  get_data()
end
|> await()

iex> (perform_this OddJob.Pool, :async, do: 10 ** 2) |> await()
100

Specs

pool_supervisor(pool()) :: pid() | not_found()

Returns the pid of the OddJob.Pool.Supervisor that supervises the job pool's workers.

Returns {:error, :not_found} if the process does not exist at the time this function is called.

There is no guarantee that the process identified by the pid will still be alive after the results are returned, as it could exit or be killed or restarted at any time. Use pool_supervisor_name/1 to obtain the persistent registered name of the supervisor.

Examples

OddJob.pool_supervisor(OddJob.Pool)
#=> #PID<0.239.0>

iex> OddJob.pool_supervisor(:does_not_exist)
{:error, :not_found}

Specs

pool_supervisor_name(pool()) :: name() | not_found()

Returns the registered name in :via of the OddJob.Pool.Supervisor that supervises the pool's workers

Returns {:error, :not_found} if the process does not exist at the time this function is called.

Examples

iex> OddJob.pool_supervisor_name(OddJob.Pool)
{:via, Registry, {OddJob.Registry, {OddJob.Pool, :pool_sup}}}

iex> OddJob.pool_supervisor_name(:does_not_exist)
{:error, :not_found}

Specs

queue(pool()) :: {pid(), queue()} | not_found()

Returns a two element tuple {pid, state} with the pid of the job queue as the first element. The second element is the state of the queue represented by an OddJob.Queue struct.

The queue state has the following fields:

• :workers - A list of the pids of all workers in the pool

• :assigned - The pids of the workers currently assigned to jobs

• :jobs - A list of OddJob.Job structs in the queue awaiting execution

• :pool - The Elixir term naming the pool that the queue belongs to

Returns {:error, :not_found} if the queue server does not exist at the time this function is called.

Examples

iex> {pid, %OddJob.Queue{pool: pool}} = OddJob.queue(OddJob.Pool)
iex> is_pid(pid)
true
iex> pool
OddJob.Pool

iex> OddJob.queue(:does_not_exist)
{:error, :not_found}

Specs

queue_name(pool()) :: name() | not_found()

Returns the registered name in :via for the queue process. See the Registry module for more information on :via process name registration.

Returns {:error, :not_found} if the queue server does not exist at the time this function is called.

Examples

iex> OddJob.queue_name(OddJob.Pool)
{:via, Registry, {OddJob.Registry, {OddJob.Pool, :queue}}}

iex> OddJob.queue_name(:does_not_exist)
{:error, :not_found}

Specs

start_link(options()) :: Supervisor.on_start()

Starts an OddJob.Pool supervision tree linked to the current process.

You can start an OddJob pool dynamically with start_link/1 to start processing concurrent jobs:

iex> {:ok, _pid} = OddJob.start_link(name: MyApp.Event, pool_size: 10)
iex> OddJob.async_perform(MyApp.Event, fn -> :do_something end) |> OddJob.await()
:do_something

In most cases you would instead start your pools inside a supervision tree:

children = [{OddJob, name: MyApp.Event, pool_size: 10}]
Supervisor.start_link(children, strategy: :one_for_one)

See OddJob.child_spec/1 for a list of available options.

See Supervisor for more on child specifications and supervision trees.

Specs

whereis(pool()) :: pid() | nil

Returns the pid of the top level supervisor in the pool supervision tree, nil if a process can't be found.

Example

iex> {:ok, pid} = OddJob.start_link(name: :whereis)
iex> OddJob.whereis(:whereis) == pid
true

iex> OddJob.whereis(:where_it_is_not) == nil
true

Specs

workers(pool()) :: [pid()] | not_found()

Returns a list of pids for the specified worker pool.

There is no guarantee that the processes will still be alive after the results are returned, as they could exit or be killed at any time.

Returns {:error, :not_found} if the pool does not exist at the time this function is called.

Examples

OddJob.workers(OddJob.Pool)
#=> [#PID<0.105.0>, #PID<0.106.0>, #PID<0.107.0>, #PID<0.108.0>, #PID<0.109.0>]

iex> OddJob.workers(:does_not_exist)
{:error, :not_found}