Roger.Partition.Global (roger v4.0.0)

Copy Markdown View Source

Coordinates the global Roger partition state

Each Roger partition has a single place where global state is kept. Global state (and global coordination) is needed for the following things:

  • Job cancellation; when cancelling a job, we store the job ID globally; when the cancelled job is started, we check the job id against this list of cancelled ids.

  • Queue keys; some jobs dictate that they cannot be queued when there is already a job queued with an identical queue key; if so, the job fails to enqueue.

  • Execution keys; jobs which have the same execution key cannot be executed concurrently and need to wait on one another.

  • Pause states; it is globally stored which queues are currently paused.

The per-partition Global process stores all this information. It provides hooks to persist the information between partition / node restarts. By default, the global state is loaded from and written to the filesystem, but it is possible to override the persister, like this:

config :roger,
  persister: Your.PersisterModule

The persister module must implement the Roger.Partition.Global.StatePersister behaviour, which provides simple load and save functions.

Summary

Functions

Mark a job id as cancelled.

Check whether a given job id has been marked cancelled

Returns a specification to start this module under a supervisor.

Check whether a given execution key has been set

Callback implementation for GenServer.init/1.

Get the set of paused queues for the given partition_id.

Cluster-wide pausing of the given queue in the given partition_id.

Cluster-wide pausing of the given queue in the given partition_id.

Check whether a given queue key has been marked enqueued

Remove the given execution key

Remove a given queue key

Functions

cancel_job(partition_id, job_id)

@spec cancel_job(partition_id :: String.t(), job_id :: String.t()) :: :ok

Mark a job id as cancelled.

This does not check for the validity of the job id. The job will not be removed from the queue, but instead will be removed as soon as it's dequeued.

When a job is currently executing, the process of a running job will be killed.

cancelled?(partition_id, job_id, remove \\ nil)

@spec cancelled?(
  partition_id :: String.t(),
  job_id :: String.t(),
  remove :: :remove | nil
) :: boolean()

Check whether a given job id has been marked cancelled

child_spec(init_arg)

Returns a specification to start this module under a supervisor.

See Supervisor.

executing?(partition_id, execution_key, add \\ nil)

@spec executing?(
  partition_id :: String.t(),
  execution_key :: String.t(),
  add :: :add | nil
) ::
  boolean()

Check whether a given execution key has been set

init(list)

Callback implementation for GenServer.init/1.

queue_get_paused(partition_id)

@spec queue_get_paused(partition_id :: String.t()) :: {:ok, MapSet.t()}

Get the set of paused queues for the given partition_id.

queue_pause(partition_id, queue)

@spec queue_pause(partition_id :: String.t(), queue :: any()) :: :ok

Cluster-wide pausing of the given queue in the given partition_id.

queue_resume(partition_id, queue)

@spec queue_resume(partition_id :: String.t(), queue :: any()) :: :ok

Cluster-wide pausing of the given queue in the given partition_id.

queued?(partition_id, queue_key, add \\ nil)

@spec queued?(partition_id :: String.t(), queue_key :: String.t(), add :: :add | nil) ::
  boolean()

Check whether a given queue key has been marked enqueued

remove_executed(partition_id, execution_key)

@spec remove_executed(partition_id :: String.t(), execution_key :: String.t()) :: :ok

Remove the given execution key

remove_queued(partition_id, queue_key)

@spec remove_queued(partition_id :: String.t(), queue_key :: String.t()) :: :ok

Remove a given queue key