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, wfGetCaller() ), $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 ) {
        $stats = \MediaWiki\MediaWikiServices::getInstance()->getStatsdDataFactory();
        $method = RequestContext::getMain()->getRequest()->getMethod();

        $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;
                }

                $name = $update instanceof DeferrableCallback
                    ? get_class( $update ) . '-' . $update->getOrigin()
                    : get_class( $update );
                $stats->increment( 'deferred_updates.' . $method . '.' . $name );
            }

            // Delegate DataUpdate execution to the DataUpdate class
            try {
                DataUpdate::runUpdates( $dataUpdates, $mode );
            } catch ( Exception $e ) {
                // Let the other updates occur if these had to rollback
                MWExceptionHandler::logException( $e );
            }
            // 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 = [];
    }
}