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 = [], $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 = [ 'class' => 'JobQueueDB' ];
  30                 }
  31                 $baseConfig['type'] = 'null';
  32                 $baseConfig['wiki'] = wfWikiID();
  33                 $variants = [
  34                         'queueRand' => [ 'order' => 'random', 'claimTTL' => 0 ],
  35                         'queueRandTTL' => [ 'order' => 'random', 'claimTTL' => 10 ],
  36                         'queueTimestamp' => [ 'order' => 'timestamp', 'claimTTL' => 0 ],
  37                         'queueTimestampTTL' => [ 'order' => 'timestamp', 'claimTTL' => 10 ],
  38                         'queueFifo' => [ 'order' => 'fifo', 'claimTTL' => 0 ],
  39                         'queueFifoTTL' => [ '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                         [
  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( [ $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( [ $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                                 [ $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                                 [ $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 = [];
 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                         [ $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                         [ $queue->getType(), $queue->getWiki() ],
 351                         $queue->getServerQueuesWithJobs(),
 352                         "Null queue in listing"
 353                 );
 354         }
 355
 356         public static function provider_queueLists() {
 357                 return [
 358                         [ 'queueRand', false, 'Random queue without ack()' ],
 359                         [ 'queueRandTTL', true, 'Random queue with ack()' ],
 360                         [ 'queueTimestamp', false, 'Time ordered queue without ack()' ],
 361                         [ 'queueTimestampTTL', true, 'Time ordered queue with ack()' ],
 362                         [ 'queueFifo', false, 'FIFO ordered queue without ack()' ],
 363                         [ 'queueFifoTTL', true, 'FIFO ordered queue with ack()' ]
 364                 ];
 365         }
 366
 367         public static function provider_fifoQueueLists() {
 368                 return [
 369                         [ 'queueFifo', false, 'Ordered queue without ack()' ],
 370                         [ 'queueFifoTTL', true, 'Ordered queue with ack()' ]
 371                 ];
 372         }
 373
 374         function newJob( $i = 0, $rootJob = [] ) {
 375                 return new NullJob( Title::newMainPage(),
 376                         [ 'lives' => 0, 'usleep' => 0, 'removeDuplicates' => 0, 'i' => $i ] + $rootJob );
 377         }
 378
 379         function newDedupedJob( $i = 0, $rootJob = [] ) {
 380                 return new NullJob( Title::newMainPage(),
 381                         [ 'lives' => 0, 'usleep' => 0, 'removeDuplicates' => 1, 'i' => $i ] + $rootJob );
 382         }
 383 }