3 * Interface and manager for deferred updates.
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.
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.
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
23 use Liuggio\StatsdClient\Factory\StatsdDataFactoryInterface
;
24 use MediaWiki\Logger\LoggerFactory
;
25 use Psr\Log\LoggerInterface
;
26 use Wikimedia\Rdbms\IDatabase
;
27 use MediaWiki\MediaWikiServices
;
28 use Wikimedia\Rdbms\LBFactory
;
29 use Wikimedia\Rdbms\ILBFactory
;
30 use Wikimedia\Rdbms\LoadBalancer
;
31 use Wikimedia\Rdbms\DBTransactionError
;
34 * Class for managing the deferred updates
36 * In web request mode, deferred updates can be run at the end of the request, either before or
37 * after the HTTP response has been sent. In either case, they run after the DB commit step. If
38 * an update runs after the response is sent, it will not block clients. If sent before, it will
39 * run synchronously. These two modes are defined via PRESEND and POSTSEND constants, the latter
40 * being the default for addUpdate() and addCallableUpdate().
42 * Updates that work through this system will be more likely to complete by the time the client
43 * makes their next request after this one than with the JobQueue system.
45 * In CLI mode, deferred updates will run:
46 * - a) During DeferredUpdates::addUpdate if no LBFactory DB handles have writes pending
47 * - b) On commit of an LBFactory DB handle if no other such handles have writes pending
48 * - c) During an LBFactory::waitForReplication call if no LBFactory DBs have writes pending
49 * - d) When the queue is large and an LBFactory DB handle commits (EnqueueableDataUpdate only)
50 * - e) At the completion of Maintenance::execute()
52 * @see Maintenance::setLBFactoryTriggers
54 * When updates are deferred, they go into one two FIFO "top-queues" (one for pre-send and one
55 * for post-send). Updates enqueued *during* doUpdate() of a "top" update go into the "sub-queue"
56 * for that update. After that method finishes, the sub-queue is run until drained. This continues
57 * for each top-queue job until the entire top queue is drained. This happens for the pre-send
58 * top-queue, and later on, the post-send top-queue, in execute().
62 class DeferredUpdates
{
63 /** @var DeferrableUpdate[] Updates to be deferred until before request end */
64 private static $preSendUpdates = [];
65 /** @var DeferrableUpdate[] Updates to be deferred until after request end */
66 private static $postSendUpdates = [];
68 const ALL
= 0; // all updates; in web requests, use only after flushing the output buffer
69 const PRESEND
= 1; // for updates that should run before flushing output buffer
70 const POSTSEND
= 2; // for updates that should run after flushing output buffer
72 const BIG_QUEUE_SIZE
= 100;
74 /** @var array|null Information about the current execute() call or null if not running */
75 private static $executeContext;
78 * Add an update to the deferred list to be run later by execute()
80 * In CLI mode, callback magic will also be used to run updates when safe
82 * @param DeferrableUpdate $update Some object that implements doUpdate()
83 * @param int $stage DeferredUpdates constant (PRESEND or POSTSEND) (since 1.27)
85 public static function addUpdate( DeferrableUpdate
$update, $stage = self
::POSTSEND
) {
86 global $wgCommandLineMode;
89 self
::$executeContext &&
90 self
::$executeContext['stage'] >= $stage &&
91 !( $update instanceof MergeableUpdate
)
93 // This is a sub-DeferredUpdate; run it right after its parent update.
94 // Also, while post-send updates are running, push any "pre-send" jobs to the
95 // active post-send queue to make sure they get run this round (or at all).
96 self
::$executeContext['subqueue'][] = $update;
101 if ( $stage === self
::PRESEND
) {
102 self
::push( self
::$preSendUpdates, $update );
104 self
::push( self
::$postSendUpdates, $update );
107 // Try to run the updates now if in CLI mode and no transaction is active.
108 // This covers scripts that don't/barely use the DB but make updates to other stores.
109 if ( $wgCommandLineMode ) {
110 self
::tryOpportunisticExecute( 'run' );
115 * Add a callable update. In a lot of cases, we just need a callback/closure,
116 * defining a new DeferrableUpdate object is not necessary
118 * @see MWCallableUpdate::__construct()
120 * @param callable $callable
121 * @param int $stage DeferredUpdates constant (PRESEND or POSTSEND) (since 1.27)
122 * @param IDatabase|IDatabase[]|null $dbw Abort if this DB is rolled back [optional] (since 1.28)
124 public static function addCallableUpdate(
125 $callable, $stage = self
::POSTSEND
, $dbw = null
127 self
::addUpdate( new MWCallableUpdate( $callable, wfGetCaller(), $dbw ), $stage );
131 * Do any deferred updates and clear the list
133 * If $stage is self::ALL then the queue of PRESEND updates will be resolved,
134 * followed by the queue of POSTSEND updates
136 * @param string $mode Use "enqueue" to use the job queue when possible [Default: "run"]
137 * @param int $stage DeferredUpdates constant (PRESEND, POSTSEND, or ALL) (since 1.27)
139 public static function doUpdates( $mode = 'run', $stage = self
::ALL
) {
140 $stageEffective = ( $stage === self
::ALL
) ? self
::POSTSEND
: $stage;
141 // For ALL mode, make sure that any PRESEND updates added along the way get run.
142 // Normally, these use the subqueue, but that isn't true for MergeableUpdate items.
144 if ( $stage === self
::ALL ||
$stage === self
::PRESEND
) {
145 self
::handleUpdateQueue( self
::$preSendUpdates, $mode, $stageEffective );
148 if ( $stage === self
::ALL ||
$stage == self
::POSTSEND
) {
149 self
::handleUpdateQueue( self
::$postSendUpdates, $mode, $stageEffective );
151 } while ( $stage === self
::ALL
&& self
::$preSendUpdates );
155 * @param DeferrableUpdate[] $queue
156 * @param DeferrableUpdate $update
158 private static function push( array &$queue, DeferrableUpdate
$update ) {
159 if ( $update instanceof MergeableUpdate
) {
160 $class = get_class( $update ); // fully-qualified class
161 if ( isset( $queue[$class] ) ) {
162 /** @var MergeableUpdate $existingUpdate */
163 $existingUpdate = $queue[$class];
164 '@phan-var MergeableUpdate $existingUpdate';
165 $existingUpdate->merge( $update );
166 // Move the update to the end to handle things like mergeable purge
167 // updates that might depend on the prior updates in the queue running
168 unset( $queue[$class] );
169 $queue[$class] = $existingUpdate;
171 $queue[$class] = $update;
179 * Immediately run or enqueue a list of updates
181 * @param DeferrableUpdate[] &$queue List of DeferrableUpdate objects
182 * @param string $mode Either "run" or "enqueue" (to use the job queue when possible)
183 * @param int $stage Class constant (PRESEND, POSTSEND) (since 1.28)
184 * @throws ErrorPageError Happens on top-level calls
185 * @throws Exception Happens on second-level calls
187 protected static function handleUpdateQueue( array &$queue, $mode, $stage ) {
188 $services = MediaWikiServices
::getInstance();
189 $stats = $services->getStatsdDataFactory();
190 $lbf = $services->getDBLoadBalancerFactory();
191 $logger = LoggerFactory
::getInstance( 'DeferredUpdates' );
192 $httpMethod = $services->getMainConfig()->get( 'CommandLineMode' )
194 : strtolower( RequestContext
::getMain()->getRequest()->getMethod() );
196 /** @var ErrorPageError $guiEx */
198 /** @var DeferrableUpdate[] $updates Snapshot of queue */
201 // Keep doing rounds of updates until none get enqueued...
203 $queue = []; // clear the queue
205 // Segregate the queue into one for DataUpdate and one for everything else
206 $dataUpdateQueue = [];
207 $genericUpdateQueue = [];
208 foreach ( $updates as $update ) {
209 if ( $update instanceof DataUpdate
) {
210 $dataUpdateQueue[] = $update;
212 $genericUpdateQueue[] = $update;
215 // Execute all DataUpdate queue followed by the DeferrableUpdate queue...
216 foreach ( [ $dataUpdateQueue, $genericUpdateQueue ] as $updateQueue ) {
217 foreach ( $updateQueue as $du ) {
218 // Enqueue the task into the job queue system instead if applicable
219 if ( $mode === 'enqueue' && $du instanceof EnqueueableDataUpdate
) {
220 self
::jobify( $du, $lbf, $logger, $stats, $httpMethod );
223 // Otherwise, execute the task and any subtasks that it spawns
224 self
::$executeContext = [ 'stage' => $stage, 'subqueue' => [] ];
226 $e = self
::run( $du, $lbf, $logger, $stats, $httpMethod );
227 $guiEx = $guiEx ?
: ( $e instanceof ErrorPageError ?
$e : null );
228 // Do the subqueue updates for $update until there are none
229 while ( self
::$executeContext['subqueue'] ) {
230 $duChild = reset( self
::$executeContext['subqueue'] );
231 $firstKey = key( self
::$executeContext['subqueue'] );
232 unset( self
::$executeContext['subqueue'][$firstKey] );
234 $e = self
::run( $duChild, $lbf, $logger, $stats, $httpMethod );
235 $guiEx = $guiEx ?
: ( $e instanceof ErrorPageError ?
$e : null );
238 // Make sure we always clean up the context.
239 // Losing updates while rewinding the stack is acceptable,
240 // losing updates that are added later is not.
241 self
::$executeContext = null;
246 $updates = $queue; // new snapshot of queue (check for new entries)
249 // Throw the first of any GUI errors as long as the context is HTTP pre-send. However,
250 // callers should check permissions *before* enqueueing updates. If the main transaction
251 // round actions succeed but some deferred updates fail due to permissions errors then
252 // there is a risk that some secondary data was not properly updated.
253 if ( $guiEx && $stage === self
::PRESEND
&& !headers_sent() ) {
259 * Run a task and catch/log any exceptions
261 * @param DeferrableUpdate $update
262 * @param LBFactory $lbFactory
263 * @param LoggerInterface $logger
264 * @param StatsdDataFactoryInterface $stats
265 * @param string $httpMethod
266 * @return Exception|Throwable|null
268 private static function run(
269 DeferrableUpdate
$update,
270 LBFactory
$lbFactory,
271 LoggerInterface
$logger,
272 StatsdDataFactoryInterface
$stats,
275 $name = get_class( $update );
276 $suffix = ( $update instanceof DeferrableCallback
) ?
"_{$update->getOrigin()}" : '';
277 $stats->increment( "deferred_updates.$httpMethod.{$name}{$suffix}" );
281 self
::attemptUpdate( $update, $lbFactory );
282 } catch ( Exception
$e ) {
283 } catch ( Throwable
$e ) {
288 "Deferred update {type} failed: {message}",
290 'type' => $name . $suffix,
291 'message' => $e->getMessage(),
292 'trace' => $e->getTraceAsString()
295 $lbFactory->rollbackMasterChanges( __METHOD__
);
296 // VW-style hack to work around T190178, so we can make sure
297 // PageMetaDataUpdater doesn't throw exceptions.
298 if ( defined( 'MW_PHPUNIT_TEST' ) ) {
307 * Push a task into the job queue system and catch/log any exceptions
309 * @param EnqueueableDataUpdate $update
310 * @param LBFactory $lbFactory
311 * @param LoggerInterface $logger
312 * @param StatsdDataFactoryInterface $stats
313 * @param string $httpMethod
315 private static function jobify(
316 EnqueueableDataUpdate
$update,
317 LBFactory
$lbFactory,
318 LoggerInterface
$logger,
319 StatsdDataFactoryInterface
$stats,
322 $stats->increment( "deferred_updates.$httpMethod." . get_class( $update ) );
326 $spec = $update->getAsJobSpecification();
327 JobQueueGroup
::singleton( $spec['domain'] ??
$spec['wiki'] )->push( $spec['job'] );
328 } catch ( Exception
$e ) {
329 } catch ( Throwable
$e ) {
334 "Job insertion of deferred update {type} failed: {message}",
336 'type' => get_class( $update ),
337 'message' => $e->getMessage(),
338 'trace' => $e->getTraceAsString()
341 $lbFactory->rollbackMasterChanges( __METHOD__
);
346 * Attempt to run an update with the appropriate transaction round state it expects
348 * DeferredUpdate classes that wrap the execution of bundles of other DeferredUpdate
349 * instances can use this method to run the updates. Any such wrapper class should
350 * always use TRX_ROUND_ABSENT itself.
352 * @param DeferrableUpdate $update
353 * @param ILBFactory $lbFactory
356 public static function attemptUpdate( DeferrableUpdate
$update, ILBFactory
$lbFactory ) {
357 $ticket = $lbFactory->getEmptyTransactionTicket( __METHOD__
);
358 if ( !$ticket ||
$lbFactory->hasTransactionRound() ) {
359 throw new DBTransactionError( null, "A database transaction round is pending." );
362 if ( $update instanceof DataUpdate
) {
363 $update->setTransactionTicket( $ticket );
366 // Designate $update::doUpdate() as the write round owner
367 $fnameTrxOwner = ( $update instanceof DeferrableCallback
)
368 ?
$update->getOrigin()
369 : get_class( $update ) . '::doUpdate';
370 // Determine whether the write round will be explicit or implicit
371 $useExplicitTrxRound = !(
372 $update instanceof TransactionRoundAwareUpdate
&&
373 $update->getTransactionRoundRequirement() == $update::TRX_ROUND_ABSENT
376 // Flush any pending changes left over from an implicit transaction round
377 if ( $useExplicitTrxRound ) {
378 $lbFactory->beginMasterChanges( $fnameTrxOwner ); // new explicit round
380 $lbFactory->commitMasterChanges( $fnameTrxOwner ); // new implicit round
382 // Run the update after any stale master view snapshots have been flushed
384 // Commit any pending changes from the explicit or implicit transaction round
385 $lbFactory->commitMasterChanges( $fnameTrxOwner );
389 * Run all deferred updates immediately if there are no DB writes active
391 * If there are many deferred updates pending, $mode is 'run', and there
392 * are still busy LBFactory database handles, then any EnqueueableDataUpdate
393 * tasks might be enqueued as jobs to be executed later.
395 * @param string $mode Use "enqueue" to use the job queue when possible
396 * @return bool Whether updates were allowed to run
399 public static function tryOpportunisticExecute( $mode = 'run' ) {
400 // execute() loop is already running
401 if ( self
::$executeContext ) {
405 // Avoiding running updates without them having outer scope
406 if ( !self
::areDatabaseTransactionsActive() ) {
407 self
::doUpdates( $mode );
411 if ( self
::pendingUpdatesCount() >= self
::BIG_QUEUE_SIZE
) {
412 // If we cannot run the updates with outer transaction context, try to
413 // at least enqueue all the updates that support queueing to job queue
414 self
::$preSendUpdates = self
::enqueueUpdates( self
::$preSendUpdates );
415 self
::$postSendUpdates = self
::enqueueUpdates( self
::$postSendUpdates );
418 return !self
::pendingUpdatesCount();
422 * Enqueue a job for each EnqueueableDataUpdate item and return the other items
424 * @param DeferrableUpdate[] $updates A list of deferred update instances
425 * @return DeferrableUpdate[] Remaining updates that do not support being queued
427 private static function enqueueUpdates( array $updates ) {
430 foreach ( $updates as $update ) {
431 if ( $update instanceof EnqueueableDataUpdate
) {
432 $spec = $update->getAsJobSpecification();
433 $domain = $spec['domain'] ??
$spec['wiki'];
434 JobQueueGroup
::singleton( $domain )->push( $spec['job'] );
436 $remaining[] = $update;
444 * @return int Number of enqueued updates
447 public static function pendingUpdatesCount() {
448 return count( self
::$preSendUpdates ) +
count( self
::$postSendUpdates );
452 * @param int $stage DeferredUpdates constant (PRESEND, POSTSEND, or ALL)
453 * @return DeferrableUpdate[]
456 public static function getPendingUpdates( $stage = self
::ALL
) {
458 if ( $stage === self
::ALL ||
$stage === self
::PRESEND
) {
459 $updates = array_merge( $updates, self
::$preSendUpdates );
461 if ( $stage === self
::ALL ||
$stage === self
::POSTSEND
) {
462 $updates = array_merge( $updates, self
::$postSendUpdates );
468 * Clear all pending updates without performing them. Generally, you don't
469 * want or need to call this. Unit tests need it though.
471 public static function clearPendingUpdates() {
472 self
::$preSendUpdates = [];
473 self
::$postSendUpdates = [];
477 * @return bool If a transaction round is active or connection is not ready for commit()
479 private static function areDatabaseTransactionsActive() {
480 $lbFactory = MediaWikiServices
::getInstance()->getDBLoadBalancerFactory();
481 if ( $lbFactory->hasTransactionRound() ||
!$lbFactory->isReadyForRoundOperations() ) {
486 $lbFactory->forEachLB( function ( LoadBalancer
$lb ) use ( &$connsBusy ) {
487 $lb->forEachOpenMasterConnection( function ( IDatabase
$conn ) use ( &$connsBusy ) {
488 if ( $conn->writesOrCallbacksPending() ||
$conn->explicitTrxActive() ) {