private $loadMonitorConfig;
/** @var array[] $aliases Map of (table => (dbname, schema, prefix) map) */
private $tableAliases = [];
+ /** @var string[] Map of (index alias => index) */
+ private $indexAliases = [];
/** @var ILoadMonitor */
private $loadMonitor;
- /** @var ChronologyProtector|null */
- private $chronProt;
+ /** @var callable|null Callback to run before the first connection attempt */
+ private $chronologyCallback;
/** @var BagOStuff */
private $srvCache;
/** @var WANObjectCache */
/** @var callable Exception logger */
private $errorLogger;
+ /** @var callable Deprecation logger */
+ private $deprecationLogger;
/** @var bool */
private $disabled = false;
- /** @var bool */
- private $chronProtInitialized = false;
+ /** @var bool Whether any connection has been attempted yet */
+ private $connectionAttempted = false;
/** @var int */
private $maxLag = self::MAX_LAG_DEFAULT;
: function ( Exception $e ) {
trigger_error( get_class( $e ) . ': ' . $e->getMessage(), E_USER_WARNING );
};
+ $this->deprecationLogger = isset( $params['deprecationLogger'] )
+ ? $params['deprecationLogger']
+ : function ( $msg ) {
+ trigger_error( $msg, E_USER_DEPRECATED );
+ };
foreach ( [ 'replLogger', 'connLogger', 'queryLogger', 'perfLogger' ] as $key ) {
$this->$key = isset( $params[$key] ) ? $params[$key] : new NullLogger();
: ( PHP_SAPI === 'cli' || PHP_SAPI === 'phpdbg' );
$this->agent = isset( $params['agent'] ) ? $params['agent'] : '';
- if ( isset( $params['chronologyProtector'] ) ) {
- $this->chronProt = $params['chronologyProtector'];
+ if ( isset( $params['chronologyCallback'] ) ) {
+ $this->chronologyCallback = $params['chronologyCallback'];
}
}
$host = $this->getServerName( $i );
if ( $lag === false && !is_infinite( $maxServerLag ) ) {
$this->replLogger->error(
- "Server {host} is not replicating?", [ 'host' => $host ] );
+ __METHOD__ .
+ ": server {host} is not replicating?", [ 'host' => $host ] );
unset( $loads[$i] );
} elseif ( $lag > $maxServerLag ) {
- $this->replLogger->info(
- "Server {host} has {lag} seconds of lag (>= {maxlag})",
+ $this->replLogger->debug(
+ __METHOD__ .
+ ": server {host} has {lag} seconds of lag (>= {maxlag})",
[ 'host' => $host, 'lag' => $lag, 'maxlag' => $maxServerLag ]
);
unset( $loads[$i] );
// Scale the configured load ratios according to each server's load and state
$this->getLoadMonitor()->scaleLoads( $loads, $domain );
- // Pick a server to use, accounting for weights, load, lag, and mWaitForPos
+ // Pick a server to use, accounting for weights, load, lag, and "waitForPos"
list( $i, $laggedReplicaMode ) = $this->pickReaderIndex( $loads, $domain );
if ( $i === false ) {
// Replica DB connection unsuccessful
if ( $this->waitForPos && $i != $this->getWriterIndex() ) {
// Before any data queries are run, wait for the server to catch up to the
// specified position. This is used to improve session consistency. Note that
- // when LoadBalancer::waitFor() sets mWaitForPos, the waiting triggers here,
+ // when LoadBalancer::waitFor() sets "waitForPos", the waiting triggers here,
// so update laggedReplicaMode as needed for consistency.
if ( !$this->doWait( $i ) ) {
$laggedReplicaMode = true;
} else {
$i = false;
if ( $this->waitForPos && $this->waitForPos->asOfTime() ) {
- // ChronologyProtecter sets mWaitForPos for session consistency.
+ // "chronologyCallback" sets "waitForPos" for session consistency.
// This triggers doWait() after connect, so it's especially good to
// avoid lagged servers so as to avoid excessive delay in that method.
$ago = microtime( true ) - $this->waitForPos->asOfTime();
}
if ( $i === false && count( $currentLoads ) != 0 ) {
// All replica DBs lagged. Switch to read-only mode
- $this->replLogger->error( "All replica DBs lagged. Switch to read-only mode" );
+ $this->replLogger->error(
+ __METHOD__ . ": all replica DBs lagged. Switch to read-only mode" );
$i = ArrayUtils::pickRandom( $currentLoads );
$laggedReplicaMode = true;
}
// If all servers were down, quit now
if ( !count( $currentLoads ) ) {
- $this->connLogger->error( "All servers down" );
+ $this->connLogger->error( __METHOD__ . ": all servers down" );
}
return [ $i, $laggedReplicaMode ];
if ( $this->loads[$i] > 0 ) {
$start = microtime( true );
$ok = $this->doWait( $i, true, $timeout ) && $ok;
- $timeout -= ( microtime( true ) - $start );
+ $timeout -= intval( microtime( true ) - $start );
if ( $timeout <= 0 ) {
break; // timeout reached
}
}
}
- /**
- * @param int $i
- * @return IDatabase|bool
- */
- public function getAnyOpenConnection( $i ) {
+ public function getAnyOpenConnection( $i, $flags = 0 ) {
+ $autocommit = ( ( $flags & self::CONN_TRX_AUTOCOMMIT ) == self::CONN_TRX_AUTOCOMMIT );
foreach ( $this->conns as $connsByServer ) {
- if ( !empty( $connsByServer[$i] ) ) {
- /** @var IDatabase[] $serverConns */
- $serverConns = $connsByServer[$i];
+ if ( !isset( $connsByServer[$i] ) ) {
+ continue;
+ }
- return reset( $serverConns );
+ foreach ( $connsByServer[$i] as $conn ) {
+ if ( !$autocommit || $conn->getLBInfo( 'autoCommitOnly' ) ) {
+ return $conn;
+ }
}
}
* @return bool
*/
protected function doWait( $index, $open = false, $timeout = null ) {
- $timeout = max( 1, $timeout ?: $this->waitTimeout );
+ $timeout = max( 1, intval( $timeout ?: $this->waitTimeout ) );
// Check if we already know that the DB has reached this point
$server = $this->getServerName( $index );
$this->replLogger->info(
__METHOD__ .
- ': Waiting for replica DB {dbserver} to catch up...',
+ ': waiting for replica DB {dbserver} to catch up...',
[ 'dbserver' => $server ]
);
);
$ok = false;
} else {
- $this->replLogger->info( __METHOD__ . ": Done" );
+ $this->replLogger->debug( __METHOD__ . ": done waiting" );
$ok = true;
// Remember that the DB reached this point
$this->srvCache->set( $key, $this->waitForPos, BagOStuff::TTL_DAY );
$domain = false; // local connection requested
}
+ if ( ( $flags & self::CONN_TRX_AUTOCOMMIT ) === self::CONN_TRX_AUTOCOMMIT ) {
+ // Assuming all servers are of the same type (or similar), which is overwhelmingly
+ // the case, use the master server information to get the attributes. The information
+ // for $i cannot be used since it might be DB_REPLICA, which might require connection
+ // attempts in order to be resolved into a real server index.
+ $attributes = $this->getServerAttributes( $this->getWriterIndex() );
+ if ( $attributes[Database::ATTR_DB_LEVEL_LOCKING] ) {
+ // Callers sometimes want to (a) escape REPEATABLE-READ stateness without locking
+ // rows (e.g. FOR UPDATE) or (b) make small commits during a larger transactions
+ // to reduce lock contention. None of these apply for sqlite and using separate
+ // connections just causes self-deadlocks.
+ $flags &= ~self::CONN_TRX_AUTOCOMMIT;
+ $this->connLogger->info( __METHOD__ .
+ ': ignoring CONN_TRX_AUTOCOMMIT to avoid deadlocks.' );
+ }
+ }
+
$groups = ( $groups === false || $groups === [] )
? [ false ] // check one "group": the generic pool
: (array)$groups;
if ( $i == self::DB_MASTER ) {
$i = $this->getWriterIndex();
- } else {
+ } elseif ( $i == self::DB_REPLICA ) {
# Try to find an available server in any the query groups (in order)
foreach ( $groups as $group ) {
$groupIndex = $this->getReaderIndex( $group, $domain );
return $conn;
}
- public function reuseConnection( $conn ) {
+ public function reuseConnection( IDatabase $conn ) {
$serverIndex = $conn->getLBInfo( 'serverIndex' );
$refCount = $conn->getLBInfo( 'foreignPoolRefCount' );
if ( $serverIndex === null || $refCount === null ) {
} elseif ( $conn instanceof DBConnRef ) {
// DBConnRef already handles calling reuseConnection() and only passes the live
// Database instance to this method. Any caller passing in a DBConnRef is broken.
- $this->connLogger->error( __METHOD__ . ": got DBConnRef instance.\n" .
+ $this->connLogger->error(
+ __METHOD__ . ": got DBConnRef instance.\n" .
( new RuntimeException() )->getTraceAsString() );
return;
$domain = false; // local connection requested
}
- if ( !$this->chronProtInitialized && $this->chronProt ) {
+ if ( !$this->connectionAttempted && $this->chronologyCallback ) {
$this->connLogger->debug( __METHOD__ . ': calling initLB() before first connection.' );
- // Load CP positions before connecting so that doWait() triggers later if needed
- $this->chronProtInitialized = true;
- $this->chronProt->initLB( $this );
+ // Load any "waitFor" positions before connecting so that doWait() is triggered
+ $this->connectionAttempted = true;
+ call_user_func( $this->chronologyCallback, $this );
}
// Check if an auto-commit connection is being requested. If so, it will not reuse the
// main set of DB connections but rather its own pool since:
// a) those are usually set to implicitly use transaction rounds via DBO_TRX
// b) those must support the use of explicit transaction rounds via beginMasterChanges()
- $autoCommit = ( ( $flags & self::CONN_TRX_AUTO ) == self::CONN_TRX_AUTO );
+ $autoCommit = ( ( $flags & self::CONN_TRX_AUTOCOMMIT ) == self::CONN_TRX_AUTOCOMMIT );
if ( $domain !== false ) {
// Connection is to a foreign domain
$conn = $this->reallyOpenConnection( $server, $this->localDomain );
$host = $this->getServerName( $i );
if ( $conn->isOpen() ) {
- $this->connLogger->debug( "Connected to database $i at '$host'." );
+ $this->connLogger->debug(
+ __METHOD__ . ": connected to database $i at '$host'." );
$this->conns[$connKey][$i][0] = $conn;
} else {
- $this->connLogger->warning( "Failed to connect to database $i at '$host'." );
+ $this->connLogger->warning(
+ __METHOD__ . ": failed to connect to database $i at '$host'." );
$this->errorConnection = $conn;
$conn = false;
}
$domainInstance = DatabaseDomain::newFromId( $domain );
$dbName = $domainInstance->getDatabase();
$prefix = $domainInstance->getTablePrefix();
- $autoCommit = ( ( $flags & self::CONN_TRX_AUTO ) == self::CONN_TRX_AUTO );
+ $autoCommit = ( ( $flags & self::CONN_TRX_AUTOCOMMIT ) == self::CONN_TRX_AUTOCOMMIT );
if ( $autoCommit ) {
$connFreeKey = self::KEY_FOREIGN_FREE_NOROUND;
return $conn;
}
+ public function getServerAttributes( $i ) {
+ return Database::attributesFromType(
+ $this->getServerType( $i ),
+ isset( $this->servers[$i]['driver'] ) ? $this->servers[$i]['driver'] : null
+ );
+ }
+
/**
* Test if the specified index represents an open connection
*
$server['connLogger'] = $this->connLogger;
$server['queryLogger'] = $this->queryLogger;
$server['errorLogger'] = $this->errorLogger;
+ $server['deprecationLogger'] = $this->deprecationLogger;
$server['profiler'] = $this->profiler;
$server['trxProfiler'] = $this->trxProfiler;
// Use the same agent and PHP mode for all DB handles
$this->getLazyConnectionRef( self::DB_MASTER, [], $db->getDomainID() )
);
$db->setTableAliases( $this->tableAliases );
+ $db->setIndexAliases( $this->indexAliases );
if ( $server['serverIndex'] === $this->getWriterIndex() ) {
if ( $this->trxRoundId !== false ) {
if ( $conn instanceof IDatabase ) {
$context['db_server'] = $conn->getServer();
$this->connLogger->warning(
- "Connection error: {last_error} ({db_server})",
+ __METHOD__ . ": connection error: {last_error} ({db_server})",
$context
);
} else {
// No last connection, probably due to all servers being too busy
$this->connLogger->error(
- "LB failure with no last connection. Connection error: {last_error}",
+ __METHOD__ .
+ ": LB failure with no last connection. Connection error: {last_error}",
$context
);
- // If all servers were busy, mLastError will contain something sensible
+ // If all servers were busy, "lastError" will contain something sensible
throw new DBConnectionError( null, $this->lastError );
}
}
return ( $name != '' ) ? $name : 'localhost';
}
+ public function getServerInfo( $i ) {
+ if ( isset( $this->servers[$i] ) ) {
+ return $this->servers[$i];
+ } else {
+ return false;
+ }
+ }
+
public function getServerType( $i ) {
return isset( $this->servers[$i]['type'] ) ? $this->servers[$i]['type'] : 'unknown';
}
public function closeAll() {
$this->forEachOpenConnection( function ( IDatabase $conn ) {
$host = $conn->getServer();
- $this->connLogger->debug( "Closing connection to database '$host'." );
+ $this->connLogger->debug(
+ __METHOD__ . ": closing connection to database '$host'." );
$conn->close();
} );
}
public function closeConnection( IDatabase $conn ) {
- $serverIndex = $conn->getLBInfo( 'serverIndex' ); // second index level of mConns
+ $serverIndex = $conn->getLBInfo( 'serverIndex' );
foreach ( $this->conns as $type => $connsByServer ) {
if ( !isset( $connsByServer[$serverIndex] ) ) {
continue;
foreach ( $connsByServer[$serverIndex] as $i => $trackedConn ) {
if ( $conn === $trackedConn ) {
$host = $this->getServerName( $i );
- $this->connLogger->debug( "Closing connection to database $i at '$host'." );
+ $this->connLogger->debug(
+ __METHOD__ . ": closing connection to database $i at '$host'." );
unset( $this->conns[$type][$serverIndex][$i] );
--$this->connsOpened;
break 2;
$e = null; // first exception
$this->forEachOpenMasterConnection( function ( Database $conn ) use ( $type, &$e ) {
$conn->setTrxEndCallbackSuppression( false );
- if ( $conn->writesOrCallbacksPending() ) {
- // This happens if onTransactionIdle() callbacks leave callbacks on *another* DB
- // (which finished its callbacks already). Warn and recover in this case. Let the
- // callbacks run in the final commitMasterChanges() in LBFactory::shutdown().
- $this->queryLogger->info( __METHOD__ . ": found writes/callbacks pending." );
+ // Callbacks run in AUTO-COMMIT mode, so make sure no transactions are pending...
+ if ( $conn->writesPending() ) {
+ // This happens if onTransactionIdle() callbacks write to *other* handles
+ // (which already finished their callbacks). Let any callbacks run in the final
+ // commitMasterChanges() in LBFactory::shutdown(), when the transaction is gone.
+ $this->queryLogger->warning( __METHOD__ . ": found writes pending." );
return;
} elseif ( $conn->trxLevel() ) {
// This happens for single-DB setups where DB_REPLICA uses the master DB,
$this->trxRoundId = false;
$this->forEachOpenMasterConnection(
function ( IDatabase $conn ) use ( $fname, $restore ) {
- if ( $conn->writesOrCallbacksPending() || $conn->explicitTrxActive() ) {
- $conn->rollback( $fname, $conn::FLUSHING_ALL_PEERS );
- }
+ $conn->rollback( $fname, $conn::FLUSHING_ALL_PEERS );
if ( $restore ) {
$this->undoTransactionRoundFlags( $conn );
}
}
/**
+ * Make all DB servers with DBO_DEFAULT/DBO_TRX set join the transaction round
+ *
+ * Some servers may have neither flag enabled, meaning that they opt out of such
+ * transaction rounds and remain in auto-commit mode. Such behavior might be desired
+ * when a DB server is used for something like simple key/value storage.
+ *
* @param IDatabase $conn
*/
private function applyTransactionRoundFlags( IDatabase $conn ) {
// DBO_TRX is controlled entirely by CLI mode presence with DBO_DEFAULT.
// Force DBO_TRX even in CLI mode since a commit round is expected soon.
$conn->setFlag( $conn::DBO_TRX, $conn::REMEMBER_PRIOR );
- // If config has explicitly requested DBO_TRX be either on or off by not
- // setting DBO_DEFAULT, then respect that. Forcing no transactions is useful
- // for things like blob stores (ExternalStore) which want auto-commit mode.
+ }
+
+ if ( $conn->getFlag( $conn::DBO_TRX ) ) {
+ $conn->setLBInfo( 'trxRoundId', $this->trxRoundId );
}
}
return; // transaction rounds do not apply to these connections
}
+ if ( $conn->getFlag( $conn::DBO_TRX ) ) {
+ $conn->setLBInfo( 'trxRoundId', false );
+ }
+
if ( $conn->getFlag( $conn::DBO_DEFAULT ) ) {
$conn->restoreFlags( $conn::RESTORE_PRIOR );
}
if ( $pos instanceof DBMasterPos ) {
$result = $conn->masterPosWait( $pos, $timeout );
if ( $result == -1 || is_null( $result ) ) {
- $msg = __METHOD__ . ': Timed out waiting on {host} pos {pos}';
+ $msg = __METHOD__ . ': timed out waiting on {host} pos {pos}';
$this->replLogger->warning( $msg, [
'host' => $conn->getServer(),
'pos' => $pos,
] );
$ok = false;
} else {
- $this->replLogger->info( __METHOD__ . ': Done' );
+ $this->replLogger->debug( __METHOD__ . ': done waiting' );
$ok = true;
}
} else {
$this->tableAliases = $aliases;
}
+ public function setIndexAliases( array $aliases ) {
+ $this->indexAliases = $aliases;
+ }
+
public function setDomainPrefix( $prefix ) {
// Find connections to explicit foreign domains still marked as in-use...
$domainsInUse = [];
dépôts / lhc / web / wiklou.git / blobdiff