update hyperdb

wp-content/db.php

@@ -5,7 +5,10 @@
 
 /** Load the wpdb class while preventing instantiation **/
 $wpdb = true;
-require_once( ABSPATH . WPINC . '/wp-db.php' );
+if ( defined('WPDB_PATH') )
+	require_once(WPDB_PATH);
+else
+	require_once( ABSPATH . WPINC . '/wp-db.php' );
 
 if ( defined('DB_CONFIG_FILE') && file_exists( DB_CONFIG_FILE ) ) {
 
@@ -29,6 +32,13 @@ if ( defined('DB_CONFIG_FILE') && file_exists( DB_CONFIG_FILE ) ) {
 
 }
 
+/**
+ * Common definitions 
+ */
+define( 'HYPERDB_LAG_OK', 1 );
+define( 'HYPERDB_LAG_BEHIND', 2 );
+define( 'HYPERDB_LAG_UNKNOWN', 3 );
+
 class hyperdb extends wpdb {
 	/**
 	 * The last table that was queried
@@ -144,10 +154,22 @@ class hyperdb extends wpdb {
 	 */
 	var $used_servers = array();
 
+	/**
+	 * Whether to save debug_backtrace in save_query_callback. You may wish
+	 * to disable this, e.g. when tracing out-of-memory problems.
+	 */
+	var $save_backtrace = true;
+
+	/**
+	 * Maximum lag in seconds. Set null to disable. Requires callbacks.
+	 * @var integer
+	 */
+	var $default_lag_threshold = null;
+
 	/**
 	 * Triggers __construct() for backwards compatibility with PHP4
 	 */
-	function db( $args = null ) {
+	function hyperdb( $args = null ) {
 		return $this->__construct($args);
 	}
 
@@ -160,6 +182,26 @@ class hyperdb extends wpdb {
 			foreach ( get_class_vars(__CLASS__) as $var => $value )
 				if ( isset($args[$var]) )
 					$this->$var = $args[$var];
+
+		$this->init_charset();
+	}
+
+	/**
+	 * Sets $this->charset and $this->collate
+	 */
+	function init_charset() {
+		if ( function_exists('is_multisite') && is_multisite() ) {
+			$this->charset = 'utf8';
+			if ( defined( 'DB_COLLATE' ) && DB_COLLATE )
+				$this->collate = DB_COLLATE;
+			else
+				$this->collate = 'utf8_general_ci';
+		} elseif ( defined( 'DB_COLLATE' ) ) {
+			$this->collate = DB_COLLATE;
+		}
+
+		if ( defined( 'DB_CHARSET' ) )
+			$this->charset = DB_CHARSET;
 	}
 
 	/**
@@ -171,6 +213,7 @@ class hyperdb extends wpdb {
 		isset($read)    or $read = 1;
 		isset($write)   or $write = 1;
 		unset($db['dataset']);
+
 		if ( $read )
 			$this->hyper_servers[ $dataset ][ 'read' ][ $read ][] = $db;
 		if ( $write )
@@ -185,10 +228,12 @@ class hyperdb extends wpdb {
 	}
 
 	/**
-	 * Add a callback to examine queries and determine dataset.
+	 * Add a callback to a group of callbacks.
+	 * The default group is 'dataset', used to examine 
+	 * queries and determine dataset.
 	 */
-	function add_callback( $callback ) {
-		$this->hyper_callbacks[] = $callback;
+	function add_callback( $callback, $group = 'dataset' ) {
+		$this->hyper_callbacks[ $group ][] = $callback;
 	}
 
 	/**
@@ -252,9 +297,8 @@ class hyperdb extends wpdb {
 	 */
 	function is_write_query( $q ) {
 		// Quick and dirty: only SELECT statements are considered read-only.
-		$q = ltrim($q, "\t (");
-		$word = strtoupper( substr( trim( $q ), 0, 6 ) );
-		return 'SELECT' != $word;
+		$q = ltrim($q, "\r\n\t (");
+		return !preg_match('/^(?:SELECT|SHOW|DESCRIBE|EXPLAIN)\s/i', $q);
 	}
 
 	/**
@@ -265,12 +309,22 @@ class hyperdb extends wpdb {
 	}
 
 	/**
-	 * Callbacks are specified in the config. They must return a dataset
-	 * or an associative array with an element called 'dataset'.
+	 * Callbacks are executed in the order in which they are registered until one
+	 * of them returns something other than null.
 	 */
-	function run_callbacks( $query ) {
-		$args = array($query, &$this);
-		foreach ( $this->hyper_callbacks as $func ) {
+	function run_callbacks( $group, $args = null) {
+		if ( !is_array( $this->hyper_callbacks[ $group ] ) )
+			return null;
+
+		if ( !isset($args) ) {
+			$args = array( &$this );
+		} elseif ( is_array( $args ) ) {
+			$args[] = &$this;
+		} else {
+			$args = array( $args, &$this );
+		}
+
+		foreach ( $this->hyper_callbacks[ $group ] as $func ) {
 			$result = call_user_func_array($func, $args);
 			if ( isset($result) )
 				return $result;
@@ -315,7 +369,7 @@ class hyperdb extends wpdb {
 		if ( isset($this->hyper_tables[$this->table]) ) {
 			$dataset = $this->hyper_tables[$this->table];
 			$this->callback_result = null;
-		} elseif ( null !== $this->callback_result = $this->run_callbacks($query) ) {
+		} elseif ( null !== $this->callback_result = $this->run_callbacks( 'dataset', $query ) ) {
 			if ( is_array($this->callback_result) )
 				extract( $this->callback_result, EXTR_OVERWRITE );
 			else
@@ -417,99 +471,176 @@ class hyperdb extends wpdb {
 			foreach ( $this->hyper_servers[$dataset][$operation] as $group => $items ) {
 				$keys = array_keys($items);
 				shuffle($keys);
-				foreach ( $keys as $key ) {
+				foreach ( $keys as $key ) 
 					$servers[] = compact('group', 'key');
-				}
 			}
-			$tries_remaining = count($servers);
-			if ( !$tries_remaining )
+
+			if ( !$tries_remaining = count( $servers ) ) 
 				return $this->bail("No database servers were found to match the query. ($this->table, $dataset)");
+			
+			if ( !isset( $unique_servers ) )
+				$unique_servers = $tries_remaining;
+
 		} while ( $tries_remaining < $this->min_tries );
 
 		// Connect to a database server
-		foreach ( $servers as $group_key ) {
-			--$tries_remaining;
-
-			// $group, $key
-			extract($group_key, EXTR_OVERWRITE);
+		do {
+			$unique_lagged_slaves = array();
+			$success = false;
 
-			// $host, $user, $password, $name, $read, $write [ $connect_function, $timeout ]
-			extract($this->hyper_servers[$dataset][$operation][$group][$key], EXTR_OVERWRITE);
+			foreach ( $servers as $group_key ) {
+				--$tries_remaining;	
+	
+				// If all servers are lagged, we need to start ignoring the lag and retry
+				if ( count( $unique_lagged_slaves ) == $unique_servers )
+					break;
 
-			list($host, $port) = explode(':', $host);
+				// $group, $key
+				extract($group_key, EXTR_OVERWRITE);
 
-			// Split host:port into $host and $port
-			if ( strpos($host, ':') )
-				list($host, $port) = explode(':', $host);
+				// $host, $user, $password, $name, $read, $write [, $lag_threshold, $connect_function, $timeout ]
+				extract($this->hyper_servers[$dataset][$operation][$group][$key], EXTR_OVERWRITE);
 
-			// Overlay $server if it was extracted from a callback
-			if ( isset($server) && is_array($server) )
-				extract($server, EXTR_OVERWRITE);
-
-			// Split again in case $server had host:port
-			if ( strpos($host, ':') )
 				list($host, $port) = explode(':', $host);
 
-			// Make sure there's always a port number
-			if ( empty($port) )
-				$port = 3306;
-
-			// Use a default timeout of 200ms
-			if ( !isset($timeout) )
-				$timeout = 0.2;
+				// Split host:port into $host and $port
+				if ( strpos($host, ':') )
+					list($host, $port) = explode(':', $host);
+
+				// Overlay $server if it was extracted from a callback
+				if ( isset($server) && is_array($server) )
+					extract($server, EXTR_OVERWRITE);
+
+				// Split again in case $server had host:port
+				if ( strpos($host, ':') )
+					list($host, $port) = explode(':', $host);
+
+				// Make sure there's always a port number
+				if ( empty($port) )
+					$port = 3306;
+
+				// Use a default timeout of 200ms
+				if ( !isset($timeout) )
+					$timeout = 0.2;
+
+				// Get the minimum group here, in case $server rewrites it
+				if ( !isset( $min_group ) || $min_group > $group )
+					$min_group = $group;
+
+				// Can be used by the lag callbacks
+				$this->lag_cache_key = "$host:$port";
+				$this->lag_threshold = isset($lag_threshold) ? $lag_threshold : $this->default_lag_threshold;
+
+				// Check for a lagged slave, if applicable
+				if ( !$use_master && !$write && !isset( $ignore_slave_lag )
+					&& isset($this->lag_threshold) && !isset( $server['host'] )
+					&& ( $lagged_status = $this->get_lag_cache() ) === HYPERDB_LAG_BEHIND
+				) {
+					// If it is the last lagged slave and it is with the best preference we will ignore its lag
+					if ( !isset( $unique_lagged_slaves[ "$host:$port" ] )
+						&& $unique_servers == count( $unique_lagged_slaves ) + 1
+						&& $group == $min_group )
+					{
+						$this->lag_threshold = null;
+					} else {
+						$unique_lagged_slaves["$host.$port"] = $this->lag;
+						continue;
+					}
+				}
 
-			$this->timer_start();
+				$this->timer_start();
 
-			// Connect if necessary or possible
-			$tcp = null;
-			if ( $use_master || !$tries_remaining || !$this->check_tcp_responsiveness || true === $tcp = $this->check_tcp_responsiveness($host, $port, $timeout) ) {
-				$this->dbhs[$dbhname] = @ $connect_function( "$host:$port", $user, $password, true );
-			} else {
-				$this->dbhs[$dbhname] = false;
-			}
+				// Connect if necessary or possible
+				$tcp = null;
+				if ( $use_master || !$tries_remaining || !$this->check_tcp_responsiveness
+					|| true === $tcp = $this->check_tcp_responsiveness($host, $port, $timeout) )
+				{
+					$this->dbhs[$dbhname] = @ $connect_function( "$host:$port", $user, $password, true );	
+				} else {
+					$this->dbhs[$dbhname] = false;
+				}
 
-			$elapsed = $this->timer_stop();
+				$elapsed = $this->timer_stop();
+
+				if ( is_resource( $this->dbhs[$dbhname] ) ) {
+					/**
+					 * If we care about lag, disconnect lagged slaves and try to find others.
+					 * We don't disconnect if it is the last lagged slave and it is with the best preference.
+					 */
+					if ( !$use_master && !$write && !isset( $ignore_slave_lag )
+						&& isset($this->lag_threshold) && !isset( $server['host'] )
+						&& $lagged_status !== HYPERDB_LAG_OK
+						&& ( $lagged_status = $this->get_lag() ) === HYPERDB_LAG_BEHIND
+						&& !(
+							!isset( $unique_lagged_slaves[ "$host:$port" ] )
+							&& $unique_servers == count( $unique_lagged_slaves ) + 1
+							&& $group == $min_group
+						)
+					) {
+						$success = false;
+						$unique_lagged_slaves["$host:$port"] = $this->lag;
+						$this->disconnect( $dbhname );
+						$this->dbhs[$dbhname] = false;
+						$msg = "Replication lag of {$this->lag}s on $host:$port ($dbhname)";
+						$this->print_error( $msg );
+						continue;
+					} elseif ( mysql_select_db( $name, $this->dbhs[ $dbhname ] ) ) {
+						$success = true;
+						$this->current_host = "$host:$port";
+						$this->dbh2host[$dbhname] = "$host:$port";
+						$queries = 1;
+						$lag = isset( $this->lag ) ? $this->lag : 0;
+						$this->last_connection = compact('dbhname', 'host', 'port', 'user', 'name', 'tcp', 'elapsed', 'success', 'querie
+s', 'lag');
+						$this->db_connections[] = $this->last_connection;
+						$this->open_connections[] = $dbhname;
+						break;
+					}
+				}
 
-			if ( is_resource($this->dbhs[$dbhname]) && mysql_select_db( $name, $this->dbhs[$dbhname] ) ) {
-				$success = true;
-				$this->current_host = "$host:$port";
-				$this->dbh2host[$dbhname] = "$host:$port";
-				$queries = 1;
-				$this->last_connection = compact('dbhname', 'host', 'port', 'user', 'name', 'tcp', 'elapsed', 'success', 'queries');
-				$this->db_connections[] = $this->last_connection;
-				$this->open_connections[] = $dbhname;
-				break;
-			} else {
 				$success = false;
 				$this->last_connection = compact('dbhname', 'host', 'port', 'user', 'name', 'tcp', 'elapsed', 'success');
 				$this->db_connections[] = $this->last_connection;
-				$error_details = array (
-					'referrer' => "{$_SERVER['HTTP_HOST']}{$_SERVER['REQUEST_URI']}",
-'server'=>$server,
+
+				$msg = date( "Y-m-d H:i:s" ) . " Can't select $dbhname - \n";
+				$msg .= "'referrer' => '{$_SERVER['HTTP_HOST']}{$_SERVER['REQUEST_URI']}',\n";
+				$msg .= "'server' => {$server},\n";
+				$msg .= "'host' => {$host},\n";
+				$msg .= "'error' => " . mysql_error() . ",\n";
+				$msg .= "'errno' => " . mysql_errno() . ",\n";
+				$msg .= "'tcp_responsive' => " . ( $tcp === true ? 'true' : $tcp ) . ",\n";
+				$msg .= "'lagged_status' => " . ( isset( $lagged_status ) ? $lagged_status : HYPERDB_LAG_UNKNOWN );
+
+				$this->print_error( $msg );
+			}
+
+			if ( !$success || !isset($this->dbhs[$dbhname]) || !is_resource( $this->dbhs[$dbhname] ) ) {
+				if ( !isset( $ignore_slave_lag ) && count( $unique_lagged_slaves ) ) { 
+					// Lagged slaves were not used. Ignore the lag for this connection attempt and retry.
+					$ignore_slave_lag = true;
+					$tries_remaining = count( $servers );
+					continue;
+				}
+				
+				$error_details = array(
 					'host' => $host,
-					'error' => mysql_error(),
-					'errno' => mysql_errno(),
-					'tcp_responsive' => $tcp === true ? 'true' : $tcp,
+					'port' => $port,
+					'operation' => $operation,
+					'table' => $this->table,
+					'dataset' => $dataset,
+					'dbhname' => $dbhname
 				);
-				$msg = date( "Y-m-d H:i:s" ) . " Can't select $dbhname - ";
-				$msg .= "\n" . print_r($error_details, true);
+				$this->run_callbacks( 'db_connection_error', $error_details );
 
-				$this->print_error( $msg );
+				return $this->bail( "Unable to connect to $host:$port to $operation table '$this->table' ($dataset)" );
 			}
-		}
 
-		if ( ! is_resource( $this->dbhs[$dbhname] ) )
-			return $this->bail("Unable to connect to $host:$port to $operation table '$this->table' ($dataset)");
+			break;
+		} while ( true );
+
+		$this->set_charset($this->dbhs[$dbhname], $charset, $collate);
 
-		if ( !empty($charset) )
-			$collation_query = "SET NAMES '$charset'";
-		elseif ( !empty($this->charset) )
-			$collation_query = "SET NAMES '$this->charset'";
-		if ( !empty($collation_query) && !empty($collate) )
-			$collation_query .= " COLLATE '$collate'";
-		if ( !empty($collation_query) && !empty($this->collation) )
-			$collation_query .= " COLLATE '$this->collation'";
-		mysql_query($collation_query, $this->dbhs[$dbhname]);
+		$this->dbh = $this->dbhs[$dbhname]; // needed by $wpdb->_real_escape()
 
 		$this->last_used_server = compact('host', 'user', 'name', 'read', 'write');
 
@@ -524,6 +655,30 @@ class hyperdb extends wpdb {
 		return $this->dbhs[$dbhname];
 	}
 
+	/**
+	 * Sets the connection's character set.
+	 * @param resource $dbh     The resource given by mysql_connect
+	 * @param string   $charset The character set (optional)
+	 * @param string   $collate The collation (optional)
+	 */
+	function set_charset($dbh, $charset = null, $collate = null) {
+		if ( !isset($charset) )
+			$charset = $this->charset;
+		if ( !isset($collate) )
+			$collate = $this->collate;
+		if ( $this->has_cap( 'collation', $dbh ) && !empty( $charset ) ) {
+			if ( function_exists( 'mysql_set_charset' ) && $this->has_cap( 'set_charset', $dbh ) ) {
+				mysql_set_charset( $charset, $dbh );
+				$this->real_escape = true;
+			} else {
+				$query = $this->prepare( 'SET NAMES %s', $charset );
+				if ( ! empty( $collate ) )
+					$query .= $this->prepare( ' COLLATE %s', $collate );
+				mysql_query( $query, $dbh );
+			}
+		}
+	}
+
 	/**
 	 * Disconnect and remove connection from open connections list
 	 * @param string $tdbhname
@@ -595,7 +750,7 @@ class hyperdb extends wpdb {
 
 			if ( $this->save_queries ) {
 				if ( is_callable($this->save_query_callback) )
-					$this->queries[] = call_user_func_array( $this->save_query_callback, array( $query, $elapsed, debug_backtrace(), &$this ) );
+					$this->queries[] = call_user_func_array( $this->save_query_callback, array( $query, $elapsed, $this->save_backtrace ? debug_backtrace( false ) : null, &$this ) );
 				else
 					$this->queries[] = array( $query, $elapsed, $this->get_caller() );
 			}
@@ -684,7 +839,8 @@ class hyperdb extends wpdb {
 		case 'group_concat' :
 		case 'subqueries' :
 			return version_compare($version, '4.1', '>=');
-			break;
+		case 'set_charset' :
+			return version_compare($version, '5.0.7', '>=');
 		endswitch;
 
 		return false;
@@ -717,7 +873,7 @@ class hyperdb extends wpdb {
 		if ( !is_callable('debug_backtrace') )
 			return '';
 
-		$bt = debug_backtrace();
+		$bt = debug_backtrace( false );
 		$caller = '';
 
 		foreach ( (array) $bt as $trace ) {
@@ -746,34 +902,33 @@ class hyperdb extends wpdb {
 	 * @return (bool) true when $host:$post responds within $float_timeout seconds, else (bool) false
 	 */
 	function check_tcp_responsiveness($host, $port, $float_timeout) {
-		if ( 1 == 2 && function_exists('apc_store') ) {
-			$use_apc = true;
-			$apc_key = "{$host}{$port}";
-			$apc_ttl = 10;
-		} else {
-			$use_apc = false;
-		}
-		if ( $use_apc ) {
-			$cached_value=apc_fetch($apc_key);
-			switch ( $cached_value ) {
-				case 'up':
-					$this->tcp_responsive = 'true';
-					return true;
-				case 'down':
-					$this->tcp_responsive = 'false';
-					return false;
-			}
-		}
-	        $socket = @ fsockopen($host, $port, $errno, $errstr, $float_timeout);
-	        if ( $socket === false ) {
-			if ( $use_apc )
-				apc_store($apc_key, 'down', $apc_ttl);
+		$socket = @ fsockopen($host, $port, $errno, $errstr, $float_timeout);
+		if ( $socket === false )
 			return "[ > $float_timeout ] ($errno) '$errstr'";
-		}
 		fclose($socket);
-		if ( $use_apc )
-			apc_store($apc_key, 'up', $apc_ttl);
-	        return true;
+		return true;
+	}
+
+	function get_lag_cache() {
+		$this->lag = $this->run_callbacks( 'get_lag_cache' );
+
+		return $this->check_lag();
+	}
+
+	function get_lag() {
+		$this->lag = $this->run_callbacks( 'get_lag' );
+
+		return $this->check_lag();
+	}
+
+	function check_lag() {
+		if ( $this->lag === false )
+			return HYPERDB_LAG_UNKNOWN;
+
+		if ( $this->lag > $this->lag_threshold )
+			return HYPERDB_LAG_BEHIND;
+
+		return HYPERDB_LAG_OK;
 	}
 
 	// Helper functions for configuration