MediaWiki master
DeferredUpdates.php
Go to the documentation of this file.
1<?php
21namespace MediaWiki\Deferred;
22
24use LogicException;
27use Throwable;
29use Wikimedia\ScopedCallback;
30
101 public const ALL = 0;
103 public const PRESEND = 1;
105 public const POSTSEND = 2;
106
108 public const STAGES = [ self::PRESEND, self::POSTSEND ];
109
111 private const BIG_QUEUE_SIZE = 100;
112
114 private static $scopeStack;
115
119 private static $preventOpportunisticUpdates = 0;
120
124 private static function getScopeStack(): DeferredUpdatesScopeStack {
125 self::$scopeStack ??= new DeferredUpdatesScopeMediaWikiStack();
126 return self::$scopeStack;
127 }
128
133 public static function setScopeStack( DeferredUpdatesScopeStack $scopeStack ): void {
134 if ( !defined( 'MW_PHPUNIT_TEST' ) ) {
135 throw new LogicException( 'Cannot reconfigure DeferredUpdates outside tests' );
136 }
137 self::$scopeStack = $scopeStack;
138 }
139
160 public static function addUpdate( DeferrableUpdate $update, $stage = self::POSTSEND ) {
161 self::getScopeStack()->current()->addUpdate( $update, $stage );
162 self::tryOpportunisticExecute();
163 }
164
175 public static function addCallableUpdate( $callable, $stage = self::POSTSEND, $dbw = null ) {
176 self::addUpdate( new MWCallableUpdate( $callable, wfGetCaller(), $dbw ), $stage );
177 }
178
186 private static function run( DeferrableUpdate $update ): ?Throwable {
187 $logger = LoggerFactory::getInstance( 'DeferredUpdates' );
188
189 $type = get_class( $update )
190 . ( $update instanceof DeferrableCallback ? '_' . $update->getOrigin() : '' );
191 $updateId = spl_object_id( $update );
192 $logger->debug( "DeferredUpdates::run: started $type #{updateId}", [ 'updateId' => $updateId ] );
193
194 $updateException = null;
195
196 $startTime = microtime( true );
197 try {
198 self::attemptUpdate( $update );
199 } catch ( Throwable $updateException ) {
200 MWExceptionHandler::logException( $updateException );
201 $logger->error(
202 "Deferred update '{deferred_type}' failed to run.",
203 [
204 'deferred_type' => $type,
205 'exception' => $updateException,
206 ]
207 );
208 self::getScopeStack()->onRunUpdateFailed( $update );
209 } finally {
210 $walltime = microtime( true ) - $startTime;
211 $logger->debug( "DeferredUpdates::run: ended $type #{updateId}, processing time: {walltime}", [
212 'updateId' => $updateId,
213 'walltime' => $walltime,
214 ] );
215 }
216
217 // Try to push the update as a job so it can run later if possible
218 if ( $updateException && $update instanceof EnqueueableDataUpdate ) {
219 try {
220 self::getScopeStack()->queueDataUpdate( $update );
221 } catch ( Throwable $jobException ) {
222 MWExceptionHandler::logException( $jobException );
223 $logger->error(
224 "Deferred update '{deferred_type}' failed to enqueue as a job.",
225 [
226 'deferred_type' => $type,
227 'exception' => $jobException,
228 ]
229 );
230 self::getScopeStack()->onRunUpdateFailed( $update );
231 }
232 }
233
234 return $updateException;
235 }
236
255 public static function doUpdates( $stage = self::ALL ) {
257 $guiError = null;
259 $exception = null;
260
261 $scope = self::getScopeStack()->current();
262
263 // T249069: recursion is not possible once explicit transaction rounds are involved
264 $activeUpdate = $scope->getActiveUpdate();
265 if ( $activeUpdate ) {
266 $class = get_class( $activeUpdate );
267 if ( !( $activeUpdate instanceof TransactionRoundAwareUpdate ) ) {
268 throw new LogicException(
269 __METHOD__ . ": reached from $class, which is not TransactionRoundAwareUpdate"
270 );
271 }
272 if ( $activeUpdate->getTransactionRoundRequirement() !== $activeUpdate::TRX_ROUND_ABSENT ) {
273 throw new LogicException(
274 __METHOD__ . ": reached from $class, which does not specify TRX_ROUND_ABSENT"
275 );
276 }
277 }
278
279 $scope->processUpdates(
280 $stage,
281 static function ( DeferrableUpdate $update, $activeStage ) use ( &$guiError, &$exception ) {
282 $scopeStack = self::getScopeStack();
283 $childScope = $scopeStack->descend( $activeStage, $update );
284 try {
285 $e = self::run( $update );
286 $guiError = $guiError ?: ( $e instanceof ErrorPageError ? $e : null );
287 $exception = $exception ?: $e;
288 // Any addUpdate() calls between descend() and ascend() used the sub-queue.
289 // In rare cases, DeferrableUpdate::doUpdates() will process them by calling
290 // doUpdates() itself. In any case, process remaining updates in the subqueue.
291 // them, enqueueing them, or transferring them to the parent scope
292 // queues as appropriate...
293 $childScope->processUpdates(
294 $activeStage,
295 static function ( DeferrableUpdate $sub ) use ( &$guiError, &$exception ) {
296 $e = self::run( $sub );
297 $guiError = $guiError ?: ( $e instanceof ErrorPageError ? $e : null );
298 $exception = $exception ?: $e;
299 }
300 );
301 } finally {
302 $scopeStack->ascend();
303 }
304 }
305 );
306
307 // VW-style hack to work around T190178, so we can make sure
308 // PageMetaDataUpdater doesn't throw exceptions.
309 if ( $exception && defined( 'MW_PHPUNIT_TEST' ) ) {
310 throw $exception;
311 }
312
313 // Throw the first of any GUI errors as long as the context is HTTP pre-send. However,
314 // callers should check permissions *before* enqueueing updates. If the main transaction
315 // round actions succeed but some deferred updates fail due to permissions errors then
316 // there is a risk that some secondary data was not properly updated.
317 if ( $guiError && $stage === self::PRESEND && !headers_sent() ) {
318 throw $guiError;
319 }
320 }
321
358 public static function tryOpportunisticExecute(): bool {
359 // Leave execution up to the current loop if an update is already in progress
360 // or if updates are explicitly disabled
361 if ( self::getRecursiveExecutionStackDepth()
362 || self::$preventOpportunisticUpdates
363 ) {
364 return false;
365 }
366
367 if ( self::getScopeStack()->allowOpportunisticUpdates() ) {
368 self::doUpdates( self::ALL );
369 return true;
370 }
371
372 if ( self::pendingUpdatesCount() >= self::BIG_QUEUE_SIZE ) {
373 // There are a large number of pending updates and none of them can run yet.
374 // The odds of losing updates due to an error increases when executing long queues
375 // and when large amounts of time pass while tasks are queued. Mitigate this by
376 // trying to eagerly move updates to the JobQueue when possible.
377 //
378 // TODO: Do we still need this now maintenance scripts automatically call
379 // tryOpportunisticExecute from addUpdate, from every commit, and every
380 // waitForReplication call?
381 $enqueuedUpdates = [];
382 self::getScopeStack()->current()->consumeMatchingUpdates(
383 self::ALL,
384 EnqueueableDataUpdate::class,
385 static function ( EnqueueableDataUpdate $update ) use ( &$enqueuedUpdates ) {
386 self::getScopeStack()->queueDataUpdate( $update );
387 $type = get_class( $update );
388 $enqueuedUpdates[$type] ??= 0;
389 $enqueuedUpdates[$type]++;
390 }
391 );
392 if ( $enqueuedUpdates ) {
393 LoggerFactory::getInstance( 'DeferredUpdates' )->debug(
394 'Enqueued {enqueuedUpdatesCount} updates as jobs',
395 [
396 'enqueuedUpdatesCount' => array_sum( $enqueuedUpdates ),
397 'enqueuedUpdates' => implode( ', ',
398 array_map( fn ( $k, $v ) => "$k: $v", array_keys( $enqueuedUpdates ), $enqueuedUpdates ) ),
399 ]
400 );
401 }
402 }
403
404 return false;
405 }
406
413 public static function preventOpportunisticUpdates() {
414 self::$preventOpportunisticUpdates++;
415 return new ScopedCallback( static function () {
416 self::$preventOpportunisticUpdates--;
417 } );
418 }
419
429 public static function pendingUpdatesCount() {
430 return self::getScopeStack()->current()->pendingUpdatesCount();
431 }
432
445 public static function getPendingUpdates( $stage = self::ALL ) {
446 return self::getScopeStack()->current()->getPendingUpdates( $stage );
447 }
448
457 public static function clearPendingUpdates() {
458 self::getScopeStack()->current()->clearPendingUpdates();
459 }
460
467 public static function getRecursiveExecutionStackDepth() {
468 return self::getScopeStack()->getRecursiveDepth();
469 }
470
483 public static function attemptUpdate( DeferrableUpdate $update ) {
484 self::getScopeStack()->onRunUpdateStart( $update );
485
486 $update->doUpdate();
487
488 self::getScopeStack()->onRunUpdateEnd( $update );
489 }
490}
491
493class_alias( DeferredUpdates::class, 'DeferredUpdates' );
wfGetCaller( $level=2)
Get the name of the function which called this function wfGetCaller( 1 ) is the function with the wfG...
run()
Run the job.
if(!defined('MW_SETUP_CALLBACK'))
Definition WebStart.php:81
An error page which can definitely be safely rendered using the OutputPage.
Handler class for MWExceptions.
static logException(Throwable $e, $catcher=self::CAUGHT_BY_OTHER, $extraData=[])
Log a throwable to the exception log (if enabled).
This class decouples DeferredUpdates's awareness of MediaWikiServices to ease unit testing.
DeferredUpdates helper class for tracking DeferrableUpdate::doUpdate() nesting levels caused by neste...
Defer callable updates to run later in the PHP process.
static getRecursiveExecutionStackDepth()
Get the number of in-progress calls to DeferredUpdates::doUpdates()
static doUpdates( $stage=self::ALL)
Consume and execute all pending updates.
static getPendingUpdates( $stage=self::ALL)
Get a list of the pending updates for the current execution context.
static pendingUpdatesCount()
Get the number of pending updates for the current execution context.
static setScopeStack(DeferredUpdatesScopeStack $scopeStack)
static tryOpportunisticExecute()
Consume and execute pending updates now if possible, instead of waiting.
static attemptUpdate(DeferrableUpdate $update)
Attempt to run an update with the appropriate transaction round state if needed.
static addUpdate(DeferrableUpdate $update, $stage=self::POSTSEND)
Add an update to the pending update queue for execution at the appropriate time.
static addCallableUpdate( $callable, $stage=self::POSTSEND, $dbw=null)
Add an update to the pending update queue that invokes the specified callback when run.
static preventOpportunisticUpdates()
Prevent opportunistic updates until the returned ScopedCallback is consumed.
static clearPendingUpdates()
Cancel all pending updates for the current execution context.
DeferrableUpdate for closure/callable.
Create PSR-3 logger objects.
Callback wrapper that has an originating method.
Interface that deferrable updates should implement.
doUpdate()
Perform the actual work.
Deferrable update that specifies whether it must run outside of any explicit LBFactory transaction ro...
Basic database interface for live and lazy-loaded relation database handles.
Definition IDatabase.php:36