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