X-Git-Url: https://git.heureux-cyclage.org/?a=blobdiff_plain;f=includes%2FLoadBalancer.php;h=4ebe26c708a3e76ab3163f85ba2774b03ce380d6;hb=e115a47d4c46a8214db907480dab8d0aee72a9ef;hp=b52cabb59dc5c57200ef5f0d4c49ec2d80ea62ff;hpb=3dc56846738f5f8bcda8c0f2bcd0edef880e2b74;p=lhc%2Fweb%2Fwiklou.git diff --git a/includes/LoadBalancer.php b/includes/LoadBalancer.php index b52cabb59d..4ebe26c708 100644 --- a/includes/LoadBalancer.php +++ b/includes/LoadBalancer.php @@ -1,55 +1,28 @@ mServers = array(); - $this->mConnections = array(); - $this->mFailFunction = false; - $this->mReadIndex = -1; - $this->mForce = -1; - $this->mLastConn = false; - } + /** + * Scale polling time so that under overload conditions, the database server + * receives a SHOW STATUS query at an average interval of this many microseconds + */ + const AVG_STATUS_POLL = 2000; - function newFromParams( $servers, $failFunction = false ) - { - $lb = new LoadBalancer; - $lb->initialise( $servers, $failFunction = false ); - return $lb; - } - - function initialise( $servers, $failFunction = false ) + function __construct( $servers, $failFunction = false, $waitTimeout = 10, $waitForMasterNow = false ) { $this->mServers = $servers; $this->mFailFunction = $failFunction; @@ -57,31 +30,57 @@ class LoadBalancer { $this->mWriteIndex = -1; $this->mForce = -1; $this->mConnections = array(); - $this->mLastConn = false; + $this->mLastIndex = -1; $this->mLoads = array(); $this->mWaitForFile = false; $this->mWaitForPos = false; + $this->mWaitTimeout = $waitTimeout; + $this->mLaggedSlaveMode = false; + $this->mErrorConnection = false; + $this->mAllowLag = false; foreach( $servers as $i => $server ) { $this->mLoads[$i] = $server['load']; + if ( isset( $server['groupLoads'] ) ) { + foreach ( $server['groupLoads'] as $group => $ratio ) { + if ( !isset( $this->mGroupLoads[$group] ) ) { + $this->mGroupLoads[$group] = array(); + } + $this->mGroupLoads[$group][$i] = $ratio; + } + } } + if ( $waitForMasterNow ) { + $this->loadMasterPos(); + } + } + + static function newFromParams( $servers, $failFunction = false, $waitTimeout = 10 ) + { + return new LoadBalancer( $servers, $failFunction, $waitTimeout ); } - - # Given an array of non-normalised probabilities, this function will select - # an element and return the appropriate key + + /** + * Given an array of non-normalised probabilities, this function will select + * an element and return the appropriate key + */ function pickRandom( $weights ) { if ( !is_array( $weights ) || count( $weights ) == 0 ) { return false; } - $sum = 0; - foreach ( $weights as $w ) { - $sum += $w; + $sum = array_sum( $weights ); + if ( $sum == 0 ) { + # No loads on any of them + # In previous versions, this triggered an unweighted random selection, + # but this feature has been removed as of April 2006 to allow for strict + # separation of query groups. + return false; } $max = mt_getrandmax(); $rand = mt_rand(0, $max) / $max * $sum; - + $sum = 0; foreach ( $weights as $i => $w ) { $sum += $w; @@ -92,191 +91,421 @@ class LoadBalancer { return $i; } - function &getReader() - { + function getRandomNonLagged( $loads ) { + # Unset excessively lagged servers + $lags = $this->getLagTimes(); + foreach ( $lags as $i => $lag ) { + if ( $i != 0 && isset( $this->mServers[$i]['max lag'] ) && + ( $lag === false || $lag > $this->mServers[$i]['max lag'] ) ) + { + unset( $loads[$i] ); + } + } + + # Find out if all the slaves with non-zero load are lagged + $sum = 0; + foreach ( $loads as $load ) { + $sum += $load; + } + if ( $sum == 0 ) { + # No appropriate DB servers except maybe the master and some slaves with zero load + # Do NOT use the master + # Instead, this function will return false, triggering read-only mode, + # and a lagged slave will be used instead. + return false; + } + + if ( count( $loads ) == 0 ) { + return false; + } + + #wfDebugLog( 'connect', var_export( $loads, true ) ); + + # Return a random representative of the remainder + return $this->pickRandom( $loads ); + } + + /** + * Get the index of the reader connection, which may be a slave + * This takes into account load ratios and lag times. It should + * always return a consistent index during a given invocation + * + * Side effect: opens connections to databases + */ + function getReaderIndex() { + global $wgReadOnly, $wgDBClusterTimeout; + + $fname = 'LoadBalancer::getReaderIndex'; + wfProfileIn( $fname ); + + $i = false; if ( $this->mForce >= 0 ) { - $conn =& $this->getConnection( $this->mForce ); + $i = $this->mForce; + } elseif ( count( $this->mServers ) == 1 ) { + # Skip the load balancing if there's only one server + $i = 0; } else { if ( $this->mReadIndex >= 0 ) { - $conn =& $this->getConnection( $this->mReadIndex ); + $i = $this->mReadIndex; } else { # $loads is $this->mLoads except with elements knocked out if they # don't work $loads = $this->mLoads; + $done = false; + $totalElapsed = 0; do { - $i = $this->pickRandom( $loads ); + if ( $wgReadOnly or $this->mAllowLagged ) { + $i = $this->pickRandom( $loads ); + } else { + $i = $this->getRandomNonLagged( $loads ); + if ( $i === false && count( $loads ) != 0 ) { + # All slaves lagged. Switch to read-only mode + $wgReadOnly = wfMsgNoDBForContent( 'readonly_lag' ); + $i = $this->pickRandom( $loads ); + } + } + $serverIndex = $i; if ( $i !== false ) { - wfDebug( "Using reader #$i: {$this->mServers[$i]['host']}\n" ); + wfDebugLog( 'connect', "$fname: Using reader #$i: {$this->mServers[$i]['host']}...\n" ); + $this->openConnection( $i ); - $conn =& $this->getConnection( $i ); - if ( !$conn->isOpen() ) { + if ( !$this->isOpen( $i ) ) { + wfDebug( "$fname: Failed\n" ); unset( $loads[$i] ); + $sleepTime = 0; + } else { + if ( isset( $this->mServers[$i]['max threads'] ) ) { + $status = $this->mConnections[$i]->getStatus("Thread%"); + if ( $status['Threads_running'] > $this->mServers[$i]['max threads'] ) { + # Too much load, back off and wait for a while. + # The sleep time is scaled by the number of threads connected, + # to produce a roughly constant global poll rate. + $sleepTime = self::AVG_STATUS_POLL * $status['Threads_connected']; + + # If we reach the timeout and exit the loop, don't use it + $i = false; + } else { + $done = true; + $sleepTime = 0; + } + } else { + $done = true; + $sleepTime = 0; + } } + } else { + $sleepTime = 500000; + } + if ( $sleepTime ) { + $totalElapsed += $sleepTime; + $x = "{$this->mServers[$serverIndex]['host']} [$serverIndex]"; + wfProfileIn( "$fname-sleep $x" ); + usleep( $sleepTime ); + wfProfileOut( "$fname-sleep $x" ); } - } while ( $i !== false && !$conn->isOpen() ); - if ( $conn->isOpen() ) { - $this->mReadIndex = $i; + } while ( count( $loads ) && !$done && $totalElapsed / 1e6 < $wgDBClusterTimeout ); + + if ( $totalElapsed / 1e6 >= $wgDBClusterTimeout ) { + $this->mErrorConnection = false; + $this->mLastError = 'All servers busy'; + } + + if ( $i !== false && $this->isOpen( $i ) ) { + # Wait for the session master pos for a short time + if ( $this->mWaitForFile ) { + if ( !$this->doWait( $i ) ) { + $this->mServers[$i]['slave pos'] = $this->mConnections[$i]->getSlavePos(); + } + } + if ( $i !== false ) { + $this->mReadIndex = $i; + } + } else { + $i = false; } } } - if ( $conn === false || !$conn->isOpen() ) { - $this->reportConnectionError( $conn ); - $conn = false; + wfProfileOut( $fname ); + return $i; + } + + /** + * Get a random server to use in a query group + */ + function getGroupIndex( $group ) { + if ( isset( $this->mGroupLoads[$group] ) ) { + $i = $this->pickRandom( $this->mGroupLoads[$group] ); + } else { + $i = false; } - return $conn; + wfDebug( "Query group $group => $i\n" ); + return $i; } - # Set the master wait position - # If a DB_SLAVE connection has been opened already, waits - # Otherwise sets a variable telling it to wait if such a connection is opened + /** + * Set the master wait position + * If a DB_SLAVE connection has been opened already, waits + * Otherwise sets a variable telling it to wait if such a connection is opened + */ function waitFor( $file, $pos ) { + $fname = 'LoadBalancer::waitFor'; + wfProfileIn( $fname ); + wfDebug( "User master pos: $file $pos\n" ); $this->mWaitForFile = false; $this->mWaitForPos = false; - if ( count( $this->mServers ) == 1 ) { - return; - } - - $this->mWaitForFile = $file; - $this->mWaitForPos = $pos; + if ( count( $this->mServers ) > 1 ) { + $this->mWaitForFile = $file; + $this->mWaitForPos = $pos; + $i = $this->mReadIndex; - if ( $this->mReadIndex > 0 ) { - $this->doWait( $this->mReadIndex ); - } + if ( $i > 0 ) { + if ( !$this->doWait( $i ) ) { + $this->mServers[$i]['slave pos'] = $this->mConnections[$i]->getSlavePos(); + $this->mLaggedSlaveMode = true; + } + } + } + wfProfileOut( $fname ); } - # Wait for a given slave to catch up to the master pos stored in $this + /** + * Wait for a given slave to catch up to the master pos stored in $this + */ function doWait( $index ) { global $wgMemc; - - $key = "masterpos:" . $index; + + $retVal = false; + + # Debugging hacks + if ( isset( $this->mServers[$index]['lagged slave'] ) ) { + return false; + } elseif ( isset( $this->mServers[$index]['fake slave'] ) ) { + return true; + } + + $key = 'masterpos:' . $index; $memcPos = $wgMemc->get( $key ); if ( $memcPos ) { list( $file, $pos ) = explode( ' ', $memcPos ); # If the saved position is later than the requested position, return now if ( $file == $this->mWaitForFile && $this->mWaitForPos <= $pos ) { - return; + $retVal = true; } } - $conn =& $this->getConnection( $index ); - wfDebug( "Waiting for slave #$index to catch up...\n" ); - $result = $conn->masterPosWait( $this->mWaitForFile, $this->mWaitForPos, MASTER_WAIT_TIMEOUT ); + if ( !$retVal && $this->isOpen( $index ) ) { + $conn =& $this->mConnections[$index]; + wfDebug( "Waiting for slave #$index to catch up...\n" ); + $result = $conn->masterPosWait( $this->mWaitForFile, $this->mWaitForPos, $this->mWaitTimeout ); - if ( $result == -1 || is_null( $result ) ) { - # Timed out waiting for slave, use master instead - # This is not the ideal solution. If there are a large number of slaves, a slow - # replicated write query will cause the master to be swamped with reads. However - # that's a relatively graceful failure mode, so it will do for now. - wfDebug( "Timed out waiting for slave #$index pos {$this->mWaitForFile} {$this->mWaitForPos}\n" ); - $this->mReadIndex = 0; - } else { - wfDebug( "Done\n" ); + if ( $result == -1 || is_null( $result ) ) { + # Timed out waiting for slave, use master instead + wfDebug( "Timed out waiting for slave #$index pos {$this->mWaitForFile} {$this->mWaitForPos}\n" ); + $retVal = false; + } else { + $retVal = true; + wfDebug( "Done\n" ); + } } - } + return $retVal; + } - function &getConnection( $i, $fail = false ) + /** + * Get a connection by index + */ + function &getConnection( $i, $fail = true, $groups = array() ) { - /* - # Task-based index - if ( $i >= DB_TASK_FIRST && $i < DB_TASK_LAST ) { - if ( $i % 2 ) { - # Odd index use writer - $i = DB_MASTER; - } else { - # Even index use reader - $i = DB_SLAVE; + global $wgDBtype; + $fname = 'LoadBalancer::getConnection'; + wfProfileIn( $fname ); + + + # Query groups + if ( !is_array( $groups ) ) { + $groupIndex = $this->getGroupIndex( $groups, $i ); + if ( $groupIndex !== false ) { + $i = $groupIndex; + } + } else { + foreach ( $groups as $group ) { + $groupIndex = $this->getGroupIndex( $group, $i ); + if ( $groupIndex !== false ) { + $i = $groupIndex; + break; + } } - }*/ + } + # For now, only go through all this for mysql databases + if ($wgDBtype != 'mysql') { + $i = $this->getWriterIndex(); + } # Operation-based index - # Note, getReader() and getWriter() will re-enter this function - if ( $i == DB_SLAVE ) { - $this->mLastConn =& $this->getReader(); + elseif ( $i == DB_SLAVE ) { + $i = $this->getReaderIndex(); } elseif ( $i == DB_MASTER ) { - $this->mLastConn =& $this->getWriter(); + $i = $this->getWriterIndex(); } elseif ( $i == DB_LAST ) { - # Just use $this->mLastConn, which should already be set - if ( $this->mLastConn === false ) { + # Just use $this->mLastIndex, which should already be set + $i = $this->mLastIndex; + if ( $i === -1 ) { # Oh dear, not set, best to use the writer for safety - $this->mLastConn =& $this->getWriter(); - } - } else { - # Explicit index - if ( !array_key_exists( $i, $this->mConnections ) || !$this->mConnections[$i]->isOpen() ) { - $this->mConnections[$i] = $this->makeConnection( $this->mServers[$i] ); - if ( $i != 0 && $this->mWaitForFile ) { - $this->doWait( $i ); - } - } - if ( !$this->mConnections[$i]->isOpen() ) { - wfDebug( "Failed to connect to database $i at {$this->mServers[$i]['host']}\n" ); - if ( $fail ) { - $this->reportConnectionError( $this->mConnections[$i] ); - } - $this->mConnections[$i] = false; + wfDebug( "Warning: DB_LAST used when there was no previous index\n" ); + $i = $this->getWriterIndex(); } - $this->mLastConn =& $this->mConnections[$i]; } - return $this->mLastConn; + # Couldn't find a working server in getReaderIndex()? + if ( $i === false ) { + $this->reportConnectionError( $this->mErrorConnection ); + } + # Now we have an explicit index into the servers array + $this->openConnection( $i, $fail ); + + wfProfileOut( $fname ); + return $this->mConnections[$i]; } - /* private */ function makeConnection( &$server ) { - extract( $server ); - # Get class for this database type - $class = 'Database' . ucfirst( $type ); - if ( !class_exists( $class ) ) { - require_once( "$class.php" ); - } + /** + * Open a connection to the server given by the specified index + * Index must be an actual index into the array + * Returns success + * @access private + */ + function openConnection( $i, $fail = false ) { + $fname = 'LoadBalancer::openConnection'; + wfProfileIn( $fname ); + $success = true; + + if ( !$this->isOpen( $i ) ) { + $this->mConnections[$i] = $this->reallyOpenConnection( $this->mServers[$i] ); + } - # Create object - return new $class( $host, $user, $password, $dbname, 1 ); + if ( !$this->isOpen( $i ) ) { + wfDebug( "Failed to connect to database $i at {$this->mServers[$i]['host']}\n" ); + if ( $fail ) { + $this->reportConnectionError( $this->mConnections[$i] ); + } + $this->mErrorConnection = $this->mConnections[$i]; + $this->mConnections[$i] = false; + $success = false; + } + $this->mLastIndex = $i; + wfProfileOut( $fname ); + return $success; } - - function reportConnectionError( &$conn ) - { - if ( !is_object( $conn ) ) { - $conn = new Database; + + /** + * Test if the specified index represents an open connection + * @access private + */ + function isOpen( $index ) { + if( !is_integer( $index ) ) { + return false; } - if ( $this->mFailFunction ) { - $conn->setFailFunction( $this->mFailFunction ); + if ( array_key_exists( $index, $this->mConnections ) && is_object( $this->mConnections[$index] ) && + $this->mConnections[$index]->isOpen() ) + { + return true; } else { - $conn->setFailFunction( "wfEmergencyAbort" ); + return false; } - $conn->reportConnectionError(); } - - function &getWriter() - { - $c =& $this->getConnection( 0 ); - if ( $c === false || !$c->isOpen() ) { - $this->reportConnectionError( $c ); - $c = false; + + /** + * Really opens a connection + * @access private + */ + function reallyOpenConnection( &$server ) { + if( !is_array( $server ) ) { + throw new MWException( 'You must update your load-balancing configuration. See DefaultSettings.php entry for $wgDBservers.' ); } - return $c; + + extract( $server ); + # Get class for this database type + $class = 'Database' . ucfirst( $type ); + + # Create object + $db = new $class( $host, $user, $password, $dbname, 1, $flags ); + $db->setLBInfo( $server ); + return $db; } - function force( $i ) + function reportConnectionError( &$conn ) { + $fname = 'LoadBalancer::reportConnectionError'; + wfProfileIn( $fname ); + # Prevent infinite recursion + + static $reporting = false; + if ( !$reporting ) { + $reporting = true; + if ( !is_object( $conn ) ) { + // No last connection, probably due to all servers being too busy + $conn = new Database; + if ( $this->mFailFunction ) { + $conn->failFunction( $this->mFailFunction ); + $conn->reportConnectionError( $this->mLastError ); + } else { + // If all servers were busy, mLastError will contain something sensible + throw new DBConnectionError( $conn, $this->mLastError ); + } + } else { + if ( $this->mFailFunction ) { + $conn->failFunction( $this->mFailFunction ); + } else { + $conn->failFunction( false ); + } + $server = $conn->getProperty( 'mServer' ); + $conn->reportConnectionError( "{$this->mLastError} ({$server})" ); + } + $reporting = false; + } + wfProfileOut( $fname ); + } + + function getWriterIndex() { + return 0; + } + + /** + * Force subsequent calls to getConnection(DB_SLAVE) to return the + * given index. Set to -1 to restore the original load balancing + * behaviour. I thought this was a good idea when I originally + * wrote this class, but it has never been used. + */ + function force( $i ) { $this->mForce = $i; } - function haveIndex( $i ) - { + /** + * Returns true if the specified index is a valid server index + */ + function haveIndex( $i ) { return array_key_exists( $i, $this->mServers ); } - # Get the number of defined servers (not the number of open connections) + /** + * Returns true if the specified index is valid and has non-zero load + */ + function isNonZeroLoad( $i ) { + return array_key_exists( $i, $this->mServers ) && $this->mLoads[$i] != 0; + } + + /** + * Get the number of defined servers (not the number of open connections) + */ function getServerCount() { return count( $this->mServers ); } - # Save master pos to the session and to memcached, if the session exists + /** + * Save master pos to the session and to memcached, if the session exists + */ function saveMasterPos() { - global $wgSessionStarted; - if ( $wgSessionStarted && count( $this->mServers ) > 1 ) { - # If this entire request was served from a slave without opening a connection to the + if ( session_id() != '' && count( $this->mServers ) > 1 ) { + # If this entire request was served from a slave without opening a connection to the # master (however unlikely that may be), then we can fetch the position from the slave. if ( empty( $this->mConnections[0] ) ) { $conn =& $this->getConnection( DB_SLAVE ); @@ -294,10 +523,127 @@ class LoadBalancer { } } - # Loads the master pos from the session, waits for it if necessary + /** + * Loads the master pos from the session, waits for it if necessary + */ function loadMasterPos() { if ( isset( $_SESSION['master_log_file'] ) && isset( $_SESSION['master_pos'] ) ) { $this->waitFor( $_SESSION['master_log_file'], $_SESSION['master_pos'] ); } } + + /** + * Close all open connections + */ + function closeAll() { + foreach( $this->mConnections as $i => $conn ) { + if ( $this->isOpen( $i ) ) { + // Need to use this syntax because $conn is a copy not a reference + $this->mConnections[$i]->close(); + } + } + } + + function commitAll() { + foreach( $this->mConnections as $i => $conn ) { + if ( $this->isOpen( $i ) ) { + // Need to use this syntax because $conn is a copy not a reference + $this->mConnections[$i]->immediateCommit(); + } + } + } + + function waitTimeout( $value = NULL ) { + return wfSetVar( $this->mWaitTimeout, $value ); + } + + function getLaggedSlaveMode() { + return $this->mLaggedSlaveMode; + } + + /* Disables/enables lag checks */ + function allowLagged($mode=null) { + if ($mode===null) + return $this->mAllowLagged; + $this->mAllowLagged=$mode; + } + + function pingAll() { + $success = true; + foreach ( $this->mConnections as $i => $conn ) { + if ( $this->isOpen( $i ) ) { + if ( !$this->mConnections[$i]->ping() ) { + $success = false; + } + } + } + return $success; + } + + /** + * Get the hostname and lag time of the most-lagged slave + * This is useful for maintenance scripts that need to throttle their updates + */ + function getMaxLag() { + $maxLag = -1; + $host = ''; + foreach ( $this->mServers as $i => $conn ) { + if ( $this->openConnection( $i ) ) { + $lag = $this->mConnections[$i]->getLag(); + if ( $lag > $maxLag ) { + $maxLag = $lag; + $host = $this->mServers[$i]['host']; + } + } + } + return array( $host, $maxLag ); + } + + /** + * Get lag time for each DB + * Results are cached for a short time in memcached + */ + function getLagTimes() { + wfProfileIn( __METHOD__ ); + $expiry = 5; + $requestRate = 10; + + global $wgMemc; + $times = $wgMemc->get( wfMemcKey( 'lag_times' ) ); + if ( $times ) { + # Randomly recache with probability rising over $expiry + $elapsed = time() - $times['timestamp']; + $chance = max( 0, ( $expiry - $elapsed ) * $requestRate ); + if ( mt_rand( 0, $chance ) != 0 ) { + unset( $times['timestamp'] ); + wfProfileOut( __METHOD__ ); + return $times; + } + wfIncrStats( 'lag_cache_miss_expired' ); + } else { + wfIncrStats( 'lag_cache_miss_absent' ); + } + + # Cache key missing or expired + + $times = array(); + foreach ( $this->mServers as $i => $conn ) { + if ($i==0) { # Master + $times[$i] = 0; + } elseif ( $this->openConnection( $i ) ) { + $times[$i] = $this->mConnections[$i]->getLag(); + } + } + + # Add a timestamp key so we know when it was cached + $times['timestamp'] = time(); + $wgMemc->set( wfMemcKey( 'lag_times' ), $times, $expiry ); + + # But don't give the timestamp to the caller + unset($times['timestamp']); + wfProfileOut( __METHOD__ ); + return $times; + } } + +?>