Skip to content
GitLab
Menu
Projects
Groups
Snippets
Loading...
Help
Help
Support
Community forum
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in / Register
Toggle navigation
Menu
Open sidebar
ruanhaishen
redis
Commits
d375595d
Unverified
Commit
d375595d
authored
Apr 27, 2022
by
Oran Agra
Committed by
GitHub
Apr 27, 2022
Browse files
Merge pull request #10652 from oranagra/redis-7.0.0
Redis 7.0.0
parents
fb4e0d40
c1f30206
Changes
124
Show whitespace changes
Inline
Side-by-side
src/eval.c
View file @
d375595d
...
...
@@ -218,31 +218,20 @@ void scriptingInit(int setup) {
lua_setglobal
(
lua
,
"redis"
);
/* Add a helper function that we use to sort the multi bulk output of non
* deterministic commands, when containing 'false' elements. */
{
char
*
compare_func
=
"function __redis__compare_helper(a,b)
\n
"
" if a == false then a = '' end
\n
"
" if b == false then b = '' end
\n
"
" return a<b
\n
"
"end
\n
"
;
luaL_loadbuffer
(
lua
,
compare_func
,
strlen
(
compare_func
),
"@cmp_func_def"
);
lua_pcall
(
lua
,
0
,
0
,
0
);
}
/* Add a helper function we use for pcall error reporting.
* Note that when the error is in the C function we want to report the
* information about the caller, that's what makes sense from the point
* of view of the user debugging a script. */
{
char
*
errh_func
=
"local dbg = debug
\n
"
"debug = nil
\n
"
"function __redis__err__handler(err)
\n
"
" local i = dbg.getinfo(2,'nSl')
\n
"
" if i and i.what == 'C' then
\n
"
" i = dbg.getinfo(3,'nSl')
\n
"
" end
\n
"
" if type(err) ~= 'table' then
\n
"
" err = {err='ERR' .. tostring(err)}"
" err = {err='ERR
' .. tostring(err)}"
" end"
" if i then
\n
"
" err['source'] = i.source
\n
"
...
...
@@ -266,10 +255,12 @@ void scriptingInit(int setup) {
lctx
.
lua_client
->
flags
|=
CLIENT_DENY_BLOCKING
;
}
/* Lua beginners often don't use "local", this is likely to introduce
* subtle bugs in their code. To prevent problems we protect accesses
* to global variables. */
luaEnableGlobalsProtection
(
lua
,
1
);
/* Lock the global table from any changes */
lua_pushvalue
(
lua
,
LUA_GLOBALSINDEX
);
luaSetErrorMetatable
(
lua
);
/* Recursively lock all tables that can be reached from the global table */
luaSetTableProtectionRecursively
(
lua
);
lua_pop
(
lua
,
1
);
lctx
.
lua
=
lua
;
}
...
...
@@ -378,35 +369,20 @@ sds luaCreateFunction(client *c, robj *body) {
sdsfreesplitres
(
parts
,
numparts
);
}
/* Build the lua function to be loaded */
sds
funcdef
=
sdsempty
();
funcdef
=
sdscat
(
funcdef
,
"function "
);
funcdef
=
sdscatlen
(
funcdef
,
funcname
,
42
);
funcdef
=
sdscatlen
(
funcdef
,
"() "
,
3
);
/* Note that in case of a shebang line we skip it but keep the line feed to conserve the user's line numbers */
funcdef
=
sdscatlen
(
funcdef
,(
char
*
)
body
->
ptr
+
shebang_len
,
sdslen
(
body
->
ptr
)
-
shebang_len
);
funcdef
=
sdscatlen
(
funcdef
,
"
\n
end"
,
4
);
if
(
luaL_loadbuffer
(
lctx
.
lua
,
funcdef
,
sdslen
(
funcdef
),
"@user_script"
))
{
if
(
luaL_loadbuffer
(
lctx
.
lua
,(
char
*
)
body
->
ptr
+
shebang_len
,
sdslen
(
body
->
ptr
)
-
shebang_len
,
"@user_script"
))
{
if
(
c
!=
NULL
)
{
addReplyErrorFormat
(
c
,
"Error compiling script (new function): %s"
,
lua_tostring
(
lctx
.
lua
,
-
1
));
}
lua_pop
(
lctx
.
lua
,
1
);
sdsfree
(
funcdef
);
return
NULL
;
}
sdsfree
(
funcdef
);
if
(
lua_pcall
(
lctx
.
lua
,
0
,
0
,
0
))
{
if
(
c
!=
NULL
)
{
addReplyErrorFormat
(
c
,
"Error running script (new function): %s"
,
lua_tostring
(
lctx
.
lua
,
-
1
));
}
lua_pop
(
lctx
.
lua
,
1
);
return
NULL
;
}
serverAssert
(
lua_isfunction
(
lctx
.
lua
,
-
1
));
lua_setfield
(
lctx
.
lua
,
LUA_REGISTRYINDEX
,
funcname
);
/* We also save a SHA1 -> Original script map in a dictionary
* so that we can replicate / write in the AOF all the
...
...
@@ -479,7 +455,7 @@ void evalGenericCommand(client *c, int evalsha) {
lua_getglobal
(
lua
,
"__redis__err__handler"
);
/* Try to lookup the Lua function */
lua_get
global
(
lua
,
funcname
);
lua_get
field
(
lua
,
LUA_REGISTRYINDEX
,
funcname
);
if
(
lua_isnil
(
lua
,
-
1
))
{
lua_pop
(
lua
,
1
);
/* remove the nil from the stack */
/* Function not defined... let's define it if we have the
...
...
@@ -497,7 +473,7 @@ void evalGenericCommand(client *c, int evalsha) {
return
;
}
/* Now the following is guaranteed to return non nil */
lua_get
global
(
lua
,
funcname
);
lua_get
field
(
lua
,
LUA_REGISTRYINDEX
,
funcname
);
serverAssert
(
!
lua_isnil
(
lua
,
-
1
));
}
...
...
src/evict.c
View file @
d375595d
...
...
@@ -492,10 +492,6 @@ static int isSafeToPerformEvictions(void) {
* expires and evictions of keys not being performed. */
if
(
checkClientPauseTimeoutAndReturnIfPaused
())
return
0
;
/* We cannot evict if we already have stuff to propagate (for example,
* CONFIG SET maxmemory inside a MULTI/EXEC) */
if
(
server
.
also_propagate
.
numops
!=
0
)
return
0
;
return
1
;
}
...
...
src/function_lua.c
View file @
d375595d
...
...
@@ -50,6 +50,7 @@
#define REGISTRY_ERROR_HANDLER_NAME "__ERROR_HANDLER__"
#define REGISTRY_LOAD_CTX_NAME "__LIBRARY_CTX__"
#define LIBRARY_API_NAME "__LIBRARY_API__"
#define GLOBALS_API_NAME "__GLOBALS_API__"
#define LOAD_TIMEOUT_MS 500
/* Lua engine ctx */
...
...
@@ -99,42 +100,23 @@ static void luaEngineLoadHook(lua_State *lua, lua_Debug *ar) {
* Return NULL on compilation error and set the error to the err variable
*/
static
int
luaEngineCreate
(
void
*
engine_ctx
,
functionLibInfo
*
li
,
sds
blob
,
sds
*
err
)
{
int
ret
=
C_ERR
;
luaEngineCtx
*
lua_engine_ctx
=
engine_ctx
;
lua_State
*
lua
=
lua_engine_ctx
->
lua
;
/* Each library will have its own global distinct table.
* We will create a new fresh Lua table and use
* lua_setfenv to set the table as the library globals
* (https://www.lua.org/manual/5.1/manual.html#lua_setfenv)
*
* At first, populate this new table with only the 'library' API
* to make sure only 'library' API is available at start. After the
* initial run is finished and all functions are registered, add
* all the default globals to the library global table and delete
* the library API.
*
* There are 2 ways to achieve the last part (add default
* globals to the new table):
*
* 1. Initialize the new table with all the default globals
* 2. Inheritance using metatable (https://www.lua.org/pil/14.3.html)
*
* For now we are choosing the second, we can change it in the future to
* achieve a better isolation between functions. */
lua_newtable
(
lua
);
/* Global table for the library */
lua_pushstring
(
lua
,
REDIS_API_NAME
);
lua_pushstring
(
lua
,
LIBRARY_API_NAME
);
lua_gettable
(
lua
,
LUA_REGISTRYINDEX
);
/* get library function from registry */
lua_settable
(
lua
,
-
3
);
/* push the library table to the new global table */
/* Set global protection on the new global table */
luaSetGlobalProtection
(
lua_engine_ctx
->
lua
);
/* set load library globals */
lua_getmetatable
(
lua
,
LUA_GLOBALSINDEX
);
lua_enablereadonlytable
(
lua
,
-
1
,
0
);
/* disable global protection */
lua_getfield
(
lua
,
LUA_REGISTRYINDEX
,
LIBRARY_API_NAME
);
lua_setfield
(
lua
,
-
2
,
"__index"
);
lua_enablereadonlytable
(
lua
,
LUA_GLOBALSINDEX
,
1
);
/* enable global protection */
lua_pop
(
lua
,
1
);
/* pop the metatable */
/* compile the code */
if
(
luaL_loadbuffer
(
lua
,
blob
,
sdslen
(
blob
),
"@user_function"
))
{
*
err
=
sdscatprintf
(
sdsempty
(),
"Error compiling function: %s"
,
lua_tostring
(
lua
,
-
1
));
lua_pop
(
lua
,
2
);
/* pops the error
and globals table
*/
return
C_ERR
;
lua_pop
(
lua
,
1
);
/* pops the error */
goto
done
;
}
serverAssert
(
lua_isfunction
(
lua
,
-
1
));
...
...
@@ -144,45 +126,31 @@ static int luaEngineCreate(void *engine_ctx, functionLibInfo *li, sds blob, sds
};
luaSaveOnRegistry
(
lua
,
REGISTRY_LOAD_CTX_NAME
,
&
load_ctx
);
/* set the function environment so only 'library' API can be accessed. */
lua_pushvalue
(
lua
,
-
2
);
/* push global table to the front */
lua_setfenv
(
lua
,
-
2
);
lua_sethook
(
lua
,
luaEngineLoadHook
,
LUA_MASKCOUNT
,
100000
);
/* Run the compiled code to allow it to register functions */
if
(
lua_pcall
(
lua
,
0
,
0
,
0
))
{
errorInfo
err_info
=
{
0
};
luaExtractErrorInformation
(
lua
,
&
err_info
);
*
err
=
sdscatprintf
(
sdsempty
(),
"Error registering functions: %s"
,
err_info
.
msg
);
lua_pop
(
lua
,
2
);
/* pops the error and globals table */
lua_sethook
(
lua
,
NULL
,
0
,
0
);
/* Disable hook */
luaSaveOnRegistry
(
lua
,
REGISTRY_LOAD_CTX_NAME
,
NULL
);
lua_pop
(
lua
,
1
);
/* pops the error */
luaErrorInformationDiscard
(
&
err_info
);
return
C_ERR
;
goto
done
;
}
lua_sethook
(
lua
,
NULL
,
0
,
0
);
/* Disable hook */
luaSaveOnRegistry
(
lua
,
REGISTRY_LOAD_CTX_NAME
,
NULL
);
/* stack contains the global table, lets rearrange it to contains the entire API. */
/* delete 'redis' API */
lua_pushstring
(
lua
,
REDIS_API_NAME
);
lua_pushnil
(
lua
);
lua_settable
(
lua
,
-
3
);
/* create metatable */
lua_newtable
(
lua
);
lua_pushstring
(
lua
,
"__index"
);
lua_pushvalue
(
lua
,
LUA_GLOBALSINDEX
);
/* push original globals */
lua_settable
(
lua
,
-
3
);
lua_pushstring
(
lua
,
"__newindex"
);
lua_pushvalue
(
lua
,
LUA_GLOBALSINDEX
);
/* push original globals */
lua_settable
(
lua
,
-
3
);
lua_setmetatable
(
lua
,
-
2
)
;
ret
=
C_OK
;
lua_pop
(
lua
,
1
);
/* pops the global table */
done:
/* restore original globals */
lua_getmetatable
(
lua
,
LUA_GLOBALSINDEX
);
lua_enablereadonlytable
(
lua
,
-
1
,
0
);
/* disable global protection */
lua_getfield
(
lua
,
LUA_REGISTRYINDEX
,
GLOBALS_API_NAME
);
lua_setfield
(
lua
,
-
2
,
"__index"
);
lua_enablereadonlytable
(
lua
,
LUA_GLOBALSINDEX
,
1
);
/* enable global protection */
lua_pop
(
lua
,
1
);
/* pop the metatable */
return
C_OK
;
lua_sethook
(
lua
,
NULL
,
0
,
0
);
/* Disable hook */
luaSaveOnRegistry
(
lua
,
REGISTRY_LOAD_CTX_NAME
,
NULL
);
return
ret
;
}
/*
...
...
@@ -458,8 +426,8 @@ int luaEngineInitEngine() {
luaRegisterRedisAPI
(
lua_engine_ctx
->
lua
);
/* Register the library commands table and fields and store it to registry */
lua_
pushstring
(
lua_engine_ctx
->
lua
,
LIBRARY_API_NAME
);
lua_newtable
(
lua_engine_ctx
->
lua
);
lua_
newtable
(
lua_engine_ctx
->
lua
);
/* load library globals */
lua_newtable
(
lua_engine_ctx
->
lua
);
/* load library `redis` table */
lua_pushstring
(
lua_engine_ctx
->
lua
,
"register_function"
);
lua_pushcfunction
(
lua_engine_ctx
->
lua
,
luaRegisterFunction
);
...
...
@@ -468,18 +436,24 @@ int luaEngineInitEngine() {
luaRegisterLogFunction
(
lua_engine_ctx
->
lua
);
luaRegisterVersion
(
lua_engine_ctx
->
lua
);
lua_settable
(
lua_engine_ctx
->
lua
,
LUA_REGISTRYINDEX
);
luaSetErrorMetatable
(
lua_engine_ctx
->
lua
);
lua_setfield
(
lua_engine_ctx
->
lua
,
-
2
,
REDIS_API_NAME
);
luaSetErrorMetatable
(
lua_engine_ctx
->
lua
);
luaSetTableProtectionRecursively
(
lua_engine_ctx
->
lua
);
/* protect load library globals */
lua_setfield
(
lua_engine_ctx
->
lua
,
LUA_REGISTRYINDEX
,
LIBRARY_API_NAME
);
/* Save error handler to registry */
lua_pushstring
(
lua_engine_ctx
->
lua
,
REGISTRY_ERROR_HANDLER_NAME
);
char
*
errh_func
=
"local dbg = debug
\n
"
"debug = nil
\n
"
"local error_handler = function (err)
\n
"
" local i = dbg.getinfo(2,'nSl')
\n
"
" if i and i.what == 'C' then
\n
"
" i = dbg.getinfo(3,'nSl')
\n
"
" end
\n
"
" if type(err) ~= 'table' then
\n
"
" err = {err='ERR' .. tostring(err)}"
" err = {err='ERR
' .. tostring(err)}"
" end"
" if i then
\n
"
" err['source'] = i.source
\n
"
...
...
@@ -492,17 +466,30 @@ int luaEngineInitEngine() {
lua_pcall
(
lua_engine_ctx
->
lua
,
0
,
1
,
0
);
lua_settable
(
lua_engine_ctx
->
lua
,
LUA_REGISTRYINDEX
);
/* Save global protection to registry */
luaRegisterGlobalProtectionFunction
(
lua_engine_ctx
->
lua
);
/* Set global protection on globals */
lua_pushvalue
(
lua_engine_ctx
->
lua
,
LUA_GLOBALSINDEX
);
luaSetGlobalProtection
(
lua_engine_ctx
->
lua
);
luaSetErrorMetatable
(
lua_engine_ctx
->
lua
);
luaSetTableProtectionRecursively
(
lua_engine_ctx
->
lua
);
/* protect globals */
lua_pop
(
lua_engine_ctx
->
lua
,
1
);
/* Save default globals to registry */
lua_pushvalue
(
lua_engine_ctx
->
lua
,
LUA_GLOBALSINDEX
);
lua_setfield
(
lua_engine_ctx
->
lua
,
LUA_REGISTRYINDEX
,
GLOBALS_API_NAME
);
/* save the engine_ctx on the registry so we can get it from the Lua interpreter */
luaSaveOnRegistry
(
lua_engine_ctx
->
lua
,
REGISTRY_ENGINE_CTX_NAME
,
lua_engine_ctx
);
/* Create new empty table to be the new globals, we will be able to control the real globals
* using metatable */
lua_newtable
(
lua_engine_ctx
->
lua
);
/* new globals */
lua_newtable
(
lua_engine_ctx
->
lua
);
/* new globals metatable */
lua_pushvalue
(
lua_engine_ctx
->
lua
,
LUA_GLOBALSINDEX
);
lua_setfield
(
lua_engine_ctx
->
lua
,
-
2
,
"__index"
);
lua_enablereadonlytable
(
lua_engine_ctx
->
lua
,
-
1
,
1
);
/* protect the metatable */
lua_setmetatable
(
lua_engine_ctx
->
lua
,
-
2
);
lua_enablereadonlytable
(
lua_engine_ctx
->
lua
,
-
1
,
1
);
/* protect the new global table */
lua_replace
(
lua_engine_ctx
->
lua
,
LUA_GLOBALSINDEX
);
/* set new global table as the new globals */
engine
*
lua_engine
=
zmalloc
(
sizeof
(
*
lua_engine
));
*
lua_engine
=
(
engine
)
{
.
engine_ctx
=
lua_engine_ctx
,
...
...
src/help.h
View file @
d375595d
/* Automatically generated by
utils
/generate-command-help.rb, do not edit. */
/* Automatically generated by
.
/generate-command-help.rb, do not edit. */
#ifndef __REDIS_HELP_H
#define __REDIS_HELP_H
...
...
@@ -130,12 +130,12 @@ struct commandHelp {
15
,
"2.6.0"
},
{
"BITFIELD"
,
"key
[
GET encoding offset]
[
SET encoding offset value
] [
INCRBY encoding offset increment
]
[OVERFLOW WRAP|SAT|FAIL]"
,
"key GET encoding offset
|[OVERFLOW WRAP|SAT|FAIL
] SET encoding offset value
|
INCRBY encoding offset increment
[GET encoding offset|
[OVERFLOW WRAP|SAT|FAIL]
SET encoding offset value|INCRBY encoding offset increment ...]
"
,
"Perform arbitrary bitfield integer operations on strings"
,
15
,
"3.2.0"
},
{
"BITFIELD_RO"
,
"key GET encoding offset"
,
"key GET encoding offset
[encoding offset ...]
"
,
"Perform arbitrary bitfield integer operations on strings. Read-only variant of BITFIELD"
,
15
,
"6.2.0"
},
...
...
@@ -690,12 +690,12 @@ struct commandHelp {
13
,
"3.2.10"
},
{
"GEOSEARCH"
,
"key
[
FROMMEMBER member
] [
FROMLONLAT longitude latitude
] [
BYRADIUS radius M|KM|FT|MI
] [
BYBOX width height M|KM|FT|MI
]
[ASC|DESC] [COUNT count [ANY]] [WITHCOORD] [WITHDIST] [WITHHASH]"
,
"key FROMMEMBER member
|
FROMLONLAT longitude latitude
BYRADIUS radius M|KM|FT|MI
|
BYBOX width height M|KM|FT|MI [ASC|DESC] [COUNT count [ANY]] [WITHCOORD] [WITHDIST] [WITHHASH]"
,
"Query a sorted set representing a geospatial index to fetch members inside an area of a box or a circle."
,
13
,
"6.2.0"
},
{
"GEOSEARCHSTORE"
,
"destination source
[
FROMMEMBER member
] [
FROMLONLAT longitude latitude
] [
BYRADIUS radius M|KM|FT|MI
] [
BYBOX width height M|KM|FT|MI
]
[ASC|DESC] [COUNT count [ANY]] [STOREDIST]"
,
"destination source FROMMEMBER member
|
FROMLONLAT longitude latitude
BYRADIUS radius M|KM|FT|MI
|
BYBOX width height M|KM|FT|MI [ASC|DESC] [COUNT count [ANY]] [STOREDIST]"
,
"Query a sorted set representing a geospatial index to fetch members inside an area of a box or a circle, and store the result in another key."
,
13
,
"6.2.0"
},
...
...
@@ -1000,7 +1000,7 @@ struct commandHelp {
1
,
"1.0.0"
},
{
"MIGRATE"
,
"host port key| destination-db timeout [COPY] [REPLACE] [AUTH password]
[AUTH2 username password] [KEYS key [key ...]]"
,
"host port key| destination-db timeout [COPY] [REPLACE]
[
[AUTH password]
|
[AUTH2 username password]
]
[KEYS key [key ...]]"
,
"Atomically transfer a key from a Redis instance to another one."
,
0
,
"2.6.0"
},
...
...
@@ -1355,7 +1355,7 @@ struct commandHelp {
8
,
"1.0.0"
},
{
"SET"
,
"key value [EX seconds|PX milliseconds|EXAT unix-time-seconds|PXAT unix-time-milliseconds|KEEPTTL]
[NX|XX] [GET]
"
,
"key value
[NX|XX] [GET]
[EX seconds|PX milliseconds|EXAT unix-time-seconds|PXAT unix-time-milliseconds|KEEPTTL]"
,
"Set the string value of a key"
,
1
,
"1.0.0"
},
...
...
src/listpack.c
View file @
d375595d
...
...
@@ -180,7 +180,8 @@ int lpStringToInt64(const char *s, unsigned long slen, int64_t *value) {
int
negative
=
0
;
uint64_t
v
;
if
(
plen
==
slen
)
/* Abort if length indicates this cannot possibly be an int */
if
(
slen
==
0
||
slen
>=
LONG_STR_SIZE
)
return
0
;
/* Special case: first and only digit is 0. */
...
...
src/listpack.h
View file @
d375595d
...
...
@@ -59,6 +59,8 @@ void lpFree(unsigned char *lp);
unsigned
char
*
lpShrinkToFit
(
unsigned
char
*
lp
);
unsigned
char
*
lpInsertString
(
unsigned
char
*
lp
,
unsigned
char
*
s
,
uint32_t
slen
,
unsigned
char
*
p
,
int
where
,
unsigned
char
**
newp
);
unsigned
char
*
lpInsertInteger
(
unsigned
char
*
lp
,
long
long
lval
,
unsigned
char
*
p
,
int
where
,
unsigned
char
**
newp
);
unsigned
char
*
lpPrepend
(
unsigned
char
*
lp
,
unsigned
char
*
s
,
uint32_t
slen
);
unsigned
char
*
lpPrependInteger
(
unsigned
char
*
lp
,
long
long
lval
);
unsigned
char
*
lpAppend
(
unsigned
char
*
lp
,
unsigned
char
*
s
,
uint32_t
slen
);
...
...
src/module.c
View file @
d375595d
...
...
@@ -464,11 +464,18 @@ static int moduleConvertArgFlags(int flags);
/* Use like malloc(). Memory allocated with this function is reported in
* Redis INFO memory, used for keys eviction according to maxmemory settings
* and in general is taken into account as memory allocated by Redis.
* You should avoid using malloc(). */
* You should avoid using malloc().
* This function panics if unable to allocate enough memory. */
void
*
RM_Alloc
(
size_t
bytes
)
{
return
zmalloc
(
bytes
);
}
/* Similar to RM_Alloc, but returns NULL in case of allocation failure, instead
* of panicking. */
void
*
RM_TryAlloc
(
size_t
bytes
)
{
return
ztrymalloc
(
bytes
);
}
/* Use like calloc(). Memory allocated with this function is reported in
* Redis INFO memory, used for keys eviction according to maxmemory settings
* and in general is taken into account as memory allocated by Redis.
...
...
@@ -710,6 +717,8 @@ void moduleFreeContext(RedisModuleCtx *ctx) {
if
(
server
.
busy_module_yield_flags
)
{
blockingOperationEnds
();
server
.
busy_module_yield_flags
=
BUSY_MODULE_YIELD_NONE
;
if
(
server
.
current_client
)
unprotectClient
(
server
.
current_client
);
unblockPostponedClients
();
}
}
...
...
@@ -1040,9 +1049,9 @@ RedisModuleCommand *moduleCreateCommandProxy(struct RedisModule *module, sds dec
* serve stale data. Don't use if you don't know what
* this means.
* * **"no-monitor"**: Don't propagate the command on monitor. Use this if
* the command has sensi
bl
e data among the arguments.
* the command has sensi
tiv
e data among the arguments.
* * **"no-slowlog"**: Don't log this command in the slowlog. Use this if
* the command has sensi
bl
e data among the arguments.
* the command has sensi
tiv
e data among the arguments.
* * **"fast"**: The command time complexity is not greater
* than O(log(N)) where N is the size of the collection or
* anything else representing the normal scalability
...
...
@@ -1924,6 +1933,7 @@ static struct redisCommandArg *moduleCopyCommandArgs(RedisModuleCommandArg *args
if
(
arg
->
token
)
realargs
[
j
].
token
=
zstrdup
(
arg
->
token
);
if
(
arg
->
summary
)
realargs
[
j
].
summary
=
zstrdup
(
arg
->
summary
);
if
(
arg
->
since
)
realargs
[
j
].
since
=
zstrdup
(
arg
->
since
);
if
(
arg
->
deprecated_since
)
realargs
[
j
].
deprecated_since
=
zstrdup
(
arg
->
deprecated_since
);
realargs
[
j
].
flags
=
moduleConvertArgFlags
(
arg
->
flags
);
if
(
arg
->
subargs
)
realargs
[
j
].
subargs
=
moduleCopyCommandArgs
(
arg
->
subargs
,
version
);
}
...
...
@@ -2079,6 +2089,12 @@ int RM_BlockedClientMeasureTimeEnd(RedisModuleBlockedClient *bc) {
* the -LOADING error)
*/
void
RM_Yield
(
RedisModuleCtx
*
ctx
,
int
flags
,
const
char
*
busy_reply
)
{
static
int
yield_nesting
=
0
;
/* Avoid nested calls to RM_Yield */
if
(
yield_nesting
)
return
;
yield_nesting
++
;
long
long
now
=
getMonotonicUs
();
if
(
now
>=
ctx
->
next_yield_time
)
{
/* In loading mode, there's no need to handle busy_module_yield_reply,
...
...
@@ -2092,10 +2108,13 @@ void RM_Yield(RedisModuleCtx *ctx, int flags, const char *busy_reply) {
server
.
busy_module_yield_reply
=
busy_reply
;
/* start the blocking operation if not already started. */
if
(
!
server
.
busy_module_yield_flags
)
{
server
.
busy_module_yield_flags
=
flags
&
REDISMODULE_YIELD_FLAG_CLIENTS
?
BUSY_MODULE_YIELD_CLIENTS
:
BUSY_MODULE_YIELD_EVENTS
;
server
.
busy_module_yield_flags
=
BUSY_MODULE_YIELD_EVENTS
;
blockingOperationStarts
();
if
(
server
.
current_client
)
protectClient
(
server
.
current_client
);
}
if
(
flags
&
REDISMODULE_YIELD_FLAG_CLIENTS
)
server
.
busy_module_yield_flags
|=
BUSY_MODULE_YIELD_CLIENTS
;
/* Let redis process events */
processEventsWhileBlocked
();
...
...
@@ -2110,6 +2129,7 @@ void RM_Yield(RedisModuleCtx *ctx, int flags, const char *busy_reply) {
/* decide when the next event should fire. */
ctx
->
next_yield_time
=
now
+
1000000
/
server
.
hz
;
}
yield_nesting
--
;
}
/* Set flags defining capabilities or behavior bit flags.
...
...
@@ -2639,9 +2659,7 @@ void RM_TrimStringAllocation(RedisModuleString *str) {
* if (argc != 3) return RedisModule_WrongArity(ctx);
*/
int
RM_WrongArity
(
RedisModuleCtx
*
ctx
)
{
addReplyErrorFormat
(
ctx
->
client
,
"wrong number of arguments for '%s' command"
,
(
char
*
)
ctx
->
client
->
argv
[
0
]
->
ptr
);
addReplyErrorArity
(
ctx
->
client
);
return
REDISMODULE_OK
;
}
...
...
@@ -3365,10 +3383,13 @@ int RM_GetClientInfoById(void *ci, uint64_t id) {
/* Publish a message to subscribers (see PUBLISH command). */
int
RM_PublishMessage
(
RedisModuleCtx
*
ctx
,
RedisModuleString
*
channel
,
RedisModuleString
*
message
)
{
UNUSED
(
ctx
);
int
receivers
=
pubsubPublishMessage
(
channel
,
message
);
if
(
server
.
cluster_enabled
)
clusterPropagatePublish
(
channel
,
message
);
return
receivers
;
return
pubsubPublishMessageAndPropagateToCluster
(
channel
,
message
,
0
);
}
/* Publish a message to shard-subscribers (see SPUBLISH command). */
int
RM_PublishMessageShard
(
RedisModuleCtx
*
ctx
,
RedisModuleString
*
channel
,
RedisModuleString
*
message
)
{
UNUSED
(
ctx
);
return
pubsubPublishMessageAndPropagateToCluster
(
channel
,
message
,
1
);
}
/* Return the currently selected DB. */
...
...
@@ -3615,7 +3636,7 @@ static void moduleInitKeyTypeSpecific(RedisModuleKey *key) {
* key does not exist, NULL is returned. However it is still safe to
* call RedisModule_CloseKey() and RedisModule_KeyType() on a NULL
* value. */
void
*
RM_OpenKey
(
RedisModuleCtx
*
ctx
,
robj
*
keyname
,
int
mode
)
{
RedisModuleKey
*
RM_OpenKey
(
RedisModuleCtx
*
ctx
,
robj
*
keyname
,
int
mode
)
{
RedisModuleKey
*
kp
;
robj
*
value
;
int
flags
=
mode
&
REDISMODULE_OPEN_KEY_NOTOUCH
?
LOOKUP_NOTOUCH
:
0
;
...
...
@@ -3633,7 +3654,7 @@ void *RM_OpenKey(RedisModuleCtx *ctx, robj *keyname, int mode) {
kp
=
zmalloc
(
sizeof
(
*
kp
));
moduleInitKey
(
kp
,
ctx
,
keyname
,
value
,
mode
);
autoMemoryAdd
(
ctx
,
REDISMODULE_AM_KEY
,
kp
);
return
(
void
*
)
kp
;
return
kp
;
}
/* Destroy a RedisModuleKey struct (freeing is the responsibility of the caller). */
...
...
@@ -5736,26 +5757,18 @@ RedisModuleCallReply *RM_Call(RedisModuleCtx *ctx, const char *cmdname, const ch
/* Lookup command now, after filters had a chance to make modifications
* if necessary.
*/
cmd
=
lookupCommand
(
c
->
argv
,
c
->
argc
);
if
(
!
cmd
)
{
cmd
=
c
->
cmd
=
c
->
lastcmd
=
c
->
realcmd
=
lookupCommand
(
c
->
argv
,
c
->
argc
);
sds
err
;
if
(
!
commandCheckExistence
(
c
,
error_as_call_replies
?
&
err
:
NULL
))
{
errno
=
ENOENT
;
if
(
error_as_call_replies
)
{
sds
msg
=
sdscatfmt
(
sdsempty
(),
"Unknown Redis "
"command '%S'."
,
c
->
argv
[
0
]
->
ptr
);
reply
=
callReplyCreateError
(
msg
,
ctx
);
}
if
(
error_as_call_replies
)
reply
=
callReplyCreateError
(
err
,
ctx
);
goto
cleanup
;
}
c
->
cmd
=
c
->
lastcmd
=
c
->
realcmd
=
cmd
;
/* Basic arity checks. */
if
((
cmd
->
arity
>
0
&&
cmd
->
arity
!=
argc
)
||
(
argc
<
-
cmd
->
arity
))
{
if
(
!
commandCheckArity
(
c
,
error_as_call_replies
?
&
err
:
NULL
))
{
errno
=
EINVAL
;
if
(
error_as_call_replies
)
{
sds
msg
=
sdscatfmt
(
sdsempty
(),
"Wrong number of "
"args calling Redis command '%S'."
,
c
->
cmd
->
fullname
);
reply
=
callReplyCreateError
(
msg
,
ctx
);
}
if
(
error_as_call_replies
)
reply
=
callReplyCreateError
(
err
,
ctx
);
goto
cleanup
;
}
...
...
@@ -5798,8 +5811,9 @@ RedisModuleCallReply *RM_Call(RedisModuleCtx *ctx, const char *cmdname, const ch
}
int
deny_write_type
=
writeCommandsDeniedByDiskError
();
int
obey_client
=
mustObeyClient
(
server
.
current_client
);
if
(
deny_write_type
!=
DISK_ERROR_TYPE_NONE
)
{
if
(
deny_write_type
!=
DISK_ERROR_TYPE_NONE
&&
!
obey_client
)
{
errno
=
ENOSPC
;
if
(
error_as_call_replies
)
{
sds
msg
=
writeCommandsGetDiskErrorMessage
(
deny_write_type
);
...
...
@@ -5841,7 +5855,7 @@ RedisModuleCallReply *RM_Call(RedisModuleCtx *ctx, const char *cmdname, const ch
/* If this is a Redis Cluster node, we need to make sure the module is not
* trying to access non-local keys, with the exception of commands
* received from our master. */
if
(
server
.
cluster_enabled
&&
!
(
ctx
->
client
->
flags
&
CLIENT_MASTER
))
{
if
(
server
.
cluster_enabled
&&
!
mustObeyClient
(
ctx
->
client
))
{
int
error_code
;
/* Duplicate relevant flags in the module client. */
c
->
flags
&=
~
(
CLIENT_READONLY
|
CLIENT_ASKING
);
...
...
@@ -5890,11 +5904,7 @@ RedisModuleCallReply *RM_Call(RedisModuleCtx *ctx, const char *cmdname, const ch
if
(
!
(
flags
&
REDISMODULE_ARGV_NO_REPLICAS
))
call_flags
|=
CMD_CALL_PROPAGATE_REPL
;
}
/* Set server.current_client */
client
*
old_client
=
server
.
current_client
;
server
.
current_client
=
c
;
call
(
c
,
call_flags
);
server
.
current_client
=
old_client
;
server
.
replication_allowed
=
prev_replication_allowed
;
serverAssert
((
c
->
flags
&
CLIENT_BLOCKED
)
==
0
);
...
...
@@ -6074,6 +6084,14 @@ const char *moduleTypeModuleName(moduleType *mt) {
return
mt
->
module
->
name
;
}
/* Return the module name from a module command */
const
char
*
moduleNameFromCommand
(
struct
redisCommand
*
cmd
)
{
serverAssert
(
cmd
->
proc
==
RedisModuleCommandDispatcher
);
RedisModuleCommand
*
cp
=
(
void
*
)(
unsigned
long
)
cmd
->
getkeys_proc
;
return
cp
->
module
->
name
;
}
/* Create a copy of a module type value using the copy callback. If failed
* or not supported, produce an error reply and return NULL.
*/
...
...
@@ -7623,6 +7641,8 @@ void moduleGILBeforeUnlock() {
if
(
server
.
busy_module_yield_flags
)
{
blockingOperationEnds
();
server
.
busy_module_yield_flags
=
BUSY_MODULE_YIELD_NONE
;
if
(
server
.
current_client
)
unprotectClient
(
server
.
current_client
);
unblockPostponedClients
();
}
}
...
...
@@ -8676,8 +8696,18 @@ int RM_ACLCheckChannelPermissions(RedisModuleUser *user, RedisModuleString *ch,
* Returns REDISMODULE_OK on success and REDISMODULE_ERR on error.
*
* For more information about ACL log, please refer to https://redis.io/commands/acl-log */
void
RM_ACLAddLogEntry
(
RedisModuleCtx
*
ctx
,
RedisModuleUser
*
user
,
RedisModuleString
*
object
)
{
addACLLogEntry
(
ctx
->
client
,
0
,
ACL_LOG_CTX_MODULE
,
-
1
,
user
->
user
->
name
,
sdsdup
(
object
->
ptr
));
int
RM_ACLAddLogEntry
(
RedisModuleCtx
*
ctx
,
RedisModuleUser
*
user
,
RedisModuleString
*
object
,
RedisModuleACLLogEntryReason
reason
)
{
int
acl_reason
;
switch
(
reason
)
{
case
REDISMODULE_ACL_LOG_AUTH
:
acl_reason
=
ACL_DENIED_AUTH
;
break
;
case
REDISMODULE_ACL_LOG_KEY
:
acl_reason
=
ACL_DENIED_KEY
;
break
;
case
REDISMODULE_ACL_LOG_CHANNEL
:
acl_reason
=
ACL_DENIED_CHANNEL
;
break
;
case
REDISMODULE_ACL_LOG_CMD
:
acl_reason
=
ACL_DENIED_CMD
;
break
;
default:
return
REDISMODULE_ERR
;
}
addACLLogEntry
(
ctx
->
client
,
acl_reason
,
ACL_LOG_CTX_MODULE
,
-
1
,
user
->
user
->
name
,
sdsdup
(
object
->
ptr
));
return
REDISMODULE_OK
;
}
/* Authenticate the client associated with the context with
...
...
@@ -9730,10 +9760,29 @@ int RM_CommandFilterArgDelete(RedisModuleCommandFilterCtx *fctx, int pos)
* with the allocation calls, since sometimes the underlying allocator
* will allocate more memory.
*/
size_t
RM_MallocSize
(
void
*
ptr
){
size_t
RM_MallocSize
(
void
*
ptr
)
{
return
zmalloc_size
(
ptr
);
}
/* Same as RM_MallocSize, except it works on RedisModuleString pointers.
*/
size_t
RM_MallocSizeString
(
RedisModuleString
*
str
)
{
serverAssert
(
str
->
type
==
OBJ_STRING
);
return
sizeof
(
*
str
)
+
getStringObjectSdsUsedMemory
(
str
);
}
/* Same as RM_MallocSize, except it works on RedisModuleDict pointers.
* Note that the returned value is only the overhead of the underlying structures,
* it does not include the allocation size of the keys and values.
*/
size_t
RM_MallocSizeDict
(
RedisModuleDict
*
dict
)
{
size_t
size
=
sizeof
(
RedisModuleDict
)
+
sizeof
(
rax
);
size
+=
dict
->
rax
->
numnodes
*
sizeof
(
raxNode
);
/* For more info about this weird line, see streamRadixTreeMemoryUsage */
size
+=
dict
->
rax
->
numnodes
*
sizeof
(
long
)
*
30
;
return
size
;
}
/* Return the a number between 0 to 1 indicating the amount of memory
* currently used, relative to the Redis "maxmemory" configuration.
*
...
...
@@ -10908,6 +10957,7 @@ int moduleFreeCommand(struct RedisModule *module, struct redisCommand *cmd) {
}
zfree
((
char
*
)
cmd
->
summary
);
zfree
((
char
*
)
cmd
->
since
);
zfree
((
char
*
)
cmd
->
deprecated_since
);
zfree
((
char
*
)
cmd
->
complexity
);
if
(
cmd
->
latency_histogram
)
{
hdr_close
(
cmd
->
latency_histogram
);
...
...
@@ -11275,6 +11325,7 @@ int moduleVerifyConfigFlags(unsigned int flags, configType type) {
|
REDISMODULE_CONFIG_HIDDEN
|
REDISMODULE_CONFIG_PROTECTED
|
REDISMODULE_CONFIG_DENY_LOADING
|
REDISMODULE_CONFIG_BITFLAGS
|
REDISMODULE_CONFIG_MEMORY
)))
{
serverLogRaw
(
LL_WARNING
,
"Invalid flag(s) for configuration"
);
return
REDISMODULE_ERR
;
...
...
@@ -11283,6 +11334,10 @@ int moduleVerifyConfigFlags(unsigned int flags, configType type) {
serverLogRaw
(
LL_WARNING
,
"Numeric flag provided for non-numeric configuration."
);
return
REDISMODULE_ERR
;
}
if
(
type
!=
ENUM_CONFIG
&&
flags
&
REDISMODULE_CONFIG_BITFLAGS
)
{
serverLogRaw
(
LL_WARNING
,
"Enum flag provided for non-enum configuration."
);
return
REDISMODULE_ERR
;
}
return
REDISMODULE_OK
;
}
...
...
@@ -11484,6 +11539,12 @@ unsigned int maskModuleNumericConfigFlags(unsigned int flags) {
return
new_flags
;
}
unsigned
int
maskModuleEnumConfigFlags
(
unsigned
int
flags
)
{
unsigned
int
new_flags
=
0
;
if
(
flags
&
REDISMODULE_CONFIG_BITFLAGS
)
new_flags
|=
MULTI_ARG_CONFIG
;
return
new_flags
;
}
/* Create a string config that Redis users can interact with via the Redis config file,
* `CONFIG SET`, `CONFIG GET`, and `CONFIG REWRITE` commands.
*
...
...
@@ -11523,6 +11584,7 @@ unsigned int maskModuleNumericConfigFlags(unsigned int flags) {
* * REDISMODULE_CONFIG_PROTECTED: This config will be only be modifiable based off the value of enable-protected-configs.
* * REDISMODULE_CONFIG_DENY_LOADING: This config is not modifiable while the server is loading data.
* * REDISMODULE_CONFIG_MEMORY: For numeric configs, this config will convert data unit notations into their byte equivalent.
* * REDISMODULE_CONFIG_BITFLAGS: For enum configs, this config will allow multiple entries to be combined as bit flags.
*
* Default values are used on startup to set the value if it is not provided via the config file
* or command line. Default values are also used to compare to on a config rewrite.
...
...
@@ -11638,7 +11700,7 @@ int RM_RegisterEnumConfig(RedisModuleCtx *ctx, const char *name, int default_val
enum_vals
[
num_enum_vals
].
name
=
NULL
;
enum_vals
[
num_enum_vals
].
val
=
0
;
listAddNodeTail
(
module
->
module_configs
,
new_config
);
flags
=
maskModuleConfigFlags
(
flags
);
flags
=
maskModuleConfigFlags
(
flags
)
|
maskModuleEnumConfigFlags
(
flags
)
;
addModuleEnumConfig
(
module
->
name
,
name
,
flags
,
new_config
,
default_val
,
enum_vals
);
return
REDISMODULE_OK
;
}
...
...
@@ -12225,6 +12287,7 @@ void moduleRegisterCoreAPI(void) {
server
.
moduleapi
=
dictCreate
(
&
moduleAPIDictType
);
server
.
sharedapi
=
dictCreate
(
&
moduleAPIDictType
);
REGISTER_API
(
Alloc
);
REGISTER_API
(
TryAlloc
);
REGISTER_API
(
Calloc
);
REGISTER_API
(
Realloc
);
REGISTER_API
(
Free
);
...
...
@@ -12490,6 +12553,7 @@ void moduleRegisterCoreAPI(void) {
REGISTER_API
(
ServerInfoGetFieldDouble
);
REGISTER_API
(
GetClientInfoById
);
REGISTER_API
(
PublishMessage
);
REGISTER_API
(
PublishMessageShard
);
REGISTER_API
(
SubscribeToServerEvent
);
REGISTER_API
(
SetLRU
);
REGISTER_API
(
GetLRU
);
...
...
@@ -12500,6 +12564,8 @@ void moduleRegisterCoreAPI(void) {
REGISTER_API
(
GetBlockedClientReadyKey
);
REGISTER_API
(
GetUsedMemoryRatio
);
REGISTER_API
(
MallocSize
);
REGISTER_API
(
MallocSizeString
);
REGISTER_API
(
MallocSizeDict
);
REGISTER_API
(
ScanCursorCreate
);
REGISTER_API
(
ScanCursorDestroy
);
REGISTER_API
(
ScanCursorRestart
);
...
...
src/modules/Makefile
View file @
d375595d
...
...
@@ -28,42 +28,42 @@ all: helloworld.so hellotype.so helloblock.so hellocluster.so hellotimer.so hell
helloworld.xo
:
../redismodule.h
helloworld.so
:
helloworld.xo
$(LD)
-o
$@
$
<
$(SHOBJ_LDFLAGS)
$(LIBS)
-lc
$(LD)
-o
$@
$
^
$(SHOBJ_LDFLAGS)
$(LIBS)
-lc
hellotype.xo
:
../redismodule.h
hellotype.so
:
hellotype.xo
$(LD)
-o
$@
$
<
$(SHOBJ_LDFLAGS)
$(LIBS)
-lc
$(LD)
-o
$@
$
^
$(SHOBJ_LDFLAGS)
$(LIBS)
-lc
helloblock.xo
:
../redismodule.h
helloblock.so
:
helloblock.xo
$(LD)
-o
$@
$
<
$(SHOBJ_LDFLAGS)
$(LIBS)
-lpthread
-lc
$(LD)
-o
$@
$
^
$(SHOBJ_LDFLAGS)
$(LIBS)
-lpthread
-lc
hellocluster.xo
:
../redismodule.h
hellocluster.so
:
hellocluster.xo
$(LD)
-o
$@
$
<
$(SHOBJ_LDFLAGS)
$(LIBS)
-lc
$(LD)
-o
$@
$
^
$(SHOBJ_LDFLAGS)
$(LIBS)
-lc
hellotimer.xo
:
../redismodule.h
hellotimer.so
:
hellotimer.xo
$(LD)
-o
$@
$
<
$(SHOBJ_LDFLAGS)
$(LIBS)
-lc
$(LD)
-o
$@
$
^
$(SHOBJ_LDFLAGS)
$(LIBS)
-lc
hellodict.xo
:
../redismodule.h
hellodict.so
:
hellodict.xo
$(LD)
-o
$@
$
<
$(SHOBJ_LDFLAGS)
$(LIBS)
-lc
$(LD)
-o
$@
$
^
$(SHOBJ_LDFLAGS)
$(LIBS)
-lc
hellohook.xo
:
../redismodule.h
hellohook.so
:
hellohook.xo
$(LD)
-o
$@
$
<
$(SHOBJ_LDFLAGS)
$(LIBS)
-lc
$(LD)
-o
$@
$
^
$(SHOBJ_LDFLAGS)
$(LIBS)
-lc
helloacl.xo
:
../redismodule.h
helloacl.so
:
helloacl.xo
$(LD)
-o
$@
$
<
$(SHOBJ_LDFLAGS)
$(LIBS)
-lc
$(LD)
-o
$@
$
^
$(SHOBJ_LDFLAGS)
$(LIBS)
-lc
clean
:
rm
-rf
*
.xo
*
.so
src/monotonic.c
View file @
d375595d
...
...
@@ -168,3 +168,13 @@ const char * monotonicInit() {
return
monotonic_info_string
;
}
const
char
*
monotonicInfoString
()
{
return
monotonic_info_string
;
}
monotonic_clock_type
monotonicGetType
()
{
if
(
getMonotonicUs
==
getMonotonicUs_posix
)
return
MONOTONIC_CLOCK_POSIX
;
return
MONOTONIC_CLOCK_HW
;
}
src/monotonic.h
View file @
d375595d
...
...
@@ -24,13 +24,22 @@ typedef uint64_t monotime;
/* Retrieve counter of micro-seconds relative to an arbitrary point in time. */
extern
monotime
(
*
getMonotonicUs
)(
void
);
typedef
enum
monotonic_clock_type
{
MONOTONIC_CLOCK_POSIX
,
MONOTONIC_CLOCK_HW
,
}
monotonic_clock_type
;
/* Call once at startup to initialize the monotonic clock. Though this only
* needs to be called once, it may be called additional times without impact.
* Returns a printable string indicating the type of clock initialized.
* (The returned string is static and doesn't need to be freed.) */
const
char
*
monotonicInit
();
const
char
*
monotonicInit
();
/* Return a string indicating the type of monotonic clock being used. */
const
char
*
monotonicInfoString
();
/* Return the type of monotonic clock being used. */
monotonic_clock_type
monotonicGetType
();
/* Functions to measure elapsed time. Example:
* monotime myTimer;
...
...
src/networking.c
View file @
d375595d
...
...
@@ -160,6 +160,7 @@ client *createClient(connection *conn) {
c
->
bulklen
=
-
1
;
c
->
sentlen
=
0
;
c
->
flags
=
0
;
c
->
slot
=
-
1
;
c
->
ctime
=
c
->
lastinteraction
=
server
.
unixtime
;
clientSetDefaultAuth
(
c
);
c
->
replstate
=
REPL_STATE_NONE
;
...
...
@@ -215,6 +216,23 @@ client *createClient(connection *conn) {
return
c
;
}
void
installClientWriteHandler
(
client
*
c
)
{
int
ae_barrier
=
0
;
/* For the fsync=always policy, we want that a given FD is never
* served for reading and writing in the same event loop iteration,
* so that in the middle of receiving the query, and serving it
* to the client, we'll call beforeSleep() that will do the
* actual fsync of AOF to disk. the write barrier ensures that. */
if
(
server
.
aof_state
==
AOF_ON
&&
server
.
aof_fsync
==
AOF_FSYNC_ALWAYS
)
{
ae_barrier
=
1
;
}
if
(
connSetWriteHandlerWithBarrier
(
c
->
conn
,
sendReplyToClient
,
ae_barrier
)
==
C_ERR
)
{
freeClientAsync
(
c
);
}
}
/* This function puts the client in the queue of clients that should write
* their output buffers to the socket. Note that it does not *yet* install
* the write handler, to start clients are put in a queue of clients that need
...
...
@@ -222,7 +240,7 @@ client *createClient(connection *conn) {
* handleClientsWithPendingWrites() function).
* If we fail and there is more data to write, compared to what the socket
* buffers can hold, then we'll really install the handler. */
void
c
lientIn
stallWriteHandler
(
client
*
c
)
{
void
putC
lientIn
PendingWriteQueue
(
client
*
c
)
{
/* Schedule the client to write the output buffers to the socket only
* if not already done and, for slaves, if the slave can actually receive
* writes at this stage. */
...
...
@@ -285,11 +303,11 @@ int prepareClientToWrite(client *c) {
* it should already be setup to do so (it has already pending data).
*
* If CLIENT_PENDING_READ is set, we're in an IO thread and should
* not
install a write handler
. Instead, it will be
done by
* handleClientsWithPendingReadsUsingThreads() upon return.
* not
put the client in pending write queue
. Instead, it will be
*
done by
handleClientsWithPendingReadsUsingThreads() upon return.
*/
if
(
!
clientHasPendingReplies
(
c
)
&&
io_threads_op
==
IO_THREADS_OP_IDLE
)
c
lientIn
stallWriteHandler
(
c
);
putC
lientIn
PendingWriteQueue
(
c
);
/* Authorize the caller to queue in the output buffer of this client. */
return
C_OK
;
...
...
@@ -521,6 +539,21 @@ void afterErrorReply(client *c, const char *s, size_t len, int flags) {
showLatestBacklog
();
}
server
.
stat_unexpected_error_replies
++
;
/* Based off the propagation error behavior, check if we need to panic here. There
* are currently two checked cases:
* * If this command was from our master and we are not a writable replica.
* * We are reading from an AOF file. */
int
panic_in_replicas
=
(
ctype
==
CLIENT_TYPE_MASTER
&&
server
.
repl_slave_ro
)
&&
(
server
.
propagation_error_behavior
==
PROPAGATION_ERR_BEHAVIOR_PANIC
||
server
.
propagation_error_behavior
==
PROPAGATION_ERR_BEHAVIOR_PANIC_ON_REPLICAS
);
int
panic_in_aof
=
c
->
id
==
CLIENT_ID_AOF
&&
server
.
propagation_error_behavior
==
PROPAGATION_ERR_BEHAVIOR_PANIC
;
if
(
panic_in_replicas
||
panic_in_aof
)
{
serverPanic
(
"This %s panicked sending an error to its %s"
" after processing the command '%s'"
,
from
,
to
,
cmdname
?
cmdname
:
"<unknown>"
);
}
}
}
...
...
@@ -1061,7 +1094,7 @@ void addReplySubcommandSyntaxError(client *c) {
sds
cmd
=
sdsnew
((
char
*
)
c
->
argv
[
0
]
->
ptr
);
sdstoupper
(
cmd
);
addReplyErrorFormat
(
c
,
"
U
nknown subcommand or wrong number of arguments for '%.128s'. Try %s HELP."
,
"
u
nknown subcommand or wrong number of arguments for '%.128s'. Try %s HELP."
,
(
char
*
)
c
->
argv
[
1
]
->
ptr
,
cmd
);
sdsfree
(
cmd
);
}
...
...
@@ -1995,20 +2028,7 @@ int handleClientsWithPendingWrites(void) {
/* If after the synchronous writes above we still have data to
* output to the client, we need to install the writable handler. */
if
(
clientHasPendingReplies
(
c
))
{
int
ae_barrier
=
0
;
/* For the fsync=always policy, we want that a given FD is never
* served for reading and writing in the same event loop iteration,
* so that in the middle of receiving the query, and serving it
* to the client, we'll call beforeSleep() that will do the
* actual fsync of AOF to disk. the write barrier ensures that. */
if
(
server
.
aof_state
==
AOF_ON
&&
server
.
aof_fsync
==
AOF_FSYNC_ALWAYS
)
{
ae_barrier
=
1
;
}
if
(
connSetWriteHandlerWithBarrier
(
c
->
conn
,
sendReplyToClient
,
ae_barrier
)
==
C_ERR
)
{
freeClientAsync
(
c
);
}
installClientWriteHandler
(
c
);
}
}
return
processed
;
...
...
@@ -2022,6 +2042,7 @@ void resetClient(client *c) {
c
->
reqtype
=
0
;
c
->
multibulklen
=
0
;
c
->
bulklen
=
-
1
;
c
->
slot
=
-
1
;
if
(
c
->
deferred_reply_errors
)
listRelease
(
c
->
deferred_reply_errors
);
...
...
@@ -2075,7 +2096,7 @@ void unprotectClient(client *c) {
c
->
flags
&=
~
CLIENT_PROTECTED
;
if
(
c
->
conn
)
{
connSetReadHandler
(
c
->
conn
,
readQueryFromClient
);
if
(
clientHasPendingReplies
(
c
))
c
lientIn
stallWriteHandler
(
c
);
if
(
clientHasPendingReplies
(
c
))
putC
lientIn
PendingWriteQueue
(
c
);
}
}
}
...
...
@@ -3808,7 +3829,7 @@ void flushSlavesOutputBuffers(void) {
}
}
/* Compute current most restictive pause type and its end time, aggregated for
/* Compute current most rest
r
ictive pause type and its end time, aggregated for
* all pause purposes. */
static
void
updateClientPauseTypeAndEndTime
(
void
)
{
pause_type
old_type
=
server
.
client_pause_type
;
...
...
@@ -4212,10 +4233,8 @@ int handleClientsWithPendingWritesUsingThreads(void) {
/* Install the write handler if there are pending writes in some
* of the clients. */
if
(
clientHasPendingReplies
(
c
)
&&
connSetWriteHandler
(
c
->
conn
,
sendReplyToClient
)
==
AE_ERR
)
{
freeClientAsync
(
c
);
if
(
clientHasPendingReplies
(
c
))
{
installClientWriteHandler
(
c
);
}
}
listEmpty
(
server
.
clients_pending_write
);
...
...
@@ -4327,10 +4346,10 @@ int handleClientsWithPendingReadsUsingThreads(void) {
}
/* We may have pending replies if a thread readQueryFromClient() produced
* replies and did not
install a write handler
(it can't).
* replies and did not
put the client in pending write queue
(it can't).
*/
if
(
!
(
c
->
flags
&
CLIENT_PENDING_WRITE
)
&&
clientHasPendingReplies
(
c
))
c
lientIn
stallWriteHandler
(
c
);
putC
lientIn
PendingWriteQueue
(
c
);
}
/* Update processed count on server */
...
...
src/notify.c
View file @
d375595d
...
...
@@ -57,6 +57,7 @@ int keyspaceEventsStringToFlags(char *classes) {
case
't'
:
flags
|=
NOTIFY_STREAM
;
break
;
case
'm'
:
flags
|=
NOTIFY_KEY_MISS
;
break
;
case
'd'
:
flags
|=
NOTIFY_MODULE
;
break
;
case
'n'
:
flags
|=
NOTIFY_NEW
;
break
;
default:
return
-
1
;
}
}
...
...
@@ -84,6 +85,7 @@ sds keyspaceEventsFlagsToString(int flags) {
if
(
flags
&
NOTIFY_EVICTED
)
res
=
sdscatlen
(
res
,
"e"
,
1
);
if
(
flags
&
NOTIFY_STREAM
)
res
=
sdscatlen
(
res
,
"t"
,
1
);
if
(
flags
&
NOTIFY_MODULE
)
res
=
sdscatlen
(
res
,
"d"
,
1
);
if
(
flags
&
NOTIFY_NEW
)
res
=
sdscatlen
(
res
,
"n"
,
1
);
}
if
(
flags
&
NOTIFY_KEYSPACE
)
res
=
sdscatlen
(
res
,
"K"
,
1
);
if
(
flags
&
NOTIFY_KEYEVENT
)
res
=
sdscatlen
(
res
,
"E"
,
1
);
...
...
@@ -124,7 +126,7 @@ void notifyKeyspaceEvent(int type, char *event, robj *key, int dbid) {
chan
=
sdscatlen
(
chan
,
"__:"
,
3
);
chan
=
sdscatsds
(
chan
,
key
->
ptr
);
chanobj
=
createObject
(
OBJ_STRING
,
chan
);
pubsubPublishMessage
(
chanobj
,
eventobj
);
pubsubPublishMessage
(
chanobj
,
eventobj
,
0
);
decrRefCount
(
chanobj
);
}
...
...
@@ -136,7 +138,7 @@ void notifyKeyspaceEvent(int type, char *event, robj *key, int dbid) {
chan
=
sdscatlen
(
chan
,
"__:"
,
3
);
chan
=
sdscatsds
(
chan
,
eventobj
->
ptr
);
chanobj
=
createObject
(
OBJ_STRING
,
chan
);
pubsubPublishMessage
(
chanobj
,
key
);
pubsubPublishMessage
(
chanobj
,
key
,
0
);
decrRefCount
(
chanobj
);
}
decrRefCount
(
eventobj
);
...
...
src/object.c
View file @
d375595d
...
...
@@ -958,7 +958,7 @@ char *strEncoding(int encoding) {
* on the insertion speed and thus the ability of the radix tree
* to compress prefixes. */
size_t
streamRadixTreeMemoryUsage
(
rax
*
rax
)
{
size_t
size
;
size_t
size
=
sizeof
(
*
rax
)
;
size
=
rax
->
numele
*
sizeof
(
streamID
);
size
+=
rax
->
numnodes
*
sizeof
(
raxNode
);
/* Add a fixed overhead due to the aux data pointer, children, ... */
...
...
src/pubsub.c
View file @
d375595d
...
...
@@ -499,16 +499,10 @@ int pubsubPublishMessageInternal(robj *channel, robj *message, pubsubtype type)
}
/* Publish a message to all the subscribers. */
int
pubsubPublishMessage
(
robj
*
channel
,
robj
*
message
)
{
return
pubsubPublishMessageInternal
(
channel
,
message
,
pubSubType
);
int
pubsubPublishMessage
(
robj
*
channel
,
robj
*
message
,
int
sharded
)
{
return
pubsubPublishMessageInternal
(
channel
,
message
,
sharded
?
pubSubShardType
:
pubSubType
);
}
/* Publish a shard message to all the subscribers. */
int
pubsubPublishMessageShard
(
robj
*
channel
,
robj
*
message
)
{
return
pubsubPublishMessageInternal
(
channel
,
message
,
pubSubShardType
);
}
/*-----------------------------------------------------------------------------
* Pubsub commands implementation
*----------------------------------------------------------------------------*/
...
...
@@ -578,6 +572,15 @@ void punsubscribeCommand(client *c) {
if
(
clientTotalPubSubSubscriptionCount
(
c
)
==
0
)
c
->
flags
&=
~
CLIENT_PUBSUB
;
}
/* This function wraps pubsubPublishMessage and also propagates the message to cluster.
* Used by the commands PUBLISH/SPUBLISH and their respective module APIs.*/
int
pubsubPublishMessageAndPropagateToCluster
(
robj
*
channel
,
robj
*
message
,
int
sharded
)
{
int
receivers
=
pubsubPublishMessage
(
channel
,
message
,
sharded
);
if
(
server
.
cluster_enabled
)
clusterPropagatePublish
(
channel
,
message
,
sharded
);
return
receivers
;
}
/* PUBLISH <channel> <message> */
void
publishCommand
(
client
*
c
)
{
if
(
server
.
sentinel_mode
)
{
...
...
@@ -585,10 +588,8 @@ void publishCommand(client *c) {
return
;
}
int
receivers
=
pubsubPublishMessage
(
c
->
argv
[
1
],
c
->
argv
[
2
]);
if
(
server
.
cluster_enabled
)
clusterPropagatePublish
(
c
->
argv
[
1
],
c
->
argv
[
2
]);
else
int
receivers
=
pubsubPublishMessageAndPropagateToCluster
(
c
->
argv
[
1
],
c
->
argv
[
2
],
0
);
if
(
!
server
.
cluster_enabled
)
forceCommandPropagation
(
c
,
PROPAGATE_REPL
);
addReplyLongLong
(
c
,
receivers
);
}
...
...
@@ -677,12 +678,9 @@ void channelList(client *c, sds pat, dict *pubsub_channels) {
/* SPUBLISH <channel> <message> */
void
spublishCommand
(
client
*
c
)
{
int
receivers
=
pubsubPublishMessageInternal
(
c
->
argv
[
1
],
c
->
argv
[
2
],
pubSubShardType
);
if
(
server
.
cluster_enabled
)
{
clusterPropagatePublishShard
(
c
->
argv
[
1
],
c
->
argv
[
2
]);
}
else
{
int
receivers
=
pubsubPublishMessageAndPropagateToCluster
(
c
->
argv
[
1
],
c
->
argv
[
2
],
1
);
if
(
!
server
.
cluster_enabled
)
forceCommandPropagation
(
c
,
PROPAGATE_REPL
);
}
addReplyLongLong
(
c
,
receivers
);
}
...
...
src/quicklist.h
View file @
d375595d
...
...
@@ -116,7 +116,7 @@ typedef struct quicklist {
typedef
struct
quicklistIter
{
quicklist
*
quicklist
;
quicklistNode
*
current
;
unsigned
char
*
zi
;
unsigned
char
*
zi
;
/* points to the current element */
long
offset
;
/* offset in current listpack */
int
direction
;
}
quicklistIter
;
...
...
@@ -141,7 +141,7 @@ typedef struct quicklistEntry {
/* quicklist compression disable */
#define QUICKLIST_NOCOMPRESS 0
/* quicklist container formats */
/* quicklist
node
container formats */
#define QUICKLIST_NODE_CONTAINER_PLAIN 1
#define QUICKLIST_NODE_CONTAINER_PACKED 2
...
...
src/rdb.c
View file @
d375595d
...
...
@@ -588,22 +588,11 @@ int rdbSaveDoubleValue(rio *rdb, double val) {
len
=
1
;
buf
[
0
]
=
(
val
<
0
)
?
255
:
254
;
}
else
{
#if (DBL_MANT_DIG >= 52) && (LLONG_MAX == 0x7fffffffffffffffLL)
/* Check if the float is in a safe range to be casted into a
* long long. We are assuming that long long is 64 bit here.
* Also we are assuming that there are no implementations around where
* double has precision < 52 bit.
*
* Under this assumptions we test if a double is inside an interval
* where casting to long long is safe. Then using two castings we
* make sure the decimal part is zero. If all this is true we use
* integer printing function that is much faster. */
double
min
=
-
4503599627370495
;
/* (2^52)-1 */
double
max
=
4503599627370496
;
/* -(2^52) */
if
(
val
>
min
&&
val
<
max
&&
val
==
((
double
)((
long
long
)
val
)))
ll2string
((
char
*
)
buf
+
1
,
sizeof
(
buf
)
-
1
,(
long
long
)
val
);
long
long
lvalue
;
/* Integer printing function is much faster, check if we can safely use it. */
if
(
double2ll
(
val
,
&
lvalue
))
ll2string
((
char
*
)
buf
+
1
,
sizeof
(
buf
)
-
1
,
lvalue
);
else
#endif
snprintf
((
char
*
)
buf
+
1
,
sizeof
(
buf
)
-
1
,
"%.17g"
,
val
);
buf
[
0
]
=
strlen
((
char
*
)
buf
+
1
);
len
=
buf
[
0
]
+
1
;
...
...
@@ -2433,6 +2422,12 @@ robj *rdbLoadObject(int rdbtype, rio *rdb, sds key, int dbid, int *error) {
return
NULL
;
}
if
(
s
->
length
&&
!
raxSize
(
s
->
rax
))
{
rdbReportCorruptRDB
(
"Stream length inconsistent with rax entries"
);
decrRefCount
(
o
);
return
NULL
;
}
/* Consumer groups loading */
uint64_t
cgroups_count
=
rdbLoadLen
(
rdb
,
NULL
);
if
(
cgroups_count
==
RDB_LENERR
)
{
...
...
src/redismodule.h
View file @
d375595d
...
...
@@ -88,6 +88,7 @@
#define REDISMODULE_CONFIG_DENY_LOADING (1ULL<<6)
/* This config is forbidden during loading. */
#define REDISMODULE_CONFIG_MEMORY (1ULL<<7)
/* Indicates if this value can be set as a memory value */
#define REDISMODULE_CONFIG_BITFLAGS (1ULL<<8)
/* Indicates if this value can be set as a multiple enum values */
/* StreamID type. */
typedef
struct
RedisModuleStreamID
{
...
...
@@ -322,6 +323,7 @@ typedef struct RedisModuleCommandArg {
const
char
*
summary
;
const
char
*
since
;
int
flags
;
/* The REDISMODULE_CMD_ARG_* macros. */
const
char
*
deprecated_since
;
struct
RedisModuleCommandArg
*
subargs
;
}
RedisModuleCommandArg
;
...
...
@@ -735,6 +737,13 @@ typedef struct RedisModuleSwapDbInfo {
#define RedisModuleSwapDbInfo RedisModuleSwapDbInfoV1
typedef
enum
{
REDISMODULE_ACL_LOG_AUTH
=
0
,
/* Authentication failure */
REDISMODULE_ACL_LOG_CMD
,
/* Command authorization failure */
REDISMODULE_ACL_LOG_KEY
,
/* Key authorization failure */
REDISMODULE_ACL_LOG_CHANNEL
/* Channel authorization failure */
}
RedisModuleACLLogEntryReason
;
/* ------------------------- End of common defines ------------------------ */
#ifndef REDISMODULE_CORE
...
...
@@ -861,6 +870,7 @@ typedef struct RedisModuleTypeMethods {
#endif
REDISMODULE_API
void
*
(
*
RedisModule_Alloc
)(
size_t
bytes
)
REDISMODULE_ATTR
;
REDISMODULE_API
void
*
(
*
RedisModule_TryAlloc
)(
size_t
bytes
)
REDISMODULE_ATTR
;
REDISMODULE_API
void
*
(
*
RedisModule_Realloc
)(
void
*
ptr
,
size_t
bytes
)
REDISMODULE_ATTR
;
REDISMODULE_API
void
(
*
RedisModule_Free
)(
void
*
ptr
)
REDISMODULE_ATTR
;
REDISMODULE_API
void
*
(
*
RedisModule_Calloc
)(
size_t
nmemb
,
size_t
size
)
REDISMODULE_ATTR
;
...
...
@@ -877,7 +887,7 @@ REDISMODULE_API int (*RedisModule_ReplyWithLongLong)(RedisModuleCtx *ctx, long l
REDISMODULE_API
int
(
*
RedisModule_GetSelectedDb
)(
RedisModuleCtx
*
ctx
)
REDISMODULE_ATTR
;
REDISMODULE_API
int
(
*
RedisModule_SelectDb
)(
RedisModuleCtx
*
ctx
,
int
newid
)
REDISMODULE_ATTR
;
REDISMODULE_API
int
(
*
RedisModule_KeyExists
)(
RedisModuleCtx
*
ctx
,
RedisModuleString
*
keyname
)
REDISMODULE_ATTR
;
REDISMODULE_API
void
*
(
*
RedisModule_OpenKey
)(
RedisModuleCtx
*
ctx
,
RedisModuleString
*
keyname
,
int
mode
)
REDISMODULE_ATTR
;
REDISMODULE_API
RedisModuleKey
*
(
*
RedisModule_OpenKey
)(
RedisModuleCtx
*
ctx
,
RedisModuleString
*
keyname
,
int
mode
)
REDISMODULE_ATTR
;
REDISMODULE_API
void
(
*
RedisModule_CloseKey
)(
RedisModuleKey
*
kp
)
REDISMODULE_ATTR
;
REDISMODULE_API
int
(
*
RedisModule_KeyType
)(
RedisModuleKey
*
kp
)
REDISMODULE_ATTR
;
REDISMODULE_API
size_t
(
*
RedisModule_ValueLength
)(
RedisModuleKey
*
kp
)
REDISMODULE_ATTR
;
...
...
@@ -990,6 +1000,7 @@ REDISMODULE_API unsigned long long (*RedisModule_GetClientId)(RedisModuleCtx *ct
REDISMODULE_API
RedisModuleString
*
(
*
RedisModule_GetClientUserNameById
)(
RedisModuleCtx
*
ctx
,
uint64_t
id
)
REDISMODULE_ATTR
;
REDISMODULE_API
int
(
*
RedisModule_GetClientInfoById
)(
void
*
ci
,
uint64_t
id
)
REDISMODULE_ATTR
;
REDISMODULE_API
int
(
*
RedisModule_PublishMessage
)(
RedisModuleCtx
*
ctx
,
RedisModuleString
*
channel
,
RedisModuleString
*
message
)
REDISMODULE_ATTR
;
REDISMODULE_API
int
(
*
RedisModule_PublishMessageShard
)(
RedisModuleCtx
*
ctx
,
RedisModuleString
*
channel
,
RedisModuleString
*
message
)
REDISMODULE_ATTR
;
REDISMODULE_API
int
(
*
RedisModule_GetContextFlags
)(
RedisModuleCtx
*
ctx
)
REDISMODULE_ATTR
;
REDISMODULE_API
int
(
*
RedisModule_AvoidReplicaTraffic
)()
REDISMODULE_ATTR
;
REDISMODULE_API
void
*
(
*
RedisModule_PoolAlloc
)(
RedisModuleCtx
*
ctx
,
size_t
bytes
)
REDISMODULE_ATTR
;
...
...
@@ -1149,6 +1160,8 @@ REDISMODULE_API int (*RedisModule_ExitFromChild)(int retcode) REDISMODULE_ATTR;
REDISMODULE_API
int
(
*
RedisModule_KillForkChild
)(
int
child_pid
)
REDISMODULE_ATTR
;
REDISMODULE_API
float
(
*
RedisModule_GetUsedMemoryRatio
)()
REDISMODULE_ATTR
;
REDISMODULE_API
size_t
(
*
RedisModule_MallocSize
)(
void
*
ptr
)
REDISMODULE_ATTR
;
REDISMODULE_API
size_t
(
*
RedisModule_MallocSizeString
)(
RedisModuleString
*
str
)
REDISMODULE_ATTR
;
REDISMODULE_API
size_t
(
*
RedisModule_MallocSizeDict
)(
RedisModuleDict
*
dict
)
REDISMODULE_ATTR
;
REDISMODULE_API
RedisModuleUser
*
(
*
RedisModule_CreateModuleUser
)(
const
char
*
name
)
REDISMODULE_ATTR
;
REDISMODULE_API
void
(
*
RedisModule_FreeModuleUser
)(
RedisModuleUser
*
user
)
REDISMODULE_ATTR
;
REDISMODULE_API
int
(
*
RedisModule_SetModuleUserACL
)(
RedisModuleUser
*
user
,
const
char
*
acl
)
REDISMODULE_ATTR
;
...
...
@@ -1157,7 +1170,7 @@ REDISMODULE_API RedisModuleUser * (*RedisModule_GetModuleUserFromUserName)(Redis
REDISMODULE_API
int
(
*
RedisModule_ACLCheckCommandPermissions
)(
RedisModuleUser
*
user
,
RedisModuleString
**
argv
,
int
argc
)
REDISMODULE_ATTR
;
REDISMODULE_API
int
(
*
RedisModule_ACLCheckKeyPermissions
)(
RedisModuleUser
*
user
,
RedisModuleString
*
key
,
int
flags
)
REDISMODULE_ATTR
;
REDISMODULE_API
int
(
*
RedisModule_ACLCheckChannelPermissions
)(
RedisModuleUser
*
user
,
RedisModuleString
*
ch
,
int
literal
)
REDISMODULE_ATTR
;
REDISMODULE_API
void
(
*
RedisModule_ACLAddLogEntry
)(
RedisModuleCtx
*
ctx
,
RedisModuleUser
*
user
,
RedisModuleString
*
object
)
REDISMODULE_ATTR
;
REDISMODULE_API
void
(
*
RedisModule_ACLAddLogEntry
)(
RedisModuleCtx
*
ctx
,
RedisModuleUser
*
user
,
RedisModuleString
*
object
,
RedisModuleACLLogEntryReason
reason
)
REDISMODULE_ATTR
;
REDISMODULE_API
int
(
*
RedisModule_AuthenticateClientWithACLUser
)(
RedisModuleCtx
*
ctx
,
const
char
*
name
,
size_t
len
,
RedisModuleUserChangedFunc
callback
,
void
*
privdata
,
uint64_t
*
client_id
)
REDISMODULE_ATTR
;
REDISMODULE_API
int
(
*
RedisModule_AuthenticateClientWithUser
)(
RedisModuleCtx
*
ctx
,
RedisModuleUser
*
user
,
RedisModuleUserChangedFunc
callback
,
void
*
privdata
,
uint64_t
*
client_id
)
REDISMODULE_ATTR
;
REDISMODULE_API
int
(
*
RedisModule_DeauthenticateAndCloseClient
)(
RedisModuleCtx
*
ctx
,
uint64_t
client_id
)
REDISMODULE_ATTR
;
...
...
@@ -1191,6 +1204,7 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int
void
*
getapifuncptr
=
((
void
**
)
ctx
)[
0
];
RedisModule_GetApi
=
(
int
(
*
)(
const
char
*
,
void
*
))
(
unsigned
long
)
getapifuncptr
;
REDISMODULE_GET_API
(
Alloc
);
REDISMODULE_GET_API
(
TryAlloc
);
REDISMODULE_GET_API
(
Calloc
);
REDISMODULE_GET_API
(
Free
);
REDISMODULE_GET_API
(
Realloc
);
...
...
@@ -1411,6 +1425,7 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int
REDISMODULE_GET_API
(
ServerInfoGetFieldDouble
);
REDISMODULE_GET_API
(
GetClientInfoById
);
REDISMODULE_GET_API
(
PublishMessage
);
REDISMODULE_GET_API
(
PublishMessageShard
);
REDISMODULE_GET_API
(
SubscribeToServerEvent
);
REDISMODULE_GET_API
(
SetLRU
);
REDISMODULE_GET_API
(
GetLRU
);
...
...
@@ -1478,6 +1493,8 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int
REDISMODULE_GET_API
(
KillForkChild
);
REDISMODULE_GET_API
(
GetUsedMemoryRatio
);
REDISMODULE_GET_API
(
MallocSize
);
REDISMODULE_GET_API
(
MallocSizeString
);
REDISMODULE_GET_API
(
MallocSizeDict
);
REDISMODULE_GET_API
(
CreateModuleUser
);
REDISMODULE_GET_API
(
FreeModuleUser
);
REDISMODULE_GET_API
(
SetModuleUserACL
);
...
...
src/replication.c
View file @
d375595d
...
...
@@ -327,9 +327,6 @@ void feedReplicationBuffer(char *s, size_t len) {
server
.
master_repl_offset
+=
len
;
server
.
repl_backlog
->
histlen
+=
len
;
/* Install write handler for all replicas. */
prepareReplicasToWrite
();
size_t
start_pos
=
0
;
/* The position of referenced block to start sending. */
listNode
*
start_node
=
NULL
;
/* Replica/backlog starts referenced node. */
int
add_new_block
=
0
;
/* Create new block if current block is total used. */
...
...
@@ -440,6 +437,10 @@ void replicationFeedSlaves(list *slaves, int dictid, robj **argv, int argc) {
/* We can't have slaves attached and no backlog. */
serverAssert
(
!
(
listLength
(
slaves
)
!=
0
&&
server
.
repl_backlog
==
NULL
));
/* Must install write handler for all replicas first before feeding
* replication stream. */
prepareReplicasToWrite
();
/* Send SELECT command to every slave if needed. */
if
(
server
.
slaveseldb
!=
dictid
)
{
robj
*
selectcmd
;
...
...
@@ -539,7 +540,12 @@ void replicationFeedStreamFromMasterStream(char *buf, size_t buflen) {
/* There must be replication backlog if having attached slaves. */
if
(
listLength
(
server
.
slaves
))
serverAssert
(
server
.
repl_backlog
!=
NULL
);
if
(
server
.
repl_backlog
)
feedReplicationBuffer
(
buf
,
buflen
);
if
(
server
.
repl_backlog
)
{
/* Must install write handler for all replicas first before feeding
* replication stream. */
prepareReplicasToWrite
();
feedReplicationBuffer
(
buf
,
buflen
);
}
}
void
replicationFeedMonitors
(
client
*
c
,
list
*
monitors
,
int
dictid
,
robj
**
argv
,
int
argc
)
{
...
...
@@ -1285,7 +1291,7 @@ void replicaStartCommandStream(client *slave) {
return
;
}
c
lientIn
stallWriteHandler
(
slave
);
putC
lientIn
PendingWriteQueue
(
slave
);
}
/* We call this function periodically to remove an RDB file that was
...
...
@@ -1969,6 +1975,20 @@ void readSyncBulkPayload(connection *conn) {
/* We need to stop any AOF rewriting child before flushing and parsing
* the RDB, otherwise we'll create a copy-on-write disaster. */
if
(
server
.
aof_state
!=
AOF_OFF
)
stopAppendOnly
();
/* Also try to stop save RDB child before flushing and parsing the RDB:
* 1. Ensure background save doesn't overwrite synced data after being loaded.
* 2. Avoid copy-on-write disaster. */
if
(
server
.
child_type
==
CHILD_TYPE_RDB
)
{
if
(
!
use_diskless_load
)
{
serverLog
(
LL_NOTICE
,
"Replica is about to load the RDB file received from the "
"master, but there is a pending RDB child running. "
"Killing process %ld and removing its temp file to avoid "
"any race"
,
(
long
)
server
.
child_pid
);
}
killRDBChild
();
}
if
(
use_diskless_load
&&
server
.
repl_diskless_load
==
REPL_DISKLESS_LOAD_SWAPDB
)
{
/* Initialize empty tempDb dictionaries. */
...
...
@@ -2100,16 +2120,6 @@ void readSyncBulkPayload(connection *conn) {
connNonBlock
(
conn
);
connRecvTimeout
(
conn
,
0
);
}
else
{
/* Ensure background save doesn't overwrite synced data */
if
(
server
.
child_type
==
CHILD_TYPE_RDB
)
{
serverLog
(
LL_NOTICE
,
"Replica is about to load the RDB file received from the "
"master, but there is a pending RDB child running. "
"Killing process %ld and removing its temp file to avoid "
"any race"
,
(
long
)
server
.
child_pid
);
killRDBChild
();
}
/* Make sure the new file (also used for persistence) is fully synced
* (not covered by earlier calls to rdb_fsync_range). */
...
...
src/resp_parser.h
View file @
d375595d
...
...
@@ -68,10 +68,10 @@ typedef struct ReplyParserCallbacks {
/* Called when the parser reaches a double (','), which is passed as an argument 'val' */
void
(
*
double_callback
)(
void
*
ctx
,
double
val
,
const
char
*
proto
,
size_t
proto_len
);
/* Called when the parser reaches a big number ('
,
'), which is passed as 'str' along with its length 'len' */
/* Called when the parser reaches a big number ('
(
'), which is passed as 'str' along with its length 'len' */
void
(
*
big_number_callback
)(
void
*
ctx
,
const
char
*
str
,
size_t
len
,
const
char
*
proto
,
size_t
proto_len
);
/* Called when the parser reaches a string, which is passed as 'str' along with its 'format' and length 'len' */
/* Called when the parser reaches a string
('=')
, which is passed as 'str' along with its 'format' and length 'len' */
void
(
*
verbatim_string_callback
)(
void
*
ctx
,
const
char
*
format
,
const
char
*
str
,
size_t
len
,
const
char
*
proto
,
size_t
proto_len
);
/* Called when the parser reaches an attribute ('|'). The attribute length is passed as an argument 'len' */
...
...
src/script.c
View file @
d375595d
...
...
@@ -36,6 +36,7 @@ scriptFlag scripts_flags_def[] = {
{.
flag
=
SCRIPT_FLAG_ALLOW_OOM
,
.
str
=
"allow-oom"
},
{.
flag
=
SCRIPT_FLAG_ALLOW_STALE
,
.
str
=
"allow-stale"
},
{.
flag
=
SCRIPT_FLAG_NO_CLUSTER
,
.
str
=
"no-cluster"
},
{.
flag
=
SCRIPT_FLAG_ALLOW_CROSS_SLOT
,
.
str
=
"allow-cross-slot-keys"
},
{.
flag
=
0
,
.
str
=
NULL
},
/* flags array end */
};
...
...
@@ -114,6 +115,7 @@ int scriptPrepareForRun(scriptRunCtx *run_ctx, client *engine_client, client *ca
int
running_stale
=
server
.
masterhost
&&
server
.
repl_state
!=
REPL_STATE_CONNECTED
&&
server
.
repl_serve_stale_data
==
0
;
int
obey_client
=
mustObeyClient
(
caller
);
if
(
!
(
script_flags
&
SCRIPT_FLAG_EVAL_COMPAT_MODE
))
{
if
((
script_flags
&
SCRIPT_FLAG_NO_CLUSTER
)
&&
server
.
cluster_enabled
)
{
...
...
@@ -139,16 +141,14 @@ int scriptPrepareForRun(scriptRunCtx *run_ctx, client *engine_client, client *ca
* 1. we are not a readonly replica
* 2. no disk error detected
* 3. command is not `fcall_ro`/`eval[sha]_ro` */
if
(
server
.
masterhost
&&
server
.
repl_slave_ro
&&
caller
->
id
!=
CLIENT_ID_AOF
&&
!
(
caller
->
flags
&
CLIENT_MASTER
))
{
if
(
server
.
masterhost
&&
server
.
repl_slave_ro
&&
!
obey_client
)
{
addReplyError
(
caller
,
"Can not run script with write flag on readonly replica"
);
return
C_ERR
;
}
/* Deny writes if we're unale to persist. */
int
deny_write_type
=
writeCommandsDeniedByDiskError
();
if
(
deny_write_type
!=
DISK_ERROR_TYPE_NONE
&&
server
.
masterhost
==
NULL
)
{
if
(
deny_write_type
!=
DISK_ERROR_TYPE_NONE
&&
!
obey_client
)
{
if
(
deny_write_type
==
DISK_ERROR_TYPE_RDB
)
addReplyError
(
caller
,
"-MISCONF Redis is configured to save RDB snapshots, "
"but it's currently unable to persist to disk. "
...
...
@@ -219,6 +219,10 @@ int scriptPrepareForRun(scriptRunCtx *run_ctx, client *engine_client, client *ca
run_ctx
->
flags
|=
SCRIPT_ALLOW_OOM
;
}
if
((
script_flags
&
SCRIPT_FLAG_EVAL_COMPAT_MODE
)
||
(
script_flags
&
SCRIPT_FLAG_ALLOW_CROSS_SLOT
))
{
run_ctx
->
flags
|=
SCRIPT_ALLOW_CROSS_SLOT
;
}
/* set the curr_run_ctx so we can use it to kill the script if needed */
curr_run_ctx
=
run_ctx
;
...
...
@@ -269,7 +273,7 @@ void scriptKill(client *c, int is_eval) {
addReplyError
(
c
,
"-NOTBUSY No scripts in execution right now."
);
return
;
}
if
(
curr_run_ctx
->
original_client
->
flags
&
CLIENT_MASTER
)
{
if
(
mustObeyClient
(
curr_run_ctx
->
original_client
)
)
{
addReplyError
(
c
,
"-UNKILLABLE The busy script was sent by a master instance in the context of replication and cannot be killed."
);
}
...
...
@@ -334,8 +338,8 @@ static int scriptVerifyWriteCommandAllow(scriptRunCtx *run_ctx, char **err) {
* of this script. */
int
deny_write_type
=
writeCommandsDeniedByDiskError
();
if
(
server
.
masterhost
&&
server
.
repl_slave_ro
&&
run_ctx
->
original_client
->
id
!=
CLIENT_ID_AOF
&&
!
(
run_ctx
->
original_client
->
flags
&
CLIENT_MASTER
))
if
(
server
.
masterhost
&&
server
.
repl_slave_ro
&&
!
mustObeyClient
(
run_ctx
->
original_client
))
{
*
err
=
sdsdup
(
shared
.
roslaveerr
->
ptr
);
return
C_ERR
;
...
...
@@ -380,8 +384,7 @@ static int scriptVerifyOOM(scriptRunCtx *run_ctx, char **err) {
* in the middle. */
if
(
server
.
maxmemory
&&
/* Maxmemory is actually enabled. */
run_ctx
->
original_client
->
id
!=
CLIENT_ID_AOF
&&
/* Don't care about mem if loading from AOF. */
!
server
.
masterhost
&&
/* Slave must execute the script. */
!
mustObeyClient
(
run_ctx
->
original_client
)
&&
/* Don't care about mem for replicas or AOF. */
!
(
run_ctx
->
flags
&
SCRIPT_WRITE_DIRTY
)
&&
/* Script had no side effects so far. */
server
.
script_oom
&&
/* Detected OOM when script start. */
(
run_ctx
->
c
->
cmd
->
flags
&
CMD_DENYOOM
))
...
...
@@ -393,8 +396,8 @@ static int scriptVerifyOOM(scriptRunCtx *run_ctx, char **err) {
return
C_OK
;
}
static
int
scriptVerifyClusterState
(
client
*
c
,
client
*
original_c
,
sds
*
err
)
{
if
(
!
server
.
cluster_enabled
||
original_c
->
id
==
CLIENT_ID_AOF
||
(
original_c
->
flags
&
CLIENT_MASTER
))
{
static
int
scriptVerifyClusterState
(
scriptRunCtx
*
run_ctx
,
client
*
c
,
client
*
original_c
,
sds
*
err
)
{
if
(
!
server
.
cluster_enabled
||
mustObeyClient
(
original_c
))
{
return
C_OK
;
}
/* If this is a Redis Cluster node, we need to make sure the script is not
...
...
@@ -404,7 +407,8 @@ static int scriptVerifyClusterState(client *c, client *original_c, sds *err) {
/* Duplicate relevant flags in the script client. */
c
->
flags
&=
~
(
CLIENT_READONLY
|
CLIENT_ASKING
);
c
->
flags
|=
original_c
->
flags
&
(
CLIENT_READONLY
|
CLIENT_ASKING
);
if
(
getNodeByQuery
(
c
,
c
->
cmd
,
c
->
argv
,
c
->
argc
,
NULL
,
&
error_code
)
!=
server
.
cluster
->
myself
)
{
int
hashslot
=
-
1
;
if
(
getNodeByQuery
(
c
,
c
->
cmd
,
c
->
argv
,
c
->
argc
,
&
hashslot
,
&
error_code
)
!=
server
.
cluster
->
myself
)
{
if
(
error_code
==
CLUSTER_REDIR_DOWN_RO_STATE
)
{
*
err
=
sdsnew
(
"Script attempted to execute a write command while the "
...
...
@@ -418,6 +422,19 @@ static int scriptVerifyClusterState(client *c, client *original_c, sds *err) {
}
return
C_ERR
;
}
/* If the script declared keys in advanced, the cross slot error would have
* already been thrown. This is only checking for cross slot keys being accessed
* that weren't pre-declared. */
if
(
hashslot
!=
-
1
&&
!
(
run_ctx
->
flags
&
SCRIPT_ALLOW_CROSS_SLOT
))
{
if
(
original_c
->
slot
==
-
1
)
{
original_c
->
slot
=
hashslot
;
}
else
if
(
original_c
->
slot
!=
hashslot
)
{
*
err
=
sdsnew
(
"Script attempted to access keys that do not hash to "
"the same slot"
);
return
C_ERR
;
}
}
return
C_OK
;
}
...
...
@@ -522,7 +539,7 @@ void scriptCall(scriptRunCtx *run_ctx, robj* *argv, int argc, sds *err) {
run_ctx
->
flags
|=
SCRIPT_WRITE_DIRTY
;
}
if
(
scriptVerifyClusterState
(
c
,
run_ctx
->
original_client
,
err
)
!=
C_OK
)
{
if
(
scriptVerifyClusterState
(
run_ctx
,
c
,
run_ctx
->
original_client
,
err
)
!=
C_OK
)
{
goto
error
;
}
...
...
Prev
1
2
3
4
5
6
7
Next
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
.
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment