includes/job/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         protected $wiki; // string; wiki ID
  33         protected $type; // string; job type
  34         protected $order; // string; job priority for pop()
  35         protected $claimTTL; // integer; seconds
  36
  37         const QoS_Atomic = 1; // integer; "all-or-nothing" job insertions
  38
  39         /**
  40          * @param $params array
  41          */
  42         protected function __construct( array $params ) {
  43                 $this->wiki     = $params['wiki'];
  44                 $this->type     = $params['type'];
  45                 $this->order    = isset( $params['order'] ) ? $params['order'] : 'random';
  46                 $this->claimTTL = isset( $params['claimTTL'] ) ? $params['claimTTL'] : 0;
  47         }
  48
  49         /**
  50          * Get a job queue object of the specified type.
  51          * $params includes:
  52          *   class    : What job class to use (determines job type)
  53          *   wiki     : wiki ID of the wiki the jobs are for (defaults to current wiki)
  54          *   type     : The name of the job types this queue handles
  55          *   order    : Order that pop() selects jobs, one of "fifo", "timestamp" or "random".
  56          *              If "fifo" is used, the queue will effectively be FIFO. Note that
  57          *              job completion will not appear to be exactly FIFO if there are multiple
  58          *              job runners since jobs can take different times to finish once popped.
  59          *              If "timestamp" is used, the queue will at least be loosely ordered
  60          *              by timestamp, allowing for some jobs to be popped off out of order.
  61          *              If "random" is used, pop() will pick jobs in random order. This might be
  62          *              useful for improving concurrency depending on the queue storage medium.
  63          *   claimTTL : If supported, the queue will recycle jobs that have been popped
  64          *              but not acknowledged as completed after this many seconds.
  65          *
  66          * @param $params array
  67          * @return JobQueue
  68          * @throws MWException
  69          */
  70         final public static function factory( array $params ) {
  71                 $class = $params['class'];
  72                 if ( !MWInit::classExists( $class ) ) {
  73                         throw new MWException( "Invalid job queue class '$class'." );
  74                 }
  75                 $obj = new $class( $params );
  76                 if ( !( $obj instanceof self ) ) {
  77                         throw new MWException( "Class '$class' is not a " . __CLASS__ . " class." );
  78                 }
  79                 return $obj;
  80         }
  81
  82         /**
  83          * @return string Wiki ID
  84          */
  85         final public function getWiki() {
  86                 return $this->wiki;
  87         }
  88
  89         /**
  90          * @return string Job type that this queue handles
  91          */
  92         final public function getType() {
  93                 return $this->type;
  94         }
  95
  96         /**
  97          * Quickly check if the queue is empty.
  98          * Queue classes should use caching if they are any slower without memcached.
  99          *
 100          * @return bool
 101          */
 102         final public function isEmpty() {
 103                 wfProfileIn( __METHOD__ );
 104                 $res = $this->doIsEmpty();
 105                 wfProfileOut( __METHOD__ );
 106                 return $res;
 107         }
 108
 109         /**
 110          * @see JobQueue::isEmpty()
 111          * @return bool
 112          */
 113         abstract protected function doIsEmpty();
 114
 115         /**
 116          * Push a batch of jobs into the queue
 117          *
 118          * @param $jobs array List of Jobs
 119          * @param $flags integer Bitfield (supports JobQueue::QoS_Atomic)
 120          * @return bool
 121          */
 122         final public function batchPush( array $jobs, $flags = 0 ) {
 123                 foreach ( $jobs as $job ) {
 124                         if ( $job->getType() !== $this->type ) {
 125                                 throw new MWException( "Got '{$job->getType()}' job; expected '{$this->type}'." );
 126                         }
 127                 }
 128                 wfProfileIn( __METHOD__ );
 129                 $ok = $this->doBatchPush( $jobs, $flags );
 130                 if ( $ok ) {
 131                         wfIncrStats( 'job-insert', count( $jobs ) );
 132                 }
 133                 wfProfileOut( __METHOD__ );
 134                 return $ok;
 135         }
 136
 137         /**
 138          * @see JobQueue::batchPush()
 139          * @return bool
 140          */
 141         abstract protected function doBatchPush( array $jobs, $flags );
 142
 143         /**
 144          * Pop a job off of the queue
 145          *
 146          * @return Job|bool Returns false on failure
 147          */
 148         final public function pop() {
 149                 wfProfileIn( __METHOD__ );
 150                 $job = $this->doPop();
 151                 if ( $job ) {
 152                         wfIncrStats( 'job-pop' );
 153                 }
 154                 wfProfileOut( __METHOD__ );
 155                 return $job;
 156         }
 157
 158         /**
 159          * @see JobQueue::pop()
 160          * @return Job
 161          */
 162         abstract protected function doPop();
 163
 164         /**
 165          * Acknowledge that a job was completed
 166          *
 167          * @param $job Job
 168          * @return bool
 169          */
 170         final public function ack( Job $job ) {
 171                 if ( $job->getType() !== $this->type ) {
 172                         throw new MWException( "Got '{$job->getType()}' job; expected '{$this->type}'." );
 173                 }
 174                 wfProfileIn( __METHOD__ );
 175                 $ok = $this->doAck( $job );
 176                 wfProfileOut( __METHOD__ );
 177                 return $ok;
 178         }
 179
 180         /**
 181          * @see JobQueue::ack()
 182          * @return bool
 183          */
 184         abstract protected function doAck( Job $job );
 185
 186         /**
 187          * Wait for any slaves or backup servers to catch up
 188          *
 189          * @return void
 190          */
 191         final public function waitForBackups() {
 192                 wfProfileIn( __METHOD__ );
 193                 $this->doWaitForBackups();
 194                 wfProfileOut( __METHOD__ );
 195         }
 196
 197         /**
 198          * @see JobQueue::waitForBackups()
 199          * @return void
 200          */
 201         protected function doWaitForBackups() {}
 202 }