76 if ( isset(
$params[
'servers'] ) ) {
77 $this->serverInfos =
$params[
'servers'];
78 $this->numServers = count( $this->serverInfos );
79 $this->serverNames =
array();
80 foreach ( $this->serverInfos
as $i => $info ) {
81 $this->serverNames[$i] = isset( $info[
'host'] ) ? $info[
'host'] :
"#$i";
83 } elseif ( isset(
$params[
'server'] ) ) {
85 $this->numServers = count( $this->serverInfos );
87 $this->serverInfos =
false;
88 $this->numServers = 1;
90 if ( isset(
$params[
'purgePeriod'] ) ) {
91 $this->purgePeriod = intval(
$params[
'purgePeriod'] );
93 if ( isset(
$params[
'tableName'] ) ) {
96 if ( isset(
$params[
'shards'] ) ) {
97 $this->shards = intval(
$params[
'shards'] );
107 protected function getDB( $serverIndex ) {
108 global $wgDebugDBTransactions;
110 if ( !isset( $this->conns[$serverIndex] ) ) {
111 if ( $serverIndex >= $this->numServers ) {
112 throw new MWException( __METHOD__ .
": Invalid server index \"$serverIndex\"" );
115 # Don't keep timing out trying to connect for each call if the DB is down
116 if ( isset( $this->connFailureErrors[$serverIndex] )
117 && ( time() - $this->connFailureTimes[$serverIndex] ) < 60
119 throw $this->connFailureErrors[$serverIndex];
122 # If server connection info was given, use that
123 if ( $this->serverInfos ) {
124 if ( $wgDebugDBTransactions ) {
125 wfDebug(
"Using provided serverInfo for SqlBagOStuff\n" );
127 $info = $this->serverInfos[$serverIndex];
128 $type = isset( $info[
'type'] ) ? $info[
'type'] :
'mysql';
129 $host = isset( $info[
'host'] ) ? $info[
'host'] :
'[unknown]';
130 wfDebug( __CLASS__ .
": connecting to $host\n" );
141 $db = $this->lb->getConnection(
DB_MASTER );
147 if ( $wgDebugDBTransactions ) {
148 wfDebug( sprintf(
"Connection %s will be used for SqlBagOStuff\n", $db ) );
150 $this->conns[$serverIndex] = $db;
153 return $this->conns[$serverIndex];
162 if ( $this->shards > 1 ) {
163 $hash = hexdec( substr( md5( $key ), 0, 8 ) ) & 0x7fffffff;
168 if ( $this->numServers > 1 ) {
171 reset( $sortedServers );
172 $serverIndex =
key( $sortedServers );
185 if ( $this->shards > 1 ) {
186 $decimals = strlen( $this->shards - 1 );
188 sprintf(
"%0{$decimals}d", $index );
199 public function get( $key, &$casToken = null ) {
201 if ( array_key_exists( $key, $values ) ) {
202 $casToken = $values[$key];
203 return $values[$key];
215 $keysByTable =
array();
218 $keysByTable[$serverIndex][
$tableName][] = $key;
224 foreach ( $keysByTable
as $serverIndex => $serverKeys ) {
226 $db = $this->
getDB( $serverIndex );
229 array(
'keyname',
'value',
'exptime' ),
230 array(
'keyname' => $tableKeys ),
232 foreach (
$res as $row ) {
233 $row->serverIndex = $serverIndex;
235 $dataRows[$row->keyname] = $row;
244 if ( isset( $dataRows[$key] ) ) {
245 $row = $dataRows[$key];
246 $this->
debug(
"get: retrieved data; expiry time is " . $row->exptime );
248 $db = $this->
getDB( $row->serverIndex );
249 if ( $this->
isExpired( $db, $row->exptime ) ) {
250 $this->
debug(
"get: key has expired, deleting" );
251 $db->commit( __METHOD__,
'flush' );
252 # Put the expiry time in the WHERE condition to avoid deleting a
253 # newly-inserted value
254 $db->delete( $row->tableName,
255 array(
'keyname' => $key,
'exptime' => $row->exptime ),
257 $db->commit( __METHOD__,
'flush' );
258 $values[$key] =
false;
260 $values[$key] = $this->
unserialize( $db->decodeBlob( $row->value ) );
266 $values[$key] =
false;
267 $this->
debug(
'get: no matching rows' );
280 public function set( $key,
$value, $exptime = 0 ) {
283 $db = $this->
getDB( $serverIndex );
284 $exptime = intval( $exptime );
286 if ( $exptime < 0 ) {
290 if ( $exptime == 0 ) {
293 if ( $exptime < 3.16e8 ) { # ~10 years
297 $encExpiry = $db->timestamp( $exptime );
299 $db->commit( __METHOD__,
'flush' );
307 'value' => $db->encodeBlob( $this->serialize(
$value ) ),
308 'exptime' => $encExpiry
310 $db->commit( __METHOD__,
'flush' );
326 public function cas( $casToken, $key,
$value, $exptime = 0 ) {
329 $db = $this->
getDB( $serverIndex );
330 $exptime = intval( $exptime );
332 if ( $exptime < 0 ) {
336 if ( $exptime == 0 ) {
339 if ( $exptime < 3.16e8 ) { # ~10 years
342 $encExpiry = $db->timestamp( $exptime );
344 $db->commit( __METHOD__,
'flush' );
351 'value' => $db->encodeBlob( $this->serialize(
$value ) ),
352 'exptime' => $encExpiry
356 'value' => $db->encodeBlob( $this->serialize( $casToken ) )
360 $db->commit( __METHOD__,
'flush' );
367 return (
bool)$db->affectedRows();
375 public function delete( $key,
$time = 0 ) {
378 $db = $this->
getDB( $serverIndex );
379 $db->commit( __METHOD__,
'flush' );
382 array(
'keyname' => $key ),
384 $db->commit( __METHOD__,
'flush' );
398 public function incr( $key, $step = 1 ) {
401 $db = $this->
getDB( $serverIndex );
402 $step = intval( $step );
403 $db->commit( __METHOD__,
'flush' );
404 $row = $db->selectRow(
406 array(
'value',
'exptime' ),
407 array(
'keyname' => $key ),
409 array(
'FOR UPDATE' ) );
410 if ( $row ===
false ) {
412 $db->commit( __METHOD__,
'flush' );
417 if ( $this->
isExpired( $db, $row->exptime ) ) {
419 $db->commit( __METHOD__,
'flush' );
424 $oldValue = intval( $this->
unserialize( $db->decodeBlob( $row->value ) ) );
425 $newValue = $oldValue + $step;
429 'value' => $db->encodeBlob( $this->serialize( $newValue ) ),
430 'exptime' => $row->exptime
431 ), __METHOD__,
'IGNORE' );
433 if ( $db->affectedRows() == 0 ) {
437 $db->commit( __METHOD__,
'flush' );
450 protected function isExpired( $db, $exptime ) {
458 if ( time() > 0x7fffffff ) {
459 return $db->timestamp( 1 << 62 );
461 return $db->timestamp( 0x7fffffff );
466 if ( !$this->purgePeriod ) {
471 if ( $this->purgePeriod !== 1 && mt_rand( 0, $this->purgePeriod - 1 ) ) {
476 if ( $now > ( $this->lastExpireAll + 1 ) ) {
477 $this->lastExpireAll = $now;
495 $db = $this->
getDB( $serverIndex );
497 $totalSeconds =
false;
498 $baseConds =
array(
'exptime < ' . $db->addQuotes( $dbTimestamp ) );
503 if ( $maxExpTime !==
false ) {
504 $conds[] =
'exptime > ' . $db->addQuotes( $maxExpTime );
508 array(
'keyname',
'exptime' ),
511 array(
'LIMIT' => 100,
'ORDER BY' =>
'exptime' ) );
512 if ( !$rows->numRows() ) {
516 $row = $rows->current();
517 $minExpTime = $row->exptime;
518 if ( $totalSeconds ===
false ) {
522 foreach ( $rows
as $row ) {
523 $keys[] = $row->keyname;
524 $maxExpTime = $row->exptime;
527 $db->commit( __METHOD__,
'flush' );
531 'exptime >= ' . $db->addQuotes( $minExpTime ),
532 'exptime < ' . $db->addQuotes( $dbTimestamp ),
536 $db->commit( __METHOD__,
'flush' );
538 if ( $progressCallback ) {
539 if ( intval( $totalSeconds ) === 0 ) {
544 if ( $remainingSeconds > $totalSeconds ) {
545 $totalSeconds = $remainingSeconds;
547 $percent = ( $i + $remainingSeconds / $totalSeconds )
548 / $this->shards * 100;
551 + ( $serverIndex / $this->numServers * 100 );
552 call_user_func( $progressCallback, $percent );
567 $db = $this->
getDB( $serverIndex );
569 $db->commit( __METHOD__,
'flush' );
571 $db->commit( __METHOD__,
'flush' );
592 if ( function_exists(
'gzdeflate' ) ) {
593 return gzdeflate( $serial );
605 if ( function_exists(
'gzinflate' ) ) {
607 $decomp = gzinflate( $serial );
610 if (
false !== $decomp ) {
627 wfDebugLog(
'SQLBagOStuff',
"DBError: {$exception->getMessage()}" );
630 wfDebug( __METHOD__ .
": ignoring connection error\n" );
633 wfDebug( __METHOD__ .
": ignoring query error\n" );
644 if ( $exception->db && $exception->db->wasReadOnlyError() ) {
646 $exception->db->rollback( __METHOD__ );
649 wfDebugLog(
'SQLBagOStuff',
"DBError: {$exception->getMessage()}" );
652 wfDebug( __METHOD__ .
": ignoring connection error\n" );
655 wfDebug( __METHOD__ .
": ignoring query error\n" );
663 if ( isset( $this->connFailureTimes[$serverIndex] ) ) {
664 if ( time() - $this->connFailureTimes[$serverIndex] >= 60 ) {
665 unset( $this->connFailureTimes[$serverIndex] );
666 unset( $this->connFailureErrors[$serverIndex] );
668 wfDebug( __METHOD__ .
": Server #$serverIndex already down\n" );
673 wfDebug( __METHOD__ .
": Server #$serverIndex down until " . ( $now + 60 ) .
"\n" );
674 $this->connFailureTimes[$serverIndex] = $now;
675 $this->connFailureErrors[$serverIndex] = $exception;
683 $db = $this->
getDB( $serverIndex );
684 if ( $db->getType() !==
'mysql' ) {
685 throw new MWException( __METHOD__ .
' is not supported on this DB server' );
689 $db->commit( __METHOD__,
'flush' );
691 'CREATE TABLE ' . $db->tableName( $this->getTableNameByShard( $i ) ) .
692 ' LIKE ' . $db->tableName(
'objectcache' ),
694 $db->commit( __METHOD__,
'flush' );