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