Commit cbd46317 authored by meir@redislabs.com's avatar meir@redislabs.com Committed by meir
Browse files

Redis Functions - Added redis function unit and Lua engine

Redis function unit is located inside functions.c
and contains Redis Function implementation:
1. FUNCTION commands:
  * FUNCTION CREATE
  * FCALL
  * FCALL_RO
  * FUNCTION DELETE
  * FUNCTION KILL
  * FUNCTION INFO
2. Register engine

In addition, this commit introduce the first engine
that uses the Redis Function capabilities, the
Lua engine.
parent f21dc38a
...@@ -443,6 +443,16 @@ This file also implements both the `SYNC` and `PSYNC` commands that are ...@@ -443,6 +443,16 @@ This file also implements both the `SYNC` and `PSYNC` commands that are
used in order to perform the first synchronization between masters and used in order to perform the first synchronization between masters and
replicas, or to continue the replication after a disconnection. replicas, or to continue the replication after a disconnection.
Script
---
The script unit is compose of 3 units
* `script.c` - integration of scripts with Redis (commands execution, set replication/resp, ..)
* `script_lua.c` - responsible to execute Lua code, uses script.c to interact with Redis from within the Lua code.
* `function_lua.c` - contains the Lua engine implementation, uses script_lua.c to execute the Lua code.
* `functions.c` - Contains Redis Functions implementation (FUNCTION command), uses functions_lua.c if the function it wants to invoke needs the Lua engine.
* `eval.c` - Contains the `eval` implementation using `script_lua.c` to invoke the Lua code.
Other C files Other C files
--- ---
...@@ -451,7 +461,6 @@ Other C files ...@@ -451,7 +461,6 @@ Other C files
* `sds.c` is the Redis string library, check https://github.com/antirez/sds for more information. * `sds.c` is the Redis string library, check https://github.com/antirez/sds for more information.
* `anet.c` is a library to use POSIX networking in a simpler way compared to the raw interface exposed by the kernel. * `anet.c` is a library to use POSIX networking in a simpler way compared to the raw interface exposed by the kernel.
* `dict.c` is an implementation of a non-blocking hash table which rehashes incrementally. * `dict.c` is an implementation of a non-blocking hash table which rehashes incrementally.
* `scripting.c` implements Lua scripting. It is completely self-contained and isolated from the rest of the Redis implementation and is simple enough to understand if you are familiar with the Lua API.
* `cluster.c` implements the Redis Cluster. Probably a good read only after being very familiar with the rest of the Redis code base. If you want to read `cluster.c` make sure to read the [Redis Cluster specification][4]. * `cluster.c` implements the Redis Cluster. Probably a good read only after being very familiar with the rest of the Redis code base. If you want to read `cluster.c` make sure to read the [Redis Cluster specification][4].
[4]: https://redis.io/topics/cluster-spec [4]: https://redis.io/topics/cluster-spec
......
...@@ -309,7 +309,7 @@ endif ...@@ -309,7 +309,7 @@ endif
REDIS_SERVER_NAME=redis-server$(PROG_SUFFIX) REDIS_SERVER_NAME=redis-server$(PROG_SUFFIX)
REDIS_SENTINEL_NAME=redis-sentinel$(PROG_SUFFIX) REDIS_SENTINEL_NAME=redis-sentinel$(PROG_SUFFIX)
REDIS_SERVER_OBJ=adlist.o quicklist.o ae.o anet.o dict.o server.o sds.o zmalloc.o lzf_c.o lzf_d.o pqsort.o zipmap.o sha1.o ziplist.o release.o networking.o util.o object.o db.o replication.o rdb.o t_string.o t_list.o t_set.o t_zset.o t_hash.o config.o aof.o pubsub.o multi.o debug.o sort.o intset.o syncio.o cluster.o crc16.o endianconv.o slowlog.o eval.o bio.o rio.o rand.o memtest.o crcspeed.o crc64.o bitops.o sentinel.o notify.o setproctitle.o blocked.o hyperloglog.o latency.o sparkline.o redis-check-rdb.o redis-check-aof.o geo.o lazyfree.o module.o evict.o expire.o geohash.o geohash_helper.o childinfo.o defrag.o siphash.o rax.o t_stream.o listpack.o localtime.o lolwut.o lolwut5.o lolwut6.o acl.o tracking.o connection.o tls.o sha256.o timeout.o setcpuaffinity.o monotonic.o mt19937-64.o resp_parser.o call_reply.o script_lua.o script.o REDIS_SERVER_OBJ=adlist.o quicklist.o ae.o anet.o dict.o server.o sds.o zmalloc.o lzf_c.o lzf_d.o pqsort.o zipmap.o sha1.o ziplist.o release.o networking.o util.o object.o db.o replication.o rdb.o t_string.o t_list.o t_set.o t_zset.o t_hash.o config.o aof.o pubsub.o multi.o debug.o sort.o intset.o syncio.o cluster.o crc16.o endianconv.o slowlog.o eval.o bio.o rio.o rand.o memtest.o crcspeed.o crc64.o bitops.o sentinel.o notify.o setproctitle.o blocked.o hyperloglog.o latency.o sparkline.o redis-check-rdb.o redis-check-aof.o geo.o lazyfree.o module.o evict.o expire.o geohash.o geohash_helper.o childinfo.o defrag.o siphash.o rax.o t_stream.o listpack.o localtime.o lolwut.o lolwut5.o lolwut6.o acl.o tracking.o connection.o tls.o sha256.o timeout.o setcpuaffinity.o monotonic.o mt19937-64.o resp_parser.o call_reply.o script_lua.o script.o functions.o function_lua.o
REDIS_CLI_NAME=redis-cli$(PROG_SUFFIX) REDIS_CLI_NAME=redis-cli$(PROG_SUFFIX)
REDIS_CLI_OBJ=anet.o adlist.o dict.o redis-cli.o zmalloc.o release.o ae.o redisassert.o crcspeed.o crc64.o siphash.o crc16.o monotonic.o cli_common.o mt19937-64.o REDIS_CLI_OBJ=anet.o adlist.o dict.o redis-cli.o zmalloc.o release.o ae.o redisassert.o crcspeed.o crc64.o siphash.o crc16.o monotonic.o cli_common.o mt19937-64.o
REDIS_BENCHMARK_NAME=redis-benchmark$(PROG_SUFFIX) REDIS_BENCHMARK_NAME=redis-benchmark$(PROG_SUFFIX)
......
...@@ -30,6 +30,7 @@ ...@@ -30,6 +30,7 @@
#include "server.h" #include "server.h"
#include "bio.h" #include "bio.h"
#include "rio.h" #include "rio.h"
#include "functions.h"
#include <signal.h> #include <signal.h>
#include <fcntl.h> #include <fcntl.h>
...@@ -754,7 +755,8 @@ int loadAppendOnlyFile(char *filename) { ...@@ -754,7 +755,8 @@ int loadAppendOnlyFile(char *filename) {
serverLog(LL_NOTICE,"Reading RDB preamble from AOF file..."); serverLog(LL_NOTICE,"Reading RDB preamble from AOF file...");
if (fseek(fp,0,SEEK_SET) == -1) goto readerr; if (fseek(fp,0,SEEK_SET) == -1) goto readerr;
rioInitWithFile(&rdb,fp); rioInitWithFile(&rdb,fp);
if (rdbLoadRio(&rdb,RDBFLAGS_AOF_PREAMBLE,NULL,server.db) != C_OK) {
if (rdbLoadRio(&rdb,RDBFLAGS_AOF_PREAMBLE,NULL) != C_OK) {
serverLog(LL_WARNING,"Error reading the RDB preamble of the AOF file, AOF loading aborted"); serverLog(LL_WARNING,"Error reading the RDB preamble of the AOF file, AOF loading aborted");
goto readerr; goto readerr;
} else { } else {
......
...@@ -1744,6 +1744,11 @@ int evalGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult * ...@@ -1744,6 +1744,11 @@ int evalGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *
return genericGetKeys(0, 2, 3, 1, argv, argc, result); return genericGetKeys(0, 2, 3, 1, argv, argc, result);
} }
int functionGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) {
UNUSED(cmd);
return genericGetKeys(0, 2, 3, 1, argv, argc, result);
}
int lmpopGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) { int lmpopGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result) {
UNUSED(cmd); UNUSED(cmd);
return genericGetKeys(0, 1, 2, 1, argv, argc, result); return genericGetKeys(0, 1, 2, 1, argv, argc, result);
......
...@@ -257,7 +257,7 @@ void scriptingInit(int setup) { ...@@ -257,7 +257,7 @@ void scriptingInit(int setup) {
/* Lua beginners often don't use "local", this is likely to introduce /* Lua beginners often don't use "local", this is likely to introduce
* subtle bugs in their code. To prevent problems we protect accesses * subtle bugs in their code. To prevent problems we protect accesses
* to global variables. */ * to global variables. */
luaEnableGlobalsProtection(lua); luaEnableGlobalsProtection(lua, 1);
lctx.lua = lua; lctx.lua = lua;
} }
...@@ -443,6 +443,8 @@ void evalGenericCommand(client *c, int evalsha) { ...@@ -443,6 +443,8 @@ void evalGenericCommand(client *c, int evalsha) {
scriptRunCtx rctx; scriptRunCtx rctx;
scriptPrepareForRun(&rctx, lctx.lua_client, c, lctx.lua_cur_script); scriptPrepareForRun(&rctx, lctx.lua_client, c, lctx.lua_cur_script);
rctx.flags |= SCRIPT_EVAL_MODE; /* mark the current run as legacy so we
will get legacy error messages and logs */
if (!lctx.lua_replicate_commands) rctx.flags |= SCRIPT_EVAL_REPLICATION; if (!lctx.lua_replicate_commands) rctx.flags |= SCRIPT_EVAL_REPLICATION;
/* This check is for EVAL_RO, EVALSHA_RO. We want to allow only read only commands */ /* This check is for EVAL_RO, EVALSHA_RO. We want to allow only read only commands */
if ((server.script_caller->cmd->proc == evalRoCommand || if ((server.script_caller->cmd->proc == evalRoCommand ||
...@@ -584,7 +586,7 @@ NULL ...@@ -584,7 +586,7 @@ NULL
addReplyBulkCBuffer(c,sha,40); addReplyBulkCBuffer(c,sha,40);
forceCommandPropagation(c,PROPAGATE_REPL|PROPAGATE_AOF); forceCommandPropagation(c,PROPAGATE_REPL|PROPAGATE_AOF);
} else if (c->argc == 2 && !strcasecmp(c->argv[1]->ptr,"kill")) { } else if (c->argc == 2 && !strcasecmp(c->argv[1]->ptr,"kill")) {
scriptKill(c); scriptKill(c, 1);
} else if (c->argc == 3 && !strcasecmp(c->argv[1]->ptr,"debug")) { } else if (c->argc == 3 && !strcasecmp(c->argv[1]->ptr,"debug")) {
if (clientHasPendingReplies(c)) { if (clientHasPendingReplies(c)) {
addReplyError(c,"SCRIPT DEBUG must be called outside a pipeline"); addReplyError(c,"SCRIPT DEBUG must be called outside a pipeline");
...@@ -610,7 +612,7 @@ NULL ...@@ -610,7 +612,7 @@ NULL
} }
unsigned long evalMemory() { unsigned long evalMemory() {
return lua_gc(lctx.lua, LUA_GCCOUNT, 0) * 1024LL; return luaMemory(lctx.lua);
} }
dict* evalScriptsDict() { dict* evalScriptsDict() {
......
/*
* Copyright (c) 2021, Redis Ltd.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions are met:
*
* * Redistributions of source code must retain the above copyright notice,
* this list of conditions and the following disclaimer.
* * Redistributions in binary form must reproduce the above copyright
* notice, this list of conditions and the following disclaimer in the
* documentation and/or other materials provided with the distribution.
* * Neither the name of Redis nor the names of its contributors may be used
* to endorse or promote products derived from this software without
* specific prior written permission.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
* AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
* ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
* LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
* CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
* SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
* INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
* CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
* POSSIBILITY OF SUCH DAMAGE.
*/
/*
* function_lua.c unit provides the Lua engine functionality.
* Including registering the engine and implementing the engine
* callbacks:
* * Create a function from blob (usually text)
* * Invoke a function
* * Free function memory
* * Get memory usage
*
* Uses script_lua.c to run the Lua code.
*/
#include "functions.h"
#include "script_lua.h"
#include <lua.h>
#include <lauxlib.h>
#include <lualib.h>
#define LUA_ENGINE_NAME "LUA"
#define REGISTRY_ENGINE_CTX_NAME "__ENGINE_CTX__"
#define REGISTRY_ERROR_HANDLER_NAME "__ERROR_HANDLER__"
/* Lua engine ctx */
typedef struct luaEngineCtx {
lua_State *lua;
} luaEngineCtx;
/* Lua function ctx */
typedef struct luaFunctionCtx {
/* Special ID that allows getting the Lua function object from the Lua registry */
int lua_function_ref;
} luaFunctionCtx;
/*
* Compile a given blob and save it on the registry.
* Return a function ctx with Lua ref that allows to later retrieve the
* function from the registry.
*
* Return NULL on compilation error and set the error to the err variable
*/
static void* luaEngineCreate(void *engine_ctx, sds blob, sds *err) {
luaEngineCtx *lua_engine_ctx = engine_ctx;
lua_State *lua = lua_engine_ctx->lua;
if (luaL_loadbuffer(lua, blob, sdslen(blob), "@user_function")) {
*err = sdsempty();
*err = sdscatprintf(*err, "Error compiling function: %s",
lua_tostring(lua, -1));
lua_pop(lua, 1);
return NULL;
}
serverAssert(lua_isfunction(lua, -1));
int lua_function_ref = luaL_ref(lua, LUA_REGISTRYINDEX);
luaFunctionCtx *f_ctx = zmalloc(sizeof(*f_ctx));
*f_ctx = (luaFunctionCtx ) { .lua_function_ref = lua_function_ref, };
return f_ctx;
}
/*
* Invole the give function with the given keys and args
*/
static void luaEngineCall(scriptRunCtx *run_ctx,
void *engine_ctx,
void *compiled_function,
robj **keys,
size_t nkeys,
robj **args,
size_t nargs)
{
luaEngineCtx *lua_engine_ctx = engine_ctx;
lua_State *lua = lua_engine_ctx->lua;
luaFunctionCtx *f_ctx = compiled_function;
/* Push error handler */
lua_pushstring(lua, REGISTRY_ERROR_HANDLER_NAME);
lua_gettable(lua, LUA_REGISTRYINDEX);
lua_rawgeti(lua, LUA_REGISTRYINDEX, f_ctx->lua_function_ref);
serverAssert(lua_isfunction(lua, -1));
luaCallFunction(run_ctx, lua, keys, nkeys, args, nargs, 0);
lua_pop(lua, 1); /* Pop error handler */
}
static size_t luaEngineGetUsedMemoy(void *engine_ctx) {
luaEngineCtx *lua_engine_ctx = engine_ctx;
return luaMemory(lua_engine_ctx->lua);
}
static size_t luaEngineFunctionMemoryOverhead(void *compiled_function) {
return zmalloc_size(compiled_function);
}
static size_t luaEngineMemoryOverhead(void *engine_ctx) {
luaEngineCtx *lua_engine_ctx = engine_ctx;
return zmalloc_size(lua_engine_ctx);
}
static void luaEngineFreeFunction(void *engine_ctx, void *compiled_function) {
luaEngineCtx *lua_engine_ctx = engine_ctx;
lua_State *lua = lua_engine_ctx->lua;
luaFunctionCtx *f_ctx = compiled_function;
lua_unref(lua, f_ctx->lua_function_ref);
zfree(f_ctx);
}
/* Initialize Lua engine, should be called once on start. */
int luaEngineInitEngine() {
luaEngineCtx *lua_engine_ctx = zmalloc(sizeof(*lua_engine_ctx));
lua_engine_ctx->lua = lua_open();
luaRegisterRedisAPI(lua_engine_ctx->lua);
/* Save error handler to registry */
lua_pushstring(lua_engine_ctx->lua, REGISTRY_ERROR_HANDLER_NAME);
char *errh_func = "local dbg = debug\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 i then\n"
" return i.source .. ':' .. i.currentline .. ': ' .. err\n"
" else\n"
" return err\n"
" end\n"
"end\n"
"return error_handler";
luaL_loadbuffer(lua_engine_ctx->lua, errh_func, strlen(errh_func), "@err_handler_def");
lua_pcall(lua_engine_ctx->lua,0,1,0);
lua_settable(lua_engine_ctx->lua, LUA_REGISTRYINDEX);
/* 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);
luaEnableGlobalsProtection(lua_engine_ctx->lua, 0);
engine *lua_engine = zmalloc(sizeof(*lua_engine));
*lua_engine = (engine) {
.engine_ctx = lua_engine_ctx,
.create = luaEngineCreate,
.call = luaEngineCall,
.get_used_memory = luaEngineGetUsedMemoy,
.get_function_memory_overhead = luaEngineFunctionMemoryOverhead,
.get_engine_memory_overhead = luaEngineMemoryOverhead,
.free_function = luaEngineFreeFunction,
};
return functionsRegisterEngine(LUA_ENGINE_NAME, lua_engine);
}
/*
* Copyright (c) 2021, Redis Ltd.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions are met:
*
* * Redistributions of source code must retain the above copyright notice,
* this list of conditions and the following disclaimer.
* * Redistributions in binary form must reproduce the above copyright
* notice, this list of conditions and the following disclaimer in the
* documentation and/or other materials provided with the distribution.
* * Neither the name of Redis nor the names of its contributors may be used
* to endorse or promote products derived from this software without
* specific prior written permission.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
* AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
* ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
* LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
* CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
* SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
* INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
* CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
* POSSIBILITY OF SUCH DAMAGE.
*/
#include "functions.h"
#include "sds.h"
#include "dict.h"
#include "adlist.h"
#include "atomicvar.h"
static size_t engine_cache_memory = 0;
/* Forward declaration */
static void engineFunctionDispose(dict *d, void *obj);
struct functionsCtx {
dict *functions; /* Function name -> Function object that can be used to run the function */
size_t cache_memory /* Overhead memory (structs, dictionaries, ..) used by all the functions */;
};
dictType engineDictType = {
dictSdsCaseHash, /* hash function */
dictSdsDup, /* key dup */
NULL, /* val dup */
dictSdsKeyCaseCompare, /* key compare */
dictSdsDestructor, /* key destructor */
NULL, /* val destructor */
NULL /* allow to expand */
};
dictType functionDictType = {
dictSdsHash, /* hash function */
dictSdsDup, /* key dup */
NULL, /* val dup */
dictSdsKeyCompare, /* key compare */
dictSdsDestructor, /* key destructor */
engineFunctionDispose,/* val destructor */
NULL /* allow to expand */
};
/* Dictionary of engines */
static dict *engines = NULL;
/* Functions Ctx.
* Contains the dictionary that map a function name to
* function object and the cache memory used by all the functions */
static functionsCtx *functions_ctx = NULL;
static size_t functionMallocSize(functionInfo *fi) {
return zmalloc_size(fi) + sdsZmallocSize(fi->name)
+ (fi->desc ? sdsZmallocSize(fi->desc) : 0)
+ sdsZmallocSize(fi->code)
+ fi->ei->engine->get_function_memory_overhead(fi->function);
}
/* Dispose function memory */
static void engineFunctionDispose(dict *d, void *obj) {
UNUSED(d);
functionInfo *fi = obj;
sdsfree(fi->code);
sdsfree(fi->name);
if (fi->desc) {
sdsfree(fi->desc);
}
engine *engine = fi->ei->engine;
engine->free_function(engine->engine_ctx, fi->function);
zfree(fi);
}
/* Free function memory and detele it from the functions dictionary */
static void engineFunctionFree(functionInfo *fi, functionsCtx *functions) {
functions->cache_memory -= functionMallocSize(fi);
dictDelete(functions->functions, fi->name);
}
/* Clear all the functions from the given functions ctx */
void functionsCtxClear(functionsCtx *functions_ctx) {
dictEmpty(functions_ctx->functions, NULL);
functions_ctx->cache_memory = 0;
}
/* Free the given functions ctx */
void functionsCtxFree(functionsCtx *functions_ctx) {
functionsCtxClear(functions_ctx);
dictRelease(functions_ctx->functions);
zfree(functions_ctx);
}
/* Swap the current functions ctx with the given one.
* Free the old functions ctx. */
void functionsCtxSwapWithCurrent(functionsCtx *new_functions_ctx) {
functionsCtxFree(functions_ctx);
functions_ctx = new_functions_ctx;
}
/* return the current functions ctx */
functionsCtx* functionsCtxGetCurrent() {
return functions_ctx;
}
/* Create a new functions ctx */
functionsCtx* functionsCtxCreate() {
functionsCtx *ret = zmalloc(sizeof(functionsCtx));
ret->functions = dictCreate(&functionDictType);
ret->cache_memory = 0;
return ret;
}
/*
* Register a function info to functions dictionary
* 1. Set the function client
* 2. Add function to functions dictionary
* 3. update cache memory
*/
static void engineFunctionRegister(functionInfo *fi, functionsCtx *functions) {
int res = dictAdd(functions->functions, fi->name, fi);
serverAssert(res == DICT_OK);
functions->cache_memory += functionMallocSize(fi);
}
/*
* Creating a function info object and register it.
* Return the created object
*/
static functionInfo* engineFunctionCreate(sds name, void *function, engineInfo *ei,
sds desc, sds code, functionsCtx *functions)
{
functionInfo *fi = zmalloc(sizeof(*fi));
*fi = (functionInfo ) {
.name = sdsdup(name),
.function = function,
.ei = ei,
.code = sdsdup(code),
.desc = desc ? sdsdup(desc) : NULL,
};
engineFunctionRegister(fi, functions);
return fi;
}
/* Register an engine, should be called once by the engine on startup and give the following:
*
* - engine_name - name of the engine to register
* - engine_ctx - the engine ctx that should be used by Redis to interact with the engine */
int functionsRegisterEngine(const char *engine_name, engine *engine) {
sds engine_name_sds = sdsnew(engine_name);
if (dictFetchValue(engines, engine_name_sds)) {
serverLog(LL_WARNING, "Same engine was registered twice");
sdsfree(engine_name_sds);
return C_ERR;
}
client *c = createClient(NULL);
c->flags |= (CLIENT_DENY_BLOCKING | CLIENT_SCRIPT);
engineInfo *ei = zmalloc(sizeof(*ei));
*ei = (engineInfo ) { .name = engine_name_sds, .engine = engine, .c = c,};
dictAdd(engines, engine_name_sds, ei);
engine_cache_memory += zmalloc_size(ei) + sdsZmallocSize(ei->name) +
zmalloc_size(engine) +
engine->get_engine_memory_overhead(engine->engine_ctx);
return C_OK;
}
/*
* FUNCTION STATS
*/
void functionsStatsCommand(client *c) {
if (scriptIsRunning() && scriptIsEval()) {
addReplyErrorObject(c, shared.slowevalerr);
return;
}
addReplyMapLen(c, 2);
addReplyBulkCString(c, "running_script");
if (!scriptIsRunning()) {
addReplyNull(c);
} else {
addReplyMapLen(c, 3);
addReplyBulkCString(c, "name");
addReplyBulkCString(c, scriptCurrFunction());
addReplyBulkCString(c, "command");
client *script_client = scriptGetCaller();
addReplyArrayLen(c, script_client->argc);
for (int i = 0 ; i < script_client->argc ; ++i) {
addReplyBulkCBuffer(c, script_client->argv[i]->ptr, sdslen(script_client->argv[i]->ptr));
}
addReplyBulkCString(c, "duration_ms");
addReplyLongLong(c, scriptRunDuration());
}
addReplyBulkCString(c, "engines");
addReplyArrayLen(c, dictSize(engines));
dictIterator *iter = dictGetIterator(engines);
dictEntry *entry = NULL;
while ((entry = dictNext(iter))) {
engineInfo *ei = dictGetVal(entry);
addReplyBulkCString(c, ei->name);
}
dictReleaseIterator(iter);
}
/*
* FUNCTION LIST
*/
void functionsListCommand(client *c) {
/* general information on all the functions */
addReplyArrayLen(c, dictSize(functions_ctx->functions));
dictIterator *iter = dictGetIterator(functions_ctx->functions);
dictEntry *entry = NULL;
while ((entry = dictNext(iter))) {
functionInfo *fi = dictGetVal(entry);
addReplyMapLen(c, 3);
addReplyBulkCString(c, "name");
addReplyBulkCBuffer(c, fi->name, sdslen(fi->name));
addReplyBulkCString(c, "engine");
addReplyBulkCBuffer(c, fi->ei->name, sdslen(fi->ei->name));
addReplyBulkCString(c, "description");
if (fi->desc) {
addReplyBulkCBuffer(c, fi->desc, sdslen(fi->desc));
} else {
addReplyNull(c);
}
}
dictReleaseIterator(iter);
}
/*
* FUNCTION INFO <FUNCTION NAME> [WITHCODE]
*/
void functionsInfoCommand(client *c) {
if (c->argc > 4) {
addReplyErrorFormat(c,"wrong number of arguments for '%s' command or subcommand", c->cmd->name);
return;
}
/* dedicated information on specific function */
robj *function_name = c->argv[2];
int with_code = 0;
if (c->argc == 4) {
robj *with_code_arg = c->argv[3];
if (!strcasecmp(with_code_arg->ptr, "withcode")) {
with_code = 1;
}
}
functionInfo *fi = dictFetchValue(functions_ctx->functions, function_name->ptr);
if (!fi) {
addReplyError(c, "Function does not exists");
return;
}
addReplyMapLen(c, with_code? 4 : 3);
addReplyBulkCString(c, "name");
addReplyBulkCBuffer(c, fi->name, sdslen(fi->name));
addReplyBulkCString(c, "engine");
addReplyBulkCBuffer(c, fi->ei->name, sdslen(fi->ei->name));
addReplyBulkCString(c, "description");
if (fi->desc) {
addReplyBulkCBuffer(c, fi->desc, sdslen(fi->desc));
} else {
addReplyNull(c);
}
if (with_code) {
addReplyBulkCString(c, "code");
addReplyBulkCBuffer(c, fi->code, sdslen(fi->code));
}
}
/*
* FUNCTION DELETE <FUNCTION NAME>
*/
void functionsDeleteCommand(client *c) {
if (server.masterhost && server.repl_slave_ro && !(c->flags & CLIENT_MASTER)) {
addReplyError(c, "Can not delete a function on a read only replica");
return;
}
robj *function_name = c->argv[2];
functionInfo *fi = dictFetchValue(functions_ctx->functions, function_name->ptr);
if (!fi) {
addReplyError(c, "Function not found");
return;
}
engineFunctionFree(fi, functions_ctx);
forceCommandPropagation(c, PROPAGATE_REPL | PROPAGATE_AOF);
addReply(c, shared.ok);
}
void functionsKillCommand(client *c) {
scriptKill(c, 0);
}
static void fcallCommandGeneric(client *c, int ro) {
robj *function_name = c->argv[1];
functionInfo *fi = dictFetchValue(functions_ctx->functions, function_name->ptr);
if (!fi) {
addReplyError(c, "Function not found");
return;
}
engine *engine = fi->ei->engine;
long long numkeys;
/* Get the number of arguments that are keys */
if (getLongLongFromObject(c->argv[2], &numkeys) != C_OK) {
addReplyError(c, "Bad number of keys provided");
return;
}
if (numkeys > (c->argc - 3)) {
addReplyError(c, "Number of keys can't be greater than number of args");
return;
} else if (numkeys < 0) {
addReplyError(c, "Number of keys can't be negative");
return;
}
scriptRunCtx run_ctx;
scriptPrepareForRun(&run_ctx, fi->ei->c, c, fi->name);
if (ro) {
run_ctx.flags |= SCRIPT_READ_ONLY;
}
engine->call(&run_ctx, engine->engine_ctx, fi->function, c->argv + 3, numkeys,
c->argv + 3 + numkeys, c->argc - 3 - numkeys);
scriptResetRun(&run_ctx);
}
/*
* FCALL <FUNCTION NAME> nkeys <key1 .. keyn> <arg1 .. argn>
*/
void fcallCommand(client *c) {
fcallCommandGeneric(c, 0);
}
/*
* FCALL_RO <FUNCTION NAME> nkeys <key1 .. keyn> <arg1 .. argn>
*/
void fcallCommandReadOnly(client *c) {
fcallCommandGeneric(c, 1);
}
void functionsHelpCommand(client *c) {
const char *help[] = {
"CREATE <ENGINE NAME> <FUNCTION NAME> [REPLACE] [DESC <FUNCTION DESCRIPTION>] <FUNCTION CODE>",
" Create a new function with the given function name and code.",
"DELETE <FUNCTION NAME>",
" Delete the given function.",
"INFO <FUNCTION NAME> [WITHCODE]",
" For each function, print the following information about the function:",
" * Function name",
" * The engine used to run the function",
" * Function description",
" * Function code (only if WITHCODE is given)",
"LIST",
" Return general information on all the functions:",
" * Function name",
" * The engine used to run the function",
" * Function description",
"STATS",
" Return information about the current function running:",
" * Function name",
" * Command used to run the function",
" * Duration in MS that the function is running",
" If not function is running, return nil",
" In addition, returns a list of available engines.",
"KILL",
" Kill the current running function.",
NULL };
addReplyHelp(c, help);
}
/* Compile and save the given function, return C_OK on success and C_ERR on failure.
* In case on failure the err out param is set with relevant error message */
int functionsCreateWithFunctionCtx(sds function_name,sds engine_name, sds desc, sds code,
int replace, sds* err, functionsCtx *functions) {
engineInfo *ei = dictFetchValue(engines, engine_name);
if (!ei) {
*err = sdsnew("Engine not found");
return C_ERR;
}
engine *engine = ei->engine;
functionInfo *fi = dictFetchValue(functions->functions, function_name);
if (fi && !replace) {
*err = sdsnew("Function already exists");
return C_ERR;
}
void *function = engine->create(engine->engine_ctx, code, err);
if (*err) {
return C_ERR;
}
if (fi) {
/* free the already existing function as we are going to replace it */
engineFunctionFree(fi, functions);
}
engineFunctionCreate(function_name, function, ei, desc, code, functions);
return C_OK;
}
/*
* FUNCTION CREATE <ENGINE NAME> <FUNCTION NAME>
* [REPLACE] [DESC <FUNCTION DESCRIPTION>] <FUNCTION CODE>
*
* ENGINE NAME - name of the engine to use the run the function
* FUNCTION NAME - name to use to invoke the function
* REPLACE - optional, replace existing function
* DESCRIPTION - optional, function description
* FUNCTION CODE - function code to pass to the engine
*/
void functionsCreateCommand(client *c) {
if (server.masterhost && server.repl_slave_ro && !(c->flags & CLIENT_MASTER)) {
addReplyError(c, "Can not create a function on a read only replica");
return;
}
robj *engine_name = c->argv[2];
robj *function_name = c->argv[3];
int replace = 0;
int argc_pos = 4;
sds desc = NULL;
while (argc_pos < c->argc - 1) {
robj *next_arg = c->argv[argc_pos++];
if (!strcasecmp(next_arg->ptr, "replace")) {
replace = 1;
continue;
}
if (!strcasecmp(next_arg->ptr, "description")) {
if (argc_pos >= c->argc) {
addReplyError(c, "Bad function description");
return;
}
desc = c->argv[argc_pos++]->ptr;
continue;
}
}
if (argc_pos >= c->argc) {
addReplyError(c, "Function code is missing");
return;
}
robj *code = c->argv[argc_pos];
sds err = NULL;
if (functionsCreateWithFunctionCtx(function_name->ptr, engine_name->ptr,
desc, code->ptr, replace, &err, functions_ctx) != C_OK)
{
addReplyErrorSds(c, err);
return;
}
forceCommandPropagation(c, PROPAGATE_REPL | PROPAGATE_AOF);
addReply(c, shared.ok);
}
/* Return memory usage of all the engines combine */
unsigned long functionsMemory() {
dictIterator *iter = dictGetIterator(engines);
dictEntry *entry = NULL;
size_t engines_nemory = 0;
while ((entry = dictNext(iter))) {
engineInfo *ei = dictGetVal(entry);
engine *engine = ei->engine;
engines_nemory += engine->get_used_memory(engine->engine_ctx);
}
dictReleaseIterator(iter);
return engines_nemory;
}
/* Return memory overhead of all the engines combine */
unsigned long functionsMemoryOverhead() {
size_t memory_overhead = dictSize(engines) * sizeof(dictEntry) +
dictSlots(engines) * sizeof(dictEntry*);
memory_overhead += dictSize(functions_ctx->functions) * sizeof(dictEntry) +
dictSlots(functions_ctx->functions) * sizeof(dictEntry*) + sizeof(functionsCtx);
memory_overhead += functions_ctx->cache_memory;
memory_overhead += engine_cache_memory;
return memory_overhead;
}
/* Returns the number of functions */
unsigned long functionsNum() {
return dictSize(functions_ctx->functions);
}
dict* functionsGet() {
return functions_ctx->functions;
}
/* Initialize engine data structures.
* Should be called once on server initialization */
int functionsInit() {
engines = dictCreate(&engineDictType);
functions_ctx = functionsCtxCreate();
if (luaEngineInitEngine() != C_OK) {
return C_ERR;
}
return C_OK;
}
/*
* Copyright (c) 2021, Redis Ltd.
* All rights reserved.
*
* Redistribution and use in source and binary forms, with or without
* modification, are permitted provided that the following conditions are met:
*
* * Redistributions of source code must retain the above copyright notice,
* this list of conditions and the following disclaimer.
* * Redistributions in binary form must reproduce the above copyright
* notice, this list of conditions and the following disclaimer in the
* documentation and/or other materials provided with the distribution.
* * Neither the name of Redis nor the names of its contributors may be used
* to endorse or promote products derived from this software without
* specific prior written permission.
*
* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
* AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
* ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
* LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
* CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
* SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
* INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
* CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
* ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
* POSSIBILITY OF SUCH DAMAGE.
*/
#ifndef __FUNCTIONS_H_
#define __FUNCTIONS_H_
/*
* functions.c unit provides the Redis Functions API:
* * FUNCTION CREATE
* * FUNCTION CALL
* * FUNCTION DELETE
* * FUNCTION KILL
* * FUNCTION INFO
*
* Also contains implementation for:
* * Save/Load function from rdb
* * Register engines
*/
#include "server.h"
#include "script.h"
#include "redismodule.h"
typedef struct engine {
/* engine specific context */
void *engine_ctx;
/* Create function callback, get the engine_ctx, and function code.
* returns NULL on error and set sds to be the error message */
void* (*create)(void *engine_ctx, sds code, sds *err);
/* Invoking a function, r_ctx is an opaque object (from engine POV).
* The r_ctx should be used by the engine to interaction with Redis,
* such interaction could be running commands, set resp, or set
* replication mode
*/
void (*call)(scriptRunCtx *r_ctx, void *engine_ctx, void *compiled_function,
robj **keys, size_t nkeys, robj **args, size_t nargs);
/* get current used memory by the engine */
size_t (*get_used_memory)(void *engine_ctx);
/* Return memory overhead for a given function,
* such memory is not counted as engine memory but as general
* structs memory that hold different information */
size_t (*get_function_memory_overhead)(void *compiled_function);
/* Return memory overhead for engine (struct size holding the engine)*/
size_t (*get_engine_memory_overhead)(void *engine_ctx);
/* free the given function */
void (*free_function)(void *engine_ctx, void *compiled_function);
} engine;
/* Hold information about an engine.
* Used on rdb.c so it must be declared here. */
typedef struct engineInfo {
sds name; /* Name of the engine */
engine *engine; /* engine callbacks that allows to interact with the engine */
client *c; /* Client that is used to run commands */
} engineInfo;
/* Hold information about the specific function.
* Used on rdb.c so it must be declared here. */
typedef struct functionInfo {
sds name; /* Function name */
void *function; /* Opaque object that set by the function's engine and allow it
to run the function, usually it's the function compiled code. */
engineInfo *ei; /* Pointer to the function engine */
sds code; /* Function code */
sds desc; /* Function description */
} functionInfo;
int functionsRegisterEngine(const char *engine_name, engine *engine_ctx);
int functionsCreateWithFunctionCtx(sds function_name, sds engine_name, sds desc, sds code,
int replace, sds* err, functionsCtx *functions);
void functionsCreateCommand(client *c);
void fcallCommand(client *c);
void fcallCommandReadOnly(client *c);
void functionsDeleteCommand(client *c);
void functionsKillCommand(client *c);
void functionsStatsCommand(client *c);
void functionsInfoCommand(client *c);
void functionsListCommand(client *c);
void functionsHelpCommand(client *c);
unsigned long functionsMemory();
unsigned long functionsMemoryOverhead();
int functionsLoad(rio *rdb, int ver);
unsigned long functionsNum();
dict* functionsGet();
functionsCtx* functionsCtxGetCurrent();
functionsCtx* functionsCtxCreate();
void functionsCtxFree(functionsCtx *functions_ctx);
void functionsCtxClear(functionsCtx *functions_ctx);
void functionsCtxSwapWithCurrent(functionsCtx *functions_ctx);
int luaEngineInitEngine();
int functionsInit();
#endif /* __FUNCTIONS_H_ */
...@@ -29,6 +29,7 @@ ...@@ -29,6 +29,7 @@
*/ */
#include "server.h" #include "server.h"
#include "functions.h"
#include <math.h> #include <math.h>
#include <ctype.h> #include <ctype.h>
...@@ -1212,6 +1213,8 @@ struct redisMemOverhead *getMemoryOverheadData(void) { ...@@ -1212,6 +1213,8 @@ struct redisMemOverhead *getMemoryOverheadData(void) {
} }
mh->lua_caches = mem; mh->lua_caches = mem;
mem_total+=mem; mem_total+=mem;
mh->functions_caches = functionsMemoryOverhead();
mem_total+=mh->functions_caches;
for (j = 0; j < server.dbnum; j++) { for (j = 0; j < server.dbnum; j++) {
redisDb *db = server.db+j; redisDb *db = server.db+j;
...@@ -1527,7 +1530,7 @@ NULL ...@@ -1527,7 +1530,7 @@ NULL
} else if (!strcasecmp(c->argv[1]->ptr,"stats") && c->argc == 2) { } else if (!strcasecmp(c->argv[1]->ptr,"stats") && c->argc == 2) {
struct redisMemOverhead *mh = getMemoryOverheadData(); struct redisMemOverhead *mh = getMemoryOverheadData();
addReplyMapLen(c,25+mh->num_dbs); addReplyMapLen(c,26+mh->num_dbs);
addReplyBulkCString(c,"peak.allocated"); addReplyBulkCString(c,"peak.allocated");
addReplyLongLong(c,mh->peak_allocated); addReplyLongLong(c,mh->peak_allocated);
...@@ -1553,6 +1556,9 @@ NULL ...@@ -1553,6 +1556,9 @@ NULL
addReplyBulkCString(c,"lua.caches"); addReplyBulkCString(c,"lua.caches");
addReplyLongLong(c,mh->lua_caches); addReplyLongLong(c,mh->lua_caches);
addReplyBulkCString(c,"functions.caches");
addReplyLongLong(c,mh->functions_caches);
for (size_t j = 0; j < mh->num_dbs; j++) { for (size_t j = 0; j < mh->num_dbs; j++) {
char dbname[32]; char dbname[32];
snprintf(dbname,sizeof(dbname),"db.%zd",mh->db[j].dbid); snprintf(dbname,sizeof(dbname),"db.%zd",mh->db[j].dbid);
......
...@@ -32,6 +32,7 @@ ...@@ -32,6 +32,7 @@
#include "zipmap.h" #include "zipmap.h"
#include "endianconv.h" #include "endianconv.h"
#include "stream.h" #include "stream.h"
#include "functions.h"
#include <math.h> #include <math.h>
#include <fcntl.h> #include <fcntl.h>
...@@ -1239,6 +1240,25 @@ int rdbSaveRio(rio *rdb, int *error, int rdbflags, rdbSaveInfo *rsi) { ...@@ -1239,6 +1240,25 @@ int rdbSaveRio(rio *rdb, int *error, int rdbflags, rdbSaveInfo *rsi) {
if (rdbSaveInfoAuxFields(rdb,rdbflags,rsi) == -1) goto werr; if (rdbSaveInfoAuxFields(rdb,rdbflags,rsi) == -1) goto werr;
if (rdbSaveModulesAux(rdb, REDISMODULE_AUX_BEFORE_RDB) == -1) goto werr; if (rdbSaveModulesAux(rdb, REDISMODULE_AUX_BEFORE_RDB) == -1) goto werr;
/* save functions */
dict *functions = functionsGet();
dictIterator *iter = dictGetIterator(functions);
dictEntry *entry = NULL;
while ((entry = dictNext(iter))) {
rdbSaveType(rdb, RDB_OPCODE_FUNCTION);
functionInfo* fi = dictGetVal(entry);
if (rdbSaveRawString(rdb, (unsigned char *) fi->name, sdslen(fi->name)) == -1) goto werr;
if (rdbSaveRawString(rdb, (unsigned char *) fi->ei->name, sdslen(fi->ei->name)) == -1) goto werr;
if (fi->desc) {
if (rdbSaveLen(rdb, 1) == -1) goto werr; /* desc exists */
if (rdbSaveRawString(rdb, (unsigned char *) fi->desc, sdslen(fi->desc)) == -1) goto werr;
} else {
if (rdbSaveLen(rdb, 0) == -1) goto werr; /* desc not exists */
}
if (rdbSaveRawString(rdb, (unsigned char *) fi->code, sdslen(fi->code)) == -1) goto werr;
}
dictReleaseIterator(iter);
for (j = 0; j < server.dbnum; j++) { for (j = 0; j < server.dbnum; j++) {
redisDb *db = server.db+j; redisDb *db = server.db+j;
dict *d = db->dict; dict *d = db->dict;
...@@ -2687,12 +2707,80 @@ void rdbLoadProgressCallback(rio *r, const void *buf, size_t len) { ...@@ -2687,12 +2707,80 @@ void rdbLoadProgressCallback(rio *r, const void *buf, size_t len) {
} }
} }
static int rdbFunctionLoad(rio *rdb, int ver, functionsCtx* functions_ctx) {
UNUSED(ver);
sds name = NULL;
sds engine_name = NULL;
sds desc = NULL;
sds blob = NULL;
sds err = NULL;
uint64_t has_desc;
int res = C_ERR;
if (!(name = rdbGenericLoadStringObject(rdb, RDB_LOAD_SDS, NULL))) {
serverLog(LL_WARNING, "Failed loading function name");
goto error;
}
if (!(engine_name = rdbGenericLoadStringObject(rdb, RDB_LOAD_SDS, NULL))) {
serverLog(LL_WARNING, "Failed loading engine name");
goto error;
}
if ((has_desc = rdbLoadLen(rdb, NULL)) == RDB_LENERR) {
serverLog(LL_WARNING, "Failed loading function desc indicator");
goto error;
}
if (has_desc && !(desc = rdbGenericLoadStringObject(rdb, RDB_LOAD_SDS, NULL))) {
serverLog(LL_WARNING, "Failed loading function desc");
goto error;
}
if (!(blob = rdbGenericLoadStringObject(rdb, RDB_LOAD_SDS, NULL))) {
serverLog(LL_WARNING, "Failed loading function blob");
goto error;
}
if (functionsCreateWithFunctionCtx(name, engine_name, desc, blob, 0, &err, functions_ctx) != C_OK) {
serverLog(LL_WARNING, "Failed compiling and saving the function %s", err);
goto error;
}
res = C_OK;
error:
if (name) sdsfree(name);
if (engine_name) sdsfree(engine_name);
if (desc) sdsfree(desc);
if (blob) sdsfree(blob);
if (err) sdsfree(err);
return res;
}
/* Load an RDB file from the rio stream 'rdb'. On success C_OK is returned, /* Load an RDB file from the rio stream 'rdb'. On success C_OK is returned,
* otherwise C_ERR is returned and 'errno' is set accordingly. */ * otherwise C_ERR is returned and 'errno' is set accordingly. */
int rdbLoadRio(rio *rdb, int rdbflags, rdbSaveInfo *rsi, redisDb *dbarray) { int rdbLoadRio(rio *rdb, int rdbflags, rdbSaveInfo *rsi) {
functionsCtx* functions_ctx = functionsCtxGetCurrent();
functionsCtxClear(functions_ctx);
rdbLoadingCtx loading_ctx = { .dbarray = server.db, .functions_ctx = functions_ctx };
int retval = rdbLoadRioWithLoadingCtx(rdb,rdbflags,rsi,&loading_ctx);
if (retval != C_OK) {
/* Loading failed, clear the function ctx */
functionsCtxClear(functions_ctx);
}
return retval;
}
/* Load an RDB file from the rio stream 'rdb'. On success C_OK is returned,
* otherwise C_ERR is returned and 'errno' is set accordingly.
* The rdb_loading_ctx argument holds objects to which the rdb will be loaded to,
* currently it only allow to set db object and functionsCtx to which the data
* will be loaded (in the future it might contains more such objects). */
int rdbLoadRioWithLoadingCtx(rio *rdb, int rdbflags, rdbSaveInfo *rsi, rdbLoadingCtx *rdb_loading_ctx) {
uint64_t dbid = 0; uint64_t dbid = 0;
int type, rdbver; int type, rdbver;
redisDb *db = dbarray+0; redisDb *db = rdb_loading_ctx->dbarray+0;
char buf[1024]; char buf[1024];
int error; int error;
long long empty_keys_skipped = 0; long long empty_keys_skipped = 0;
...@@ -2764,7 +2852,7 @@ int rdbLoadRio(rio *rdb, int rdbflags, rdbSaveInfo *rsi, redisDb *dbarray) { ...@@ -2764,7 +2852,7 @@ int rdbLoadRio(rio *rdb, int rdbflags, rdbSaveInfo *rsi, redisDb *dbarray) {
"databases. Exiting\n", server.dbnum); "databases. Exiting\n", server.dbnum);
exit(1); exit(1);
} }
db = dbarray+dbid; db = rdb_loading_ctx->dbarray+dbid;
continue; /* Read next opcode. */ continue; /* Read next opcode. */
} else if (type == RDB_OPCODE_RESIZEDB) { } else if (type == RDB_OPCODE_RESIZEDB) {
/* RESIZEDB: Hint about the size of the keys in the currently /* RESIZEDB: Hint about the size of the keys in the currently
...@@ -2895,6 +2983,12 @@ int rdbLoadRio(rio *rdb, int rdbflags, rdbSaveInfo *rsi, redisDb *dbarray) { ...@@ -2895,6 +2983,12 @@ int rdbLoadRio(rio *rdb, int rdbflags, rdbSaveInfo *rsi, redisDb *dbarray) {
decrRefCount(aux); decrRefCount(aux);
continue; /* Read next opcode. */ continue; /* Read next opcode. */
} }
} else if (type == RDB_OPCODE_FUNCTION) {
if (rdbFunctionLoad(rdb, rdbver, rdb_loading_ctx->functions_ctx) != C_OK) {
serverLog(LL_WARNING,"Failed loading function");
goto eoferr;
}
continue;
} }
/* Read key */ /* Read key */
...@@ -3044,7 +3138,9 @@ int rdbLoad(char *filename, rdbSaveInfo *rsi, int rdbflags) { ...@@ -3044,7 +3138,9 @@ int rdbLoad(char *filename, rdbSaveInfo *rsi, int rdbflags) {
if ((fp = fopen(filename,"r")) == NULL) return C_ERR; if ((fp = fopen(filename,"r")) == NULL) return C_ERR;
startLoadingFile(fp, filename,rdbflags); startLoadingFile(fp, filename,rdbflags);
rioInitWithFile(&rdb,fp); rioInitWithFile(&rdb,fp);
retval = rdbLoadRio(&rdb,rdbflags,rsi,server.db);
retval = rdbLoadRio(&rdb,rdbflags,rsi);
fclose(fp); fclose(fp);
stopLoading(retval==C_OK); stopLoading(retval==C_OK);
return retval; return retval;
......
...@@ -100,6 +100,7 @@ ...@@ -100,6 +100,7 @@
#define rdbIsObjectType(t) ((t >= 0 && t <= 7) || (t >= 9 && t <= 18)) #define rdbIsObjectType(t) ((t >= 0 && t <= 7) || (t >= 9 && t <= 18))
/* Special RDB opcodes (saved/loaded with rdbSaveType/rdbLoadType). */ /* Special RDB opcodes (saved/loaded with rdbSaveType/rdbLoadType). */
#define RDB_OPCODE_FUNCTION 246 /* engine data */
#define RDB_OPCODE_MODULE_AUX 247 /* Module auxiliary data. */ #define RDB_OPCODE_MODULE_AUX 247 /* Module auxiliary data. */
#define RDB_OPCODE_IDLE 248 /* LRU idle time. */ #define RDB_OPCODE_IDLE 248 /* LRU idle time. */
#define RDB_OPCODE_FREQ 249 /* LFU frequency. */ #define RDB_OPCODE_FREQ 249 /* LFU frequency. */
...@@ -166,7 +167,8 @@ int rdbSaveBinaryDoubleValue(rio *rdb, double val); ...@@ -166,7 +167,8 @@ int rdbSaveBinaryDoubleValue(rio *rdb, double val);
int rdbLoadBinaryDoubleValue(rio *rdb, double *val); int rdbLoadBinaryDoubleValue(rio *rdb, double *val);
int rdbSaveBinaryFloatValue(rio *rdb, float val); int rdbSaveBinaryFloatValue(rio *rdb, float val);
int rdbLoadBinaryFloatValue(rio *rdb, float *val); int rdbLoadBinaryFloatValue(rio *rdb, float *val);
int rdbLoadRio(rio *rdb, int rdbflags, rdbSaveInfo *rsi, redisDb *db); int rdbLoadRio(rio *rdb, int rdbflags, rdbSaveInfo *rsi);
int rdbLoadRioWithLoadingCtx(rio *rdb, int rdbflags, rdbSaveInfo *rsi, rdbLoadingCtx *rdb_loading_ctx);
int rdbSaveRio(rio *rdb, int *error, int rdbflags, rdbSaveInfo *rsi); int rdbSaveRio(rio *rdb, int *error, int rdbflags, rdbSaveInfo *rsi);
rdbSaveInfo *rdbPopulateSaveInfo(rdbSaveInfo *rsi); rdbSaveInfo *rdbPopulateSaveInfo(rdbSaveInfo *rsi);
......
...@@ -32,6 +32,7 @@ ...@@ -32,6 +32,7 @@
#include "server.h" #include "server.h"
#include "cluster.h" #include "cluster.h"
#include "bio.h" #include "bio.h"
#include "functions.h"
#include <memory.h> #include <memory.h>
#include <sys/time.h> #include <sys/time.h>
...@@ -1738,6 +1739,7 @@ void readSyncBulkPayload(connection *conn) { ...@@ -1738,6 +1739,7 @@ void readSyncBulkPayload(connection *conn) {
ssize_t nread, readlen, nwritten; ssize_t nread, readlen, nwritten;
int use_diskless_load = useDisklessLoad(); int use_diskless_load = useDisklessLoad();
redisDb *diskless_load_tempDb = NULL; redisDb *diskless_load_tempDb = NULL;
functionsCtx* temp_functions_ctx = NULL;
int empty_db_flags = server.repl_slave_lazy_flush ? EMPTYDB_ASYNC : int empty_db_flags = server.repl_slave_lazy_flush ? EMPTYDB_ASYNC :
EMPTYDB_NO_FLAGS; EMPTYDB_NO_FLAGS;
off_t left; off_t left;
...@@ -1913,6 +1915,7 @@ void readSyncBulkPayload(connection *conn) { ...@@ -1913,6 +1915,7 @@ void readSyncBulkPayload(connection *conn) {
if (use_diskless_load && server.repl_diskless_load == REPL_DISKLESS_LOAD_SWAPDB) { if (use_diskless_load && server.repl_diskless_load == REPL_DISKLESS_LOAD_SWAPDB) {
/* Initialize empty tempDb dictionaries. */ /* Initialize empty tempDb dictionaries. */
diskless_load_tempDb = disklessLoadInitTempDb(); diskless_load_tempDb = disklessLoadInitTempDb();
temp_functions_ctx = functionsCtxCreate();
moduleFireServerEvent(REDISMODULE_EVENT_REPL_ASYNC_LOAD, moduleFireServerEvent(REDISMODULE_EVENT_REPL_ASYNC_LOAD,
REDISMODULE_SUBEVENT_REPL_ASYNC_LOAD_STARTED, REDISMODULE_SUBEVENT_REPL_ASYNC_LOAD_STARTED,
...@@ -1935,6 +1938,7 @@ void readSyncBulkPayload(connection *conn) { ...@@ -1935,6 +1938,7 @@ void readSyncBulkPayload(connection *conn) {
if (use_diskless_load) { if (use_diskless_load) {
rio rdb; rio rdb;
redisDb *dbarray; redisDb *dbarray;
functionsCtx* functions_ctx;
int asyncLoading = 0; int asyncLoading = 0;
if (server.repl_diskless_load == REPL_DISKLESS_LOAD_SWAPDB) { if (server.repl_diskless_load == REPL_DISKLESS_LOAD_SWAPDB) {
...@@ -1947,8 +1951,11 @@ void readSyncBulkPayload(connection *conn) { ...@@ -1947,8 +1951,11 @@ void readSyncBulkPayload(connection *conn) {
asyncLoading = 1; asyncLoading = 1;
} }
dbarray = diskless_load_tempDb; dbarray = diskless_load_tempDb;
functions_ctx = temp_functions_ctx;
} else { } else {
dbarray = server.db; dbarray = server.db;
functions_ctx = functionsCtxGetCurrent();
functionsCtxClear(functions_ctx);
} }
rioInitWithConn(&rdb,conn,server.repl_transfer_size); rioInitWithConn(&rdb,conn,server.repl_transfer_size);
...@@ -1960,7 +1967,8 @@ void readSyncBulkPayload(connection *conn) { ...@@ -1960,7 +1967,8 @@ void readSyncBulkPayload(connection *conn) {
startLoading(server.repl_transfer_size, RDBFLAGS_REPLICATION, asyncLoading); startLoading(server.repl_transfer_size, RDBFLAGS_REPLICATION, asyncLoading);
int loadingFailed = 0; int loadingFailed = 0;
if (rdbLoadRio(&rdb,RDBFLAGS_REPLICATION,&rsi,dbarray) != C_OK) { rdbLoadingCtx loadingCtx = { .dbarray = dbarray, .functions_ctx = functions_ctx };
if (rdbLoadRioWithLoadingCtx(&rdb,RDBFLAGS_REPLICATION,&rsi,&loadingCtx) != C_OK) {
/* RDB loading failed. */ /* RDB loading failed. */
serverLog(LL_WARNING, serverLog(LL_WARNING,
"Failed trying to load the MASTER synchronization DB " "Failed trying to load the MASTER synchronization DB "
...@@ -1988,6 +1996,7 @@ void readSyncBulkPayload(connection *conn) { ...@@ -1988,6 +1996,7 @@ void readSyncBulkPayload(connection *conn) {
NULL); NULL);
disklessLoadDiscardTempDb(diskless_load_tempDb); disklessLoadDiscardTempDb(diskless_load_tempDb);
functionsCtxFree(temp_functions_ctx);
serverLog(LL_NOTICE, "MASTER <-> REPLICA sync: Discarding temporary DB in background"); serverLog(LL_NOTICE, "MASTER <-> REPLICA sync: Discarding temporary DB in background");
} else { } else {
/* Remove the half-loaded data in case we started with an empty replica. */ /* Remove the half-loaded data in case we started with an empty replica. */
...@@ -2010,6 +2019,9 @@ void readSyncBulkPayload(connection *conn) { ...@@ -2010,6 +2019,9 @@ void readSyncBulkPayload(connection *conn) {
serverLog(LL_NOTICE, "MASTER <-> REPLICA sync: Swapping active DB with loaded DB"); serverLog(LL_NOTICE, "MASTER <-> REPLICA sync: Swapping active DB with loaded DB");
swapMainDbWithTempDb(diskless_load_tempDb); swapMainDbWithTempDb(diskless_load_tempDb);
/* swap existing functions ctx with the temporary one */
functionsCtxSwapWithCurrent(temp_functions_ctx);
moduleFireServerEvent(REDISMODULE_EVENT_REPL_ASYNC_LOAD, moduleFireServerEvent(REDISMODULE_EVENT_REPL_ASYNC_LOAD,
REDISMODULE_SUBEVENT_REPL_ASYNC_LOAD_COMPLETED, REDISMODULE_SUBEVENT_REPL_ASYNC_LOAD_COMPLETED,
NULL); NULL);
......
...@@ -60,6 +60,11 @@ client* scriptGetClient() { ...@@ -60,6 +60,11 @@ client* scriptGetClient() {
return curr_run_ctx->c; return curr_run_ctx->c;
} }
client* scriptGetCaller() {
serverAssert(scriptIsRunning());
return curr_run_ctx->original_client;
}
/* interrupt function for scripts, should be call /* interrupt function for scripts, should be call
* from time to time to reply some special command (like ping) * from time to time to reply some special command (like ping)
* and also check if the run should be terminated. */ * and also check if the run should be terminated. */
...@@ -78,8 +83,8 @@ int scriptInterrupt(scriptRunCtx *run_ctx) { ...@@ -78,8 +83,8 @@ int scriptInterrupt(scriptRunCtx *run_ctx) {
serverLog(LL_WARNING, serverLog(LL_WARNING,
"Slow script detected: still in execution after %lld milliseconds. " "Slow script detected: still in execution after %lld milliseconds. "
"You can try killing the script using the SCRIPT KILL command.", "You can try killing the script using the %s command.",
elapsed); elapsed, (run_ctx->flags & SCRIPT_EVAL_MODE) ? "SCRIPT KILL" : "FUNCTION KILL");
enterScriptTimedoutMode(run_ctx); enterScriptTimedoutMode(run_ctx);
/* Once the script timeouts we reenter the event loop to permit others /* Once the script timeouts we reenter the event loop to permit others
...@@ -159,8 +164,18 @@ int scriptIsRunning() { ...@@ -159,8 +164,18 @@ int scriptIsRunning() {
return curr_run_ctx != NULL; return curr_run_ctx != NULL;
} }
const char* scriptCurrFunction() {
serverAssert(scriptIsRunning());
return curr_run_ctx->funcname;
}
int scriptIsEval() {
serverAssert(scriptIsRunning());
return curr_run_ctx->flags & SCRIPT_EVAL_MODE;
}
/* Kill the current running script */ /* Kill the current running script */
void scriptKill(client *c) { void scriptKill(client *c, int is_eval) {
if (!curr_run_ctx) { if (!curr_run_ctx) {
addReplyError(c, "-NOTBUSY No scripts in execution right now."); addReplyError(c, "-NOTBUSY No scripts in execution right now.");
return; return;
...@@ -177,6 +192,16 @@ void scriptKill(client *c) { ...@@ -177,6 +192,16 @@ void scriptKill(client *c) {
"using the SHUTDOWN NOSAVE command."); "using the SHUTDOWN NOSAVE command.");
return; return;
} }
if (is_eval && !(curr_run_ctx->flags & SCRIPT_EVAL_MODE)) {
/* Kill a function with 'SCRIPT KILL' is not allow */
addReplyErrorObject(c, shared.slowscripterr);
return;
}
if (!is_eval && (curr_run_ctx->flags & SCRIPT_EVAL_MODE)) {
/* Kill an eval with 'FUNCTION KILL' is not allow */
addReplyErrorObject(c, shared.slowevalerr);
return;
}
curr_run_ctx->flags |= SCRIPT_KILLED; curr_run_ctx->flags |= SCRIPT_KILLED;
addReply(c, shared.ok); addReply(c, shared.ok);
} }
...@@ -430,3 +455,10 @@ mstime_t scriptTimeSnapshot() { ...@@ -430,3 +455,10 @@ mstime_t scriptTimeSnapshot() {
serverAssert(!curr_run_ctx); serverAssert(!curr_run_ctx);
return curr_run_ctx->snapshot_time; return curr_run_ctx->snapshot_time;
} }
long long scriptRunDuration() {
serverAssert(scriptIsRunning());
return elapsedMs(curr_run_ctx->start_time);
}
...@@ -69,6 +69,7 @@ ...@@ -69,6 +69,7 @@
#define SCRIPT_READ_ONLY (1ULL<<5) /* indicate that the current script should only perform read commands */ #define SCRIPT_READ_ONLY (1ULL<<5) /* indicate that the current script should only perform read commands */
#define SCRIPT_EVAL_REPLICATION (1ULL<<6) /* mode for eval, indicate that we replicate the #define SCRIPT_EVAL_REPLICATION (1ULL<<6) /* mode for eval, indicate that we replicate the
script invocation and not the effects */ script invocation and not the effects */
#define SCRIPT_EVAL_MODE (1ULL<<7) /* Indicate that the current script called from legacy Lua */
typedef struct scriptRunCtx scriptRunCtx; typedef struct scriptRunCtx scriptRunCtx;
struct scriptRunCtx { struct scriptRunCtx {
...@@ -87,10 +88,14 @@ int scriptSetResp(scriptRunCtx *r_ctx, int resp); ...@@ -87,10 +88,14 @@ int scriptSetResp(scriptRunCtx *r_ctx, int resp);
int scriptSetRepl(scriptRunCtx *r_ctx, int repl); int scriptSetRepl(scriptRunCtx *r_ctx, int repl);
void scriptCall(scriptRunCtx *r_ctx, robj **argv, int argc, sds *err); void scriptCall(scriptRunCtx *r_ctx, robj **argv, int argc, sds *err);
int scriptInterrupt(scriptRunCtx *r_ctx); int scriptInterrupt(scriptRunCtx *r_ctx);
void scriptKill(client *c); void scriptKill(client *c, int is_eval);
int scriptIsRunning(); int scriptIsRunning();
const char* scriptCurrFunction();
int scriptIsEval();
int scriptIsTimedout(); int scriptIsTimedout();
client* scriptGetClient(); client* scriptGetClient();
client* scriptGetCaller();
mstime_t scriptTimeSnapshot(); mstime_t scriptTimeSnapshot();
long long scriptRunDuration();
#endif /* __SCRIPT_H_ */ #endif /* __SCRIPT_H_ */
...@@ -867,6 +867,8 @@ cleanup: ...@@ -867,6 +867,8 @@ cleanup:
} }
c->user = NULL; c->user = NULL;
c->argv = NULL;
c->argc = 0;
if (raise_error) { if (raise_error) {
/* If we are here we should have an error in the stack, in the /* If we are here we should have an error in the stack, in the
...@@ -1067,8 +1069,12 @@ static void luaRemoveUnsupportedFunctions(lua_State *lua) { ...@@ -1067,8 +1069,12 @@ static void luaRemoveUnsupportedFunctions(lua_State *lua) {
* the creation of globals accidentally. * the creation of globals accidentally.
* *
* It should be the last to be called in the scripting engine initialization * It should be the last to be called in the scripting engine initialization
* sequence, because it may interact with creation of globals. */ * sequence, because it may interact with creation of globals.
void luaEnableGlobalsProtection(lua_State *lua) { *
* On Legacy Lua (eval) we need to check 'w ~= \"main\"' otherwise we will not be able
* to create the global 'function <sha> ()' variable. On Lua engine we do not use this trick
* so its not needed. */
void luaEnableGlobalsProtection(lua_State *lua, int is_eval) {
char *s[32]; char *s[32];
sds code = sdsempty(); sds code = sdsempty();
int j = 0; int j = 0;
...@@ -1081,7 +1087,7 @@ void luaEnableGlobalsProtection(lua_State *lua) { ...@@ -1081,7 +1087,7 @@ void luaEnableGlobalsProtection(lua_State *lua) {
s[j++]="mt.__newindex = function (t, n, v)\n"; s[j++]="mt.__newindex = function (t, n, v)\n";
s[j++]=" if dbg.getinfo(2) then\n"; s[j++]=" if dbg.getinfo(2) then\n";
s[j++]=" local w = dbg.getinfo(2, \"S\").what\n"; s[j++]=" local w = dbg.getinfo(2, \"S\").what\n";
s[j++]=" if w ~= \"main\" and w ~= \"C\" then\n"; s[j++]= is_eval ? " if w ~= \"main\" and w ~= \"C\" then\n" : " if w ~= \"C\" then\n";
s[j++]=" error(\"Script attempted to create global variable '\"..tostring(n)..\"'\", 2)\n"; s[j++]=" error(\"Script attempted to create global variable '\"..tostring(n)..\"'\", 2)\n";
s[j++]=" end\n"; s[j++]=" end\n";
s[j++]=" end\n"; s[j++]=" end\n";
...@@ -1336,3 +1342,7 @@ void luaCallFunction(scriptRunCtx* run_ctx, lua_State *lua, robj** keys, size_t ...@@ -1336,3 +1342,7 @@ void luaCallFunction(scriptRunCtx* run_ctx, lua_State *lua, robj** keys, size_t
/* remove run_ctx from registry, its only applicable for the current script. */ /* remove run_ctx from registry, its only applicable for the current script. */
luaSaveOnRegistry(lua, REGISTRY_RUN_CTX_NAME, NULL); luaSaveOnRegistry(lua, REGISTRY_RUN_CTX_NAME, NULL);
} }
unsigned long luaMemory(lua_State *lua) {
return lua_gc(lua, LUA_GCCOUNT, 0) * 1024LL;
}
...@@ -57,10 +57,11 @@ ...@@ -57,10 +57,11 @@
#define REGISTRY_RUN_CTX_NAME "__RUN_CTX__" #define REGISTRY_RUN_CTX_NAME "__RUN_CTX__"
void luaRegisterRedisAPI(lua_State* lua); void luaRegisterRedisAPI(lua_State* lua);
void luaEnableGlobalsProtection(lua_State *lua); void luaEnableGlobalsProtection(lua_State *lua, int is_eval);
void luaSaveOnRegistry(lua_State* lua, const char* name, void* ptr); void luaSaveOnRegistry(lua_State* lua, const char* name, void* ptr);
void* luaGetFromRegistry(lua_State* lua, const char* name); void* luaGetFromRegistry(lua_State* lua, const char* name);
void luaCallFunction(scriptRunCtx* r_ctx, lua_State *lua, robj** keys, size_t nkeys, robj** args, size_t nargs, int debug_enabled); void luaCallFunction(scriptRunCtx* r_ctx, lua_State *lua, robj** keys, size_t nkeys, robj** args, size_t nargs, int debug_enabled);
unsigned long luaMemory(lua_State *lua);
#endif /* __SCRIPT_LUA_H_ */ #endif /* __SCRIPT_LUA_H_ */
...@@ -35,7 +35,7 @@ ...@@ -35,7 +35,7 @@
#include "latency.h" #include "latency.h"
#include "atomicvar.h" #include "atomicvar.h"
#include "mt19937-64.h" #include "mt19937-64.h"
#include "script.h" #include "functions.h"
#include <time.h> #include <time.h>
#include <signal.h> #include <signal.h>
...@@ -472,6 +472,31 @@ struct redisCommand scriptSubcommands[] = { ...@@ -472,6 +472,31 @@ struct redisCommand scriptSubcommands[] = {
{NULL}, {NULL},
}; };
struct redisCommand functionSubcommands[] = {
{"create",functionsCreateCommand,-5,
"may-replicate no-script @scripting"},
{"delete",functionsDeleteCommand,3,
"may-replicate no-script @scripting"},
{"kill",functionsKillCommand,2,
"no-script @scripting"},
{"info",functionsInfoCommand,-3,
"no-script @scripting"},
{"list",functionsListCommand,2,
"no-script @scripting"},
{"stats",functionsStatsCommand,2,
"no-script @scripting"},
{"help",functionsHelpCommand,2,
"ok-loading ok-stale @scripting"},
{NULL},
};
struct redisCommand clientSubcommands[] = { struct redisCommand clientSubcommands[] = {
{"caching",clientCommand,3, {"caching",clientCommand,3,
"no-script ok-loading ok-stale @connection"}, "no-script ok-loading ok-stale @connection"},
...@@ -2033,7 +2058,25 @@ struct redisCommand redisCommandTable[] = { ...@@ -2033,7 +2058,25 @@ struct redisCommand redisCommandTable[] = {
"no-auth no-script ok-stale ok-loading fast @connection"}, "no-auth no-script ok-stale ok-loading fast @connection"},
{"failover",failoverCommand,-1, {"failover",failoverCommand,-1,
"admin no-script ok-stale"} "admin no-script ok-stale"},
{"function",NULL,-2,
"",
.subcommands=functionSubcommands},
{"fcall",fcallCommand,-3,
"no-script no-monitor may-replicate no-mandatory-keys @scripting",
{{"read write", /* We pass both read and write because these flag are worst-case-scenario */
KSPEC_BS_INDEX,.bs.index={2},
KSPEC_FK_KEYNUM,.fk.keynum={0,1,1}}},
functionGetKeys},
{"fcall_ro",fcallCommandReadOnly,-3,
"no-script no-monitor no-mandatory-keys @scripting",
{{"read",
KSPEC_BS_INDEX,.bs.index={2},
KSPEC_FK_KEYNUM,.fk.keynum={0,1,1}}},
functionGetKeys},
}; };
/*============================ Utility functions ============================ */ /*============================ Utility functions ============================ */
...@@ -2209,6 +2252,11 @@ void dictSdsDestructor(dict *d, void *val) ...@@ -2209,6 +2252,11 @@ void dictSdsDestructor(dict *d, void *val)
sdsfree(val); sdsfree(val);
} }
void *dictSdsDup(dict *d, const void *key) {
UNUSED(d);
return sdsdup((const sds) key);
}
int dictObjKeyCompare(dict *d, const void *key1, int dictObjKeyCompare(dict *d, const void *key1,
const void *key2) const void *key2)
{ {
...@@ -3517,8 +3565,10 @@ void createSharedObjects(void) { ...@@ -3517,8 +3565,10 @@ void createSharedObjects(void) {
"-NOSCRIPT No matching script. Please use EVAL.\r\n")); "-NOSCRIPT No matching script. Please use EVAL.\r\n"));
shared.loadingerr = createObject(OBJ_STRING,sdsnew( shared.loadingerr = createObject(OBJ_STRING,sdsnew(
"-LOADING Redis is loading the dataset in memory\r\n")); "-LOADING Redis is loading the dataset in memory\r\n"));
shared.slowscripterr = createObject(OBJ_STRING,sdsnew( shared.slowevalerr = createObject(OBJ_STRING,sdsnew(
"-BUSY Redis is busy running a script. You can only call SCRIPT KILL or SHUTDOWN NOSAVE.\r\n")); "-BUSY Redis is busy running a script. You can only call SCRIPT KILL or SHUTDOWN NOSAVE.\r\n"));
shared.slowscripterr = createObject(OBJ_STRING,sdsnew(
"-BUSY Redis is busy running a script. You can only call FUNCTION KILL or SHUTDOWN NOSAVE.\r\n"));
shared.masterdownerr = createObject(OBJ_STRING,sdsnew( shared.masterdownerr = createObject(OBJ_STRING,sdsnew(
"-MASTERDOWN Link with MASTER is down and replica-serve-stale-data is set to 'no'.\r\n")); "-MASTERDOWN Link with MASTER is down and replica-serve-stale-data is set to 'no'.\r\n"));
shared.bgsaveerr = createObject(OBJ_STRING,sdsnew( shared.bgsaveerr = createObject(OBJ_STRING,sdsnew(
...@@ -4345,6 +4395,7 @@ void initServer(void) { ...@@ -4345,6 +4395,7 @@ void initServer(void) {
if (server.cluster_enabled) clusterInit(); if (server.cluster_enabled) clusterInit();
replicationScriptCacheInit(); replicationScriptCacheInit();
scriptingInit(1); scriptingInit(1);
functionsInit();
slowlogInit(); slowlogInit();
latencyMonitorInit(); latencyMonitorInit();
...@@ -5448,9 +5499,15 @@ int processCommand(client *c) { ...@@ -5448,9 +5499,15 @@ int processCommand(client *c) {
tolower(((char*)c->argv[1]->ptr)[0]) == 'n') && tolower(((char*)c->argv[1]->ptr)[0]) == 'n') &&
!(c->cmd->proc == scriptCommand && !(c->cmd->proc == scriptCommand &&
c->argc == 2 && c->argc == 2 &&
tolower(((char*)c->argv[1]->ptr)[0]) == 'k')) tolower(((char*)c->argv[1]->ptr)[0]) == 'k') &&
!(c->cmd->proc == functionsKillCommand) &&
!(c->cmd->proc == functionsStatsCommand))
{ {
rejectCommand(c, shared.slowscripterr); if (scriptIsEval()) {
rejectCommand(c, shared.slowevalerr);
} else {
rejectCommand(c, shared.slowscripterr);
}
return C_OK; return C_OK;
} }
...@@ -6281,6 +6338,7 @@ sds genRedisInfoString(const char *section) { ...@@ -6281,6 +6338,7 @@ sds genRedisInfoString(const char *section) {
char peak_hmem[64]; char peak_hmem[64];
char total_system_hmem[64]; char total_system_hmem[64];
char used_memory_lua_hmem[64]; char used_memory_lua_hmem[64];
char used_memory_vm_total_hmem[64];
char used_memory_scripts_hmem[64]; char used_memory_scripts_hmem[64];
char used_memory_rss_hmem[64]; char used_memory_rss_hmem[64];
char maxmemory_hmem[64]; char maxmemory_hmem[64];
...@@ -6288,6 +6346,7 @@ sds genRedisInfoString(const char *section) { ...@@ -6288,6 +6346,7 @@ sds genRedisInfoString(const char *section) {
size_t total_system_mem = server.system_memory_size; size_t total_system_mem = server.system_memory_size;
const char *evict_policy = evictPolicyToString(); const char *evict_policy = evictPolicyToString();
long long memory_lua = evalMemory(); long long memory_lua = evalMemory();
long long memory_functions = functionsMemory();
struct redisMemOverhead *mh = getMemoryOverheadData(); struct redisMemOverhead *mh = getMemoryOverheadData();
/* Peak memory is updated from time to time by serverCron() so it /* Peak memory is updated from time to time by serverCron() so it
...@@ -6301,7 +6360,8 @@ sds genRedisInfoString(const char *section) { ...@@ -6301,7 +6360,8 @@ sds genRedisInfoString(const char *section) {
bytesToHuman(peak_hmem,server.stat_peak_memory); bytesToHuman(peak_hmem,server.stat_peak_memory);
bytesToHuman(total_system_hmem,total_system_mem); bytesToHuman(total_system_hmem,total_system_mem);
bytesToHuman(used_memory_lua_hmem,memory_lua); bytesToHuman(used_memory_lua_hmem,memory_lua);
bytesToHuman(used_memory_scripts_hmem,mh->lua_caches); bytesToHuman(used_memory_vm_total_hmem,memory_functions + memory_lua);
bytesToHuman(used_memory_scripts_hmem,mh->lua_caches + mh->functions_caches);
bytesToHuman(used_memory_rss_hmem,server.cron_malloc_stats.process_rss); bytesToHuman(used_memory_rss_hmem,server.cron_malloc_stats.process_rss);
bytesToHuman(maxmemory_hmem,server.maxmemory); bytesToHuman(maxmemory_hmem,server.maxmemory);
...@@ -6324,11 +6384,18 @@ sds genRedisInfoString(const char *section) { ...@@ -6324,11 +6384,18 @@ sds genRedisInfoString(const char *section) {
"allocator_resident:%zu\r\n" "allocator_resident:%zu\r\n"
"total_system_memory:%lu\r\n" "total_system_memory:%lu\r\n"
"total_system_memory_human:%s\r\n" "total_system_memory_human:%s\r\n"
"used_memory_lua:%lld\r\n" "used_memory_lua:%lld\r\n" /* deprecated, renamed to used_memory_vm_eval */
"used_memory_lua_human:%s\r\n" "used_memory_vm_eval:%lld\r\n"
"used_memory_lua_human:%s\r\n" /* deprecated */
"used_memory_scripts_eval:%lld\r\n"
"number_of_cached_scripts:%lu\r\n"
"number_of_functions:%lu\r\n"
"used_memory_vm_functions:%lld\r\n"
"used_memory_vm_total:%lld\r\n"
"used_memory_vm_total_human:%s\r\n"
"used_memory_functions:%lld\r\n"
"used_memory_scripts:%lld\r\n" "used_memory_scripts:%lld\r\n"
"used_memory_scripts_human:%s\r\n" "used_memory_scripts_human:%s\r\n"
"number_of_cached_scripts:%lu\r\n"
"maxmemory:%lld\r\n" "maxmemory:%lld\r\n"
"maxmemory_human:%s\r\n" "maxmemory_human:%s\r\n"
"maxmemory_policy:%s\r\n" "maxmemory_policy:%s\r\n"
...@@ -6367,10 +6434,17 @@ sds genRedisInfoString(const char *section) { ...@@ -6367,10 +6434,17 @@ sds genRedisInfoString(const char *section) {
(unsigned long)total_system_mem, (unsigned long)total_system_mem,
total_system_hmem, total_system_hmem,
memory_lua, memory_lua,
memory_lua,
used_memory_lua_hmem, used_memory_lua_hmem,
(long long) mh->lua_caches, (long long) mh->lua_caches,
used_memory_scripts_hmem,
dictSize(evalScriptsDict()), dictSize(evalScriptsDict()),
functionsNum(),
memory_functions,
memory_functions + memory_lua,
used_memory_vm_total_hmem,
(long long) mh->functions_caches,
(long long) mh->lua_caches + (long long) mh->functions_caches,
used_memory_scripts_hmem,
server.maxmemory, server.maxmemory,
maxmemory_hmem, maxmemory_hmem,
evict_policy, evict_policy,
......
...@@ -822,6 +822,19 @@ typedef struct redisDb { ...@@ -822,6 +822,19 @@ typedef struct redisDb {
clusterSlotToKeyMapping *slots_to_keys; /* Array of slots to keys. Only used in cluster mode (db 0). */ clusterSlotToKeyMapping *slots_to_keys; /* Array of slots to keys. Only used in cluster mode (db 0). */
} redisDb; } redisDb;
/* forward declaration for functions ctx */
typedef struct functionsCtx functionsCtx;
/* Holding object that need to be populated during
* rdb loading. On loading end it is possible to decide
* whether not to set those objects on their rightful place.
* For example: dbarray need to be set as main database on
* successful loading and dropped on failure. */
typedef struct rdbLoadingCtx {
redisDb* dbarray;
functionsCtx* functions_ctx;
}rdbLoadingCtx;
/* Client MULTI/EXEC state */ /* Client MULTI/EXEC state */
typedef struct multiCmd { typedef struct multiCmd {
robj **argv; robj **argv;
...@@ -1122,7 +1135,7 @@ struct sharedObjectsStruct { ...@@ -1122,7 +1135,7 @@ struct sharedObjectsStruct {
robj *crlf, *ok, *err, *emptybulk, *czero, *cone, *pong, *space, robj *crlf, *ok, *err, *emptybulk, *czero, *cone, *pong, *space,
*queued, *null[4], *nullarray[4], *emptymap[4], *emptyset[4], *queued, *null[4], *nullarray[4], *emptymap[4], *emptyset[4],
*emptyarray, *wrongtypeerr, *nokeyerr, *syntaxerr, *sameobjecterr, *emptyarray, *wrongtypeerr, *nokeyerr, *syntaxerr, *sameobjecterr,
*outofrangeerr, *noscripterr, *loadingerr, *slowscripterr, *bgsaveerr, *outofrangeerr, *noscripterr, *loadingerr, *slowevalerr, *slowscripterr, *bgsaveerr,
*masterdownerr, *roslaveerr, *execaborterr, *noautherr, *noreplicaserr, *masterdownerr, *roslaveerr, *execaborterr, *noautherr, *noreplicaserr,
*busykeyerr, *oomerr, *plus, *messagebulk, *pmessagebulk, *subscribebulk, *busykeyerr, *oomerr, *plus, *messagebulk, *pmessagebulk, *subscribebulk,
*unsubscribebulk, *psubscribebulk, *punsubscribebulk, *del, *unlink, *unsubscribebulk, *psubscribebulk, *punsubscribebulk, *del, *unlink,
...@@ -1203,6 +1216,7 @@ struct redisMemOverhead { ...@@ -1203,6 +1216,7 @@ struct redisMemOverhead {
size_t clients_normal; size_t clients_normal;
size_t aof_buffer; size_t aof_buffer;
size_t lua_caches; size_t lua_caches;
size_t functions_caches;
size_t overhead_total; size_t overhead_total;
size_t dataset; size_t dataset;
size_t total_keys; size_t total_keys;
...@@ -2670,6 +2684,7 @@ int sintercardGetKeys(struct redisCommand *cmd,robj **argv, int argc, getKeysRes ...@@ -2670,6 +2684,7 @@ int sintercardGetKeys(struct redisCommand *cmd,robj **argv, int argc, getKeysRes
int zunionInterDiffGetKeys(struct redisCommand *cmd,robj **argv, int argc, getKeysResult *result); int zunionInterDiffGetKeys(struct redisCommand *cmd,robj **argv, int argc, getKeysResult *result);
int zunionInterDiffStoreGetKeys(struct redisCommand *cmd,robj **argv, int argc, getKeysResult *result); int zunionInterDiffStoreGetKeys(struct redisCommand *cmd,robj **argv, int argc, getKeysResult *result);
int evalGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result); int evalGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result);
int functionGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result);
int sortGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result); int sortGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result);
int migrateGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result); int migrateGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result);
int georadiusGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result); int georadiusGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult *result);
...@@ -2761,6 +2776,7 @@ uint64_t dictSdsCaseHash(const void *key); ...@@ -2761,6 +2776,7 @@ uint64_t dictSdsCaseHash(const void *key);
int dictSdsKeyCompare(dict *d, const void *key1, const void *key2); int dictSdsKeyCompare(dict *d, const void *key1, const void *key2);
int dictSdsKeyCaseCompare(dict *d, const void *key1, const void *key2); int dictSdsKeyCaseCompare(dict *d, const void *key1, const void *key2);
void dictSdsDestructor(dict *d, void *val); void dictSdsDestructor(dict *d, void *val);
void *dictSdsDup(dict *d, const void *key);
/* Git SHA1 */ /* Git SHA1 */
char *redisGitSHA1(void); char *redisGitSHA1(void);
......
...@@ -521,6 +521,12 @@ foreach testType {Successful Aborted} { ...@@ -521,6 +521,12 @@ foreach testType {Successful Aborted} {
# Set a key value on replica to check status during loading, on failure and after swapping db # Set a key value on replica to check status during loading, on failure and after swapping db
$replica set mykey myvalue $replica set mykey myvalue
# Set a function value on replica to check status during loading, on failure and after swapping db
$replica function create LUA test {return 'hello1'}
# Set a function value on master to check it reaches the replica when replication ends
$master function create LUA test {return 'hello2'}
# Force the replica to try another full sync (this time it will have matching master replid) # Force the replica to try another full sync (this time it will have matching master replid)
$master multi $master multi
$master client kill type replica $master client kill type replica
...@@ -552,6 +558,9 @@ foreach testType {Successful Aborted} { ...@@ -552,6 +558,9 @@ foreach testType {Successful Aborted} {
# Ensure we still see old values while async_loading is in progress and also not LOADING status # Ensure we still see old values while async_loading is in progress and also not LOADING status
assert_equal [$replica get mykey] "myvalue" assert_equal [$replica get mykey] "myvalue"
# Ensure we still can call old function while async_loading is in progress
assert_equal [$replica fcall test 0] "hello1"
# Make sure we're still async_loading to validate previous assertion # Make sure we're still async_loading to validate previous assertion
assert_equal [s -1 async_loading] 1 assert_equal [s -1 async_loading] 1
...@@ -576,6 +585,9 @@ foreach testType {Successful Aborted} { ...@@ -576,6 +585,9 @@ foreach testType {Successful Aborted} {
# Ensure we see old values from replica # Ensure we see old values from replica
assert_equal [$replica get mykey] "myvalue" assert_equal [$replica get mykey] "myvalue"
# Ensure we still can call old function
assert_equal [$replica fcall test 0] "hello1"
# Make sure amount of replica keys didn't change # Make sure amount of replica keys didn't change
assert_equal [$replica dbsize] 2001 assert_equal [$replica dbsize] 2001
} }
...@@ -595,6 +607,9 @@ foreach testType {Successful Aborted} { ...@@ -595,6 +607,9 @@ foreach testType {Successful Aborted} {
# Ensure we don't see anymore the key that was stored only to replica and also that we don't get LOADING status # Ensure we don't see anymore the key that was stored only to replica and also that we don't get LOADING status
assert_equal [$replica GET mykey] "" assert_equal [$replica GET mykey] ""
# Ensure we got the new function
assert_equal [$replica fcall test 0] "hello2"
# Make sure amount of keys matches master # Make sure amount of keys matches master
assert_equal [$replica dbsize] 1010 assert_equal [$replica dbsize] 1010
} }
...@@ -624,6 +639,10 @@ test {diskless loading short read} { ...@@ -624,6 +639,10 @@ test {diskless loading short read} {
$replica config set dynamic-hz no $replica config set dynamic-hz no
# Try to fill the master with all types of data types / encodings # Try to fill the master with all types of data types / encodings
set start [clock clicks -milliseconds] set start [clock clicks -milliseconds]
# Set a function value to check short read handling on functions
r function create LUA test {return 'hello1'}
for {set k 0} {$k < 3} {incr k} { for {set k 0} {$k < 3} {incr k} {
for {set i 0} {$i < 10} {incr i} { for {set i 0} {$i < 10} {incr i} {
r set "$k int_$i" [expr {int(rand()*10000)}] r set "$k int_$i" [expr {int(rand()*10000)}]
......
start_server {tags {"scripting"}} {
test {FUNCTION - Basic usage} {
r function create LUA test {return 'hello'}
r fcall test 0
} {hello}
test {FUNCTION - Create an already exiting function raise error} {
catch {
r function create LUA test {return 'hello1'}
} e
set _ $e
} {*Function already exists*}
test {FUNCTION - Create function with unexisting engine} {
catch {
r function create bad_engine test {return 'hello1'}
} e
set _ $e
} {*Engine not found*}
test {FUNCTION - Test uncompiled script} {
catch {
r function create LUA test1 {bad script}
} e
set _ $e
} {*Error compiling function*}
test {FUNCTION - test replace argument} {
r function create LUA test REPLACE {return 'hello1'}
r fcall test 0
} {hello1}
test {FUNCTION - test replace argument with function creation failure keeps old function} {
catch {r function create LUA test REPLACE {error}}
r fcall test 0
} {hello1}
test {FUNCTION - test function delete} {
r function delete test
catch {
r fcall test 0
} e
set _ $e
} {*Function not found*}
test {FUNCTION - test description argument} {
r function create LUA test DESCRIPTION {some description} {return 'hello'}
r function list
} {{name test engine LUA description {some description}}}
test {FUNCTION - test info specific function} {
r function info test WITHCODE
} {name test engine LUA description {some description} code {return 'hello'}}
test {FUNCTION - test info without code} {
r function info test
} {name test engine LUA description {some description}}
test {FUNCTION - test info on function that does not exists} {
catch {
r function info bad_function_name
} e
set _ $e
} {*Function does not exists*}
test {FUNCTION - test info with bad number of arguments} {
catch {
r function info test WITHCODE bad_arg
} e
set _ $e
} {*wrong number of arguments*}
test {FUNCTION - test fcall bad arguments} {
catch {
r fcall test bad_arg
} e
set _ $e
} {*Bad number of keys provided*}
test {FUNCTION - test fcall bad number of keys arguments} {
catch {
r fcall test 10 key1
} e
set _ $e
} {*Number of keys can't be greater than number of args*}
test {FUNCTION - test fcall negative number of keys} {
catch {
r fcall test -1 key1
} e
set _ $e
} {*Number of keys can't be negative*}
test {FUNCTION - test function delete on not exiting function} {
catch {
r function delete test1
} e
set _ $e
} {*Function not found*}
test {FUNCTION - test function kill when function is not running} {
catch {
r function kill
} e
set _ $e
} {*No scripts in execution*}
test {FUNCTION - test wrong subcommand} {
catch {
r function bad_subcommand
} e
set _ $e
} {*Unknown subcommand*}
test {FUNCTION - test loading from rdb} {
r debug reload
r fcall test 0
} {hello}
test {FUNCTION - test fcall_ro with write command} {
r function create lua test REPLACE {return redis.call('set', 'x', '1')}
catch { r fcall_ro test 0 } e
set _ $e
} {*Write commands are not allowed from read-only scripts*}
test {FUNCTION - test fcall_ro with read only commands} {
r function create lua test REPLACE {return redis.call('get', 'x')}
r set x 1
r fcall_ro test 0
} {1}
test {FUNCTION - test keys and argv} {
r function create lua test REPLACE {return redis.call('set', KEYS[1], ARGV[1])}
r fcall test 1 x foo
r get x
} {foo}
test {FUNCTION - test command get keys on fcall} {
r COMMAND GETKEYS fcall test 1 x foo
} {x}
test {FUNCTION - test command get keys on fcall_ro} {
r COMMAND GETKEYS fcall_ro test 1 x foo
} {x}
test {FUNCTION - test function kill} {
set rd [redis_deferring_client]
r config set script-time-limit 10
r function create lua test REPLACE {local a = 1 while true do a = a + 1 end}
$rd fcall test 0
after 200
catch {r ping} e
assert_match {BUSY*} $e
assert_match {running_script {name test command {fcall test 0} duration_ms *} engines LUA} [r FUNCTION STATS]
r function kill
after 200 ; # Give some time to Lua to call the hook again...
assert_equal [r ping] "PONG"
}
test {FUNCTION - test script kill not working on function} {
set rd [redis_deferring_client]
r config set script-time-limit 10
r function create lua test REPLACE {local a = 1 while true do a = a + 1 end}
$rd fcall test 0
after 200
catch {r ping} e
assert_match {BUSY*} $e
catch {r script kill} e
assert_match {BUSY*} $e
r function kill
after 200 ; # Give some time to Lua to call the hook again...
assert_equal [r ping] "PONG"
}
test {FUNCTION - test function kill not working on eval} {
set rd [redis_deferring_client]
r config set script-time-limit 10
$rd eval {local a = 1 while true do a = a + 1 end} 0
after 200
catch {r ping} e
assert_match {BUSY*} $e
catch {r function kill} e
assert_match {BUSY*} $e
r script kill
after 200 ; # Give some time to Lua to call the hook again...
assert_equal [r ping] "PONG"
}
}
start_server {tags {"scripting repl"}} {
start_server {} {
test "Connect a replica to the master instance" {
r -1 slaveof [srv 0 host] [srv 0 port]
wait_for_condition 50 100 {
[s -1 role] eq {slave} &&
[string match {*master_link_status:up*} [r -1 info replication]]
} else {
fail "Can't turn the instance into a replica"
}
}
test {FUNCTION - creation is replicated to replica} {
r function create LUA test DESCRIPTION {some description} {return 'hello'}
wait_for_condition 50 100 {
[r -1 function list] eq {{name test engine LUA description {some description}}}
} else {
fail "Failed waiting for function to replicate to replica"
}
}
test {FUNCTION - call on replica} {
r -1 fcall test 0
} {hello}
test {FUNCTION - delete is replicated to replica} {
r function delete test
wait_for_condition 50 100 {
[r -1 function list] eq {}
} else {
fail "Failed waiting for function to replicate to replica"
}
}
test "Disconnecting the replica from master instance" {
r -1 slaveof no one
# creating a function after disconnect to make sure function
# is replicated on rdb phase
r function create LUA test DESCRIPTION {some description} {return 'hello'}
# reconnect the replica
r -1 slaveof [srv 0 host] [srv 0 port]
wait_for_condition 50 100 {
[s -1 role] eq {slave} &&
[string match {*master_link_status:up*} [r -1 info replication]]
} else {
fail "Can't turn the instance into a replica"
}
}
test "FUNCTION - test replication to replica on rdb phase" {
r -1 fcall test 0
} {hello}
test "FUNCTION - test replication to replica on rdb phase info command" {
r -1 function info test WITHCODE
} {name test engine LUA description {some description} code {return 'hello'}}
test "FUNCTION - create on read only replica" {
catch {
r -1 function create LUA test DESCRIPTION {some description} {return 'hello'}
} e
set _ $e
} {*Can not create a function on a read only replica*}
test "FUNCTION - delete on read only replica" {
catch {
r -1 function delete test
} e
set _ $e
} {*Can not delete a function on a read only replica*}
test "FUNCTION - function effect is replicated to replica" {
r function create LUA test REPLACE {return redis.call('set', 'x', '1')}
r fcall test 0
assert {[r get x] eq {1}}
wait_for_condition 50 100 {
[r -1 get x] eq {1}
} else {
fail "Failed waiting function effect to be replicated to replica"
}
}
test "FUNCTION - modify key space of read only replica" {
catch {
r -1 fcall test 0
} e
set _ $e
} {*can't write against a read only replica*}
}
}
\ No newline at end of file
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