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