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;
69 if ( isset(
$params[
'server'] ) ) {
70 $this->server =
$params[
'server'];
72 $this->server[
'flags'] ??= 0;
73 $this->server[
'flags'] &= ~( IDatabase::DBO_TRX | IDatabase::DBO_DEFAULT );
74 } elseif ( isset(
$params[
'cluster'] ) && is_string(
$params[
'cluster'] ) ) {
75 $this->cluster =
$params[
'cluster'];
80 return [
'random',
'timestamp',
'fifo' ];
95 $found = (bool)$dbr->newSelectQueryBuilder()
98 ->where( [
'job_cmd' => $this->type,
'job_token' =>
'' ] )
99 ->caller( __METHOD__ )->fetchField();
114 $size = $this->wanCache->get( $key );
115 if ( is_int( $size ) ) {
121 $size = $dbr->newSelectQueryBuilder()
123 ->where( [
'job_cmd' => $this->type,
'job_token' =>
'' ] )
124 ->caller( __METHOD__ )->fetchRowCount();
128 $this->wanCache->set( $key, $size, self::CACHE_TTL_SHORT );
138 if ( $this->claimTTL <= 0 ) {
144 $count = $this->wanCache->get( $key );
145 if ( is_int( $count ) ) {
151 $count = $dbr->newSelectQueryBuilder()
154 'job_cmd' => $this->type,
155 $dbr->expr(
'job_token',
'!=',
'' ),
157 ->caller( __METHOD__ )->fetchRowCount();
161 $this->wanCache->set( $key, $count, self::CACHE_TTL_SHORT );
173 if ( $this->claimTTL <= 0 ) {
179 $count = $this->wanCache->get( $key );
180 if ( is_int( $count ) ) {
186 $count = $dbr->newSelectQueryBuilder()
190 'job_cmd' => $this->type,
191 $dbr->expr(
'job_token',
'!=',
'' ),
192 $dbr->expr(
'job_attempts',
'>=', $this->maxTries ),
195 ->caller( __METHOD__ )->fetchRowCount();
200 $this->wanCache->set( $key, $count, self::CACHE_TTL_SHORT );
215 $scope = $transactionProfiler->silenceForScope();
217 ScopedCallback::consume( $scope );
228 $dbw->onTransactionPreCommitOrIdle(
229 function (
IDatabase $dbw ) use ( $jobs, $flags, $fname ) {
248 if ( $jobs === [] ) {
254 foreach ( $jobs as
$job ) {
256 if (
$job->ignoreDuplicates() ) {
257 $rowSet[$row[
'job_sha1']] = $row;
263 if ( $flags & self::QOS_ATOMIC ) {
268 if ( count( $rowSet ) ) {
270 ->select(
'job_sha1' )
275 'job_sha1' => array_map(
'strval', array_keys( $rowSet ) ),
279 ->caller( $method )->fetchResultSet();
280 foreach ( $res as $row ) {
281 wfDebug(
"Job with hash '{$row->job_sha1}' is a duplicate." );
282 unset( $rowSet[$row->job_sha1] );
286 $rows = array_merge( $rowList, array_values( $rowSet ) );
290 $scope = $transactionProfiler->silenceForScope();
292 foreach ( array_chunk( $rows, 50 ) as $rowBatch ) {
294 ->insertInto(
'job' )
296 ->caller( $method )->execute();
298 ScopedCallback::consume( $scope );
299 $this->
incrStats(
'inserts', $this->type, count( $rows ) );
300 $this->
incrStats(
'dupe_inserts', $this->type,
301 count( $rowSet ) + count( $rowList ) - count( $rows )
306 if ( $flags & self::QOS_ATOMIC ) {
321 if ( in_array( $this->order, [
'fifo',
'timestamp' ] ) ) {
324 $rand = mt_rand( 0, self::MAX_JOB_RANDOM );
325 $gte = (bool)mt_rand( 0, 1 );
339 if ( !
$job || mt_rand( 0, 9 ) == 0 ) {
362 $tinyQueue = $this->wanCache->get( $this->
getCacheKey(
'small' ) );
364 $invertedDirection =
false;
373 $row = $dbw->newSelectQueryBuilder()
374 ->select( self::selectFields() )
378 'job_cmd' => $this->type,
380 $dbw->expr(
'job_random', $gte ?
'>=' :
'<=', $rand )
385 $gte ? SelectQueryBuilder::SORT_ASC : SelectQueryBuilder::SORT_DESC
387 ->caller( __METHOD__ )->fetchRow();
388 if ( !$row && !$invertedDirection ) {
390 $invertedDirection =
true;
397 $row = $dbw->newSelectQueryBuilder()
398 ->select( self::selectFields() )
402 'job_cmd' => $this->type,
406 ->offset( mt_rand( 0, self::MAX_OFFSET ) )
407 ->caller( __METHOD__ )->fetchRow();
410 $this->wanCache->set( $this->
getCacheKey(
'small' ), 1, 30 );
419 $dbw->newUpdateQueryBuilder()
422 'job_token' => $uuid,
423 'job_token_timestamp' => $dbw->timestamp(),
424 'job_attempts' =>
new RawSQLValue(
'job_attempts+1' ),
427 'job_cmd' => $this->type,
428 'job_id' => $row->job_id,
431 ->caller( __METHOD__ )->execute();
435 if ( !$dbw->affectedRows() ) {
454 if ( $dbw->getType() ===
'mysql' ) {
459 $dbw->query(
"UPDATE {$dbw->tableName( 'job' )} " .
461 "job_token = {$dbw->addQuotes( $uuid ) }, " .
462 "job_token_timestamp = {$dbw->addQuotes( $dbw->timestamp() )}, " .
463 "job_attempts = job_attempts+1 " .
465 "job_cmd = {$dbw->addQuotes( $this->type )} " .
466 "AND job_token = {$dbw->addQuotes( '' )} " .
467 ") ORDER BY job_id ASC LIMIT 1",
473 $qb = $dbw->newSelectQueryBuilder()
476 ->where( [
'job_cmd' => $this->type,
'job_token' =>
'' ] )
477 ->orderBy(
'job_id', SelectQueryBuilder::SORT_ASC )
480 $dbw->newUpdateQueryBuilder()
483 'job_token' => $uuid,
484 'job_token_timestamp' => $dbw->timestamp(),
485 'job_attempts' =>
new RawSQLValue(
'job_attempts+1' ),
487 ->where( [
'job_id' =>
new RawSQLValue(
'(' . $qb->getSQL() .
')' ) ] )
488 ->caller( __METHOD__ )->execute();
491 if ( !$dbw->affectedRows() ) {
496 $row = $dbw->newSelectQueryBuilder()
497 ->select( self::selectFields() )
499 ->where( [
'job_cmd' => $this->type,
'job_token' => $uuid ] )
500 ->caller( __METHOD__ )->fetchRow();
502 wfDebug(
"Row deleted as duplicate by another process." );
516 $id =
$job->getMetadata(
'id' );
517 if ( $id ===
null ) {
518 throw new UnexpectedValueException(
"Job of type '{$job->getType()}' has no ID." );
524 $dbw->newDeleteQueryBuilder()
525 ->deleteFrom(
'job' )
526 ->where( [
'job_cmd' => $this->type,
'job_id' => $id ] )
527 ->caller( __METHOD__ )->execute();
548 $dbw->onTransactionCommitOrIdle(
549 function () use (
$job ) {
550 parent::doDeduplicateRootJob(
$job );
565 $dbw->newDeleteQueryBuilder()
566 ->deleteFrom(
'job' )
567 ->where( [
'job_cmd' => $this->type ] )
568 ->caller( __METHOD__ )->execute();
581 if ( $this->server ) {
585 $lbFactory = MediaWikiServices::getInstance()->getDBLoadBalancerFactory();
586 $lbFactory->waitForReplication();
593 foreach ( [
'size',
'acquiredcount' ] as
$type ) {
594 $this->wanCache->delete( $this->
getCacheKey( $type ) );
612 return $this->
getJobIterator( [
'job_cmd' => $this->
getType(), $dbr->expr(
'job_token',
'>',
'' ) ] );
623 $dbr->expr(
'job_token',
'>',
'' ),
624 $dbr->expr(
'job_attempts',
'>=', intval( $this->maxTries ) ),
634 $qb = $dbr->newSelectQueryBuilder()
635 ->select( self::selectFields() )
640 $qb->caller( __METHOD__ )->fetchResultSet(),
651 if ( $this->server ) {
655 return is_string( $this->cluster )
656 ?
"DBCluster:{$this->cluster}:{$this->domain}"
657 :
"LBFactory:{$this->domain}";
666 $res = $dbr->newSelectQueryBuilder()
667 ->select(
'job_cmd' )
670 ->where( [
'job_cmd' => $types ] )
671 ->caller( __METHOD__ )->fetchResultSet();
674 foreach ( $res as $row ) {
675 $types[] = $row->job_cmd;
684 $res = $dbr->newSelectQueryBuilder()
685 ->select( [
'job_cmd',
'count' =>
'COUNT(*)' ] )
687 ->where( [
'job_cmd' => $types ] )
688 ->groupBy(
'job_cmd' )
689 ->caller( __METHOD__ )->fetchResultSet();
692 foreach ( $res as $row ) {
693 $sizes[$row->job_cmd] = (int)$row->count;
710 if ( !$dbw->lock(
"jobqueue-recycle-{$this->type}", __METHOD__, 1 ) ) {
715 if ( $this->claimTTL > 0 ) {
716 $claimCutoff = $dbw->timestamp( $now - $this->claimTTL );
720 $res = $dbw->newSelectQueryBuilder()
725 'job_cmd' => $this->type,
726 $dbw->expr(
'job_token',
'!=',
'' ),
727 $dbw->expr(
'job_token_timestamp',
'<', $claimCutoff ),
728 $dbw->expr(
'job_attempts',
'<', $this->maxTries ),
731 ->caller( __METHOD__ )->fetchResultSet();
733 static function ( $o ) {
735 }, iterator_to_array( $res )
737 if ( count( $ids ) ) {
741 $dbw->newUpdateQueryBuilder()
745 'job_token_timestamp' => $dbw->timestamp( $now )
749 $dbw->expr(
'job_token',
'!=',
'' ),
751 ->caller( __METHOD__ )->execute();
753 $affected = $dbw->affectedRows();
755 $this->
incrStats(
'recycles', $this->type, $affected );
760 $pruneCutoff = $dbw->timestamp( $now - self::MAX_AGE_PRUNE );
761 $qb = $dbw->newSelectQueryBuilder()
766 'job_cmd' => $this->type,
767 $dbw->expr(
'job_token',
'!=',
'' ),
768 $dbw->expr(
'job_token_timestamp',
'<', $pruneCutoff )
771 if ( $this->claimTTL > 0 ) {
772 $qb->andWhere( $dbw->expr(
'job_attempts',
'>=', $this->maxTries ) );
776 $res = $qb->caller( __METHOD__ )->fetchResultSet();
778 static function ( $o ) {
780 }, iterator_to_array( $res )
782 if ( count( $ids ) ) {
783 $dbw->newDeleteQueryBuilder()
784 ->deleteFrom(
'job' )
785 ->where( [
'job_id' => $ids ] )
786 ->caller( __METHOD__ )->execute();
787 $affected = $dbw->affectedRows();
789 $this->
incrStats(
'abandons', $this->type, $affected );
792 $dbw->unlock(
"jobqueue-recycle-{$this->type}", __METHOD__ );
808 'job_cmd' =>
$job->getType(),
810 'job_title' =>
$job->getParams()[
'title'] ??
'',
811 'job_params' => self::makeBlob(
$job->getParams() ),
814 'job_sha1' => Wikimedia\base_convert(
815 sha1( serialize(
$job->getDeduplicationInfo() ) ),
818 'job_random' => mt_rand( 0, self::MAX_JOB_RANDOM )
851 protected function getDB( $index ) {
852 if ( $this->server ) {
853 if ( $this->conn instanceof
IDatabase ) {
855 } elseif ( $this->conn instanceof
DBError ) {
860 $this->conn = MediaWikiServices::getInstance()->getDatabaseFactory()->create(
861 $this->server[
'type'],
871 $lbFactory = MediaWikiServices::getInstance()->getDBLoadBalancerFactory();
872 $lb = is_string( $this->cluster )
873 ? $lbFactory->getExternalLB( $this->cluster )
874 : $lbFactory->getMainLB( $this->domain );
876 if ( $lb->getServerType( ServerInfo::WRITER_INDEX ) !==
'sqlite' ) {
879 $flags = $lb::CONN_TRX_AUTOCOMMIT;
885 return $lb->getMaintenanceConnectionRef( $index, [], $this->domain, $flags );
894 $cluster = is_string( $this->cluster ) ? $this->cluster :
'main';
896 return $this->wanCache->makeGlobalKey(
922 $params = ( (string)$row->job_params !==
'' ) ? unserialize( $row->job_params ) : [];
924 throw new UnexpectedValueException(
925 "Could not unserialize job with ID '{$row->job_id}'." );
928 $params += [
'namespace' => $row->job_namespace,
'title' => $row->job_title ];
930 $job->setMetadata(
'id', $row->job_id );
931 $job->setMetadata(
'timestamp', $row->job_timestamp );
941 return new JobQueueError( get_class( $e ) .
": " . $e->getMessage() );
960 '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