includes/db/loadbalancer/LoadBalancer.php

   1 <?php
   2 /**
   3  * Database load balancing.
   4  *
   5  * This program is free software; you can redistribute it and/or modify
   6  * it under the terms of the GNU General Public License as published by
   7  * the Free Software Foundation; either version 2 of the License, or
   8  * (at your option) any later version.
   9  *
  10  * This program is distributed in the hope that it will be useful,
  11  * but WITHOUT ANY WARRANTY; without even the implied warranty of
  12  * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
  13  * GNU General Public License for more details.
  14  *
  15  * You should have received a copy of the GNU General Public License along
  16  * with this program; if not, write to the Free Software Foundation, Inc.,
  17  * 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
  18  * http://www.gnu.org/copyleft/gpl.html
  19  *
  20  * @file
  21  * @ingroup Database
  22  */
  23
  24 /**
  25  * Database load balancing object
  26  *
  27  * @todo document
  28  * @ingroup Database
  29  */
  30 class LoadBalancer {
  31         /** @var array[] Map of (server index => server config array) */
  32         private $mServers;
  33         /** @var array[] Map of (local/foreignUsed/foreignFree => server index => DatabaseBase array) */
  34         private $mConns;
  35         /** @var array Map of (server index => weight) */
  36         private $mLoads;
  37         /** @var array[] Map of (group => server index => weight) */
  38         private $mGroupLoads;
  39         /** @var bool Whether to disregard slave lag as a factor in slave selection */
  40         private $mAllowLagged;
  41         /** @var integer Seconds to spend waiting on slave lag to resolve */
  42         private $mWaitTimeout;
  43
  44         /** @var array LBFactory information */
  45         private $mParentInfo;
  46         /** @var string The LoadMonitor subclass name */
  47         private $mLoadMonitorClass;
  48         /** @var LoadMonitor */
  49         private $mLoadMonitor;
  50
  51         /** @var bool|DatabaseBase Database connection that caused a problem */
  52         private $mErrorConnection;
  53         /** @var integer The generic (not query grouped) slave index (of $mServers) */
  54         private $mReadIndex;
  55         /** @var bool|DBMasterPos False if not set */
  56         private $mWaitForPos;
  57         /** @var bool Whether the generic reader fell back to a lagged slave */
  58         private $mLaggedSlaveMode;
  59         /** @var string The last DB selection or connection error */
  60         private $mLastError = 'Unknown error';
  61         /** @var integer Total connections opened */
  62         private $connsOpened = 0;
  63
  64         /** @var integer Warn when this many connection are held */
  65         const CONN_HELD_WARN_THRESHOLD = 10;
  66         /** @var integer Default 'max lag' when unspecified */
  67         const MAX_LAG = 30;
  68
  69         /**
  70          * @param array $params Array with keys:
  71          *   servers           Required. Array of server info structures.
  72          *   loadMonitor       Name of a class used to fetch server lag and load.
  73          * @throws MWException
  74          */
  75         public function __construct( array $params ) {
  76                 if ( !isset( $params['servers'] ) ) {
  77                         throw new MWException( __CLASS__ . ': missing servers parameter' );
  78                 }
  79                 $this->mServers = $params['servers'];
  80                 $this->mWaitTimeout = 10;
  81
  82                 $this->mReadIndex = -1;
  83                 $this->mWriteIndex = -1;
  84                 $this->mConns = array(
  85                         'local' => array(),
  86                         'foreignUsed' => array(),
  87                         'foreignFree' => array() );
  88                 $this->mLoads = array();
  89                 $this->mWaitForPos = false;
  90                 $this->mLaggedSlaveMode = false;
  91                 $this->mErrorConnection = false;
  92                 $this->mAllowLagged = false;
  93
  94                 if ( isset( $params['loadMonitor'] ) ) {
  95                         $this->mLoadMonitorClass = $params['loadMonitor'];
  96                 } else {
  97                         $master = reset( $params['servers'] );
  98                         if ( isset( $master['type'] ) && $master['type'] === 'mysql' ) {
  99                                 $this->mLoadMonitorClass = 'LoadMonitorMySQL';
 100                         } else {
 101                                 $this->mLoadMonitorClass = 'LoadMonitorNull';
 102                         }
 103                 }
 104
 105                 foreach ( $params['servers'] as $i => $server ) {
 106                         $this->mLoads[$i] = $server['load'];
 107                         if ( isset( $server['groupLoads'] ) ) {
 108                                 foreach ( $server['groupLoads'] as $group => $ratio ) {
 109                                         if ( !isset( $this->mGroupLoads[$group] ) ) {
 110                                                 $this->mGroupLoads[$group] = array();
 111                                         }
 112                                         $this->mGroupLoads[$group][$i] = $ratio;
 113                                 }
 114                         }
 115                 }
 116         }
 117
 118         /**
 119          * Get a LoadMonitor instance
 120          *
 121          * @return LoadMonitor
 122          */
 123         private function getLoadMonitor() {
 124                 if ( !isset( $this->mLoadMonitor ) ) {
 125                         $class = $this->mLoadMonitorClass;
 126                         $this->mLoadMonitor = new $class( $this );
 127                 }
 128
 129                 return $this->mLoadMonitor;
 130         }
 131
 132         /**
 133          * Get or set arbitrary data used by the parent object, usually an LBFactory
 134          * @param mixed $x
 135          * @return mixed
 136          */
 137         public function parentInfo( $x = null ) {
 138                 return wfSetVar( $this->mParentInfo, $x );
 139         }
 140
 141         /**
 142          * Given an array of non-normalised probabilities, this function will select
 143          * an element and return the appropriate key
 144          *
 145          * @deprecated since 1.21, use ArrayUtils::pickRandom()
 146          *
 147          * @param array $weights
 148          * @return bool|int|string
 149          */
 150         public function pickRandom( array $weights ) {
 151                 return ArrayUtils::pickRandom( $weights );
 152         }
 153
 154         /**
 155          * @param array $loads
 156          * @param bool|string $wiki Wiki to get non-lagged for
 157          * @param float $maxLag Restrict the maximum allowed lag to this many seconds
 158          * @return bool|int|string
 159          */
 160         private function getRandomNonLagged( array $loads, $wiki = false, $maxLag = self::MAX_LAG ) {
 161                 $lags = $this->getLagTimes( $wiki );
 162
 163                 # Unset excessively lagged servers
 164                 foreach ( $lags as $i => $lag ) {
 165                         if ( $i != 0 ) {
 166                                 $maxServerLag = $maxLag;
 167                                 if ( isset( $this->mServers[$i]['max lag'] ) ) {
 168                                         $maxServerLag = min( $maxServerLag, $this->mServers[$i]['max lag'] );
 169                                 }
 170                                 if ( $lag === false ) {
 171                                         wfDebugLog( 'replication', "Server #$i is not replicating" );
 172                                         unset( $loads[$i] );
 173                                 } elseif ( $lag > $maxServerLag ) {
 174                                         wfDebugLog( 'replication', "Server #$i is excessively lagged ($lag seconds)" );
 175                                         unset( $loads[$i] );
 176                                 }
 177                         }
 178                 }
 179
 180                 # Find out if all the slaves with non-zero load are lagged
 181                 $sum = 0;
 182                 foreach ( $loads as $load ) {
 183                         $sum += $load;
 184                 }
 185                 if ( $sum == 0 ) {
 186                         # No appropriate DB servers except maybe the master and some slaves with zero load
 187                         # Do NOT use the master
 188                         # Instead, this function will return false, triggering read-only mode,
 189                         # and a lagged slave will be used instead.
 190                         return false;
 191                 }
 192
 193                 if ( count( $loads ) == 0 ) {
 194                         return false;
 195                 }
 196
 197                 # wfDebugLog( 'connect', var_export( $loads, true ) );
 198
 199                 # Return a random representative of the remainder
 200                 return ArrayUtils::pickRandom( $loads );
 201         }
 202
 203         /**
 204          * Get the index of the reader connection, which may be a slave
 205          * This takes into account load ratios and lag times. It should
 206          * always return a consistent index during a given invocation
 207          *
 208          * Side effect: opens connections to databases
 209          * @param string|bool $group Query group, or false for the generic reader
 210          * @param string|bool $wiki Wiki ID, or false for the current wiki
 211          * @throws MWException
 212          * @return bool|int|string
 213          */
 214         public function getReaderIndex( $group = false, $wiki = false ) {
 215                 global $wgDBtype;
 216
 217                 # @todo FIXME: For now, only go through all this for mysql databases
 218                 if ( $wgDBtype != 'mysql' ) {
 219                         return $this->getWriterIndex();
 220                 }
 221
 222                 if ( count( $this->mServers ) == 1 ) {
 223                         # Skip the load balancing if there's only one server
 224                         return 0;
 225                 } elseif ( $group === false && $this->mReadIndex >= 0 ) {
 226                         # Shortcut if generic reader exists already
 227                         return $this->mReadIndex;
 228                 }
 229
 230                 # Find the relevant load array
 231                 if ( $group !== false ) {
 232                         if ( isset( $this->mGroupLoads[$group] ) ) {
 233                                 $nonErrorLoads = $this->mGroupLoads[$group];
 234                         } else {
 235                                 # No loads for this group, return false and the caller can use some other group
 236                                 wfDebug( __METHOD__ . ": no loads for group $group\n" );
 237
 238                                 return false;
 239                         }
 240                 } else {
 241                         $nonErrorLoads = $this->mLoads;
 242                 }
 243
 244                 if ( !count( $nonErrorLoads ) ) {
 245                         throw new MWException( "Empty server array given to LoadBalancer" );
 246                 }
 247
 248                 # Scale the configured load ratios according to the dynamic load (if the load monitor supports it)
 249                 $this->getLoadMonitor()->scaleLoads( $nonErrorLoads, $group, $wiki );
 250
 251                 $laggedSlaveMode = false;
 252
 253                 # No server found yet
 254                 $i = false;
 255                 $conn = false;
 256                 # First try quickly looking through the available servers for a server that
 257                 # meets our criteria
 258                 $currentLoads = $nonErrorLoads;
 259                 while ( count( $currentLoads ) ) {
 260                         if ( $this->mAllowLagged || $laggedSlaveMode ) {
 261                                 $i = ArrayUtils::pickRandom( $currentLoads );
 262                         } else {
 263                                 $i = false;
 264                                 if ( $this->mWaitForPos && $this->mWaitForPos->asOfTime() ) {
 265                                         # ChronologyProtecter causes mWaitForPos to be set via sessions.
 266                                         # This triggers doWait() after connect, so it's especially good to
 267                                         # avoid lagged servers so as to avoid just blocking in that method.
 268                                         $ago = microtime( true ) - $this->mWaitForPos->asOfTime();
 269                                         # Aim for <= 1 second of waiting (being too picky can backfire)
 270                                         $i = $this->getRandomNonLagged( $currentLoads, $wiki, $ago + 1 );
 271                                 }
 272                                 if ( $i === false ) {
 273                                         # Any server with less lag than it's 'max lag' param is preferable
 274                                         $i = $this->getRandomNonLagged( $currentLoads, $wiki );
 275                                 }
 276                                 if ( $i === false && count( $currentLoads ) != 0 ) {
 277                                         # All slaves lagged. Switch to read-only mode
 278                                         wfDebugLog( 'replication', "All slaves lagged. Switch to read-only mode" );
 279                                         $i = ArrayUtils::pickRandom( $currentLoads );
 280                                         $laggedSlaveMode = true;
 281                                 }
 282                         }
 283
 284                         if ( $i === false ) {
 285                                 # pickRandom() returned false
 286                                 # This is permanent and means the configuration or the load monitor
 287                                 # wants us to return false.
 288                                 wfDebugLog( 'connect', __METHOD__ . ": pickRandom() returned false" );
 289
 290                                 return false;
 291                         }
 292
 293                         $serverName = $this->getServerName( $i );
 294                         wfDebugLog( 'connect', __METHOD__ . ": Using reader #$i: $serverName..." );
 295
 296                         $conn = $this->openConnection( $i, $wiki );
 297                         if ( !$conn ) {
 298                                 wfDebugLog( 'connect', __METHOD__ . ": Failed connecting to $i/$wiki" );
 299                                 unset( $nonErrorLoads[$i] );
 300                                 unset( $currentLoads[$i] );
 301                                 $i = false;
 302                                 continue;
 303                         }
 304
 305                         // Decrement reference counter, we are finished with this connection.
 306                         // It will be incremented for the caller later.
 307                         if ( $wiki !== false ) {
 308                                 $this->reuseConnection( $conn );
 309                         }
 310
 311                         # Return this server
 312                         break;
 313                 }
 314
 315                 # If all servers were down, quit now
 316                 if ( !count( $nonErrorLoads ) ) {
 317                         wfDebugLog( 'connect', "All servers down" );
 318                 }
 319
 320                 if ( $i !== false ) {
 321                         # Slave connection successful
 322                         # Wait for the session master pos for a short time
 323                         if ( $this->mWaitForPos && $i > 0 ) {
 324                                 if ( !$this->doWait( $i ) ) {
 325                                         $this->mServers[$i]['slave pos'] = $conn->getSlavePos();
 326                                 }
 327                         }
 328                         if ( $this->mReadIndex <= 0 && $this->mLoads[$i] > 0 && $group === false ) {
 329                                 $this->mReadIndex = $i;
 330                                 # Record if the generic reader index is in "lagged slave" mode
 331                                 if ( $laggedSlaveMode ) {
 332                                         $this->mLaggedSlaveMode = true;
 333                                 }
 334                         }
 335                         $serverName = $this->getServerName( $i );
 336                         wfDebug( __METHOD__ . ": using server $serverName for group '$group'\n" );
 337                 }
 338
 339                 return $i;
 340         }
 341
 342         /**
 343          * Set the master wait position
 344          * If a DB_SLAVE connection has been opened already, waits
 345          * Otherwise sets a variable telling it to wait if such a connection is opened
 346          * @param DBMasterPos $pos
 347          */
 348         public function waitFor( $pos ) {
 349                 $this->mWaitForPos = $pos;
 350                 $i = $this->mReadIndex;
 351
 352                 if ( $i > 0 ) {
 353                         if ( !$this->doWait( $i ) ) {
 354                                 $this->mServers[$i]['slave pos'] = $this->getAnyOpenConnection( $i )->getSlavePos();
 355                                 $this->mLaggedSlaveMode = true;
 356                         }
 357                 }
 358         }
 359
 360         /**
 361          * Set the master wait position and wait for a "generic" slave to catch up to it
 362          *
 363          * This can be used a faster proxy for waitForAll()
 364          *
 365          * @param DBMasterPos $pos
 366          * @param int $timeout Max seconds to wait; default is mWaitTimeout
 367          * @return bool Success (able to connect and no timeouts reached)
 368          * @since 1.26
 369          */
 370         public function waitForOne( $pos, $timeout = null ) {
 371                 $this->mWaitForPos = $pos;
 372
 373                 $i = $this->mReadIndex;
 374                 if ( $i <= 0 ) {
 375                         // Pick a generic slave if there isn't one yet
 376                         $readLoads = $this->mLoads;
 377                         unset( $readLoads[$this->getWriterIndex()] ); // slaves only
 378                         $readLoads = array_filter( $readLoads ); // with non-zero load
 379                         $i = ArrayUtils::pickRandom( $readLoads );
 380                 }
 381
 382                 if ( $i > 0 ) {
 383                         $ok = $this->doWait( $i, true, $timeout );
 384                 } else {
 385                         $ok = true; // no applicable loads
 386                 }
 387
 388                 return $ok;
 389         }
 390
 391         /**
 392          * Set the master wait position and wait for ALL slaves to catch up to it
 393          * @param DBMasterPos $pos
 394          * @param int $timeout Max seconds to wait; default is mWaitTimeout
 395          * @return bool Success (able to connect and no timeouts reached)
 396          */
 397         public function waitForAll( $pos, $timeout = null ) {
 398                 $this->mWaitForPos = $pos;
 399                 $serverCount = count( $this->mServers );
 400
 401                 $ok = true;
 402                 for ( $i = 1; $i < $serverCount; $i++ ) {
 403                         if ( $this->mLoads[$i] > 0 ) {
 404                                 $ok = $this->doWait( $i, true, $timeout ) && $ok;
 405                         }
 406                 }
 407
 408                 return $ok;
 409         }
 410
 411         /**
 412          * Get any open connection to a given server index, local or foreign
 413          * Returns false if there is no connection open
 414          *
 415          * @param int $i
 416          * @return DatabaseBase|bool False on failure
 417          */
 418         public function getAnyOpenConnection( $i ) {
 419                 foreach ( $this->mConns as $conns ) {
 420                         if ( !empty( $conns[$i] ) ) {
 421                                 return reset( $conns[$i] );
 422                         }
 423                 }
 424
 425                 return false;
 426         }
 427
 428         /**
 429          * Wait for a given slave to catch up to the master pos stored in $this
 430          * @param int $index Server index
 431          * @param bool $open Check the server even if a new connection has to be made
 432          * @param int $timeout Max seconds to wait; default is mWaitTimeout
 433          * @return bool
 434          */
 435         protected function doWait( $index, $open = false, $timeout = null ) {
 436                 $close = false; // close the connection afterwards
 437
 438                 # Find a connection to wait on, creating one if needed and allowed
 439                 $conn = $this->getAnyOpenConnection( $index );
 440                 if ( !$conn ) {
 441                         if ( !$open ) {
 442                                 wfDebug( __METHOD__ . ": no connection open\n" );
 443
 444                                 return false;
 445                         } else {
 446                                 $conn = $this->openConnection( $index, '' );
 447                                 if ( !$conn ) {
 448                                         wfDebug( __METHOD__ . ": failed to open connection\n" );
 449
 450                                         return false;
 451                                 }
 452                                 // Avoid connection spam in waitForAll() when connections
 453                                 // are made just for the sake of doing this lag check.
 454                                 $close = true;
 455                         }
 456                 }
 457
 458                 wfDebug( __METHOD__ . ": Waiting for slave #$index to catch up...\n" );
 459                 $timeout = $timeout ?: $this->mWaitTimeout;
 460                 $result = $conn->masterPosWait( $this->mWaitForPos, $timeout );
 461
 462                 if ( $result == -1 || is_null( $result ) ) {
 463                         # Timed out waiting for slave, use master instead
 464                         $server = $server = $this->getServerName( $index );
 465                         $msg = __METHOD__ . ": Timed out waiting on $server pos {$this->mWaitForPos}";
 466                         wfDebug( "$msg\n" );
 467                         wfDebugLog( 'DBPerformance', "$msg:\n" . wfBacktrace( true ) );
 468                         $ok = false;
 469                 } else {
 470                         wfDebug( __METHOD__ . ": Done\n" );
 471                         $ok = true;
 472                 }
 473
 474                 if ( $close ) {
 475                         $this->closeConnection( $conn );
 476                 }
 477
 478                 return $ok;
 479         }
 480
 481         /**
 482          * Get a connection by index
 483          * This is the main entry point for this class.
 484          *
 485          * @param int $i Server index
 486          * @param array|string|bool $groups Query group(s), or false for the generic reader
 487          * @param string|bool $wiki Wiki ID, or false for the current wiki
 488          *
 489          * @throws MWException
 490          * @return DatabaseBase
 491          */
 492         public function getConnection( $i, $groups = array(), $wiki = false ) {
 493                 if ( $i === null || $i === false ) {
 494                         throw new MWException( 'Attempt to call ' . __METHOD__ .
 495                                 ' with invalid server index' );
 496                 }
 497
 498                 if ( $wiki === wfWikiID() ) {
 499                         $wiki = false;
 500                 }
 501
 502                 $groups = ( $groups === false || $groups === array() )
 503                         ? array( false ) // check one "group": the generic pool
 504                         : (array)$groups;
 505
 506                 $masterOnly = ( $i == DB_MASTER || $i == $this->getWriterIndex() );
 507                 $oldConnsOpened = $this->connsOpened; // connections open now
 508
 509                 if ( $i == DB_MASTER ) {
 510                         $i = $this->getWriterIndex();
 511                 } else {
 512                         # Try to find an available server in any the query groups (in order)
 513                         foreach ( $groups as $group ) {
 514                                 $groupIndex = $this->getReaderIndex( $group, $wiki );
 515                                 if ( $groupIndex !== false ) {
 516                                         $i = $groupIndex;
 517                                         break;
 518                                 }
 519                         }
 520                 }
 521
 522                 # Operation-based index
 523                 if ( $i == DB_SLAVE ) {
 524                         $this->mLastError = 'Unknown error'; // reset error string
 525                         # Try the general server pool if $groups are unavailable.
 526                         $i = in_array( false, $groups, true )
 527                                 ? false // don't bother with this if that is what was tried above
 528                                 : $this->getReaderIndex( false, $wiki );
 529                         # Couldn't find a working server in getReaderIndex()?
 530                         if ( $i === false ) {
 531                                 $this->mLastError = 'No working slave server: ' . $this->mLastError;
 532
 533                                 return $this->reportConnectionError();
 534                         }
 535                 }
 536
 537                 # Now we have an explicit index into the servers array
 538                 $conn = $this->openConnection( $i, $wiki );
 539                 if ( !$conn ) {
 540                         return $this->reportConnectionError();
 541                 }
 542
 543                 # Profile any new connections that happen
 544                 if ( $this->connsOpened > $oldConnsOpened ) {
 545                         $host = $conn->getServer();
 546                         $dbname = $conn->getDBname();
 547                         $trxProf = Profiler::instance()->getTransactionProfiler();
 548                         $trxProf->recordConnection( $host, $dbname, $masterOnly );
 549                 }
 550
 551                 # Make master connections read only if in lagged slave mode
 552                 if ( $masterOnly && $this->getServerCount() > 1 && $this->getLaggedSlaveMode() ) {
 553                         $conn->setLBInfo( 'readOnlyReason',
 554                                 'The database has been automatically locked ' .
 555                                 'while the slave database servers catch up to the master'
 556                         );
 557                 }
 558
 559                 return $conn;
 560         }
 561
 562         /**
 563          * Mark a foreign connection as being available for reuse under a different
 564          * DB name or prefix. This mechanism is reference-counted, and must be called
 565          * the same number of times as getConnection() to work.
 566          *
 567          * @param DatabaseBase $conn
 568          * @throws MWException
 569          */
 570         public function reuseConnection( $conn ) {
 571                 $serverIndex = $conn->getLBInfo( 'serverIndex' );
 572                 $refCount = $conn->getLBInfo( 'foreignPoolRefCount' );
 573                 if ( $serverIndex === null || $refCount === null ) {
 574                         wfDebug( __METHOD__ . ": this connection was not opened as a foreign connection\n" );
 575
 576                         /**
 577                          * This can happen in code like:
 578                          *   foreach ( $dbs as $db ) {
 579                          *     $conn = $lb->getConnection( DB_SLAVE, array(), $db );
 580                          *     ...
 581                          *     $lb->reuseConnection( $conn );
 582                          *   }
 583                          * When a connection to the local DB is opened in this way, reuseConnection()
 584                          * should be ignored
 585                          */
 586
 587                         return;
 588                 }
 589
 590                 $dbName = $conn->getDBname();
 591                 $prefix = $conn->tablePrefix();
 592                 if ( strval( $prefix ) !== '' ) {
 593                         $wiki = "$dbName-$prefix";
 594                 } else {
 595                         $wiki = $dbName;
 596                 }
 597                 if ( $this->mConns['foreignUsed'][$serverIndex][$wiki] !== $conn ) {
 598                         throw new MWException( __METHOD__ . ": connection not found, has " .
 599                                 "the connection been freed already?" );
 600                 }
 601                 $conn->setLBInfo( 'foreignPoolRefCount', --$refCount );
 602                 if ( $refCount <= 0 ) {
 603                         $this->mConns['foreignFree'][$serverIndex][$wiki] = $conn;
 604                         unset( $this->mConns['foreignUsed'][$serverIndex][$wiki] );
 605                         wfDebug( __METHOD__ . ": freed connection $serverIndex/$wiki\n" );
 606                 } else {
 607                         wfDebug( __METHOD__ . ": reference count for $serverIndex/$wiki reduced to $refCount\n" );
 608                 }
 609         }
 610
 611         /**
 612          * Get a database connection handle reference
 613          *
 614          * The handle's methods wrap simply wrap those of a DatabaseBase handle
 615          *
 616          * @see LoadBalancer::getConnection() for parameter information
 617          *
 618          * @param int $db
 619          * @param array|string|bool $groups Query group(s), or false for the generic reader
 620          * @param string|bool $wiki Wiki ID, or false for the current wiki
 621          * @return DBConnRef
 622          */
 623         public function getConnectionRef( $db, $groups = array(), $wiki = false ) {
 624                 return new DBConnRef( $this, $this->getConnection( $db, $groups, $wiki ) );
 625         }
 626
 627         /**
 628          * Get a database connection handle reference without connecting yet
 629          *
 630          * The handle's methods wrap simply wrap those of a DatabaseBase handle
 631          *
 632          * @see LoadBalancer::getConnection() for parameter information
 633          *
 634          * @param int $db
 635          * @param array|string|bool $groups Query group(s), or false for the generic reader
 636          * @param string|bool $wiki Wiki ID, or false for the current wiki
 637          * @return DBConnRef
 638          */
 639         public function getLazyConnectionRef( $db, $groups = array(), $wiki = false ) {
 640                 return new DBConnRef( $this, array( $db, $groups, $wiki ) );
 641         }
 642
 643         /**
 644          * Open a connection to the server given by the specified index
 645          * Index must be an actual index into the array.
 646          * If the server is already open, returns it.
 647          *
 648          * On error, returns false, and the connection which caused the
 649          * error will be available via $this->mErrorConnection.
 650          *
 651          * @param int $i Server index
 652          * @param string|bool $wiki Wiki ID, or false for the current wiki
 653          * @return DatabaseBase
 654          *
 655          * @access private
 656          */
 657         public function openConnection( $i, $wiki = false ) {
 658                 if ( $wiki !== false ) {
 659                         $conn = $this->openForeignConnection( $i, $wiki );
 660                 } elseif ( isset( $this->mConns['local'][$i][0] ) ) {
 661                         $conn = $this->mConns['local'][$i][0];
 662                 } else {
 663                         $server = $this->mServers[$i];
 664                         $server['serverIndex'] = $i;
 665                         $conn = $this->reallyOpenConnection( $server, false );
 666                         $serverName = $this->getServerName( $i );
 667                         if ( $conn->isOpen() ) {
 668                                 wfDebug( "Connected to database $i at $serverName\n" );
 669                                 $this->mConns['local'][$i][0] = $conn;
 670                         } else {
 671                                 wfDebug( "Failed to connect to database $i at $serverName\n" );
 672                                 $this->mErrorConnection = $conn;
 673                                 $conn = false;
 674                         }
 675                 }
 676
 677                 if ( $conn && !$conn->isOpen() ) {
 678                         // Connection was made but later unrecoverably lost for some reason.
 679                         // Do not return a handle that will just throw exceptions on use,
 680                         // but let the calling code (e.g. getReaderIndex) try another server.
 681                         // See DatabaseMyslBase::ping() for how this can happen.
 682                         $this->mErrorConnection = $conn;
 683                         $conn = false;
 684                 }
 685
 686                 return $conn;
 687         }
 688
 689         /**
 690          * Open a connection to a foreign DB, or return one if it is already open.
 691          *
 692          * Increments a reference count on the returned connection which locks the
 693          * connection to the requested wiki. This reference count can be
 694          * decremented by calling reuseConnection().
 695          *
 696          * If a connection is open to the appropriate server already, but with the wrong
 697          * database, it will be switched to the right database and returned, as long as
 698          * it has been freed first with reuseConnection().
 699          *
 700          * On error, returns false, and the connection which caused the
 701          * error will be available via $this->mErrorConnection.
 702          *
 703          * @param int $i Server index
 704          * @param string $wiki Wiki ID to open
 705          * @return DatabaseBase
 706          */
 707         private function openForeignConnection( $i, $wiki ) {
 708                 list( $dbName, $prefix ) = wfSplitWikiID( $wiki );
 709                 if ( isset( $this->mConns['foreignUsed'][$i][$wiki] ) ) {
 710                         // Reuse an already-used connection
 711                         $conn = $this->mConns['foreignUsed'][$i][$wiki];
 712                         wfDebug( __METHOD__ . ": reusing connection $i/$wiki\n" );
 713                 } elseif ( isset( $this->mConns['foreignFree'][$i][$wiki] ) ) {
 714                         // Reuse a free connection for the same wiki
 715                         $conn = $this->mConns['foreignFree'][$i][$wiki];
 716                         unset( $this->mConns['foreignFree'][$i][$wiki] );
 717                         $this->mConns['foreignUsed'][$i][$wiki] = $conn;
 718                         wfDebug( __METHOD__ . ": reusing free connection $i/$wiki\n" );
 719                 } elseif ( !empty( $this->mConns['foreignFree'][$i] ) ) {
 720                         // Reuse a connection from another wiki
 721                         $conn = reset( $this->mConns['foreignFree'][$i] );
 722                         $oldWiki = key( $this->mConns['foreignFree'][$i] );
 723
 724                         // The empty string as a DB name means "don't care".
 725                         // DatabaseMysqlBase::open() already handle this on connection.
 726                         if ( $dbName !== '' && !$conn->selectDB( $dbName ) ) {
 727                                 $this->mLastError = "Error selecting database $dbName on server " .
 728                                         $conn->getServer() . " from client host " . wfHostname() . "\n";
 729                                 $this->mErrorConnection = $conn;
 730                                 $conn = false;
 731                         } else {
 732                                 $conn->tablePrefix( $prefix );
 733                                 unset( $this->mConns['foreignFree'][$i][$oldWiki] );
 734                                 $this->mConns['foreignUsed'][$i][$wiki] = $conn;
 735                                 wfDebug( __METHOD__ . ": reusing free connection from $oldWiki for $wiki\n" );
 736                         }
 737                 } else {
 738                         // Open a new connection
 739                         $server = $this->mServers[$i];
 740                         $server['serverIndex'] = $i;
 741                         $server['foreignPoolRefCount'] = 0;
 742                         $server['foreign'] = true;
 743                         $conn = $this->reallyOpenConnection( $server, $dbName );
 744                         if ( !$conn->isOpen() ) {
 745                                 wfDebug( __METHOD__ . ": error opening connection for $i/$wiki\n" );
 746                                 $this->mErrorConnection = $conn;
 747                                 $conn = false;
 748                         } else {
 749                                 $conn->tablePrefix( $prefix );
 750                                 $this->mConns['foreignUsed'][$i][$wiki] = $conn;
 751                                 wfDebug( __METHOD__ . ": opened new connection for $i/$wiki\n" );
 752                         }
 753                 }
 754
 755                 // Increment reference count
 756                 if ( $conn ) {
 757                         $refCount = $conn->getLBInfo( 'foreignPoolRefCount' );
 758                         $conn->setLBInfo( 'foreignPoolRefCount', $refCount + 1 );
 759                 }
 760
 761                 return $conn;
 762         }
 763
 764         /**
 765          * Test if the specified index represents an open connection
 766          *
 767          * @param int $index Server index
 768          * @access private
 769          * @return bool
 770          */
 771         private function isOpen( $index ) {
 772                 if ( !is_integer( $index ) ) {
 773                         return false;
 774                 }
 775
 776                 return (bool)$this->getAnyOpenConnection( $index );
 777         }
 778
 779         /**
 780          * Really opens a connection. Uncached.
 781          * Returns a Database object whether or not the connection was successful.
 782          * @access private
 783          *
 784          * @param array $server
 785          * @param bool $dbNameOverride
 786          * @throws MWException
 787          * @return DatabaseBase
 788          */
 789         protected function reallyOpenConnection( $server, $dbNameOverride = false ) {
 790                 if ( !is_array( $server ) ) {
 791                         throw new MWException( 'You must update your load-balancing configuration. ' .
 792                                 'See DefaultSettings.php entry for $wgDBservers.' );
 793                 }
 794
 795                 if ( $dbNameOverride !== false ) {
 796                         $server['dbname'] = $dbNameOverride;
 797                 }
 798
 799                 // Log when many connection are made on requests
 800                 if ( ++$this->connsOpened >= self::CONN_HELD_WARN_THRESHOLD ) {
 801                         $masterAddr = $this->getServerName( 0 );
 802                         wfDebugLog( 'DBPerformance', __METHOD__ . ": " .
 803                                 "{$this->connsOpened}+ connections made (master=$masterAddr)\n" .
 804                                 wfBacktrace( true ) );
 805                 }
 806
 807                 # Create object
 808                 try {
 809                         $db = DatabaseBase::factory( $server['type'], $server );
 810                 } catch ( DBConnectionError $e ) {
 811                         // FIXME: This is probably the ugliest thing I have ever done to
 812                         // PHP. I'm half-expecting it to segfault, just out of disgust. -- TS
 813                         $db = $e->db;
 814                 }
 815
 816                 $db->setLBInfo( $server );
 817
 818                 return $db;
 819         }
 820
 821         /**
 822          * @throws DBConnectionError
 823          * @return bool
 824          */
 825         private function reportConnectionError() {
 826                 $conn = $this->mErrorConnection; // The connection which caused the error
 827                 $context = array(
 828                         'method' => __METHOD__,
 829                         'last_error' => $this->mLastError,
 830                 );
 831
 832                 if ( !is_object( $conn ) ) {
 833                         // No last connection, probably due to all servers being too busy
 834                         wfLogDBError(
 835                                 "LB failure with no last connection. Connection error: {last_error}",
 836                                 $context
 837                         );
 838
 839                         // If all servers were busy, mLastError will contain something sensible
 840                         throw new DBConnectionError( null, $this->mLastError );
 841                 } else {
 842                         $context['db_server'] = $conn->getProperty( 'mServer' );
 843                         wfLogDBError(
 844                                 "Connection error: {last_error} ({db_server})",
 845                                 $context
 846                         );
 847
 848                         // throws DBConnectionError
 849                         $conn->reportConnectionError( "{$this->mLastError} ({$context['db_server']})" );
 850                 }
 851
 852                 return false; /* not reached */
 853         }
 854
 855         /**
 856          * @return int
 857          * @since 1.26
 858          */
 859         public function getWriterIndex() {
 860                 return 0;
 861         }
 862
 863         /**
 864          * Returns true if the specified index is a valid server index
 865          *
 866          * @param string $i
 867          * @return bool
 868          */
 869         public function haveIndex( $i ) {
 870                 return array_key_exists( $i, $this->mServers );
 871         }
 872
 873         /**
 874          * Returns true if the specified index is valid and has non-zero load
 875          *
 876          * @param string $i
 877          * @return bool
 878          */
 879         public function isNonZeroLoad( $i ) {
 880                 return array_key_exists( $i, $this->mServers ) && $this->mLoads[$i] != 0;
 881         }
 882
 883         /**
 884          * Get the number of defined servers (not the number of open connections)
 885          *
 886          * @return int
 887          */
 888         public function getServerCount() {
 889                 return count( $this->mServers );
 890         }
 891
 892         /**
 893          * Get the host name or IP address of the server with the specified index
 894          * Prefer a readable name if available.
 895          * @param string $i
 896          * @return string
 897          */
 898         public function getServerName( $i ) {
 899                 if ( isset( $this->mServers[$i]['hostName'] ) ) {
 900                         $name = $this->mServers[$i]['hostName'];
 901                 } elseif ( isset( $this->mServers[$i]['host'] ) ) {
 902                         $name = $this->mServers[$i]['host'];
 903                 } else {
 904                         $name = '';
 905                 }
 906
 907                 return ( $name != '' ) ? $name : 'localhost';
 908         }
 909
 910         /**
 911          * Return the server info structure for a given index, or false if the index is invalid.
 912          * @param int $i
 913          * @return array|bool
 914          */
 915         public function getServerInfo( $i ) {
 916                 if ( isset( $this->mServers[$i] ) ) {
 917                         return $this->mServers[$i];
 918                 } else {
 919                         return false;
 920                 }
 921         }
 922
 923         /**
 924          * Sets the server info structure for the given index. Entry at index $i
 925          * is created if it doesn't exist
 926          * @param int $i
 927          * @param array $serverInfo
 928          */
 929         public function setServerInfo( $i, array $serverInfo ) {
 930                 $this->mServers[$i] = $serverInfo;
 931         }
 932
 933         /**
 934          * Get the current master position for chronology control purposes
 935          * @return mixed
 936          */
 937         public function getMasterPos() {
 938                 # If this entire request was served from a slave without opening a connection to the
 939                 # master (however unlikely that may be), then we can fetch the position from the slave.
 940                 $masterConn = $this->getAnyOpenConnection( 0 );
 941                 if ( !$masterConn ) {
 942                         $serverCount = count( $this->mServers );
 943                         for ( $i = 1; $i < $serverCount; $i++ ) {
 944                                 $conn = $this->getAnyOpenConnection( $i );
 945                                 if ( $conn ) {
 946                                         wfDebug( "Master pos fetched from slave\n" );
 947
 948                                         return $conn->getSlavePos();
 949                                 }
 950                         }
 951                 } else {
 952                         wfDebug( "Master pos fetched from master\n" );
 953
 954                         return $masterConn->getMasterPos();
 955                 }
 956
 957                 return false;
 958         }
 959
 960         /**
 961          * Close all open connections
 962          */
 963         public function closeAll() {
 964                 foreach ( $this->mConns as $conns2 ) {
 965                         foreach ( $conns2 as $conns3 ) {
 966                                 /** @var DatabaseBase $conn */
 967                                 foreach ( $conns3 as $conn ) {
 968                                         $conn->close();
 969                                 }
 970                         }
 971                 }
 972                 $this->mConns = array(
 973                         'local' => array(),
 974                         'foreignFree' => array(),
 975                         'foreignUsed' => array(),
 976                 );
 977                 $this->connsOpened = 0;
 978         }
 979
 980         /**
 981          * Close a connection
 982          * Using this function makes sure the LoadBalancer knows the connection is closed.
 983          * If you use $conn->close() directly, the load balancer won't update its state.
 984          * @param DatabaseBase $conn
 985          */
 986         public function closeConnection( $conn ) {
 987                 $done = false;
 988                 foreach ( $this->mConns as $i1 => $conns2 ) {
 989                         foreach ( $conns2 as $i2 => $conns3 ) {
 990                                 foreach ( $conns3 as $i3 => $candidateConn ) {
 991                                         if ( $conn === $candidateConn ) {
 992                                                 $conn->close();
 993                                                 unset( $this->mConns[$i1][$i2][$i3] );
 994                                                 --$this->connsOpened;
 995                                                 $done = true;
 996                                                 break;
 997                                         }
 998                                 }
 999                         }
1000                 }
1001                 if ( !$done ) {
1002                         $conn->close();
1003                 }
1004         }
1005
1006         /**
1007          * Commit transactions on all open connections
1008          */
1009         public function commitAll() {
1010                 foreach ( $this->mConns as $conns2 ) {
1011                         foreach ( $conns2 as $conns3 ) {
1012                                 /** @var DatabaseBase[] $conns3 */
1013                                 foreach ( $conns3 as $conn ) {
1014                                         if ( $conn->trxLevel() ) {
1015                                                 $conn->commit( __METHOD__, 'flush' );
1016                                         }
1017                                 }
1018                         }
1019                 }
1020         }
1021
1022         /**
1023          *  Issue COMMIT only on master, only if queries were done on connection
1024          */
1025         public function commitMasterChanges() {
1026                 $masterIndex = $this->getWriterIndex();
1027                 foreach ( $this->mConns as $conns2 ) {
1028                         if ( empty( $conns2[$masterIndex] ) ) {
1029                                 continue;
1030                         }
1031                         /** @var DatabaseBase $conn */
1032                         foreach ( $conns2[$masterIndex] as $conn ) {
1033                                 if ( $conn->trxLevel() && $conn->writesOrCallbacksPending() ) {
1034                                         $conn->commit( __METHOD__, 'flush' );
1035                                 }
1036                         }
1037                 }
1038         }
1039
1040         /**
1041          * Issue ROLLBACK only on master, only if queries were done on connection
1042          * @since 1.23
1043          */
1044         public function rollbackMasterChanges() {
1045                 $failedServers = array();
1046
1047                 $masterIndex = $this->getWriterIndex();
1048                 foreach ( $this->mConns as $conns2 ) {
1049                         if ( empty( $conns2[$masterIndex] ) ) {
1050                                 continue;
1051                         }
1052                         /** @var DatabaseBase $conn */
1053                         foreach ( $conns2[$masterIndex] as $conn ) {
1054                                 if ( $conn->trxLevel() && $conn->writesOrCallbacksPending() ) {
1055                                         try {
1056                                                 $conn->rollback( __METHOD__, 'flush' );
1057                                         } catch ( DBError $e ) {
1058                                                 MWExceptionHandler::logException( $e );
1059                                                 $failedServers[] = $conn->getServer();
1060                                         }
1061                                 }
1062                         }
1063                 }
1064
1065                 if ( $failedServers ) {
1066                         throw new DBExpectedError( null, "Rollback failed on server(s) " .
1067                                 implode( ', ', array_unique( $failedServers ) ) );
1068                 }
1069         }
1070
1071         /**
1072          * @return bool Whether a master connection is already open
1073          * @since 1.24
1074          */
1075         public function hasMasterConnection() {
1076                 return $this->isOpen( $this->getWriterIndex() );
1077         }
1078
1079         /**
1080          * Determine if there are pending changes in a transaction by this thread
1081          * @since 1.23
1082          * @return bool
1083          */
1084         public function hasMasterChanges() {
1085                 $masterIndex = $this->getWriterIndex();
1086                 foreach ( $this->mConns as $conns2 ) {
1087                         if ( empty( $conns2[$masterIndex] ) ) {
1088                                 continue;
1089                         }
1090                         /** @var DatabaseBase $conn */
1091                         foreach ( $conns2[$masterIndex] as $conn ) {
1092                                 if ( $conn->trxLevel() && $conn->writesOrCallbacksPending() ) {
1093                                         return true;
1094                                 }
1095                         }
1096                 }
1097                 return false;
1098         }
1099
1100         /**
1101          * Get the timestamp of the latest write query done by this thread
1102          * @since 1.25
1103          * @return float|bool UNIX timestamp or false
1104          */
1105         public function lastMasterChangeTimestamp() {
1106                 $lastTime = false;
1107                 $masterIndex = $this->getWriterIndex();
1108                 foreach ( $this->mConns as $conns2 ) {
1109                         if ( empty( $conns2[$masterIndex] ) ) {
1110                                 continue;
1111                         }
1112                         /** @var DatabaseBase $conn */
1113                         foreach ( $conns2[$masterIndex] as $conn ) {
1114                                 $lastTime = max( $lastTime, $conn->lastDoneWrites() );
1115                         }
1116                 }
1117                 return $lastTime;
1118         }
1119
1120         /**
1121          * Check if this load balancer object had any recent or still
1122          * pending writes issued against it by this PHP thread
1123          *
1124          * @param float $age How many seconds ago is "recent" [defaults to mWaitTimeout]
1125          * @return bool
1126          * @since 1.25
1127          */
1128         public function hasOrMadeRecentMasterChanges( $age = null ) {
1129                 $age = ( $age === null ) ? $this->mWaitTimeout : $age;
1130
1131                 return ( $this->hasMasterChanges()
1132                         || $this->lastMasterChangeTimestamp() > microtime( true ) - $age );
1133         }
1134
1135         /**
1136          * @param mixed $value
1137          * @return mixed
1138          */
1139         public function waitTimeout( $value = null ) {
1140                 return wfSetVar( $this->mWaitTimeout, $value );
1141         }
1142
1143         /**
1144          * @note This method will trigger a DB connection if not yet done
1145          * @return bool Whether the generic connection for reads is highly "lagged"
1146          */
1147         public function getLaggedSlaveMode() {
1148                 # Get a generic reader connection
1149                 $this->getConnection( DB_SLAVE );
1150
1151                 return $this->mLaggedSlaveMode;
1152         }
1153
1154         /**
1155          * @note This method will never cause a new DB connection
1156          * @return bool Whether any generic connection used for reads was highly "lagged"
1157          * @since 1.27
1158          */
1159         public function laggedSlaveUsed() {
1160                 return $this->mLaggedSlaveMode;
1161         }
1162
1163         /**
1164          * Disables/enables lag checks
1165          * @param null|bool $mode
1166          * @return bool
1167          */
1168         public function allowLagged( $mode = null ) {
1169                 if ( $mode === null ) {
1170                         return $this->mAllowLagged;
1171                 }
1172                 $this->mAllowLagged = $mode;
1173
1174                 return $this->mAllowLagged;
1175         }
1176
1177         /**
1178          * @return bool
1179          */
1180         public function pingAll() {
1181                 $success = true;
1182                 foreach ( $this->mConns as $conns2 ) {
1183                         foreach ( $conns2 as $conns3 ) {
1184                                 /** @var DatabaseBase[] $conns3 */
1185                                 foreach ( $conns3 as $conn ) {
1186                                         if ( !$conn->ping() ) {
1187                                                 $success = false;
1188                                         }
1189                                 }
1190                         }
1191                 }
1192
1193                 return $success;
1194         }
1195
1196         /**
1197          * Call a function with each open connection object
1198          * @param callable $callback
1199          * @param array $params
1200          */
1201         public function forEachOpenConnection( $callback, array $params = array() ) {
1202                 foreach ( $this->mConns as $conns2 ) {
1203                         foreach ( $conns2 as $conns3 ) {
1204                                 foreach ( $conns3 as $conn ) {
1205                                         $mergedParams = array_merge( array( $conn ), $params );
1206                                         call_user_func_array( $callback, $mergedParams );
1207                                 }
1208                         }
1209                 }
1210         }
1211
1212         /**
1213          * Get the hostname and lag time of the most-lagged slave
1214          *
1215          * This is useful for maintenance scripts that need to throttle their updates.
1216          * May attempt to open connections to slaves on the default DB. If there is
1217          * no lag, the maximum lag will be reported as -1.
1218          *
1219          * @param bool|string $wiki Wiki ID, or false for the default database
1220          * @return array ( host, max lag, index of max lagged host )
1221          */
1222         public function getMaxLag( $wiki = false ) {
1223                 $maxLag = -1;
1224                 $host = '';
1225                 $maxIndex = 0;
1226
1227                 if ( $this->getServerCount() <= 1 ) {
1228                         return array( $host, $maxLag, $maxIndex ); // no replication = no lag
1229                 }
1230
1231                 $lagTimes = $this->getLagTimes( $wiki );
1232                 foreach ( $lagTimes as $i => $lag ) {
1233                         if ( $lag > $maxLag ) {
1234                                 $maxLag = $lag;
1235                                 $host = $this->mServers[$i]['host'];
1236                                 $maxIndex = $i;
1237                         }
1238                 }
1239
1240                 return array( $host, $maxLag, $maxIndex );
1241         }
1242
1243         /**
1244          * Get an estimate of replication lag (in seconds) for each server
1245          *
1246          * Results are cached for a short time in memcached/process cache
1247          *
1248          * Values may be "false" if replication is too broken to estimate
1249          *
1250          * @param string|bool $wiki
1251          * @return int[] Map of (server index => float|int|bool)
1252          */
1253         public function getLagTimes( $wiki = false ) {
1254                 if ( $this->getServerCount() <= 1 ) {
1255                         return array( 0 => 0 ); // no replication = no lag
1256                 }
1257
1258                 # Send the request to the load monitor
1259                 return $this->getLoadMonitor()->getLagTimes( array_keys( $this->mServers ), $wiki );
1260         }
1261
1262         /**
1263          * Get the lag in seconds for a given connection, or zero if this load
1264          * balancer does not have replication enabled.
1265          *
1266          * This should be used in preference to Database::getLag() in cases where
1267          * replication may not be in use, since there is no way to determine if
1268          * replication is in use at the connection level without running
1269          * potentially restricted queries such as SHOW SLAVE STATUS. Using this
1270          * function instead of Database::getLag() avoids a fatal error in this
1271          * case on many installations.
1272          *
1273          * @param DatabaseBase $conn
1274          * @return int
1275          */
1276         public function safeGetLag( $conn ) {
1277                 if ( $this->getServerCount() == 1 ) {
1278                         return 0;
1279                 } else {
1280                         return $conn->getLag();
1281                 }
1282         }
1283
1284         /**
1285          * Clear the cache for slag lag delay times
1286          *
1287          * This is only used for testing
1288          */
1289         public function clearLagTimeCache() {
1290                 $this->getLoadMonitor()->clearCaches();
1291         }
1292 }