JobQueueFederated.php

Go to the documentation of this file.

<?php
class JobQueueFederated extends JobQueue {
    protected $partitionMap = array();

    protected $partitionQueues = array();

    protected $partitionPushRing;

    protected $cache;

    protected $maxPartitionsTry;

    const CACHE_TTL_SHORT = 30; // integer; seconds to cache info without re-validating
    const CACHE_TTL_LONG = 300; // integer; seconds to cache info that is kept up to date

    protected function __construct( array $params ) {
        parent::__construct( $params );
        $section = isset( $params['sectionsByWiki'][$this->wiki] )
            ? $params['sectionsByWiki'][$this->wiki]
            : 'default';
        if ( !isset( $params['partitionsBySection'][$section] ) ) {
            throw new MWException( "No configuration for section '$section'." );
        }
        $this->maxPartitionsTry = isset( $params['maxPartitionsTry'] )
            ? $params['maxPartitionsTry']
            : 2;
        // Get the full partition map
        $this->partitionMap = $params['partitionsBySection'][$section];
        arsort( $this->partitionMap, SORT_NUMERIC );
        // Get the partitions jobs can actually be pushed to
        $partitionPushMap = $this->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
        ) {
            unset( $baseConfig[$o] ); // partition queue doesn't care about this
        }
        // Get the partition queue objects
        foreach ( $this->partitionMap as $partition => $w ) {
            if ( !isset( $params['configByPartition'][$partition] ) ) {
                throw new MWException( "No configuration for partition '$partition'." );
            }
            $this->partitionQueues[$partition] = JobQueue::factory(
                $baseConfig + $params['configByPartition'][$partition] );
        }
        // Get the ring of partitions to push jobs into
        $this->partitionPushRing = new HashRing( $partitionPushMap );
        // Aggregate cache some per-queue values if there are multiple partition queues
        $this->cache = count( $this->partitionMap ) > 1 ? wfGetMainCache() : new EmptyBagOStuff();
    }

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

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

    protected function supportsDelayedJobs() {
        return true; // defer checks to the partitions
    }

    protected function doIsEmpty() {
        $key = $this->getCacheKey( 'empty' );

        $isEmpty = $this->cache->get( $key );
        if ( $isEmpty === 'true' ) {
            return true;
        } elseif ( $isEmpty === 'false' ) {
            return false;
        }

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

        $this->cache->add( $key, $empty ? 'true' : 'false', self::CACHE_TTL_LONG );
        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' );
    }

    protected function getCrossPartitionSum( $type, $method ) {
        $key = $this->getCacheKey( $type );

        $count = $this->cache->get( $key );
        if ( is_int( $count ) ) {
            return $count;
        }

        $failed = 0;
        foreach ( $this->partitionQueues as $queue ) {
            try {
                $count += $queue->$method();
            } catch ( JobQueueError $e ) {
                ++$failed;
                MWExceptionHandler::logException( $e );
            }
        }
        $this->throwErrorIfAllPartitionsDown( $failed );

        $this->cache->set( $key, $count, self::CACHE_TTL_SHORT );

        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->partitionPushRing;
        // Try to insert the jobs and update $partitionsTry on any failures.
        // Retry to insert any remaning jobs again, ignoring the bad partitions.
        $jobsLeft = $jobs;
        for ( $i = $this->maxPartitionsTry; $i > 0 && count( $jobsLeft ); --$i ) {
            $jobsLeft = $this->tryJobInsertions( $jobsLeft, $partitionRing, $flags );
        }
        if ( count( $jobsLeft ) ) {
            throw new JobQueueError(
                "Could not insert job(s), {$this->maxPartitionsTry} partitions tried." );
        }

        return true;
    }

    protected function tryJobInsertions( array $jobs, HashRing &$partitionRing, $flags ) {
        $jobsLeft = array();

        // 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 = array(); // (partition name => job list)
        foreach ( $jobs as $key => $job ) {
            if ( $job->ignoreDuplicates() ) {
                $sha1 = sha1( serialize( $job->getDeduplicationInfo() ) );
                $uJobsByPartition[$partitionRing->getLocation( $sha1 )][] = $job;
                unset( $jobs[$key] );
            }
        }
        // Get the batches of jobs that are not de-duplicated
        if ( $flags & self::QOS_ATOMIC ) {
            $nuJobBatches = array( $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 ) {
            $queue = $this->partitionQueues[$partition];
            try {
                $ok = $queue->doBatchPush( $jobBatch, $flags | self::QOS_ATOMIC );
            } catch ( JobQueueError $e ) {
                $ok = false;
                MWExceptionHandler::logException( $e );
            }
            if ( $ok ) {
                $key = $this->getCacheKey( 'empty' );
                $this->cache->set( $key, 'false', JobQueueDB::CACHE_TTL_LONG );
            } else {
                $partitionRing = $partitionRing->newWithoutLocation( $partition ); // blacklist
                if ( !$partitionRing ) {
                    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->getLocationWeights() );
            $queue = $this->partitionQueues[$partition];
            try {
                $ok = $queue->doBatchPush( $jobBatch, $flags | self::QOS_ATOMIC );
            } catch ( JobQueueError $e ) {
                $ok = false;
                MWExceptionHandler::logException( $e );
            }
            if ( $ok ) {
                $key = $this->getCacheKey( 'empty' );
                $this->cache->set( $key, 'false', JobQueueDB::CACHE_TTL_LONG );
            } else {
                $partitionRing = $partitionRing->newWithoutLocation( $partition ); // blacklist
                if ( !$partitionRing ) {
                    throw new JobQueueError( "Could not insert job(s), no partitions available." );
                }
                $jobsLeft = array_merge( $jobsLeft, $jobBatch ); // not inserted
            }
        }

        return $jobsLeft;
    }

    protected function doPop() {
        $key = $this->getCacheKey( 'empty' );

        $isEmpty = $this->cache->get( $key );
        if ( $isEmpty === 'true' ) {
            return false;
        }

        $partitionsTry = $this->partitionMap; // (partition => weight)

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

            $queue = $this->partitionQueues[$partition];
            try {
                $job = $queue->pop();
            } catch ( JobQueueError $e ) {
                ++$failed;
                MWExceptionHandler::logException( $e );
                $job = false;
            }
            if ( $job ) {
                $job->metadata['QueuePartition'] = $partition;

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

        $this->cache->set( $key, 'true', JobQueueDB::CACHE_TTL_LONG );

        return false;
    }

    protected function doAck( Job $job ) {
        if ( !isset( $job->metadata['QueuePartition'] ) ) {
            throw new MWException( "The given job has no defined partition name." );
        }

        return $this->partitionQueues[$job->metadata['QueuePartition']]->ack( $job );
    }

    protected function doIsRootJobOldDuplicate( Job $job ) {
        $params = $job->getRootJobParams();
        $partitions = $this->partitionPushRing->getLocations( $params['rootJobSignature'], 2 );
        try {
            return $this->partitionQueues[$partitions[0]]->doIsRootJobOldDuplicate( $job );
        } catch ( JobQueueError $e ) {
            if ( isset( $partitions[1] ) ) { // check fallback partition
                return $this->partitionQueues[$partitions[1]]->doIsRootJobOldDuplicate( $job );
            }
        }

        return false;
    }

    protected function doDeduplicateRootJob( Job $job ) {
        $params = $job->getRootJobParams();
        $partitions = $this->partitionPushRing->getLocations( $params['rootJobSignature'], 2 );
        try {
            return $this->partitionQueues[$partitions[0]]->doDeduplicateRootJob( $job );
        } catch ( JobQueueError $e ) {
            if ( isset( $partitions[1] ) ) { // check fallback partition
                return $this->partitionQueues[$partitions[1]]->doDeduplicateRootJob( $job );
            }
        }

        return false;
    }

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

    protected function doWaitForBackups() {
        $failed = 0;
        foreach ( $this->partitionQueues as $queue ) {
            try {
                $queue->waitForBackups();
            } catch ( JobQueueError $e ) {
                ++$failed;
                MWExceptionHandler::logException( $e );
            }
        }
        $this->throwErrorIfAllPartitionsDown( $failed );
    }

    protected function doGetPeriodicTasks() {
        $tasks = array();
        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'
        );

        foreach ( $types as $type ) {
            $this->cache->delete( $this->getCacheKey( $type ) );
        }

        foreach ( $this->partitionQueues as $queue ) {
            $queue->doFlushCaches();
        }
    }

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

        foreach ( $this->partitionQueues as $queue ) {
            $iterator->append( $queue->getAllQueuedJobs() );
        }

        return $iterator;
    }

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

        foreach ( $this->partitionQueues as $queue ) {
            $iterator->append( $queue->getAllDelayedJobs() );
        }

        return $iterator;
    }

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

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

        $failed = 0;
        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;
                MWExceptionHandler::logException( $e );
            }
        }
        $this->throwErrorIfAllPartitionsDown( $failed );

        return array_values( $result );
    }

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

        return $result;
    }

    protected function throwErrorIfAllPartitionsDown( $down ) {
        if ( $down >= count( $this->partitionQueues ) ) {
            throw new JobQueueError( 'No queue partitions available.' );
        }
    }

    public function setTestingPrefix( $key ) {
        foreach ( $this->partitionQueues as $queue ) {
            $queue->setTestingPrefix( $key );
        }
    }

    private function getCacheKey( $property ) {
        list( $db, $prefix ) = wfSplitWikiID( $this->wiki );

        return wfForeignMemcKey( $db, $prefix, 'jobqueue', $this->type, $property );
    }
}