3 * Job queue code for federated queues.
5 * This program is free software; you can redistribute it and/or modify
6 * it under the terms of the GNU General Public License as published by
7 * the Free Software Foundation; either version 2 of the License, or
8 * (at your option) any later version.
10 * This program is distributed in the hope that it will be useful,
11 * but WITHOUT ANY WARRANTY; without even the implied warranty of
12 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13 * GNU General Public License for more details.
15 * You should have received a copy of the GNU General Public License along
16 * with this program; if not, write to the Free Software Foundation, Inc.,
17 * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
18 * http://www.gnu.org/copyleft/gpl.html
21 * @author Aaron Schulz
25 * Class to handle enqueueing and running of background jobs for federated queues
27 * This class allows for queues to be partitioned into smaller queues.
28 * A partition is defined by the configuration for a JobQueue instance.
29 * For example, one can set $wgJobTypeConf['refreshLinks'] to point to a
30 * JobQueueFederated instance, which itself would consist of three JobQueueRedis
31 * instances, each using their own redis server. This would allow for the jobs
32 * to be split (evenly or based on weights) accross multiple servers if a single
33 * server becomes impractical or expensive. Different JobQueue classes can be mixed.
35 * The basic queue configuration (e.g. "order", "claimTTL") of a federated queue
36 * is inherited by the partition queues. Additional configuration defines what
37 * section each wiki is in, what partition queues each section uses (and their weight),
38 * and the JobQueue configuration for each partition. Some sections might only need a
39 * single queue partition, like the sections for groups of small wikis.
41 * If used for performance, then $wgMainCacheType should be set to memcached/redis.
42 * Note that "fifo" cannot be used for the ordering, since the data is distributed.
43 * One can still use "timestamp" instead, as in "roughly timestamp ordered".
48 class JobQueueFederated
extends JobQueue
{
49 /** @var Array (wiki ID => section name) */
50 protected $sectionsByWiki = array();
51 /** @var Array (section name => (partition name => weight)) */
52 protected $partitionsBySection = array();
53 /** @var Array (section name => config array) */
54 protected $configByPartition = array();
55 /** @var Array (partition names => integer) */
56 protected $partitionsNoPush = array();
59 protected $partitionRing;
60 /** @var Array (partition name => JobQueue) */
61 protected $partitionQueues = array();
65 const CACHE_TTL_SHORT
= 30; // integer; seconds to cache info without re-validating
66 const CACHE_TTL_LONG
= 300; // integer; seconds to cache info that is kept up to date
70 * - sectionsByWiki : A map of wiki IDs to section names.
71 * Wikis will default to using the section "default".
72 * - partitionsBySection : Map of section names to maps of (partition name => weight).
73 * A section called 'default' must be defined if not all wikis
74 * have explicitly defined sections.
75 * - configByPartition : Map of queue partition names to configuration arrays.
76 * These configuration arrays are passed to JobQueue::factory().
77 * The options set here are overriden by those passed to this
78 * the federated queue itself (e.g. 'order' and 'claimTTL').
79 * - partitionsNoPush : List of partition names that can handle pop() but not push().
80 * This can be used to migrate away from a certain partition.
81 * @param array $params
83 protected function __construct( array $params ) {
84 parent
::__construct( $params );
85 $this->sectionsByWiki
= isset( $params['sectionsByWiki'] )
86 ?
$params['sectionsByWiki']
87 : array(); // all in "default" section
88 $this->partitionsBySection
= $params['partitionsBySection'];
89 $this->configByPartition
= $params['configByPartition'];
90 if ( isset( $params['partitionsNoPush'] ) ) {
91 $this->partitionsNoPush
= array_flip( $params['partitionsNoPush'] );
93 $baseConfig = $params;
94 foreach ( array( 'class', 'sectionsByWiki',
95 'partitionsBySection', 'configByPartition', 'partitionsNoPush' ) as $o )
97 unset( $baseConfig[$o] );
99 foreach ( $this->getPartitionMap() as $partition => $w ) {
100 if ( !isset( $this->configByPartition
[$partition] ) ) {
101 throw new MWException( "No configuration for partition '$partition'." );
103 $this->partitionQueues
[$partition] = JobQueue
::factory(
104 $baseConfig +
$this->configByPartition
[$partition]
107 // Get the ring of partitions to push job de-duplication information into
108 $partitionsTry = array_diff_key(
109 $this->getPartitionMap(),
110 $this->partitionsNoPush
111 ); // (partition => weight)
112 $this->partitionRing
= new HashRing( $partitionsTry );
113 // Aggregate cache some per-queue values if there are multiple partition queues
114 $this->cache
= $this->isFederated() ?
wfGetMainCache() : new EmptyBagOStuff();
117 protected function supportedOrders() {
118 // No FIFO due to partitioning, though "rough timestamp order" is supported
119 return array( 'undefined', 'random', 'timestamp' );
122 protected function optimalOrder() {
123 return 'undefined'; // defer to the partitions
126 protected function supportsDelayedJobs() {
127 return true; // defer checks to the partitions
130 protected function doIsEmpty() {
131 $key = $this->getCacheKey( 'empty' );
133 $isEmpty = $this->cache
->get( $key );
134 if ( $isEmpty === 'true' ) {
136 } elseif ( $isEmpty === 'false' ) {
140 foreach ( $this->partitionQueues
as $queue ) {
142 if ( !$queue->doIsEmpty() ) {
143 $this->cache
->add( $key, 'false', self
::CACHE_TTL_LONG
);
146 } catch ( JobQueueError
$e ) {
147 wfDebugLog( 'exception', $e->getLogMessage() );
151 $this->cache
->add( $key, 'true', self
::CACHE_TTL_LONG
);
155 protected function doGetSize() {
156 return $this->getCrossPartitionSum( 'size', 'doGetSize' );
159 protected function doGetAcquiredCount() {
160 return $this->getCrossPartitionSum( 'acquiredcount', 'doGetAcquiredCount' );
163 protected function doGetDelayedCount() {
164 return $this->getCrossPartitionSum( 'delayedcount', 'doGetDelayedCount' );
167 protected function doGetAbandonedCount() {
168 return $this->getCrossPartitionSum( 'abandonedcount', 'doGetAbandonedCount' );
172 * @param string $type
173 * @param string $method
176 protected function getCrossPartitionSum( $type, $method ) {
177 $key = $this->getCacheKey( $type );
179 $count = $this->cache
->get( $key );
180 if ( is_int( $count ) ) {
185 foreach ( $this->partitionQueues
as $queue ) {
187 $count +
= $queue->$method();
188 } catch ( JobQueueError
$e ) {
189 wfDebugLog( 'exception', $e->getLogMessage() );
193 $this->cache
->set( $key, $count, self
::CACHE_TTL_SHORT
);
197 protected function doBatchPush( array $jobs, $flags ) {
198 if ( !count( $jobs ) ) {
199 return true; // nothing to do
202 $partitionsTry = array_diff_key(
203 $this->getPartitionMap(),
204 $this->partitionsNoPush
205 ); // (partition => weight)
207 // Try to insert the jobs and update $partitionsTry on any failures
208 $jobsLeft = $this->tryJobInsertions( $jobs, $partitionsTry, $flags );
209 if ( count( $jobsLeft ) ) { // some jobs failed to insert?
210 // Try to insert the remaning jobs once more, ignoring the bad partitions
211 return !count( $this->tryJobInsertions( $jobsLeft, $partitionsTry, $flags ) );
219 * @param array $partitionsTry
220 * @param integer $flags
221 * @return array List of Job object that could not be inserted
223 protected function tryJobInsertions( array $jobs, array &$partitionsTry, $flags ) {
224 if ( !count( $partitionsTry ) ) {
225 return $jobs; // can't insert anything
230 $partitionRing = new HashRing( $partitionsTry );
231 // Because jobs are spread across partitions, per-job de-duplication needs
232 // to use a consistent hash to avoid allowing duplicate jobs per partition.
233 // When inserting a batch of de-duplicated jobs, QOS_ATOMIC is disregarded.
234 $uJobsByPartition = array(); // (partition name => job list)
235 foreach ( $jobs as $key => $job ) {
236 if ( $job->ignoreDuplicates() ) {
237 $sha1 = sha1( serialize( $job->getDeduplicationInfo() ) );
238 $uJobsByPartition[$partitionRing->getLocation( $sha1 )][] = $job;
239 unset( $jobs[$key] );
242 // Get the batches of jobs that are not de-duplicated
243 if ( $flags & self
::QOS_ATOMIC
) {
244 $nuJobBatches = array( $jobs ); // all or nothing
246 // Split the jobs into batches and spread them out over servers if there
247 // are many jobs. This helps keep the partitions even. Otherwise, send all
248 // the jobs to a single partition queue to avoids the extra connections.
249 $nuJobBatches = array_chunk( $jobs, 300 );
252 // Insert the de-duplicated jobs into the queues...
253 foreach ( $uJobsByPartition as $partition => $jobBatch ) {
254 $queue = $this->partitionQueues
[$partition];
256 $ok = $queue->doBatchPush( $jobBatch, $flags );
257 } catch ( JobQueueError
$e ) {
259 wfDebugLog( 'exception', $e->getLogMessage() );
262 $key = $this->getCacheKey( 'empty' );
263 $this->cache
->set( $key, 'false', JobQueueDB
::CACHE_TTL_LONG
);
265 unset( $partitionsTry[$partition] ); // blacklist partition
266 $jobsLeft = array_merge( $jobsLeft, $jobBatch ); // not inserted
269 // Insert the jobs that are not de-duplicated into the queues...
270 foreach ( $nuJobBatches as $jobBatch ) {
271 $partition = ArrayUtils
::pickRandom( $partitionsTry );
272 if ( $partition === false ) { // all partitions at 0 weight?
273 $jobsLeft = array_merge( $jobsLeft, $jobBatch ); // not inserted
275 $queue = $this->partitionQueues
[$partition];
277 $ok = $queue->doBatchPush( $jobBatch, $flags );
278 } catch ( JobQueueError
$e ) {
280 wfDebugLog( 'exception', $e->getLogMessage() );
283 $key = $this->getCacheKey( 'empty' );
284 $this->cache
->set( $key, 'false', JobQueueDB
::CACHE_TTL_LONG
);
286 unset( $partitionsTry[$partition] ); // blacklist partition
287 $jobsLeft = array_merge( $jobsLeft, $jobBatch ); // not inserted
295 protected function doPop() {
296 $key = $this->getCacheKey( 'empty' );
298 $isEmpty = $this->cache
->get( $key );
299 if ( $isEmpty === 'true' ) {
303 $partitionsTry = $this->getPartitionMap(); // (partition => weight)
305 while ( count( $partitionsTry ) ) {
306 $partition = ArrayUtils
::pickRandom( $partitionsTry );
307 if ( $partition === false ) {
308 break; // all partitions at 0 weight
310 $queue = $this->partitionQueues
[$partition];
312 $job = $queue->pop();
313 } catch ( JobQueueError
$e ) {
315 wfDebugLog( 'exception', $e->getLogMessage() );
318 $job->metadata
['QueuePartition'] = $partition;
321 unset( $partitionsTry[$partition] ); // blacklist partition
325 $this->cache
->set( $key, 'true', JobQueueDB
::CACHE_TTL_LONG
);
329 protected function doAck( Job
$job ) {
330 if ( !isset( $job->metadata
['QueuePartition'] ) ) {
331 throw new MWException( "The given job has no defined partition name." );
333 return $this->partitionQueues
[$job->metadata
['QueuePartition']]->ack( $job );
336 protected function doIsRootJobOldDuplicate( Job
$job ) {
337 $params = $job->getRootJobParams();
338 $partitions = $this->partitionRing
->getLocations( $params['rootJobSignature'], 2 );
340 return $this->partitionQueues
[$partitions[0]]->doIsRootJobOldDuplicate( $job );
341 } catch ( MWException
$e ) {
342 if ( isset( $partitions[1] ) ) { // check fallback partition
343 return $this->partitionQueues
[$partitions[1]]->doIsRootJobOldDuplicate( $job );
349 protected function doDeduplicateRootJob( Job
$job ) {
350 $params = $job->getRootJobParams();
351 $partitions = $this->partitionRing
->getLocations( $params['rootJobSignature'], 2 );
353 return $this->partitionQueues
[$partitions[0]]->doDeduplicateRootJob( $job );
354 } catch ( MWException
$e ) {
355 if ( isset( $partitions[1] ) ) { // check fallback partition
356 return $this->partitionQueues
[$partitions[1]]->doDeduplicateRootJob( $job );
362 protected function doDelete() {
363 foreach ( $this->partitionQueues
as $queue ) {
366 } catch ( JobQueueError
$e ) {
367 wfDebugLog( 'exception', $e->getLogMessage() );
372 protected function doWaitForBackups() {
373 foreach ( $this->partitionQueues
as $queue ) {
375 $queue->waitForBackups();
376 } catch ( JobQueueError
$e ) {
377 wfDebugLog( 'exception', $e->getLogMessage() );
382 protected function doGetPeriodicTasks() {
384 foreach ( $this->partitionQueues
as $partition => $queue ) {
385 foreach ( $queue->getPeriodicTasks() as $task => $def ) {
386 $tasks["{$partition}:{$task}"] = $def;
392 protected function doFlushCaches() {
393 static $types = array(
400 foreach ( $types as $type ) {
401 $this->cache
->delete( $this->getCacheKey( $type ) );
403 foreach ( $this->partitionQueues
as $queue ) {
404 $queue->doFlushCaches();
408 public function getAllQueuedJobs() {
409 $iterator = new AppendIterator();
410 foreach ( $this->partitionQueues
as $queue ) {
411 $iterator->append( $queue->getAllQueuedJobs() );
416 public function getAllDelayedJobs() {
417 $iterator = new AppendIterator();
418 foreach ( $this->partitionQueues
as $queue ) {
419 $iterator->append( $queue->getAllDelayedJobs() );
424 public function setTestingPrefix( $key ) {
425 foreach ( $this->partitionQueues
as $queue ) {
426 $queue->setTestingPrefix( $key );
431 * @return Array Map of (partition name => weight)
433 protected function getPartitionMap() {
434 $section = isset( $this->sectionsByWiki
[$this->wiki
] )
435 ?
$this->sectionsByWiki
[$this->wiki
]
437 if ( !isset( $this->partitionsBySection
[$section] ) ) {
438 throw new MWException( "No configuration for section '$section'." );
440 return $this->partitionsBySection
[$section];
444 * @return bool The queue is actually split up across multiple queue partitions
446 protected function isFederated() {
447 return ( count( $this->getPartitionMap() ) > 1 );
453 private function getCacheKey( $property ) {
454 list( $db, $prefix ) = wfSplitWikiID( $this->wiki
);
455 return wfForeignMemcKey( $db, $prefix, 'jobqueue', $this->type
, $property );