115 return MediaWikiServices::getInstance()->getJobQueueGroupFactory()->makeJobQueueGroup(
$domain );
135 if ( isset( $this->jobTypeConfiguration[
$type] ) ) {
136 $conf += $this->jobTypeConfiguration[
$type];
138 $conf += $this->jobTypeConfiguration[
'default'];
140 if ( !isset( $conf[
'readOnlyReason'] ) ) {
141 $conf[
'readOnlyReason'] = $this->readOnlyMode->getReason();
161 public function push( $jobs ) {
162 if ( $this->invalidDomain ) {
164 $e =
new LogicException(
"Domain '{$this->domain}' is not recognized." );
169 $jobs = is_array( $jobs ) ? $jobs : [ $jobs ];
170 if ( $jobs === [] ) {
177 foreach ( $jobs as
$job ) {
179 if ( isset( $this->jobTypeConfiguration[
$type] ) ) {
183 isset( $this->jobTypeConfiguration[
'default'][
'typeAgnostic'] ) &&
184 $this->jobTypeConfiguration[
'default'][
'typeAgnostic']
186 $jobsByType[
'default'][] =
$job;
193 foreach ( $jobsByType as
$type => $jobs ) {
197 if ( $this->cache->hasField(
'queues-ready',
'list' ) ) {
198 $list = $this->cache->getField(
'queues-ready',
'list' );
199 if ( count( array_diff( array_keys( $jobsByType ), $list ) ) ) {
200 $this->cache->clear(
'queues-ready' );
206 $cache->makeGlobalKey(
'jobqueue', $this->domain,
'hasjobs', self::TYPE_ANY ),
210 if ( array_diff( array_keys( $jobsByType ), $this->jobTypesExcludedFromDefaultQueue ) ) {
212 $cache->makeGlobalKey(
'jobqueue', $this->domain,
'hasjobs', self::TYPE_DEFAULT ),
227 if ( $this->invalidDomain ) {
229 throw new LogicException(
"Domain '{$this->domain}' is not recognized." );
232 if ( PHP_SAPI ===
'cli' || PHP_SAPI ===
'phpdbg' ) {
233 $this->
push( $jobs );
237 $jobs = is_array( $jobs ) ? $jobs : [ $jobs ];
256 public function pop( $qtype = self::TYPE_DEFAULT, $flags = 0, array $ignored = [] ) {
261 "Cannot pop '{$qtype}' job off foreign '{$this->domain}' wiki queue." );
262 } elseif ( is_string( $qtype ) && !isset( $this->jobClasses[$qtype] ) ) {
264 throw new JobQueueError(
"Unrecognized job type '$qtype'." );
267 if ( is_string( $qtype ) ) {
268 if ( !in_array( $qtype, $ignored ) ) {
269 $job = $this->
get( $qtype )->
pop();
272 if ( $flags & self::USE_CACHE ) {
273 if ( !$this->cache->hasField(
'queues-ready',
'list', self::PROC_CACHE_TTL ) ) {
276 $types = $this->cache->getField(
'queues-ready',
'list' );
281 if ( $qtype == self::TYPE_DEFAULT ) {
285 $types = array_diff( $types, $ignored );
288 foreach ( $types as
$type ) {
293 $this->cache->clear(
'queues-ready' );
331 foreach ( $this->jobTypeConfiguration as
$type => $conf ) {
351 return array_diff( $this->
getQueueTypes(), $this->jobTypesExcludedFromDefaultQueue );
363 $key =
$cache->makeGlobalKey(
'jobqueue', $this->domain,
'hasjobs',
$type );
366 if ( $value ===
false ) {
368 if (
$type == self::TYPE_DEFAULT ) {
371 $value = count( $queues ) ?
'true' :
'false';
372 $cache->add( $key, $value, 15 );
375 return ( $value ===
'true' );
389 if ( is_array( $nonEmpty ) ) {
390 $types = array_merge( $types, $nonEmpty );
392 foreach ( $info[
'types'] as
$type ) {
393 if ( !$this->
get( $type )->isEmpty() ) {
414 if ( is_array( $sizes ) ) {
417 foreach ( $info[
'types'] as
$type ) {
418 $sizeMap[
$type] = $this->
get(
$type )->getSize();
431 if ( $this->coalescedQueues ===
null ) {
432 $this->coalescedQueues = [];
433 foreach ( $this->jobTypeConfiguration as
$type => $conf ) {
435 $conf[
'type'] =
'null';
441 $loc =
$queue->getCoalesceLocationInternal();
442 if ( !isset( $this->coalescedQueues[$loc] ) ) {
443 $this->coalescedQueues[$loc][
'queue'] =
$queue;
444 $this->coalescedQueues[$loc][
'types'] = [];
446 if (
$type ===
'default' ) {
447 $this->coalescedQueues[$loc][
'types'] = array_merge(
448 $this->coalescedQueues[$loc][
'types'],
449 array_diff( $this->
getQueueTypes(), array_keys( $this->jobTypeConfiguration ) )
452 $this->coalescedQueues[$loc][
'types'][] =
$type;
467 return $GLOBALS[$name];
470 $cache = MediaWikiServices::getInstance()->getMainWANObjectCache();
472 $cache->makeGlobalKey(
'jobqueue',
'configvalue', $this->domain, $name ),
473 $cache::TTL_DAY + mt_rand( 0, $cache::TTL_DAY ),
474 static function () use ( $wiki, $name ) {
477 return [
'v' =>
$wgConf->getConfig( $wiki, $name ) ];
479 [
'pcTTL' => WANObjectCache::TTL_PROC_LONG ]
491 foreach ( $jobs as
$job ) {
494 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.
getCachedConfigVar( $name)
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)
static singleton( $domain=false)
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.
static destroySingletons()
Destroy the singleton instances.
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.
GlobalIdGenerator $globalIdGenerator
static JobQueueGroup[] $instances
array $jobTypesExcludedFromDefaultQueue
getQueuesWithJobs()
Get the list of job types that have non-empty queues.
assertValidJobs(array $jobs)
getQueueTypes()
Get the list of queue types.
IBufferingStatsdDataFactory $statsdDataFactory
deduplicateRootJob(RunnableJob $job)
Register the "root job" of a given job into the queue for de-duplication.
array $jobTypeConfiguration
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.
static getWikiIdFromDbDomain( $domain)
Get the wiki ID of a database domain.
static isCurrentWikiDbDomain( $domain)
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