DeferredUpdates.php

Go to the documentation of this file.

<?php
class DeferredUpdates {
    private static $preSendUpdates = [];
    private static $postSendUpdates = [];
 
    const ALL = 0; // all updates
    const PRESEND = 1; // for updates that should run before flushing output buffer
    const POSTSEND = 2; // for updates that should run after flushing output buffer
 
    public static function addUpdate( DeferrableUpdate $update, $type = self::POSTSEND ) {
        if ( $type === self::PRESEND ) {
            self::push( self::$preSendUpdates, $update );
        } else {
            self::push( self::$postSendUpdates, $update );
        }
    }
 
    public static function addCallableUpdate( $callable, $type = self::POSTSEND ) {
        self::addUpdate( new MWCallableUpdate( $callable ), $type );
    }
 
    public static function doUpdates( $mode = 'run', $type = self::ALL ) {
        if ( $type === self::ALL || $type == self::PRESEND ) {
            self::execute( self::$preSendUpdates, $mode );
        }
 
        if ( $type === self::ALL || $type == self::POSTSEND ) {
            self::execute( self::$postSendUpdates, $mode );
        }
    }
 
    private static function push( array &$queue, DeferrableUpdate $update ) {
        global $wgCommandLineMode;
 
        if ( $update instanceof MergeableUpdate ) {
            $class = get_class( $update ); // fully-qualified class
            if ( isset( $queue[$class] ) ) {
                $existingUpdate = $queue[$class];
                $existingUpdate->merge( $update );
            } else {
                $queue[$class] = $update;
            }
        } else {
            $queue[] = $update;
        }
 
        // CLI scripts may forget to periodically flush these updates,
        // so try to handle that rather than OOMing and losing them entirely.
        // Try to run the updates as soon as there is no current wiki transaction.
        static $waitingOnTrx = false; // de-duplicate callback
        if ( $wgCommandLineMode && !$waitingOnTrx ) {
            $lb = wfGetLB();
            $dbw = $lb->getAnyOpenConnection( $lb->getWriterIndex() );
            // Do the update as soon as there is no transaction
            if ( $dbw && $dbw->trxLevel() ) {
                $waitingOnTrx = true;
                $dbw->onTransactionIdle( function() use ( &$waitingOnTrx ) {
                    DeferredUpdates::doUpdates();
                    $waitingOnTrx = false;
                } );
            } else {
                self::doUpdates();
            }
        }
    }
 
    public static function execute( array &$queue, $mode ) {
        $updates = $queue; // snapshot of queue
 
        // Keep doing rounds of updates until none get enqueued
        while ( count( $updates ) ) {
            $queue = []; // clear the queue
            $dataUpdates = [];
            $otherUpdates = [];
            foreach ( $updates as $update ) {
                if ( $update instanceof DataUpdate ) {
                    $dataUpdates[] = $update;
                } else {
                    $otherUpdates[] = $update;
                }
            }
 
            // Delegate DataUpdate execution to the DataUpdate class
            DataUpdate::runUpdates( $dataUpdates, $mode );
            // Execute the non-DataUpdate tasks
            foreach ( $otherUpdates as $update ) {
                try {
                    $update->doUpdate();
                    wfGetLBFactory()->commitMasterChanges( __METHOD__ );
                } catch ( Exception $e ) {
                    // We don't want exceptions thrown during deferred updates to
                    // be reported to the user since the output is already sent
                    if ( !$e instanceof ErrorPageError ) {
                        MWExceptionHandler::logException( $e );
                    }
                    // Make sure incomplete transactions are not committed and end any
                    // open atomic sections so that other DB updates have a chance to run
                    wfGetLBFactory()->rollbackMasterChanges( __METHOD__ );
                }
            }
 
            $updates = $queue; // new snapshot of queue (check for new entries)
        }
    }
 
    public static function clearPendingUpdates() {
        self::$preSendUpdates = [];
        self::$postSendUpdates = [];
    }
}