Skip to content

Commit 878ee46

Browse files
0x501D0x501D
committed
Add caching of sharding function
The ddl.bucket_id() function needs to know a sharding function. It is costly to obtain the function declaration / definition stored in the _ddl_sharding_func space. This cache adds sharding function cache divided into two parts: raw and processed. Raw part is used for get_schema() method. Raw cache stored as is. Processed part is used for bucket_id(). Processed sharding_func cache entry may be: * table with parsed dot notation (like {'foo', 'bar'}) * function ready to call, this offloads using of loadstring() * string with an error Cache will be rebuilded if: * _ddl_sharding_func space changed: cache sets _ddl_sharding_func:on_replace trigger * schema changed: cache checks box.internal.schema_version changes This patch does not serve hot reload techniques. This entails an on_replace trigger duplication if hot reload occurs. Hot reload support will be done in separate task: #87 Closes #82
1 parent 4f0fbd1 commit 878ee46

File tree

4 files changed

+322
-11
lines changed

4 files changed

+322
-11
lines changed

‎ddl/cache.lua

Lines changed: 136 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,136 @@
1+
local fiber = require('fiber')
2+
3+
local cache = nil
4+
5+
local CACHE_LOCK_TIMEOUT = 3
6+
local SPACE_NAME_IDX = 1
7+
local SHARD_FUNC_NAME_IDX = 2
8+
local SHARD_FUNC_BODY_IDX = 3
9+
10+
-- Function decorator that is used to prevent cache_build() from being
11+
-- called concurrently by different fibers.
12+
local function locked(f)
13+
return function(...)
14+
local ok = cache.lock:put(true, CACHE_LOCK_TIMEOUT)
15+
16+
if not ok then
17+
error("cache lock timeout is exceeded")
18+
end
19+
20+
local status, err = pcall(f, ...)
21+
cache.lock:get()
22+
23+
if not status or err ~= nil then
24+
return err
25+
end
26+
end
27+
end
28+
29+
-- Build cache.
30+
--
31+
-- Cache structure format:
32+
-- cache = {
33+
-- spaces = {
34+
-- 'space_name' = {
35+
-- raw = {}, -- raw sharding metadata, used for ddl.get()
36+
-- processed = {}, -- table with parsed dot notation (like {'foo', 'bar'})
37+
-- -- or a function ready to call (or a string with an error)
38+
-- }
39+
-- },
40+
-- lock, -- locking based on fiber.channel()
41+
-- schema_version, -- current schema version
42+
-- }
43+
--
44+
-- function returns nothing
45+
local cache_build = locked(function()
46+
-- clear cache
47+
cache.spaces = {}
48+
49+
if box.space._ddl_sharding_func == nil then
50+
return
51+
end
52+
53+
for _, tuple in box.space._ddl_sharding_func:pairs() do
54+
local space_name = tuple[SPACE_NAME_IDX]
55+
local func_name = tuple[SHARD_FUNC_NAME_IDX]
56+
local func_body = tuple[SHARD_FUNC_BODY_IDX]
57+
58+
cache.spaces[space_name] = {
59+
raw = tuple
60+
}
61+
62+
if func_body ~= nil then
63+
local sharding_func, err = loadstring('return ' .. func_body)
64+
if sharding_func == nil then
65+
cache.spaces[space_name].processed =
66+
string.format("Body is incorrect in sharding_func for space (%s): %s",
67+
space_name, err)
68+
else
69+
cache.spaces[space_name].processed =
70+
sharding_func()
71+
end
72+
elseif func_name ~= nil then
73+
local chunks = string.split(func_name, '.')
74+
cache.spaces[space_name].processed = chunks
75+
end
76+
end
77+
78+
end)
79+
80+
-- Rebuild cache if _ddl_sharding_func space changed.
81+
local function cache_set_trigger()
82+
if box.space._ddl_sharding_func == nil then
83+
return
84+
end
85+
86+
local trigger_found = false
87+
88+
for _, func in pairs(box.space._ddl_sharding_func:on_replace()) do
89+
if func == cache_build then
90+
trigger_found = true
91+
break
92+
end
93+
end
94+
95+
if not trigger_found then
96+
box.space._ddl_sharding_func:on_replace(cache_build)
97+
end
98+
end
99+
100+
-- Get data from cache.
101+
-- Returns all cached data for "space_name" or nil.
102+
local function cache_get(space_name)
103+
if space_name == nil then
104+
return nil
105+
end
106+
107+
-- using tarantool internal API.
108+
-- this is not reliable, but it is the only way to track
109+
-- schema_version changes. Fix it if a public method appears:
110+
-- https://github.com/tarantool/tarantool/issues/6544
111+
local schema_version = box.internal.schema_version()
112+
113+
if not cache then
114+
cache = {
115+
lock = fiber.channel(1)
116+
}
117+
cache_build()
118+
cache_set_trigger()
119+
cache.schema_version = schema_version
120+
end
121+
122+
-- rebuild cache if database schema changed
123+
if schema_version ~= cache.schema_version then
124+
cache_build()
125+
cache_set_trigger()
126+
cache.schema_version = schema_version
127+
end
128+
129+
return cache.spaces[space_name]
130+
end
131+
132+
return {
133+
internal = {
134+
get = cache_get,
135+
}
136+
}

