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". Also,
44 * queue classes used by this should ignore down servers (with TTL) to avoid slowness.
49 class JobQueueFederated
extends JobQueue
{
51 protected $partitionRing;
53 protected $partitionPushRing;
54 /** @var array (partition name => JobQueue) reverse sorted by weight */
55 protected $partitionQueues = array();
60 /** @var int Maximum number of partitions to try */
61 protected $maxPartitionsTry;
63 const CACHE_TTL_SHORT
= 30; // integer; seconds to cache info without re-validating
64 const CACHE_TTL_LONG
= 300; // integer; seconds to cache info that is kept up to date
67 * @param array $params Possible keys:
68 * - sectionsByWiki : A map of wiki IDs to section names.
69 * Wikis will default to using the section "default".
70 * - partitionsBySection : Map of section names to maps of (partition name => weight).
71 * A section called 'default' must be defined if not all wikis
72 * have explicitly defined sections.
73 * - configByPartition : Map of queue partition names to configuration arrays.
74 * These configuration arrays are passed to JobQueue::factory().
75 * The options set here are overriden by those passed to this
76 * the federated queue itself (e.g. 'order' and 'claimTTL').
77 * - partitionsNoPush : List of partition names that can handle pop() but not push().
78 * This can be used to migrate away from a certain partition.
79 * - maxPartitionsTry : Maximum number of times to attempt job insertion using
80 * different partition queues. This improves availability
81 * during failure, at the cost of added latency and somewhat
82 * less reliable job de-duplication mechanisms.
85 protected function __construct( array $params ) {
86 parent
::__construct( $params );
87 $section = isset( $params['sectionsByWiki'][$this->wiki
] )
88 ?
$params['sectionsByWiki'][$this->wiki
]
90 if ( !isset( $params['partitionsBySection'][$section] ) ) {
91 throw new MWException( "No configuration for section '$section'." );
93 $this->maxPartitionsTry
= isset( $params['maxPartitionsTry'] )
94 ?
$params['maxPartitionsTry']
96 // Get the full partition map
97 $partitionMap = $params['partitionsBySection'][$section];
98 arsort( $partitionMap, SORT_NUMERIC
);
99 // Get the partitions jobs can actually be pushed to
100 $partitionPushMap = $partitionMap;
101 if ( isset( $params['partitionsNoPush'] ) ) {
102 foreach ( $params['partitionsNoPush'] as $partition ) {
103 unset( $partitionPushMap[$partition] );
106 // Get the config to pass to merge into each partition queue config
107 $baseConfig = $params;
108 foreach ( array( 'class', 'sectionsByWiki', 'maxPartitionsTry',
109 'partitionsBySection', 'configByPartition', 'partitionsNoPush' ) as $o
111 unset( $baseConfig[$o] ); // partition queue doesn't care about this
113 // Get the partition queue objects
114 foreach ( $partitionMap as $partition => $w ) {
115 if ( !isset( $params['configByPartition'][$partition] ) ) {
116 throw new MWException( "No configuration for partition '$partition'." );
118 $this->partitionQueues
[$partition] = JobQueue
::factory(
119 $baseConfig +
$params['configByPartition'][$partition] );
121 // Ring of all partitions
122 $this->partitionRing
= new HashRing( $partitionMap );
123 // Get the ring of partitions to push jobs into
124 if ( count( $partitionPushMap ) === count( $partitionMap ) ) {
125 $this->partitionPushRing
= clone $this->partitionRing
; // faster
127 $this->partitionPushRing
= new HashRing( $partitionPushMap );
129 // Aggregate cache some per-queue values if there are multiple partition queues
130 $this->cache
= count( $partitionMap ) > 1 ?
wfGetMainCache() : new EmptyBagOStuff();
133 protected function supportedOrders() {
134 // No FIFO due to partitioning, though "rough timestamp order" is supported
135 return array( 'undefined', 'random', 'timestamp' );
138 protected function optimalOrder() {
139 return 'undefined'; // defer to the partitions
142 protected function supportsDelayedJobs() {
143 return true; // defer checks to the partitions
146 protected function doIsEmpty() {
147 $key = $this->getCacheKey( 'empty' );
149 $isEmpty = $this->cache
->get( $key );
150 if ( $isEmpty === 'true' ) {
152 } elseif ( $isEmpty === 'false' ) {
158 foreach ( $this->partitionQueues
as $queue ) {
160 $empty = $empty && $queue->doIsEmpty();
161 } catch ( JobQueueError
$e ) {
163 MWExceptionHandler
::logException( $e );
166 $this->throwErrorIfAllPartitionsDown( $failed );
168 $this->cache
->add( $key, $empty ?
'true' : 'false', self
::CACHE_TTL_LONG
);
172 protected function doGetSize() {
173 return $this->getCrossPartitionSum( 'size', 'doGetSize' );
176 protected function doGetAcquiredCount() {
177 return $this->getCrossPartitionSum( 'acquiredcount', 'doGetAcquiredCount' );
180 protected function doGetDelayedCount() {
181 return $this->getCrossPartitionSum( 'delayedcount', 'doGetDelayedCount' );
184 protected function doGetAbandonedCount() {
185 return $this->getCrossPartitionSum( 'abandonedcount', 'doGetAbandonedCount' );
189 * @param string $type
190 * @param string $method
193 protected function getCrossPartitionSum( $type, $method ) {
194 $key = $this->getCacheKey( $type );
196 $count = $this->cache
->get( $key );
197 if ( $count !== false ) {
202 foreach ( $this->partitionQueues
as $queue ) {
204 $count +
= $queue->$method();
205 } catch ( JobQueueError
$e ) {
207 MWExceptionHandler
::logException( $e );
210 $this->throwErrorIfAllPartitionsDown( $failed );
212 $this->cache
->set( $key, $count, self
::CACHE_TTL_SHORT
);
217 protected function doBatchPush( array $jobs, $flags ) {
218 // Local ring variable that may be changed to point to a new ring on failure
219 $partitionRing = $this->partitionPushRing
;
220 // Try to insert the jobs and update $partitionsTry on any failures.
221 // Retry to insert any remaning jobs again, ignoring the bad partitions.
223 // @codingStandardsIgnoreStart Generic.CodeAnalysis.ForLoopWithTestFunctionCall.NotAllowed
224 for ( $i = $this->maxPartitionsTry
; $i > 0 && count( $jobsLeft ); --$i ) {
225 // @codingStandardsIgnoreEnd
227 $partitionRing->getLiveRing();
228 } catch ( UnexpectedValueException
$e ) {
229 break; // all servers down; nothing to insert to
231 $jobsLeft = $this->tryJobInsertions( $jobsLeft, $partitionRing, $flags );
233 if ( count( $jobsLeft ) ) {
234 throw new JobQueueError(
235 "Could not insert job(s), {$this->maxPartitionsTry} partitions tried." );
241 * @param HashRing $partitionRing
243 * @throws JobQueueError
244 * @return array List of Job object that could not be inserted
246 protected function tryJobInsertions( array $jobs, HashRing
&$partitionRing, $flags ) {
249 // Because jobs are spread across partitions, per-job de-duplication needs
250 // to use a consistent hash to avoid allowing duplicate jobs per partition.
251 // When inserting a batch of de-duplicated jobs, QOS_ATOMIC is disregarded.
252 $uJobsByPartition = array(); // (partition name => job list)
254 foreach ( $jobs as $key => $job ) {
255 if ( $job->ignoreDuplicates() ) {
256 $sha1 = sha1( serialize( $job->getDeduplicationInfo() ) );
257 $uJobsByPartition[$partitionRing->getLiveLocation( $sha1 )][] = $job;
258 unset( $jobs[$key] );
261 // Get the batches of jobs that are not de-duplicated
262 if ( $flags & self
::QOS_ATOMIC
) {
263 $nuJobBatches = array( $jobs ); // all or nothing
265 // Split the jobs into batches and spread them out over servers if there
266 // are many jobs. This helps keep the partitions even. Otherwise, send all
267 // the jobs to a single partition queue to avoids the extra connections.
268 $nuJobBatches = array_chunk( $jobs, 300 );
271 // Insert the de-duplicated jobs into the queues...
272 foreach ( $uJobsByPartition as $partition => $jobBatch ) {
273 /** @var JobQueue $queue */
274 $queue = $this->partitionQueues
[$partition];
277 $queue->doBatchPush( $jobBatch, $flags | self
::QOS_ATOMIC
);
278 } catch ( JobQueueError
$e ) {
280 MWExceptionHandler
::logException( $e );
283 $key = $this->getCacheKey( 'empty' );
284 $this->cache
->set( $key, 'false', self
::CACHE_TTL_LONG
);
286 if ( !$partitionRing->ejectFromLiveRing( $partition, 5 ) ) { // blacklist
287 throw new JobQueueError( "Could not insert job(s), no partitions available." );
289 $jobsLeft = array_merge( $jobsLeft, $jobBatch ); // not inserted
293 // Insert the jobs that are not de-duplicated into the queues...
294 foreach ( $nuJobBatches as $jobBatch ) {
295 $partition = ArrayUtils
::pickRandom( $partitionRing->getLiveLocationWeights() );
296 $queue = $this->partitionQueues
[$partition];
299 $queue->doBatchPush( $jobBatch, $flags | self
::QOS_ATOMIC
);
300 } catch ( JobQueueError
$e ) {
302 MWExceptionHandler
::logException( $e );
305 $key = $this->getCacheKey( 'empty' );
306 $this->cache
->set( $key, 'false', self
::CACHE_TTL_LONG
);
308 if ( !$partitionRing->ejectFromLiveRing( $partition, 5 ) ) { // blacklist
309 throw new JobQueueError( "Could not insert job(s), no partitions available." );
311 $jobsLeft = array_merge( $jobsLeft, $jobBatch ); // not inserted
318 protected function doPop() {
319 $partitionsTry = $this->partitionRing
->getLiveLocationWeights(); // (partition => weight)
322 while ( count( $partitionsTry ) ) {
323 $partition = ArrayUtils
::pickRandom( $partitionsTry );
324 if ( $partition === false ) {
325 break; // all partitions at 0 weight
328 /** @var JobQueue $queue */
329 $queue = $this->partitionQueues
[$partition];
331 $job = $queue->pop();
332 } catch ( JobQueueError
$e ) {
334 MWExceptionHandler
::logException( $e );
338 $job->metadata
['QueuePartition'] = $partition;
342 unset( $partitionsTry[$partition] ); // blacklist partition
345 $this->throwErrorIfAllPartitionsDown( $failed );
347 $key = $this->getCacheKey( 'empty' );
348 $this->cache
->set( $key, 'true', self
::CACHE_TTL_LONG
);
353 protected function doAck( Job
$job ) {
354 if ( !isset( $job->metadata
['QueuePartition'] ) ) {
355 throw new MWException( "The given job has no defined partition name." );
358 return $this->partitionQueues
[$job->metadata
['QueuePartition']]->ack( $job );
361 protected function doIsRootJobOldDuplicate( Job
$job ) {
362 $params = $job->getRootJobParams();
363 $sigature = $params['rootJobSignature'];
364 $partition = $this->partitionPushRing
->getLiveLocation( $sigature );
366 return $this->partitionQueues
[$partition]->doIsRootJobOldDuplicate( $job );
367 } catch ( JobQueueError
$e ) {
368 if ( $this->partitionPushRing
->ejectFromLiveRing( $partition, 5 ) ) {
369 $partition = $this->partitionPushRing
->getLiveLocation( $sigature );
370 return $this->partitionQueues
[$partition]->doIsRootJobOldDuplicate( $job );
377 protected function doDeduplicateRootJob( Job
$job ) {
378 $params = $job->getRootJobParams();
379 $sigature = $params['rootJobSignature'];
380 $partition = $this->partitionPushRing
->getLiveLocation( $sigature );
382 return $this->partitionQueues
[$partition]->doDeduplicateRootJob( $job );
383 } catch ( JobQueueError
$e ) {
384 if ( $this->partitionPushRing
->ejectFromLiveRing( $partition, 5 ) ) {
385 $partition = $this->partitionPushRing
->getLiveLocation( $sigature );
386 return $this->partitionQueues
[$partition]->doDeduplicateRootJob( $job );
393 protected function doDelete() {
395 /** @var JobQueue $queue */
396 foreach ( $this->partitionQueues
as $queue ) {
399 } catch ( JobQueueError
$e ) {
401 MWExceptionHandler
::logException( $e );
404 $this->throwErrorIfAllPartitionsDown( $failed );
408 protected function doWaitForBackups() {
410 /** @var JobQueue $queue */
411 foreach ( $this->partitionQueues
as $queue ) {
413 $queue->waitForBackups();
414 } catch ( JobQueueError
$e ) {
416 MWExceptionHandler
::logException( $e );
419 $this->throwErrorIfAllPartitionsDown( $failed );
422 protected function doGetPeriodicTasks() {
424 /** @var JobQueue $queue */
425 foreach ( $this->partitionQueues
as $partition => $queue ) {
426 foreach ( $queue->getPeriodicTasks() as $task => $def ) {
427 $tasks["{$partition}:{$task}"] = $def;
434 protected function doFlushCaches() {
435 static $types = array(
443 foreach ( $types as $type ) {
444 $this->cache
->delete( $this->getCacheKey( $type ) );
447 /** @var JobQueue $queue */
448 foreach ( $this->partitionQueues
as $queue ) {
449 $queue->doFlushCaches();
453 public function getAllQueuedJobs() {
454 $iterator = new AppendIterator();
456 /** @var JobQueue $queue */
457 foreach ( $this->partitionQueues
as $queue ) {
458 $iterator->append( $queue->getAllQueuedJobs() );
464 public function getAllDelayedJobs() {
465 $iterator = new AppendIterator();
467 /** @var JobQueue $queue */
468 foreach ( $this->partitionQueues
as $queue ) {
469 $iterator->append( $queue->getAllDelayedJobs() );
475 public function getCoalesceLocationInternal() {
476 return "JobQueueFederated:wiki:{$this->wiki}" .
477 sha1( serialize( array_keys( $this->partitionQueues
) ) );
480 protected function doGetSiblingQueuesWithJobs( array $types ) {
484 /** @var JobQueue $queue */
485 foreach ( $this->partitionQueues
as $queue ) {
487 $nonEmpty = $queue->doGetSiblingQueuesWithJobs( $types );
488 if ( is_array( $nonEmpty ) ) {
489 $result = array_unique( array_merge( $result, $nonEmpty ) );
491 return null; // not supported on all partitions; bail
493 if ( count( $result ) == count( $types ) ) {
494 break; // short-circuit
496 } catch ( JobQueueError
$e ) {
498 MWExceptionHandler
::logException( $e );
501 $this->throwErrorIfAllPartitionsDown( $failed );
503 return array_values( $result );
506 protected function doGetSiblingQueueSizes( array $types ) {
509 /** @var JobQueue $queue */
510 foreach ( $this->partitionQueues
as $queue ) {
512 $sizes = $queue->doGetSiblingQueueSizes( $types );
513 if ( is_array( $sizes ) ) {
514 foreach ( $sizes as $type => $size ) {
515 $result[$type] = isset( $result[$type] ) ?
$result[$type] +
$size : $size;
518 return null; // not supported on all partitions; bail
520 } catch ( JobQueueError
$e ) {
522 MWExceptionHandler
::logException( $e );
525 $this->throwErrorIfAllPartitionsDown( $failed );
531 * Throw an error if no partitions available
533 * @param int $down The number of up partitions down
535 * @throws JobQueueError
537 protected function throwErrorIfAllPartitionsDown( $down ) {
538 if ( $down >= count( $this->partitionQueues
) ) {
539 throw new JobQueueError( 'No queue partitions available.' );
543 public function setTestingPrefix( $key ) {
544 /** @var JobQueue $queue */
545 foreach ( $this->partitionQueues
as $queue ) {
546 $queue->setTestingPrefix( $key );
551 * @param string $property
554 private function getCacheKey( $property ) {
555 list( $db, $prefix ) = wfSplitWikiID( $this->wiki
);
557 return wfForeignMemcKey( $db, $prefix, 'jobqueue', $this->type
, $property );