X-Git-Url: https://git.heureux-cyclage.org/?a=blobdiff_plain;f=maintenance%2FrunJobs.php;h=b488fe94b22e9c9058fff8f6824cd5625d245356;hb=f856a480b9dc99e62bdeb33df5abe213ac9e0f4f;hp=429edf427b7dc0cc7023f2a4bcaf0266e4576702;hpb=59ea520aeeee211d2123a2be1468ad1147fe6733;p=lhc%2Fweb%2Fwiklou.git diff --git a/maintenance/runJobs.php b/maintenance/runJobs.php index 429edf427b..b488fe94b2 100644 --- a/maintenance/runJobs.php +++ b/maintenance/runJobs.php @@ -2,10 +2,6 @@ /** * Run pending jobs. * - * Options: - * --maxjobs (default 10000) - * --type - * * This program is free software; you can redistribute it and/or modify * it under the terms of the GNU General Public License as published by * the Free Software Foundation; either version 2 of the License, or @@ -40,19 +36,19 @@ class RunJobs extends Maintenance { $this->addOption( 'maxtime', 'Maximum amount of wall-clock time', false, true ); $this->addOption( 'type', 'Type of job to run', false, true ); $this->addOption( 'procs', 'Number of processes to use', false, true ); + $this->addOption( 'nothrottle', 'Ignore job throttling configuration', false, false ); } public function memoryLimit() { if ( $this->hasOption( 'memory-limit' ) ) { return parent::memoryLimit(); } + // Don't eat all memory on the machine if we get a bad job. return "150M"; } public function execute() { - global $wgTitle; - if ( wfReadOnly() ) { $this->error( "Unable to run jobs; the wiki is in read-only mode.", 1 ); // die } @@ -68,12 +64,12 @@ class RunJobs extends Maintenance { } } } + + $type = $this->getOption( 'type', false ); $maxJobs = $this->getOption( 'maxjobs', false ); $maxTime = $this->getOption( 'maxtime', false ); + $noThrottle = $this->hasOption( 'nothrottle' ); $startTime = time(); - $type = $this->getOption( 'type', false ); - $wgTitle = Title::newFromText( 'RunJobs.php' ); - $jobsRun = 0; // counter $group = JobQueueGroup::singleton(); // Handle any required periodic queue maintenance @@ -82,18 +78,31 @@ class RunJobs extends Maintenance { $this->runJobsLog( "Executed $count periodic queue task(s)." ); } - $flags = JobQueueGroup::USE_CACHE | JobQueueGroup::USE_PRIORITY; + $backoffs = $this->loadBackoffs(); // map of (type => UNIX expiry) + $startingBackoffs = $backoffs; // avoid unnecessary writes + $backoffExpireFunc = function ( $t ) { + return $t > time(); + }; + + $jobsRun = 0; // counter + $flags = JobQueueGroup::USE_CACHE; $lastTime = time(); // time since last slave check do { - $job = ( $type === false ) - ? $group->pop( JobQueueGroup::TYPE_DEFAULT, $flags ) - : $group->pop( $type ); // job from a single queue + $backoffs = array_filter( $backoffs, $backoffExpireFunc ); + $blacklist = $noThrottle ? array() : array_keys( $backoffs ); + if ( $type === false ) { + $job = $group->pop( JobQueueGroup::TYPE_DEFAULT, $flags, $blacklist ); + } elseif ( in_array( $type, $blacklist ) ) { + $job = false; // requested queue in backoff state + } else { + $job = $group->pop( $type ); // job from a single queue + } if ( $job ) { // found a job ++$jobsRun; $this->runJobsLog( $job->toString() . " STARTING" ); // Set timer to stop the job if too much CPU time is used - set_time_limit( $maxTime ?: 0 ); + set_time_limit( $maxTime ? : 0 ); // Run the job... wfProfileIn( __METHOD__ . '-' . get_class( $job ) ); $t = microtime( true ); @@ -101,6 +110,7 @@ class RunJobs extends Maintenance { $status = $job->run(); $error = $job->getLastError(); } catch ( MWException $e ) { + MWExceptionHandler::rollbackMasterChangesAndLog( $e ); $status = false; $error = get_class( $e ) . ': ' . $e->getMessage(); $e->report(); // write error to STDERR and the log @@ -121,6 +131,14 @@ class RunJobs extends Maintenance { $this->runJobsLog( $job->toString() . " t=$timeMs good" ); } + // Back off of certain jobs for a while + $ttw = $this->getBackoffTimeToWait( $job ); + if ( $ttw > 0 ) { + $jType = $job->getType(); + $backoffs[$jType] = isset( $backoffs[$jType] ) ? $backoffs[$jType] : 0; + $backoffs[$jType] = max( $backoffs[$jType], time() + $ttw ); + } + // Break out if we hit the job count or wall time limits... if ( $maxJobs && $jobsRun >= $maxJobs ) { break; @@ -143,10 +161,87 @@ class RunJobs extends Maintenance { $this->assertMemoryOK(); } } while ( $job ); // stop when there are no jobs + // Sync the persistent backoffs for the next runJobs.php pass + $backoffs = array_filter( $backoffs, $backoffExpireFunc ); + if ( $backoffs !== $startingBackoffs ) { + $this->syncBackoffs( $backoffs ); + } + } + + /** + * @param Job $job + * @return int Seconds for this runner to avoid doing more jobs of this type + * @see $wgJobBackoffThrottling + */ + private function getBackoffTimeToWait( Job $job ) { + global $wgJobBackoffThrottling; + + if ( !isset( $wgJobBackoffThrottling[$job->getType()] ) ) { + return 0; // not throttled + } + $itemsPerSecond = $wgJobBackoffThrottling[$job->getType()]; + if ( $itemsPerSecond <= 0 ) { + return 0; // not throttled + } + + $seconds = 0; + if ( $job->workItemCount() > 0 ) { + $seconds = floor( $job->workItemCount() / $itemsPerSecond ); + $remainder = $job->workItemCount() % $itemsPerSecond; + $seconds += ( mt_rand( 1, $itemsPerSecond ) <= $remainder ) ? 1 : 0; + } + + return (int)$seconds; + } + + /** + * Get the previous backoff expiries from persistent storage + * + * @return array Map of (job type => backoff expiry timestamp) + */ + private function loadBackoffs() { + $section = new ProfileSection( __METHOD__ ); + + $backoffs = array(); + $file = wfTempDir() . '/mw-runJobs-backoffs.json'; + if ( is_file( $file ) ) { + $handle = fopen( $file, 'rb' ); + flock( $handle, LOCK_SH ); + $content = stream_get_contents( $handle ); + flock( $handle, LOCK_UN ); + fclose( $handle ); + $backoffs = json_decode( $content, true ) ? : array(); + } + + return $backoffs; + } + + /** + * Merge the current backoff expiries from persistent storage + * + * @param array $backoffs Map of (job type => backoff expiry timestamp) + */ + private function syncBackoffs( array $backoffs ) { + $section = new ProfileSection( __METHOD__ ); + + $file = wfTempDir() . '/mw-runJobs-backoffs.json'; + $handle = fopen( $file, 'wb+' ); + flock( $handle, LOCK_EX ); + $content = stream_get_contents( $handle ); + $cBackoffs = json_decode( $content, true ) ? : array(); + foreach ( $backoffs as $type => $timestamp ) { + $cBackoffs[$type] = isset( $cBackoffs[$type] ) ? $cBackoffs[$type] : 0; + $cBackoffs[$type] = max( $cBackoffs[$type], $backoffs[$type] ); + } + ftruncate( $handle, 0 ); + fwrite( $handle, json_encode( $backoffs ) ); + flock( $handle, LOCK_UN ); + fclose( $handle ); } /** - * Make sure that this script is not too close to the memory usage limit + * Make sure that this script is not too close to the memory usage limit. + * It is better to die in between jobs than OOM right in the middle of one. * @throws MWException */ private function assertMemoryOK() { @@ -155,7 +250,7 @@ class RunJobs extends Maintenance { $m = array(); if ( preg_match( '!^(\d+)(k|m|g|)$!i', ini_get( 'memory_limit' ), $m ) ) { list( , $num, $unit ) = $m; - $conv = array( 'g' => 1024 * 1024 * 1024, 'm' => 1024 * 1024, 'k' => 1024, '' => 1 ); + $conv = array( 'g' => 1073741824, 'm' => 1048576, 'k' => 1024, '' => 1 ); $maxBytes = $num * $conv[strtolower( $unit )]; } else { $maxBytes = 0; @@ -169,7 +264,7 @@ class RunJobs extends Maintenance { /** * Log the job message - * @param $msg String The message to log + * @param string $msg The message to log */ private function runJobsLog( $msg ) { $this->output( wfTimestamp( TS_DB ) . " $msg\n" );