JobQueueAggregatorMemc.php

Go to the documentation of this file.

<?php
class JobQueueAggregatorMemc extends JobQueueAggregator {
        protected $cache;

        protected $cacheTTL; // integer; seconds

        protected function __construct( array $params ) {
                parent::__construct( $params );
                $this->cache = isset( $params['objectCache'] )
                        ? wfGetCache( $params['objectCache'] )
                        : wfGetMainCache();
                $this->cacheTTL = isset( $params['cacheTTL'] ) ? $params['cacheTTL'] : 180; // 3 min
        }

        protected function doNotifyQueueEmpty( $wiki, $type ) {
                $key = $this->getReadyQueueCacheKey();
                // Delist the queue from the "ready queue" list
                if ( $this->cache->add( "$key:lock", 1, 60 ) ) { // lock
                        $curInfo = $this->cache->get( $key );
                        if ( is_array( $curInfo ) && isset( $curInfo['pendingDBs'][$type] ) ) {
                                if ( in_array( $wiki, $curInfo['pendingDBs'][$type] ) ) {
                                        $curInfo['pendingDBs'][$type] = array_diff(
                                                $curInfo['pendingDBs'][$type], array( $wiki ) );
                                        $this->cache->set( $key, $curInfo );
                                }
                        }
                        $this->cache->delete( "$key:lock" ); // unlock
                }
                return true;
        }

        protected function doNotifyQueueNonEmpty( $wiki, $type ) {
                return true; // updated periodically
        }

        protected function doGetAllReadyWikiQueues() {
                $key = $this->getReadyQueueCacheKey();
                // If the cache entry wasn't present, is stale, or in .1% of cases otherwise,
                // regenerate the cache. Use any available stale cache if another process is
                // currently regenerating the pending DB information.
                $pendingDbInfo = $this->cache->get( $key );
                if ( !is_array( $pendingDbInfo )
                        || ( time() - $pendingDbInfo['timestamp'] ) > $this->cacheTTL
                        || mt_rand( 0, 999 ) == 0
                ) {
                        if ( $this->cache->add( "$key:rebuild", 1, 1800 ) ) { // lock
                                $pendingDbInfo = array(
                                        'pendingDBs' => $this->findPendingWikiQueues(),
                                        'timestamp'  => time()
                                );
                                for ( $attempts=1; $attempts <= 25; ++$attempts ) {
                                        if ( $this->cache->add( "$key:lock", 1, 60 ) ) { // lock
                                                $this->cache->set( $key, $pendingDbInfo );
                                                $this->cache->delete( "$key:lock" ); // unlock
                                                break;
                                        }
                                }
                                $this->cache->delete( "$key:rebuild" ); // unlock
                        }
                }
                return is_array( $pendingDbInfo )
                        ? $pendingDbInfo['pendingDBs']
                        : array(); // cache is both empty and locked
        }

        private function getReadyQueueCacheKey() {
                return "jobqueue:aggregator:ready-queues:v1"; // global
        }
}