/** @var StatsdDataFactoryInterface */
protected $stats;
- /** @var BagOStuff */
- protected $dupCache;
+ /** @var WANObjectCache */
+ protected $wanCache;
const QOS_ATOMIC = 1; // integer; "all-or-nothing" job insertions
/**
* @param array $params
+ * - type : A job type
+ * - domain : A DB domain ID
+ * - wanCache : An instance of WANObjectCache to use for caching [default: none]
+ * - stats : An instance of StatsdDataFactoryInterface [default: none]
+ * - claimTTL : Seconds a job can be claimed for exclusive execution [default: forever]
+ * - maxTries : Total times a job can be tried, assuming claims expire [default: 3]
+ * - order : Queue order, one of ("fifo", "timestamp", "random") [default: variable]
+ * - readOnlyReason : Mark the queue as read-only with this reason [default: false]
* @throws JobQueueError
*/
protected function __construct( array $params ) {
}
$this->readOnlyReason = $params['readOnlyReason'] ?? false;
$this->stats = $params['stats'] ?? new NullStatsdDataFactory();
- $this->dupCache = $params['stash'] ?? new EmptyBagOStuff();
+ $this->wanCache = $params['wanCache'] ?? WANObjectCache::newEmpty();
}
/**
* @return bool
*/
protected function doDeduplicateRootJob( IJobSpecification $job ) {
- if ( !$job->hasRootJobParams() ) {
+ $params = $job->hasRootJobParams() ? $job->getRootJobParams() : null;
+ if ( !$params ) {
throw new JobQueueError( "Cannot register root job; missing parameters." );
}
- $params = $job->getRootJobParams();
$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. 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'] ) {
+ // 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->dupCache->set( $key, $params['rootJobTimestamp'], self::ROOTJOB_TTL );
+ return $this->wanCache->set( $key, $params['rootJobTimestamp'], self::ROOTJOB_TTL );
}
/**
if ( $job->getType() !== $this->type ) {
throw new JobQueueError( "Got '{$job->getType()}' job; expected '{$this->type}'." );
}
- $isDuplicate = $this->doIsRootJobOldDuplicate( $job );
- return $isDuplicate;
+ return $this->doIsRootJobOldDuplicate( $job );
}
/**
* @return bool
*/
protected function doIsRootJobOldDuplicate( IJobSpecification $job ) {
- if ( !$job->hasRootJobParams() ) {
+ $params = $job->hasRootJobParams() ? $job->getRootJobParams() : null;
+ if ( !$params ) {
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 );
+ $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'] );
* @return string
*/
protected function getRootJobCacheKey( $signature ) {
- return $this->dupCache->makeGlobalKey(
+ return $this->wanCache->makeGlobalKey(
'jobqueue',
$this->domain,
$this->type,
use Wikimedia\Rdbms\DBConnectionError;
use Wikimedia\Rdbms\DBError;
use MediaWiki\MediaWikiServices;
+use Wikimedia\Rdbms\IMaintainableDatabase;
use Wikimedia\ScopedCallback;
/**
const MAX_JOB_RANDOM = 2147483647; // integer; 2^31 - 1, used for job_random
const MAX_OFFSET = 255; // integer; maximum number of rows to skip
- /** @var WANObjectCache */
- protected $cache;
- /** @var IDatabase|DBError|null */
+ /** @var IMaintainableDatabase|DBError|null */
protected $conn;
/** @var array|null Server configuration array */
* If not specified, the primary DB cluster for the wiki will be used.
* This can be overridden with a custom cluster so that DB handles will
* be retrieved via LBFactory::getExternalLB() and getConnection().
- * - wanCache : An instance of WANObjectCache to use for caching.
* @param array $params
*/
protected function __construct( array $params ) {
} elseif ( isset( $params['cluster'] ) && is_string( $params['cluster'] ) ) {
$this->cluster = $params['cluster'];
}
-
- $this->cache = $params['wanCache'] ?? WANObjectCache::newEmpty();
}
protected function supportedOrders() {
protected function doGetSize() {
$key = $this->getCacheKey( 'size' );
- $size = $this->cache->get( $key );
+ $size = $this->wanCache->get( $key );
if ( is_int( $size ) ) {
return $size;
}
} catch ( DBError $e ) {
throw $this->getDBException( $e );
}
- $this->cache->set( $key, $size, self::CACHE_TTL_SHORT );
+ $this->wanCache->set( $key, $size, self::CACHE_TTL_SHORT );
return $size;
}
$key = $this->getCacheKey( 'acquiredcount' );
- $count = $this->cache->get( $key );
+ $count = $this->wanCache->get( $key );
if ( is_int( $count ) ) {
return $count;
}
} catch ( DBError $e ) {
throw $this->getDBException( $e );
}
- $this->cache->set( $key, $count, self::CACHE_TTL_SHORT );
+ $this->wanCache->set( $key, $count, self::CACHE_TTL_SHORT );
return $count;
}
$key = $this->getCacheKey( 'abandonedcount' );
- $count = $this->cache->get( $key );
+ $count = $this->wanCache->get( $key );
if ( is_int( $count ) ) {
return $count;
}
throw $this->getDBException( $e );
}
- $this->cache->set( $key, $count, self::CACHE_TTL_SHORT );
+ $this->wanCache->set( $key, $count, self::CACHE_TTL_SHORT );
return $count;
}
/** @noinspection PhpUnusedLocalVariableInspection */
$scope = $this->getScopedNoTrxFlag( $dbw );
// Check cache to see if the queue has <= OFFSET items
- $tinyQueue = $this->cache->get( $this->getCacheKey( 'small' ) );
+ $tinyQueue = $this->wanCache->get( $this->getCacheKey( 'small' ) );
$invertedDirection = false; // whether one job_random direction was already scanned
// This uses a replication safe method for acquiring jobs. One could use UPDATE+LIMIT
);
if ( !$row ) {
$tinyQueue = true; // we know the queue must have <= MAX_OFFSET rows
- $this->cache->set( $this->getCacheKey( 'small' ), 1, 30 );
+ $this->wanCache->set( $this->getCacheKey( 'small' ), 1, 30 );
continue; // use job_random
}
}
* @return bool
*/
protected function doDeduplicateRootJob( IJobSpecification $job ) {
- $params = $job->getParams();
- if ( !isset( $params['rootJobSignature'] ) ) {
- throw new MWException( "Cannot register root job; missing 'rootJobSignature'." );
- } elseif ( !isset( $params['rootJobTimestamp'] ) ) {
- throw new MWException( "Cannot register root job; missing 'rootJobTimestamp'." );
- }
- $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. Since the insert is
- // deferred till "transaction idle", do the same here, so that the ordering is
+ // Callers should call JobQueueGroup::push() before this method 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.
$dbw = $this->getMasterDB();
/** @noinspection PhpUnusedLocalVariableInspection */
$scope = $this->getScopedNoTrxFlag( $dbw );
-
- $cache = $this->dupCache;
$dbw->onTransactionCommitOrIdle(
- function () use ( $cache, $params, $key ) {
- $timestamp = $cache->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 $cache->set( $key, $params['rootJobTimestamp'], JobQueueDB::ROOTJOB_TTL );
+ function () use ( $job ) {
+ parent::doDeduplicateRootJob( $job );
},
__METHOD__
);
*/
protected function doFlushCaches() {
foreach ( [ 'size', 'acquiredcount' ] as $type ) {
- $this->cache->delete( $this->getCacheKey( $type ) );
+ $this->wanCache->delete( $this->getCacheKey( $type ) );
}
}
/**
* @throws JobQueueConnectionError
- * @return IDatabase
+ * @return IMaintainableDatabase
*/
protected function getMasterDB() {
try {
/**
* @param int $index (DB_REPLICA/DB_MASTER)
- * @return IDatabase
+ * @return IMaintainableDatabase
*/
protected function getDB( $index ) {
if ( $this->server ) {
? $lbFactory->getExternalLB( $this->cluster )
: $lbFactory->getMainLB( $this->domain );
- return ( $lb->getServerType( $lb->getWriterIndex() ) !== 'sqlite' )
+ if ( $lb->getServerType( $lb->getWriterIndex() ) !== 'sqlite' ) {
// Keep a separate connection to avoid contention and deadlocks;
// However, SQLite has the opposite behavior due to DB-level locking.
- ? $lb->getConnectionRef( $index, [], $this->domain, $lb::CONN_TRX_AUTOCOMMIT )
+ $flags = $lb::CONN_TRX_AUTOCOMMIT;
+ } else {
// Jobs insertion will be defered until the PRESEND stage to reduce contention.
- : $lb->getConnectionRef( $index, [], $this->domain );
+ $flags = 0;
+ }
+
+ return $lb->getMaintenanceConnectionRef( $index, [], $this->domain, $flags );
}
}
private function getCacheKey( $property ) {
$cluster = is_string( $this->cluster ) ? $this->cluster : 'main';
- return $this->cache->makeGlobalKey(
+ return $this->wanCache->makeGlobalKey(
'jobqueue',
$this->domain,
$cluster,
dépôts / lhc / web / wiklou.git / commitdiff