‎ddl/get.lua

Lines changed: 24 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -1,4 +1,5 @@
11
local utils = require('ddl.utils')
2+
local cache = require('ddl.cache')
23
local ddl_check = require('ddl.check')
34

45
local function _get_index_field_path(space, index_part)
@@ -66,11 +67,24 @@ local function get_metadata(space_name, metadata_name)
6667
end
6768

6869
local function get_sharding_func(space_name)
69-
local record = get_metadata(space_name, "sharding_func")
70+
local record = cache.internal.get(space_name)
71+
7072
if not record then
7173
return nil
7274
end
7375

76+
return record.processed
77+
end
78+
79+
local function get_sharding_func_raw(space_name)
80+
local record = cache.internal.get(space_name)
81+
82+
if not record or not record.raw then
83+
return nil
84+
end
85+
86+
record = record.raw
87+
7488
if record.sharding_func_body ~= nil then
7589
return {body = record.sharding_func_body}
7690
end
@@ -97,7 +111,7 @@ local function get_space_schema(space_name)
97111
space_ddl.engine = box_space.engine
98112
space_ddl.format = box_space:format()
99113
space_ddl.sharding_key = get_sharding_key(space_name)
100-
space_ddl.sharding_func = get_sharding_func(space_name)
114+
space_ddl.sharding_func = get_sharding_func_raw(space_name)
101115
for _, field in ipairs(space_ddl.format) do
102116
if field.is_nullable == nil then
103117
field.is_nullable = false
@@ -115,21 +129,21 @@ local function get_space_schema(space_name)
115129
end
116130

117131
local function prepare_sharding_func_for_call(space_name, sharding_func_def)
118-
if type(sharding_func_def) == 'string' then
132+
if type(sharding_func_def) == 'table' then
119133
local sharding_func = utils.get_G_function(sharding_func_def)
120134
if sharding_func ~= nil and
121135
ddl_check.internal.is_callable(sharding_func) == true then
122136
return sharding_func
123137
end
124138
end
125139

126-
if type(sharding_func_def) == 'table' then
127-
local sharding_func, err = loadstring('return ' .. sharding_func_def.body)
128-
if sharding_func == nil then
129-
return nil, string.format(
130-
"Body is incorrect in sharding_func for space (%s): %s", space_name, err)
131-
end
132-
return sharding_func()
140+
if type(sharding_func_def) == 'function' then
141+
return sharding_func_def
142+
end
143+
144+
-- error from cache
145+
if type(sharding_func_def) == 'string' then
146+
return nil, sharding_func_def
133147
end
134148

