From eb03cf3f7dda40c16627d41304b9abf17fcb7b44 Mon Sep 17 00:00:00 2001 From: Aaron Schulz Date: Sat, 15 Mar 2014 14:17:16 -0700 Subject: [PATCH] Added a Redis pool counter class * This should be easier to set up for typical installs Change-Id: Icb4a7481b944fa0818c4635e3edbe12d08af9924 --- RELEASE-NOTES-1.23 | 2 + includes/AutoLoader.php | 1 + includes/PoolCounterRedis.php | 412 ++++++++++++++++++++++++++++++ languages/i18n/en.json | 1 + languages/i18n/qqq.json | 1 + maintenance/language/messages.inc | 1 + 6 files changed, 418 insertions(+) create mode 100644 includes/PoolCounterRedis.php diff --git a/RELEASE-NOTES-1.23 b/RELEASE-NOTES-1.23 index 393b2687ee..d430a4d788 100644 --- a/RELEASE-NOTES-1.23 +++ b/RELEASE-NOTES-1.23 @@ -140,6 +140,8 @@ production. * Add new hook ChangesListInitRows accessed via ChangesList::initChangesListRows. If called by the ChangesList consumer this gives extensions a chance to batch process the result set prior to rendering. +* A PoolCounterRedis class was added which can be make use of in $wgPoolCounterConf. + This requires at least one Redis 2.6+ server. === Bug fixes in 1.23 === * (bug 41759) The "updated since last visit" markers (on history pages, recent diff --git a/includes/AutoLoader.php b/includes/AutoLoader.php index 8428adecf0..d9bb4bf613 100644 --- a/includes/AutoLoader.php +++ b/includes/AutoLoader.php @@ -157,6 +157,7 @@ $wgAutoloadLocalClasses = array( 'PhpHttpRequest' => 'includes/HttpFunctions.php', 'PoolCounter' => 'includes/PoolCounter.php', 'PoolCounter_Stub' => 'includes/PoolCounter.php', + 'PoolCounterRedis' => 'includes/PoolCounterRedis.php', 'PoolCounterWork' => 'includes/PoolCounter.php', 'PoolCounterWorkViaCallback' => 'includes/PoolCounter.php', 'PoolWorkArticleView' => 'includes/WikiPage.php', diff --git a/includes/PoolCounterRedis.php b/includes/PoolCounterRedis.php new file mode 100644 index 0000000000..1bc10f21f4 --- /dev/null +++ b/includes/PoolCounterRedis.php @@ -0,0 +1,412 @@ + host) map */ + protected $serversByLabel; + /** @var string SHA-1 of the key */ + protected $keySha1; + /** @var integer TTL for locks to expire (work should finish in this time) */ + protected $lockTTL; + + /** @var RedisConnRef */ + protected $conn; + /** @var string Pool slot value */ + protected $slot; + /** @var integer AWAKE_* constant */ + protected $onRelease; + /** @var string Unique string to identify this process */ + protected $session; + /** @var integer UNIX timestamp */ + protected $slotTime; + + const AWAKE_ONE = 1; // wake-up if when a slot can be taken from an existing process + const AWAKE_ALL = 2; // wake-up if an existing process finishes and wake up such others + + /** @var Array List of active PoolCounterRedis objects in this script */ + protected static $active = null; + + function __construct( $conf, $type, $key ) { + parent::__construct( $conf, $type, $key ); + + $this->serversByLabel = $conf['servers']; + $this->ring = new HashRing( array_fill_keys( array_keys( $conf['servers'] ), 100 ) ); + + $conf['redisConfig']['serializer'] = 'none'; // for use with Lua + $this->pool = RedisConnectionPool::singleton( $conf['redisConfig'] ); + + $this->keySha1 = sha1( $this->key ); + $met = ini_get( 'max_execution_time' ); // usually 0 in CLI mode + $this->lockTTL = $met ? 2*$met : 3600; + + if ( self::$active === null ) { + self::$active = array(); + register_shutdown_function( array( __CLASS__, 'releaseAll' ) ); + } + } + + /** + * @return Status Uses RediConnRef as value on success + */ + protected function getConnection() { + if ( !isset( $this->conn ) ) { + $conn = false; + $servers = $this->ring->getLocations( $this->key, 3 ); + ArrayUtils::consistentHashSort( $servers, $this->key ); + foreach ( $servers as $server ) { + $conn = $this->pool->getConnection( $this->serversByLabel[$server] ); + if ( $conn ) { + break; + } + } + if ( !$conn ) { + return Status::newFatal( 'pool-servererror', implode( ', ', $servers ) ); + } + $this->conn = $conn; + } + return Status::newGood( $this->conn ); + } + + function acquireForMe() { + $section = new ProfileSection( __METHOD__ ); + + return $this->waitForSlotOrNotif( self::AWAKE_ONE ); + } + + function acquireForAnyone() { + $section = new ProfileSection( __METHOD__ ); + + return $this->waitForSlotOrNotif( self::AWAKE_ALL ); + } + + function release() { + $section = new ProfileSection( __METHOD__ ); + + if ( $this->slot === null ) { + return Status::newGood( PoolCounter::NOT_LOCKED ); // not locked + } + + $status = $this->getConnection(); + if ( !$status->isOK() ) { + return $status; + } + $conn = $status->value; + + static $script = +<<= (1*rMaxWorkers - 1) then + -- Clear list to save space; it will re-init as needed + redis.call('del',kSlots,kSlotsNextRelease) + else + -- Add slot back to pool and update the "next release" time + redis.call('rPush',kSlots,rSlot) + redis.call('zAdd',kSlotsNextRelease,rTime + 30,rSlot) + -- Always keep renewing the expiry on use + redis.call('expireAt',kSlots,math.ceil(rTime + rExpiry)) + redis.call('expireAt',kSlotsNextRelease,math.ceil(rTime + rExpiry)) + end + end + -- Update an ephemeral list to wake up other clients that can + -- reuse any cached work from this process. Only do this if no + -- slots are currently free (e.g. clients could be waiting). + if 1*rAwakeAll == 1 then + local count = redis.call('zCard',kWaiting) + for i = 1,count do + redis.call('rPush',kWakeup,'w') + end + redis.call('pexpire',kWakeup,1) + end + return 1 +LUA; + try { + $res = $conn->luaEval( $script, + array( + $this->getSlotListKey(), + $this->getSlotRTimeSetKey(), + $this->getWakeupListKey(), + $this->getWaitSetKey(), + $this->workers, + $this->lockTTL, + $this->slot, + $this->slotTime, // used for CAS-style sanity check + ( $this->onRelease === self::AWAKE_ALL ) ? 1 : 0, + microtime( true ) + ), + 4 # number of first argument(s) that are keys + ); + } catch ( RedisException $e ) { + return Status::newFatal( 'pool-error-unknown', $e->getMessage() ); + } + + $this->slot = null; + $this->slotTime = null; + $this->onRelease = null; + unset( self::$active[$this->session] ); + + return Status::newGood( PoolCounter::RELEASED ); + } + + /** + * @param int $doWakeup AWAKE_* constant + * @return Status + */ + protected function waitForSlotOrNotif( $doWakeup ) { + if ( $this->slot !== null ) { + return Status::newGood( PoolCounter::LOCK_HELD ); // already acquired + } + + $status = $this->getConnection(); + if ( !$status->isOK() ) { + return $status; + } + $conn = $status->value; + + $now = microtime( true ); + try { + $slot = $this->initAndPopPoolSlotList( $conn, $now ); + if ( ctype_digit( $slot ) ) { + // Pool slot acquired by this process + $slotTime = $now; + } elseif ( $slot === 'QUEUE_FULL' ) { + // Too many processes are waiting for pooled processes to finish + return Status::newGood( PoolCounter::QUEUE_FULL ); + } elseif ( $slot === 'QUEUE_WAIT' ) { + // This process is now registered as waiting + $keys = ( $doWakeup == self::AWAKE_ALL ) + // Wait for an open slot or wake-up signal (preferring the later) + ? array( $this->getWakeupListKey(), $this->getSlotListKey() ) + // Just wait for an actual pool slot + : array( $this->getSlotListKey() ); + + $res = $conn->blPop( $keys, $this->timeout ); + if ( $res === array() ) { + $conn->zRem( $this->getWaitSetKey(), $this->session ); // no longer waiting + return Status::newGood( PoolCounter::TIMEOUT ); + } + + $slot = $res[1]; // pool slot or "w" for wake-up notifications + $slotTime = microtime( true ); // last microtime() was a few RTTs ago + // Unregister this process as waiting and bump slot "next release" time + $this->registerAcquisitionTime( $conn, $slot, $slotTime ); + } else { + return Status::newFatal( 'pool-error-unknown', "Server gave slot '$slot'." ); + } + } catch ( RedisException $e ) { + return Status::newFatal( 'pool-error-unknown', $e->getMessage() ); + } + + if ( $slot !== 'w' ) { + $this->slot = $slot; + $this->slotTime = $slotTime; + $this->onRelease = $doWakeup; + self::$active[$this->session] = $this; + } + + return Status::newGood( $slot === 'w' ? PoolCounter::DONE : PoolCounter::LOCKED ); + } + + /** + * @param RedisConnRef $conn + * @param float UNIX timestamp + * @return string|bool False on failure + */ + protected function initAndPopPoolSlotList( RedisConnRef $conn, $now ) { + static $script = +<< 0 then + slot = redis.call('lPop',kSlots) + -- Update the slot "next release" time + redis.call('zAdd',kSlotsNextRelease,rTime + rExpiry,slot) + elseif redis.call('zCard',kSlotWaits) >= 1*rMaxQueue then + slot = 'QUEUE_FULL' + else + slot = 'QUEUE_WAIT' + -- Register this process as waiting + redis.call('zAdd',kSlotWaits,rTime,rSess) + redis.call('expireAt',kSlotWaits,math.ceil(rTime + 2*rTimeout)) + end + -- Always keep renewing the expiry on use + redis.call('expireAt',kSlots,math.ceil(rTime + rExpiry)) + redis.call('expireAt',kSlotsNextRelease,math.ceil(rTime + rExpiry)) + return slot +LUA; + return $conn->luaEval( $script, + array( + $this->getSlotListKey(), + $this->getSlotRTimeSetKey(), + $this->getWaitSetKey(), + $this->workers, + $this->maxqueue, + $this->timeout, + $this->lockTTL, + $this->session, + $now + ), + 3 # number of first argument(s) that are keys + ); + } + + /** + * @param RedisConnRef $conn + * @param string $slot + * @param float $now + * @return int|bool False on failure + */ + protected function registerAcquisitionTime( RedisConnRef $conn, $slot, $now ) { + static $script = +<<luaEval( $script, + array( + $this->getSlotListKey(), + $this->getSlotRTimeSetKey(), + $this->getWaitSetKey(), + $slot, + $this->lockTTL, + $this->session, + $now + ), + 3 # number of first argument(s) that are keys + ); + } + + /** + * @return string + */ + protected function getSlotListKey() { + return "poolcounter:l-slots-{$this->keySha1}-{$this->workers}"; + } + + /** + * @return string + */ + protected function getSlotRTimeSetKey() { + return "poolcounter:z-renewtime-{$this->keySha1}-{$this->workers}"; + } + + /** + * @return string + */ + protected function getWaitSetKey() { + return "poolcounter:z-wait-{$this->keySha1}-{$this->workers}"; + } + + /** + * @return string + */ + protected function getWakeupListKey() { + return "poolcounter:l-wakeup-{$this->keySha1}-{$this->workers}"; + } + + /** + * Try to make sure that locks get released (even with exceptions and fatals) + */ + public static function releaseAll() { + foreach ( self::$active as $poolCounter ) { + try { + if ( $poolCounter->slot !== null ) { + $poolCounter->release(); + } + } catch ( Exception $e ) {} + } + } +} diff --git a/languages/i18n/en.json b/languages/i18n/en.json index 4da19a7c31..9223bb2e39 100644 --- a/languages/i18n/en.json +++ b/languages/i18n/en.json @@ -228,6 +228,7 @@ "pool-timeout": "Timeout waiting for the lock", "pool-queuefull": "Pool queue is full", "pool-errorunknown": "Unknown error", + "pool-servererror": "The pool counter service is not available ($1)", "aboutsite": "About {{SITENAME}}", "aboutpage": "Project:About", "copyright": "Content is available under $1 unless otherwise noted.", diff --git a/languages/i18n/qqq.json b/languages/i18n/qqq.json index c6cb48b531..5b3c939115 100644 --- a/languages/i18n/qqq.json +++ b/languages/i18n/qqq.json @@ -383,6 +383,7 @@ "pool-timeout": "Part of {{msg-mw|view-pool-error}}.\n\nFor explanation of 'lock' see [[w:Lock_(computer_science)|wikipedia]].", "pool-queuefull": "Part of {{msg-mw|view-pool-error}}\n\n\"Pool\" refers to a pool of processes.", "pool-errorunknown": "Part of {{msg-mw|view-pool-error}}.\n{{Identical|Unknown error}}", + "pool-servererror": "Error message. Parameters:\n* $1 - list of server addresses", "aboutsite": "Used as the label of the link that appears at the footer of every page on the wiki (in most of the skins) and leads to the page that contains the site description. The link target is {{msg-mw|aboutpage}}.\n\n[[mw:Manual:Interface/Aboutsite|MediaWiki manual]].\n\n{{doc-important|Do not change {{SITENAME}}.}}\n\n{{Identical|About}}", "aboutpage": "Used as the target of the link that appears at the footer of every page on the wiki (in most of the skins) and leads to the page that contains the site description. Therefore the content should be the same with the page name of the site description page. Only the message in the [[mw:Manual:$wgLanguageCode|site language]] ([[MediaWiki:Aboutpage]]) is used. The link label is {{msg-mw|aboutsite}}.\n\n{{doc-important|Do not translate \"Project:\" part, for this is the namespace prefix.}}", "copyright": "Parameters:\n* $1 - license name\n'''See also'''\n* {{msg-mw|Mobile-frontend-copyright}}", diff --git a/maintenance/language/messages.inc b/maintenance/language/messages.inc index a953d3e6a9..66a78a9294 100644 --- a/maintenance/language/messages.inc +++ b/maintenance/language/messages.inc @@ -268,6 +268,7 @@ $wgMessageStructure = array( 'pool-timeout', 'pool-queuefull', 'pool-errorunknown', + 'pool-servererror', ), 'links' => array( 'aboutsite', -- 2.20.1