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
    }

    protected function doPurge() {
        return $this->cache->delete( $this->getReadyQueueCacheKey() );
    }

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