JobQueue.php

Go to the documentation of this file.

<?php
abstract class JobQueue {
    protected $wiki;

    protected $type;

    protected $order;

    protected $claimTTL;

    protected $maxTries;

    protected $checkDelay;

    protected $dupCache;

    const QOS_ATOMIC = 1; // integer; "all-or-nothing" job insertions

    const ROOTJOB_TTL = 2419200; // integer; seconds to remember root jobs (28 days)

    protected function __construct( array $params ) {
        $this->wiki = $params['wiki'];
        $this->type = $params['type'];
        $this->claimTTL = isset( $params['claimTTL'] ) ? $params['claimTTL'] : 0;
        $this->maxTries = isset( $params['maxTries'] ) ? $params['maxTries'] : 3;
        if ( isset( $params['order'] ) && $params['order'] !== 'any' ) {
            $this->order = $params['order'];
        } else {
            $this->order = $this->optimalOrder();
        }
        if ( !in_array( $this->order, $this->supportedOrders() ) ) {
            throw new MWException( __CLASS__ . " does not support '{$this->order}' order." );
        }
        $this->checkDelay = !empty( $params['checkDelay'] );
        if ( $this->checkDelay && !$this->supportsDelayedJobs() ) {
            throw new MWException( __CLASS__ . " does not support delayed jobs." );
        }
        $this->dupCache = wfGetCache( CACHE_ANYTHING );
    }

    final public static function factory( array $params ) {
        $class = $params['class'];
        if ( !class_exists( $class ) ) {
            throw new MWException( "Invalid job queue class '$class'." );
        }
        $obj = new $class( $params );
        if ( !( $obj instanceof self ) ) {
            throw new MWException( "Class '$class' is not a " . __CLASS__ . " class." );
        }

        return $obj;
    }

    final public function getWiki() {
        return $this->wiki;
    }

    final public function getType() {
        return $this->type;
    }

    final public function getOrder() {
        return $this->order;
    }

    final public function delayedJobsEnabled() {
        return $this->checkDelay;
    }

    abstract protected function supportedOrders();

    abstract protected function optimalOrder();

    protected function supportsDelayedJobs() {
        return false; // not implemented
    }

    final public function isEmpty() {
        wfProfileIn( __METHOD__ );
        $res = $this->doIsEmpty();
        wfProfileOut( __METHOD__ );

        return $res;
    }

    abstract protected function doIsEmpty();

    final public function getSize() {
        wfProfileIn( __METHOD__ );
        $res = $this->doGetSize();
        wfProfileOut( __METHOD__ );

        return $res;
    }

    abstract protected function doGetSize();

    final public function getAcquiredCount() {
        wfProfileIn( __METHOD__ );
        $res = $this->doGetAcquiredCount();
        wfProfileOut( __METHOD__ );

        return $res;
    }

    abstract protected function doGetAcquiredCount();

    final public function getDelayedCount() {
        wfProfileIn( __METHOD__ );
        $res = $this->doGetDelayedCount();
        wfProfileOut( __METHOD__ );

        return $res;
    }

    protected function doGetDelayedCount() {
        return 0; // not implemented
    }

    final public function getAbandonedCount() {
        wfProfileIn( __METHOD__ );
        $res = $this->doGetAbandonedCount();
        wfProfileOut( __METHOD__ );

        return $res;
    }

    protected function doGetAbandonedCount() {
        return 0; // not implemented
    }

    final public function push( $jobs, $flags = 0 ) {
        return $this->batchPush( is_array( $jobs ) ? $jobs : array( $jobs ), $flags );
    }

    final public function batchPush( array $jobs, $flags = 0 ) {
        if ( !count( $jobs ) ) {
            return true; // nothing to do
        }

        foreach ( $jobs as $job ) {
            if ( $job->getType() !== $this->type ) {
                throw new MWException(
                    "Got '{$job->getType()}' job; expected a '{$this->type}' job." );
            } elseif ( $job->getReleaseTimestamp() && !$this->checkDelay ) {
                throw new MWException(
                    "Got delayed '{$job->getType()}' job; delays are not supported." );
            }
        }

        wfProfileIn( __METHOD__ );
        $ok = $this->doBatchPush( $jobs, $flags );
        wfProfileOut( __METHOD__ );

        return $ok;
    }

    abstract protected function doBatchPush( array $jobs, $flags );

    final public function pop() {
        global $wgJobClasses;

        if ( $this->wiki !== wfWikiID() ) {
            throw new MWException( "Cannot pop '{$this->type}' job off foreign wiki queue." );
        } elseif ( !isset( $wgJobClasses[$this->type] ) ) {
            // Do not pop jobs if there is no class for the queue type
            throw new MWException( "Unrecognized job type '{$this->type}'." );
        }

        wfProfileIn( __METHOD__ );
        $job = $this->doPop();
        wfProfileOut( __METHOD__ );

        // Flag this job as an old duplicate based on its "root" job...
        try {
            if ( $job && $this->isRootJobOldDuplicate( $job ) ) {
                JobQueue::incrStats( 'job-pop-duplicate', $this->type, 1, $this->wiki );
                $job = DuplicateJob::newFromJob( $job ); // convert to a no-op
            }
        } catch ( MWException $e ) {
            // don't lose jobs over this
        }

        return $job;
    }

