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