Permalink
Browse files

enhance mysql connection pool

make varz_details execution in serial to prevent pool blockage
export few connection pool metrics

Change-Id: I3b4ae4f24ba40f0caa58a9749baf92df498386d4
  • Loading branch information...
1 parent c555b7d commit 5ae6c5fd498da423422d2c8b474fdf4e3a001d25 @andl andl committed Sep 17, 2012
Showing with 51 additions and 1 deletion.
  1. +6 −0 mysql/lib/mysql_service/node.rb
  2. +45 −1 mysql/lib/mysql_service/util.rb
@@ -73,6 +73,7 @@ def initialize(options)
@kill_long_queries_lock = Mutex.new
@kill_long_transaction_lock = Mutex.new
@enforce_quota_lock = Mutex.new
+ @varz_lock = Mutex.new
@connection_wait_timeout = options[:connection_wait_timeout]
Mysql2::Client.default_timeout = @connection_wait_timeout
@@ -596,6 +597,8 @@ def exe_cmd(cmd, stdin=nil)
end
def varz_details()
+ acquired = @varz_lock.try_lock
+ return unless acquired
varz = {}
# how many queries served since startup
varz[:queries_since_startup] = get_queries_status
@@ -624,10 +627,13 @@ def varz_details()
rescue => e
@logger.error("Error get instance list: #{e}")
end
+ varz[:connection_pool] = @pool.inspect
varz
rescue => e
@logger.error("Error during generate varz: #{e}")
{}
+ ensure
+ @varz_lock.unlock if acquired
end
def get_status(instance)
@@ -134,6 +134,7 @@ def kill_database_session(connection, database)
end
class Connection
+ attr_accessor :checked_out_by
attr_reader :conn
def initialize(opts)
@@ -175,16 +176,59 @@ def initialize(options)
@cond = @connections.new_cond
@reserved_connections = {}
@checked_out = []
+ @metrix_lock = Mutex.new
+ @latency_sum = 0
+ @queries_served = 1
+ @worst_latency = 0
for i in 1..@size do
@connections << Connection.new(@options)
end
end
+ def inspect
+ {
+ :size => @size,
+ :checked_out_size => @checked_out.size,
+ :checked_out_by => @checked_out.map{|conn| conn.checked_out_by },
+ :average_latency_ms => @latency_sum / @queries_served,
+ :worst_latency_ms => @worst_latency
+ }
+ rescue => e
+ @logger.warn("Error in inspect: #{e}")
+ nil
+ end
+
+ def timing
+ t1 = Time.now
+ yield
+ ensure
+ update_latency_metric (Time.now - t1) * 1000
+ end
+
+ def update_latency_metric(latency)
+ @metrix_lock.synchronize {
+ @latency_sum += latency
+ @queries_served += 1
+ @worst_latency = latency if latency > @worst_latency
+ }
+ end
+
+ def parse_caller(callstack)
+ frame = callstack[0]
+ method = nil
+ method = $1 if frame =~ /in `([^']+)/
+ method
+ end
+
def with_connection
connection_id = current_connection_id
fresh_connection = !@reserved_connections.has_key?(connection_id)
connection = (@reserved_connections[connection_id] ||= checkout)
- yield connection.conn
+
+ direct_caller = parse_caller(caller(1))
+ connection.checked_out_by = direct_caller
+
+ timing { yield connection.conn }
ensure
release_connection(connection_id) if fresh_connection
end

0 comments on commit 5ae6c5f

Please sign in to comment.