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