MediaWiki 1.42.0
JobRunner.php
Go to the documentation of this file.
1<?php
21use Liuggio\StatsdClient\Factory\StatsdDataFactoryInterface;
27use Psr\Log\LoggerInterface;
32
39class JobRunner {
40
44 public const CONSTRUCTOR_OPTIONS = [
45 MainConfigNames::JobBackoffThrottling,
46 MainConfigNames::JobClasses,
47 MainConfigNames::MaxJobDBWriteDuration,
48 MainConfigNames::TrxProfilerLimits,
49 ];
50
52 private $options;
53
55 private $lbFactory;
56
58 private $jobQueueGroup;
59
61 private $readOnlyMode;
62
64 private $linkCache;
65
67 private $stats;
68
70 private $debug;
71
73 private $logger;
74
76 private const MAX_ALLOWED_LAG = 3;
78 private const SYNC_TIMEOUT = self::MAX_ALLOWED_LAG;
80 private const LAG_CHECK_PERIOD = 1.0;
82 private const ERROR_BACKOFF_TTL = 1;
84 private const READONLY_BACKOFF_TTL = 30;
85
89 public function setDebugHandler( $debug ) {
90 $this->debug = $debug;
91 }
92
103 public function __construct(
104 ServiceOptions $serviceOptions,
105 ILBFactory $lbFactory,
106 JobQueueGroup $jobQueueGroup,
107 ReadOnlyMode $readOnlyMode,
108 LinkCache $linkCache,
109 StatsdDataFactoryInterface $statsdDataFactory,
110 LoggerInterface $logger
111 ) {
112 $serviceOptions->assertRequiredOptions( self::CONSTRUCTOR_OPTIONS );
113 $this->options = $serviceOptions;
114 $this->lbFactory = $lbFactory;
115 $this->jobQueueGroup = $jobQueueGroup;
116 $this->readOnlyMode = $readOnlyMode;
117 $this->linkCache = $linkCache;
118 $this->stats = $statsdDataFactory;
119 $this->logger = $logger;
120 }
121
147 public function run( array $options ) {
148 $type = $options['type'] ?? false;
149 $maxJobs = $options['maxJobs'] ?? false;
150 $maxTime = $options['maxTime'] ?? false;
151 $throttle = $options['throttle'] ?? true;
152
153 $jobClasses = $this->options->get( MainConfigNames::JobClasses );
154 $profilerLimits = $this->options->get( MainConfigNames::TrxProfilerLimits );
155
156 $response = [ 'jobs' => [], 'reached' => 'none-ready' ];
157
158 if ( $type !== false && !isset( $jobClasses[$type] ) ) {
159 // Invalid job type specified
160 $response['reached'] = 'none-possible';
161 return $response;
162 }
163
164 if ( $this->readOnlyMode->isReadOnly() ) {
165 // Any jobs popped off the queue might fail to run and thus might end up lost
166 $response['reached'] = 'read-only';
167 return $response;
168 }
169
170 [ , $maxLag ] = $this->lbFactory->getMainLB()->getMaxLag();
171 if ( $maxLag >= self::MAX_ALLOWED_LAG ) {
172 // DB lag is already too high; caller can immediately try other wikis if applicable
173 $response['reached'] = 'replica-lag-limit';
174 return $response;
175 }
176
177 // Narrow DB query expectations for this HTTP request
178 $this->lbFactory->getTransactionProfiler()
179 ->setExpectations( $profilerLimits['JobRunner'], __METHOD__ );
180
181 // Error out if an explicit DB transaction round is somehow active
182 if ( $this->lbFactory->hasTransactionRound() ) {
183 throw new LogicException( __METHOD__ . ' called with an active transaction round.' );
184 }
185
186 // Some jobs types should not run until a certain timestamp
187 $backoffs = []; // map of (type => UNIX expiry)
188 $backoffDeltas = []; // map of (type => seconds)
189 $wait = 'wait'; // block to read backoffs the first time
190
191 $loopStartTime = microtime( true );
192 $jobsPopped = 0;
193 $timeMsTotal = 0;
194 $lastSyncTime = 1; // initialize "last sync check timestamp" to "ages ago"
195 // Keep popping and running jobs until there are no more...
196 do {
197 // Sync the persistent backoffs with concurrent runners
198 $backoffs = $this->syncBackoffDeltas( $backoffs, $backoffDeltas, $wait );
199 $backoffKeys = $throttle ? array_keys( $backoffs ) : [];
200 $wait = 'nowait'; // less important now
201
202 if ( $type === false ) {
203 // Treat the default job type queues as a single queue and pop off a job
204 $job = $this->jobQueueGroup
205 ->pop( JobQueueGroup::TYPE_DEFAULT, JobQueueGroup::USE_CACHE, $backoffKeys );
206 } else {
207 // Pop off a job from the specified job type queue unless the execution of
208 // that type of job is currently rate-limited by the back-off list
209 $job = in_array( $type, $backoffKeys ) ? false : $this->jobQueueGroup->pop( $type );
210 }
211
212 if ( $job ) {
213 ++$jobsPopped;
214 $jType = $job->getType();
215
216 // Back off of certain jobs for a while (for throttling and for errors)
217 $ttw = $this->getBackoffTimeToWait( $job );
218 if ( $ttw > 0 ) {
219 // Always add the delta for other runners in case the time running the
220 // job negated the backoff for each individually but not collectively.
221 $backoffDeltas[$jType] = ( $backoffDeltas[$jType] ?? 0 ) + $ttw;
222 $backoffs = $this->syncBackoffDeltas( $backoffs, $backoffDeltas, $wait );
223 }
224
225 $info = $this->executeJob( $job );
226
227 // Mark completed or "one shot only" jobs as resolved
228 if ( $info['status'] !== false || !$job->allowRetries() ) {
229 $this->jobQueueGroup->ack( $job );
230 }
231
232 // Back off of certain jobs for a while (for throttling and for errors)
233 if ( $info['status'] === false && mt_rand( 0, 49 ) == 0 ) {
234 $ttw = max( $ttw, $this->getErrorBackoffTTL( $info['caught'] ) );
235 $backoffDeltas[$jType] = ( $backoffDeltas[$jType] ?? 0 ) + $ttw;
236 }
237
238 $response['jobs'][] = [
239 'type' => $jType,
240 'status' => ( $info['status'] === false ) ? 'failed' : 'ok',
241 'error' => $info['error'],
242 'time' => $info['timeMs']
243 ];
244 $timeMsTotal += $info['timeMs'];
245
246 // Break out if we hit the job count or wall time limits
247 if ( $maxJobs && $jobsPopped >= $maxJobs ) {
248 $response['reached'] = 'job-limit';
249 break;
250 } elseif ( $maxTime && ( microtime( true ) - $loopStartTime ) > $maxTime ) {
251 $response['reached'] = 'time-limit';
252 break;
253 }
254
255 // Stop if we caught a DBConnectionError. In theory it would be
256 // possible to explicitly reconnect, but the present behaviour
257 // is to just throw more exceptions every time something database-
258 // related is attempted.
259 if ( in_array( DBConnectionError::class, $info['caught'], true ) ) {
260 $response['reached'] = 'exception';
261 break;
262 }
263
264 // Don't let any of the main DB replica DBs get backed up.
265 // This only waits for so long before exiting and letting
266 // other wikis in the farm (on different masters) get a chance.
267 $timePassed = microtime( true ) - $lastSyncTime;
268 if ( $timePassed >= self::LAG_CHECK_PERIOD || $timePassed < 0 ) {
269 $opts = [ 'ifWritesSince' => $lastSyncTime, 'timeout' => self::SYNC_TIMEOUT ];
270 if ( !$this->lbFactory->waitForReplication( $opts ) ) {
271 $response['reached'] = 'replica-lag-limit';
272 break;
273 }
274 $lastSyncTime = microtime( true );
275 }
276
277 // Abort if nearing OOM to avoid erroring out in the middle of a job
278 if ( !$this->checkMemoryOK() ) {
279 $response['reached'] = 'memory-limit';
280 break;
281 }
282 }
283 } while ( $job );
284
285 // Sync the persistent backoffs for the next runJobs.php pass
286 if ( $backoffDeltas ) {
287 $this->syncBackoffDeltas( $backoffs, $backoffDeltas, 'wait' );
288 }
289
290 $response['backoffs'] = $backoffs;
291 $response['elapsed'] = $timeMsTotal;
292
293 return $response;
294 }
295
314 public function executeJob( RunnableJob $job ) {
315 $telemetry = Telemetry::getInstance();
316 $oldRequestId = $telemetry->getRequestId();
317
318 if ( $job->getRequestId() !== null ) {
319 // Temporarily inherit the original ID of the web request that spawned this job
320 $telemetry->overrideRequestId( $job->getRequestId() );
321 } else {
322 // TODO: do we need to regenerate if job doesn't have the request id?
323 // If JobRunner was called with X-Request-ID header, regeneration will generate the
324 // same value
325 $telemetry->regenerateRequestId();
326 }
327 // Use an appropriate timeout to balance lag avoidance and job progress
328 $oldTimeout = $this->lbFactory->setDefaultReplicationWaitTimeout( self::SYNC_TIMEOUT );
329 try {
330 return $this->doExecuteJob( $job );
331 } finally {
332 $this->lbFactory->setDefaultReplicationWaitTimeout( $oldTimeout );
333 $telemetry->overrideRequestId( $oldRequestId );
334 }
335 }
336
345 private function doExecuteJob( RunnableJob $job ) {
346 $jType = $job->getType();
347 $msg = $job->toString() . " STARTING";
348 $this->logger->debug( $msg, [ 'job_type' => $job->getType() ] );
349 $this->debugCallback( $msg );
350
351 // Clear out title cache data from prior snapshots
352 // (e.g. from before JobRunner was invoked in this process)
353 $this->linkCache->clear();
354
355 // Run the job...
356 $caught = [];
357 $rssStart = $this->getMaxRssKb();
358 $jobStartTime = microtime( true );
359 try {
360 $fnameTrxOwner = get_class( $job ) . '::run'; // give run() outer scope
361 // Flush any pending changes left over from an implicit transaction round
362 if ( $job->hasExecutionFlag( $job::JOB_NO_EXPLICIT_TRX_ROUND ) ) {
363 $this->lbFactory->commitPrimaryChanges( $fnameTrxOwner ); // new implicit round
364 } else {
365 $this->lbFactory->beginPrimaryChanges( $fnameTrxOwner ); // new explicit round
366 }
367 // Clear any stale REPEATABLE-READ snapshots from replica DB connections
368 $this->lbFactory->flushReplicaSnapshots( $fnameTrxOwner );
369 $status = $job->run();
370 $error = $job->getLastError();
371 // Commit all pending changes from this job
372 $this->lbFactory->commitPrimaryChanges(
373 $fnameTrxOwner,
374 // Abort if any transaction was too big
375 $this->options->get( MainConfigNames::MaxJobDBWriteDuration )
376 );
377 // Run any deferred update tasks; doUpdates() manages transactions itself
378 DeferredUpdates::doUpdates();
379 } catch ( Throwable $e ) {
381 $status = false;
382 $error = get_class( $e ) . ': ' . $e->getMessage() . ' in '
383 . $e->getFile() . ' on line ' . $e->getLine();
384 $caught[] = get_class( $e );
385 }
386 // Always attempt to call teardown(), even if Job throws exception
387 try {
388 $job->tearDown( $status );
389 } catch ( Throwable $e ) {
391 }
392
393 $timeMs = intval( ( microtime( true ) - $jobStartTime ) * 1000 );
394 $rssEnd = $this->getMaxRssKb();
395
396 // Record how long jobs wait before getting popped
397 $readyTs = $job->getReadyTimestamp();
398 if ( $readyTs ) {
399 $pickupDelay = max( 0, $jobStartTime - $readyTs );
400 $this->stats->timing( 'jobqueue.pickup_delay.all', 1000 * $pickupDelay );
401 $this->stats->timing( "jobqueue.pickup_delay.$jType", 1000 * $pickupDelay );
402 }
403 // Record root job age for jobs being run
404 $rootTimestamp = $job->getRootJobParams()['rootJobTimestamp'];
405 if ( $rootTimestamp ) {
406 $age = max( 0, $jobStartTime - (int)wfTimestamp( TS_UNIX, $rootTimestamp ) );
407 $this->stats->timing( "jobqueue.pickup_root_age.$jType", 1000 * $age );
408 }
409 // Track the execution time for jobs
410 $this->stats->timing( "jobqueue.run.$jType", $timeMs );
411 // Track RSS increases for jobs (in case of memory leaks)
412 if ( $rssStart && $rssEnd ) {
413 $this->stats->updateCount( "jobqueue.rss_delta.$jType", $rssEnd - $rssStart );
414 }
415
416 if ( $status === false ) {
417 $msg = $job->toString() . " t={job_duration} error={job_error}";
418 $this->logger->error( $msg, [
419 'job_type' => $job->getType(),
420 'job_duration' => $timeMs,
421 'job_error' => $error,
422 ] );
423
424 $msg = $job->toString() . " t=$timeMs error={$error}";
425 $this->debugCallback( $msg );
426 } else {
427 $msg = $job->toString() . " t={job_duration} good";
428 $this->logger->info( $msg, [
429 'job_type' => $job->getType(),
430 'job_duration' => $timeMs,
431 ] );
432
433 $msg = $job->toString() . " t=$timeMs good";
434 $this->debugCallback( $msg );
435 }
436
437 return [
438 'status' => $status,
439 'error' => $error,
440 'caught' => $caught,
441 'timeMs' => $timeMs
442 ];
443 }
444
449 private function getErrorBackoffTTL( array $caught ) {
450 return in_array( DBReadOnlyError::class, $caught )
451 ? self::READONLY_BACKOFF_TTL
452 : self::ERROR_BACKOFF_TTL;
453 }
454
458 private function getMaxRssKb() {
459 $info = getrusage( 0 /* RUSAGE_SELF */ );
460 // see https://linux.die.net/man/2/getrusage
461 return isset( $info['ru_maxrss'] ) ? (int)$info['ru_maxrss'] : null;
462 }
463
469 private function getBackoffTimeToWait( RunnableJob $job ) {
470 $throttling = $this->options->get( MainConfigNames::JobBackoffThrottling );
471
472 if ( !isset( $throttling[$job->getType()] ) || $job instanceof DuplicateJob ) {
473 return 0; // not throttled
474 }
475
476 $itemsPerSecond = $throttling[$job->getType()];
477 if ( $itemsPerSecond <= 0 ) {
478 return 0; // not throttled
479 }
480
481 $seconds = 0;
482 if ( $job->workItemCount() > 0 ) {
483 $exactSeconds = $job->workItemCount() / $itemsPerSecond;
484 // use randomized rounding
485 $seconds = floor( $exactSeconds );
486 $remainder = $exactSeconds - $seconds;
487 $seconds += ( mt_rand() / mt_getrandmax() < $remainder ) ? 1 : 0;
488 }
489
490 return (int)$seconds;
491 }
492
501 private function loadBackoffs( array $backoffs, $mode = 'wait' ) {
502 $file = wfTempDir() . '/mw-runJobs-backoffs.json';
503 if ( is_file( $file ) ) {
504 $noblock = ( $mode === 'nowait' ) ? LOCK_NB : 0;
505 $handle = fopen( $file, 'rb' );
506 if ( !flock( $handle, LOCK_SH | $noblock ) ) {
507 fclose( $handle );
508 return $backoffs; // don't wait on lock
509 }
510 $content = stream_get_contents( $handle );
511 flock( $handle, LOCK_UN );
512 fclose( $handle );
513 $ctime = microtime( true );
514 $cBackoffs = json_decode( $content, true ) ?: [];
515 foreach ( $cBackoffs as $type => $timestamp ) {
516 if ( $timestamp < $ctime ) {
517 unset( $cBackoffs[$type] );
518 }
519 }
520 } else {
521 $cBackoffs = [];
522 }
523
524 return $cBackoffs;
525 }
526
538 private function syncBackoffDeltas( array $backoffs, array &$deltas, $mode = 'wait' ) {
539 if ( !$deltas ) {
540 return $this->loadBackoffs( $backoffs, $mode );
541 }
542
543 $noblock = ( $mode === 'nowait' ) ? LOCK_NB : 0;
544 $file = wfTempDir() . '/mw-runJobs-backoffs.json';
545 $handle = fopen( $file, 'wb+' );
546 if ( !flock( $handle, LOCK_EX | $noblock ) ) {
547 fclose( $handle );
548 return $backoffs; // don't wait on lock
549 }
550 $ctime = microtime( true );
551 $content = stream_get_contents( $handle );
552 $cBackoffs = json_decode( $content, true ) ?: [];
553 foreach ( $deltas as $type => $seconds ) {
554 $cBackoffs[$type] = isset( $cBackoffs[$type] ) && $cBackoffs[$type] >= $ctime
555 ? $cBackoffs[$type] + $seconds
556 : $ctime + $seconds;
557 }
558 foreach ( $cBackoffs as $type => $timestamp ) {
559 if ( $timestamp < $ctime ) {
560 unset( $cBackoffs[$type] );
561 }
562 }
563 ftruncate( $handle, 0 );
564 fwrite( $handle, json_encode( $cBackoffs ) );
565 flock( $handle, LOCK_UN );
566 fclose( $handle );
567
568 $deltas = [];
569
570 return $cBackoffs;
571 }
572
578 private function checkMemoryOK() {
579 static $maxBytes = null;
580 if ( $maxBytes === null ) {
581 $m = [];
582 if ( preg_match( '!^(\d+)(k|m|g|)$!i', ini_get( 'memory_limit' ), $m ) ) {
583 [ , $num, $unit ] = $m;
584 $conv = [ 'g' => 1073741824, 'm' => 1048576, 'k' => 1024, '' => 1 ];
585 $maxBytes = (int)$num * $conv[strtolower( $unit )];
586 } else {
587 $maxBytes = 0;
588 }
589 }
590 $usedBytes = memory_get_usage();
591 if ( $maxBytes && $usedBytes >= 0.95 * $maxBytes ) {
592 $msg = "Detected excessive memory usage ({used_bytes}/{max_bytes}).";
593 $this->logger->error( $msg, [
594 'used_bytes' => $usedBytes,
595 'max_bytes' => $maxBytes,
596 ] );
597
598 $msg = "Detected excessive memory usage ($usedBytes/$maxBytes).";
599 $this->debugCallback( $msg );
600
601 return false;
602 }
603
604 return true;
605 }
606
611 private function debugCallback( $msg ) {
612 if ( $this->debug ) {
613 call_user_func_array( $this->debug, [ wfTimestamp( TS_DB ) . " $msg\n" ] );
614 }
615 }
616}
wfTempDir()
Tries to get the system directory for temporary files.
wfTimestamp( $outputtype=TS_UNIX, $ts=0)
Get a timestamp string in one of various formats.
No-op job that does nothing.
Handle enqueueing of background jobs.
Job queue runner utility methods.
Definition JobRunner.php:39
setDebugHandler( $debug)
Definition JobRunner.php:89
run(array $options)
Run jobs of the specified number/type for the specified time.
const CONSTRUCTOR_OPTIONS
Definition JobRunner.php:44
executeJob(RunnableJob $job)
Run a specific job in a manner appropriate for mass use by job dispatchers.
__construct(ServiceOptions $serviceOptions, ILBFactory $lbFactory, JobQueueGroup $jobQueueGroup, ReadOnlyMode $readOnlyMode, LinkCache $linkCache, StatsdDataFactoryInterface $statsdDataFactory, LoggerInterface $logger)
static rollbackPrimaryChangesAndLog(Throwable $e, $catcher=self::CAUGHT_BY_OTHER)
Roll back any open database transactions and log the stack trace of the throwable.
static logException(Throwable $e, $catcher=self::CAUGHT_BY_OTHER, $extraData=[])
Log a throwable to the exception log (if enabled).
Cache for article titles (prefixed DB keys) and ids linked from one source.
Definition LinkCache.php:53
A class for passing options to services.
assertRequiredOptions(array $expectedKeys)
Assert that the list of options provided in this instance exactly match $expectedKeys,...
Defer callable updates to run later in the PHP process.
Service for handling telemetry data.
Definition Telemetry.php:29
A class containing constants representing the names of configuration variables.
Determine whether a site is currently in read-only mode.
Job that has a run() method and metadata accessors for JobQueue::pop() and JobQueue::ack().
Manager of ILoadBalancer objects and, indirectly, IDatabase connections.
if(count( $args)< 1) $job