wikimedia /
mediawiki
Complex classes like JobRunner often do a lot of different things. To break such a class down, we need to identify a cohesive component within that class. A common approach to find such a component is to look for fields/methods that share the same prefixes, or suffixes. You can also have a look at the cohesion graph to spot any un-connected, or weakly-connected components.
Once you have determined the fields that belong together, you can apply the Extract Class refactoring. If the component makes sense as a sub-class, Extract Subclass is also a candidate, and is often faster.
While breaking up the class, it is a good idea to analyze how other classes use JobRunner, and based on these observations, apply Extract Interface, too.
| 1 | <?php |
||
| 34 | class JobRunner implements LoggerAwareInterface { |
||
| 35 | /** @var callable|null Debug output handler */ |
||
| 36 | protected $debug; |
||
| 37 | |||
| 38 | /** |
||
| 39 | * @var LoggerInterface $logger |
||
| 40 | */ |
||
| 41 | protected $logger; |
||
| 42 | |||
| 43 | const MAX_ALLOWED_LAG = 3; // abort if more than this much DB lag is present |
||
| 44 | const LAG_CHECK_PERIOD = 1.0; // check slave lag this many seconds |
||
| 45 | const ERROR_BACKOFF_TTL = 1; // seconds to back off a queue due to errors |
||
| 46 | |||
| 47 | /** |
||
| 48 | * @param callable $debug Optional debug output handler |
||
| 49 | */ |
||
| 50 | public function setDebugHandler( $debug ) { |
||
| 51 | $this->debug = $debug; |
||
| 52 | } |
||
| 53 | |||
| 54 | /** |
||
| 55 | * @param LoggerInterface $logger |
||
| 56 | * @return void |
||
| 57 | */ |
||
| 58 | public function setLogger( LoggerInterface $logger ) { |
||
| 59 | $this->logger = $logger; |
||
| 60 | } |
||
| 61 | |||
| 62 | /** |
||
| 63 | * @param LoggerInterface $logger |
||
| 64 | */ |
||
| 65 | public function __construct( LoggerInterface $logger = null ) { |
||
| 66 | if ( $logger === null ) { |
||
| 67 | $logger = LoggerFactory::getInstance( 'runJobs' ); |
||
| 68 | } |
||
| 69 | $this->setLogger( $logger ); |
||
| 70 | } |
||
| 71 | |||
| 72 | /** |
||
| 73 | * Run jobs of the specified number/type for the specified time |
||
| 74 | * |
||
| 75 | * The response map has a 'job' field that lists status of each job, including: |
||
| 76 | * - type : the job type |
||
| 77 | * - status : ok/failed |
||
| 78 | * - error : any error message string |
||
| 79 | * - time : the job run time in ms |
||
| 80 | * The response map also has: |
||
| 81 | * - backoffs : the (job type => seconds) map of backoff times |
||
| 82 | * - elapsed : the total time spent running tasks in ms |
||
| 83 | * - reached : the reason the script finished, one of (none-ready, job-limit, time-limit, |
||
| 84 | * memory-limit) |
||
| 85 | * |
||
| 86 | * This method outputs status information only if a debug handler was set. |
||
| 87 | * Any exceptions are caught and logged, but are not reported as output. |
||
| 88 | * |
||
| 89 | * @param array $options Map of parameters: |
||
| 90 | * - type : the job type (or false for the default types) |
||
| 91 | * - maxJobs : maximum number of jobs to run |
||
| 92 | * - maxTime : maximum time in seconds before stopping |
||
| 93 | * - throttle : whether to respect job backoff configuration |
||
| 94 | * @return array Summary response that can easily be JSON serialized |
||
| 95 | */ |
||
| 96 | public function run( array $options ) { |
||
| 97 | global $wgJobClasses, $wgTrxProfilerLimits; |
||
| 98 | |||
| 99 | $response = [ 'jobs' => [], 'reached' => 'none-ready' ]; |
||
| 100 | |||
| 101 | $type = isset( $options['type'] ) ? $options['type'] : false; |
||
| 102 | $maxJobs = isset( $options['maxJobs'] ) ? $options['maxJobs'] : false; |
||
| 103 | $maxTime = isset( $options['maxTime'] ) ? $options['maxTime'] : false; |
||
| 104 | $noThrottle = isset( $options['throttle'] ) && !$options['throttle']; |
||
| 105 | |||
| 106 | // Bail if job type is invalid |
||
| 107 | if ( $type !== false && !isset( $wgJobClasses[$type] ) ) { |
||
| 108 | $response['reached'] = 'none-possible'; |
||
| 109 | return $response; |
||
| 110 | } |
||
| 111 | // Bail out if DB is in read-only mode |
||
| 112 | if ( wfReadOnly() ) { |
||
| 113 | $response['reached'] = 'read-only'; |
||
| 114 | return $response; |
||
| 115 | } |
||
| 116 | // Bail out if there is too much DB lag. |
||
| 117 | // This check should not block as we want to try other wiki queues. |
||
| 118 | list( , $maxLag ) = wfGetLB( wfWikiID() )->getMaxLag(); |
||
|
|
|||
| 119 | if ( $maxLag >= self::MAX_ALLOWED_LAG ) { |
||
| 120 | $response['reached'] = 'slave-lag-limit'; |
||
| 121 | return $response; |
||
| 122 | } |
||
| 123 | |||
| 124 | // Flush any pending DB writes for sanity |
||
| 125 | wfGetLBFactory()->commitAll( __METHOD__ ); |
||
| 126 | |||
| 127 | // Catch huge single updates that lead to slave lag |
||
| 128 | $trxProfiler = Profiler::instance()->getTransactionProfiler(); |
||
| 129 | $trxProfiler->setLogger( LoggerFactory::getInstance( 'DBPerformance' ) ); |
||
| 130 | $trxProfiler->setExpectations( $wgTrxProfilerLimits['JobRunner'], __METHOD__ ); |
||
| 131 | |||
| 132 | // Some jobs types should not run until a certain timestamp |
||
| 133 | $backoffs = []; // map of (type => UNIX expiry) |
||
| 134 | $backoffDeltas = []; // map of (type => seconds) |
||
| 135 | $wait = 'wait'; // block to read backoffs the first time |
||
| 136 | |||
| 137 | $group = JobQueueGroup::singleton(); |
||
| 138 | $stats = RequestContext::getMain()->getStats(); |
||
| 139 | $jobsPopped = 0; |
||
| 140 | $timeMsTotal = 0; |
||
| 141 | $startTime = microtime( true ); // time since jobs started running |
||
| 142 | $lastCheckTime = 1; // timestamp of last slave check |
||
| 143 | do { |
||
| 144 | // Sync the persistent backoffs with concurrent runners |
||
| 145 | $backoffs = $this->syncBackoffDeltas( $backoffs, $backoffDeltas, $wait ); |
||
| 146 | $blacklist = $noThrottle ? [] : array_keys( $backoffs ); |
||
| 147 | $wait = 'nowait'; // less important now |
||
| 148 | |||
| 149 | if ( $type === false ) { |
||
| 150 | $job = $group->pop( |
||
| 151 | JobQueueGroup::TYPE_DEFAULT, |
||
| 152 | JobQueueGroup::USE_CACHE, |
||
| 153 | $blacklist |
||
| 154 | ); |
||
| 155 | } elseif ( in_array( $type, $blacklist ) ) { |
||
| 156 | $job = false; // requested queue in backoff state |
||
| 157 | } else { |
||
| 158 | $job = $group->pop( $type ); // job from a single queue |
||
| 159 | } |
||
| 160 | |||
| 161 | if ( $job ) { // found a job |
||
| 162 | ++$jobsPopped; |
||
| 163 | $popTime = time(); |
||
| 164 | $jType = $job->getType(); |
||
| 165 | |||
| 166 | WebRequest::overrideRequestId( $job->getRequestId() ); |
||
| 167 | |||
| 168 | // Back off of certain jobs for a while (for throttling and for errors) |
||
| 169 | $ttw = $this->getBackoffTimeToWait( $job ); |
||
| 170 | if ( $ttw > 0 ) { |
||
| 171 | // Always add the delta for other runners in case the time running the |
||
| 172 | // job negated the backoff for each individually but not collectively. |
||
| 173 | $backoffDeltas[$jType] = isset( $backoffDeltas[$jType] ) |
||
| 174 | ? $backoffDeltas[$jType] + $ttw |
||
| 175 | : $ttw; |
||
| 176 | $backoffs = $this->syncBackoffDeltas( $backoffs, $backoffDeltas, $wait ); |
||
| 177 | } |
||
| 178 | |||
| 179 | $info = $this->executeJob( $job, $stats, $popTime ); |
||
| 180 | if ( $info['status'] !== false || !$job->allowRetries() ) { |
||
| 181 | $group->ack( $job ); // succeeded or job cannot be retried |
||
| 182 | } |
||
| 183 | |||
| 184 | // Back off of certain jobs for a while (for throttling and for errors) |
||
| 185 | if ( $info['status'] === false && mt_rand( 0, 49 ) == 0 ) { |
||
| 186 | $ttw = max( $ttw, self::ERROR_BACKOFF_TTL ); // too many errors |
||
| 187 | $backoffDeltas[$jType] = isset( $backoffDeltas[$jType] ) |
||
| 188 | ? $backoffDeltas[$jType] + $ttw |
||
| 189 | : $ttw; |
||
| 190 | } |
||
| 191 | |||
| 192 | $response['jobs'][] = [ |
||
| 193 | 'type' => $jType, |
||
| 194 | 'status' => ( $info['status'] === false ) ? 'failed' : 'ok', |
||
| 195 | 'error' => $info['error'], |
||
| 196 | 'time' => $info['timeMs'] |
||
| 197 | ]; |
||
| 198 | $timeMsTotal += $info['timeMs']; |
||
| 199 | |||
| 200 | // Break out if we hit the job count or wall time limits... |
||
| 201 | if ( $maxJobs && $jobsPopped >= $maxJobs ) { |
||
| 202 | $response['reached'] = 'job-limit'; |
||
| 203 | break; |
||
| 204 | } elseif ( $maxTime && ( microtime( true ) - $startTime ) > $maxTime ) { |
||
| 205 | $response['reached'] = 'time-limit'; |
||
| 206 | break; |
||
| 207 | } |
||
| 208 | |||
| 209 | // Don't let any of the main DB slaves get backed up. |
||
| 210 | // This only waits for so long before exiting and letting |
||
| 211 | // other wikis in the farm (on different masters) get a chance. |
||
| 212 | $timePassed = microtime( true ) - $lastCheckTime; |
||
| 213 | if ( $timePassed >= self::LAG_CHECK_PERIOD || $timePassed < 0 ) { |
||
| 214 | try { |
||
| 215 | wfGetLBFactory()->waitForReplication( [ |
||
| 216 | 'ifWritesSince' => $lastCheckTime, |
||
| 217 | 'timeout' => self::MAX_ALLOWED_LAG |
||
| 218 | ] ); |
||
| 219 | } catch ( DBReplicationWaitError $e ) { |
||
| 220 | $response['reached'] = 'slave-lag-limit'; |
||
| 221 | break; |
||
| 222 | } |
||
| 223 | $lastCheckTime = microtime( true ); |
||
| 224 | } |
||
| 225 | // Don't let any queue slaves/backups fall behind |
||
| 226 | if ( $jobsPopped > 0 && ( $jobsPopped % 100 ) == 0 ) { |
||
| 227 | $group->waitForBackups(); |
||
| 228 | } |
||
| 229 | |||
| 230 | // Bail if near-OOM instead of in a job |
||
| 231 | if ( !$this->checkMemoryOK() ) { |
||
| 232 | $response['reached'] = 'memory-limit'; |
||
| 233 | break; |
||
| 234 | } |
||
| 235 | } |
||
| 236 | } while ( $job ); // stop when there are no jobs |
||
| 237 | |||
| 238 | // Sync the persistent backoffs for the next runJobs.php pass |
||
| 239 | if ( $backoffDeltas ) { |
||
| 240 | $this->syncBackoffDeltas( $backoffs, $backoffDeltas, 'wait' ); |
||
| 241 | } |
||
| 242 | |||
| 243 | $response['backoffs'] = $backoffs; |
||
| 244 | $response['elapsed'] = $timeMsTotal; |
||
| 245 | |||
| 246 | return $response; |
||
| 247 | } |
||
| 248 | |||
| 249 | /** |
||
| 250 | * @param Job $job |
||
| 251 | * @param BufferingStatsdDataFactory $stats |
||
| 252 | * @param float $popTime |
||
| 253 | * @return array Map of status/error/timeMs |
||
| 254 | */ |
||
| 255 | private function executeJob( Job $job, $stats, $popTime ) { |
||
| 325 | |||
| 326 | /** |
||
| 327 | * @return int|null Max memory RSS in kilobytes |
||
| 328 | */ |
||
| 329 | private function getMaxRssKb() { |
||
| 334 | |||
| 335 | /** |
||
| 336 | * @param Job $job |
||
| 337 | * @return int Seconds for this runner to avoid doing more jobs of this type |
||
| 338 | * @see $wgJobBackoffThrottling |
||
| 339 | */ |
||
| 340 | private function getBackoffTimeToWait( Job $job ) { |
||
| 365 | |||
| 366 | /** |
||
| 367 | * Get the previous backoff expiries from persistent storage |
||
| 368 | * On I/O or lock acquisition failure this returns the original $backoffs. |
||
| 369 | * |
||
| 370 | * @param array $backoffs Map of (job type => UNIX timestamp) |
||
| 371 | * @param string $mode Lock wait mode - "wait" or "nowait" |
||
| 372 | * @return array Map of (job type => backoff expiry timestamp) |
||
| 373 | */ |
||
| 374 | private function loadBackoffs( array $backoffs, $mode = 'wait' ) { |
||
| 399 | |||
| 400 | /** |
||
| 401 | * Merge the current backoff expiries from persistent storage |
||
| 402 | * |
||
| 403 | * The $deltas map is set to an empty array on success. |
||
| 404 | * On I/O or lock acquisition failure this returns the original $backoffs. |
||
| 405 | * |
||
| 406 | * @param array $backoffs Map of (job type => UNIX timestamp) |
||
| 407 | * @param array $deltas Map of (job type => seconds) |
||
| 408 | * @param string $mode Lock wait mode - "wait" or "nowait" |
||
| 409 | * @return array The new backoffs account for $backoffs and the latest file data |
||
| 410 | */ |
||
| 411 | private function syncBackoffDeltas( array $backoffs, array &$deltas, $mode = 'wait' ) { |
||
| 445 | |||
| 446 | /** |
||
| 447 | * Make sure that this script is not too close to the memory usage limit. |
||
| 448 | * It is better to die in between jobs than OOM right in the middle of one. |
||
| 449 | * @return bool |
||
| 450 | */ |
||
| 451 | private function checkMemoryOK() { |
||
| 474 | |||
| 475 | /** |
||
| 476 | * Log the job message |
||
| 477 | * @param string $msg The message to log |
||
| 478 | */ |
||
| 479 | private function debugCallback( $msg ) { |
||
| 484 | |||
| 485 | /** |
||
| 486 | * Issue a commit on all masters who are currently in a transaction and have |
||
| 487 | * made changes to the database. It also supports sometimes waiting for the |
||
| 488 | * local wiki's slaves to catch up. See the documentation for |
||
| 489 | * $wgJobSerialCommitThreshold for more. |
||
| 490 | * |
||
| 491 | * @param Job $job |
||
| 492 | * @throws DBError |
||
| 493 | */ |
||
| 494 | private function commitMasterChanges( Job $job ) { |
||
| 548 | } |
||
| 549 |
Loading history...
This function has been deprecated. The supplier of the file has supplied an explanatory message.
The explanatory message should give you some clue as to whether and when the function will be removed from the class and what other function to use instead.