JobQueueFederated.php
No OneTemporary
Actions

Size

15 KB

Referenced Files

None

Subscribers

None

JobQueueFederated.php
View Options

	<?php
	/**
	* 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
	*/

	/**
	* Enqueue and run background jobs via a federated queue, for wiki farms.
	*
	* This class allows for queues to be partitioned into smaller queues.
	* A partition is defined by the configuration for a JobQueue instance.
	* For example, one can set $wgJobTypeConf['refreshLinks'] to point to a
	* JobQueueFederated instance, which itself would consist of three JobQueueRedis
	* instances, each using their own redis server. This would allow for the jobs
	* to be split (evenly or based on weights) across multiple servers if a single
	* server becomes impractical or expensive. Different JobQueue classes can be mixed.
	*
	* The basic queue configuration (e.g. "order", "claimTTL") of a federated queue
	* is inherited by the partition queues. Additional configuration defines what
	* section each wiki is in, what partition queues each section uses (and their weight),
	* and the JobQueue configuration for each partition. Some sections might only need a
	* single queue partition, like the sections for groups of small wikis.
	*
	* If used for performance, then $wgMainCacheType should be set to memcached/redis.
	* Note that "fifo" cannot be used for the ordering, since the data is distributed.
	* One can still use "timestamp" instead, as in "roughly timestamp ordered". Also,
	* queue classes used by this should ignore down servers (with TTL) to avoid slowness.
	*
	* @since 1.22
	* @ingroup JobQueue
	*/
	class JobQueueFederated extends JobQueue {
	/** @var HashRing */
	protected $partitionRing;
	/** @var JobQueue[] (partition name => JobQueue) reverse sorted by weight */
	protected $partitionQueues = [];

	/** @var int Maximum number of partitions to try */
	protected $maxPartitionsTry;

	/**
	* @param array $params Possible keys:
	* - 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 overridden by those passed to this
	* the federated queue itself (e.g. 'order' and 'claimTTL').
	* - maxPartitionsTry : Maximum number of times to attempt job insertion using
	* different partition queues. This improves availability
	* during failure, at the cost of added latency and somewhat
	* less reliable job de-duplication mechanisms.
	*/
	protected function __construct( array $params ) {
	parent::__construct( $params );
	$section = $params['sectionsByWiki'][$this->domain] ?? 'default';
	if ( !isset( $params['partitionsBySection'][$section] ) ) {
	throw new InvalidArgumentException( "No configuration for section '$section'." );
	}
	$this->maxPartitionsTry = $params['maxPartitionsTry'] ?? 2;
	// Get the full partition map
	$partitionMap = $params['partitionsBySection'][$section];
	arsort( $partitionMap, SORT_NUMERIC );
	// Get the config to pass to merge into each partition queue config
	$baseConfig = $params;
	foreach ( [ 'class', 'sectionsByWiki', 'maxPartitionsTry',
	'partitionsBySection', 'configByPartition', ] as $o
	) {
	unset( $baseConfig[$o] ); // partition queue doesn't care about this
	}
	// Get the partition queue objects
	foreach ( $partitionMap as $partition => $w ) {
	if ( !isset( $params['configByPartition'][$partition] ) ) {
	throw new InvalidArgumentException( "No configuration for partition '$partition'." );
	}
	$this->partitionQueues[$partition] = JobQueue::factory(
	$baseConfig + $params['configByPartition'][$partition] );
	}
	// Ring of all partitions
	$this->partitionRing = new HashRing( $partitionMap );
	}

	protected function supportedOrders() {
	// No FIFO due to partitioning, though "rough timestamp order" is supported
	return [ 'undefined', 'random', 'timestamp' ];
	}

	protected function optimalOrder() {
	return 'undefined'; // defer to the partitions
	}

	protected function supportsDelayedJobs() {
	foreach ( $this->partitionQueues as $queue ) {
	if ( !$queue->supportsDelayedJobs() ) {
	return false;
	}
	}

	return true;
	}

	protected function doIsEmpty() {
	$empty = true;
	$failed = 0;
	foreach ( $this->partitionQueues as $queue ) {
	try {
	$empty = $empty && $queue->doIsEmpty();
	} catch ( JobQueueError $e ) {
	++$failed;
	$this->logException( $e );
	}
	}
	$this->throwErrorIfAllPartitionsDown( $failed );

	return $empty;
	}

	protected function doGetSize() {
	return $this->getCrossPartitionSum( 'size', 'doGetSize' );
	}

	protected function doGetAcquiredCount() {
	return $this->getCrossPartitionSum( 'acquiredcount', 'doGetAcquiredCount' );
	}

	protected function doGetDelayedCount() {
	return $this->getCrossPartitionSum( 'delayedcount', 'doGetDelayedCount' );
	}

	protected function doGetAbandonedCount() {
	return $this->getCrossPartitionSum( 'abandonedcount', 'doGetAbandonedCount' );
	}

	/**
	* @param string $type
	* @param string $method
	* @return int
	*/
	protected function getCrossPartitionSum( $type, $method ) {
	$count = 0;
	$failed = 0;
	foreach ( $this->partitionQueues as $queue ) {
	try {
	$count += $queue->$method();
	} catch ( JobQueueError $e ) {
	++$failed;
	$this->logException( $e );
	}
	}
	$this->throwErrorIfAllPartitionsDown( $failed );

	return $count;
	}

	protected function doBatchPush( array $jobs, $flags ) {
	// Local ring variable that may be changed to point to a new ring on failure
	$partitionRing = $this->partitionRing;
	// Try to insert the jobs and update $partitionsTry on any failures.
	// Retry to insert any remaining jobs again, ignoring the bad partitions.
	$jobsLeft = $jobs;
	for ( $i = $this->maxPartitionsTry; $i > 0 && count( $jobsLeft ); --$i ) {
	try {
	$partitionRing->getLiveLocationWeights();
	} catch ( UnexpectedValueException $e ) {
	break; // all servers down; nothing to insert to
	}
	$jobsLeft = $this->tryJobInsertions( $jobsLeft, $partitionRing, $flags );
	}
	if ( count( $jobsLeft ) ) {
	throw new JobQueueError(
	"Could not insert job(s), {$this->maxPartitionsTry} partitions tried." );
	}
	}

	/**
	* @param array $jobs
	* @param HashRing &$partitionRing
	* @param int $flags
	* @throws JobQueueError
	* @return IJobSpecification[] List of Job object that could not be inserted
	*/
	protected function tryJobInsertions( array $jobs, HashRing &$partitionRing, $flags ) {
	$jobsLeft = [];

	// Because jobs are spread across partitions, per-job de-duplication needs
	// to use a consistent hash to avoid allowing duplicate jobs per partition.
	// When inserting a batch of de-duplicated jobs, QOS_ATOMIC is disregarded.
	$uJobsByPartition = []; // (partition name => job list)
	/** @var Job $job */
	foreach ( $jobs as $key => $job ) {
	if ( $job->ignoreDuplicates() ) {
	$sha1 = sha1( serialize( $job->getDeduplicationInfo() ) );
	$uJobsByPartition[$partitionRing->getLiveLocation( $sha1 )][] = $job;
	unset( $jobs[$key] );
	}
	}
	// Get the batches of jobs that are not de-duplicated
	if ( $flags & self::QOS_ATOMIC ) {
	$nuJobBatches = [ $jobs ]; // all or nothing
	} else {
	// Split the jobs into batches and spread them out over servers if there
	// are many jobs. This helps keep the partitions even. Otherwise, send all
	// the jobs to a single partition queue to avoids the extra connections.
	$nuJobBatches = array_chunk( $jobs, 300 );
	}

	// Insert the de-duplicated jobs into the queues...
	foreach ( $uJobsByPartition as $partition => $jobBatch ) {
	/** @var JobQueue $queue */
	$queue = $this->partitionQueues[$partition];
	try {
	$ok = true;
	$queue->doBatchPush( $jobBatch, $flags \| self::QOS_ATOMIC );
	} catch ( JobQueueError $e ) {
	$ok = false;
	$this->logException( $e );
	}
	if ( !$ok ) {
	if ( !$partitionRing->ejectFromLiveRing( $partition, 5 ) ) {
	throw new JobQueueError( "Could not insert job(s), no partitions available." );
	}
	$jobsLeft = array_merge( $jobsLeft, $jobBatch ); // not inserted
	}
	}

	// Insert the jobs that are not de-duplicated into the queues...
	foreach ( $nuJobBatches as $jobBatch ) {
	$partition = ArrayUtils::pickRandom( $partitionRing->getLiveLocationWeights() );
	$queue = $this->partitionQueues[$partition];
	try {
	$ok = true;
	$queue->doBatchPush( $jobBatch, $flags \| self::QOS_ATOMIC );
	} catch ( JobQueueError $e ) {
	$ok = false;
	$this->logException( $e );
	}
	if ( !$ok ) {
	if ( !$partitionRing->ejectFromLiveRing( $partition, 5 ) ) {
	throw new JobQueueError( "Could not insert job(s), no partitions available." );
	}
	$jobsLeft = array_merge( $jobsLeft, $jobBatch ); // not inserted
	}
	}

	return $jobsLeft;
	}

	protected function doPop() {
	$partitionsTry = $this->partitionRing->getLiveLocationWeights(); // (partition => weight)

	$failed = 0;
	while ( count( $partitionsTry ) ) {
	$partition = ArrayUtils::pickRandom( $partitionsTry );
	if ( $partition === false ) {
	break; // all partitions at 0 weight
	}

	/** @var JobQueue $queue */
	$queue = $this->partitionQueues[$partition];
	try {
	$job = $queue->pop();
	} catch ( JobQueueError $e ) {
	++$failed;
	$this->logException( $e );
	$job = false;
	}
	if ( $job ) {
	$job->setMetadata( 'QueuePartition', $partition );

	return $job;
	} else {
	unset( $partitionsTry[$partition] );
	}
	}
	$this->throwErrorIfAllPartitionsDown( $failed );

	return false;
	}

	protected function doAck( RunnableJob $job ) {
	$partition = $job->getMetadata( 'QueuePartition' );
	if ( $partition === null ) {
	throw new UnexpectedValueException( "The given job has no defined partition name." );
	}

	$this->partitionQueues[$partition]->ack( $job );
	}

	protected function doIsRootJobOldDuplicate( IJobSpecification $job ) {
	$signature = $job->getRootJobParams()['rootJobSignature'];
	$partition = $this->partitionRing->getLiveLocation( $signature );
	try {
	return $this->partitionQueues[$partition]->doIsRootJobOldDuplicate( $job );
	} catch ( JobQueueError $e ) {
	if ( $this->partitionRing->ejectFromLiveRing( $partition, 5 ) ) {
	$partition = $this->partitionRing->getLiveLocation( $signature );
	return $this->partitionQueues[$partition]->doIsRootJobOldDuplicate( $job );
	}
	}

	return false;
	}

	protected function doDeduplicateRootJob( IJobSpecification $job ) {
	$signature = $job->getRootJobParams()['rootJobSignature'];
	$partition = $this->partitionRing->getLiveLocation( $signature );
	try {
	return $this->partitionQueues[$partition]->doDeduplicateRootJob( $job );
	} catch ( JobQueueError $e ) {
	if ( $this->partitionRing->ejectFromLiveRing( $partition, 5 ) ) {
	$partition = $this->partitionRing->getLiveLocation( $signature );
	return $this->partitionQueues[$partition]->doDeduplicateRootJob( $job );
	}
	}

	return false;
	}

	protected function doDelete() {
	$failed = 0;
	/** @var JobQueue $queue */
	foreach ( $this->partitionQueues as $queue ) {
	try {
	$queue->doDelete();
	} catch ( JobQueueError $e ) {
	++$failed;
	$this->logException( $e );
	}
	}
	$this->throwErrorIfAllPartitionsDown( $failed );
	return true;
	}

	protected function doWaitForBackups() {
	$failed = 0;
	/** @var JobQueue $queue */
	foreach ( $this->partitionQueues as $queue ) {
	try {
	$queue->waitForBackups();
	} catch ( JobQueueError $e ) {
	++$failed;
	$this->logException( $e );
	}
	}
	$this->throwErrorIfAllPartitionsDown( $failed );
	}

	protected function doFlushCaches() {
	/** @var JobQueue $queue */
	foreach ( $this->partitionQueues as $queue ) {
	$queue->doFlushCaches();
	}
	}

	public function getAllQueuedJobs() {
	$iterator = new AppendIterator();

	/** @var JobQueue $queue */
	foreach ( $this->partitionQueues as $queue ) {
	$iterator->append( $queue->getAllQueuedJobs() );
	}

	return $iterator;
	}

	public function getAllDelayedJobs() {
	$iterator = new AppendIterator();

	/** @var JobQueue $queue */
	foreach ( $this->partitionQueues as $queue ) {
	$iterator->append( $queue->getAllDelayedJobs() );
	}

	return $iterator;
	}

	public function getAllAcquiredJobs() {
	$iterator = new AppendIterator();

	/** @var JobQueue $queue */
	foreach ( $this->partitionQueues as $queue ) {
	$iterator->append( $queue->getAllAcquiredJobs() );
	}

	return $iterator;
	}

	public function getAllAbandonedJobs() {
	$iterator = new AppendIterator();

	/** @var JobQueue $queue */
	foreach ( $this->partitionQueues as $queue ) {
	$iterator->append( $queue->getAllAbandonedJobs() );
	}

	return $iterator;
	}

	public function getCoalesceLocationInternal() {
	return "JobQueueFederated:wiki:{$this->domain}" .
	sha1( serialize( array_keys( $this->partitionQueues ) ) );
	}

	protected function doGetSiblingQueuesWithJobs( array $types ) {
	$result = [];

	$failed = 0;
	/** @var JobQueue $queue */
	foreach ( $this->partitionQueues as $queue ) {
	try {
	$nonEmpty = $queue->doGetSiblingQueuesWithJobs( $types );
	if ( is_array( $nonEmpty ) ) {
	$result = array_unique( array_merge( $result, $nonEmpty ) );
	} else {
	return null; // not supported on all partitions; bail
	}
	if ( count( $result ) == count( $types ) ) {
	break; // short-circuit
	}
	} catch ( JobQueueError $e ) {
	++$failed;
	$this->logException( $e );
	}
	}
	$this->throwErrorIfAllPartitionsDown( $failed );

	return array_values( $result );
	}

	protected function doGetSiblingQueueSizes( array $types ) {
	$result = [];
	$failed = 0;
	/** @var JobQueue $queue */
	foreach ( $this->partitionQueues as $queue ) {
	try {
	$sizes = $queue->doGetSiblingQueueSizes( $types );
	if ( is_array( $sizes ) ) {
	foreach ( $sizes as $type => $size ) {
	$result[$type] = ( $result[$type] ?? 0 ) + $size;
	}
	} else {
	return null; // not supported on all partitions; bail
	}
	} catch ( JobQueueError $e ) {
	++$failed;
	$this->logException( $e );
	}
	}
	$this->throwErrorIfAllPartitionsDown( $failed );

	return $result;
	}

	protected function logException( Exception $e ) {
	wfDebugLog( 'JobQueue', $e->getMessage() . "\n" . $e->getTraceAsString() );
	}

	/**
	* Throw an error if no partitions available
	*
	* @param int $down The number of up partitions down
	* @return void
	* @throws JobQueueError
	*/
	protected function throwErrorIfAllPartitionsDown( $down ) {
	if ( $down >= count( $this->partitionQueues ) ) {
	throw new JobQueueError( 'No queue partitions available.' );
	}
	}
	}

File Metadata

Mime Type: text/x-php
Expires: Aug 20 2026, 02:31 (11 w, 18 h ago)
Storage Engine: local-disk
Storage Format: Raw Data
Storage Handle: 8f/e5/afe124eb49a1586337a43ccfef66
Default Alt Text: JobQueueFederated.php (15 KB)

JobQueueFederated.phpNo OneTemporaryActions

JobQueueFederated.phpView Options

File Metadata

Event Timeline

JobQueueFederated.php
No OneTemporary
Actions

JobQueueFederated.php
View Options