Redis Functions - Move code to make review process easier.

This commit is only move code around without changing it.
The reason behind this is to make review process easier
by allowing the reviewer to simply ignore all code movements.

changes:
1. rename scripting.c to eval.c
2. introduce and new file, script_lua.c, and move parts of Lua
   functionality to this new file. script_lua.c will eventually
   contains the shared code between legacy lua and lua engine.

This commit does not compiled on purpose. Its only purpose is to move
code and rename files.

1 files changed, 1423 insertions, 0 deletions

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);
+    }
+}