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.21
  29  */
  30 class JobQueueDB extends JobQueue {
  31         const CACHE_TTL      = 300; // integer; seconds to cache queue information
  32         const MAX_AGE_PRUNE  = 604800; // integer; seconds a job can live once claimed
  33         const MAX_ATTEMPTS   = 3; // integer; number of times to try a job
  34         const MAX_JOB_RANDOM = 2147483647; // integer; 2^31 - 1, used for job_random
  35
  36         /**
  37          * @see JobQueue::doIsEmpty()
  38          * @return bool
  39          */
  40         protected function doIsEmpty() {
  41                 global $wgMemc;
  42
  43                 $key = $this->getEmptinessCacheKey();
  44
  45                 $isEmpty = $wgMemc->get( $key );
  46                 if ( $isEmpty === 'true' ) {
  47                         return true;
  48                 } elseif ( $isEmpty === 'false' ) {
  49                         return false;
  50                 }
  51
  52                 $found = $this->getSlaveDB()->selectField( // unclaimed job
  53                         'job', '1', array( 'job_cmd' => $this->type, 'job_token' => '' ), __METHOD__
  54                 );
  55                 $wgMemc->add( $key, $found ? 'false' : 'true', self::CACHE_TTL );
  56
  57                 return !$found;
  58         }
  59
  60         /**
  61          * @see JobQueue::doBatchPush()
  62          * @param array $jobs
  63          * @param $flags
  64          * @throws DBError|Exception
  65          * @return bool
  66          */
  67         protected function doBatchPush( array $jobs, $flags ) {
  68                 if ( count( $jobs ) ) {
  69                         $dbw = $this->getMasterDB();
  70
  71                         $rowSet = array(); // (sha1 => job) map for jobs that are de-duplicated
  72                         $rowList = array(); // list of jobs for jobs that are are not de-duplicated
  73
  74                         foreach ( $jobs as $job ) {
  75                                 $row = $this->insertFields( $job );
  76                                 if ( $job->ignoreDuplicates() ) {
  77                                         $rowSet[$row['job_sha1']] = $row;
  78                                 } else {
  79                                         $rowList[] = $row;
  80                                 }
  81                         }
  82
  83                         $atomic = ( $flags & self::QoS_Atomic );
  84                         $key    = $this->getEmptinessCacheKey();
  85                         $ttl    = self::CACHE_TTL;
  86
  87                         $dbw->onTransactionIdle(
  88                                 function() use ( $dbw, $rowSet, $rowList, $atomic, $key, $ttl
  89                         ) {
  90                                 global $wgMemc;
  91
  92                                 if ( $atomic ) {
  93                                         $dbw->begin( __METHOD__ ); // wrap all the job additions in one transaction
  94                                 }
  95                                 try {
  96                                         // Strip out any duplicate jobs that are already in the queue...
  97                                         if ( count( $rowSet ) ) {
  98                                                 $res = $dbw->select( 'job', 'job_sha1',
  99                                                         array(
 100                                                                 // No job_type condition since it's part of the job_sha1 hash
 101                                                                 'job_sha1'  => array_keys( $rowSet ),
 102                                                                 'job_token' => '' // unclaimed
 103                                                         ),
 104                                                         __METHOD__
 105                                                 );
 106                                                 foreach ( $res as $row ) {
 107                                                         wfDebug( "Job with hash '{$row->job_sha1}' is a duplicate." );
 108                                                         unset( $rowSet[$row->job_sha1] ); // already enqueued
 109                                                 }
 110                                         }
 111                                         // Build the full list of job rows to insert
 112                                         $rows = array_merge( $rowList, array_values( $rowSet ) );
 113                                         // Insert the job rows in chunks to avoid slave lag...
 114                                         foreach ( array_chunk( $rows, 50 ) as $rowBatch ) {
 115                                                 $dbw->insert( 'job', $rowBatch, __METHOD__ );
 116                                         }
 117                                         wfIncrStats( 'job-insert', count( $rows ) );
 118                                 } catch ( DBError $e ) {
 119                                         if ( $atomic ) {
 120                                                 $dbw->rollback( __METHOD__ );
 121                                         }
 122                                         throw $e;
 123                                 }
 124                                 if ( $atomic ) {
 125                                         $dbw->commit( __METHOD__ );
 126                                 }
 127
 128                                 $wgMemc->set( $key, 'false', $ttl ); // queue is not empty
 129                         } );
 130                 }
 131
 132                 return true;
 133         }
 134
 135         /**
 136          * @see JobQueue::doPop()
 137          * @return Job|bool
 138          */
 139         protected function doPop() {
 140                 global $wgMemc;
 141
 142                 if ( $wgMemc->get( $this->getEmptinessCacheKey() ) === 'true' ) {
 143                         return false; // queue is empty
 144                 }
 145
 146                 $dbw = $this->getMasterDB();
 147                 $dbw->commit( __METHOD__, 'flush' ); // flush existing transaction
 148
 149                 $uuid = wfRandomString( 32 ); // pop attempt
 150                 $job = false; // job popped off
 151                 // Occasionally recycle jobs back into the queue that have been claimed too long
 152                 if ( mt_rand( 0, 99 ) == 0 ) {
 153                         $this->recycleStaleJobs();
 154                 }
 155                 do { // retry when our row is invalid or deleted as a duplicate
 156                         // Try to reserve a row in the DB...
 157                         if ( in_array( $this->order, array( 'fifo', 'timestamp' ) ) ) {
 158                                 $row = $this->claimOldest( $uuid );
 159                         } else { // random first
 160                                 $rand = mt_rand( 0, self::MAX_JOB_RANDOM ); // encourage concurrent UPDATEs
 161                                 $gte  = (bool)mt_rand( 0, 1 ); // find rows with rand before/after $rand
 162                                 $row  = $this->claimRandom( $uuid, $rand, $gte );
 163                                 if ( !$row ) { // need to try the other direction
 164                                         $row = $this->claimRandom( $uuid, $rand, !$gte );
 165                                 }
 166                         }
 167                         // Check if we found a row to reserve...
 168                         if ( !$row ) {
 169                                 $wgMemc->set( $this->getEmptinessCacheKey(), 'true', self::CACHE_TTL );
 170                                 break; // nothing to do
 171                         }
 172                         wfIncrStats( 'job-pop' );
 173                         // Get the job object from the row...
 174                         $title = Title::makeTitleSafe( $row->job_namespace, $row->job_title );
 175                         if ( !$title ) {
 176                                 $dbw->delete( 'job', array( 'job_id' => $row->job_id ), __METHOD__ );
 177                                 wfDebugLog( 'JobQueueDB', "Row has invalid title '{$row->job_title}'." );
 178                                 continue; // try again
 179                         }
 180                         $job = Job::factory( $row->job_cmd, $title,
 181                                 self::extractBlob( $row->job_params ), $row->job_id );
 182                         $job->id = $row->job_id; // XXX: work around broken subclasses
 183                         // Flag this job as an old duplicate based on its "root" job...
 184                         if ( $this->isRootJobOldDuplicate( $job ) ) {
 185                                 $job = DuplicateJob::newFromJob( $job ); // convert to a no-op
 186                         }
 187                         break; // done
 188                 } while( true );
 189
 190                 return $job;
 191         }
 192
 193         /**
 194          * Reserve a row with a single UPDATE without holding row locks over RTTs...
 195          *
 196          * @param $uuid string 32 char hex string
 197          * @param $rand integer Random unsigned integer (31 bits)
 198          * @param $gte bool Search for job_random >= $random (otherwise job_random <= $random)
 199          * @return Row|false
 200          */
 201         protected function claimRandom( $uuid, $rand, $gte ) {
 202                 $dbw  = $this->getMasterDB();
 203                 $ineq = $gte ? '>=' : '<=';
 204
 205                 $row = false; // the row acquired
 206                 // This uses a replication safe method for acquiring jobs. One could use UPDATE+LIMIT
 207                 // instead, but that either uses ORDER BY (in which case it deadlocks in MySQL) or is
 208                 // not replication safe. Due to http://bugs.mysql.com/bug.php?id=6980, subqueries cannot
 209                 // be used here with MySQL.
 210                 do {
 211                         $row = $dbw->selectRow( 'job', '*', // find a random job
 212                                 array(
 213                                         'job_cmd'   => $this->type,
 214                                         'job_token' => '',
 215                                         "job_random {$ineq} {$dbw->addQuotes( $rand )}" ),
 216                                 __METHOD__
 217                                 // Bug 42614: "ORDER BY job_random" causes slowness on mysql for some reason
 218                         );
 219                         if ( $row ) { // claim the job
 220                                 $dbw->update( 'job', // update by PK
 221                                         array(
 222                                                 'job_token'           => $uuid,
 223                                                 'job_token_timestamp' => $dbw->timestamp(),
 224                                                 'job_attempts = job_attempts+1' ),
 225                                         array( 'job_cmd' => $this->type, 'job_id' => $row->job_id, 'job_token' => '' ),
 226                                         __METHOD__
 227                                 );
 228                                 // This might get raced out by another runner when claiming the previously
 229                                 // selected row. The use of job_random should minimize this problem, however.
 230                                 if ( !$dbw->affectedRows() ) {
 231                                         $row = false; // raced out
 232                                 }
 233                         } else {
 234                                 break; // nothing to do
 235                         }
 236                 } while ( !$row );
 237
 238                 return $row;
 239         }
 240
 241         /**
 242          * Reserve a row with a single UPDATE without holding row locks over RTTs...
 243          *
 244          * @param $uuid string 32 char hex string
 245          * @return Row|false
 246          */
 247         protected function claimOldest( $uuid ) {
 248                 $dbw  = $this->getMasterDB();
 249
 250                 $row = false; // the row acquired
 251                 do {
 252                         if ( $dbw->getType() === 'mysql' ) {
 253                                 // Per http://bugs.mysql.com/bug.php?id=6980, we can't use subqueries on the
 254                                 // same table being changed in an UPDATE query in MySQL (gives Error: 1093).
 255                                 // Oracle and Postgre have no such limitation. However, MySQL offers an
 256                                 // alternative here by supporting ORDER BY + LIMIT for UPDATE queries.
 257                                 $dbw->query( "UPDATE {$dbw->tableName( 'job' )} " .
 258                                         "SET " .
 259                                                 "job_token = {$dbw->addQuotes( $uuid ) }, " .
 260                                                 "job_token_timestamp = {$dbw->addQuotes( $dbw->timestamp() )}, " .
 261                                                 "job_attempts = job_attempts+1 " .
 262                                         "WHERE ( " .
 263                                                 "job_cmd = {$dbw->addQuotes( $this->type )} " .
 264                                                 "AND job_token = {$dbw->addQuotes( '' )} " .
 265                                         ") ORDER BY job_id ASC LIMIT 1",
 266                                         __METHOD__
 267                                 );
 268                         } else {
 269                                 // Use a subquery to find the job, within an UPDATE to claim it.
 270                                 // This uses as much of the DB wrapper functions as possible.
 271                                 $dbw->update( 'job',
 272                                         array(
 273                                                 'job_token'           => $uuid,
 274                                                 'job_token_timestamp' => $dbw->timestamp(),
 275                                                 'job_attempts = job_attempts+1' ),
 276                                         array( 'job_id = (' .
 277                                                 $dbw->selectSQLText( 'job', 'job_id',
 278                                                         array( 'job_cmd' => $this->type, 'job_token' => '' ),
 279                                                         __METHOD__,
 280                                                         array( 'ORDER BY' => 'job_id ASC', 'LIMIT' => 1 ) ) .
 281                                                 ')'
 282                                         ),
 283                                         __METHOD__
 284                                 );
 285                         }
 286                         // Fetch any row that we just reserved...
 287                         if ( $dbw->affectedRows() ) {
 288                                 $row = $dbw->selectRow( 'job', '*',
 289                                         array( 'job_cmd' => $this->type, 'job_token' => $uuid ), __METHOD__
 290                                 );
 291                                 if ( !$row ) { // raced out by duplicate job removal
 292                                         wfDebugLog( 'JobQueueDB', "Row deleted as duplicate by another process." );
 293                                 }
 294                         } else {
 295                                 break; // nothing to do
 296                         }
 297                 } while ( !$row );
 298
 299                 return $row;
 300         }
 301
 302         /**
 303          * Recycle or destroy any jobs that have been claimed for too long
 304          *
 305          * @return integer Number of jobs recycled/deleted
 306          */
 307         protected function recycleStaleJobs() {
 308                 $now   = time();
 309                 $dbw   = $this->getMasterDB();
 310                 $count = 0; // affected rows
 311
 312                 if ( !$dbw->lock( "jobqueue-recycle-{$this->type}", __METHOD__, 1 ) ) {
 313                         return $count; // already in progress
 314                 }
 315
 316                 // Remove claims on jobs acquired for too long if enabled...
 317                 if ( $this->claimTTL > 0 ) {
 318                         $claimCutoff = $dbw->timestamp( $now - $this->claimTTL );
 319                         // Get the IDs of jobs that have be claimed but not finished after too long.
 320                         // These jobs can be recycled into the queue by expiring the claim. Selecting
 321                         // the IDs first means that the UPDATE can be done by primary key (less deadlocks).
 322                         $res = $dbw->select( 'job', 'job_id',
 323                                 array(
 324                                         'job_cmd' => $this->type,
 325                                         "job_token != {$dbw->addQuotes( '' )}", // was acquired
 326                                         "job_token_timestamp < {$dbw->addQuotes( $claimCutoff )}", // stale
 327                                         "job_attempts < {$dbw->addQuotes( self::MAX_ATTEMPTS )}" ), // retries left
 328                                 __METHOD__
 329                         );
 330                         $ids = array_map( function( $o ) { return $o->job_id; }, iterator_to_array( $res ) );
 331                         if ( count( $ids ) ) {
 332                                 // Reset job_token for these jobs so that other runners will pick them up.
 333                                 // Set the timestamp to the current time, as it is useful to now that the job
 334                                 // was already tried before (the timestamp becomes the "released" time).
 335                                 $dbw->update( 'job',
 336                                         array(
 337                                                 'job_token' => '',
 338                                                 'job_token_timestamp' => $dbw->timestamp( $now ) ), // time of release
 339                                         array(
 340                                                 'job_id' => $ids ),
 341                                         __METHOD__
 342                                 );
 343                                 $count += $dbw->affectedRows();
 344                         }
 345                 }
 346
 347                 // Just destroy any stale jobs...
 348                 $pruneCutoff = $dbw->timestamp( $now - self::MAX_AGE_PRUNE );
 349                 $conds = array(
 350                         'job_cmd' => $this->type,
 351                         "job_token != {$dbw->addQuotes( '' )}", // was acquired
 352                         "job_token_timestamp < {$dbw->addQuotes( $pruneCutoff )}" // stale
 353                 );
 354                 if ( $this->claimTTL > 0 ) { // only prune jobs attempted too many times...
 355                         $conds[] = "job_attempts >= {$dbw->addQuotes( self::MAX_ATTEMPTS )}";
 356                 }
 357                 // Get the IDs of jobs that are considered stale and should be removed. Selecting
 358                 // the IDs first means that the UPDATE can be done by primary key (less deadlocks).
 359                 $res = $dbw->select( 'job', 'job_id', $conds, __METHOD__ );
 360                 $ids = array_map( function( $o ) { return $o->job_id; }, iterator_to_array( $res ) );
 361                 if ( count( $ids ) ) {
 362                         $dbw->delete( 'job', array( 'job_id' => $ids ), __METHOD__ );
 363                         $count += $dbw->affectedRows();
 364                 }
 365
 366                 $dbw->unlock( "jobqueue-recycle-{$this->type}", __METHOD__ );
 367
 368                 return $count;
 369         }
 370
 371         /**
 372          * @see JobQueue::doAck()
 373          * @param Job $job
 374          * @throws MWException
 375          * @return Job|bool
 376          */
 377         protected function doAck( Job $job ) {
 378                 if ( !$job->getId() ) {
 379                         throw new MWException( "Job of type '{$job->getType()}' has no ID." );
 380                 }
 381
 382                 $dbw = $this->getMasterDB();
 383                 $dbw->commit( __METHOD__, 'flush' ); // flush existing transaction
 384
 385                 // Delete a row with a single DELETE without holding row locks over RTTs...
 386                 $dbw->delete( 'job',
 387                         array( 'job_cmd' => $this->type, 'job_id' => $job->getId() ), __METHOD__ );
 388
 389                 return true;
 390         }
 391
 392         /**
 393          * @see JobQueue::doDeduplicateRootJob()
 394          * @param Job $job
 395          * @throws MWException
 396          * @return bool
 397          */
 398         protected function doDeduplicateRootJob( Job $job ) {
 399                 $params = $job->getParams();
 400                 if ( !isset( $params['rootJobSignature'] ) ) {
 401                         throw new MWException( "Cannot register root job; missing 'rootJobSignature'." );
 402                 } elseif ( !isset( $params['rootJobTimestamp'] ) ) {
 403                         throw new MWException( "Cannot register root job; missing 'rootJobTimestamp'." );
 404                 }
 405                 $key = $this->getRootJobCacheKey( $params['rootJobSignature'] );
 406                 // Callers should call batchInsert() and then this function so that if the insert
 407                 // fails, the de-duplication registration will be aborted. Since the insert is
 408                 // deferred till "transaction idle", do the same here, so that the ordering is
 409                 // maintained. Having only the de-duplication registration succeed would cause
 410                 // jobs to become no-ops without any actual jobs that made them redundant.
 411                 $this->getMasterDB()->onTransactionIdle( function() use ( $params, $key ) {
 412                         global $wgMemc;
 413
 414                         $timestamp = $wgMemc->get( $key ); // current last timestamp of this job
 415                         if ( $timestamp && $timestamp >= $params['rootJobTimestamp'] ) {
 416                                 return true; // a newer version of this root job was enqueued
 417                         }
 418
 419                         // Update the timestamp of the last root job started at the location...
 420                         return $wgMemc->set( $key, $params['rootJobTimestamp'], 14*86400 ); // 2 weeks
 421                 } );
 422
 423                 return true;
 424         }
 425
 426         /**
 427          * Check if the "root" job of a given job has been superseded by a newer one
 428          *
 429          * @param $job Job
 430          * @return bool
 431          */
 432         protected function isRootJobOldDuplicate( Job $job ) {
 433                 global $wgMemc;
 434
 435                 $params = $job->getParams();
 436                 if ( !isset( $params['rootJobSignature'] ) ) {
 437                         return false; // job has no de-deplication info
 438                 } elseif ( !isset( $params['rootJobTimestamp'] ) ) {
 439                         trigger_error( "Cannot check root job; missing 'rootJobTimestamp'." );
 440                         return false;
 441                 }
 442
 443                 // Get the last time this root job was enqueued
 444                 $timestamp = $wgMemc->get( $this->getRootJobCacheKey( $params['rootJobSignature'] ) );
 445
 446                 // Check if a new root job was started at the location after this one's...
 447                 return ( $timestamp && $timestamp > $params['rootJobTimestamp'] );
 448         }
 449
 450         /**
 451          * @see JobQueue::doWaitForBackups()
 452          * @return void
 453          */
 454         protected function doWaitForBackups() {
 455                 wfWaitForSlaves();
 456         }
 457
 458         /**
 459          * @return DatabaseBase
 460          */
 461         protected function getSlaveDB() {
 462                 return wfGetDB( DB_SLAVE, array(), $this->wiki );
 463         }
 464
 465         /**
 466          * @return DatabaseBase
 467          */
 468         protected function getMasterDB() {
 469                 return wfGetDB( DB_MASTER, array(), $this->wiki );
 470         }
 471
 472         /**
 473          * @param $job Job
 474          * @return array
 475          */
 476         protected function insertFields( Job $job ) {
 477                 $dbw = $this->getMasterDB();
 478                 return array(
 479                         // Fields that describe the nature of the job
 480                         'job_cmd'       => $job->getType(),
 481                         'job_namespace' => $job->getTitle()->getNamespace(),
 482                         'job_title'     => $job->getTitle()->getDBkey(),
 483                         'job_params'    => self::makeBlob( $job->getParams() ),
 484                         // Additional job metadata
 485                         'job_id'        => $dbw->nextSequenceValue( 'job_job_id_seq' ),
 486                         'job_timestamp' => $dbw->timestamp(),
 487                         'job_sha1'      => wfBaseConvert(
 488                                 sha1( serialize( $job->getDeduplicationInfo() ) ),
 489                                 16, 36, 31
 490                         ),
 491                         'job_random'    => mt_rand( 0, self::MAX_JOB_RANDOM )
 492                 );
 493         }
 494
 495         /**
 496          * @return string
 497          */
 498         private function getEmptinessCacheKey() {
 499                 list( $db, $prefix ) = wfSplitWikiID( $this->wiki );
 500                 return wfForeignMemcKey( $db, $prefix, 'jobqueue', $this->type, 'isempty' );
 501         }
 502
 503         /**
 504          * @param string $signature Hash identifier of the root job
 505          * @return string
 506          */
 507         private function getRootJobCacheKey( $signature ) {
 508                 list( $db, $prefix ) = wfSplitWikiID( $this->wiki );
 509                 return wfForeignMemcKey( $db, $prefix, 'jobqueue', $this->type, 'rootjob', $signature );
 510         }
 511
 512         /**
 513          * @param $params
 514          * @return string
 515          */
 516         protected static function makeBlob( $params ) {
 517                 if ( $params !== false ) {
 518                         return serialize( $params );
 519                 } else {
 520                         return '';
 521                 }
 522         }
 523
 524         /**
 525          * @param $blob
 526          * @return bool|mixed
 527          */
 528         protected static function extractBlob( $blob ) {
 529                 if ( (string)$blob !== '' ) {
 530                         return unserialize( $blob );
 531                 } else {
 532                         return false;
 533                 }
 534         }
 535 }