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