23use Psr\Log\LoggerInterface;
82 private const MAX_PUSH_SIZE = 25;
99 $params[
'redisConfig'][
'serializer'] =
'none';
100 $this->server =
$params[
'redisServer'];
101 $this->compression =
$params[
'compression'] ??
'none';
102 $this->redisPool = RedisConnectionPool::singleton(
$params[
'redisConfig'] );
103 if ( empty(
$params[
'daemonized'] ) ) {
104 throw new InvalidArgumentException(
105 "Non-daemonized mode is no longer supported. Please install the " .
106 "mediawiki/services/jobrunner service and update \$wgJobTypeConf as needed." );
108 $this->logger = LoggerFactory::getInstance(
'redis' );
112 return [
'timestamp',
'fifo' ];
140 return $conn->lLen( $this->getQueueKey(
'l-unclaimed' ) );
141 }
catch ( RedisException $e ) {
154 $conn->multi( Redis::PIPELINE );
155 $conn->zCard( $this->getQueueKey(
'z-claimed' ) );
156 $conn->zCard( $this->getQueueKey(
'z-abandoned' ) );
158 return array_sum( $conn->exec() );
159 }
catch ( RedisException $e ) {
172 return $conn->zCard( $this->getQueueKey(
'z-delayed' ) );
173 }
catch ( RedisException $e ) {
186 return $conn->zCard( $this->getQueueKey(
'z-abandoned' ) );
187 }
catch ( RedisException $e ) {
202 foreach ( $jobs as
$job ) {
204 if ( strlen( $item[
'sha1'] ) ) {
205 $items[$item[
'sha1']] = $item;
207 $items[$item[
'uuid']] = $item;
211 if ( $items === [] ) {
218 if ( $flags & self::QOS_ATOMIC ) {
219 $batches = [ $items ];
221 $batches = array_chunk( $items, self::MAX_PUSH_SIZE );
225 foreach ( $batches as $itemBatch ) {
226 $added = $this->
pushBlobs( $conn, $itemBatch );
227 if ( is_int( $added ) ) {
230 $failed += count( $itemBatch );
233 $this->
incrStats(
'inserts', $this->type, count( $items ) );
234 $this->
incrStats(
'inserts_actual', $this->type, $pushed );
235 $this->
incrStats(
'dupe_inserts', $this->type,
236 count( $items ) - $failed - $pushed );
238 $err =
"Could not insert {$failed} {$this->type} job(s).";
240 throw new RedisException( $err );
242 }
catch ( RedisException $e ) {
254 $args = [ $this->encodeQueueName() ];
256 foreach ( $items as $item ) {
257 $args[] = (string)$item[
'uuid'];
258 $args[] = (string)$item[
'sha1'];
259 $args[] = (string)$item[
'rtimestamp'];
260 $args[] = (string)$this->
serialize( $item );
265 local kUnclaimed, kSha1ById, kIdBySha1, kDelayed, kData, kQwJobs = unpack(KEYS)
266 -- First argument is the queue ID
267 local queueId = ARGV[1]
268 -- Next arguments all come in 4s (one per job)
269 local variadicArgCount = #ARGV - 1
270 if variadicArgCount % 4 ~= 0 then
271 return redis.error_reply(
'Unmatched arguments')
273 -- Insert each job into
this queue as needed
276 local id,sha1,rtimestamp,blob = ARGV[i],ARGV[i+1],ARGV[i+2],ARGV[i+3]
277 if sha1 ==
'' or redis.call(
'hExists',kIdBySha1,sha1) == 0 then
278 if 1*rtimestamp > 0 then
279 -- Insert into delayed queue (release time as score)
280 redis.call(
'zAdd',kDelayed,rtimestamp,
id)
282 -- Insert into unclaimed queue
283 redis.call(
'lPush',kUnclaimed,
id)
286 redis.call(
'hSet',kSha1ById,
id,sha1)
287 redis.call(
'hSet',kIdBySha1,sha1,
id)
289 redis.call(
'hSet',kData,
id,blob)
293 -- Mark
this queue as having jobs
294 redis.call(
'sAdd',kQwJobs,queueId)
297 return $conn->
luaEval( $script,
300 $this->getQueueKey(
'l-unclaimed' ), # KEYS[1]
301 $this->getQueueKey(
'h-sha1ById' ), # KEYS[2]
302 $this->getQueueKey(
'h-idBySha1' ), # KEYS[3]
303 $this->getQueueKey(
'z-delayed' ), # KEYS[4]
304 $this->getQueueKey(
'h-data' ), # KEYS[5]
305 $this->getGlobalKey(
's-queuesWithJobs' ), # KEYS[6]
309 6 # number of first argument(s) that are keys
325 if ( !is_string( $blob ) ) {
331 if ( $item ===
false ) {
332 wfDebugLog(
'JobQueue',
"Could not unserialize {$this->type} job." );
339 }
catch ( RedisException $e ) {
355 local kUnclaimed, kSha1ById, kIdBySha1, kClaimed, kAttempts, kData = unpack(KEYS)
356 local rTime = unpack(ARGV)
357 -- Pop an item off the queue
358 local
id = redis.call(
'rPop',kUnclaimed)
362 -- Allow
new duplicates of
this job
363 local sha1 = redis.call(
'hGet',kSha1ById,
id)
364 if sha1 then redis.call(
'hDel',kIdBySha1,sha1) end
365 redis.call(
'hDel',kSha1ById,
id)
366 -- Mark the jobs as claimed and
return it
367 redis.call(
'zAdd',kClaimed,rTime,
id)
368 redis.call(
'hIncrBy',kAttempts,
id,1)
369 return redis.call(
'hGet',kData,
id)
371 return $conn->
luaEval( $script,
373 $this->getQueueKey(
'l-unclaimed' ), # KEYS[1]
374 $this->getQueueKey(
'h-sha1ById' ), # KEYS[2]
375 $this->getQueueKey(
'h-idBySha1' ), # KEYS[3]
376 $this->getQueueKey(
'z-claimed' ), # KEYS[4]
377 $this->getQueueKey(
'h-attempts' ), # KEYS[5]
378 $this->getQueueKey(
'h-data' ), # KEYS[6]
379 time(), # ARGV[1] (injected to be replication-safe)
381 6 # number of first argument(s) that are keys
393 $uuid =
$job->getMetadata(
'uuid' );
394 if ( $uuid ===
null ) {
395 throw new UnexpectedValueException(
"Job of type '{$job->getType()}' has no UUID." );
403 local kClaimed, kAttempts, kData = unpack(KEYS)
404 local
id = unpack(ARGV)
405 -- Unmark the job as claimed
406 local removed = redis.call(
'zRem',kClaimed,
id)
407 -- Check
if the job was recycled
411 -- Delete the retry data
412 redis.call(
'hDel',kAttempts,
id)
413 -- Delete the job data itself
414 return redis.call(
'hDel',kData,
id)
416 $res = $conn->luaEval( $script,
418 $this->getQueueKey(
'z-claimed' ), # KEYS[1]
419 $this->getQueueKey(
'h-attempts' ), # KEYS[2]
420 $this->getQueueKey(
'h-data' ), # KEYS[3]
423 3 # number of first argument(s) that are keys
427 wfDebugLog(
'JobQueue',
"Could not acknowledge {$this->type} job $uuid." );
433 }
catch ( RedisException $e ) {
448 if ( !
$job->hasRootJobParams() ) {
449 throw new LogicException(
"Cannot register root job; missing parameters." );
457 $timestamp = $conn->get( $key );
458 if ( $timestamp && $timestamp >=
$params[
'rootJobTimestamp'] ) {
463 return $conn->set( $key,
$params[
'rootJobTimestamp'], self::ROOTJOB_TTL );
464 }
catch ( RedisException $e ) {
476 if ( !
$job->hasRootJobParams() ) {
485 }
catch ( RedisException $e ) {
490 return ( $timestamp && $timestamp >
$params[
'rootJobTimestamp'] );
499 static $props = [
'l-unclaimed',
'z-claimed',
'z-abandoned',
500 'z-delayed',
'h-idBySha1',
'h-sha1ById',
'h-attempts',
'h-data' ];
505 foreach ( $props as $prop ) {
506 $keys[] = $this->getQueueKey( $prop );
509 $ok = ( $conn->del( $keys ) !== false );
510 $conn->sRem( $this->getGlobalKey(
's-queuesWithJobs' ), $this->encodeQueueName() );
513 }
catch ( RedisException $e ) {
526 $uids = $conn->lRange( $this->getQueueKey(
'l-unclaimed' ), 0, -1 );
527 }
catch ( RedisException $e ) {
542 $uids = $conn->zRange( $this->getQueueKey(
'z-delayed' ), 0, -1 );
543 }
catch ( RedisException $e ) {
558 $uids = $conn->zRange( $this->getQueueKey(
'z-claimed' ), 0, -1 );
559 }
catch ( RedisException $e ) {
574 $uids = $conn->zRange( $this->getQueueKey(
'z-abandoned' ), 0, -1 );
575 }
catch ( RedisException $e ) {
590 function ( $uid ) use ( $conn ) {
593 [
'accept' =>
static function (
$job ) {
594 return is_object(
$job );
609 $types = array_values( $types );
612 $conn->multi( Redis::PIPELINE );
613 foreach ( $types as
$type ) {
614 $conn->lLen( $this->getQueueKey(
'l-unclaimed',
$type ) );
616 $res = $conn->exec();
617 if ( is_array( $res ) ) {
618 foreach ( $res as $i => $size ) {
619 $sizes[$types[$i]] = $size;
622 }
catch ( RedisException $e ) {
640 $data = $conn->hGet( $this->getQueueKey(
'h-data' ), $uid );
641 if ( $data ===
false ) {
645 if ( !is_array( $item ) ) {
646 throw new UnexpectedValueException(
"Could not unserialize job with ID '$uid'." );
650 $params += [
'namespace' => $item[
'namespace'],
'title' => $item[
'title'] ];
652 $job->setMetadata(
'uuid', $item[
'uuid'] );
653 $job->setMetadata(
'timestamp', $item[
'timestamp'] );
655 $job->setMetadata(
'attempts',
656 $conn->hGet( $this->getQueueKey(
'h-attempts' ), $uid ) );
659 }
catch ( RedisException $e ) {
674 $set = $conn->sMembers( $this->getGlobalKey(
's-queuesWithJobs' ) );
675 foreach ( $set as $queue ) {
676 $queues[] = $this->decodeQueueName( $queue );
678 }
catch ( RedisException $e ) {
692 'type' =>
$job->getType(),
694 'title' =>
$job->getParams()[
'title'] ??
'',
695 'params' =>
$job->getParams(),
697 'rtimestamp' =>
$job->getReleaseTimestamp() ?: 0,
699 'uuid' => $this->idGenerator->newRawUUIDv4(),
700 'sha1' =>
$job->ignoreDuplicates()
701 ? Wikimedia\base_convert( sha1(
serialize(
$job->getDeduplicationInfo() ) ), 16, 36, 31 )
703 'timestamp' => time()
713 $params += [
'namespace' => $fields[
'namespace'],
'title' => $fields[
'title'] ];
716 $job->setMetadata(
'uuid', $fields[
'uuid'] );
717 $job->setMetadata(
'timestamp', $fields[
'timestamp'] );
728 if ( $this->compression ===
'gzip'
729 && strlen( $blob ) >= 1024
730 && function_exists(
'gzdeflate' )
732 $object = (object)[
'blob' => gzdeflate( $blob ),
'enc' =>
'gzip' ];
735 return ( strlen( $blobz ) < strlen( $blob ) ) ? $blobz : $blob;
747 if ( is_object( $fields ) ) {
748 if ( $fields->enc ===
'gzip' && function_exists(
'gzinflate' ) ) {
749 $fields =
unserialize( gzinflate( $fields->blob ) );
755 return is_array( $fields ) ? $fields :
false;
765 $conn = $this->redisPool->getConnection( $this->server, $this->logger );
768 "Unable to connect to redis server {$this->server}." );
780 $this->redisPool->handleError( $conn, $e );
781 return new JobQueueError(
"Redis server error: {$e->getMessage()}\n" );
787 private function encodeQueueName() {
788 return json_encode( [ $this->type, $this->domain ] );
795 private function decodeQueueName( $name ) {
796 return json_decode( $name );
803 private function getGlobalKey( $name ) {
804 $parts = [
'global',
'jobqueue', $name ];
805 foreach ( $parts as $part ) {
806 if ( !preg_match(
'/[a-zA-Z0-9_-]+/', $part ) ) {
807 throw new InvalidArgumentException(
"Key part characters are out of range." );
811 return implode(
':', $parts );
819 private function getQueueKey( $prop,
$type =
null ) {
823 $keyspace = WikiMap::getWikiIdFromDbDomain( $this->domain );
825 $parts = [ $keyspace,
'jobqueue',
$type, $prop ];
828 return implode(
':', array_map(
'rawurlencode', $parts ) );
wfDebugLog( $logGroup, $text, $dest='all', array $context=[])
Send a line to a supplementary debug log file, if configured, or main debug log if not.
array $params
The job parameters.
Redis-backed job queue storage.
doDeduplicateRootJob(IJobSpecification $job)
__construct(array $params)
popAndAcquireBlob(RedisConnRef $conn)
getCoalesceLocationInternal()
Do not use this function outside of JobQueue/JobQueueGroup.
getJobFromUidInternal( $uid, $conn)
This function should not be called outside JobQueueRedis.
doGetSiblingQueuesWithJobs(array $types)
doGetSiblingQueueSizes(array $types)
getServerQueuesWithJobs()
RedisConnectionPool $redisPool
pushBlobs(RedisConnRef $conn, array $items)
string $server
Server address.
supportedOrders()
Get the allowed queue orders for configuration validation.
supportsDelayedJobs()
Find out if delayed jobs are supported for configuration validation.
string $compression
Compression method to use.
doIsRootJobOldDuplicate(IJobSpecification $job)
doBatchPush(array $jobs, $flags)
getJobIterator(RedisConnRef $conn, array $uids)
getConnection()
Get a connection to the server that handles all sub-queues for this queue.
getJobFromFields(array $fields)
optimalOrder()
Get the default queue order to use if configuration does not specify one.
handleErrorAndMakeException(RedisConnRef $conn, $e)
getNewJobFields(IJobSpecification $job)
Base class for queueing and running background jobs from a storage backend.
incrStats( $key, $type, $delta=1)
Call StatsdDataFactoryInterface::updateCount() for the queue overall and for the queue type.
factoryJob( $command, $params)
getRootJobCacheKey( $signature, $type)
Convenience class for generating iterators from iterators.
Helper class to handle automatically marking connections as reusable (via RAII pattern)
luaEval( $script, array $params, $numKeys)
Helper class to manage Redis connections.
Interface for serializable objects that describe a job queue task.
Job that has a run() method and metadata accessors for JobQueue::pop() and JobQueue::ack().
if(count( $args)< 1) $job