Skip to content

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
Browse files

Updated included memcache-client to the 1.5.0.5 version which include…

…s fixes from fiveruns and 37signals to deal with failover and timeouts (Joshua Sierles) [#1535 state:committed]

Signed-off-by: David Heinemeier Hansson <david@loudthinking.com>
  • Loading branch information...
commit e54f17920fddebdfa241a5d9253ad5810972af87 1 parent 9adcf95
@dhh dhh authored
View
2  activesupport/CHANGELOG
@@ -1,5 +1,7 @@
*2.3.0 [Edge]*
+* Updated included memcache-client to the 1.5.0.5 version which includes fixes from fiveruns and 37signals to deal with failover and timeouts #1535 [Joshua Sierles]
+
* Multibyte: add multibyte-safe Chars#ord rather than falling back to String#ord. #1483 [Jason Cheow]
* I18n support for Array#to_sentence. Introduces support.array.words_connector, .two_words_connector, and .last_word_connector translation keys. #1397 [Akira Matsuda]
View
4 activesupport/lib/active_support/vendor.rb
@@ -9,9 +9,9 @@
require 'builder'
begin
- gem 'memcache-client', '~> 1.5.1'
+ gem 'memcache-client', '~> 1.5.0.5'
rescue Gem::LoadError
- $:.unshift "#{File.dirname(__FILE__)}/vendor/memcache-client-1.5.1"
+ $:.unshift "#{File.dirname(__FILE__)}/vendor/memcache-client-1.5.0.5"
end
begin
View
461 ..._support/vendor/memcache-client-1.5.1/memcache.rb → ...upport/vendor/memcache-client-1.5.0.5/memcache.rb
@@ -26,36 +26,13 @@
# OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE,
# EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+$TESTING = defined?($TESTING) && $TESTING
require 'socket'
require 'thread'
require 'timeout'
require 'rubygems'
-
-class String
-
- ##
- # Uses the ITU-T polynomial in the CRC32 algorithm.
-
- def crc32_ITU_T
- n = length
- r = 0xFFFFFFFF
-
- n.times do |i|
- r ^= self[i]
- 8.times do
- if (r & 1) != 0 then
- r = (r>>1) ^ 0xEDB88320
- else
- r >>= 1
- end
- end
- end
-
- r ^ 0xFFFFFFFF
- end
-
-end
+require 'zlib'
##
# A Ruby client library for memcached.
@@ -69,7 +46,7 @@ class MemCache
##
# The version of MemCache you are using.
- VERSION = '1.5.0'
+ VERSION = '1.5.0.5'
##
# Default options for the cache object.
@@ -78,6 +55,7 @@ class MemCache
:namespace => nil,
:readonly => false,
:multithread => false,
+ :failover => true
}
##
@@ -113,6 +91,10 @@ class MemCache
attr_reader :servers
##
+ # Whether this client should failover reads and writes to another server
+
+ attr_accessor :failover
+ ##
# Accepts a list of +servers+ and a list of +opts+. +servers+ may be
# omitted. See +servers=+ for acceptable server list arguments.
#
@@ -148,6 +130,7 @@ def initialize(*args)
@namespace = opts[:namespace]
@readonly = opts[:readonly]
@multithread = opts[:multithread]
+ @failover = opts[:failover]
@mutex = Mutex.new if @multithread
@buckets = []
self.servers = servers
@@ -182,7 +165,7 @@ def readonly?
def servers=(servers)
# Create the server objects.
- @servers = servers.collect do |server|
+ @servers = Array(servers).collect do |server|
case server
when String
host, port, weight = server.split ':', 3
@@ -212,15 +195,12 @@ def servers=(servers)
# 0. +key+ can not be decremented below 0.
def decr(key, amount = 1)
- server, cache_key = request_setup key
-
- if @multithread then
- threadsafe_cache_decr server, cache_key, amount
- else
+ raise MemCacheError, "Update of readonly cache" if @readonly
+ with_server(key) do |server, cache_key|
cache_decr server, cache_key, amount
end
- rescue TypeError, SocketError, SystemCallError, IOError => err
- handle_error server, err
+ rescue TypeError => err
+ handle_error nil, err
end
##
@@ -228,21 +208,14 @@ def decr(key, amount = 1)
# unmarshalled.
def get(key, raw = false)
- server, cache_key = request_setup key
-
- value = if @multithread then
- threadsafe_cache_get server, cache_key
- else
- cache_get server, cache_key
- end
-
- return nil if value.nil?
-
- value = Marshal.load value unless raw
-
- return value
- rescue TypeError, SocketError, SystemCallError, IOError => err
- handle_error server, err
+ with_server(key) do |server, cache_key|
+ value = cache_get server, cache_key
+ return nil if value.nil?
+ value = Marshal.load value unless raw
+ return value
+ end
+ rescue TypeError => err
+ handle_error nil, err
end
##
@@ -280,36 +253,29 @@ def get_multi(*keys)
server_keys.each do |server, keys_for_server|
keys_for_server = keys_for_server.join ' '
- values = if @multithread then
- threadsafe_cache_get_multi server, keys_for_server
- else
- cache_get_multi server, keys_for_server
- end
+ values = cache_get_multi server, keys_for_server
values.each do |key, value|
results[cache_keys[key]] = Marshal.load value
end
end
return results
- rescue TypeError, SocketError, SystemCallError, IOError => err
- handle_error server, err
+ rescue TypeError, IndexError => err
+ handle_error nil, err
end
##
- # Increments the value for +key+ by +amount+ and retruns the new value.
+ # Increments the value for +key+ by +amount+ and returns the new value.
# +key+ must already exist. If +key+ is not an integer, it is assumed to be
# 0.
def incr(key, amount = 1)
- server, cache_key = request_setup key
-
- if @multithread then
- threadsafe_cache_incr server, cache_key, amount
- else
+ raise MemCacheError, "Update of readonly cache" if @readonly
+ with_server(key) do |server, cache_key|
cache_incr server, cache_key, amount
end
- rescue TypeError, SocketError, SystemCallError, IOError => err
- handle_error server, err
+ rescue TypeError => err
+ handle_error nil, err
end
##
@@ -321,23 +287,23 @@ def incr(key, amount = 1)
def set(key, value, expiry = 0, raw = false)
raise MemCacheError, "Update of readonly cache" if @readonly
- server, cache_key = request_setup key
- socket = server.socket
+ with_server(key) do |server, cache_key|
- value = Marshal.dump value unless raw
- command = "set #{cache_key} 0 #{expiry} #{value.size}\r\n#{value}\r\n"
+ value = Marshal.dump value unless raw
+ command = "set #{cache_key} 0 #{expiry} #{value.to_s.size}\r\n#{value}\r\n"
- begin
- @mutex.lock if @multithread
- socket.write command
- result = socket.gets
- raise_on_error_response! result
- result
- rescue SocketError, SystemCallError, IOError => err
- server.close
- raise MemCacheError, err.message
- ensure
- @mutex.unlock if @multithread
+ with_socket_management(server) do |socket|
+ socket.write command
+ result = socket.gets
+ raise_on_error_response! result
+
+ if result.nil?
+ server.close
+ raise MemCacheError, "lost connection to #{server.host}:#{server.port}"
+ end
+
+ result
+ end
end
end
@@ -351,23 +317,16 @@ def set(key, value, expiry = 0, raw = false)
def add(key, value, expiry = 0, raw = false)
raise MemCacheError, "Update of readonly cache" if @readonly
- server, cache_key = request_setup key
- socket = server.socket
-
- value = Marshal.dump value unless raw
- command = "add #{cache_key} 0 #{expiry} #{value.size}\r\n#{value}\r\n"
-
- begin
- @mutex.lock if @multithread
- socket.write command
- result = socket.gets
- raise_on_error_response! result
- result
- rescue SocketError, SystemCallError, IOError => err
- server.close
- raise MemCacheError, err.message
- ensure
- @mutex.unlock if @multithread
+ with_server(key) do |server, cache_key|
+ value = Marshal.dump value unless raw
+ command = "add #{cache_key} 0 #{expiry} #{value.size}\r\n#{value}\r\n"
+
+ with_socket_management(server) do |socket|
+ socket.write command
+ result = socket.gets
+ raise_on_error_response! result
+ result
+ end
end
end
@@ -375,26 +334,15 @@ def add(key, value, expiry = 0, raw = false)
# Removes +key+ from the cache in +expiry+ seconds.
def delete(key, expiry = 0)
- @mutex.lock if @multithread
-
- raise MemCacheError, "No active servers" unless active?
- cache_key = make_cache_key key
- server = get_server_for_key cache_key
-
- sock = server.socket
- raise MemCacheError, "No connection to server" if sock.nil?
-
- begin
- sock.write "delete #{cache_key} #{expiry}\r\n"
- result = sock.gets
- raise_on_error_response! result
- result
- rescue SocketError, SystemCallError, IOError => err
- server.close
- raise MemCacheError, err.message
+ raise MemCacheError, "Update of readonly cache" if @readonly
+ with_server(key) do |server, cache_key|
+ with_socket_management(server) do |socket|
+ socket.write "delete #{cache_key} #{expiry}\r\n"
+ result = socket.gets
+ raise_on_error_response! result
+ result
+ end
end
- ensure
- @mutex.unlock if @multithread
end
##
@@ -403,21 +351,19 @@ def delete(key, expiry = 0)
def flush_all
raise MemCacheError, 'No active servers' unless active?
raise MemCacheError, "Update of readonly cache" if @readonly
+
begin
@mutex.lock if @multithread
@servers.each do |server|
- begin
- sock = server.socket
- raise MemCacheError, "No connection to server" if sock.nil?
- sock.write "flush_all\r\n"
- result = sock.gets
+ with_socket_management(server) do |socket|
+ socket.write "flush_all\r\n"
+ result = socket.gets
raise_on_error_response! result
result
- rescue SocketError, SystemCallError, IOError => err
- server.close
- raise MemCacheError, err.message
end
end
+ rescue IndexError => err
+ handle_error nil, err
ensure
@mutex.unlock if @multithread
end
@@ -469,14 +415,13 @@ def stats
server_stats = {}
@servers.each do |server|
- sock = server.socket
- raise MemCacheError, "No connection to server" if sock.nil?
+ next unless server.alive?
- value = nil
- begin
- sock.write "stats\r\n"
+ with_socket_management(server) do |socket|
+ value = nil
+ socket.write "stats\r\n"
stats = {}
- while line = sock.gets do
+ while line = socket.gets do
raise_on_error_response! line
break if line == "END\r\n"
if line =~ /\ASTAT ([\w]+) ([\w\.\:]+)/ then
@@ -498,12 +443,10 @@ def stats
end
end
server_stats["#{server.host}:#{server.port}"] = stats
- rescue SocketError, SystemCallError, IOError => err
- server.close
- raise MemCacheError, err.message
end
end
+ raise MemCacheError, "No active servers" if server_stats.empty?
server_stats
end
@@ -520,7 +463,7 @@ def []=(key, value)
set key, value
end
- protected
+ protected unless $TESTING
##
# Create a key for the cache, incorporating the namespace qualifier if
@@ -537,7 +480,7 @@ def make_cache_key(key)
##
# Pick a server to handle the request based on a hash of the key.
- def get_server_for_key(key)
+ def get_server_for_key(key, options = {})
raise ArgumentError, "illegal character in key #{key.inspect}" if
key =~ /\s/
raise ArgumentError, "key too long #{key.inspect}" if key.length > 250
@@ -545,13 +488,17 @@ def get_server_for_key(key)
return @servers.first if @servers.length == 1
hkey = hash_for key
-
- 20.times do |try|
- server = @buckets[hkey % @buckets.nitems]
- return server if server.alive?
- hkey += hash_for "#{try}#{key}"
+
+ if @failover
+ 20.times do |try|
+ server = @buckets[hkey % @buckets.compact.size]
+ return server if server.alive?
+ hkey += hash_for "#{try}#{key}"
+ end
+ else
+ return @buckets[hkey % @buckets.compact.size]
end
-
+
raise MemCacheError, "No servers available"
end
@@ -560,7 +507,7 @@ def get_server_for_key(key)
# sketchy for down servers).
def hash_for(key)
- (key.crc32_ITU_T >> 16) & 0x7fff
+ (Zlib.crc32(key) >> 16) & 0x7fff
end
##
@@ -568,12 +515,13 @@ def hash_for(key)
# found.
def cache_decr(server, cache_key, amount)
- socket = server.socket
- socket.write "decr #{cache_key} #{amount}\r\n"
- text = socket.gets
- raise_on_error_response! text
- return nil if text == "NOT_FOUND\r\n"
- return text.to_i
+ with_socket_management(server) do |socket|
+ socket.write "decr #{cache_key} #{amount}\r\n"
+ text = socket.gets
+ raise_on_error_response! text
+ return nil if text == "NOT_FOUND\r\n"
+ return text.to_i
+ end
end
##
@@ -581,52 +529,54 @@ def cache_decr(server, cache_key, amount)
# miss.
def cache_get(server, cache_key)
- socket = server.socket
- socket.write "get #{cache_key}\r\n"
- keyline = socket.gets # "VALUE <key> <flags> <bytes>\r\n"
+ with_socket_management(server) do |socket|
+ socket.write "get #{cache_key}\r\n"
+ keyline = socket.gets # "VALUE <key> <flags> <bytes>\r\n"
- if keyline.nil? then
- server.close
- raise MemCacheError, "lost connection to #{server.host}:#{server.port}"
- end
+ if keyline.nil? then
+ server.close
+ raise MemCacheError, "lost connection to #{server.host}:#{server.port}"
+ end
- raise_on_error_response! keyline
- return nil if keyline == "END\r\n"
+ raise_on_error_response! keyline
+ return nil if keyline == "END\r\n"
- unless keyline =~ /(\d+)\r/ then
- server.close
- raise MemCacheError, "unexpected response #{keyline.inspect}"
+ unless keyline =~ /(\d+)\r/ then
+ server.close
+ raise MemCacheError, "unexpected response #{keyline.inspect}"
+ end
+ value = socket.read $1.to_i
+ socket.read 2 # "\r\n"
+ socket.gets # "END\r\n"
+ return value
end
- value = socket.read $1.to_i
- socket.read 2 # "\r\n"
- socket.gets # "END\r\n"
- return value
end
##
# Fetches +cache_keys+ from +server+ using a multi-get.
def cache_get_multi(server, cache_keys)
- values = {}
- socket = server.socket
- socket.write "get #{cache_keys}\r\n"
+ with_socket_management(server) do |socket|
+ values = {}
+ socket.write "get #{cache_keys}\r\n"
- while keyline = socket.gets do
- return values if keyline == "END\r\n"
- raise_on_error_response! keyline
+ while keyline = socket.gets do
+ return values if keyline == "END\r\n"
+ raise_on_error_response! keyline
- unless keyline =~ /\AVALUE (.+) (.+) (.+)/ then
- server.close
- raise MemCacheError, "unexpected response #{keyline.inspect}"
+ unless keyline =~ /\AVALUE (.+) (.+) (.+)/ then
+ server.close
+ raise MemCacheError, "unexpected response #{keyline.inspect}"
+ end
+
+ key, data_length = $1, $3
+ values[$1] = socket.read data_length.to_i
+ socket.read(2) # "\r\n"
end
- key, data_length = $1, $3
- values[$1] = socket.read data_length.to_i
- socket.read(2) # "\r\n"
+ server.close
+ raise MemCacheError, "lost connection to #{server.host}:#{server.port}" # TODO: retry here too
end
-
- server.close
- raise MemCacheError, "lost connection to #{server.host}:#{server.port}"
end
##
@@ -634,18 +584,76 @@ def cache_get_multi(server, cache_keys)
# found.
def cache_incr(server, cache_key, amount)
- socket = server.socket
- socket.write "incr #{cache_key} #{amount}\r\n"
- text = socket.gets
- raise_on_error_response! text
- return nil if text == "NOT_FOUND\r\n"
- return text.to_i
+ with_socket_management(server) do |socket|
+ socket.write "incr #{cache_key} #{amount}\r\n"
+ text = socket.gets
+ raise_on_error_response! text
+ return nil if text == "NOT_FOUND\r\n"
+ return text.to_i
+ end
+ end
+
+ ##
+ # Gets or creates a socket connected to the given server, and yields it
+ # to the block, wrapped in a mutex synchronization if @multithread is true.
+ #
+ # If a socket error (SocketError, SystemCallError, IOError) or protocol error
+ # (MemCacheError) is raised by the block, closes the socket, attempts to
+ # connect again, and retries the block (once). If an error is again raised,
+ # reraises it as MemCacheError.
+ #
+ # If unable to connect to the server (or if in the reconnect wait period),
+ # raises MemCacheError. Note that the socket connect code marks a server
+ # dead for a timeout period, so retrying does not apply to connection attempt
+ # failures (but does still apply to unexpectedly lost connections etc.).
+
+ def with_socket_management(server, &block)
+ @mutex.lock if @multithread
+ retried = false
+
+ begin
+ socket = server.socket
+
+ # Raise an IndexError to show this server is out of whack. If were inside
+ # a with_server block, we'll catch it and attempt to restart the operation.
+
+ raise IndexError, "No connection to server (#{server.status})" if socket.nil?
+
+ block.call(socket)
+
+ rescue SocketError => err
+ server.mark_dead(err.message)
+ handle_error(server, err)
+
+ rescue MemCacheError, SocketError, SystemCallError, IOError => err
+ handle_error(server, err) if retried || socket.nil?
+ retried = true
+ retry
+ end
+ ensure
+ @mutex.unlock if @multithread
+ end
+
+ def with_server(key)
+ retried = false
+ begin
+ server, cache_key = request_setup(key)
+ yield server, cache_key
+ rescue IndexError => e
+ if !retried && @servers.size > 1
+ puts "Connection to server #{server.inspect} DIED! Retrying operation..."
+ retried = true
+ retry
+ end
+ handle_error(nil, e)
+ end
end
##
# Handles +error+ from +server+.
def handle_error(server, error)
+ raise error if error.is_a?(MemCacheError)
server.close if server
new_error = MemCacheError.new error.message
new_error.set_backtrace error.backtrace
@@ -660,45 +668,15 @@ def request_setup(key)
raise MemCacheError, 'No active servers' unless active?
cache_key = make_cache_key key
server = get_server_for_key cache_key
- raise MemCacheError, 'No connection to server' if server.socket.nil?
return server, cache_key
end
- def threadsafe_cache_decr(server, cache_key, amount) # :nodoc:
- @mutex.lock
- cache_decr server, cache_key, amount
- ensure
- @mutex.unlock
- end
-
- def threadsafe_cache_get(server, cache_key) # :nodoc:
- @mutex.lock
- cache_get server, cache_key
- ensure
- @mutex.unlock
- end
-
- def threadsafe_cache_get_multi(socket, cache_keys) # :nodoc:
- @mutex.lock
- cache_get_multi socket, cache_keys
- ensure
- @mutex.unlock
- end
-
- def threadsafe_cache_incr(server, cache_key, amount) # :nodoc:
- @mutex.lock
- cache_incr server, cache_key, amount
- ensure
- @mutex.unlock
- end
-
def raise_on_error_response!(response)
- if response =~ /\A(?:CLIENT_|SERVER_)?ERROR (.*)/
+ if response =~ /\A(?:CLIENT_|SERVER_)?ERROR(.*)/
raise MemCacheError, $1.strip
end
end
-
##
# This class represents a memcached server instance.
@@ -712,6 +690,13 @@ class Server
CONNECT_TIMEOUT = 0.25
##
+ # The amount of time to wait for a response from a memcached server.
+ # If a response isn't received within this time limit,
+ # the server will be marked as down.
+
+ SOCKET_TIMEOUT = 0.5
+
+ ##
# The amount of time to wait before attempting to re-establish a
# connection with a server that is marked dead.
@@ -795,9 +780,9 @@ def socket
# Attempt to connect if not already connected.
begin
- @sock = timeout CONNECT_TIMEOUT do
- TCPSocket.new @host, @port
- end
+
+ @sock = TCPTimeoutSocket.new @host, @port
+
if Socket.constants.include? 'TCP_NODELAY' then
@sock.setsockopt Socket::IPPROTO_TCP, Socket::TCP_NODELAY, 1
end
@@ -826,8 +811,6 @@ def close
@mutex.unlock if @multithread
end
- private
-
##
# Mark the server as dead and close its socket.
@@ -836,8 +819,9 @@ def mark_dead(reason = "Unknown error")
@sock = nil
@retry = Time.now + RETRY_DELAY
- @status = sprintf "DEAD: %s, will retry at %s", reason, @retry
+ @status = sprintf "%s:%s DEAD: %s, will retry at %s", @host, @port, reason, @retry
end
+
end
##
@@ -847,3 +831,38 @@ class MemCacheError < RuntimeError; end
end
+# TCPSocket facade class which implements timeouts.
+class TCPTimeoutSocket
+ def initialize(*args)
+ Timeout::timeout(MemCache::Server::CONNECT_TIMEOUT, SocketError) do
+ @sock = TCPSocket.new(*args)
+ @len = MemCache::Server::SOCKET_TIMEOUT.to_f || 0.5
+ end
+ end
+
+ def write(*args)
+ Timeout::timeout(@len, SocketError) do
+ @sock.write(*args)
+ end
+ end
+
+ def gets(*args)
+ Timeout::timeout(@len, SocketError) do
+ @sock.gets(*args)
+ end
+ end
+
+ def read(*args)
+ Timeout::timeout(@len, SocketError) do
+ @sock.read(*args)
+ end
+ end
+
+ def _socket
+ @sock
+ end
+
+ def method_missing(meth, *args)
+ @sock.__send__(meth, *args)
+ end
+end
Please sign in to comment.
Something went wrong with that request. Please try again.