includes/jobqueue/JobQueueFederated.php

   1 <?php
   2 /**
   3  * Job queue code for federated queues.
   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  * @author Aaron Schulz
  22  */
  23
  24 /**
  25  * Class to handle enqueueing and running of background jobs for federated queues
  26  *
  27  * This class allows for queues to be partitioned into smaller queues.
  28  * A partition is defined by the configuration for a JobQueue instance.
  29  * For example, one can set $wgJobTypeConf['refreshLinks'] to point to a
  30  * JobQueueFederated instance, which itself would consist of three JobQueueRedis
  31  * instances, each using their own redis server. This would allow for the jobs
  32  * to be split (evenly or based on weights) accross multiple servers if a single
  33  * server becomes impractical or expensive. Different JobQueue classes can be mixed.
  34  *
  35  * The basic queue configuration (e.g. "order", "claimTTL") of a federated queue
  36  * is inherited by the partition queues. Additional configuration defines what
  37  * section each wiki is in, what partition queues each section uses (and their weight),
  38  * and the JobQueue configuration for each partition. Some sections might only need a
  39  * single queue partition, like the sections for groups of small wikis.
  40  *
  41  * If used for performance, then $wgMainCacheType should be set to memcached/redis.
  42  * Note that "fifo" cannot be used for the ordering, since the data is distributed.
  43  * One can still use "timestamp" instead, as in "roughly timestamp ordered". Also,
  44  * queue classes used by this should ignore down servers (with TTL) to avoid slowness.
  45  *
  46  * @ingroup JobQueue
  47  * @since 1.22
  48  */
  49 class JobQueueFederated extends JobQueue {
  50         /** @var HashRing */
  51         protected $partitionRing;
  52         /** @var JobQueue[] (partition name => JobQueue) reverse sorted by weight */
  53         protected $partitionQueues = [];
  54
  55         /** @var int Maximum number of partitions to try */
  56         protected $maxPartitionsTry;
  57
  58         /**
  59          * @param array $params Possible keys:
  60          *  - sectionsByWiki      : A map of wiki IDs to section names.
  61          *                          Wikis will default to using the section "default".
  62          *  - partitionsBySection : Map of section names to maps of (partition name => weight).
  63          *                          A section called 'default' must be defined if not all wikis
  64          *                          have explicitly defined sections.
  65          *  - configByPartition   : Map of queue partition names to configuration arrays.
  66          *                          These configuration arrays are passed to JobQueue::factory().
  67          *                          The options set here are overridden by those passed to this
  68          *                          the federated queue itself (e.g. 'order' and 'claimTTL').
  69          *  - maxPartitionsTry    : Maximum number of times to attempt job insertion using
  70          *                          different partition queues. This improves availability
  71          *                          during failure, at the cost of added latency and somewhat
  72          *                          less reliable job de-duplication mechanisms.
  73          * @throws MWException
  74          */
  75         protected function __construct( array $params ) {
  76                 parent::__construct( $params );
  77                 $section = isset( $params['sectionsByWiki'][$this->wiki] )
  78                         ? $params['sectionsByWiki'][$this->wiki]
  79                         : 'default';
  80                 if ( !isset( $params['partitionsBySection'][$section] ) ) {
  81                         throw new MWException( "No configuration for section '$section'." );
  82                 }
  83                 $this->maxPartitionsTry = isset( $params['maxPartitionsTry'] )
  84                         ? $params['maxPartitionsTry']
  85                         : 2;
  86                 // Get the full partition map
  87                 $partitionMap = $params['partitionsBySection'][$section];
  88                 arsort( $partitionMap, SORT_NUMERIC );
  89                 // Get the config to pass to merge into each partition queue config
  90                 $baseConfig = $params;
  91                 foreach ( [ 'class', 'sectionsByWiki', 'maxPartitionsTry',
  92                         'partitionsBySection', 'configByPartition', ] as $o
  93                 ) {
  94                         unset( $baseConfig[$o] ); // partition queue doesn't care about this
  95                 }
  96                 // The class handles all aggregator calls already
  97                 unset( $baseConfig['aggregator'] );
  98                 // Get the partition queue objects
  99                 foreach ( $partitionMap as $partition => $w ) {
 100                         if ( !isset( $params['configByPartition'][$partition] ) ) {
 101                                 throw new MWException( "No configuration for partition '$partition'." );
 102                         }
 103                         $this->partitionQueues[$partition] = JobQueue::factory(
 104                                 $baseConfig + $params['configByPartition'][$partition] );
 105                 }
 106                 // Ring of all partitions
 107                 $this->partitionRing = new HashRing( $partitionMap );
 108         }
 109
 110         protected function supportedOrders() {
 111                 // No FIFO due to partitioning, though "rough timestamp order" is supported
 112                 return [ 'undefined', 'random', 'timestamp' ];
 113         }
 114
 115         protected function optimalOrder() {
 116                 return 'undefined'; // defer to the partitions
 117         }
 118
 119         protected function supportsDelayedJobs() {
 120                 foreach ( $this->partitionQueues as $queue ) {
 121                         if ( !$queue->supportsDelayedJobs() ) {
 122                                 return false;
 123                         }
 124                 }
 125
 126                 return true;
 127         }
 128
 129         protected function doIsEmpty() {
 130                 $empty = true;
 131                 $failed = 0;
 132                 foreach ( $this->partitionQueues as $queue ) {
 133                         try {
 134                                 $empty = $empty && $queue->doIsEmpty();
 135                         } catch ( JobQueueError $e ) {
 136                                 ++$failed;
 137                                 $this->logException( $e );
 138                         }
 139                 }
 140                 $this->throwErrorIfAllPartitionsDown( $failed );
 141
 142                 return $empty;
 143         }
 144
 145         protected function doGetSize() {
 146                 return $this->getCrossPartitionSum( 'size', 'doGetSize' );
 147         }
 148
 149         protected function doGetAcquiredCount() {
 150                 return $this->getCrossPartitionSum( 'acquiredcount', 'doGetAcquiredCount' );
 151         }
 152
 153         protected function doGetDelayedCount() {
 154                 return $this->getCrossPartitionSum( 'delayedcount', 'doGetDelayedCount' );
 155         }
 156
 157         protected function doGetAbandonedCount() {
 158                 return $this->getCrossPartitionSum( 'abandonedcount', 'doGetAbandonedCount' );
 159         }
 160
 161         /**
 162          * @param string $type
 163          * @param string $method
 164          * @return int
 165          */
 166         protected function getCrossPartitionSum( $type, $method ) {
 167                 $count = 0;
 168                 $failed = 0;
 169                 foreach ( $this->partitionQueues as $queue ) {
 170                         try {
 171                                 $count += $queue->$method();
 172                         } catch ( JobQueueError $e ) {
 173                                 ++$failed;
 174                                 $this->logException( $e );
 175                         }
 176                 }
 177                 $this->throwErrorIfAllPartitionsDown( $failed );
 178
 179                 return $count;
 180         }
 181
 182         protected function doBatchPush( array $jobs, $flags ) {
 183                 // Local ring variable that may be changed to point to a new ring on failure
 184                 $partitionRing = $this->partitionRing;
 185                 // Try to insert the jobs and update $partitionsTry on any failures.
 186                 // Retry to insert any remaning jobs again, ignoring the bad partitions.
 187                 $jobsLeft = $jobs;
 188                 // @codingStandardsIgnoreStart Generic.CodeAnalysis.ForLoopWithTestFunctionCall.NotAllowed
 189                 for ( $i = $this->maxPartitionsTry; $i > 0 && count( $jobsLeft ); --$i ) {
 190                         // @codingStandardsIgnoreEnd
 191                         try {
 192                                 $partitionRing->getLiveRing();
 193                         } catch ( UnexpectedValueException $e ) {
 194                                 break; // all servers down; nothing to insert to
 195                         }
 196                         $jobsLeft = $this->tryJobInsertions( $jobsLeft, $partitionRing, $flags );
 197                 }
 198                 if ( count( $jobsLeft ) ) {
 199                         throw new JobQueueError(
 200                                 "Could not insert job(s), {$this->maxPartitionsTry} partitions tried." );
 201                 }
 202         }
 203
 204         /**
 205          * @param array $jobs
 206          * @param HashRing $partitionRing
 207          * @param int $flags
 208          * @throws JobQueueError
 209          * @return array List of Job object that could not be inserted
 210          */
 211         protected function tryJobInsertions( array $jobs, HashRing &$partitionRing, $flags ) {
 212                 $jobsLeft = [];
 213
 214                 // Because jobs are spread across partitions, per-job de-duplication needs
 215                 // to use a consistent hash to avoid allowing duplicate jobs per partition.
 216                 // When inserting a batch of de-duplicated jobs, QOS_ATOMIC is disregarded.
 217                 $uJobsByPartition = []; // (partition name => job list)
 218                 /** @var Job $job */
 219                 foreach ( $jobs as $key => $job ) {
 220                         if ( $job->ignoreDuplicates() ) {
 221                                 $sha1 = sha1( serialize( $job->getDeduplicationInfo() ) );
 222                                 $uJobsByPartition[$partitionRing->getLiveLocation( $sha1 )][] = $job;
 223                                 unset( $jobs[$key] );
 224                         }
 225                 }
 226                 // Get the batches of jobs that are not de-duplicated
 227                 if ( $flags & self::QOS_ATOMIC ) {
 228                         $nuJobBatches = [ $jobs ]; // all or nothing
 229                 } else {
 230                         // Split the jobs into batches and spread them out over servers if there
 231                         // are many jobs. This helps keep the partitions even. Otherwise, send all
 232                         // the jobs to a single partition queue to avoids the extra connections.
 233                         $nuJobBatches = array_chunk( $jobs, 300 );
 234                 }
 235
 236                 // Insert the de-duplicated jobs into the queues...
 237                 foreach ( $uJobsByPartition as $partition => $jobBatch ) {
 238                         /** @var JobQueue $queue */
 239                         $queue = $this->partitionQueues[$partition];
 240                         try {
 241                                 $ok = true;
 242                                 $queue->doBatchPush( $jobBatch, $flags | self::QOS_ATOMIC );
 243                         } catch ( JobQueueError $e ) {
 244                                 $ok = false;
 245                                 $this->logException( $e );
 246                         }
 247                         if ( !$ok ) {
 248                                 if ( !$partitionRing->ejectFromLiveRing( $partition, 5 ) ) { // blacklist
 249                                         throw new JobQueueError( "Could not insert job(s), no partitions available." );
 250                                 }
 251                                 $jobsLeft = array_merge( $jobsLeft, $jobBatch ); // not inserted
 252                         }
 253                 }
 254
 255                 // Insert the jobs that are not de-duplicated into the queues...
 256                 foreach ( $nuJobBatches as $jobBatch ) {
 257                         $partition = ArrayUtils::pickRandom( $partitionRing->getLiveLocationWeights() );
 258                         $queue = $this->partitionQueues[$partition];
 259                         try {
 260                                 $ok = true;
 261                                 $queue->doBatchPush( $jobBatch, $flags | self::QOS_ATOMIC );
 262                         } catch ( JobQueueError $e ) {
 263                                 $ok = false;
 264                                 $this->logException( $e );
 265                         }
 266                         if ( !$ok ) {
 267                                 if ( !$partitionRing->ejectFromLiveRing( $partition, 5 ) ) { // blacklist
 268                                         throw new JobQueueError( "Could not insert job(s), no partitions available." );
 269                                 }
 270                                 $jobsLeft = array_merge( $jobsLeft, $jobBatch ); // not inserted
 271                         }
 272                 }
 273
 274                 return $jobsLeft;
 275         }
 276
 277         protected function doPop() {
 278                 $partitionsTry = $this->partitionRing->getLiveLocationWeights(); // (partition => weight)
 279
 280                 $failed = 0;
 281                 while ( count( $partitionsTry ) ) {
 282                         $partition = ArrayUtils::pickRandom( $partitionsTry );
 283                         if ( $partition === false ) {
 284                                 break; // all partitions at 0 weight
 285                         }
 286
 287                         /** @var JobQueue $queue */
 288                         $queue = $this->partitionQueues[$partition];
 289                         try {
 290                                 $job = $queue->pop();
 291                         } catch ( JobQueueError $e ) {
 292                                 ++$failed;
 293                                 $this->logException( $e );
 294                                 $job = false;
 295                         }
 296                         if ( $job ) {
 297                                 $job->metadata['QueuePartition'] = $partition;
 298
 299                                 return $job;
 300                         } else {
 301                                 unset( $partitionsTry[$partition] ); // blacklist partition
 302                         }
 303                 }
 304                 $this->throwErrorIfAllPartitionsDown( $failed );
 305
 306                 return false;
 307         }
 308
 309         protected function doAck( Job $job ) {
 310                 if ( !isset( $job->metadata['QueuePartition'] ) ) {
 311                         throw new MWException( "The given job has no defined partition name." );
 312                 }
 313
 314                 $this->partitionQueues[$job->metadata['QueuePartition']]->ack( $job );
 315         }
 316
 317         protected function doIsRootJobOldDuplicate( Job $job ) {
 318                 $signature = $job->getRootJobParams()['rootJobSignature'];
 319                 $partition = $this->partitionRing->getLiveLocation( $signature );
 320                 try {
 321                         return $this->partitionQueues[$partition]->doIsRootJobOldDuplicate( $job );
 322                 } catch ( JobQueueError $e ) {
 323                         if ( $this->partitionRing->ejectFromLiveRing( $partition, 5 ) ) {
 324                                 $partition = $this->partitionRing->getLiveLocation( $signature );
 325                                 return $this->partitionQueues[$partition]->doIsRootJobOldDuplicate( $job );
 326                         }
 327                 }
 328
 329                 return false;
 330         }
 331
 332         protected function doDeduplicateRootJob( IJobSpecification $job ) {
 333                 $signature = $job->getRootJobParams()['rootJobSignature'];
 334                 $partition = $this->partitionRing->getLiveLocation( $signature );
 335                 try {
 336                         return $this->partitionQueues[$partition]->doDeduplicateRootJob( $job );
 337                 } catch ( JobQueueError $e ) {
 338                         if ( $this->partitionRing->ejectFromLiveRing( $partition, 5 ) ) {
 339                                 $partition = $this->partitionRing->getLiveLocation( $signature );
 340                                 return $this->partitionQueues[$partition]->doDeduplicateRootJob( $job );
 341                         }
 342                 }
 343
 344                 return false;
 345         }
 346
 347         protected function doDelete() {
 348                 $failed = 0;
 349                 /** @var JobQueue $queue */
 350                 foreach ( $this->partitionQueues as $queue ) {
 351                         try {
 352                                 $queue->doDelete();
 353                         } catch ( JobQueueError $e ) {
 354                                 ++$failed;
 355                                 $this->logException( $e );
 356                         }
 357                 }
 358                 $this->throwErrorIfAllPartitionsDown( $failed );
 359                 return true;
 360         }
 361
 362         protected function doWaitForBackups() {
 363                 $failed = 0;
 364                 /** @var JobQueue $queue */
 365                 foreach ( $this->partitionQueues as $queue ) {
 366                         try {
 367                                 $queue->waitForBackups();
 368                         } catch ( JobQueueError $e ) {
 369                                 ++$failed;
 370                                 $this->logException( $e );
 371                         }
 372                 }
 373                 $this->throwErrorIfAllPartitionsDown( $failed );
 374         }
 375
 376         protected function doFlushCaches() {
 377                 /** @var JobQueue $queue */
 378                 foreach ( $this->partitionQueues as $queue ) {
 379                         $queue->doFlushCaches();
 380                 }
 381         }
 382
 383         public function getAllQueuedJobs() {
 384                 $iterator = new AppendIterator();
 385
 386                 /** @var JobQueue $queue */
 387                 foreach ( $this->partitionQueues as $queue ) {
 388                         $iterator->append( $queue->getAllQueuedJobs() );
 389                 }
 390
 391                 return $iterator;
 392         }
 393
 394         public function getAllDelayedJobs() {
 395                 $iterator = new AppendIterator();
 396
 397                 /** @var JobQueue $queue */
 398                 foreach ( $this->partitionQueues as $queue ) {
 399                         $iterator->append( $queue->getAllDelayedJobs() );
 400                 }
 401
 402                 return $iterator;
 403         }
 404
 405         public function getAllAcquiredJobs() {
 406                 $iterator = new AppendIterator();
 407
 408                 /** @var JobQueue $queue */
 409                 foreach ( $this->partitionQueues as $queue ) {
 410                         $iterator->append( $queue->getAllAcquiredJobs() );
 411                 }
 412
 413                 return $iterator;
 414         }
 415
 416         public function getAllAbandonedJobs() {
 417                 $iterator = new AppendIterator();
 418
 419                 /** @var JobQueue $queue */
 420                 foreach ( $this->partitionQueues as $queue ) {
 421                         $iterator->append( $queue->getAllAbandonedJobs() );
 422                 }
 423
 424                 return $iterator;
 425         }
 426
 427         public function getCoalesceLocationInternal() {
 428                 return "JobQueueFederated:wiki:{$this->wiki}" .
 429                         sha1( serialize( array_keys( $this->partitionQueues ) ) );
 430         }
 431
 432         protected function doGetSiblingQueuesWithJobs( array $types ) {
 433                 $result = [];
 434
 435                 $failed = 0;
 436                 /** @var JobQueue $queue */
 437                 foreach ( $this->partitionQueues as $queue ) {
 438                         try {
 439                                 $nonEmpty = $queue->doGetSiblingQueuesWithJobs( $types );
 440                                 if ( is_array( $nonEmpty ) ) {
 441                                         $result = array_unique( array_merge( $result, $nonEmpty ) );
 442                                 } else {
 443                                         return null; // not supported on all partitions; bail
 444                                 }
 445                                 if ( count( $result ) == count( $types ) ) {
 446                                         break; // short-circuit
 447                                 }
 448                         } catch ( JobQueueError $e ) {
 449                                 ++$failed;
 450                                 $this->logException( $e );
 451                         }
 452                 }
 453                 $this->throwErrorIfAllPartitionsDown( $failed );
 454
 455                 return array_values( $result );
 456         }
 457
 458         protected function doGetSiblingQueueSizes( array $types ) {
 459                 $result = [];
 460                 $failed = 0;
 461                 /** @var JobQueue $queue */
 462                 foreach ( $this->partitionQueues as $queue ) {
 463                         try {
 464                                 $sizes = $queue->doGetSiblingQueueSizes( $types );
 465                                 if ( is_array( $sizes ) ) {
 466                                         foreach ( $sizes as $type => $size ) {
 467                                                 $result[$type] = isset( $result[$type] ) ? $result[$type] + $size : $size;
 468                                         }
 469                                 } else {
 470                                         return null; // not supported on all partitions; bail
 471                                 }
 472                         } catch ( JobQueueError $e ) {
 473                                 ++$failed;
 474                                 $this->logException( $e );
 475                         }
 476                 }
 477                 $this->throwErrorIfAllPartitionsDown( $failed );
 478
 479                 return $result;
 480         }
 481
 482         protected function logException( Exception $e ) {
 483                 wfDebugLog( 'JobQueueFederated', $e->getMessage() . "\n" . $e->getTraceAsString() );
 484         }
 485
 486         /**
 487          * Throw an error if no partitions available
 488          *
 489          * @param int $down The number of up partitions down
 490          * @return void
 491          * @throws JobQueueError
 492          */
 493         protected function throwErrorIfAllPartitionsDown( $down ) {
 494                 if ( $down >= count( $this->partitionQueues ) ) {
 495                         throw new JobQueueError( 'No queue partitions available.' );
 496                 }
 497         }
 498 }