JobQueue.php

Go to the documentation of this file.

<?php
use Liuggio\StatsdClient\Factory\StatsdDataFactoryInterface;
 
abstract class JobQueue {
    protected $domain;
    protected $type;
    protected $order;
    protected $claimTTL;
    protected $maxTries;
    protected $readOnlyReason;
    protected $stats;
 
    protected $wanCache;
 
    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->domain = $params['domain'] ?? $params['wiki']; // b/c
        $this->type = $params['type'];
        $this->claimTTL = $params['claimTTL'] ?? 0;
        $this->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 JobQueueError( __CLASS__ . " does not support '{$this->order}' order." );
        }
        $this->readOnlyReason = $params['readOnlyReason'] ?? false;
        $this->stats = $params['stats'] ?? new NullStatsdDataFactory();
        $this->wanCache = $params['wanCache'] ?? WANObjectCache::newEmpty();
    }
 
    final public static function factory( array $params ) {
        $class = $params['class'];
        if ( !class_exists( $class ) ) {
            throw new JobQueueError( "Invalid job queue class '$class'." );
        }
        $obj = new $class( $params );
        if ( !( $obj instanceof self ) ) {
            throw new JobQueueError( "Class '$class' is not a " . __CLASS__ . " class." );
        }
 
        return $obj;
    }
 
    final public function getDomain() {
        return $this->domain;
    }
 
    final public function getWiki() {
        return WikiMap::getWikiIdFromDbDomain( $this->domain );
    }
 
    final public function getType() {
        return $this->type;
    }
 
    final public function getOrder() {
        return $this->order;
    }
 
    abstract protected function supportedOrders();
 
    abstract protected function optimalOrder();
 
    protected function supportsDelayedJobs() {
        return false; // not implemented
    }
 
    final public function delayedJobsEnabled() {
        return $this->supportsDelayedJobs();
    }
 
    public function getReadOnlyReason() {
        return $this->readOnlyReason;
    }
 
    final public function isEmpty() {
        $res = $this->doIsEmpty();
 
        return $res;
    }
 
    abstract protected function doIsEmpty();
 
    final public function getSize() {
        $res = $this->doGetSize();
 
        return $res;
    }
 
    abstract protected function doGetSize();
 
    final public function getAcquiredCount() {
        $res = $this->doGetAcquiredCount();
 
        return $res;
    }
 
    abstract protected function doGetAcquiredCount();
 
    final public function getDelayedCount() {
        $res = $this->doGetDelayedCount();
 
        return $res;
    }
 
    protected function doGetDelayedCount() {
        return 0; // not implemented
    }
 
    final public function getAbandonedCount() {
        $res = $this->doGetAbandonedCount();
 
        return $res;
    }
 
    protected function doGetAbandonedCount() {
        return 0; // not implemented
    }
 
    final public function push( $jobs, $flags = 0 ) {
        $jobs = is_array( $jobs ) ? $jobs : [ $jobs ];
        $this->batchPush( $jobs, $flags );
    }
 
    final public function batchPush( array $jobs, $flags = 0 ) {
        $this->assertNotReadOnly();
 
        if ( $jobs === [] ) {
            return; // nothing to do
        }
 
        foreach ( $jobs as $job ) {
            if ( $job->getType() !== $this->type ) {
                throw new JobQueueError(
                    "Got '{$job->getType()}' job; expected a '{$this->type}' job." );
            } elseif ( $job->getReleaseTimestamp() && !$this->supportsDelayedJobs() ) {
                throw new JobQueueError(
                    "Got delayed '{$job->getType()}' job; delays are not supported." );
            }
        }
 
        $this->doBatchPush( $jobs, $flags );
 
        foreach ( $jobs as $job ) {
            if ( $job->isRootJob() ) {
                $this->deduplicateRootJob( $job );
            }
        }
    }
 
    abstract protected function doBatchPush( array $jobs, $flags );
 
    final public function pop() {
        $this->assertNotReadOnly();
 
        $job = $this->doPop();
 
        // Flag this job as an old duplicate based on its "root" job...
        try {
            if ( $job && $this->isRootJobOldDuplicate( $job ) ) {
                $this->incrStats( 'dupe_pops', $this->type );
                $job = DuplicateJob::newFromJob( $job ); // convert to a no-op
            }
        } catch ( Exception $e ) {
            // don't lose jobs over this
        }
 
        return $job;
    }
 
    abstract protected function doPop();
 
    final public function ack( RunnableJob $job ) {
        $this->assertNotReadOnly();
        if ( $job->getType() !== $this->type ) {
            throw new JobQueueError( "Got '{$job->getType()}' job; expected '{$this->type}'." );
        }
 
        $this->doAck( $job );
    }
 
    abstract protected function doAck( RunnableJob $job );
 
    final public function deduplicateRootJob( IJobSpecification $job ) {
        $this->assertNotReadOnly();
        if ( $job->getType() !== $this->type ) {
            throw new JobQueueError( "Got '{$job->getType()}' job; expected '{$this->type}'." );
        }
 
        return $this->doDeduplicateRootJob( $job );
    }
 
    protected function doDeduplicateRootJob( IJobSpecification $job ) {
        $params = $job->hasRootJobParams() ? $job->getRootJobParams() : null;
        if ( !$params ) {
            throw new JobQueueError( "Cannot register root job; missing parameters." );
        }
 
        $key = $this->getRootJobCacheKey( $params['rootJobSignature'] );
        // Callers should call JobQueueGroup::push() before this method so that if the
        // insert fails, the de-duplication registration will be aborted. Having only the
        // de-duplication registration succeed would cause jobs to become no-ops without
        // any actual jobs that made them redundant.
        $timestamp = $this->wanCache->get( $key ); // last known timestamp of such a root job
        if ( $timestamp !== false && $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->wanCache->set( $key, $params['rootJobTimestamp'], self::ROOTJOB_TTL );
    }
 
    final protected function isRootJobOldDuplicate( IJobSpecification $job ) {
        if ( $job->getType() !== $this->type ) {
            throw new JobQueueError( "Got '{$job->getType()}' job; expected '{$this->type}'." );
        }
 
        return $this->doIsRootJobOldDuplicate( $job );
    }
 
    protected function doIsRootJobOldDuplicate( IJobSpecification $job ) {
        $params = $job->hasRootJobParams() ? $job->getRootJobParams() : null;
        if ( !$params ) {
            return false; // job has no de-deplication info
        }
 
        $key = $this->getRootJobCacheKey( $params['rootJobSignature'] );
        // Get the last time this root job was enqueued
        $timestamp = $this->wanCache->get( $key );
        if ( $timestamp === false || $params['rootJobTimestamp'] > $timestamp ) {
            // Update the timestamp of the last known root job started at the location...
            $this->wanCache->set( $key, $params['rootJobTimestamp'], self::ROOTJOB_TTL );
        }
 
        // Check if a new root job was started at the location after this one's...
        return ( $timestamp && $timestamp > $params['rootJobTimestamp'] );
    }
 
    protected function getRootJobCacheKey( $signature ) {
        return $this->wanCache->makeGlobalKey(
            'jobqueue',
            $this->domain,
            $this->type,
            'rootjob',
            $signature
        );
    }
 
    final public function delete() {
        $this->assertNotReadOnly();
 
        $this->doDelete();
    }
 
    protected function doDelete() {
        throw new JobQueueError( "This method is not implemented." );
    }
 
    final public function waitForBackups() {
        $this->doWaitForBackups();
    }
 
    protected function doWaitForBackups() {
    }
 
    final public function flushCaches() {
        $this->doFlushCaches();
    }
 
    protected function doFlushCaches() {
    }
 
    abstract public function getAllQueuedJobs();
 
    public function getAllDelayedJobs() {
        return new ArrayIterator( [] ); // not implemented
    }
 
    public function getAllAcquiredJobs() {
        return new ArrayIterator( [] ); // not implemented
    }
 
    public function getAllAbandonedJobs() {
        return new ArrayIterator( [] ); // not implemented
    }
 
    public function getCoalesceLocationInternal() {
        return null;
    }
 
    final public function getSiblingQueuesWithJobs( array $types ) {
        return $this->doGetSiblingQueuesWithJobs( $types );
    }
 
    protected function doGetSiblingQueuesWithJobs( array $types ) {
        return null; // not supported
    }
 
    final public function getSiblingQueueSizes( array $types ) {
        return $this->doGetSiblingQueueSizes( $types );
    }
 
    protected function doGetSiblingQueueSizes( array $types ) {
        return null; // not supported
    }
 
    protected function factoryJob( $command, $params ) {
        // @TODO: dependency inject this as a callback
        return Job::factory( $command, $params );
    }
 
    protected function assertNotReadOnly() {
        if ( $this->readOnlyReason !== false ) {
            throw new JobQueueReadOnlyError( "Job queue is read-only: {$this->readOnlyReason}" );
        }
    }
 
    protected function incrStats( $key, $type, $delta = 1 ) {
        $this->stats->updateCount( "jobqueue.{$key}.all", $delta );
        $this->stats->updateCount( "jobqueue.{$key}.{$type}", $delta );
    }
}