includes/jobqueue/JobQueue.php

   1 <?php
   2 /**
   3  * Job queue base code.
   4  *
   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.
   9  *
  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.
  14  *
  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
  19  *
  20  * @file
  21  * @defgroup JobQueue JobQueue
  22  * @author Aaron Schulz
  23  */
  24
  25 /**
  26  * Class to handle enqueueing and running of background jobs
  27  *
  28  * @ingroup JobQueue
  29  * @since 1.21
  30  */
  31 abstract class JobQueue {
  32         /** @var string Wiki ID */
  33         protected $wiki;
  34
  35         /** @var string Job type */
  36         protected $type;
  37
  38         /** @var string Job priority for pop() */
  39         protected $order;
  40
  41         /** @var int Time to live in seconds */
  42         protected $claimTTL;
  43
  44         /** @var int Maximum number of times to try a job */
  45         protected $maxTries;
  46
  47         /** @var BagOStuff */
  48         protected $dupCache;
  49         /** @var JobQueueAggregator */
  50         protected $aggr;
  51
  52         const QOS_ATOMIC = 1; // integer; "all-or-nothing" job insertions
  53
  54         const ROOTJOB_TTL = 2419200; // integer; seconds to remember root jobs (28 days)
  55
  56         /**
  57          * @param array $params
  58          * @throws MWException
  59          */
  60         protected function __construct( array $params ) {
  61                 $this->wiki = $params['wiki'];
  62                 $this->type = $params['type'];
  63                 $this->claimTTL = isset( $params['claimTTL'] ) ? $params['claimTTL'] : 0;
  64                 $this->maxTries = isset( $params['maxTries'] ) ? $params['maxTries'] : 3;
  65                 if ( isset( $params['order'] ) && $params['order'] !== 'any' ) {
  66                         $this->order = $params['order'];
  67                 } else {
  68                         $this->order = $this->optimalOrder();
  69                 }
  70                 if ( !in_array( $this->order, $this->supportedOrders() ) ) {
  71                         throw new MWException( __CLASS__ . " does not support '{$this->order}' order." );
  72                 }
  73                 $this->dupCache = wfGetCache( CACHE_ANYTHING );
  74                 $this->aggr = isset( $params['aggregator'] )
  75                         ? $params['aggregator']
  76                         : new JobQueueAggregatorNull( array() );
  77         }
  78
  79         /**
  80          * Get a job queue object of the specified type.
  81          * $params includes:
  82          *   - class      : What job class to use (determines job type)
  83          *   - wiki       : wiki ID of the wiki the jobs are for (defaults to current wiki)
  84          *   - type       : The name of the job types this queue handles
  85          *   - order      : Order that pop() selects jobs, one of "fifo", "timestamp" or "random".
  86          *                  If "fifo" is used, the queue will effectively be FIFO. Note that job
  87          *                  completion will not appear to be exactly FIFO if there are multiple
  88          *                  job runners since jobs can take different times to finish once popped.
  89          *                  If "timestamp" is used, the queue will at least be loosely ordered
  90          *                  by timestamp, allowing for some jobs to be popped off out of order.
  91          *                  If "random" is used, pop() will pick jobs in random order.
  92          *                  Note that it may only be weakly random (e.g. a lottery of the oldest X).
  93          *                  If "any" is choosen, the queue will use whatever order is the fastest.
  94          *                  This might be useful for improving concurrency for job acquisition.
  95          *   - claimTTL   : If supported, the queue will recycle jobs that have been popped
  96          *                  but not acknowledged as completed after this many seconds. Recycling
  97          *                  of jobs simply means re-inserting them into the queue. Jobs can be
  98          *                  attempted up to three times before being discarded.
  99          *
 100          * Queue classes should throw an exception if they do not support the options given.
 101          *
 102          * @param array $params
 103          * @return JobQueue
 104          * @throws MWException
 105          */
 106         final public static function factory( array $params ) {
 107                 $class = $params['class'];
 108                 if ( !class_exists( $class ) ) {
 109                         throw new MWException( "Invalid job queue class '$class'." );
 110                 }
 111                 $obj = new $class( $params );
 112                 if ( !( $obj instanceof self ) ) {
 113                         throw new MWException( "Class '$class' is not a " . __CLASS__ . " class." );
 114                 }
 115
 116                 return $obj;
 117         }
 118
 119         /**
 120          * @return string Wiki ID
 121          */
 122         final public function getWiki() {
 123                 return $this->wiki;
 124         }
 125
 126         /**
 127          * @return string Job type that this queue handles
 128          */
 129         final public function getType() {
 130                 return $this->type;
 131         }
 132
 133         /**
 134          * @return string One of (random, timestamp, fifo, undefined)
 135          */
 136         final public function getOrder() {
 137                 return $this->order;
 138         }
 139
 140         /**
 141          * Get the allowed queue orders for configuration validation
 142          *
 143          * @return array Subset of (random, timestamp, fifo, undefined)
 144          */
 145         abstract protected function supportedOrders();
 146
 147         /**
 148          * Get the default queue order to use if configuration does not specify one
 149          *
 150          * @return string One of (random, timestamp, fifo, undefined)
 151          */
 152         abstract protected function optimalOrder();
 153
 154         /**
 155          * Find out if delayed jobs are supported for configuration validation
 156          *
 157          * @return bool Whether delayed jobs are supported
 158          */
 159         protected function supportsDelayedJobs() {
 160                 return false; // not implemented
 161         }
 162
 163         /**
 164          * @return bool Whether delayed jobs are enabled
 165          * @since 1.22
 166          */
 167         final public function delayedJobsEnabled() {
 168                 return $this->supportsDelayedJobs();
 169         }
 170
 171         /**
 172          * Quickly check if the queue has no available (unacquired, non-delayed) jobs.
 173          * Queue classes should use caching if they are any slower without memcached.
 174          *
 175          * If caching is used, this might return false when there are actually no jobs.
 176          * If pop() is called and returns false then it should correct the cache. Also,
 177          * calling flushCaches() first prevents this. However, this affect is typically
 178          * not distinguishable from the race condition between isEmpty() and pop().
 179          *
 180          * @return bool
 181          * @throws JobQueueError
 182          */
 183         final public function isEmpty() {
 184                 $res = $this->doIsEmpty();
 185
 186                 return $res;
 187         }
 188
 189         /**
 190          * @see JobQueue::isEmpty()
 191          * @return bool
 192          */
 193         abstract protected function doIsEmpty();
 194
 195         /**
 196          * Get the number of available (unacquired, non-delayed) jobs in the queue.
 197          * Queue classes should use caching if they are any slower without memcached.
 198          *
 199          * If caching is used, this number might be out of date for a minute.
 200          *
 201          * @return int
 202          * @throws JobQueueError
 203          */
 204         final public function getSize() {
 205                 $res = $this->doGetSize();
 206
 207                 return $res;
 208         }
 209
 210         /**
 211          * @see JobQueue::getSize()
 212          * @return int
 213          */
 214         abstract protected function doGetSize();
 215
 216         /**
 217          * Get the number of acquired jobs (these are temporarily out of the queue).
 218          * Queue classes should use caching if they are any slower without memcached.
 219          *
 220          * If caching is used, this number might be out of date for a minute.
 221          *
 222          * @return int
 223          * @throws JobQueueError
 224          */
 225         final public function getAcquiredCount() {
 226                 $res = $this->doGetAcquiredCount();
 227
 228                 return $res;
 229         }
 230
 231         /**
 232          * @see JobQueue::getAcquiredCount()
 233          * @return int
 234          */
 235         abstract protected function doGetAcquiredCount();
 236
 237         /**
 238          * Get the number of delayed jobs (these are temporarily out of the queue).
 239          * Queue classes should use caching if they are any slower without memcached.
 240          *
 241          * If caching is used, this number might be out of date for a minute.
 242          *
 243          * @return int
 244          * @throws JobQueueError
 245          * @since 1.22
 246          */
 247         final public function getDelayedCount() {
 248                 $res = $this->doGetDelayedCount();
 249
 250                 return $res;
 251         }
 252
 253         /**
 254          * @see JobQueue::getDelayedCount()
 255          * @return int
 256          */
 257         protected function doGetDelayedCount() {
 258                 return 0; // not implemented
 259         }
 260
 261         /**
 262          * Get the number of acquired jobs that can no longer be attempted.
 263          * Queue classes should use caching if they are any slower without memcached.
 264          *
 265          * If caching is used, this number might be out of date for a minute.
 266          *
 267          * @return int
 268          * @throws JobQueueError
 269          */
 270         final public function getAbandonedCount() {
 271                 $res = $this->doGetAbandonedCount();
 272
 273                 return $res;
 274         }
 275
 276         /**
 277          * @see JobQueue::getAbandonedCount()
 278          * @return int
 279          */
 280         protected function doGetAbandonedCount() {
 281                 return 0; // not implemented
 282         }
 283
 284         /**
 285          * Push one or more jobs into the queue.
 286          * This does not require $wgJobClasses to be set for the given job type.
 287          * Outside callers should use JobQueueGroup::push() instead of this function.
 288          *
 289          * @param JobSpecification|JobSpecification[] $jobs
 290          * @param int $flags Bitfield (supports JobQueue::QOS_ATOMIC)
 291          * @return void
 292          * @throws JobQueueError
 293          */
 294         final public function push( $jobs, $flags = 0 ) {
 295                 $jobs = is_array( $jobs ) ? $jobs : array( $jobs );
 296                 $this->batchPush( $jobs, $flags );
 297         }
 298
 299         /**
 300          * Push a batch of jobs into the queue.
 301          * This does not require $wgJobClasses to be set for the given job type.
 302          * Outside callers should use JobQueueGroup::push() instead of this function.
 303          *
 304          * @param JobSpecification[] $jobs
 305          * @param int $flags Bitfield (supports JobQueue::QOS_ATOMIC)
 306          * @return void
 307          * @throws MWException
 308          */
 309         final public function batchPush( array $jobs, $flags = 0 ) {
 310                 if ( !count( $jobs ) ) {
 311                         return; // nothing to do
 312                 }
 313
 314                 foreach ( $jobs as $job ) {
 315                         if ( $job->getType() !== $this->type ) {
 316                                 throw new MWException(
 317                                         "Got '{$job->getType()}' job; expected a '{$this->type}' job." );
 318                         } elseif ( $job->getReleaseTimestamp() && !$this->supportsDelayedJobs() ) {
 319                                 throw new MWException(
 320                                         "Got delayed '{$job->getType()}' job; delays are not supported." );
 321                         }
 322                 }
 323
 324                 $this->doBatchPush( $jobs, $flags );
 325                 $this->aggr->notifyQueueNonEmpty( $this->wiki, $this->type );
 326
 327                 foreach ( $jobs as $job ) {
 328                         if ( $job->isRootJob() ) {
 329                                 $this->deduplicateRootJob( $job );
 330                         }
 331                 }
 332         }
 333
 334         /**
 335          * @see JobQueue::batchPush()
 336          * @param JobSpecification[] $jobs
 337          * @param int $flags
 338          */
 339         abstract protected function doBatchPush( array $jobs, $flags );
 340
 341         /**
 342          * Pop a job off of the queue.
 343          * This requires $wgJobClasses to be set for the given job type.
 344          * Outside callers should use JobQueueGroup::pop() instead of this function.
 345          *
 346          * @throws MWException
 347          * @return Job|bool Returns false if there are no jobs
 348          */
 349         final public function pop() {
 350                 global $wgJobClasses;
 351
 352                 if ( $this->wiki !== wfWikiID() ) {
 353                         throw new MWException( "Cannot pop '{$this->type}' job off foreign wiki queue." );
 354                 } elseif ( !isset( $wgJobClasses[$this->type] ) ) {
 355                         // Do not pop jobs if there is no class for the queue type
 356                         throw new MWException( "Unrecognized job type '{$this->type}'." );
 357                 }
 358
 359                 $job = $this->doPop();
 360
 361                 if ( !$job ) {
 362                         $this->aggr->notifyQueueEmpty( $this->wiki, $this->type );
 363                 }
 364
 365                 // Flag this job as an old duplicate based on its "root" job...
 366                 try {
 367                         if ( $job && $this->isRootJobOldDuplicate( $job ) ) {
 368                                 JobQueue::incrStats( 'dupe_pops', $this->type );
 369                                 $job = DuplicateJob::newFromJob( $job ); // convert to a no-op
 370                         }
 371                 } catch ( Exception $e ) {
 372                         // don't lose jobs over this
 373                 }
 374
 375                 return $job;
 376         }
 377
 378         /**
 379          * @see JobQueue::pop()
 380          * @return Job
 381          */
 382         abstract protected function doPop();
 383
 384         /**
 385          * Acknowledge that a job was completed.
 386          *
 387          * This does nothing for certain queue classes or if "claimTTL" is not set.
 388          * Outside callers should use JobQueueGroup::ack() instead of this function.
 389          *
 390          * @param Job $job
 391          * @return void
 392          * @throws MWException
 393          */
 394         final public function ack( Job $job ) {
 395                 if ( $job->getType() !== $this->type ) {
 396                         throw new MWException( "Got '{$job->getType()}' job; expected '{$this->type}'." );
 397                 }
 398                 $this->doAck( $job );
 399         }
 400
 401         /**
 402          * @see JobQueue::ack()
 403          * @param Job $job
 404          */
 405         abstract protected function doAck( Job $job );
 406
 407         /**
 408          * Register the "root job" of a given job into the queue for de-duplication.
 409          * This should only be called right *after* all the new jobs have been inserted.
 410          * This is used to turn older, duplicate, job entries into no-ops. The root job
 411          * information will remain in the registry until it simply falls out of cache.
 412          *
 413          * This requires that $job has two special fields in the "params" array:
 414          *   - rootJobSignature : hash (e.g. SHA1) that identifies the task
 415          *   - rootJobTimestamp : TS_MW timestamp of this instance of the task
 416          *
 417          * A "root job" is a conceptual job that consist of potentially many smaller jobs
 418          * that are actually inserted into the queue. For example, "refreshLinks" jobs are
 419          * spawned when a template is edited. One can think of the task as "update links
 420          * of pages that use template X" and an instance of that task as a "root job".
 421          * However, what actually goes into the queue are range and leaf job subtypes.
 422          * Since these jobs include things like page ID ranges and DB master positions,
 423          * and can morph into smaller jobs recursively, simple duplicate detection
 424          * for individual jobs being identical (like that of job_sha1) is not useful.
 425          *
 426          * In the case of "refreshLinks", if these jobs are still in the queue when the template
 427          * is edited again, we want all of these old refreshLinks jobs for that template to become
 428          * no-ops. This can greatly reduce server load, since refreshLinks jobs involves parsing.
 429          * Essentially, the new batch of jobs belong to a new "root job" and the older ones to a
 430          * previous "root job" for the same task of "update links of pages that use template X".
 431          *
 432          * This does nothing for certain queue classes.
 433          *
 434          * @param IJobSpecification $job
 435          * @throws MWException
 436          * @return bool
 437          */
 438         final public function deduplicateRootJob( IJobSpecification $job ) {
 439                 if ( $job->getType() !== $this->type ) {
 440                         throw new MWException( "Got '{$job->getType()}' job; expected '{$this->type}'." );
 441                 }
 442                 $ok = $this->doDeduplicateRootJob( $job );
 443
 444                 return $ok;
 445         }
 446
 447         /**
 448          * @see JobQueue::deduplicateRootJob()
 449          * @param IJobSpecification $job
 450          * @throws MWException
 451          * @return bool
 452          */
 453         protected function doDeduplicateRootJob( IJobSpecification $job ) {
 454                 if ( !$job->hasRootJobParams() ) {
 455                         throw new MWException( "Cannot register root job; missing parameters." );
 456                 }
 457                 $params = $job->getRootJobParams();
 458
 459                 $key = $this->getRootJobCacheKey( $params['rootJobSignature'] );
 460                 // Callers should call batchInsert() and then this function so that if the insert
 461                 // fails, the de-duplication registration will be aborted. Since the insert is
 462                 // deferred till "transaction idle", do the same here, so that the ordering is
 463                 // maintained. Having only the de-duplication registration succeed would cause
 464                 // jobs to become no-ops without any actual jobs that made them redundant.
 465                 $timestamp = $this->dupCache->get( $key ); // current last timestamp of this job
 466                 if ( $timestamp && $timestamp >= $params['rootJobTimestamp'] ) {
 467                         return true; // a newer version of this root job was enqueued
 468                 }
 469
 470                 // Update the timestamp of the last root job started at the location...
 471                 return $this->dupCache->set( $key, $params['rootJobTimestamp'], JobQueueDB::ROOTJOB_TTL );
 472         }
 473
 474         /**
 475          * Check if the "root" job of a given job has been superseded by a newer one
 476          *
 477          * @param Job $job
 478          * @throws MWException
 479          * @return bool
 480          */
 481         final protected function isRootJobOldDuplicate( Job $job ) {
 482                 if ( $job->getType() !== $this->type ) {
 483                         throw new MWException( "Got '{$job->getType()}' job; expected '{$this->type}'." );
 484                 }
 485                 $isDuplicate = $this->doIsRootJobOldDuplicate( $job );
 486
 487                 return $isDuplicate;
 488         }
 489
 490         /**
 491          * @see JobQueue::isRootJobOldDuplicate()
 492          * @param Job $job
 493          * @return bool
 494          */
 495         protected function doIsRootJobOldDuplicate( Job $job ) {
 496                 if ( !$job->hasRootJobParams() ) {
 497                         return false; // job has no de-deplication info
 498                 }
 499                 $params = $job->getRootJobParams();
 500
 501                 $key = $this->getRootJobCacheKey( $params['rootJobSignature'] );
 502                 // Get the last time this root job was enqueued
 503                 $timestamp = $this->dupCache->get( $key );
 504
 505                 // Check if a new root job was started at the location after this one's...
 506                 return ( $timestamp && $timestamp > $params['rootJobTimestamp'] );
 507         }
 508
 509         /**
 510          * @param string $signature Hash identifier of the root job
 511          * @return string
 512          */
 513         protected function getRootJobCacheKey( $signature ) {
 514                 list( $db, $prefix ) = wfSplitWikiID( $this->wiki );
 515
 516                 return wfForeignMemcKey( $db, $prefix, 'jobqueue', $this->type, 'rootjob', $signature );
 517         }
 518
 519         /**
 520          * Deleted all unclaimed and delayed jobs from the queue
 521          *
 522          * @throws JobQueueError
 523          * @since 1.22
 524          * @return void
 525          */
 526         final public function delete() {
 527                 $this->doDelete();
 528         }
 529
 530         /**
 531          * @see JobQueue::delete()
 532          * @throws MWException
 533          */
 534         protected function doDelete() {
 535                 throw new MWException( "This method is not implemented." );
 536         }
 537
 538         /**
 539          * Wait for any slaves or backup servers to catch up.
 540          *
 541          * This does nothing for certain queue classes.
 542          *
 543          * @return void
 544          * @throws JobQueueError
 545          */
 546         final public function waitForBackups() {
 547                 $this->doWaitForBackups();
 548         }
 549
 550         /**
 551          * @see JobQueue::waitForBackups()
 552          * @return void
 553          */
 554         protected function doWaitForBackups() {
 555         }
 556
 557         /**
 558          * Clear any process and persistent caches
 559          *
 560          * @return void
 561          */
 562         final public function flushCaches() {
 563                 $this->doFlushCaches();
 564         }
 565
 566         /**
 567          * @see JobQueue::flushCaches()
 568          * @return void
 569          */
 570         protected function doFlushCaches() {
 571         }
 572
 573         /**
 574          * Get an iterator to traverse over all available jobs in this queue.
 575          * This does not include jobs that are currently acquired or delayed.
 576          * Note: results may be stale if the queue is concurrently modified.
 577          *
 578          * @return Iterator
 579          * @throws JobQueueError
 580          */
 581         abstract public function getAllQueuedJobs();
 582
 583         /**
 584          * Get an iterator to traverse over all delayed jobs in this queue.
 585          * Note: results may be stale if the queue is concurrently modified.
 586          *
 587          * @return Iterator
 588          * @throws JobQueueError
 589          * @since 1.22
 590          */
 591         public function getAllDelayedJobs() {
 592                 return new ArrayIterator( array() ); // not implemented
 593         }
 594
 595         /**
 596          * Get an iterator to traverse over all claimed jobs in this queue
 597          *
 598          * Callers should be quick to iterator over it or few results
 599          * will be returned due to jobs being acknowledged and deleted
 600          *
 601          * @return Iterator
 602          * @throws JobQueueError
 603          * @since 1.26
 604          */
 605         public function getAllAcquiredJobs() {
 606                 return new ArrayIterator( array() ); // not implemented
 607         }
 608
 609         /**
 610          * Get an iterator to traverse over all abandoned jobs in this queue
 611          *
 612          * @return Iterator
 613          * @throws JobQueueError
 614          * @since 1.25
 615          */
 616         public function getAllAbandonedJobs() {
 617                 return new ArrayIterator( array() ); // not implemented
 618         }
 619
 620         /**
 621          * Do not use this function outside of JobQueue/JobQueueGroup
 622          *
 623          * @return string
 624          * @since 1.22
 625          */
 626         public function getCoalesceLocationInternal() {
 627                 return null;
 628         }
 629
 630         /**
 631          * Check whether each of the given queues are empty.
 632          * This is used for batching checks for queues stored at the same place.
 633          *
 634          * @param array $types List of queues types
 635          * @return array|null (list of non-empty queue types) or null if unsupported
 636          * @throws MWException
 637          * @since 1.22
 638          */
 639         final public function getSiblingQueuesWithJobs( array $types ) {
 640                 return $this->doGetSiblingQueuesWithJobs( $types );
 641         }
 642
 643         /**
 644          * @see JobQueue::getSiblingQueuesWithJobs()
 645          * @param array $types List of queues types
 646          * @return array|null (list of queue types) or null if unsupported
 647          */
 648         protected function doGetSiblingQueuesWithJobs( array $types ) {
 649                 return null; // not supported
 650         }
 651
 652         /**
 653          * Check the size of each of the given queues.
 654          * For queues not served by the same store as this one, 0 is returned.
 655          * This is used for batching checks for queues stored at the same place.
 656          *
 657          * @param array $types List of queues types
 658          * @return array|null (job type => whether queue is empty) or null if unsupported
 659          * @throws MWException
 660          * @since 1.22
 661          */
 662         final public function getSiblingQueueSizes( array $types ) {
 663                 return $this->doGetSiblingQueueSizes( $types );
 664         }
 665
 666         /**
 667          * @see JobQueue::getSiblingQueuesSize()
 668          * @param array $types List of queues types
 669          * @return array|null (list of queue types) or null if unsupported
 670          */
 671         protected function doGetSiblingQueueSizes( array $types ) {
 672                 return null; // not supported
 673         }
 674
 675         /**
 676          * Call wfIncrStats() for the queue overall and for the queue type
 677          *
 678          * @param string $key Event type
 679          * @param string $type Job type
 680          * @param int $delta
 681          * @since 1.22
 682          */
 683         public static function incrStats( $key, $type, $delta = 1 ) {
 684                 static $stats;
 685                 if ( !$stats ) {
 686                         $stats = RequestContext::getMain()->getStats();
 687                 }
 688                 $stats->updateCount( "jobqueue.{$key}.all", $delta );
 689                 $stats->updateCount( "jobqueue.{$key}.{$type}", $delta );
 690         }
 691
 692         /**
 693          * Namespace the queue with a key to isolate it for testing
 694          *
 695          * @param string $key
 696          * @return void
 697          * @throws MWException
 698          */
 699         public function setTestingPrefix( $key ) {
 700                 throw new MWException( "Queue namespacing not supported for this queue type." );
 701         }
 702 }
 703
 704 /**
 705  * @ingroup JobQueue
 706  * @since 1.22
 707  */
 708 class JobQueueError extends MWException {
 709 }
 710
 711 class JobQueueConnectionError extends JobQueueError {
 712 }