includes/job/JobQueueDB.php

   1 <?php
   2 /**
   3  * Database-backed job queue 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  * @author Aaron Schulz
  22  */
  23
  24 /**
  25  * Class to handle job queues stored in the DB
  26  *
  27  * @ingroup JobQueue
  28  * @since 1.20
  29  */
  30 class JobQueueDB extends JobQueue {
  31         const CACHE_TTL      = 30; // integer; seconds
  32         const MAX_JOB_RANDOM = 2147483647; // 2^31 - 1; used for job_random
  33
  34         /**
  35          * @see JobQueue::doIsEmpty()
  36          * @return bool
  37          */
  38         protected function doIsEmpty() {
  39                 global $wgMemc;
  40
  41                 $key = $this->getEmptinessCacheKey();
  42
  43                 $isEmpty = $wgMemc->get( $key );
  44                 if ( $isEmpty === 'true' ) {
  45                         return true;
  46                 } elseif ( $isEmpty === 'false' ) {
  47                         return false;
  48                 }
  49
  50                 $found = $this->getSlaveDB()->selectField(
  51                         'job', '1', array( 'job_cmd' => $this->type ), __METHOD__
  52                 );
  53
  54                 $wgMemc->add( $key, $found ? 'false' : 'true', self::CACHE_TTL );
  55         }
  56
  57         /**
  58          * @see JobQueue::doBatchPush()
  59          * @return bool
  60          */
  61         protected function doBatchPush( array $jobs, $flags ) {
  62                 if ( count( $jobs ) ) {
  63                         $dbw = $this->getMasterDB();
  64
  65                         $rows = array();
  66                         foreach ( $jobs as $job ) {
  67                                 $rows[] = $this->insertFields( $job );
  68                         }
  69                         $atomic = ( $flags & self::QoS_Atomic );
  70                         $key    = $this->getEmptinessCacheKey();
  71                         $ttl    = self::CACHE_TTL;
  72
  73                         $dbw->onTransactionIdle( function() use ( $dbw, $rows, $atomic, $key, $ttl ) {
  74                                 global $wgMemc;
  75
  76                                 $autoTrx = $dbw->getFlag( DBO_TRX ); // automatic begin() enabled?
  77                                 if ( $atomic ) {
  78                                         $dbw->begin(); // wrap all the job additions in one transaction
  79                                 } else {
  80                                         $dbw->clearFlag( DBO_TRX ); // make each query its own transaction
  81                                 }
  82                                 try {
  83                                         foreach ( array_chunk( $rows, 50 ) as $rowBatch ) { // avoid slave lag
  84                                                 $dbw->insert( 'job', $rowBatch, __METHOD__ );
  85                                         }
  86                                 } catch ( DBError $e ) {
  87                                         if ( $atomic ) {
  88                                                 $dbw->rollback();
  89                                         } else {
  90                                                 $dbw->setFlag( $autoTrx ? DBO_TRX : 0 ); // restore automatic begin()
  91                                         }
  92                                         throw $e;
  93                                 }
  94                                 if ( $atomic ) {
  95                                         $dbw->commit();
  96                                 } else {
  97                                         $dbw->setFlag( $autoTrx ? DBO_TRX : 0 ); // restore automatic begin()
  98                                 }
  99
 100                                 $wgMemc->set( $key, 'false', $ttl );
 101                         } );
 102                 }
 103
 104                 return true;
 105         }
 106
 107         /**
 108          * @see JobQueue::doPop()
 109          * @return Job|bool
 110          */
 111         protected function doPop() {
 112                 global $wgMemc;
 113
 114                 if ( $wgMemc->get( $this->getEmptinessCacheKey() ) === 'true' ) {
 115                         return false; // queue is empty
 116                 }
 117
 118                 $dbw = $this->getMasterDB();
 119                 $dbw->commit( __METHOD__, 'flush' ); // flush existing transaction
 120
 121                 $uuid = wfRandomString( 32 ); // pop attempt
 122                 $job = false; // job popped off
 123                 $autoTrx = $dbw->getFlag( DBO_TRX ); // automatic begin() enabled?
 124                 $dbw->clearFlag( DBO_TRX ); // make each query its own transaction
 125                 try {
 126                         do { // retry when our row is invalid or deleted as a duplicate
 127                                 // Try to reserve a row in the DB...
 128                                 if ( $this->order === 'timestamp' ) { // oldest first
 129                                         $row = $this->claimOldest( $uuid );
 130                                 } else { // random first
 131                                         $rand = mt_rand( 0, self::MAX_JOB_RANDOM ); // encourage concurrent UPDATEs
 132                                         $gte  = (bool)mt_rand( 0, 1 ); // find rows with rand before/after $rand
 133                                         $row  = $this->claimRandom( $uuid, $rand, $gte );
 134                                         if ( !$row ) { // need to try the other direction
 135                                                 $row = $this->claimRandom( $uuid, $rand, !$gte );
 136                                         }
 137                                 }
 138                                 // Check if we found a row to reserve...
 139                                 if ( !$row ) {
 140                                         $wgMemc->set( $this->getEmptinessCacheKey(), 'true', self::CACHE_TTL );
 141                                         break; // nothing to do
 142                                 }
 143                                 // Get the job object from the row...
 144                                 $title = Title::makeTitleSafe( $row->job_namespace, $row->job_title );
 145                                 if ( !$title ) {
 146                                         $dbw->delete( 'job', array( 'job_id' => $row->job_id ), __METHOD__ );
 147                                         wfIncrStats( 'job-pop' );
 148                                         wfDebugLog( 'JobQueueDB', "Row has invalid title '{$row->job_title}'." );
 149                                         continue; // try again
 150                                 }
 151                                 $job = Job::factory( $row->job_cmd, $title,
 152                                         self::extractBlob( $row->job_params ), $row->job_id );
 153                                 // Delete any *other* duplicate jobs in the queue...
 154                                 if ( $job->ignoreDuplicates() && strlen( $row->job_sha1 ) ) {
 155                                         $dbw->delete( 'job',
 156                                                 array( 'job_sha1' => $row->job_sha1,
 157                                                         "job_id != {$dbw->addQuotes( $row->job_id )}" ),
 158                                                 __METHOD__
 159                                         );
 160                                         wfIncrStats( 'job-pop', $dbw->affectedRows() );
 161                                 }
 162                                 break; // done
 163                         } while( true );
 164                 } catch ( DBError $e ) {
 165                         $dbw->setFlag( $autoTrx ? DBO_TRX : 0 ); // restore automatic begin()
 166                         throw $e;
 167                 }
 168                 $dbw->setFlag( $autoTrx ? DBO_TRX : 0 ); // restore automatic begin()
 169
 170                 return $job;
 171         }
 172
 173         /**
 174          * Reserve a row with a single UPDATE without holding row locks over RTTs...
 175          *
 176          * @param $uuid string 32 char hex string
 177          * @param $rand integer Random unsigned integer (31 bits)
 178          * @param $gte bool Search for job_random >= $random (otherwise job_random <= $random)
 179          * @return Row|false
 180          */
 181         protected function claimRandom( $uuid, $rand, $gte ) {
 182                 $dbw  = $this->getMasterDB();
 183                 $dir  = $gte ? 'ASC' : 'DESC';
 184                 $ineq = $gte ? '>=' : '<=';
 185
 186                 $row = false; // the row acquired
 187                 // This uses a replication safe method for acquiring jobs. One could use UPDATE+LIMIT
 188                 // instead, but that either uses ORDER BY (in which case it deadlocks in MySQL) or is
 189                 // not replication safe. Due to http://bugs.mysql.com/bug.php?id=6980, subqueries cannot
 190                 // be used here with MySQL.
 191                 do {
 192                         $row = $dbw->selectRow( 'job', '*', // find a random job
 193                                 array(
 194                                         'job_cmd'   => $this->type,
 195                                         'job_token' => '',
 196                                         "job_random {$ineq} {$dbw->addQuotes( $rand )}" ),
 197                                 __METHOD__,
 198                                 array( 'ORDER BY' => "job_random {$dir}" )
 199                         );
 200                         if ( $row ) { // claim the job
 201                                 $dbw->update( 'job', // update by PK
 202                                         array( 'job_token' => $uuid, 'job_token_timestamp' => $dbw->timestamp() ),
 203                                         array( 'job_cmd' => $this->type, 'job_id' => $row->job_id, 'job_token' => '' ),
 204                                         __METHOD__
 205                                 );
 206                                 // This might get raced out by another runner when claiming the previously
 207                                 // selected row. The use of job_random should minimize this problem, however.
 208                                 if ( !$dbw->affectedRows() ) {
 209                                         $row = false; // raced out
 210                                 }
 211                         } else {
 212                                 break; // nothing to do
 213                         }
 214                 } while ( !$row );
 215
 216                 return $row;
 217         }
 218
 219         /**
 220          * Reserve a row with a single UPDATE without holding row locks over RTTs...
 221          *
 222          * @param $uuid string 32 char hex string
 223          * @return Row|false
 224          */
 225         protected function claimOldest( $uuid ) {
 226                 $dbw  = $this->getMasterDB();
 227
 228                 $row = false; // the row acquired
 229                 do {
 230                         if ( $dbw->getType() === 'mysql' ) {
 231                                 // Per http://bugs.mysql.com/bug.php?id=6980, we can't use subqueries on the
 232                                 // same table being changed in an UPDATE query in MySQL (gives Error: 1093).
 233                                 // Oracle and Postgre have no such limitation. However, MySQL offers an
 234                                 // alternative here by supporting ORDER BY + LIMIT for UPDATE queries.
 235                                 $dbw->query( "UPDATE {$dbw->tableName( 'job' )}
 236                                         SET
 237                                                 job_token = {$dbw->addQuotes( $uuid ) },
 238                                                 job_token_timestamp = {$dbw->addQuotes( $dbw->timestamp() )}
 239                                         WHERE (
 240                                                 job_cmd = {$dbw->addQuotes( $this->type )}
 241                                                 AND job_token = {$dbw->addQuotes( '' )}
 242                                         ) ORDER BY job_random ASC LIMIT 1",
 243                                         __METHOD__
 244                                 );
 245                         } else {
 246                                 // Use a subquery to find the job, within an UPDATE to claim it.
 247                                 // This uses as much of the DB wrapper functions as possible.
 248                                 $dbw->update( 'job',
 249                                         array( 'job_token' => $uuid, 'job_token_timestamp' => $dbw->timestamp() ),
 250                                         array( 'job_id = (' .
 251                                                 $dbw->selectSQLText( 'job', 'job_id',
 252                                                         array( 'job_cmd' => $this->type, 'job_token' => '' ),
 253                                                         __METHOD__,
 254                                                         array( 'ORDER BY' => 'job_random ASC', 'LIMIT' => 1 ) ) .
 255                                                 ')'
 256                                         ),
 257                                         __METHOD__
 258                                 );
 259                         }
 260                         // Fetch any row that we just reserved...
 261                         if ( $dbw->affectedRows() ) {
 262                                 $row = $dbw->selectRow( 'job', '*',
 263                                         array( 'job_cmd' => $this->type, 'job_token' => $uuid ), __METHOD__
 264                                 );
 265                                 if ( !$row ) { // raced out by duplicate job removal
 266                                         wfDebugLog( 'JobQueueDB', "Row deleted as duplicate by another process." );
 267                                 }
 268                         } else {
 269                                 break; // nothing to do
 270                         }
 271                 } while ( !$row );
 272
 273                 return $row;
 274         }
 275
 276         /**
 277          * @see JobQueue::doAck()
 278          * @return Job|bool
 279          */
 280         protected function doAck( Job $job ) {
 281                 $dbw = $this->getMasterDB();
 282                 $dbw->commit( __METHOD__, 'flush' ); // flush existing transaction
 283
 284                 $autoTrx = $dbw->getFlag( DBO_TRX ); // automatic begin() enabled?
 285                 $dbw->clearFlag( DBO_TRX ); // make each query its own transaction
 286                 try {
 287                         // Delete a row with a single DELETE without holding row locks over RTTs...
 288                         $dbw->delete( 'job', array( 'job_cmd' => $this->type, 'job_id' => $job->getId() ) );
 289                 } catch ( Exception $e ) {
 290                         $dbw->setFlag( $autoTrx ? DBO_TRX : 0 ); // restore automatic begin()
 291                         throw $e;
 292                 }
 293                 $dbw->setFlag( $autoTrx ? DBO_TRX : 0 ); // restore automatic begin()
 294
 295                 return true;
 296         }
 297
 298         /**
 299          * @see JobQueue::doWaitForBackups()
 300          * @return void
 301          */
 302         protected function doWaitForBackups() {
 303                 wfWaitForSlaves();
 304         }
 305
 306         /**
 307          * @return DatabaseBase
 308          */
 309         protected function getSlaveDB() {
 310                 return wfGetDB( DB_SLAVE, array(), $this->wiki );
 311         }
 312
 313         /**
 314          * @return DatabaseBase
 315          */
 316         protected function getMasterDB() {
 317                 return wfGetDB( DB_MASTER, array(), $this->wiki );
 318         }
 319
 320         /**
 321          * @param $job Job
 322          * @return array
 323          */
 324         protected function insertFields( Job $job ) {
 325                 // Rows that describe the nature of the job
 326                 $descFields = array(
 327                         'job_cmd'       => $job->getType(),
 328                         'job_namespace' => $job->getTitle()->getNamespace(),
 329                         'job_title'     => $job->getTitle()->getDBkey(),
 330                         'job_params'    => self::makeBlob( $job->getParams() ),
 331                 );
 332                 // Additional job metadata
 333                 if ( $this->order === 'timestamp' ) { // oldest first
 334                         $random = time() - 1325376000; // seconds since "January 1, 2012"
 335                 } else { // random first
 336                         $random = mt_rand( 0, self::MAX_JOB_RANDOM );
 337                 }
 338                 $dbw = $this->getMasterDB();
 339                 $metaFields = array(
 340                         'job_id'        => $dbw->nextSequenceValue( 'job_job_id_seq' ),
 341                         'job_timestamp' => $dbw->timestamp(),
 342                         'job_sha1'      => wfBaseConvert( sha1( serialize( $descFields ) ), 16, 36, 32 ),
 343                         'job_random'    => mt_rand( 0, self::MAX_JOB_RANDOM )
 344                 );
 345                 return ( $descFields + $metaFields );
 346         }
 347
 348         /**
 349          * @return string
 350          */
 351         private function getEmptinessCacheKey() {
 352                 list( $db, $prefix ) = wfSplitWikiID( $this->wiki );
 353                 return wfForeignMemcKey( $db, $prefix, 'jobqueue', $this->type, 'isempty' );
 354         }
 355
 356         /**
 357          * @param $params
 358          * @return string
 359          */
 360         protected static function makeBlob( $params ) {
 361                 if ( $params !== false ) {
 362                         return serialize( $params );
 363                 } else {
 364                         return '';
 365                 }
 366         }
 367
 368         /**
 369          * @param $blob
 370          * @return bool|mixed
 371          */
 372         protected static function extractBlob( $blob ) {
 373                 if ( (string)$blob !== '' ) {
 374                         return unserialize( $blob );
 375                 } else {
 376                         return false;
 377                 }
 378         }
 379 }