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