Scripting: native lua.exists() implementation.

This was a test to check how much faster native implementation would be.
In initial tests it does not look like is this huge win.

After all this is at least in part obvious. Now scripting.c tries to
avoid allocations of argument vectors, and turning ":1" accumulated in
the client buffer into a Lua type is a fast operation.

src/blocked.c

@@ -118,9 +118,7 @@ void processUnblockedClients(void) {
 
         /* Process remaining data in the input buffer. */
         if (c->querybuf && sdslen(c->querybuf) > 0) {
-            server.current_client = c;
             processInputBuffer(c);
-            server.current_client = NULL;
         }
     }
 }

src/cluster.c

@@ -4584,7 +4584,7 @@ try_again:
             /* On error assume that last_dbid is no longer valid. */
             cs->last_dbid = -1;
             addReplyErrorFormat(c,"Target instance replied with error: %s",
-                (cs->last_dbid != dbid && buf1[0] == '-') ? buf1+1 : buf2+1);
+                (select && buf1[0] == '-') ? buf1+1 : buf2+1);
         } else {
             /* Update the last_dbid in migrateCachedSocket */
             cs->last_dbid = dbid;

src/db.c

@@ -62,7 +62,7 @@ robj *lookupKeyRead(redisDb *db, robj *key) {
 
     if (expireIfNeeded(db,key) == 1) {
         /* Key expired. If we are in the context of a master, expireIfNeeded()
-         * returns 0 only when the key does not exist at all, so it's save
+         * returns 0 only when the key does not exist at all, so it's safe
          * to return NULL ASAP. */
         if (server.masterhost == NULL) return NULL;
 

src/latency.c

@@ -248,7 +248,7 @@ sds createLatencyReport(void) {
     dictEntry *de;
     int eventnum = 0;
 
-    di = dictGetIterator(server.latency_events);
+    di = dictGetSafeIterator(server.latency_events);
     while((de = dictNext(di)) != NULL) {
         char *event = dictGetKey(de);
         struct latencyTimeSeries *ts = dictGetVal(de);

src/multi.c

@@ -304,6 +304,9 @@ void touchWatchedKeysOnFlush(int dbid) {
     }
 }
 
+/* WATCH key key key ...
+ * Add keys to the set of watched keys. If those keys are modified
+ * before the next transaction is executed, the transaction aborts. */
 void watchCommand(redisClient *c) {
     int j;
 
@@ -316,8 +319,47 @@ void watchCommand(redisClient *c) {
     addReply(c,shared.ok);
 }
 
+/* UNWATCH
+ * Flush the set of watched keys. */
 void unwatchCommand(redisClient *c) {
     unwatchAllKeys(c);
     c->flags &= (~REDIS_DIRTY_CAS);
     addReply(c,shared.ok);
 }
+
+/* IF XX key key key ...
+ * IF NX key key key ...
+ * Abort transaction if condition is not met. */
+void ifCommand(redisClient *c) {
+    int j;
+    int xx = 0, nx = 0;
+
+    if (!(c->flags & REDIS_MULTI)) {
+        addReplyError(c,"IF outside MULTI is not allowed");
+        return;
+    }
+
+    /* Check if it's xx or nx option, trying to do it reasonably fast. */
+    char *a = c->argv[1]->ptr;
+    if ((a[0] == 'n' || a[0] == 'N') &&
+        (a[1] == 'x' || a[1] == 'X') && a[2] == '\0')
+    {
+        nx = 1;
+    } else if ((a[0] == 'x' || a[0] == 'X') &&
+               (a[1] == 'x' || a[1] == 'X') && a[2] == '\0')
+    {
+        xx = 1;
+    } else {
+        addReply(c,shared.syntaxerr);
+        return;
+    }
+
+    /* Check if keys exist / don't exist, and flag the transaction if
+     * at least one key fails the test according to NX / XX option. */
+    for (j = 2; j < c->argc; j++) {
+        robj *o = lookupKeyRead(c->db,c->argv[j]);
+        if ((o != NULL && nx) || (o == NULL && xx))
+            c->flags |= REDIS_DIRTY_CAS;
+    }
+    addReply(c,shared.ok);
+}

src/networking.c

@@ -1106,18 +1106,19 @@ int processMultibulkBuffer(redisClient *c) {
 }
 
 void processInputBuffer(redisClient *c) {
+    server.current_client = c;
     /* Keep processing while there is something in the input buffer */
     while(sdslen(c->querybuf)) {
         /* Return if clients are paused. */
-        if (!(c->flags & REDIS_SLAVE) && clientsArePaused()) return;
+        if (!(c->flags & REDIS_SLAVE) && clientsArePaused()) break;
 
         /* Immediately abort if the client is in the middle of something. */
-        if (c->flags & REDIS_BLOCKED) return;
+        if (c->flags & REDIS_BLOCKED) break;
 
         /* REDIS_CLOSE_AFTER_REPLY closes the connection once the reply is
          * written to the client. Make sure to not let the reply grow after
          * this flag has been set (i.e. don't process more commands). */
-        if (c->flags & REDIS_CLOSE_AFTER_REPLY) return;
+        if (c->flags & REDIS_CLOSE_AFTER_REPLY) break;
 
         /* Determine request type when unknown. */
         if (!c->reqtype) {
@@ -1145,6 +1146,7 @@ void processInputBuffer(redisClient *c) {
                 resetClient(c);
         }
     }
+    server.current_client = NULL;
 }
 
 void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) {
@@ -1154,7 +1156,6 @@ void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) {
     REDIS_NOTUSED(el);
     REDIS_NOTUSED(mask);
 
-    server.current_client = c;
     readlen = REDIS_IOBUF_LEN;
     /* If this is a multi bulk request, and we are processing a bulk reply
      * that is large enough, try to maximize the probability that the query
@@ -1176,7 +1177,7 @@ void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) {
     nread = read(fd, c->querybuf+qblen, readlen);
     if (nread == -1) {
         if (errno == EAGAIN) {
-            nread = 0;
+            return;
         } else {
             redisLog(REDIS_VERBOSE, "Reading from client: %s",strerror(errno));
             freeClient(c);
@@ -1187,15 +1188,11 @@ void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) {
         freeClient(c);
         return;
     }
-    if (nread) {
-        sdsIncrLen(c->querybuf,nread);
-        c->lastinteraction = server.unixtime;
-        if (c->flags & REDIS_MASTER) c->reploff += nread;
-        server.stat_net_input_bytes += nread;
-    } else {
-        server.current_client = NULL;
-        return;
-    }
+
+    sdsIncrLen(c->querybuf,nread);
+    c->lastinteraction = server.unixtime;
+    if (c->flags & REDIS_MASTER) c->reploff += nread;
+    server.stat_net_input_bytes += nread;
     if (sdslen(c->querybuf) > server.client_max_querybuf_len) {
         sds ci = catClientInfoString(sdsempty(),c), bytes = sdsempty();
 
@@ -1207,7 +1204,6 @@ void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask) {
         return;
     }
     processInputBuffer(c);
-    server.current_client = NULL;
 }
 
 void getClientsMaxBuffers(unsigned long *longest_output_list,

src/object.c

@@ -529,9 +529,7 @@ size_t stringObjectLen(robj *o) {
     if (sdsEncodedObject(o)) {
         return sdslen(o->ptr);
     } else {
-        char buf[32];
-
-        return ll2string(buf,32,(long)o->ptr);
+        return sdigits10((long)o->ptr);
     }
 }
 

src/redis.c

@@ -202,6 +202,7 @@ struct redisCommand redisCommandTable[] = {
     {"hincrbyfloat",hincrbyfloatCommand,4,"wmF",0,NULL,1,1,1,0,0},
     {"hdel",hdelCommand,-3,"wF",0,NULL,1,1,1,0,0},
     {"hlen",hlenCommand,2,"rF",0,NULL,1,1,1,0,0},
+    {"hstrlen",hstrlenCommand,3,"rF",0,NULL,1,1,1,0,0},
     {"hkeys",hkeysCommand,2,"rS",0,NULL,1,1,1,0,0},
     {"hvals",hvalsCommand,2,"rS",0,NULL,1,1,1,0,0},
     {"hgetall",hgetallCommand,2,"r",0,NULL,1,1,1,0,0},
@@ -260,6 +261,7 @@ struct redisCommand redisCommandTable[] = {
     {"pubsub",pubsubCommand,-2,"pltrR",0,NULL,0,0,0,0,0},
     {"watch",watchCommand,-2,"rsF",0,NULL,1,-1,1,0,0},
     {"unwatch",unwatchCommand,1,"rsF",0,NULL,0,0,0,0,0},
+    {"if",ifCommand,-2,"rsF",0,NULL,2,-1,1,0,0},
     {"cluster",clusterCommand,-2,"ar",0,NULL,0,0,0,0,0},
     {"restore",restoreCommand,-4,"wm",0,NULL,1,1,1,0,0},
     {"restore-asking",restoreCommand,-4,"wmk",0,NULL,1,1,1,0,0},
@@ -2335,7 +2337,7 @@ int processCommand(redisClient *c) {
 
     /* Exec the command */
     if (c->flags & REDIS_MULTI &&
-        c->cmd->proc != execCommand && c->cmd->proc != discardCommand &&
+        c->cmd->proc != execCommand && c->cmd->proc != ifCommand &&
         c->cmd->proc != multiCommand && c->cmd->proc != watchCommand)
     {
         queueMultiCommand(c);

src/redis.h

@@ -1516,6 +1516,7 @@ void hmsetCommand(redisClient *c);
 void hmgetCommand(redisClient *c);
 void hdelCommand(redisClient *c);
 void hlenCommand(redisClient *c);
+void hstrlenCommand(redisClient *c);
 void zremrangebyrankCommand(redisClient *c);
 void zunionstoreCommand(redisClient *c);
 void zinterstoreCommand(redisClient *c);
@@ -1560,6 +1561,7 @@ void pfcountCommand(redisClient *c);
 void pfmergeCommand(redisClient *c);
 void pfdebugCommand(redisClient *c);
 void latencyCommand(redisClient *c);
+void ifCommand(redisClient *c);
 
 #if defined(__GNUC__)
 void *calloc(size_t count, size_t size) __attribute__ ((deprecated));

src/scripting.c

@@ -458,6 +458,35 @@ int luaRedisPCallCommand(lua_State *lua) {
     return luaRedisGenericCommand(lua,0);
 }
 
+int luaExistsCommand(lua_State *lua) {
+    int argc = lua_gettop(lua);
+    redisClient *c = server.lua_client;
+    char *obj_s;
+    size_t obj_len;
+    robj *key, *val;
+
+    /* Require at least one argument */
+    if (argc != 1) {
+        luaPushError(lua,
+            "redis.exists() needs exactly one argument");
+        return 1;
+    }
+
+    obj_s = (char*)lua_tolstring(lua,1,&obj_len);
+    if (obj_s == NULL) {
+        luaPushError(lua,
+            "redis.exists() argument must be a string");
+        return 1;
+    }
+
+    key = createStringObject(obj_s,obj_len);
+    val = lookupKeyRead(c->db,key);
+    decrRefCount(key);
+
+    lua_pushnumber(lua,val != NULL);
+    return 1;
+}
+
 /* This adds redis.sha1hex(string) to Lua scripts using the same hashing
  * function used for sha1ing lua scripts. */
 int luaRedisSha1hexCommand(lua_State *lua) {
@@ -664,6 +693,11 @@ void scriptingInit(void) {
     lua_pushcfunction(lua,luaRedisPCallCommand);
     lua_settable(lua,-3);
 
+    /* redis.exists */
+    lua_pushstring(lua,"exists");
+    lua_pushcfunction(lua,luaExistsCommand);
+    lua_settable(lua,-3);
+
     /* redis.log and log levels. */
     lua_pushstring(lua,"log");
     lua_pushcfunction(lua,luaLogCommand);

src/t_hash.c

@@ -130,7 +130,6 @@ robj *hashTypeGetObject(robj *o, robj *field) {
                 value = createStringObjectFromLongLong(vll);
             }
         }
-
     } else if (o->encoding == REDIS_ENCODING_HT) {
         robj *aux;
 
@@ -144,6 +143,29 @@ robj *hashTypeGetObject(robj *o, robj *field) {
     return value;
 }
 
+/* Higher level function using hashTypeGet*() to return the length of the
+ * object associated with the requested field, or 0 if the field does not
+ * exist. */
+size_t hashTypeGetValueLength(robj *o, robj *field) {
+    size_t len = 0;
+    if (o->encoding == REDIS_ENCODING_ZIPLIST) {
+        unsigned char *vstr = NULL;
+        unsigned int vlen = UINT_MAX;
+        long long vll = LLONG_MAX;
+
+        if (hashTypeGetFromZiplist(o, field, &vstr, &vlen, &vll) == 0)
+            len = vstr ? vlen : sdigits10(vll);
+    } else if (o->encoding == REDIS_ENCODING_HT) {
+        robj *aux;
+
+        if (hashTypeGetFromHashTable(o, field, &aux) == 0)
+            len = stringObjectLen(aux);
+    } else {
+        redisPanic("Unknown hash encoding");
+    }
+    return len;
+}
+
 /* Test if the specified field exists in the given hash. Returns 1 if the field
  * exists, and 0 when it doesn't. */
 int hashTypeExists(robj *o, robj *field) {
@@ -679,12 +701,21 @@ void hdelCommand(redisClient *c) {
 
 void hlenCommand(redisClient *c) {
     robj *o;
+
     if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL ||
         checkType(c,o,REDIS_HASH)) return;
 
     addReplyLongLong(c,hashTypeLength(o));
 }
 
+void hstrlenCommand(redisClient *c) {
+    robj *o;
+
+    if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL ||
+        checkType(c,o,REDIS_HASH)) return;
+    addReplyLongLong(c,hashTypeGetValueLength(o,c->argv[2]));
+}
+
 static void addHashIteratorCursorToReply(redisClient *c, hashTypeIterator *hi, int what) {
     if (hi->encoding == REDIS_ENCODING_ZIPLIST) {
         unsigned char *vstr = NULL;

src/util.c

@@ -251,6 +251,18 @@ uint32_t digits10(uint64_t v) {
     return 12 + digits10(v / 1000000000000UL);
 }
 
+/* Like digits10() but for signed values. */
+uint32_t sdigits10(int64_t v) {
+    if (v < 0) {
+        /* Abs value of LLONG_MIN requires special handling. */
+        uint64_t uv = (v != LLONG_MIN) ?
+                      (uint64_t)-v : ((uint64_t) LLONG_MAX)+1;
+        return digits10(uv)+1; /* +1 for the minus. */
+    } else {
+        return digits10(v);
+    }
+}
+
 /* Convert a long long into a string. Returns the number of
  * characters needed to represent the number.
  * If the buffer is not big enough to store the string, 0 is returned.

src/util.h

@@ -30,11 +30,14 @@
 #ifndef __REDIS_UTIL_H
 #define __REDIS_UTIL_H
 
+#include <stdint.h>
 #include "sds.h"
 
 int stringmatchlen(const char *p, int plen, const char *s, int slen, int nocase);
 int stringmatch(const char *p, const char *s, int nocase);
 long long memtoll(const char *p, int *err);
+uint32_t digits10(uint64_t v);
+uint32_t sdigits10(int64_t v);
 int ll2string(char *s, size_t len, long long value);
 int string2ll(const char *s, size_t slen, long long *value);
 int string2l(const char *s, size_t slen, long *value);

tests/unit/type/hash.tcl

@@ -390,6 +390,51 @@ start_server {tags {"hash"}} {
         lappend rv [string match "ERR*not*float*" $bigerr]
     } {1 1}
 
+    test {HSTRLEN against the small hash} {
+        set err {}
+        foreach k [array names smallhash *] {
+            if {[string length $smallhash($k)] ne [r hstrlen smallhash $k]} {
+                set err "[string length $smallhash($k)] != [r hstrlen smallhash $k]"
+                break
+            }
+        }
+        set _ $err
+    } {}
+
+    test {HSTRLEN against the big hash} {
+        set err {}
+        foreach k [array names bighash *] {
+            if {[string length $bighash($k)] ne [r hstrlen bighash $k]} {
+                set err "[string length $bighash($k)] != [r hstrlen bighash $k]"
+                break
+            }
+        }
+        set _ $err
+    } {}
+
+    test {HSTRLEN against non existing field} {
+        set rv {}
+        lappend rv [r hstrlen smallhash __123123123__]
+        lappend rv [r hstrlen bighash __123123123__]
+        set _ $rv
+    } {0 0}
+
+    test {HSTRLEN corner cases} {
+        set vals {
+            -9223372036854775808 9223372036854775807 9223372036854775808
+            {} 0 -1 x
+        }
+        foreach v $vals {
+            r hmset smallhash field $v
+            r hmset bighash field $v
+            set len1 [string length $v]
+            set len2 [r hstrlen smallhash field]
+            set len3 [r hstrlen bighash field]
+            assert {$len1 == $len2}
+            assert {$len2 == $len3}
+        }
+    }
+
     test {Hash ziplist regression test for large keys} {
         r hset hash kkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkk a
         r hset hash kkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkkk b