Move ChronologyProtector/TransactionProfiler to Rdbms namespace
[mediawiki.git] / includes / objectcache / SqlBagOStuff.php
blob5fd3348a8cbe20767fee870e7f1ebd764a2fd129
1 <?php
2 /**
3 * Object caching using a SQL database.
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
20 * @file
21 * @ingroup Cache
24 use \MediaWiki\MediaWikiServices;
25 use \Wikimedia\WaitConditionLoop;
26 use \Wikimedia\Rdbms\TransactionProfiler;
28 /**
29 * Class to store objects in the database
31 * @ingroup Cache
33 class SqlBagOStuff extends BagOStuff {
34 /** @var array[] (server index => server config) */
35 protected $serverInfos;
36 /** @var string[] (server index => tag/host name) */
37 protected $serverTags;
38 /** @var int */
39 protected $numServers;
40 /** @var int */
41 protected $lastExpireAll = 0;
42 /** @var int */
43 protected $purgePeriod = 100;
44 /** @var int */
45 protected $shards = 1;
46 /** @var string */
47 protected $tableName = 'objectcache';
48 /** @var bool */
49 protected $replicaOnly = false;
50 /** @var int */
51 protected $syncTimeout = 3;
53 /** @var LoadBalancer|null */
54 protected $separateMainLB;
55 /** @var array */
56 protected $conns;
57 /** @var array UNIX timestamps */
58 protected $connFailureTimes = [];
59 /** @var array Exceptions */
60 protected $connFailureErrors = [];
62 /**
63 * Constructor. Parameters are:
64 * - server: A server info structure in the format required by each
65 * element in $wgDBServers.
67 * - servers: An array of server info structures describing a set of database servers
68 * to distribute keys to. If this is specified, the "server" option will be
69 * ignored. If string keys are used, then they will be used for consistent
70 * hashing *instead* of the host name (from the server config). This is useful
71 * when a cluster is replicated to another site (with different host names)
72 * but each server has a corresponding replica in the other cluster.
74 * - purgePeriod: The average number of object cache requests in between
75 * garbage collection operations, where expired entries
76 * are removed from the database. Or in other words, the
77 * reciprocal of the probability of purging on any given
78 * request. If this is set to zero, purging will never be
79 * done.
81 * - tableName: The table name to use, default is "objectcache".
83 * - shards: The number of tables to use for data storage on each server.
84 * If this is more than 1, table names will be formed in the style
85 * objectcacheNNN where NNN is the shard index, between 0 and
86 * shards-1. The number of digits will be the minimum number
87 * required to hold the largest shard index. Data will be
88 * distributed across all tables by key hash. This is for
89 * MySQL bugs 61735 and 61736.
90 * - slaveOnly: Whether to only use replica DBs and avoid triggering
91 * garbage collection logic of expired items. This only
92 * makes sense if the primary DB is used and only if get()
93 * calls will be used. This is used by ReplicatedBagOStuff.
94 * - syncTimeout: Max seconds to wait for replica DBs to catch up for WRITE_SYNC.
96 * @param array $params
98 public function __construct( $params ) {
99 parent::__construct( $params );
101 $this->attrMap[self::ATTR_EMULATION] = self::QOS_EMULATION_SQL;
102 $this->attrMap[self::ATTR_SYNCWRITES] = self::QOS_SYNCWRITES_NONE;
104 if ( isset( $params['servers'] ) ) {
105 $this->serverInfos = [];
106 $this->serverTags = [];
107 $this->numServers = count( $params['servers'] );
108 $index = 0;
109 foreach ( $params['servers'] as $tag => $info ) {
110 $this->serverInfos[$index] = $info;
111 if ( is_string( $tag ) ) {
112 $this->serverTags[$index] = $tag;
113 } else {
114 $this->serverTags[$index] = isset( $info['host'] ) ? $info['host'] : "#$index";
116 ++$index;
118 } elseif ( isset( $params['server'] ) ) {
119 $this->serverInfos = [ $params['server'] ];
120 $this->numServers = count( $this->serverInfos );
121 } else {
122 // Default to using the main wiki's database servers
123 $this->serverInfos = false;
124 $this->numServers = 1;
125 $this->attrMap[self::ATTR_SYNCWRITES] = self::QOS_SYNCWRITES_BE;
127 if ( isset( $params['purgePeriod'] ) ) {
128 $this->purgePeriod = intval( $params['purgePeriod'] );
130 if ( isset( $params['tableName'] ) ) {
131 $this->tableName = $params['tableName'];
133 if ( isset( $params['shards'] ) ) {
134 $this->shards = intval( $params['shards'] );
136 if ( isset( $params['syncTimeout'] ) ) {
137 $this->syncTimeout = $params['syncTimeout'];
139 $this->replicaOnly = !empty( $params['slaveOnly'] );
142 protected function getSeparateMainLB() {
143 global $wgDBtype;
145 if ( $wgDBtype === 'mysql' && $this->usesMainDB() ) {
146 if ( !$this->separateMainLB ) {
147 // We must keep a separate connection to MySQL in order to avoid deadlocks
148 $lbFactory = MediaWikiServices::getInstance()->getDBLoadBalancerFactory();
149 $this->separateMainLB = $lbFactory->newMainLB();
151 return $this->separateMainLB;
152 } else {
153 // However, SQLite has an opposite behavior. And PostgreSQL needs to know
154 // if we are in transaction or not (@TODO: find some PostgreSQL work-around).
155 return null;
160 * Get a connection to the specified database
162 * @param int $serverIndex
163 * @return IDatabase
164 * @throws MWException
166 protected function getDB( $serverIndex ) {
167 if ( !isset( $this->conns[$serverIndex] ) ) {
168 if ( $serverIndex >= $this->numServers ) {
169 throw new MWException( __METHOD__ . ": Invalid server index \"$serverIndex\"" );
172 # Don't keep timing out trying to connect for each call if the DB is down
173 if ( isset( $this->connFailureErrors[$serverIndex] )
174 && ( time() - $this->connFailureTimes[$serverIndex] ) < 60
176 throw $this->connFailureErrors[$serverIndex];
179 # If server connection info was given, use that
180 if ( $this->serverInfos ) {
181 $info = $this->serverInfos[$serverIndex];
182 $type = isset( $info['type'] ) ? $info['type'] : 'mysql';
183 $host = isset( $info['host'] ) ? $info['host'] : '[unknown]';
184 $this->logger->debug( __CLASS__ . ": connecting to $host" );
185 // Use a blank trx profiler to ignore expections as this is a cache
186 $info['trxProfiler'] = new TransactionProfiler();
187 $db = Database::factory( $type, $info );
188 $db->clearFlag( DBO_TRX );
189 } else {
190 $index = $this->replicaOnly ? DB_REPLICA : DB_MASTER;
191 if ( $this->getSeparateMainLB() ) {
192 $db = $this->getSeparateMainLB()->getConnection( $index );
193 $db->clearFlag( DBO_TRX ); // auto-commit mode
194 } else {
195 $db = wfGetDB( $index );
196 // Can't mess with transaction rounds (DBO_TRX) :(
199 $this->logger->debug( sprintf( "Connection %s will be used for SqlBagOStuff", $db ) );
200 $this->conns[$serverIndex] = $db;
203 return $this->conns[$serverIndex];
207 * Get the server index and table name for a given key
208 * @param string $key
209 * @return array Server index and table name
211 protected function getTableByKey( $key ) {
212 if ( $this->shards > 1 ) {
213 $hash = hexdec( substr( md5( $key ), 0, 8 ) ) & 0x7fffffff;
214 $tableIndex = $hash % $this->shards;
215 } else {
216 $tableIndex = 0;
218 if ( $this->numServers > 1 ) {
219 $sortedServers = $this->serverTags;
220 ArrayUtils::consistentHashSort( $sortedServers, $key );
221 reset( $sortedServers );
222 $serverIndex = key( $sortedServers );
223 } else {
224 $serverIndex = 0;
226 return [ $serverIndex, $this->getTableNameByShard( $tableIndex ) ];
230 * Get the table name for a given shard index
231 * @param int $index
232 * @return string
234 protected function getTableNameByShard( $index ) {
235 if ( $this->shards > 1 ) {
236 $decimals = strlen( $this->shards - 1 );
237 return $this->tableName .
238 sprintf( "%0{$decimals}d", $index );
239 } else {
240 return $this->tableName;
244 protected function doGet( $key, $flags = 0 ) {
245 $casToken = null;
247 return $this->getWithToken( $key, $casToken, $flags );
250 protected function getWithToken( $key, &$casToken, $flags = 0 ) {
251 $values = $this->getMulti( [ $key ] );
252 if ( array_key_exists( $key, $values ) ) {
253 $casToken = $values[$key];
254 return $values[$key];
256 return false;
259 public function getMulti( array $keys, $flags = 0 ) {
260 $values = []; // array of (key => value)
262 $keysByTable = [];
263 foreach ( $keys as $key ) {
264 list( $serverIndex, $tableName ) = $this->getTableByKey( $key );
265 $keysByTable[$serverIndex][$tableName][] = $key;
268 $this->garbageCollect(); // expire old entries if any
270 $dataRows = [];
271 foreach ( $keysByTable as $serverIndex => $serverKeys ) {
272 try {
273 $db = $this->getDB( $serverIndex );
274 foreach ( $serverKeys as $tableName => $tableKeys ) {
275 $res = $db->select( $tableName,
276 [ 'keyname', 'value', 'exptime' ],
277 [ 'keyname' => $tableKeys ],
278 __METHOD__,
279 // Approximate write-on-the-fly BagOStuff API via blocking.
280 // This approximation fails if a ROLLBACK happens (which is rare).
281 // We do not want to flush the TRX as that can break callers.
282 $db->trxLevel() ? [ 'LOCK IN SHARE MODE' ] : []
284 if ( $res === false ) {
285 continue;
287 foreach ( $res as $row ) {
288 $row->serverIndex = $serverIndex;
289 $row->tableName = $tableName;
290 $dataRows[$row->keyname] = $row;
293 } catch ( DBError $e ) {
294 $this->handleReadError( $e, $serverIndex );
298 foreach ( $keys as $key ) {
299 if ( isset( $dataRows[$key] ) ) { // HIT?
300 $row = $dataRows[$key];
301 $this->debug( "get: retrieved data; expiry time is " . $row->exptime );
302 $db = null;
303 try {
304 $db = $this->getDB( $row->serverIndex );
305 if ( $this->isExpired( $db, $row->exptime ) ) { // MISS
306 $this->debug( "get: key has expired" );
307 } else { // HIT
308 $values[$key] = $this->unserialize( $db->decodeBlob( $row->value ) );
310 } catch ( DBQueryError $e ) {
311 $this->handleWriteError( $e, $db, $row->serverIndex );
313 } else { // MISS
314 $this->debug( 'get: no matching rows' );
318 return $values;
321 public function setMulti( array $data, $expiry = 0 ) {
322 $keysByTable = [];
323 foreach ( $data as $key => $value ) {
324 list( $serverIndex, $tableName ) = $this->getTableByKey( $key );
325 $keysByTable[$serverIndex][$tableName][] = $key;
328 $this->garbageCollect(); // expire old entries if any
330 $result = true;
331 $exptime = (int)$expiry;
332 foreach ( $keysByTable as $serverIndex => $serverKeys ) {
333 $db = null;
334 try {
335 $db = $this->getDB( $serverIndex );
336 } catch ( DBError $e ) {
337 $this->handleWriteError( $e, $db, $serverIndex );
338 $result = false;
339 continue;
342 if ( $exptime < 0 ) {
343 $exptime = 0;
346 if ( $exptime == 0 ) {
347 $encExpiry = $this->getMaxDateTime( $db );
348 } else {
349 $exptime = $this->convertExpiry( $exptime );
350 $encExpiry = $db->timestamp( $exptime );
352 foreach ( $serverKeys as $tableName => $tableKeys ) {
353 $rows = [];
354 foreach ( $tableKeys as $key ) {
355 $rows[] = [
356 'keyname' => $key,
357 'value' => $db->encodeBlob( $this->serialize( $data[$key] ) ),
358 'exptime' => $encExpiry,
362 try {
363 $db->replace(
364 $tableName,
365 [ 'keyname' ],
366 $rows,
367 __METHOD__
369 } catch ( DBError $e ) {
370 $this->handleWriteError( $e, $db, $serverIndex );
371 $result = false;
378 return $result;
381 public function set( $key, $value, $exptime = 0, $flags = 0 ) {
382 $ok = $this->setMulti( [ $key => $value ], $exptime );
383 if ( ( $flags & self::WRITE_SYNC ) == self::WRITE_SYNC ) {
384 $ok = $this->waitForReplication() && $ok;
387 return $ok;
390 protected function cas( $casToken, $key, $value, $exptime = 0 ) {
391 list( $serverIndex, $tableName ) = $this->getTableByKey( $key );
392 $db = null;
393 try {
394 $db = $this->getDB( $serverIndex );
395 $exptime = intval( $exptime );
397 if ( $exptime < 0 ) {
398 $exptime = 0;
401 if ( $exptime == 0 ) {
402 $encExpiry = $this->getMaxDateTime( $db );
403 } else {
404 $exptime = $this->convertExpiry( $exptime );
405 $encExpiry = $db->timestamp( $exptime );
407 // (bug 24425) use a replace if the db supports it instead of
408 // delete/insert to avoid clashes with conflicting keynames
409 $db->update(
410 $tableName,
412 'keyname' => $key,
413 'value' => $db->encodeBlob( $this->serialize( $value ) ),
414 'exptime' => $encExpiry
417 'keyname' => $key,
418 'value' => $db->encodeBlob( $this->serialize( $casToken ) )
420 __METHOD__
422 } catch ( DBQueryError $e ) {
423 $this->handleWriteError( $e, $db, $serverIndex );
425 return false;
428 return (bool)$db->affectedRows();
431 public function delete( $key ) {
432 list( $serverIndex, $tableName ) = $this->getTableByKey( $key );
433 $db = null;
434 try {
435 $db = $this->getDB( $serverIndex );
436 $db->delete(
437 $tableName,
438 [ 'keyname' => $key ],
439 __METHOD__ );
440 } catch ( DBError $e ) {
441 $this->handleWriteError( $e, $db, $serverIndex );
442 return false;
445 return true;
448 public function incr( $key, $step = 1 ) {
449 list( $serverIndex, $tableName ) = $this->getTableByKey( $key );
450 $db = null;
451 try {
452 $db = $this->getDB( $serverIndex );
453 $step = intval( $step );
454 $row = $db->selectRow(
455 $tableName,
456 [ 'value', 'exptime' ],
457 [ 'keyname' => $key ],
458 __METHOD__,
459 [ 'FOR UPDATE' ] );
460 if ( $row === false ) {
461 // Missing
463 return null;
465 $db->delete( $tableName, [ 'keyname' => $key ], __METHOD__ );
466 if ( $this->isExpired( $db, $row->exptime ) ) {
467 // Expired, do not reinsert
469 return null;
472 $oldValue = intval( $this->unserialize( $db->decodeBlob( $row->value ) ) );
473 $newValue = $oldValue + $step;
474 $db->insert( $tableName,
476 'keyname' => $key,
477 'value' => $db->encodeBlob( $this->serialize( $newValue ) ),
478 'exptime' => $row->exptime
479 ], __METHOD__, 'IGNORE' );
481 if ( $db->affectedRows() == 0 ) {
482 // Race condition. See bug 28611
483 $newValue = null;
485 } catch ( DBError $e ) {
486 $this->handleWriteError( $e, $db, $serverIndex );
487 return null;
490 return $newValue;
493 public function merge( $key, callable $callback, $exptime = 0, $attempts = 10, $flags = 0 ) {
494 $ok = $this->mergeViaCas( $key, $callback, $exptime, $attempts );
495 if ( ( $flags & self::WRITE_SYNC ) == self::WRITE_SYNC ) {
496 $ok = $this->waitForReplication() && $ok;
499 return $ok;
502 public function changeTTL( $key, $expiry = 0 ) {
503 list( $serverIndex, $tableName ) = $this->getTableByKey( $key );
504 $db = null;
505 try {
506 $db = $this->getDB( $serverIndex );
507 $db->update(
508 $tableName,
509 [ 'exptime' => $db->timestamp( $this->convertExpiry( $expiry ) ) ],
510 [ 'keyname' => $key, 'exptime > ' . $db->addQuotes( $db->timestamp( time() ) ) ],
511 __METHOD__
513 if ( $db->affectedRows() == 0 ) {
514 return false;
516 } catch ( DBError $e ) {
517 $this->handleWriteError( $e, $db, $serverIndex );
518 return false;
521 return true;
525 * @param IDatabase $db
526 * @param string $exptime
527 * @return bool
529 protected function isExpired( $db, $exptime ) {
530 return $exptime != $this->getMaxDateTime( $db ) && wfTimestamp( TS_UNIX, $exptime ) < time();
534 * @param IDatabase $db
535 * @return string
537 protected function getMaxDateTime( $db ) {
538 if ( time() > 0x7fffffff ) {
539 return $db->timestamp( 1 << 62 );
540 } else {
541 return $db->timestamp( 0x7fffffff );
545 protected function garbageCollect() {
546 if ( !$this->purgePeriod || $this->replicaOnly ) {
547 // Disabled
548 return;
550 // Only purge on one in every $this->purgePeriod requests.
551 if ( $this->purgePeriod !== 1 && mt_rand( 0, $this->purgePeriod - 1 ) ) {
552 return;
554 $now = time();
555 // Avoid repeating the delete within a few seconds
556 if ( $now > ( $this->lastExpireAll + 1 ) ) {
557 $this->lastExpireAll = $now;
558 $this->expireAll();
562 public function expireAll() {
563 $this->deleteObjectsExpiringBefore( wfTimestampNow() );
567 * Delete objects from the database which expire before a certain date.
568 * @param string $timestamp
569 * @param bool|callable $progressCallback
570 * @return bool
572 public function deleteObjectsExpiringBefore( $timestamp, $progressCallback = false ) {
573 for ( $serverIndex = 0; $serverIndex < $this->numServers; $serverIndex++ ) {
574 $db = null;
575 try {
576 $db = $this->getDB( $serverIndex );
577 $dbTimestamp = $db->timestamp( $timestamp );
578 $totalSeconds = false;
579 $baseConds = [ 'exptime < ' . $db->addQuotes( $dbTimestamp ) ];
580 for ( $i = 0; $i < $this->shards; $i++ ) {
581 $maxExpTime = false;
582 while ( true ) {
583 $conds = $baseConds;
584 if ( $maxExpTime !== false ) {
585 $conds[] = 'exptime > ' . $db->addQuotes( $maxExpTime );
587 $rows = $db->select(
588 $this->getTableNameByShard( $i ),
589 [ 'keyname', 'exptime' ],
590 $conds,
591 __METHOD__,
592 [ 'LIMIT' => 100, 'ORDER BY' => 'exptime' ] );
593 if ( $rows === false || !$rows->numRows() ) {
594 break;
596 $keys = [];
597 $row = $rows->current();
598 $minExpTime = $row->exptime;
599 if ( $totalSeconds === false ) {
600 $totalSeconds = wfTimestamp( TS_UNIX, $timestamp )
601 - wfTimestamp( TS_UNIX, $minExpTime );
603 foreach ( $rows as $row ) {
604 $keys[] = $row->keyname;
605 $maxExpTime = $row->exptime;
608 $db->delete(
609 $this->getTableNameByShard( $i ),
611 'exptime >= ' . $db->addQuotes( $minExpTime ),
612 'exptime < ' . $db->addQuotes( $dbTimestamp ),
613 'keyname' => $keys
615 __METHOD__ );
617 if ( $progressCallback ) {
618 if ( intval( $totalSeconds ) === 0 ) {
619 $percent = 0;
620 } else {
621 $remainingSeconds = wfTimestamp( TS_UNIX, $timestamp )
622 - wfTimestamp( TS_UNIX, $maxExpTime );
623 if ( $remainingSeconds > $totalSeconds ) {
624 $totalSeconds = $remainingSeconds;
626 $processedSeconds = $totalSeconds - $remainingSeconds;
627 $percent = ( $i + $processedSeconds / $totalSeconds )
628 / $this->shards * 100;
630 $percent = ( $percent / $this->numServers )
631 + ( $serverIndex / $this->numServers * 100 );
632 call_user_func( $progressCallback, $percent );
636 } catch ( DBError $e ) {
637 $this->handleWriteError( $e, $db, $serverIndex );
638 return false;
641 return true;
645 * Delete content of shard tables in every server.
646 * Return true if the operation is successful, false otherwise.
647 * @return bool
649 public function deleteAll() {
650 for ( $serverIndex = 0; $serverIndex < $this->numServers; $serverIndex++ ) {
651 $db = null;
652 try {
653 $db = $this->getDB( $serverIndex );
654 for ( $i = 0; $i < $this->shards; $i++ ) {
655 $db->delete( $this->getTableNameByShard( $i ), '*', __METHOD__ );
657 } catch ( DBError $e ) {
658 $this->handleWriteError( $e, $db, $serverIndex );
659 return false;
662 return true;
666 * Serialize an object and, if possible, compress the representation.
667 * On typical message and page data, this can provide a 3X decrease
668 * in storage requirements.
670 * @param mixed $data
671 * @return string
673 protected function serialize( &$data ) {
674 $serial = serialize( $data );
676 if ( function_exists( 'gzdeflate' ) ) {
677 return gzdeflate( $serial );
678 } else {
679 return $serial;
684 * Unserialize and, if necessary, decompress an object.
685 * @param string $serial
686 * @return mixed
688 protected function unserialize( $serial ) {
689 if ( function_exists( 'gzinflate' ) ) {
690 MediaWiki\suppressWarnings();
691 $decomp = gzinflate( $serial );
692 MediaWiki\restoreWarnings();
694 if ( false !== $decomp ) {
695 $serial = $decomp;
699 $ret = unserialize( $serial );
701 return $ret;
705 * Handle a DBError which occurred during a read operation.
707 * @param DBError $exception
708 * @param int $serverIndex
710 protected function handleReadError( DBError $exception, $serverIndex ) {
711 if ( $exception instanceof DBConnectionError ) {
712 $this->markServerDown( $exception, $serverIndex );
714 $this->logger->error( "DBError: {$exception->getMessage()}" );
715 if ( $exception instanceof DBConnectionError ) {
716 $this->setLastError( BagOStuff::ERR_UNREACHABLE );
717 $this->logger->debug( __METHOD__ . ": ignoring connection error" );
718 } else {
719 $this->setLastError( BagOStuff::ERR_UNEXPECTED );
720 $this->logger->debug( __METHOD__ . ": ignoring query error" );
725 * Handle a DBQueryError which occurred during a write operation.
727 * @param DBError $exception
728 * @param IDatabase|null $db DB handle or null if connection failed
729 * @param int $serverIndex
730 * @throws Exception
732 protected function handleWriteError( DBError $exception, IDatabase $db = null, $serverIndex ) {
733 if ( !$db ) {
734 $this->markServerDown( $exception, $serverIndex );
735 } elseif ( $db->wasReadOnlyError() ) {
736 if ( $db->trxLevel() && $this->usesMainDB() ) {
737 // Errors like deadlocks and connection drops already cause rollback.
738 // For consistency, we have no choice but to throw an error and trigger
739 // complete rollback if the main DB is also being used as the cache DB.
740 throw $exception;
744 $this->logger->error( "DBError: {$exception->getMessage()}" );
745 if ( $exception instanceof DBConnectionError ) {
746 $this->setLastError( BagOStuff::ERR_UNREACHABLE );
747 $this->logger->debug( __METHOD__ . ": ignoring connection error" );
748 } else {
749 $this->setLastError( BagOStuff::ERR_UNEXPECTED );
750 $this->logger->debug( __METHOD__ . ": ignoring query error" );
755 * Mark a server down due to a DBConnectionError exception
757 * @param DBError $exception
758 * @param int $serverIndex
760 protected function markServerDown( DBError $exception, $serverIndex ) {
761 unset( $this->conns[$serverIndex] ); // bug T103435
763 if ( isset( $this->connFailureTimes[$serverIndex] ) ) {
764 if ( time() - $this->connFailureTimes[$serverIndex] >= 60 ) {
765 unset( $this->connFailureTimes[$serverIndex] );
766 unset( $this->connFailureErrors[$serverIndex] );
767 } else {
768 $this->logger->debug( __METHOD__ . ": Server #$serverIndex already down" );
769 return;
772 $now = time();
773 $this->logger->info( __METHOD__ . ": Server #$serverIndex down until " . ( $now + 60 ) );
774 $this->connFailureTimes[$serverIndex] = $now;
775 $this->connFailureErrors[$serverIndex] = $exception;
779 * Create shard tables. For use from eval.php.
781 public function createTables() {
782 for ( $serverIndex = 0; $serverIndex < $this->numServers; $serverIndex++ ) {
783 $db = $this->getDB( $serverIndex );
784 if ( $db->getType() !== 'mysql' ) {
785 throw new MWException( __METHOD__ . ' is not supported on this DB server' );
788 for ( $i = 0; $i < $this->shards; $i++ ) {
789 $db->query(
790 'CREATE TABLE ' . $db->tableName( $this->getTableNameByShard( $i ) ) .
791 ' LIKE ' . $db->tableName( 'objectcache' ),
792 __METHOD__ );
798 * @return bool Whether the main DB is used, e.g. wfGetDB( DB_MASTER )
800 protected function usesMainDB() {
801 return !$this->serverInfos;
804 protected function waitForReplication() {
805 if ( !$this->usesMainDB() ) {
806 // Custom DB server list; probably doesn't use replication
807 return true;
810 $lb = $this->getSeparateMainLB()
811 ?: MediaWikiServices::getInstance()->getDBLoadBalancer();
813 if ( $lb->getServerCount() <= 1 ) {
814 return true; // no replica DBs
817 // Main LB is used; wait for any replica DBs to catch up
818 $masterPos = $lb->getMasterPos();
820 $loop = new WaitConditionLoop(
821 function () use ( $lb, $masterPos ) {
822 return $lb->waitForAll( $masterPos, 1 );
824 $this->syncTimeout,
825 $this->busyCallbacks
828 return ( $loop->invoke() === $loop::CONDITION_REACHED );