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
