includes/jobqueue/JobQueueRedis.php

   1 <?php
   2 /**
   3  * Redis-backed job queue code.
   4  *
   5  * This program is free software; you can redistribute it and/or modify
   6  * it under the terms of the GNU General Public License as published by
   7  * the Free Software Foundation; either version 2 of the License, or
   8  * (at your option) any later version.
   9  *
  10  * This program is distributed in the hope that it will be useful,
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  13  * GNU General Public License for more details.
  14  *
  15  * You should have received a copy of the GNU General Public License along
  16  * with this program; if not, write to the Free Software Foundation, Inc.,
  17  * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
  18  * http://www.gnu.org/copyleft/gpl.html
  19  *
  20  * @file
  21  */
  22 use Psr\Log\LoggerInterface;
  23
  24 /**
  25  * Class to handle job queues stored in Redis
  26  *
  27  * This is a faster and less resource-intensive job queue than JobQueueDB.
  28  * All data for a queue using this class is placed into one redis server.
  29  * The mediawiki/services/jobrunner background service must be set up and running.
  30  *
  31  * There are eight main redis keys (per queue) used to track jobs:
  32  *   - l-unclaimed  : A list of job IDs used for ready unclaimed jobs
  33  *   - z-claimed    : A sorted set of (job ID, UNIX timestamp as score) used for job retries
  34  *   - z-abandoned  : A sorted set of (job ID, UNIX timestamp as score) used for broken jobs
  35  *   - z-delayed    : A sorted set of (job ID, UNIX timestamp as score) used for delayed jobs
  36  *   - h-idBySha1   : A hash of (SHA1 => job ID) for unclaimed jobs used for de-duplication
  37  *   - h-sha1ById   : A hash of (job ID => SHA1) for unclaimed jobs used for de-duplication
  38  *   - h-attempts   : A hash of (job ID => attempt count) used for job claiming/retries
  39  *   - h-data       : A hash of (job ID => serialized blobs) for job storage
  40  * A job ID can be in only one of z-delayed, l-unclaimed, z-claimed, and z-abandoned.
  41  * If an ID appears in any of those lists, it should have a h-data entry for its ID.
  42  * If a job has a SHA1 de-duplication value and its ID is in l-unclaimed or z-delayed, then
  43  * there should be no other such jobs with that SHA1. Every h-idBySha1 entry has an h-sha1ById
  44  * entry and every h-sha1ById must refer to an ID that is l-unclaimed. If a job has its
  45  * ID in z-claimed or z-abandoned, then it must also have an h-attempts entry for its ID.
  46  *
  47  * The following keys are used to track queue states:
  48  *   - s-queuesWithJobs : A set of all queues with non-abandoned jobs
  49  *
  50  * The background service takes care of undelaying, recycling, and pruning jobs as well as
  51  * removing s-queuesWithJobs entries as queues empty.
  52  *
  53  * Additionally, "rootjob:* keys track "root jobs" used for additional de-duplication.
  54  * Aside from root job keys, all keys have no expiry, and are only removed when jobs are run.
  55  * All the keys are prefixed with the relevant wiki ID information.
  56  *
  57  * This class requires Redis 2.6 as it makes use Lua scripts for fast atomic operations.
  58  * Additionally, it should be noted that redis has different persistence modes, such
  59  * as rdb snapshots, journaling, and no persistence. Appropriate configuration should be
  60  * made on the servers based on what queues are using it and what tolerance they have.
  61  *
  62  * @ingroup JobQueue
  63  * @ingroup Redis
  64  * @since 1.22
  65  */
  66 class JobQueueRedis extends JobQueue {
  67         /** @var RedisConnectionPool */
  68         protected $redisPool;
  69         /** @var LoggerInterface */
  70         protected $logger;
  71
  72         /** @var string Server address */
  73         protected $server;
  74         /** @var string Compression method to use */
  75         protected $compression;
  76
  77         const MAX_PUSH_SIZE = 25; // avoid tying up the server
  78
  79         /**
  80          * @param array $params Possible keys:
  81          *   - redisConfig : An array of parameters to RedisConnectionPool::__construct().
  82          *                   Note that the serializer option is ignored as "none" is always used.
  83          *   - redisServer : A hostname/port combination or the absolute path of a UNIX socket.
  84          *                   If a hostname is specified but no port, the standard port number
  85          *                   6379 will be used. Required.
  86          *   - compression : The type of compression to use; one of (none,gzip).
  87          *   - daemonized  : Set to true if the redisJobRunnerService runs in the background.
  88          *                   This will disable job recycling/undelaying from the MediaWiki side
  89          *                   to avoid redundance and out-of-sync configuration.
  90          * @throws InvalidArgumentException
  91          */
  92         public function __construct( array $params ) {
  93                 parent::__construct( $params );
  94                 $params['redisConfig']['serializer'] = 'none'; // make it easy to use Lua
  95                 $this->server = $params['redisServer'];
  96                 $this->compression = isset( $params['compression'] ) ? $params['compression'] : 'none';
  97                 $this->redisPool = RedisConnectionPool::singleton( $params['redisConfig'] );
  98                 if ( empty( $params['daemonized'] ) ) {
  99                         throw new InvalidArgumentException(
 100                                 "Non-daemonized mode is no longer supported. Please install the " .
 101                                 "mediawiki/services/jobrunner service and update \$wgJobTypeConf as needed." );
 102                 }
 103                 $this->logger = \MediaWiki\Logger\LoggerFactory::getInstance( 'redis' );
 104         }
 105
 106         protected function supportedOrders() {
 107                 return [ 'timestamp', 'fifo' ];
 108         }
 109
 110         protected function optimalOrder() {
 111                 return 'fifo';
 112         }
 113
 114         protected function supportsDelayedJobs() {
 115                 return true;
 116         }
 117
 118         /**
 119          * @see JobQueue::doIsEmpty()
 120          * @return bool
 121          * @throws JobQueueError
 122          */
 123         protected function doIsEmpty() {
 124                 return $this->doGetSize() == 0;
 125         }
 126
 127         /**
 128          * @see JobQueue::doGetSize()
 129          * @return int
 130          * @throws JobQueueError
 131          */
 132         protected function doGetSize() {
 133                 $conn = $this->getConnection();
 134                 try {
 135                         return $conn->lSize( $this->getQueueKey( 'l-unclaimed' ) );
 136                 } catch ( RedisException $e ) {
 137                         $this->throwRedisException( $conn, $e );
 138                 }
 139         }
 140
 141         /**
 142          * @see JobQueue::doGetAcquiredCount()
 143          * @return int
 144          * @throws JobQueueError
 145          */
 146         protected function doGetAcquiredCount() {
 147                 $conn = $this->getConnection();
 148                 try {
 149                         $conn->multi( Redis::PIPELINE );
 150                         $conn->zSize( $this->getQueueKey( 'z-claimed' ) );
 151                         $conn->zSize( $this->getQueueKey( 'z-abandoned' ) );
 152
 153                         return array_sum( $conn->exec() );
 154                 } catch ( RedisException $e ) {
 155                         $this->throwRedisException( $conn, $e );
 156                 }
 157         }
 158
 159         /**
 160          * @see JobQueue::doGetDelayedCount()
 161          * @return int
 162          * @throws JobQueueError
 163          */
 164         protected function doGetDelayedCount() {
 165                 $conn = $this->getConnection();
 166                 try {
 167                         return $conn->zSize( $this->getQueueKey( 'z-delayed' ) );
 168                 } catch ( RedisException $e ) {
 169                         $this->throwRedisException( $conn, $e );
 170                 }
 171         }
 172
 173         /**
 174          * @see JobQueue::doGetAbandonedCount()
 175          * @return int
 176          * @throws JobQueueError
 177          */
 178         protected function doGetAbandonedCount() {
 179                 $conn = $this->getConnection();
 180                 try {
 181                         return $conn->zSize( $this->getQueueKey( 'z-abandoned' ) );
 182                 } catch ( RedisException $e ) {
 183                         $this->throwRedisException( $conn, $e );
 184                 }
 185         }
 186
 187         /**
 188          * @see JobQueue::doBatchPush()
 189          * @param IJobSpecification[] $jobs
 190          * @param int $flags
 191          * @return void
 192          * @throws JobQueueError
 193          */
 194         protected function doBatchPush( array $jobs, $flags ) {
 195                 // Convert the jobs into field maps (de-duplicated against each other)
 196                 $items = []; // (job ID => job fields map)
 197                 foreach ( $jobs as $job ) {
 198                         $item = $this->getNewJobFields( $job );
 199                         if ( strlen( $item['sha1'] ) ) { // hash identifier => de-duplicate
 200                                 $items[$item['sha1']] = $item;
 201                         } else {
 202                                 $items[$item['uuid']] = $item;
 203                         }
 204                 }
 205
 206                 if ( !count( $items ) ) {
 207                         return; // nothing to do
 208                 }
 209
 210                 $conn = $this->getConnection();
 211                 try {
 212                         // Actually push the non-duplicate jobs into the queue...
 213                         if ( $flags & self::QOS_ATOMIC ) {
 214                                 $batches = [ $items ]; // all or nothing
 215                         } else {
 216                                 $batches = array_chunk( $items, self::MAX_PUSH_SIZE );
 217                         }
 218                         $failed = 0;
 219                         $pushed = 0;
 220                         foreach ( $batches as $itemBatch ) {
 221                                 $added = $this->pushBlobs( $conn, $itemBatch );
 222                                 if ( is_int( $added ) ) {
 223                                         $pushed += $added;
 224                                 } else {
 225                                         $failed += count( $itemBatch );
 226                                 }
 227                         }
 228                         JobQueue::incrStats( 'inserts', $this->type, count( $items ) );
 229                         JobQueue::incrStats( 'inserts_actual', $this->type, $pushed );
 230                         JobQueue::incrStats( 'dupe_inserts', $this->type,
 231                                 count( $items ) - $failed - $pushed );
 232                         if ( $failed > 0 ) {
 233                                 $err = "Could not insert {$failed} {$this->type} job(s).";
 234                                 wfDebugLog( 'JobQueueRedis', $err );
 235                                 throw new RedisException( $err );
 236                         }
 237                 } catch ( RedisException $e ) {
 238                         $this->throwRedisException( $conn, $e );
 239                 }
 240         }
 241
 242         /**
 243          * @param RedisConnRef $conn
 244          * @param array $items List of results from JobQueueRedis::getNewJobFields()
 245          * @return int Number of jobs inserted (duplicates are ignored)
 246          * @throws RedisException
 247          */
 248         protected function pushBlobs( RedisConnRef $conn, array $items ) {
 249                 $args = [ $this->encodeQueueName() ];
 250                 // Next args come in 4s ([id, sha1, rtime, blob [, id, sha1, rtime, blob ... ] ] )
 251                 foreach ( $items as $item ) {
 252                         $args[] = (string)$item['uuid'];
 253                         $args[] = (string)$item['sha1'];
 254                         $args[] = (string)$item['rtimestamp'];
 255                         $args[] = (string)$this->serialize( $item );
 256                 }
 257                 static $script =
 258                 /** @lang Lua */
 259 <<<LUA
 260                 local kUnclaimed, kSha1ById, kIdBySha1, kDelayed, kData, kQwJobs = unpack(KEYS)
 261                 -- First argument is the queue ID
 262                 local queueId = ARGV[1]
 263                 -- Next arguments all come in 4s (one per job)
 264                 local variadicArgCount = #ARGV - 1
 265                 if variadicArgCount % 4 ~= 0 then
 266                         return redis.error_reply('Unmatched arguments')
 267                 end
 268                 -- Insert each job into this queue as needed
 269                 local pushed = 0
 270                 for i = 2,#ARGV,4 do
 271                         local id,sha1,rtimestamp,blob = ARGV[i],ARGV[i+1],ARGV[i+2],ARGV[i+3]
 272                         if sha1 == '' or redis.call('hExists',kIdBySha1,sha1) == 0 then
 273                                 if 1*rtimestamp > 0 then
 274                                         -- Insert into delayed queue (release time as score)
 275                                         redis.call('zAdd',kDelayed,rtimestamp,id)
 276                                 else
 277                                         -- Insert into unclaimed queue
 278                                         redis.call('lPush',kUnclaimed,id)
 279                                 end
 280                                 if sha1 ~= '' then
 281                                         redis.call('hSet',kSha1ById,id,sha1)
 282                                         redis.call('hSet',kIdBySha1,sha1,id)
 283                                 end
 284                                 redis.call('hSet',kData,id,blob)
 285                                 pushed = pushed + 1
 286                         end
 287                 end
 288                 -- Mark this queue as having jobs
 289                 redis.call('sAdd',kQwJobs,queueId)
 290                 return pushed
 291 LUA;
 292                 return $conn->luaEval( $script,
 293                         array_merge(
 294                                 [
 295                                         $this->getQueueKey( 'l-unclaimed' ), # KEYS[1]
 296                                         $this->getQueueKey( 'h-sha1ById' ), # KEYS[2]
 297                                         $this->getQueueKey( 'h-idBySha1' ), # KEYS[3]
 298                                         $this->getQueueKey( 'z-delayed' ), # KEYS[4]
 299                                         $this->getQueueKey( 'h-data' ), # KEYS[5]
 300                                         $this->getGlobalKey( 's-queuesWithJobs' ), # KEYS[6]
 301                                 ],
 302                                 $args
 303                         ),
 304                         6 # number of first argument(s) that are keys
 305                 );
 306         }
 307
 308         /**
 309          * @see JobQueue::doPop()
 310          * @return Job|bool
 311          * @throws JobQueueError
 312          */
 313         protected function doPop() {
 314                 $job = false;
 315
 316                 $conn = $this->getConnection();
 317                 try {
 318                         do {
 319                                 $blob = $this->popAndAcquireBlob( $conn );
 320                                 if ( !is_string( $blob ) ) {
 321                                         break; // no jobs; nothing to do
 322                                 }
 323
 324                                 JobQueue::incrStats( 'pops', $this->type );
 325                                 $item = $this->unserialize( $blob );
 326                                 if ( $item === false ) {
 327                                         wfDebugLog( 'JobQueueRedis', "Could not unserialize {$this->type} job." );
 328                                         continue;
 329                                 }
 330
 331                                 // If $item is invalid, the runner loop recyling will cleanup as needed
 332                                 $job = $this->getJobFromFields( $item ); // may be false
 333                         } while ( !$job ); // job may be false if invalid
 334                 } catch ( RedisException $e ) {
 335                         $this->throwRedisException( $conn, $e );
 336                 }
 337
 338                 return $job;
 339         }
 340
 341         /**
 342          * @param RedisConnRef $conn
 343          * @return array Serialized string or false
 344          * @throws RedisException
 345          */
 346         protected function popAndAcquireBlob( RedisConnRef $conn ) {
 347                 static $script =
 348                 /** @lang Lua */
 349 <<<LUA
 350                 local kUnclaimed, kSha1ById, kIdBySha1, kClaimed, kAttempts, kData = unpack(KEYS)
 351                 local rTime = unpack(ARGV)
 352                 -- Pop an item off the queue
 353                 local id = redis.call('rPop',kUnclaimed)
 354                 if not id then
 355                         return false
 356                 end
 357                 -- Allow new duplicates of this job
 358                 local sha1 = redis.call('hGet',kSha1ById,id)
 359                 if sha1 then redis.call('hDel',kIdBySha1,sha1) end
 360                 redis.call('hDel',kSha1ById,id)
 361                 -- Mark the jobs as claimed and return it
 362                 redis.call('zAdd',kClaimed,rTime,id)
 363                 redis.call('hIncrBy',kAttempts,id,1)
 364                 return redis.call('hGet',kData,id)
 365 LUA;
 366                 return $conn->luaEval( $script,
 367                         [
 368                                 $this->getQueueKey( 'l-unclaimed' ), # KEYS[1]
 369                                 $this->getQueueKey( 'h-sha1ById' ), # KEYS[2]
 370                                 $this->getQueueKey( 'h-idBySha1' ), # KEYS[3]
 371                                 $this->getQueueKey( 'z-claimed' ), # KEYS[4]
 372                                 $this->getQueueKey( 'h-attempts' ), # KEYS[5]
 373                                 $this->getQueueKey( 'h-data' ), # KEYS[6]
 374                                 time(), # ARGV[1] (injected to be replication-safe)
 375                         ],
 376                         6 # number of first argument(s) that are keys
 377                 );
 378         }
 379
 380         /**
 381          * @see JobQueue::doAck()
 382          * @param Job $job
 383          * @return Job|bool
 384          * @throws UnexpectedValueException
 385          * @throws JobQueueError
 386          */
 387         protected function doAck( Job $job ) {
 388                 if ( !isset( $job->metadata['uuid'] ) ) {
 389                         throw new UnexpectedValueException( "Job of type '{$job->getType()}' has no UUID." );
 390                 }
 391
 392                 $uuid = $job->metadata['uuid'];
 393                 $conn = $this->getConnection();
 394                 try {
 395                         static $script =
 396                         /** @lang Lua */
 397 <<<LUA
 398                         local kClaimed, kAttempts, kData = unpack(KEYS)
 399                         local id = unpack(ARGV)
 400                         -- Unmark the job as claimed
 401                         local removed = redis.call('zRem',kClaimed,id)
 402                         -- Check if the job was recycled
 403                         if removed == 0 then
 404                                 return 0
 405                         end
 406                         -- Delete the retry data
 407                         redis.call('hDel',kAttempts,id)
 408                         -- Delete the job data itself
 409                         return redis.call('hDel',kData,id)
 410 LUA;
 411                         $res = $conn->luaEval( $script,
 412                                 [
 413                                         $this->getQueueKey( 'z-claimed' ), # KEYS[1]
 414                                         $this->getQueueKey( 'h-attempts' ), # KEYS[2]
 415                                         $this->getQueueKey( 'h-data' ), # KEYS[3]
 416                                         $uuid # ARGV[1]
 417                                 ],
 418                                 3 # number of first argument(s) that are keys
 419                         );
 420
 421                         if ( !$res ) {
 422                                 wfDebugLog( 'JobQueueRedis', "Could not acknowledge {$this->type} job $uuid." );
 423
 424                                 return false;
 425                         }
 426
 427                         JobQueue::incrStats( 'acks', $this->type );
 428                 } catch ( RedisException $e ) {
 429                         $this->throwRedisException( $conn, $e );
 430                 }
 431
 432                 return true;
 433         }
 434
 435         /**
 436          * @see JobQueue::doDeduplicateRootJob()
 437          * @param IJobSpecification $job
 438          * @return bool
 439          * @throws JobQueueError
 440          * @throws LogicException
 441          */
 442         protected function doDeduplicateRootJob( IJobSpecification $job ) {
 443                 if ( !$job->hasRootJobParams() ) {
 444                         throw new LogicException( "Cannot register root job; missing parameters." );
 445                 }
 446                 $params = $job->getRootJobParams();
 447
 448                 $key = $this->getRootJobCacheKey( $params['rootJobSignature'] );
 449
 450                 $conn = $this->getConnection();
 451                 try {
 452                         $timestamp = $conn->get( $key ); // current last timestamp of this job
 453                         if ( $timestamp && $timestamp >= $params['rootJobTimestamp'] ) {
 454                                 return true; // a newer version of this root job was enqueued
 455                         }
 456
 457                         // Update the timestamp of the last root job started at the location...
 458                         return $conn->set( $key, $params['rootJobTimestamp'], self::ROOTJOB_TTL ); // 2 weeks
 459                 } catch ( RedisException $e ) {
 460                         $this->throwRedisException( $conn, $e );
 461                 }
 462         }
 463
 464         /**
 465          * @see JobQueue::doIsRootJobOldDuplicate()
 466          * @param Job $job
 467          * @return bool
 468          * @throws JobQueueError
 469          */
 470         protected function doIsRootJobOldDuplicate( Job $job ) {
 471                 if ( !$job->hasRootJobParams() ) {
 472                         return false; // job has no de-deplication info
 473                 }
 474                 $params = $job->getRootJobParams();
 475
 476                 $conn = $this->getConnection();
 477                 try {
 478                         // Get the last time this root job was enqueued
 479                         $timestamp = $conn->get( $this->getRootJobCacheKey( $params['rootJobSignature'] ) );
 480                 } catch ( RedisException $e ) {
 481                         $timestamp = false;
 482                         $this->throwRedisException( $conn, $e );
 483                 }
 484
 485                 // Check if a new root job was started at the location after this one's...
 486                 return ( $timestamp && $timestamp > $params['rootJobTimestamp'] );
 487         }
 488
 489         /**
 490          * @see JobQueue::doDelete()
 491          * @return bool
 492          * @throws JobQueueError
 493          */
 494         protected function doDelete() {
 495                 static $props = [ 'l-unclaimed', 'z-claimed', 'z-abandoned',
 496                         'z-delayed', 'h-idBySha1', 'h-sha1ById', 'h-attempts', 'h-data' ];
 497
 498                 $conn = $this->getConnection();
 499                 try {
 500                         $keys = [];
 501                         foreach ( $props as $prop ) {
 502                                 $keys[] = $this->getQueueKey( $prop );
 503                         }
 504
 505                         $ok = ( $conn->delete( $keys ) !== false );
 506                         $conn->sRem( $this->getGlobalKey( 's-queuesWithJobs' ), $this->encodeQueueName() );
 507
 508                         return $ok;
 509                 } catch ( RedisException $e ) {
 510                         $this->throwRedisException( $conn, $e );
 511                 }
 512         }
 513
 514         /**
 515          * @see JobQueue::getAllQueuedJobs()
 516          * @return Iterator
 517          * @throws JobQueueError
 518          */
 519         public function getAllQueuedJobs() {
 520                 $conn = $this->getConnection();
 521                 try {
 522                         $uids = $conn->lRange( $this->getQueueKey( 'l-unclaimed' ), 0, -1 );
 523                 } catch ( RedisException $e ) {
 524                         $this->throwRedisException( $conn, $e );
 525                 }
 526
 527                 return $this->getJobIterator( $conn, $uids );
 528         }
 529
 530         /**
 531          * @see JobQueue::getAllDelayedJobs()
 532          * @return Iterator
 533          * @throws JobQueueError
 534          */
 535         public function getAllDelayedJobs() {
 536                 $conn = $this->getConnection();
 537                 try {
 538                         $uids = $conn->zRange( $this->getQueueKey( 'z-delayed' ), 0, -1 );
 539                 } catch ( RedisException $e ) {
 540                         $this->throwRedisException( $conn, $e );
 541                 }
 542
 543                 return $this->getJobIterator( $conn, $uids );
 544         }
 545
 546         /**
 547          * @see JobQueue::getAllAcquiredJobs()
 548          * @return Iterator
 549          * @throws JobQueueError
 550          */
 551         public function getAllAcquiredJobs() {
 552                 $conn = $this->getConnection();
 553                 try {
 554                         $uids = $conn->zRange( $this->getQueueKey( 'z-claimed' ), 0, -1 );
 555                 } catch ( RedisException $e ) {
 556                         $this->throwRedisException( $conn, $e );
 557                 }
 558
 559                 return $this->getJobIterator( $conn, $uids );
 560         }
 561
 562         /**
 563          * @see JobQueue::getAllAbandonedJobs()
 564          * @return Iterator
 565          * @throws JobQueueError
 566          */
 567         public function getAllAbandonedJobs() {
 568                 $conn = $this->getConnection();
 569                 try {
 570                         $uids = $conn->zRange( $this->getQueueKey( 'z-abandoned' ), 0, -1 );
 571                 } catch ( RedisException $e ) {
 572                         $this->throwRedisException( $conn, $e );
 573                 }
 574
 575                 return $this->getJobIterator( $conn, $uids );
 576         }
 577
 578         /**
 579          * @param RedisConnRef $conn
 580          * @param array $uids List of job UUIDs
 581          * @return MappedIterator
 582          */
 583         protected function getJobIterator( RedisConnRef $conn, array $uids ) {
 584                 return new MappedIterator(
 585                         $uids,
 586                         function ( $uid ) use ( $conn ) {
 587                                 return $this->getJobFromUidInternal( $uid, $conn );
 588                         },
 589                         [ 'accept' => function ( $job ) {
 590                                 return is_object( $job );
 591                         } ]
 592                 );
 593         }
 594
 595         public function getCoalesceLocationInternal() {
 596                 return "RedisServer:" . $this->server;
 597         }
 598
 599         protected function doGetSiblingQueuesWithJobs( array $types ) {
 600                 return array_keys( array_filter( $this->doGetSiblingQueueSizes( $types ) ) );
 601         }
 602
 603         protected function doGetSiblingQueueSizes( array $types ) {
 604                 $sizes = []; // (type => size)
 605                 $types = array_values( $types ); // reindex
 606                 $conn = $this->getConnection();
 607                 try {
 608                         $conn->multi( Redis::PIPELINE );
 609                         foreach ( $types as $type ) {
 610                                 $conn->lSize( $this->getQueueKey( 'l-unclaimed', $type ) );
 611                         }
 612                         $res = $conn->exec();
 613                         if ( is_array( $res ) ) {
 614                                 foreach ( $res as $i => $size ) {
 615                                         $sizes[$types[$i]] = $size;
 616                                 }
 617                         }
 618                 } catch ( RedisException $e ) {
 619                         $this->throwRedisException( $conn, $e );
 620                 }
 621
 622                 return $sizes;
 623         }
 624
 625         /**
 626          * This function should not be called outside JobQueueRedis
 627          *
 628          * @param string $uid
 629          * @param RedisConnRef $conn
 630          * @return Job|bool Returns false if the job does not exist
 631          * @throws JobQueueError
 632          * @throws UnexpectedValueException
 633          */
 634         public function getJobFromUidInternal( $uid, RedisConnRef $conn ) {
 635                 try {
 636                         $data = $conn->hGet( $this->getQueueKey( 'h-data' ), $uid );
 637                         if ( $data === false ) {
 638                                 return false; // not found
 639                         }
 640                         $item = $this->unserialize( $data );
 641                         if ( !is_array( $item ) ) { // this shouldn't happen
 642                                 throw new UnexpectedValueException( "Could not find job with ID '$uid'." );
 643                         }
 644                         $title = Title::makeTitle( $item['namespace'], $item['title'] );
 645                         $job = Job::factory( $item['type'], $title, $item['params'] );
 646                         $job->metadata['uuid'] = $item['uuid'];
 647                         $job->metadata['timestamp'] = $item['timestamp'];
 648                         // Add in attempt count for debugging at showJobs.php
 649                         $job->metadata['attempts'] = $conn->hGet( $this->getQueueKey( 'h-attempts' ), $uid );
 650
 651                         return $job;
 652                 } catch ( RedisException $e ) {
 653                         $this->throwRedisException( $conn, $e );
 654                 }
 655         }
 656
 657         /**
 658          * @return array List of (wiki,type) tuples for queues with non-abandoned jobs
 659          * @throws JobQueueConnectionError
 660          * @throws JobQueueError
 661          */
 662         public function getServerQueuesWithJobs() {
 663                 $queues = [];
 664
 665                 $conn = $this->getConnection();
 666                 try {
 667                         $set = $conn->sMembers( $this->getGlobalKey( 's-queuesWithJobs' ) );
 668                         foreach ( $set as $queue ) {
 669                                 $queues[] = $this->decodeQueueName( $queue );
 670                         }
 671                 } catch ( RedisException $e ) {
 672                         $this->throwRedisException( $conn, $e );
 673                 }
 674
 675                 return $queues;
 676         }
 677
 678         /**
 679          * @param IJobSpecification $job
 680          * @return array
 681          */
 682         protected function getNewJobFields( IJobSpecification $job ) {
 683                 return [
 684                         // Fields that describe the nature of the job
 685                         'type' => $job->getType(),
 686                         'namespace' => $job->getTitle()->getNamespace(),
 687                         'title' => $job->getTitle()->getDBkey(),
 688                         'params' => $job->getParams(),
 689                         // Some jobs cannot run until a "release timestamp"
 690                         'rtimestamp' => $job->getReleaseTimestamp() ?: 0,
 691                         // Additional job metadata
 692                         'uuid' => UIDGenerator::newRawUUIDv4( UIDGenerator::QUICK_RAND ),
 693                         'sha1' => $job->ignoreDuplicates()
 694                                 ? Wikimedia\base_convert( sha1( serialize( $job->getDeduplicationInfo() ) ), 16, 36, 31 )
 695                                 : '',
 696                         'timestamp' => time() // UNIX timestamp
 697                 ];
 698         }
 699
 700         /**
 701          * @param array $fields
 702          * @return Job|bool
 703          */
 704         protected function getJobFromFields( array $fields ) {
 705                 $title = Title::makeTitle( $fields['namespace'], $fields['title'] );
 706                 $job = Job::factory( $fields['type'], $title, $fields['params'] );
 707                 $job->metadata['uuid'] = $fields['uuid'];
 708                 $job->metadata['timestamp'] = $fields['timestamp'];
 709
 710                 return $job;
 711         }
 712
 713         /**
 714          * @param array $fields
 715          * @return string Serialized and possibly compressed version of $fields
 716          */
 717         protected function serialize( array $fields ) {
 718                 $blob = serialize( $fields );
 719                 if ( $this->compression === 'gzip'
 720                         && strlen( $blob ) >= 1024
 721                         && function_exists( 'gzdeflate' )
 722                 ) {
 723                         $object = (object)[ 'blob' => gzdeflate( $blob ), 'enc' => 'gzip' ];
 724                         $blobz = serialize( $object );
 725
 726                         return ( strlen( $blobz ) < strlen( $blob ) ) ? $blobz : $blob;
 727                 } else {
 728                         return $blob;
 729                 }
 730         }
 731
 732         /**
 733          * @param string $blob
 734          * @return array|bool Unserialized version of $blob or false
 735          */
 736         protected function unserialize( $blob ) {
 737                 $fields = unserialize( $blob );
 738                 if ( is_object( $fields ) ) {
 739                         if ( $fields->enc === 'gzip' && function_exists( 'gzinflate' ) ) {
 740                                 $fields = unserialize( gzinflate( $fields->blob ) );
 741                         } else {
 742                                 $fields = false;
 743                         }
 744                 }
 745
 746                 return is_array( $fields ) ? $fields : false;
 747         }
 748
 749         /**
 750          * Get a connection to the server that handles all sub-queues for this queue
 751          *
 752          * @return RedisConnRef
 753          * @throws JobQueueConnectionError
 754          */
 755         protected function getConnection() {
 756                 $conn = $this->redisPool->getConnection( $this->server, $this->logger );
 757                 if ( !$conn ) {
 758                         throw new JobQueueConnectionError(
 759                                 "Unable to connect to redis server {$this->server}." );
 760                 }
 761
 762                 return $conn;
 763         }
 764
 765         /**
 766          * @param RedisConnRef $conn
 767          * @param RedisException $e
 768          * @throws JobQueueError
 769          */
 770         protected function throwRedisException( RedisConnRef $conn, $e ) {
 771                 $this->redisPool->handleError( $conn, $e );
 772                 throw new JobQueueError( "Redis server error: {$e->getMessage()}\n" );
 773         }
 774
 775         /**
 776          * @return string JSON
 777          */
 778         private function encodeQueueName() {
 779                 return json_encode( [ $this->type, $this->wiki ] );
 780         }
 781
 782         /**
 783          * @param string $name JSON
 784          * @return array (type, wiki)
 785          */
 786         private function decodeQueueName( $name ) {
 787                 return json_decode( $name );
 788         }
 789
 790         /**
 791          * @param string $name
 792          * @return string
 793          */
 794         private function getGlobalKey( $name ) {
 795                 $parts = [ 'global', 'jobqueue', $name ];
 796                 foreach ( $parts as $part ) {
 797                         if ( !preg_match( '/[a-zA-Z0-9_-]+/', $part ) ) {
 798                                 throw new InvalidArgumentException( "Key part characters are out of range." );
 799                         }
 800                 }
 801
 802                 return implode( ':', $parts );
 803         }
 804
 805         /**
 806          * @param string $prop
 807          * @param string|null $type Override this for sibling queues
 808          * @return string
 809          */
 810         private function getQueueKey( $prop, $type = null ) {
 811                 $type = is_string( $type ) ? $type : $this->type;
 812                 list( $db, $prefix ) = wfSplitWikiID( $this->wiki );
 813                 $keyspace = $prefix ? "$db-$prefix" : $db;
 814
 815                 $parts = [ $keyspace, 'jobqueue', $type, $prop ];
 816
 817                 // Parts are typically ASCII, but encode for sanity to escape ":"
 818                 return implode( ':', array_map( 'rawurlencode', $parts ) );
 819         }
 820 }