includes/LoadBalancer.php

   1 <?php
   2 /**
   3  *
   4  */
   5
   6
   7 /**
   8  * Database load balancing object
   9  *
  10  * @todo document
  11  */
  12 class LoadBalancer {
  13         /* private */ var $mServers, $mConnections, $mLoads, $mGroupLoads;
  14         /* private */ var $mFailFunction, $mErrorConnection;
  15         /* private */ var $mForce, $mReadIndex, $mLastIndex, $mAllowLagged;
  16         /* private */ var $mWaitForFile, $mWaitForPos, $mWaitTimeout;
  17         /* private */ var $mLaggedSlaveMode, $mLastError = 'Unknown error';
  18
  19         /**
  20          * Scale polling time so that under overload conditions, the database server
  21          * receives a SHOW STATUS query at an average interval of this many microseconds
  22          */
  23         const AVG_STATUS_POLL = 30000;
  24
  25         function __construct( $servers, $failFunction = false, $waitTimeout = 10, $waitForMasterNow = false )
  26         {
  27                 $this->mServers = $servers;
  28                 $this->mFailFunction = $failFunction;
  29                 $this->mReadIndex = -1;
  30                 $this->mWriteIndex = -1;
  31                 $this->mForce = -1;
  32                 $this->mConnections = array();
  33                 $this->mLastIndex = -1;
  34                 $this->mLoads = array();
  35                 $this->mWaitForFile = false;
  36                 $this->mWaitForPos = false;
  37                 $this->mWaitTimeout = $waitTimeout;
  38                 $this->mLaggedSlaveMode = false;
  39                 $this->mErrorConnection = false;
  40                 $this->mAllowLag = false;
  41
  42                 foreach( $servers as $i => $server ) {
  43                         $this->mLoads[$i] = $server['load'];
  44                         if ( isset( $server['groupLoads'] ) ) {
  45                                 foreach ( $server['groupLoads'] as $group => $ratio ) {
  46                                         if ( !isset( $this->mGroupLoads[$group] ) ) {
  47                                                 $this->mGroupLoads[$group] = array();
  48                                         }
  49                                         $this->mGroupLoads[$group][$i] = $ratio;
  50                                 }
  51                         }
  52                 }
  53                 if ( $waitForMasterNow ) {
  54                         $this->loadMasterPos();
  55                 }
  56         }
  57
  58         static function newFromParams( $servers, $failFunction = false, $waitTimeout = 10 )
  59         {
  60                 return new LoadBalancer( $servers, $failFunction, $waitTimeout );
  61         }
  62
  63         /**
  64          * Given an array of non-normalised probabilities, this function will select
  65          * an element and return the appropriate key
  66          */
  67         function pickRandom( $weights )
  68         {
  69                 if ( !is_array( $weights ) || count( $weights ) == 0 ) {
  70                         return false;
  71                 }
  72
  73                 $sum = array_sum( $weights );
  74                 if ( $sum == 0 ) {
  75                         # No loads on any of them
  76                         # In previous versions, this triggered an unweighted random selection,
  77                         # but this feature has been removed as of April 2006 to allow for strict
  78                         # separation of query groups.
  79                         return false;
  80                 }
  81                 $max = mt_getrandmax();
  82                 $rand = mt_rand(0, $max) / $max * $sum;
  83
  84                 $sum = 0;
  85                 foreach ( $weights as $i => $w ) {
  86                         $sum += $w;
  87                         if ( $sum >= $rand ) {
  88                                 break;
  89                         }
  90                 }
  91                 return $i;
  92         }
  93
  94         function getRandomNonLagged( $loads ) {
  95                 # Unset excessively lagged servers
  96                 $lags = $this->getLagTimes();
  97                 foreach ( $lags as $i => $lag ) {
  98                         if ( $i != 0 && isset( $this->mServers[$i]['max lag'] ) &&
  99                                 ( $lag === false || $lag > $this->mServers[$i]['max lag'] ) )
 100                         {
 101                                 unset( $loads[$i] );
 102                         }
 103                 }
 104
 105                 # Find out if all the slaves with non-zero load are lagged
 106                 $sum = 0;
 107                 foreach ( $loads as $load ) {
 108                         $sum += $load;
 109                 }
 110                 if ( $sum == 0 ) {
 111                         # No appropriate DB servers except maybe the master and some slaves with zero load
 112                         # Do NOT use the master
 113                         # Instead, this function will return false, triggering read-only mode,
 114                         # and a lagged slave will be used instead.
 115                         return false;
 116                 }
 117
 118                 if ( count( $loads ) == 0 ) {
 119                         return false;
 120                 }
 121
 122                 #wfDebugLog( 'connect', var_export( $loads, true ) );
 123
 124                 # Return a random representative of the remainder
 125                 return $this->pickRandom( $loads );
 126         }
 127
 128         /**
 129          * Get the index of the reader connection, which may be a slave
 130          * This takes into account load ratios and lag times. It should
 131          * always return a consistent index during a given invocation
 132          *
 133          * Side effect: opens connections to databases
 134          */
 135         function getReaderIndex() {
 136                 global $wgReadOnly, $wgDBClusterTimeout;
 137
 138                 $fname = 'LoadBalancer::getReaderIndex';
 139                 wfProfileIn( $fname );
 140
 141                 $i = false;
 142                 if ( $this->mForce >= 0 ) {
 143                         $i = $this->mForce;
 144                 } elseif ( count( $this->mServers ) == 1 )  {
 145                         # Skip the load balancing if there's only one server
 146                         $i = 0;
 147                 } else {
 148                         if ( $this->mReadIndex >= 0 ) {
 149                                 $i = $this->mReadIndex;
 150                         } else {
 151                                 # $loads is $this->mLoads except with elements knocked out if they
 152                                 # don't work
 153                                 $loads = $this->mLoads;
 154                                 $done = false;
 155                                 $totalElapsed = 0;
 156                                 do {
 157                                         if ( $wgReadOnly or $this->mAllowLagged ) {
 158                                                 $i = $this->pickRandom( $loads );
 159                                         } else {
 160                                                 $i = $this->getRandomNonLagged( $loads );
 161                                                 if ( $i === false && count( $loads ) != 0 )  {
 162                                                         # All slaves lagged. Switch to read-only mode
 163                                                         $wgReadOnly = wfMsgNoDBForContent( 'readonly_lag' );
 164                                                         $i = $this->pickRandom( $loads );
 165                                                 }
 166                                         }
 167                                         $serverIndex = $i;
 168                                         if ( $i !== false ) {
 169                                                 wfDebugLog( 'connect', "$fname: Using reader #$i: {$this->mServers[$i]['host']}...\n" );
 170                                                 $this->openConnection( $i );
 171
 172                                                 if ( !$this->isOpen( $i ) ) {
 173                                                         wfDebug( "$fname: Failed\n" );
 174                                                         unset( $loads[$i] );
 175                                                         $sleepTime = 0;
 176                                                 } else {
 177                                                         if ( isset( $this->mServers[$i]['max threads'] ) ) {
 178                                                             $status = $this->mConnections[$i]->getStatus("Thread%");
 179                                                             if ( $status['Threads_running'] > $this->mServers[$i]['max threads'] ) {
 180                                                                 # Too much load, back off and wait for a while.
 181                                                                 # The sleep time is scaled by the number of threads connected,
 182                                                                 # to produce a roughly constant global poll rate.
 183                                                                 $sleepTime = self::AVG_STATUS_POLL * $status['Threads_connected'];
 184
 185                                                                 # If we reach the timeout and exit the loop, don't use it
 186                                                                 $i = false;
 187                                                             } else {
 188                                                                 $done = true;
 189                                                                 $sleepTime = 0;
 190                                                             }
 191                                                         } else {
 192                                                             $done = true;
 193                                                             $sleepTime = 0;
 194                                                         }
 195                                                 }
 196                                         } else {
 197                                                 $sleepTime = 500000;
 198                                         }
 199                                         if ( $sleepTime ) {
 200                                                         $totalElapsed += $sleepTime;
 201                                                         $x = "{$this->mServers[$serverIndex]['host']} [$serverIndex]";
 202                                                         wfProfileIn( "$fname-sleep $x" );
 203                                                         usleep( $sleepTime );
 204                                                         wfProfileOut( "$fname-sleep $x" );
 205                                         }
 206                                 } while ( count( $loads ) && !$done && $totalElapsed / 1e6 < $wgDBClusterTimeout );
 207
 208                                 if ( $totalElapsed / 1e6 >= $wgDBClusterTimeout ) {
 209                                         $this->mErrorConnection = false;
 210                                         $this->mLastError = 'All servers busy';
 211                                 }
 212
 213                                 if ( $i !== false && $this->isOpen( $i ) ) {
 214                                         # Wait for the session master pos for a short time
 215                                         if ( $this->mWaitForFile ) {
 216                                                 if ( !$this->doWait( $i ) ) {
 217                                                         $this->mServers[$i]['slave pos'] = $this->mConnections[$i]->getSlavePos();
 218                                                 }
 219                                         }
 220                                         if ( $i !== false ) {
 221                                                 $this->mReadIndex = $i;
 222                                         }
 223                                 } else {
 224                                         $i = false;
 225                                 }
 226                         }
 227                 }
 228                 wfProfileOut( $fname );
 229                 return $i;
 230         }
 231
 232         /**
 233          * Get a random server to use in a query group
 234          */
 235         function getGroupIndex( $group ) {
 236                 if ( isset( $this->mGroupLoads[$group] ) ) {
 237                         $i = $this->pickRandom( $this->mGroupLoads[$group] );
 238                 } else {
 239                         $i = false;
 240                 }
 241                 wfDebug( "Query group $group => $i\n" );
 242                 return $i;
 243         }
 244
 245         /**
 246          * Set the master wait position
 247          * If a DB_SLAVE connection has been opened already, waits
 248          * Otherwise sets a variable telling it to wait if such a connection is opened
 249          */
 250         function waitFor( $file, $pos ) {
 251                 $fname = 'LoadBalancer::waitFor';
 252                 wfProfileIn( $fname );
 253
 254                 wfDebug( "User master pos: $file $pos\n" );
 255                 $this->mWaitForFile = false;
 256                 $this->mWaitForPos = false;
 257
 258                 if ( count( $this->mServers ) > 1 ) {
 259                         $this->mWaitForFile = $file;
 260                         $this->mWaitForPos = $pos;
 261                         $i = $this->mReadIndex;
 262
 263                         if ( $i > 0 ) {
 264                                 if ( !$this->doWait( $i ) ) {
 265                                         $this->mServers[$i]['slave pos'] = $this->mConnections[$i]->getSlavePos();
 266                                         $this->mLaggedSlaveMode = true;
 267                                 }
 268                         }
 269                 }
 270                 wfProfileOut( $fname );
 271         }
 272
 273         /**
 274          * Wait for a given slave to catch up to the master pos stored in $this
 275          */
 276         function doWait( $index ) {
 277                 global $wgMemc;
 278
 279                 $retVal = false;
 280
 281                 # Debugging hacks
 282                 if ( isset( $this->mServers[$index]['lagged slave'] ) ) {
 283                         return false;
 284                 } elseif ( isset( $this->mServers[$index]['fake slave'] ) ) {
 285                         return true;
 286                 }
 287
 288                 $key = 'masterpos:' . $index;
 289                 $memcPos = $wgMemc->get( $key );
 290                 if ( $memcPos ) {
 291                         list( $file, $pos ) = explode( ' ', $memcPos );
 292                         # If the saved position is later than the requested position, return now
 293                         if ( $file == $this->mWaitForFile && $this->mWaitForPos <= $pos ) {
 294                                 $retVal = true;
 295                         }
 296                 }
 297
 298                 if ( !$retVal && $this->isOpen( $index ) ) {
 299                         $conn =& $this->mConnections[$index];
 300                         wfDebug( "Waiting for slave #$index to catch up...\n" );
 301                         $result = $conn->masterPosWait( $this->mWaitForFile, $this->mWaitForPos, $this->mWaitTimeout );
 302
 303                         if ( $result == -1 || is_null( $result ) ) {
 304                                 # Timed out waiting for slave, use master instead
 305                                 wfDebug( "Timed out waiting for slave #$index pos {$this->mWaitForFile} {$this->mWaitForPos}\n" );
 306                                 $retVal = false;
 307                         } else {
 308                                 $retVal = true;
 309                                 wfDebug( "Done\n" );
 310                         }
 311                 }
 312                 return $retVal;
 313         }
 314
 315         /**
 316          * Get a connection by index
 317          */
 318         function &getConnection( $i, $fail = true, $groups = array() )
 319         {
 320                 global $wgDBtype;
 321                 $fname = 'LoadBalancer::getConnection';
 322                 wfProfileIn( $fname );
 323
 324
 325                 # Query groups
 326                 if ( !is_array( $groups ) ) {
 327                         $groupIndex = $this->getGroupIndex( $groups );
 328                         if ( $groupIndex !== false ) {
 329                                 $i = $groupIndex;
 330                         }
 331                 } else {
 332                         foreach ( $groups as $group ) {
 333                                 $groupIndex = $this->getGroupIndex( $group );
 334                                 if ( $groupIndex !== false ) {
 335                                         $i = $groupIndex;
 336                                         break;
 337                                 }
 338                         }
 339                 }
 340
 341                 # For now, only go through all this for mysql databases
 342                 if ($wgDBtype != 'mysql') {
 343                         $i = $this->getWriterIndex();
 344                 }
 345                 # Operation-based index
 346                 elseif ( $i == DB_SLAVE ) {
 347                         $i = $this->getReaderIndex();
 348                 } elseif ( $i == DB_MASTER ) {
 349                         $i = $this->getWriterIndex();
 350                 } elseif ( $i == DB_LAST ) {
 351                         # Just use $this->mLastIndex, which should already be set
 352                         $i = $this->mLastIndex;
 353                         if ( $i === -1 ) {
 354                                 # Oh dear, not set, best to use the writer for safety
 355                                 wfDebug( "Warning: DB_LAST used when there was no previous index\n" );
 356                                 $i = $this->getWriterIndex();
 357                         }
 358                 }
 359                 # Couldn't find a working server in getReaderIndex()?
 360                 if ( $i === false ) {
 361                         $this->reportConnectionError( $this->mErrorConnection );
 362                 }
 363                 # Now we have an explicit index into the servers array
 364                 $this->openConnection( $i, $fail );
 365
 366                 wfProfileOut( $fname );
 367                 return $this->mConnections[$i];
 368         }
 369
 370         /**
 371          * Open a connection to the server given by the specified index
 372          * Index must be an actual index into the array
 373          * Returns success
 374          * @access private
 375          */
 376         function openConnection( $i, $fail = false ) {
 377                 $fname = 'LoadBalancer::openConnection';
 378                 wfProfileIn( $fname );
 379                 $success = true;
 380
 381                 if ( !$this->isOpen( $i ) ) {
 382                         $this->mConnections[$i] = $this->reallyOpenConnection( $this->mServers[$i] );
 383                 }
 384
 385                 if ( !$this->isOpen( $i ) ) {
 386                         wfDebug( "Failed to connect to database $i at {$this->mServers[$i]['host']}\n" );
 387                         if ( $fail ) {
 388                                 $this->reportConnectionError( $this->mConnections[$i] );
 389                         }
 390                         $this->mErrorConnection = $this->mConnections[$i];
 391                         $this->mConnections[$i] = false;
 392                         $success = false;
 393                 }
 394                 $this->mLastIndex = $i;
 395                 wfProfileOut( $fname );
 396                 return $success;
 397         }
 398
 399         /**
 400          * Test if the specified index represents an open connection
 401          * @access private
 402          */
 403         function isOpen( $index ) {
 404                 if( !is_integer( $index ) ) {
 405                         return false;
 406                 }
 407                 if ( array_key_exists( $index, $this->mConnections ) && is_object( $this->mConnections[$index] ) &&
 408                   $this->mConnections[$index]->isOpen() )
 409                 {
 410                         return true;
 411                 } else {
 412                         return false;
 413                 }
 414         }
 415
 416         /**
 417          * Really opens a connection
 418          * @access private
 419          */
 420         function reallyOpenConnection( &$server ) {
 421                 if( !is_array( $server ) ) {
 422                         throw new MWException( 'You must update your load-balancing configuration. See DefaultSettings.php entry for $wgDBservers.' );
 423                 }
 424
 425                 extract( $server );
 426                 # Get class for this database type
 427                 $class = 'Database' . ucfirst( $type );
 428
 429                 # Create object
 430                 $db = new $class( $host, $user, $password, $dbname, 1, $flags );
 431                 $db->setLBInfo( $server );
 432                 return $db;
 433         }
 434
 435         function reportConnectionError( &$conn ) {
 436                 $fname = 'LoadBalancer::reportConnectionError';
 437                 wfProfileIn( $fname );
 438                 # Prevent infinite recursion
 439
 440                 static $reporting = false;
 441                 if ( !$reporting ) {
 442                         $reporting = true;
 443                         if ( !is_object( $conn ) ) {
 444                                 // No last connection, probably due to all servers being too busy
 445                                 $conn = new Database;
 446                                 if ( $this->mFailFunction ) {
 447                                         $conn->failFunction( $this->mFailFunction );
 448                                         $conn->reportConnectionError( $this->mLastError );
 449                                 } else {
 450                                         // If all servers were busy, mLastError will contain something sensible
 451                                         throw new DBConnectionError( $conn, $this->mLastError );
 452                                 }
 453                         } else {
 454                                 if ( $this->mFailFunction ) {
 455                                         $conn->failFunction( $this->mFailFunction );
 456                                 } else {
 457                                         $conn->failFunction( false );
 458                                 }
 459                                 $server = $conn->getProperty( 'mServer' );
 460                                 $conn->reportConnectionError( "{$this->mLastError} ({$server})" );
 461                         }
 462                         $reporting = false;
 463                 }
 464                 wfProfileOut( $fname );
 465         }
 466
 467         function getWriterIndex() {
 468                 return 0;
 469         }
 470
 471         /**
 472          * Force subsequent calls to getConnection(DB_SLAVE) to return the
 473          * given index. Set to -1 to restore the original load balancing
 474          * behaviour. I thought this was a good idea when I originally
 475          * wrote this class, but it has never been used.
 476          */
 477         function force( $i ) {
 478                 $this->mForce = $i;
 479         }
 480
 481         /**
 482          * Returns true if the specified index is a valid server index
 483          */
 484         function haveIndex( $i ) {
 485                 return array_key_exists( $i, $this->mServers );
 486         }
 487
 488         /**
 489          * Returns true if the specified index is valid and has non-zero load
 490          */
 491         function isNonZeroLoad( $i ) {
 492                 return array_key_exists( $i, $this->mServers ) && $this->mLoads[$i] != 0;
 493         }
 494
 495         /**
 496          * Get the number of defined servers (not the number of open connections)
 497          */
 498         function getServerCount() {
 499                 return count( $this->mServers );
 500         }
 501
 502         /**
 503          * Save master pos to the session and to memcached, if the session exists
 504          */
 505         function saveMasterPos() {
 506                 if ( session_id() != '' && count( $this->mServers ) > 1 ) {
 507                         # If this entire request was served from a slave without opening a connection to the
 508                         # master (however unlikely that may be), then we can fetch the position from the slave.
 509                         if ( empty( $this->mConnections[0] ) ) {
 510                                 $conn =& $this->getConnection( DB_SLAVE );
 511                                 list( $file, $pos ) = $conn->getSlavePos();
 512                                 wfDebug( "Saving master pos fetched from slave: $file $pos\n" );
 513                         } else {
 514                                 $conn =& $this->getConnection( 0 );
 515                                 list( $file, $pos ) = $conn->getMasterPos();
 516                                 wfDebug( "Saving master pos: $file $pos\n" );
 517                         }
 518                         if ( $file !== false ) {
 519                                 $_SESSION['master_log_file'] = $file;
 520                                 $_SESSION['master_pos'] = $pos;
 521                         }
 522                 }
 523         }
 524
 525         /**
 526          * Loads the master pos from the session, waits for it if necessary
 527          */
 528         function loadMasterPos() {
 529                 if ( isset( $_SESSION['master_log_file'] ) && isset( $_SESSION['master_pos'] ) ) {
 530                         $this->waitFor( $_SESSION['master_log_file'], $_SESSION['master_pos'] );
 531                 }
 532         }
 533
 534         /**
 535          * Close all open connections
 536          */
 537         function closeAll() {
 538                 foreach( $this->mConnections as $i => $conn ) {
 539                         if ( $this->isOpen( $i ) ) {
 540                                 // Need to use this syntax because $conn is a copy not a reference
 541                                 $this->mConnections[$i]->close();
 542                         }
 543                 }
 544         }
 545
 546         function commitAll() {
 547                 foreach( $this->mConnections as $i => $conn ) {
 548                         if ( $this->isOpen( $i ) ) {
 549                                 // Need to use this syntax because $conn is a copy not a reference
 550                                 $this->mConnections[$i]->immediateCommit();
 551                         }
 552                 }
 553         }
 554
 555         /* Issue COMMIT only on master, only if queries were done on connection */
 556         function commitMasterChanges() {
 557                 // Always 0, but who knows.. :)
 558                 $i = $this->getWriterIndex();
 559                 if (array_key_exists($i,$this->mConnections)) {
 560                         if ($this->mConnections[$i]->lastQuery() != '') {
 561                                 $this->mConnections[$i]->immediateCommit();
 562                         }
 563                 }
 564         }
 565
 566         function waitTimeout( $value = NULL ) {
 567                 return wfSetVar( $this->mWaitTimeout, $value );
 568         }
 569
 570         function getLaggedSlaveMode() {
 571                 return $this->mLaggedSlaveMode;
 572         }
 573
 574         /* Disables/enables lag checks */
 575         function allowLagged($mode=null) {
 576                 if ($mode===null)
 577                         return $this->mAllowLagged;
 578                 $this->mAllowLagged=$mode;
 579         }
 580
 581         function pingAll() {
 582                 $success = true;
 583                 foreach ( $this->mConnections as $i => $conn ) {
 584                         if ( $this->isOpen( $i ) ) {
 585                                 if ( !$this->mConnections[$i]->ping() ) {
 586                                         $success = false;
 587                                 }
 588                         }
 589                 }
 590                 return $success;
 591         }
 592
 593         /**
 594          * Get the hostname and lag time of the most-lagged slave
 595          * This is useful for maintenance scripts that need to throttle their updates
 596          */
 597         function getMaxLag() {
 598                 $maxLag = -1;
 599                 $host = '';
 600                 foreach ( $this->mServers as $i => $conn ) {
 601                         if ( $this->openConnection( $i ) ) {
 602                                 $lag = $this->mConnections[$i]->getLag();
 603                                 if ( $lag > $maxLag ) {
 604                                         $maxLag = $lag;
 605                                         $host = $this->mServers[$i]['host'];
 606                                 }
 607                         }
 608                 }
 609                 return array( $host, $maxLag );
 610         }
 611
 612         /**
 613          * Get lag time for each DB
 614          * Results are cached for a short time in memcached
 615          */
 616         function getLagTimes() {
 617                 wfProfileIn( __METHOD__ );
 618                 $expiry = 5;
 619                 $requestRate = 10;
 620
 621                 global $wgMemc;
 622                 $times = $wgMemc->get( wfMemcKey( 'lag_times' ) );
 623                 if ( $times ) {
 624                         # Randomly recache with probability rising over $expiry
 625                         $elapsed = time() - $times['timestamp'];
 626                         $chance = max( 0, ( $expiry - $elapsed ) * $requestRate );
 627                         if ( mt_rand( 0, $chance ) != 0 ) {
 628                                 unset( $times['timestamp'] );
 629                                 wfProfileOut( __METHOD__ );
 630                                 return $times;
 631                         }
 632                         wfIncrStats( 'lag_cache_miss_expired' );
 633                 } else {
 634                         wfIncrStats( 'lag_cache_miss_absent' );
 635                 }
 636
 637                 # Cache key missing or expired
 638
 639                 $times = array();
 640                 foreach ( $this->mServers as $i => $conn ) {
 641                         if ($i==0) { # Master
 642                                 $times[$i] = 0;
 643                         } elseif ( $this->openConnection( $i ) ) {
 644                                 $times[$i] = $this->mConnections[$i]->getLag();
 645                         }
 646                 }
 647
 648                 # Add a timestamp key so we know when it was cached
 649                 $times['timestamp'] = time();
 650                 $wgMemc->set( wfMemcKey( 'lag_times' ), $times, $expiry );
 651
 652                 # But don't give the timestamp to the caller
 653                 unset($times['timestamp']);
 654                 wfProfileOut( __METHOD__ );
 655                 return $times;
 656         }
 657 }
 658
 659