5 * This program is free software; you can redistribute it and/or modify
6 * it under the terms of the GNU General Public License as published by
7 * the Free Software Foundation; either version 2 of the License, or
8 * (at your option) any later version.
10 * This program is distributed in the hope that it will be useful,
11 * but WITHOUT ANY WARRANTY; without even the implied warranty of
12 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13 * GNU General Public License for more details.
15 * You should have received a copy of the GNU General Public License along
16 * with this program; if not, write to the Free Software Foundation, Inc.,
17 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
18 * http://www.gnu.org/copyleft/gpl.html
21 * @defgroup JobQueue JobQueue
23 use Liuggio\StatsdClient\Factory\StatsdDataFactoryInterface
;
26 * Class to handle enqueueing and running of background jobs
31 abstract class JobQueue
{
32 /** @var string DB domain ID */
34 /** @var string Job type */
36 /** @var string Job priority for pop() */
38 /** @var int Time to live in seconds */
40 /** @var int Maximum number of times to try a job */
42 /** @var string|bool Read only rationale (or false if r/w) */
43 protected $readOnlyReason;
44 /** @var StatsdDataFactoryInterface */
47 /** @var WANObjectCache */
50 const QOS_ATOMIC
= 1; // integer; "all-or-nothing" job insertions
52 const ROOTJOB_TTL
= 2419200; // integer; seconds to remember root jobs (28 days)
55 * @param array $params
57 * - domain : A DB domain ID
58 * - wanCache : An instance of WANObjectCache to use for caching [default: none]
59 * - stats : An instance of StatsdDataFactoryInterface [default: none]
60 * - claimTTL : Seconds a job can be claimed for exclusive execution [default: forever]
61 * - maxTries : Total times a job can be tried, assuming claims expire [default: 3]
62 * - order : Queue order, one of ("fifo", "timestamp", "random") [default: variable]
63 * - readOnlyReason : Mark the queue as read-only with this reason [default: false]
64 * @throws JobQueueError
66 protected function __construct( array $params ) {
67 $this->domain
= $params['domain'] ??
$params['wiki']; // b/c
68 $this->type
= $params['type'];
69 $this->claimTTL
= $params['claimTTL'] ??
0;
70 $this->maxTries
= $params['maxTries'] ??
3;
71 if ( isset( $params['order'] ) && $params['order'] !== 'any' ) {
72 $this->order
= $params['order'];
74 $this->order
= $this->optimalOrder();
76 if ( !in_array( $this->order
, $this->supportedOrders() ) ) {
77 throw new JobQueueError( __CLASS__
. " does not support '{$this->order}' order." );
79 $this->readOnlyReason
= $params['readOnlyReason'] ??
false;
80 $this->stats
= $params['stats'] ??
new NullStatsdDataFactory();
81 $this->wanCache
= $params['wanCache'] ?? WANObjectCache
::newEmpty();
85 * Get a job queue object of the specified type.
87 * - class : What job class to use (determines job type)
88 * - domain : Database domain ID of the wiki the jobs are for (defaults to current wiki)
89 * - type : The name of the job types this queue handles
90 * - order : Order that pop() selects jobs, one of "fifo", "timestamp" or "random".
91 * If "fifo" is used, the queue will effectively be FIFO. Note that job
92 * completion will not appear to be exactly FIFO if there are multiple
93 * job runners since jobs can take different times to finish once popped.
94 * If "timestamp" is used, the queue will at least be loosely ordered
95 * by timestamp, allowing for some jobs to be popped off out of order.
96 * If "random" is used, pop() will pick jobs in random order.
97 * Note that it may only be weakly random (e.g. a lottery of the oldest X).
98 * If "any" is choosen, the queue will use whatever order is the fastest.
99 * This might be useful for improving concurrency for job acquisition.
100 * - claimTTL : If supported, the queue will recycle jobs that have been popped
101 * but not acknowledged as completed after this many seconds. Recycling
102 * of jobs simply means re-inserting them into the queue. Jobs can be
103 * attempted up to three times before being discarded.
104 * - readOnlyReason : Set this to a string to make the queue read-only.
105 * - stash : A BagOStuff instance that can be used for root job deduplication
106 * - stats : A StatsdDataFactoryInterface [optional]
108 * Queue classes should throw an exception if they do not support the options given.
110 * @param array $params
112 * @throws JobQueueError
114 final public static function factory( array $params ) {
115 $class = $params['class'];
116 if ( !class_exists( $class ) ) {
117 throw new JobQueueError( "Invalid job queue class '$class'." );
119 $obj = new $class( $params );
120 if ( !( $obj instanceof self
) ) {
121 throw new JobQueueError( "Class '$class' is not a " . __CLASS__
. " class." );
128 * @return string Database domain ID
130 final public function getDomain() {
131 return $this->domain
;
135 * @return string Wiki ID
138 final public function getWiki() {
139 return WikiMap
::getWikiIdFromDbDomain( $this->domain
);
143 * @return string Job type that this queue handles
145 final public function getType() {
150 * @return string One of (random, timestamp, fifo, undefined)
152 final public function getOrder() {
157 * Get the allowed queue orders for configuration validation
159 * @return array Subset of (random, timestamp, fifo, undefined)
161 abstract protected function supportedOrders();
164 * Get the default queue order to use if configuration does not specify one
166 * @return string One of (random, timestamp, fifo, undefined)
168 abstract protected function optimalOrder();
171 * Find out if delayed jobs are supported for configuration validation
173 * @return bool Whether delayed jobs are supported
175 protected function supportsDelayedJobs() {
176 return false; // not implemented
180 * @return bool Whether delayed jobs are enabled
183 final public function delayedJobsEnabled() {
184 return $this->supportsDelayedJobs();
188 * @return string|bool Read-only rational or false if r/w
191 public function getReadOnlyReason() {
192 return $this->readOnlyReason
;
196 * Quickly check if the queue has no available (unacquired, non-delayed) jobs.
197 * Queue classes should use caching if they are any slower without memcached.
199 * If caching is used, this might return false when there are actually no jobs.
200 * If pop() is called and returns false then it should correct the cache. Also,
201 * calling flushCaches() first prevents this. However, this affect is typically
202 * not distinguishable from the race condition between isEmpty() and pop().
205 * @throws JobQueueError
207 final public function isEmpty() {
208 $res = $this->doIsEmpty();
214 * @see JobQueue::isEmpty()
217 abstract protected function doIsEmpty();
220 * Get the number of available (unacquired, non-delayed) jobs in the queue.
221 * Queue classes should use caching if they are any slower without memcached.
223 * If caching is used, this number might be out of date for a minute.
226 * @throws JobQueueError
228 final public function getSize() {
229 $res = $this->doGetSize();
235 * @see JobQueue::getSize()
238 abstract protected function doGetSize();
241 * Get the number of acquired jobs (these are temporarily out of the queue).
242 * Queue classes should use caching if they are any slower without memcached.
244 * If caching is used, this number might be out of date for a minute.
247 * @throws JobQueueError
249 final public function getAcquiredCount() {
250 $res = $this->doGetAcquiredCount();
256 * @see JobQueue::getAcquiredCount()
259 abstract protected function doGetAcquiredCount();
262 * Get the number of delayed jobs (these are temporarily out of the queue).
263 * Queue classes should use caching if they are any slower without memcached.
265 * If caching is used, this number might be out of date for a minute.
268 * @throws JobQueueError
271 final public function getDelayedCount() {
272 $res = $this->doGetDelayedCount();
278 * @see JobQueue::getDelayedCount()
281 protected function doGetDelayedCount() {
282 return 0; // not implemented
286 * Get the number of acquired jobs that can no longer be attempted.
287 * Queue classes should use caching if they are any slower without memcached.
289 * If caching is used, this number might be out of date for a minute.
292 * @throws JobQueueError
294 final public function getAbandonedCount() {
295 $res = $this->doGetAbandonedCount();
301 * @see JobQueue::getAbandonedCount()
304 protected function doGetAbandonedCount() {
305 return 0; // not implemented
309 * Push one or more jobs into the queue.
310 * This does not require $wgJobClasses to be set for the given job type.
311 * Outside callers should use JobQueueGroup::push() instead of this function.
313 * @param IJobSpecification|IJobSpecification[] $jobs
314 * @param int $flags Bitfield (supports JobQueue::QOS_ATOMIC)
316 * @throws JobQueueError
318 final public function push( $jobs, $flags = 0 ) {
319 $jobs = is_array( $jobs ) ?
$jobs : [ $jobs ];
320 $this->batchPush( $jobs, $flags );
324 * Push a batch of jobs into the queue.
325 * This does not require $wgJobClasses to be set for the given job type.
326 * Outside callers should use JobQueueGroup::push() instead of this function.
328 * @param IJobSpecification[] $jobs
329 * @param int $flags Bitfield (supports JobQueue::QOS_ATOMIC)
331 * @throws JobQueueError
333 final public function batchPush( array $jobs, $flags = 0 ) {
334 $this->assertNotReadOnly();
336 if ( $jobs === [] ) {
337 return; // nothing to do
340 foreach ( $jobs as $job ) {
341 if ( $job->getType() !== $this->type
) {
342 throw new JobQueueError(
343 "Got '{$job->getType()}' job; expected a '{$this->type}' job." );
344 } elseif ( $job->getReleaseTimestamp() && !$this->supportsDelayedJobs() ) {
345 throw new JobQueueError(
346 "Got delayed '{$job->getType()}' job; delays are not supported." );
350 $this->doBatchPush( $jobs, $flags );
352 foreach ( $jobs as $job ) {
353 if ( $job->isRootJob() ) {
354 $this->deduplicateRootJob( $job );
360 * @see JobQueue::batchPush()
361 * @param IJobSpecification[] $jobs
364 abstract protected function doBatchPush( array $jobs, $flags );
367 * Pop a job off of the queue.
368 * This requires $wgJobClasses to be set for the given job type.
369 * Outside callers should use JobQueueGroup::pop() instead of this function.
371 * @throws JobQueueError
372 * @return RunnableJob|bool Returns false if there are no jobs
374 final public function pop() {
375 $this->assertNotReadOnly();
377 $job = $this->doPop();
379 // Flag this job as an old duplicate based on its "root" job...
381 if ( $job && $this->isRootJobOldDuplicate( $job ) ) {
382 $this->incrStats( 'dupe_pops', $this->type
);
383 $job = DuplicateJob
::newFromJob( $job ); // convert to a no-op
385 } catch ( Exception
$e ) {
386 // don't lose jobs over this
393 * @see JobQueue::pop()
394 * @return RunnableJob|bool
396 abstract protected function doPop();
399 * Acknowledge that a job was completed.
401 * This does nothing for certain queue classes or if "claimTTL" is not set.
402 * Outside callers should use JobQueueGroup::ack() instead of this function.
404 * @param RunnableJob $job
406 * @throws JobQueueError
408 final public function ack( RunnableJob
$job ) {
409 $this->assertNotReadOnly();
410 if ( $job->getType() !== $this->type
) {
411 throw new JobQueueError( "Got '{$job->getType()}' job; expected '{$this->type}'." );
414 $this->doAck( $job );
418 * @see JobQueue::ack()
419 * @param RunnableJob $job
421 abstract protected function doAck( RunnableJob
$job );
424 * Register the "root job" of a given job into the queue for de-duplication.
425 * This should only be called right *after* all the new jobs have been inserted.
426 * This is used to turn older, duplicate, job entries into no-ops. The root job
427 * information will remain in the registry until it simply falls out of cache.
429 * This requires that $job has two special fields in the "params" array:
430 * - rootJobSignature : hash (e.g. SHA1) that identifies the task
431 * - rootJobTimestamp : TS_MW timestamp of this instance of the task
433 * A "root job" is a conceptual job that consist of potentially many smaller jobs
434 * that are actually inserted into the queue. For example, "refreshLinks" jobs are
435 * spawned when a template is edited. One can think of the task as "update links
436 * of pages that use template X" and an instance of that task as a "root job".
437 * However, what actually goes into the queue are range and leaf job subtypes.
438 * Since these jobs include things like page ID ranges and DB master positions,
439 * and can morph into smaller jobs recursively, simple duplicate detection
440 * for individual jobs being identical (like that of job_sha1) is not useful.
442 * In the case of "refreshLinks", if these jobs are still in the queue when the template
443 * is edited again, we want all of these old refreshLinks jobs for that template to become
444 * no-ops. This can greatly reduce server load, since refreshLinks jobs involves parsing.
445 * Essentially, the new batch of jobs belong to a new "root job" and the older ones to a
446 * previous "root job" for the same task of "update links of pages that use template X".
448 * This does nothing for certain queue classes.
450 * @param IJobSpecification $job
451 * @throws JobQueueError
454 final public function deduplicateRootJob( IJobSpecification
$job ) {
455 $this->assertNotReadOnly();
456 if ( $job->getType() !== $this->type
) {
457 throw new JobQueueError( "Got '{$job->getType()}' job; expected '{$this->type}'." );
460 return $this->doDeduplicateRootJob( $job );
464 * @see JobQueue::deduplicateRootJob()
465 * @param IJobSpecification $job
466 * @throws JobQueueError
469 protected function doDeduplicateRootJob( IJobSpecification
$job ) {
470 $params = $job->hasRootJobParams() ?
$job->getRootJobParams() : null;
472 throw new JobQueueError( "Cannot register root job; missing parameters." );
475 $key = $this->getRootJobCacheKey( $params['rootJobSignature'] );
476 // Callers should call JobQueueGroup::push() before this method so that if the
477 // insert fails, the de-duplication registration will be aborted. Having only the
478 // de-duplication registration succeed would cause jobs to become no-ops without
479 // any actual jobs that made them redundant.
480 $timestamp = $this->wanCache
->get( $key ); // last known timestamp of such a root job
481 if ( $timestamp !== false && $timestamp >= $params['rootJobTimestamp'] ) {
482 return true; // a newer version of this root job was enqueued
485 // Update the timestamp of the last root job started at the location...
486 return $this->wanCache
->set( $key, $params['rootJobTimestamp'], self
::ROOTJOB_TTL
);
490 * Check if the "root" job of a given job has been superseded by a newer one
492 * @param IJobSpecification $job
493 * @throws JobQueueError
496 final protected function isRootJobOldDuplicate( IJobSpecification
$job ) {
497 if ( $job->getType() !== $this->type
) {
498 throw new JobQueueError( "Got '{$job->getType()}' job; expected '{$this->type}'." );
501 return $this->doIsRootJobOldDuplicate( $job );
505 * @see JobQueue::isRootJobOldDuplicate()
506 * @param IJobSpecification $job
509 protected function doIsRootJobOldDuplicate( IJobSpecification
$job ) {
510 $params = $job->hasRootJobParams() ?
$job->getRootJobParams() : null;
512 return false; // job has no de-deplication info
515 $key = $this->getRootJobCacheKey( $params['rootJobSignature'] );
516 // Get the last time this root job was enqueued
517 $timestamp = $this->wanCache
->get( $key );
518 if ( $timestamp === false ||
$params['rootJobTimestamp'] > $timestamp ) {
519 // Update the timestamp of the last known root job started at the location...
520 $this->wanCache
->set( $key, $params['rootJobTimestamp'], self
::ROOTJOB_TTL
);
523 // Check if a new root job was started at the location after this one's...
524 return ( $timestamp && $timestamp > $params['rootJobTimestamp'] );
528 * @param string $signature Hash identifier of the root job
531 protected function getRootJobCacheKey( $signature ) {
532 return $this->wanCache
->makeGlobalKey(
542 * Deleted all unclaimed and delayed jobs from the queue
544 * @throws JobQueueError
548 final public function delete() {
549 $this->assertNotReadOnly();
555 * @see JobQueue::delete()
556 * @throws JobQueueError
558 protected function doDelete() {
559 throw new JobQueueError( "This method is not implemented." );
563 * Wait for any replica DBs or backup servers to catch up.
565 * This does nothing for certain queue classes.
568 * @throws JobQueueError
570 final public function waitForBackups() {
571 $this->doWaitForBackups();
575 * @see JobQueue::waitForBackups()
578 protected function doWaitForBackups() {
582 * Clear any process and persistent caches
586 final public function flushCaches() {
587 $this->doFlushCaches();
591 * @see JobQueue::flushCaches()
594 protected function doFlushCaches() {
598 * Get an iterator to traverse over all available jobs in this queue.
599 * This does not include jobs that are currently acquired or delayed.
600 * Note: results may be stale if the queue is concurrently modified.
603 * @throws JobQueueError
605 abstract public function getAllQueuedJobs();
608 * Get an iterator to traverse over all delayed jobs in this queue.
609 * Note: results may be stale if the queue is concurrently modified.
612 * @throws JobQueueError
615 public function getAllDelayedJobs() {
616 return new ArrayIterator( [] ); // not implemented
620 * Get an iterator to traverse over all claimed jobs in this queue
622 * Callers should be quick to iterator over it or few results
623 * will be returned due to jobs being acknowledged and deleted
626 * @throws JobQueueError
629 public function getAllAcquiredJobs() {
630 return new ArrayIterator( [] ); // not implemented
634 * Get an iterator to traverse over all abandoned jobs in this queue
637 * @throws JobQueueError
640 public function getAllAbandonedJobs() {
641 return new ArrayIterator( [] ); // not implemented
645 * Do not use this function outside of JobQueue/JobQueueGroup
650 public function getCoalesceLocationInternal() {
655 * Check whether each of the given queues are empty.
656 * This is used for batching checks for queues stored at the same place.
658 * @param array $types List of queues types
659 * @return array|null (list of non-empty queue types) or null if unsupported
660 * @throws JobQueueError
663 final public function getSiblingQueuesWithJobs( array $types ) {
664 return $this->doGetSiblingQueuesWithJobs( $types );
668 * @see JobQueue::getSiblingQueuesWithJobs()
669 * @param array $types List of queues types
670 * @return array|null (list of queue types) or null if unsupported
672 protected function doGetSiblingQueuesWithJobs( array $types ) {
673 return null; // not supported
677 * Check the size of each of the given queues.
678 * For queues not served by the same store as this one, 0 is returned.
679 * This is used for batching checks for queues stored at the same place.
681 * @param array $types List of queues types
682 * @return array|null (job type => whether queue is empty) or null if unsupported
683 * @throws JobQueueError
686 final public function getSiblingQueueSizes( array $types ) {
687 return $this->doGetSiblingQueueSizes( $types );
691 * @see JobQueue::getSiblingQueuesSize()
692 * @param array $types List of queues types
693 * @return array|null (list of queue types) or null if unsupported
695 protected function doGetSiblingQueueSizes( array $types ) {
696 return null; // not supported
700 * @param string $command
701 * @param array $params
704 protected function factoryJob( $command, $params ) {
705 // @TODO: dependency inject this as a callback
706 return Job
::factory( $command, $params );
710 * @throws JobQueueReadOnlyError
712 protected function assertNotReadOnly() {
713 if ( $this->readOnlyReason
!== false ) {
714 throw new JobQueueReadOnlyError( "Job queue is read-only: {$this->readOnlyReason}" );
719 * Call wfIncrStats() for the queue overall and for the queue type
721 * @param string $key Event type
722 * @param string $type Job type
726 protected function incrStats( $key, $type, $delta = 1 ) {
727 $this->stats
->updateCount( "jobqueue.{$key}.all", $delta );
728 $this->stats
->updateCount( "jobqueue.{$key}.{$type}", $delta );