30 use \Wikimedia\WaitConditionLoop;
31 use \Wikimedia\Rdbms\TransactionProfiler;
105 parent::__construct(
$params );
110 if ( isset(
$params[
'servers'] ) ) {
111 $this->serverInfos = [];
112 $this->serverTags = [];
116 $this->serverInfos[$index] = $info;
117 if ( is_string(
$tag ) ) {
118 $this->serverTags[$index] =
$tag;
120 $this->serverTags[$index] = isset( $info[
'host'] ) ? $info[
'host'] :
"#$index";
124 } elseif ( isset(
$params[
'server'] ) ) {
125 $this->serverInfos = [
$params[
'server'] ];
126 $this->numServers =
count( $this->serverInfos );
129 $this->serverInfos =
false;
130 $this->numServers = 1;
133 if ( isset(
$params[
'purgePeriod'] ) ) {
134 $this->purgePeriod = intval(
$params[
'purgePeriod'] );
136 if ( isset(
$params[
'tableName'] ) ) {
139 if ( isset(
$params[
'shards'] ) ) {
140 $this->shards = intval(
$params[
'shards'] );
142 if ( isset(
$params[
'syncTimeout'] ) ) {
143 $this->syncTimeout =
$params[
'syncTimeout'];
145 $this->replicaOnly = !empty(
$params[
'slaveOnly'] );
151 if ( $this->
usesMainDB() && $wgDBtype !==
'sqlite' ) {
152 if ( !$this->separateMainLB ) {
154 $lbFactory = MediaWikiServices::getInstance()->getDBLoadBalancerFactory();
155 $this->separateMainLB =
$lbFactory->newMainLB();
171 protected function getDB( $serverIndex ) {
172 if ( !isset( $this->conns[$serverIndex] ) ) {
173 if ( $serverIndex >= $this->numServers ) {
174 throw new MWException( __METHOD__ .
": Invalid server index \"$serverIndex\"" );
177 # Don't keep timing out trying to connect for each call if the DB is down
178 if ( isset( $this->connFailureErrors[$serverIndex] )
179 && ( time() - $this->connFailureTimes[$serverIndex] ) < 60
181 throw $this->connFailureErrors[$serverIndex];
184 # If server connection info was given, use that
185 if ( $this->serverInfos ) {
186 $info = $this->serverInfos[$serverIndex];
187 $type = isset( $info[
'type'] ) ? $info[
'type'] :
'mysql';
188 $host = isset( $info[
'host'] ) ? $info[
'host'] :
'[unknown]';
189 $this->logger->debug( __CLASS__ .
": connecting to $host" );
192 $db = Database::factory(
$type, $info );
204 $this->logger->debug( sprintf(
"Connection %s will be used for SqlBagOStuff", $db ) );
205 $this->conns[$serverIndex] = $db;
208 return $this->conns[$serverIndex];
217 if ( $this->shards > 1 ) {
218 $hash = hexdec( substr( md5( $key ), 0, 8 ) ) & 0x7fffffff;
223 if ( $this->numServers > 1 ) {
226 reset( $sortedServers );
227 $serverIndex =
key( $sortedServers );
240 if ( $this->shards > 1 ) {
241 $decimals = strlen( $this->shards - 1 );
243 sprintf(
"%0{$decimals}d", $index );
256 $values = $this->
getMulti( [ $key ] );
257 if ( array_key_exists( $key, $values ) ) {
258 $casToken = $values[$key];
259 return $values[$key];
270 $keysByTable[$serverIndex][
$tableName][] = $key;
276 foreach ( $keysByTable
as $serverIndex => $serverKeys ) {
278 $db = $this->
getDB( $serverIndex );
281 [
'keyname',
'value',
'exptime' ],
282 [
'keyname' => $tableKeys ],
287 $db->trxLevel() ? [
'LOCK IN SHARE MODE' ] : []
289 if (
$res ===
false ) {
292 foreach (
$res as $row ) {
293 $row->serverIndex = $serverIndex;
295 $dataRows[$row->keyname] = $row;
304 if ( isset( $dataRows[$key] ) ) {
305 $row = $dataRows[$key];
306 $this->
debug(
"get: retrieved data; expiry time is " . $row->exptime );
309 $db = $this->
getDB( $row->serverIndex );
310 if ( $this->
isExpired( $db, $row->exptime ) ) {
311 $this->
debug(
"get: key has expired" );
313 $values[$key] = $this->
unserialize( $db->decodeBlob( $row->value ) );
319 $this->
debug(
'get: no matching rows' );
328 foreach ( $data
as $key =>
$value ) {
330 $keysByTable[$serverIndex][
$tableName][] = $key;
336 $exptime = (int)$expiry;
337 foreach ( $keysByTable
as $serverIndex => $serverKeys ) {
340 $db = $this->
getDB( $serverIndex );
347 if ( $exptime < 0 ) {
351 if ( $exptime == 0 ) {
355 $encExpiry = $db->timestamp( $exptime );
359 foreach ( $tableKeys
as $key ) {
362 'value' => $db->encodeBlob( $this->
serialize( $data[$key] ) ),
363 'exptime' => $encExpiry,
388 if ( (
$flags & self::WRITE_SYNC ) == self::WRITE_SYNC ) {
395 protected function cas( $casToken, $key,
$value, $exptime = 0 ) {
399 $db = $this->
getDB( $serverIndex );
400 $exptime = intval( $exptime );
402 if ( $exptime < 0 ) {
406 if ( $exptime == 0 ) {
410 $encExpiry = $db->timestamp( $exptime );
418 'value' => $db->encodeBlob( $this->serialize(
$value ) ),
419 'exptime' => $encExpiry
423 'value' => $db->encodeBlob( $this->serialize( $casToken ) )
433 return (
bool)$db->affectedRows();
436 public function delete( $key ) {
440 $db = $this->
getDB( $serverIndex );
443 [
'keyname' => $key ],
453 public function incr( $key, $step = 1 ) {
457 $db = $this->
getDB( $serverIndex );
458 $step = intval( $step );
459 $row = $db->selectRow(
461 [
'value',
'exptime' ],
462 [
'keyname' => $key ],
465 if ( $row ===
false ) {
470 $db->delete(
$tableName, [
'keyname' => $key ], __METHOD__ );
471 if ( $this->
isExpired( $db, $row->exptime ) ) {
477 $oldValue = intval( $this->
unserialize( $db->decodeBlob( $row->value ) ) );
478 $newValue = $oldValue + $step;
482 'value' => $db->encodeBlob( $this->serialize( $newValue ) ),
483 'exptime' => $row->exptime
484 ], __METHOD__,
'IGNORE' );
486 if ( $db->affectedRows() == 0 ) {
498 public function merge( $key, callable $callback, $exptime = 0, $attempts = 10,
$flags = 0 ) {
499 $ok = $this->
mergeViaCas( $key, $callback, $exptime, $attempts );
500 if ( (
$flags & self::WRITE_SYNC ) == self::WRITE_SYNC ) {
511 $db = $this->
getDB( $serverIndex );
514 [
'exptime' => $db->timestamp( $this->convertExpiry( $expiry ) ) ],
515 [
'keyname' => $key,
'exptime > ' . $db->addQuotes( $db->timestamp( time() ) ) ],
518 if ( $db->affectedRows() == 0 ) {
543 if ( time() > 0x7fffffff ) {
544 return $db->timestamp( 1 << 62 );
546 return $db->timestamp( 0x7fffffff );
551 if ( !$this->purgePeriod || $this->replicaOnly ) {
556 if ( $this->purgePeriod !== 1 && mt_rand( 0, $this->purgePeriod - 1 ) ) {
561 if ( $now > ( $this->lastExpireAll + 1 ) ) {
562 $this->lastExpireAll = $now;
581 $db = $this->
getDB( $serverIndex );
582 $dbTimestamp = $db->timestamp( $timestamp );
583 $totalSeconds =
false;
584 $baseConds = [
'exptime < ' . $db->addQuotes( $dbTimestamp ) ];
589 if ( $maxExpTime !==
false ) {
590 $conds[] =
'exptime > ' . $db->addQuotes( $maxExpTime );
594 [
'keyname',
'exptime' ],
597 [
'LIMIT' => 100,
'ORDER BY' =>
'exptime' ] );
598 if ( $rows ===
false || !$rows->numRows() ) {
602 $row = $rows->current();
603 $minExpTime = $row->exptime;
604 if ( $totalSeconds ===
false ) {
608 foreach ( $rows
as $row ) {
609 $keys[] = $row->keyname;
610 $maxExpTime = $row->exptime;
616 'exptime >= ' . $db->addQuotes( $minExpTime ),
617 'exptime < ' . $db->addQuotes( $dbTimestamp ),
622 if ( $progressCallback ) {
623 if ( intval( $totalSeconds ) === 0 ) {
626 $remainingSeconds =
wfTimestamp( TS_UNIX, $timestamp )
628 if ( $remainingSeconds > $totalSeconds ) {
629 $totalSeconds = $remainingSeconds;
631 $processedSeconds = $totalSeconds - $remainingSeconds;
632 $percent = ( $i + $processedSeconds / $totalSeconds )
633 / $this->shards * 100;
636 + ( $serverIndex / $this->numServers * 100 );
637 call_user_func( $progressCallback, $percent );
658 $db = $this->
getDB( $serverIndex );
681 if ( function_exists(
'gzdeflate' ) ) {
682 return gzdeflate( $serial );
694 if ( function_exists(
'gzinflate' ) ) {
695 MediaWiki\suppressWarnings();
696 $decomp = gzinflate( $serial );
697 MediaWiki\restoreWarnings();
699 if (
false !== $decomp ) {
719 $this->logger->error(
"DBError: {$exception->getMessage()}" );
722 $this->logger->debug( __METHOD__ .
": ignoring connection error" );
725 $this->logger->debug( __METHOD__ .
": ignoring query error" );
740 } elseif ( $db->wasReadOnlyError() ) {
741 if ( $db->trxLevel() && $this->
usesMainDB() ) {
749 $this->logger->error(
"DBError: {$exception->getMessage()}" );
752 $this->logger->debug( __METHOD__ .
": ignoring connection error" );
755 $this->logger->debug( __METHOD__ .
": ignoring query error" );
766 unset( $this->conns[$serverIndex] );
768 if ( isset( $this->connFailureTimes[$serverIndex] ) ) {
769 if ( time() - $this->connFailureTimes[$serverIndex] >= 60 ) {
770 unset( $this->connFailureTimes[$serverIndex] );
771 unset( $this->connFailureErrors[$serverIndex] );
773 $this->logger->debug( __METHOD__ .
": Server #$serverIndex already down" );
778 $this->logger->info( __METHOD__ .
": Server #$serverIndex down until " . ( $now + 60 ) );
779 $this->connFailureTimes[$serverIndex] = $now;
780 $this->connFailureErrors[$serverIndex] = $exception;
788 $db = $this->
getDB( $serverIndex );
789 if ( $db->getType() !==
'mysql' ) {
790 throw new MWException( __METHOD__ .
' is not supported on this DB server' );
795 'CREATE TABLE ' . $db->tableName( $this->getTableNameByShard( $i ) ) .
796 ' LIKE ' . $db->tableName(
'objectcache' ),
816 ?: MediaWikiServices::getInstance()->getDBLoadBalancer();
818 if ( $lb->getServerCount() <= 1 ) {
823 $masterPos = $lb->getMasterPos();
825 $loop =
new WaitConditionLoop(
826 function ()
use ( $lb, $masterPos ) {
827 return $lb->waitForAll( $masterPos, 1 );
833 return ( $loop->invoke() === $loop::CONDITION_REACHED );