name: Formal RFC about: Submit a formal Request For Comments for consideration by the team. title: ‘Background jobs with FoundationDB’ labels: rfc, discussion assignees: ''
This document describes a data model, implementation, and an API for running CouchDB background jobs with FoundationDB.
CouchDB background jobs are used for things like index building, replication and couch-peruser processing. We present a generalized model which allows creation, running, and monitoring of these jobs.
The key words “MUST”, “MUST NOT”, “REQUIRED”, “SHALL”, “SHALL NOT”, “SHOULD”, “SHOULD NOT”, “RECOMMENDED”, “MAY”, and “OPTIONAL” in this document are to be interpreted as described in RFC 2119.
In the discussion below a job is considered to be an abstract unit of work. It is identified by a JobId
and has a JobType
. Client code creates a job which is then is executed by a job processor. A job processor is language-specific execution unit that runs the job. It could be an Erlang process, a thread, or just a function.
The API used to create jobs is called the Job Creation API
and the API used by the job processors to run jobs is called the Job Processing API
.
Jobs in the system can be in 3 states. After a job is added and is waiting to run, the job is considered to be pending
. A job executed by a job processor is considered to be running
. When a job is neither running
, nor pending
, it is considered to be finished
. This is the state transition diagram:
+------------>+ | | | v -->[PENDING] [RUNNING]--->[FINISHED] ^ | | | v | +-------------+<----------+
The general pattern of using this API might look like:
Job creators:
add/4,5
to add a jobremove/3
to remove itJob processors:
accept/1,2
and wait until it gets a job to process.update/2,3
to prevent the job from being re-enqueued due to idleness.finish/2,3
add(Tx, Type, JobId, JobData[, ScheduledTime]) -> ok | {error, Error}
JobData
is map with a job type-specific data in it. It MAY contain any data as long as it can be properly encoded as JSON.ScheduledTime
is an optional parameter to schedule the job to be executed at a later time. The format is an integer seconds since UNIX epoch.JobId
exists:pending
, then the ScheduledTime
is updated.running
then the job is flagged to be resubmitted when it finishes running.finished
then it will be re-enqueued as pending
remove(Tx, Type, JobId) -> ok | {error, Error}
get_job_data(Job) -> {ok, JobData} | {error, Error}
JobData
associated with the job.get_job_state(Job) -> {ok, pending | running | finished} | {error, Error}
set_type_timeout(Type, TimeoutSec) -> ok
get_type_timeout(Type) -> {ok, TimeoutSec} | {error, Error}
subscribe(Type, JobId) -> {ok, SubscriptionId, JobState}
wait/2,3
calls.unsubscribe(SubscriptionId) -> ok
wait(SubscriptionId, Timeout) -> {Type, JobId, JobState} | timeout wait([SubscriptionId], Timeout) -> {Type, JobId, JobState} | timeout
wait(SubscriptionId, Type, Timeout) -> {Type, JobId, JobState} | timeout wait([SubscriptionId], Type, Timeout) -> {Type, JobId, JobState} | timeout
accept(Type[, OptionsMap]) -> {ok, Job} | {error, Error}
pending
job and start running it. OptionsMap
is a map that MAY have these parameters:no_schedule
= true
| false
Use a more optimized dequeueing strategy if time-based scheduling is not used and job IDs are known to start with a random looking (UUID-like) prefix.max_sched_time
= SecondsSinceEpoch
: Only accept jobs which have been scheduled before or at SecondsSinceEpoch
UNIX time.timeout
= TimeoutMSec
: Maximum timeout to wait when there are no pending jobs available. 0
means don't wait at all and return {error, not_found}
immediately, effectively making accept/1,2
non-blocking.update(Tx, Job[, JobData]) -> {ok, Job} | {error, halt | Error}
JobData
. It MUST be called at least as often as the configured timeout value for the job’s type. Not doing this will result in the job being re-enqueued. If halt
is returned, the job processor MUST stop running the job. Job processors MUST call update/2,3
in any write transactions it performs in order to guarantee mutual exclusion that at most one job processor is executing a particular job at a time.finish(Tx, Job[, JobData]) -> ok | {error, halt | Error}
JobData
parameter MAY contain a final result. If halt
is returned, it means that the JobData
value wasn't updated. Job processors MUST call update/2,3
or finish/2,3
in any write transactions it performs in order to guarantee mutual exclusion that at most one job processor is executing a particular job at a time.resubmit(Tx, Job[, ScheduledTime]) -> {ok, Job} | {error, Error}
finish/2,3
is called.is_resubmitted(Job) -> true | false
Job
object that gets returned from the update/2,3
function to determine if job creator had requested the job to be resubmitted. The job won't actually be re-enqueued until finish/2,3
function is called.This section discusses how some of the framework functionality is implemented.
All the coordination between job creation and job processing is done via FoundationDB. There is a top level "couch_jobs"
subspace. All the subspaces mentioned below will be under this subspace.
Each job managed by the framework will have an entry in the main jobs table
. Pending jobs are added to a pending queue
subspace. When they are accepted by a jobs processor, the jobs are removed from the pending queue and added to the active jobs
subspace.
Job states referenced in the API section are essentially defined based on the presence in any of these subspaces:
pending queue
it is considered pending
active jobs
subspace, then it is running
pending
or running
then it is considered finished
Job processors may suddenly crash and stop running their jobs. In that case the framework will automatically make those jobs pending
after a timeout. That ensures the jobs continue to make progress. To avoid getting re-enqueued as pending
due the timeout, each job processor must periodically call the update/2,3
function. That functionality is implemented by the activity monitor
. It periodically watches a per-type versionstamp-ed key, then scans active jobs
subspace for any running
jobs which haven't updated their entries during the timeout period.
Subscription notifications are managed separately for each job type. They use a per-type versionstamp-ed watch to monitor which jobs have updated since the last time it delivered notifications to the subscribers.
("couch_jobs", "data", Type, JobId) = (Sequence, JobLock, ScheduledTime, Resubmit, JobData)
("couch_jobs", "pending", Type, ScheduledTime, JobId) = ""
("couch_jobs", "watches_pending", Type) = Sequence
("couch_jobs", "watches_activity", Type) = Sequence
("couch_jobs", "activity_timeout", Type) = ActivityTimeout
("couch_jobs", "activity", Type, Sequence) = JobId
This section describes how the framework implements some of the API functions.
add/4,5
:
JobId
exists, resubmit the job."pending"
watch for the type with a new versionstamp and bump its counter.JobLock
is set to null
.remove/3
:
update/2,3
call will get a halt
error and know to stop running the job.accept/1,2
:
JobLock
UUID.JobLock
in the jobs table."activity"
subspace."pending"
queue and wait until it fires, then try again.update/2,3
:
halt
.JobLock
matches, otherwise return halt
."activity"
sequence entry.JobData
."activity"
sequence entry and in main job table."watches"
sequence for that job type.finish/2,3
:
halt
.JobLock
matches, otherwise returns halt
."activity"
sequence entry.Resubmit
field is true
, re-enqueue the job, and set Resubmit
to false
.JobLock
to null
resubmit/2,3
:
Resubmit
field to true
.finish/2,3
is called.The main advantage is having a central way to coordinate batch processing across a cluster, with a single, unified API.
Since all job keys and values are just FDB tuples and JSON encoded objects, in the future it might be possible to accept external jobs, not just jobs defined by the CouchDB internals. Also, since workers could be written in any language as long as they can talk to the FDB cluster, and follow the behavior describes in the design, it opens the possibility to have custom (user defined) workers of different types. But that is out of scope in the current RFC discussion.
Replication, indexing, couch-peruser
None. However, in the future, it might be useful to have an API to query and monitor the state of all the queues and workers.
None have been identified.
None have been identified.
Original mailing list discussion