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