Skip to content

Commit 4729923

Browse files
shuxiao9058shuxiao9058
committed
cache slots_info in memory
1 parent 3cc5be5 commit 4729923

File tree

2 files changed

+146
-50
lines changed

2 files changed

+146
-50
lines changed

‎README.md

Lines changed: 29 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -61,10 +61,11 @@ While building the client, thanks for https://github.com/cuiweixie/lua-resty-red
6161

6262
```lua
6363
local config = {
64-
dict_name = "test_locks", --shared dictionary name for locks, if default value is not used
65-
refresh_lock_key = "refresh_lock", --shared dictionary name prefix for lock of each worker, if default value is not used
66-
name = "testCluster", --rediscluster name
67-
serv_list = { --redis cluster node list(host and port),
64+
dict_name = "test_locks", --shared dictionary name for locks, if default value is not used
65+
refresh_lock_key = "refresh_lock", --shared dictionary name prefix for lock of each worker, if default value is not used
66+
slots_info_dict_name = "test_slots_info", --shared dictionary name for slots_info
67+
name = "testCluster", --rediscluster name
68+
serv_list = { --redis cluster node list(host and port),
6869
{ ip = "127.0.0.1", port = 7001 },
6970
{ ip = "127.0.0.1", port = 7002 },
7071
{ ip = "127.0.0.1", port = 7003 },
@@ -93,10 +94,11 @@ end
9394

9495
```lua
9596
local config = {
96-
dict_name = "test_locks", --shared dictionary name for locks, if default value is not used
97-
refresh_lock_key = "refresh_lock", --shared dictionary name prefix for lock of each worker, if default value is not used
98-
name = "testCluster", --rediscluster name
99-
serv_list = { --redis cluster node list(host and port),
97+
dict_name = "test_locks", --shared dictionary name for locks, if default value is not used
98+
refresh_lock_key = "refresh_lock", --shared dictionary name prefix for lock of each worker, if default value is not used
99+
slots_info_dict_name = "test_slots_info", --shared dictionary name for slots_info
100+
name = "testCluster", --rediscluster name
101+
serv_list = { --redis cluster node list(host and port),
100102
{ ip = "127.0.0.1", port = 7001 },
101103
{ ip = "127.0.0.1", port = 7002 },
102104
{ ip = "127.0.0.1", port = 7003 },
@@ -131,8 +133,9 @@ end
131133
local cjson = require "cjson"
132134

133135
local config = {
134-
dict_name = "test_locks", --shared dictionary name for locks, if default value is not used
135-
refresh_lock_key = "refresh_lock", --shared dictionary name prefix for lock of each worker, if default value is not used
136+
dict_name = "test_locks", --shared dictionary name for locks, if default value is not used
137+
refresh_lock_key = "refresh_lock", --shared dictionary name prefix for lock of each worker, if default value is not used
138+
slots_info_dict_name = "test_slots_info", --shared dictionary name for slots_info
136139
name = "testCluster",
137140
serv_list = {
138141
{ ip = "127.0.0.1", port = 7001 },
@@ -181,8 +184,9 @@ end
181184
local cjson = require "cjson"
182185

183186
local config = {
184-
dict_name = "test_locks", --shared dictionary name for locks, if default value is not used
185-
refresh_lock_key = "refresh_lock", --shared dictionary name prefix for lock of each worker, if default value is not used
187+
dict_name = "test_locks", --shared dictionary name for locks, if default value is not used
188+
refresh_lock_key = "refresh_lock", --shared dictionary name prefix for lock of each worker, if default value is not used
189+
slots_info_dict_name = "test_slots_info", --shared dictionary name for slots_info
186190
name = "testCluster",
187191
enable_slave_read = true,
188192
serv_list = {
@@ -218,8 +222,9 @@ end
218222
local cjson = require "cjson"
219223

220224
local config = {
221-
dict_name = "test_locks", --shared dictionary name for locks, if default value is not used
222-
refresh_lock_key = "refresh_lock", --shared dictionary name prefix for lock of each worker, if default value is not used
225+
dict_name = "test_locks", --shared dictionary name for locks, if default value is not used
226+
refresh_lock_key = "refresh_lock", --shared dictionary name prefix for lock of each worker, if default value is not used
227+
slots_info_dict_name = "test_slots_info", --shared dictionary name for slots_info
223228
name = "testCluster",
224229
enable_slave_read = true,
225230
serv_list = {
@@ -261,10 +266,11 @@ end
261266

262267
```lua
263268
local config = {
264-
dict_name = "test_locks", --shared dictionary name for locks, if default value is not used
265-
refresh_lock_key = "refresh_lock", --shared dictionary name prefix for lock of each worker, if default value is not used
266-
name = "testCluster", --rediscluster name
267-
serv_list = { --redis cluster node list(host and port),
269+
dict_name = "test_locks", --shared dictionary name for locks, if default value is not used
270+
refresh_lock_key = "refresh_lock", --shared dictionary name prefix for lock of each worker, if default value is not used
271+
slots_info_dict_name = "test_slots_info", --shared dictionary name for slots_info
272+
name = "testCluster", --rediscluster name
273+
serv_list = { --redis cluster node list(host and port),
268274
{ ip = "127.0.0.1", port = 7001 },
269275
{ ip = "127.0.0.1", port = 7002 },
270276
{ ip = "127.0.0.1", port = 7003 },
@@ -298,10 +304,11 @@ end
298304

299305
```lua
300306
local config = {
301-
dict_name = "test_locks", --shared dictionary name for locks, if default value is not used
302-
refresh_lock_key = "refresh_lock", --shared dictionary name prefix for lock of each worker, if default value is not used
303-
name = "testCluster", --rediscluster name
304-
serv_list = { --redis cluster node list(host and port),
307+
dict_name = "test_locks", --shared dictionary name for locks, if default value is not used
308+
refresh_lock_key = "refresh_lock", --shared dictionary name prefix for lock of each worker, if default value is not used
309+
slots_info_dict_name = "test_slots_info", --shared dictionary name for slots_info
310+
name = "testCluster", --rediscluster name
311+
serv_list = { --redis cluster node list(host and port),
305312
{ ip = "127.0.0.1", port = 7001 },
306313
{ ip = "127.0.0.1", port = 7002 },
307314
{ ip = "127.0.0.1", port = 7003 },

‎lib/resty/rediscluster.lua

Lines changed: 117 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -19,7 +19,12 @@ local table_insert = table.insert
1919
local string_find = string.find
2020
local redis_crc = xmodem.redis_crc
2121

22+
local cjson = require('cjson.safe')
23+
local cjson_encode = cjson.encode
24+
local cjson_deocde = cjson.decode
25+
2226
local DEFAULT_SHARED_DICT_NAME = "redis_cluster_slot_locks"
27+
local DEFAULT_SLOTS_INFO_DICT_NAME = "redis_cluster_slots_info"
2328
local DEFAULT_REFRESH_DICT_NAME = "refresh_lock"
2429
local DEFAULT_MAX_REDIRECTION = 5
2530
local DEFAULT_MAX_CONNECTION_ATTEMPTS = 3
@@ -96,6 +101,42 @@ local function split(s, delimiter)
96101
return result;
97102
end
98103

104+
local function generate_full_slots_cache_info(slots_info)
105+
if not slots_info then
106+
return nil, nil, 'slots_info is nil'
107+
end
108+
109+
local slots = {}
110+
-- while slots are updated, create a list of servers present in cluster
111+
-- this can differ from self.config.serv_list if a cluster is resized (added/removed nodes)
112+
local servers = { serv_list = {} }
113+
for n = 1, #slots_info do
114+
local sub_info = slots_info[n]
115+
-- slot info item 1 and 2 are the subrange start end slots
116+
local start_slot, end_slot = sub_info[1], sub_info[2]
117+
local list = { serv_list = {} }
118+
--from 3, here lists the host/port/nodeid of in charge nodes
119+
for j = 3, #sub_info do
120+
table.insert(list.serv_list,{
121+
ip = sub_info[j][1],
122+
port = sub_info[j][2],
123+
slave = (j > 3) -- first node in the list is the master
124+
})
125+
end
126+
127+
for slot = start_slot, end_slot do
128+
slots[slot] = list
129+
end
130+
131+
-- append to the list of all servers
132+
for _, serv in ipairs(list.serv_list) do
133+
table.insert(servers.serv_list,serv)
134+
end
135+
end
136+
137+
return slots, servers
138+
end
139+
99140
local function try_hosts_slots(self, serv_list)
100141
local start_time = ngx.now()
101142
local errors = {}
@@ -140,36 +181,16 @@ local function try_hosts_slots(self, serv_list)
140181
local slots_info
141182
slots_info, err = redis_client:cluster("slots")
142183
if slots_info then
143-
local slots = {}
144-
-- while slots are updated, create a list of servers present in cluster
145-
-- this can differ from self.config.serv_list if a cluster is resized (added/removed nodes)
146-
local servers = { serv_list = {} }
147-
for n = 1, #slots_info do
148-
local sub_info = slots_info[n]
149-
-- slot info item 1 and 2 are the subrange start end slots
150-
local start_slot, end_slot = sub_info[1], sub_info[2]
151-
local list = { serv_list = {} }
152-
--from 3, here lists the host/port/nodeid of in charge nodes
153-
for j = 3, #sub_info do
154-
table.insert(list.serv_list,{
155-
ip = sub_info[j][1],
156-
port = sub_info[j][2],
157-
slave = (j > 3) -- first node in the list is the master
158-
})
159-
end
160-
161-
for slot = start_slot, end_slot do
162-
slots[slot] = list
163-
end
164-
165-
-- append to the list of all servers
166-
for _, serv in ipairs(list.serv_list) do
167-
table.insert(servers.serv_list,serv)
168-
end
169-
end
184+
local slots, servers = generate_full_slots_cache_info(slots_info)
170185
--ngx.log(ngx.NOTICE, "finished initializing slotcache...")
171186
slot_cache[self.config.name] = slots
172187
slot_cache[self.config.name .. "serv_list"] = servers
188+
189+
-- cache slots_info to memory
190+
_, err = self:try_cache_slots_info_to_memory(slots_info)
191+
if err then
192+
ngx.log(ngx.ERR, 'failed to cache slots to memory: ', err)
193+
end
173194
else
174195
table_insert(errors, err)
175196
end
@@ -223,6 +244,56 @@ function _M.fetch_slots(self)
223244
end
224245
end
225246

247+
function _M.try_load_slots_from_memory_cache(self)
248+
local dict_name = self.config.slots_info_dict_name or DEFAULT_SLOTS_INFO_DICT_NAME
249+
local slots_cache_dict = ngx.shared[dict_name]
250+
if slots_cache_dict == nil then
251+
return false, dict_name ..' is nil'
252+
end
253+
254+
255+
local slots_info_str = slots_cache_dict:get(self.config.name)
256+
if not slots_info_str or slots_info_str == '' then
257+
ngx.log(ngx.ERR, 'slots_info_str: ', slots_info_str)
258+
return false, 'slots_info_str is nil or empty'
259+
end
260+
261+
local slots_info = cjson_decode(slots_info_str)
262+
if not slots_info then
263+
return false , 'slots_info is nil'
264+
end
265+
266+
local slots, servers = generate_full_slots_cache_info(slots_info)
267+
if not slots or not servers then
268+
return false, 'slots or servers is nil'
269+
end
270+
271+
--ngx.log(ngx.NOTICE, "finished initializing slotcache...")
272+
slot_cache[self.config.name] = slots
273+
slot_cache[self.config.name .. "serv_list"] = servers
274+
275+
return true
276+
end
277+
278+
function _M.try_cache_slots_info_to_memory(self, slots_info)
279+
local dict_name = self.config.slots_info_dict_name or DEFAULT_SLOTS_INFO_DICT_NAME
280+
local slots_cache_dict = ngx.shared[dict_name]
281+
if slots_cache_dict == nil then
282+
return false, dict_name ..' is nil'
283+
end
284+
285+
if not slots_info then
286+
return false, 'slots_info is nil'
287+
end
288+
289+
local slots_info_str = cjson_encode(slots_info)
290+
local success, err = slots_cache_dict:set(self.config.name, slots_info_str)
291+
if not success then
292+
ngx.log(ngx.ERR, 'error set slots_info: ', err, ', slots_info_str: ', slots_info_str)
293+
return false, err
294+
end
295+
return true
296+
end
226297

227298
function _M.refresh_slots(self)
228299
local worker_id = ngx.worker.id()
@@ -275,6 +346,19 @@ function _M.init_slots(self)
275346
return true
276347
end
277348

349+
-- fetch slots from memory cache
350+
ok, err = self:try_load_slots_from_memory_cache()
351+
if err then
352+
ngx.log(ngx.ERR, 'failed to fetch slots from memory cache: ', err)
353+
end
354+
if ok then
355+
ok, err = lock:unlock()
356+
if not ok then
357+
ngx.log(ngx.ERR, "failed to unlock in initialization slot cache:", err)
358+
end
359+
return true
360+
end
361+
278362
local _, errs = self:fetch_slots()
279363
if errs then
280364
ok, err = lock:unlock()
@@ -487,8 +571,13 @@ local function handle_command_with_retry(self, target_ip, target_port, asking, c
487571
return res, err
488572
end
489573
else
574+
-- `too many waiting connect operations` means queued connect operations is out of backlog
575+
-- `timeout` means timeout while wait for connection release
576+
-- if connect timeout caused by server's issue, the connerr is `connection timed out`
577+
if connerr ~= 'too many waiting connect operations' and connerr ~= 'timeout' then
578+
self:refresh_slots()
579+
end
490580
--There might be node fail, we should also refresh slot cache
491-
self:refresh_slots()
492581
if k == config.max_redirection or k == DEFAULT_MAX_REDIRECTION then
493582
-- only return after allowing for `k` attempts
494583
return nil, connerr

0 commit comments

Comments
 (0)