29use Wikimedia\ScopedCallback;
39 private const CACHE_TTL_SHORT = 30;
41 private const MAX_AGE_PRUNE = 7 * 24 * 3600;
46 private const MAX_JOB_RANDOM = 2_147_483_647;
48 private const MAX_OFFSET = 255;
70 if ( isset(
$params[
'server'] ) ) {
71 $this->server =
$params[
'server'];
73 $this->server[
'flags'] ??= 0;
74 $this->server[
'flags'] &= ~( IDatabase::DBO_TRX | IDatabase::DBO_DEFAULT );
75 } elseif ( isset(
$params[
'cluster'] ) && is_string(
$params[
'cluster'] ) ) {
76 $this->cluster =
$params[
'cluster'];
81 return [
'random',
'timestamp',
'fifo' ];
96 $found = (bool)$dbr->newSelectQueryBuilder()
99 ->where( [
'job_cmd' => $this->type,
'job_token' =>
'' ] )
100 ->caller( __METHOD__ )->fetchField();
115 $size = $this->wanCache->get( $key );
116 if ( is_int( $size ) ) {
122 $size = $dbr->newSelectQueryBuilder()
124 ->where( [
'job_cmd' => $this->type,
'job_token' =>
'' ] )
125 ->caller( __METHOD__ )->fetchRowCount();
129 $this->wanCache->set( $key, $size, self::CACHE_TTL_SHORT );
139 if ( $this->claimTTL <= 0 ) {
145 $count = $this->wanCache->get( $key );
146 if ( is_int( $count ) ) {
152 $count = $dbr->newSelectQueryBuilder()
155 'job_cmd' => $this->type,
156 $dbr->expr(
'job_token',
'!=',
'' ),
158 ->caller( __METHOD__ )->fetchRowCount();
162 $this->wanCache->set( $key, $count, self::CACHE_TTL_SHORT );
174 if ( $this->claimTTL <= 0 ) {
180 $count = $this->wanCache->get( $key );
181 if ( is_int( $count ) ) {
187 $count = $dbr->newSelectQueryBuilder()
191 'job_cmd' => $this->type,
192 $dbr->expr(
'job_token',
'!=',
'' ),
193 $dbr->expr(
'job_attempts',
'>=', $this->maxTries ),
196 ->caller( __METHOD__ )->fetchRowCount();
201 $this->wanCache->set( $key, $count, self::CACHE_TTL_SHORT );
216 $scope = $transactionProfiler->silenceForScope();
218 ScopedCallback::consume( $scope );
229 $dbw->onTransactionPreCommitOrIdle(
230 function (
IDatabase $dbw ) use ( $jobs, $flags, $fname ) {
249 if ( $jobs === [] ) {
255 foreach ( $jobs as
$job ) {
257 if (
$job->ignoreDuplicates() ) {
258 $rowSet[$row[
'job_sha1']] = $row;
264 if ( $flags & self::QOS_ATOMIC ) {
269 if ( count( $rowSet ) ) {
271 ->select(
'job_sha1' )
276 'job_sha1' => array_map(
'strval', array_keys( $rowSet ) ),
280 ->caller( $method )->fetchResultSet();
281 foreach ( $res as $row ) {
282 wfDebug(
"Job with hash '{$row->job_sha1}' is a duplicate." );
283 unset( $rowSet[$row->job_sha1] );
287 $rows = array_merge( $rowList, array_values( $rowSet ) );
291 $scope = $transactionProfiler->silenceForScope();
293 foreach ( array_chunk( $rows, 50 ) as $rowBatch ) {
295 ->insertInto(
'job' )
297 ->caller( $method )->execute();
299 ScopedCallback::consume( $scope );
300 $this->
incrStats(
'inserts', $this->type, count( $rows ) );
301 $this->
incrStats(
'dupe_inserts', $this->type,
302 count( $rowSet ) + count( $rowList ) - count( $rows )
307 if ( $flags & self::QOS_ATOMIC ) {
322 if ( in_array( $this->order, [
'fifo',
'timestamp' ] ) ) {
325 $rand = mt_rand( 0, self::MAX_JOB_RANDOM );
326 $gte = (bool)mt_rand( 0, 1 );
340 if ( !
$job || mt_rand( 0, 9 ) == 0 ) {
363 $tinyQueue = $this->wanCache->get( $this->
getCacheKey(
'small' ) );
365 $invertedDirection =
false;
374 $row = $dbw->newSelectQueryBuilder()
375 ->select( self::selectFields() )
379 'job_cmd' => $this->type,
381 $dbw->expr(
'job_random', $gte ?
'>=' :
'<=', $rand )
386 $gte ? SelectQueryBuilder::SORT_ASC : SelectQueryBuilder::SORT_DESC
388 ->caller( __METHOD__ )->fetchRow();
389 if ( !$row && !$invertedDirection ) {
391 $invertedDirection =
true;
398 $row = $dbw->newSelectQueryBuilder()
399 ->select( self::selectFields() )
403 'job_cmd' => $this->type,
407 ->offset( mt_rand( 0, self::MAX_OFFSET ) )
408 ->caller( __METHOD__ )->fetchRow();
411 $this->wanCache->set( $this->
getCacheKey(
'small' ), 1, 30 );
420 $dbw->newUpdateQueryBuilder()
423 'job_token' => $uuid,
424 'job_token_timestamp' => $dbw->timestamp(),
425 'job_attempts' =>
new RawSQLValue(
'job_attempts+1' ),
428 'job_cmd' => $this->type,
429 'job_id' => $row->job_id,
432 ->caller( __METHOD__ )->execute();
436 if ( !$dbw->affectedRows() ) {
455 if ( $dbw->getType() ===
'mysql' ) {
460 $dbw->query(
"UPDATE {$dbw->tableName( 'job' )} " .
462 "job_token = {$dbw->addQuotes( $uuid ) }, " .
463 "job_token_timestamp = {$dbw->addQuotes( $dbw->timestamp() )}, " .
464 "job_attempts = job_attempts+1 " .
466 "job_cmd = {$dbw->addQuotes( $this->type )} " .
467 "AND job_token = {$dbw->addQuotes( '' )} " .
468 ") ORDER BY job_id ASC LIMIT 1",
474 $qb = $dbw->newSelectQueryBuilder()
477 ->where( [
'job_cmd' => $this->type,
'job_token' =>
'' ] )
478 ->orderBy(
'job_id', SelectQueryBuilder::SORT_ASC )
481 $dbw->newUpdateQueryBuilder()
484 'job_token' => $uuid,
485 'job_token_timestamp' => $dbw->timestamp(),
486 'job_attempts' =>
new RawSQLValue(
'job_attempts+1' ),
488 ->where( [
'job_id' =>
new RawSQLValue(
'(' . $qb->getSQL() .
')' ) ] )
489 ->caller( __METHOD__ )->execute();
492 if ( !$dbw->affectedRows() ) {
497 $row = $dbw->newSelectQueryBuilder()
498 ->select( self::selectFields() )
500 ->where( [
'job_cmd' => $this->type,
'job_token' => $uuid ] )
501 ->caller( __METHOD__ )->fetchRow();
503 wfDebug(
"Row deleted as duplicate by another process." );
517 $id =
$job->getMetadata(
'id' );
518 if ( $id ===
null ) {
519 throw new UnexpectedValueException(
"Job of type '{$job->getType()}' has no ID." );
525 $dbw->newDeleteQueryBuilder()
526 ->deleteFrom(
'job' )
527 ->where( [
'job_cmd' => $this->type,
'job_id' => $id ] )
528 ->caller( __METHOD__ )->execute();
549 $dbw->onTransactionCommitOrIdle(
550 function () use (
$job ) {
551 parent::doDeduplicateRootJob(
$job );
566 $dbw->newDeleteQueryBuilder()
567 ->deleteFrom(
'job' )
568 ->where( [
'job_cmd' => $this->type ] )
569 ->caller( __METHOD__ )->execute();
582 if ( $this->server ) {
586 $lbFactory = MediaWikiServices::getInstance()->getDBLoadBalancerFactory();
587 $lbFactory->waitForReplication();
594 foreach ( [
'size',
'acquiredcount' ] as
$type ) {
595 $this->wanCache->delete( $this->
getCacheKey( $type ) );
613 return $this->
getJobIterator( [
'job_cmd' => $this->
getType(), $dbr->expr(
'job_token',
'>',
'' ) ] );
624 $dbr->expr(
'job_token',
'>',
'' ),
625 $dbr->expr(
'job_attempts',
'>=', intval( $this->maxTries ) ),
635 $qb = $dbr->newSelectQueryBuilder()
636 ->select( self::selectFields() )
641 $qb->caller( __METHOD__ )->fetchResultSet(),
652 if ( $this->server ) {
656 return is_string( $this->cluster )
657 ?
"DBCluster:{$this->cluster}:{$this->domain}"
658 :
"LBFactory:{$this->domain}";
667 $res = $dbr->newSelectQueryBuilder()
668 ->select(
'job_cmd' )
671 ->where( [
'job_cmd' => $types ] )
672 ->caller( __METHOD__ )->fetchResultSet();
675 foreach ( $res as $row ) {
676 $types[] = $row->job_cmd;
685 $res = $dbr->newSelectQueryBuilder()
686 ->select( [
'job_cmd',
'count' =>
'COUNT(*)' ] )
688 ->where( [
'job_cmd' => $types ] )
689 ->groupBy(
'job_cmd' )
690 ->caller( __METHOD__ )->fetchResultSet();
693 foreach ( $res as $row ) {
694 $sizes[$row->job_cmd] = (int)$row->count;
711 if ( !$dbw->lock(
"jobqueue-recycle-{$this->type}", __METHOD__, 1 ) ) {
716 if ( $this->claimTTL > 0 ) {
717 $claimCutoff = $dbw->timestamp( $now - $this->claimTTL );
721 $res = $dbw->newSelectQueryBuilder()
726 'job_cmd' => $this->type,
727 $dbw->expr(
'job_token',
'!=',
'' ),
728 $dbw->expr(
'job_token_timestamp',
'<', $claimCutoff ),
729 $dbw->expr(
'job_attempts',
'<', $this->maxTries ),
732 ->caller( __METHOD__ )->fetchResultSet();
734 static function ( $o ) {
736 }, iterator_to_array( $res )
738 if ( count( $ids ) ) {
742 $dbw->newUpdateQueryBuilder()
746 'job_token_timestamp' => $dbw->timestamp( $now )
750 $dbw->expr(
'job_token',
'!=',
'' ),
752 ->caller( __METHOD__ )->execute();
754 $affected = $dbw->affectedRows();
756 $this->
incrStats(
'recycles', $this->type, $affected );
761 $pruneCutoff = $dbw->timestamp( $now - self::MAX_AGE_PRUNE );
762 $qb = $dbw->newSelectQueryBuilder()
767 'job_cmd' => $this->type,
768 $dbw->expr(
'job_token',
'!=',
'' ),
769 $dbw->expr(
'job_token_timestamp',
'<', $pruneCutoff )
772 if ( $this->claimTTL > 0 ) {
773 $qb->andWhere( $dbw->expr(
'job_attempts',
'>=', $this->maxTries ) );
777 $res = $qb->caller( __METHOD__ )->fetchResultSet();
779 static function ( $o ) {
781 }, iterator_to_array( $res )
783 if ( count( $ids ) ) {
784 $dbw->newDeleteQueryBuilder()
785 ->deleteFrom(
'job' )
786 ->where( [
'job_id' => $ids ] )
787 ->caller( __METHOD__ )->execute();
788 $affected = $dbw->affectedRows();
790 $this->
incrStats(
'abandons', $this->type, $affected );
793 $dbw->unlock(
"jobqueue-recycle-{$this->type}", __METHOD__ );
809 'job_cmd' =>
$job->getType(),
811 'job_title' =>
$job->getParams()[
'title'] ??
'',
812 'job_params' => self::makeBlob(
$job->getParams() ),
815 'job_sha1' => Wikimedia\base_convert(
816 sha1( serialize(
$job->getDeduplicationInfo() ) ),
819 'job_random' => mt_rand( 0, self::MAX_JOB_RANDOM )
852 protected function getDB( $index ) {
853 if ( $this->server ) {
854 if ( $this->conn instanceof
IDatabase ) {
856 } elseif ( $this->conn instanceof
DBError ) {
861 $this->conn = MediaWikiServices::getInstance()->getDatabaseFactory()->create(
862 $this->server[
'type'],
872 $lbFactory = MediaWikiServices::getInstance()->getDBLoadBalancerFactory();
873 $lb = is_string( $this->cluster )
874 ? $lbFactory->getExternalLB( $this->cluster )
875 : $lbFactory->getMainLB( $this->domain );
877 if ( $lb->getServerType( ServerInfo::WRITER_INDEX ) !==
'sqlite' ) {
880 $flags = $lb::CONN_TRX_AUTOCOMMIT;
886 return $lb->getMaintenanceConnectionRef( $index, [], $this->domain, $flags );
895 $cluster = is_string( $this->cluster ) ? $this->cluster :
'main';
897 return $this->wanCache->makeGlobalKey(
923 $params = ( (string)$row->job_params !==
'' ) ? unserialize( $row->job_params ) : [];
925 throw new UnexpectedValueException(
926 "Could not unserialize job with ID '{$row->job_id}'." );
929 $params += [
'namespace' => $row->job_namespace,
'title' => $row->job_title ];
931 $job->setMetadata(
'id', $row->job_id );
932 $job->setMetadata(
'timestamp', $row->job_timestamp );
942 return new JobQueueError( get_class( $e ) .
": " . $e->getMessage() );
961 'job_token_timestamp',
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.
array $params
The job parameters.
getCacheKey()
Get the cache key used to store status.
Database-backed job queue storage.
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)
__construct(array $params)
Additional parameters include:
getDBException(DBError $e)
doBatchPush(array $jobs, $flags)
insertFields(IJobSpecification $job, IReadableDatabase $db)
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.
Base class for queueing and running background jobs from a storage backend.
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