Merge "mw.ui: button: Update focus state"
[lhc/web/wiklou.git] / includes / jobqueue / JobQueue.php
1 <?php
2 /**
3 * Job queue base code.
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 * @defgroup JobQueue JobQueue
22 * @author Aaron Schulz
23 */
24
25 /**
26 * Class to handle enqueueing and running of background jobs
27 *
28 * @ingroup JobQueue
29 * @since 1.21
30 */
31 abstract class JobQueue {
32 /** @var string Wiki ID */
33 protected $wiki;
34
35 /** @var string Job type */
36 protected $type;
37
38 /** @var string Job priority for pop() */
39 protected $order;
40
41 /** @var int Time to live in seconds */
42 protected $claimTTL;
43
44 /** @var int Maximum number of times to try a job */
45 protected $maxTries;
46
47 /** @var bool Allow delayed jobs */
48 protected $checkDelay;
49
50 /** @var BagOStuff */
51 protected $dupCache;
52 /** @var JobQueueAggregator */
53 protected $aggr;
54
55 const QOS_ATOMIC = 1; // integer; "all-or-nothing" job insertions
56
57 const ROOTJOB_TTL = 2419200; // integer; seconds to remember root jobs (28 days)
58
59 /**
60 * @param array $params
61 * @throws MWException
62 */
63 protected function __construct( array $params ) {
64 $this->wiki = $params['wiki'];
65 $this->type = $params['type'];
66 $this->claimTTL = isset( $params['claimTTL'] ) ? $params['claimTTL'] : 0;
67 $this->maxTries = isset( $params['maxTries'] ) ? $params['maxTries'] : 3;
68 if ( isset( $params['order'] ) && $params['order'] !== 'any' ) {
69 $this->order = $params['order'];
70 } else {
71 $this->order = $this->optimalOrder();
72 }
73 if ( !in_array( $this->order, $this->supportedOrders() ) ) {
74 throw new MWException( __CLASS__ . " does not support '{$this->order}' order." );
75 }
76 $this->checkDelay = !empty( $params['checkDelay'] );
77 if ( $this->checkDelay && !$this->supportsDelayedJobs() ) {
78 throw new MWException( __CLASS__ . " does not support delayed jobs." );
79 }
80 $this->dupCache = wfGetCache( CACHE_ANYTHING );
81 $this->aggr = isset( $params['aggregator'] )
82 ? $params['aggregator']
83 : new JobQueueAggregatorNull( array() );
84 }
85
86 /**
87 * Get a job queue object of the specified type.
88 * $params includes:
89 * - class : What job class to use (determines job type)
90 * - wiki : wiki ID of the wiki the jobs are for (defaults to current wiki)
91 * - type : The name of the job types this queue handles
92 * - order : Order that pop() selects jobs, one of "fifo", "timestamp" or "random".
93 * If "fifo" is used, the queue will effectively be FIFO. Note that job
94 * completion will not appear to be exactly FIFO if there are multiple
95 * job runners since jobs can take different times to finish once popped.
96 * If "timestamp" is used, the queue will at least be loosely ordered
97 * by timestamp, allowing for some jobs to be popped off out of order.
98 * If "random" is used, pop() will pick jobs in random order.
99 * Note that it may only be weakly random (e.g. a lottery of the oldest X).
100 * If "any" is choosen, the queue will use whatever order is the fastest.
101 * This might be useful for improving concurrency for job acquisition.
102 * - claimTTL : If supported, the queue will recycle jobs that have been popped
103 * but not acknowledged as completed after this many seconds. Recycling
104 * of jobs simple means re-inserting them into the queue. Jobs can be
105 * attempted up to three times before being discarded.
106 * - checkDelay : If supported, respect Job::getReleaseTimestamp() in the push functions.
107 * This lets delayed jobs wait in a staging area until a given timestamp is
108 * reached, at which point they will enter the queue. If this is not enabled
109 * or not supported, an exception will be thrown on delayed job insertion.
110 *
111 * Queue classes should throw an exception if they do not support the options given.
112 *
113 * @param array $params
114 * @return JobQueue
115 * @throws MWException
116 */
117 final public static function factory( array $params ) {
118 $class = $params['class'];
119 if ( !class_exists( $class ) ) {
120 throw new MWException( "Invalid job queue class '$class'." );
121 }
122 $obj = new $class( $params );
123 if ( !( $obj instanceof self ) ) {
124 throw new MWException( "Class '$class' is not a " . __CLASS__ . " class." );
125 }
126
127 return $obj;
128 }
129
130 /**
131 * @return string Wiki ID
132 */
133 final public function getWiki() {
134 return $this->wiki;
135 }
136
137 /**
138 * @return string Job type that this queue handles
139 */
140 final public function getType() {
141 return $this->type;
142 }
143
144 /**
145 * @return string One of (random, timestamp, fifo, undefined)
146 */
147 final public function getOrder() {
148 return $this->order;
149 }
150
151 /**
152 * @return bool Whether delayed jobs are enabled
153 * @since 1.22
154 */
155 final public function delayedJobsEnabled() {
156 return $this->checkDelay;
157 }
158
159 /**
160 * Get the allowed queue orders for configuration validation
161 *
162 * @return array Subset of (random, timestamp, fifo, undefined)
163 */
164 abstract protected function supportedOrders();
165
166 /**
167 * Get the default queue order to use if configuration does not specify one
168 *
169 * @return string One of (random, timestamp, fifo, undefined)
170 */
171 abstract protected function optimalOrder();
172
173 /**
174 * Find out if delayed jobs are supported for configuration validation
175 *
176 * @return bool Whether delayed jobs are supported
177 */
178 protected function supportsDelayedJobs() {
179 return false; // not implemented
180 }
181
182 /**
183 * Quickly check if the queue has no available (unacquired, non-delayed) jobs.
184 * Queue classes should use caching if they are any slower without memcached.
185 *
186 * If caching is used, this might return false when there are actually no jobs.
187 * If pop() is called and returns false then it should correct the cache. Also,
188 * calling flushCaches() first prevents this. However, this affect is typically
189 * not distinguishable from the race condition between isEmpty() and pop().
190 *
191 * @return bool
192 * @throws JobQueueError
193 */
194 final public function isEmpty() {
195 $res = $this->doIsEmpty();
196
197 return $res;
198 }
199
200 /**
201 * @see JobQueue::isEmpty()
202 * @return bool
203 */
204 abstract protected function doIsEmpty();
205
206 /**
207 * Get the number of available (unacquired, non-delayed) jobs in the queue.
208 * Queue classes should use caching if they are any slower without memcached.
209 *
210 * If caching is used, this number might be out of date for a minute.
211 *
212 * @return int
213 * @throws JobQueueError
214 */
215 final public function getSize() {
216 $res = $this->doGetSize();
217
218 return $res;
219 }
220
221 /**
222 * @see JobQueue::getSize()
223 * @return int
224 */
225 abstract protected function doGetSize();
226
227 /**
228 * Get the number of acquired jobs (these are temporarily out of the queue).
229 * Queue classes should use caching if they are any slower without memcached.
230 *
231 * If caching is used, this number might be out of date for a minute.
232 *
233 * @return int
234 * @throws JobQueueError
235 */
236 final public function getAcquiredCount() {
237 $res = $this->doGetAcquiredCount();
238
239 return $res;
240 }
241
242 /**
243 * @see JobQueue::getAcquiredCount()
244 * @return int
245 */
246 abstract protected function doGetAcquiredCount();
247
248 /**
249 * Get the number of delayed jobs (these are temporarily out of the queue).
250 * Queue classes should use caching if they are any slower without memcached.
251 *
252 * If caching is used, this number might be out of date for a minute.
253 *
254 * @return int
255 * @throws JobQueueError
256 * @since 1.22
257 */
258 final public function getDelayedCount() {
259 $res = $this->doGetDelayedCount();
260
261 return $res;
262 }
263
264 /**
265 * @see JobQueue::getDelayedCount()
266 * @return int
267 */
268 protected function doGetDelayedCount() {
269 return 0; // not implemented
270 }
271
272 /**
273 * Get the number of acquired jobs that can no longer be attempted.
274 * Queue classes should use caching if they are any slower without memcached.
275 *
276 * If caching is used, this number might be out of date for a minute.
277 *
278 * @return int
279 * @throws JobQueueError
280 */
281 final public function getAbandonedCount() {
282 $res = $this->doGetAbandonedCount();
283
284 return $res;
285 }
286
287 /**
288 * @see JobQueue::getAbandonedCount()
289 * @return int
290 */
291 protected function doGetAbandonedCount() {
292 return 0; // not implemented
293 }
294
295 /**
296 * Push one or more jobs into the queue.
297 * This does not require $wgJobClasses to be set for the given job type.
298 * Outside callers should use JobQueueGroup::push() instead of this function.
299 *
300 * @param Job|array $jobs A single job or an array of Jobs
301 * @param int $flags Bitfield (supports JobQueue::QOS_ATOMIC)
302 * @return void
303 * @throws JobQueueError
304 */
305 final public function push( $jobs, $flags = 0 ) {
306 $jobs = is_array( $jobs ) ? $jobs : array( $jobs );
307 $this->batchPush( $jobs, $flags );
308 }
309
310 /**
311 * Push a batch of jobs into the queue.
312 * This does not require $wgJobClasses to be set for the given job type.
313 * Outside callers should use JobQueueGroup::push() instead of this function.
314 *
315 * @param array $jobs List of Jobs
316 * @param int $flags Bitfield (supports JobQueue::QOS_ATOMIC)
317 * @return void
318 * @throws MWException
319 */
320 final public function batchPush( array $jobs, $flags = 0 ) {
321 if ( !count( $jobs ) ) {
322 return; // nothing to do
323 }
324
325 foreach ( $jobs as $job ) {
326 if ( $job->getType() !== $this->type ) {
327 throw new MWException(
328 "Got '{$job->getType()}' job; expected a '{$this->type}' job." );
329 } elseif ( $job->getReleaseTimestamp() && !$this->checkDelay ) {
330 throw new MWException(
331 "Got delayed '{$job->getType()}' job; delays are not supported." );
332 }
333 }
334
335 $this->doBatchPush( $jobs, $flags );
336 $this->aggr->notifyQueueNonEmpty( $this->wiki, $this->type );
337 }
338
339 /**
340 * @see JobQueue::batchPush()
341 * @param array $jobs
342 * @param int $flags
343 */
344 abstract protected function doBatchPush( array $jobs, $flags );
345
346 /**
347 * Pop a job off of the queue.
348 * This requires $wgJobClasses to be set for the given job type.
349 * Outside callers should use JobQueueGroup::pop() instead of this function.
350 *
351 * @throws MWException
352 * @return Job|bool Returns false if there are no jobs
353 */
354 final public function pop() {
355 global $wgJobClasses;
356
357 if ( $this->wiki !== wfWikiID() ) {
358 throw new MWException( "Cannot pop '{$this->type}' job off foreign wiki queue." );
359 } elseif ( !isset( $wgJobClasses[$this->type] ) ) {
360 // Do not pop jobs if there is no class for the queue type
361 throw new MWException( "Unrecognized job type '{$this->type}'." );
362 }
363
364 $job = $this->doPop();
365
366 if ( !$job ) {
367 $this->aggr->notifyQueueEmpty( $this->wiki, $this->type );
368 }
369
370 // Flag this job as an old duplicate based on its "root" job...
371 try {
372 if ( $job && $this->isRootJobOldDuplicate( $job ) ) {
373 JobQueue::incrStats( 'job-pop-duplicate', $this->type, 1, $this->wiki );
374 $job = DuplicateJob::newFromJob( $job ); // convert to a no-op
375 }
376 } catch ( Exception $e ) {
377 // don't lose jobs over this
378 }
379
380 return $job;
381 }
382
383 /**
384 * @see JobQueue::pop()
385 * @return Job
386 */
387 abstract protected function doPop();
388
389 /**
390 * Acknowledge that a job was completed.
391 *
392 * This does nothing for certain queue classes or if "claimTTL" is not set.
393 * Outside callers should use JobQueueGroup::ack() instead of this function.
394 *
395 * @param Job $job
396 * @return void
397 * @throws MWException
398 */
399 final public function ack( Job $job ) {
400 if ( $job->getType() !== $this->type ) {
401 throw new MWException( "Got '{$job->getType()}' job; expected '{$this->type}'." );
402 }
403 $this->doAck( $job );
404 }
405
406 /**
407 * @see JobQueue::ack()
408 * @param Job $job
409 */
410 abstract protected function doAck( Job $job );
411
412 /**
413 * Register the "root job" of a given job into the queue for de-duplication.
414 * This should only be called right *after* all the new jobs have been inserted.
415 * This is used to turn older, duplicate, job entries into no-ops. The root job
416 * information will remain in the registry until it simply falls out of cache.
417 *
418 * This requires that $job has two special fields in the "params" array:
419 * - rootJobSignature : hash (e.g. SHA1) that identifies the task
420 * - rootJobTimestamp : TS_MW timestamp of this instance of the task
421 *
422 * A "root job" is a conceptual job that consist of potentially many smaller jobs
423 * that are actually inserted into the queue. For example, "refreshLinks" jobs are
424 * spawned when a template is edited. One can think of the task as "update links
425 * of pages that use template X" and an instance of that task as a "root job".
426 * However, what actually goes into the queue are range and leaf job subtypes.
427 * Since these jobs include things like page ID ranges and DB master positions,
428 * and can morph into smaller jobs recursively, simple duplicate detection
429 * for individual jobs being identical (like that of job_sha1) is not useful.
430 *
431 * In the case of "refreshLinks", if these jobs are still in the queue when the template
432 * is edited again, we want all of these old refreshLinks jobs for that template to become
433 * no-ops. This can greatly reduce server load, since refreshLinks jobs involves parsing.
434 * Essentially, the new batch of jobs belong to a new "root job" and the older ones to a
435 * previous "root job" for the same task of "update links of pages that use template X".
436 *
437 * This does nothing for certain queue classes.
438 *
439 * @param Job $job
440 * @throws MWException
441 * @return bool
442 */
443 final public function deduplicateRootJob( Job $job ) {
444 if ( $job->getType() !== $this->type ) {
445 throw new MWException( "Got '{$job->getType()}' job; expected '{$this->type}'." );
446 }
447 $ok = $this->doDeduplicateRootJob( $job );
448
449 return $ok;
450 }
451
452 /**
453 * @see JobQueue::deduplicateRootJob()
454 * @param Job $job
455 * @throws MWException
456 * @return bool
457 */
458 protected function doDeduplicateRootJob( Job $job ) {
459 if ( !$job->hasRootJobParams() ) {
460 throw new MWException( "Cannot register root job; missing parameters." );
461 }
462 $params = $job->getRootJobParams();
463
464 $key = $this->getRootJobCacheKey( $params['rootJobSignature'] );
465 // Callers should call batchInsert() and then this function so that if the insert
466 // fails, the de-duplication registration will be aborted. Since the insert is
467 // deferred till "transaction idle", do the same here, so that the ordering is
468 // maintained. Having only the de-duplication registration succeed would cause
469 // jobs to become no-ops without any actual jobs that made them redundant.
470 $timestamp = $this->dupCache->get( $key ); // current last timestamp of this job
471 if ( $timestamp && $timestamp >= $params['rootJobTimestamp'] ) {
472 return true; // a newer version of this root job was enqueued
473 }
474
475 // Update the timestamp of the last root job started at the location...
476 return $this->dupCache->set( $key, $params['rootJobTimestamp'], JobQueueDB::ROOTJOB_TTL );
477 }
478
479 /**
480 * Check if the "root" job of a given job has been superseded by a newer one
481 *
482 * @param Job $job
483 * @throws MWException
484 * @return bool
485 */
486 final protected function isRootJobOldDuplicate( Job $job ) {
487 if ( $job->getType() !== $this->type ) {
488 throw new MWException( "Got '{$job->getType()}' job; expected '{$this->type}'." );
489 }
490 $isDuplicate = $this->doIsRootJobOldDuplicate( $job );
491
492 return $isDuplicate;
493 }
494
495 /**
496 * @see JobQueue::isRootJobOldDuplicate()
497 * @param Job $job
498 * @return bool
499 */
500 protected function doIsRootJobOldDuplicate( Job $job ) {
501 if ( !$job->hasRootJobParams() ) {
502 return false; // job has no de-deplication info
503 }
504 $params = $job->getRootJobParams();
505
506 $key = $this->getRootJobCacheKey( $params['rootJobSignature'] );
507 // Get the last time this root job was enqueued
508 $timestamp = $this->dupCache->get( $key );
509
510 // Check if a new root job was started at the location after this one's...
511 return ( $timestamp && $timestamp > $params['rootJobTimestamp'] );
512 }
513
514 /**
515 * @param string $signature Hash identifier of the root job
516 * @return string
517 */
518 protected function getRootJobCacheKey( $signature ) {
519 list( $db, $prefix ) = wfSplitWikiID( $this->wiki );
520
521 return wfForeignMemcKey( $db, $prefix, 'jobqueue', $this->type, 'rootjob', $signature );
522 }
523
524 /**
525 * Deleted all unclaimed and delayed jobs from the queue
526 *
527 * @throws JobQueueError
528 * @since 1.22
529 * @return void
530 */
531 final public function delete() {
532 $this->doDelete();
533 }
534
535 /**
536 * @see JobQueue::delete()
537 * @throws MWException
538 */
539 protected function doDelete() {
540 throw new MWException( "This method is not implemented." );
541 }
542
543 /**
544 * Wait for any slaves or backup servers to catch up.
545 *
546 * This does nothing for certain queue classes.
547 *
548 * @return void
549 * @throws JobQueueError
550 */
551 final public function waitForBackups() {
552 $this->doWaitForBackups();
553 }
554
555 /**
556 * @see JobQueue::waitForBackups()
557 * @return void
558 */
559 protected function doWaitForBackups() {
560 }
561
562 /**
563 * Return a map of task names to task definition maps.
564 * A "task" is a fast periodic queue maintenance action.
565 * Mutually exclusive tasks must implement their own locking in the callback.
566 *
567 * Each task value is an associative array with:
568 * - name : the name of the task
569 * - callback : a PHP callable that performs the task
570 * - period : the period in seconds corresponding to the task frequency
571 *
572 * @return array
573 */
574 final public function getPeriodicTasks() {
575 $tasks = $this->doGetPeriodicTasks();
576 foreach ( $tasks as $name => &$def ) {
577 $def['name'] = $name;
578 }
579
580 return $tasks;
581 }
582
583 /**
584 * @see JobQueue::getPeriodicTasks()
585 * @return array
586 */
587 protected function doGetPeriodicTasks() {
588 return array();
589 }
590
591 /**
592 * Clear any process and persistent caches
593 *
594 * @return void
595 */
596 final public function flushCaches() {
597 $this->doFlushCaches();
598 }
599
600 /**
601 * @see JobQueue::flushCaches()
602 * @return void
603 */
604 protected function doFlushCaches() {
605 }
606
607 /**
608 * Get an iterator to traverse over all available jobs in this queue.
609 * This does not include jobs that are currently acquired or delayed.
610 * Note: results may be stale if the queue is concurrently modified.
611 *
612 * @return Iterator
613 * @throws JobQueueError
614 */
615 abstract public function getAllQueuedJobs();
616
617 /**
618 * Get an iterator to traverse over all delayed jobs in this queue.
619 * Note: results may be stale if the queue is concurrently modified.
620 *
621 * @return Iterator
622 * @throws JobQueueError
623 * @since 1.22
624 */
625 public function getAllDelayedJobs() {
626 return new ArrayIterator( array() ); // not implemented
627 }
628
629 /**
630 * Do not use this function outside of JobQueue/JobQueueGroup
631 *
632 * @return string
633 * @since 1.22
634 */
635 public function getCoalesceLocationInternal() {
636 return null;
637 }
638
639 /**
640 * Check whether each of the given queues are empty.
641 * This is used for batching checks for queues stored at the same place.
642 *
643 * @param array $types List of queues types
644 * @return array|null (list of non-empty queue types) or null if unsupported
645 * @throws MWException
646 * @since 1.22
647 */
648 final public function getSiblingQueuesWithJobs( array $types ) {
649
650 return $this->doGetSiblingQueuesWithJobs( $types );
651 }
652
653 /**
654 * @see JobQueue::getSiblingQueuesWithJobs()
655 * @param array $types List of queues types
656 * @return array|null (list of queue types) or null if unsupported
657 */
658 protected function doGetSiblingQueuesWithJobs( array $types ) {
659 return null; // not supported
660 }
661
662 /**
663 * Check the size of each of the given queues.
664 * For queues not served by the same store as this one, 0 is returned.
665 * This is used for batching checks for queues stored at the same place.
666 *
667 * @param array $types List of queues types
668 * @return array|null (job type => whether queue is empty) or null if unsupported
669 * @throws MWException
670 * @since 1.22
671 */
672 final public function getSiblingQueueSizes( array $types ) {
673
674 return $this->doGetSiblingQueueSizes( $types );
675 }
676
677 /**
678 * @see JobQueue::getSiblingQueuesSize()
679 * @param array $types List of queues types
680 * @return array|null (list of queue types) or null if unsupported
681 */
682 protected function doGetSiblingQueueSizes( array $types ) {
683 return null; // not supported
684 }
685
686 /**
687 * Call wfIncrStats() for the queue overall and for the queue type
688 *
689 * @param string $key Event type
690 * @param string $type Job type
691 * @param int $delta
692 * @param string $wiki Wiki ID (added in 1.23)
693 * @since 1.22
694 */
695 public static function incrStats( $key, $type, $delta = 1, $wiki = null ) {
696 wfIncrStats( $key, $delta );
697 wfIncrStats( "{$key}-{$type}", $delta );
698 if ( $wiki !== null ) {
699 wfIncrStats( "{$key}-{$type}-{$wiki}", $delta );
700 }
701 }
702
703 /**
704 * Namespace the queue with a key to isolate it for testing
705 *
706 * @param string $key
707 * @return void
708 * @throws MWException
709 */
710 public function setTestingPrefix( $key ) {
711 throw new MWException( "Queue namespacing not supported for this queue type." );
712 }
713 }
714
715 /**
716 * @ingroup JobQueue
717 * @since 1.22
718 */
719 class JobQueueError extends MWException {
720 }
721
722 class JobQueueConnectionError extends JobQueueError {
723 }