28use Wikimedia\ScopedCallback;
37 private const CACHE_TTL_SHORT = 30;
38 private const MAX_AGE_PRUNE = 604800;
39 private const MAX_JOB_RANDOM = 2147483647;
40 private const MAX_OFFSET = 255;
60 parent::__construct( $params );
62 if ( isset( $params[
'server'] ) ) {
63 $this->server = $params[
'server'];
64 } elseif ( isset( $params[
'cluster'] ) && is_string( $params[
'cluster'] ) ) {
65 $this->cluster = $params[
'cluster'];
70 return [
'random',
'timestamp',
'fifo' ];
84 $scope = $this->getScopedNoTrxFlag(
$dbr );
87 $found = (bool)
$dbr->selectField(
'job',
'1',
88 [
'job_cmd' => $this->type,
'job_token' =>
'' ],
103 $key = $this->getCacheKey(
'size' );
105 $size = $this->wanCache->get( $key );
106 if ( is_int( $size ) ) {
112 $scope = $this->getScopedNoTrxFlag(
$dbr );
114 $size = (int)
$dbr->selectField(
'job',
'COUNT(*)',
115 [
'job_cmd' => $this->type,
'job_token' =>
'' ],
121 $this->wanCache->set( $key, $size, self::CACHE_TTL_SHORT );
131 if ( $this->claimTTL <= 0 ) {
135 $key = $this->getCacheKey(
'acquiredcount' );
137 $count = $this->wanCache->get( $key );
138 if ( is_int( $count ) ) {
144 $scope = $this->getScopedNoTrxFlag(
$dbr );
146 $count = (int)
$dbr->selectField(
'job',
'COUNT(*)',
147 [
'job_cmd' => $this->type,
"job_token != {$dbr->addQuotes( '' )}" ],
153 $this->wanCache->set( $key, $count, self::CACHE_TTL_SHORT );
164 if ( $this->claimTTL <= 0 ) {
168 $key = $this->getCacheKey(
'abandonedcount' );
170 $count = $this->wanCache->get( $key );
171 if ( is_int( $count ) ) {
177 $scope = $this->getScopedNoTrxFlag(
$dbr );
179 $count = (int)
$dbr->selectField(
'job',
'COUNT(*)',
181 'job_cmd' => $this->type,
182 "job_token != {$dbr->addQuotes( '' )}",
183 "job_attempts >= " .
$dbr->addQuotes( $this->maxTries )
191 $this->wanCache->set( $key, $count, self::CACHE_TTL_SHORT );
206 $scope = $this->getScopedNoTrxFlag( $dbw );
217 $dbw->onTransactionPreCommitOrIdle(
218 function (
IDatabase $dbw ) use ( $jobs, $flags, $fname ) {
237 if ( $jobs === [] ) {
243 foreach ( $jobs as
$job ) {
245 if (
$job->ignoreDuplicates() ) {
246 $rowSet[$row[
'job_sha1']] = $row;
252 if ( $flags & self::QOS_ATOMIC ) {
257 if ( count( $rowSet ) ) {
261 'job_sha1' => array_map(
'strval', array_keys( $rowSet ) ),
266 foreach (
$res as $row ) {
267 wfDebug(
"Job with hash '{$row->job_sha1}' is a duplicate." );
268 unset( $rowSet[$row->job_sha1] );
272 $rows = array_merge( $rowList, array_values( $rowSet ) );
274 foreach ( array_chunk( $rows, 50 ) as $rowBatch ) {
275 $dbw->
insert(
'job', $rowBatch, $method );
277 $this->
incrStats(
'inserts', $this->type, count( $rows ) );
278 $this->
incrStats(
'dupe_inserts', $this->type,
279 count( $rowSet ) + count( $rowList ) - count( $rows )
284 if ( $flags & self::QOS_ATOMIC ) {
296 $scope = $this->getScopedNoTrxFlag( $dbw );
303 if ( in_array( $this->order, [
'fifo',
'timestamp' ] ) ) {
306 $rand = mt_rand( 0, self::MAX_JOB_RANDOM );
307 $gte = (bool)mt_rand( 0, 1 );
321 if ( !
$job || mt_rand( 0, 9 ) == 0 ) {
344 $scope = $this->getScopedNoTrxFlag( $dbw );
346 $tinyQueue = $this->wanCache->get( $this->getCacheKey(
'small' ) );
348 $invertedDirection =
false;
357 $ineq = $gte ?
'>=' :
'<=';
358 $dir = $gte ?
'ASC' :
'DESC';
359 $row = $dbw->selectRow(
'job', self::selectFields(),
361 'job_cmd' => $this->type,
363 "job_random {$ineq} {$dbw->addQuotes( $rand )}" ],
365 [
'ORDER BY' =>
"job_random {$dir}" ]
367 if ( !$row && !$invertedDirection ) {
369 $invertedDirection =
true;
376 $row = $dbw->selectRow(
'job', self::selectFields(),
378 'job_cmd' => $this->type,
382 [
'OFFSET' => mt_rand( 0, self::MAX_OFFSET ) ]
386 $this->wanCache->set( $this->getCacheKey(
'small' ), 1, 30 );
397 'job_token' => $uuid,
398 'job_token_timestamp' => $dbw->timestamp(),
399 'job_attempts = job_attempts+1' ],
400 [
'job_cmd' => $this->type,
'job_id' => $row->job_id,
'job_token' =>
'' ],
405 if ( !$dbw->affectedRows() ) {
422 $scope = $this->getScopedNoTrxFlag( $dbw );
426 if ( $dbw->getType() ===
'mysql' ) {
431 $dbw->query(
"UPDATE {$dbw->tableName( 'job' )} " .
433 "job_token = {$dbw->addQuotes( $uuid ) }, " .
434 "job_token_timestamp = {$dbw->addQuotes( $dbw->timestamp() )}, " .
435 "job_attempts = job_attempts+1 " .
437 "job_cmd = {$dbw->addQuotes( $this->type )} " .
438 "AND job_token = {$dbw->addQuotes( '' )} " .
439 ") ORDER BY job_id ASC LIMIT 1",
447 'job_token' => $uuid,
448 'job_token_timestamp' => $dbw->timestamp(),
449 'job_attempts = job_attempts+1' ],
451 $dbw->selectSQLText(
'job',
'job_id',
452 [
'job_cmd' => $this->type,
'job_token' =>
'' ],
454 [
'ORDER BY' =>
'job_id ASC',
'LIMIT' => 1 ] ) .
461 if ( !$dbw->affectedRows() ) {
466 $row = $dbw->selectRow(
'job', self::selectFields(),
467 [
'job_cmd' => $this->type,
'job_token' => $uuid ], __METHOD__
470 wfDebug(
"Row deleted as duplicate by another process." );
483 $id =
$job->getMetadata(
'id' );
484 if ( $id ===
null ) {
485 throw new MWException(
"Job of type '{$job->getType()}' has no ID." );
490 $scope = $this->getScopedNoTrxFlag( $dbw );
495 [
'job_cmd' => $this->type,
'job_id' => $id ],
519 $scope = $this->getScopedNoTrxFlag( $dbw );
520 $dbw->onTransactionCommitOrIdle(
521 function () use (
$job ) {
522 parent::doDeduplicateRootJob(
$job );
537 $scope = $this->getScopedNoTrxFlag( $dbw );
539 $dbw->delete(
'job', [
'job_cmd' => $this->type ], __METHOD__ );
552 if ( $this->server ) {
556 $lbFactory = MediaWikiServices::getInstance()->getDBLoadBalancerFactory();
557 $lbFactory->waitForReplication( [
558 'domain' => $this->domain,
559 'cluster' => is_string( $this->cluster ) ? $this->cluster : false
567 foreach ( [
'size',
'acquiredcount' ] as
$type ) {
568 $this->wanCache->delete( $this->getCacheKey(
$type ) );
596 "job_attempts >= " . intval( $this->maxTries )
607 $scope = $this->getScopedNoTrxFlag(
$dbr );
610 $dbr->select(
'job', self::selectFields(), $conds, __METHOD__ ),
621 if ( $this->server ) {
625 return is_string( $this->cluster )
626 ?
"DBCluster:{$this->cluster}:{$this->domain}"
627 :
"LBFactory:{$this->domain}";
633 $scope = $this->getScopedNoTrxFlag(
$dbr );
638 $res =
$dbr->select(
'job',
'DISTINCT job_cmd',
639 [
'job_cmd' => $types ], __METHOD__ );
642 foreach (
$res as $row ) {
643 $types[] = $row->job_cmd;
652 $scope = $this->getScopedNoTrxFlag(
$dbr );
654 $res =
$dbr->select(
'job', [
'job_cmd',
'count' =>
'COUNT(*)' ],
655 [
'job_cmd' => $types ], __METHOD__, [
'GROUP BY' =>
'job_cmd' ] );
658 foreach (
$res as $row ) {
659 $sizes[$row->job_cmd] = (int)$row->count;
675 $scope = $this->getScopedNoTrxFlag( $dbw );
678 if ( !$dbw->lock(
"jobqueue-recycle-{$this->type}", __METHOD__, 1 ) ) {
683 if ( $this->claimTTL > 0 ) {
684 $claimCutoff = $dbw->timestamp( $now - $this->claimTTL );
688 $res = $dbw->select(
'job',
'job_id',
690 'job_cmd' => $this->type,
691 "job_token != {$dbw->addQuotes( '' )}",
692 "job_token_timestamp < {$dbw->addQuotes( $claimCutoff )}",
693 "job_attempts < {$dbw->addQuotes( $this->maxTries )}" ],
697 static function ( $o ) {
699 }, iterator_to_array(
$res )
701 if ( count( $ids ) ) {
708 'job_token_timestamp' => $dbw->timestamp( $now )
710 [
'job_id' => $ids,
"job_token != {$dbw->addQuotes( '' )}" ],
713 $affected = $dbw->affectedRows();
715 $this->
incrStats(
'recycles', $this->type, $affected );
720 $pruneCutoff = $dbw->timestamp( $now - self::MAX_AGE_PRUNE );
723 "job_token != {$dbw->addQuotes( '' )}",
724 "job_token_timestamp < {$dbw->addQuotes( $pruneCutoff )}"
726 if ( $this->claimTTL > 0 ) {
727 $conds[] =
"job_attempts >= {$dbw->addQuotes( $this->maxTries )}";
731 $res = $dbw->select(
'job',
'job_id', $conds, __METHOD__ );
733 static function ( $o ) {
735 }, iterator_to_array(
$res )
737 if ( count( $ids ) ) {
738 $dbw->delete(
'job', [
'job_id' => $ids ], __METHOD__ );
739 $affected = $dbw->affectedRows();
741 $this->
incrStats(
'abandons', $this->type, $affected );
744 $dbw->unlock(
"jobqueue-recycle-{$this->type}", __METHOD__ );
760 'job_cmd' =>
$job->getType(),
762 'job_title' =>
$job->getParams()[
'title'] ??
'',
763 'job_params' => self::makeBlob(
$job->getParams() ),
766 'job_sha1' => Wikimedia\base_convert(
770 'job_random' => mt_rand( 0, self::MAX_JOB_RANDOM )
813 protected function getDB( $index ) {
814 if ( $this->server ) {
815 if ( $this->conn instanceof
IDatabase ) {
817 } elseif ( $this->conn instanceof
DBError ) {
822 $this->conn = Database::factory( $this->server[
'type'], $this->server );
830 $lbFactory = MediaWikiServices::getInstance()->getDBLoadBalancerFactory();
831 $lb = is_string( $this->cluster )
832 ? $lbFactory->getExternalLB( $this->cluster )
833 : $lbFactory->getMainLB( $this->domain );
835 if ( $lb->getServerType( $lb->getWriterIndex() ) !==
'sqlite' ) {
838 $flags = $lb::CONN_TRX_AUTOCOMMIT;
844 return $lb->getMaintenanceConnectionRef( $index, [], $this->domain, $flags );
852 private function getScopedNoTrxFlag(
IDatabase $db ) {
856 return new ScopedCallback(
static function () use ( $db, $autoTrx ) {
867 private function getCacheKey( $property ) {
868 $cluster = is_string( $this->cluster ) ? $this->cluster :
'main';
870 return $this->wanCache->makeGlobalKey(
884 if ( $params !==
false ) {
896 $params = ( (string)$row->job_params !==
'' ) ?
unserialize( $row->job_params ) : [];
897 if ( !is_array( $params ) ) {
898 throw new UnexpectedValueException(
899 "Could not unserialize job with ID '{$row->job_id}'." );
902 $params += [
'namespace' => $row->job_namespace,
'title' => $row->job_title ];
904 $job->setMetadata(
'id', $row->job_id );
905 $job->setMetadata(
'timestamp', $row->job_timestamp );
915 return new JobQueueError( get_class( $e ) .
": " . $e->getMessage() );
934 'job_token_timestamp',
unserialize( $serialized)
wfDebug( $text, $dest='all', array $context=[])
Sends a line to the debug log if enabled or, optionally, to a comment in output.
wfRandomString( $length=32)
Get a random string containing a number of pseudo-random hex characters.
wfDeprecated( $function, $version=false, $component=false, $callerOffset=2)
Logs a warning that a deprecated feature was used.
Class to handle job queues stored in the DB.
claimOldest( $uuid)
Reserve a row with a single UPDATE without holding row locks over RTTs...
supportedOrders()
Get the allowed queue orders for configuration validation.
doGetSiblingQueueSizes(array $types)
insertFields(IJobSpecification $job, IDatabase $db)
__construct(array $params)
Additional parameters include:
getDBException(DBError $e)
doBatchPush(array $jobs, $flags)
static makeBlob( $params)
claimRandom( $uuid, $rand, $gte)
Reserve a row with a single UPDATE without holding row locks over RTTs...
doGetSiblingQueuesWithJobs(array $types)
recycleAndDeleteStaleJobs()
Recycle or destroy any jobs that have been claimed for too long.
doBatchPushInternal(IDatabase $dbw, array $jobs, $flags, $method)
This function should not be called outside of JobQueueDB.
optimalOrder()
Get the default queue order to use if configuration does not specify one.
string null $cluster
Name of an external DB cluster or null for the local DB cluster.
IMaintainableDatabase DBError null $conn
getCoalesceLocationInternal()
Do not use this function outside of JobQueue/JobQueueGroup.
doDeduplicateRootJob(IJobSpecification $job)
static selectFields()
Return the list of job fields that should be selected.
getJobIterator(array $conds)
array null $server
Server configuration array.
Class to handle enqueueing and running of background jobs.
incrStats( $key, $type, $delta=1)
Call StatsdDataFactoryInterface::updateCount() for the queue overall and for the queue type.
factoryJob( $command, $params)
Convenience class for generating iterators from iterators.
Interface for serializable objects that describe a job queue task.
Job that has a run() method and metadata accessors for JobQueue::pop() and JobQueue::ack()
Advanced database interface for IDatabase handles that include maintenance methods.
if(count( $args)< 1) $job