MediaWiki
REL1_22
|
00001 <?php 00028 require_once __DIR__ . '/Maintenance.php'; 00029 00035 class RunJobs extends Maintenance { 00036 public function __construct() { 00037 parent::__construct(); 00038 $this->mDescription = "Run pending jobs"; 00039 $this->addOption( 'maxjobs', 'Maximum number of jobs to run', false, true ); 00040 $this->addOption( 'maxtime', 'Maximum amount of wall-clock time', false, true ); 00041 $this->addOption( 'type', 'Type of job to run', false, true ); 00042 $this->addOption( 'procs', 'Number of processes to use', false, true ); 00043 } 00044 00045 public function memoryLimit() { 00046 if ( $this->hasOption( 'memory-limit' ) ) { 00047 return parent::memoryLimit(); 00048 } 00049 // Don't eat all memory on the machine if we get a bad job. 00050 return "150M"; 00051 } 00052 00053 public function execute() { 00054 global $wgTitle; 00055 00056 if ( wfReadOnly() ) { 00057 $this->error( "Unable to run jobs; the wiki is in read-only mode.", 1 ); // die 00058 } 00059 00060 if ( $this->hasOption( 'procs' ) ) { 00061 $procs = intval( $this->getOption( 'procs' ) ); 00062 if ( $procs < 1 || $procs > 1000 ) { 00063 $this->error( "Invalid argument to --procs", true ); 00064 } elseif ( $procs != 1 ) { 00065 $fc = new ForkController( $procs ); 00066 if ( $fc->start() != 'child' ) { 00067 exit( 0 ); 00068 } 00069 } 00070 } 00071 $maxJobs = $this->getOption( 'maxjobs', false ); 00072 $maxTime = $this->getOption( 'maxtime', false ); 00073 $startTime = time(); 00074 $type = $this->getOption( 'type', false ); 00075 $wgTitle = Title::newFromText( 'RunJobs.php' ); 00076 $jobsRun = 0; // counter 00077 00078 $group = JobQueueGroup::singleton(); 00079 // Handle any required periodic queue maintenance 00080 $count = $group->executeReadyPeriodicTasks(); 00081 if ( $count > 0 ) { 00082 $this->runJobsLog( "Executed $count periodic queue task(s)." ); 00083 } 00084 00085 $flags = JobQueueGroup::USE_CACHE | JobQueueGroup::USE_PRIORITY; 00086 $lastTime = time(); // time since last slave check 00087 do { 00088 $job = ( $type === false ) 00089 ? $group->pop( JobQueueGroup::TYPE_DEFAULT, $flags ) 00090 : $group->pop( $type ); // job from a single queue 00091 if ( $job ) { // found a job 00092 ++$jobsRun; 00093 $this->runJobsLog( $job->toString() . " STARTING" ); 00094 00095 // Set timer to stop the job if too much CPU time is used 00096 set_time_limit( $maxTime ?: 0 ); 00097 // Run the job... 00098 wfProfileIn( __METHOD__ . '-' . get_class( $job ) ); 00099 $t = microtime( true ); 00100 try { 00101 $status = $job->run(); 00102 $error = $job->getLastError(); 00103 } catch ( MWException $e ) { 00104 $status = false; 00105 $error = get_class( $e ) . ': ' . $e->getMessage(); 00106 $e->report(); // write error to STDERR and the log 00107 } 00108 $timeMs = intval( ( microtime( true ) - $t ) * 1000 ); 00109 wfProfileOut( __METHOD__ . '-' . get_class( $job ) ); 00110 // Disable the timer 00111 set_time_limit( 0 ); 00112 00113 // Mark the job as done on success or when the job cannot be retried 00114 if ( $status !== false || !$job->allowRetries() ) { 00115 $group->ack( $job ); // done 00116 } 00117 00118 if ( $status === false ) { 00119 $this->runJobsLog( $job->toString() . " t=$timeMs error={$error}" ); 00120 } else { 00121 $this->runJobsLog( $job->toString() . " t=$timeMs good" ); 00122 } 00123 00124 // Break out if we hit the job count or wall time limits... 00125 if ( $maxJobs && $jobsRun >= $maxJobs ) { 00126 break; 00127 } elseif ( $maxTime && ( time() - $startTime ) > $maxTime ) { 00128 break; 00129 } 00130 00131 // Don't let any of the main DB slaves get backed up 00132 $timePassed = time() - $lastTime; 00133 if ( $timePassed >= 5 || $timePassed < 0 ) { 00134 wfWaitForSlaves(); 00135 $lastTime = time(); 00136 } 00137 // Don't let any queue slaves/backups fall behind 00138 if ( $jobsRun > 0 && ( $jobsRun % 100 ) == 0 ) { 00139 $group->waitForBackups(); 00140 } 00141 00142 // Bail if near-OOM instead of in a job 00143 $this->assertMemoryOK(); 00144 } 00145 } while ( $job ); // stop when there are no jobs 00146 } 00147 00152 private function assertMemoryOK() { 00153 static $maxBytes = null; 00154 if ( $maxBytes === null ) { 00155 $m = array(); 00156 if ( preg_match( '!^(\d+)(k|m|g|)$!i', ini_get( 'memory_limit' ), $m ) ) { 00157 list( , $num, $unit ) = $m; 00158 $conv = array( 'g' => 1024 * 1024 * 1024, 'm' => 1024 * 1024, 'k' => 1024, '' => 1 ); 00159 $maxBytes = $num * $conv[strtolower( $unit )]; 00160 } else { 00161 $maxBytes = 0; 00162 } 00163 } 00164 $usedBytes = memory_get_usage(); 00165 if ( $maxBytes && $usedBytes >= 0.95 * $maxBytes ) { 00166 throw new MWException( "Detected excessive memory usage ($usedBytes/$maxBytes)." ); 00167 } 00168 } 00169 00174 private function runJobsLog( $msg ) { 00175 $this->output( wfTimestamp( TS_DB ) . " $msg\n" ); 00176 wfDebugLog( 'runJobs', $msg ); 00177 } 00178 } 00179 00180 $maintClass = "RunJobs"; 00181 require_once RUN_MAINTENANCE_IF_MAIN;