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
24 * Class to handle enqueueing and running of background jobs for federated queues
26 * This class allows for queues to be partitioned into smaller queues.
27 * A partition is defined by the configuration for a JobQueue instance.
28 * For example, one can set $wgJobTypeConf['refreshLinks'] to point to a
29 * JobQueueFederated instance, which itself would consist of three JobQueueRedis
30 * instances, each using their own redis server. This would allow for the jobs
31 * to be split (evenly or based on weights) across multiple servers if a single
32 * server becomes impractical or expensive. Different JobQueue classes can be mixed.
34 * The basic queue configuration (e.g. "order", "claimTTL") of a federated queue
35 * is inherited by the partition queues. Additional configuration defines what
36 * section each wiki is in, what partition queues each section uses (and their weight),
37 * and the JobQueue configuration for each partition. Some sections might only need a
38 * single queue partition, like the sections for groups of small wikis.
40 * If used for performance, then $wgMainCacheType should be set to memcached/redis.
41 * Note that "fifo" cannot be used for the ordering, since the data is distributed.
42 * One can still use "timestamp" instead, as in "roughly timestamp ordered". Also,
43 * queue classes used by this should ignore down servers (with TTL) to avoid slowness.
48 class JobQueueFederated
extends JobQueue
{
50 protected $partitionRing;
51 /** @var JobQueue[] (partition name => JobQueue) reverse sorted by weight */
52 protected $partitionQueues = [];
54 /** @var int Maximum number of partitions to try */
55 protected $maxPartitionsTry;
58 * @param array $params Possible keys:
59 * - sectionsByWiki : A map of wiki IDs to section names.
60 * Wikis will default to using the section "default".
61 * - partitionsBySection : Map of section names to maps of (partition name => weight).
62 * A section called 'default' must be defined if not all wikis
63 * have explicitly defined sections.
64 * - configByPartition : Map of queue partition names to configuration arrays.
65 * These configuration arrays are passed to JobQueue::factory().
66 * The options set here are overridden by those passed to this
67 * the federated queue itself (e.g. 'order' and 'claimTTL').
68 * - maxPartitionsTry : Maximum number of times to attempt job insertion using
69 * different partition queues. This improves availability
70 * during failure, at the cost of added latency and somewhat
71 * less reliable job de-duplication mechanisms.
74 protected function __construct( array $params ) {
75 parent
::__construct( $params );
76 $section = isset( $params['sectionsByWiki'][$this->wiki
] )
77 ?
$params['sectionsByWiki'][$this->wiki
]
79 if ( !isset( $params['partitionsBySection'][$section] ) ) {
80 throw new MWException( "No configuration for section '$section'." );
82 $this->maxPartitionsTry
= isset( $params['maxPartitionsTry'] )
83 ?
$params['maxPartitionsTry']
85 // Get the full partition map
86 $partitionMap = $params['partitionsBySection'][$section];
87 arsort( $partitionMap, SORT_NUMERIC
);
88 // Get the config to pass to merge into each partition queue config
89 $baseConfig = $params;
90 foreach ( [ 'class', 'sectionsByWiki', 'maxPartitionsTry',
91 'partitionsBySection', 'configByPartition', ] as $o
93 unset( $baseConfig[$o] ); // partition queue doesn't care about this
95 // The class handles all aggregator calls already
96 unset( $baseConfig['aggregator'] );
97 // Get the partition queue objects
98 foreach ( $partitionMap as $partition => $w ) {
99 if ( !isset( $params['configByPartition'][$partition] ) ) {
100 throw new MWException( "No configuration for partition '$partition'." );
102 $this->partitionQueues
[$partition] = JobQueue
::factory(
103 $baseConfig +
$params['configByPartition'][$partition] );
105 // Ring of all partitions
106 $this->partitionRing
= new HashRing( $partitionMap );
109 protected function supportedOrders() {
110 // No FIFO due to partitioning, though "rough timestamp order" is supported
111 return [ 'undefined', 'random', 'timestamp' ];
114 protected function optimalOrder() {
115 return 'undefined'; // defer to the partitions
118 protected function supportsDelayedJobs() {
119 foreach ( $this->partitionQueues
as $queue ) {
120 if ( !$queue->supportsDelayedJobs() ) {
128 protected function doIsEmpty() {
131 foreach ( $this->partitionQueues
as $queue ) {
133 $empty = $empty && $queue->doIsEmpty();
134 } catch ( JobQueueError
$e ) {
136 $this->logException( $e );
139 $this->throwErrorIfAllPartitionsDown( $failed );
144 protected function doGetSize() {
145 return $this->getCrossPartitionSum( 'size', 'doGetSize' );
148 protected function doGetAcquiredCount() {
149 return $this->getCrossPartitionSum( 'acquiredcount', 'doGetAcquiredCount' );
152 protected function doGetDelayedCount() {
153 return $this->getCrossPartitionSum( 'delayedcount', 'doGetDelayedCount' );
156 protected function doGetAbandonedCount() {
157 return $this->getCrossPartitionSum( 'abandonedcount', 'doGetAbandonedCount' );
161 * @param string $type
162 * @param string $method
165 protected function getCrossPartitionSum( $type, $method ) {
168 foreach ( $this->partitionQueues
as $queue ) {
170 $count +
= $queue->$method();
171 } catch ( JobQueueError
$e ) {
173 $this->logException( $e );
176 $this->throwErrorIfAllPartitionsDown( $failed );
181 protected function doBatchPush( array $jobs, $flags ) {
182 // Local ring variable that may be changed to point to a new ring on failure
183 $partitionRing = $this->partitionRing
;
184 // Try to insert the jobs and update $partitionsTry on any failures.
185 // Retry to insert any remaning jobs again, ignoring the bad partitions.
187 // @codingStandardsIgnoreStart Generic.CodeAnalysis.ForLoopWithTestFunctionCall.NotAllowed
188 for ( $i = $this->maxPartitionsTry
; $i > 0 && count( $jobsLeft ); --$i ) {
189 // @codingStandardsIgnoreEnd
191 $partitionRing->getLiveRing();
192 } catch ( UnexpectedValueException
$e ) {
193 break; // all servers down; nothing to insert to
195 $jobsLeft = $this->tryJobInsertions( $jobsLeft, $partitionRing, $flags );
197 if ( count( $jobsLeft ) ) {
198 throw new JobQueueError(
199 "Could not insert job(s), {$this->maxPartitionsTry} partitions tried." );
205 * @param HashRing $partitionRing
207 * @throws JobQueueError
208 * @return array List of Job object that could not be inserted
210 protected function tryJobInsertions( array $jobs, HashRing
&$partitionRing, $flags ) {
213 // Because jobs are spread across partitions, per-job de-duplication needs
214 // to use a consistent hash to avoid allowing duplicate jobs per partition.
215 // When inserting a batch of de-duplicated jobs, QOS_ATOMIC is disregarded.
216 $uJobsByPartition = []; // (partition name => job list)
218 foreach ( $jobs as $key => $job ) {
219 if ( $job->ignoreDuplicates() ) {
220 $sha1 = sha1( serialize( $job->getDeduplicationInfo() ) );
221 $uJobsByPartition[$partitionRing->getLiveLocation( $sha1 )][] = $job;
222 unset( $jobs[$key] );
225 // Get the batches of jobs that are not de-duplicated
226 if ( $flags & self
::QOS_ATOMIC
) {
227 $nuJobBatches = [ $jobs ]; // all or nothing
229 // Split the jobs into batches and spread them out over servers if there
230 // are many jobs. This helps keep the partitions even. Otherwise, send all
231 // the jobs to a single partition queue to avoids the extra connections.
232 $nuJobBatches = array_chunk( $jobs, 300 );
235 // Insert the de-duplicated jobs into the queues...
236 foreach ( $uJobsByPartition as $partition => $jobBatch ) {
237 /** @var JobQueue $queue */
238 $queue = $this->partitionQueues
[$partition];
241 $queue->doBatchPush( $jobBatch, $flags | self
::QOS_ATOMIC
);
242 } catch ( JobQueueError
$e ) {
244 $this->logException( $e );
247 if ( !$partitionRing->ejectFromLiveRing( $partition, 5 ) ) { // blacklist
248 throw new JobQueueError( "Could not insert job(s), no partitions available." );
250 $jobsLeft = array_merge( $jobsLeft, $jobBatch ); // not inserted
254 // Insert the jobs that are not de-duplicated into the queues...
255 foreach ( $nuJobBatches as $jobBatch ) {
256 $partition = ArrayUtils
::pickRandom( $partitionRing->getLiveLocationWeights() );
257 $queue = $this->partitionQueues
[$partition];
260 $queue->doBatchPush( $jobBatch, $flags | self
::QOS_ATOMIC
);
261 } catch ( JobQueueError
$e ) {
263 $this->logException( $e );
266 if ( !$partitionRing->ejectFromLiveRing( $partition, 5 ) ) { // blacklist
267 throw new JobQueueError( "Could not insert job(s), no partitions available." );
269 $jobsLeft = array_merge( $jobsLeft, $jobBatch ); // not inserted
276 protected function doPop() {
277 $partitionsTry = $this->partitionRing
->getLiveLocationWeights(); // (partition => weight)
280 while ( count( $partitionsTry ) ) {
281 $partition = ArrayUtils
::pickRandom( $partitionsTry );
282 if ( $partition === false ) {
283 break; // all partitions at 0 weight
286 /** @var JobQueue $queue */
287 $queue = $this->partitionQueues
[$partition];
289 $job = $queue->pop();
290 } catch ( JobQueueError
$e ) {
292 $this->logException( $e );
296 $job->metadata
['QueuePartition'] = $partition;
300 unset( $partitionsTry[$partition] ); // blacklist partition
303 $this->throwErrorIfAllPartitionsDown( $failed );
308 protected function doAck( Job
$job ) {
309 if ( !isset( $job->metadata
['QueuePartition'] ) ) {
310 throw new MWException( "The given job has no defined partition name." );
313 $this->partitionQueues
[$job->metadata
['QueuePartition']]->ack( $job );
316 protected function doIsRootJobOldDuplicate( Job
$job ) {
317 $signature = $job->getRootJobParams()['rootJobSignature'];
318 $partition = $this->partitionRing
->getLiveLocation( $signature );
320 return $this->partitionQueues
[$partition]->doIsRootJobOldDuplicate( $job );
321 } catch ( JobQueueError
$e ) {
322 if ( $this->partitionRing
->ejectFromLiveRing( $partition, 5 ) ) {
323 $partition = $this->partitionRing
->getLiveLocation( $signature );
324 return $this->partitionQueues
[$partition]->doIsRootJobOldDuplicate( $job );
331 protected function doDeduplicateRootJob( IJobSpecification
$job ) {
332 $signature = $job->getRootJobParams()['rootJobSignature'];
333 $partition = $this->partitionRing
->getLiveLocation( $signature );
335 return $this->partitionQueues
[$partition]->doDeduplicateRootJob( $job );
336 } catch ( JobQueueError
$e ) {
337 if ( $this->partitionRing
->ejectFromLiveRing( $partition, 5 ) ) {
338 $partition = $this->partitionRing
->getLiveLocation( $signature );
339 return $this->partitionQueues
[$partition]->doDeduplicateRootJob( $job );
346 protected function doDelete() {
348 /** @var JobQueue $queue */
349 foreach ( $this->partitionQueues
as $queue ) {
352 } catch ( JobQueueError
$e ) {
354 $this->logException( $e );
357 $this->throwErrorIfAllPartitionsDown( $failed );
361 protected function doWaitForBackups() {
363 /** @var JobQueue $queue */
364 foreach ( $this->partitionQueues
as $queue ) {
366 $queue->waitForBackups();
367 } catch ( JobQueueError
$e ) {
369 $this->logException( $e );
372 $this->throwErrorIfAllPartitionsDown( $failed );
375 protected function doFlushCaches() {
376 /** @var JobQueue $queue */
377 foreach ( $this->partitionQueues
as $queue ) {
378 $queue->doFlushCaches();
382 public function getAllQueuedJobs() {
383 $iterator = new AppendIterator();
385 /** @var JobQueue $queue */
386 foreach ( $this->partitionQueues
as $queue ) {
387 $iterator->append( $queue->getAllQueuedJobs() );
393 public function getAllDelayedJobs() {
394 $iterator = new AppendIterator();
396 /** @var JobQueue $queue */
397 foreach ( $this->partitionQueues
as $queue ) {
398 $iterator->append( $queue->getAllDelayedJobs() );
404 public function getAllAcquiredJobs() {
405 $iterator = new AppendIterator();
407 /** @var JobQueue $queue */
408 foreach ( $this->partitionQueues
as $queue ) {
409 $iterator->append( $queue->getAllAcquiredJobs() );
415 public function getAllAbandonedJobs() {
416 $iterator = new AppendIterator();
418 /** @var JobQueue $queue */
419 foreach ( $this->partitionQueues
as $queue ) {
420 $iterator->append( $queue->getAllAbandonedJobs() );
426 public function getCoalesceLocationInternal() {
427 return "JobQueueFederated:wiki:{$this->wiki}" .
428 sha1( serialize( array_keys( $this->partitionQueues
) ) );
431 protected function doGetSiblingQueuesWithJobs( array $types ) {
435 /** @var JobQueue $queue */
436 foreach ( $this->partitionQueues
as $queue ) {
438 $nonEmpty = $queue->doGetSiblingQueuesWithJobs( $types );
439 if ( is_array( $nonEmpty ) ) {
440 $result = array_unique( array_merge( $result, $nonEmpty ) );
442 return null; // not supported on all partitions; bail
444 if ( count( $result ) == count( $types ) ) {
445 break; // short-circuit
447 } catch ( JobQueueError
$e ) {
449 $this->logException( $e );
452 $this->throwErrorIfAllPartitionsDown( $failed );
454 return array_values( $result );
457 protected function doGetSiblingQueueSizes( array $types ) {
460 /** @var JobQueue $queue */
461 foreach ( $this->partitionQueues
as $queue ) {
463 $sizes = $queue->doGetSiblingQueueSizes( $types );
464 if ( is_array( $sizes ) ) {
465 foreach ( $sizes as $type => $size ) {
466 $result[$type] = isset( $result[$type] ) ?
$result[$type] +
$size : $size;
469 return null; // not supported on all partitions; bail
471 } catch ( JobQueueError
$e ) {
473 $this->logException( $e );
476 $this->throwErrorIfAllPartitionsDown( $failed );
481 protected function logException( Exception
$e ) {
482 wfDebugLog( 'JobQueueFederated', $e->getMessage() . "\n" . $e->getTraceAsString() );
486 * Throw an error if no partitions available
488 * @param int $down The number of up partitions down
490 * @throws JobQueueError
492 protected function throwErrorIfAllPartitionsDown( $down ) {
493 if ( $down >= count( $this->partitionQueues
) ) {
494 throw new JobQueueError( 'No queue partitions available.' );