77 if ( !isset( self::$instances[
$wiki] ) ) {
82 self::$instances[
$wiki]->invalidWiki =
true;
86 return self::$instances[
$wiki];
95 self::$instances = [];
114 if ( $this->readOnlyReason !==
false ) {
131 public function push( $jobs ) {
134 if ( $this->invalidWiki ) {
136 $e =
new LogicException(
"Domain '{$this->wiki}' is not recognized." );
137 MWExceptionHandler::logException(
$e );
141 $jobs = is_array( $jobs ) ? $jobs : [ $jobs ];
142 if ( !count( $jobs ) ) {
149 foreach ( $jobs as
$job ) {
150 $jobsByType[
$job->getType()][] =
$job;
153 foreach ( $jobsByType as
$type => $jobs ) {
157 if ( $this->
cache->has(
'queues-ready',
'list' ) ) {
158 $list = $this->
cache->get(
'queues-ready',
'list' );
159 if ( count( array_diff( array_keys( $jobsByType ), $list ) ) ) {
160 $this->
cache->clear(
'queues-ready' );
164 $cache = ObjectCache::getLocalClusterInstance();
166 $cache->makeGlobalKey(
'jobqueue', $this->wiki,
'hasjobs', self::TYPE_ANY ),
172 $cache->makeGlobalKey(
'jobqueue', $this->wiki,
'hasjobs', self::TYPE_DEFAULT ),
191 if ( $this->invalidWiki ) {
193 throw new LogicException(
"Domain '{$this->wiki}' is not recognized." );
196 if ( PHP_SAPI ===
'cli' || PHP_SAPI ===
'phpdbg' ) {
197 $this->
push( $jobs );
201 $jobs = is_array( $jobs ) ? $jobs : [ $jobs ];
206 $this->bufferedJobs = array_merge( $this->bufferedJobs, $jobs );
216 foreach ( self::$instances as $group ) {
218 $group->push( $group->bufferedJobs );
219 $group->bufferedJobs = [];
220 }
catch ( Exception
$e ) {
222 MWExceptionHandler::logException(
$e );
238 public function pop( $qtype = self::TYPE_DEFAULT,
$flags = 0, array $blacklist = [] ) {
241 if ( is_string( $qtype ) ) {
242 if ( !in_array( $qtype, $blacklist ) ) {
243 $job = $this->
get( $qtype )->
pop();
246 if (
$flags & self::USE_CACHE ) {
247 if ( !$this->
cache->has(
'queues-ready',
'list', self::PROC_CACHE_TTL ) ) {
250 $types = $this->
cache->get(
'queues-ready',
'list' );
255 if ( $qtype == self::TYPE_DEFAULT ) {
259 $types = array_diff( $types, $blacklist );
262 foreach ( $types as
$type ) {
267 $this->
cache->clear(
'queues-ready' );
329 return array_diff( $this->
getQueueTypes(), $wgJobTypesExcludedFromDefaultQueue );
340 $cache = ObjectCache::getLocalClusterInstance();
341 $key =
$cache->makeGlobalKey(
'jobqueue', $this->wiki,
'hasjobs',
$type );
346 if (
$type == self::TYPE_DEFAULT ) {
349 $value = count( $queues ) ?
'true' :
'false';
353 return (
$value ===
'true' );
364 $nonEmpty = $info[
'queue']->getSiblingQueuesWithJobs( $this->
getQueueTypes() );
365 if ( is_array( $nonEmpty ) ) {
366 $types = array_merge( $types, $nonEmpty );
368 foreach ( $info[
'types'] as
$type ) {
369 if ( !$this->
get( $type )->isEmpty() ) {
387 $sizes = $info[
'queue']->getSiblingQueueSizes( $this->
getQueueTypes() );
388 if ( is_array( $sizes ) ) {
389 $sizeMap = $sizeMap + $sizes;
391 foreach ( $info[
'types'] as
$type ) {
392 $sizeMap[
$type] = $this->
get(
$type )->getSize();
406 if ( $this->coalescedQueues ===
null ) {
407 $this->coalescedQueues = [];
410 [
'wiki' => $this->wiki,
'type' =>
'null' ] + $conf );
411 $loc =
$queue->getCoalesceLocationInternal();
412 if ( !isset( $this->coalescedQueues[$loc] ) ) {
413 $this->coalescedQueues[$loc][
'queue'] =
$queue;
414 $this->coalescedQueues[$loc][
'types'] = [];
416 if (
$type ===
'default' ) {
417 $this->coalescedQueues[$loc][
'types'] = array_merge(
418 $this->coalescedQueues[$loc][
'types'],
422 $this->coalescedQueues[$loc][
'types'][] =
$type;
440 $cache = ObjectCache::getMainWANInstance();
442 $cache->makeGlobalKey(
'jobqueue',
'configvalue',
$wiki, $name ),
443 $cache::TTL_DAY + mt_rand( 0, $cache::TTL_DAY ),
444 function () use (
$wiki, $name ) {
449 [
'pcTTL' => WANObjectCache::TTL_PROC_LONG ]
461 foreach ( $jobs as
$job ) {
463 throw new InvalidArgumentException(
"Expected IJobSpecification objects" );
469 $n = count( $this->bufferedJobs );
471 $type = implode(
', ', array_unique( array_map(
'get_class', $this->bufferedJobs ) ) );
472 trigger_error( __METHOD__ .
": $n buffered job(s) of type(s) $type never inserted." );
$GLOBALS['wgAutoloadClasses']['LocalisationUpdate']
$wgJobTypeConf
Map of job types to configuration arrays.
$wgLocalDatabases
Other wikis on this site, can be administered from a single developer account.
$wgJobTypesExcludedFromDefaultQueue
Jobs that must be explicitly requested, i.e.
$wgConf
wgConf hold the site configuration.
wfConfiguredReadOnlyReason()
Get the value of $wgReadOnly or the contents of $wgReadOnlyFile.
wfWikiID()
Get an ASCII string identifying this wiki This is used as a prefix in memcached keys.
Class to handle enqueueing of background jobs.
getCachedConfigVar( $name)
push( $jobs)
Insert jobs into the respective queues of which they belong.
getQueueSizes()
Get the size of the queus for a list of job types.
ack(Job $job)
Acknowledge that a job was completed.
static pushLazyJobs()
Push all jobs buffered via lazyPush() into their respective queues.
waitForBackups()
Wait for any replica DBs or backup queue servers to catch up.
deduplicateRootJob(Job $job)
Register the "root job" of a given job into the queue for de-duplication.
array $coalescedQueues
Map of (bucket => (queue => JobQueue, types => list of types)
__construct( $wiki, $readOnlyReason)
bool $invalidWiki
Whether the wiki is not recognized in configuration.
string bool $readOnlyReason
Read only rationale (or false if r/w)
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 $blacklist=[])
Pop a job off one of the job queues.
static destroySingletons()
Destroy the singleton instances.
queuesHaveJobs( $type=self::TYPE_ANY)
Check if there are any queues with jobs (this is cached)
static JobQueueGroup[] $instances
static singleton( $wiki=false)
getQueuesWithJobs()
Get the list of job types that have non-empty queues.
assertValidJobs(array $jobs)
getQueueTypes()
Get the list of queue types.
static factory(array $params)
Get a job queue object of the specified type.
Class to both describe a background job and handle jobs.
Handles per process caching of items.
get( $key, $prop)
Get a property field for a cache entry.
set( $key, $prop, $value)
Set a property field for a cache entry.
it s the revision text itself In either if gzip is the revision text is gzipped $flags
Allows to change the fields on the form that will be generated $name
processing should stop and the error should be shown to the user * false
returning false will NOT prevent logging $e
Job queue task description interface.
you have access to all of the normal MediaWiki so you can get a DB use the cache
if(count( $args)< 1) $job