diff options
Diffstat (limited to 'includes/job/aggregator')
| -rw-r--r-- | includes/job/aggregator/JobQueueAggregator.php | 156 | ||||
| -rw-r--r-- | includes/job/aggregator/JobQueueAggregatorMemc.php | 124 | ||||
| -rw-r--r-- | includes/job/aggregator/JobQueueAggregatorRedis.php | 193 |
3 files changed, 0 insertions, 473 deletions
diff --git a/includes/job/aggregator/JobQueueAggregator.php b/includes/job/aggregator/JobQueueAggregator.php deleted file mode 100644 index a8186abd..00000000 --- a/includes/job/aggregator/JobQueueAggregator.php +++ /dev/null @@ -1,156 +0,0 @@ -<?php -/** - * Job queue aggregator code. - * - * This program is free software; you can redistribute it and/or modify - * it under the terms of the GNU General Public License as published by - * the Free Software Foundation; either version 2 of the License, or - * (at your option) any later version. - * - * This program is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - * GNU General Public License for more details. - * - * You should have received a copy of the GNU General Public License along - * with this program; if not, write to the Free Software Foundation, Inc., - * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. - * http://www.gnu.org/copyleft/gpl.html - * - * @file - * @author Aaron Schulz - */ - -/** - * Class to handle tracking information about all queues - * - * @ingroup JobQueue - * @since 1.21 - */ -abstract class JobQueueAggregator { - /** @var JobQueueAggregator */ - protected static $instance = null; - - /** - * @param array $params - */ - protected function __construct( array $params ) {} - - /** - * @return JobQueueAggregator - */ - final public static function singleton() { - global $wgJobQueueAggregator; - - if ( !isset( self::$instance ) ) { - $class = $wgJobQueueAggregator['class']; - $obj = new $class( $wgJobQueueAggregator ); - if ( !( $obj instanceof JobQueueAggregator ) ) { - throw new MWException( "Class '$class' is not a JobQueueAggregator class." ); - } - self::$instance = $obj; - } - - return self::$instance; - } - - /** - * Destroy the singleton instance - * - * @return void - */ - final public static function destroySingleton() { - self::$instance = null; - } - - /** - * Mark a queue as being empty - * - * @param string $wiki - * @param string $type - * @return bool Success - */ - final public function notifyQueueEmpty( $wiki, $type ) { - wfProfileIn( __METHOD__ ); - $ok = $this->doNotifyQueueEmpty( $wiki, $type ); - wfProfileOut( __METHOD__ ); - return $ok; - } - - /** - * @see JobQueueAggregator::notifyQueueEmpty() - */ - abstract protected function doNotifyQueueEmpty( $wiki, $type ); - - /** - * Mark a queue as being non-empty - * - * @param string $wiki - * @param string $type - * @return bool Success - */ - final public function notifyQueueNonEmpty( $wiki, $type ) { - wfProfileIn( __METHOD__ ); - $ok = $this->doNotifyQueueNonEmpty( $wiki, $type ); - wfProfileOut( __METHOD__ ); - return $ok; - } - - /** - * @see JobQueueAggregator::notifyQueueNonEmpty() - */ - abstract protected function doNotifyQueueNonEmpty( $wiki, $type ); - - /** - * Get the list of all of the queues with jobs - * - * @return Array (job type => (list of wiki IDs)) - */ - final public function getAllReadyWikiQueues() { - wfProfileIn( __METHOD__ ); - $res = $this->doGetAllReadyWikiQueues(); - wfProfileOut( __METHOD__ ); - return $res; - } - - /** - * @see JobQueueAggregator::getAllReadyWikiQueues() - */ - abstract protected function doGetAllReadyWikiQueues(); - - /** - * Purge all of the aggregator information - * - * @return bool Success - */ - final public function purge() { - wfProfileIn( __METHOD__ ); - $res = $this->doPurge(); - wfProfileOut( __METHOD__ ); - return $res; - } - - /** - * @see JobQueueAggregator::purge() - */ - abstract protected function doPurge(); - - /** - * Get all databases that have a pending job. - * This poll all the queues and is this expensive. - * - * @return Array (job type => (list of wiki IDs)) - */ - protected function findPendingWikiQueues() { - global $wgLocalDatabases; - - $pendingDBs = array(); // (job type => (db list)) - foreach ( $wgLocalDatabases as $db ) { - foreach ( JobQueueGroup::singleton( $db )->getQueuesWithJobs() as $type ) { - $pendingDBs[$type][] = $db; - } - } - - return $pendingDBs; - } -} diff --git a/includes/job/aggregator/JobQueueAggregatorMemc.php b/includes/job/aggregator/JobQueueAggregatorMemc.php deleted file mode 100644 index 9434da04..00000000 --- a/includes/job/aggregator/JobQueueAggregatorMemc.php +++ /dev/null @@ -1,124 +0,0 @@ -<?php -/** - * Job queue aggregator code that uses BagOStuff. - * - * This program is free software; you can redistribute it and/or modify - * it under the terms of the GNU General Public License as published by - * the Free Software Foundation; either version 2 of the License, or - * (at your option) any later version. - * - * This program is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - * GNU General Public License for more details. - * - * You should have received a copy of the GNU General Public License along - * with this program; if not, write to the Free Software Foundation, Inc., - * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. - * http://www.gnu.org/copyleft/gpl.html - * - * @file - * @author Aaron Schulz - */ - -/** - * Class to handle tracking information about all queues using BagOStuff - * - * @ingroup JobQueue - * @since 1.21 - */ -class JobQueueAggregatorMemc extends JobQueueAggregator { - /** @var BagOStuff */ - protected $cache; - - protected $cacheTTL; // integer; seconds - - /** - * @params include: - * - objectCache : Name of an object cache registered in $wgObjectCaches. - * This defaults to the one specified by $wgMainCacheType. - * - cacheTTL : Seconds to cache the aggregate data before regenerating. - * @param array $params - */ - 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 - } - - /** - * @see JobQueueAggregator::doNotifyQueueEmpty() - */ - 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; - } - - /** - * @see JobQueueAggregator::doNotifyQueueNonEmpty() - */ - protected function doNotifyQueueNonEmpty( $wiki, $type ) { - return true; // updated periodically - } - - /** - * @see JobQueueAggregator::doAllGetReadyWikiQueues() - */ - 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 - } - - /** - * @see JobQueueAggregator::doPurge() - */ - protected function doPurge() { - return $this->cache->delete( $this->getReadyQueueCacheKey() ); - } - - /** - * @return string - */ - private function getReadyQueueCacheKey() { - return "jobqueue:aggregator:ready-queues:v1"; // global - } -} diff --git a/includes/job/aggregator/JobQueueAggregatorRedis.php b/includes/job/aggregator/JobQueueAggregatorRedis.php deleted file mode 100644 index c6a799df..00000000 --- a/includes/job/aggregator/JobQueueAggregatorRedis.php +++ /dev/null @@ -1,193 +0,0 @@ -<?php -/** - * Job queue aggregator code that uses PhpRedis. - * - * This program is free software; you can redistribute it and/or modify - * it under the terms of the GNU General Public License as published by - * the Free Software Foundation; either version 2 of the License, or - * (at your option) any later version. - * - * This program is distributed in the hope that it will be useful, - * but WITHOUT ANY WARRANTY; without even the implied warranty of - * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the - * GNU General Public License for more details. - * - * You should have received a copy of the GNU General Public License along - * with this program; if not, write to the Free Software Foundation, Inc., - * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA. - * http://www.gnu.org/copyleft/gpl.html - * - * @file - * @author Aaron Schulz - */ - -/** - * Class to handle tracking information about all queues using PhpRedis - * - * @ingroup JobQueue - * @ingroup Redis - * @since 1.21 - */ -class JobQueueAggregatorRedis extends JobQueueAggregator { - /** @var RedisConnectionPool */ - protected $redisPool; - - /** - * @params include: - * - redisConfig : An array of parameters to RedisConnectionPool::__construct(). - * - redisServer : A hostname/port combination or the absolute path of a UNIX socket. - * If a hostname is specified but no port, the standard port number - * 6379 will be used. Required. - * @param array $params - */ - protected function __construct( array $params ) { - parent::__construct( $params ); - $this->server = $params['redisServer']; - $this->redisPool = RedisConnectionPool::singleton( $params['redisConfig'] ); - } - - /** - * @see JobQueueAggregator::doNotifyQueueEmpty() - */ - protected function doNotifyQueueEmpty( $wiki, $type ) { - $conn = $this->getConnection(); - if ( !$conn ) { - return false; - } - try { - $conn->hDel( $this->getReadyQueueKey(), $this->encQueueName( $type, $wiki ) ); - return true; - } catch ( RedisException $e ) { - $this->handleException( $conn, $e ); - return false; - } - } - - /** - * @see JobQueueAggregator::doNotifyQueueNonEmpty() - */ - protected function doNotifyQueueNonEmpty( $wiki, $type ) { - $conn = $this->getConnection(); - if ( !$conn ) { - return false; - } - try { - $conn->hSet( $this->getReadyQueueKey(), $this->encQueueName( $type, $wiki ), time() ); - return true; - } catch ( RedisException $e ) { - $this->handleException( $conn, $e ); - return false; - } - } - - /** - * @see JobQueueAggregator::doAllGetReadyWikiQueues() - */ - protected function doGetAllReadyWikiQueues() { - $conn = $this->getConnection(); - if ( !$conn ) { - return array(); - } - try { - $conn->multi( Redis::PIPELINE ); - $conn->exists( $this->getReadyQueueKey() ); - $conn->hGetAll( $this->getReadyQueueKey() ); - list( $exists, $map ) = $conn->exec(); - - if ( $exists ) { // cache hit - $pendingDBs = array(); // (type => list of wikis) - foreach ( $map as $key => $time ) { - list( $type, $wiki ) = $this->dencQueueName( $key ); - $pendingDBs[$type][] = $wiki; - } - } else { // cache miss - // Avoid duplicated effort - $conn->multi( Redis::MULTI ); - $conn->setnx( $this->getReadyQueueKey() . ":lock", 1 ); - $conn->expire( $this->getReadyQueueKey() . ":lock", 3600 ); - if ( $conn->exec() !== array( true, true ) ) { // lock - return array(); // already in progress - } - - $pendingDBs = $this->findPendingWikiQueues(); // (type => list of wikis) - - $conn->delete( $this->getReadyQueueKey() . ":lock" ); // unlock - - $now = time(); - $map = array(); - foreach ( $pendingDBs as $type => $wikis ) { - foreach ( $wikis as $wiki ) { - $map[$this->encQueueName( $type, $wiki )] = $now; - } - } - $conn->hMSet( $this->getReadyQueueKey(), $map ); - } - - return $pendingDBs; - } catch ( RedisException $e ) { - $this->handleException( $conn, $e ); - return array(); - } - } - - /** - * @see JobQueueAggregator::doPurge() - */ - protected function doPurge() { - $conn = $this->getConnection(); - if ( !$conn ) { - return false; - } - try { - $conn->delete( $this->getReadyQueueKey() ); - } catch ( RedisException $e ) { - $this->handleException( $conn, $e ); - return false; - } - return true; - } - - /** - * Get a connection to the server that handles all sub-queues for this queue - * - * @return Array (server name, Redis instance) - * @throws MWException - */ - protected function getConnection() { - return $this->redisPool->getConnection( $this->server ); - } - - /** - * @param RedisConnRef $conn - * @param RedisException $e - * @return void - */ - protected function handleException( RedisConnRef $conn, $e ) { - $this->redisPool->handleException( $this->server, $conn, $e ); - } - - /** - * @return string - */ - private function getReadyQueueKey() { - return "jobqueue:aggregator:h-ready-queues:v1"; // global - } - - /** - * @param string $type - * @param string $wiki - * @return string - */ - private function encQueueName( $type, $wiki ) { - return rawurlencode( $type ) . '/' . rawurlencode( $wiki ); - } - - /** - * @param string $name - * @return string - */ - private function dencQueueName( $name ) { - list( $type, $wiki ) = explode( '/', $name, 2 ); - return array( rawurldecode( $type ), rawurldecode( $wiki ) ); - } -} |