3 use MediaWiki\MediaWikiServices
;
10 class JobQueueTest
extends MediaWikiTestCase
{
12 protected $queueRand, $queueRandTTL, $queueFifo, $queueFifoTTL;
14 function __construct( $name = null, array $data = [], $dataName = '' ) {
15 parent
::__construct( $name, $data, $dataName );
17 $this->tablesUsed
[] = 'job';
20 protected function setUp() {
21 global $wgJobTypeConf;
24 if ( $this->getCliArg( 'use-jobqueue' ) ) {
25 $name = $this->getCliArg( 'use-jobqueue' );
26 if ( !isset( $wgJobTypeConf[$name] ) ) {
27 throw new MWException( "No \$wgJobTypeConf entry for '$name'." );
29 $baseConfig = $wgJobTypeConf[$name];
31 $baseConfig = [ 'class' => JobQueueDBSingle
::class ];
33 $baseConfig['type'] = 'null';
34 $baseConfig['domain'] = WikiMap
::getCurrentWikiDbDomain()->getId();
36 'queueRand' => [ 'order' => 'random', 'claimTTL' => 0 ],
37 'queueRandTTL' => [ 'order' => 'random', 'claimTTL' => 10 ],
38 'queueTimestamp' => [ 'order' => 'timestamp', 'claimTTL' => 0 ],
39 'queueTimestampTTL' => [ 'order' => 'timestamp', 'claimTTL' => 10 ],
40 'queueFifo' => [ 'order' => 'fifo', 'claimTTL' => 0 ],
41 'queueFifoTTL' => [ 'order' => 'fifo', 'claimTTL' => 10 ],
43 foreach ( $variants as $q => $settings ) {
45 $this->$q = JobQueue
::factory( $settings +
$baseConfig );
46 } catch ( MWException
$e ) {
48 // @todo What if it was another error?
53 protected function tearDown() {
57 'queueRand', 'queueRandTTL', 'queueTimestamp', 'queueTimestampTTL',
58 'queueFifo', 'queueFifoTTL'
69 * @dataProvider provider_queueLists
70 * @covers JobQueue::getWiki
72 public function testGetWiki( $queue, $recycles, $desc ) {
73 $queue = $this->$queue;
75 $this->markTestSkipped( $desc );
77 $this->assertEquals( wfWikiID(), $queue->getWiki(), "Proper wiki ID ($desc)" );
79 WikiMap
::getCurrentWikiDbDomain()->getId(),
81 "Proper wiki ID ($desc)" );
85 * @dataProvider provider_queueLists
86 * @covers JobQueue::getType
88 public function testGetType( $queue, $recycles, $desc ) {
89 $queue = $this->$queue;
91 $this->markTestSkipped( $desc );
93 $this->assertEquals( 'null', $queue->getType(), "Proper job type ($desc)" );
97 * @dataProvider provider_queueLists
100 public function testBasicOperations( $queue, $recycles, $desc ) {
101 $queue = $this->$queue;
103 $this->markTestSkipped( $desc );
106 $this->assertTrue( $queue->isEmpty(), "Queue is empty ($desc)" );
108 $queue->flushCaches();
109 $this->assertEquals( 0, $queue->getSize(), "Queue is empty ($desc)" );
110 $this->assertEquals( 0, $queue->getAcquiredCount(), "Queue is empty ($desc)" );
112 $this->assertNull( $queue->push( $this->newJob() ), "Push worked ($desc)" );
113 $this->assertNull( $queue->batchPush( [ $this->newJob() ] ), "Push worked ($desc)" );
115 $this->assertFalse( $queue->isEmpty(), "Queue is not empty ($desc)" );
117 $queue->flushCaches();
118 $this->assertEquals( 2, $queue->getSize(), "Queue size is correct ($desc)" );
119 $this->assertEquals( 0, $queue->getAcquiredCount(), "No jobs active ($desc)" );
120 $jobs = iterator_to_array( $queue->getAllQueuedJobs() );
121 $this->assertEquals( 2, count( $jobs ), "Queue iterator size is correct ($desc)" );
123 $job1 = $queue->pop();
124 $this->assertFalse( $queue->isEmpty(), "Queue is not empty ($desc)" );
126 $queue->flushCaches();
127 $this->assertEquals( 1, $queue->getSize(), "Queue size is correct ($desc)" );
129 $queue->flushCaches();
131 $this->assertEquals( 1, $queue->getAcquiredCount(), "Active job count ($desc)" );
134 $job2 = $queue->pop();
135 $this->assertTrue( $queue->isEmpty(), "Queue is empty ($desc)" );
136 $this->assertEquals( 0, $queue->getSize(), "Queue is empty ($desc)" );
138 $queue->flushCaches();
140 $this->assertEquals( 2, $queue->getAcquiredCount(), "Active job count ($desc)" );
143 $queue->ack( $job1 );
145 $queue->flushCaches();
147 $this->assertEquals( 1, $queue->getAcquiredCount(), "Active job count ($desc)" );
150 $queue->ack( $job2 );
152 $queue->flushCaches();
153 $this->assertEquals( 0, $queue->getAcquiredCount(), "Active job count ($desc)" );
155 $this->assertNull( $queue->batchPush( [ $this->newJob(), $this->newJob() ] ),
156 "Push worked ($desc)" );
157 $this->assertFalse( $queue->isEmpty(), "Queue is not empty ($desc)" );
160 $queue->flushCaches();
161 $this->assertTrue( $queue->isEmpty(), "Queue is empty ($desc)" );
162 $this->assertEquals( 0, $queue->getSize(), "Queue is empty ($desc)" );
166 * @dataProvider provider_queueLists
169 public function testBasicDeduplication( $queue, $recycles, $desc ) {
170 $queue = $this->$queue;
172 $this->markTestSkipped( $desc );
175 $this->assertTrue( $queue->isEmpty(), "Queue is empty ($desc)" );
177 $queue->flushCaches();
178 $this->assertEquals( 0, $queue->getSize(), "Queue is empty ($desc)" );
179 $this->assertEquals( 0, $queue->getAcquiredCount(), "Queue is empty ($desc)" );
183 [ $this->newDedupedJob(), $this->newDedupedJob(), $this->newDedupedJob() ]
185 "Push worked ($desc)" );
187 $this->assertFalse( $queue->isEmpty(), "Queue is not empty ($desc)" );
189 $queue->flushCaches();
190 $this->assertEquals( 1, $queue->getSize(), "Queue size is correct ($desc)" );
191 $this->assertEquals( 0, $queue->getAcquiredCount(), "No jobs active ($desc)" );
195 [ $this->newDedupedJob(), $this->newDedupedJob(), $this->newDedupedJob() ]
197 "Push worked ($desc)"
200 $this->assertFalse( $queue->isEmpty(), "Queue is not empty ($desc)" );
202 $queue->flushCaches();
203 $this->assertEquals( 1, $queue->getSize(), "Queue size is correct ($desc)" );
204 $this->assertEquals( 0, $queue->getAcquiredCount(), "No jobs active ($desc)" );
206 $job1 = $queue->pop();
207 $this->assertTrue( $queue->isEmpty(), "Queue is empty ($desc)" );
209 $queue->flushCaches();
210 $this->assertEquals( 0, $queue->getSize(), "Queue is empty ($desc)" );
212 $this->assertEquals( 1, $queue->getAcquiredCount(), "Active job count ($desc)" );
215 $queue->ack( $job1 );
217 $queue->flushCaches();
218 $this->assertEquals( 0, $queue->getAcquiredCount(), "Active job count ($desc)" );
222 * @dataProvider provider_queueLists
225 public function testDeduplicationWhileClaimed( $queue, $recycles, $desc ) {
226 $queue = $this->$queue;
228 $this->markTestSkipped( $desc );
231 $job = $this->newDedupedJob();
232 $queue->push( $job );
234 // De-duplication does not apply to already-claimed jobs
236 $queue->push( $job );
240 // Make sure ack() of the twin did not delete the sibling data
241 $this->assertType( NullJob
::class, $j );
245 * @dataProvider provider_queueLists
248 public function testRootDeduplication( $queue, $recycles, $desc ) {
249 $queue = $this->$queue;
251 $this->markTestSkipped( $desc );
254 $this->assertTrue( $queue->isEmpty(), "Queue is empty ($desc)" );
256 $queue->flushCaches();
257 $this->assertEquals( 0, $queue->getSize(), "Queue is empty ($desc)" );
258 $this->assertEquals( 0, $queue->getAcquiredCount(), "Queue is empty ($desc)" );
260 $id = wfRandomString( 32 );
261 $root1 = Job
::newRootJobParams( "nulljobspam:$id" ); // task ID/timestamp
262 for ( $i = 0; $i < 5; ++
$i ) {
263 $this->assertNull( $queue->push( $this->newJob( 0, $root1 ) ), "Push worked ($desc)" );
265 $queue->deduplicateRootJob( $this->newJob( 0, $root1 ) );
268 # Add a second to UNIX epoch and format back to TS_MW
269 $root2_ts = strtotime( $root2['rootJobTimestamp'] );
271 $root2['rootJobTimestamp'] = wfTimestamp( TS_MW
, $root2_ts );
273 $this->assertNotEquals( $root1['rootJobTimestamp'], $root2['rootJobTimestamp'],
274 "Root job signatures have different timestamps." );
275 for ( $i = 0; $i < 5; ++
$i ) {
276 $this->assertNull( $queue->push( $this->newJob( 0, $root2 ) ), "Push worked ($desc)" );
278 $queue->deduplicateRootJob( $this->newJob( 0, $root2 ) );
280 $this->assertFalse( $queue->isEmpty(), "Queue is not empty ($desc)" );
282 $queue->flushCaches();
283 $this->assertEquals( 10, $queue->getSize(), "Queue size is correct ($desc)" );
284 $this->assertEquals( 0, $queue->getAcquiredCount(), "No jobs active ($desc)" );
289 $job = $queue->pop();
294 if ( $job instanceof DuplicateJob
) {
299 $this->assertEquals( 10, count( $jobs ), "Correct number of jobs popped ($desc)" );
300 $this->assertEquals( 5, $dupcount, "Correct number of duplicate jobs popped ($desc)" );
304 * @dataProvider provider_fifoQueueLists
307 public function testJobOrder( $queue, $recycles, $desc ) {
308 $queue = $this->$queue;
310 $this->markTestSkipped( $desc );
313 $this->assertTrue( $queue->isEmpty(), "Queue is empty ($desc)" );
315 $queue->flushCaches();
316 $this->assertEquals( 0, $queue->getSize(), "Queue is empty ($desc)" );
317 $this->assertEquals( 0, $queue->getAcquiredCount(), "Queue is empty ($desc)" );
319 for ( $i = 0; $i < 10; ++
$i ) {
320 $this->assertNull( $queue->push( $this->newJob( $i ) ), "Push worked ($desc)" );
323 for ( $i = 0; $i < 10; ++
$i ) {
324 $job = $queue->pop();
325 $this->assertTrue( $job instanceof Job
, "Jobs popped from queue ($desc)" );
326 $params = $job->getParams();
327 $this->assertEquals( $i, $params['i'], "Job popped from queue is FIFO ($desc)" );
331 $this->assertFalse( $queue->pop(), "Queue is not empty ($desc)" );
333 $queue->flushCaches();
334 $this->assertEquals( 0, $queue->getSize(), "Queue is empty ($desc)" );
335 $this->assertEquals( 0, $queue->getAcquiredCount(), "No jobs active ($desc)" );
341 public function testQueueAggregateTable() {
342 $queue = $this->queueFifo
;
343 if ( !$queue ||
!method_exists( $queue, 'getServerQueuesWithJobs' ) ) {
344 $this->markTestSkipped();
347 $this->assertNotContains(
348 [ $queue->getType(), $queue->getWiki() ],
349 $queue->getServerQueuesWithJobs(),
350 "Null queue not in listing"
353 $queue->push( $this->newJob( 0 ) );
355 $this->assertContains(
356 [ $queue->getType(), $queue->getWiki() ],
357 $queue->getServerQueuesWithJobs(),
358 "Null queue in listing"
362 public static function provider_queueLists() {
364 [ 'queueRand', false, 'Random queue without ack()' ],
365 [ 'queueRandTTL', true, 'Random queue with ack()' ],
366 [ 'queueTimestamp', false, 'Time ordered queue without ack()' ],
367 [ 'queueTimestampTTL', true, 'Time ordered queue with ack()' ],
368 [ 'queueFifo', false, 'FIFO ordered queue without ack()' ],
369 [ 'queueFifoTTL', true, 'FIFO ordered queue with ack()' ]
373 public static function provider_fifoQueueLists() {
375 [ 'queueFifo', false, 'Ordered queue without ack()' ],
376 [ 'queueFifoTTL', true, 'Ordered queue with ack()' ]
380 function newJob( $i = 0, $rootJob = [] ) {
381 return new NullJob( Title
::newMainPage(),
382 [ 'lives' => 0, 'usleep' => 0, 'removeDuplicates' => 0, 'i' => $i ] +
$rootJob );
385 function newDedupedJob( $i = 0, $rootJob = [] ) {
386 return new NullJob( Title
::newMainPage(),
387 [ 'lives' => 0, 'usleep' => 0, 'removeDuplicates' => 1, 'i' => $i ] +
$rootJob );
391 class JobQueueDBSingle
extends JobQueueDB
{
392 protected function getDB( $index ) {
393 $lb = MediaWikiServices
::getInstance()->getDBLoadBalancer();
394 // Override to not use CONN_TRX_AUTOCOMMIT so that we see the same temporary `job` table
395 return $lb->getConnection( $index, [], $this->domain
);