Aaron Schulz has uploaded a new change for review. https://gerrit.wikimedia.org/r/51823
Change subject: [JobQueue] Added JobQueueFederated class for partitioned queues. ...................................................................... [JobQueue] Added JobQueueFederated class for partitioned queues. * Moved the root job handling up to the base class and provided some default function implementations. * Also bumped ROOTJOB_TTL from 14 to 28 days. Change-Id: I44d59b67cf417dca28a3e9b25371dac5a7ffcb47 --- M includes/AutoLoader.php M includes/job/JobQueue.php M includes/job/JobQueueDB.php A includes/job/JobQueueFederated.php M includes/job/JobQueueRedis.php 5 files changed, 370 insertions(+), 65 deletions(-) git pull ssh://gerrit.wikimedia.org:29418/mediawiki/core refs/changes/23/51823/1 diff --git a/includes/AutoLoader.php b/includes/AutoLoader.php index ace1348..bfe2a5f 100644 --- a/includes/AutoLoader.php +++ b/includes/AutoLoader.php @@ -659,6 +659,7 @@ 'JobQueueAggregatorRedis' => 'includes/job/JobQueueAggregatorRedis.php', 'JobQueueDB' => 'includes/job/JobQueueDB.php', 'JobQueueGroup' => 'includes/job/JobQueueGroup.php', + 'JobQueueFederated' => 'includes/job/JobQueueFederated.php', 'JobQueueRedis' => 'includes/job/JobQueueRedis.php', # includes/job/jobs diff --git a/includes/job/JobQueue.php b/includes/job/JobQueue.php index 4996a9e..8fb6259 100644 --- a/includes/job/JobQueue.php +++ b/includes/job/JobQueue.php @@ -37,6 +37,8 @@ const QoS_Atomic = 1; // integer; "all-or-nothing" job insertions + const ROOTJOB_TTL = 2419200; // integer; seconds to remember root jobs (28 days) + /** * @param $params array */ @@ -228,6 +230,15 @@ wfProfileIn( __METHOD__ ); $job = $this->doPop(); wfProfileOut( __METHOD__ ); + + // Flag this job as an old duplicate based on its "root" job... + try { + if ( $job && $this->isRootJobOldDuplicate( $job ) ) { + wfIncrStats( 'job-pop-duplicate' ); + $job = DuplicateJob::newFromJob( $job ); // convert to a no-op + } + } catch ( MWException $e ) {} // don't lose jobs over this + return $job; } @@ -310,7 +321,76 @@ * @return bool */ protected function doDeduplicateRootJob( Job $job ) { - return true; + global $wgMemc; + + $params = $job->getParams(); + if ( !isset( $params['rootJobSignature'] ) ) { + throw new MWException( "Cannot register root job; missing 'rootJobSignature'." ); + } elseif ( !isset( $params['rootJobTimestamp'] ) ) { + throw new MWException( "Cannot register root job; missing 'rootJobTimestamp'." ); + } + $key = $this->getRootJobCacheKey( $params['rootJobSignature'] ); + // Callers should call batchInsert() and then this function so that if the insert + // fails, the de-duplication registration will be aborted. Since the insert is + // deferred till "transaction idle", do the same here, so that the ordering is + // maintained. Having only the de-duplication registration succeed would cause + // jobs to become no-ops without any actual jobs that made them redundant. + $timestamp = $wgMemc->get( $key ); // current last timestamp of this job + if ( $timestamp && $timestamp >= $params['rootJobTimestamp'] ) { + return true; // a newer version of this root job was enqueued + } + + // Update the timestamp of the last root job started at the location... + return $wgMemc->set( $key, $params['rootJobTimestamp'], JobQueueDB::ROOTJOB_TTL ); + } + + /** + * Check if the "root" job of a given job has been superseded by a newer one + * + * @param $job Job + * @return bool + * @throws MWException + */ + final protected function isRootJobOldDuplicate( Job $job ) { + if ( $job->getType() !== $this->type ) { + throw new MWException( "Got '{$job->getType()}' job; expected '{$this->type}'." ); + } + wfProfileIn( __METHOD__ ); + $isDuplicate = $this->doIsRootJobOldDuplicate( $job ); + wfProfileOut( __METHOD__ ); + return $isDuplicate; + } + + /** + * @see JobQueue::isRootJobOldDuplicate() + * @param Job $job + * @return bool + */ + protected function doIsRootJobOldDuplicate( Job $job ) { + global $wgMemc; + + $params = $job->getParams(); + if ( !isset( $params['rootJobSignature'] ) ) { + return false; // job has no de-deplication info + } elseif ( !isset( $params['rootJobTimestamp'] ) ) { + trigger_error( "Cannot check root job; missing 'rootJobTimestamp'." ); + return false; + } + + // Get the last time this root job was enqueued + $timestamp = $wgMemc->get( $this->getRootJobCacheKey( $params['rootJobSignature'] ) ); + + // Check if a new root job was started at the location after this one's... + return ( $timestamp && $timestamp > $params['rootJobTimestamp'] ); + } + + /** + * @param string $signature Hash identifier of the root job + * @return string + */ + protected function getRootJobCacheKey( $signature ) { + list( $db, $prefix ) = wfSplitWikiID( $this->wiki ); + return wfForeignMemcKey( $db, $prefix, 'jobqueue', $this->type, 'rootjob', $signature ); } /** diff --git a/includes/job/JobQueueDB.php b/includes/job/JobQueueDB.php index fd64895..130394e 100644 --- a/includes/job/JobQueueDB.php +++ b/includes/job/JobQueueDB.php @@ -28,7 +28,6 @@ * @since 1.21 */ class JobQueueDB extends JobQueue { - const ROOTJOB_TTL = 1209600; // integer; seconds to remember root jobs (14 days) const CACHE_TTL_SHORT = 30; // integer; seconds to cache info without re-validating const CACHE_TTL_LONG = 300; // integer; seconds to cache info that is kept up to date const MAX_AGE_PRUNE = 604800; // integer; seconds a job can live once claimed @@ -244,11 +243,6 @@ $job = Job::factory( $row->job_cmd, $title, self::extractBlob( $row->job_params ), $row->job_id ); $job->id = $row->job_id; // XXX: work around broken subclasses - // Flag this job as an old duplicate based on its "root" job... - if ( $this->isRootJobOldDuplicate( $job ) ) { - wfIncrStats( 'job-pop-duplicate' ); - $job = DuplicateJob::newFromJob( $job ); // convert to a no-op - } break; // done } while( true ); @@ -526,30 +520,6 @@ } /** - * Check if the "root" job of a given job has been superseded by a newer one - * - * @param $job Job - * @return bool - */ - protected function isRootJobOldDuplicate( Job $job ) { - global $wgMemc; - - $params = $job->getParams(); - if ( !isset( $params['rootJobSignature'] ) ) { - return false; // job has no de-deplication info - } elseif ( !isset( $params['rootJobTimestamp'] ) ) { - trigger_error( "Cannot check root job; missing 'rootJobTimestamp'." ); - return false; - } - - // Get the last time this root job was enqueued - $timestamp = $wgMemc->get( $this->getRootJobCacheKey( $params['rootJobSignature'] ) ); - - // Check if a new root job was started at the location after this one's... - return ( $timestamp && $timestamp > $params['rootJobTimestamp'] ); - } - - /** * @see JobQueue::doWaitForBackups() * @return void */ @@ -640,15 +610,6 @@ private function getCacheKey( $property ) { list( $db, $prefix ) = wfSplitWikiID( $this->wiki ); return wfForeignMemcKey( $db, $prefix, 'jobqueue', $this->type, $property ); - } - - /** - * @param string $signature Hash identifier of the root job - * @return string - */ - private function getRootJobCacheKey( $signature ) { - list( $db, $prefix ) = wfSplitWikiID( $this->wiki ); - return wfForeignMemcKey( $db, $prefix, 'jobqueue', $this->type, 'rootjob', $signature ); } /** diff --git a/includes/job/JobQueueFederated.php b/includes/job/JobQueueFederated.php new file mode 100644 index 0000000..43e9eda --- /dev/null +++ b/includes/job/JobQueueFederated.php @@ -0,0 +1,283 @@ +<?php +/** + * Job queue code for federated queues. + * + * 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 enqueueing and running of background jobs for federated queues + * + * @ingroup JobQueue + * @since 1.21 + */ +class JobQueueFederated extends JobQueue { + /** @var Array (wiki ID => section name) */ + protected $sectionsByWiki = array(); + /** @var Array (section name => (partition name => weight)) */ + protected $partitionsBySection = array(); + /** @var Array (section name => config array) */ + protected $configByPartition = array(); + + /** @var BagOStuff */ + protected $cache; + + const CACHE_TTL_SHORT = 30; // integer; seconds to cache info without re-validating + const CACHE_TTL_LONG = 300; // integer; seconds to cache info that is kept up to date + + /** + * @params include: + * - sectionsByWiki : A map of wiki IDs to section names. + * Wikis will default to using the section "DEFAULT". + * - partitionsBySection : Map of section names to maps of (partition name => weight). + * A section called 'default' must be defined if not all wikis + * have explicitly defined sections. + * - configByPartition : Map of queue partition names to configuration arrays. + * These configuration arrays are passed to JobQueue::factory(). + * The options set here are overriden by those passed to this + * the federated queue itself (e.g. 'order' and 'claimTTL'). + * @param array $params + */ + protected function __construct( array $params ) { + parent::__construct( $params ); + $this->sectionsByWiki = $params['sectionsByWiki']; + $this->partitionsBySection = $params['partitionsBySection']; + $this->configByPartition = $params['configByPartition']; + if ( $this->order === 'fifo' ) { + throw new MWException( 'Federated queues do not support FIFO.' ); + } + $this->cache = wfGetMainCache(); + } + + /** + * @see JobQueue::isEmpty() + * @return bool + */ + protected function doIsEmpty() { + $key = $this->getCacheKey( 'empty' ); + + $isEmpty = $this->cache->get( $key ); + if ( $isEmpty === 'true' ) { + return true; + } elseif ( $isEmpty === 'false' ) { + return false; + } + + foreach ( $this->getQueuePartitions() as $queue ) { + if ( !$queue->doIsEmpty() ) { + $this->cache->add( $key, 'false', self::CACHE_TTL_LONG ); + return false; + } + } + + $this->cache->add( $key, 'true', self::CACHE_TTL_LONG ); + return true; + } + + /** + * @see JobQueue::getSize() + * @return integer + */ + protected function doGetSize() { + $key = $this->getCacheKey( 'size' ); + + $size = $this->cache->get( $key ); + if ( is_int( $size ) ) { + return $size; + } + + $size = 0; + foreach ( $this->getQueuePartitions() as $queue ) { + $size += $queue->doGetSize(); + } + + $this->cache->set( $key, $size, self::CACHE_TTL_SHORT ); + return $size; + } + + /** + * @see JobQueue::getAcquiredCount() + * @return integer + */ + protected function doGetAcquiredCount() { + $key = $this->getCacheKey( 'acquiredcount' ); + + $count = $this->cache->get( $key ); + if ( is_int( $count ) ) { + return $count; + } + + $count = 0; + foreach ( $this->getQueuePartitions() as $queue ) { + $count += $queue->doGetAcquiredCount(); + } + + $this->cache->set( $key, $count, self::CACHE_TTL_SHORT ); + return $count; + } + + /** + * @see JobQueue::batchPush() + * @return bool + */ + protected function doBatchPush( array $jobs, $flags ) { + if ( !count( $jobs ) ) { + return true; // nothing to do + } + + // Insert *all* the jobs in to one random parition queue to avoid RTTs. + // The also means that the JobQueue::QoS_Atomic flag is respected if given. + // A side effect is that large insertions are more even when split up. + $partitionsTry = $this->getPartitionMap(); // (partition => weight) + for ( $attempts=1; $attempts <= 2 && count( $partitionsTry ); ++$attempts ) { + $partition = wfPickRandom( $partitionsTry ); + $queue = $this->getPartitionQueue( $partition ); + if ( $queue->doBatchPush( $jobs, $flags ) ) { + $key = $this->getCacheKey( 'empty' ); + $this->cache->set( $key, 'false', JobQueueDB::CACHE_TTL_LONG ); + return true; + } + unset( $partitionsTry[$partition] ); + } + + return false; + } + + /** + * @see JobQueue::pop() + * @return Job + */ + protected function doPop() { + $partitionsTry = $this->getPartitionMap(); // (partition => weight) + while ( count( $partitionsTry ) ) { + $partition = wfPickRandom( $partitionsTry ); + $queue = $this->getPartitionQueue( $partition ); + $job = $queue->pop(); + if ( $job ) { + $job->metadata['QueuePartition'] = $partition; + return $job; + } + unset( $partitionsTry[$partition] ); + } + $key = $this->getCacheKey( 'empty' ); + $this->cache->set( $key, 'true', JobQueueDB::CACHE_TTL_LONG ); + + return false; + } + + /** + * @see JobQueue::ack() + * @return bool + */ + protected function doAck( Job $job ) { + if ( !isset( $job->metadata['QueuePartition'] ) ) { + throw new MWException( "The given job has no defined partition name." ); + } + return $this->getPartitionQueue( $job->metadata['QueuePartition'] )->ack( $job ); + } + + /** + * @see JobQueue::waitForBackups() + * @return void + */ + protected function doWaitForBackups() { + foreach ( $this->getQueuePartitions() as $queue ) { + $queue->waitForBackups(); + } + } + + /** + * @see JobQueue::getPeriodicTasks() + * @return Array + */ + protected function doGetPeriodicTasks() { + $tasks = array(); + foreach ( $this->getQueuePartitions() as $partition => $queue ) { + foreach ( $queue->getPeriodicTasks() as $task => $def ) { + $tasks["{$partition}:{$task}"] = $def; + } + } + return $tasks; + } + + /** + * @see JobQueue::flushCaches() + * @return void + */ + protected function doFlushCaches() { + foreach ( $this->getQueuePartitions() as $queue ) { + $queue->doFlushCaches(); + } + } + + /** + * @return string + */ + private function getCacheKey( $property ) { + list( $db, $prefix ) = wfSplitWikiID( $this->wiki ); + return wfForeignMemcKey( $db, $prefix, 'jobqueue', $this->type, $property ); + } + + /** + * @see JobQueue::setTestingPrefix() + * @param $key string + * @return void + * @throws MWException + */ + public function setTestingPrefix( $key ) { + foreach ( $this->getQueuePartitions() as $queue ) { + $queue->setTestingPrefix( $key ); + } + } + + /** + * @return Array Map of (partition name => weight) + */ + protected function getPartitionMap() { + $section = isset( $this->sectionsByWiki[$this->wiki] ) + ? $this->sectionsByWiki[$this->wiki] + : 'default'; + return $this->partitionsBySection[$section]; + } + + /** + * @param $partition string + * @return JobQueue + */ + protected function getPartitionQueue( $partition ) { + if ( !isset( $this->configByPartition[$partition] ) ) { + throw new MWException( "No configuration for partition '$partition'." ); + } + return JobQueue::factory( + array( 'wiki' => $this->wiki, 'type' => $this->type ) + + $this->configByPartition[$partition] + ); + } + + /** + * @return Array Map of (partition name => JobQueue) + */ + protected function getQueuePartitions() { + $queues = array(); + foreach ( $this->getPartitionMap( $this->wiki ) as $partition => $weight ) { + $queues[$partition] = $this->getPartitionQueue( $partition ); + } + return $queues; + } +} diff --git a/includes/job/JobQueueRedis.php b/includes/job/JobQueueRedis.php index 2ce47bb..156c2a1 100644 --- a/includes/job/JobQueueRedis.php +++ b/includes/job/JobQueueRedis.php @@ -33,7 +33,6 @@ protected $server; // string; server address - const ROOTJOB_TTL = 1209600; // integer; seconds to remember root jobs (14 days) const MAX_AGE_PRUNE = 604800; // integer; seconds a job can live once claimed (7 days) protected $key; // string; key to prefix the queue keys with (used for testing) @@ -224,14 +223,6 @@ $this->throwRedisException( $this->server, $conn, $e ); } - // Flag this job as an old duplicate based on its "root" job... - try { - if ( $job && $this->isRootJobOldDuplicate( $job ) ) { - wfIncrStats( 'job-pop-duplicate' ); - return DuplicateJob::newFromJob( $job ); // convert to a no-op - } - } catch ( MWException $e ) {} // don't lose jobs over this - return $job; } @@ -286,7 +277,7 @@ } elseif ( !isset( $params['rootJobTimestamp'] ) ) { throw new MWException( "Cannot register root job; missing 'rootJobTimestamp'." ); } - $key = $this->getRootJobKey( $params['rootJobSignature'] ); + $key = $this->getRootJobCacheKey( $params['rootJobSignature'] ); $conn = $this->getConnection(); try { @@ -302,13 +293,11 @@ } /** - * Check if the "root" job of a given job has been superseded by a newer one - * - * @param $job Job + * @see JobQueue::doIsRootJobOldDuplicate() + * @param Job $job * @return bool - * @throws MWException */ - protected function isRootJobOldDuplicate( Job $job ) { + protected function doIsRootJobOldDuplicate( Job $job ) { $params = $job->getParams(); if ( !isset( $params['rootJobSignature'] ) ) { return false; // job has no de-deplication info @@ -320,7 +309,7 @@ $conn = $this->getConnection(); try { // Get the last time this root job was enqueued - $timestamp = $conn->get( $this->getRootJobKey( $params['rootJobSignature'] ) ); + $timestamp = $conn->get( $this->getRootJobCacheKey( $params['rootJobSignature'] ) ); } catch ( RedisException $e ) { $this->throwRedisException( $this->server, $conn, $e ); } @@ -561,15 +550,6 @@ } else { return wfForeignMemcKey( $db, $prefix, 'jobqueue', $this->type, $prop ); } - } - - /** - * @param string $signature Hash identifier of the root job - * @return string - */ - private function getRootJobKey( $signature ) { - list( $db, $prefix ) = wfSplitWikiID( $this->wiki ); - return wfForeignMemcKey( $db, $prefix, 'jobqueue', $this->type, 'rootjob', $signature ); } /** -- To view, visit https://gerrit.wikimedia.org/r/51823 To unsubscribe, visit https://gerrit.wikimedia.org/r/settings Gerrit-MessageType: newchange Gerrit-Change-Id: I44d59b67cf417dca28a3e9b25371dac5a7ffcb47 Gerrit-PatchSet: 1 Gerrit-Project: mediawiki/core Gerrit-Branch: master Gerrit-Owner: Aaron Schulz <asch...@wikimedia.org> _______________________________________________ MediaWiki-commits mailing list MediaWiki-commits@lists.wikimedia.org https://lists.wikimedia.org/mailman/listinfo/mediawiki-commits