2 # Database load balancing object
4 require_once( "Database.php" );
6 # Valid database indexes
7 # Operation-based indexes
8 define( "DB_SLAVE", -1 ); # Read from the slave (or only server)
9 define( "DB_MASTER", -2 ); # Write to master (or only server)
10 define( "DB_LAST", -3 ); # Whatever database was used last
13 define( "DB_READ", -1 );
14 define( "DB_WRITE", -2 );
17 # ***NOT USED YET, EXPERIMENTAL***
18 # These may be defined in $wgDBservers. If they aren't, the default reader or writer will be used
19 # Even numbers are always readers, odd numbers are writers
20 define( "DB_TASK_FIRST", 1000 ); # First in list
21 define( "DB_SEARCH_R", 1000 ); # Search read
22 define( "DB_SEARCH_W", 1001 ); # Search write
23 define( "DB_ASKSQL_R", 1002 ); # Special:Asksql read
24 define( "DB_WATCHLIST_R", 1004 ); # Watchlist read
25 define( "DB_TASK_LAST", 1004) ; # Last in list
27 define( "MASTER_WAIT_TIMEOUT", 15 ); # Time to wait for a slave to synchronise
30 /* private */ var $mServers, $mConnections, $mLoads;
31 /* private */ var $mFailFunction;
32 /* private */ var $mForce, $mReadIndex, $mLastIndex;
33 /* private */ var $mWaitForFile, $mWaitForPos;
35 function LoadBalancer()
37 $this->mServers
= array();
38 $this->mConnections
= array();
39 $this->mFailFunction
= false;
40 $this->mReadIndex
= -1;
42 $this->mLastIndex
= -1;
45 function newFromParams( $servers, $failFunction = false )
47 $lb = new LoadBalancer
;
48 $lb->initialise( $servers, $failFunction = false );
52 function initialise( $servers, $failFunction = false )
54 $this->mServers
= $servers;
55 $this->mFailFunction
= $failFunction;
56 $this->mReadIndex
= -1;
57 $this->mWriteIndex
= -1;
59 $this->mConnections
= array();
60 $this->mLastIndex
= 1;
61 $this->mLoads
= array();
62 $this->mWaitForFile
= false;
63 $this->mWaitForPos
= false;
65 foreach( $servers as $i => $server ) {
66 $this->mLoads
[$i] = $server['load'];
70 # Given an array of non-normalised probabilities, this function will select
71 # an element and return the appropriate key
72 function pickRandom( $weights )
74 if ( !is_array( $weights ) ||
count( $weights ) == 0 ) {
79 foreach ( $weights as $w ) {
82 $max = mt_getrandmax();
83 $rand = mt_rand(0, $max) / $max * $sum;
86 foreach ( $weights as $i => $w ) {
88 if ( $sum >= $rand ) {
95 function getReaderIndex()
97 $fname = 'LoadBalancer::getReaderIndex';
98 wfProfileIn( $fname );
101 if ( $this->mForce
>= 0 ) {
104 if ( $this->mReadIndex
>= 0 ) {
105 $i = $this->mReadIndex
;
107 # $loads is $this->mLoads except with elements knocked out if they
109 $loads = $this->mLoads
;
111 $i = $this->pickRandom( $loads );
112 if ( $i !== false ) {
113 wfDebug( "Using reader #$i: {$this->mServers[$i]['host']}\n" );
115 $this->openConnection( $i );
117 if ( !$this->isOpen( $i ) ) {
121 } while ( $i !== false && !$this->isOpen( $i ) );
123 if ( $this->isOpen( $i ) ) {
124 $this->mReadIndex
= $i;
130 wfProfileOut( $fname );
134 # Set the master wait position
135 # If a DB_SLAVE connection has been opened already, waits
136 # Otherwise sets a variable telling it to wait if such a connection is opened
137 function waitFor( $file, $pos ) {
138 $fname = "LoadBalancer::waitFor";
139 wfProfileIn( $fname );
141 wfDebug( "User master pos: $file $pos\n" );
142 $this->mWaitForFile
= false;
143 $this->mWaitForPos
= false;
145 if ( count( $this->mServers
) > 1 ) {
146 $this->mWaitForFile
= $file;
147 $this->mWaitForPos
= $pos;
149 if ( $this->mReadIndex
> 0 ) {
150 if ( !$this->doWait( $this->mReadIndex
) ) {
152 $this->mReadIndex
= 0;
156 wfProfileOut( $fname );
159 # Wait for a given slave to catch up to the master pos stored in $this
160 function doWait( $index ) {
165 $key = "masterpos:" . $index;
166 $memcPos = $wgMemc->get( $key );
168 list( $file, $pos ) = explode( ' ', $memcPos );
169 # If the saved position is later than the requested position, return now
170 if ( $file == $this->mWaitForFile
&& $this->mWaitForPos
<= $pos ) {
175 if ( !$retVal && $this->isOpen( $index ) ) {
176 $conn =& $this->mConnections( $index );
177 wfDebug( "Waiting for slave #$index to catch up...\n" );
178 $result = $conn->masterPosWait( $this->mWaitForFile
, $this->mWaitForPos
, MASTER_WAIT_TIMEOUT
);
180 if ( $result == -1 ||
is_null( $result ) ) {
181 # Timed out waiting for slave, use master instead
182 wfDebug( "Timed out waiting for slave #$index pos {$this->mWaitForFile} {$this->mWaitForPos}\n" );
192 # Get a connection by index
193 function &getConnection( $i, $fail = true )
195 $fname = "LoadBalancer::getConnection";
196 wfProfileIn( $fname );
199 if ( $i >= DB_TASK_FIRST && $i < DB_TASK_LAST ) {
201 # Odd index use writer
204 # Even index use reader
209 # Operation-based index
210 if ( $i == DB_SLAVE
) {
211 $i = $this->getReaderIndex();
212 } elseif ( $i == DB_MASTER
) {
213 $i = $this->getWriterIndex();
214 } elseif ( $i == DB_LAST
) {
215 # Just use $this->mLastIndex, which should already be set
216 $i = $this->mLastIndex
;
218 # Oh dear, not set, best to use the writer for safety
219 wfDebug( "Warning: DB_LAST used when there was no previous index\n" );
220 $i = $this->getWriterIndex();
223 # Now we have an explicit index into the servers array
224 $this->openConnection( $i, $fail );
225 wfProfileOut( $fname );
226 return $this->mConnections
[$i];
229 # Open a connection to the server given by the specified index
230 # Index must be an actual index into the array
231 /* private */ function openConnection( $i, $fail = false ) {
232 $fname = 'LoadBalancer::openConnection';
233 wfProfileIn( $fname );
235 if ( !$this->isOpen( $i ) ) {
236 $this->mConnections
[$i] = $this->reallyOpenConnection( $this->mServers
[$i] );
238 if ( $i != 0 && $this->mWaitForFile
) {
239 if ( !$this->doWait( $i ) ) {
240 # Error waiting for this slave, use master instead
241 $this->mReadIndex
= 0;
243 if ( !$this->isOpen( 0 ) ) {
244 $this->mConnections
[0] = $this->reallyOpenConnection( $this->mServers
[0] );
246 wfDebug( "Failed over to {$this->mConnections[0]->mServer}\n" );
250 if ( !$this->isOpen( $i ) ) {
251 wfDebug( "Failed to connect to database $i at {$this->mServers[$i]['host']}\n" );
253 $this->reportConnectionError( $this->mConnections
[$i] );
255 $this->mConnections
[$i] = false;
257 $this->mLastIndex
= $i;
258 wfProfileOut( $fname );
261 # Test if the specified index represents an open connection
262 /* private */ function isOpen( $index ) {
263 if( !is_integer( $index ) ) {
266 if ( array_key_exists( $index, $this->mConnections
) && is_object( $this->mConnections
[$index] ) &&
267 $this->mConnections
[$index]->isOpen() )
275 # Really opens a connection
276 /* private */ function reallyOpenConnection( &$server ) {
278 # Get class for this database type
279 $class = 'Database' . ucfirst( $type );
280 if ( !class_exists( $class ) ) {
281 require_once( "$class.php" );
285 return new $class( $host, $user, $password, $dbname, 1, $flags );
288 function reportConnectionError( &$conn )
290 $fname = 'LoadBalancer::reportConnectionError';
291 wfProfileIn( $fname );
292 # Prevent infinite recursion
294 static $reporting = false;
297 if ( !is_object( $conn ) ) {
298 $conn = new Database
;
300 if ( $this->mFailFunction
) {
301 $conn->failFunction( $this->mFailFunction
);
303 $conn->failFunction( "wfEmergencyAbort" );
305 $conn->reportConnectionError();
308 wfProfileOut( $fname );
311 function getWriterIndex()
321 function haveIndex( $i )
323 return array_key_exists( $i, $this->mServers
);
326 # Get the number of defined servers (not the number of open connections)
327 function getServerCount() {
328 return count( $this->mServers
);
331 # Save master pos to the session and to memcached, if the session exists
332 function saveMasterPos() {
333 global $wgSessionStarted;
334 if ( $wgSessionStarted && count( $this->mServers
) > 1 ) {
335 # If this entire request was served from a slave without opening a connection to the
336 # master (however unlikely that may be), then we can fetch the position from the slave.
337 if ( empty( $this->mConnections
[0] ) ) {
338 $conn =& $this->getConnection( DB_SLAVE
);
339 list( $file, $pos ) = $conn->getSlavePos();
340 wfDebug( "Saving master pos fetched from slave: $file $pos\n" );
342 $conn =& $this->getConnection( 0 );
343 list( $file, $pos ) = $conn->getMasterPos();
344 wfDebug( "Saving master pos: $file $pos\n" );
346 if ( $file !== false ) {
347 $_SESSION['master_log_file'] = $file;
348 $_SESSION['master_pos'] = $pos;
353 # Loads the master pos from the session, waits for it if necessary
354 function loadMasterPos() {
355 if ( isset( $_SESSION['master_log_file'] ) && isset( $_SESSION['master_pos'] ) ) {
356 $this->waitFor( $_SESSION['master_log_file'], $_SESSION['master_pos'] );
360 # Close all open connections
361 function closeAll() {
362 foreach( $this->mConnections
as $i => $conn ) {
363 if ( $this->isOpen( $i ) ) {
369 function commitAll() {
370 foreach( $this->mConnections
as $i => $conn ) {
371 if ( $this->isOpen( $i ) ) {
372 $conn->immediateCommit();