JobQueueGroup.php

Go to the documentation of this file.

<?php
class JobQueueGroup {
        protected static $instances = array();

        protected $cache;

        protected $wiki; // string; wiki ID

        const TYPE_DEFAULT = 1; // integer; jobs popped by default
        const TYPE_ANY = 2; // integer; any job

        const USE_CACHE = 1; // integer; use process or persistent cache

        const PROC_CACHE_TTL = 15; // integer; seconds

        const CACHE_VERSION = 1; // integer; cache version

        protected function __construct( $wiki ) {
                $this->wiki = $wiki;
                $this->cache = new ProcessCacheLRU( 10 );
        }

        public static function singleton( $wiki = false ) {
                $wiki = ( $wiki === false ) ? wfWikiID() : $wiki;
                if ( !isset( self::$instances[$wiki] ) ) {
                        self::$instances[$wiki] = new self( $wiki );
                }
                return self::$instances[$wiki];
        }

        public static function destroySingletons() {
                self::$instances = array();
        }

        public function get( $type ) {
                global $wgJobTypeConf;

                $conf = array( 'wiki' => $this->wiki, 'type' => $type );
                if ( isset( $wgJobTypeConf[$type] ) ) {
                        $conf = $conf + $wgJobTypeConf[$type];
                } else {
                        $conf = $conf + $wgJobTypeConf['default'];
                }

                return JobQueue::factory( $conf );
        }

        public function push( $jobs ) {
                $jobs = is_array( $jobs ) ? $jobs : array( $jobs );

                $jobsByType = array(); // (job type => list of jobs)
                foreach ( $jobs as $job ) {
                        if ( $job instanceof Job ) {
                                $jobsByType[$job->getType()][] = $job;
                        } else {
                                throw new MWException( "Attempted to push a non-Job object into a queue." );
                        }
                }

                $ok = true;
                foreach ( $jobsByType as $type => $jobs ) {
                        if ( $this->get( $type )->push( $jobs ) ) {
                                JobQueueAggregator::singleton()->notifyQueueNonEmpty( $this->wiki, $type );
                        } else {
                                $ok = false;
                        }
                }

                if ( $this->cache->has( 'queues-ready', 'list' ) ) {
                        $list = $this->cache->get( 'queues-ready', 'list' );
                        if ( count( array_diff( array_keys( $jobsByType ), $list ) ) ) {
                                $this->cache->clear( 'queues-ready' );
                        }
                }

                return $ok;
        }

        public function pop( $qtype = self::TYPE_DEFAULT, $flags = 0 ) {
                if ( is_string( $qtype ) ) { // specific job type
                        $job = $this->get( $qtype )->pop();
                        if ( !$job ) {
                                JobQueueAggregator::singleton()->notifyQueueEmpty( $this->wiki, $qtype );
                        }
                        return $job;
                } else { // any job in the "default" jobs types
                        if ( $flags & self::USE_CACHE ) {
                                if ( !$this->cache->has( 'queues-ready', 'list', self::PROC_CACHE_TTL ) ) {
                                        $this->cache->set( 'queues-ready', 'list', $this->getQueuesWithJobs() );
                                }
                                $types = $this->cache->get( 'queues-ready', 'list' );
                        } else {
                                $types = $this->getQueuesWithJobs();
                        }

                        if ( $qtype == self::TYPE_DEFAULT ) {
                                $types = array_intersect( $types, $this->getDefaultQueueTypes() );
                        }
                        shuffle( $types ); // avoid starvation

                        foreach ( $types as $type ) { // for each queue...
                                $job = $this->get( $type )->pop();
                                if ( $job ) { // found
                                        return $job;
                                } else { // not found
                                        JobQueueAggregator::singleton()->notifyQueueEmpty( $this->wiki, $type );
                                        $this->cache->clear( 'queues-ready' );
                                }
                        }

                        return false; // no jobs found
                }
        }

        public function ack( Job $job ) {
                return $this->get( $job->getType() )->ack( $job );
        }

