MediaWiki  master
JobRunner.php
Go to the documentation of this file.
1 <?php
24 use Liuggio\StatsdClient\Factory\StatsdDataFactoryInterface;
27 use Psr\Log\LoggerInterface;
32 use Wikimedia\ScopedCallback;
33 
40 class JobRunner {
41 
45  public const CONSTRUCTOR_OPTIONS = [
46  MainConfigNames::JobBackoffThrottling,
47  MainConfigNames::JobClasses,
48  MainConfigNames::JobSerialCommitThreshold,
49  MainConfigNames::MaxJobDBWriteDuration,
50  MainConfigNames::TrxProfilerLimits,
51  ];
52 
54  private $options;
55 
57  private $lbFactory;
58 
60  private $jobQueueGroup;
61 
63  private $readOnlyMode;
64 
66  private $linkCache;
67 
69  private $stats;
70 
72  private $debug;
73 
75  private $logger;
76 
78  private const MAX_ALLOWED_LAG = 3;
80  private const SYNC_TIMEOUT = self::MAX_ALLOWED_LAG;
82  private const LAG_CHECK_PERIOD = 1.0;
84  private const ERROR_BACKOFF_TTL = 1;
86  private const READONLY_BACKOFF_TTL = 30;
87 
91  public function setDebugHandler( $debug ) {
92  $this->debug = $debug;
93  }
94 
105  public function __construct(
106  ServiceOptions $serviceOptions,
107  ILBFactory $lbFactory,
108  JobQueueGroup $jobQueueGroup,
109  ReadOnlyMode $readOnlyMode,
110  LinkCache $linkCache,
111  StatsdDataFactoryInterface $statsdDataFactory,
112  LoggerInterface $logger
113  ) {
114  $serviceOptions->assertRequiredOptions( self::CONSTRUCTOR_OPTIONS );
115  $this->options = $serviceOptions;
116  $this->lbFactory = $lbFactory;
117  $this->jobQueueGroup = $jobQueueGroup;
118  $this->readOnlyMode = $readOnlyMode;
119  $this->linkCache = $linkCache;
120  $this->stats = $statsdDataFactory;
121  $this->logger = $logger;
122  }
123 
149  public function run( array $options ) {
150  $type = $options['type'] ?? false;
151  $maxJobs = $options['maxJobs'] ?? false;
152  $maxTime = $options['maxTime'] ?? false;
153  $throttle = $options['throttle'] ?? true;
154 
155  $jobClasses = $this->options->get( MainConfigNames::JobClasses );
156  $profilerLimits = $this->options->get( MainConfigNames::TrxProfilerLimits );
157 
158  $response = [ 'jobs' => [], 'reached' => 'none-ready' ];
159 
160  if ( $type !== false && !isset( $jobClasses[$type] ) ) {
161  // Invalid job type specified
162  $response['reached'] = 'none-possible';
163  return $response;
164  }
165 
166  if ( $this->readOnlyMode->isReadOnly() ) {
167  // Any jobs popped off the queue might fail to run and thus might end up lost
168  $response['reached'] = 'read-only';
169  return $response;
170  }
171 
172  [ , $maxLag ] = $this->lbFactory->getMainLB()->getMaxLag();
173  if ( $maxLag >= self::MAX_ALLOWED_LAG ) {
174  // DB lag is already too high; caller can immediately try other wikis if applicable
175  $response['reached'] = 'replica-lag-limit';
176  return $response;
177  }
178 
179  // Narrow DB query expectations for this HTTP request
180  $this->lbFactory->getTransactionProfiler()
181  ->setExpectations( $profilerLimits['JobRunner'], __METHOD__ );
182 
183  // Error out if an explicit DB transaction round is somehow active
184  if ( $this->lbFactory->hasTransactionRound() ) {
185  throw new LogicException( __METHOD__ . ' called with an active transaction round.' );
186  }
187 
188  // Some jobs types should not run until a certain timestamp
189  $backoffs = []; // map of (type => UNIX expiry)
190  $backoffDeltas = []; // map of (type => seconds)
191  $wait = 'wait'; // block to read backoffs the first time
192 
193  $loopStartTime = microtime( true );
194  $jobsPopped = 0;
195  $timeMsTotal = 0;
196  $lastSyncTime = 1; // initialize "last sync check timestamp" to "ages ago"
197  // Keep popping and running jobs until there are no more...
198  do {
199  // Sync the persistent backoffs with concurrent runners
200  $backoffs = $this->syncBackoffDeltas( $backoffs, $backoffDeltas, $wait );
201  $backoffKeys = $throttle ? array_keys( $backoffs ) : [];
202  $wait = 'nowait'; // less important now
203 
204  if ( $type === false ) {
205  // Treat the default job type queues as a single queue and pop off a job
206  $job = $this->jobQueueGroup
208  } else {
209  // Pop off a job from the specified job type queue unless the execution of
210  // that type of job is currently rate-limited by the back-off list
211  $job = in_array( $type, $backoffKeys ) ? false : $this->jobQueueGroup->pop( $type );
212  }
213 
214  if ( $job ) {
215  ++$jobsPopped;
216  $jType = $job->getType();
217 
218  // Back off of certain jobs for a while (for throttling and for errors)
219  $ttw = $this->getBackoffTimeToWait( $job );
220  if ( $ttw > 0 ) {
221  // Always add the delta for other runners in case the time running the
222  // job negated the backoff for each individually but not collectively.
223  $backoffDeltas[$jType] = ( $backoffDeltas[$jType] ?? 0 ) + $ttw;
224  $backoffs = $this->syncBackoffDeltas( $backoffs, $backoffDeltas, $wait );
225  }
226 
227  $info = $this->executeJob( $job );
228 
229  // Mark completed or "one shot only" jobs as resolved
230  if ( $info['status'] !== false || !$job->allowRetries() ) {
231  $this->jobQueueGroup->ack( $job );
232  }
233 
234  // Back off of certain jobs for a while (for throttling and for errors)
235  if ( $info['status'] === false && mt_rand( 0, 49 ) == 0 ) {
236  $ttw = max( $ttw, $this->getErrorBackoffTTL( $info['caught'] ) );
237  $backoffDeltas[$jType] = ( $backoffDeltas[$jType] ?? 0 ) + $ttw;
238  }
239 
240  $response['jobs'][] = [
241  'type' => $jType,
242  'status' => ( $info['status'] === false ) ? 'failed' : 'ok',
243  'error' => $info['error'],
244  'time' => $info['timeMs']
245  ];
246  $timeMsTotal += $info['timeMs'];
247 
248  // Break out if we hit the job count or wall time limits
249  if ( $maxJobs && $jobsPopped >= $maxJobs ) {
250  $response['reached'] = 'job-limit';
251  break;
252  } elseif ( $maxTime && ( microtime( true ) - $loopStartTime ) > $maxTime ) {
253  $response['reached'] = 'time-limit';
254  break;
255  }
256 
257  // Stop if we caught a DBConnectionError. In theory it would be
258  // possible to explicitly reconnect, but the present behaviour
259  // is to just throw more exceptions every time something database-
260  // related is attempted.
261  if ( in_array( DBConnectionError::class, $info['caught'], true ) ) {
262  $response['reached'] = 'exception';
263  break;
264  }
265 
266  // Don't let any of the main DB replica DBs get backed up.
267  // This only waits for so long before exiting and letting
268  // other wikis in the farm (on different masters) get a chance.
269  $timePassed = microtime( true ) - $lastSyncTime;
270  if ( $timePassed >= self::LAG_CHECK_PERIOD || $timePassed < 0 ) {
271  $opts = [ 'ifWritesSince' => $lastSyncTime, 'timeout' => self::SYNC_TIMEOUT ];
272  if ( !$this->lbFactory->waitForReplication( $opts ) ) {
273  $response['reached'] = 'replica-lag-limit';
274  break;
275  }
276  $lastSyncTime = microtime( true );
277  }
278 
279  // Abort if nearing OOM to avoid erroring out in the middle of a job
280  if ( !$this->checkMemoryOK() ) {
281  $response['reached'] = 'memory-limit';
282  break;
283  }
284  }
285  } while ( $job );
286 
287  // Sync the persistent backoffs for the next runJobs.php pass
288  if ( $backoffDeltas ) {
289  $this->syncBackoffDeltas( $backoffs, $backoffDeltas, 'wait' );
290  }
291 
292  $response['backoffs'] = $backoffs;
293  $response['elapsed'] = $timeMsTotal;
294 
295  return $response;
296  }
297 
316  public function executeJob( RunnableJob $job ) {
317  $oldRequestId = WebRequest::getRequestId();
318  // Temporarily inherit the original ID of the web request that spawned this job
319  WebRequest::overrideRequestId( $job->getRequestId() );
320  // Use an appropriate timeout to balance lag avoidance and job progress
321  $oldTimeout = $this->lbFactory->setDefaultReplicationWaitTimeout( self::SYNC_TIMEOUT );
322  try {
323  return $this->doExecuteJob( $job );
324  } finally {
325  $this->lbFactory->setDefaultReplicationWaitTimeout( $oldTimeout );
326  WebRequest::overrideRequestId( $oldRequestId );
327  }
328  }
329 
338  private function doExecuteJob( RunnableJob $job ) {
339  $jType = $job->getType();
340  $msg = $job->toString() . " STARTING";
341  $this->logger->debug( $msg, [ 'job_type' => $job->getType() ] );
342  $this->debugCallback( $msg );
343 
344  // Clear out title cache data from prior snapshots
345  // (e.g. from before JobRunner was invoked in this process)
346  $this->linkCache->clear();
347 
348  // Run the job...
349  $caught = [];
350  $rssStart = $this->getMaxRssKb();
351  $jobStartTime = microtime( true );
352  try {
353  $fnameTrxOwner = get_class( $job ) . '::run'; // give run() outer scope
354  // Flush any pending changes left over from an implicit transaction round
355  if ( $job->hasExecutionFlag( $job::JOB_NO_EXPLICIT_TRX_ROUND ) ) {
356  $this->lbFactory->commitPrimaryChanges( $fnameTrxOwner ); // new implicit round
357  } else {
358  $this->lbFactory->beginPrimaryChanges( $fnameTrxOwner ); // new explicit round
359  }
360  // Clear any stale REPEATABLE-READ snapshots from replica DB connections
361  $this->lbFactory->flushReplicaSnapshots( $fnameTrxOwner );
362  $status = $job->run();
363  $error = $job->getLastError();
364  // Commit all pending changes from this job
365  $this->commitPrimaryChanges( $job, $fnameTrxOwner );
366  // Run any deferred update tasks; doUpdates() manages transactions itself
368  } catch ( Throwable $e ) {
370  $status = false;
371  $error = get_class( $e ) . ': ' . $e->getMessage() . ' in '
372  . $e->getFile() . ' on line ' . $e->getLine();
373  $caught[] = get_class( $e );
374  }
375  // Always attempt to call teardown(), even if Job throws exception
376  try {
377  $job->tearDown( $status );
378  } catch ( Throwable $e ) {
380  }
381 
382  $timeMs = intval( ( microtime( true ) - $jobStartTime ) * 1000 );
383  $rssEnd = $this->getMaxRssKb();
384 
385  // Record how long jobs wait before getting popped
386  $readyTs = $job->getReadyTimestamp();
387  if ( $readyTs ) {
388  $pickupDelay = max( 0, $jobStartTime - $readyTs );
389  $this->stats->timing( 'jobqueue.pickup_delay.all', 1000 * $pickupDelay );
390  $this->stats->timing( "jobqueue.pickup_delay.$jType", 1000 * $pickupDelay );
391  }
392  // Record root job age for jobs being run
393  $rootTimestamp = $job->getRootJobParams()['rootJobTimestamp'];
394  if ( $rootTimestamp ) {
395  $age = max( 0, $jobStartTime - (int)wfTimestamp( TS_UNIX, $rootTimestamp ) );
396  $this->stats->timing( "jobqueue.pickup_root_age.$jType", 1000 * $age );
397  }
398  // Track the execution time for jobs
399  $this->stats->timing( "jobqueue.run.$jType", $timeMs );
400  // Track RSS increases for jobs (in case of memory leaks)
401  if ( $rssStart && $rssEnd ) {
402  $this->stats->updateCount( "jobqueue.rss_delta.$jType", $rssEnd - $rssStart );
403  }
404 
405  if ( $status === false ) {
406  $msg = $job->toString() . " t={job_duration} error={job_error}";
407  $this->logger->error( $msg, [
408  'job_type' => $job->getType(),
409  'job_duration' => $timeMs,
410  'job_error' => $error,
411  ] );
412 
413  $msg = $job->toString() . " t=$timeMs error={$error}";
414  $this->debugCallback( $msg );
415  } else {
416  $msg = $job->toString() . " t={job_duration} good";
417  $this->logger->info( $msg, [
418  'job_type' => $job->getType(),
419  'job_duration' => $timeMs,
420  ] );
421 
422  $msg = $job->toString() . " t=$timeMs good";
423  $this->debugCallback( $msg );
424  }
425 
426  return [
427  'status' => $status,
428  'error' => $error,
429  'caught' => $caught,
430  'timeMs' => $timeMs
431  ];
432  }
433 
438  private function getErrorBackoffTTL( array $caught ) {
439  return in_array( DBReadOnlyError::class, $caught )
440  ? self::READONLY_BACKOFF_TTL
441  : self::ERROR_BACKOFF_TTL;
442  }
443 
447  private function getMaxRssKb() {
448  $info = getrusage( 0 /* RUSAGE_SELF */ );
449  // see https://linux.die.net/man/2/getrusage
450  return isset( $info['ru_maxrss'] ) ? (int)$info['ru_maxrss'] : null;
451  }
452 
458  private function getBackoffTimeToWait( RunnableJob $job ) {
459  $throttling = $this->options->get( MainConfigNames::JobBackoffThrottling );
460 
461  if ( !isset( $throttling[$job->getType()] ) || $job instanceof DuplicateJob ) {
462  return 0; // not throttled
463  }
464 
465  $itemsPerSecond = $throttling[$job->getType()];
466  if ( $itemsPerSecond <= 0 ) {
467  return 0; // not throttled
468  }
469 
470  $seconds = 0;
471  if ( $job->workItemCount() > 0 ) {
472  $exactSeconds = $job->workItemCount() / $itemsPerSecond;
473  // use randomized rounding
474  $seconds = floor( $exactSeconds );
475  $remainder = $exactSeconds - $seconds;
476  $seconds += ( mt_rand() / mt_getrandmax() < $remainder ) ? 1 : 0;
477  }
478 
479  return (int)$seconds;
480  }
481 
490  private function loadBackoffs( array $backoffs, $mode = 'wait' ) {
491  $file = wfTempDir() . '/mw-runJobs-backoffs.json';
492  if ( is_file( $file ) ) {
493  $noblock = ( $mode === 'nowait' ) ? LOCK_NB : 0;
494  $handle = fopen( $file, 'rb' );
495  if ( !flock( $handle, LOCK_SH | $noblock ) ) {
496  fclose( $handle );
497  return $backoffs; // don't wait on lock
498  }
499  $content = stream_get_contents( $handle );
500  flock( $handle, LOCK_UN );
501  fclose( $handle );
502  $ctime = microtime( true );
503  $cBackoffs = json_decode( $content, true ) ?: [];
504  foreach ( $cBackoffs as $type => $timestamp ) {
505  if ( $timestamp < $ctime ) {
506  unset( $cBackoffs[$type] );
507  }
508  }
509  } else {
510  $cBackoffs = [];
511  }
512 
513  return $cBackoffs;
514  }
515 
527  private function syncBackoffDeltas( array $backoffs, array &$deltas, $mode = 'wait' ) {
528  if ( !$deltas ) {
529  return $this->loadBackoffs( $backoffs, $mode );
530  }
531 
532  $noblock = ( $mode === 'nowait' ) ? LOCK_NB : 0;
533  $file = wfTempDir() . '/mw-runJobs-backoffs.json';
534  $handle = fopen( $file, 'wb+' );
535  if ( !flock( $handle, LOCK_EX | $noblock ) ) {
536  fclose( $handle );
537  return $backoffs; // don't wait on lock
538  }
539  $ctime = microtime( true );
540  $content = stream_get_contents( $handle );
541  $cBackoffs = json_decode( $content, true ) ?: [];
542  foreach ( $deltas as $type => $seconds ) {
543  $cBackoffs[$type] = isset( $cBackoffs[$type] ) && $cBackoffs[$type] >= $ctime
544  ? $cBackoffs[$type] + $seconds
545  : $ctime + $seconds;
546  }
547  foreach ( $cBackoffs as $type => $timestamp ) {
548  if ( $timestamp < $ctime ) {
549  unset( $cBackoffs[$type] );
550  }
551  }
552  ftruncate( $handle, 0 );
553  fwrite( $handle, json_encode( $cBackoffs ) );
554  flock( $handle, LOCK_UN );
555  fclose( $handle );
556 
557  $deltas = [];
558 
559  return $cBackoffs;
560  }
561 
567  private function checkMemoryOK() {
568  static $maxBytes = null;
569  if ( $maxBytes === null ) {
570  $m = [];
571  if ( preg_match( '!^(\d+)(k|m|g|)$!i', ini_get( 'memory_limit' ), $m ) ) {
572  [ , $num, $unit ] = $m;
573  $conv = [ 'g' => 1073741824, 'm' => 1048576, 'k' => 1024, '' => 1 ];
574  $maxBytes = (int)$num * $conv[strtolower( $unit )];
575  } else {
576  $maxBytes = 0;
577  }
578  }
579  $usedBytes = memory_get_usage();
580  if ( $maxBytes && $usedBytes >= 0.95 * $maxBytes ) {
581  $msg = "Detected excessive memory usage ({used_bytes}/{max_bytes}).";
582  $this->logger->error( $msg, [
583  'used_bytes' => $usedBytes,
584  'max_bytes' => $maxBytes,
585  ] );
586 
587  $msg = "Detected excessive memory usage ($usedBytes/$maxBytes).";
588  $this->debugCallback( $msg );
589 
590  return false;
591  }
592 
593  return true;
594  }
595 
600  private function debugCallback( $msg ) {
601  if ( $this->debug ) {
602  call_user_func_array( $this->debug, [ wfTimestamp( TS_DB ) . " $msg\n" ] );
603  }
604  }
605 
616  private function commitPrimaryChanges( RunnableJob $job, $fnameTrxOwner ) {
617  $syncThreshold = $this->options->get( MainConfigNames::JobSerialCommitThreshold );
618 
619  $time = false;
620  $lb = $this->lbFactory->getMainLB();
621  if ( $syncThreshold !== false && $lb->hasStreamingReplicaServers() ) {
622  // Generally, there is one primary connection to the local DB
623  $dbwSerial = $lb->getAnyOpenConnection( $lb->getWriterIndex() );
624  // We need natively blocking fast locks
625  if ( $dbwSerial && $dbwSerial->namedLocksEnqueue() ) {
626  $time = $dbwSerial->pendingWriteQueryDuration( $dbwSerial::ESTIMATE_DB_APPLY );
627  if ( $time < $syncThreshold ) {
628  $dbwSerial = false;
629  }
630  } else {
631  $dbwSerial = false;
632  }
633  } else {
634  // There are no replica DBs or writes are all to foreign DB (we don't handle that)
635  $dbwSerial = false;
636  }
637 
638  if ( !$dbwSerial ) {
639  $this->lbFactory->commitPrimaryChanges(
640  $fnameTrxOwner,
641  // Abort if any transaction was too big
642  $this->options->get( MainConfigNames::MaxJobDBWriteDuration )
643  );
644 
645  return;
646  }
647 
648  $ms = intval( 1000 * $time );
649 
650  $msg = $job->toString() . " COMMIT ENQUEUED [{job_commit_write_ms}ms of writes]";
651  $this->logger->info( $msg, [
652  'job_type' => $job->getType(),
653  'job_commit_write_ms' => $ms,
654  ] );
655 
656  $msg = $job->toString() . " COMMIT ENQUEUED [{$ms}ms of writes]";
657  $this->debugCallback( $msg );
658 
659  // Wait for an exclusive lock to commit
660  if ( !$dbwSerial->lock( 'jobrunner-serial-commit', $fnameTrxOwner, 30 ) ) {
661  // This will trigger a rollback in the main loop
662  throw new DBError( $dbwSerial, "Timed out waiting on commit queue." );
663  }
664  $unlocker = new ScopedCallback( static function () use ( $dbwSerial, $fnameTrxOwner ) {
665  $dbwSerial->unlock( 'jobrunner-serial-commit', $fnameTrxOwner );
666  } );
667 
668  // Wait for the replica DBs to catch up
669  $pos = $lb->getPrimaryPos();
670  if ( $pos ) {
671  $lb->waitForAll( $pos );
672  }
673 
674  // Actually commit the DB primary changes
675  $this->lbFactory->commitPrimaryChanges(
676  $fnameTrxOwner,
677  // Abort if any transaction was too big
678  $this->options->get( MainConfigNames::MaxJobDBWriteDuration )
679  );
680  ScopedCallback::consume( $unlocker );
681  }
682 }
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.
static doUpdates( $stage=self::ALL)
Consume and execute all pending updates.
No-op job that does nothing.
Class to handle enqueueing of background jobs.
Job queue runner utility methods.
Definition: JobRunner.php:40
setDebugHandler( $debug)
Definition: JobRunner.php:91
run(array $options)
Run jobs of the specified number/type for the specified time.
Definition: JobRunner.php:149
const CONSTRUCTOR_OPTIONS
Definition: JobRunner.php:45
executeJob(RunnableJob $job)
Run a specific job in a manner appropriate for mass use by job dispatchers.
Definition: JobRunner.php:316
__construct(ServiceOptions $serviceOptions, ILBFactory $lbFactory, JobQueueGroup $jobQueueGroup, ReadOnlyMode $readOnlyMode, LinkCache $linkCache, StatsdDataFactoryInterface $statsdDataFactory, LoggerInterface $logger)
Definition: JobRunner.php:105
Cache for article titles (prefixed DB keys) and ids linked from one source.
Definition: LinkCache.php:42
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).
A class for passing options to services.
assertRequiredOptions(array $expectedKeys)
Assert that the list of options provided in this instance exactly match $expectedKeys,...
A class containing constants representing the names of configuration variables.
A service class for fetching the wiki's current read-only mode.
static getRequestId()
Get the current request ID.
Definition: WebRequest.php:344
static overrideRequestId( $id)
Override the unique request ID.
Definition: WebRequest.php:366
Database error base class.
Definition: DBError.php:31
Job that has a run() method and metadata accessors for JobQueue::pop() and JobQueue::ack()
Definition: RunnableJob.php:37
Manager of ILoadBalancer objects and, indirectly, IDatabase connections.
Definition: ILBFactory.php:46
if(count( $args)< 1) $job
$content
Definition: router.php:76
if(PHP_SAPI !='cli-server') if(!isset( $_SERVER['SCRIPT_FILENAME'])) $file
Item class for a filearchive table row.
Definition: router.php:42