    abstract protected function doPop();

    final public function ack( Job $job ) {
        if ( $job->getType() !== $this->type ) {
            throw new MWException( "Got '{$job->getType()}' job; expected '{$this->type}'." );
        }
        wfProfileIn( __METHOD__ );
        $ok = $this->doAck( $job );
        wfProfileOut( __METHOD__ );

        return $ok;
    }

    abstract protected function doAck( Job $job );

    final public function deduplicateRootJob( Job $job ) {
        if ( $job->getType() !== $this->type ) {
            throw new MWException( "Got '{$job->getType()}' job; expected '{$this->type}'." );
        }
        wfProfileIn( __METHOD__ );
        $ok = $this->doDeduplicateRootJob( $job );
        wfProfileOut( __METHOD__ );

        return $ok;
    }

    protected function doDeduplicateRootJob( Job $job ) {
        if ( !$job->hasRootJobParams() ) {
            throw new MWException( "Cannot register root job; missing parameters." );
        }
        $params = $job->getRootJobParams();

        $key = $this->getRootJobCacheKey( $params['rootJobSignature'] );
        // Callers should call batchInsert() and then this function so that if the insert
        // fails, the de-duplication registration will be aborted. Since the insert is
        // deferred till "transaction idle", do the same here, so that the ordering is
        // maintained. Having only the de-duplication registration succeed would cause
        // jobs to become no-ops without any actual jobs that made them redundant.
        $timestamp = $this->dupCache->get( $key ); // current last timestamp of this job
        if ( $timestamp && $timestamp >= $params['rootJobTimestamp'] ) {
            return true; // a newer version of this root job was enqueued
        }

        // Update the timestamp of the last root job started at the location...
        return $this->dupCache->set( $key, $params['rootJobTimestamp'], JobQueueDB::ROOTJOB_TTL );
    }

    final protected function isRootJobOldDuplicate( Job $job ) {
        if ( $job->getType() !== $this->type ) {
            throw new MWException( "Got '{$job->getType()}' job; expected '{$this->type}'." );
        }
        wfProfileIn( __METHOD__ );
        $isDuplicate = $this->doIsRootJobOldDuplicate( $job );
        wfProfileOut( __METHOD__ );

        return $isDuplicate;
    }

    protected function doIsRootJobOldDuplicate( Job $job ) {
        if ( !$job->hasRootJobParams() ) {
            return false; // job has no de-deplication info
        }
        $params = $job->getRootJobParams();

        $key = $this->getRootJobCacheKey( $params['rootJobSignature'] );
        // Get the last time this root job was enqueued
        $timestamp = $this->dupCache->get( $key );

        // Check if a new root job was started at the location after this one's...
        return ( $timestamp && $timestamp > $params['rootJobTimestamp'] );
    }

    protected function getRootJobCacheKey( $signature ) {
        list( $db, $prefix ) = wfSplitWikiID( $this->wiki );

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

    final public function delete() {
        wfProfileIn( __METHOD__ );
        $res = $this->doDelete();
        wfProfileOut( __METHOD__ );

        return $res;
    }

    protected function doDelete() {
        throw new MWException( "This method is not implemented." );
    }

    final public function waitForBackups() {
        wfProfileIn( __METHOD__ );
        $this->doWaitForBackups();
        wfProfileOut( __METHOD__ );
    }

    protected function doWaitForBackups() {
    }

    final public function getPeriodicTasks() {
        $tasks = $this->doGetPeriodicTasks();
        foreach ( $tasks as $name => &$def ) {
            $def['name'] = $name;
        }

        return $tasks;
    }

    protected function doGetPeriodicTasks() {
        return array();
    }

    final public function flushCaches() {
        wfProfileIn( __METHOD__ );
        $this->doFlushCaches();
        wfProfileOut( __METHOD__ );
    }

    protected function doFlushCaches() {
    }

    abstract public function getAllQueuedJobs();

    public function getAllDelayedJobs() {
        return new ArrayIterator( array() ); // not implemented
    }

    public function getCoalesceLocationInternal() {
        return null;
    }

    final public function getSiblingQueuesWithJobs( array $types ) {
        $section = new ProfileSection( __METHOD__ );

        return $this->doGetSiblingQueuesWithJobs( $types );
    }

    protected function doGetSiblingQueuesWithJobs( array $types ) {
        return null; // not supported
    }

    final public function getSiblingQueueSizes( array $types ) {
        $section = new ProfileSection( __METHOD__ );

        return $this->doGetSiblingQueueSizes( $types );
    }

    protected function doGetSiblingQueueSizes( array $types ) {
        return null; // not supported
    }

    public static function incrStats( $key, $type, $delta = 1, $wiki = null ) {
        wfIncrStats( $key, $delta );
        wfIncrStats( "{$key}-{$type}", $delta );
        if ( $wiki !== null ) {
            wfIncrStats( "{$key}-{$type}-{$wiki}", $delta );
        }
    }

    public function setTestingPrefix( $key ) {
        throw new MWException( "Queue namespacing not supported for this queue type." );
    }
}

class JobQueueError extends MWException {
}

class JobQueueConnectionError extends JobQueueError {
}