        public function deduplicateRootJob( Job $job ) {
                return $this->get( $job->getType() )->deduplicateRootJob( $job );
        }

        public function waitForBackups() {
                global $wgJobTypeConf;

                wfProfileIn( __METHOD__ );
                // Try to avoid doing this more than once per queue storage medium
                foreach ( $wgJobTypeConf as $type => $conf ) {
                        $this->get( $type )->waitForBackups();
                }
                wfProfileOut( __METHOD__ );
        }

        public function getQueueTypes() {
                return array_keys( $this->getCachedConfigVar( 'wgJobClasses' ) );
        }

        public function getDefaultQueueTypes() {
                global $wgJobTypesExcludedFromDefaultQueue;

                return array_diff( $this->getQueueTypes(), $wgJobTypesExcludedFromDefaultQueue );
        }

        public function getQueuesWithJobs() {
                $types = array();
                foreach ( $this->getQueueTypes() as $type ) {
                        if ( !$this->get( $type )->isEmpty() ) {
                                $types[] = $type;
                        }
                }
                return $types;
        }

        public function isQueueDeprioritized( $type ) {
                if ( $type === 'refreshLinks2' ) {
                        // Don't keep converting refreshLinks2 => refreshLinks jobs if the
                        // later jobs have not been done yet. This helps throttle queue spam.
                        return !$this->get( 'refreshLinks' )->isEmpty();
                }
                return false;
        }

        public function executeReadyPeriodicTasks() {
                global $wgMemc;

                list( $db, $prefix ) = wfSplitWikiID( $this->wiki );
                $key = wfForeignMemcKey( $db, $prefix, 'jobqueuegroup', 'taskruns', 'v1' );
                $lastRuns = $wgMemc->get( $key ); // (queue => task => UNIX timestamp)

                $count = 0;
                $tasksRun = array(); // (queue => task => UNIX timestamp)
                foreach ( $this->getQueueTypes() as $type ) {
                        $queue = $this->get( $type );
                        foreach ( $queue->getPeriodicTasks() as $task => $definition ) {
                                if ( $definition['period'] <= 0 ) {
                                        continue; // disabled
                                } elseif ( !isset( $lastRuns[$type][$task] )
                                        || $lastRuns[$type][$task] < ( time() - $definition['period'] ) )
                                {
                                        if ( call_user_func( $definition['callback'] ) !== null ) {
                                                $tasksRun[$type][$task] = time();
                                                ++$count;
                                        }
                                }
                        }
                }

                $wgMemc->merge( $key, function( $cache, $key, $lastRuns ) use ( $tasksRun ) {
                        if ( is_array( $lastRuns ) ) {
                                foreach ( $tasksRun as $type => $tasks ) {
                                        foreach ( $tasks as $task => $timestamp ) {
                                                if ( !isset( $lastRuns[$type][$task] )
                                                        || $timestamp > $lastRuns[$type][$task] )
                                                {
                                                        $lastRuns[$type][$task] = $timestamp;
                                                }
                                        }
                                }
                        } else {
                                $lastRuns = $tasksRun;
                        }
                        return $lastRuns;
                } );

                return $count;
        }

        private function getCachedConfigVar( $name ) {
                global $wgConf, $wgMemc;

                if ( $this->wiki === wfWikiID() ) {
                        return $GLOBALS[$name]; // common case
                } else {
                        list( $db, $prefix ) = wfSplitWikiID( $this->wiki );
                        $key = wfForeignMemcKey( $db, $prefix, 'configvalue', $name );
                        $value = $wgMemc->get( $key ); // ('v' => ...) or false
                        if ( is_array( $value ) ) {
                                return $value['v'];
                        } else {
                                $value = $wgConf->getConfig( $this->wiki, $name );
                                $wgMemc->set( $key, array( 'v' => $value ), 86400 + mt_rand( 0, 86400 ) );
                                return $value;
                        }
                }
        }
}