dépôts / lhc / web / wiklou.git / blob
? search:


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