24use Liuggio\StatsdClient\Factory\StatsdDataFactoryInterface;
28use Psr\Log\LoggerAwareInterface;
29use Psr\Log\LoggerInterface;
33use Wikimedia\ScopedCallback;
44 'JobBackoffThrottling',
46 'JobSerialCommitThreshold',
47 'MaxJobDBWriteDuration',
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;
98 public function setLogger( LoggerInterface $logger ) {
100 $this->logger = $logger;
115 $serviceOptions =
null,
120 StatsdDataFactoryInterface $statsdDataFactory =
null,
121 LoggerInterface $logger =
null
123 if ( !$serviceOptions || $serviceOptions instanceof LoggerInterface ) {
127 static::CONSTRUCTOR_OPTIONS,
128 MediaWikiServices::getInstance()->getMainConfig()
132 $this->options = $serviceOptions;
133 $this->lbFactory =
$lbFactory ?? MediaWikiServices::getInstance()->getDBLoadBalancerFactory();
134 $this->jobQueueGroup =
$jobQueueGroup ?? JobQueueGroup::singleton();
135 $this->readOnlyMode =
$readOnlyMode ?: MediaWikiServices::getInstance()->getReadOnlyMode();
136 $this->linkCache =
$linkCache ?? MediaWikiServices::getInstance()->getLinkCache();
137 $this->stats = $statsdDataFactory ?? MediaWikiServices::getInstance()->getStatsdDataFactory();
138 $this->logger =
$logger ?? LoggerFactory::getInstance(
'runJobs' );
166 public function run( array $options ) {
167 $type = $options[
'type'] ??
false;
168 $maxJobs = $options[
'maxJobs'] ??
false;
169 $maxTime = $options[
'maxTime'] ??
false;
170 $throttle = $options[
'throttle'] ??
true;
172 $jobClasses = $this->options->get(
'JobClasses' );
173 $profilerLimits = $this->options->get(
'TrxProfilerLimits' );
175 $response = [
'jobs' => [],
'reached' =>
'none-ready' ];
177 if (
$type !==
false && !isset( $jobClasses[
$type] ) ) {
179 $response[
'reached'] =
'none-possible';
183 if ( $this->readOnlyMode->isReadOnly() ) {
185 $response[
'reached'] =
'read-only';
189 list( , $maxLag ) = $this->lbFactory->getMainLB()->getMaxLag();
190 if ( $maxLag >= self::MAX_ALLOWED_LAG ) {
192 $response[
'reached'] =
'replica-lag-limit';
197 $this->lbFactory->getTransactionProfiler()
198 ->setExpectations( $profilerLimits[
'JobRunner'], __METHOD__ );
201 if ( $this->lbFactory->hasTransactionRound() ) {
202 throw new LogicException( __METHOD__ .
' called with an active transaction round.' );
210 $loopStartTime = microtime(
true );
218 $blacklist = $throttle ? array_keys( $backoffs ) : [];
221 if (
$type ===
false ) {
223 $job = $this->jobQueueGroup
224 ->pop( JobQueueGroup::TYPE_DEFAULT, JobQueueGroup::USE_CACHE, $blacklist );
228 $job = in_array(
$type, $blacklist ) ? false : $this->jobQueueGroup->pop(
$type );
233 $jType =
$job->getType();
240 $backoffDeltas[$jType] = ( $backoffDeltas[$jType] ?? 0 ) + $ttw;
247 if ( $info[
'status'] !==
false || !
$job->allowRetries() ) {
248 $this->jobQueueGroup->ack(
$job );
252 if ( $info[
'status'] ===
false && mt_rand( 0, 49 ) == 0 ) {
254 $backoffDeltas[$jType] = isset( $backoffDeltas[$jType] )
255 ? $backoffDeltas[$jType] + $ttw
259 $response[
'jobs'][] = [
261 'status' => ( $info[
'status'] === false ) ?
'failed' :
'ok',
262 'error' => $info[
'error'],
263 'time' => $info[
'timeMs']
265 $timeMsTotal += $info[
'timeMs'];
268 if ( $maxJobs && $jobsPopped >= $maxJobs ) {
269 $response[
'reached'] =
'job-limit';
271 } elseif ( $maxTime && ( microtime(
true ) - $loopStartTime ) > $maxTime ) {
272 $response[
'reached'] =
'time-limit';
279 $timePassed = microtime(
true ) - $lastSyncTime;
280 if ( $timePassed >= self::LAG_CHECK_PERIOD || $timePassed < 0 ) {
281 $opts = [
'ifWritesSince' => $lastSyncTime,
'timeout' => self::SYNC_TIMEOUT ];
282 if ( !$this->lbFactory->waitForReplication( $opts ) ) {
283 $response[
'reached'] =
'replica-lag-limit';
286 $lastSyncTime = microtime(
true );
291 $response[
'reached'] =
'memory-limit';
298 if ( $backoffDeltas ) {
302 $response[
'backoffs'] = $backoffs;
303 $response[
'elapsed'] = $timeMsTotal;
327 $oldRequestId = WebRequest::getRequestId();
329 WebRequest::overrideRequestId(
$job->getRequestId() );
331 $oldTimeout = $this->lbFactory->setDefaultReplicationWaitTimeout( self::SYNC_TIMEOUT );
335 $this->lbFactory->setDefaultReplicationWaitTimeout( $oldTimeout );
336 WebRequest::overrideRequestId( $oldRequestId );
349 $jType =
$job->getType();
350 $msg =
$job->toString() .
" STARTING";
351 $this->logger->debug( $msg, [
'job_type' =>
$job->getType() ] );
356 $this->linkCache->clear();
361 $jobStartTime = microtime(
true );
363 $fnameTrxOwner = get_class(
$job ) .
'::run';
365 if (
$job->hasExecutionFlag( $job::JOB_NO_EXPLICIT_TRX_ROUND ) ) {
366 $this->lbFactory->commitMasterChanges( $fnameTrxOwner );
368 $this->lbFactory->beginMasterChanges( $fnameTrxOwner );
371 $this->lbFactory->flushReplicaSnapshots( $fnameTrxOwner );
372 $status =
$job->run();
373 $error =
$job->getLastError();
377 DeferredUpdates::doUpdates();
378 }
catch ( Throwable $e ) {
379 MWExceptionHandler::rollbackMasterChangesAndLog( $e );
381 $error = get_class( $e ) .
': ' . $e->getMessage();
382 $caught[] = get_class( $e );
386 $job->tearDown( $status );
387 }
catch ( Throwable $e ) {
388 MWExceptionHandler::logException( $e );
391 $timeMs = intval( ( microtime(
true ) - $jobStartTime ) * 1000 );
395 $readyTs =
$job->getReadyTimestamp();
397 $pickupDelay = max( 0, $jobStartTime - $readyTs );
398 $this->stats->timing(
'jobqueue.pickup_delay.all', 1000 * $pickupDelay );
399 $this->stats->timing(
"jobqueue.pickup_delay.$jType", 1000 * $pickupDelay );
402 $rootTimestamp =
$job->getRootJobParams()[
'rootJobTimestamp'];
403 if ( $rootTimestamp ) {
404 $age = max( 0, $jobStartTime -
wfTimestamp( TS_UNIX, $rootTimestamp ) );
405 $this->stats->timing(
"jobqueue.pickup_root_age.$jType", 1000 * $age );
408 $this->stats->timing(
"jobqueue.run.$jType", $timeMs );
410 if ( $rssStart && $rssEnd ) {
411 $this->stats->updateCount(
"jobqueue.rss_delta.$jType", $rssEnd - $rssStart );
414 if ( $status ===
false ) {
415 $msg =
$job->toString() .
" t={job_duration} error={job_error}";
416 $this->logger->error( $msg, [
417 'job_type' =>
$job->getType(),
418 'job_duration' => $timeMs,
419 'job_error' => $error,
422 $msg =
$job->toString() .
" t=$timeMs error={$error}";
425 $msg =
$job->toString() .
" t={job_duration} good";
426 $this->logger->info( $msg, [
427 'job_type' =>
$job->getType(),
428 'job_duration' => $timeMs,
431 $msg =
$job->toString() .
" t=$timeMs good";
448 return in_array( DBReadOnlyError::class, $caught )
449 ? self::READONLY_BACKOFF_TTL
450 : self::ERROR_BACKOFF_TTL;
457 $info = getrusage( 0 );
459 return isset( $info[
'ru_maxrss'] ) ? (int)$info[
'ru_maxrss'] :
null;
468 $throttling = $this->options->get(
'JobBackoffThrottling' );
474 $itemsPerSecond = $throttling[
$job->getType()];
475 if ( $itemsPerSecond <= 0 ) {
480 if (
$job->workItemCount() > 0 ) {
481 $exactSeconds =
$job->workItemCount() / $itemsPerSecond;
483 $seconds = floor( $exactSeconds );
484 $remainder = $exactSeconds - $seconds;
485 $seconds += ( mt_rand() / mt_getrandmax() < $remainder ) ? 1 : 0;
488 return (
int)$seconds;
501 if ( is_file(
$file ) ) {
502 $noblock = ( $mode ===
'nowait' ) ? LOCK_NB : 0;
503 $handle = fopen(
$file,
'rb' );
504 if ( !flock( $handle, LOCK_SH | $noblock ) ) {
508 $content = stream_get_contents( $handle );
509 flock( $handle, LOCK_UN );
511 $ctime = microtime(
true );
512 $cBackoffs = json_decode(
$content,
true ) ?: [];
513 foreach ( $cBackoffs as
$type => $timestamp ) {
514 if ( $timestamp < $ctime ) {
515 unset( $cBackoffs[
$type] );
541 $noblock = ( $mode ===
'nowait' ) ? LOCK_NB : 0;
543 $handle = fopen(
$file,
'wb+' );
544 if ( !flock( $handle, LOCK_EX | $noblock ) ) {
548 $ctime = microtime(
true );
549 $content = stream_get_contents( $handle );
550 $cBackoffs = json_decode(
$content,
true ) ?: [];
551 foreach ( $deltas as
$type => $seconds ) {
552 $cBackoffs[
$type] = isset( $cBackoffs[
$type] ) && $cBackoffs[
$type] >= $ctime
553 ? $cBackoffs[
$type] + $seconds
556 foreach ( $cBackoffs as
$type => $timestamp ) {
557 if ( $timestamp < $ctime ) {
558 unset( $cBackoffs[
$type] );
561 ftruncate( $handle, 0 );
562 fwrite( $handle, json_encode( $cBackoffs ) );
563 flock( $handle, LOCK_UN );
577 static $maxBytes =
null;
578 if ( $maxBytes ===
null ) {
580 if ( preg_match(
'!^(\d+)(k|m|g|)$!i', ini_get(
'memory_limit' ), $m ) ) {
581 list( , $num, $unit ) = $m;
582 $conv = [
'g' => 1073741824,
'm' => 1048576,
'k' => 1024,
'' => 1 ];
583 $maxBytes = $num * $conv[strtolower( $unit )];
588 $usedBytes = memory_get_usage();
589 if ( $maxBytes && $usedBytes >= 0.95 * $maxBytes ) {
590 $msg =
"Detected excessive memory usage ({used_bytes}/{max_bytes}).";
591 $this->logger->error( $msg, [
592 'used_bytes' => $usedBytes,
593 'max_bytes' => $maxBytes,
596 $msg =
"Detected excessive memory usage ($usedBytes/$maxBytes).";
610 if ( $this->debug ) {
611 call_user_func_array( $this->debug, [
wfTimestamp( TS_DB ) .
" $msg\n" ] );
626 $syncThreshold = $this->options->get(
'JobSerialCommitThreshold' );
629 $lb = $this->lbFactory->getMainLB();
630 if ( $syncThreshold !==
false && $lb->hasStreamingReplicaServers() ) {
632 $dbwSerial = $lb->getAnyOpenConnection( $lb->getWriterIndex() );
634 if ( $dbwSerial && $dbwSerial->namedLocksEnqueue() ) {
635 $time = $dbwSerial->pendingWriteQueryDuration( $dbwSerial::ESTIMATE_DB_APPLY );
636 if ( $time < $syncThreshold ) {
648 $this->lbFactory->commitMasterChanges(
651 [
'maxWriteDuration' => $this->options->get(
'MaxJobDBWriteDuration' ) ]
657 $ms = intval( 1000 * $time );
659 $msg =
$job->toString() .
" COMMIT ENQUEUED [{job_commit_write_ms}ms of writes]";
660 $this->logger->info( $msg, [
661 'job_type' =>
$job->getType(),
662 'job_commit_write_ms' => $ms,
665 $msg =
$job->toString() .
" COMMIT ENQUEUED [{$ms}ms of writes]";
669 if ( !$dbwSerial->lock(
'jobrunner-serial-commit', $fnameTrxOwner, 30 ) ) {
671 throw new DBError( $dbwSerial,
"Timed out waiting on commit queue." );
673 $unlocker =
new ScopedCallback(
function () use ( $dbwSerial, $fnameTrxOwner ) {
674 $dbwSerial->unlock(
'jobrunner-serial-commit', $fnameTrxOwner );
678 $pos = $lb->getMasterPos();
680 $lb->waitForAll( $pos );
684 $this->lbFactory->commitMasterChanges(
687 [
'maxWriteDuration' => $this->options->get(
'MaxJobDBWriteDuration' ) ]
689 ScopedCallback::consume( $unlocker );
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.
wfDeprecated( $function, $version=false, $component=false, $callerOffset=2)
Logs a warning that $function is deprecated.
No-op job that does nothing.
Class to handle enqueueing of background jobs.
Job queue runner utility methods.
commitMasterChanges(RunnableJob $job, $fnameTrxOwner)
Issue a commit on all masters who are currently in a transaction and have made changes to the databas...
run(array $options)
Run jobs of the specified number/type for the specified time.
getErrorBackoffTTL(array $caught)
ReadOnlyMode $readOnlyMode
const CONSTRUCTOR_OPTIONS
callable null $debug
Debug output handler.
setLogger(LoggerInterface $logger)
executeJob(RunnableJob $job)
Run a specific job in a manner appropriate for mass use by job dispatchers.
syncBackoffDeltas(array $backoffs, array &$deltas, $mode='wait')
Merge the current backoff expiries from persistent storage.
JobQueueGroup $jobQueueGroup
StatsdDataFactoryInterface $stats
doExecuteJob(RunnableJob $job)
debugCallback( $msg)
Log the job message.
getBackoffTimeToWait(RunnableJob $job)
checkMemoryOK()
Make sure that this script is not too close to the memory usage limit.
__construct( $serviceOptions=null, ILBFactory $lbFactory=null, JobQueueGroup $jobQueueGroup=null, ReadOnlyMode $readOnlyMode=null, LinkCache $linkCache=null, StatsdDataFactoryInterface $statsdDataFactory=null, LoggerInterface $logger=null)
Calling this directly is deprecated.
loadBackoffs(array $backoffs, $mode='wait')
Get the previous backoff expiries from persistent storage On I/O or lock acquisition failure this ret...
Cache for article titles (prefixed DB keys) and ids linked from one source.
A service class for fetching the wiki's current read-only mode.
Job that has a run() method and metadata accessors for JobQueue::pop() and JobQueue::ack()
if(count( $args)< 1) $job
if(PHP_SAPI !='cli-server') if(!isset( $_SERVER['SCRIPT_FILENAME'])) $file
Item class for a filearchive table row.