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