46 private $jobTypeConfiguration;
48 private $jobTypesExcludedFromDefaultQueue;
50 private $statsdDataFactory;
54 private $globalIdGenerator;
60 private const TYPE_ANY = 2;
64 private const PROC_CACHE_TTL = 15;
66 private const CACHE_VERSION = 1;
86 array $jobTypeConfiguration,
87 array $jobTypesExcludedFromDefaultQueue,
96 $this->jobClasses = $jobClasses;
97 $this->jobTypeConfiguration = $jobTypeConfiguration;
98 $this->jobTypesExcludedFromDefaultQueue = $jobTypesExcludedFromDefaultQueue;
99 $this->statsdDataFactory = $statsdDataFactory;
100 $this->wanCache = $wanCache;
101 $this->globalIdGenerator = $globalIdGenerator;
112 $conf += $this->jobTypeConfiguration[
$type] ?? $this->jobTypeConfiguration[
'default'];
113 if ( !isset( $conf[
'readOnlyReason'] ) ) {
114 $conf[
'readOnlyReason'] = $this->readOnlyMode->getReason();
117 $conf[
'stats'] = $this->statsdDataFactory;
118 $conf[
'wanCache'] = $this->wanCache;
119 $conf[
'idGenerator'] = $this->globalIdGenerator;
134 public function push( $jobs ) {
135 if ( $this->invalidDomain ) {
137 $e =
new LogicException(
"Domain '{$this->domain}' is not recognized." );
142 $jobs = is_array( $jobs ) ? $jobs : [ $jobs ];
143 if ( $jobs === [] ) {
147 $this->assertValidJobs( $jobs );
150 foreach ( $jobs as
$job ) {
152 if ( isset( $this->jobTypeConfiguration[
$type] ) ) {
156 isset( $this->jobTypeConfiguration[
'default'][
'typeAgnostic'] ) &&
157 $this->jobTypeConfiguration[
'default'][
'typeAgnostic']
159 $jobsByType[
'default'][] =
$job;
166 foreach ( $jobsByType as
$type => $jobs ) {
170 if ( $this->cache->hasField(
'queues-ready',
'list' ) ) {
171 $list = $this->cache->getField(
'queues-ready',
'list' );
172 if ( count( array_diff( array_keys( $jobsByType ), $list ) ) ) {
173 $this->cache->clear(
'queues-ready' );
179 $cache->makeGlobalKey(
'jobqueue', $this->domain,
'hasjobs', self::TYPE_ANY ),
183 if ( array_diff( array_keys( $jobsByType ), $this->jobTypesExcludedFromDefaultQueue ) ) {
185 $cache->makeGlobalKey(
'jobqueue', $this->domain,
'hasjobs', self::TYPE_DEFAULT ),
200 if ( $this->invalidDomain ) {
202 throw new LogicException(
"Domain '{$this->domain}' is not recognized." );
205 if ( PHP_SAPI ===
'cli' || PHP_SAPI ===
'phpdbg' ) {
206 $this->
push( $jobs );
210 $jobs = is_array( $jobs ) ? $jobs : [ $jobs ];
213 $this->assertValidJobs( $jobs );
229 public function pop( $qtype = self::TYPE_DEFAULT, $flags = 0, array $ignored = [] ) {
232 if ( !WikiMap::isCurrentWikiDbDomain( $this->domain ) ) {
234 "Cannot pop '{$qtype}' job off foreign '{$this->domain}' wiki queue." );
235 } elseif ( is_string( $qtype ) && !isset( $this->jobClasses[$qtype] ) ) {
237 throw new JobQueueError(
"Unrecognized job type '$qtype'." );
240 if ( is_string( $qtype ) ) {
241 if ( !in_array( $qtype, $ignored ) ) {
242 $job = $this->
get( $qtype )->
pop();
245 if ( $flags & self::USE_CACHE ) {
246 if ( !$this->cache->hasField(
'queues-ready',
'list', self::PROC_CACHE_TTL ) ) {
249 $types = $this->cache->getField(
'queues-ready',
'list' );
254 if ( $qtype == self::TYPE_DEFAULT ) {
258 $types = array_diff( $types, $ignored );
261 foreach ( $types as
$type ) {
266 $this->cache->clear(
'queues-ready' );
304 foreach ( $this->jobTypeConfiguration as
$type => $conf ) {
315 return array_keys( $this->getCachedConfigVar(
'wgJobClasses' ) );
324 return array_diff( $this->
getQueueTypes(), $this->jobTypesExcludedFromDefaultQueue );
336 $key =
$cache->makeGlobalKey(
'jobqueue', $this->domain,
'hasjobs',
$type );
339 if ( $value ===
false ) {
341 if (
$type == self::TYPE_DEFAULT ) {
344 $value = count( $queues ) ?
'true' :
'false';
345 $cache->add( $key, $value, 15 );
348 return ( $value ===
'true' );
362 if ( is_array( $nonEmpty ) ) {
363 $types = array_merge( $types, $nonEmpty );
365 foreach ( $info[
'types'] as
$type ) {
366 if ( !$this->
get( $type )->isEmpty() ) {
387 if ( is_array( $sizes ) ) {
390 foreach ( $info[
'types'] as
$type ) {
391 $sizeMap[
$type] = $this->
get(
$type )->getSize();
404 if ( $this->coalescedQueues ===
null ) {
405 $this->coalescedQueues = [];
406 foreach ( $this->jobTypeConfiguration as
$type => $conf ) {
408 $conf[
'type'] =
'null';
409 $conf[
'stats'] = $this->statsdDataFactory;
410 $conf[
'wanCache'] = $this->wanCache;
411 $conf[
'idGenerator'] = $this->globalIdGenerator;
414 $loc =
$queue->getCoalesceLocationInternal();
415 if ( !isset( $this->coalescedQueues[$loc] ) ) {
416 $this->coalescedQueues[$loc][
'queue'] =
$queue;
417 $this->coalescedQueues[$loc][
'types'] = [];
419 if (
$type ===
'default' ) {
420 $this->coalescedQueues[$loc][
'types'] = array_merge(
421 $this->coalescedQueues[$loc][
'types'],
422 array_diff( $this->
getQueueTypes(), array_keys( $this->jobTypeConfiguration ) )
425 $this->coalescedQueues[$loc][
'types'][] =
$type;
437 private function getCachedConfigVar( $name ) {
439 if ( WikiMap::isCurrentWikiDbDomain( $this->domain ) ) {
440 return $GLOBALS[$name];
442 $wiki = WikiMap::getWikiIdFromDbDomain( $this->domain );
443 $cache = MediaWikiServices::getInstance()->getMainWANObjectCache();
445 $cache->makeGlobalKey(
'jobqueue',
'configvalue', $this->domain, $name ),
446 $cache::TTL_DAY + mt_rand( 0, $cache::TTL_DAY ),
447 static function () use ( $wiki, $name ) {
450 return [
'v' =>
$wgConf->getConfig( $wiki, $name ) ];
452 [
'pcTTL' => WANObjectCache::TTL_PROC_LONG ]
463 private function assertValidJobs( array $jobs ) {
464 foreach ( $jobs as
$job ) {
467 throw new InvalidArgumentException(
"Expected IJobSpecification objects, got " .
$type );
$wgConf
$wgConf hold the site configuration.
static addUpdate(DeferrableUpdate $update, $stage=self::POSTSEND)
Add an update to the pending update queue for execution at the appropriate time.
Enqueue lazy-pushed jobs that have accumulated from JobQueueGroup.
Class to handle enqueueing of background jobs.
push( $jobs)
Insert jobs into the respective queues of which they belong.
string $domain
Wiki domain ID.
getQueueSizes()
Get the size of the queues for a list of job types.
__construct( $domain, ConfiguredReadOnlyMode $readOnlyMode, bool $invalidDomain, array $jobClasses, array $jobTypeConfiguration, array $jobTypesExcludedFromDefaultQueue, IBufferingStatsdDataFactory $statsdDataFactory, WANObjectCache $wanCache, GlobalIdGenerator $globalIdGenerator)
waitForBackups()
Wait for any replica DBs or backup queue servers to catch up.
array $coalescedQueues
Map of (bucket => (queue => JobQueue, types => list of types)
getDefaultQueueTypes()
Get the list of default queue types.
lazyPush( $jobs)
Buffer jobs for insertion via push() or call it now if in CLI mode.
pop( $qtype=self::TYPE_DEFAULT, $flags=0, array $ignored=[])
Pop a job off one of the job queues.
ack(RunnableJob $job)
Acknowledge that a job was completed.
ConfiguredReadOnlyMode $readOnlyMode
Read only mode.
queuesHaveJobs( $type=self::TYPE_ANY)
Check if there are any queues with jobs (this is cached)
bool $invalidDomain
Whether the wiki is not recognized in configuration.
getQueuesWithJobs()
Get the list of job types that have non-empty queues.
getQueueTypes()
Get the list of queue types.
deduplicateRootJob(RunnableJob $job)
Register the "root job" of a given job into the queue for de-duplication.
static factory(array $params)
Get a job queue object of the specified type.
static logException(Throwable $e, $catcher=self::CAUGHT_BY_OTHER, $extraData=[])
Log a throwable to the exception log (if enabled).
Handles a simple LRU key/value map with a maximum number of entries.
set( $key, $value, $rank=self::RANK_TOP)
Set a key/value pair.
get( $key, $maxAge=INF, $default=null)
Get the value for a key.
getWithSetCallback( $key, callable $callback, $rank=self::RANK_TOP, $maxAge=INF)
Get an item with the given key, producing and setting it if not found.
static getLocalClusterInstance()
Get the main cluster-local cache object.
Multi-datacenter aware caching interface.
MediaWiki adaptation of StatsdDataFactory that provides buffering functionality.
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()
if(count( $args)< 1) $job