diff --git a/src/Makefile b/src/Makefile
index 34b5c3566..469e8eb54 100644
--- a/src/Makefile
+++ b/src/Makefile
@@ -309,7 +309,7 @@ endif
 
 REDIS_SERVER_NAME=redis-server$(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
 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_BENCHMARK_NAME=redis-benchmark$(PROG_SUFFIX)
diff --git a/src/scripting.c b/src/eval.c
similarity index 55%
rename from src/scripting.c
rename to src/eval.c
index f27840956..6e6aeac7c 100644
--- a/src/scripting.c
+++ b/src/eval.c
@@ -33,6 +33,7 @@
 #include "cluster.h"
 #include "monotonic.h"
 #include "resp_parser.h"
+#include "script_lua.h"
 
 #include <lua.h>
 #include <lauxlib.h>
@@ -40,23 +41,6 @@
 #include <ctype.h>
 #include <math.h>
 
-static void redisProtocolToLuaType_Int(void *ctx, long long val, const char *proto, size_t proto_len);
-static void redisProtocolToLuaType_BulkString(void *ctx, const char *str, size_t len, const char *proto, size_t proto_len);
-static void redisProtocolToLuaType_NullBulkString(void *ctx, const char *proto, size_t proto_len);
-static void redisProtocolToLuaType_NullArray(void *ctx, const char *proto, size_t proto_len);
-static void redisProtocolToLuaType_Status(void *ctx, const char *str, size_t len, const char *proto, size_t proto_len);
-static void redisProtocolToLuaType_Error(void *ctx, const char *str, size_t len, const char *proto, size_t proto_len);
-static void redisProtocolToLuaType_Array(struct ReplyParser *parser, void *ctx, size_t len, const char *proto);
-static void redisProtocolToLuaType_Map(struct ReplyParser *parser, void *ctx, size_t len, const char *proto);
-static void redisProtocolToLuaType_Set(struct ReplyParser *parser, void *ctx, size_t len, const char *proto);
-static void redisProtocolToLuaType_Null(void *ctx, const char *proto, size_t proto_len);
-static void redisProtocolToLuaType_Bool(void *ctx, int val, const char *proto, size_t proto_len);
-static void redisProtocolToLuaType_Double(void *ctx, double d, const char *proto, size_t proto_len);
-static void redisProtocolToLuaType_BigNumber(void *ctx, const char *str, size_t len, const char *proto, size_t proto_len);
-static void redisProtocolToLuaType_VerbatimString(void *ctx, const char *format, const char *str, size_t len, const char *proto, size_t proto_len);
-static void redisProtocolToLuaType_Attribute(struct ReplyParser *parser, void *ctx, size_t len, const char *proto);
-int redis_math_random (lua_State *L);
-int redis_math_randomseed (lua_State *L);
 void ldbInit(void);
 void ldbDisable(client *c);
 void ldbEnable(client *c);
@@ -115,1035 +99,6 @@ void sha1hex(char *digest, char *script, size_t len) {
     digest[40] = '\0';
 }
 
-/* ---------------------------------------------------------------------------
- * Redis reply to Lua type conversion functions.
- * ------------------------------------------------------------------------- */
-
-/* Take a Redis reply in the Redis protocol format and convert it into a
- * Lua type. Thanks to this function, and the introduction of not connected
- * clients, it is trivial to implement the redis() lua function.
- *
- * Basically we take the arguments, execute the Redis command in the context
- * of a non connected client, then take the generated reply and convert it
- * into a suitable Lua type. With this trick the scripting feature does not
- * need the introduction of a full Redis internals API. The script
- * is like a normal client that bypasses all the slow I/O paths.
- *
- * Note: in this function we do not do any sanity check as the reply is
- * generated by Redis directly. This allows us to go faster.
- *
- * Errors are returned as a table with a single 'err' field set to the
- * error string.
- */
-
-static const ReplyParserCallbacks DefaultLuaTypeParserCallbacks = {
-    .null_array_callback = redisProtocolToLuaType_NullArray,
-    .bulk_string_callback = redisProtocolToLuaType_BulkString,
-    .null_bulk_string_callback = redisProtocolToLuaType_NullBulkString,
-    .error_callback = redisProtocolToLuaType_Error,
-    .simple_str_callback = redisProtocolToLuaType_Status,
-    .long_callback = redisProtocolToLuaType_Int,
-    .array_callback = redisProtocolToLuaType_Array,
-    .set_callback = redisProtocolToLuaType_Set,
-    .map_callback = redisProtocolToLuaType_Map,
-    .bool_callback = redisProtocolToLuaType_Bool,
-    .double_callback = redisProtocolToLuaType_Double,
-    .null_callback = redisProtocolToLuaType_Null,
-    .big_number_callback = redisProtocolToLuaType_BigNumber,
-    .verbatim_string_callback = redisProtocolToLuaType_VerbatimString,
-    .attribute_callback = redisProtocolToLuaType_Attribute,
-    .error = NULL,
-};
-
-void redisProtocolToLuaType(lua_State *lua, char* reply) {
-    ReplyParser parser = {.curr_location = reply, .callbacks = DefaultLuaTypeParserCallbacks};
-
-    parseReply(&parser, lua);
-}
-
-static void redisProtocolToLuaType_Int(void *ctx, long long val, const char *proto, size_t proto_len) {
-    UNUSED(proto);
-    UNUSED(proto_len);
-    if (!ctx) {
-        return;
-    }
-
-    lua_State *lua = ctx;
-    if (!lua_checkstack(lua, 1)) {
-        /* Increase the Lua stack if needed, to make sure there is enough room
-         * to push elements to the stack. On failure, exit with panic. */
-        serverPanic("lua stack limit reach when parsing redis.call reply");
-    }
-    lua_pushnumber(lua,(lua_Number)val);
-}
-
-static void redisProtocolToLuaType_NullBulkString(void *ctx, const char *proto, size_t proto_len) {
-    UNUSED(proto);
-    UNUSED(proto_len);
-    if (!ctx) {
-        return;
-    }
-
-    lua_State *lua = ctx;
-    if (!lua_checkstack(lua, 1)) {
-        /* Increase the Lua stack if needed, to make sure there is enough room
-         * to push elements to the stack. On failure, exit with panic. */
-        serverPanic("lua stack limit reach when parsing redis.call reply");
-    }
-    lua_pushboolean(lua,0);
-}
-
-static void redisProtocolToLuaType_NullArray(void *ctx, const char *proto, size_t proto_len) {
-    UNUSED(proto);
-    UNUSED(proto_len);
-    if (!ctx) {
-        return;
-    }
-    lua_State *lua = ctx;
-    if (!lua_checkstack(lua, 1)) {
-        /* Increase the Lua stack if needed, to make sure there is enough room
-         * to push elements to the stack. On failure, exit with panic. */
-        serverPanic("lua stack limit reach when parsing redis.call reply");
-    }
-    lua_pushboolean(lua,0);
-}
-
-
-static void redisProtocolToLuaType_BulkString(void *ctx, const char *str, size_t len, const char *proto, size_t proto_len) {
-    UNUSED(proto);
-    UNUSED(proto_len);
-    if (!ctx) {
-        return;
-    }
-
-    lua_State *lua = ctx;
-    if (!lua_checkstack(lua, 1)) {
-        /* Increase the Lua stack if needed, to make sure there is enough room
-         * to push elements to the stack. On failure, exit with panic. */
-        serverPanic("lua stack limit reach when parsing redis.call reply");
-    }
-    lua_pushlstring(lua,str,len);
-}
-
-static void redisProtocolToLuaType_Status(void *ctx, const char *str, size_t len, const char *proto, size_t proto_len) {
-    UNUSED(proto);
-    UNUSED(proto_len);
-    if (!ctx) {
-        return;
-    }
-
-    lua_State *lua = ctx;
-    if (!lua_checkstack(lua, 3)) {
-        /* Increase the Lua stack if needed, to make sure there is enough room
-         * to push elements to the stack. On failure, exit with panic. */
-        serverPanic("lua stack limit reach when parsing redis.call reply");
-    }
-    lua_newtable(lua);
-    lua_pushstring(lua,"ok");
-    lua_pushlstring(lua,str,len);
-    lua_settable(lua,-3);
-}
-
-static void redisProtocolToLuaType_Error(void *ctx, const char *str, size_t len, const char *proto, size_t proto_len) {
-    UNUSED(proto);
-    UNUSED(proto_len);
-    if (!ctx) {
-        return;
-    }
-
-    lua_State *lua = ctx;
-    if (!lua_checkstack(lua, 3)) {
-        /* Increase the Lua stack if needed, to make sure there is enough room
-         * to push elements to the stack. On failure, exit with panic. */
-        serverPanic("lua stack limit reach when parsing redis.call reply");
-    }
-    lua_newtable(lua);
-    lua_pushstring(lua,"err");
-    lua_pushlstring(lua,str,len);
-    lua_settable(lua,-3);
-}
-
-static void redisProtocolToLuaType_Map(struct ReplyParser *parser, void *ctx, size_t len, const char *proto) {
-    UNUSED(proto);
-    lua_State *lua = ctx;
-    if (lua) {
-        if (!lua_checkstack(lua, 3)) {
-            /* Increase the Lua stack if needed, to make sure there is enough room
-             * to push elements to the stack. On failure, exit with panic. */
-            serverPanic("lua stack limit reach when parsing redis.call reply");
-        }
-        lua_newtable(lua);
-        lua_pushstring(lua, "map");
-        lua_newtable(lua);
-    }
-    for (size_t j = 0; j < len; j++) {
-        parseReply(parser,lua);
-        parseReply(parser,lua);
-        if (lua) lua_settable(lua,-3);
-    }
-    if (lua) lua_settable(lua,-3);
-}
-
-static void redisProtocolToLuaType_Set(struct ReplyParser *parser, void *ctx, size_t len, const char *proto) {
-    UNUSED(proto);
-
-    lua_State *lua = ctx;
-    if (lua) {
-        if (!lua_checkstack(lua, 3)) {
-            /* Increase the Lua stack if needed, to make sure there is enough room
-             * to push elements to the stack. On failure, exit with panic. */
-            serverPanic("lua stack limit reach when parsing redis.call reply");
-        }
-        lua_newtable(lua);
-        lua_pushstring(lua, "set");
-        lua_newtable(lua);
-    }
-    for (size_t j = 0; j < len; j++) {
-        parseReply(parser,lua);
-        if (lua) {
-            if (!lua_checkstack(lua, 1)) {
-                /* Increase the Lua stack if needed, to make sure there is enough room
-                 * to push elements to the stack. On failure, exit with panic.
-                 * Notice that here we need to check the stack again because the recursive
-                 * call to redisProtocolToLuaType might have use the room allocated in the stack*/
-                serverPanic("lua stack limit reach when parsing redis.call reply");
-            }
-            lua_pushboolean(lua,1);
-            lua_settable(lua,-3);
-        }
-    }
-    if (lua) lua_settable(lua,-3);
-}
-
-static void redisProtocolToLuaType_Array(struct ReplyParser *parser, void *ctx, size_t len, const char *proto) {
-    UNUSED(proto);
-
-    lua_State *lua = ctx;
-    if (lua){
-        if (!lua_checkstack(lua, 2)) {
-            /* Increase the Lua stack if needed, to make sure there is enough room
-             * to push elements to the stack. On failure, exit with panic. */
-            serverPanic("lua stack limit reach when parsing redis.call reply");
-        }
-        lua_newtable(lua);
-    }
-    for (size_t j = 0; j < len; j++) {
-        if (lua) lua_pushnumber(lua,j+1);
-        parseReply(parser,lua);
-        if (lua) lua_settable(lua,-3);
-    }
-}
-
-static void redisProtocolToLuaType_Attribute(struct ReplyParser *parser, void *ctx, size_t len, const char *proto) {
-    UNUSED(proto);
-
-    /* Parse the attribute reply.
-     * Currently, we do not expose the attribute to the Lua script so
-     * we just need to continue parsing and ignore it (the NULL ensures that the
-     * reply will be ignored). */
-    for (size_t j = 0; j < len; j++) {
-        parseReply(parser,NULL);
-        parseReply(parser,NULL);
-    }
-
-    /* Parse the reply itself. */
-    parseReply(parser,ctx);
-}
-
-static void redisProtocolToLuaType_VerbatimString(void *ctx, const char *format, const char *str, size_t len, const char *proto, size_t proto_len) {
-    UNUSED(proto);
-    UNUSED(proto_len);
-    if (!ctx) {
-        return;
-    }
-
-    lua_State *lua = ctx;
-    if (!lua_checkstack(lua, 5)) {
-        /* Increase the Lua stack if needed, to make sure there is enough room
-         * to push elements to the stack. On failure, exit with panic. */
-        serverPanic("lua stack limit reach when parsing redis.call reply");
-    }
-    lua_newtable(lua);
-    lua_pushstring(lua,"verbatim_string");
-    lua_newtable(lua);
-    lua_pushstring(lua,"string");
-    lua_pushlstring(lua,str,len);
-    lua_settable(lua,-3);
-    lua_pushstring(lua,"format");
-    lua_pushlstring(lua,format,3);
-    lua_settable(lua,-3);
-    lua_settable(lua,-3);
-}
-
-static void redisProtocolToLuaType_BigNumber(void *ctx, const char *str, size_t len, const char *proto, size_t proto_len) {
-    UNUSED(proto);
-    UNUSED(proto_len);
-    if (!ctx) {
-        return;
-    }
-
-    lua_State *lua = ctx;
-    if (!lua_checkstack(lua, 3)) {
-        /* Increase the Lua stack if needed, to make sure there is enough room
-         * to push elements to the stack. On failure, exit with panic. */
-        serverPanic("lua stack limit reach when parsing redis.call reply");
-    }
-    lua_newtable(lua);
-    lua_pushstring(lua,"big_number");
-    lua_pushlstring(lua,str,len);
-    lua_settable(lua,-3);
-}
-
-static void redisProtocolToLuaType_Null(void *ctx, const char *proto, size_t proto_len) {
-    UNUSED(proto);
-    UNUSED(proto_len);
-    if (!ctx) {
-        return;
-    }
-
-    lua_State *lua = ctx;
-    if (!lua_checkstack(lua, 1)) {
-        /* Increase the Lua stack if needed, to make sure there is enough room
-         * to push elements to the stack. On failure, exit with panic. */
-        serverPanic("lua stack limit reach when parsing redis.call reply");
-    }
-    lua_pushnil(lua);
-}
-
-static void redisProtocolToLuaType_Bool(void *ctx, int val, const char *proto, size_t proto_len) {
-    UNUSED(proto);
-    UNUSED(proto_len);
-    if (!ctx) {
-        return;
-    }
-
-    lua_State *lua = ctx;
-    if (!lua_checkstack(lua, 1)) {
-        /* Increase the Lua stack if needed, to make sure there is enough room
-         * to push elements to the stack. On failure, exit with panic. */
-        serverPanic("lua stack limit reach when parsing redis.call reply");
-    }
-    lua_pushboolean(lua,val);
-}
-
-static void redisProtocolToLuaType_Double(void *ctx, double d, const char *proto, size_t proto_len) {
-    UNUSED(proto);
-    UNUSED(proto_len);
-    if (!ctx) {
-        return;
-    }
-
-    lua_State *lua = ctx;
-    if (!lua_checkstack(lua, 3)) {
-        /* Increase the Lua stack if needed, to make sure there is enough room
-         * to push elements to the stack. On failure, exit with panic. */
-        serverPanic("lua stack limit reach when parsing redis.call reply");
-    }
-    lua_newtable(lua);
-    lua_pushstring(lua,"double");
-    lua_pushnumber(lua,d);
-    lua_settable(lua,-3);
-}
-
-/* This function is used in order to push an error on the Lua stack in the
- * format used by redis.pcall to return errors, which is a lua table
- * with a single "err" field set to the error string. Note that this
- * table is never a valid reply by proper commands, since the returned
- * tables are otherwise always indexed by integers, never by strings. */
-void luaPushError(lua_State *lua, char *error) {
-    lua_Debug dbg;
-
-    /* If debugging is active and in step mode, log errors resulting from
-     * Redis commands. */
-    if (ldb.active && ldb.step) {
-        ldbLog(sdscatprintf(sdsempty(),"<error> %s",error));
-    }
-
-    lua_newtable(lua);
-    lua_pushstring(lua,"err");
-
-    /* Attempt to figure out where this function was called, if possible */
-    if(lua_getstack(lua, 1, &dbg) && lua_getinfo(lua, "nSl", &dbg)) {
-        sds msg = sdscatprintf(sdsempty(), "%s: %d: %s",
-            dbg.source, dbg.currentline, error);
-        lua_pushstring(lua, msg);
-        sdsfree(msg);
-    } else {
-        lua_pushstring(lua, error);
-    }
-    lua_settable(lua,-3);
-}
-
-/* In case the error set into the Lua stack by luaPushError() was generated
- * by the non-error-trapping version of redis.pcall(), which is redis.call(),
- * this function will raise the Lua error so that the execution of the
- * script will be halted. */
-int luaRaiseError(lua_State *lua) {
-    lua_pushstring(lua,"err");
-    lua_gettable(lua,-2);
-    return lua_error(lua);
-}
-
-/* Sort the array currently in the stack. We do this to make the output
- * of commands like KEYS or SMEMBERS something deterministic when called
- * from Lua (to play well with AOf/replication).
- *
- * The array is sorted using table.sort itself, and assuming all the
- * list elements are strings. */
-void luaSortArray(lua_State *lua) {
-    /* Initial Stack: array */
-    lua_getglobal(lua,"table");
-    lua_pushstring(lua,"sort");
-    lua_gettable(lua,-2);       /* Stack: array, table, table.sort */
-    lua_pushvalue(lua,-3);      /* Stack: array, table, table.sort, array */
-    if (lua_pcall(lua,1,0,0)) {
-        /* Stack: array, table, error */
-
-        /* We are not interested in the error, we assume that the problem is
-         * that there are 'false' elements inside the array, so we try
-         * again with a slower function but able to handle this case, that
-         * is: table.sort(table, __redis__compare_helper) */
-        lua_pop(lua,1);             /* Stack: array, table */
-        lua_pushstring(lua,"sort"); /* Stack: array, table, sort */
-        lua_gettable(lua,-2);       /* Stack: array, table, table.sort */
-        lua_pushvalue(lua,-3);      /* Stack: array, table, table.sort, array */
-        lua_getglobal(lua,"__redis__compare_helper");
-        /* Stack: array, table, table.sort, array, __redis__compare_helper */
-        lua_call(lua,2,0);
-    }
-    /* Stack: array (sorted), table */
-    lua_pop(lua,1);             /* Stack: array (sorted) */
-}
-
-/* ---------------------------------------------------------------------------
- * Lua reply to Redis reply conversion functions.
- * ------------------------------------------------------------------------- */
-
-/* Reply to client 'c' converting the top element in the Lua stack to a
- * Redis reply. As a side effect the element is consumed from the stack.  */
-void luaReplyToRedisReply(client *c, lua_State *lua) {
-    int t = lua_type(lua,-1);
-
-    if (!lua_checkstack(lua, 4)) {
-        /* Increase the Lua stack if needed to make sure there is enough room
-         * to push 4 elements to the stack. On failure, return error.
-         * Notice that we need, in the worst case, 4 elements because returning a map might
-         * require push 4 elements to the Lua stack.*/
-        addReplyErrorFormat(c, "reached lua stack limit");
-        lua_pop(lua,1); /* pop the element from the stack */
-        return;
-    }
-
-    switch(t) {
-    case LUA_TSTRING:
-        addReplyBulkCBuffer(c,(char*)lua_tostring(lua,-1),lua_strlen(lua,-1));
-        break;
-    case LUA_TBOOLEAN:
-        if (server.lua_client->resp == 2)
-            addReply(c,lua_toboolean(lua,-1) ? shared.cone :
-                                               shared.null[c->resp]);
-        else
-            addReplyBool(c,lua_toboolean(lua,-1));
-        break;
-    case LUA_TNUMBER:
-        addReplyLongLong(c,(long long)lua_tonumber(lua,-1));
-        break;
-    case LUA_TTABLE:
-        /* We need to check if it is an array, an error, or a status reply.
-         * Error are returned as a single element table with 'err' field.
-         * Status replies are returned as single element table with 'ok'
-         * field. */
-
-        /* Handle error reply. */
-        /* we took care of the stack size on function start */
-        lua_pushstring(lua,"err");
-        lua_gettable(lua,-2);
-        t = lua_type(lua,-1);
-        if (t == LUA_TSTRING) {
-            addReplyErrorFormat(c,"-%s",lua_tostring(lua,-1));
-            lua_pop(lua,2);
-            return;
-        }
-        lua_pop(lua,1); /* Discard field name pushed before. */
-
-        /* Handle status reply. */
-        lua_pushstring(lua,"ok");
-        lua_gettable(lua,-2);
-        t = lua_type(lua,-1);
-        if (t == LUA_TSTRING) {
-            sds ok = sdsnew(lua_tostring(lua,-1));
-            sdsmapchars(ok,"\r\n","  ",2);
-            addReplySds(c,sdscatprintf(sdsempty(),"+%s\r\n",ok));
-            sdsfree(ok);
-            lua_pop(lua,2);
-            return;
-        }
-        lua_pop(lua,1); /* Discard field name pushed before. */
-
-        /* Handle double reply. */
-        lua_pushstring(lua,"double");
-        lua_gettable(lua,-2);
-        t = lua_type(lua,-1);
-        if (t == LUA_TNUMBER) {
-            addReplyDouble(c,lua_tonumber(lua,-1));
-            lua_pop(lua,2);
-            return;
-        }
-        lua_pop(lua,1); /* Discard field name pushed before. */
-
-        /* Handle big number reply. */
-        lua_pushstring(lua,"big_number");
-        lua_gettable(lua,-2);
-        t = lua_type(lua,-1);
-        if (t == LUA_TSTRING) {
-            sds big_num = sdsnewlen(lua_tostring(lua,-1), lua_strlen(lua,-1));
-            sdsmapchars(big_num,"\r\n","  ",2);
-            addReplyBigNum(c,big_num,sdslen(big_num));
-            sdsfree(big_num);
-            lua_pop(lua,2);
-            return;
-        }
-        lua_pop(lua,1); /* Discard field name pushed before. */
-
-        /* Handle verbatim reply. */
-        lua_pushstring(lua,"verbatim_string");
-        lua_gettable(lua,-2);
-        t = lua_type(lua,-1);
-        if (t == LUA_TTABLE) {
-            lua_pushstring(lua,"format");
-            lua_gettable(lua,-2);
-            t = lua_type(lua,-1);
-            if (t == LUA_TSTRING){
-                char* format = (char*)lua_tostring(lua,-1);
-                lua_pushstring(lua,"string");
-                lua_gettable(lua,-3);
-                t = lua_type(lua,-1);
-                if (t == LUA_TSTRING){
-                    size_t len;
-                    char* str = (char*)lua_tolstring(lua,-1,&len);
-                    addReplyVerbatim(c, str, len, format);
-                    lua_pop(lua,4);
-                    return;
-                }
-                lua_pop(lua,1);
-            }
-            lua_pop(lua,1);
-        }
-        lua_pop(lua,1); /* Discard field name pushed before. */
-
-        /* Handle map reply. */
-        lua_pushstring(lua,"map");
-        lua_gettable(lua,-2);
-        t = lua_type(lua,-1);
-        if (t == LUA_TTABLE) {
-            int maplen = 0;
-            void *replylen = addReplyDeferredLen(c);
-            /* we took care of the stack size on function start */
-            lua_pushnil(lua); /* Use nil to start iteration. */
-            while (lua_next(lua,-2)) {
-                /* Stack now: table, key, value */
-                lua_pushvalue(lua,-2);        /* Dup key before consuming. */
-                luaReplyToRedisReply(c, lua); /* Return key. */
-                luaReplyToRedisReply(c, lua); /* Return value. */
-                /* Stack now: table, key. */
-                maplen++;
-            }
-            setDeferredMapLen(c,replylen,maplen);
-            lua_pop(lua,2);
-            return;
-        }
-        lua_pop(lua,1); /* Discard field name pushed before. */
-
-        /* Handle set reply. */
-        lua_pushstring(lua,"set");
-        lua_gettable(lua,-2);
-        t = lua_type(lua,-1);
-        if (t == LUA_TTABLE) {
-            int setlen = 0;
-            void *replylen = addReplyDeferredLen(c);
-            /* we took care of the stack size on function start */
-            lua_pushnil(lua); /* Use nil to start iteration. */
-            while (lua_next(lua,-2)) {
-                /* Stack now: table, key, true */
-                lua_pop(lua,1);               /* Discard the boolean value. */
-                lua_pushvalue(lua,-1);        /* Dup key before consuming. */
-                luaReplyToRedisReply(c, lua); /* Return key. */
-                /* Stack now: table, key. */
-                setlen++;
-            }
-            setDeferredSetLen(c,replylen,setlen);
-            lua_pop(lua,2);
-            return;
-        }
-        lua_pop(lua,1); /* Discard field name pushed before. */
-
-        /* Handle the array reply. */
-        void *replylen = addReplyDeferredLen(c);
-        int j = 1, mbulklen = 0;
-        while(1) {
-            /* we took care of the stack size on function start */
-            lua_pushnumber(lua,j++);
-            lua_gettable(lua,-2);
-            t = lua_type(lua,-1);
-            if (t == LUA_TNIL) {
-                lua_pop(lua,1);
-                break;
-            }
-            luaReplyToRedisReply(c, lua);
-            mbulklen++;
-        }
-        setDeferredArrayLen(c,replylen,mbulklen);
-        break;
-    default:
-        addReplyNull(c);
-    }
-    lua_pop(lua,1);
-}
-
-/* ---------------------------------------------------------------------------
- * Lua redis.* functions implementations.
- * ------------------------------------------------------------------------- */
-
-#define LUA_CMD_OBJCACHE_SIZE 32
-#define LUA_CMD_OBJCACHE_MAX_LEN 64
-int luaRedisGenericCommand(lua_State *lua, int raise_error) {
-    int j, argc = lua_gettop(lua);
-    struct redisCommand *cmd;
-    client *c = server.lua_client;
-    sds reply;
-
-    /* Cached across calls. */
-    static robj **argv = NULL;
-    static int argv_size = 0;
-    static robj *cached_objects[LUA_CMD_OBJCACHE_SIZE];
-    static size_t cached_objects_len[LUA_CMD_OBJCACHE_SIZE];
-    static int inuse = 0;   /* Recursive calls detection. */
-
-    /* By using Lua debug hooks it is possible to trigger a recursive call
-     * to luaRedisGenericCommand(), which normally should never happen.
-     * To make this function reentrant is futile and makes it slower, but
-     * we should at least detect such a misuse, and abort. */
-    if (inuse) {
-        char *recursion_warning =
-            "luaRedisGenericCommand() recursive call detected. "
-            "Are you doing funny stuff with Lua debug hooks?";
-        serverLog(LL_WARNING,"%s",recursion_warning);
-        luaPushError(lua,recursion_warning);
-        return 1;
-    }
-    inuse++;
-
-    /* Require at least one argument */
-    if (argc == 0) {
-        luaPushError(lua,
-            "Please specify at least one argument for redis.call()");
-        inuse--;
-        return raise_error ? luaRaiseError(lua) : 1;
-    }
-
-    /* Build the arguments vector */
-    if (argv_size < argc) {
-        argv = zrealloc(argv,sizeof(robj*)*argc);
-        argv_size = argc;
-    }
-
-    for (j = 0; j < argc; j++) {
-        char *obj_s;
-        size_t obj_len;
-        char dbuf[64];
-
-        if (lua_type(lua,j+1) == LUA_TNUMBER) {
-            /* We can't use lua_tolstring() for number -> string conversion
-             * since Lua uses a format specifier that loses precision. */
-            lua_Number num = lua_tonumber(lua,j+1);
-
-            obj_len = snprintf(dbuf,sizeof(dbuf),"%.17g",(double)num);
-            obj_s = dbuf;
-        } else {
-            obj_s = (char*)lua_tolstring(lua,j+1,&obj_len);
-            if (obj_s == NULL) break; /* Not a string. */
-        }
-
-        /* Try to use a cached object. */
-        if (j < LUA_CMD_OBJCACHE_SIZE && cached_objects[j] &&
-            cached_objects_len[j] >= obj_len)
-        {
-            sds s = cached_objects[j]->ptr;
-            argv[j] = cached_objects[j];
-            cached_objects[j] = NULL;
-            memcpy(s,obj_s,obj_len+1);
-            sdssetlen(s, obj_len);
-        } else {
-            argv[j] = createStringObject(obj_s, obj_len);
-        }
-    }
-
-    /* Check if one of the arguments passed by the Lua script
-     * is not a string or an integer (lua_isstring() return true for
-     * integers as well). */
-    if (j != argc) {
-        j--;
-        while (j >= 0) {
-            decrRefCount(argv[j]);
-            j--;
-        }
-        luaPushError(lua,
-            "Lua redis() command arguments must be strings or integers");
-        inuse--;
-        return raise_error ? luaRaiseError(lua) : 1;
-    }
-
-    /* Pop all arguments from the stack, we do not need them anymore
-     * and this way we guaranty we will have room on the stack for the result. */
-    lua_pop(lua, argc);
-
-    /* Setup our fake client for command execution */
-    c->argv = argv;
-    c->argc = argc;
-    c->user = server.lua_caller->user;
-
-    /* Process module hooks */
-    moduleCallCommandFilters(c);
-    argv = c->argv;
-    argc = c->argc;
-
-    /* Log the command if debugging is active. */
-    if (ldb.active && ldb.step) {
-        sds cmdlog = sdsnew("<redis>");
-        for (j = 0; j < c->argc; j++) {
-            if (j == 10) {
-                cmdlog = sdscatprintf(cmdlog," ... (%d more)",
-                    c->argc-j-1);
-                break;
-            } else {
-                cmdlog = sdscatlen(cmdlog," ",1);
-                cmdlog = sdscatsds(cmdlog,c->argv[j]->ptr);
-            }
-        }
-        ldbLog(cmdlog);
-    }
-
-    /* Command lookup */
-    cmd = lookupCommand(argv,argc);
-    if (!cmd || ((cmd->arity > 0 && cmd->arity != argc) ||
-                   (argc < -cmd->arity)))
-    {
-        if (cmd)
-            luaPushError(lua,
-                "Wrong number of args calling Redis command From Lua script");
-        else
-            luaPushError(lua,"Unknown Redis command called from Lua script");
-        goto cleanup;
-    }
-    c->cmd = c->lastcmd = cmd;
-
-    /* There are commands that are not allowed inside scripts. */
-    if (!server.lua_disable_deny_script && (cmd->flags & CMD_NOSCRIPT)) {
-        luaPushError(lua, "This Redis command is not allowed from scripts");
-        goto cleanup;
-    }
-
-    /* This check is for EVAL_RO, EVALSHA_RO. We want to allow only read only commands */
-    if ((server.lua_caller->cmd->proc == evalRoCommand ||
-         server.lua_caller->cmd->proc == evalShaRoCommand) &&
-         (cmd->flags & CMD_WRITE))
-    {
-        luaPushError(lua, "Write commands are not allowed from read-only scripts");
-        goto cleanup;
-    }
-
-    /* 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:
-            luaPushError(lua, "The user executing the script can't run this "
-                              "command or subcommand");
-            break;
-        case ACL_DENIED_KEY:
-            luaPushError(lua, "The user executing the script can't access "
-                              "at least one of the keys mentioned in the "
-                              "command arguments");
-            break;
-        case ACL_DENIED_CHANNEL:
-            luaPushError(lua, "The user executing the script can't publish "
-                              "to the channel mentioned in the command");
-            break;
-        default:
-            luaPushError(lua, "The user executing the script is lacking the "
-                              "permissions for the command");
-            break;
-        }
-        goto cleanup;
-    }
-
-    /* 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. */
-    if (cmd->flags & CMD_WRITE) {
-        int deny_write_type = writeCommandsDeniedByDiskError();
-        if (server.lua_random_dirty && !server.lua_replicate_commands) {
-            luaPushError(lua,
-                "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.");
-            goto cleanup;
-        } else if (server.masterhost && server.repl_slave_ro &&
-                   server.lua_caller->id != CLIENT_ID_AOF &&
-                   !(server.lua_caller->flags & CLIENT_MASTER))
-        {
-            luaPushError(lua, shared.roslaveerr->ptr);
-            goto cleanup;
-        } else if (deny_write_type != DISK_ERROR_TYPE_NONE) {
-            if (deny_write_type == DISK_ERROR_TYPE_RDB) {
-                luaPushError(lua, shared.bgsaveerr->ptr);
-            } else {
-                sds aof_write_err = sdscatfmt(sdsempty(),
-                    "-MISCONF Errors writing to the AOF file: %s\r\n",
-                    strerror(server.aof_last_write_errno));
-                luaPushError(lua, aof_write_err);
-                sdsfree(aof_write_err);
-            }
-            goto cleanup;
-        }
-    }
-
-    /* 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. */
-        server.lua_caller->id != CLIENT_ID_AOF &&  /* Don't care about mem if loading from AOF. */
-        !server.masterhost &&                      /* Slave must execute the script. */
-        server.lua_write_dirty == 0 &&             /* Script had no side effects so far. */
-        server.lua_oom &&                          /* Detected OOM when script start. */
-        (cmd->flags & CMD_DENYOOM))
-    {
-        luaPushError(lua, shared.oomerr->ptr);
-        goto cleanup;
-    }
-
-    if (cmd->flags & CMD_RANDOM) server.lua_random_dirty = 1;
-    if (cmd->flags & CMD_WRITE) server.lua_write_dirty = 1;
-
-    /* If this is a Redis Cluster node, we need to make sure Lua 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. */
-    if (server.cluster_enabled && server.lua_caller->id != CLIENT_ID_AOF &&
-        !(server.lua_caller->flags & CLIENT_MASTER))
-    {
-        int error_code;
-        /* Duplicate relevant flags in the lua client. */
-        c->flags &= ~(CLIENT_READONLY|CLIENT_ASKING);
-        c->flags |= server.lua_caller->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) {
-                luaPushError(lua,
-                    "Lua script attempted to execute a write command while the "
-                    "cluster is down and readonly");
-            } else if (error_code == CLUSTER_REDIR_DOWN_STATE) {
-                luaPushError(lua,
-                    "Lua script attempted to execute a command while the "
-                    "cluster is down");
-            } else {
-                luaPushError(lua,
-                    "Lua script attempted to access a non local key in a "
-                    "cluster node");
-            }
-
-            goto cleanup;
-        }
-    }
-
-    /* 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. */
-    if (server.lua_replicate_commands &&
-        !server.lua_multi_emitted &&
-        !(server.lua_caller->flags & CLIENT_MULTI) &&
-        server.lua_write_dirty &&
-        server.lua_repl != PROPAGATE_NONE)
-    {
-        execCommandPropagateMulti(server.lua_caller->db->id);
-        server.lua_multi_emitted = 1;
-        /* Now we are in the MULTI context, the lua_client should be
-         * flag as CLIENT_MULTI. */
-        c->flags |= CLIENT_MULTI;
-    }
-
-    /* Run the command */
-    int call_flags = CMD_CALL_SLOWLOG | CMD_CALL_STATS;
-    if (server.lua_replicate_commands) {
-        /* Set flags according to redis.set_repl() settings. */
-        if (server.lua_repl & PROPAGATE_AOF)
-            call_flags |= CMD_CALL_PROPAGATE_AOF;
-        if (server.lua_repl & PROPAGATE_REPL)
-            call_flags |= CMD_CALL_PROPAGATE_REPL;
-    }
-    call(c,call_flags);
-    serverAssert((c->flags & CLIENT_BLOCKED) == 0);
-
-    /* Convert the result of the Redis command into a suitable Lua type.
-     * The first thing we need is to create a single string from the client
-     * output buffers. */
-    if (listLength(c->reply) == 0 && (size_t)c->bufpos < c->buf_usable_size) {
-        /* This is a fast path for the common case of a reply inside the
-         * client static buffer. Don't create an SDS string but just use
-         * the client buffer directly. */
-        c->buf[c->bufpos] = '\0';
-        reply = c->buf;
-        c->bufpos = 0;
-    } else {
-        reply = sdsnewlen(c->buf,c->bufpos);
-        c->bufpos = 0;
-        while(listLength(c->reply)) {
-            clientReplyBlock *o = listNodeValue(listFirst(c->reply));
-
-            reply = sdscatlen(reply,o->buf,o->used);
-            listDelNode(c->reply,listFirst(c->reply));
-        }
-    }
-    if (raise_error && reply[0] != '-') raise_error = 0;
-    redisProtocolToLuaType(lua,reply);
-
-    /* If the debugger is active, log the reply from Redis. */
-    if (ldb.active && ldb.step)
-        ldbLogRedisReply(reply);
-
-    /* Sort the output array if needed, assuming it is a non-null multi bulk
-     * reply as expected. */
-    if ((cmd->flags & CMD_SORT_FOR_SCRIPT) &&
-        (server.lua_replicate_commands == 0) &&
-        (reply[0] == '*' && reply[1] != '-')) {
-            luaSortArray(lua);
-    }
-    if (reply != c->buf) sdsfree(reply);
-    c->reply_bytes = 0;
-
-cleanup:
-    /* Clean up. Command code may have changed argv/argc so we use the
-     * argv/argc of the client instead of the local variables. */
-    for (j = 0; j < c->argc; j++) {
-        robj *o = c->argv[j];
-
-        /* Try to cache the object in the cached_objects array.
-         * The object must be small, SDS-encoded, and with refcount = 1
-         * (we must be the only owner) for us to cache it. */
-        if (j < LUA_CMD_OBJCACHE_SIZE &&
-            o->refcount == 1 &&
-            (o->encoding == OBJ_ENCODING_RAW ||
-             o->encoding == OBJ_ENCODING_EMBSTR) &&
-            sdslen(o->ptr) <= LUA_CMD_OBJCACHE_MAX_LEN)
-        {
-            sds s = o->ptr;
-            if (cached_objects[j]) decrRefCount(cached_objects[j]);
-            cached_objects[j] = o;
-            cached_objects_len[j] = sdsalloc(s);
-        } else {
-            decrRefCount(o);
-        }
-    }
-
-    if (c->argv != argv) {
-        zfree(c->argv);
-        argv = NULL;
-        argv_size = 0;
-    }
-
-    c->user = NULL;
-
-    if (raise_error) {
-        /* If we are here we should have an error in the stack, in the
-         * form of a table with an "err" field. Extract the string to
-         * return the plain error. */
-        inuse--;
-        return luaRaiseError(lua);
-    }
-    inuse--;
-    return 1;
-}
-
-/* redis.call() */
-int luaRedisCallCommand(lua_State *lua) {
-    return luaRedisGenericCommand(lua,1);
-}
-
-/* redis.pcall() */
-int luaRedisPCallCommand(lua_State *lua) {
-    return luaRedisGenericCommand(lua,0);
-}
-
-/* This adds redis.sha1hex(string) to Lua scripts using the same hashing
- * function used for sha1ing lua scripts. */
-int luaRedisSha1hexCommand(lua_State *lua) {
-    int argc = lua_gettop(lua);
-    char digest[41];
-    size_t len;
-    char *s;
-
-    if (argc != 1) {
-        lua_pushstring(lua, "wrong number of arguments");
-        return lua_error(lua);
-    }
-
-    s = (char*)lua_tolstring(lua,1,&len);
-    sha1hex(digest,s,len);
-    lua_pushstring(lua,digest);
-    return 1;
-}
-
-/* Returns a table with a single field 'field' set to the string value
- * passed as argument. This helper function is handy when returning
- * a Redis Protocol error or status reply from Lua:
- *
- * return redis.error_reply("ERR Some Error")
- * return redis.status_reply("ERR Some Error")
- */
-int luaRedisReturnSingleFieldTable(lua_State *lua, char *field) {
-    if (lua_gettop(lua) != 1 || lua_type(lua,-1) != LUA_TSTRING) {
-        luaPushError(lua, "wrong number or type of arguments");
-        return 1;
-    }
-
-    lua_newtable(lua);
-    lua_pushstring(lua, field);
-    lua_pushvalue(lua, -3);
-    lua_settable(lua, -3);
-    return 1;
-}
-
-/* redis.error_reply() */
-int luaRedisErrorReplyCommand(lua_State *lua) {
-    return luaRedisReturnSingleFieldTable(lua,"err");
-}
-
-/* redis.status_reply() */
-int luaRedisStatusReplyCommand(lua_State *lua) {
-    return luaRedisReturnSingleFieldTable(lua,"ok");
-}
-
-/* redis.replicate_commands()
- *
- * 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 luaRedisReplicateCommandsCommand(lua_State *lua) {
-    if (server.lua_write_dirty) {
-        lua_pushboolean(lua,0);
-    } else {
-        server.lua_replicate_commands = 1;
-        /* When we switch to single commands replication, we can provide
-         * different math.random() sequences at every call, which is what
-         * the user normally expects. */
-        redisSrand48(rand());
-        lua_pushboolean(lua,1);
-    }
-    return 1;
-}
-
 /* redis.breakpoint()
  *
  * Allows to stop execution during a debugging session from within
@@ -1176,165 +131,24 @@ int luaRedisDebugCommand(lua_State *lua) {
     return 0;
 }
 
-/* redis.set_repl()
+/* redis.replicate_commands()
  *
- * Set the propagation of write commands executed in the context of the
- * script to on/off for AOF and slaves. */
-int luaRedisSetReplCommand(lua_State *lua) {
-    int argc = lua_gettop(lua);
-    int flags;
-
-    if (server.lua_replicate_commands == 0) {
-        lua_pushstring(lua, "You can set the replication behavior only after turning on single commands replication with redis.replicate_commands().");
-        return lua_error(lua);
-    } else if (argc != 1) {
-        lua_pushstring(lua, "redis.set_repl() requires two arguments.");
-        return lua_error(lua);
+ * 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 luaRedisReplicateCommandsCommand(lua_State *lua) {
+    if (server.lua_write_dirty) {
+        lua_pushboolean(lua,0);
+    } else {
+        server.lua_replicate_commands = 1;
+        /* When we switch to single commands replication, we can provide
+         * different math.random() sequences at every call, which is what
+         * the user normally expects. */
+        redisSrand48(rand());
+        lua_pushboolean(lua,1);
     }
-
-    flags = lua_tonumber(lua,-1);
-    if ((flags & ~(PROPAGATE_AOF|PROPAGATE_REPL)) != 0) {
-        lua_pushstring(lua, "Invalid replication flags. Use REPL_AOF, REPL_REPLICA, REPL_ALL or REPL_NONE.");
-        return lua_error(lua);
-    }
-    server.lua_repl = flags;
-    return 0;
-}
-
-/* redis.log() */
-int luaLogCommand(lua_State *lua) {
-    int j, argc = lua_gettop(lua);
-    int level;
-    sds log;
-
-    if (argc < 2) {
-        lua_pushstring(lua, "redis.log() requires two arguments or more.");
-        return lua_error(lua);
-    } else if (!lua_isnumber(lua,-argc)) {
-        lua_pushstring(lua, "First argument must be a number (log level).");
-        return lua_error(lua);
-    }
-    level = lua_tonumber(lua,-argc);
-    if (level < LL_DEBUG || level > LL_WARNING) {
-        lua_pushstring(lua, "Invalid debug level.");
-        return lua_error(lua);
-    }
-    if (level < server.verbosity) return 0;
-
-    /* Glue together all the arguments */
-    log = sdsempty();
-    for (j = 1; j < argc; j++) {
-        size_t len;
-        char *s;
-
-        s = (char*)lua_tolstring(lua,(-argc)+j,&len);
-        if (s) {
-            if (j != 1) log = sdscatlen(log," ",1);
-            log = sdscatlen(log,s,len);
-        }
-    }
-    serverLogRaw(level,log);
-    sdsfree(log);
-    return 0;
-}
-
-/* redis.setresp() */
-int luaSetResp(lua_State *lua) {
-    int argc = lua_gettop(lua);
-
-    if (argc != 1) {
-        lua_pushstring(lua, "redis.setresp() requires one argument.");
-        return lua_error(lua);
-    }
-
-    int resp = lua_tonumber(lua,-argc);
-    if (resp != 2 && resp != 3) {
-        lua_pushstring(lua, "RESP version must be 2 or 3.");
-        return lua_error(lua);
-    }
-
-    server.lua_client->resp = resp;
-    return 0;
-}
-
-/* ---------------------------------------------------------------------------
- * Lua engine initialization and reset.
- * ------------------------------------------------------------------------- */
-
-void luaLoadLib(lua_State *lua, const char *libname, lua_CFunction luafunc) {
-  lua_pushcfunction(lua, luafunc);
-  lua_pushstring(lua, libname);
-  lua_call(lua, 1, 0);
-}
-
-LUALIB_API int (luaopen_cjson) (lua_State *L);
-LUALIB_API int (luaopen_struct) (lua_State *L);
-LUALIB_API int (luaopen_cmsgpack) (lua_State *L);
-LUALIB_API int (luaopen_bit) (lua_State *L);
-
-void luaLoadLibraries(lua_State *lua) {
-    luaLoadLib(lua, "", luaopen_base);
-    luaLoadLib(lua, LUA_TABLIBNAME, luaopen_table);
-    luaLoadLib(lua, LUA_STRLIBNAME, luaopen_string);
-    luaLoadLib(lua, LUA_MATHLIBNAME, luaopen_math);
-    luaLoadLib(lua, LUA_DBLIBNAME, luaopen_debug);
-    luaLoadLib(lua, "cjson", luaopen_cjson);
-    luaLoadLib(lua, "struct", luaopen_struct);
-    luaLoadLib(lua, "cmsgpack", luaopen_cmsgpack);
-    luaLoadLib(lua, "bit", luaopen_bit);
-
-#if 0 /* Stuff that we don't load currently, for sandboxing concerns. */
-    luaLoadLib(lua, LUA_LOADLIBNAME, luaopen_package);
-    luaLoadLib(lua, LUA_OSLIBNAME, luaopen_os);
-#endif
-}
-
-/* Remove a functions that we don't want to expose to the Redis scripting
- * environment. */
-void luaRemoveUnsupportedFunctions(lua_State *lua) {
-    lua_pushnil(lua);
-    lua_setglobal(lua,"loadfile");
-    lua_pushnil(lua);
-    lua_setglobal(lua,"dofile");
-}
-
-/* This function installs metamethods in the global table _G that prevent
- * the creation of globals accidentally.
- *
- * It should be the last to be called in the scripting engine initialization
- * sequence, because it may interact with creation of globals. */
-void scriptingEnableGlobalsProtection(lua_State *lua) {
-    char *s[32];
-    sds code = sdsempty();
-    int j = 0;
-
-    /* strict.lua from: http://metalua.luaforge.net/src/lib/strict.lua.html.
-     * Modified to be adapted to Redis. */
-    s[j++]="local dbg=debug\n";
-    s[j++]="local mt = {}\n";
-    s[j++]="setmetatable(_G, mt)\n";
-    s[j++]="mt.__newindex = function (t, n, v)\n";
-    s[j++]="  if dbg.getinfo(2) then\n";
-    s[j++]="    local w = dbg.getinfo(2, \"S\").what\n";
-    s[j++]="    if w ~= \"main\" and w ~= \"C\" then\n";
-    s[j++]="      error(\"Script attempted to create global variable '\"..tostring(n)..\"'\", 2)\n";
-    s[j++]="    end\n";
-    s[j++]="  end\n";
-    s[j++]="  rawset(t, n, v)\n";
-    s[j++]="end\n";
-    s[j++]="mt.__index = function (t, n)\n";
-    s[j++]="  if dbg.getinfo(2) and dbg.getinfo(2, \"S\").what ~= \"C\" then\n";
-    s[j++]="    error(\"Script attempted to access nonexistent global variable '\"..tostring(n)..\"'\", 2)\n";
-    s[j++]="  end\n";
-    s[j++]="  return rawget(t, n)\n";
-    s[j++]="end\n";
-    s[j++]="debug = nil\n";
-    s[j++]=NULL;
-
-    for (j = 0; s[j] != NULL; j++) code = sdscatlen(code,s[j],strlen(s[j]));
-    luaL_loadbuffer(lua,code,sdslen(code),"@enable_strict_lua");
-    lua_pcall(lua,0,0,0);
-    sdsfree(code);
+    return 1;
 }
 
 /* Initialize the scripting environment.
@@ -1359,96 +173,16 @@ void scriptingInit(int setup) {
         ldbInit();
     }
 
-    luaLoadLibraries(lua);
-    luaRemoveUnsupportedFunctions(lua);
-
     /* Initialize a dictionary we use to map SHAs to scripts.
      * This is useful for replication, as we need to replicate EVALSHA
      * as EVAL, so we need to remember the associated script. */
     server.lua_scripts = dictCreate(&shaScriptObjectDictType);
     server.lua_scripts_mem = 0;
 
-    /* Register the redis commands table and fields */
-    lua_newtable(lua);
+    luaEngineRegisterRedisAPI(lua);
 
-    /* redis.call */
-    lua_pushstring(lua,"call");
-    lua_pushcfunction(lua,luaRedisCallCommand);
-    lua_settable(lua,-3);
-
-    /* redis.pcall */
-    lua_pushstring(lua,"pcall");
-    lua_pushcfunction(lua,luaRedisPCallCommand);
-    lua_settable(lua,-3);
-
-    /* redis.log and log levels. */
-    lua_pushstring(lua,"log");
-    lua_pushcfunction(lua,luaLogCommand);
-    lua_settable(lua,-3);
-
-    /* redis.setresp */
-    lua_pushstring(lua,"setresp");
-    lua_pushcfunction(lua,luaSetResp);
-    lua_settable(lua,-3);
-
-    lua_pushstring(lua,"LOG_DEBUG");
-    lua_pushnumber(lua,LL_DEBUG);
-    lua_settable(lua,-3);
-
-    lua_pushstring(lua,"LOG_VERBOSE");
-    lua_pushnumber(lua,LL_VERBOSE);
-    lua_settable(lua,-3);
-
-    lua_pushstring(lua,"LOG_NOTICE");
-    lua_pushnumber(lua,LL_NOTICE);
-    lua_settable(lua,-3);
-
-    lua_pushstring(lua,"LOG_WARNING");
-    lua_pushnumber(lua,LL_WARNING);
-    lua_settable(lua,-3);
-
-    /* redis.sha1hex */
-    lua_pushstring(lua, "sha1hex");
-    lua_pushcfunction(lua, luaRedisSha1hexCommand);
-    lua_settable(lua, -3);
-
-    /* redis.error_reply and redis.status_reply */
-    lua_pushstring(lua, "error_reply");
-    lua_pushcfunction(lua, luaRedisErrorReplyCommand);
-    lua_settable(lua, -3);
-    lua_pushstring(lua, "status_reply");
-    lua_pushcfunction(lua, luaRedisStatusReplyCommand);
-    lua_settable(lua, -3);
-
-    /* redis.replicate_commands */
-    lua_pushstring(lua, "replicate_commands");
-    lua_pushcfunction(lua, luaRedisReplicateCommandsCommand);
-    lua_settable(lua, -3);
-
-    /* redis.set_repl and associated flags. */
-    lua_pushstring(lua,"set_repl");
-    lua_pushcfunction(lua,luaRedisSetReplCommand);
-    lua_settable(lua,-3);
-
-    lua_pushstring(lua,"REPL_NONE");
-    lua_pushnumber(lua,PROPAGATE_NONE);
-    lua_settable(lua,-3);
-
-    lua_pushstring(lua,"REPL_AOF");
-    lua_pushnumber(lua,PROPAGATE_AOF);
-    lua_settable(lua,-3);
-
-    lua_pushstring(lua,"REPL_SLAVE");
-    lua_pushnumber(lua,PROPAGATE_REPL);
-    lua_settable(lua,-3);
-
-    lua_pushstring(lua,"REPL_REPLICA");
-    lua_pushnumber(lua,PROPAGATE_REPL);
-    lua_settable(lua,-3);
-
-    lua_pushstring(lua,"REPL_ALL");
-    lua_pushnumber(lua,PROPAGATE_AOF|PROPAGATE_REPL);
-    lua_settable(lua,-3);
+    /* register debug commands */
+    lua_getglobal(lua,"redis");
 
     /* redis.breakpoint */
     lua_pushstring(lua,"breakpoint");
@@ -1460,22 +194,13 @@ void scriptingInit(int setup) {
     lua_pushcfunction(lua,luaRedisDebugCommand);
     lua_settable(lua,-3);
 
-    /* Finally set the table as 'redis' global var. */
+    /* redis.replicate_commands */
+    lua_pushstring(lua, "replicate_commands");
+    lua_pushcfunction(lua, luaRedisReplicateCommandsCommand);
+    lua_settable(lua, -3);
+
     lua_setglobal(lua,"redis");
 
-    /* Replace math.random and math.randomseed with our implementations. */
-    lua_getglobal(lua,"math");
-
-    lua_pushstring(lua,"random");
-    lua_pushcfunction(lua,redis_math_random);
-    lua_settable(lua,-3);
-
-    lua_pushstring(lua,"randomseed");
-    lua_pushcfunction(lua,redis_math_randomseed);
-    lua_settable(lua,-3);
-
-    lua_setglobal(lua,"math");
-
     /* Add a helper function that we use to sort the multi bulk output of non
      * deterministic commands, when containing 'false' elements. */
     {
@@ -1545,62 +270,6 @@ void scriptingReset(int async) {
     scriptingInit(0);
 }
 
-/* Set an array of Redis String Objects as a Lua array (table) stored into a
- * global variable. */
-void luaSetGlobalArray(lua_State *lua, char *var, robj **elev, int elec) {
-    int j;
-
-    lua_newtable(lua);
-    for (j = 0; j < elec; j++) {
-        lua_pushlstring(lua,(char*)elev[j]->ptr,sdslen(elev[j]->ptr));
-        lua_rawseti(lua,-2,j+1);
-    }
-    lua_setglobal(lua,var);
-}
-
-/* ---------------------------------------------------------------------------
- * Redis provided math.random
- * ------------------------------------------------------------------------- */
-
-/* We replace math.random() with our implementation that is not affected
- * by specific libc random() implementations and will output the same sequence
- * (for the same seed) in every arch. */
-
-/* The following implementation is the one shipped with Lua itself but with
- * rand() replaced by redisLrand48(). */
-int redis_math_random (lua_State *L) {
-  /* the `%' avoids the (rare) case of r==1, and is needed also because on
-     some systems (SunOS!) `rand()' may return a value larger than RAND_MAX */
-  lua_Number r = (lua_Number)(redisLrand48()%REDIS_LRAND48_MAX) /
-                                (lua_Number)REDIS_LRAND48_MAX;
-  switch (lua_gettop(L)) {  /* check number of arguments */
-    case 0: {  /* no arguments */
-      lua_pushnumber(L, r);  /* Number between 0 and 1 */
-      break;
-    }
-    case 1: {  /* only upper limit */
-      int u = luaL_checkint(L, 1);
-      luaL_argcheck(L, 1<=u, 1, "interval is empty");
-      lua_pushnumber(L, floor(r*u)+1);  /* int between 1 and `u' */
-      break;
-    }
-    case 2: {  /* lower and upper limits */
-      int l = luaL_checkint(L, 1);
-      int u = luaL_checkint(L, 2);
-      luaL_argcheck(L, l<=u, 2, "interval is empty");
-      lua_pushnumber(L, floor(r*(u-l+1))+l);  /* int between `l' and `u' */
-      break;
-    }
-    default: return luaL_error(L, "wrong number of arguments");
-  }
-  return 1;
-}
-
-int redis_math_randomseed (lua_State *L) {
-  redisSrand48(luaL_checkint(L, 1));
-  return 0;
-}
-
 /* ---------------------------------------------------------------------------
  * EVAL and SCRIPT commands implementation
  * ------------------------------------------------------------------------- */
@@ -1676,45 +345,6 @@ sds luaCreateFunction(client *c, lua_State *lua, robj *body) {
     return sha;
 }
 
-/* This is the Lua script "count" hook that we use to detect scripts timeout. */
-void luaMaskCountHook(lua_State *lua, lua_Debug *ar) {
-    long long elapsed = elapsedMs(server.lua_time_start);
-    UNUSED(ar);
-    UNUSED(lua);
-
-    /* Set the timeout condition if not already set and the maximum
-     * execution time was reached. */
-    if (elapsed >= server.lua_time_limit && server.lua_timedout == 0) {
-        serverLog(LL_WARNING,
-            "Lua slow script detected: still in execution after %lld milliseconds. "
-            "You can try killing the script using the SCRIPT KILL command. "
-            "Script SHA1 is: %s",
-            elapsed, server.lua_cur_script);
-        server.lua_timedout = 1;
-        blockingOperationStarts();
-        /* Once the script timeouts we reenter the event loop to permit others
-         * to call SCRIPT KILL or SHUTDOWN NOSAVE if needed. 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(server.lua_caller);
-    }
-    if (server.lua_timedout) processEventsWhileBlocked();
-    if (server.lua_kill) {
-        serverLog(LL_WARNING,"Lua script killed by user with SCRIPT KILL.");
-
-        /*
-         * Set the hook to invoke all the time so the user
-         * will not be able to catch the error with pcall and invoke
-         * pcall again which will prevent the script from ever been killed
-         */
-        lua_sethook(lua, luaMaskCountHook, LUA_MASKLINE, 0);
-
-        lua_pushstring(lua,"Script killed by user with SCRIPT KILL...");
-        lua_error(lua);
-    }
-}
-
 void prepareLuaClient(void) {
     /* Select the right DB in the context of the Lua client */
     selectDb(server.lua_client,server.lua_caller->db->id);
diff --git a/src/script_lua.c b/src/script_lua.c
new file mode 100644
index 000000000..0685a4bb3
--- /dev/null
+++ b/src/script_lua.c
@@ -0,0 +1,1423 @@
+/*
+ * Copyright (c) 2009-2021, Redis Labs 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 "script_lua.h"
+
+#include "server.h"
+#include "sha1.h"
+#include "rand.h"
+#include "cluster.h"
+#include "monotonic.h"
+#include "resp_parser.h"
+#include <lauxlib.h>
+#include <lualib.h>
+#include <ctype.h>
+#include <math.h>
+#include "functions.h"
+
+int redis_math_random (lua_State *L);
+int redis_math_randomseed (lua_State *L);
+static void redisProtocolToLuaType_Int(void *ctx, long long val, const char *proto, size_t proto_len);
+static void redisProtocolToLuaType_BulkString(void *ctx, const char *str, size_t len, const char *proto, size_t proto_len);
+static void redisProtocolToLuaType_NullBulkString(void *ctx, const char *proto, size_t proto_len);
+static void redisProtocolToLuaType_NullArray(void *ctx, const char *proto, size_t proto_len);
+static void redisProtocolToLuaType_Status(void *ctx, const char *str, size_t len, const char *proto, size_t proto_len);
+static void redisProtocolToLuaType_Error(void *ctx, const char *str, size_t len, const char *proto, size_t proto_len);
+static void redisProtocolToLuaType_Array(struct ReplyParser *parser, void *ctx, size_t len, const char *proto);
+static void redisProtocolToLuaType_Map(struct ReplyParser *parser, void *ctx, size_t len, const char *proto);
+static void redisProtocolToLuaType_Set(struct ReplyParser *parser, void *ctx, size_t len, const char *proto);
+static void redisProtocolToLuaType_Null(void *ctx, const char *proto, size_t proto_len);
+static void redisProtocolToLuaType_Bool(void *ctx, int val, const char *proto, size_t proto_len);
+static void redisProtocolToLuaType_Double(void *ctx, double d, const char *proto, size_t proto_len);
+static void redisProtocolToLuaType_BigNumber(void *ctx, const char *str, size_t len, const char *proto, size_t proto_len);
+static void redisProtocolToLuaType_VerbatimString(void *ctx, const char *format, const char *str, size_t len, const char *proto, size_t proto_len);
+static void redisProtocolToLuaType_Attribute(struct ReplyParser *parser, void *ctx, size_t len, const char *proto);
+
+/* ---------------------------------------------------------------------------
+ * Redis reply to Lua type conversion functions.
+ * ------------------------------------------------------------------------- */
+
+/* Take a Redis reply in the Redis protocol format and convert it into a
+ * Lua type. Thanks to this function, and the introduction of not connected
+ * clients, it is trivial to implement the redis() lua function.
+ *
+ * Basically we take the arguments, execute the Redis command in the context
+ * of a non connected client, then take the generated reply and convert it
+ * into a suitable Lua type. With this trick the scripting feature does not
+ * need the introduction of a full Redis internals API. The script
+ * is like a normal client that bypasses all the slow I/O paths.
+ *
+ * Note: in this function we do not do any sanity check as the reply is
+ * generated by Redis directly. This allows us to go faster.
+ *
+ * Errors are returned as a table with a single 'err' field set to the
+ * error string.
+ */
+
+static const ReplyParserCallbacks DefaultLuaTypeParserCallbacks = {
+    .null_array_callback = redisProtocolToLuaType_NullArray,
+    .bulk_string_callback = redisProtocolToLuaType_BulkString,
+    .null_bulk_string_callback = redisProtocolToLuaType_NullBulkString,
+    .error_callback = redisProtocolToLuaType_Error,
+    .simple_str_callback = redisProtocolToLuaType_Status,
+    .long_callback = redisProtocolToLuaType_Int,
+    .array_callback = redisProtocolToLuaType_Array,
+    .set_callback = redisProtocolToLuaType_Set,
+    .map_callback = redisProtocolToLuaType_Map,
+    .bool_callback = redisProtocolToLuaType_Bool,
+    .double_callback = redisProtocolToLuaType_Double,
+    .null_callback = redisProtocolToLuaType_Null,
+    .big_number_callback = redisProtocolToLuaType_BigNumber,
+    .verbatim_string_callback = redisProtocolToLuaType_VerbatimString,
+    .attribute_callback = redisProtocolToLuaType_Attribute,
+    .error = NULL,
+};
+
+void redisProtocolToLuaType(lua_State *lua, char* reply) {
+    ReplyParser parser = {.curr_location = reply, .callbacks = DefaultLuaTypeParserCallbacks};
+
+    parseReply(&parser, lua);
+}
+
+static void redisProtocolToLuaType_Int(void *ctx, long long val, const char *proto, size_t proto_len) {
+    UNUSED(proto);
+    UNUSED(proto_len);
+    if (!ctx) {
+        return;
+    }
+
+    lua_State *lua = ctx;
+    if (!lua_checkstack(lua, 1)) {
+        /* Increase the Lua stack if needed, to make sure there is enough room
+         * to push elements to the stack. On failure, exit with panic. */
+        serverPanic("lua stack limit reach when parsing redis.call reply");
+    }
+    lua_pushnumber(lua,(lua_Number)val);
+}
+
+static void redisProtocolToLuaType_NullBulkString(void *ctx, const char *proto, size_t proto_len) {
+    UNUSED(proto);
+    UNUSED(proto_len);
+    if (!ctx) {
+        return;
+    }
+
+    lua_State *lua = ctx;
+    if (!lua_checkstack(lua, 1)) {
+        /* Increase the Lua stack if needed, to make sure there is enough room
+         * to push elements to the stack. On failure, exit with panic. */
+        serverPanic("lua stack limit reach when parsing redis.call reply");
+    }
+    lua_pushboolean(lua,0);
+}
+
+static void redisProtocolToLuaType_NullArray(void *ctx, const char *proto, size_t proto_len) {
+    UNUSED(proto);
+    UNUSED(proto_len);
+    if (!ctx) {
+        return;
+    }
+    lua_State *lua = ctx;
+    if (!lua_checkstack(lua, 1)) {
+        /* Increase the Lua stack if needed, to make sure there is enough room
+         * to push elements to the stack. On failure, exit with panic. */
+        serverPanic("lua stack limit reach when parsing redis.call reply");
+    }
+    lua_pushboolean(lua,0);
+}
+
+
+static void redisProtocolToLuaType_BulkString(void *ctx, const char *str, size_t len, const char *proto, size_t proto_len) {
+    UNUSED(proto);
+    UNUSED(proto_len);
+    if (!ctx) {
+        return;
+    }
+
+    lua_State *lua = ctx;
+    if (!lua_checkstack(lua, 1)) {
+        /* Increase the Lua stack if needed, to make sure there is enough room
+         * to push elements to the stack. On failure, exit with panic. */
+        serverPanic("lua stack limit reach when parsing redis.call reply");
+    }
+    lua_pushlstring(lua,str,len);
+}
+
+static void redisProtocolToLuaType_Status(void *ctx, const char *str, size_t len, const char *proto, size_t proto_len) {
+    UNUSED(proto);
+    UNUSED(proto_len);
+    if (!ctx) {
+        return;
+    }
+
+    lua_State *lua = ctx;
+    if (!lua_checkstack(lua, 3)) {
+        /* Increase the Lua stack if needed, to make sure there is enough room
+         * to push elements to the stack. On failure, exit with panic. */
+        serverPanic("lua stack limit reach when parsing redis.call reply");
+    }
+    lua_newtable(lua);
+    lua_pushstring(lua,"ok");
+    lua_pushlstring(lua,str,len);
+    lua_settable(lua,-3);
+}
+
+static void redisProtocolToLuaType_Error(void *ctx, const char *str, size_t len, const char *proto, size_t proto_len) {
+    UNUSED(proto);
+    UNUSED(proto_len);
+    if (!ctx) {
+        return;
+    }
+
+    lua_State *lua = ctx;
+    if (!lua_checkstack(lua, 3)) {
+        /* Increase the Lua stack if needed, to make sure there is enough room
+         * to push elements to the stack. On failure, exit with panic. */
+        serverPanic("lua stack limit reach when parsing redis.call reply");
+    }
+    lua_newtable(lua);
+    lua_pushstring(lua,"err");
+    lua_pushlstring(lua,str,len);
+    lua_settable(lua,-3);
+}
+
+static void redisProtocolToLuaType_Map(struct ReplyParser *parser, void *ctx, size_t len, const char *proto) {
+    UNUSED(proto);
+    lua_State *lua = ctx;
+    if (lua) {
+        if (!lua_checkstack(lua, 3)) {
+            /* Increase the Lua stack if needed, to make sure there is enough room
+             * to push elements to the stack. On failure, exit with panic. */
+            serverPanic("lua stack limit reach when parsing redis.call reply");
+        }
+        lua_newtable(lua);
+        lua_pushstring(lua, "map");
+        lua_newtable(lua);
+    }
+    for (size_t j = 0; j < len; j++) {
+        parseReply(parser,lua);
+        parseReply(parser,lua);
+        if (lua) lua_settable(lua,-3);
+    }
+    if (lua) lua_settable(lua,-3);
+}
+
+static void redisProtocolToLuaType_Set(struct ReplyParser *parser, void *ctx, size_t len, const char *proto) {
+    UNUSED(proto);
+
+    lua_State *lua = ctx;
+    if (lua) {
+        if (!lua_checkstack(lua, 3)) {
+            /* Increase the Lua stack if needed, to make sure there is enough room
+             * to push elements to the stack. On failure, exit with panic. */
+            serverPanic("lua stack limit reach when parsing redis.call reply");
+        }
+        lua_newtable(lua);
+        lua_pushstring(lua, "set");
+        lua_newtable(lua);
+    }
+    for (size_t j = 0; j < len; j++) {
+        parseReply(parser,lua);
+        if (lua) {
+            if (!lua_checkstack(lua, 1)) {
+                /* Increase the Lua stack if needed, to make sure there is enough room
+                 * to push elements to the stack. On failure, exit with panic.
+                 * Notice that here we need to check the stack again because the recursive
+                 * call to redisProtocolToLuaType might have use the room allocated in the stack*/
+                serverPanic("lua stack limit reach when parsing redis.call reply");
+            }
+            lua_pushboolean(lua,1);
+            lua_settable(lua,-3);
+        }
+    }
+    if (lua) lua_settable(lua,-3);
+}
+
+static void redisProtocolToLuaType_Array(struct ReplyParser *parser, void *ctx, size_t len, const char *proto) {
+    UNUSED(proto);
+
+    lua_State *lua = ctx;
+    if (lua){
+        if (!lua_checkstack(lua, 2)) {
+            /* Increase the Lua stack if needed, to make sure there is enough room
+             * to push elements to the stack. On failure, exit with panic. */
+            serverPanic("lua stack limit reach when parsing redis.call reply");
+        }
+        lua_newtable(lua);
+    }
+    for (size_t j = 0; j < len; j++) {
+        if (lua) lua_pushnumber(lua,j+1);
+        parseReply(parser,lua);
+        if (lua) lua_settable(lua,-3);
+    }
+}
+
+static void redisProtocolToLuaType_Attribute(struct ReplyParser *parser, void *ctx, size_t len, const char *proto) {
+    UNUSED(proto);
+
+    /* Parse the attribute reply.
+     * Currently, we do not expose the attribute to the Lua script so
+     * we just need to continue parsing and ignore it (the NULL ensures that the
+     * reply will be ignored). */
+    for (size_t j = 0; j < len; j++) {
+        parseReply(parser,NULL);
+        parseReply(parser,NULL);
+    }
+
+    /* Parse the reply itself. */
+    parseReply(parser,ctx);
+}
+
+static void redisProtocolToLuaType_VerbatimString(void *ctx, const char *format, const char *str, size_t len, const char *proto, size_t proto_len) {
+    UNUSED(proto);
+    UNUSED(proto_len);
+    if (!ctx) {
+        return;
+    }
+
+    lua_State *lua = ctx;
+    if (!lua_checkstack(lua, 5)) {
+        /* Increase the Lua stack if needed, to make sure there is enough room
+         * to push elements to the stack. On failure, exit with panic. */
+        serverPanic("lua stack limit reach when parsing redis.call reply");
+    }
+    lua_newtable(lua);
+    lua_pushstring(lua,"verbatim_string");
+    lua_newtable(lua);
+    lua_pushstring(lua,"string");
+    lua_pushlstring(lua,str,len);
+    lua_settable(lua,-3);
+    lua_pushstring(lua,"format");
+    lua_pushlstring(lua,format,3);
+    lua_settable(lua,-3);
+    lua_settable(lua,-3);
+}
+
+static void redisProtocolToLuaType_BigNumber(void *ctx, const char *str, size_t len, const char *proto, size_t proto_len) {
+    UNUSED(proto);
+    UNUSED(proto_len);
+    if (!ctx) {
+        return;
+    }
+
+    lua_State *lua = ctx;
+    if (!lua_checkstack(lua, 3)) {
+        /* Increase the Lua stack if needed, to make sure there is enough room
+         * to push elements to the stack. On failure, exit with panic. */
+        serverPanic("lua stack limit reach when parsing redis.call reply");
+    }
+    lua_newtable(lua);
+    lua_pushstring(lua,"big_number");
+    lua_pushlstring(lua,str,len);
+    lua_settable(lua,-3);
+}
+
+static void redisProtocolToLuaType_Null(void *ctx, const char *proto, size_t proto_len) {
+    UNUSED(proto);
+    UNUSED(proto_len);
+    if (!ctx) {
+        return;
+    }
+
+    lua_State *lua = ctx;
+    if (!lua_checkstack(lua, 1)) {
+        /* Increase the Lua stack if needed, to make sure there is enough room
+         * to push elements to the stack. On failure, exit with panic. */
+        serverPanic("lua stack limit reach when parsing redis.call reply");
+    }
+    lua_pushnil(lua);
+}
+
+static void redisProtocolToLuaType_Bool(void *ctx, int val, const char *proto, size_t proto_len) {
+    UNUSED(proto);
+    UNUSED(proto_len);
+    if (!ctx) {
+        return;
+    }
+
+    lua_State *lua = ctx;
+    if (!lua_checkstack(lua, 1)) {
+        /* Increase the Lua stack if needed, to make sure there is enough room
+         * to push elements to the stack. On failure, exit with panic. */
+        serverPanic("lua stack limit reach when parsing redis.call reply");
+    }
+    lua_pushboolean(lua,val);
+}
+
+static void redisProtocolToLuaType_Double(void *ctx, double d, const char *proto, size_t proto_len) {
+    UNUSED(proto);
+    UNUSED(proto_len);
+    if (!ctx) {
+        return;
+    }
+
+    lua_State *lua = ctx;
+    if (!lua_checkstack(lua, 3)) {
+        /* Increase the Lua stack if needed, to make sure there is enough room
+         * to push elements to the stack. On failure, exit with panic. */
+        serverPanic("lua stack limit reach when parsing redis.call reply");
+    }
+    lua_newtable(lua);
+    lua_pushstring(lua,"double");
+    lua_pushnumber(lua,d);
+    lua_settable(lua,-3);
+}
+
+/* This function is used in order to push an error on the Lua stack in the
+ * format used by redis.pcall to return errors, which is a lua table
+ * with a single "err" field set to the error string. Note that this
+ * table is never a valid reply by proper commands, since the returned
+ * tables are otherwise always indexed by integers, never by strings. */
+void luaPushError(lua_State *lua, char *error) {
+    lua_Debug dbg;
+
+    /* If debugging is active and in step mode, log errors resulting from
+     * Redis commands. */
+    if (ldb.active && ldb.step) {
+        ldbLog(sdscatprintf(sdsempty(),"<error> %s",error));
+    }
+
+    lua_newtable(lua);
+    lua_pushstring(lua,"err");
+
+    /* Attempt to figure out where this function was called, if possible */
+    if(lua_getstack(lua, 1, &dbg) && lua_getinfo(lua, "nSl", &dbg)) {
+        sds msg = sdscatprintf(sdsempty(), "%s: %d: %s",
+            dbg.source, dbg.currentline, error);
+        lua_pushstring(lua, msg);
+        sdsfree(msg);
+    } else {
+        lua_pushstring(lua, error);
+    }
+    lua_settable(lua,-3);
+}
+
+/* In case the error set into the Lua stack by luaPushError() was generated
+ * by the non-error-trapping version of redis.pcall(), which is redis.call(),
+ * this function will raise the Lua error so that the execution of the
+ * script will be halted. */
+int luaRaiseError(lua_State *lua) {
+    lua_pushstring(lua,"err");
+    lua_gettable(lua,-2);
+    return lua_error(lua);
+}
+
+/* Sort the array currently in the stack. We do this to make the output
+ * of commands like KEYS or SMEMBERS something deterministic when called
+ * from Lua (to play well with AOf/replication).
+ *
+ * The array is sorted using table.sort itself, and assuming all the
+ * list elements are strings. */
+void luaSortArray(lua_State *lua) {
+    /* Initial Stack: array */
+    lua_getglobal(lua,"table");
+    lua_pushstring(lua,"sort");
+    lua_gettable(lua,-2);       /* Stack: array, table, table.sort */
+    lua_pushvalue(lua,-3);      /* Stack: array, table, table.sort, array */
+    if (lua_pcall(lua,1,0,0)) {
+        /* Stack: array, table, error */
+
+        /* We are not interested in the error, we assume that the problem is
+         * that there are 'false' elements inside the array, so we try
+         * again with a slower function but able to handle this case, that
+         * is: table.sort(table, __redis__compare_helper) */
+        lua_pop(lua,1);             /* Stack: array, table */
+        lua_pushstring(lua,"sort"); /* Stack: array, table, sort */
+        lua_gettable(lua,-2);       /* Stack: array, table, table.sort */
+        lua_pushvalue(lua,-3);      /* Stack: array, table, table.sort, array */
+        lua_getglobal(lua,"__redis__compare_helper");
+        /* Stack: array, table, table.sort, array, __redis__compare_helper */
+        lua_call(lua,2,0);
+    }
+    /* Stack: array (sorted), table */
+    lua_pop(lua,1);             /* Stack: array (sorted) */
+}
+
+/* ---------------------------------------------------------------------------
+ * Lua reply to Redis reply conversion functions.
+ * ------------------------------------------------------------------------- */
+
+/* Reply to client 'c' converting the top element in the Lua stack to a
+ * Redis reply. As a side effect the element is consumed from the stack.  */
+void luaReplyToRedisReply(client *c, lua_State *lua) {
+    int t = lua_type(lua,-1);
+
+    if (!lua_checkstack(lua, 4)) {
+        /* Increase the Lua stack if needed to make sure there is enough room
+         * to push 4 elements to the stack. On failure, return error.
+         * Notice that we need, in the worst case, 4 elements because returning a map might
+         * require push 4 elements to the Lua stack.*/
+        addReplyErrorFormat(c, "reached lua stack limit");
+        lua_pop(lua,1); /* pop the element from the stack */
+        return;
+    }
+
+    switch(t) {
+    case LUA_TSTRING:
+        addReplyBulkCBuffer(c,(char*)lua_tostring(lua,-1),lua_strlen(lua,-1));
+        break;
+    case LUA_TBOOLEAN:
+        if (server.lua_client->resp == 2)
+            addReply(c,lua_toboolean(lua,-1) ? shared.cone :
+                                               shared.null[c->resp]);
+        else
+            addReplyBool(c,lua_toboolean(lua,-1));
+        break;
+    case LUA_TNUMBER:
+        addReplyLongLong(c,(long long)lua_tonumber(lua,-1));
+        break;
+    case LUA_TTABLE:
+        /* We need to check if it is an array, an error, or a status reply.
+         * Error are returned as a single element table with 'err' field.
+         * Status replies are returned as single element table with 'ok'
+         * field. */
+
+        /* Handle error reply. */
+        /* we took care of the stack size on function start */
+        lua_pushstring(lua,"err");
+        lua_gettable(lua,-2);
+        t = lua_type(lua,-1);
+        if (t == LUA_TSTRING) {
+            addReplyErrorFormat(c,"-%s",lua_tostring(lua,-1));
+            lua_pop(lua,2);
+            return;
+        }
+        lua_pop(lua,1); /* Discard field name pushed before. */
+
+        /* Handle status reply. */
+        lua_pushstring(lua,"ok");
+        lua_gettable(lua,-2);
+        t = lua_type(lua,-1);
+        if (t == LUA_TSTRING) {
+            sds ok = sdsnew(lua_tostring(lua,-1));
+            sdsmapchars(ok,"\r\n","  ",2);
+            addReplySds(c,sdscatprintf(sdsempty(),"+%s\r\n",ok));
+            sdsfree(ok);
+            lua_pop(lua,2);
+            return;
+        }
+        lua_pop(lua,1); /* Discard field name pushed before. */
+
+        /* Handle double reply. */
+        lua_pushstring(lua,"double");
+        lua_gettable(lua,-2);
+        t = lua_type(lua,-1);
+        if (t == LUA_TNUMBER) {
+            addReplyDouble(c,lua_tonumber(lua,-1));
+            lua_pop(lua,2);
+            return;
+        }
+        lua_pop(lua,1); /* Discard field name pushed before. */
+
+        /* Handle big number reply. */
+        lua_pushstring(lua,"big_number");
+        lua_gettable(lua,-2);
+        t = lua_type(lua,-1);
+        if (t == LUA_TSTRING) {
+            sds big_num = sdsnewlen(lua_tostring(lua,-1), lua_strlen(lua,-1));
+            sdsmapchars(big_num,"\r\n","  ",2);
+            addReplyBigNum(c,big_num,sdslen(big_num));
+            sdsfree(big_num);
+            lua_pop(lua,2);
+            return;
+        }
+        lua_pop(lua,1); /* Discard field name pushed before. */
+
+        /* Handle verbatim reply. */
+        lua_pushstring(lua,"verbatim_string");
+        lua_gettable(lua,-2);
+        t = lua_type(lua,-1);
+        if (t == LUA_TTABLE) {
+            lua_pushstring(lua,"format");
+            lua_gettable(lua,-2);
+            t = lua_type(lua,-1);
+            if (t == LUA_TSTRING){
+                char* format = (char*)lua_tostring(lua,-1);
+                lua_pushstring(lua,"string");
+                lua_gettable(lua,-3);
+                t = lua_type(lua,-1);
+                if (t == LUA_TSTRING){
+                    size_t len;
+                    char* str = (char*)lua_tolstring(lua,-1,&len);
+                    addReplyVerbatim(c, str, len, format);
+                    lua_pop(lua,4);
+                    return;
+                }
+                lua_pop(lua,1);
+            }
+            lua_pop(lua,1);
+        }
+        lua_pop(lua,1); /* Discard field name pushed before. */
+
+        /* Handle map reply. */
+        lua_pushstring(lua,"map");
+        lua_gettable(lua,-2);
+        t = lua_type(lua,-1);
+        if (t == LUA_TTABLE) {
+            int maplen = 0;
+            void *replylen = addReplyDeferredLen(c);
+            /* we took care of the stack size on function start */
+            lua_pushnil(lua); /* Use nil to start iteration. */
+            while (lua_next(lua,-2)) {
+                /* Stack now: table, key, value */
+                lua_pushvalue(lua,-2);        /* Dup key before consuming. */
+                luaReplyToRedisReply(c, lua); /* Return key. */
+                luaReplyToRedisReply(c, lua); /* Return value. */
+                /* Stack now: table, key. */
+                maplen++;
+            }
+            setDeferredMapLen(c,replylen,maplen);
+            lua_pop(lua,2);
+            return;
+        }
+        lua_pop(lua,1); /* Discard field name pushed before. */
+
+        /* Handle set reply. */
+        lua_pushstring(lua,"set");
+        lua_gettable(lua,-2);
+        t = lua_type(lua,-1);
+        if (t == LUA_TTABLE) {
+            int setlen = 0;
+            void *replylen = addReplyDeferredLen(c);
+            /* we took care of the stack size on function start */
+            lua_pushnil(lua); /* Use nil to start iteration. */
+            while (lua_next(lua,-2)) {
+                /* Stack now: table, key, true */
+                lua_pop(lua,1);               /* Discard the boolean value. */
+                lua_pushvalue(lua,-1);        /* Dup key before consuming. */
+                luaReplyToRedisReply(c, lua); /* Return key. */
+                /* Stack now: table, key. */
+                setlen++;
+            }
+            setDeferredSetLen(c,replylen,setlen);
+            lua_pop(lua,2);
+            return;
+        }
+        lua_pop(lua,1); /* Discard field name pushed before. */
+
+        /* Handle the array reply. */
+        void *replylen = addReplyDeferredLen(c);
+        int j = 1, mbulklen = 0;
+        while(1) {
+            /* we took care of the stack size on function start */
+            lua_pushnumber(lua,j++);
+            lua_gettable(lua,-2);
+            t = lua_type(lua,-1);
+            if (t == LUA_TNIL) {
+                lua_pop(lua,1);
+                break;
+            }
+            luaReplyToRedisReply(c, lua);
+            mbulklen++;
+        }
+        setDeferredArrayLen(c,replylen,mbulklen);
+        break;
+    default:
+        addReplyNull(c);
+    }
+    lua_pop(lua,1);
+}
+
+/* ---------------------------------------------------------------------------
+ * Lua redis.* functions implementations.
+ * ------------------------------------------------------------------------- */
+
+#define LUA_CMD_OBJCACHE_SIZE 32
+#define LUA_CMD_OBJCACHE_MAX_LEN 64
+int luaRedisGenericCommand(lua_State *lua, int raise_error) {
+    int j, argc = lua_gettop(lua);
+    struct redisCommand *cmd;
+    client *c = server.lua_client;
+    sds reply;
+
+    /* Cached across calls. */
+    static robj **argv = NULL;
+    static int argv_size = 0;
+    static robj *cached_objects[LUA_CMD_OBJCACHE_SIZE];
+    static size_t cached_objects_len[LUA_CMD_OBJCACHE_SIZE];
+    static int inuse = 0;   /* Recursive calls detection. */
+
+    /* By using Lua debug hooks it is possible to trigger a recursive call
+     * to luaRedisGenericCommand(), which normally should never happen.
+     * To make this function reentrant is futile and makes it slower, but
+     * we should at least detect such a misuse, and abort. */
+    if (inuse) {
+        char *recursion_warning =
+            "luaRedisGenericCommand() recursive call detected. "
+            "Are you doing funny stuff with Lua debug hooks?";
+        serverLog(LL_WARNING,"%s",recursion_warning);
+        luaPushError(lua,recursion_warning);
+        return 1;
+    }
+    inuse++;
+
+    /* Require at least one argument */
+    if (argc == 0) {
+        luaPushError(lua,
+            "Please specify at least one argument for redis.call()");
+        inuse--;
+        return raise_error ? luaRaiseError(lua) : 1;
+    }
+
+    /* Build the arguments vector */
+    if (argv_size < argc) {
+        argv = zrealloc(argv,sizeof(robj*)*argc);
+        argv_size = argc;
+    }
+
+    for (j = 0; j < argc; j++) {
+        char *obj_s;
+        size_t obj_len;
+        char dbuf[64];
+
+        if (lua_type(lua,j+1) == LUA_TNUMBER) {
+            /* We can't use lua_tolstring() for number -> string conversion
+             * since Lua uses a format specifier that loses precision. */
+            lua_Number num = lua_tonumber(lua,j+1);
+
+            obj_len = snprintf(dbuf,sizeof(dbuf),"%.17g",(double)num);
+            obj_s = dbuf;
+        } else {
+            obj_s = (char*)lua_tolstring(lua,j+1,&obj_len);
+            if (obj_s == NULL) break; /* Not a string. */
+        }
+
+        /* Try to use a cached object. */
+        if (j < LUA_CMD_OBJCACHE_SIZE && cached_objects[j] &&
+            cached_objects_len[j] >= obj_len)
+        {
+            sds s = cached_objects[j]->ptr;
+            argv[j] = cached_objects[j];
+            cached_objects[j] = NULL;
+            memcpy(s,obj_s,obj_len+1);
+            sdssetlen(s, obj_len);
+        } else {
+            argv[j] = createStringObject(obj_s, obj_len);
+        }
+    }
+
+    /* Check if one of the arguments passed by the Lua script
+     * is not a string or an integer (lua_isstring() return true for
+     * integers as well). */
+    if (j != argc) {
+        j--;
+        while (j >= 0) {
+            decrRefCount(argv[j]);
+            j--;
+        }
+        luaPushError(lua,
+            "Lua redis() command arguments must be strings or integers");
+        inuse--;
+        return raise_error ? luaRaiseError(lua) : 1;
+    }
+
+    /* Pop all arguments from the stack, we do not need them anymore
+     * and this way we guaranty we will have room on the stack for the result. */
+    lua_pop(lua, argc);
+
+    /* Setup our fake client for command execution */
+    c->argv = argv;
+    c->argc = argc;
+    c->user = server.lua_caller->user;
+
+    /* Process module hooks */
+    moduleCallCommandFilters(c);
+    argv = c->argv;
+    argc = c->argc;
+
+    /* Log the command if debugging is active. */
+    if (ldb.active && ldb.step) {
+        sds cmdlog = sdsnew("<redis>");
+        for (j = 0; j < c->argc; j++) {
+            if (j == 10) {
+                cmdlog = sdscatprintf(cmdlog," ... (%d more)",
+                    c->argc-j-1);
+                break;
+            } else {
+                cmdlog = sdscatlen(cmdlog," ",1);
+                cmdlog = sdscatsds(cmdlog,c->argv[j]->ptr);
+            }
+        }
+        ldbLog(cmdlog);
+    }
+
+    /* Command lookup */
+    cmd = lookupCommand(argv[0]->ptr);
+    if (!cmd || ((cmd->arity > 0 && cmd->arity != argc) ||
+                   (argc < -cmd->arity)))
+    {
+        if (cmd)
+            luaPushError(lua,
+                "Wrong number of args calling Redis command From Lua script");
+        else
+            luaPushError(lua,"Unknown Redis command called from Lua script");
+        goto cleanup;
+    }
+    c->cmd = c->lastcmd = cmd;
+
+    /* There are commands that are not allowed inside scripts. */
+    if (!server.lua_disable_deny_script && (cmd->flags & CMD_NOSCRIPT)) {
+        luaPushError(lua, "This Redis command is not allowed from scripts");
+        goto cleanup;
+    }
+
+    /* This check is for EVAL_RO, EVALSHA_RO. We want to allow only read only commands */
+    if ((server.lua_caller->cmd->proc == evalRoCommand ||
+         server.lua_caller->cmd->proc == evalShaRoCommand) &&
+         (cmd->flags & CMD_WRITE))
+    {
+        luaPushError(lua, "Write commands are not allowed from read-only scripts");
+        goto cleanup;
+    }
+
+    /* 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:
+            luaPushError(lua, "The user executing the script can't run this "
+                              "command or subcommand");
+            break;
+        case ACL_DENIED_KEY:
+            luaPushError(lua, "The user executing the script can't access "
+                              "at least one of the keys mentioned in the "
+                              "command arguments");
+            break;
+        case ACL_DENIED_CHANNEL:
+            luaPushError(lua, "The user executing the script can't publish "
+                              "to the channel mentioned in the command");
+            break;
+        default:
+            luaPushError(lua, "The user executing the script is lacking the "
+                              "permissions for the command");
+            break;
+        }
+        goto cleanup;
+    }
+
+    /* 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. */
+    if (cmd->flags & CMD_WRITE) {
+        int deny_write_type = writeCommandsDeniedByDiskError();
+        if (server.lua_random_dirty && !server.lua_replicate_commands) {
+            luaPushError(lua,
+                "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.");
+            goto cleanup;
+        } else if (server.masterhost && server.repl_slave_ro &&
+                   server.lua_caller->id != CLIENT_ID_AOF &&
+                   !(server.lua_caller->flags & CLIENT_MASTER))
+        {
+            luaPushError(lua, shared.roslaveerr->ptr);
+            goto cleanup;
+        } else if (deny_write_type != DISK_ERROR_TYPE_NONE) {
+            if (deny_write_type == DISK_ERROR_TYPE_RDB) {
+                luaPushError(lua, shared.bgsaveerr->ptr);
+            } else {
+                sds aof_write_err = sdscatfmt(sdsempty(),
+                    "-MISCONF Errors writing to the AOF file: %s\r\n",
+                    strerror(server.aof_last_write_errno));
+                luaPushError(lua, aof_write_err);
+                sdsfree(aof_write_err);
+            }
+            goto cleanup;
+        }
+    }
+
+    /* 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. */
+        server.lua_caller->id != CLIENT_ID_AOF &&  /* Don't care about mem if loading from AOF. */
+        !server.masterhost &&                      /* Slave must execute the script. */
+        server.lua_write_dirty == 0 &&             /* Script had no side effects so far. */
+        server.lua_oom &&                          /* Detected OOM when script start. */
+        (cmd->flags & CMD_DENYOOM))
+    {
+        luaPushError(lua, shared.oomerr->ptr);
+        goto cleanup;
+    }
+
+    if (cmd->flags & CMD_RANDOM) server.lua_random_dirty = 1;
+    if (cmd->flags & CMD_WRITE) server.lua_write_dirty = 1;
+
+    /* If this is a Redis Cluster node, we need to make sure Lua 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. */
+    if (server.cluster_enabled && server.lua_caller->id != CLIENT_ID_AOF &&
+        !(server.lua_caller->flags & CLIENT_MASTER))
+    {
+        int error_code;
+        /* Duplicate relevant flags in the lua client. */
+        c->flags &= ~(CLIENT_READONLY|CLIENT_ASKING);
+        c->flags |= server.lua_caller->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) {
+                luaPushError(lua,
+                    "Lua script attempted to execute a write command while the "
+                    "cluster is down and readonly");
+            } else if (error_code == CLUSTER_REDIR_DOWN_STATE) {
+                luaPushError(lua,
+                    "Lua script attempted to execute a command while the "
+                    "cluster is down");
+            } else {
+                luaPushError(lua,
+                    "Lua script attempted to access a non local key in a "
+                    "cluster node");
+            }
+
+            goto cleanup;
+        }
+    }
+
+    /* 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. */
+    if (server.lua_replicate_commands &&
+        !server.lua_multi_emitted &&
+        !(server.lua_caller->flags & CLIENT_MULTI) &&
+        server.lua_write_dirty &&
+        server.lua_repl != PROPAGATE_NONE)
+    {
+        execCommandPropagateMulti(server.lua_caller->db->id);
+        server.lua_multi_emitted = 1;
+        /* Now we are in the MULTI context, the lua_client should be
+         * flag as CLIENT_MULTI. */
+        c->flags |= CLIENT_MULTI;
+    }
+
+    /* Run the command */
+    int call_flags = CMD_CALL_SLOWLOG | CMD_CALL_STATS;
+    if (server.lua_replicate_commands) {
+        /* Set flags according to redis.set_repl() settings. */
+        if (server.lua_repl & PROPAGATE_AOF)
+            call_flags |= CMD_CALL_PROPAGATE_AOF;
+        if (server.lua_repl & PROPAGATE_REPL)
+            call_flags |= CMD_CALL_PROPAGATE_REPL;
+    }
+    call(c,call_flags);
+    serverAssert((c->flags & CLIENT_BLOCKED) == 0);
+
+    /* Convert the result of the Redis command into a suitable Lua type.
+     * The first thing we need is to create a single string from the client
+     * output buffers. */
+    if (listLength(c->reply) == 0 && (size_t)c->bufpos < c->buf_usable_size) {
+        /* This is a fast path for the common case of a reply inside the
+         * client static buffer. Don't create an SDS string but just use
+         * the client buffer directly. */
+        c->buf[c->bufpos] = '\0';
+        reply = c->buf;
+        c->bufpos = 0;
+    } else {
+        reply = sdsnewlen(c->buf,c->bufpos);
+        c->bufpos = 0;
+        while(listLength(c->reply)) {
+            clientReplyBlock *o = listNodeValue(listFirst(c->reply));
+
+            reply = sdscatlen(reply,o->buf,o->used);
+            listDelNode(c->reply,listFirst(c->reply));
+        }
+    }
+    if (raise_error && reply[0] != '-') raise_error = 0;
+    redisProtocolToLuaType(lua,reply);
+
+    /* If the debugger is active, log the reply from Redis. */
+    if (ldb.active && ldb.step)
+        ldbLogRedisReply(reply);
+
+    /* Sort the output array if needed, assuming it is a non-null multi bulk
+     * reply as expected. */
+    if ((cmd->flags & CMD_SORT_FOR_SCRIPT) &&
+        (server.lua_replicate_commands == 0) &&
+        (reply[0] == '*' && reply[1] != '-')) {
+            luaSortArray(lua);
+    }
+    if (reply != c->buf) sdsfree(reply);
+    c->reply_bytes = 0;
+
+cleanup:
+    /* Clean up. Command code may have changed argv/argc so we use the
+     * argv/argc of the client instead of the local variables. */
+    for (j = 0; j < c->argc; j++) {
+        robj *o = c->argv[j];
+
+        /* Try to cache the object in the cached_objects array.
+         * The object must be small, SDS-encoded, and with refcount = 1
+         * (we must be the only owner) for us to cache it. */
+        if (j < LUA_CMD_OBJCACHE_SIZE &&
+            o->refcount == 1 &&
+            (o->encoding == OBJ_ENCODING_RAW ||
+             o->encoding == OBJ_ENCODING_EMBSTR) &&
+            sdslen(o->ptr) <= LUA_CMD_OBJCACHE_MAX_LEN)
+        {
+            sds s = o->ptr;
+            if (cached_objects[j]) decrRefCount(cached_objects[j]);
+            cached_objects[j] = o;
+            cached_objects_len[j] = sdsalloc(s);
+        } else {
+            decrRefCount(o);
+        }
+    }
+
+    if (c->argv != argv) {
+        zfree(c->argv);
+        argv = NULL;
+        argv_size = 0;
+    }
+
+    c->user = NULL;
+
+    if (raise_error) {
+        /* If we are here we should have an error in the stack, in the
+         * form of a table with an "err" field. Extract the string to
+         * return the plain error. */
+        inuse--;
+        return luaRaiseError(lua);
+    }
+    inuse--;
+    return 1;
+}
+
+/* redis.call() */
+int luaRedisCallCommand(lua_State *lua) {
+    return luaRedisGenericCommand(lua,1);
+}
+
+/* redis.pcall() */
+int luaRedisPCallCommand(lua_State *lua) {
+    return luaRedisGenericCommand(lua,0);
+}
+
+/* This adds redis.sha1hex(string) to Lua scripts using the same hashing
+ * function used for sha1ing lua scripts. */
+int luaRedisSha1hexCommand(lua_State *lua) {
+    int argc = lua_gettop(lua);
+    char digest[41];
+    size_t len;
+    char *s;
+
+    if (argc != 1) {
+        lua_pushstring(lua, "wrong number of arguments");
+        return lua_error(lua);
+    }
+
+    s = (char*)lua_tolstring(lua,1,&len);
+    sha1hex(digest,s,len);
+    lua_pushstring(lua,digest);
+    return 1;
+}
+
+/* Returns a table with a single field 'field' set to the string value
+ * passed as argument. This helper function is handy when returning
+ * a Redis Protocol error or status reply from Lua:
+ *
+ * return redis.error_reply("ERR Some Error")
+ * return redis.status_reply("ERR Some Error")
+ */
+int luaRedisReturnSingleFieldTable(lua_State *lua, char *field) {
+    if (lua_gettop(lua) != 1 || lua_type(lua,-1) != LUA_TSTRING) {
+        luaPushError(lua, "wrong number or type of arguments");
+        return 1;
+    }
+
+    lua_newtable(lua);
+    lua_pushstring(lua, field);
+    lua_pushvalue(lua, -3);
+    lua_settable(lua, -3);
+    return 1;
+}
+
+/* redis.error_reply() */
+int luaRedisErrorReplyCommand(lua_State *lua) {
+    return luaRedisReturnSingleFieldTable(lua,"err");
+}
+
+/* redis.status_reply() */
+int luaRedisStatusReplyCommand(lua_State *lua) {
+    return luaRedisReturnSingleFieldTable(lua,"ok");
+}
+
+/* redis.set_repl()
+ *
+ * Set the propagation of write commands executed in the context of the
+ * script to on/off for AOF and slaves. */
+int luaRedisSetReplCommand(lua_State *lua) {
+    int argc = lua_gettop(lua);
+    int flags;
+
+    if (server.lua_replicate_commands == 0) {
+        lua_pushstring(lua, "You can set the replication behavior only after turning on single commands replication with redis.replicate_commands().");
+        return lua_error(lua);
+    } else if (argc != 1) {
+        lua_pushstring(lua, "redis.set_repl() requires two arguments.");
+        return lua_error(lua);
+    }
+
+    flags = lua_tonumber(lua,-1);
+    if ((flags & ~(PROPAGATE_AOF|PROPAGATE_REPL)) != 0) {
+        lua_pushstring(lua, "Invalid replication flags. Use REPL_AOF, REPL_REPLICA, REPL_ALL or REPL_NONE.");
+        return lua_error(lua);
+    }
+    server.lua_repl = flags;
+    return 0;
+}
+
+/* redis.log() */
+int luaLogCommand(lua_State *lua) {
+    int j, argc = lua_gettop(lua);
+    int level;
+    sds log;
+
+    if (argc < 2) {
+        lua_pushstring(lua, "redis.log() requires two arguments or more.");
+        return lua_error(lua);
+    } else if (!lua_isnumber(lua,-argc)) {
+        lua_pushstring(lua, "First argument must be a number (log level).");
+        return lua_error(lua);
+    }
+    level = lua_tonumber(lua,-argc);
+    if (level < LL_DEBUG || level > LL_WARNING) {
+        lua_pushstring(lua, "Invalid debug level.");
+        return lua_error(lua);
+    }
+    if (level < server.verbosity) return 0;
+
+    /* Glue together all the arguments */
+    log = sdsempty();
+    for (j = 1; j < argc; j++) {
+        size_t len;
+        char *s;
+
+        s = (char*)lua_tolstring(lua,(-argc)+j,&len);
+        if (s) {
+            if (j != 1) log = sdscatlen(log," ",1);
+            log = sdscatlen(log,s,len);
+        }
+    }
+    serverLogRaw(level,log);
+    sdsfree(log);
+    return 0;
+}
+
+/* redis.setresp() */
+int luaSetResp(lua_State *lua) {
+    int argc = lua_gettop(lua);
+
+    if (argc != 1) {
+        lua_pushstring(lua, "redis.setresp() requires one argument.");
+        return lua_error(lua);
+    }
+
+    int resp = lua_tonumber(lua,-argc);
+    if (resp != 2 && resp != 3) {
+        lua_pushstring(lua, "RESP version must be 2 or 3.");
+        return lua_error(lua);
+    }
+
+    server.lua_client->resp = resp;
+    return 0;
+}
+
+/* ---------------------------------------------------------------------------
+ * Lua engine initialization and reset.
+ * ------------------------------------------------------------------------- */
+
+void luaLoadLib(lua_State *lua, const char *libname, lua_CFunction luafunc) {
+  lua_pushcfunction(lua, luafunc);
+  lua_pushstring(lua, libname);
+  lua_call(lua, 1, 0);
+}
+
+LUALIB_API int (luaopen_cjson) (lua_State *L);
+LUALIB_API int (luaopen_struct) (lua_State *L);
+LUALIB_API int (luaopen_cmsgpack) (lua_State *L);
+LUALIB_API int (luaopen_bit) (lua_State *L);
+
+void luaLoadLibraries(lua_State *lua) {
+    luaLoadLib(lua, "", luaopen_base);
+    luaLoadLib(lua, LUA_TABLIBNAME, luaopen_table);
+    luaLoadLib(lua, LUA_STRLIBNAME, luaopen_string);
+    luaLoadLib(lua, LUA_MATHLIBNAME, luaopen_math);
+    luaLoadLib(lua, LUA_DBLIBNAME, luaopen_debug);
+    luaLoadLib(lua, "cjson", luaopen_cjson);
+    luaLoadLib(lua, "struct", luaopen_struct);
+    luaLoadLib(lua, "cmsgpack", luaopen_cmsgpack);
+    luaLoadLib(lua, "bit", luaopen_bit);
+
+#if 0 /* Stuff that we don't load currently, for sandboxing concerns. */
+    luaLoadLib(lua, LUA_LOADLIBNAME, luaopen_package);
+    luaLoadLib(lua, LUA_OSLIBNAME, luaopen_os);
+#endif
+}
+
+/* Remove a functions that we don't want to expose to the Redis scripting
+ * environment. */
+void luaRemoveUnsupportedFunctions(lua_State *lua) {
+    lua_pushnil(lua);
+    lua_setglobal(lua,"loadfile");
+    lua_pushnil(lua);
+    lua_setglobal(lua,"dofile");
+}
+
+/* This function installs metamethods in the global table _G that prevent
+ * the creation of globals accidentally.
+ *
+ * It should be the last to be called in the scripting engine initialization
+ * sequence, because it may interact with creation of globals. */
+void scriptingEnableGlobalsProtection(lua_State *lua) {
+    char *s[32];
+    sds code = sdsempty();
+    int j = 0;
+
+    /* strict.lua from: http://metalua.luaforge.net/src/lib/strict.lua.html.
+     * Modified to be adapted to Redis. */
+    s[j++]="local dbg=debug\n";
+    s[j++]="local mt = {}\n";
+    s[j++]="setmetatable(_G, mt)\n";
+    s[j++]="mt.__newindex = function (t, n, v)\n";
+    s[j++]="  if dbg.getinfo(2) then\n";
+    s[j++]="    local w = dbg.getinfo(2, \"S\").what\n";
+    s[j++]="    if w ~= \"main\" and w ~= \"C\" then\n";
+    s[j++]="      error(\"Script attempted to create global variable '\"..tostring(n)..\"'\", 2)\n";
+    s[j++]="    end\n";
+    s[j++]="  end\n";
+    s[j++]="  rawset(t, n, v)\n";
+    s[j++]="end\n";
+    s[j++]="mt.__index = function (t, n)\n";
+    s[j++]="  if dbg.getinfo(2) and dbg.getinfo(2, \"S\").what ~= \"C\" then\n";
+    s[j++]="    error(\"Script attempted to access nonexistent global variable '\"..tostring(n)..\"'\", 2)\n";
+    s[j++]="  end\n";
+    s[j++]="  return rawget(t, n)\n";
+    s[j++]="end\n";
+    s[j++]="debug = nil\n";
+    s[j++]=NULL;
+
+    for (j = 0; s[j] != NULL; j++) code = sdscatlen(code,s[j],strlen(s[j]));
+    luaL_loadbuffer(lua,code,sdslen(code),"@enable_strict_lua");
+    lua_pcall(lua,0,0,0);
+    sdsfree(code);
+}
+
+void luaEngineRegisterRedisAPI(lua_State* lua) {
+    luaLoadLibraries(lua);
+    luaRemoveUnsupportedFunctions(lua);
+
+    /* Register the redis commands table and fields */
+    lua_newtable(lua);
+
+    /* redis.call */
+    lua_pushstring(lua,"call");
+    lua_pushcfunction(lua,luaRedisCallCommand);
+    lua_settable(lua,-3);
+
+    /* redis.pcall */
+    lua_pushstring(lua,"pcall");
+    lua_pushcfunction(lua,luaRedisPCallCommand);
+    lua_settable(lua,-3);
+
+    /* redis.log and log levels. */
+    lua_pushstring(lua,"log");
+    lua_pushcfunction(lua,luaLogCommand);
+    lua_settable(lua,-3);
+
+    /* redis.setresp */
+    lua_pushstring(lua,"setresp");
+    lua_pushcfunction(lua,luaSetResp);
+    lua_settable(lua,-3);
+
+    lua_pushstring(lua,"LOG_DEBUG");
+    lua_pushnumber(lua,LL_DEBUG);
+    lua_settable(lua,-3);
+
+    lua_pushstring(lua,"LOG_VERBOSE");
+    lua_pushnumber(lua,LL_VERBOSE);
+    lua_settable(lua,-3);
+
+    lua_pushstring(lua,"LOG_NOTICE");
+    lua_pushnumber(lua,LL_NOTICE);
+    lua_settable(lua,-3);
+
+    lua_pushstring(lua,"LOG_WARNING");
+    lua_pushnumber(lua,LL_WARNING);
+    lua_settable(lua,-3);
+
+    /* redis.sha1hex */
+    lua_pushstring(lua, "sha1hex");
+    lua_pushcfunction(lua, luaRedisSha1hexCommand);
+    lua_settable(lua, -3);
+
+    /* redis.error_reply and redis.status_reply */
+    lua_pushstring(lua, "error_reply");
+    lua_pushcfunction(lua, luaRedisErrorReplyCommand);
+    lua_settable(lua, -3);
+    lua_pushstring(lua, "status_reply");
+    lua_pushcfunction(lua, luaRedisStatusReplyCommand);
+    lua_settable(lua, -3);
+
+    /* redis.set_repl and associated flags. */
+    lua_pushstring(lua,"set_repl");
+    lua_pushcfunction(lua,luaRedisSetReplCommand);
+    lua_settable(lua,-3);
+
+    lua_pushstring(lua,"REPL_NONE");
+    lua_pushnumber(lua,PROPAGATE_NONE);
+    lua_settable(lua,-3);
+
+    lua_pushstring(lua,"REPL_AOF");
+    lua_pushnumber(lua,PROPAGATE_AOF);
+    lua_settable(lua,-3);
+
+    lua_pushstring(lua,"REPL_SLAVE");
+    lua_pushnumber(lua,PROPAGATE_REPL);
+    lua_settable(lua,-3);
+
+    lua_pushstring(lua,"REPL_REPLICA");
+    lua_pushnumber(lua,PROPAGATE_REPL);
+    lua_settable(lua,-3);
+
+    lua_pushstring(lua,"REPL_ALL");
+    lua_pushnumber(lua,PROPAGATE_AOF|PROPAGATE_REPL);
+
+    lua_settable(lua,-3);
+    /* Finally set the table as 'redis' global var. */
+    lua_setglobal(lua,"redis");
+
+    /* Replace math.random and math.randomseed with our implementations. */
+    lua_getglobal(lua,"math");
+
+    lua_pushstring(lua,"random");
+    lua_pushcfunction(lua,redis_math_random);
+    lua_settable(lua,-3);
+
+    lua_pushstring(lua,"randomseed");
+    lua_pushcfunction(lua,redis_math_randomseed);
+    lua_settable(lua,-3);
+
+    lua_setglobal(lua,"math");
+}
+
+/* Set an array of Redis String Objects as a Lua array (table) stored into a
+ * global variable. */
+void luaSetGlobalArray(lua_State *lua, char *var, robj **elev, int elec) {
+    int j;
+
+    lua_newtable(lua);
+    for (j = 0; j < elec; j++) {
+        lua_pushlstring(lua,(char*)elev[j]->ptr,sdslen(elev[j]->ptr));
+        lua_rawseti(lua,-2,j+1);
+    }
+    lua_setglobal(lua,var);
+}
+
+/* ---------------------------------------------------------------------------
+ * Redis provided math.random
+ * ------------------------------------------------------------------------- */
+
+/* We replace math.random() with our implementation that is not affected
+ * by specific libc random() implementations and will output the same sequence
+ * (for the same seed) in every arch. */
+
+/* The following implementation is the one shipped with Lua itself but with
+ * rand() replaced by redisLrand48(). */
+int redis_math_random (lua_State *L) {
+  /* the `%' avoids the (rare) case of r==1, and is needed also because on
+     some systems (SunOS!) `rand()' may return a value larger than RAND_MAX */
+  lua_Number r = (lua_Number)(redisLrand48()%REDIS_LRAND48_MAX) /
+                                (lua_Number)REDIS_LRAND48_MAX;
+  switch (lua_gettop(L)) {  /* check number of arguments */
+    case 0: {  /* no arguments */
+      lua_pushnumber(L, r);  /* Number between 0 and 1 */
+      break;
+    }
+    case 1: {  /* only upper limit */
+      int u = luaL_checkint(L, 1);
+      luaL_argcheck(L, 1<=u, 1, "interval is empty");
+      lua_pushnumber(L, floor(r*u)+1);  /* int between 1 and `u' */
+      break;
+    }
+    case 2: {  /* lower and upper limits */
+      int l = luaL_checkint(L, 1);
+      int u = luaL_checkint(L, 2);
+      luaL_argcheck(L, l<=u, 2, "interval is empty");
+      lua_pushnumber(L, floor(r*(u-l+1))+l);  /* int between `l' and `u' */
+      break;
+    }
+    default: return luaL_error(L, "wrong number of arguments");
+  }
+  return 1;
+}
+
+int redis_math_randomseed (lua_State *L) {
+  redisSrand48(luaL_checkint(L, 1));
+  return 0;
+}
+
+/* This is the Lua script "count" hook that we use to detect scripts timeout. */
+void luaMaskCountHook(lua_State *lua, lua_Debug *ar) {
+    long long elapsed = elapsedMs(server.lua_time_start);
+    UNUSED(ar);
+    UNUSED(lua);
+
+    /* Set the timeout condition if not already set and the maximum
+     * execution time was reached. */
+    if (elapsed >= server.lua_time_limit && server.lua_timedout == 0) {
+        serverLog(LL_WARNING,
+            "Lua slow script detected: still in execution after %lld milliseconds. "
+            "You can try killing the script using the SCRIPT KILL command. "
+            "Script SHA1 is: %s",
+            elapsed, server.lua_cur_script);
+        server.lua_timedout = 1;
+        blockingOperationStarts();
+        /* Once the script timeouts we reenter the event loop to permit others
+         * to call SCRIPT KILL or SHUTDOWN NOSAVE if needed. 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(server.lua_caller);
+    }
+    if (server.lua_timedout) processEventsWhileBlocked();
+    if (server.lua_kill) {
+        serverLog(LL_WARNING,"Lua script killed by user with SCRIPT KILL.");
+
+        /*
+         * Set the hook to invoke all the time so the user
+         * will not be able to catch the error with pcall and invoke
+         * pcall again which will prevent the script from ever been killed
+         */
+        lua_sethook(lua, luaMaskCountHook, LUA_MASKLINE, 0);
+
+        lua_pushstring(lua,"Script killed by user with SCRIPT KILL...");
+        lua_error(lua);
+    }
+}