From 2798c778c71b0e0d43a6e543df852f297da34a34 Mon Sep 17 00:00:00 2001 From: sundream Date: Sat, 3 Jun 2017 12:26:19 +0800 Subject: [PATCH] redis cluster client --- lualib/skynet/db/redis/cluster.lua | 383 +++++++++++++++++++++++++++++ lualib/skynet/db/redis/crc16.lua | 62 +++++ test/testrediscluster.lua | 114 +++++++++ 3 files changed, 559 insertions(+) create mode 100644 lualib/skynet/db/redis/cluster.lua create mode 100644 lualib/skynet/db/redis/crc16.lua create mode 100644 test/testrediscluster.lua diff --git a/lualib/skynet/db/redis/cluster.lua b/lualib/skynet/db/redis/cluster.lua new file mode 100644 index 00000000..99e37577 --- /dev/null +++ b/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 diff --git a/lualib/skynet/db/redis/crc16.lua b/lualib/skynet/db/redis/crc16.lua new file mode 100644 index 00000000..1acab9f5 --- /dev/null +++ b/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 diff --git a/test/testrediscluster.lua b/test/testrediscluster.lua new file mode 100644 index 00000000..5c6efdf0 --- /dev/null +++ b/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)