Merge pull request #651 from sundream/upstream

redis cluster client

lualib/skynet/db/redis/cluster.lua

@@ -0,0 +1,383 @@
+-- a simple redis-cluster client
+-- rewrite from https://github.com/antirez/redis-rb-cluster
+
+local skynet = require "skynet"
+local redis = require "skynet.db.redis"
+local crc16 = require "skynet.db.redis.crc16"
+
+local RedisClusterHashSlots = 16384
+local RedisClusterRequestTTL = 16
+
+
+local _M = {}
+
+local rediscluster = {}
+rediscluster.__index = rediscluster
+_M.rediscluster = rediscluster
+
+function _M.new(startup_nodes,opt)
+	if #startup_nodes == 0 then
+		startup_nodes = {startup_nodes,}
+	end
+	opt = opt or {}
+	local self = {
+		startup_nodes = startup_nodes,
+		max_connections = opt.max_connections or 16,
+		connections = setmetatable({},{__mode = "kv"}),
+		opt = opt,
+		refresh_table_asap = false,
+	}
+	setmetatable(self,rediscluster)
+	self:initialize_slots_cache()
+	return self
+end
+
+local function nodename(node)
+	return string.format("%s:%s",node.host,node.port)
+end
+
+function rediscluster:get_redis_link(node)
+	local conf = {
+		host = node.host,
+		port = node.port,
+		auth = self.opt.auth,
+		db = self.opt.db or 0,
+	}
+	return redis.connect(conf)
+end
+ 
+-- Given a node (that is just a Ruby hash) give it a name just
+-- concatenating the host and port. We use the node name as a key
+-- to cache connections to that node.
+function rediscluster:set_node_name(node)
+	if not node.name then
+		node.name = nodename(node)
+	end
+	if not node.slaves then
+		local oldnode = self.name_node[node.name]
+		if oldnode then
+			node.slaves = oldnode.slaves
+		end
+	end
+	self.name_node[node.name] = node
+end
+
+-- Contact the startup nodes and try to fetch the hash slots -> instances
+-- map in order to initialize the @slots hash.
+function rediscluster:initialize_slots_cache()
+	self.slots = {}
+	self.nodes = {}
+	self.name_node = {}
+	for _,startup_node in ipairs(self.startup_nodes) do
+		local ok = pcall(function ()
+			local name = nodename(startup_node)
+			local conn = self.connections[name] or self:get_redis_link(startup_node)
+			local list = conn:cluster("slots")
+			for _,result in ipairs(list) do
+				local ip,port,runid = table.unpack(result[3])
+				local master_node = {
+					host = ip,
+					port = port,
+					runid = runid,
+					slaves = {},
+				}
+				self:set_node_name(master_node)
+				for i=4,#result do
+					local ip,port,runid = table.unpack(result[i])
+					local slave_node = {
+						host = ip,
+						port = port,
+						runid = runid,
+					}
+					self:set_node_name(slave_node)
+					table.insert(master_node.slaves,slave_node)
+				end
+				for slot=tonumber(result[1]),tonumber(result[2]) do
+					table.insert(self.nodes,master_node)
+					self.slots[slot] = master_node
+				end
+			end
+			self.refresh_table_asap = false
+			if not self.connections[name] then
+				self.connections[name] = conn
+			end
+		end)
+		-- Exit the loop as long as the first node replies
+		if ok then
+			break
+		end
+	end
+end
+
+-- Flush the cache, mostly useful for debugging when we want to force
+-- redirection.
+function rediscluster:flush_slots_cache()
+	self.slots = {}
+end
+
+-- Return the hash slot from the key.
+function rediscluster:keyslot(key)
+	-- Only hash what is inside {...} if there is such a pattern in the key.
+	-- Note that the specification requires the content that is between
+	-- the first { and the first } after the first {. If we found {} without
+	-- nothing in the middle, the whole key is hashed as usually.
+	local startpos = string.find(key,"{",1,true)
+	if startpos then
+		local endpos = string.find(key,"}",startpos+1,true)
+		if endpos and endpos ~= startpos + 1 then
+			key = string.sub(key,startpos+1,endpos-1)
+		end
+	end
+	return crc16(key) % RedisClusterHashSlots
+end
+
+-- Return the first key in the command arguments.
+--
+-- Currently we just return argv[1], that is, the first argument
+-- after the command name.
+--
+-- This is indeed the key for most commands, and when it is not true
+-- the cluster redirection will point us to the right node anyway.
+--
+-- For commands we want to explicitly bad as they don't make sense
+-- in the context of cluster, nil is returned.
+function rediscluster:get_key_from_command(argv)
+	local cmd,key = table.unpack(argv)
+	cmd = string.lower(cmd)
+	if cmd == "info" or
+		cmd == "multi" or
+		cmd == "exec" or
+		cmd == "slaveof" or
+		cmd == "config" or
+		cmd == "shutdown" then
+		return nil
+	end
+	-- Unknown commands, and all the commands having the key
+	-- as first argument are handled here:
+	-- set, get, ...
+	return key
+end
+
+-- If the current number of connections is already the maximum number
+-- allowed, close a random connection. This should be called every time
+-- we cache a new connection in the @connections hash.
+function rediscluster:close_existing_connection()
+	local length = 0
+	for name,conn in pairs(self.connections) do
+		length = length + 1
+	end
+	if length >= self.max_connections then
+		pcall(function ()
+			local name,conn = next(self.connections)
+			self.connections[name] = nil
+			conn:disconnect()
+		end)
+	end
+end
+
+function rediscluster:close_all_connection()
+	local connections = self.connections
+	self.connections = setmetatable({},{__mode = "kv"})
+	for name,conn in pairs(connections) do
+		pcall(conn.disconnect,conn)
+	end
+end
+
+function rediscluster:get_connection(node)
+	if type(node) == "string" then
+		local ip,port = string.match(node,"^([^:]+):([^:]+)$")
+		node = {host=ip,port=port}
+	end
+	local name = node.name or nodename(node)
+	local conn = self.connections[name]
+	if not conn then
+		conn = self:get_redis_link(node)
+		self.connections[name] = conn
+	end
+	return self.connections[name]
+end
+
+-- Return a link to a random node, or raise an error if no node can be
+-- contacted. This function is only called when we can't reach the node
+-- associated with a given hash slot, or when we don't know the right
+-- mapping.
+-- The function will try to get a successful reply to the PING command,
+-- otherwise the next node is tried.
+function rediscluster:get_random_connection()
+	-- shuffle
+	local shuffle_idx = {}
+	local startpos = 1
+	local endpos = #self.nodes
+	for i=startpos,endpos do
+		shuffle_idx[i] = i
+	end
+	for i=startpos,endpos do
+		local idx = math.random(i,endpos)
+		local tmp = shuffle_idx[i]
+		shuffle_idx[i] = shuffle_idx[idx]
+		shuffle_idx[idx] = tmp
+	end
+	for i,idx in ipairs(shuffle_idx) do
+		local ok,conn = pcall(function ()
+			local node = self.nodes[idx]
+			local conn = self.connections[node.name]
+			if not conn then
+				-- Connect the node if it is not connected
+				conn = self:get_redis_link(node)
+				if conn:ping() == "PONG" then
+					self:close_existing_connection()
+					self.connections[node.name] = conn
+					return conn
+				else
+					-- If the connection is not good close it ASAP in order
+					-- to avoid waiting for the GC finalizer. File
+					-- descriptors are a rare resource.
+					conn:disconnect()
+				end
+			else
+				-- The node was already connected, test the connection.
+				if conn:ping() == "PONG" then
+					return conn
+				end
+			end
+		end)
+		if ok and conn then
+			return conn
+		end
+	end
+	error("Can't reach a single startup node.")
+end
+
+-- Given a slot return the link (Redis instance) to the mapped node.
+-- Make sure to create a connection with the node if we don't have
+-- one.
+function rediscluster:get_connection_by_slot(slot)
+	local node = self.slots[slot]
+	-- If we don't know what the mapping is, return a random node.
+	if not node then
+		return self:get_random_connection()
+	end
+	if not self.connections[node.name] then
+		local ok = pcall(function ()
+			self:close_existing_connection()
+			self.connections[node.name] = self:get_redis_link(node)
+		end)
+		if not ok then
+			if self.opt.read_slave and node.slaves and #node.slaves > 0 then
+				local slave_node = node.slaves[math.random(1,#node.slaves)]
+				local ok2,conn = pcall(self.get_connection,self,slave_node)
+				if ok2 then
+					conn:readonly()		-- allow this connection read-slave
+					return conn
+				end
+			end
+			-- This will probably never happen with recent redis-rb
+			-- versions because the connection is enstablished in a lazy
+			-- way only when a command is called. However it is wise to
+			-- handle an instance creation error of some kind.
+			return self:get_random_connection()
+		end
+	end
+	return self.connections[node.name]
+end
+
+-- Dispatch commands.
+function rediscluster:call(...)
+	local argv = table.pack(...)
+	if self.refresh_table_asap then
+		self:initialize_slots_cache()
+	end
+	local ttl = RedisClusterRequestTTL	-- Max number of redirections
+	local err
+	local asking = false
+	local try_random_node = false
+	while ttl > 0 do
+		ttl = ttl - 1
+		local key = self:get_key_from_command(argv)
+		if not key then
+			error("No way to dispatch this command to Redis Cluster: " .. tostring(argv[1]))
+		end
+		local conn
+		local slot = self:keyslot(key)
+		if asking then
+			conn = self:get_connection(asking)
+		elseif try_random_node then
+			conn = self:get_random_connection()
+			try_random_node = false
+		else
+			conn = self:get_connection_by_slot(slot)
+		end
+		local result = {pcall(function ()
+			-- TODO: use pipelining to send asking and save a rtt.
+			if asking then
+				conn:asking()
+			end
+			asking = false
+			local cmd = argv[1]
+			local func = conn[cmd]
+			return func(conn,table.unpack(argv,2))
+		end)}
+		local ok = result[1]
+		if not ok then
+			err = table.unpack(result,2)
+			err = tostring(err)
+			if err == "[Error: socket]" then
+				-- may be nerver come here?
+				try_random_node = true
+				if ttl < RedisClusterRequestTTL/2 then
+					skynet.sleep(10)
+				end
+			else
+				-- err: ./lualib/skynet/socketchannel.lua:371: xxx
+				err = string.match(err,".+:%d+:%s(.*)$") or err
+				local errlist = {}
+				for e in string.gmatch(err,"([^%s]+)%s?") do
+					table.insert(errlist,e)
+				end
+				if (errlist[1] ~= "MOVED" and errlist[1] ~= "ASK") then
+					error(err)
+				else
+					if errlist[1] == "ASK" then
+						asking = true
+					else
+						-- Serve replied with MOVED. It's better for us to
+						-- ask for CLUSTER SLOTS the next time.
+						self.refresh_table_asap = true
+					end
+					local newslot = tonumber(errlist[2])
+					local node_ip,node_port = string.match(errlist[3],"^([^:]+):([^:]+)$")
+					local node = {
+						host = node_ip,
+						port = tonumber(node_port),
+					}
+					if not asking then
+						self:set_node_name(node)
+						self.slots[newslot] = node
+					else
+						asking = node
+					end
+				end
+			end
+		else
+			return table.unpack(result,2)
+		end
+	end
+	error(string.format("Too many Cluster redirections?,maybe node is disconnected (last error: %q)",err))
+end
+
+-- Currently we handle all the commands using method_missing for
+-- simplicity. For a Cluster client actually it will be better to have
+-- every single command as a method with the right arity and possibly
+-- additional checks (example: RPOPLPUSH with same src/dst key, SORT
+-- without GET or BY, and so forth).
+setmetatable(rediscluster,{
+	__index = function (t,cmd)
+		t[cmd] = function (self,...)
+			return self:call(cmd,...)
+		end
+		return t[cmd]
+	end,
+})
+
+
+return _M

lualib/skynet/db/redis/crc16.lua

@@ -0,0 +1,62 @@
+--/*
+-- This is the CRC16 algorithm used by Redis Cluster to hash keys.
+-- Implementation according to CCITT standards.
+--
+-- This is actually the XMODEM CRC 16 algorithm, using the
+-- following parameters:
+--
+-- Name                       : "XMODEM", also known as "ZMODEM", "CRC-16/ACORN"
+-- Width                      : 16 bit
+-- Poly                       : 1021 (That is actually x^16 + x^12 + x^5 + 1)
+-- Initialization             : 0000
+-- Reflect Input byte         : False
+-- Reflect Output CRC         : False
+-- Xor constant to output CRC : 0000
+-- Output for "123456789"     : 31C3
+--*/
+
+
+
+local XMODEMCRC16Lookup = {
+	0x0000,0x1021,0x2042,0x3063,0x4084,0x50a5,0x60c6,0x70e7,
+	0x8108,0x9129,0xa14a,0xb16b,0xc18c,0xd1ad,0xe1ce,0xf1ef,
+	0x1231,0x0210,0x3273,0x2252,0x52b5,0x4294,0x72f7,0x62d6,
+	0x9339,0x8318,0xb37b,0xa35a,0xd3bd,0xc39c,0xf3ff,0xe3de,
+	0x2462,0x3443,0x0420,0x1401,0x64e6,0x74c7,0x44a4,0x5485,
+	0xa56a,0xb54b,0x8528,0x9509,0xe5ee,0xf5cf,0xc5ac,0xd58d,
+	0x3653,0x2672,0x1611,0x0630,0x76d7,0x66f6,0x5695,0x46b4,
+	0xb75b,0xa77a,0x9719,0x8738,0xf7df,0xe7fe,0xd79d,0xc7bc,
+	0x48c4,0x58e5,0x6886,0x78a7,0x0840,0x1861,0x2802,0x3823,
+	0xc9cc,0xd9ed,0xe98e,0xf9af,0x8948,0x9969,0xa90a,0xb92b,
+	0x5af5,0x4ad4,0x7ab7,0x6a96,0x1a71,0x0a50,0x3a33,0x2a12,
+	0xdbfd,0xcbdc,0xfbbf,0xeb9e,0x9b79,0x8b58,0xbb3b,0xab1a,
+	0x6ca6,0x7c87,0x4ce4,0x5cc5,0x2c22,0x3c03,0x0c60,0x1c41,
+	0xedae,0xfd8f,0xcdec,0xddcd,0xad2a,0xbd0b,0x8d68,0x9d49,
+	0x7e97,0x6eb6,0x5ed5,0x4ef4,0x3e13,0x2e32,0x1e51,0x0e70,
+	0xff9f,0xefbe,0xdfdd,0xcffc,0xbf1b,0xaf3a,0x9f59,0x8f78,
+	0x9188,0x81a9,0xb1ca,0xa1eb,0xd10c,0xc12d,0xf14e,0xe16f,
+	0x1080,0x00a1,0x30c2,0x20e3,0x5004,0x4025,0x7046,0x6067,
+	0x83b9,0x9398,0xa3fb,0xb3da,0xc33d,0xd31c,0xe37f,0xf35e,
+	0x02b1,0x1290,0x22f3,0x32d2,0x4235,0x5214,0x6277,0x7256,
+	0xb5ea,0xa5cb,0x95a8,0x8589,0xf56e,0xe54f,0xd52c,0xc50d,
+	0x34e2,0x24c3,0x14a0,0x0481,0x7466,0x6447,0x5424,0x4405,
+	0xa7db,0xb7fa,0x8799,0x97b8,0xe75f,0xf77e,0xc71d,0xd73c,
+	0x26d3,0x36f2,0x0691,0x16b0,0x6657,0x7676,0x4615,0x5634,
+	0xd94c,0xc96d,0xf90e,0xe92f,0x99c8,0x89e9,0xb98a,0xa9ab,
+	0x5844,0x4865,0x7806,0x6827,0x18c0,0x08e1,0x3882,0x28a3,
+	0xcb7d,0xdb5c,0xeb3f,0xfb1e,0x8bf9,0x9bd8,0xabbb,0xbb9a,
+	0x4a75,0x5a54,0x6a37,0x7a16,0x0af1,0x1ad0,0x2ab3,0x3a92,
+	0xfd2e,0xed0f,0xdd6c,0xcd4d,0xbdaa,0xad8b,0x9de8,0x8dc9,
+	0x7c26,0x6c07,0x5c64,0x4c45,0x3ca2,0x2c83,0x1ce0,0x0cc1,
+	0xef1f,0xff3e,0xcf5d,0xdf7c,0xaf9b,0xbfba,0x8fd9,0x9ff8,
+	0x6e17,0x7e36,0x4e55,0x5e74,0x2e93,0x3eb2,0x0ed1,0x1ef0
+}
+
+return function(bytes)
+	local crc = 0
+	for i=1,#bytes do
+		local b = string.byte(bytes,i,i)
+		crc = ((crc<<8) & 0xffff) ~ XMODEMCRC16Lookup[(((crc>>8)~b) & 0xff) + 1]
+	end
+	return tonumber(crc)
+end

test/testrediscluster.lua

@@ -0,0 +1,114 @@
+local skynet = require "skynet"
+local rediscluster = require "skynet.db.redis.cluster"
+
+local test_more = ...
+
+skynet.start(function ()
+	local db = rediscluster.new({
+		{host="127.0.0.1",port=7000},
+		{host="127.0.0.1",port=7001},},
+		{read_slave=true,auth=nil,db=0,}
+	)
+	db:del("list")
+	db:del("map")
+	db:rpush("list",1,2,3)
+	local list = db:lrange("list",0,-1)
+	for i,v in ipairs(list) do
+		print(v)
+	end
+	db:hmset("map","key1",1,"key2",2)
+	local map = db:hgetall("map")
+	for i=1,#map,2 do
+		local key = map[i]
+		local val = map[i+1]
+		print(key,val)
+	end
+	-- test MOVED
+	db:flush_slots_cache()
+	print(db:set("A",1))
+	print(db:get("A"))
+	-- reconnect
+	local cnt = 0
+	for name,conn in pairs(db.connections) do
+		print(name,conn)
+		cnt = cnt + 1
+	end
+	print("cnt:",cnt)
+	db:close_all_connection()
+	print(db:set("A",1))
+	print(db:del("A"))
+
+	local slot = db:keyslot("{foo}")
+	local conn = db:get_connection_by_slot(slot)
+	-- You must ensure keys at one slot: use same key or hash tags
+	local ret = conn:pipeline({
+		{"hincrby", "{foo}hello", 1, 1},
+		{"del", "{foo}hello"},
+		{"hmset", "{foo}hello", 1, 1, 2, 2, 3, 3},
+		{"hgetall", "{foo}hello"},
+	},{})
+	print(ret[1].ok,ret[1].out)
+	print(ret[2].ok,ret[2].out)
+	print(ret[3].ok,ret[3].out)
+	print(ret[4].ok)
+	if ret[4].ok then
+		for i,v in ipairs(ret[4].out) do
+			print(v)
+		end
+	else
+		print(ret[4].out)
+	end
+	-- dbsize/info/keys
+	local conn = db:get_random_connection()
+	print("dbsize:",conn:dbsize())
+	print("info:",conn:info())
+	local keys = conn:keys("list*")
+	for i,key in ipairs(keys) do
+		print(key)
+	end
+	print("cluster nodes")
+	local nodes = db:cluster("nodes")
+	print(nodes)
+	print("cluster slots")
+	local slots = db:cluster("slots")
+	for i,slot_map in ipairs(slots) do
+		local start_slot = slot_map[1]
+		local end_slot = slot_map[2]
+		local master_node = slot_map[3]
+		print(start_slot,end_slot)
+		for i,v in ipairs(master_node) do
+			print(v)
+		end
+		for i=4,#slot_map do
+			local slave_node = slot_map[i]
+			for i,v in ipairs(slave_node) do
+				print(v)
+			end
+		end
+	end
+
+	if not test_more then
+		skynet.exit()
+		return
+	end
+	local last = false
+	while not last do
+		last = db:get("__last__")
+		if last == nil then
+			last = 0
+		end
+		last = tonumber(last)
+	end
+	for val=last,1000000000 do
+		local ok,errmsg = pcall(function ()
+			local key = string.format("foo%s",val)
+			db:set(key,val)
+			print(key,db:get(key))
+			db:set("__last__",val)
+		end)
+		if not ok then
+			print("error:",errmsg)
+		end
+	end
+	skynet.exit()
+end)