Commit 20618c71 authored by Oran Agra's avatar Oran Agra
Browse files

Merge remote-tracking branch 'origin/unstable' into 7.0

parents fb4e0d40 89772ed8
......@@ -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;
}
......
......@@ -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,
......
/* 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" },
......
......@@ -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. */
......
......@@ -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);
......
......@@ -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 sensible data among the arguments.
* the command has sensitive data among the arguments.
* * **"no-slowlog"**: Don't log this command in the slowlog. Use this if
* the command has sensible data among the arguments.
* the command has sensitive 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);
......
......@@ -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
......@@ -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;
}
......@@ -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;
......
......@@ -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 clientInstallWriteHandler(client *c) {
void putClientInPendingWriteQueue(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)
clientInstallWriteHandler(c);
putClientInPendingWriteQueue(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,
"Unknown subcommand or wrong number of arguments for '%.128s'. Try %s HELP.",
"unknown 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)) clientInstallWriteHandler(c);
if (clientHasPendingReplies(c)) putClientInPendingWriteQueue(c);
}
}
}
......@@ -3808,7 +3829,7 @@ void flushSlavesOutputBuffers(void) {
}
}
/* Compute current most restictive pause type and its end time, aggregated for
/* Compute current most restrictive 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))
clientInstallWriteHandler(c);
putClientInPendingWriteQueue(c);
}
/* Update processed count on server */
......
......@@ -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);
......
......@@ -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, ... */
......
......@@ -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);
}
......
......@@ -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
......
......@@ -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) {
......
......@@ -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);
......
......@@ -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;
}
clientInstallWriteHandler(slave);
putClientInPendingWriteQueue(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). */
......
......@@ -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' */
......
......@@ -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;
}
......
......@@ -64,6 +64,7 @@
#define SCRIPT_READ_ONLY (1ULL<<5) /* indicate that the current script should only perform read commands */
#define SCRIPT_ALLOW_OOM (1ULL<<6) /* indicate to allow any command even if OOM reached */
#define SCRIPT_EVAL_MODE (1ULL<<7) /* Indicate that the current script called from legacy Lua */
#define SCRIPT_ALLOW_CROSS_SLOT (1ULL<<8) /* Indicate that the current script may access keys from multiple slots */
typedef struct scriptRunCtx scriptRunCtx;
struct scriptRunCtx {
......@@ -82,6 +83,7 @@ struct scriptRunCtx {
#define SCRIPT_FLAG_ALLOW_STALE (1ULL<<2)
#define SCRIPT_FLAG_NO_CLUSTER (1ULL<<3)
#define SCRIPT_FLAG_EVAL_COMPAT_MODE (1ULL<<4) /* EVAL Script backwards compatible behavior, no shebang provided */
#define SCRIPT_FLAG_ALLOW_CROSS_SLOT (1ULL<<5)
/* Defines a script flags */
typedef struct scriptFlag {
......
Markdown is supported
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment