23use InvalidArgumentException;
30use Psr\Log\LoggerInterface;
33use UnexpectedValueException;
94 private const MAX_PUSH_SIZE = 25;
109 parent::__construct( $params );
110 $params[
'redisConfig'][
'serializer'] =
'none';
111 $this->server = $params[
'redisServer'];
112 $this->compression = $params[
'compression'] ??
'none';
114 if ( empty( $params[
'daemonized'] ) ) {
115 throw new InvalidArgumentException(
116 "Non-daemonized mode is no longer supported. Please install the " .
117 "mediawiki/services/jobrunner service and update \$wgJobTypeConf as needed." );
119 $this->logger = LoggerFactory::getInstance(
'redis' );
123 return [
'timestamp',
'fifo' ];
151 return $conn->lLen( $this->getQueueKey(
'l-unclaimed' ) );
152 }
catch ( RedisException $e ) {
165 $conn->multi( Redis::PIPELINE );
166 $conn->zCard( $this->getQueueKey(
'z-claimed' ) );
167 $conn->zCard( $this->getQueueKey(
'z-abandoned' ) );
169 return array_sum( $conn->exec() );
170 }
catch ( RedisException $e ) {
183 return $conn->zCard( $this->getQueueKey(
'z-delayed' ) );
184 }
catch ( RedisException $e ) {
197 return $conn->zCard( $this->getQueueKey(
'z-abandoned' ) );
198 }
catch ( RedisException $e ) {
213 foreach ( $jobs as
$job ) {
215 if ( $item[
'sha1'] !==
'' ) {
216 $items[$item[
'sha1']] = $item;
218 $items[$item[
'uuid']] = $item;
222 if ( $items === [] ) {
229 if ( $flags & self::QOS_ATOMIC ) {
230 $batches = [ $items ];
232 $batches = array_chunk( $items, self::MAX_PUSH_SIZE );
236 foreach ( $batches as $itemBatch ) {
237 $added = $this->
pushBlobs( $conn, $itemBatch );
238 if ( is_int( $added ) ) {
241 $failed += count( $itemBatch );
244 $this->
incrStats(
'inserts', $this->type, count( $items ) );
245 $this->
incrStats(
'inserts_actual', $this->type, $pushed );
246 $this->
incrStats(
'dupe_inserts', $this->type,
247 count( $items ) - $failed - $pushed );
249 $err =
"Could not insert {$failed} {$this->type} job(s).";
251 throw new RedisException( $err );
253 }
catch ( RedisException $e ) {
265 $args = [ $this->encodeQueueName() ];
267 foreach ( $items as $item ) {
268 $args[] = (string)$item[
'uuid'];
269 $args[] = (string)$item[
'sha1'];
270 $args[] = (string)$item[
'rtimestamp'];
271 $args[] = (string)$this->
serialize( $item );
276 local kUnclaimed, kSha1ById, kIdBySha1, kDelayed, kData, kQwJobs = unpack(KEYS)
277 -- First argument is the queue ID
278 local queueId = ARGV[1]
279 -- Next arguments all come in 4s (one per job)
280 local variadicArgCount = #ARGV - 1
281 if variadicArgCount % 4 ~= 0 then
282 return redis.error_reply(
'Unmatched arguments')
284 -- Insert each job into
this queue as needed
287 local id,sha1,rtimestamp,blob = ARGV[i],ARGV[i+1],ARGV[i+2],ARGV[i+3]
288 if sha1 ==
'' or redis.call(
'hExists',kIdBySha1,sha1) == 0 then
289 if 1*rtimestamp > 0 then
290 -- Insert into delayed queue (release time as score)
291 redis.call(
'zAdd',kDelayed,rtimestamp,
id)
293 -- Insert into unclaimed queue
294 redis.call(
'lPush',kUnclaimed,
id)
297 redis.call(
'hSet',kSha1ById,
id,sha1)
298 redis.call(
'hSet',kIdBySha1,sha1,
id)
300 redis.call(
'hSet',kData,
id,blob)
304 -- Mark
this queue as having jobs
305 redis.call(
'sAdd',kQwJobs,queueId)
308 return $conn->
luaEval( $script,
311 $this->getQueueKey(
'l-unclaimed' ), # KEYS[1]
312 $this->getQueueKey(
'h-sha1ById' ), # KEYS[2]
313 $this->getQueueKey(
'h-idBySha1' ), # KEYS[3]
314 $this->getQueueKey(
'z-delayed' ), # KEYS[4]
315 $this->getQueueKey(
'h-data' ), # KEYS[5]
316 $this->getGlobalKey(
's-queuesWithJobs' ), # KEYS[6]
320 6 # number of first argument(s) that are keys
336 if ( !is_string( $blob ) ) {
342 if ( $item ===
false ) {
343 wfDebugLog(
'JobQueue',
"Could not unserialize {$this->type} job." );
350 }
catch ( RedisException $e ) {
366 local kUnclaimed, kSha1ById, kIdBySha1, kClaimed, kAttempts, kData = unpack(KEYS)
367 local rTime = unpack(ARGV)
368 -- Pop an item off the queue
369 local
id = redis.call(
'rPop',kUnclaimed)
373 -- Allow
new duplicates of
this job
374 local sha1 = redis.call(
'hGet',kSha1ById,
id)
375 if sha1 then redis.call(
'hDel',kIdBySha1,sha1) end
376 redis.call(
'hDel',kSha1ById,
id)
377 -- Mark the jobs as claimed and
return it
378 redis.call(
'zAdd',kClaimed,rTime,
id)
379 redis.call(
'hIncrBy',kAttempts,
id,1)
380 return redis.call(
'hGet',kData,
id)
382 return $conn->
luaEval( $script,
384 $this->getQueueKey(
'l-unclaimed' ), # KEYS[1]
385 $this->getQueueKey(
'h-sha1ById' ), # KEYS[2]
386 $this->getQueueKey(
'h-idBySha1' ), # KEYS[3]
387 $this->getQueueKey(
'z-claimed' ), # KEYS[4]
388 $this->getQueueKey(
'h-attempts' ), # KEYS[5]
389 $this->getQueueKey(
'h-data' ), # KEYS[6]
390 time(), # ARGV[1] (injected to be replication-safe)
392 6 # number of first argument(s) that are keys
404 $uuid =
$job->getMetadata(
'uuid' );
405 if ( $uuid ===
null ) {
406 throw new UnexpectedValueException(
"Job of type '{$job->getType()}' has no UUID." );
414 local kClaimed, kAttempts, kData = unpack(KEYS)
415 local
id = unpack(ARGV)
416 -- Unmark the job as claimed
417 local removed = redis.call(
'zRem',kClaimed,
id)
418 -- Check
if the job was recycled
422 -- Delete the retry data
423 redis.call(
'hDel',kAttempts,
id)
424 -- Delete the job data itself
425 return redis.call(
'hDel',kData,
id)
427 $res = $conn->luaEval( $script,
429 $this->getQueueKey(
'z-claimed' ), # KEYS[1]
430 $this->getQueueKey(
'h-attempts' ), # KEYS[2]
431 $this->getQueueKey(
'h-data' ), # KEYS[3]
434 3 # number of first argument(s) that are keys
438 wfDebugLog(
'JobQueue',
"Could not acknowledge {$this->type} job $uuid." );
444 }
catch ( RedisException $e ) {
458 if ( !
$job->hasRootJobParams() ) {
459 throw new LogicException(
"Cannot register root job; missing parameters." );
461 $params =
$job->getRootJobParams();
467 $timestamp = $conn->get( $key );
468 if ( $timestamp && $timestamp >= $params[
'rootJobTimestamp'] ) {
473 return $conn->set( $key, $params[
'rootJobTimestamp'], self::ROOTJOB_TTL );
474 }
catch ( RedisException $e ) {
486 if ( !
$job->hasRootJobParams() ) {
489 $params =
$job->getRootJobParams();
495 }
catch ( RedisException $e ) {
500 return ( $timestamp && $timestamp > $params[
'rootJobTimestamp'] );
509 static $props = [
'l-unclaimed',
'z-claimed',
'z-abandoned',
510 'z-delayed',
'h-idBySha1',
'h-sha1ById',
'h-attempts',
'h-data' ];
515 foreach ( $props as $prop ) {
516 $keys[] = $this->getQueueKey( $prop );
519 $ok = ( $conn->del( $keys ) !== false );
520 $conn->sRem( $this->getGlobalKey(
's-queuesWithJobs' ), $this->encodeQueueName() );
523 }
catch ( RedisException $e ) {
536 $uids = $conn->lRange( $this->getQueueKey(
'l-unclaimed' ), 0, -1 );
537 }
catch ( RedisException $e ) {
552 $uids = $conn->zRange( $this->getQueueKey(
'z-delayed' ), 0, -1 );
553 }
catch ( RedisException $e ) {
568 $uids = $conn->zRange( $this->getQueueKey(
'z-claimed' ), 0, -1 );
569 }
catch ( RedisException $e ) {
584 $uids = $conn->zRange( $this->getQueueKey(
'z-abandoned' ), 0, -1 );
585 }
catch ( RedisException $e ) {
600 function ( $uid ) use ( $conn ) {
603 [
'accept' =>
static function (
$job ) {
604 return is_object(
$job );
619 $types = array_values( $types );
622 $conn->multi( Redis::PIPELINE );
623 foreach ( $types as
$type ) {
624 $conn->lLen( $this->getQueueKey(
'l-unclaimed',
$type ) );
626 $res = $conn->exec();
627 if ( is_array( $res ) ) {
628 foreach ( $res as $i => $size ) {
629 $sizes[$types[$i]] = $size;
632 }
catch ( RedisException $e ) {
650 $data = $conn->hGet( $this->getQueueKey(
'h-data' ), $uid );
651 if ( $data ===
false ) {
655 if ( !is_array( $item ) ) {
656 throw new UnexpectedValueException(
"Could not unserialize job with ID '$uid'." );
659 $params = $item[
'params'];
660 $params += [
'namespace' => $item[
'namespace'],
'title' => $item[
'title'] ];
662 $job->setMetadata(
'uuid', $item[
'uuid'] );
663 $job->setMetadata(
'timestamp', $item[
'timestamp'] );
665 $job->setMetadata(
'attempts',
666 $conn->hGet( $this->getQueueKey(
'h-attempts' ), $uid ) );
669 }
catch ( RedisException $e ) {
684 $set = $conn->sMembers( $this->getGlobalKey(
's-queuesWithJobs' ) );
685 foreach ( $set as $queue ) {
686 $queues[] = $this->decodeQueueName( $queue );
688 }
catch ( RedisException $e ) {
702 'type' =>
$job->getType(),
704 'title' =>
$job->getParams()[
'title'] ??
'',
705 'params' =>
$job->getParams(),
707 'rtimestamp' =>
$job->getReleaseTimestamp() ?: 0,
709 'uuid' => $this->idGenerator->newRawUUIDv4(),
710 'sha1' =>
$job->ignoreDuplicates()
711 ? \Wikimedia\base_convert( sha1(
serialize(
$job->getDeduplicationInfo() ) ), 16, 36, 31 )
713 'timestamp' => time()
722 $params = $fields[
'params'];
723 $params += [
'namespace' => $fields[
'namespace'],
'title' => $fields[
'title'] ];
726 $job->setMetadata(
'uuid', $fields[
'uuid'] );
727 $job->setMetadata(
'timestamp', $fields[
'timestamp'] );
738 if ( $this->compression ===
'gzip'
739 && strlen( $blob ) >= 1024
740 && function_exists(
'gzdeflate' )
742 $object = (object)[
'blob' => gzdeflate( $blob ),
'enc' =>
'gzip' ];
745 return ( strlen( $blobz ) < strlen( $blob ) ) ? $blobz : $blob;
757 if ( is_object( $fields ) ) {
758 if ( $fields->enc ===
'gzip' && function_exists(
'gzinflate' ) ) {
759 $fields =
unserialize( gzinflate( $fields->blob ) );
765 return is_array( $fields ) ? $fields :
false;
775 $conn = $this->redisPool->getConnection( $this->server, $this->logger );
778 "Unable to connect to redis server {$this->server}." );
790 $this->redisPool->handleError( $conn, $e );
791 return new JobQueueError(
"Redis server error: {$e->getMessage()}\n" );
797 private function encodeQueueName() {
798 return json_encode( [ $this->type, $this->domain ] );
805 private function decodeQueueName( $name ) {
806 return json_decode( $name );
813 private function getGlobalKey( $name ) {
814 $parts = [
'global',
'jobqueue', $name ];
815 foreach ( $parts as $part ) {
816 if ( !preg_match(
'/[a-zA-Z0-9_-]+/', $part ) ) {
817 throw new InvalidArgumentException(
"Key part characters are out of range." );
821 return implode(
':', $parts );
829 private function getQueueKey( $prop,
$type =
null ) {
833 $keyspace = WikiMap::getWikiIdFromDbDomain( $this->domain );
835 $parts = [ $keyspace,
'jobqueue',
$type, $prop ];
838 return implode(
':', array_map(
'rawurlencode', $parts ) );
843class_alias( JobQueueRedis::class,
'JobQueueRedis' );
wfDebugLog( $logGroup, $text, $dest='all', array $context=[])
Send a line to a supplementary debug log file, if configured, or main debug log if not.
Convenience class for generating iterators from iterators.
if(count( $args)< 1) $job