includes/JobQueue.php

   1 <?php
   2
   3 if ( !defined( 'MEDIAWIKI' ) ) {
   4         die( "This file is part of MediaWiki, it is not a valid entry point\n" );
   5 }
   6
   7 class Job {
   8         var $command,
   9                 $title,
  10                 $params,
  11                 $id,
  12                 $removeDuplicates,
  13                 $error;
  14
  15         /*-------------------------------------------------------------------------
  16          * Static functions
  17          *------------------------------------------------------------------------*/
  18         /**
  19          * Add an array of refreshLinks jobs to the queue
  20          * @param array $titles Array of title objects.
  21          * @static
  22          */
  23         function queueLinksJobs( $titles ) {
  24                 $fname = 'Job::queueLinksJobs';
  25                 wfProfileIn( $fname );
  26                 foreach ( $titles as $title ) {
  27                         $job = new Job( 'refreshLinks', $title );
  28                         $job->insert();
  29                 }
  30                 wfProfileOut( $fname );
  31         }
  32
  33         /**
  34          * Pop a job off the front of the queue
  35          * @static
  36          * @return Job or false if there's no jobs
  37          */
  38         function pop() {
  39                 $fname = 'Job::pop';
  40                 wfProfileIn( $fname );
  41
  42                 $dbr =& wfGetDB( DB_SLAVE );
  43
  44                 // Get a job from the slave
  45                 $row = $dbr->selectRow( 'job', '*', '', $fname,
  46                         array( 'ORDER BY' => 'job_id', 'LIMIT' => 1 )
  47                 );
  48
  49                 if ( $row === false ) {
  50                         wfProfileOut( $fname );
  51                         return false;
  52                 }
  53
  54                 // Try to delete it from the master
  55                 $dbw =& wfGetDB( DB_MASTER );
  56                 $dbw->delete( 'job', array( 'job_id' => $row->job_id ), $fname );
  57                 $affected = $dbw->affectedRows();
  58                 $dbw->immediateCommit();
  59
  60                 if ( !$affected ) {
  61                         // Failed, someone else beat us to it
  62                         // Try getting a random row
  63                         $row = $dbw->selectRow( 'job', array( 'MIN(job_id) as minjob',
  64                                 'MAX(job_id) as maxjob' ), '', $fname );
  65                         if ( $row === false || is_null( $row->minjob ) || is_null( $row->maxjob ) ) {
  66                                 // No jobs to get
  67                                 wfProfileOut( $fname );
  68                                 return false;
  69                         }
  70                         // Get the random row
  71                         $row = $dbw->selectRow( 'job', '*',
  72                                 array( 'job_id' => mt_rand( $row->minjob, $row->maxjob ) ),     $fname );
  73                         if ( $row === false ) {
  74                                 // Random job gone before we got the chance to select it
  75                                 // Give up
  76                                 wfProfileOut( $fname );
  77                                 return false;
  78                         }
  79                         // Delete the random row
  80                         $dbw->delete( 'job', array( 'job_id' => $row->job_id ), $fname );
  81                         $affected = $dbw->affectedRows();
  82                         $dbw->immediateCommit();
  83
  84                         if ( !$affected ) {
  85                                 // Random job gone before we exclusively deleted it
  86                                 // Give up
  87                                 wfProfileOut( $fname );
  88                                 return false;
  89                         }
  90                 }
  91
  92                 // If execution got to here, there's a row in $row that has been deleted from the database
  93                 // by this thread. Hence the concurrent pop was successful.
  94                 $namespace = $row->job_namespace;
  95                 $dbkey = $row->job_title;
  96                 $title = Title::makeTitleSafe( $namespace, $dbkey );
  97                 $job = new Job( $row->job_cmd, $title, $row->job_params, $row->job_id );
  98                 wfProfileOut( $fname );
  99                 return $job;
 100         }
 101
 102         /*-------------------------------------------------------------------------
 103          * Non-static functions
 104          *------------------------------------------------------------------------*/
 105
 106         function Job( $command, $title, $params = '', $id = 0 ) {
 107                 $this->command = $command;
 108                 $this->title = $title;
 109                 $this->params = $params;
 110                 $this->id = $id;
 111
 112                 // A bit of premature generalisation
 113                 // Oh well, the whole class is premature generalisation really
 114                 $this->removeDuplicates = true;
 115         }
 116
 117         function insert() {
 118                 $fname = 'Job::insert';
 119
 120                 $fields = array(
 121                         'job_cmd' => $this->command,
 122                         'job_namespace' => $this->title->getNamespace(),
 123                         'job_title' => $this->title->getDBkey(),
 124                         'job_params' => $this->params
 125                 );
 126
 127                 $dbw =& wfGetDB( DB_MASTER );
 128
 129                 if ( $this->removeDuplicates ) {
 130                         $res = $dbw->select( 'job', array( '1' ), $fields, $fname );
 131                         if ( $dbw->numRows( $res ) ) {
 132                                 return;
 133                         }
 134                 }
 135                 $fields['job_id'] = $dbw->nextSequenceValue( 'job_job_id_seq' );
 136                 $dbw->insert( 'job', $fields, $fname );
 137         }
 138
 139         /**
 140          * Run the job
 141          * @return boolean success
 142          */
 143         function run() {
 144                 $fname = 'Job::run';
 145                 wfProfileIn( $fname );
 146                 switch ( $this->command ) {
 147                         case 'refreshLinks':
 148                                 $retval = $this->refreshLinks();
 149                                 break;
 150                         default:
 151                                 $this->error = "Invalid job type {$this->command}, ignoring";
 152                                 wfDebug( $this->error . "\n" );
 153                                 $retval = false;
 154                 }
 155                 wfProfileOut( $fname );
 156                 return $retval;
 157         }
 158
 159         /**
 160          * Run a refreshLinks job
 161          * @return boolean success
 162          */
 163         function refreshLinks() {
 164                 global $wgParser;
 165                 $fname = 'Job::refreshLinks';
 166                 wfProfileIn( $fname );
 167
 168                 $dbw =& wfGetDB( DB_MASTER );
 169
 170                 $linkCache =& LinkCache::singleton();
 171                 $linkCache->clear();
 172
 173                 if ( is_null( $this->title ) ) {
 174                         $this->error = "refreshLinks: Invalid title";
 175                         wfProfileOut( $fname );
 176                         return false;
 177                 }
 178
 179                 $revision = Revision::newFromTitle( $this->title );
 180                 if ( !$revision ) {
 181                         $this->error = 'refreshLinks: Article not found "' . $this->title->getPrefixedDBkey() . '"';
 182                         wfProfileOut( $fname );
 183                         return false;
 184                 }
 185
 186                 wfProfileIn( "$fname-parse" );
 187                 $options = new ParserOptions;
 188                 $parserOutput = $wgParser->parse( $revision->getText(), $this->title, $options, true, true, $revision->getId() );
 189                 wfProfileOut( "$fname-parse" );
 190                 wfProfileIn( "$fname-update" );
 191                 $update = new LinksUpdate( $this->title, $parserOutput, false );
 192                 $update->doUpdate();
 193                 wfProfileOut( "$fname-update" );
 194                 wfProfileOut( $fname );
 195                 return true;
 196         }
 197
 198         function toString() {
 199                 if ( is_object( $this->title ) ) {
 200                         $s = "{$this->command} " . $this->title->getPrefixedDBkey();
 201                         if ( $this->params !== '' ) {
 202                                 $s .= ', ' . $this->params;
 203                         }
 204                         return $s;
 205                 } else {
 206                         return "{$this->command} {$this->params}";
 207                 }
 208         }
 209
 210         function getLastError() {
 211                 return $this->error;
 212         }
 213 }
 214 ?>