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
22 use MediaWiki\MediaWikiServices
;
25 * Class for managing the deferred updates
27 * In web request mode, deferred updates can be run at the end of the request, either before or
28 * after the HTTP response has been sent. In either case, they run after the DB commit step. If
29 * an update runs after the response is sent, it will not block clients. If sent before, it will
30 * run synchronously. If such an update works via queueing, it will be more likely to complete by
31 * the time the client makes their next request after this one.
33 * In CLI mode, updates are only deferred until the current wiki has no DB write transaction
34 * active within this request.
36 * When updates are deferred, they use a FIFO queue (one for pre-send and one for post-send).
40 class DeferredUpdates
{
41 /** @var DeferrableUpdate[] Updates to be deferred until before request end */
42 private static $preSendUpdates = [];
43 /** @var DeferrableUpdate[] Updates to be deferred until after request end */
44 private static $postSendUpdates = [];
46 const ALL
= 0; // all updates
47 const PRESEND
= 1; // for updates that should run before flushing output buffer
48 const POSTSEND
= 2; // for updates that should run after flushing output buffer
50 const BIG_QUEUE_SIZE
= 100;
53 * Add an update to the deferred list to be run later by execute()
55 * In CLI mode, callback magic will also be used to run updates when safe
57 * @param DeferrableUpdate $update Some object that implements doUpdate()
58 * @param integer $type DeferredUpdates constant (PRESEND or POSTSEND) (since 1.27)
60 public static function addUpdate( DeferrableUpdate
$update, $type = self
::POSTSEND
) {
61 global $wgCommandLineMode;
63 if ( $type === self
::PRESEND
) {
64 self
::push( self
::$preSendUpdates, $update );
66 self
::push( self
::$postSendUpdates, $update );
69 // Try to run the updates now if in CLI mode and no transaction is active.
70 // This covers scripts that don't/barely use the DB but make updates to other stores.
71 if ( $wgCommandLineMode ) {
72 self
::tryOpportunisticExecute( 'run' );
77 * Add a callable update. In a lot of cases, we just need a callback/closure,
78 * defining a new DeferrableUpdate object is not necessary
80 * @see MWCallableUpdate::__construct()
82 * @param callable $callable
83 * @param integer $type DeferredUpdates constant (PRESEND or POSTSEND) (since 1.27)
84 * @param IDatabase|null $dbw Abort if this DB is rolled back [optional] (since 1.28)
86 public static function addCallableUpdate(
87 $callable, $type = self
::POSTSEND
, IDatabase
$dbw = null
89 self
::addUpdate( new MWCallableUpdate( $callable, wfGetCaller(), $dbw ), $type );
93 * Do any deferred updates and clear the list
95 * @param string $mode Use "enqueue" to use the job queue when possible [Default: "run"]
96 * @param integer $type DeferredUpdates constant (PRESEND, POSTSEND, or ALL) (since 1.27)
98 public static function doUpdates( $mode = 'run', $type = self
::ALL
) {
99 if ( $type === self
::ALL ||
$type == self
::PRESEND
) {
100 self
::execute( self
::$preSendUpdates, $mode );
103 if ( $type === self
::ALL ||
$type == self
::POSTSEND
) {
104 self
::execute( self
::$postSendUpdates, $mode );
108 private static function push( array &$queue, DeferrableUpdate
$update ) {
109 if ( $update instanceof MergeableUpdate
) {
110 $class = get_class( $update ); // fully-qualified class
111 if ( isset( $queue[$class] ) ) {
112 /** @var $existingUpdate MergeableUpdate */
113 $existingUpdate = $queue[$class];
114 $existingUpdate->merge( $update );
116 $queue[$class] = $update;
123 public static function execute( array &$queue, $mode ) {
124 $stats = \MediaWiki\MediaWikiServices
::getInstance()->getStatsdDataFactory();
125 $method = RequestContext
::getMain()->getRequest()->getMethod();
127 $updates = $queue; // snapshot of queue
128 // Keep doing rounds of updates until none get enqueued
129 while ( count( $updates ) ) {
130 $queue = []; // clear the queue
131 /** @var DataUpdate[] $dataUpdates */
133 /** @var DeferrableUpdate[] $otherUpdates */
135 foreach ( $updates as $update ) {
136 if ( $update instanceof DataUpdate
) {
137 $dataUpdates[] = $update;
139 $otherUpdates[] = $update;
142 $name = $update instanceof DeferrableCallback
143 ?
get_class( $update ) . '-' . $update->getOrigin()
144 : get_class( $update );
145 $stats->increment( 'deferred_updates.' . $method . '.' . $name );
148 // Delegate DataUpdate execution to the DataUpdate class
150 DataUpdate
::runUpdates( $dataUpdates, $mode );
151 } catch ( Exception
$e ) {
152 // Let the other updates occur if these had to rollback
153 MWExceptionHandler
::logException( $e );
155 // Execute the non-DataUpdate tasks
156 foreach ( $otherUpdates as $update ) {
159 wfGetLBFactory()->commitMasterChanges( __METHOD__
);
160 } catch ( Exception
$e ) {
161 // We don't want exceptions thrown during deferred updates to
162 // be reported to the user since the output is already sent
163 if ( !$e instanceof ErrorPageError
) {
164 MWExceptionHandler
::logException( $e );
166 // Make sure incomplete transactions are not committed and end any
167 // open atomic sections so that other DB updates have a chance to run
168 wfGetLBFactory()->rollbackMasterChanges( __METHOD__
);
172 $updates = $queue; // new snapshot of queue (check for new entries)
177 * Run all deferred updates immediately if there are no DB writes active
179 * If $mode is 'run' but there are busy databates, EnqueueableDataUpdate
180 * tasks will be enqueued anyway for the sake of progress.
182 * @param string $mode Use "enqueue" to use the job queue when possible
183 * @return bool Whether updates were allowed to run
186 public static function tryOpportunisticExecute( $mode = 'run' ) {
187 if ( !self
::getBusyDbConnections() ) {
188 self
::doUpdates( $mode );
192 if ( self
::pendingUpdatesCount() >= self
::BIG_QUEUE_SIZE
) {
193 // If we cannot run the updates with outer transaction context, try to
194 // at least enqueue all the updates that support queueing to job queue
195 self
::$preSendUpdates = self
::enqueueUpdates( self
::$preSendUpdates );
196 self
::$postSendUpdates = self
::enqueueUpdates( self
::$postSendUpdates );
199 return !self
::pendingUpdatesCount();
203 * Enqueue a job for each EnqueueableDataUpdate item and return the other items
205 * @param DeferrableUpdate[] $updates A list of deferred update instances
206 * @return DeferrableUpdate[] Remaining updates that do not support being queued
208 private static function enqueueUpdates( array $updates ) {
211 foreach ( $updates as $update ) {
212 if ( $update instanceof EnqueueableDataUpdate
) {
213 $spec = $update->getAsJobSpecification();
214 JobQueueGroup
::singleton( $spec['wiki'] )->push( $spec['job'] );
216 $remaining[] = $update;
224 * @return integer Number of enqueued updates
227 public static function pendingUpdatesCount() {
228 return count( self
::$preSendUpdates ) +
count( self
::$postSendUpdates );
232 * Clear all pending updates without performing them. Generally, you don't
233 * want or need to call this. Unit tests need it though.
235 public static function clearPendingUpdates() {
236 self
::$preSendUpdates = [];
237 self
::$postSendUpdates = [];
241 * Set the rollback/commit watcher on a DB to trigger update runs when safe
243 * @TODO: use this to replace DB logic in push()
244 * @param LoadBalancer $lb
247 public static function installDBListener( LoadBalancer
$lb ) {
248 static $triggers = [ IDatabase
::TRIGGER_COMMIT
, IDatabase
::TRIGGER_ROLLBACK
];
249 // Hook into active master connections to find a moment where no writes are pending
250 $lb->setTransactionListener(
252 function ( $trigger, IDatabase
$conn ) use ( $triggers ) {
253 global $wgCommandLineMode;
255 if ( $wgCommandLineMode && in_array( $trigger, $triggers ) ) {
256 DeferredUpdates
::tryOpportunisticExecute();
263 * @return IDatabase[] Connection where commit() cannot be called yet
265 private static function getBusyDbConnections() {
268 $lbFactory = MediaWikiServices
::getInstance()->getDBLoadBalancerFactory();
269 $lbFactory->forEachLB( function ( LoadBalancer
$lb ) use ( &$connsBusy ) {
270 $lb->forEachOpenMasterConnection( function ( IDatabase
$conn ) use ( &$connsBusy ) {
271 if ( $conn->writesOrCallbacksPending() ||
$conn->explicitTrxActive() ) {
272 $connsBusy[] = $conn;