Unverified Commit 64f61596 authored by Oran Agra's avatar Oran Agra Committed by GitHub
Browse files

Merge Redis Functions PR (#9780)

# Redis Function
This PR added the Redis Functions capabilities that were suggested on #8693.
The PR also introduce a big refactoring to the current Lua implementation
(i.e `scripting.c`). The main purpose of the refactoring is to have better
code sharing between the Lua implementation that exists today on Redis
(`scripting.c`) and the new Lua engine that is introduced on this PR.
The refactoring includes code movements and file name changes as well as some
logic changes that need to be carefully reviewed. To make the review easier,
the PR was split into multiple commits. Each commit is deeply described later on
but the main concept is that some commits are just moving code around without
making any logical changes, those commits are less likely to cause any issues
or regressions and can be reviewed fast. Other commits, which perform code and
logic changes, need to be reviewed carefully, but those commits were created
after the code movements so it's pretty easy to see what was changed. To sum up,
it is highly recommended to review this PR commit by commit as it will be easier
to see the changes, it is also recommended to read each commit description
(written below) to understand what was changed on the commit and whether or not
it's just a huge code movement or a logic changes.

## Terminology
Currently, the terminology in Redis is not clearly defined. Scripts refer to Lua
scripts and eval also refers only to Lua. Introducing Redis Function requires
redefining those terms to be able to clearly understand what is been discussed
on each context.
* eval - legacy Lua script implementation.
* Function - new scripting implementation (currently implemented in Lua but in
  the future, it might be other languages like javascript).
* Engine - the component that is responsible for executing functions.
* Script - Function or legacy Lua (executed with `eval` or `evalsha`)

## Refactoring New Structure
Today, the entire scripting logic is located on `scripting.c`. This logic can
be split into 3 main groups:
1. Script management - responsible for storing the scripts that were sent to
   Redis and retrieving them when they need to be run (base on the script sha
   on the current implementation).
2. Script invocation - invoke the script given on `eval` or `evalsha` command
   (this part includes finding the relevant script, preparing the arguments, ..)
3. Interact back with Redis (command invocation)

Those 3 groups are tightly coupled on `scripting.c`. Redis Functions also need
to use those groups logics, for example,  to interact back with Redis or to
execute Lua code. The refactoring attempts to split those 3 groups and define
APIs so that we can reuse the code both on legacy Lua scripts and Redis Functions.

In order to do so we define the following units:
1. script.c: responsible for interaction with Redis from within a script.
2. script_lua.c: responsible to execute Lua code, uses `script.c` to interact
   with Redis from within the Lua code.
3. function_lua.c: contains the Lua engine implementation, uses `script_lua.c`
   to execute the Lua code.
4. functions.c: Contains Redis Functions implementation (`FUNCTION` command,),
   uses `functions_lua.c` if the function it wants to invoke needs the Lua
   engine.
4. eval.c: the original `scripting.c` contains the Lua legacy implementation and
   was refactored to use `script_lua.c` to invoke the Lua code.

## Commits breakdown
Notice: Some small commits are omitted from this list as they are small and
insignificant (for example build fixes)

### First commit - code movements
This commit rename `scripting.c` -> `eval.c` and introduce the new `script_lua.c`
unit. The commit moves relevant code from `eval.c` (`scripting.c`) to
`script_lua.c`, the purpose of moving the code is so that later we will be able
to re-use the code on the Lua engine (`function_lua.c`). The commit only moves
the code without modifying even a single line, so there is a very low risk of
breaking anything and it also makes it much easier to see the changes on the
following commits.
Because the commit does not change the code (only moves it), it does not compile.
But we do not care about it as the only purpose here is to make the review
processes simpler.

### Second commit - move legacy Lua variables into `eval.c`
Today, all Lua-related variables are located on the server struct. The commit
attempt to identify those variable and take them out from the server struct,
leaving only script related variables (variables that later need to be used
also by engines)
The following variable where renamed and left on the server struct:
   * lua_caller 			-> script_caller
   * lua_time_limit 		-> script_time_limit
   * lua_timedout 		-> script_timedout
   * lua_oom 			-> script_oom
   * lua_disable_deny_script 	-> script_disable_deny_script
   * in_eval			-> in_script

The following variables where moved to lctx under eval.c
   * lua
   * lua_client
   * lua_cur_script
   * lua_scripts
   * lua_scripts_mem
   * lua_replicate_commands
   * lua_write_dirty
   * lua_random_dirty
   * lua_multi_emitted
   * lua_repl
   * lua_kill
   * lua_time_start
   * lua_time_snapshot

This commit is in a low risk of introducing any issues and it is just moving
variables around and not changing any logic.

### Third commit - introducing script unit
This commit introduces the `script.c` unit. Its purpose (as described above) is
to provide an API for scripts to interact with Redis. Interaction includes
mostly executing commands, but also other functionalities. The interaction is
done using a `ScriptRunCtx` object that needs to be created by the user and
initialized using `scriptPrepareForRun`. A detailed list of functionalities
expose by the unit:
1. Calling commands (including all the validation checks such as
   acl, cluster, read only run, ...)
2. Set Resp
3. Set Replication method (AOF/REPLICATION/NONE)
4. Call Redis back on long-running scripts to allow Redis to reply to clients
   and perform script kill

The commit introduces the new unit and uses it on eval commands to interact with
Redis.

### Fourth commit - Moved functionality of invoke Lua code to `script_lua.c`
This commit moves the logic of invoking the Lua code into `script_lua.c` so
later it can be used also by Lua engine (`function_lua.c`). The code is located
on `callFunction` function and assumes the Lua function already located on the
top of the Lua stack. This commit also change `eval.c` to use the new
functionality to invoke Lua code.

### Fith commit - Added Redis Functions unit (`functions.c`) and Lua engine
(`function_lua.c`)
Added Redis Functions unit under `functions.c`, included:
1. FUNCTION command:
     * FUNCTION CREATE
     * FUNCTION CALL
     * FUNCTION DELETE
     * FUNCTION KILL
     * FUNCTION INFO
     * FUNCTION STATS
2. Register engines

In addition, this commit introduces the first engine that uses the Redis
Functions capabilities, the Lua engine (`function_lua.c`)

## API Changes
### `lua-time-limit`
configuration was renamed to `script-time-limit` (keep `lua-time-limit` as alias
for backward compatibility).

### Error log changes
When integrating with Redis from within a Lua script, the `Lua` term was removed
from all the error messages and instead we write only `script`. For example:
`Wrong number of args calling Redis command From Lua script` -> `Wrong number
of args calling Redis command From script`

### `info memory` changes:
Before stating all the changes made to memory stats we will try to explain the
reason behind them and what we want to see on those metrics:
* memory metrics should show both totals (for all scripting frameworks), as well
  as a breakdown per framework / vm.
* The totals metrics should have "human" metrics while the breakdown shouldn't.
* We did try to maintain backward compatibility in some way, that said we did
  make some repurpose to existing metrics where it looks reasonable.
* We separate between memory used by the script framework (part of redis's
  used_memory), and memory used by the VM (not part of redis's used_memory)

A full breakdown of `info memory` changes:
* `used_memory_lua` and `used_memory_lua_human` was deprecated,
  `used_memory_vm_eval` has the same meaning as `used_memory_lua`
* `used_memory_scripts` was renamed to `used_memory_scripts_eval`
* `used_memory_scripts` and `used_memory_scripts_human` were repurposed and now
  return the total memory used by functions and eval (not including vm memory,
  only code cache, and structs).
* `used_memory_vm_function` was added and represents the total memory used by
  functions vm's
* `used_memory_functions` was added and represents the total memory by functions
  (not including vm memory, only code cache, and structs)
* `used_memory_vm_total` and `used_memory_vm_total_human` was added and
  represents the total memory used by vm's (functions and eval combined)

### `functions.caches`
`functions.caches` field was added to `memory stats`, representing the memory
used by engines that are not functions (this memory includes data structures
like dictionaries, arrays, ...)

## New API
### FUNCTION CREATE

Usage: FUNCTION CREATE `ENGINE` `NAME` `[REPLACE]` `[DESC <DESCRIPTION>]` `<CODE>`

* `ENGINE` - The name of the engine to use to create the script.
* `NAME` - the name of the function that can be used later to call the function
  using `FUNCTION CALL` command.
* `REPLACE` - if given, replace the given function with the existing function
  (if exists).
* `DESCRIPTION` - optional argument describing the function and what it does
* `CODE` - function code.

The command will return `OK` if created successfully or error in the following
cases:
* The given engine name does not exist
* The function name is already taken and `REPLACE` was not used.
* The given function failed on the compilation.

### FCALL and FCALL_RO

Usage: FCALL/FCALL_RO `NAME` `NUM_KEYS key1 key2` … ` arg1 arg2`

Call and execute the function specified by `NAME`. The function will receive
all arguments given after `NUM_KEYS`. The return value from the function will
be returned to the user as a result.

* `NAME` - Name of the function to run.
* The rest is as today with EVALSHA command.

The command will return an error in the following cases:
* `NAME` does not exist
* The function itself returned an error.

The `FCALL_RO` is equivalent to `EVAL_RO` and allows only read-only commands to
be invoked from the script.

### FUNCTION DELETE

Usage: FUNCTION DELETE `NAME`

Delete a function identified by `NAME`. Return `OK` on success or error on one
of the following:
* The given function does not exist

### FUNCTION INFO

Usage: FUNCTION INFO `NAME` [WITHCODE]

Return information about a function by function name:
* Function name
* Engine name
* Description
* Raw code (only if WITHCODE argument is given)

### FUNCTION LIST

Usage: FUNCTION LIST

Return general information about all the functions:
* Function name
* Engine name
* Description

### FUNCTION STATS

Usage: FUNCTION STATS

Return information about the current running function:
* Function name
* Command that was used to invoke the function
* Duration in MS that the function is already running

If no function is currently running, this section is just a RESP nil.

Additionally, return a list of all the available engines.

### FUNCTION KILL

Usage: `FUNCTION KILL`

Kill the currently executing function. The command will fail if the function
already initiated a write command.

## Notes
Note: Function creation/deletion is replicated to AOF but AOFRW is not
implemented sense its going to be removed: #9794
parents e57a4db5 cbd46317
...@@ -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 scripting.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 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)
......
...@@ -1897,7 +1897,7 @@ void addACLLogEntry(client *c, int reason, int context, int argpos, sds username ...@@ -1897,7 +1897,7 @@ void addACLLogEntry(client *c, int reason, int context, int argpos, sds username
} }
client *realclient = c; client *realclient = c;
if (realclient->flags & CLIENT_LUA) realclient = server.lua_caller; if (realclient->flags & CLIENT_SCRIPT) realclient = server.script_caller;
le->cinfo = catClientInfoString(sdsempty(),realclient); le->cinfo = catClientInfoString(sdsempty(),realclient);
le->context = context; le->context = context;
......
...@@ -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 {
......
...@@ -2650,7 +2650,7 @@ standardConfig configs[] = { ...@@ -2650,7 +2650,7 @@ standardConfig configs[] = {
createULongConfig("acllog-max-len", NULL, MODIFIABLE_CONFIG, 0, LONG_MAX, server.acllog_max_len, 128, INTEGER_CONFIG, NULL, NULL), createULongConfig("acllog-max-len", NULL, MODIFIABLE_CONFIG, 0, LONG_MAX, server.acllog_max_len, 128, INTEGER_CONFIG, NULL, NULL),
/* Long Long configs */ /* Long Long configs */
createLongLongConfig("lua-time-limit", NULL, MODIFIABLE_CONFIG, 0, LONG_MAX, server.lua_time_limit, 5000, INTEGER_CONFIG, NULL, NULL),/* milliseconds */ createLongLongConfig("script-time-limit", "lua-time-limit", MODIFIABLE_CONFIG, 0, LONG_MAX, server.script_time_limit, 5000, INTEGER_CONFIG, NULL, NULL),/* milliseconds */
createLongLongConfig("cluster-node-timeout", NULL, MODIFIABLE_CONFIG, 0, LLONG_MAX, server.cluster_node_timeout, 15000, INTEGER_CONFIG, NULL, NULL), createLongLongConfig("cluster-node-timeout", NULL, MODIFIABLE_CONFIG, 0, LLONG_MAX, server.cluster_node_timeout, 15000, INTEGER_CONFIG, NULL, NULL),
createLongLongConfig("slowlog-log-slower-than", NULL, MODIFIABLE_CONFIG, -1, LLONG_MAX, server.slowlog_log_slower_than, 10000, INTEGER_CONFIG, NULL, NULL), createLongLongConfig("slowlog-log-slower-than", NULL, MODIFIABLE_CONFIG, -1, LLONG_MAX, server.slowlog_log_slower_than, 10000, INTEGER_CONFIG, NULL, NULL),
createLongLongConfig("latency-monitor-threshold", NULL, MODIFIABLE_CONFIG, 0, LLONG_MAX, server.latency_monitor_threshold, 0, INTEGER_CONFIG, NULL, NULL), createLongLongConfig("latency-monitor-threshold", NULL, MODIFIABLE_CONFIG, 0, LLONG_MAX, server.latency_monitor_threshold, 0, INTEGER_CONFIG, NULL, NULL),
......
...@@ -31,6 +31,7 @@ ...@@ -31,6 +31,7 @@
#include "cluster.h" #include "cluster.h"
#include "atomicvar.h" #include "atomicvar.h"
#include "latency.h" #include "latency.h"
#include "script.h"
#include <signal.h> #include <signal.h>
#include <ctype.h> #include <ctype.h>
...@@ -88,7 +89,7 @@ robj *lookupKey(redisDb *db, robj *key, int flags) { ...@@ -88,7 +89,7 @@ robj *lookupKey(redisDb *db, robj *key, int flags) {
* commands is to make writable replicas behave consistently. It * commands is to make writable replicas behave consistently. It
* shall not be used in readonly commands. Modules are accepted so * shall not be used in readonly commands. Modules are accepted so
* that we don't break old modules. */ * that we don't break old modules. */
client *c = server.in_eval ? server.lua_client : server.current_client; client *c = server.in_script ? scriptGetClient() : server.current_client;
serverAssert(!c || !c->cmd || (c->cmd->flags & (CMD_WRITE|CMD_MODULE))); serverAssert(!c || !c->cmd || (c->cmd->flags & (CMD_WRITE|CMD_MODULE)));
} }
if (expireIfNeeded(db, key, force_delete_expired)) { if (expireIfNeeded(db, key, force_delete_expired)) {
...@@ -1501,8 +1502,8 @@ int keyIsExpired(redisDb *db, robj *key) { ...@@ -1501,8 +1502,8 @@ int keyIsExpired(redisDb *db, robj *key) {
* only the first time it is accessed and not in the middle of the * only the first time it is accessed and not in the middle of the
* script execution, making propagation to slaves / AOF consistent. * script execution, making propagation to slaves / AOF consistent.
* See issue #1525 on Github for more information. */ * See issue #1525 on Github for more information. */
if (server.lua_caller) { if (server.script_caller) {
now = server.lua_time_snapshot; now = evalTimeSnapshot();
} }
/* If we are in the middle of a command execution, we still want to use /* If we are in the middle of a command execution, we still want to use
* a reference time that does not change: in that case we just use the * a reference time that does not change: in that case we just use the
...@@ -1743,6 +1744,11 @@ int evalGetKeys(struct redisCommand *cmd, robj **argv, int argc, getKeysResult * ...@@ -1743,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);
......
...@@ -919,7 +919,7 @@ NULL ...@@ -919,7 +919,7 @@ NULL
addReplyStatus(c,"Apparently Redis did not crash: test passed"); addReplyStatus(c,"Apparently Redis did not crash: test passed");
} else if (!strcasecmp(c->argv[1]->ptr,"set-disable-deny-scripts") && c->argc == 3) } else if (!strcasecmp(c->argv[1]->ptr,"set-disable-deny-scripts") && c->argc == 3)
{ {
server.lua_disable_deny_script = atoi(c->argv[2]->ptr);; server.script_disable_deny_script = atoi(c->argv[2]->ptr);;
addReply(c,shared.ok); addReply(c,shared.ok);
} else if (!strcasecmp(c->argv[1]->ptr,"config-rewrite-force-all") && c->argc == 2) } else if (!strcasecmp(c->argv[1]->ptr,"config-rewrite-force-all") && c->argc == 2)
{ {
......
...@@ -939,7 +939,7 @@ long defragOtherGlobals() { ...@@ -939,7 +939,7 @@ long defragOtherGlobals() {
/* there are many more pointers to defrag (e.g. client argv, output / aof buffers, etc. /* there are many more pointers to defrag (e.g. client argv, output / aof buffers, etc.
* but we assume most of these are short lived, we only need to defrag allocations * but we assume most of these are short lived, we only need to defrag allocations
* that remain static for a long time */ * that remain static for a long time */
defragged += activeDefragSdsDict(server.lua_scripts, DEFRAG_SDS_DICT_VAL_IS_STROB); defragged += activeDefragSdsDict(evalScriptsDict(), DEFRAG_SDS_DICT_VAL_IS_STROB);
defragged += activeDefragSdsListAndDict(server.repl_scriptcache_fifo, server.repl_scriptcache_dict, DEFRAG_SDS_DICT_NO_VAL); defragged += activeDefragSdsListAndDict(server.repl_scriptcache_fifo, server.repl_scriptcache_dict, DEFRAG_SDS_DICT_NO_VAL);
defragged += moduleDefragGlobals(); defragged += moduleDefragGlobals();
return defragged; return defragged;
......
This diff is collapsed.
...@@ -33,6 +33,7 @@ ...@@ -33,6 +33,7 @@
#include "server.h" #include "server.h"
#include "bio.h" #include "bio.h"
#include "atomicvar.h" #include "atomicvar.h"
#include "script.h"
#include <math.h> #include <math.h>
/* ---------------------------------------------------------------------------- /* ----------------------------------------------------------------------------
...@@ -472,7 +473,7 @@ static int evictionTimeProc( ...@@ -472,7 +473,7 @@ static int evictionTimeProc(
static int isSafeToPerformEvictions(void) { static int isSafeToPerformEvictions(void) {
/* - There must be no script in timeout condition. /* - There must be no script in timeout condition.
* - Nor we are loading data right now. */ * - Nor we are loading data right now. */
if (server.lua_timedout || server.loading) return 0; if (scriptIsTimedout() || server.loading) return 0;
/* By default replicas should ignore maxmemory /* By default replicas should ignore maxmemory
* and just be masters exact copies. */ * and just be masters exact copies. */
......
/*
* 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_ */
...@@ -629,7 +629,7 @@ void moduleHandlePropagationAfterCommandCallback(RedisModuleCtx *ctx) { ...@@ -629,7 +629,7 @@ void moduleHandlePropagationAfterCommandCallback(RedisModuleCtx *ctx) {
/* If this command is executed from with Lua or MULTI/EXEC we do not /* If this command is executed from with Lua or MULTI/EXEC we do not
* need to propagate EXEC */ * need to propagate EXEC */
if (server.in_eval || server.in_exec) return; if (server.in_script || server.in_exec) return;
/* Handle the replication of the final EXEC, since whatever a command /* Handle the replication of the final EXEC, since whatever a command
* emits is always wrapped around MULTI/EXEC. */ * emits is always wrapped around MULTI/EXEC. */
...@@ -2333,7 +2333,7 @@ int RM_ReplyWithLongDouble(RedisModuleCtx *ctx, long double ld) { ...@@ -2333,7 +2333,7 @@ int RM_ReplyWithLongDouble(RedisModuleCtx *ctx, long double ld) {
void moduleReplicateMultiIfNeeded(RedisModuleCtx *ctx) { void moduleReplicateMultiIfNeeded(RedisModuleCtx *ctx) {
/* Skip this if client explicitly wrap the command with MULTI, or if /* Skip this if client explicitly wrap the command with MULTI, or if
* the module command was called by a script. */ * the module command was called by a script. */
if (server.in_eval || server.in_exec) return; if (server.in_script || server.in_exec) return;
/* If we already emitted MULTI return ASAP. */ /* If we already emitted MULTI return ASAP. */
if (server.propagate_in_transaction) return; if (server.propagate_in_transaction) return;
/* If this is a thread safe context, we do not want to wrap commands /* If this is a thread safe context, we do not want to wrap commands
...@@ -2709,7 +2709,7 @@ int RM_GetContextFlags(RedisModuleCtx *ctx) { ...@@ -2709,7 +2709,7 @@ int RM_GetContextFlags(RedisModuleCtx *ctx) {
} }
} }
if (server.in_eval) if (server.in_script)
flags |= REDISMODULE_CTX_FLAGS_LUA; flags |= REDISMODULE_CTX_FLAGS_LUA;
if (server.in_exec) if (server.in_exec)
...@@ -6215,7 +6215,7 @@ void unblockClientFromModule(client *c) { ...@@ -6215,7 +6215,7 @@ void unblockClientFromModule(client *c) {
*/ */
RedisModuleBlockedClient *moduleBlockClient(RedisModuleCtx *ctx, RedisModuleCmdFunc reply_callback, RedisModuleCmdFunc timeout_callback, void (*free_privdata)(RedisModuleCtx*,void*), long long timeout_ms, RedisModuleString **keys, int numkeys, void *privdata) { RedisModuleBlockedClient *moduleBlockClient(RedisModuleCtx *ctx, RedisModuleCmdFunc reply_callback, RedisModuleCmdFunc timeout_callback, void (*free_privdata)(RedisModuleCtx*,void*), long long timeout_ms, RedisModuleString **keys, int numkeys, void *privdata) {
client *c = ctx->client; client *c = ctx->client;
int islua = server.in_eval; int islua = server.in_script;
int ismulti = server.in_exec; int ismulti = server.in_exec;
c->bpop.module_blocked_handle = zmalloc(sizeof(RedisModuleBlockedClient)); c->bpop.module_blocked_handle = zmalloc(sizeof(RedisModuleBlockedClient));
......
...@@ -30,6 +30,7 @@ ...@@ -30,6 +30,7 @@
#include "server.h" #include "server.h"
#include "atomicvar.h" #include "atomicvar.h"
#include "cluster.h" #include "cluster.h"
#include "script.h"
#include <sys/socket.h> #include <sys/socket.h>
#include <sys/uio.h> #include <sys/uio.h>
#include <math.h> #include <math.h>
...@@ -260,7 +261,7 @@ void clientInstallWriteHandler(client *c) { ...@@ -260,7 +261,7 @@ void clientInstallWriteHandler(client *c) {
int prepareClientToWrite(client *c) { int prepareClientToWrite(client *c) {
/* If it's the Lua client we always return ok without installing any /* If it's the Lua client we always return ok without installing any
* handler since there is no socket at all. */ * handler since there is no socket at all. */
if (c->flags & (CLIENT_LUA|CLIENT_MODULE)) return C_OK; if (c->flags & (CLIENT_SCRIPT|CLIENT_MODULE)) return C_OK;
/* If CLIENT_CLOSE_ASAP flag is set, we need not write anything. */ /* If CLIENT_CLOSE_ASAP flag is set, we need not write anything. */
if (c->flags & CLIENT_CLOSE_ASAP) return C_ERR; if (c->flags & CLIENT_CLOSE_ASAP) return C_ERR;
...@@ -1491,7 +1492,7 @@ void freeClientAsync(client *c) { ...@@ -1491,7 +1492,7 @@ void freeClientAsync(client *c) {
* may access the list while Redis uses I/O threads. All the other accesses * may access the list while Redis uses I/O threads. All the other accesses
* are in the context of the main thread while the other threads are * are in the context of the main thread while the other threads are
* idle. */ * idle. */
if (c->flags & CLIENT_CLOSE_ASAP || c->flags & CLIENT_LUA) return; if (c->flags & CLIENT_CLOSE_ASAP || c->flags & CLIENT_SCRIPT) return;
c->flags |= CLIENT_CLOSE_ASAP; c->flags |= CLIENT_CLOSE_ASAP;
if (server.io_threads_num == 1) { if (server.io_threads_num == 1) {
/* no need to bother with locking if there's just one thread (the main thread) */ /* no need to bother with locking if there's just one thread (the main thread) */
...@@ -2199,7 +2200,7 @@ int processInputBuffer(client *c) { ...@@ -2199,7 +2200,7 @@ int processInputBuffer(client *c) {
* condition on the slave. We want just to accumulate the replication * condition on the slave. We want just to accumulate the replication
* stream (instead of replying -BUSY like we do with other clients) and * stream (instead of replying -BUSY like we do with other clients) and
* later resume the processing. */ * later resume the processing. */
if (server.lua_timedout && c->flags & CLIENT_MASTER) break; if (scriptIsTimedout() && c->flags & CLIENT_MASTER) break;
/* CLIENT_CLOSE_AFTER_REPLY closes the connection once the reply is /* CLIENT_CLOSE_AFTER_REPLY closes the connection once the reply is
* written to the client. Make sure to not let the reply grow after * written to the client. Make sure to not let the reply grow after
......
...@@ -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>
...@@ -1203,9 +1204,7 @@ struct redisMemOverhead *getMemoryOverheadData(void) { ...@@ -1203,9 +1204,7 @@ struct redisMemOverhead *getMemoryOverheadData(void) {
mh->aof_buffer = mem; mh->aof_buffer = mem;
mem_total+=mem; mem_total+=mem;
mem = server.lua_scripts_mem; mem = evalScriptsMemory();
mem += dictSize(server.lua_scripts) * sizeof(dictEntry) +
dictSlots(server.lua_scripts) * sizeof(dictEntry*);
mem += dictSize(server.repl_scriptcache_dict) * sizeof(dictEntry) + mem += dictSize(server.repl_scriptcache_dict) * sizeof(dictEntry) +
dictSlots(server.repl_scriptcache_dict) * sizeof(dictEntry*); dictSlots(server.repl_scriptcache_dict) * sizeof(dictEntry*);
if (listLength(server.repl_scriptcache_fifo) > 0) { if (listLength(server.repl_scriptcache_fifo) > 0) {
...@@ -1214,6 +1213,8 @@ struct redisMemOverhead *getMemoryOverheadData(void) { ...@@ -1214,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;
...@@ -1325,7 +1326,7 @@ sds getMemoryDoctorReport(void) { ...@@ -1325,7 +1326,7 @@ sds getMemoryDoctorReport(void) {
} }
/* Too many scripts are cached? */ /* Too many scripts are cached? */
if (dictSize(server.lua_scripts) > 1000) { if (dictSize(evalScriptsDict()) > 1000) {
many_scripts = 1; many_scripts = 1;
num_reports++; num_reports++;
} }
...@@ -1529,7 +1530,7 @@ NULL ...@@ -1529,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);
...@@ -1555,6 +1556,9 @@ NULL ...@@ -1555,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;
...@@ -1303,8 +1323,8 @@ int rdbSaveRio(rio *rdb, int *error, int rdbflags, rdbSaveInfo *rsi) { ...@@ -1303,8 +1323,8 @@ int rdbSaveRio(rio *rdb, int *error, int rdbflags, rdbSaveInfo *rsi) {
* the script cache as well: on successful PSYNC after a restart, we need * the script cache as well: on successful PSYNC after a restart, we need
* to be able to process any EVALSHA inside the replication backlog the * to be able to process any EVALSHA inside the replication backlog the
* master will send us. */ * master will send us. */
if (rsi && dictSize(server.lua_scripts)) { if (rsi && dictSize(evalScriptsDict())) {
di = dictGetIterator(server.lua_scripts); di = dictGetIterator(evalScriptsDict());
while((de = dictNext(di)) != NULL) { while((de = dictNext(di)) != NULL) {
robj *body = dictGetVal(de); robj *body = dictGetVal(de);
if (rdbSaveAuxField(rdb,"lua",3,body->ptr,sdslen(body->ptr)) == -1) if (rdbSaveAuxField(rdb,"lua",3,body->ptr,sdslen(body->ptr)) == -1)
...@@ -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
...@@ -2808,7 +2896,7 @@ int rdbLoadRio(rio *rdb, int rdbflags, rdbSaveInfo *rsi, redisDb *dbarray) { ...@@ -2808,7 +2896,7 @@ int rdbLoadRio(rio *rdb, int rdbflags, rdbSaveInfo *rsi, redisDb *dbarray) {
if (rsi) rsi->repl_offset = strtoll(auxval->ptr,NULL,10); if (rsi) rsi->repl_offset = strtoll(auxval->ptr,NULL,10);
} else if (!strcasecmp(auxkey->ptr,"lua")) { } else if (!strcasecmp(auxkey->ptr,"lua")) {
/* Load the script back in memory. */ /* Load the script back in memory. */
if (luaCreateFunction(NULL,server.lua,auxval) == NULL) { if (luaCreateFunction(NULL, auxval) == NULL) {
rdbReportCorruptRDB( rdbReportCorruptRDB(
"Can't load Lua script from RDB file! " "Can't load Lua script from RDB file! "
"BODY: %s", (char*)auxval->ptr); "BODY: %s", (char*)auxval->ptr);
...@@ -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>
...@@ -546,7 +547,7 @@ void replicationFeedMonitors(client *c, list *monitors, int dictid, robj **argv, ...@@ -546,7 +547,7 @@ void replicationFeedMonitors(client *c, list *monitors, int dictid, robj **argv,
gettimeofday(&tv,NULL); gettimeofday(&tv,NULL);
cmdrepr = sdscatprintf(cmdrepr,"%ld.%06ld ",(long)tv.tv_sec,(long)tv.tv_usec); cmdrepr = sdscatprintf(cmdrepr,"%ld.%06ld ",(long)tv.tv_sec,(long)tv.tv_usec);
if (c->flags & CLIENT_LUA) { if (c->flags & CLIENT_SCRIPT) {
cmdrepr = sdscatprintf(cmdrepr,"[%d lua] ",dictid); cmdrepr = sdscatprintf(cmdrepr,"[%d lua] ",dictid);
} else if (c->flags & CLIENT_UNIX_SOCKET) { } else if (c->flags & CLIENT_UNIX_SOCKET) {
cmdrepr = sdscatprintf(cmdrepr,"[%d unix:%s] ",dictid,server.unixsocket); cmdrepr = sdscatprintf(cmdrepr,"[%d unix:%s] ",dictid,server.unixsocket);
...@@ -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);
......
/*
* Copyright (c) 2009-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 "server.h"
#include "script.h"
#include "cluster.h"
/* On script invocation, holding the current run context */
static scriptRunCtx *curr_run_ctx = NULL;
static void exitScriptTimedoutMode(scriptRunCtx *run_ctx) {
serverAssert(run_ctx == curr_run_ctx);
serverAssert(scriptIsTimedout());
run_ctx->flags &= ~SCRIPT_TIMEDOUT;
blockingOperationEnds();
/* if we are a replica and we have an active master, set it for continue processing */
if (server.masterhost && server.master) queueClientForReprocessing(server.master);
}
static void enterScriptTimedoutMode(scriptRunCtx *run_ctx) {
serverAssert(run_ctx == curr_run_ctx);
serverAssert(!scriptIsTimedout());
/* Mark script as timedout */
run_ctx->flags |= SCRIPT_TIMEDOUT;
blockingOperationStarts();
}
int scriptIsTimedout() {
return scriptIsRunning() && (curr_run_ctx->flags & SCRIPT_TIMEDOUT);
}
client* scriptGetClient() {
serverAssert(scriptIsRunning());
return curr_run_ctx->c;
}
client* scriptGetCaller() {
serverAssert(scriptIsRunning());
return curr_run_ctx->original_client;
}
/* interrupt function for scripts, should be call
* from time to time to reply some special command (like ping)
* and also check if the run should be terminated. */
int scriptInterrupt(scriptRunCtx *run_ctx) {
if (run_ctx->flags & SCRIPT_TIMEDOUT) {
/* script already timedout
we just need to precess some events and return */
processEventsWhileBlocked();
return (run_ctx->flags & SCRIPT_KILLED) ? SCRIPT_KILL : SCRIPT_CONTINUE;
}
long long elapsed = elapsedMs(run_ctx->start_time);
if (elapsed < server.script_time_limit) {
return SCRIPT_CONTINUE;
}
serverLog(LL_WARNING,
"Slow script detected: still in execution after %lld milliseconds. "
"You can try killing the script using the %s command.",
elapsed, (run_ctx->flags & SCRIPT_EVAL_MODE) ? "SCRIPT KILL" : "FUNCTION KILL");
enterScriptTimedoutMode(run_ctx);
/* Once the script timeouts we reenter the event loop to permit others
* some commands execution. For this reason
* we need to mask the client executing the script from the event loop.
* If we don't do that the client may disconnect and could no longer be
* here when the EVAL command will return. */
protectClient(run_ctx->original_client);
processEventsWhileBlocked();
return (run_ctx->flags & SCRIPT_KILLED) ? SCRIPT_KILL : SCRIPT_CONTINUE;
}
/* Prepare the given run ctx for execution */
void scriptPrepareForRun(scriptRunCtx *run_ctx, client *engine_client, client *caller, const char *funcname) {
serverAssert(!curr_run_ctx);
/* set the curr_run_ctx so we can use it to kill the script if needed */
curr_run_ctx = run_ctx;
run_ctx->c = engine_client;
run_ctx->original_client = caller;
run_ctx->funcname = funcname;
client *script_client = run_ctx->c;
client *curr_client = run_ctx->original_client;
server.script_caller = curr_client;
/* Select the right DB in the context of the Lua client */
selectDb(script_client, curr_client->db->id);
script_client->resp = 2; /* Default is RESP2, scripts can change it. */
/* If we are in MULTI context, flag Lua client as CLIENT_MULTI. */
if (curr_client->flags & CLIENT_MULTI) {
script_client->flags |= CLIENT_MULTI;
}
server.in_script = 1;
run_ctx->start_time = getMonotonicUs();
run_ctx->snapshot_time = mstime();
run_ctx->flags = 0;
run_ctx->repl_flags = PROPAGATE_AOF | PROPAGATE_REPL;
}
/* Reset the given run ctx after execution */
void scriptResetRun(scriptRunCtx *run_ctx) {
serverAssert(curr_run_ctx);
/* After the script done, remove the MULTI state. */
run_ctx->c->flags &= ~CLIENT_MULTI;
server.in_script = 0;
server.script_caller = NULL;
if (scriptIsTimedout()) {
exitScriptTimedoutMode(run_ctx);
/* Restore the client that was protected when the script timeout
* was detected. */
unprotectClient(run_ctx->original_client);
}
if (!(run_ctx->flags & SCRIPT_EVAL_REPLICATION)) {
preventCommandPropagation(run_ctx->original_client);
if (run_ctx->flags & SCRIPT_MULTI_EMMITED) {
execCommandPropagateExec(run_ctx->original_client->db->id);
}
}
/* unset curr_run_ctx so we will know there is no running script */
curr_run_ctx = NULL;
}
/* return true if a script is currently running */
int scriptIsRunning() {
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 */
void scriptKill(client *c, int is_eval) {
if (!curr_run_ctx) {
addReplyError(c, "-NOTBUSY No scripts in execution right now.");
return;
}
if (curr_run_ctx->original_client->flags & CLIENT_MASTER) {
addReplyError(c,
"-UNKILLABLE The busy script was sent by a master instance in the context of replication and cannot be killed.");
}
if (curr_run_ctx->flags & SCRIPT_WRITE_DIRTY) {
addReplyError(c,
"-UNKILLABLE Sorry the script already executed write "
"commands against the dataset. You can either wait the "
"script termination or kill the server in a hard way "
"using the SHUTDOWN NOSAVE command.");
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;
addReply(c, shared.ok);
}
static int scriptVerifyCommandArity(struct redisCommand *cmd, int argc, sds *err) {
if (!cmd || ((cmd->arity > 0 && cmd->arity != argc) || (argc < cmd->arity))) {
if (cmd)
*err = sdsnew("Wrong number of args calling Redis command from script");
else
*err = sdsnew("Unknown Redis command called from script");
return C_ERR;
}
return C_OK;
}
static int scriptVerifyACL(client *c, sds *err) {
/* Check the ACLs. */
int acl_errpos;
int acl_retval = ACLCheckAllPerm(c, &acl_errpos);
if (acl_retval != ACL_OK) {
addACLLogEntry(c,acl_retval,ACL_LOG_CTX_LUA,acl_errpos,NULL,NULL);
switch (acl_retval) {
case ACL_DENIED_CMD:
*err = sdsnew("The user executing the script can't run this "
"command or subcommand");
break;
case ACL_DENIED_KEY:
*err = sdsnew("The user executing the script can't access "
"at least one of the keys mentioned in the "
"command arguments");
break;
case ACL_DENIED_CHANNEL:
*err = sdsnew("The user executing the script can't publish "
"to the channel mentioned in the command");
break;
default:
*err = sdsnew("The user executing the script is lacking the "
"permissions for the command");
break;
}
return C_ERR;
}
return C_OK;
}
static int scriptVerifyWriteCommandAllow(scriptRunCtx *run_ctx, char **err) {
if (!(run_ctx->c->cmd->flags & CMD_WRITE)) {
return C_OK;
}
if (run_ctx->flags & SCRIPT_READ_ONLY) {
/* We know its a write command, on a read only run we do not allow it. */
*err = sdsnew("Write commands are not allowed from read-only scripts.");
return C_ERR;
}
if ((run_ctx->flags & SCRIPT_RANDOM_DIRTY) && (run_ctx->flags & SCRIPT_EVAL_REPLICATION)) {
*err = sdsnew("Write commands not allowed after non deterministic commands. Call redis.replicate_commands() at the start of your script in order to switch to single commands replication mode.");
return C_ERR;
}
/* Write commands are forbidden against read-only slaves, or if a
* command marked as non-deterministic was already called in the context
* of this script. */
int deny_write_type = writeCommandsDeniedByDiskError();
if (server.masterhost && server.repl_slave_ro && run_ctx->original_client->flags != CLIENT_ID_AOF
&& !(run_ctx->original_client->flags & CLIENT_MASTER))
{
*err = sdsdup(shared.roslaveerr->ptr);
return C_ERR;
}
if (deny_write_type != DISK_ERROR_TYPE_NONE) {
if (deny_write_type == DISK_ERROR_TYPE_RDB) {
*err = sdsdup(shared.bgsaveerr->ptr);
} else {
*err = sdsempty();
*err = sdscatfmt(*err,
"MISCONF Errors writing to the AOF file: %s\r\n",
strerror(server.aof_last_write_errno));
}
return C_ERR;
}
return C_OK;
}
static int scriptVerifyOOM(scriptRunCtx *run_ctx, char **err) {
/* If we reached the memory limit configured via maxmemory, commands that
* could enlarge the memory usage are not allowed, but only if this is the
* first write in the context of this script, otherwise we can't stop
* in the middle. */
if (server.maxmemory && /* Maxmemory is actually enabled. */
run_ctx->original_client->id != CLIENT_ID_AOF && /* Don't care about mem if loading from AOF. */
!server.masterhost && /* Slave must execute the script. */
!(run_ctx->flags & SCRIPT_WRITE_DIRTY) && /* Script had no side effects so far. */
server.script_oom && /* Detected OOM when script start. */
(run_ctx->c->cmd->flags & CMD_DENYOOM))
{
*err = sdsdup(shared.oomerr->ptr);
return C_ERR;
}
return C_OK;
}
static int scriptVerifyClusterState(client *c, client *original_c, sds *err) {
if (!server.cluster_enabled || original_c->id == CLIENT_ID_AOF || (original_c->flags & CLIENT_MASTER)) {
return C_OK;
}
/* If this is a Redis Cluster node, we need to make sure the script is not
* trying to access non-local keys, with the exception of commands
* received from our master or when loading the AOF back in memory. */
int error_code;
/* Duplicate relevant flags in the script client. */
c->flags &= ~(CLIENT_READONLY | CLIENT_ASKING);
c->flags |= original_c->flags & (CLIENT_READONLY | CLIENT_ASKING);
if (getNodeByQuery(c, c->cmd, c->argv, c->argc, NULL, &error_code) != server.cluster->myself) {
if (error_code == CLUSTER_REDIR_DOWN_RO_STATE) {
*err = sdsnew(
"Script attempted to execute a write command while the "
"cluster is down and readonly");
} else if (error_code == CLUSTER_REDIR_DOWN_STATE) {
*err = sdsnew("Script attempted to execute a command while the "
"cluster is down");
} else {
*err = sdsnew("Script attempted to access a non local key in a "
"cluster node");
}
return C_ERR;
}
return C_OK;
}
static void scriptEmitMultiIfNeeded(scriptRunCtx *run_ctx) {
/* If we are using single commands replication, we need to wrap what
* we propagate into a MULTI/EXEC block, so that it will be atomic like
* a Lua script in the context of AOF and slaves. */
client *c = run_ctx->c;
if (!(run_ctx->flags & SCRIPT_EVAL_REPLICATION)
&& !(run_ctx->flags & SCRIPT_MULTI_EMMITED)
&& !(run_ctx->original_client->flags & CLIENT_MULTI)
&& (run_ctx->flags & SCRIPT_WRITE_DIRTY)
&& ((run_ctx->repl_flags & PROPAGATE_AOF)
|| (run_ctx->repl_flags & PROPAGATE_REPL)))
{
execCommandPropagateMulti(run_ctx->original_client->db->id);
run_ctx->flags |= SCRIPT_MULTI_EMMITED;
/* Now we are in the MULTI context, the lua_client should be
* flag as CLIENT_MULTI. */
c->flags |= CLIENT_MULTI;
}
}
/* set RESP for a given run_ctx */
int scriptSetResp(scriptRunCtx *run_ctx, int resp) {
if (resp != 2 && resp != 3) {
return C_ERR;
}
run_ctx->c->resp = resp;
return C_OK;
}
/* set Repl for a given run_ctx
* either: PROPAGATE_AOF | PROPAGATE_REPL*/
int scriptSetRepl(scriptRunCtx *run_ctx, int repl) {
if ((repl & ~(PROPAGATE_AOF | PROPAGATE_REPL)) != 0) {
return C_ERR;
}
run_ctx->repl_flags = repl;
return C_OK;
}
/* Call a Redis command.
* The reply is written to the run_ctx client and it is
* up to the engine to take and parse.
* The err out variable is set only if error occurs and describe the error.
* If err is set on reply is written to the run_ctx client. */
void scriptCall(scriptRunCtx *run_ctx, robj* *argv, int argc, sds *err) {
client *c = run_ctx->c;
/* Setup our fake client for command execution */
c->argv = argv;
c->argc = argc;
c->user = run_ctx->original_client->user;
/* Process module hooks */
moduleCallCommandFilters(c);
argv = c->argv;
argc = c->argc;
struct redisCommand *cmd = lookupCommand(argv, argc);
if (scriptVerifyCommandArity(cmd, argc, err) != C_OK) {
return;
}
c->cmd = c->lastcmd = cmd;
/* There are commands that are not allowed inside scripts. */
if (!server.script_disable_deny_script && (cmd->flags & CMD_NOSCRIPT)) {
*err = sdsnew("This Redis command is not allowed from script");
return;
}
if (scriptVerifyACL(c, err) != C_OK) {
return;
}
if (scriptVerifyWriteCommandAllow(run_ctx, err) != C_OK) {
return;
}
if (scriptVerifyOOM(run_ctx, err) != C_OK) {
return;
}
if (cmd->flags & CMD_WRITE) {
/* signify that we already change the data in this execution */
run_ctx->flags |= SCRIPT_WRITE_DIRTY;
}
if (cmd->flags & CMD_RANDOM) {
/* signify that we already perform a random command in this execution */
run_ctx->flags |= SCRIPT_RANDOM_DIRTY;
}
if (scriptVerifyClusterState(c, run_ctx->original_client, err) != C_OK) {
return;
}
scriptEmitMultiIfNeeded(run_ctx);
int call_flags = CMD_CALL_SLOWLOG | CMD_CALL_STATS;
if (!(run_ctx->flags & SCRIPT_EVAL_REPLICATION)) {
if (run_ctx->repl_flags & PROPAGATE_AOF) {
call_flags |= CMD_CALL_PROPAGATE_AOF;
}
if (run_ctx->repl_flags & PROPAGATE_REPL) {
call_flags |= CMD_CALL_PROPAGATE_REPL;
}
}
call(c, call_flags);
serverAssert((c->flags & CLIENT_BLOCKED) == 0);
}
/* Returns the time when the script invocation started */
mstime_t scriptTimeSnapshot() {
serverAssert(!curr_run_ctx);
return curr_run_ctx->snapshot_time;
}
long long scriptRunDuration() {
serverAssert(scriptIsRunning());
return elapsedMs(curr_run_ctx->start_time);
}
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