135149
return nil, string.format(

‎ddl/utils.lua

Lines changed: 11 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -189,9 +189,19 @@ end
189189
-- split sharding func name in dot notation by dot
190190
-- foo.bar.baz -> chunks: foo bar baz
191191
-- foo -> chunks: foo
192+
--
193+
-- func_name parameter may be a string in dot notation or table
194+
-- if func_name type is of type table it is assumed that it is already split
192195
local function get_G_function(func_name)
193-
local chunks = string.split(func_name, '.')
194196
local sharding_func = _G
197+
local chunks
198+
199+
if type(func_name) == 'string' then
200+
chunks = string.split(func_name, '.')
201+
else
202+
chunks = func_name
203+
end
204+
195205
-- check is the each chunk an identifier
196206
for _, chunk in pairs(chunks) do
197207
if not check_name_isident(chunk) or sharding_func == nil then

‎test/cache_test.lua

Lines changed: 151 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,151 @@
1+
#!/usr/bin/env tarantool
2+
3+
local t = require('luatest')
4+
local db = require('test.db')
5+
local ddl = require('ddl')
6+
local cache = require('ddl.cache')
7+
local helper = require('test.helper')
8+
9+
local SPACE_NAME_IDX = 1
10+
local SHARD_FUNC_NAME_IDX = 2
11+
local SHARD_FUNC_BODY_IDX = 3
12+
13+
local primary_index = {
14+
type = 'HASH',
15+
unique = true,
16+
parts = {
17+
{path = 'string_nonnull', is_nullable = false, type = 'string'},
18+
{path = 'unsigned_nonnull', is_nullable = false, type = 'unsigned'},
19+
},
20+
name = 'primary'
21+
}
22+
23+
local bucket_id_idx = {
24+
type = 'TREE',
25+
unique = false,
26+
parts = {{path = 'bucket_id', type = 'unsigned', is_nullable = false}},
27+
name = 'bucket_id'
28+
}
29+
30+
local func_body_first = 'function() return 42 end'
31+
local func_body_second = 'function() return 24 end'
32+
33+
local function space_init(g)
34+
db.drop_all()
35+
36+
g.space = {
37+
engine = 'memtx',
38+
is_local = true,
39+
temporary = false,
40+
format = table.deepcopy(helper.test_space_format())
41+
}
42+
table.insert(g.space.format, 1, {
43+
name = 'bucket_id', type = 'unsigned', is_nullable = false
44+
})
45+
46+
g.space.indexes = {
47+
table.deepcopy(primary_index),
48+
table.deepcopy(bucket_id_idx)
49+
}
50+
g.space.sharding_key = {'unsigned_nonnull', 'integer_nonnull'}
51+
g.schema = {
52+
spaces = {
53+
space = g.space,
54+
}
55+
}
56+
end
57+
58+
local g = t.group()
59+
g.before_all(db.init)
60+
g.before_each(space_init)
61+
62+
function g.test_cache_processed_func_body()
63+
g.schema.spaces.space.sharding_func = {
64+
body = func_body_first
65+
}
66+
local ok, err = ddl.set_schema(g.schema)
67+
t.assert_equals(err, nil)
68+
t.assert_equals(ok, true)
69+
70+
local res = cache.internal.get('space')
71+
t.assert(res)
72+
t.assert(res.processed)
73+
t.assert(type(res.processed) == 'function')
74+
t.assert_equals(res.processed(), 42)
75+
end
76+
77+
function g.test_cache_processed_func_name()
78+
local sharding_func_name = 'sharding_func'
79+
rawset(_G, sharding_func_name, function(key) return key end)
80+
g.schema.spaces.space.sharding_func = sharding_func_name
81+
82+
local ok, err = ddl.set_schema(g.schema)
83+
t.assert_equals(err, nil)
84+
t.assert_equals(ok, true)
85+
86+
local res = cache.internal.get('space')
87+
t.assert(res)
88+
t.assert(res.processed)
89+
t.assert(type(res.processed) == 'table')
90+
t.assert_equals(res.processed[1], 'sharding_func')
91+
92+
rawset(_G, sharding_func_name, nil)
93+
end
94+
95+
function g.test_cache_schema_changed()
96+
g.schema.spaces.space.sharding_func = {
97+
body = func_body_first
98+
}
99+
local ok, err = ddl.set_schema(g.schema)
100+
t.assert_equals(err, nil)
101+
t.assert_equals(ok, true)
102+
103+
local res = cache.internal.get('space')
104+
t.assert(res)
105+
t.assert(res.raw)
106+
t.assert_equals(res.raw[SPACE_NAME_IDX], 'space')
107+
t.assert_equals(res.raw[SHARD_FUNC_NAME_IDX], nil)
108+
t.assert_equals(res.raw[SHARD_FUNC_BODY_IDX], func_body_first)
109+
110+
space_init(g)
111+
112+
g.schema.spaces.space.sharding_func = {
113+
body = func_body_second
114+
}
115+
local ok, err = ddl.set_schema(g.schema)
116+
t.assert_equals(err, nil)
117+
t.assert_equals(ok, true)
118+
119+
local res = cache.internal.get('space')
120+
t.assert(res)
121+
t.assert(res.raw)
122+
t.assert_equals(res.raw[SPACE_NAME_IDX], 'space')
123+
t.assert_equals(res.raw[SHARD_FUNC_NAME_IDX], nil)
124+
t.assert_equals(res.raw[SHARD_FUNC_BODY_IDX], func_body_second)
125+
end
126+
127+
function g.test_cache_space_updated()
128+
g.schema.spaces.space.sharding_func = {
129+
body = func_body_first
130+
}
131+
local ok, err = ddl.set_schema(g.schema)
132+
t.assert_equals(err, nil)
133+
t.assert_equals(ok, true)
134+
135+
local res = cache.internal.get('space')
136+
t.assert(res)
137+
t.assert(res.raw)
138+
t.assert_equals(res.raw[SPACE_NAME_IDX], 'space')
139+
t.assert_equals(res.raw[SHARD_FUNC_NAME_IDX], nil)
140+
t.assert_equals(res.raw[SHARD_FUNC_BODY_IDX], func_body_first)
141+
142+
box.space._ddl_sharding_func
143+
:update({'space'}, {{'=', SHARD_FUNC_BODY_IDX, func_body_second}})
144+
145+
local res = cache.internal.get('space')
146+
t.assert(res)
147+
t.assert(res.raw)
148+
t.assert_equals(res.raw[SPACE_NAME_IDX], 'space')
149+
t.assert_equals(res.raw[SHARD_FUNC_NAME_IDX], nil)
150+
t.assert_equals(res.raw[SHARD_FUNC_BODY_IDX], func_body_second)
151+
end

0 commit comments

Comments
 (0)