2010-11-16 20:43:59 +00:00
|
|
|
# encoding: UTF-8
|
|
|
|
|
|
|
|
# --
|
2011-01-17 17:26:32 +00:00
|
|
|
# Copyright (C) 2008-2011 10gen Inc.
|
2010-11-16 20:43:59 +00:00
|
|
|
#
|
|
|
|
# Licensed under the Apache License, Version 2.0 (the "License");
|
|
|
|
# you may not use this file except in compliance with the License.
|
|
|
|
# You may obtain a copy of the License at
|
|
|
|
#
|
|
|
|
# http://www.apache.org/licenses/LICENSE-2.0
|
|
|
|
#
|
|
|
|
# Unless required by applicable law or agreed to in writing, software
|
|
|
|
# distributed under the License is distributed on an "AS IS" BASIS,
|
|
|
|
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
|
|
|
# See the License for the specific language governing permissions and
|
|
|
|
# limitations under the License.
|
|
|
|
|
|
|
|
module Mongo
|
|
|
|
class Pool
|
2011-08-22 15:52:11 +00:00
|
|
|
PING_ATTEMPTS = 6
|
2011-10-17 18:41:09 +00:00
|
|
|
MAX_PING_TIME = 1_000_000
|
2010-11-16 20:43:59 +00:00
|
|
|
|
2011-10-12 21:13:48 +00:00
|
|
|
attr_accessor :host, :port, :address,
|
2011-11-03 22:37:23 +00:00
|
|
|
:size, :timeout, :safe, :checked_out, :connection,
|
|
|
|
:sockets_low
|
2010-11-16 20:43:59 +00:00
|
|
|
|
|
|
|
# Create a new pool of connections.
|
2011-01-05 16:30:20 +00:00
|
|
|
def initialize(connection, host, port, opts={})
|
2010-11-16 20:43:59 +00:00
|
|
|
@connection = connection
|
|
|
|
|
|
|
|
@host, @port = host, port
|
|
|
|
|
2011-08-22 15:52:11 +00:00
|
|
|
# A Mongo::Node object.
|
|
|
|
@node = opts[:node]
|
|
|
|
|
2011-10-12 21:13:48 +00:00
|
|
|
# The string address
|
|
|
|
@address = "#{@host}:#{@port}"
|
|
|
|
|
2010-11-16 20:43:59 +00:00
|
|
|
# Pool size and timeout.
|
2011-11-03 22:37:23 +00:00
|
|
|
@size = opts[:size] || 10000
|
2011-01-05 16:30:20 +00:00
|
|
|
@timeout = opts[:timeout] || 5.0
|
2010-11-16 20:43:59 +00:00
|
|
|
|
|
|
|
# Mutex for synchronizing pool access
|
|
|
|
@connection_mutex = Mutex.new
|
|
|
|
|
|
|
|
# Condition variable for signal and wait
|
|
|
|
@queue = ConditionVariable.new
|
|
|
|
|
2011-01-31 19:47:05 +00:00
|
|
|
# Operations to perform on a socket
|
|
|
|
@socket_ops = Hash.new { |h, k| h[k] = [] }
|
|
|
|
|
2011-11-03 22:37:23 +00:00
|
|
|
@sockets_low = true
|
2010-11-16 20:43:59 +00:00
|
|
|
@sockets = []
|
2011-03-18 02:14:31 +00:00
|
|
|
@pids = {}
|
2010-11-16 20:43:59 +00:00
|
|
|
@checked_out = []
|
2011-11-03 22:37:23 +00:00
|
|
|
@threads = {}
|
2011-08-30 19:59:04 +00:00
|
|
|
@ping_time = nil
|
|
|
|
@last_ping = nil
|
2011-10-17 18:41:09 +00:00
|
|
|
@closed = false
|
2010-11-16 20:43:59 +00:00
|
|
|
end
|
|
|
|
|
|
|
|
def close
|
2011-09-13 21:50:01 +00:00
|
|
|
@connection_mutex.synchronize do
|
2011-10-14 13:52:11 +00:00
|
|
|
(@sockets - @checked_out).each do |sock|
|
2011-09-13 21:50:01 +00:00
|
|
|
begin
|
|
|
|
sock.close
|
|
|
|
rescue IOError => ex
|
|
|
|
warn "IOError when attempting to close socket connected to #{@host}:#{@port}: #{ex.inspect}"
|
|
|
|
end
|
2010-12-14 22:38:52 +00:00
|
|
|
end
|
2011-09-13 21:50:01 +00:00
|
|
|
@sockets.clear
|
|
|
|
@pids.clear
|
|
|
|
@checked_out.clear
|
2011-10-17 18:41:09 +00:00
|
|
|
@closed = true
|
2010-11-16 20:43:59 +00:00
|
|
|
end
|
|
|
|
end
|
|
|
|
|
2011-10-17 18:41:09 +00:00
|
|
|
def closed?
|
|
|
|
@closed
|
|
|
|
end
|
|
|
|
|
2011-11-03 22:37:23 +00:00
|
|
|
def sockets_low?
|
|
|
|
@sockets_low
|
|
|
|
end
|
|
|
|
|
2011-08-31 21:34:06 +00:00
|
|
|
def inspect
|
|
|
|
"#<Mongo::Pool:0x#{self.object_id.to_s(16)} @host=#{@host} @port=#{port} " +
|
2011-09-15 22:44:02 +00:00
|
|
|
"@ping_time=#{@ping_time} #{@checked_out.size}/#{@size} sockets available.>"
|
2011-08-31 21:34:06 +00:00
|
|
|
end
|
|
|
|
|
2011-08-24 22:34:00 +00:00
|
|
|
def host_string
|
|
|
|
"#{@host}:#{@port}"
|
|
|
|
end
|
|
|
|
|
2011-08-25 22:52:20 +00:00
|
|
|
def host_port
|
|
|
|
[@host, @port]
|
|
|
|
end
|
|
|
|
|
2011-08-30 19:59:04 +00:00
|
|
|
# Refresh ping time only if we haven't
|
|
|
|
# checked within the last five minutes.
|
|
|
|
def ping_time
|
|
|
|
if !@last_ping
|
|
|
|
@last_ping = Time.now
|
|
|
|
@ping_time = refresh_ping_time
|
|
|
|
elsif Time.now - @last_ping > 300
|
|
|
|
@last_ping = Time.now
|
|
|
|
@ping_time = refresh_ping_time
|
|
|
|
else
|
|
|
|
@ping_time
|
|
|
|
end
|
|
|
|
end
|
|
|
|
|
2011-08-22 15:52:11 +00:00
|
|
|
# Return the time it takes on average
|
|
|
|
# to do a round-trip against this node.
|
2011-08-30 19:59:04 +00:00
|
|
|
def refresh_ping_time
|
2011-08-24 22:34:00 +00:00
|
|
|
trials = []
|
2011-10-17 18:41:09 +00:00
|
|
|
PING_ATTEMPTS.times do
|
|
|
|
t1 = Time.now
|
|
|
|
if !self.ping
|
|
|
|
return MAX_PING_TIME
|
|
|
|
end
|
|
|
|
trials << (Time.now - t1) * 1000
|
2011-08-24 22:34:00 +00:00
|
|
|
end
|
2011-08-22 15:52:11 +00:00
|
|
|
|
|
|
|
trials.sort!
|
2011-08-30 19:59:04 +00:00
|
|
|
|
|
|
|
# Delete shortest and longest times
|
2011-08-22 15:52:11 +00:00
|
|
|
trials.delete_at(trials.length-1)
|
|
|
|
trials.delete_at(0)
|
|
|
|
|
|
|
|
total = 0.0
|
|
|
|
trials.each { |t| total += t }
|
|
|
|
|
2011-08-30 19:59:04 +00:00
|
|
|
(total / trials.length).ceil
|
2011-08-22 15:52:11 +00:00
|
|
|
end
|
|
|
|
|
2011-10-17 18:41:09 +00:00
|
|
|
def ping
|
|
|
|
begin
|
|
|
|
return self.connection['admin'].command({:ping => 1}, :socket => @node.socket)
|
|
|
|
rescue OperationFailure, SocketError, SystemCallError, IOError => ex
|
|
|
|
return false
|
|
|
|
end
|
|
|
|
end
|
|
|
|
|
2010-11-16 20:43:59 +00:00
|
|
|
# Return a socket to the pool.
|
|
|
|
def checkin(socket)
|
|
|
|
@connection_mutex.synchronize do
|
2011-11-03 22:37:23 +00:00
|
|
|
puts "deleting #{socket}, size: #{@checked_out.size}"
|
2010-11-16 20:43:59 +00:00
|
|
|
@checked_out.delete(socket)
|
2011-11-03 22:37:23 +00:00
|
|
|
puts "size now: #{@checked_out.size}"
|
2010-11-16 20:43:59 +00:00
|
|
|
@queue.signal
|
|
|
|
end
|
|
|
|
true
|
|
|
|
end
|
|
|
|
|
|
|
|
# Adds a new socket to the pool and checks it out.
|
|
|
|
#
|
|
|
|
# This method is called exclusively from #checkout;
|
|
|
|
# therefore, it runs within a mutex.
|
|
|
|
def checkout_new_socket
|
|
|
|
begin
|
2011-11-02 19:21:46 +00:00
|
|
|
puts "Creating connection in pool to #{@host}:#{@port}"
|
2011-08-26 21:35:40 +00:00
|
|
|
socket = self.connection.socket_class.new(@host, @port)
|
2011-08-25 15:27:58 +00:00
|
|
|
socket.setsockopt(Socket::IPPROTO_TCP, Socket::TCP_NODELAY, 1)
|
2010-11-16 20:43:59 +00:00
|
|
|
rescue => ex
|
2011-08-25 15:27:58 +00:00
|
|
|
socket.close if socket
|
2011-02-15 21:48:29 +00:00
|
|
|
raise ConnectionFailure, "Failed to connect to host #{@host} and port #{@port}: #{ex}"
|
2011-09-15 19:44:12 +00:00
|
|
|
@node.close if @node
|
2010-11-16 20:43:59 +00:00
|
|
|
end
|
2011-01-31 19:47:05 +00:00
|
|
|
|
|
|
|
# If any saved authentications exist, we want to apply those
|
|
|
|
# when creating new sockets.
|
|
|
|
@connection.apply_saved_authentication(:socket => socket)
|
|
|
|
|
2010-11-16 20:43:59 +00:00
|
|
|
@sockets << socket
|
2011-03-18 02:14:31 +00:00
|
|
|
@pids[socket] = Process.pid
|
2010-11-16 20:43:59 +00:00
|
|
|
@checked_out << socket
|
2011-11-03 22:37:23 +00:00
|
|
|
@threads[socket] = Thread.current.object_id
|
2010-11-16 20:43:59 +00:00
|
|
|
socket
|
|
|
|
end
|
|
|
|
|
2011-02-02 16:26:31 +00:00
|
|
|
# If a user calls DB#authenticate, and several sockets exist,
|
2011-01-31 19:47:05 +00:00
|
|
|
# then we need a way to apply the authentication on each socket.
|
|
|
|
# So we store the apply_authentication method, and this will be
|
|
|
|
# applied right before the next use of each socket.
|
|
|
|
def authenticate_existing
|
|
|
|
@connection_mutex.synchronize do
|
|
|
|
@sockets.each do |socket|
|
|
|
|
@socket_ops[socket] << Proc.new do
|
|
|
|
@connection.apply_saved_authentication(:socket => socket)
|
|
|
|
end
|
|
|
|
end
|
|
|
|
end
|
|
|
|
end
|
|
|
|
|
|
|
|
# Store the logout op for each existing socket to be applied before
|
|
|
|
# the next use of each socket.
|
|
|
|
def logout_existing(db)
|
|
|
|
@connection_mutex.synchronize do
|
|
|
|
@sockets.each do |socket|
|
|
|
|
@socket_ops[socket] << Proc.new do
|
|
|
|
@connection.db(db).issue_logout(:socket => socket)
|
|
|
|
end
|
|
|
|
end
|
|
|
|
end
|
|
|
|
end
|
|
|
|
|
2010-11-16 20:43:59 +00:00
|
|
|
# Checks out the first available socket from the pool.
|
|
|
|
#
|
2011-03-18 02:14:31 +00:00
|
|
|
# If the pid has changed, remove the socket and check out
|
|
|
|
# new one.
|
|
|
|
#
|
2010-11-16 20:43:59 +00:00
|
|
|
# This method is called exclusively from #checkout;
|
|
|
|
# therefore, it runs within a mutex.
|
|
|
|
def checkout_existing_socket
|
|
|
|
socket = (@sockets - @checked_out).first
|
2011-03-18 02:14:31 +00:00
|
|
|
if @pids[socket] != Process.pid
|
|
|
|
@pids[socket] = nil
|
|
|
|
@sockets.delete(socket)
|
2011-09-13 21:50:01 +00:00
|
|
|
socket.close if socket
|
2011-03-18 02:14:31 +00:00
|
|
|
checkout_new_socket
|
|
|
|
else
|
|
|
|
@checked_out << socket
|
2011-11-03 22:37:23 +00:00
|
|
|
@threads[socket] = Thread.current.object_id
|
2011-03-18 02:14:31 +00:00
|
|
|
socket
|
|
|
|
end
|
2010-11-16 20:43:59 +00:00
|
|
|
end
|
|
|
|
|
2011-11-03 22:37:23 +00:00
|
|
|
def cleanup
|
|
|
|
return unless @sockets.size > @size
|
|
|
|
puts "-----CLEANUP*****"
|
|
|
|
alive = {}
|
|
|
|
Thread.list.each do |t|
|
|
|
|
if t.alive?
|
|
|
|
alive[t.object_id] = true
|
|
|
|
end
|
|
|
|
end
|
|
|
|
|
|
|
|
@checked_out.each do |socket|
|
|
|
|
if !alive[@threads[socket]]
|
|
|
|
@checked_out.delete(socket)
|
|
|
|
if @sockets.size > @size
|
|
|
|
puts "CLEANING: #{socket}"
|
|
|
|
socket.close
|
|
|
|
@sockets.delete(socket)
|
|
|
|
end
|
|
|
|
end
|
|
|
|
end
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
end
|
|
|
|
|
2010-11-16 20:43:59 +00:00
|
|
|
# Check out an existing socket or create a new socket if the maximum
|
|
|
|
# pool size has not been exceeded. Otherwise, wait for the next
|
|
|
|
# available socket.
|
|
|
|
def checkout
|
|
|
|
@connection.connect if !@connection.connected?
|
|
|
|
start_time = Time.now
|
|
|
|
loop do
|
|
|
|
if (Time.now - start_time) > @timeout
|
|
|
|
raise ConnectionTimeoutError, "could not obtain connection within " +
|
|
|
|
"#{@timeout} seconds. The max pool size is currently #{@size}; " +
|
|
|
|
"consider increasing the pool size or timeout."
|
|
|
|
end
|
|
|
|
|
2011-11-03 22:37:23 +00:00
|
|
|
puts "CHECKING OUT"
|
|
|
|
#@sockets_low = @checked_out.size > @size / 2
|
2010-11-16 20:43:59 +00:00
|
|
|
@connection_mutex.synchronize do
|
2011-11-03 22:37:23 +00:00
|
|
|
if @sockets.size > 0.7 * @size
|
|
|
|
@sockets_low = true
|
|
|
|
else
|
|
|
|
@sockets_low = false
|
|
|
|
end
|
2010-11-16 20:43:59 +00:00
|
|
|
socket = if @checked_out.size < @sockets.size
|
2011-11-03 22:37:23 +00:00
|
|
|
p "checkout existing from size #{@sockets.size}"
|
2010-11-16 20:43:59 +00:00
|
|
|
checkout_existing_socket
|
2011-11-03 22:37:23 +00:00
|
|
|
else
|
2010-11-16 20:43:59 +00:00
|
|
|
checkout_new_socket
|
|
|
|
end
|
|
|
|
|
2011-01-31 19:47:05 +00:00
|
|
|
if socket
|
2011-11-03 22:37:23 +00:00
|
|
|
# This calls all procs, in order, scoped to existing sockets.
|
|
|
|
# At the moment, we use this to lazily authenticate and
|
|
|
|
# logout existing socket connections.
|
|
|
|
@socket_ops[socket].reject! do |op|
|
|
|
|
op.call
|
|
|
|
end
|
2010-11-16 20:43:59 +00:00
|
|
|
|
2011-01-31 19:47:05 +00:00
|
|
|
return socket
|
|
|
|
else
|
|
|
|
# Otherwise, wait
|
|
|
|
@queue.wait(@connection_mutex)
|
2010-11-16 20:43:59 +00:00
|
|
|
end
|
|
|
|
end
|
|
|
|
end
|
|
|
|
end
|
|
|
|
end
|
|
|
|
end
|