redis-trib: when loading node info also fetch the output of CLUSTER INFO.

[redis.git] / src / t_hash.c

diff --git a/src/t_hash.c b/src/t_hash.c
index 21fe9e10970f2419c520f5610b5255f24c18c19b..83ca5b2754c3fce9d9e9bede7e1a1d95e90ba7f2 100644 (file)
--- a/src/t_hash.c
+++ b/src/t_hash.c
@@ -185,24 +185,50 @@ int hashTypeNext(hashTypeIterator *hi) {
 }
 
 /* Get key or value object at current iteration position.
- * This increases the refcount of the field object by 1. */
-robj *hashTypeCurrent(hashTypeIterator *hi, int what) {
-    robj *o;
+ * The returned item differs with the hash object encoding:
+ * - When encoding is REDIS_ENCODING_HT, the objval pointer is populated
+ *   with the original object.
+ * - When encoding is REDIS_ENCODING_ZIPMAP, a pointer to the string and
+ *   its length is retunred populating the v and vlen pointers.
+ * This function is copy on write friendly as accessing objects in read only
+ * does not require writing to any memory page.
+ *
+ * The function returns the encoding of the object, so that the caller
+ * can underestand if the key or value was returned as object or C string. */
+int hashTypeCurrent(hashTypeIterator *hi, int what, robj **objval, unsigned char **v, unsigned int *vlen) {
     if (hi->encoding == REDIS_ENCODING_ZIPMAP) {
         if (what & REDIS_HASH_KEY) {
-            o = createStringObject((char*)hi->zk,hi->zklen);
+            *v = hi->zk;
+            *vlen = hi->zklen;
         } else {
-            o = createStringObject((char*)hi->zv,hi->zvlen);
+            *v = hi->zv;
+            *vlen = hi->zvlen;
         }
     } else {
-        if (what & REDIS_HASH_KEY) {
-            o = dictGetEntryKey(hi->de);
-        } else {
-            o = dictGetEntryVal(hi->de);
-        }
-        incrRefCount(o);
+        if (what & REDIS_HASH_KEY)
+            *objval = dictGetEntryKey(hi->de);
+        else
+            *objval = dictGetEntryVal(hi->de);
+    }
+    return hi->encoding;
+}
+
+/* A non copy-on-write friendly but higher level version of hashTypeCurrent()
+ * that always returns an object with refcount incremented by one (or a new
+ * object), so it's up to the caller to decrRefCount() the object if no
+ * reference is retained. */
+robj *hashTypeCurrentObject(hashTypeIterator *hi, int what) {
+    robj *obj;
+    unsigned char *v = NULL;
+    unsigned int vlen = 0;
+    int encoding = hashTypeCurrent(hi,what,&obj,&v,&vlen);
+
+    if (encoding == REDIS_ENCODING_HT) {
+        incrRefCount(obj);
+        return obj;
+    } else {
+        return createStringObject((char*)v,vlen);
     }
-    return o;
 }
 
 robj *hashTypeLookupWriteOrCreate(redisClient *c, robj *key) {
@@ -253,7 +279,7 @@ void hsetCommand(redisClient *c) {
     hashTypeTryObjectEncoding(o,&c->argv[2], &c->argv[3]);
     update = hashTypeSet(o,c->argv[2],c->argv[3]);
     addReply(c, update ? shared.czero : shared.cone);
-    touchWatchedKey(c->db,c->argv[1]);
+    signalModifiedKey(c->db,c->argv[1]);
     server.dirty++;
 }
 
@@ -268,7 +294,7 @@ void hsetnxCommand(redisClient *c) {
         hashTypeTryObjectEncoding(o,&c->argv[2], &c->argv[3]);
         hashTypeSet(o,c->argv[2],c->argv[3]);
         addReply(c, shared.cone);
-        touchWatchedKey(c->db,c->argv[1]);
+        signalModifiedKey(c->db,c->argv[1]);
         server.dirty++;
     }
 }
@@ -289,7 +315,7 @@ void hmsetCommand(redisClient *c) {
         hashTypeSet(o,c->argv[i],c->argv[i+1]);
     }
     addReply(c, shared.ok);
-    touchWatchedKey(c->db,c->argv[1]);
+    signalModifiedKey(c->db,c->argv[1]);
     server.dirty++;
 }
 
@@ -316,7 +342,7 @@ void hincrbyCommand(redisClient *c) {
     hashTypeSet(o,c->argv[2],new);
     decrRefCount(new);
     addReplyLongLong(c,value);
-    touchWatchedKey(c->db,c->argv[1]);
+    signalModifiedKey(c->db,c->argv[1]);
     server.dirty++;
 }
 
@@ -370,17 +396,25 @@ void hmgetCommand(redisClient *c) {
 
 void hdelCommand(redisClient *c) {
     robj *o;
+    int j, deleted = 0;
+
     if ((o = lookupKeyWriteOrReply(c,c->argv[1],shared.czero)) == NULL ||
         checkType(c,o,REDIS_HASH)) return;
 
-    if (hashTypeDelete(o,c->argv[2])) {
-        if (hashTypeLength(o) == 0) dbDelete(c->db,c->argv[1]);
-        addReply(c,shared.cone);
-        touchWatchedKey(c->db,c->argv[1]);
-        server.dirty++;
-    } else {
-        addReply(c,shared.czero);
+    for (j = 2; j < c->argc; j++) {
+        if (hashTypeDelete(o,c->argv[j])) {
+            deleted++;
+            if (hashTypeLength(o) == 0) {
+                dbDelete(c->db,c->argv[1]);
+                break;
+            }
+        }
     }
+    if (deleted) {
+        signalModifiedKey(c->db,c->argv[1]);
+        server.dirty += deleted;
+    }
+    addReplyLongLong(c,deleted);
 }
 
 void hlenCommand(redisClient *c) {
@@ -392,7 +426,7 @@ void hlenCommand(redisClient *c) {
 }
 
 void genericHgetallCommand(redisClient *c, int flags) {
-    robj *o, *obj;
+    robj *o;
     unsigned long count = 0;
     hashTypeIterator *hi;
     void *replylen = NULL;
@@ -403,16 +437,25 @@ void genericHgetallCommand(redisClient *c, int flags) {
     replylen = addDeferredMultiBulkLength(c);
     hi = hashTypeInitIterator(o);
     while (hashTypeNext(hi) != REDIS_ERR) {
+        robj *obj;
+        unsigned char *v = NULL;
+        unsigned int vlen = 0;
+        int encoding;
+
         if (flags & REDIS_HASH_KEY) {
-            obj = hashTypeCurrent(hi,REDIS_HASH_KEY);
-            addReplyBulk(c,obj);
-            decrRefCount(obj);
+            encoding = hashTypeCurrent(hi,REDIS_HASH_KEY,&obj,&v,&vlen);
+            if (encoding == REDIS_ENCODING_HT)
+                addReplyBulk(c,obj);
+            else
+                addReplyBulkCBuffer(c,v,vlen);
             count++;
         }
         if (flags & REDIS_HASH_VALUE) {
-            obj = hashTypeCurrent(hi,REDIS_HASH_VALUE);
-            addReplyBulk(c,obj);
-            decrRefCount(obj);
+            encoding = hashTypeCurrent(hi,REDIS_HASH_VALUE,&obj,&v,&vlen);
+            if (encoding == REDIS_ENCODING_HT)
+                addReplyBulk(c,obj);
+            else
+                addReplyBulkCBuffer(c,v,vlen);
             count++;
         }
     }