Merge "[debug] Remove BC code from AvroFormatter"

[lhc/web/wiklou.git] / includes / jobqueue / JobQueueFederated.php

diff --git a/includes/jobqueue/JobQueueFederated.php b/includes/jobqueue/JobQueueFederated.php
index 66bbb9c..07f2c93 100644 (file)
--- a/includes/jobqueue/JobQueueFederated.php
+++ b/includes/jobqueue/JobQueueFederated.php
@@ -49,8 +49,6 @@
 class JobQueueFederated extends JobQueue {
        /** @var HashRing */
        protected $partitionRing;
-       /** @var HashRing */
-       protected $partitionPushRing;
        /** @var array (partition name => JobQueue) reverse sorted by weight */
        protected $partitionQueues = array();
 
@@ -68,8 +66,6 @@ class JobQueueFederated extends JobQueue {
         *                          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').
-        *  - partitionsNoPush    : List of partition names that can handle pop() but not push().
-        *                          This can be used to migrate away from a certain partition.
         *  - 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
@@ -90,20 +86,15 @@ class JobQueueFederated extends JobQueue {
                // Get the full partition map
                $partitionMap = $params['partitionsBySection'][$section];
                arsort( $partitionMap, SORT_NUMERIC );
-               // Get the partitions jobs can actually be pushed to
-               $partitionPushMap = $partitionMap;
-               if ( isset( $params['partitionsNoPush'] ) ) {
-                       foreach ( $params['partitionsNoPush'] as $partition ) {
-                               unset( $partitionPushMap[$partition] );
-                       }
-               }
                // Get the config to pass to merge into each partition queue config
                $baseConfig = $params;
                foreach ( array( 'class', 'sectionsByWiki', 'maxPartitionsTry',
-                       'partitionsBySection', 'configByPartition', 'partitionsNoPush' ) as $o
+                       'partitionsBySection', 'configByPartition', ) as $o
                ) {
                        unset( $baseConfig[$o] ); // partition queue doesn't care about this
                }
+               // The class handles all aggregator calls already
+               unset( $baseConfig['aggregator'] );
                // Get the partition queue objects
                foreach ( $partitionMap as $partition => $w ) {
                        if ( !isset( $params['configByPartition'][$partition] ) ) {
@@ -114,12 +105,6 @@ class JobQueueFederated extends JobQueue {
                }
                // Ring of all partitions
                $this->partitionRing = new HashRing( $partitionMap );
-               // Get the ring of partitions to push jobs into
-               if ( count( $partitionPushMap ) === count( $partitionMap ) ) {
-                       $this->partitionPushRing = clone $this->partitionRing; // faster
-               } else {
-                       $this->partitionPushRing = new HashRing( $partitionPushMap );
-               }
        }
 
        protected function supportedOrders() {
@@ -196,7 +181,7 @@ class JobQueueFederated extends JobQueue {
 
        protected function doBatchPush( array $jobs, $flags ) {
                // Local ring variable that may be changed to point to a new ring on failure
-               $partitionRing = $this->partitionPushRing;
+               $partitionRing = $this->partitionRing;
                // Try to insert the jobs and update $partitionsTry on any failures.
                // Retry to insert any remaning jobs again, ignoring the bad partitions.
                $jobsLeft = $jobs;
@@ -332,12 +317,12 @@ class JobQueueFederated extends JobQueue {
        protected function doIsRootJobOldDuplicate( Job $job ) {
                $params = $job->getRootJobParams();
                $sigature = $params['rootJobSignature'];
-               $partition = $this->partitionPushRing->getLiveLocation( $sigature );
+               $partition = $this->partitionRing->getLiveLocation( $sigature );
                try {
                        return $this->partitionQueues[$partition]->doIsRootJobOldDuplicate( $job );
                } catch ( JobQueueError $e ) {
-                       if ( $this->partitionPushRing->ejectFromLiveRing( $partition, 5 ) ) {
-                               $partition = $this->partitionPushRing->getLiveLocation( $sigature );
+                       if ( $this->partitionRing->ejectFromLiveRing( $partition, 5 ) ) {
+                               $partition = $this->partitionRing->getLiveLocation( $sigature );
                                return $this->partitionQueues[$partition]->doIsRootJobOldDuplicate( $job );
                        }
                }
@@ -345,15 +330,15 @@ class JobQueueFederated extends JobQueue {
                return false;
        }
 
-       protected function doDeduplicateRootJob( Job $job ) {
+       protected function doDeduplicateRootJob( IJobSpecification $job ) {
                $params = $job->getRootJobParams();
                $sigature = $params['rootJobSignature'];
-               $partition = $this->partitionPushRing->getLiveLocation( $sigature );
+               $partition = $this->partitionRing->getLiveLocation( $sigature );
                try {
                        return $this->partitionQueues[$partition]->doDeduplicateRootJob( $job );
                } catch ( JobQueueError $e ) {
-                       if ( $this->partitionPushRing->ejectFromLiveRing( $partition, 5 ) ) {
-                               $partition = $this->partitionPushRing->getLiveLocation( $sigature );
+                       if ( $this->partitionRing->ejectFromLiveRing( $partition, 5 ) ) {
+                               $partition = $this->partitionRing->getLiveLocation( $sigature );
                                return $this->partitionQueues[$partition]->doDeduplicateRootJob( $job );
                        }
                }
@@ -390,27 +375,7 @@ class JobQueueFederated extends JobQueue {
                $this->throwErrorIfAllPartitionsDown( $failed );
        }
 
-       protected function doGetPeriodicTasks() {
-               $tasks = array();
-               /** @var JobQueue $queue */
-               foreach ( $this->partitionQueues as $partition => $queue ) {
-                       foreach ( $queue->getPeriodicTasks() as $task => $def ) {
-                               $tasks["{$partition}:{$task}"] = $def;
-                       }
-               }
-
-               return $tasks;
-       }
-
        protected function doFlushCaches() {
-               static $types = array(
-                       'empty',
-                       'size',
-                       'acquiredcount',
-                       'delayedcount',
-                       'abandonedcount'
-               );
-
                /** @var JobQueue $queue */
                foreach ( $this->partitionQueues as $queue ) {
                        $queue->doFlushCaches();
@@ -439,6 +404,28 @@ class JobQueueFederated extends JobQueue {
                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->wiki}" .
                        sha1( serialize( array_keys( $this->partitionQueues ) ) );
@@ -510,11 +497,4 @@ class JobQueueFederated extends JobQueue {
                        throw new JobQueueError( 'No queue partitions available.' );
                }
        }
-
-       public function setTestingPrefix( $key ) {
-               /** @var JobQueue $queue */
-               foreach ( $this->partitionQueues as $queue ) {
-                       $queue->setTestingPrefix( $key );
-               }
-       }
 }