25use InvalidArgumentException;
27use UnexpectedValueException;
82 parent::__construct( $params );
83 $section = $params[
'sectionsByWiki'][
$this->domain] ??
'default';
84 if ( !isset( $params[
'partitionsBySection'][$section] ) ) {
85 throw new InvalidArgumentException(
"No configuration for section '$section'." );
87 $this->maxPartitionsTry = $params[
'maxPartitionsTry'] ?? 2;
89 $partitionMap = $params[
'partitionsBySection'][$section];
90 arsort( $partitionMap, SORT_NUMERIC );
92 $baseConfig = $params;
93 foreach ( [
'class',
'sectionsByWiki',
'maxPartitionsTry',
94 'partitionsBySection',
'configByPartition', ] as $o
96 unset( $baseConfig[$o] );
99 foreach ( $partitionMap as $partition => $w ) {
100 if ( !isset( $params[
'configByPartition'][$partition] ) ) {
101 throw new InvalidArgumentException(
"No configuration for partition '$partition'." );
104 $baseConfig + $params[
'configByPartition'][$partition] );
107 $this->partitionRing =
new HashRing( $partitionMap );
112 return [
'undefined',
'random',
'timestamp' ];
120 foreach ( $this->partitionQueues as $queue ) {
121 if ( !$queue->supportsDelayedJobs() ) {
132 foreach ( $this->partitionQueues as $queue ) {
134 $empty = $empty && $queue->doIsEmpty();
169 foreach ( $this->partitionQueues as $queue ) {
171 $count += $queue->$method();
188 for ( $i = $this->maxPartitionsTry; $i > 0 && count( $jobsLeft ); --$i ) {
191 }
catch ( UnexpectedValueException ) {
196 if ( count( $jobsLeft ) ) {
198 "Could not insert job(s), {$this->maxPartitionsTry} partitions tried." );
215 $uJobsByPartition = [];
217 foreach ( $jobs as $key =>
$job ) {
218 if (
$job->ignoreDuplicates() ) {
219 $sha1 = sha1( serialize(
$job->getDeduplicationInfo() ) );
221 unset( $jobs[$key] );
225 if ( $flags & self::QOS_ATOMIC ) {
226 $nuJobBatches = [ $jobs ];
231 $nuJobBatches = array_chunk( $jobs, 300 );
235 foreach ( $uJobsByPartition as $partition => $jobBatch ) {
237 $queue = $this->partitionQueues[$partition];
240 $queue->doBatchPush( $jobBatch, $flags | self::QOS_ATOMIC );
247 throw new JobQueueError(
"Could not insert job(s), no partitions available." );
249 $jobsLeft = array_merge( $jobsLeft, $jobBatch );
254 foreach ( $nuJobBatches as $jobBatch ) {
256 $queue = $this->partitionQueues[$partition];
259 $queue->doBatchPush( $jobBatch, $flags | self::QOS_ATOMIC );
266 throw new JobQueueError(
"Could not insert job(s), no partitions available." );
268 $jobsLeft = array_merge( $jobsLeft, $jobBatch );
276 $partitionsTry = $this->partitionRing->getLiveLocationWeights();
279 while ( count( $partitionsTry ) ) {
280 $partition = ArrayUtils::pickRandom( $partitionsTry );
281 if ( $partition ===
false ) {
286 $queue = $this->partitionQueues[$partition];
288 $job = $queue->pop();
295 $job->setMetadata(
'QueuePartition', $partition );
299 unset( $partitionsTry[$partition] );
308 $partition =
$job->getMetadata(
'QueuePartition' );
309 if ( $partition ===
null ) {
310 throw new UnexpectedValueException(
"The given job has no defined partition name." );
313 $this->partitionQueues[$partition]->ack(
$job );
317 $signature =
$job->getRootJobParams()[
'rootJobSignature'];
318 $partition = $this->partitionRing->getLiveLocation( $signature );
320 return $this->partitionQueues[$partition]->doIsRootJobOldDuplicate(
$job );
322 if ( $this->partitionRing->ejectFromLiveRing( $partition, 5 ) ) {
323 $partition = $this->partitionRing->getLiveLocation( $signature );
324 return $this->partitionQueues[$partition]->doIsRootJobOldDuplicate(
$job );
332 $signature =
$job->getRootJobParams()[
'rootJobSignature'];
333 $partition = $this->partitionRing->getLiveLocation( $signature );
335 return $this->partitionQueues[$partition]->doDeduplicateRootJob(
$job );
337 if ( $this->partitionRing->ejectFromLiveRing( $partition, 5 ) ) {
338 $partition = $this->partitionRing->getLiveLocation( $signature );
339 return $this->partitionQueues[$partition]->doDeduplicateRootJob(
$job );
349 foreach ( $this->partitionQueues as $queue ) {
364 foreach ( $this->partitionQueues as $queue ) {
366 $queue->waitForBackups();
377 foreach ( $this->partitionQueues as $queue ) {
378 $queue->doFlushCaches();
383 $iterator =
new AppendIterator();
386 foreach ( $this->partitionQueues as $queue ) {
387 $iterator->append( $queue->getAllQueuedJobs() );
394 $iterator =
new AppendIterator();
397 foreach ( $this->partitionQueues as $queue ) {
398 $iterator->append( $queue->getAllDelayedJobs() );
405 $iterator =
new AppendIterator();
408 foreach ( $this->partitionQueues as $queue ) {
409 $iterator->append( $queue->getAllAcquiredJobs() );
416 $iterator =
new AppendIterator();
419 foreach ( $this->partitionQueues as $queue ) {
420 $iterator->append( $queue->getAllAbandonedJobs() );
427 return "JobQueueFederated:wiki:{$this->domain}" .
428 sha1( serialize( array_keys( $this->partitionQueues ) ) );
436 foreach ( $this->partitionQueues as $queue ) {
438 $nonEmpty = $queue->doGetSiblingQueuesWithJobs( $types );
439 if ( is_array( $nonEmpty ) ) {
440 $result = array_unique( array_merge( $result, $nonEmpty ) );
444 if ( count( $result ) == count( $types ) ) {
454 return array_values( $result );
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] = ( $result[
$type] ?? 0 ) + $size;
482 wfDebugLog(
'JobQueue', $e->getMessage() .
"\n" . $e->getTraceAsString() );
493 if ( $down >= count( $this->partitionQueues ) ) {
500class_alias( JobQueueFederated::class,
'JobQueueFederated' );
wfDebugLog( $logGroup, $text, $dest='all', array $context=[])
Send a line to a supplementary debug log file, if configured, or main debug log if not.
if(count( $args)< 1) $job