*/
use MediaWiki\MediaWikiServices;
+use Wikimedia\AtEase\AtEase;
use Wikimedia\Rdbms\Database;
use Wikimedia\Rdbms\IDatabase;
use Wikimedia\Rdbms\DBError;
use Wikimedia\Rdbms\IMaintainableDatabase;
use Wikimedia\Rdbms\LoadBalancer;
use Wikimedia\ScopedCallback;
+use Wikimedia\Timestamp\ConvertibleTimestamp;
use Wikimedia\WaitConditionLoop;
/**
$this->numServerShards = count( $this->serverInfos );
} else {
// Default to using the main wiki's database servers
- $this->serverInfos = false;
+ $this->serverInfos = [];
$this->numServerShards = 1;
$this->attrMap[self::ATTR_SYNCWRITES] = self::QOS_SYNCWRITES_BE;
}
* @return IMaintainableDatabase
* @throws MWException
*/
- private function getDB( $shardIndex ) {
+ private function getConnection( $shardIndex ) {
if ( $shardIndex >= $this->numServerShards ) {
throw new MWException( __METHOD__ . ": Invalid server index \"$shardIndex\"" );
}
$type = $info['type'] ?? 'mysql';
$host = $info['host'] ?? '[unknown]';
$this->logger->debug( __CLASS__ . ": connecting to $host" );
- $db = Database::factory( $type, $info );
- $db->clearFlag( DBO_TRX ); // auto-commit mode
- $this->conns[$shardIndex] = $db;
+ $conn = Database::factory( $type, $info );
+ $conn->clearFlag( DBO_TRX ); // auto-commit mode
+ $this->conns[$shardIndex] = $conn;
}
- $db = $this->conns[$shardIndex];
+ $conn = $this->conns[$shardIndex];
} 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->getConnectionRef( $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->getConnectionRef( $index );
+ // If the RDBMS has row-level locking, use the autocommit connection to avoid
+ // contention and deadlocks. Do not do this if it only has DB-level locking since
+ // that would just cause deadlocks.
+ $attribs = $lb->getServerAttributes( $lb->getWriterIndex() );
+ $flags = $attribs[Database::ATTR_DB_LEVEL_LOCKING] ? 0 : $lb::CONN_TRX_AUTOCOMMIT;
+ $conn = $lb->getMaintenanceConnectionRef( $index, [], false, $flags );
+ // Automatically create the objectcache table for sqlite as needed
+ if ( $conn->getType() === 'sqlite' ) {
+ $this->initSqliteDatabase( $conn );
}
}
- $this->logger->debug( sprintf( "Connection %s will be used for SqlBagOStuff", $db ) );
+ $this->logger->debug( sprintf( "Connection %s will be used for SqlBagOStuff", $conn ) );
- return $db;
+ return $conn;
}
/**
$dataRows = [];
foreach ( $keysByTable as $shardIndex => $serverKeys ) {
try {
- $db = $this->getDB( $shardIndex );
+ $db = $this->getConnection( $shardIndex );
foreach ( $serverKeys as $tableName => $tableKeys ) {
$res = $db->select( $tableName,
[ 'keyname', 'value', 'exptime' ],
$this->debug( "get: retrieved data; expiry time is " . $row->exptime );
$db = null; // in case of connection failure
try {
- $db = $this->getDB( $row->shardIndex );
+ $db = $this->getConnection( $row->shardIndex );
if ( $this->isExpired( $db, $row->exptime ) ) { // MISS
$this->debug( "get: key has expired" );
} else { // HIT
foreach ( $keysByTable as $shardIndex => $serverKeys ) {
$db = null; // in case of connection failure
try {
- $db = $this->getDB( $shardIndex );
+ $db = $this->getConnection( $shardIndex );
$this->occasionallyGarbageCollect( $db ); // expire old entries if any
$dbExpiry = $exptime ? $db->timestamp( $exptime ) : $this->getMaxDateTime( $db );
} catch ( DBError $e ) {
}
}
- if ( ( $flags & self::WRITE_SYNC ) == self::WRITE_SYNC ) {
+ if ( $this->fieldHasFlags( $flags, self::WRITE_SYNC ) ) {
$result = $this->waitForReplication() && $result;
}
$silenceScope = $this->silenceTransactionProfiler();
$db = null; // in case of connection failure
try {
- $db = $this->getDB( $shardIndex );
+ $db = $this->getConnection( $shardIndex );
// (T26425) use a replace if the db supports it instead of
// delete/insert to avoid clashes with conflicting keynames
$db->update(
return false;
}
- return (bool)$db->affectedRows();
+ $success = (bool)$db->affectedRows();
+ if ( $this->fieldHasFlags( $flags, self::WRITE_SYNC ) ) {
+ $success = $this->waitForReplication() && $success;
+ }
+
+ return $success;
}
protected function doDeleteMulti( array $keys, $flags = 0 ) {
return $this->modifyMulti( [ $key => null ], 0, $flags, self::$OP_DELETE );
}
- public function incr( $key, $step = 1 ) {
+ public function incr( $key, $step = 1, $flags = 0 ) {
list( $shardIndex, $tableName ) = $this->getTableByKey( $key );
$newCount = false;
$silenceScope = $this->silenceTransactionProfiler();
$db = null; // in case of connection failure
try {
- $db = $this->getDB( $shardIndex );
+ $db = $this->getConnection( $shardIndex );
$encTimestamp = $db->addQuotes( $db->timestamp() );
$db->update(
$tableName,
return $newCount;
}
- public function merge( $key, callable $callback, $exptime = 0, $attempts = 10, $flags = 0 ) {
- $ok = $this->mergeViaCas( $key, $callback, $exptime, $attempts, $flags );
- if ( ( $flags & self::WRITE_SYNC ) == self::WRITE_SYNC ) {
- $ok = $this->waitForReplication() && $ok;
- }
-
- return $ok;
+ public function decr( $key, $value = 1, $flags = 0 ) {
+ return $this->incr( $key, -$value, $flags );
}
public function changeTTLMulti( array $keys, $exptime, $flags = 0 ) {
* @param string $exptime
* @return bool
*/
- private function isExpired( $db, $exptime ) {
+ private function isExpired( IDatabase $db, $exptime ) {
return (
$exptime != $this->getMaxDateTime( $db ) &&
- wfTimestamp( TS_UNIX, $exptime ) < $this->getCurrentTime()
+ ConvertibleTimestamp::convert( TS_UNIX, $exptime ) < $this->getCurrentTime()
);
}
foreach ( $shardIndexes as $numServersDone => $shardIndex ) {
$db = null; // in case of connection failure
try {
- $db = $this->getDB( $shardIndex );
+ $db = $this->getConnection( $shardIndex );
$this->deleteServerObjectsExpiringBefore(
$db,
$timestamp,
$serversDoneCount = 0,
&$keysDeletedCount = 0
) {
- $cutoffUnix = wfTimestamp( TS_UNIX, $timestamp );
+ $cutoffUnix = ConvertibleTimestamp::convert( TS_UNIX, $timestamp );
$shardIndexes = range( 0, $this->numTableShards - 1 );
shuffle( $shardIndexes );
if ( $res->numRows() ) {
$row = $res->current();
if ( $lag === null ) {
- $lag = max( $cutoffUnix - wfTimestamp( TS_UNIX, $row->exptime ), 1 );
+ $rowExpUnix = ConvertibleTimestamp::convert( TS_UNIX, $row->exptime );
+ $lag = max( $cutoffUnix - $rowExpUnix, 1 );
}
$keys = [];
if ( is_callable( $progressCallback ) ) {
if ( $lag ) {
- $remainingLag = $cutoffUnix - wfTimestamp( TS_UNIX, $continue );
+ $continueUnix = ConvertibleTimestamp::convert( TS_UNIX, $continue );
+ $remainingLag = $cutoffUnix - $continueUnix;
$processedLag = max( $lag - $remainingLag, 0 );
$doneRatio = ( $numShardsDone + $processedLag / $lag ) / $this->numTableShards;
} else {
for ( $shardIndex = 0; $shardIndex < $this->numServerShards; $shardIndex++ ) {
$db = null; // in case of connection failure
try {
- $db = $this->getDB( $shardIndex );
+ $db = $this->getConnection( $shardIndex );
for ( $i = 0; $i < $this->numTableShards; $i++ ) {
$db->delete( $this->getTableNameByShard( $i ), '*', __METHOD__ );
}
$db = null; // in case of connection failure
try {
- $db = $this->getDB( $shardIndex );
+ $db = $this->getConnection( $shardIndex );
$ok = $db->lock( $key, __METHOD__, $timeout );
if ( $ok ) {
$this->locks[$key] = [ 'class' => $rclass, 'depth' => 1 ];
$db = null; // in case of connection failure
try {
- $db = $this->getDB( $shardIndex );
+ $db = $this->getConnection( $shardIndex );
$ok = $db->unlock( $key, __METHOD__ );
if ( !$ok ) {
$this->logger->warning(
}
if ( function_exists( 'gzinflate' ) ) {
- Wikimedia\suppressWarnings();
+ AtEase::suppressWarnings();
$decomp = gzinflate( $serial );
- Wikimedia\restoreWarnings();
+ AtEase::restoreWarnings();
if ( $decomp !== false ) {
$serial = $decomp;
}
/**
- * Create shard tables. For use from eval.php.
+ * @param IMaintainableDatabase $db
+ * @throws DBError
+ */
+ private function initSqliteDatabase( IMaintainableDatabase $db ) {
+ if ( $db->tableExists( 'objectcache' ) ) {
+ return;
+ }
+ // Use one table for SQLite; sharding does not seem to have much benefit
+ $db->query( "PRAGMA journal_mode=WAL" ); // this is permanent
+ $db->startAtomic( __METHOD__ ); // atomic DDL
+ try {
+ $encTable = $db->tableName( 'objectcache' );
+ $encExptimeIndex = $db->addIdentifierQuotes( $db->tablePrefix() . 'exptime' );
+ $db->query(
+ "CREATE TABLE $encTable (\n" .
+ " keyname BLOB NOT NULL default '' PRIMARY KEY,\n" .
+ " value BLOB,\n" .
+ " exptime TEXT\n" .
+ ")",
+ __METHOD__
+ );
+ $db->query( "CREATE INDEX $encExptimeIndex ON $encTable (exptime)" );
+ $db->endAtomic( __METHOD__ );
+ } catch ( DBError $e ) {
+ $db->rollback( __METHOD__ );
+ throw $e;
+ }
+ }
+
+ /**
+ * Create the shard tables on all databases (e.g. via eval.php/shell.php)
*/
public function createTables() {
for ( $shardIndex = 0; $shardIndex < $this->numServerShards; $shardIndex++ ) {
- $db = $this->getDB( $shardIndex );
- if ( $db->getType() !== 'mysql' ) {
- throw new MWException( __METHOD__ . ' is not supported on this DB server' );
- }
-
- for ( $i = 0; $i < $this->numTableShards; $i++ ) {
- $db->query(
- 'CREATE TABLE ' . $db->tableName( $this->getTableNameByShard( $i ) ) .
- ' LIKE ' . $db->tableName( 'objectcache' ),
- __METHOD__ );
+ $db = $this->getConnection( $shardIndex );
+ if ( in_array( $db->getType(), [ 'mysql', 'postgres' ], true ) ) {
+ for ( $i = 0; $i < $this->numTableShards; $i++ ) {
+ $encBaseTable = $db->tableName( 'objectcache' );
+ $encShardTable = $db->tableName( $this->getTableNameByShard( $i ) );
+ $db->query( "CREATE TABLE $encShardTable LIKE $encBaseTable" );
+ }
}
}
}
dépôts / lhc / web / wiklou.git / blobdiff