Unverified Commit ae020e3d authored by Meir Shpilraien (Spielrein)'s avatar Meir Shpilraien (Spielrein) Committed by GitHub
Browse files

Functions: Move library meta data to be part of the library payload. (#10500)

## Move library meta data to be part of the library payload.

Following the discussion on https://github.com/redis/redis/issues/10429 and the intention to add (in the future) library versioning support, we believe that the entire library metadata (like name and engine) should be part of the library payload and not provided by the `FUNCTION LOAD` command. The reasoning behind this is that the programmer who developed the library should be the one who set those values (name, engine, and in the future also version). **It is not the responsibility of the admin who load the library into the database.**

The PR moves all the library metadata (engine and function name) to be part of the library payload. The metadata needs to be provided on the first line of the payload using the shebang format (`#!<engine> name=<name>`), example:

```lua
#!lua name=test
redis.register_function('foo', function() return 1 end)
```

The above script will run on the Lua engine and will create a library called `test`.

## API Changes (compare to 7.0 rc2)

* `FUNCTION LOAD` command was change and now it simply gets the library payload and extract the engine and name from the payload. In addition, the command will now return the function name which can later be used on `FUNCTION DELETE` and `FUNCTION LIST`.
* The description field was completely removed from`FUNCTION LOAD`, and `FUNCTION LIST`


## Breaking Changes (compare to 7.0 rc2)

* Library description was removed (we can re-add it in the future either as part of the shebang line or an additional line).
* Loading an AOF file that was generated by either 7.0 rc1 or 7.0 rc2 will fail because the old command syntax is invalid.

## Notes

* Loading an RDB file that was generated by rc1 / rc2 **is** supported, Redis will automatically add the shebang to the libraries payloads (we can probably delete that code after 7.0.3 or so since there's no need to keep supporting upgrades from an RC build).
parent 2db0d898
......@@ -2142,19 +2142,9 @@ static int rewriteFunctions(rio *aof) {
dictEntry *entry = NULL;
while ((entry = dictNext(iter))) {
functionLibInfo *li = dictGetVal(entry);
if (li->desc) {
if (rioWrite(aof, "*7\r\n", 4) == 0) goto werr;
} else {
if (rioWrite(aof, "*5\r\n", 4) == 0) goto werr;
}
if (rioWrite(aof, "*3\r\n", 4) == 0) goto werr;
char function_load[] = "$8\r\nFUNCTION\r\n$4\r\nLOAD\r\n";
if (rioWrite(aof, function_load, sizeof(function_load) - 1) == 0) goto werr;
if (rioWriteBulkString(aof, li->ei->name, sdslen(li->ei->name)) == 0) goto werr;
if (rioWriteBulkString(aof, li->name, sdslen(li->name)) == 0) goto werr;
if (li->desc) {
if (rioWriteBulkString(aof, "description", 11) == 0) goto werr;
if (rioWriteBulkString(aof, li->desc, sdslen(li->desc)) == 0) goto werr;
}
if (rioWriteBulkString(aof, li->code, sdslen(li->code)) == 0) goto werr;
}
dictReleaseIterator(iter);
......
......@@ -3425,10 +3425,7 @@ NULL
/* FUNCTION LOAD argument table */
struct redisCommandArg FUNCTION_LOAD_Args[] = {
{"engine-name",ARG_TYPE_STRING,-1,NULL,NULL,NULL,CMD_ARG_NONE},
{"library-name",ARG_TYPE_STRING,-1,NULL,NULL,NULL,CMD_ARG_NONE},
{"replace",ARG_TYPE_PURE_TOKEN,-1,"REPLACE",NULL,NULL,CMD_ARG_OPTIONAL},
{"library-description",ARG_TYPE_STRING,-1,"DESCRIPTION",NULL,NULL,CMD_ARG_OPTIONAL},
{"function-code",ARG_TYPE_STRING,-1,NULL,NULL,NULL,CMD_ARG_NONE},
{0}
};
......@@ -3481,7 +3478,7 @@ struct redisCommand FUNCTION_Subcommands[] = {
{"help","Show helpful text about the different subcommands","O(1)","7.0.0",CMD_DOC_NONE,NULL,NULL,COMMAND_GROUP_SCRIPTING,FUNCTION_HELP_History,FUNCTION_HELP_tips,functionHelpCommand,2,CMD_LOADING|CMD_STALE,ACL_CATEGORY_SCRIPTING},
{"kill","Kill the function currently in execution.","O(1)","7.0.0",CMD_DOC_NONE,NULL,NULL,COMMAND_GROUP_SCRIPTING,FUNCTION_KILL_History,FUNCTION_KILL_tips,functionKillCommand,2,CMD_NOSCRIPT|CMD_ALLOW_BUSY,ACL_CATEGORY_SCRIPTING},
{"list","List information about all the functions","O(N) where N is the number of functions","7.0.0",CMD_DOC_NONE,NULL,NULL,COMMAND_GROUP_SCRIPTING,FUNCTION_LIST_History,FUNCTION_LIST_tips,functionListCommand,-2,CMD_NOSCRIPT,ACL_CATEGORY_SCRIPTING,.args=FUNCTION_LIST_Args},
{"load","Create a function with the given arguments (name, code, description)","O(1) (considering compilation time is redundant)","7.0.0",CMD_DOC_NONE,NULL,NULL,COMMAND_GROUP_SCRIPTING,FUNCTION_LOAD_History,FUNCTION_LOAD_tips,functionLoadCommand,-5,CMD_NOSCRIPT|CMD_WRITE|CMD_DENYOOM,ACL_CATEGORY_SCRIPTING,.args=FUNCTION_LOAD_Args},
{"load","Create a function with the given arguments (name, code, description)","O(1) (considering compilation time is redundant)","7.0.0",CMD_DOC_NONE,NULL,NULL,COMMAND_GROUP_SCRIPTING,FUNCTION_LOAD_History,FUNCTION_LOAD_tips,functionLoadCommand,-3,CMD_NOSCRIPT|CMD_WRITE|CMD_DENYOOM,ACL_CATEGORY_SCRIPTING,.args=FUNCTION_LOAD_Args},
{"restore","Restore all the functions on the given payload","O(N) where N is the number of functions on the payload","7.0.0",CMD_DOC_NONE,NULL,NULL,COMMAND_GROUP_SCRIPTING,FUNCTION_RESTORE_History,FUNCTION_RESTORE_tips,functionRestoreCommand,-3,CMD_NOSCRIPT|CMD_WRITE|CMD_DENYOOM,ACL_CATEGORY_SCRIPTING,.args=FUNCTION_RESTORE_Args},
{"stats","Return information about the function currently running (name, description, duration)","O(1)","7.0.0",CMD_DOC_NONE,NULL,NULL,COMMAND_GROUP_SCRIPTING,FUNCTION_STATS_History,FUNCTION_STATS_tips,functionStatsCommand,2,CMD_NOSCRIPT|CMD_ALLOW_BUSY,ACL_CATEGORY_SCRIPTING},
{0}
......
......@@ -4,7 +4,7 @@
"complexity": "O(1) (considering compilation time is redundant)",
"group": "scripting",
"since": "7.0.0",
"arity": -5,
"arity": -3,
"container": "FUNCTION",
"function": "functionLoadCommand",
"command_flags": [
......@@ -20,26 +20,12 @@
"RESPONSE_POLICY:ALL_SUCCEEDED"
],
"arguments": [
{
"name": "engine-name",
"type": "string"
},
{
"name": "library-name",
"type": "string"
},
{
"name": "replace",
"type": "pure-token",
"token": "REPLACE",
"optional": true
},
{
"name": "library-description",
"type": "string",
"token": "DESCRIPTION",
"optional": true
},
{
"name": "function-code",
"type": "string"
......
......@@ -57,6 +57,12 @@ struct functionsLibCtx {
dict *engines_stats; /* Per engine statistics */
};
typedef struct functionsLibMataData {
sds engine;
sds name;
sds code;
} functionsLibMataData;
dictType engineDictType = {
dictSdsCaseHash, /* hash function */
dictSdsDup, /* key dup */
......@@ -124,7 +130,6 @@ static size_t functionMallocSize(functionInfo *fi) {
static size_t libraryMallocSize(functionLibInfo *li) {
return zmalloc_size(li) + sdsZmallocSize(li->name)
+ (li->desc ? sdsZmallocSize(li->desc) : 0)
+ sdsZmallocSize(li->code);
}
......@@ -157,7 +162,6 @@ static void engineLibraryFree(functionLibInfo* li) {
dictRelease(li->functions);
sdsfree(li->name);
sdsfree(li->code);
if (li->desc) sdsfree(li->desc);
zfree(li);
}
......@@ -265,14 +269,13 @@ int functionLibCreateFunction(sds name, void *function, functionLibInfo *li, sds
return C_OK;
}
static functionLibInfo* engineLibraryCreate(sds name, engineInfo *ei, sds desc, sds code) {
static functionLibInfo* engineLibraryCreate(sds name, engineInfo *ei, sds code) {
functionLibInfo *li = zmalloc(sizeof(*li));
*li = (functionLibInfo) {
.name = sdsdup(name),
.functions = dictCreate(&libraryFunctionDictType),
.ei = ei,
.code = sdsdup(code),
.desc = desc ? sdsdup(desc) : NULL,
};
return li;
}
......@@ -540,17 +543,11 @@ void functionListCommand(client *c) {
}
}
++reply_len;
addReplyMapLen(c, with_code? 5 : 4);
addReplyMapLen(c, with_code? 4 : 3);
addReplyBulkCString(c, "library_name");
addReplyBulkCBuffer(c, li->name, sdslen(li->name));
addReplyBulkCString(c, "engine");
addReplyBulkCBuffer(c, li->ei->name, sdslen(li->ei->name));
addReplyBulkCString(c, "description");
if (li->desc) {
addReplyBulkCBuffer(c, li->desc, sdslen(li->desc));
} else {
addReplyNull(c);
}
addReplyBulkCString(c, "functions");
addReplyArrayLen(c, dictSize(li->functions));
......@@ -745,11 +742,11 @@ void functionRestoreCommand(client *c) {
err = sdsnew("can not read data type");
goto load_error;
}
if (type != RDB_OPCODE_FUNCTION) {
if (type != RDB_OPCODE_FUNCTION && type != RDB_OPCODE_FUNCTION2) {
err = sdsnew("given type is not a function");
goto load_error;
}
if (rdbFunctionLoad(&payload, rdbver, functions_lib_ctx, RDBFLAGS_NONE, &err) != C_OK) {
if (rdbFunctionLoad(&payload, rdbver, functions_lib_ctx, type, RDBFLAGS_NONE, &err) != C_OK) {
if (!err) {
err = sdsnew("failed loading the given functions payload");
}
......@@ -868,36 +865,111 @@ static int functionsVerifyName(sds name) {
return C_OK;
}
/* Compile and save the given library, 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 functionsCreateWithLibraryCtx(sds lib_name,sds engine_name, sds desc, sds code,
int replace, sds* err, functionsLibCtx *lib_ctx) {
int functionExtractLibMetaData(sds payload, functionsLibMataData *md, sds *err) {
sds name = NULL;
sds desc = NULL;
sds engine = NULL;
sds code = NULL;
if (strncmp(payload, "#!", 2) != 0) {
*err = sdsnew("Missing library metadata");
return C_ERR;
}
char *shebang_end = strchr(payload, '\n');
if (shebang_end == NULL) {
*err = sdsnew("Invalid library metadata");
return C_ERR;
}
size_t shebang_len = shebang_end - payload;
sds shebang = sdsnewlen(payload, shebang_len);
int numparts;
sds *parts = sdssplitargs(shebang, &numparts);
sdsfree(shebang);
if (!parts || numparts == 0) {
*err = sdsnew("Invalid library metadata");
sdsfreesplitres(parts, numparts);
return C_ERR;
}
engine = sdsdup(parts[0]);
sdsrange(engine, 2, -1);
for (int i = 1 ; i < numparts ; ++i) {
sds part = parts[i];
if (strncasecmp(part, "name=", 5) == 0) {
if (name) {
*err = sdscatfmt(sdsempty(), "Invalid metadata value, name argument was given multiple times");
goto error;
}
name = sdsdup(part);
sdsrange(name, 5, -1);
continue;
}
*err = sdscatfmt(sdsempty(), "Invalid metadata value given: %s", part);
goto error;
}
if (!name) {
*err = sdsnew("Library name was not given");
goto error;
}
sdsfreesplitres(parts, numparts);
md->name = name;
md->code = sdsnewlen(shebang_end, sdslen(payload) - shebang_len);
md->engine = engine;
return C_OK;
error:
if (name) sdsfree(name);
if (desc) sdsfree(desc);
if (engine) sdsfree(engine);
if (code) sdsfree(code);
sdsfreesplitres(parts, numparts);
return C_ERR;
}
void functionFreeLibMetaData(functionsLibMataData *md) {
if (md->code) sdsfree(md->code);
if (md->name) sdsfree(md->name);
if (md->engine) sdsfree(md->engine);
}
/* Compile and save the given library, return the loaded library name on success
* and NULL on failure. In case on failure the err out param is set with relevant error message */
sds functionsCreateWithLibraryCtx(sds code, int replace, sds* err, functionsLibCtx *lib_ctx) {
dictIterator *iter = NULL;
dictEntry *entry = NULL;
if (functionsVerifyName(lib_name)) {
functionLibInfo *new_li = NULL;
functionLibInfo *old_li = NULL;
functionsLibMataData md = {0};
if (functionExtractLibMetaData(code, &md, err) != C_OK) {
return NULL;
}
if (functionsVerifyName(md.name)) {
*err = sdsnew("Library names can only contain letters and numbers and must be at least one character long");
return C_ERR;
goto error;
}
engineInfo *ei = dictFetchValue(engines, engine_name);
engineInfo *ei = dictFetchValue(engines, md.engine);
if (!ei) {
*err = sdsnew("Engine not found");
return C_ERR;
*err = sdscatfmt(sdsempty(), "Engine '%S' not found", md.engine);
goto error;
}
engine *engine = ei->engine;
functionLibInfo *old_li = dictFetchValue(lib_ctx->libraries, lib_name);
old_li = dictFetchValue(lib_ctx->libraries, md.name);
if (old_li && !replace) {
*err = sdsnew("Library already exists");
return C_ERR;
*err = sdscatfmt(sdsempty(), "Library '%S' already exists", md.name);
goto error;
}
if (old_li) {
libraryUnlink(lib_ctx, old_li);
}
functionLibInfo *new_li = engineLibraryCreate(lib_name, ei, desc, code);
if (engine->create(engine->engine_ctx, new_li, code, err) != C_OK) {
new_li = engineLibraryCreate(md.name, ei, code);
if (engine->create(engine->engine_ctx, new_li, md.code, err) != C_OK) {
goto error;
}
......@@ -925,48 +997,34 @@ int functionsCreateWithLibraryCtx(sds lib_name,sds engine_name, sds desc, sds co
engineLibraryFree(old_li);
}
return C_OK;
sds loaded_lib_name = md.name;
md.name = NULL;
functionFreeLibMetaData(&md);
return loaded_lib_name;
error:
if (iter) dictReleaseIterator(iter);
engineLibraryFree(new_li);
if (old_li) {
libraryLink(lib_ctx, old_li);
}
return C_ERR;
if (new_li) engineLibraryFree(new_li);
if (old_li) libraryLink(lib_ctx, old_li);
functionFreeLibMetaData(&md);
return NULL;
}
/*
* FUNCTION LOAD <ENGINE NAME> <LIBRARY NAME>
* [REPLACE] [DESC <LIBRARY DESCRIPTION>] <LIBRARY CODE>
*
* ENGINE NAME - name of the engine to use the run the library
* LIBRARY NAME - name of the library
* FUNCTION LOAD [REPLACE] <LIBRARY CODE>
* REPLACE - optional, replace existing library
* DESCRIPTION - optional, library description
* LIBRARY CODE - library code to pass to the engine
*/
void functionLoadCommand(client *c) {
robj *engine_name = c->argv[2];
robj *library_name = c->argv[3];
int replace = 0;
int argc_pos = 4;
sds desc = NULL;
int argc_pos = 2;
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;
}
addReplyErrorFormat(c, "Unknown option given: %s", (char*)next_arg->ptr);
return;
}
......@@ -978,8 +1036,8 @@ void functionLoadCommand(client *c) {
robj *code = c->argv[argc_pos];
sds err = NULL;
if (functionsCreateWithLibraryCtx(library_name->ptr, engine_name->ptr,
desc, code->ptr, replace, &err, curr_functions_lib_ctx) != C_OK)
sds library_name = NULL;
if (!(library_name = functionsCreateWithLibraryCtx(code->ptr, replace, &err, curr_functions_lib_ctx)))
{
addReplyErrorSds(c, err);
return;
......@@ -987,7 +1045,7 @@ void functionLoadCommand(client *c) {
/* Indicate that the command changed the data so it will be replicated and
* counted as a data change (for persistence configuration) */
server.dirty++;
addReply(c, shared.ok);
addReplyBulkSds(c, library_name);
}
/* Return memory usage of all the engines combine */
......
......@@ -106,12 +106,10 @@ struct functionLibInfo {
dict *functions; /* Functions dictionary */
engineInfo *ei; /* Pointer to the function engine */
sds code; /* Library code */
sds desc; /* Library description */
};
int functionsRegisterEngine(const char *engine_name, engine *engine_ctx);
int functionsCreateWithLibraryCtx(sds lib_name, sds engine_name, sds desc, sds code,
int replace, sds* err, functionsLibCtx *lib_ctx);
sds functionsCreateWithLibraryCtx(sds code, int replace, sds* err, functionsLibCtx *lib_ctx);
unsigned long functionsMemory();
unsigned long functionsMemoryOverhead();
unsigned long functionsNum();
......
......@@ -1242,24 +1242,9 @@ ssize_t rdbSaveFunctions(rio *rdb) {
ssize_t written = 0;
ssize_t ret;
while ((entry = dictNext(iter))) {
if ((ret = rdbSaveType(rdb, RDB_OPCODE_FUNCTION)) < 0) goto werr;
if ((ret = rdbSaveType(rdb, RDB_OPCODE_FUNCTION2)) < 0) goto werr;
written += ret;
functionLibInfo *li = dictGetVal(entry);
if ((ret = rdbSaveRawString(rdb, (unsigned char *) li->name, sdslen(li->name))) < 0) goto werr;
written += ret;
if ((ret = rdbSaveRawString(rdb, (unsigned char *) li->ei->name, sdslen(li->ei->name))) < 0) goto werr;
written += ret;
if (li->desc) {
/* desc exists */
if ((ret = rdbSaveLen(rdb, 1)) < 0) goto werr;
written += ret;
if ((ret = rdbSaveRawString(rdb, (unsigned char *) li->desc, sdslen(li->desc))) < 0) goto werr;
written += ret;
} else {
/* desc not exists */
if ((ret = rdbSaveLen(rdb, 0)) < 0) goto werr;
written += ret;
}
if ((ret = rdbSaveRawString(rdb, (unsigned char *) li->code, sdslen(li->code))) < 0) goto werr;
written += ret;
}
......@@ -2811,56 +2796,79 @@ void rdbLoadProgressCallback(rio *r, const void *buf, size_t len) {
*
* The lib_ctx argument is also optional. If NULL is given, only verify rdb
* structure with out performing the actual functions loading. */
int rdbFunctionLoad(rio *rdb, int ver, functionsLibCtx* lib_ctx, int rdbflags, sds *err) {
int rdbFunctionLoad(rio *rdb, int ver, functionsLibCtx* lib_ctx, int type, int rdbflags, sds *err) {
UNUSED(ver);
sds error = NULL;
sds final_payload = NULL;
int res = C_ERR;
if (type == RDB_OPCODE_FUNCTION) {
/* RDB that was generated on versions 7.0 rc1 and 7.0 rc2 has another
* an old format that contains the library name, engine and description.
* To support this format we must read those values. */
sds name = NULL;
sds engine_name = NULL;
sds desc = NULL;
sds blob = NULL;
uint64_t has_desc;
sds error = NULL;
int res = C_ERR;
if (!(name = rdbGenericLoadStringObject(rdb, RDB_LOAD_SDS, NULL))) {
error = sdsnew("Failed loading library name");
goto error;
goto cleanup;
}
if (!(engine_name = rdbGenericLoadStringObject(rdb, RDB_LOAD_SDS, NULL))) {
error = sdsnew("Failed loading engine name");
goto error;
goto cleanup;
}
if ((has_desc = rdbLoadLen(rdb, NULL)) == RDB_LENERR) {
error = sdsnew("Failed loading library description indicator");
goto error;
goto cleanup;
}
if (has_desc && !(desc = rdbGenericLoadStringObject(rdb, RDB_LOAD_SDS, NULL))) {
error = sdsnew("Failed loading library description");
goto error;
goto cleanup;
}
if (!(blob = rdbGenericLoadStringObject(rdb, RDB_LOAD_SDS, NULL))) {
error = sdsnew("Failed loading library blob");
goto error;
goto cleanup;
}
/* Translate old format (versions 7.0 rc1 and 7.0 rc2) to new format.
* The new format has the library name and engine inside the script payload.
* Add those parameters to the original script payload (ignore the description if exists). */
final_payload = sdscatfmt(sdsempty(), "#!%s name=%s\n%s", engine_name, name, blob);
cleanup:
if (name) sdsfree(name);
if (engine_name) sdsfree(engine_name);
if (desc) sdsfree(desc);
if (blob) sdsfree(blob);
if (error) goto done;
} else if (type == RDB_OPCODE_FUNCTION2) {
if (!(final_payload = rdbGenericLoadStringObject(rdb, RDB_LOAD_SDS, NULL))) {
error = sdsnew("Failed loading library payload");
goto done;
}
} else {
serverPanic("Bad function type was given to rdbFunctionLoad");
}
if (lib_ctx) {
if (functionsCreateWithLibraryCtx(name, engine_name, desc, blob, rdbflags & RDBFLAGS_ALLOW_DUP, &error, lib_ctx) != C_OK) {
sds library_name = NULL;
if (!(library_name = functionsCreateWithLibraryCtx(final_payload, rdbflags & RDBFLAGS_ALLOW_DUP, &error, lib_ctx))) {
if (!error) {
error = sdsnew("Failed creating the library");
}
goto error;
goto done;
}
sdsfree(library_name);
}
res = C_OK;
error:
if (name) sdsfree(name);
if (engine_name) sdsfree(engine_name);
if (desc) sdsfree(desc);
if (blob) sdsfree(blob);
done:
if (final_payload) sdsfree(final_payload);
if (error) {
if (err) {
*err = error;
......@@ -3091,9 +3099,9 @@ int rdbLoadRioWithLoadingCtx(rio *rdb, int rdbflags, rdbSaveInfo *rsi, rdbLoadin
decrRefCount(aux);
continue; /* Read next opcode. */
}
} else if (type == RDB_OPCODE_FUNCTION) {
} else if (type == RDB_OPCODE_FUNCTION || type == RDB_OPCODE_FUNCTION2) {
sds err = NULL;
if (rdbFunctionLoad(rdb, rdbver, rdb_loading_ctx->functions_lib_ctx, rdbflags, &err) != C_OK) {
if (rdbFunctionLoad(rdb, rdbver, rdb_loading_ctx->functions_lib_ctx, type, rdbflags, &err) != C_OK) {
serverLog(LL_WARNING,"Failed loading library, %s", err);
sdsfree(err);
goto eoferr;
......
......@@ -101,7 +101,8 @@
#define rdbIsObjectType(t) ((t >= 0 && t <= 7) || (t >= 9 && t <= 19))
/* Special RDB opcodes (saved/loaded with rdbSaveType/rdbLoadType). */
#define RDB_OPCODE_FUNCTION 246 /* engine data */
#define RDB_OPCODE_FUNCTION2 245 /* function library data */
#define RDB_OPCODE_FUNCTION 246 /* old function library data for 7.0 rc1 and rc2 */
#define RDB_OPCODE_MODULE_AUX 247 /* Module auxiliary data. */
#define RDB_OPCODE_IDLE 248 /* LRU idle time. */
#define RDB_OPCODE_FREQ 249 /* LFU frequency. */
......@@ -170,7 +171,7 @@ int rdbSaveBinaryFloatValue(rio *rdb, float val);
int rdbLoadBinaryFloatValue(rio *rdb, float *val);
int rdbLoadRio(rio *rdb, int rdbflags, rdbSaveInfo *rsi);
int rdbLoadRioWithLoadingCtx(rio *rdb, int rdbflags, rdbSaveInfo *rsi, rdbLoadingCtx *rdb_loading_ctx);
int rdbFunctionLoad(rio *rdb, int ver, functionsLibCtx* lib_ctx, int rdbflags, sds *err);
int rdbFunctionLoad(rio *rdb, int ver, functionsLibCtx* lib_ctx, int type, int rdbflags, sds *err);
int rdbSaveRio(int req, rio *rdb, int *error, int rdbflags, rdbSaveInfo *rsi);
ssize_t rdbSaveFunctions(rio *rdb);
rdbSaveInfo *rdbPopulateSaveInfo(rdbSaveInfo *rsi);
......
......@@ -63,6 +63,7 @@ struct {
#define RDB_CHECK_DOING_READ_LEN 6
#define RDB_CHECK_DOING_READ_AUX 7
#define RDB_CHECK_DOING_READ_MODULE_AUX 8
#define RDB_CHECK_DOING_READ_FUNCTIONS 9
char *rdb_check_doing_string[] = {
"start",
......@@ -73,7 +74,8 @@ char *rdb_check_doing_string[] = {
"check-sum",
"read-len",
"read-aux",
"read-module-aux"
"read-module-aux",
"read-functions"
};
char *rdb_type_string[] = {
......@@ -303,9 +305,10 @@ int redis_check_rdb(char *rdbfilename, FILE *fp) {
robj *o = rdbLoadCheckModuleValue(&rdb,name);
decrRefCount(o);
continue; /* Read type again. */
} else if (type == RDB_OPCODE_FUNCTION) {
} else if (type == RDB_OPCODE_FUNCTION || type == RDB_OPCODE_FUNCTION2) {
sds err = NULL;
if (rdbFunctionLoad(&rdb, rdbver, NULL, 0, &err) != C_OK) {
rdbstate.doing = RDB_CHECK_DOING_READ_FUNCTIONS;
if (rdbFunctionLoad(&rdb, rdbver, NULL, type, 0, &err) != C_OK) {
rdbCheckError("Failed loading library, %s", err);
sdsfree(err);
goto err;
......
......@@ -64,7 +64,7 @@ test "It is possible to write and read from the cluster" {
}
test "Function no-cluster flag" {
R 1 function load lua test {
R 1 function load {#!lua name=test
redis.register_function{function_name='f1', callback=function() return 'hello' end, flags={'no-cluster'}}
}
catch {R 1 fcall f1 0} e
......
......@@ -346,7 +346,7 @@ if {!$::tls} { ;# fake_redis_node doesn't support TLS
set dir [lindex [r config get dir] 1]
assert_equal "OK" [r debug populate 100000 key 1000]
assert_equal "OK" [r function load lua lib1 "redis.register_function('func1', function() return 123 end)"]
assert_equal "lib1" [r function load "#!lua name=lib1\nredis.register_function('func1', function() return 123 end)"]
if {$functions_only} {
set args "--functions-rdb $dir/cli.rdb"
} else {
......@@ -359,10 +359,10 @@ if {!$::tls} { ;# fake_redis_node doesn't support TLS
file rename "$dir/cli.rdb" "$dir/dump.rdb"
assert_equal "OK" [r set should-not-exist 1]
assert_equal "OK" [r function load lua should_not_exist_func "redis.register_function('should_not_exist_func', function() return 456 end)"]
assert_equal "should_not_exist_func" [r function load "#!lua name=should_not_exist_func\nredis.register_function('should_not_exist_func', function() return 456 end)"]
assert_equal "OK" [r debug reload nosave]
assert_equal {} [r get should-not-exist]
assert_equal {{library_name lib1 engine LUA description {} functions {{name func1 description {} flags {}}}}} [r function list]
assert_equal {{library_name lib1 engine LUA functions {{name func1 description {} flags {}}}}} [r function list]
if {$functions_only} {
assert_equal 0 [r dbsize]
} else {
......
......@@ -47,7 +47,7 @@ start_server {tags {"repl external:skip"}} {
set slave [srv 0 client]
# Load some functions to be used later
$master FUNCTION load lua test replace {
$master FUNCTION load replace {#!lua name=test
redis.register_function{function_name='f_default_flags', callback=function(keys, args) return redis.call('get',keys[1]) end, flags={}}
redis.register_function{function_name='f_no_writes', callback=function(keys, args) return redis.call('get',keys[1]) end, flags={'no-writes'}}
}
......
......@@ -523,10 +523,14 @@ foreach testType {Successful Aborted} {
$replica set mykey myvalue
# Set a function value on replica to check status during loading, on failure and after swapping db
$replica function load LUA test {redis.register_function('test', function() return 'hello1' end)}
$replica function load {#!lua name=test
redis.register_function('test', function() return 'hello1' end)
}
# Set a function value on master to check it reaches the replica when replication ends
$master function load LUA test {redis.register_function('test', function() return 'hello2' end)}
$master function load {#!lua name=test
redis.register_function('test', function() return 'hello2' end)
}
# Force the replica to try another full sync (this time it will have matching master replid)
$master multi
......@@ -659,7 +663,9 @@ test {diskless loading short read} {
set start [clock clicks -milliseconds]
# Set a function value to check short read handling on functions
r function load LUA test {redis.register_function('test', function() return 'hello1' end)}
r function load {#!lua name=test
redis.register_function('test', function() return 'hello1' end)
}
for {set k 0} {$k < 3} {incr k} {
for {set i 0} {$i < 10} {incr i} {
......
......@@ -188,6 +188,10 @@ proc ::redis::__method__readraw {id fd val} {
set ::redis::readraw($id) $val
}
proc ::redis::__method__readingraw {id fd} {
return $::redis::readraw($id)
}
proc ::redis::__method__attributes {id fd} {
set _ $::redis::attributes($id)
}
......
......@@ -185,7 +185,7 @@ start_server {tags {"aofrw external:skip"} overrides {aof-use-rdb-preamble no}}
test "AOF rewrite functions" {
r flushall
r FUNCTION LOAD LUA test DESCRIPTION {desc} {
r FUNCTION LOAD {#!lua name=test
redis.register_function('test', function() return 1 end)
}
r bgrewriteaof
......@@ -194,7 +194,7 @@ start_server {tags {"aofrw external:skip"} overrides {aof-use-rdb-preamble no}}
r debug loadaof
assert_equal [r fcall test 0] 1
r FUNCTION LIST
} {{library_name test engine LUA description desc functions {{name test description {} flags {}}}}}
} {{library_name test engine LUA functions {{name test description {} flags {}}}}}
test {BGREWRITEAOF is delayed if BGSAVE is in progress} {
r flushall
......
......@@ -173,7 +173,9 @@ start_multiple_servers 5 [list overrides $base_conf] {
# upload a function to all the cluster
exec src/redis-cli --cluster-yes --cluster call 127.0.0.1:[srv 0 port] \
FUNCTION LOAD LUA TEST {redis.register_function('test', function() return 'hello' end)}
FUNCTION LOAD {#!lua name=TEST
redis.register_function('test', function() return 'hello' end)
}
# adding node to the cluster
exec src/redis-cli --cluster-yes --cluster add-node \
......@@ -190,13 +192,15 @@ start_multiple_servers 5 [list overrides $base_conf] {
}
# make sure 'test' function was added to the new node
assert_equal {{library_name TEST engine LUA description {} functions {{name test description {} flags {}}}}} [$node4_rd FUNCTION LIST]
assert_equal {{library_name TEST engine LUA functions {{name test description {} flags {}}}}} [$node4_rd FUNCTION LIST]
# add function to node 5
assert_equal {OK} [$node5_rd FUNCTION LOAD LUA TEST {redis.register_function('test', function() return 'hello' end)}]
assert_equal {TEST} [$node5_rd FUNCTION LOAD {#!lua name=TEST
redis.register_function('test', function() return 'hello' end)
}]
# make sure functions was added to node 5
assert_equal {{library_name TEST engine LUA description {} functions {{name test description {} flags {}}}}} [$node5_rd FUNCTION LIST]
assert_equal {{library_name TEST engine LUA functions {{name test description {} flags {}}}}} [$node5_rd FUNCTION LIST]
# adding node 5 to the cluster should failed because it already contains the 'test' function
catch {
......
This diff is collapsed.
......@@ -15,17 +15,25 @@ if {$is_eval == 1} {
}
} else {
proc run_script {args} {
r function load LUA test replace [format "redis.register_function('test', function(KEYS, ARGV)\n %s \nend)" [lindex $args 0]]
r function load replace [format "#!lua name=test\nredis.register_function('test', function(KEYS, ARGV)\n %s \nend)" [lindex $args 0]]
if {[r readingraw] eq 1} {
# read name
assert_equal {test} [r read]
}
r fcall test {*}[lrange $args 1 end]
}
proc run_script_ro {args} {
r function load LUA test replace [format "redis.register_function{function_name='test', callback=function(KEYS, ARGV)\n %s \nend, flags={'no-writes'}}" [lindex $args 0]]
r function load replace [format "#!lua name=test\nredis.register_function{function_name='test', callback=function(KEYS, ARGV)\n %s \nend, flags={'no-writes'}}" [lindex $args 0]]
if {[r readingraw] eq 1} {
# read name
assert_equal {test} [r read]
}
r fcall_ro test {*}[lrange $args 1 end]
}
proc run_script_on_connection {args} {
set rd [lindex $args 0]
$rd function load LUA test replace [format "redis.register_function('test', function(KEYS, ARGV)\n %s \nend)" [lindex $args 1]]
# read the ok reply of function create
$rd function load replace [format "#!lua name=test\nredis.register_function('test', function(KEYS, ARGV)\n %s \nend)" [lindex $args 1]]
# read name
$rd read
$rd fcall test {*}[lrange $args 2 end]
}
......@@ -784,7 +792,7 @@ start_server {tags {"scripting"}} {
set buf "*3\r\n\$4\r\neval\r\n\$33\r\nwhile 1 do redis.call('ping') end\r\n\$1\r\n0\r\n"
append buf "*1\r\n\$4\r\nping\r\n"
} else {
set buf "*6\r\n\$8\r\nfunction\r\n\$4\r\nload\r\n\$3\r\nlua\r\n\$4\r\ntest\r\n\$7\r\nreplace\r\n\$81\r\nredis.register_function('test', function() while 1 do redis.call('ping') end end)\r\n"
set buf "*4\r\n\$8\r\nfunction\r\n\$4\r\nload\r\n\$7\r\nreplace\r\n\$97\r\n#!lua name=test\nredis.register_function('test', function() while 1 do redis.call('ping') end end)\r\n"
append buf "*3\r\n\$5\r\nfcall\r\n\$4\r\ntest\r\n\$1\r\n0\r\n"
append buf "*1\r\n\$4\r\nping\r\n"
}
......@@ -808,8 +816,8 @@ start_server {tags {"scripting"}} {
assert_equal [r ping] "PONG"
if {$is_eval == 0} {
# read the ok reply of function create
assert_match {OK} [$rd read]
# read the function name
assert_match {test} [$rd read]
}
catch {$rd read} res
......
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