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 HashRing */
  53         protected $partitionPushRing;
  54         /** @var array (partition name => JobQueue) reverse sorted by weight */
  55         protected $partitionQueues = array();
  56
  57         /** @var BagOStuff */
  58         protected $cache;
  59
  60         /** @var int Maximum number of partitions to try */
  61         protected $maxPartitionsTry;
  62
  63         const CACHE_TTL_SHORT = 30; // integer; seconds to cache info without re-validating
  64         const CACHE_TTL_LONG = 300; // integer; seconds to cache info that is kept up to date
  65
  66         /**
  67          * @params include:
  68          *  - sectionsByWiki      : A map of wiki IDs to section names.
  69          *                          Wikis will default to using the section "default".
  70          *  - partitionsBySection : Map of section names to maps of (partition name => weight).
  71          *                          A section called 'default' must be defined if not all wikis
  72          *                          have explicitly defined sections.
  73          *  - configByPartition   : Map of queue partition names to configuration arrays.
  74          *                          These configuration arrays are passed to JobQueue::factory().
  75          *                          The options set here are overriden by those passed to this
  76          *                          the federated queue itself (e.g. 'order' and 'claimTTL').
  77          *  - partitionsNoPush    : List of partition names that can handle pop() but not push().
  78          *                          This can be used to migrate away from a certain partition.
  79          *  - maxPartitionsTry    : Maximum number of times to attempt job insertion using
  80          *                          different partition queues. This improves availability
  81          *                          during failure, at the cost of added latency and somewhat
  82          *                          less reliable job de-duplication mechanisms.
  83          * @param array $params
  84          * @throws MWException
  85          */
  86         protected function __construct( array $params ) {
  87                 parent::__construct( $params );
  88                 $section = isset( $params['sectionsByWiki'][$this->wiki] )
  89                         ? $params['sectionsByWiki'][$this->wiki]
  90                         : 'default';
  91                 if ( !isset( $params['partitionsBySection'][$section] ) ) {
  92                         throw new MWException( "No configuration for section '$section'." );
  93                 }
  94                 $this->maxPartitionsTry = isset( $params['maxPartitionsTry'] )
  95                         ? $params['maxPartitionsTry']
  96                         : 2;
  97                 // Get the full partition map
  98                 $partitionMap = $params['partitionsBySection'][$section];
  99                 arsort( $partitionMap, SORT_NUMERIC );
 100                 // Get the partitions jobs can actually be pushed to
 101                 $partitionPushMap = $partitionMap;
 102                 if ( isset( $params['partitionsNoPush'] ) ) {
 103                         foreach ( $params['partitionsNoPush'] as $partition ) {
 104                                 unset( $partitionPushMap[$partition] );
 105                         }
 106                 }
 107                 // Get the config to pass to merge into each partition queue config
 108                 $baseConfig = $params;
 109                 foreach ( array( 'class', 'sectionsByWiki', 'maxPartitionsTry',
 110                         'partitionsBySection', 'configByPartition', 'partitionsNoPush' ) as $o
 111                 ) {
 112                         unset( $baseConfig[$o] ); // partition queue doesn't care about this
 113                 }
 114                 // Get the partition queue objects
 115                 foreach ( $partitionMap as $partition => $w ) {
 116                         if ( !isset( $params['configByPartition'][$partition] ) ) {
 117                                 throw new MWException( "No configuration for partition '$partition'." );
 118                         }
 119                         $this->partitionQueues[$partition] = JobQueue::factory(
 120                                 $baseConfig + $params['configByPartition'][$partition] );
 121                 }
 122                 // Ring of all partitions
 123                 $this->partitionRing = new HashRing( $partitionMap );
 124                 // Get the ring of partitions to push jobs into
 125                 if ( count( $partitionPushMap ) === count( $partitionMap ) ) {
 126                         $this->partitionPushRing = clone $this->partitionRing; // faster
 127                 } else {
 128                         $this->partitionPushRing = new HashRing( $partitionPushMap );
 129                 }
 130                 // Aggregate cache some per-queue values if there are multiple partition queues
 131                 $this->cache = count( $partitionMap ) > 1 ? wfGetMainCache() : new EmptyBagOStuff();
 132         }
 133
 134         protected function supportedOrders() {
 135                 // No FIFO due to partitioning, though "rough timestamp order" is supported
 136                 return array( 'undefined', 'random', 'timestamp' );
 137         }
 138
 139         protected function optimalOrder() {
 140                 return 'undefined'; // defer to the partitions
 141         }
 142
 143         protected function supportsDelayedJobs() {
 144                 return true; // defer checks to the partitions
 145         }
 146
 147         protected function doIsEmpty() {
 148                 $key = $this->getCacheKey( 'empty' );
 149
 150                 $isEmpty = $this->cache->get( $key );
 151                 if ( $isEmpty === 'true' ) {
 152                         return true;
 153                 } elseif ( $isEmpty === 'false' ) {
 154                         return false;
 155                 }
 156
 157                 $empty = true;
 158                 $failed = 0;
 159                 foreach ( $this->partitionQueues as $queue ) {
 160                         try {
 161                                 $empty = $empty && $queue->doIsEmpty();
 162                         } catch ( JobQueueError $e ) {
 163                                 ++$failed;
 164                                 MWExceptionHandler::logException( $e );
 165                         }
 166                 }
 167                 $this->throwErrorIfAllPartitionsDown( $failed );
 168
 169                 $this->cache->add( $key, $empty ? 'true' : 'false', self::CACHE_TTL_LONG );
 170                 return $empty;
 171         }
 172
 173         protected function doGetSize() {
 174                 return $this->getCrossPartitionSum( 'size', 'doGetSize' );
 175         }
 176
 177         protected function doGetAcquiredCount() {
 178                 return $this->getCrossPartitionSum( 'acquiredcount', 'doGetAcquiredCount' );
 179         }
 180
 181         protected function doGetDelayedCount() {
 182                 return $this->getCrossPartitionSum( 'delayedcount', 'doGetDelayedCount' );
 183         }
 184
 185         protected function doGetAbandonedCount() {
 186                 return $this->getCrossPartitionSum( 'abandonedcount', 'doGetAbandonedCount' );
 187         }
 188
 189         /**
 190          * @param string $type
 191          * @param string $method
 192          * @return int
 193          */
 194         protected function getCrossPartitionSum( $type, $method ) {
 195                 $key = $this->getCacheKey( $type );
 196
 197                 $count = $this->cache->get( $key );
 198                 if ( is_int( $count ) ) {
 199                         return $count;
 200                 }
 201
 202                 $failed = 0;
 203                 foreach ( $this->partitionQueues as $queue ) {
 204                         try {
 205                                 $count += $queue->$method();
 206                         } catch ( JobQueueError $e ) {
 207                                 ++$failed;
 208                                 MWExceptionHandler::logException( $e );
 209                         }
 210                 }
 211                 $this->throwErrorIfAllPartitionsDown( $failed );
 212
 213                 $this->cache->set( $key, $count, self::CACHE_TTL_SHORT );
 214
 215                 return $count;
 216         }
 217
 218         protected function doBatchPush( array $jobs, $flags ) {
 219                 // Local ring variable that may be changed to point to a new ring on failure
 220                 $partitionRing = $this->partitionPushRing;
 221                 // Try to insert the jobs and update $partitionsTry on any failures.
 222                 // Retry to insert any remaning jobs again, ignoring the bad partitions.
 223                 $jobsLeft = $jobs;
 224                 // @codingStandardsIgnoreStart Generic.CodeAnalysis.ForLoopWithTestFunctionCall.NotAllowed
 225                 for ( $i = $this->maxPartitionsTry; $i > 0 && count( $jobsLeft ); --$i ) {
 226                         // @codingStandardsIgnoreEnd
 227                         try {
 228                                 $partitionRing->getLiveRing();
 229                         } catch ( UnexpectedValueException $e ) {
 230                                 break; // all servers down; nothing to insert to
 231                         }
 232                         $jobsLeft = $this->tryJobInsertions( $jobsLeft, $partitionRing, $flags );
 233                 }
 234                 if ( count( $jobsLeft ) ) {
 235                         throw new JobQueueError(
 236                                 "Could not insert job(s), {$this->maxPartitionsTry} partitions tried." );
 237                 }
 238
 239                 return true;
 240         }
 241
 242         /**
 243          * @param array $jobs
 244          * @param HashRing $partitionRing
 245          * @param int $flags
 246          * @throws JobQueueError
 247          * @return array List of Job object that could not be inserted
 248          */
 249         protected function tryJobInsertions( array $jobs, HashRing &$partitionRing, $flags ) {
 250                 $jobsLeft = array();
 251
 252                 // Because jobs are spread across partitions, per-job de-duplication needs
 253                 // to use a consistent hash to avoid allowing duplicate jobs per partition.
 254                 // When inserting a batch of de-duplicated jobs, QOS_ATOMIC is disregarded.
 255                 $uJobsByPartition = array(); // (partition name => job list)
 256                 /** @var Job $job */
 257                 foreach ( $jobs as $key => $job ) {
 258                         if ( $job->ignoreDuplicates() ) {
 259                                 $sha1 = sha1( serialize( $job->getDeduplicationInfo() ) );
 260                                 $uJobsByPartition[$partitionRing->getLiveLocation( $sha1 )][] = $job;
 261                                 unset( $jobs[$key] );
 262                         }
 263                 }
 264                 // Get the batches of jobs that are not de-duplicated
 265                 if ( $flags & self::QOS_ATOMIC ) {
 266                         $nuJobBatches = array( $jobs ); // all or nothing
 267                 } else {
 268                         // Split the jobs into batches and spread them out over servers if there
 269                         // are many jobs. This helps keep the partitions even. Otherwise, send all
 270                         // the jobs to a single partition queue to avoids the extra connections.
 271                         $nuJobBatches = array_chunk( $jobs, 300 );
 272                 }
 273
 274                 // Insert the de-duplicated jobs into the queues...
 275                 foreach ( $uJobsByPartition as $partition => $jobBatch ) {
 276                         /** @var JobQueue $queue */
 277                         $queue = $this->partitionQueues[$partition];
 278                         try {
 279                                 $ok = true;
 280                                 $queue->doBatchPush( $jobBatch, $flags | self::QOS_ATOMIC );
 281                         } catch ( JobQueueError $e ) {
 282                                 $ok = false;
 283                                 MWExceptionHandler::logException( $e );
 284                         }
 285                         if ( $ok ) {
 286                                 $key = $this->getCacheKey( 'empty' );
 287                                 $this->cache->set( $key, 'false', JobQueueDB::CACHE_TTL_LONG );
 288                         } else {
 289                                 if ( !$partitionRing->ejectFromLiveRing( $partition, 5 ) ) { // blacklist
 290                                         throw new JobQueueError( "Could not insert job(s), no partitions available." );
 291                                 }
 292                                 $jobsLeft = array_merge( $jobsLeft, $jobBatch ); // not inserted
 293                         }
 294                 }
 295
 296                 // Insert the jobs that are not de-duplicated into the queues...
 297                 foreach ( $nuJobBatches as $jobBatch ) {
 298                         $partition = ArrayUtils::pickRandom( $partitionRing->getLiveLocationWeights() );
 299                         $queue = $this->partitionQueues[$partition];
 300                         try {
 301                                 $ok = true;
 302                                 $queue->doBatchPush( $jobBatch, $flags | self::QOS_ATOMIC );
 303                         } catch ( JobQueueError $e ) {
 304                                 $ok = false;
 305                                 MWExceptionHandler::logException( $e );
 306                         }
 307                         if ( $ok ) {
 308                                 $key = $this->getCacheKey( 'empty' );
 309                                 $this->cache->set( $key, 'false', JobQueueDB::CACHE_TTL_LONG );
 310                         } else {
 311                                 if ( !$partitionRing->ejectFromLiveRing( $partition, 5 ) ) { // blacklist
 312                                         throw new JobQueueError( "Could not insert job(s), no partitions available." );
 313                                 }
 314                                 $jobsLeft = array_merge( $jobsLeft, $jobBatch ); // not inserted
 315                         }
 316                 }
 317
 318                 return $jobsLeft;
 319         }
 320
 321         protected function doPop() {
 322                 $key = $this->getCacheKey( 'empty' );
 323
 324                 $isEmpty = $this->cache->get( $key );
 325                 if ( $isEmpty === 'true' ) {
 326                         return false;
 327                 }
 328
 329                 $partitionsTry = $this->partitionRing->getLiveLocationWeights(); // (partition => weight)
 330
 331                 $failed = 0;
 332                 while ( count( $partitionsTry ) ) {
 333                         $partition = ArrayUtils::pickRandom( $partitionsTry );
 334                         if ( $partition === false ) {
 335                                 break; // all partitions at 0 weight
 336                         }
 337
 338                         /** @var JobQueue $queue */
 339                         $queue = $this->partitionQueues[$partition];
 340                         try {
 341                                 $job = $queue->pop();
 342                         } catch ( JobQueueError $e ) {
 343                                 ++$failed;
 344                                 MWExceptionHandler::logException( $e );
 345                                 $job = false;
 346                         }
 347                         if ( $job ) {
 348                                 $job->metadata['QueuePartition'] = $partition;
 349
 350                                 return $job;
 351                         } else {
 352                                 unset( $partitionsTry[$partition] ); // blacklist partition
 353                         }
 354                 }
 355                 $this->throwErrorIfAllPartitionsDown( $failed );
 356
 357                 $this->cache->set( $key, 'true', JobQueueDB::CACHE_TTL_LONG );
 358
 359                 return false;
 360         }
 361
 362         protected function doAck( Job $job ) {
 363                 if ( !isset( $job->metadata['QueuePartition'] ) ) {
 364                         throw new MWException( "The given job has no defined partition name." );
 365                 }
 366
 367                 return $this->partitionQueues[$job->metadata['QueuePartition']]->ack( $job );
 368         }
 369
 370         protected function doIsRootJobOldDuplicate( Job $job ) {
 371                 $params = $job->getRootJobParams();
 372                 $sigature = $params['rootJobSignature'];
 373                 $partition = $this->partitionPushRing->getLiveLocation( $sigature );
 374                 try {
 375                         return $this->partitionQueues[$partition]->doIsRootJobOldDuplicate( $job );
 376                 } catch ( JobQueueError $e ) {
 377                         if ( $this->partitionPushRing->ejectFromLiveRing( $partition, 5 ) ) {
 378                                 $partition = $this->partitionPushRing->getLiveLocation( $sigature );
 379                                 return $this->partitionQueues[$partition]->doIsRootJobOldDuplicate( $job );
 380                         }
 381                 }
 382
 383                 return false;
 384         }
 385
 386         protected function doDeduplicateRootJob( Job $job ) {
 387                 $params = $job->getRootJobParams();
 388                 $sigature = $params['rootJobSignature'];
 389                 $partition = $this->partitionPushRing->getLiveLocation( $sigature );
 390                 try {
 391                         return $this->partitionQueues[$partition]->doDeduplicateRootJob( $job );
 392                 } catch ( JobQueueError $e ) {
 393                         if ( $this->partitionPushRing->ejectFromLiveRing( $partition, 5 ) ) {
 394                                 $partition = $this->partitionPushRing->getLiveLocation( $sigature );
 395                                 return $this->partitionQueues[$partition]->doDeduplicateRootJob( $job );
 396                         }
 397                 }
 398
 399                 return false;
 400         }
 401
 402         protected function doDelete() {
 403                 $failed = 0;
 404                 /** @var JobQueue $queue */
 405                 foreach ( $this->partitionQueues as $queue ) {
 406                         try {
 407                                 $queue->doDelete();
 408                         } catch ( JobQueueError $e ) {
 409                                 ++$failed;
 410                                 MWExceptionHandler::logException( $e );
 411                         }
 412                 }
 413                 $this->throwErrorIfAllPartitionsDown( $failed );
 414                 return true;
 415         }
 416
 417         protected function doWaitForBackups() {
 418                 $failed = 0;
 419                 /** @var JobQueue $queue */
 420                 foreach ( $this->partitionQueues as $queue ) {
 421                         try {
 422                                 $queue->waitForBackups();
 423                         } catch ( JobQueueError $e ) {
 424                                 ++$failed;
 425                                 MWExceptionHandler::logException( $e );
 426                         }
 427                 }
 428                 $this->throwErrorIfAllPartitionsDown( $failed );
 429         }
 430
 431         protected function doGetPeriodicTasks() {
 432                 $tasks = array();
 433                 /** @var JobQueue $queue */
 434                 foreach ( $this->partitionQueues as $partition => $queue ) {
 435                         foreach ( $queue->getPeriodicTasks() as $task => $def ) {
 436                                 $tasks["{$partition}:{$task}"] = $def;
 437                         }
 438                 }
 439
 440                 return $tasks;
 441         }
 442
 443         protected function doFlushCaches() {
 444                 static $types = array(
 445                         'empty',
 446                         'size',
 447                         'acquiredcount',
 448                         'delayedcount',
 449                         'abandonedcount'
 450                 );
 451
 452                 foreach ( $types as $type ) {
 453                         $this->cache->delete( $this->getCacheKey( $type ) );
 454                 }
 455
 456                 /** @var JobQueue $queue */
 457                 foreach ( $this->partitionQueues as $queue ) {
 458                         $queue->doFlushCaches();
 459                 }
 460         }
 461
 462         public function getAllQueuedJobs() {
 463                 $iterator = new AppendIterator();
 464
 465                 /** @var JobQueue $queue */
 466                 foreach ( $this->partitionQueues as $queue ) {
 467                         $iterator->append( $queue->getAllQueuedJobs() );
 468                 }
 469
 470                 return $iterator;
 471         }
 472
 473         public function getAllDelayedJobs() {
 474                 $iterator = new AppendIterator();
 475
 476                 /** @var JobQueue $queue */
 477                 foreach ( $this->partitionQueues as $queue ) {
 478                         $iterator->append( $queue->getAllDelayedJobs() );
 479                 }
 480
 481                 return $iterator;
 482         }
 483
 484         public function getCoalesceLocationInternal() {
 485                 return "JobQueueFederated:wiki:{$this->wiki}" .
 486                         sha1( serialize( array_keys( $this->partitionQueues ) ) );
 487         }
 488
 489         protected function doGetSiblingQueuesWithJobs( array $types ) {
 490                 $result = array();
 491
 492                 $failed = 0;
 493                 /** @var JobQueue $queue */
 494                 foreach ( $this->partitionQueues as $queue ) {
 495                         try {
 496                                 $nonEmpty = $queue->doGetSiblingQueuesWithJobs( $types );
 497                                 if ( is_array( $nonEmpty ) ) {
 498                                         $result = array_unique( array_merge( $result, $nonEmpty ) );
 499                                 } else {
 500                                         return null; // not supported on all partitions; bail
 501                                 }
 502                                 if ( count( $result ) == count( $types ) ) {
 503                                         break; // short-circuit
 504                                 }
 505                         } catch ( JobQueueError $e ) {
 506                                 ++$failed;
 507                                 MWExceptionHandler::logException( $e );
 508                         }
 509                 }
 510                 $this->throwErrorIfAllPartitionsDown( $failed );
 511
 512                 return array_values( $result );
 513         }
 514
 515         protected function doGetSiblingQueueSizes( array $types ) {
 516                 $result = array();
 517                 $failed = 0;
 518                 /** @var JobQueue $queue */
 519                 foreach ( $this->partitionQueues as $queue ) {
 520                         try {
 521                                 $sizes = $queue->doGetSiblingQueueSizes( $types );
 522                                 if ( is_array( $sizes ) ) {
 523                                         foreach ( $sizes as $type => $size ) {
 524                                                 $result[$type] = isset( $result[$type] ) ? $result[$type] + $size : $size;
 525                                         }
 526                                 } else {
 527                                         return null; // not supported on all partitions; bail
 528                                 }
 529                         } catch ( JobQueueError $e ) {
 530                                 ++$failed;
 531                                 MWExceptionHandler::logException( $e );
 532                         }
 533                 }
 534                 $this->throwErrorIfAllPartitionsDown( $failed );
 535
 536                 return $result;
 537         }
 538
 539         /**
 540          * Throw an error if no partitions available
 541          *
 542          * @param int $down The number of up partitions down
 543          * @return void
 544          * @throws JobQueueError
 545          */
 546         protected function throwErrorIfAllPartitionsDown( $down ) {
 547                 if ( $down >= count( $this->partitionQueues ) ) {
 548                         throw new JobQueueError( 'No queue partitions available.' );
 549                 }
 550         }
 551
 552         public function setTestingPrefix( $key ) {
 553                 /** @var JobQueue $queue */
 554                 foreach ( $this->partitionQueues as $queue ) {
 555                         $queue->setTestingPrefix( $key );
 556                 }
 557         }
 558
 559         /**
 560          * @param string $property
 561          * @return string
 562          */
 563         private function getCacheKey( $property ) {
 564                 list( $db, $prefix ) = wfSplitWikiID( $this->wiki );
 565
 566                 return wfForeignMemcKey( $db, $prefix, 'jobqueue', $this->type, $property );
 567         }
 568 }