tests/phpunit/includes/jobqueue/JobQueueTest.php

   1 <?php
   2
   3 /**
   4  * @group JobQueue
   5  * @group medium
   6  * @group Database
   7  */
   8 class JobQueueTest extends MediaWikiTestCase {
   9         protected $key;
  10         protected $queueRand, $queueRandTTL, $queueFifo, $queueFifoTTL;
  11
  12         function __construct( $name = null, array $data = array(), $dataName = '' ) {
  13                 parent::__construct( $name, $data, $dataName );
  14
  15                 $this->tablesUsed[] = 'job';
  16         }
  17
  18         protected function setUp() {
  19                 global $wgJobTypeConf;
  20                 parent::setUp();
  21
  22                 if ( $this->getCliArg( 'use-jobqueue' ) ) {
  23                         $name = $this->getCliArg( 'use-jobqueue' );
  24                         if ( !isset( $wgJobTypeConf[$name] ) ) {
  25                                 throw new MWException( "No \$wgJobTypeConf entry for '$name'." );
  26                         }
  27                         $baseConfig = $wgJobTypeConf[$name];
  28                 } else {
  29                         $baseConfig = array( 'class' => 'JobQueueDB' );
  30                 }
  31                 $baseConfig['type'] = 'null';
  32                 $baseConfig['wiki'] = wfWikiID();
  33                 $variants = array(
  34                         'queueRand' => array( 'order' => 'random', 'claimTTL' => 0 ),
  35                         'queueRandTTL' => array( 'order' => 'random', 'claimTTL' => 10 ),
  36                         'queueTimestamp' => array( 'order' => 'timestamp', 'claimTTL' => 0 ),
  37                         'queueTimestampTTL' => array( 'order' => 'timestamp', 'claimTTL' => 10 ),
  38                         'queueFifo' => array( 'order' => 'fifo', 'claimTTL' => 0 ),
  39                         'queueFifoTTL' => array( 'order' => 'fifo', 'claimTTL' => 10 ),
  40                 );
  41                 foreach ( $variants as $q => $settings ) {
  42                         try {
  43                                 $this->$q = JobQueue::factory( $settings + $baseConfig );
  44                         } catch ( MWException $e ) {
  45                                 // unsupported?
  46                                 // @todo What if it was another error?
  47                         };
  48                 }
  49         }
  50
  51         protected function tearDown() {
  52                 parent::tearDown();
  53                 foreach (
  54                         array(
  55                                 'queueRand', 'queueRandTTL', 'queueTimestamp', 'queueTimestampTTL',
  56                                 'queueFifo', 'queueFifoTTL'
  57                         ) as $q
  58                 ) {
  59                         if ( $this->$q ) {
  60                                 $this->$q->delete();
  61                         }
  62                         $this->$q = null;
  63                 }
  64         }
  65
  66         /**
  67          * @dataProvider provider_queueLists
  68          * @covers JobQueue::getWiki
  69          */
  70         public function testGetWiki( $queue, $recycles, $desc ) {
  71                 $queue = $this->$queue;
  72                 if ( !$queue ) {
  73                         $this->markTestSkipped( $desc );
  74                 }
  75                 $this->assertEquals( wfWikiID(), $queue->getWiki(), "Proper wiki ID ($desc)" );
  76         }
  77
  78         /**
  79          * @dataProvider provider_queueLists
  80          * @covers JobQueue::getType
  81          */
  82         public function testGetType( $queue, $recycles, $desc ) {
  83                 $queue = $this->$queue;
  84                 if ( !$queue ) {
  85                         $this->markTestSkipped( $desc );
  86                 }
  87                 $this->assertEquals( 'null', $queue->getType(), "Proper job type ($desc)" );
  88         }
  89
  90         /**
  91          * @dataProvider provider_queueLists
  92          * @covers JobQueue
  93          */
  94         public function testBasicOperations( $queue, $recycles, $desc ) {
  95                 $queue = $this->$queue;
  96                 if ( !$queue ) {
  97                         $this->markTestSkipped( $desc );
  98                 }
  99
 100                 $this->assertTrue( $queue->isEmpty(), "Queue is empty ($desc)" );
 101
 102                 $queue->flushCaches();
 103                 $this->assertEquals( 0, $queue->getSize(), "Queue is empty ($desc)" );
 104                 $this->assertEquals( 0, $queue->getAcquiredCount(), "Queue is empty ($desc)" );
 105
 106                 $this->assertNull( $queue->push( $this->newJob() ), "Push worked ($desc)" );
 107                 $this->assertNull( $queue->batchPush( array( $this->newJob() ) ), "Push worked ($desc)" );
 108
 109                 $this->assertFalse( $queue->isEmpty(), "Queue is not empty ($desc)" );
 110
 111                 $queue->flushCaches();
 112                 $this->assertEquals( 2, $queue->getSize(), "Queue size is correct ($desc)" );
 113                 $this->assertEquals( 0, $queue->getAcquiredCount(), "No jobs active ($desc)" );
 114                 $jobs = iterator_to_array( $queue->getAllQueuedJobs() );
 115                 $this->assertEquals( 2, count( $jobs ), "Queue iterator size is correct ($desc)" );
 116
 117                 $job1 = $queue->pop();
 118                 $this->assertFalse( $queue->isEmpty(), "Queue is not empty ($desc)" );
 119
 120                 $queue->flushCaches();
 121                 $this->assertEquals( 1, $queue->getSize(), "Queue size is correct ($desc)" );
 122
 123                 $queue->flushCaches();
 124                 if ( $recycles ) {
 125                         $this->assertEquals( 1, $queue->getAcquiredCount(), "Active job count ($desc)" );
 126                 }
 127
 128                 $job2 = $queue->pop();
 129                 $this->assertTrue( $queue->isEmpty(), "Queue is empty ($desc)" );
 130                 $this->assertEquals( 0, $queue->getSize(), "Queue is empty ($desc)" );
 131
 132                 $queue->flushCaches();
 133                 if ( $recycles ) {
 134                         $this->assertEquals( 2, $queue->getAcquiredCount(), "Active job count ($desc)" );
 135                 }
 136
 137                 $queue->ack( $job1 );
 138
 139                 $queue->flushCaches();
 140                 if ( $recycles ) {
 141                         $this->assertEquals( 1, $queue->getAcquiredCount(), "Active job count ($desc)" );
 142                 }
 143
 144                 $queue->ack( $job2 );
 145
 146                 $queue->flushCaches();
 147                 $this->assertEquals( 0, $queue->getAcquiredCount(), "Active job count ($desc)" );
 148
 149                 $this->assertNull( $queue->batchPush( array( $this->newJob(), $this->newJob() ) ),
 150                         "Push worked ($desc)" );
 151                 $this->assertFalse( $queue->isEmpty(), "Queue is not empty ($desc)" );
 152
 153                 $queue->delete();
 154                 $queue->flushCaches();
 155                 $this->assertTrue( $queue->isEmpty(), "Queue is empty ($desc)" );
 156                 $this->assertEquals( 0, $queue->getSize(), "Queue is empty ($desc)" );
 157         }
 158
 159         /**
 160          * @dataProvider provider_queueLists
 161          * @covers JobQueue
 162          */
 163         public function testBasicDeduplication( $queue, $recycles, $desc ) {
 164                 $queue = $this->$queue;
 165                 if ( !$queue ) {
 166                         $this->markTestSkipped( $desc );
 167                 }
 168
 169                 $this->assertTrue( $queue->isEmpty(), "Queue is empty ($desc)" );
 170
 171                 $queue->flushCaches();
 172                 $this->assertEquals( 0, $queue->getSize(), "Queue is empty ($desc)" );
 173                 $this->assertEquals( 0, $queue->getAcquiredCount(), "Queue is empty ($desc)" );
 174
 175                 $this->assertNull(
 176                         $queue->batchPush(
 177                                 array( $this->newDedupedJob(), $this->newDedupedJob(), $this->newDedupedJob() )
 178                         ),
 179                         "Push worked ($desc)" );
 180
 181                 $this->assertFalse( $queue->isEmpty(), "Queue is not empty ($desc)" );
 182
 183                 $queue->flushCaches();
 184                 $this->assertEquals( 1, $queue->getSize(), "Queue size is correct ($desc)" );
 185                 $this->assertEquals( 0, $queue->getAcquiredCount(), "No jobs active ($desc)" );
 186
 187                 $this->assertNull(
 188                         $queue->batchPush(
 189                                 array( $this->newDedupedJob(), $this->newDedupedJob(), $this->newDedupedJob() )
 190                         ),
 191                         "Push worked ($desc)"
 192                 );
 193
 194                 $this->assertFalse( $queue->isEmpty(), "Queue is not empty ($desc)" );
 195
 196                 $queue->flushCaches();
 197                 $this->assertEquals( 1, $queue->getSize(), "Queue size is correct ($desc)" );
 198                 $this->assertEquals( 0, $queue->getAcquiredCount(), "No jobs active ($desc)" );
 199
 200                 $job1 = $queue->pop();
 201                 $this->assertTrue( $queue->isEmpty(), "Queue is empty ($desc)" );
 202
 203                 $queue->flushCaches();
 204                 $this->assertEquals( 0, $queue->getSize(), "Queue is empty ($desc)" );
 205                 if ( $recycles ) {
 206                         $this->assertEquals( 1, $queue->getAcquiredCount(), "Active job count ($desc)" );
 207                 }
 208
 209                 $queue->ack( $job1 );
 210
 211                 $queue->flushCaches();
 212                 $this->assertEquals( 0, $queue->getAcquiredCount(), "Active job count ($desc)" );
 213         }
 214
 215         /**
 216          * @dataProvider provider_queueLists
 217          * @covers JobQueue
 218          */
 219         public function testDeduplicationWhileClaimed( $queue, $recycles, $desc ) {
 220                 $queue = $this->$queue;
 221                 if ( !$queue ) {
 222                         $this->markTestSkipped( $desc );
 223                 }
 224
 225                 $job = $this->newDedupedJob();
 226                 $queue->push( $job );
 227
 228                 // De-duplication does not apply to already-claimed jobs
 229                 $j = $queue->pop();
 230                 $queue->push( $job );
 231                 $queue->ack( $j );
 232
 233                 $j = $queue->pop();
 234                 // Make sure ack() of the twin did not delete the sibling data
 235                 $this->assertType( 'NullJob', $j );
 236         }
 237
 238         /**
 239          * @dataProvider provider_queueLists
 240          * @covers JobQueue
 241          */
 242         public function testRootDeduplication( $queue, $recycles, $desc ) {
 243                 $queue = $this->$queue;
 244                 if ( !$queue ) {
 245                         $this->markTestSkipped( $desc );
 246                 }
 247
 248                 $this->assertTrue( $queue->isEmpty(), "Queue is empty ($desc)" );
 249
 250                 $queue->flushCaches();
 251                 $this->assertEquals( 0, $queue->getSize(), "Queue is empty ($desc)" );
 252                 $this->assertEquals( 0, $queue->getAcquiredCount(), "Queue is empty ($desc)" );
 253
 254                 $id = wfRandomString( 32 );
 255                 $root1 = Job::newRootJobParams( "nulljobspam:$id" ); // task ID/timestamp
 256                 for ( $i = 0; $i < 5; ++$i ) {
 257                         $this->assertNull( $queue->push( $this->newJob( 0, $root1 ) ), "Push worked ($desc)" );
 258                 }
 259                 $queue->deduplicateRootJob( $this->newJob( 0, $root1 ) );
 260
 261                 $root2 = $root1;
 262                 # Add a second to UNIX epoch and format back to TS_MW
 263                 $root2_ts = strtotime( $root2['rootJobTimestamp'] );
 264                 $root2_ts++;
 265                 $root2['rootJobTimestamp'] = wfTimestamp( TS_MW, $root2_ts );
 266
 267                 $this->assertNotEquals( $root1['rootJobTimestamp'], $root2['rootJobTimestamp'],
 268                         "Root job signatures have different timestamps." );
 269                 for ( $i = 0; $i < 5; ++$i ) {
 270                         $this->assertNull( $queue->push( $this->newJob( 0, $root2 ) ), "Push worked ($desc)" );
 271                 }
 272                 $queue->deduplicateRootJob( $this->newJob( 0, $root2 ) );
 273
 274                 $this->assertFalse( $queue->isEmpty(), "Queue is not empty ($desc)" );
 275
 276                 $queue->flushCaches();
 277                 $this->assertEquals( 10, $queue->getSize(), "Queue size is correct ($desc)" );
 278                 $this->assertEquals( 0, $queue->getAcquiredCount(), "No jobs active ($desc)" );
 279
 280                 $dupcount = 0;
 281                 $jobs = array();
 282                 do {
 283                         $job = $queue->pop();
 284                         if ( $job ) {
 285                                 $jobs[] = $job;
 286                                 $queue->ack( $job );
 287                         }
 288                         if ( $job instanceof DuplicateJob ) {
 289                                 ++$dupcount;
 290                         }
 291                 } while ( $job );
 292
 293                 $this->assertEquals( 10, count( $jobs ), "Correct number of jobs popped ($desc)" );
 294                 $this->assertEquals( 5, $dupcount, "Correct number of duplicate jobs popped ($desc)" );
 295         }
 296
 297         /**
 298          * @dataProvider provider_fifoQueueLists
 299          * @covers JobQueue
 300          */
 301         public function testJobOrder( $queue, $recycles, $desc ) {
 302                 $queue = $this->$queue;
 303                 if ( !$queue ) {
 304                         $this->markTestSkipped( $desc );
 305                 }
 306
 307                 $this->assertTrue( $queue->isEmpty(), "Queue is empty ($desc)" );
 308
 309                 $queue->flushCaches();
 310                 $this->assertEquals( 0, $queue->getSize(), "Queue is empty ($desc)" );
 311                 $this->assertEquals( 0, $queue->getAcquiredCount(), "Queue is empty ($desc)" );
 312
 313                 for ( $i = 0; $i < 10; ++$i ) {
 314                         $this->assertNull( $queue->push( $this->newJob( $i ) ), "Push worked ($desc)" );
 315                 }
 316
 317                 for ( $i = 0; $i < 10; ++$i ) {
 318                         $job = $queue->pop();
 319                         $this->assertTrue( $job instanceof Job, "Jobs popped from queue ($desc)" );
 320                         $params = $job->getParams();
 321                         $this->assertEquals( $i, $params['i'], "Job popped from queue is FIFO ($desc)" );
 322                         $queue->ack( $job );
 323                 }
 324
 325                 $this->assertFalse( $queue->pop(), "Queue is not empty ($desc)" );
 326
 327                 $queue->flushCaches();
 328                 $this->assertEquals( 0, $queue->getSize(), "Queue is empty ($desc)" );
 329                 $this->assertEquals( 0, $queue->getAcquiredCount(), "No jobs active ($desc)" );
 330         }
 331
 332         /**
 333          * @covers JobQueue
 334          */
 335         public function testQueueAggregateTable() {
 336                 $queue = $this->queueFifo;
 337                 if ( !$queue || !method_exists( $queue, 'getServerQueuesWithJobs' ) ) {
 338                         $this->markTestSkipped();
 339                 }
 340
 341                 $this->assertNotContains(
 342                         array( $queue->getType(), $queue->getWiki() ),
 343                         $queue->getServerQueuesWithJobs(),
 344                         "Null queue not in listing"
 345                 );
 346
 347                 $queue->push( $this->newJob( 0 ) );
 348
 349                 $this->assertContains(
 350                         array( $queue->getType(), $queue->getWiki() ),
 351                         $queue->getServerQueuesWithJobs(),
 352                         "Null queue in listing"
 353                 );
 354         }
 355
 356         public static function provider_queueLists() {
 357                 return array(
 358                         array( 'queueRand', false, 'Random queue without ack()' ),
 359                         array( 'queueRandTTL', true, 'Random queue with ack()' ),
 360                         array( 'queueTimestamp', false, 'Time ordered queue without ack()' ),
 361                         array( 'queueTimestampTTL', true, 'Time ordered queue with ack()' ),
 362                         array( 'queueFifo', false, 'FIFO ordered queue without ack()' ),
 363                         array( 'queueFifoTTL', true, 'FIFO ordered queue with ack()' )
 364                 );
 365         }
 366
 367         public static function provider_fifoQueueLists() {
 368                 return array(
 369                         array( 'queueFifo', false, 'Ordered queue without ack()' ),
 370                         array( 'queueFifoTTL', true, 'Ordered queue with ack()' )
 371                 );
 372         }
 373
 374         function newJob( $i = 0, $rootJob = array() ) {
 375                 return new NullJob( Title::newMainPage(),
 376                         array( 'lives' => 0, 'usleep' => 0, 'removeDuplicates' => 0, 'i' => $i ) + $rootJob );
 377         }
 378
 379         function newDedupedJob( $i = 0, $rootJob = array() ) {
 380                 return new NullJob( Title::newMainPage(),
 381                         array( 'lives' => 0, 'usleep' => 0, 'removeDuplicates' => 1, 'i' => $i ) + $rootJob );
 382         }
 383 }