*
* @ingroup Cache
*/
-class SqlBagOStuff extends BagOStuff {
+class SqlBagOStuff extends MediumSpecificBagOStuff {
/** @var array[] (server index => server config) */
protected $serverInfos;
/** @var string[] (server index => tag/host name) */
protected $serverTags;
/** @var int */
protected $numServers;
- /** @var int */
- protected $lastExpireAll = 0;
+ /** @var int UNIX timestamp */
+ protected $lastGarbageCollect = 0;
/** @var int */
protected $purgePeriod = 10;
/** @var int */
protected $tableName = 'objectcache';
/** @var bool */
protected $replicaOnly = false;
- /** @var int */
- protected $syncTimeout = 3;
/** @var LoadBalancer|null */
protected $separateMainLB;
/** @var array Exceptions */
protected $connFailureErrors = [];
+ /** @var int */
+ private static $GARBAGE_COLLECT_DELAY_SEC = 1;
+
+ /** @var string */
+ private static $OP_SET = 'set';
+ /** @var string */
+ private static $OP_ADD = 'add';
+ /** @var string */
+ private static $OP_TOUCH = 'touch';
+ /** @var string */
+ private static $OP_DELETE = 'delete';
+
/**
* Constructor. Parameters are:
* - server: A server info structure in the format required by each
* MySQL bugs 61735 <https://bugs.mysql.com/bug.php?id=61735>
* and 61736 <https://bugs.mysql.com/bug.php?id=61736>.
*
- * - slaveOnly: Whether to only use replica DBs and avoid triggering
+ * - replicaOnly: Whether to only use replica DBs and avoid triggering
* garbage collection logic of expired items. This only
* makes sense if the primary DB is used and only if get()
* calls will be used. This is used by ReplicatedBagOStuff.
if ( isset( $params['shards'] ) ) {
$this->shards = intval( $params['shards'] );
}
- if ( isset( $params['syncTimeout'] ) ) {
- $this->syncTimeout = $params['syncTimeout'];
- }
- $this->replicaOnly = !empty( $params['slaveOnly'] );
+ // Backwards-compatibility for < 1.34
+ $this->replicaOnly = $params['replicaOnly'] ?? ( $params['slaveOnly'] ?? false );
}
/**
* @throws MWException
*/
protected function getDB( $serverIndex ) {
- if ( !isset( $this->conns[$serverIndex] ) ) {
- if ( $serverIndex >= $this->numServers ) {
- throw new MWException( __METHOD__ . ": Invalid server index \"$serverIndex\"" );
- }
+ if ( $serverIndex >= $this->numServers ) {
+ throw new MWException( __METHOD__ . ": Invalid server index \"$serverIndex\"" );
+ }
- # Don't keep timing out trying to connect for each call if the DB is down
- if ( isset( $this->connFailureErrors[$serverIndex] )
- && ( time() - $this->connFailureTimes[$serverIndex] ) < 60
- ) {
- throw $this->connFailureErrors[$serverIndex];
- }
+ # Don't keep timing out trying to connect for each call if the DB is down
+ if (
+ isset( $this->connFailureErrors[$serverIndex] ) &&
+ ( time() - $this->connFailureTimes[$serverIndex] ) < 60
+ ) {
+ throw $this->connFailureErrors[$serverIndex];
+ }
- if ( $this->serverInfos ) {
+ if ( $this->serverInfos ) {
+ if ( !isset( $this->conns[$serverIndex] ) ) {
// Use custom database defined by server connection info
$info = $this->serverInfos[$serverIndex];
$type = $info['type'] ?? 'mysql';
$this->logger->debug( __CLASS__ . ": connecting to $host" );
$db = Database::factory( $type, $info );
$db->clearFlag( DBO_TRX ); // auto-commit mode
+ $this->conns[$serverIndex] = $db;
+ }
+ $db = $this->conns[$serverIndex];
+ } else {
+ // Use the main LB database
+ $lb = MediaWikiServices::getInstance()->getDBLoadBalancer();
+ $index = $this->replicaOnly ? DB_REPLICA : DB_MASTER;
+ if ( $lb->getServerType( $lb->getWriterIndex() ) !== 'sqlite' ) {
+ // Keep a separate connection to avoid contention and deadlocks
+ $db = $lb->getConnection( $index, [], false, $lb::CONN_TRX_AUTOCOMMIT );
} else {
- // Use the main LB database
- $lb = MediaWikiServices::getInstance()->getDBLoadBalancer();
- $index = $this->replicaOnly ? DB_REPLICA : DB_MASTER;
- if ( $lb->getServerType( $lb->getWriterIndex() ) !== 'sqlite' ) {
- // Keep a separate connection to avoid contention and deadlocks
- $db = $lb->getConnection( $index, [], false, $lb::CONN_TRX_AUTOCOMMIT );
- } else {
- // However, SQLite has the opposite behavior due to DB-level locking.
- // Stock sqlite MediaWiki installs use a separate sqlite cache DB instead.
- $db = $lb->getConnection( $index );
- }
+ // However, SQLite has the opposite behavior due to DB-level locking.
+ // Stock sqlite MediaWiki installs use a separate sqlite cache DB instead.
+ $db = $lb->getConnection( $index );
}
-
- $this->logger->debug( sprintf( "Connection %s will be used for SqlBagOStuff", $db ) );
- $this->conns[$serverIndex] = $db;
}
- return $this->conns[$serverIndex];
+ $this->logger->debug( sprintf( "Connection %s will be used for SqlBagOStuff", $db ) );
+
+ return $db;
}
/**
if ( isset( $dataRows[$key] ) ) { // HIT?
$row = $dataRows[$key];
$this->debug( "get: retrieved data; expiry time is " . $row->exptime );
- $db = null;
+ $db = null; // in case of connection failure
try {
$db = $this->getDB( $row->serverIndex );
if ( $this->isExpired( $db, $row->exptime ) ) { // MISS
return $values;
}
- public function setMulti( array $data, $expiry = 0, $flags = 0 ) {
- return $this->insertMulti( $data, $expiry, $flags, true );
+ protected function doSetMulti( array $data, $exptime = 0, $flags = 0 ) {
+ return $this->modifyMulti( $data, $exptime, $flags, self::$OP_SET );
}
- private function insertMulti( array $data, $expiry, $flags, $replace ) {
+ /**
+ * @param mixed[]|null[] $data Map of (key => new value or null)
+ * @param int $exptime UNIX timestamp, TTL in seconds, or 0 (no expiration)
+ * @param int $flags Bitfield of BagOStuff::WRITE_* constants
+ * @param string $op Cache operation
+ * @return bool
+ */
+ private function modifyMulti( array $data, $exptime, $flags, $op ) {
$keysByTable = [];
foreach ( $data as $key => $value ) {
list( $serverIndex, $tableName ) = $this->getTableByKey( $key );
$keysByTable[$serverIndex][$tableName][] = $key;
}
- $this->garbageCollect(); // expire old entries if any
+ $exptime = $this->convertToExpiry( $exptime );
$result = true;
- $exptime = (int)$expiry;
/** @noinspection PhpUnusedLocalVariableInspection */
$silenceScope = $this->silenceTransactionProfiler();
foreach ( $keysByTable as $serverIndex => $serverKeys ) {
- $db = null;
+ $db = null; // in case of connection failure
try {
$db = $this->getDB( $serverIndex );
+ $this->occasionallyGarbageCollect( $db ); // expire old entries if any
+ $dbExpiry = $exptime ? $db->timestamp( $exptime ) : $this->getMaxDateTime( $db );
} catch ( DBError $e ) {
$this->handleWriteError( $e, $db, $serverIndex );
$result = false;
continue;
}
- if ( $exptime < 0 ) {
- $exptime = 0;
- }
-
- if ( $exptime == 0 ) {
- $encExpiry = $this->getMaxDateTime( $db );
- } else {
- $exptime = $this->convertToExpiry( $exptime );
- $encExpiry = $db->timestamp( $exptime );
- }
foreach ( $serverKeys as $tableName => $tableKeys ) {
- $rows = [];
- foreach ( $tableKeys as $key ) {
- $rows[] = [
- 'keyname' => $key,
- 'value' => $db->encodeBlob( $this->serialize( $data[$key] ) ),
- 'exptime' => $encExpiry,
- ];
- }
-
try {
- if ( $replace ) {
- $db->replace( $tableName, [ 'keyname' ], $rows, __METHOD__ );
- } else {
- $db->insert( $tableName, $rows, __METHOD__, [ 'IGNORE' ] );
- $result = ( $db->affectedRows() > 0 && $result );
- }
+ $result = $this->updateTableKeys(
+ $op,
+ $db,
+ $tableName,
+ $tableKeys,
+ $data,
+ $dbExpiry
+ ) && $result;
} catch ( DBError $e ) {
$this->handleWriteError( $e, $db, $serverIndex );
$result = false;
return $result;
}
- protected function doSet( $key, $value, $exptime = 0, $flags = 0 ) {
- $ok = $this->insertMulti( [ $key => $value ], $exptime, $flags, true );
+ /**
+ * @param string $op
+ * @param IDatabase $db
+ * @param string $table
+ * @param string[] $tableKeys Keys in $data to update
+ * @param mixed[]|null[] $data Map of (key => new value or null)
+ * @param string $dbExpiry DB-encoded expiry
+ * @return bool
+ * @throws DBError
+ * @throws InvalidArgumentException
+ */
+ private function updateTableKeys( $op, $db, $table, $tableKeys, $data, $dbExpiry ) {
+ $success = true;
- return $ok;
+ if ( $op === self::$OP_ADD ) {
+ $rows = [];
+ foreach ( $tableKeys as $key ) {
+ $rows[] = [
+ 'keyname' => $key,
+ 'value' => $db->encodeBlob( $this->serialize( $data[$key] ) ),
+ 'exptime' => $dbExpiry
+ ];
+ }
+ $db->delete(
+ $table,
+ [
+ 'keyname' => $tableKeys,
+ 'exptime <= ' . $db->addQuotes( $db->timestamp() )
+ ],
+ __METHOD__
+ );
+ $db->insert( $table, $rows, __METHOD__, [ 'IGNORE' ] );
+
+ $success = ( $db->affectedRows() == count( $rows ) );
+ } elseif ( $op === self::$OP_SET ) {
+ $rows = [];
+ foreach ( $tableKeys as $key ) {
+ $rows[] = [
+ 'keyname' => $key,
+ 'value' => $db->encodeBlob( $this->serialize( $data[$key] ) ),
+ 'exptime' => $dbExpiry
+ ];
+ }
+ $db->replace( $table, [ 'keyname' ], $rows, __METHOD__ );
+ } elseif ( $op === self::$OP_DELETE ) {
+ $db->delete( $table, [ 'keyname' => $tableKeys ], __METHOD__ );
+ } elseif ( $op === self::$OP_TOUCH ) {
+ $db->update(
+ $table,
+ [ 'exptime' => $dbExpiry ],
+ [
+ 'keyname' => $tableKeys,
+ 'exptime > ' . $db->addQuotes( $db->timestamp() )
+ ],
+ __METHOD__
+ );
+
+ $success = ( $db->affectedRows() == count( $tableKeys ) );
+ } else {
+ throw new InvalidArgumentException( "Invalid operation '$op'" );
+ }
+
+ return $success;
}
- public function add( $key, $value, $exptime = 0, $flags = 0 ) {
- $added = $this->insertMulti( [ $key => $value ], $exptime, $flags, false );
+ protected function doSet( $key, $value, $exptime = 0, $flags = 0 ) {
+ return $this->modifyMulti( [ $key => $value ], $exptime, $flags, self::$OP_SET );
+ }
- return $added;
+ public function add( $key, $value, $exptime = 0, $flags = 0 ) {
+ return $this->modifyMulti( [ $key => $value ], $exptime, $flags, self::$OP_ADD );
}
protected function cas( $casToken, $key, $value, $exptime = 0, $flags = 0 ) {
list( $serverIndex, $tableName ) = $this->getTableByKey( $key );
- $db = null;
+ $exptime = $this->convertToExpiry( $exptime );
+
/** @noinspection PhpUnusedLocalVariableInspection */
$silenceScope = $this->silenceTransactionProfiler();
+ $db = null; // in case of connection failure
try {
$db = $this->getDB( $serverIndex );
- $exptime = intval( $exptime );
-
- if ( $exptime < 0 ) {
- $exptime = 0;
- }
-
- if ( $exptime == 0 ) {
- $encExpiry = $this->getMaxDateTime( $db );
- } else {
- $exptime = $this->convertToExpiry( $exptime );
- $encExpiry = $db->timestamp( $exptime );
- }
// (T26425) use a replace if the db supports it instead of
// delete/insert to avoid clashes with conflicting keynames
$db->update(
[
'keyname' => $key,
'value' => $db->encodeBlob( $this->serialize( $value ) ),
- 'exptime' => $encExpiry
+ 'exptime' => $exptime
+ ? $db->timestamp( $exptime )
+ : $this->getMaxDateTime( $db )
],
[
'keyname' => $key,
- 'value' => $db->encodeBlob( $casToken )
+ 'value' => $db->encodeBlob( $casToken ),
+ 'exptime > ' . $db->addQuotes( $db->timestamp() )
],
__METHOD__
);
return (bool)$db->affectedRows();
}
- public function deleteMulti( array $keys, $flags = 0 ) {
- return $this->purgeMulti( $keys, $flags );
- }
-
- public function purgeMulti( array $keys, $flags = 0 ) {
- $keysByTable = [];
- foreach ( $keys as $key ) {
- list( $serverIndex, $tableName ) = $this->getTableByKey( $key );
- $keysByTable[$serverIndex][$tableName][] = $key;
- }
-
- $result = true;
- /** @noinspection PhpUnusedLocalVariableInspection */
- $silenceScope = $this->silenceTransactionProfiler();
- foreach ( $keysByTable as $serverIndex => $serverKeys ) {
- $db = null;
- try {
- $db = $this->getDB( $serverIndex );
- } catch ( DBError $e ) {
- $this->handleWriteError( $e, $db, $serverIndex );
- $result = false;
- continue;
- }
-
- foreach ( $serverKeys as $tableName => $tableKeys ) {
- try {
- $db->delete( $tableName, [ 'keyname' => $tableKeys ], __METHOD__ );
- } catch ( DBError $e ) {
- $this->handleWriteError( $e, $db, $serverIndex );
- $result = false;
- }
-
- }
- }
-
- if ( ( $flags & self::WRITE_SYNC ) == self::WRITE_SYNC ) {
- $result = $this->waitForReplication() && $result;
- }
-
- return $result;
+ protected function doDeleteMulti( array $keys, $flags = 0 ) {
+ return $this->modifyMulti(
+ array_fill_keys( $keys, null ),
+ 0,
+ $flags,
+ self::$OP_DELETE
+ );
}
protected function doDelete( $key, $flags = 0 ) {
- $ok = $this->purgeMulti( [ $key ], $flags );
-
- return $ok;
+ return $this->modifyMulti( [ $key => null ], 0, $flags, self::$OP_DELETE );
}
public function incr( $key, $step = 1 ) {
list( $serverIndex, $tableName ) = $this->getTableByKey( $key );
- $db = null;
+
+ $newCount = false;
/** @noinspection PhpUnusedLocalVariableInspection */
$silenceScope = $this->silenceTransactionProfiler();
+ $db = null; // in case of connection failure
try {
$db = $this->getDB( $serverIndex );
- $step = intval( $step );
- $row = $db->selectRow(
- $tableName,
- [ 'value', 'exptime' ],
- [ 'keyname' => $key ],
- __METHOD__,
- [ 'FOR UPDATE' ]
- );
- if ( $row === false ) {
- // Missing
- return false;
- }
- $db->delete( $tableName, [ 'keyname' => $key ], __METHOD__ );
- if ( $this->isExpired( $db, $row->exptime ) ) {
- // Expired, do not reinsert
- return false;
- }
-
- $oldValue = intval( $this->unserialize( $db->decodeBlob( $row->value ) ) );
- $newValue = $oldValue + $step;
- $db->insert(
+ $encTimestamp = $db->addQuotes( $db->timestamp() );
+ $db->update(
$tableName,
- [
- 'keyname' => $key,
- 'value' => $db->encodeBlob( $this->serialize( $newValue ) ),
- 'exptime' => $row->exptime
- ],
- __METHOD__,
- [ 'IGNORE' ]
+ [ 'value = value + ' . (int)$step ],
+ [ 'keyname' => $key, "exptime > $encTimestamp" ],
+ __METHOD__
);
-
- if ( $db->affectedRows() == 0 ) {
- // Race condition. See T30611
- $newValue = false;
+ if ( $db->affectedRows() > 0 ) {
+ $newValue = $db->selectField(
+ $tableName,
+ 'value',
+ [ 'keyname' => $key, "exptime > $encTimestamp" ],
+ __METHOD__
+ );
+ if ( $this->isInteger( $newValue ) ) {
+ $newCount = (int)$newValue;
+ }
}
} catch ( DBError $e ) {
$this->handleWriteError( $e, $db, $serverIndex );
- return false;
}
- return $newValue;
+ return $newCount;
}
public function merge( $key, callable $callback, $exptime = 0, $attempts = 10, $flags = 0 ) {
return $ok;
}
- public function changeTTL( $key, $exptime = 0, $flags = 0 ) {
- list( $serverIndex, $tableName ) = $this->getTableByKey( $key );
- $db = null;
- /** @noinspection PhpUnusedLocalVariableInspection */
- $silenceScope = $this->silenceTransactionProfiler();
- try {
- $db = $this->getDB( $serverIndex );
- if ( $exptime == 0 ) {
- $timestamp = $this->getMaxDateTime( $db );
- } else {
- $timestamp = $db->timestamp( $this->convertToExpiry( $exptime ) );
- }
- $db->update(
- $tableName,
- [ 'exptime' => $timestamp ],
- [ 'keyname' => $key, 'exptime > ' . $db->addQuotes( $db->timestamp( time() ) ) ],
- __METHOD__
- );
- if ( $db->affectedRows() == 0 ) {
- $exists = (bool)$db->selectField(
- $tableName,
- 1,
- [ 'keyname' => $key, 'exptime' => $timestamp ],
- __METHOD__,
- [ 'FOR UPDATE' ]
- );
-
- return $exists;
- }
- } catch ( DBError $e ) {
- $this->handleWriteError( $e, $db, $serverIndex );
- return false;
- }
+ protected function doChangeTTLMulti( array $keys, $exptime, $flags = 0 ) {
+ return $this->modifyMulti(
+ array_fill_keys( $keys, null ),
+ $exptime,
+ $flags,
+ self::$OP_TOUCH
+ );
+ }
- return true;
+ protected function doChangeTTL( $key, $exptime, $flags ) {
+ return $this->modifyMulti( [ $key => null ], $exptime, $flags, self::$OP_TOUCH );
}
/**
* @return bool
*/
protected function isExpired( $db, $exptime ) {
- return $exptime != $this->getMaxDateTime( $db ) && wfTimestamp( TS_UNIX, $exptime ) < time();
+ return (
+ $exptime != $this->getMaxDateTime( $db ) &&
+ wfTimestamp( TS_UNIX, $exptime ) < time()
+ );
}
/**
}
}
- protected function garbageCollect() {
- if ( !$this->purgePeriod || $this->replicaOnly ) {
- // Disabled
- return;
- }
- // Only purge on one in every $this->purgePeriod writes
- if ( $this->purgePeriod !== 1 && mt_rand( 0, $this->purgePeriod - 1 ) ) {
- return;
- }
- $now = time();
- // Avoid repeating the delete within a few seconds
- if ( $now > ( $this->lastExpireAll + 1 ) ) {
- $this->lastExpireAll = $now;
- $this->deleteObjectsExpiringBefore(
- wfTimestamp( TS_MW, $now ),
- false,
- $this->purgeLimit
- );
+ /**
+ * @param IDatabase $db
+ * @throws DBError
+ */
+ protected function occasionallyGarbageCollect( IDatabase $db ) {
+ if (
+ // Random purging is enabled
+ $this->purgePeriod &&
+ // Only purge on one in every $this->purgePeriod writes
+ mt_rand( 0, $this->purgePeriod - 1 ) == 0 &&
+ // Avoid repeating the delete within a few seconds
+ ( time() - $this->lastGarbageCollect ) > self::$GARBAGE_COLLECT_DELAY_SEC
+ ) {
+ $garbageCollector = function () use ( $db ) {
+ $this->deleteServerObjectsExpiringBefore( $db, time(), null, $this->purgeLimit );
+ $this->lastGarbageCollect = time();
+ };
+ if ( $this->asyncHandler ) {
+ $this->lastGarbageCollect = time(); // avoid duplicate enqueues
+ ( $this->asyncHandler )( $garbageCollector );
+ } else {
+ $garbageCollector();
+ }
}
}
public function expireAll() {
- $this->deleteObjectsExpiringBefore( wfTimestampNow() );
+ $this->deleteObjectsExpiringBefore( time() );
}
public function deleteObjectsExpiringBefore(
$timestamp,
- $progressCallback = false,
+ callable $progress = null,
$limit = INF
) {
/** @noinspection PhpUnusedLocalVariableInspection */
$silenceScope = $this->silenceTransactionProfiler();
- $count = 0;
- for ( $serverIndex = 0; $serverIndex < $this->numServers; $serverIndex++ ) {
- $db = null;
+ $serverIndexes = range( 0, $this->numServers - 1 );
+ shuffle( $serverIndexes );
+
+ $ok = true;
+
+ $keysDeletedCount = 0;
+ foreach ( $serverIndexes as $numServersDone => $serverIndex ) {
+ $db = null; // in case of connection failure
try {
$db = $this->getDB( $serverIndex );
- $dbTimestamp = $db->timestamp( $timestamp );
- $totalSeconds = false;
- $baseConds = [ 'exptime < ' . $db->addQuotes( $dbTimestamp ) ];
- for ( $i = 0; $i < $this->shards; $i++ ) {
- $maxExpTime = false;
- while ( true ) {
- $conds = $baseConds;
- if ( $maxExpTime !== false ) {
- $conds[] = 'exptime >= ' . $db->addQuotes( $maxExpTime );
- }
- $rows = $db->select(
- $this->getTableNameByShard( $i ),
- [ 'keyname', 'exptime' ],
- $conds,
- __METHOD__,
- [ 'LIMIT' => 100, 'ORDER BY' => 'exptime' ]
- );
- if ( $rows === false || !$rows->numRows() ) {
- break;
- }
- $keys = [];
- $row = $rows->current();
- $minExpTime = $row->exptime;
- if ( $totalSeconds === false ) {
- $totalSeconds = wfTimestamp( TS_UNIX, $timestamp )
- - wfTimestamp( TS_UNIX, $minExpTime );
- }
- foreach ( $rows as $row ) {
- $keys[] = $row->keyname;
- $maxExpTime = $row->exptime;
- }
-
- $db->delete(
- $this->getTableNameByShard( $i ),
- [
- 'exptime >= ' . $db->addQuotes( $minExpTime ),
- 'exptime < ' . $db->addQuotes( $dbTimestamp ),
- 'keyname' => $keys
- ],
- __METHOD__
- );
- $count += $db->affectedRows();
- if ( $count >= $limit ) {
- return true;
- }
-
- if ( is_callable( $progressCallback ) ) {
- if ( intval( $totalSeconds ) === 0 ) {
- $percent = 0;
- } else {
- $remainingSeconds = wfTimestamp( TS_UNIX, $timestamp )
- - wfTimestamp( TS_UNIX, $maxExpTime );
- if ( $remainingSeconds > $totalSeconds ) {
- $totalSeconds = $remainingSeconds;
- }
- $processedSeconds = $totalSeconds - $remainingSeconds;
- $percent = ( $i + $processedSeconds / $totalSeconds )
- / $this->shards * 100;
- }
- $percent = ( $percent / $this->numServers )
- + ( $serverIndex / $this->numServers * 100 );
- call_user_func( $progressCallback, $percent );
- }
- }
- }
+ $this->deleteServerObjectsExpiringBefore(
+ $db,
+ $timestamp,
+ $progress,
+ $limit,
+ $numServersDone,
+ $keysDeletedCount
+ );
} catch ( DBError $e ) {
$this->handleWriteError( $e, $db, $serverIndex );
- return false;
+ $ok = false;
}
}
- return true;
+ return $ok;
+ }
+
+ /**
+ * @param IDatabase $db
+ * @param string|int $timestamp
+ * @param callable|null $progressCallback
+ * @param int $limit
+ * @param int $serversDoneCount
+ * @param int &$keysDeletedCount
+ * @throws DBError
+ */
+ private function deleteServerObjectsExpiringBefore(
+ IDatabase $db,
+ $timestamp,
+ $progressCallback,
+ $limit,
+ $serversDoneCount = 0,
+ &$keysDeletedCount = 0
+ ) {
+ $cutoffUnix = wfTimestamp( TS_UNIX, $timestamp );
+ $shardIndexes = range( 0, $this->shards - 1 );
+ shuffle( $shardIndexes );
+
+ foreach ( $shardIndexes as $numShardsDone => $shardIndex ) {
+ $continue = null; // last exptime
+ $lag = null; // purge lag
+ do {
+ $res = $db->select(
+ $this->getTableNameByShard( $shardIndex ),
+ [ 'keyname', 'exptime' ],
+ array_merge(
+ [ 'exptime < ' . $db->addQuotes( $db->timestamp( $cutoffUnix ) ) ],
+ $continue ? [ 'exptime >= ' . $db->addQuotes( $continue ) ] : []
+ ),
+ __METHOD__,
+ [ 'LIMIT' => min( $limit, 100 ), 'ORDER BY' => 'exptime' ]
+ );
+
+ if ( $res->numRows() ) {
+ $row = $res->current();
+ if ( $lag === null ) {
+ $lag = max( $cutoffUnix - wfTimestamp( TS_UNIX, $row->exptime ), 1 );
+ }
+
+ $keys = [];
+ foreach ( $res as $row ) {
+ $keys[] = $row->keyname;
+ $continue = $row->exptime;
+ }
+
+ $db->delete(
+ $this->getTableNameByShard( $shardIndex ),
+ [
+ 'exptime < ' . $db->addQuotes( $db->timestamp( $cutoffUnix ) ),
+ 'keyname' => $keys
+ ],
+ __METHOD__
+ );
+ $keysDeletedCount += $db->affectedRows();
+ }
+
+ if ( is_callable( $progressCallback ) ) {
+ if ( $lag ) {
+ $remainingLag = $cutoffUnix - wfTimestamp( TS_UNIX, $continue );
+ $processedLag = max( $lag - $remainingLag, 0 );
+ $doneRatio = ( $numShardsDone + $processedLag / $lag ) / $this->shards;
+ } else {
+ $doneRatio = 1;
+ }
+
+ $overallRatio = ( $doneRatio / $this->numServers )
+ + ( $serversDoneCount / $this->numServers );
+ call_user_func( $progressCallback, $overallRatio * 100 );
+ }
+ } while ( $res->numRows() && $keysDeletedCount < $limit );
+ }
}
/**
/** @noinspection PhpUnusedLocalVariableInspection */
$silenceScope = $this->silenceTransactionProfiler();
for ( $serverIndex = 0; $serverIndex < $this->numServers; $serverIndex++ ) {
- $db = null;
+ $db = null; // in case of connection failure
try {
$db = $this->getDB( $serverIndex );
for ( $i = 0; $i < $this->shards; $i++ ) {
}
list( $serverIndex ) = $this->getTableByKey( $key );
+
+ $db = null; // in case of connection failure
try {
$db = $this->getDB( $serverIndex );
$ok = $db->lock( $key, __METHOD__, $timeout );
unset( $this->locks[$key] );
list( $serverIndex ) = $this->getTableByKey( $key );
+
+ $db = null; // in case of connection failure
try {
$db = $this->getDB( $serverIndex );
$ok = $db->unlock( $key, __METHOD__ );
$this->handleWriteError( $e, $db, $serverIndex );
$ok = false;
}
- } else {
- $ok = false;
+
+ return $ok;
}
- return $ok;
+ return true;
}
/**
* in storage requirements.
*
* @param mixed $data
- * @return string
+ * @return string|int
*/
protected function serialize( $data ) {
- $serial = serialize( $data );
+ if ( is_int( $data ) ) {
+ return $data;
+ }
+ $serial = serialize( $data );
if ( function_exists( 'gzdeflate' ) ) {
- return gzdeflate( $serial );
- } else {
- return $serial;
+ $serial = gzdeflate( $serial );
}
+
+ return $serial;
}
/**
* @return mixed
*/
protected function unserialize( $serial ) {
+ if ( $this->isInteger( $serial ) ) {
+ return (int)$serial;
+ }
+
if ( function_exists( 'gzinflate' ) ) {
Wikimedia\suppressWarnings();
$decomp = gzinflate( $serial );
}
}
- $ret = unserialize( $serial );
-
- return $ret;
+ return unserialize( $serial );
}
/**
dépôts / lhc / web / wiklou.git / blobdiff