includes/deferred/DeferredUpdates.php

   1 <?php
   2 /**
   3  * Interface and manager for deferred updates.
   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  */
  22
  23 /**
  24  * Class for managing the deferred updates
  25  *
  26  * In web request mode, deferred updates can be run at the end of the request, either before or
  27  * after the HTTP response has been sent. In either case, they run after the DB commit step. If
  28  * an update runs after the response is sent, it will not block clients. If sent before, it will
  29  * run synchronously. If such an update works via queueing, it will be more likely to complete by
  30  * the time the client makes their next request after this one.
  31  *
  32  * In CLI mode, updates are only deferred until the current wiki has no DB write transaction
  33  * active within this request.
  34  *
  35  * When updates are deferred, they use a FIFO queue (one for pre-send and one for post-send).
  36  *
  37  * @since 1.19
  38  */
  39 class DeferredUpdates {
  40         /** @var DeferrableUpdate[] Updates to be deferred until before request end */
  41         private static $preSendUpdates = [];
  42         /** @var DeferrableUpdate[] Updates to be deferred until after request end */
  43         private static $postSendUpdates = [];
  44
  45         const ALL = 0; // all updates
  46         const PRESEND = 1; // for updates that should run before flushing output buffer
  47         const POSTSEND = 2; // for updates that should run after flushing output buffer
  48
  49         /**
  50          * Add an update to the deferred list
  51          *
  52          * @param DeferrableUpdate $update Some object that implements doUpdate()
  53          * @param integer $type DeferredUpdates constant (PRESEND or POSTSEND) (since 1.27)
  54          */
  55         public static function addUpdate( DeferrableUpdate $update, $type = self::POSTSEND ) {
  56                 if ( $type === self::PRESEND ) {
  57                         self::push( self::$preSendUpdates, $update );
  58                 } else {
  59                         self::push( self::$postSendUpdates, $update );
  60                 }
  61         }
  62
  63         /**
  64          * Add a callable update. In a lot of cases, we just need a callback/closure,
  65          * defining a new DeferrableUpdate object is not necessary
  66          *
  67          * @see MWCallableUpdate::__construct()
  68          *
  69          * @param callable $callable
  70          * @param integer $type DeferredUpdates constant (PRESEND or POSTSEND) (since 1.27)
  71          * @param IDatabase|null $dbw Abort if this DB is rolled back [optional] (since 1.28)
  72          */
  73         public static function addCallableUpdate(
  74                 $callable, $type = self::POSTSEND, IDatabase $dbw = null
  75         ) {
  76                 self::addUpdate( new MWCallableUpdate( $callable, wfGetCaller(), $dbw ), $type );
  77         }
  78
  79         /**
  80          * Do any deferred updates and clear the list
  81          *
  82          * @param string $mode Use "enqueue" to use the job queue when possible [Default: "run"]
  83          * @param integer $type DeferredUpdates constant (PRESEND, POSTSEND, or ALL) (since 1.27)
  84          */
  85         public static function doUpdates( $mode = 'run', $type = self::ALL ) {
  86                 if ( $type === self::ALL || $type == self::PRESEND ) {
  87                         self::execute( self::$preSendUpdates, $mode );
  88                 }
  89
  90                 if ( $type === self::ALL || $type == self::POSTSEND ) {
  91                         self::execute( self::$postSendUpdates, $mode );
  92                 }
  93         }
  94
  95         private static function push( array &$queue, DeferrableUpdate $update ) {
  96                 global $wgCommandLineMode;
  97
  98                 if ( $update instanceof MergeableUpdate ) {
  99                         $class = get_class( $update ); // fully-qualified class
 100                         if ( isset( $queue[$class] ) ) {
 101                                 /** @var $existingUpdate MergeableUpdate */
 102                                 $existingUpdate = $queue[$class];
 103                                 $existingUpdate->merge( $update );
 104                         } else {
 105                                 $queue[$class] = $update;
 106                         }
 107                 } else {
 108                         $queue[] = $update;
 109                 }
 110
 111                 // CLI scripts may forget to periodically flush these updates,
 112                 // so try to handle that rather than OOMing and losing them entirely.
 113                 // Try to run the updates as soon as there is no current wiki transaction.
 114                 static $waitingOnTrx = false; // de-duplicate callback
 115                 if ( $wgCommandLineMode && !$waitingOnTrx ) {
 116                         $lb = wfGetLB();
 117                         $dbw = $lb->getAnyOpenConnection( $lb->getWriterIndex() );
 118                         // Do the update as soon as there is no transaction
 119                         if ( $dbw && $dbw->trxLevel() ) {
 120                                 $waitingOnTrx = true;
 121                                 $dbw->onTransactionIdle( function() use ( &$waitingOnTrx ) {
 122                                         DeferredUpdates::doUpdates();
 123                                         $waitingOnTrx = false;
 124                                 } );
 125                         } else {
 126                                 self::doUpdates();
 127                         }
 128                 }
 129         }
 130
 131         public static function execute( array &$queue, $mode ) {
 132                 $stats = \MediaWiki\MediaWikiServices::getInstance()->getStatsdDataFactory();
 133                 $method = RequestContext::getMain()->getRequest()->getMethod();
 134
 135                 $updates = $queue; // snapshot of queue
 136                 // Keep doing rounds of updates until none get enqueued
 137                 while ( count( $updates ) ) {
 138                         $queue = []; // clear the queue
 139                         /** @var DataUpdate[] $dataUpdates */
 140                         $dataUpdates = [];
 141                         /** @var DeferrableUpdate[] $otherUpdates */
 142                         $otherUpdates = [];
 143                         foreach ( $updates as $update ) {
 144                                 if ( $update instanceof DataUpdate ) {
 145                                         $dataUpdates[] = $update;
 146                                 } else {
 147                                         $otherUpdates[] = $update;
 148                                 }
 149
 150                                 $name = $update instanceof DeferrableCallback
 151                                         ? get_class( $update ) . '-' . $update->getOrigin()
 152                                         : get_class( $update );
 153                                 $stats->increment( 'deferred_updates.' . $method . '.' . $name );
 154                         }
 155
 156                         // Delegate DataUpdate execution to the DataUpdate class
 157                         try {
 158                                 DataUpdate::runUpdates( $dataUpdates, $mode );
 159                         } catch ( Exception $e ) {
 160                                 // Let the other updates occur if these had to rollback
 161                                 MWExceptionHandler::logException( $e );
 162                         }
 163                         // Execute the non-DataUpdate tasks
 164                         foreach ( $otherUpdates as $update ) {
 165                                 try {
 166                                         $update->doUpdate();
 167                                         wfGetLBFactory()->commitMasterChanges( __METHOD__ );
 168                                 } catch ( Exception $e ) {
 169                                         // We don't want exceptions thrown during deferred updates to
 170                                         // be reported to the user since the output is already sent
 171                                         if ( !$e instanceof ErrorPageError ) {
 172                                                 MWExceptionHandler::logException( $e );
 173                                         }
 174                                         // Make sure incomplete transactions are not committed and end any
 175                                         // open atomic sections so that other DB updates have a chance to run
 176                                         wfGetLBFactory()->rollbackMasterChanges( __METHOD__ );
 177                                 }
 178                         }
 179
 180                         $updates = $queue; // new snapshot of queue (check for new entries)
 181                 }
 182         }
 183
 184         /**
 185          * Clear all pending updates without performing them. Generally, you don't
 186          * want or need to call this. Unit tests need it though.
 187          */
 188         public static function clearPendingUpdates() {
 189                 self::$preSendUpdates = [];
 190                 self::$postSendUpdates = [];
 191         }
 192 }