46 private $localJobClasses;
48 private $jobTypeConfiguration;
50 private $jobTypesExcludedFromDefaultQueue;
52 private $statsdDataFactory;
56 private $globalIdGenerator;
62 private const TYPE_ANY = 2;
66 private const PROC_CACHE_TTL = 15;
84 ?array $localJobClasses,
85 array $jobTypeConfiguration,
86 array $jobTypesExcludedFromDefaultQueue,
91 $this->domain = $domain;
92 $this->readOnlyMode = $readOnlyMode;
94 $this->localJobClasses = $localJobClasses;
95 $this->jobTypeConfiguration = $jobTypeConfiguration;
96 $this->jobTypesExcludedFromDefaultQueue = $jobTypesExcludedFromDefaultQueue;
97 $this->statsdDataFactory = $statsdDataFactory;
98 $this->wanCache = $wanCache;
99 $this->globalIdGenerator = $globalIdGenerator;
108 public function get( $type ) {
109 $conf = [
'domain' => $this->domain,
'type' => $type ];
110 $conf += $this->jobTypeConfiguration[$type] ?? $this->jobTypeConfiguration[
'default'];
111 if ( !isset( $conf[
'readOnlyReason'] ) ) {
112 $conf[
'readOnlyReason'] = $this->readOnlyMode->getConfiguredReason();
115 $conf[
'stats'] = $this->statsdDataFactory;
116 $conf[
'wanCache'] = $this->wanCache;
117 $conf[
'idGenerator'] = $this->globalIdGenerator;
131 public function push( $jobs ) {
132 $jobs = is_array( $jobs ) ? $jobs : [ $jobs ];
133 if ( $jobs === [] ) {
137 $this->assertValidJobs( $jobs );
140 foreach ( $jobs as
$job ) {
141 $type =
$job->getType();
142 if ( isset( $this->jobTypeConfiguration[$type] ) ) {
143 $jobsByType[$type][] =
$job;
146 isset( $this->jobTypeConfiguration[
'default'][
'typeAgnostic'] ) &&
147 $this->jobTypeConfiguration[
'default'][
'typeAgnostic']
149 $jobsByType[
'default'][] =
$job;
151 $jobsByType[$type][] =
$job;
156 foreach ( $jobsByType as $type => $jobs ) {
157 $this->
get( $type )->
push( $jobs );
160 if ( $this->cache->hasField(
'queues-ready',
'list' ) ) {
161 $list = $this->cache->getField(
'queues-ready',
'list' );
162 if ( count( array_diff( array_keys( $jobsByType ), $list ) ) ) {
163 $this->cache->clear(
'queues-ready' );
167 $cache = MediaWikiServices::getInstance()->getObjectCacheFactory()->getLocalClusterInstance();
169 $cache->makeGlobalKey(
'jobqueue', $this->domain,
'hasjobs', self::TYPE_ANY ),
173 if ( array_diff( array_keys( $jobsByType ), $this->jobTypesExcludedFromDefaultQueue ) ) {
175 $cache->makeGlobalKey(
'jobqueue', $this->domain,
'hasjobs', self::TYPE_DEFAULT ),
190 if ( PHP_SAPI ===
'cli' || PHP_SAPI ===
'phpdbg' ) {
191 $this->
push( $jobs );
195 $jobs = is_array( $jobs ) ? $jobs : [ $jobs ];
198 $this->assertValidJobs( $jobs );
217 public function pop( $qtype = self::TYPE_DEFAULT, $flags = 0, array $ignored = [] ) {
220 if ( !$this->localJobClasses ) {
222 "Cannot pop '{$qtype}' job off foreign '{$this->domain}' wiki queue." );
224 if ( is_string( $qtype ) && !isset( $this->localJobClasses[$qtype] ) ) {
226 throw new JobQueueError(
"Unrecognized job type '$qtype'." );
229 if ( is_string( $qtype ) ) {
230 if ( !in_array( $qtype, $ignored ) ) {
231 $job = $this->
get( $qtype )->
pop();
234 if ( $flags & self::USE_CACHE ) {
235 if ( !$this->cache->hasField(
'queues-ready',
'list', self::PROC_CACHE_TTL ) ) {
238 $types = $this->cache->getField(
'queues-ready',
'list' );
243 if ( $qtype == self::TYPE_DEFAULT ) {
247 $types = array_diff( $types, $ignored );
250 foreach ( $types as $type ) {
251 $job = $this->
get( $type )->
pop();
255 $this->cache->clear(
'queues-ready' );
298 foreach ( $this->jobTypeConfiguration as $type => $conf ) {
311 if ( !$this->localJobClasses ) {
312 throw new JobQueueError(
'Cannot inspect job queue from foreign wiki' );
314 return array_keys( $this->localJobClasses );
325 return array_diff( $this->
getQueueTypes(), $this->jobTypesExcludedFromDefaultQueue );
338 $cache = MediaWikiServices::getInstance()->getObjectCacheFactory()->getLocalClusterInstance();
339 $key = $cache->makeGlobalKey(
'jobqueue', $this->domain,
'hasjobs', $type );
341 $value = $cache->
get( $key );
342 if ( $value ===
false ) {
344 if ( $type == self::TYPE_DEFAULT ) {
347 $value = count( $queues ) ?
'true' :
'false';
348 $cache->add( $key, $value, 15 );
351 return ( $value ===
'true' );
365 $queue = $info[
'queue'];
366 $nonEmpty = $queue->getSiblingQueuesWithJobs( $this->
getQueueTypes() );
367 if ( is_array( $nonEmpty ) ) {
368 $types = array_merge( $types, $nonEmpty );
370 foreach ( $info[
'types'] as $type ) {
371 if ( !$this->
get( $type )->isEmpty() ) {
392 $queue = $info[
'queue'];
393 $sizes = $queue->getSiblingQueueSizes( $this->
getQueueTypes() );
394 if ( is_array( $sizes ) ) {
397 foreach ( $info[
'types'] as $type ) {
398 $sizeMap[$type] = $this->
get( $type )->getSize();
411 if ( $this->coalescedQueues ===
null ) {
412 $this->coalescedQueues = [];
413 foreach ( $this->jobTypeConfiguration as $type => $conf ) {
414 $conf[
'domain'] = $this->domain;
415 $conf[
'type'] =
'null';
416 $conf[
'stats'] = $this->statsdDataFactory;
417 $conf[
'wanCache'] = $this->wanCache;
418 $conf[
'idGenerator'] = $this->globalIdGenerator;
421 $loc = $queue->getCoalesceLocationInternal();
422 if ( !isset( $this->coalescedQueues[$loc] ) ) {
423 $this->coalescedQueues[$loc][
'queue'] = $queue;
424 $this->coalescedQueues[$loc][
'types'] = [];
426 if ( $type ===
'default' ) {
427 $this->coalescedQueues[$loc][
'types'] = array_merge(
428 $this->coalescedQueues[$loc][
'types'],
429 array_diff( $this->
getQueueTypes(), array_keys( $this->jobTypeConfiguration ) )
432 $this->coalescedQueues[$loc][
'types'][] = $type;
437 return $this->coalescedQueues;
444 private function assertValidJobs( array $jobs ) {
445 foreach ( $jobs as
$job ) {
447 $type = is_object(
$job ) ? get_class(
$job ) : gettype(
$job );
448 throw new InvalidArgumentException(
"Expected IJobSpecification objects, got " . $type );
wfDeprecated( $function, $version=false, $component=false, $callerOffset=2)
Logs a warning that a deprecated feature was used.
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.
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 one job off a job queue.
ack(RunnableJob $job)
Acknowledge that a job was completed.
queuesHaveJobs( $type=self::TYPE_ANY)
Check if there are any queues with jobs (this is cached)
ReadOnlyMode $readOnlyMode
Read only mode.
__construct( $domain, ReadOnlyMode $readOnlyMode, ?array $localJobClasses, array $jobTypeConfiguration, array $jobTypesExcludedFromDefaultQueue, IBufferingStatsdDataFactory $statsdDataFactory, WANObjectCache $wanCache, GlobalIdGenerator $globalIdGenerator)
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.
Store key-value entries in a size-limited in-memory LRU cache.
set( $key, $value, $rank=self::RANK_TOP)
Set a key/value pair.
get( $key, $maxAge=INF, $default=null)
Get the value for a key.
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