26use Wikimedia\ScopedCallback;
36 private const CACHE_TTL_SHORT = 30;
38 private const MAX_AGE_PRUNE = 7 * 24 * 3600;
43 private const MAX_JOB_RANDOM = 2_147_483_647;
45 private const MAX_OFFSET = 255;
67 if ( isset(
$params[
'server'] ) ) {
68 $this->server =
$params[
'server'];
69 } elseif ( isset(
$params[
'cluster'] ) && is_string(
$params[
'cluster'] ) ) {
70 $this->cluster =
$params[
'cluster'];
75 return [
'random',
'timestamp',
'fifo' ];
89 $scope = $this->getScopedNoTrxFlag( $dbr );
92 $found = (bool)$dbr->newSelectQueryBuilder()
95 ->where( [
'job_cmd' => $this->type,
'job_token' =>
'' ] )
96 ->caller( __METHOD__ )->fetchField();
111 $size = $this->wanCache->get( $key );
112 if ( is_int( $size ) ) {
118 $scope = $this->getScopedNoTrxFlag( $dbr );
120 $size = $dbr->newSelectQueryBuilder()
122 ->where( [
'job_cmd' => $this->type,
'job_token' =>
'' ] )
123 ->caller( __METHOD__ )->fetchRowCount();
127 $this->wanCache->set( $key, $size, self::CACHE_TTL_SHORT );
137 if ( $this->claimTTL <= 0 ) {
143 $count = $this->wanCache->get( $key );
144 if ( is_int( $count ) ) {
150 $scope = $this->getScopedNoTrxFlag( $dbr );
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 $scope = $this->getScopedNoTrxFlag( $dbr );
189 $count = $dbr->newSelectQueryBuilder()
193 'job_cmd' => $this->type,
194 $dbr->expr(
'job_token',
'!=',
'' ),
195 $dbr->expr(
'job_attempts',
'>=', $this->maxTries ),
198 ->caller( __METHOD__ )->fetchRowCount();
203 $this->wanCache->set( $key, $count, self::CACHE_TTL_SHORT );
218 $scope = $this->getScopedNoTrxFlag( $dbw );
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 ) );
289 foreach ( array_chunk( $rows, 50 ) as $rowBatch ) {
291 ->insertInto(
'job' )
293 ->caller( $method )->execute();
295 $this->
incrStats(
'inserts', $this->type, count( $rows ) );
296 $this->
incrStats(
'dupe_inserts', $this->type,
297 count( $rowSet ) + count( $rowList ) - count( $rows )
302 if ( $flags & self::QOS_ATOMIC ) {
314 $scope = $this->getScopedNoTrxFlag( $dbw );
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 $scope = $this->getScopedNoTrxFlag( $dbw );
364 $tinyQueue = $this->wanCache->get( $this->
getCacheKey(
'small' ) );
366 $invertedDirection =
false;
375 $row = $dbw->newSelectQueryBuilder()
376 ->select( self::selectFields() )
380 'job_cmd' => $this->type,
382 $dbw->expr(
'job_random', $gte ?
'>=' :
'<=', $rand )
387 $gte ? SelectQueryBuilder::SORT_ASC : SelectQueryBuilder::SORT_DESC
389 ->caller( __METHOD__ )->fetchRow();
390 if ( !$row && !$invertedDirection ) {
392 $invertedDirection =
true;
399 $row = $dbw->newSelectQueryBuilder()
400 ->select( self::selectFields() )
404 'job_cmd' => $this->type,
408 ->offset( mt_rand( 0, self::MAX_OFFSET ) )
409 ->caller( __METHOD__ )->fetchRow();
412 $this->wanCache->set( $this->
getCacheKey(
'small' ), 1, 30 );
421 $dbw->newUpdateQueryBuilder()
424 'job_token' => $uuid,
425 'job_token_timestamp' => $dbw->timestamp(),
426 'job_attempts = job_attempts+1'
429 'job_cmd' => $this->type,
430 'job_id' => $row->job_id,
433 ->caller( __METHOD__ )->execute();
437 if ( !$dbw->affectedRows() ) {
454 $scope = $this->getScopedNoTrxFlag( $dbw );
458 if ( $dbw->getType() ===
'mysql' ) {
463 $dbw->query(
"UPDATE {$dbw->tableName( 'job' )} " .
465 "job_token = {$dbw->addQuotes( $uuid ) }, " .
466 "job_token_timestamp = {$dbw->addQuotes( $dbw->timestamp() )}, " .
467 "job_attempts = job_attempts+1 " .
469 "job_cmd = {$dbw->addQuotes( $this->type )} " .
470 "AND job_token = {$dbw->addQuotes( '' )} " .
471 ") ORDER BY job_id ASC LIMIT 1",
477 $qb = $dbw->newSelectQueryBuilder()
480 ->where( [
'job_cmd' => $this->type,
'job_token' =>
'' ] )
481 ->orderBy(
'job_id', SelectQueryBuilder::SORT_ASC )
484 $dbw->newUpdateQueryBuilder()
487 'job_token' => $uuid,
488 'job_token_timestamp' => $dbw->timestamp(),
489 'job_attempts = job_attempts+1'
491 ->where( [
'job_id = (' . $qb->getSQL() .
')' ] )
492 ->caller( __METHOD__ )->execute();
495 if ( !$dbw->affectedRows() ) {
500 $row = $dbw->newSelectQueryBuilder()
501 ->select( self::selectFields() )
503 ->where( [
'job_cmd' => $this->type,
'job_token' => $uuid ] )
504 ->caller( __METHOD__ )->fetchRow();
506 wfDebug(
"Row deleted as duplicate by another process." );
520 $id =
$job->getMetadata(
'id' );
521 if ( $id ===
null ) {
522 throw new UnexpectedValueException(
"Job of type '{$job->getType()}' has no ID." );
527 $scope = $this->getScopedNoTrxFlag( $dbw );
530 $dbw->newDeleteQueryBuilder()
531 ->deleteFrom(
'job' )
532 ->where( [
'job_cmd' => $this->type,
'job_id' => $id ] )
533 ->caller( __METHOD__ )->execute();
555 $scope = $this->getScopedNoTrxFlag( $dbw );
556 $dbw->onTransactionCommitOrIdle(
557 function () use (
$job ) {
558 parent::doDeduplicateRootJob(
$job );
573 $scope = $this->getScopedNoTrxFlag( $dbw );
575 $dbw->newDeleteQueryBuilder()
576 ->deleteFrom(
'job' )
577 ->where( [
'job_cmd' => $this->type ] )
578 ->caller( __METHOD__ )->execute();
591 if ( $this->server ) {
595 $lbFactory = MediaWikiServices::getInstance()->getDBLoadBalancerFactory();
596 $lbFactory->waitForReplication();
603 foreach ( [
'size',
'acquiredcount' ] as
$type ) {
604 $this->wanCache->delete( $this->
getCacheKey( $type ) );
632 "job_attempts >= " . intval( $this->maxTries )
643 $scope = $this->getScopedNoTrxFlag( $dbr );
644 $qb = $dbr->newSelectQueryBuilder()
645 ->select( self::selectFields() )
650 $qb->caller( __METHOD__ )->fetchResultSet(),
661 if ( $this->server ) {
665 return is_string( $this->cluster )
666 ?
"DBCluster:{$this->cluster}:{$this->domain}"
667 :
"LBFactory:{$this->domain}";
673 $scope = $this->getScopedNoTrxFlag( $dbr );
678 $res = $dbr->newSelectQueryBuilder()
679 ->select(
'job_cmd' )
682 ->where( [
'job_cmd' => $types ] )
683 ->caller( __METHOD__ )->fetchResultSet();
686 foreach ( $res as $row ) {
687 $types[] = $row->job_cmd;
696 $scope = $this->getScopedNoTrxFlag( $dbr );
698 $res = $dbr->newSelectQueryBuilder()
699 ->select( [
'job_cmd',
'count' =>
'COUNT(*)' ] )
701 ->where( [
'job_cmd' => $types ] )
702 ->groupBy(
'job_cmd' )
703 ->caller( __METHOD__ )->fetchResultSet();
706 foreach ( $res as $row ) {
707 $sizes[$row->job_cmd] = (int)$row->count;
723 $scope = $this->getScopedNoTrxFlag( $dbw );
726 if ( !$dbw->lock(
"jobqueue-recycle-{$this->type}", __METHOD__, 1 ) ) {
731 if ( $this->claimTTL > 0 ) {
732 $claimCutoff = $dbw->timestamp( $now - $this->claimTTL );
736 $res = $dbw->newSelectQueryBuilder()
741 'job_cmd' => $this->type,
742 $dbw->expr(
'job_token',
'!=',
'' ),
743 $dbw->expr(
'job_token_timestamp',
'<', $claimCutoff ),
744 $dbw->expr(
'job_attempts',
'<', $this->maxTries ),
747 ->caller( __METHOD__ )->fetchResultSet();
749 static function ( $o ) {
751 }, iterator_to_array( $res )
753 if ( count( $ids ) ) {
757 $dbw->newUpdateQueryBuilder()
761 'job_token_timestamp' => $dbw->timestamp( $now )
765 $dbw->expr(
'job_token',
'!=',
'' ),
767 ->caller( __METHOD__ )->execute();
769 $affected = $dbw->affectedRows();
771 $this->
incrStats(
'recycles', $this->type, $affected );
776 $pruneCutoff = $dbw->timestamp( $now - self::MAX_AGE_PRUNE );
777 $qb = $dbw->newSelectQueryBuilder()
782 'job_cmd' => $this->type,
783 $dbw->expr(
'job_token',
'!=',
'' ),
784 $dbw->expr(
'job_token_timestamp',
'<', $pruneCutoff )
787 if ( $this->claimTTL > 0 ) {
788 $qb->andWhere( $dbw->expr(
'job_attempts',
'>=', $this->maxTries ) );
792 $res = $qb->caller( __METHOD__ )->fetchResultSet();
794 static function ( $o ) {
796 }, iterator_to_array( $res )
798 if ( count( $ids ) ) {
799 $dbw->newDeleteQueryBuilder()
800 ->deleteFrom(
'job' )
801 ->where( [
'job_id' => $ids ] )
802 ->caller( __METHOD__ )->execute();
803 $affected = $dbw->affectedRows();
805 $this->
incrStats(
'abandons', $this->type, $affected );
808 $dbw->unlock(
"jobqueue-recycle-{$this->type}", __METHOD__ );
824 'job_cmd' =>
$job->getType(),
826 'job_title' =>
$job->getParams()[
'title'] ??
'',
827 'job_params' => self::makeBlob(
$job->getParams() ),
830 'job_sha1' => Wikimedia\base_convert(
831 sha1( serialize(
$job->getDeduplicationInfo() ) ),
834 'job_random' => mt_rand( 0, self::MAX_JOB_RANDOM )
867 protected function getDB( $index ) {
868 if ( $this->server ) {
869 if ( $this->conn instanceof
IDatabase ) {
871 } elseif ( $this->conn instanceof
DBError ) {
876 $this->conn = MediaWikiServices::getInstance()->getDatabaseFactory()->create(
877 $this->server[
'type'],
887 $lbFactory = MediaWikiServices::getInstance()->getDBLoadBalancerFactory();
888 $lb = is_string( $this->cluster )
889 ? $lbFactory->getExternalLB( $this->cluster )
890 : $lbFactory->getMainLB( $this->domain );
892 if ( $lb->getServerType( $lb->getWriterIndex() ) !==
'sqlite' ) {
895 $flags = $lb::CONN_TRX_AUTOCOMMIT;
901 return $lb->getMaintenanceConnectionRef( $index, [], $this->domain, $flags );
909 private function getScopedNoTrxFlag(
IDatabase $db ) {
913 return new ScopedCallback(
static function () use ( $db, $autoTrx ) {
925 $cluster = is_string( $this->cluster ) ? $this->cluster :
'main';
927 return $this->wanCache->makeGlobalKey(
953 $params = ( (string)$row->job_params !==
'' ) ? unserialize( $row->job_params ) : [];
955 throw new UnexpectedValueException(
956 "Could not unserialize job with ID '{$row->job_id}'." );
959 $params += [
'namespace' => $row->job_namespace,
'title' => $row->job_title ];
961 $job->setMetadata(
'id', $row->job_id );
962 $job->setMetadata(
'timestamp', $row->job_timestamp );
972 return new JobQueueError( get_class( $e ) .
": " . $e->getMessage() );
991 '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)
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.
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