gvsafronov/futriix
1
0
Fork 0
Code Issues Pull Requests Actions 4 Packages Projects Releases Wiki Activity
futriix/src/eval.c
uriyage bbfd041895
Async IO threads (#758) 
This PR is 1 of 3 PRs intended to achieve the goal of 1 million requests
per second, as detailed by [dan touitou](https://github.com/touitou-dan)
in https://github.com/valkey-io/valkey/issues/22. This PR modifies the
IO threads to be fully asynchronous, which is a first and necessary step
to allow more work offloading and better utilization of the IO threads.

### Current IO threads state:

Valkey IO threads were introduced in Redis 6.0 to allow better
utilization of multi-core machines. Before this, Redis was
single-threaded and could only use one CPU core for network and command
processing. The introduction of IO threads helps in offloading the IO
operations to multiple threads.

**Current IO Threads flow:**

1. Initialization: When Redis starts, it initializes a specified number
of IO threads. These threads are in addition to the main thread, each
thread starts with an empty list, the main thread will populate that
list in each event-loop with pending-read-clients or
pending-write-clients.
2. Read Phase: The main thread accepts incoming connections and reads
requests from clients. The reading of requests are offloaded to IO
threads. The main thread puts the clients ready-to-read in a list and
set the global io_threads_op to IO_THREADS_OP_READ, the IO threads pick
the clients up, perform the read operation and parse the first incoming
command.
3. Command Processing: After reading the requests, command processing is
still single-threaded and handled by the main thread.
4. Write Phase: Similar to the read phase, the write phase is also be
offloaded to IO threads. The main thread prepares the response in the
clients’ output buffer then the main thread puts the client in the list,
and sets the global io_threads_op to the IO_THREADS_OP_WRITE. The IO
threads then pick the clients up and perform the write operation to send
the responses back to clients.
5. Synchronization: The main-thread communicate with the threads on how
many jobs left per each thread with atomic counter. The main-thread
doesn’t access the clients while being handled by the IO threads.

**Issues with current implementation:**

* Underutilized Cores: The current implementation of IO-threads leads to
the underutilization of CPU cores.
* The main thread remains responsible for a significant portion of
IO-related tasks that could be offloaded to IO-threads.
* When the main-thread is processing client’s commands, the IO threads
are idle for a considerable amount of time.
* Notably, the main thread's performance during the IO-related tasks is
constrained by the speed of the slowest IO-thread.
* Limited Offloading: Currently, Since the Main-threads waits
synchronously for the IO threads, the Threads perform only read-parse,
and write operations, with parsing done only for the first command. If
the threads can do work asynchronously we may offload more work to the
threads reducing the load from the main-thread.
* TLS: Currently, we don't support IO threads with TLS (where offloading
IO would be more beneficial) since TLS read/write operations are not
thread-safe with the current implementation.

### Suggested change

Non-blocking main thread - The main thread and IO threads will operate
in parallel to maximize efficiency. The main thread will not be blocked
by IO operations. It will continue to process commands independently of
the IO thread's activities.

**Implementation details**

**Inter-thread communication.**

* We use a static, lock-free ring buffer of fixed size (2048 jobs) for
the main thread to send jobs and for the IO to receive them. If the ring
buffer fills up, the main thread will handle the task itself, acting as
back pressure (in case IO operations are more expensive than command
processing). A static ring buffer is a better candidate than a dynamic
job queue as it eliminates the need for allocation/freeing per job.
* An IO job will be in the format: ` [void* function-call-back | void
*data] `where data is either a client to read/write from and the
function-ptr is the function to be called with the data for example
readQueryFromClient using this format we can use it later to offload
other types of works to the IO threads.
* The Ring buffer is one way from the main-thread to the IO thread, Upon
read/write event the main thread will send a read/write job then in
before sleep it will iterate over the pending read/write clients to
checking for each client if the IO threads has already finished handling
it. The IO thread signals it has finished handling a client read/write
by toggling an atomic flag read_state / write_state on the client
struct.

**Thread Safety**

As suggested in this solution, the IO threads are reading from and
writing to the clients' buffers while the main thread may access those
clients.
We must ensure no race conditions or unsafe access occurs while keeping
the Valkey code simple and lock free.

Minimal Action in the IO Threads
The main change is to limit the IO thread operations to the bare
minimum. The IO thread will access only the client's struct and only the
necessary fields in this struct.
The IO threads will be responsible for the following:

* Read Operation: The IO thread will only read and parse a single
command. It will not update the server stats, handle read errors, or
parsing errors. These tasks will be taken care of by the main thread.
* Write Operation: The IO thread will only write the available data. It
will not free the client's replies, handle write errors, or update the
server statistics.


To achieve this without code duplication, the read/write code has been
refactored into smaller, independent components:

* Functions that perform only the read/parse/write calls.
* Functions that handle the read/parse/write results.

This refactor accounts for the majority of the modifications in this PR.

**Client Struct Safe Access**

As we ensure that the IO threads access memory only within the client
struct, we need to ensure thread safety only for the client's struct's
shared fields.

* Query Buffer 
* Command parsing - The main thread will not try to parse a command from
the query buffer when a client is offloaded to the IO thread.
* Client's memory checks in client-cron - The main thread will not
access the client query buffer if it is offloaded and will handle the
querybuf grow/shrink when the client is back.
* CLIENT LIST command - The main thread will busy-wait for the IO thread
to finish handling the client, falling back to the current behavior
where the main thread waits for the IO thread to finish their
processing.
* Output Buffer 
* The IO thread will not change the client's bufpos and won't free the
client's reply lists. These actions will be done by the main thread on
the client's return from the IO thread.
* bufpos / block→used: As the main thread may change the bufpos, the
reply-block→used, or add/delete blocks to the reply list while the IO
thread writes, we add two fields to the client struct: io_last_bufpos
and io_last_reply_block. The IO thread will write until the
io_last_bufpos, which was set by the main-thread before sending the
client to the IO thread. If more data has been added to the cob in
between, it will be written in the next write-job. In addition, the main
thread will not trim or merge reply blocks while the client is
offloaded.
* Parsing Fields 
    * Client's cmd, argc, argv, reqtype, etc., are set during parsing.
* The main thread will indicate to the IO thread not to parse a cmd if
the client is not reset. In this case, the IO thread will only read from
the network and won't attempt to parse a new command.
* The main thread won't access the c→cmd/c→argv in the CLIENT LIST
command as stated before it will busy wait for the IO threads.
* Client Flags 
* c→flags, which may be changed by the main thread in multiple places,
won't be accessed by the IO thread. Instead, the main thread will set
the c→io_flags with the information necessary for the IO thread to know
the client's state.
* Client Close 
* On freeClient, the main thread will busy wait for the IO thread to
finish processing the client's read/write before proceeding to free the
client.
* Client's Memory Limits 
* The IO thread won't handle the qb/cob limits. In case a client crosses
the qb limit, the IO thread will stop reading for it, letting the main
thread know that the client crossed the limit.

**TLS**

TLS is currently not supported with IO threads for the following
reasons:

1. Pending reads - If SSL has pending data that has already been read
from the socket, there is a risk of not calling the read handler again.
To handle this, a list is used to hold the pending clients. With IO
threads, multiple threads can access the list concurrently.
2. Event loop modification - Currently, the TLS code
registers/unregisters the file descriptor from the event loop depending
on the read/write results. With IO threads, multiple threads can modify
the event loop struct simultaneously.
3. The same client can be sent to 2 different threads concurrently
(https://github.com/redis/redis/issues/12540).

Those issues were handled in the current PR:

1. The IO thread only performs the read operation. The main thread will
check for pending reads after the client returns from the IO thread and
will be the only one to access the pending list.
2. The registering/unregistering of events will be similarly postponed
and handled by the main thread only.
3. Each client is being sent to the same dedicated thread (c→id %
num_of_threads).


**Sending Replies Immediately with IO threads.**

Currently, after processing a command, we add the client to the
pending_writes_list. Only after processing all the clients do we send
all the replies. Since the IO threads are now working asynchronously, we
can send the reply immediately after processing the client’s requests,
reducing the command latency. However, if we are using AOF=always, we
must wait for the AOF buffer to be written, in which case we revert to
the current behavior.

**IO threads dynamic adjustment**

Currently, we use an all-or-nothing approach when activating the IO
threads. The current logic is as follows: if the number of pending write
clients is greater than twice the number of threads (including the main
thread), we enable all threads; otherwise, we enable none. For example,
if 8 IO threads are defined, we enable all 8 threads if there are 16
pending clients; else, we enable none.
It makes more sense to enable partial activation of the IO threads. If
we have 10 pending clients, we will enable 5 threads, and so on. This
approach allows for a more granular and efficient allocation of
resources based on the current workload.

In addition, the user will now be able to change the number of I/O
threads at runtime. For example, when decreasing the number of threads
from 4 to 2, threads 3 and 4 will be closed after flushing their job
queues.

**Tests**

Currently, we run the io-threads tests with 4 IO threads
(443d80f168/.github/workflows/daily.yml (L353)).
This means that we will not activate the IO threads unless there are 8
(threads * 2) pending write clients per single loop, which is unlikely
to happened in most of tests, meaning the IO threads are not currently
being tested.

To enforce the main thread to always offload work to the IO threads,
regardless of the number of pending events, we add an
events-per-io-thread configuration with a default value of 2. When set
to 0, this configuration will force the main thread to always offload
work to the IO threads.

When we offload every single read/write operation to the IO threads, the
IO-threads are running with 100% CPU when running multiple tests
concurrently some tests fail as a result of larger than expected command
latencies. To address this issue, we have to add some after or wait_for
calls to some of the tests to ensure they pass with IO threads as well.

Signed-off-by: Uri Yagelnik <uriy@amazon.com>
2024-07-08 20:01:39 -07:00

1754 lines
66 KiB
C
Raw Permalink Blame History

/*
* Copyright (c) 2009-2012, Salvatore Sanfilippo <antirez at gmail dot com>
* 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.
*/
/*
* This file initializes the global LUA object and registers functions to call Valkey API from within the LUA language.
* It heavily invokes LUA's C API documented at https://www.lua.org/pil/24.html. There are 2 entrypoint functions in
* this file:
* 1. evalCommand() - Gets invoked every time a user runs LUA script via eval command on Valkey.
* 2. scriptingInit() - initServer() function from server.c invokes this to initialize LUA at startup.
* It is also invoked between 2 eval invocations to reset Lua.
*/
#include "server.h"
#include "sha1.h"
#include "rand.h"
#include "cluster.h"
#include "monotonic.h"
#include "resp_parser.h"
#include "script_lua.h"
#include <lua.h>
#include <lauxlib.h>
#include <lualib.h>
#include <ctype.h>
#include <math.h>
void ldbInit(void);
void ldbDisable(client *c);
void ldbEnable(client *c);
void evalGenericCommandWithDebugging(client *c, int evalsha);
sds ldbCatStackValue(sds s, lua_State *lua, int idx);
listNode *luaScriptsLRUAdd(client *c, sds sha, int evalsha);
static void dictLuaScriptDestructor(dict *d, void *val) {
UNUSED(d);
if (val == NULL) return; /* Lazy freeing will set value to NULL. */
decrRefCount(((luaScript *)val)->body);
zfree(val);
}
static uint64_t dictStrCaseHash(const void *key) {
return dictGenCaseHashFunction((unsigned char *)key, strlen((char *)key));
}
/* lctx.lua_scripts sha (as sds string) -> scripts (as luaScript) cache. */
dictType shaScriptObjectDictType = {
dictStrCaseHash, /* hash function */
NULL, /* key dup */
dictSdsKeyCaseCompare, /* key compare */
dictSdsDestructor, /* key destructor */
dictLuaScriptDestructor, /* val destructor */
NULL /* allow to expand */
};
/* Lua context */
struct luaCtx {
lua_State *lua; /* The Lua interpreter. We use just one for all clients */
client *lua_client; /* The "fake client" to query the server from Lua */
dict *lua_scripts; /* A dictionary of SHA1 -> Lua scripts */
list *lua_scripts_lru_list; /* A list of SHA1, first in first out LRU eviction. */
unsigned long long lua_scripts_mem; /* Cached scripts' memory + oh */
} lctx;
/* Debugger shared state is stored inside this global structure. */
#define LDB_BREAKPOINTS_MAX 64 /* Max number of breakpoints. */
#define LDB_MAX_LEN_DEFAULT 256 /* Default len limit for replies / var dumps. */
struct ldbState {
connection *conn; /* Connection of the debugging client. */
int active; /* Are we debugging EVAL right now? */
int forked; /* Is this a fork()ed debugging session? */
list *logs; /* List of messages to send to the client. */
list *traces; /* Messages about commands executed since last stop.*/
list *children; /* All forked debugging sessions pids. */
int bp[LDB_BREAKPOINTS_MAX]; /* An array of breakpoints line numbers. */
int bpcount; /* Number of valid entries inside bp. */
int step; /* Stop at next line regardless of breakpoints. */
int luabp; /* Stop at next line because server.breakpoint() was called. */
sds *src; /* Lua script source code split by line. */
int lines; /* Number of lines in 'src'. */
int currentline; /* Current line number. */
sds cbuf; /* Debugger client command buffer. */
size_t maxlen; /* Max var dump / reply length. */
int maxlen_hint_sent; /* Did we already hint about "set maxlen"? */
} ldb;
/* ---------------------------------------------------------------------------
* Utility functions.
* ------------------------------------------------------------------------- */
/* Perform the SHA1 of the input string. We use this both for hashing script
* bodies in order to obtain the Lua function name, and in the implementation
* of server.sha1().
*
* 'digest' should point to a 41 bytes buffer: 40 for SHA1 converted into an
* hexadecimal number, plus 1 byte for null term. */
void sha1hex(char *digest, char *script, size_t len) {
SHA1_CTX ctx;
unsigned char hash[20];
char *cset = "0123456789abcdef";
int j;
SHA1Init(&ctx);
SHA1Update(&ctx, (unsigned char *)script, len);
SHA1Final(hash, &ctx);
for (j = 0; j < 20; j++) {
digest[j * 2] = cset[((hash[j] & 0xF0) >> 4)];
digest[j * 2 + 1] = cset[(hash[j] & 0xF)];
}
digest[40] = '\0';
}
/* server.breakpoint()
*
* Allows to stop execution during a debugging session from within
* the Lua code implementation, like if a breakpoint was set in the code
* immediately after the function. */
int luaServerBreakpointCommand(lua_State *lua) {
if (ldb.active) {
ldb.luabp = 1;
lua_pushboolean(lua, 1);
} else {
lua_pushboolean(lua, 0);
}
return 1;
}
/* server.debug()
*
* Log a string message into the output console.
* Can take multiple arguments that will be separated by commas.
* Nothing is returned to the caller. */
int luaServerDebugCommand(lua_State *lua) {
if (!ldb.active) return 0;
int argc = lua_gettop(lua);
sds log = sdscatprintf(sdsempty(), "<debug> line %d: ", ldb.currentline);
while (argc--) {
log = ldbCatStackValue(log, lua, -1 - argc);
if (argc != 0) log = sdscatlen(log, ", ", 2);
}
ldbLog(log);
return 0;
}
/* server.replicate_commands()
*
* DEPRECATED: Now do nothing and always return true.
* Turn on single commands replication if the script never called
* a write command so far, and returns true. Otherwise if the script
* already started to write, returns false and stick to whole scripts
* replication, which is our default. */
int luaServerReplicateCommandsCommand(lua_State *lua) {
lua_pushboolean(lua, 1);
return 1;
}
/* Initialize the scripting environment.
*
* This function is called the first time at server startup with
* the 'setup' argument set to 1.
*
* It can be called again multiple times during the lifetime of the
* process, with 'setup' set to 0, and following a scriptingRelease() call,
* in order to reset the Lua scripting environment.
*
* However it is simpler to just call scriptingReset() that does just that. */
void scriptingInit(int setup) {
lua_State *lua = lua_open();
if (setup) {
lctx.lua_client = NULL;
server.script_disable_deny_script = 0;
ldbInit();
}
/* Initialize a dictionary we use to map SHAs to scripts.
* Initialize a list we use for lua script evictions, it shares the
* sha with the dictionary, so free fn is not set. */
lctx.lua_scripts = dictCreate(&shaScriptObjectDictType);
lctx.lua_scripts_lru_list = listCreate();
lctx.lua_scripts_mem = 0;
luaRegisterServerAPI(lua);
/* register debug commands */
lua_getglobal(lua, "server");
/* server.breakpoint */
lua_pushstring(lua, "breakpoint");
lua_pushcfunction(lua, luaServerBreakpointCommand);
lua_settable(lua, -3);
/* server.debug */
lua_pushstring(lua, "debug");
lua_pushcfunction(lua, luaServerDebugCommand);
lua_settable(lua, -3);
/* server.replicate_commands */
lua_pushstring(lua, "replicate_commands");
lua_pushcfunction(lua, luaServerReplicateCommandsCommand);
lua_settable(lua, -3);
lua_setglobal(lua, "server");
/* Add a helper function we use for pcall error reporting.
* Note that when the error is in the C function we want to report the
* information about the caller, that's what makes sense from the point
* of view of the user debugging a script. */
{
char *errh_func = "local dbg = debug\n"
"debug = nil\n"
"function __redis__err__handler(err)\n"
" local i = dbg.getinfo(2,'nSl')\n"
" if i and i.what == 'C' then\n"
" i = dbg.getinfo(3,'nSl')\n"
" end\n"
" if type(err) ~= 'table' then\n"
" err = {err='ERR ' .. tostring(err)}"
" end"
" if i then\n"
" err['source'] = i.source\n"
" err['line'] = i.currentline\n"
" end"
" return err\n"
"end\n";
luaL_loadbuffer(lua, errh_func, strlen(errh_func), "@err_handler_def");
lua_pcall(lua, 0, 0, 0);
}
/* Create the (non connected) client that we use to execute server commands
* inside the Lua interpreter.
* Note: there is no need to create it again when this function is called
* by scriptingReset(). */
if (lctx.lua_client == NULL) {
lctx.lua_client = createClient(NULL);
lctx.lua_client->flag.script = 1;
/* We do not want to allow blocking commands inside Lua */
lctx.lua_client->flag.deny_blocking = 1;
}
/* Lock the global table from any changes */
lua_pushvalue(lua, LUA_GLOBALSINDEX);
luaSetErrorMetatable(lua);
/* Recursively lock all tables that can be reached from the global table */
luaSetTableProtectionRecursively(lua);
lua_pop(lua, 1);
lctx.lua = lua;
}
/* Free lua_scripts dict and close lua interpreter. */
void freeLuaScriptsSync(dict *lua_scripts, list *lua_scripts_lru_list, lua_State *lua) {
dictRelease(lua_scripts);
listRelease(lua_scripts_lru_list);
lua_close(lua);
#if !defined(USE_LIBC)
/* The lua interpreter may hold a lot of memory internally, and lua is
* using libc. libc may take a bit longer to return the memory to the OS,
* so after lua_close, we call malloc_trim try to purge it earlier.
*
* We do that only when the server itself does not use libc. When Lua and the server
* use different allocators, one won't use the fragmentation holes of the
* other, and released memory can take a long time until it is returned to
* the OS. */
zlibc_trim();
#endif
}
/* Release resources related to Lua scripting.
* This function is used in order to reset the scripting environment. */
void scriptingRelease(int async) {
if (async)
freeLuaScriptsAsync(lctx.lua_scripts, lctx.lua_scripts_lru_list, lctx.lua);
else
freeLuaScriptsSync(lctx.lua_scripts, lctx.lua_scripts_lru_list, lctx.lua);
}
void scriptingReset(int async) {
scriptingRelease(async);
scriptingInit(0);
}
/* ---------------------------------------------------------------------------
* EVAL and SCRIPT commands implementation
* ------------------------------------------------------------------------- */
static void evalCalcFunctionName(int evalsha, sds script, char *out_funcname) {
/* We obtain the script SHA1, then check if this function is already
* defined into the Lua state */
out_funcname[0] = 'f';
out_funcname[1] = '_';
if (!evalsha) {
/* Hash the code if this is an EVAL call */
sha1hex(out_funcname + 2, script, sdslen(script));
} else {
/* We already have the SHA if it is an EVALSHA */
int j;
char *sha = script;
/* Convert to lowercase. We don't use tolower since the function
* managed to always show up in the profiler output consuming
* a non trivial amount of time. */
for (j = 0; j < 40; j++) out_funcname[j + 2] = (sha[j] >= 'A' && sha[j] <= 'Z') ? sha[j] + ('a' - 'A') : sha[j];
out_funcname[42] = '\0';
}
}
/* Helper function to try and extract shebang flags from the script body.
* If no shebang is found, return with success and COMPAT mode flag.
* The err arg is optional, can be used to get a detailed error string.
* The out_shebang_len arg is optional, can be used to trim the shebang from the script.
* Returns C_OK on success, and C_ERR on error. */
int evalExtractShebangFlags(sds body, uint64_t *out_flags, ssize_t *out_shebang_len, sds *err) {
ssize_t shebang_len = 0;
uint64_t script_flags = SCRIPT_FLAG_EVAL_COMPAT_MODE;
if (!strncmp(body, "#!", 2)) {
int numparts, j;
char *shebang_end = strchr(body, '\n');
if (shebang_end == NULL) {
if (err) *err = sdsnew("Invalid script shebang");
return C_ERR;
}
shebang_len = shebang_end - body;
sds shebang = sdsnewlen(body, shebang_len);
sds *parts = sdssplitargs(shebang, &numparts);
sdsfree(shebang);
if (!parts || numparts == 0) {
if (err) *err = sdsnew("Invalid engine in script shebang");
sdsfreesplitres(parts, numparts);
return C_ERR;
}
/* Verify lua interpreter was specified */
if (strcmp(parts[0], "#!lua")) {
if (err) *err = sdscatfmt(sdsempty(), "Unexpected engine in script shebang: %s", parts[0]);
sdsfreesplitres(parts, numparts);
return C_ERR;
}
script_flags &= ~SCRIPT_FLAG_EVAL_COMPAT_MODE;
for (j = 1; j < numparts; j++) {
if (!strncmp(parts[j], "flags=", 6)) {
sdsrange(parts[j], 6, -1);
int numflags, jj;
sds *flags = sdssplitlen(parts[j], sdslen(parts[j]), ",", 1, &numflags);
for (jj = 0; jj < numflags; jj++) {
scriptFlag *sf;
for (sf = scripts_flags_def; sf->flag; sf++) {
if (!strcmp(flags[jj], sf->str)) break;
}
if (!sf->flag) {
if (err) *err = sdscatfmt(sdsempty(), "Unexpected flag in script shebang: %s", flags[jj]);
sdsfreesplitres(flags, numflags);
sdsfreesplitres(parts, numparts);
return C_ERR;
}
script_flags |= sf->flag;
}
sdsfreesplitres(flags, numflags);
} else {
/* We only support function flags options for lua scripts */
if (err) *err = sdscatfmt(sdsempty(), "Unknown lua shebang option: %s", parts[j]);
sdsfreesplitres(parts, numparts);
return C_ERR;
}
}
sdsfreesplitres(parts, numparts);
}
if (out_shebang_len) *out_shebang_len = shebang_len;
*out_flags = script_flags;
return C_OK;
}
/* Try to extract command flags if we can, returns the modified flags.
* Note that it does not guarantee the command arguments are right. */
uint64_t evalGetCommandFlags(client *c, uint64_t cmd_flags) {
char funcname[43];
int evalsha = c->cmd->proc == evalShaCommand || c->cmd->proc == evalShaRoCommand;
if (evalsha && sdslen(c->argv[1]->ptr) != 40) return cmd_flags;
uint64_t script_flags;
evalCalcFunctionName(evalsha, c->argv[1]->ptr, funcname);
char *lua_cur_script = funcname + 2;
c->cur_script = dictFind(lctx.lua_scripts, lua_cur_script);
if (!c->cur_script) {
if (evalsha) return cmd_flags;
if (evalExtractShebangFlags(c->argv[1]->ptr, &script_flags, NULL, NULL) == C_ERR) return cmd_flags;
} else {
luaScript *l = dictGetVal(c->cur_script);
script_flags = l->flags;
}
if (script_flags & SCRIPT_FLAG_EVAL_COMPAT_MODE) return cmd_flags;
return scriptFlagsToCmdFlags(cmd_flags, script_flags);
}
/* Define a Lua function with the specified body.
* The function name will be generated in the following form:
*
* f_<hex sha1 sum>
*
* The function increments the reference count of the 'body' object as a
* side effect of a successful call.
*
* On success a pointer to an SDS string representing the function SHA1 of the
* just added function is returned (and will be valid until the next call
* to scriptingReset() function), otherwise NULL is returned.
*
* The function handles the fact of being called with a script that already
* exists, and in such a case, it behaves like in the success case.
*
* If 'c' is not NULL, on error the client is informed with an appropriate
* error describing the nature of the problem and the Lua interpreter error.
*
* 'evalsha' indicating whether the lua function is created from the EVAL context
* or from the SCRIPT LOAD. */
sds luaCreateFunction(client *c, robj *body, int evalsha) {
char funcname[43];
dictEntry *de;
uint64_t script_flags;
funcname[0] = 'f';
funcname[1] = '_';
sha1hex(funcname + 2, body->ptr, sdslen(body->ptr));
if ((de = dictFind(lctx.lua_scripts, funcname + 2)) != NULL) {
/* If the script was previously added via EVAL, we promote it to
* SCRIPT LOAD, prevent it from being evicted later. */
luaScript *l = dictGetVal(de);
if (evalsha && l->node) {
listDelNode(lctx.lua_scripts_lru_list, l->node);
l->node = NULL;
}
return dictGetKey(de);
}
/* Handle shebang header in script code */
ssize_t shebang_len = 0;
sds err = NULL;
if (evalExtractShebangFlags(body->ptr, &script_flags, &shebang_len, &err) == C_ERR) {
if (c != NULL) {
addReplyErrorSds(c, err);
}
return NULL;
}
/* Note that in case of a shebang line we skip it but keep the line feed to conserve the user's line numbers */
if (luaL_loadbuffer(lctx.lua, (char *)body->ptr + shebang_len, sdslen(body->ptr) - shebang_len, "@user_script")) {
if (c != NULL) {
addReplyErrorFormat(c, "Error compiling script (new function): %s", lua_tostring(lctx.lua, -1));
}
lua_pop(lctx.lua, 1);
return NULL;
}
serverAssert(lua_isfunction(lctx.lua, -1));
lua_setfield(lctx.lua, LUA_REGISTRYINDEX, funcname);
/* We also save a SHA1 -> Original script map in a dictionary
* so that we can replicate / write in the AOF all the
* EVALSHA commands as EVAL using the original script. */
luaScript *l = zcalloc(sizeof(luaScript));
l->body = body;
l->flags = script_flags;
sds sha = sdsnewlen(funcname + 2, 40);
l->node = luaScriptsLRUAdd(c, sha, evalsha);
int retval = dictAdd(lctx.lua_scripts, sha, l);
serverAssertWithInfo(c ? c : lctx.lua_client, NULL, retval == DICT_OK);
lctx.lua_scripts_mem += sdsZmallocSize(sha) + getStringObjectSdsUsedMemory(body);
incrRefCount(body);
return sha;
}
/* Delete a Lua function with the specified sha.
*
* This will delete the lua function from the lua interpreter and delete
* the lua function from server. */
void luaDeleteFunction(client *c, sds sha) {
/* Delete the script from lua interpreter. */
char funcname[43];
funcname[0] = 'f';
funcname[1] = '_';
memcpy(funcname + 2, sha, 40);
funcname[42] = '\0';
lua_pushnil(lctx.lua);
lua_setfield(lctx.lua, LUA_REGISTRYINDEX, funcname);
/* Delete the script from server. */
dictEntry *de = dictUnlink(lctx.lua_scripts, sha);
serverAssertWithInfo(c ? c : lctx.lua_client, NULL, de);
luaScript *l = dictGetVal(de);
/* We only delete `EVAL` scripts, which must exist in the LRU list. */
serverAssert(l->node);
listDelNode(lctx.lua_scripts_lru_list, l->node);
lctx.lua_scripts_mem -= sdsZmallocSize(sha) + getStringObjectSdsUsedMemory(l->body);
dictFreeUnlinkedEntry(lctx.lua_scripts, de);
}
/* Users who abuse EVAL will generate a new lua script on each call, which can
* consume large amounts of memory over time. Since EVAL is mostly the one that
* abuses the lua cache, and these won't have pipeline issues (scripts won't
* disappear when EVALSHA needs it and cause failure), we implement script eviction
* only for these (not for one loaded with SCRIPT LOAD). Considering that we don't
* have many scripts, then unlike keys, we don't need to worry about the memory
* usage of keeping a true sorted LRU linked list.
*
* 'evalsha' indicating whether the lua function is added from the EVAL context
* or from the SCRIPT LOAD.
*
* Returns the corresponding node added, which is used to save it in luaScript
* and use it for quick removal and re-insertion into an LRU list each time the
* script is used. */
#define LRU_LIST_LENGTH 500
listNode *luaScriptsLRUAdd(client *c, sds sha, int evalsha) {
/* Script eviction only applies to EVAL, not SCRIPT LOAD. */
if (evalsha) return NULL;
/* Evict oldest. */
while (listLength(lctx.lua_scripts_lru_list) >= LRU_LIST_LENGTH) {
listNode *ln = listFirst(lctx.lua_scripts_lru_list);
sds oldest = listNodeValue(ln);
luaDeleteFunction(c, oldest);
server.stat_evictedscripts++;
}
/* Add current. */
listAddNodeTail(lctx.lua_scripts_lru_list, sha);
return listLast(lctx.lua_scripts_lru_list);
}
void evalGenericCommand(client *c, int evalsha) {
lua_State *lua = lctx.lua;
char funcname[43];
long long numkeys;
/* Get the number of arguments that are keys */
if (getLongLongFromObjectOrReply(c, c->argv[2], &numkeys, NULL) != C_OK) 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;
}
if (c->cur_script) {
funcname[0] = 'f', funcname[1] = '_';
memcpy(funcname + 2, dictGetKey(c->cur_script), 40);
funcname[42] = '\0';
} else
evalCalcFunctionName(evalsha, c->argv[1]->ptr, funcname);
/* Push the pcall error handler function on the stack. */
lua_getglobal(lua, "__redis__err__handler");
/* Try to lookup the Lua function */
lua_getfield(lua, LUA_REGISTRYINDEX, funcname);
if (lua_isnil(lua, -1)) {
lua_pop(lua, 1); /* remove the nil from the stack */
/* Function not defined... let's define it if we have the
* body of the function. If this is an EVALSHA call we can just
* return an error. */
if (evalsha) {
lua_pop(lua, 1); /* remove the error handler from the stack. */
addReplyErrorObject(c, shared.noscripterr);
return;
}
if (luaCreateFunction(c, c->argv[1], evalsha) == NULL) {
lua_pop(lua, 1); /* remove the error handler from the stack. */
/* The error is sent to the client by luaCreateFunction()
* itself when it returns NULL. */
return;
}
/* Now the following is guaranteed to return non nil */
lua_getfield(lua, LUA_REGISTRYINDEX, funcname);
serverAssert(!lua_isnil(lua, -1));
}
char *lua_cur_script = funcname + 2;
dictEntry *de = c->cur_script;
if (!de) de = dictFind(lctx.lua_scripts, lua_cur_script);
luaScript *l = dictGetVal(de);
int ro = c->cmd->proc == evalRoCommand || c->cmd->proc == evalShaRoCommand;
scriptRunCtx rctx;
if (scriptPrepareForRun(&rctx, lctx.lua_client, c, lua_cur_script, l->flags, ro) != C_OK) {
lua_pop(lua, 2); /* Remove the function and error handler. */
return;
}
rctx.flags |= SCRIPT_EVAL_MODE; /* mark the current run as EVAL (as opposed to FCALL) so we'll
get appropriate error messages and logs */
luaCallFunction(&rctx, lua, c->argv + 3, numkeys, c->argv + 3 + numkeys, c->argc - 3 - numkeys, ldb.active);
lua_pop(lua, 1); /* Remove the error handler. */
scriptResetRun(&rctx);
if (l->node) {
/* Quick removal and re-insertion after the script is called to
* maintain the LRU list. */
listUnlinkNode(lctx.lua_scripts_lru_list, l->node);
listLinkNodeTail(lctx.lua_scripts_lru_list, l->node);
}
}
void evalCommand(client *c) {
/* Explicitly feed monitor here so that lua commands appear after their
* script command. */
replicationFeedMonitors(c, server.monitors, c->db->id, c->argv, c->argc);
if (!c->flag.lua_debug)
evalGenericCommand(c, 0);
else
evalGenericCommandWithDebugging(c, 0);
}
void evalRoCommand(client *c) {
evalCommand(c);
}
void evalShaCommand(client *c) {
/* Explicitly feed monitor here so that lua commands appear after their
* script command. */
replicationFeedMonitors(c, server.monitors, c->db->id, c->argv, c->argc);
if (sdslen(c->argv[1]->ptr) != 40) {
/* We know that a match is not possible if the provided SHA is
* not the right length. So we return an error ASAP, this way
* evalGenericCommand() can be implemented without string length
* sanity check */
addReplyErrorObject(c, shared.noscripterr);
return;
}
if (!c->flag.lua_debug)
evalGenericCommand(c, 1);
else {
addReplyError(c, "Please use EVAL instead of EVALSHA for debugging");
return;
}
}
void evalShaRoCommand(client *c) {
evalShaCommand(c);
}
void scriptCommand(client *c) {
if (c->argc == 2 && !strcasecmp(c->argv[1]->ptr, "help")) {
/* clang-format off */
const char *help[] = {
"DEBUG (YES|SYNC|NO)",
" Set the debug mode for subsequent scripts executed.",
"EXISTS <sha1> [<sha1> ...]",
" Return information about the existence of the scripts in the script cache.",
"FLUSH [ASYNC|SYNC]",
" Flush the Lua scripts cache. Very dangerous on replicas.",
" When called without the optional mode argument, the behavior is determined",
" by the lazyfree-lazy-user-flush configuration directive. Valid modes are:",
" * ASYNC: Asynchronously flush the scripts cache.",
" * SYNC: Synchronously flush the scripts cache.",
"KILL",
" Kill the currently executing Lua script.",
"LOAD <script>",
" Load a script into the scripts cache without executing it.",
"SHOW <sha1>",
" Show a script from the scripts cache.",
NULL
};
/* clang-format on */
addReplyHelp(c, help);
} else if (c->argc >= 2 && !strcasecmp(c->argv[1]->ptr, "flush")) {
int async = 0;
if (c->argc == 3 && !strcasecmp(c->argv[2]->ptr, "sync")) {
async = 0;
} else if (c->argc == 3 && !strcasecmp(c->argv[2]->ptr, "async")) {
async = 1;
} else if (c->argc == 2) {
async = server.lazyfree_lazy_user_flush ? 1 : 0;
} else {
addReplyError(c, "SCRIPT FLUSH only support SYNC|ASYNC option");
return;
}
scriptingReset(async);
addReply(c, shared.ok);
} else if (c->argc >= 2 && !strcasecmp(c->argv[1]->ptr, "exists")) {
int j;
addReplyArrayLen(c, c->argc - 2);
for (j = 2; j < c->argc; j++) {
if (dictFind(lctx.lua_scripts, c->argv[j]->ptr))
addReply(c, shared.cone);
else
addReply(c, shared.czero);
}
} else if (c->argc == 3 && !strcasecmp(c->argv[1]->ptr, "load")) {
sds sha = luaCreateFunction(c, c->argv[2], 1);
if (sha == NULL) return; /* The error was sent by luaCreateFunction(). */
addReplyBulkCBuffer(c, sha, 40);
} else if (c->argc == 2 && !strcasecmp(c->argv[1]->ptr, "kill")) {
scriptKill(c, 1);
} else if (c->argc == 3 && !strcasecmp(c->argv[1]->ptr, "debug")) {
if (clientHasPendingReplies(c)) {
addReplyError(c, "SCRIPT DEBUG must be called outside a pipeline");
return;
}
if (!strcasecmp(c->argv[2]->ptr, "no")) {
ldbDisable(c);
addReply(c, shared.ok);
} else if (!strcasecmp(c->argv[2]->ptr, "yes")) {
ldbEnable(c);
addReply(c, shared.ok);
} else if (!strcasecmp(c->argv[2]->ptr, "sync")) {
ldbEnable(c);
addReply(c, shared.ok);
c->flag.lua_debug_sync = 1;
} else {
addReplyError(c, "Use SCRIPT DEBUG YES/SYNC/NO");
return;
}
} else if (c->argc == 3 && !strcasecmp(c->argv[1]->ptr, "show")) {
dictEntry *de;
luaScript *ls;
if (sdslen(c->argv[2]->ptr) == 40 && (de = dictFind(lctx.lua_scripts, c->argv[2]->ptr))) {
ls = dictGetVal(de);
addReplyBulk(c, ls->body);
} else {
addReplyErrorObject(c, shared.noscripterr);
}
} else {
addReplySubcommandSyntaxError(c);
}
}
unsigned long evalMemory(void) {
return luaMemory(lctx.lua);
}
dict *evalScriptsDict(void) {
return lctx.lua_scripts;
}
unsigned long evalScriptsMemory(void) {
return lctx.lua_scripts_mem + dictMemUsage(lctx.lua_scripts) + dictSize(lctx.lua_scripts) * sizeof(luaScript) +
listLength(lctx.lua_scripts_lru_list) * sizeof(listNode);
}
/* ---------------------------------------------------------------------------
* LDB: Lua debugging facilities
* ------------------------------------------------------------------------- */
/* Initialize Lua debugger data structures. */
void ldbInit(void) {
ldb.conn = NULL;
ldb.active = 0;
ldb.logs = listCreate();
listSetFreeMethod(ldb.logs, (void (*)(void *))sdsfree);
ldb.children = listCreate();
ldb.src = NULL;
ldb.lines = 0;
ldb.cbuf = sdsempty();
}
/* Remove all the pending messages in the specified list. */
void ldbFlushLog(list *log) {
listNode *ln;
while ((ln = listFirst(log)) != NULL) listDelNode(log, ln);
}
int ldbIsEnabled(void) {
return ldb.active && ldb.step;
}
/* Enable debug mode of Lua scripts for this client. */
void ldbEnable(client *c) {
c->flag.lua_debug = 1;
ldbFlushLog(ldb.logs);
ldb.conn = c->conn;
ldb.step = 1;
ldb.bpcount = 0;
ldb.luabp = 0;
sdsfree(ldb.cbuf);
ldb.cbuf = sdsempty();
ldb.maxlen = LDB_MAX_LEN_DEFAULT;
ldb.maxlen_hint_sent = 0;
}
/* Exit debugging mode from the POV of client. This function is not enough
* to properly shut down a client debugging session, see ldbEndSession()
* for more information. */
void ldbDisable(client *c) {
c->flag.lua_debug = 0;
c->flag.lua_debug_sync = 0;
}
/* Append a log entry to the specified LDB log. */
void ldbLog(sds entry) {
listAddNodeTail(ldb.logs, entry);
}
/* A version of ldbLog() which prevents producing logs greater than
* ldb.maxlen. The first time the limit is reached a hint is generated
* to inform the user that reply trimming can be disabled using the
* debugger "maxlen" command. */
void ldbLogWithMaxLen(sds entry) {
int trimmed = 0;
if (ldb.maxlen && sdslen(entry) > ldb.maxlen) {
sdsrange(entry, 0, ldb.maxlen - 1);
entry = sdscatlen(entry, " ...", 4);
trimmed = 1;
}
ldbLog(entry);
if (trimmed && ldb.maxlen_hint_sent == 0) {
ldb.maxlen_hint_sent = 1;
ldbLog(sdsnew("<hint> The above reply was trimmed. Use 'maxlen 0' to disable trimming."));
}
}
/* Send ldb.logs to the debugging client as a multi-bulk reply
* consisting of simple strings. Log entries which include newlines have them
* replaced with spaces. The entries sent are also consumed. */
void ldbSendLogs(void) {
sds proto = sdsempty();
proto = sdscatfmt(proto, "*%i\r\n", (int)listLength(ldb.logs));
while (listLength(ldb.logs)) {
listNode *ln = listFirst(ldb.logs);
proto = sdscatlen(proto, "+", 1);
sdsmapchars(ln->value, "\r\n", " ", 2);
proto = sdscatsds(proto, ln->value);
proto = sdscatlen(proto, "\r\n", 2);
listDelNode(ldb.logs, ln);
}
if (connWrite(ldb.conn, proto, sdslen(proto)) == -1) {
/* Avoid warning. We don't check the return value of write()
* since the next read() will catch the I/O error and will
* close the debugging session. */
}
sdsfree(proto);
}
/* Start a debugging session before calling EVAL implementation.
* The technique we use is to capture the client socket file descriptor,
* in order to perform direct I/O with it from within Lua hooks. This
* way we don't have to re-enter the server in order to handle I/O.
*
* The function returns 1 if the caller should proceed to call EVAL,
* and 0 if instead the caller should abort the operation (this happens
* for the parent in a forked session, since it's up to the children
* to continue, or when fork returned an error).
*
* The caller should call ldbEndSession() only if ldbStartSession()
* returned 1. */
int ldbStartSession(client *c) {
ldb.forked = !c->flag.lua_debug_sync;
if (ldb.forked) {
pid_t cp = serverFork(CHILD_TYPE_LDB);
if (cp == -1) {
addReplyErrorFormat(c, "Fork() failed: can't run EVAL in debugging mode: %s", strerror(errno));
return 0;
} else if (cp == 0) {
/* Child. Let's ignore important signals handled by the parent. */
struct sigaction act;
sigemptyset(&act.sa_mask);
act.sa_flags = 0;
act.sa_handler = SIG_IGN;
sigaction(SIGTERM, &act, NULL);
sigaction(SIGINT, &act, NULL);
/* Log the creation of the child and close the listening
* socket to make sure if the parent crashes a reset is sent
* to the clients. */
serverLog(LL_NOTICE, "%s forked for debugging eval", SERVER_TITLE);
} else {
/* Parent */
listAddNodeTail(ldb.children, (void *)(unsigned long)cp);
freeClientAsync(c); /* Close the client in the parent side. */
return 0;
}
} else {
serverLog(LL_NOTICE, "%s synchronous debugging eval session started", SERVER_TITLE);
}
/* Setup our debugging session. */
connBlock(ldb.conn);
connSendTimeout(ldb.conn, 5000);
ldb.active = 1;
/* First argument of EVAL is the script itself. We split it into different
* lines since this is the way the debugger accesses the source code. */
sds srcstring = sdsdup(c->argv[1]->ptr);
size_t srclen = sdslen(srcstring);
while (srclen && (srcstring[srclen - 1] == '\n' || srcstring[srclen - 1] == '\r')) {
srcstring[--srclen] = '\0';
}
sdssetlen(srcstring, srclen);
ldb.src = sdssplitlen(srcstring, sdslen(srcstring), "\n", 1, &ldb.lines);
sdsfree(srcstring);
return 1;
}
/* End a debugging session after the EVAL call with debugging enabled
* returned. */
void ldbEndSession(client *c) {
/* Emit the remaining logs and an <endsession> mark. */
ldbLog(sdsnew("<endsession>"));
ldbSendLogs();
/* If it's a fork()ed session, we just exit. */
if (ldb.forked) {
writeToClient(c);
serverLog(LL_NOTICE, "Lua debugging session child exiting");
exitFromChild(0);
} else {
serverLog(LL_NOTICE, "%s synchronous debugging eval session ended", SERVER_TITLE);
}
/* Otherwise let's restore client's state. */
connNonBlock(ldb.conn);
connSendTimeout(ldb.conn, 0);
/* Close the client connection after sending the final EVAL reply
* in order to signal the end of the debugging session. */
c->flag.close_after_reply = 1;
/* Cleanup. */
sdsfreesplitres(ldb.src, ldb.lines);
ldb.lines = 0;
ldb.active = 0;
}
/* If the specified pid is among the list of children spawned for
* forked debugging sessions, it is removed from the children list.
* If the pid was found non-zero is returned. */
int ldbRemoveChild(pid_t pid) {
listNode *ln = listSearchKey(ldb.children, (void *)(unsigned long)pid);
if (ln) {
listDelNode(ldb.children, ln);
return 1;
}
return 0;
}
/* Return the number of children we still did not receive termination
* acknowledge via wait() in the parent process. */
int ldbPendingChildren(void) {
return listLength(ldb.children);
}
/* Kill all the forked sessions. */
void ldbKillForkedSessions(void) {
listIter li;
listNode *ln;
listRewind(ldb.children, &li);
while ((ln = listNext(&li))) {
pid_t pid = (unsigned long)ln->value;
serverLog(LL_NOTICE, "Killing debugging session %ld", (long)pid);
kill(pid, SIGKILL);
}
listRelease(ldb.children);
ldb.children = listCreate();
}
/* Wrapper for EVAL / EVALSHA that enables debugging, and makes sure
* that when EVAL returns, whatever happened, the session is ended. */
void evalGenericCommandWithDebugging(client *c, int evalsha) {
if (ldbStartSession(c)) {
evalGenericCommand(c, evalsha);
ldbEndSession(c);
} else {
ldbDisable(c);
}
}
/* Return a pointer to ldb.src source code line, considering line to be
* one-based, and returning a special string for out of range lines. */
char *ldbGetSourceLine(int line) {
int idx = line - 1;
if (idx < 0 || idx >= ldb.lines) return "<out of range source code line>";
return ldb.src[idx];
}
/* Return true if there is a breakpoint in the specified line. */
int ldbIsBreakpoint(int line) {
int j;
for (j = 0; j < ldb.bpcount; j++)
if (ldb.bp[j] == line) return 1;
return 0;
}
/* Add the specified breakpoint. Ignore it if we already reached the max.
* Returns 1 if the breakpoint was added (or was already set). 0 if there is
* no space for the breakpoint or if the line is invalid. */
int ldbAddBreakpoint(int line) {
if (line <= 0 || line > ldb.lines) return 0;
if (!ldbIsBreakpoint(line) && ldb.bpcount != LDB_BREAKPOINTS_MAX) {
ldb.bp[ldb.bpcount++] = line;
return 1;
}
return 0;
}
/* Remove the specified breakpoint, returning 1 if the operation was
* performed or 0 if there was no such breakpoint. */
int ldbDelBreakpoint(int line) {
int j;
for (j = 0; j < ldb.bpcount; j++) {
if (ldb.bp[j] == line) {
ldb.bpcount--;
memmove(ldb.bp + j, ldb.bp + j + 1, ldb.bpcount - j);
return 1;
}
}
return 0;
}
/* Expect a valid multi-bulk command in the debugging client query buffer.
* On success the command is parsed and returned as an array of SDS strings,
* otherwise NULL is returned and there is to read more buffer. */
sds *ldbReplParseCommand(int *argcp, char **err) {
static char *protocol_error = "protocol error";
sds *argv = NULL;
int argc = 0;
if (sdslen(ldb.cbuf) == 0) return NULL;
/* Working on a copy is simpler in this case. We can modify it freely
* for the sake of simpler parsing. */
sds copy = sdsdup(ldb.cbuf);
char *p = copy;
/* This RESP parser is a joke... just the simplest thing that
* works in this context. It is also very forgiving regarding broken
* protocol. */
/* Seek and parse *<count>\r\n. */
p = strchr(p, '*');
if (!p) goto protoerr;
char *plen = p + 1; /* Multi bulk len pointer. */
p = strstr(p, "\r\n");
if (!p) goto keep_reading;
*p = '\0';
p += 2;
*argcp = atoi(plen);
if (*argcp <= 0 || *argcp > 1024) goto protoerr;
/* Parse each argument. */
argv = zmalloc(sizeof(sds) * (*argcp));
argc = 0;
while (argc < *argcp) {
/* reached the end but there should be more data to read */
if (*p == '\0') goto keep_reading;
if (*p != '$') goto protoerr;
plen = p + 1; /* Bulk string len pointer. */
p = strstr(p, "\r\n");
if (!p) goto keep_reading;
*p = '\0';
p += 2;
int slen = atoi(plen); /* Length of this arg. */
if (slen <= 0 || slen > 1024) goto protoerr;
if ((size_t)(p + slen + 2 - copy) > sdslen(copy)) goto keep_reading;
argv[argc++] = sdsnewlen(p, slen);
p += slen; /* Skip the already parsed argument. */
if (p[0] != '\r' || p[1] != '\n') goto protoerr;
p += 2; /* Skip \r\n. */
}
sdsfree(copy);
return argv;
protoerr:
*err = protocol_error;
keep_reading:
sdsfreesplitres(argv, argc);
sdsfree(copy);
return NULL;
}
/* Log the specified line in the Lua debugger output. */
void ldbLogSourceLine(int lnum) {
char *line = ldbGetSourceLine(lnum);
char *prefix;
int bp = ldbIsBreakpoint(lnum);
int current = ldb.currentline == lnum;
if (current && bp)
prefix = "->#";
else if (current)
prefix = "-> ";
else if (bp)
prefix = " #";
else
prefix = " ";
sds thisline = sdscatprintf(sdsempty(), "%s%-3d %s", prefix, lnum, line);
ldbLog(thisline);
}
/* Implement the "list" command of the Lua debugger. If around is 0
* the whole file is listed, otherwise only a small portion of the file
* around the specified line is shown. When a line number is specified
* the amount of context (lines before/after) is specified via the
* 'context' argument. */
void ldbList(int around, int context) {
int j;
for (j = 1; j <= ldb.lines; j++) {
if (around != 0 && abs(around - j) > context) continue;
ldbLogSourceLine(j);
}
}
/* Append a human readable representation of the Lua value at position 'idx'
* on the stack of the 'lua' state, to the SDS string passed as argument.
* The new SDS string with the represented value attached is returned.
* Used in order to implement ldbLogStackValue().
*
* The element is not automatically removed from the stack, nor it is
* converted to a different type. */
#define LDB_MAX_VALUES_DEPTH (LUA_MINSTACK / 2)
sds ldbCatStackValueRec(sds s, lua_State *lua, int idx, int level) {
int t = lua_type(lua, idx);
if (level++ == LDB_MAX_VALUES_DEPTH) return sdscat(s, "<max recursion level reached! Nested table?>");
switch (t) {
case LUA_TSTRING: {
size_t strl;
char *strp = (char *)lua_tolstring(lua, idx, &strl);
s = sdscatrepr(s, strp, strl);
} break;
case LUA_TBOOLEAN: s = sdscat(s, lua_toboolean(lua, idx) ? "true" : "false"); break;
case LUA_TNUMBER: s = sdscatprintf(s, "%g", (double)lua_tonumber(lua, idx)); break;
case LUA_TNIL: s = sdscatlen(s, "nil", 3); break;
case LUA_TTABLE: {
int expected_index = 1; /* First index we expect in an array. */
int is_array = 1; /* Will be set to null if check fails. */
/* Note: we create two representations at the same time, one
* assuming the table is an array, one assuming it is not. At the
* end we know what is true and select the right one. */
sds repr1 = sdsempty();
sds repr2 = sdsempty();
lua_pushnil(lua); /* The first key to start the iteration is nil. */
while (lua_next(lua, idx - 1)) {
/* Test if so far the table looks like an array. */
if (is_array && (lua_type(lua, -2) != LUA_TNUMBER || lua_tonumber(lua, -2) != expected_index)) is_array = 0;
/* Stack now: table, key, value */
/* Array repr. */
repr1 = ldbCatStackValueRec(repr1, lua, -1, level);
repr1 = sdscatlen(repr1, "; ", 2);
/* Full repr. */
repr2 = sdscatlen(repr2, "[", 1);
repr2 = ldbCatStackValueRec(repr2, lua, -2, level);
repr2 = sdscatlen(repr2, "]=", 2);
repr2 = ldbCatStackValueRec(repr2, lua, -1, level);
repr2 = sdscatlen(repr2, "; ", 2);
lua_pop(lua, 1); /* Stack: table, key. Ready for next iteration. */
expected_index++;
}
/* Strip the last " ;" from both the representations. */
if (sdslen(repr1)) sdsrange(repr1, 0, -3);
if (sdslen(repr2)) sdsrange(repr2, 0, -3);
/* Select the right one and discard the other. */
s = sdscatlen(s, "{", 1);
s = sdscatsds(s, is_array ? repr1 : repr2);
s = sdscatlen(s, "}", 1);
sdsfree(repr1);
sdsfree(repr2);
} break;
case LUA_TFUNCTION:
case LUA_TUSERDATA:
case LUA_TTHREAD:
case LUA_TLIGHTUSERDATA: {
const void *p = lua_topointer(lua, idx);
char *typename = "unknown";
if (t == LUA_TFUNCTION)
typename = "function";
else if (t == LUA_TUSERDATA)
typename = "userdata";
else if (t == LUA_TTHREAD)
typename = "thread";
else if (t == LUA_TLIGHTUSERDATA)
typename = "light-userdata";
s = sdscatprintf(s, "\"%s@%p\"", typename, p);
} break;
default: s = sdscat(s, "\"<unknown-lua-type>\""); break;
}
return s;
}
/* Higher level wrapper for ldbCatStackValueRec() that just uses an initial
* recursion level of '0'. */
sds ldbCatStackValue(sds s, lua_State *lua, int idx) {
return ldbCatStackValueRec(s, lua, idx, 0);
}
/* Produce a debugger log entry representing the value of the Lua object
* currently on the top of the stack. The element is not popped nor modified.
* Check ldbCatStackValue() for the actual implementation. */
void ldbLogStackValue(lua_State *lua, char *prefix) {
sds s = sdsnew(prefix);
s = ldbCatStackValue(s, lua, -1);
ldbLogWithMaxLen(s);
}
char *ldbRespToHuman_Int(sds *o, char *reply);
char *ldbRespToHuman_Bulk(sds *o, char *reply);
char *ldbRespToHuman_Status(sds *o, char *reply);
char *ldbRespToHuman_MultiBulk(sds *o, char *reply);
char *ldbRespToHuman_Set(sds *o, char *reply);
char *ldbRespToHuman_Map(sds *o, char *reply);
char *ldbRespToHuman_Null(sds *o, char *reply);
char *ldbRespToHuman_Bool(sds *o, char *reply);
char *ldbRespToHuman_Double(sds *o, char *reply);
/* Get RESP from 'reply' and appends it in human readable form to
* the passed SDS string 'o'.
*
* Note that the SDS string is passed by reference (pointer of pointer to
* char*) so that we can return a modified pointer, as for SDS semantics. */
char *ldbRespToHuman(sds *o, char *reply) {
char *p = reply;
switch (*p) {
case ':': p = ldbRespToHuman_Int(o, reply); break;
case '$': p = ldbRespToHuman_Bulk(o, reply); break;
case '+': p = ldbRespToHuman_Status(o, reply); break;
case '-': p = ldbRespToHuman_Status(o, reply); break;
case '*': p = ldbRespToHuman_MultiBulk(o, reply); break;
case '~': p = ldbRespToHuman_Set(o, reply); break;
case '%': p = ldbRespToHuman_Map(o, reply); break;
case '_': p = ldbRespToHuman_Null(o, reply); break;
case '#': p = ldbRespToHuman_Bool(o, reply); break;
case ',': p = ldbRespToHuman_Double(o, reply); break;
}
return p;
}
/* The following functions are helpers for ldbRespToHuman(), each
* take care of a given RESP return type. */
char *ldbRespToHuman_Int(sds *o, char *reply) {
char *p = strchr(reply + 1, '\r');
*o = sdscatlen(*o, reply + 1, p - reply - 1);
return p + 2;
}
char *ldbRespToHuman_Bulk(sds *o, char *reply) {
char *p = strchr(reply + 1, '\r');
long long bulklen;
string2ll(reply + 1, p - reply - 1, &bulklen);
if (bulklen == -1) {
*o = sdscatlen(*o, "NULL", 4);
return p + 2;
} else {
*o = sdscatrepr(*o, p + 2, bulklen);
return p + 2 + bulklen + 2;
}
}
char *ldbRespToHuman_Status(sds *o, char *reply) {
char *p = strchr(reply + 1, '\r');
*o = sdscatrepr(*o, reply, p - reply);
return p + 2;
}
char *ldbRespToHuman_MultiBulk(sds *o, char *reply) {
char *p = strchr(reply + 1, '\r');
long long mbulklen;
int j = 0;
string2ll(reply + 1, p - reply - 1, &mbulklen);
p += 2;
if (mbulklen == -1) {
*o = sdscatlen(*o, "NULL", 4);
return p;
}
*o = sdscatlen(*o, "[", 1);
for (j = 0; j < mbulklen; j++) {
p = ldbRespToHuman(o, p);
if (j != mbulklen - 1) *o = sdscatlen(*o, ",", 1);
}
*o = sdscatlen(*o, "]", 1);
return p;
}
char *ldbRespToHuman_Set(sds *o, char *reply) {
char *p = strchr(reply + 1, '\r');
long long mbulklen;
int j = 0;
string2ll(reply + 1, p - reply - 1, &mbulklen);
p += 2;
*o = sdscatlen(*o, "~(", 2);
for (j = 0; j < mbulklen; j++) {
p = ldbRespToHuman(o, p);
if (j != mbulklen - 1) *o = sdscatlen(*o, ",", 1);
}
*o = sdscatlen(*o, ")", 1);
return p;
}
char *ldbRespToHuman_Map(sds *o, char *reply) {
char *p = strchr(reply + 1, '\r');
long long mbulklen;
int j = 0;
string2ll(reply + 1, p - reply - 1, &mbulklen);
p += 2;
*o = sdscatlen(*o, "{", 1);
for (j = 0; j < mbulklen; j++) {
p = ldbRespToHuman(o, p);
*o = sdscatlen(*o, " => ", 4);
p = ldbRespToHuman(o, p);
if (j != mbulklen - 1) *o = sdscatlen(*o, ",", 1);
}
*o = sdscatlen(*o, "}", 1);
return p;
}
char *ldbRespToHuman_Null(sds *o, char *reply) {
char *p = strchr(reply + 1, '\r');
*o = sdscatlen(*o, "(null)", 6);
return p + 2;
}
char *ldbRespToHuman_Bool(sds *o, char *reply) {
char *p = strchr(reply + 1, '\r');
if (reply[1] == 't')
*o = sdscatlen(*o, "#true", 5);
else
*o = sdscatlen(*o, "#false", 6);
return p + 2;
}
char *ldbRespToHuman_Double(sds *o, char *reply) {
char *p = strchr(reply + 1, '\r');
*o = sdscatlen(*o, "(double) ", 9);
*o = sdscatlen(*o, reply + 1, p - reply - 1);
return p + 2;
}
/* Log a RESP reply as debugger output, in a human readable format.
* If the resulting string is longer than 'len' plus a few more chars
* used as prefix, it gets truncated. */
void ldbLogRespReply(char *reply) {
sds log = sdsnew("<reply> ");
ldbRespToHuman(&log, reply);
ldbLogWithMaxLen(log);
}
/* Implements the "print <var>" command of the Lua debugger. It scans for Lua
* var "varname" starting from the current stack frame up to the top stack
* frame. The first matching variable is printed. */
void ldbPrint(lua_State *lua, char *varname) {
lua_Debug ar;
int l = 0; /* Stack level. */
while (lua_getstack(lua, l, &ar) != 0) {
l++;
const char *name;
int i = 1; /* Variable index. */
while ((name = lua_getlocal(lua, &ar, i)) != NULL) {
i++;
if (strcmp(varname, name) == 0) {
ldbLogStackValue(lua, "<value> ");
lua_pop(lua, 1);
return;
} else {
lua_pop(lua, 1); /* Discard the var name on the stack. */
}
}
}
/* Let's try with global vars in two selected cases */
if (!strcmp(varname, "ARGV") || !strcmp(varname, "KEYS")) {
lua_getglobal(lua, varname);
ldbLogStackValue(lua, "<value> ");
lua_pop(lua, 1);
} else {
ldbLog(sdsnew("No such variable."));
}
}
/* Implements the "print" command (without arguments) of the Lua debugger.
* Prints all the variables in the current stack frame. */
void ldbPrintAll(lua_State *lua) {
lua_Debug ar;
int vars = 0;
if (lua_getstack(lua, 0, &ar) != 0) {
const char *name;
int i = 1; /* Variable index. */
while ((name = lua_getlocal(lua, &ar, i)) != NULL) {
i++;
if (!strstr(name, "(*temporary)")) {
sds prefix = sdscatprintf(sdsempty(), "<value> %s = ", name);
ldbLogStackValue(lua, prefix);
sdsfree(prefix);
vars++;
}
lua_pop(lua, 1);
}
}
if (vars == 0) {
ldbLog(sdsnew("No local variables in the current context."));
}
}
/* Implements the break command to list, add and remove breakpoints. */
void ldbBreak(sds *argv, int argc) {
if (argc == 1) {
if (ldb.bpcount == 0) {
ldbLog(sdsnew("No breakpoints set. Use 'b <line>' to add one."));
return;
} else {
ldbLog(sdscatfmt(sdsempty(), "%i breakpoints set:", ldb.bpcount));
int j;
for (j = 0; j < ldb.bpcount; j++) ldbLogSourceLine(ldb.bp[j]);
}
} else {
int j;
for (j = 1; j < argc; j++) {
char *arg = argv[j];
long line;
if (!string2l(arg, sdslen(arg), &line)) {
ldbLog(sdscatfmt(sdsempty(), "Invalid argument:'%s'", arg));
} else {
if (line == 0) {
ldb.bpcount = 0;
ldbLog(sdsnew("All breakpoints removed."));
} else if (line > 0) {
if (ldb.bpcount == LDB_BREAKPOINTS_MAX) {
ldbLog(sdsnew("Too many breakpoints set."));
} else if (ldbAddBreakpoint(line)) {
ldbList(line, 1);
} else {
ldbLog(sdsnew("Wrong line number."));
}
} else if (line < 0) {
if (ldbDelBreakpoint(-line))
ldbLog(sdsnew("Breakpoint removed."));
else
ldbLog(sdsnew("No breakpoint in the specified line."));
}
}
}
}
}
/* Implements the Lua debugger "eval" command. It just compiles the user
* passed fragment of code and executes it, showing the result left on
* the stack. */
void ldbEval(lua_State *lua, sds *argv, int argc) {
/* Glue the script together if it is composed of multiple arguments. */
sds code = sdsjoinsds(argv + 1, argc - 1, " ", 1);
sds expr = sdscatsds(sdsnew("return "), code);
/* Try to compile it as an expression, prepending "return ". */
if (luaL_loadbuffer(lua, expr, sdslen(expr), "@ldb_eval")) {
lua_pop(lua, 1);
/* Failed? Try as a statement. */
if (luaL_loadbuffer(lua, code, sdslen(code), "@ldb_eval")) {
ldbLog(sdscatfmt(sdsempty(), "<error> %s", lua_tostring(lua, -1)));
lua_pop(lua, 1);
sdsfree(code);
sdsfree(expr);
return;
}
}
/* Call it. */
sdsfree(code);
sdsfree(expr);
if (lua_pcall(lua, 0, 1, 0)) {
ldbLog(sdscatfmt(sdsempty(), "<error> %s", lua_tostring(lua, -1)));
lua_pop(lua, 1);
return;
}
ldbLogStackValue(lua, "<retval> ");
lua_pop(lua, 1);
}
/* Implement the debugger "server" command. We use a trick in order to make
* the implementation very simple: we just call the Lua server.call() command
* implementation, with ldb.step enabled, so as a side effect the command
* and its reply are logged. */
void ldbServer(lua_State *lua, sds *argv, int argc) {
int j;
if (!lua_checkstack(lua, argc + 1)) {
/* Increase the Lua stack if needed to make sure there is enough room
* to push 'argc + 1' elements to the stack. On failure, return error.
* Notice that we need, in worst case, 'argc + 1' elements because we push all the arguments
* given by the user (without the first argument) and we also push the 'server' global table and
* 'server.call' function so:
* (1 (server table)) + (1 (server.call function)) + (argc - 1 (all arguments without the first)) = argc + 1*/
ldbLogRespReply("max lua stack reached");
return;
}
lua_getglobal(lua, "server");
lua_pushstring(lua, "call");
lua_gettable(lua, -2); /* Stack: server, server.call */
for (j = 1; j < argc; j++) lua_pushlstring(lua, argv[j], sdslen(argv[j]));
ldb.step = 1; /* Force server.call() to log. */
lua_pcall(lua, argc - 1, 1, 0); /* Stack: server, result */
ldb.step = 0; /* Disable logging. */
lua_pop(lua, 2); /* Discard the result and clean the stack. */
}
/* Implements "trace" command of the Lua debugger. It just prints a backtrace
* querying Lua starting from the current callframe back to the outer one. */
void ldbTrace(lua_State *lua) {
lua_Debug ar;
int level = 0;
while (lua_getstack(lua, level, &ar)) {
lua_getinfo(lua, "Snl", &ar);
if (strstr(ar.short_src, "user_script") != NULL) {
ldbLog(sdscatprintf(sdsempty(), "%s %s:", (level == 0) ? "In" : "From", ar.name ? ar.name : "top level"));
ldbLogSourceLine(ar.currentline);
}
level++;
}
if (level == 0) {
ldbLog(sdsnew("<error> Can't retrieve Lua stack."));
}
}
/* Implements the debugger "maxlen" command. It just queries or sets the
* ldb.maxlen variable. */
void ldbMaxlen(sds *argv, int argc) {
if (argc == 2) {
int newval = atoi(argv[1]);
ldb.maxlen_hint_sent = 1; /* User knows about this command. */
if (newval != 0 && newval <= 60) newval = 60;
ldb.maxlen = newval;
}
if (ldb.maxlen) {
ldbLog(sdscatprintf(sdsempty(), "<value> replies are truncated at %d bytes.", (int)ldb.maxlen));
} else {
ldbLog(sdscatprintf(sdsempty(), "<value> replies are unlimited."));
}
}
/* Read debugging commands from client.
* Return C_OK if the debugging session is continuing, otherwise
* C_ERR if the client closed the connection or is timing out. */
int ldbRepl(lua_State *lua) {
sds *argv;
int argc;
char *err = NULL;
/* We continue processing commands until a command that should return
* to the Lua interpreter is found. */
while (1) {
while ((argv = ldbReplParseCommand(&argc, &err)) == NULL) {
char buf[1024];
if (err) {
luaPushError(lua, err);
luaError(lua);
}
int nread = connRead(ldb.conn, buf, sizeof(buf));
if (nread <= 0) {
/* Make sure the script runs without user input since the
* client is no longer connected. */
ldb.step = 0;
ldb.bpcount = 0;
return C_ERR;
}
ldb.cbuf = sdscatlen(ldb.cbuf, buf, nread);
/* after 1M we will exit with an error
* so that the client will not blow the memory
*/
if (sdslen(ldb.cbuf) > 1 << 20) {
sdsfree(ldb.cbuf);
ldb.cbuf = sdsempty();
luaPushError(lua, "max client buffer reached");
luaError(lua);
}
}
/* Flush the old buffer. */
sdsfree(ldb.cbuf);
ldb.cbuf = sdsempty();
/* Execute the command. */
if (!strcasecmp(argv[0], "h") || !strcasecmp(argv[0], "help")) {
ldbLog(sdsnew("Lua debugger help:"));
ldbLog(sdsnew("[h]elp Show this help."));
ldbLog(sdsnew("[s]tep Run current line and stop again."));
ldbLog(sdsnew("[n]ext Alias for step."));
ldbLog(sdsnew("[c]ontinue Run till next breakpoint."));
ldbLog(sdsnew("[l]ist List source code around current line."));
ldbLog(sdsnew("[l]ist [line] List source code around [line]."));
ldbLog(sdsnew(" line = 0 means: current position."));
ldbLog(sdsnew("[l]ist [line] [ctx] In this form [ctx] specifies how many lines"));
ldbLog(sdsnew(" to show before/after [line]."));
ldbLog(sdsnew("[w]hole List all source code. Alias for 'list 1 1000000'."));
ldbLog(sdsnew("[p]rint Show all the local variables."));
ldbLog(sdsnew("[p]rint <var> Show the value of the specified variable."));
ldbLog(sdsnew(" Can also show global vars KEYS and ARGV."));
ldbLog(sdsnew("[b]reak Show all breakpoints."));
ldbLog(sdsnew("[b]reak <line> Add a breakpoint to the specified line."));
ldbLog(sdsnew("[b]reak -<line> Remove breakpoint from the specified line."));
ldbLog(sdsnew("[b]reak 0 Remove all breakpoints."));
ldbLog(sdsnew("[t]race Show a backtrace."));
ldbLog(sdsnew("[e]val <code> Execute some Lua code (in a different callframe)."));
ldbLog(sdsnew("[v]alkey <cmd> Execute a command."));
ldbLog(sdsnew("[m]axlen [len] Trim logged replies and Lua var dumps to len."));
ldbLog(sdsnew(" Specifying zero as <len> means unlimited."));
ldbLog(sdsnew("[a]bort Stop the execution of the script. In sync"));
ldbLog(sdsnew(" mode dataset changes will be retained."));
ldbLog(sdsnew(""));
ldbLog(sdsnew("Debugger functions you can call from Lua scripts:"));
ldbLog(sdsnew("server.debug() Produce logs in the debugger console."));
ldbLog(sdsnew("server.breakpoint() Stop execution like if there was a breakpoint in the"));
ldbLog(sdsnew(" next line of code."));
ldbSendLogs();
} else if (!strcasecmp(argv[0], "s") || !strcasecmp(argv[0], "step") || !strcasecmp(argv[0], "n") ||
!strcasecmp(argv[0], "next")) {
ldb.step = 1;
break;
} else if (!strcasecmp(argv[0], "c") || !strcasecmp(argv[0], "continue")) {
break;
} else if (!strcasecmp(argv[0], "t") || !strcasecmp(argv[0], "trace")) {
ldbTrace(lua);
ldbSendLogs();
} else if (!strcasecmp(argv[0], "m") || !strcasecmp(argv[0], "maxlen")) {
ldbMaxlen(argv, argc);
ldbSendLogs();
} else if (!strcasecmp(argv[0], "b") || !strcasecmp(argv[0], "break")) {
ldbBreak(argv, argc);
ldbSendLogs();
} else if (!strcasecmp(argv[0], "e") || !strcasecmp(argv[0], "eval")) {
ldbEval(lua, argv, argc);
ldbSendLogs();
} else if (!strcasecmp(argv[0], "a") || !strcasecmp(argv[0], "abort")) {
luaPushError(lua, "script aborted for user request");
luaError(lua);
} else if (argc > 1 && ((!strcasecmp(argv[0], "r") || !strcasecmp(argv[0], "redis")) ||
(!strcasecmp(argv[0], "v") || !strcasecmp(argv[0], "valkey")) ||
!strcasecmp(argv[0], SERVER_API_NAME))) {
/* [r]redis or [v]alkey calls a command. We accept "server" too, but
* not "s" because that's "step". Neither can we use [c]all because
* "c" is continue. */
ldbServer(lua, argv, argc);
ldbSendLogs();
} else if ((!strcasecmp(argv[0], "p") || !strcasecmp(argv[0], "print"))) {
if (argc == 2)
ldbPrint(lua, argv[1]);
else
ldbPrintAll(lua);
ldbSendLogs();
} else if (!strcasecmp(argv[0], "l") || !strcasecmp(argv[0], "list")) {
int around = ldb.currentline, ctx = 5;
if (argc > 1) {
int num = atoi(argv[1]);
if (num > 0) around = num;
}
if (argc > 2) ctx = atoi(argv[2]);
ldbList(around, ctx);
ldbSendLogs();
} else if (!strcasecmp(argv[0], "w") || !strcasecmp(argv[0], "whole")) {
ldbList(1, 1000000);
ldbSendLogs();
} else {
ldbLog(sdsnew("<error> Unknown Lua debugger command or "
"wrong number of arguments."));
ldbSendLogs();
}
/* Free the command vector. */
sdsfreesplitres(argv, argc);
}
/* Free the current command argv if we break inside the while loop. */
sdsfreesplitres(argv, argc);
return C_OK;
}
/* This is the core of our Lua debugger, called each time Lua is about
* to start executing a new line. */
void luaLdbLineHook(lua_State *lua, lua_Debug *ar) {
scriptRunCtx *rctx = luaGetFromRegistry(lua, REGISTRY_RUN_CTX_NAME);
serverAssert(rctx); /* Only supported inside script invocation */
lua_getstack(lua, 0, ar);
lua_getinfo(lua, "Sl", ar);
ldb.currentline = ar->currentline;
int bp = ldbIsBreakpoint(ldb.currentline) || ldb.luabp;
int timeout = 0;
/* Events outside our script are not interesting. */
if (strstr(ar->short_src, "user_script") == NULL) return;
/* Check if a timeout occurred. */
if (ar->event == LUA_HOOKCOUNT && ldb.step == 0 && bp == 0) {
mstime_t elapsed = elapsedMs(rctx->start_time);
mstime_t timelimit = server.busy_reply_threshold ? server.busy_reply_threshold : 5000;
if (elapsed >= timelimit) {
timeout = 1;
ldb.step = 1;
} else {
return; /* No timeout, ignore the COUNT event. */
}
}
if (ldb.step || bp) {
char *reason = "step over";
if (bp)
reason = ldb.luabp ? "server.breakpoint() called" : "break point";
else if (timeout)
reason = "timeout reached, infinite loop?";
ldb.step = 0;
ldb.luabp = 0;
ldbLog(sdscatprintf(sdsempty(), "* Stopped at %d, stop reason = %s", ldb.currentline, reason));
ldbLogSourceLine(ldb.currentline);
ldbSendLogs();
if (ldbRepl(lua) == C_ERR && timeout) {
/* If the client closed the connection and we have a timeout
* connection, let's kill the script otherwise the process
* will remain blocked indefinitely. */
luaPushError(lua, "timeout during Lua debugging with client closing connection");
luaError(lua);
}
rctx->start_time = getMonotonicUs();
}
}
Reference in New Issue View Git Blame Copy Permalink