SPOP replication tests.

The old version of SPOP with "count" argument used an API call of dict.c
which was actually designed for a different goal, and was not capable of
good distribution. We follow a different three-cases approach optimized
for different ratiion between sets and requested number of elements.

The implementation is simpler and allowed the removal of a large amount
of code.

src/dict.c

@@ -211,6 +211,9 @@ int dictExpand(dict *d, unsigned long size)
     if (dictIsRehashing(d) || d->ht[0].used > size)
         return DICT_ERR;
 
+    /* Rehashing to the same table size is not useful. */
+    if (realsize == d->ht[0].size) return DICT_ERR;
+
     /* Allocate the new hash table and initialize all pointers to NULL */
     n.size = realsize;
     n.sizemask = realsize-1;
@@ -232,27 +235,27 @@ int dictExpand(dict *d, unsigned long size)
 
 /* Performs N steps of incremental rehashing. Returns 1 if there are still
  * keys to move from the old to the new hash table, otherwise 0 is returned.
+ *
  * Note that a rehashing step consists in moving a bucket (that may have more
- * than one key as we use chaining) from the old to the new hash table. */
+ * than one key as we use chaining) from the old to the new hash table, however
+ * since part of the hash table may be composed of empty spaces, it is not
+ * guaranteed that this function will rehash even a single bucket, since it
+ * will visit at max N*10 empty buckets in total, otherwise the amount of
+ * work it does would be unbound and the function may block for a long time. */
 int dictRehash(dict *d, int n) {
+    int empty_visits = n*10; /* Max number of empty buckets to visit. */
     if (!dictIsRehashing(d)) return 0;
 
-    while(n--) {
+    while(n-- && d->ht[0].used != 0) {
         dictEntry *de, *nextde;
 
-        /* Check if we already rehashed the whole table... */
-        if (d->ht[0].used == 0) {
-            zfree(d->ht[0].table);
-            d->ht[0] = d->ht[1];
-            _dictReset(&d->ht[1]);
-            d->rehashidx = -1;
-            return 0;
-        }
-
         /* Note that rehashidx can't overflow as we are sure there are more
          * elements because ht[0].used != 0 */
         assert(d->ht[0].size > (unsigned long)d->rehashidx);
-        while(d->ht[0].table[d->rehashidx] == NULL) d->rehashidx++;
+        while(d->ht[0].table[d->rehashidx] == NULL) {
+            d->rehashidx++;
+            if (--empty_visits == 0) return 1;
+        }
         de = d->ht[0].table[d->rehashidx];
         /* Move all the keys in this bucket from the old to the new hash HT */
         while(de) {
@@ -270,6 +273,17 @@ int dictRehash(dict *d, int n) {
         d->ht[0].table[d->rehashidx] = NULL;
         d->rehashidx++;
     }
+
+    /* Check if we already rehashed the whole table... */
+    if (d->ht[0].used == 0) {
+        zfree(d->ht[0].table);
+        d->ht[0] = d->ht[1];
+        _dictReset(&d->ht[1]);
+        d->rehashidx = -1;
+        return 0;
+    }
+
+    /* More to rehash... */
     return 1;
 }
 
@@ -619,7 +633,11 @@ dictEntry *dictGetRandomKey(dict *d)
     if (dictIsRehashing(d)) _dictRehashStep(d);
     if (dictIsRehashing(d)) {
         do {
-            h = random() % (d->ht[0].size+d->ht[1].size);
+            /* We are sure there are no elements in indexes from 0
+             * to rehashidx-1 */
+            h = d->rehashidx + (random() % (d->ht[0].size +
+                                            d->ht[1].size -
+                                            d->rehashidx));
             he = (h >= d->ht[0].size) ? d->ht[1].table[h - d->ht[0].size] :
                                       d->ht[0].table[h];
         } while(he == NULL);
@@ -646,9 +664,12 @@ dictEntry *dictGetRandomKey(dict *d)
     return he;
 }
 
-/* This is a version of dictGetRandomKey() that is modified in order to
- * return multiple entries by jumping at a random place of the hash table
- * and scanning linearly for entries.
+/* This function samples the dictionary to return a few keys from random
+ * locations.
+ *
+ * It does not guarantee to return all the keys specified in 'count', nor
+ * it does guarantee to return non-duplicated elements, however it will make
+ * some effort to do both things.
  *
  * Returned pointers to hash table entries are stored into 'des' that
  * points to an array of dictEntry pointers. The array must have room for
@@ -657,28 +678,65 @@ dictEntry *dictGetRandomKey(dict *d)
  *
  * The function returns the number of items stored into 'des', that may
  * be less than 'count' if the hash table has less than 'count' elements
- * inside.
+ * inside, or if not enough elements were found in a reasonable amount of
+ * steps.
  *
  * Note that this function is not suitable when you need a good distribution
  * of the returned items, but only when you need to "sample" a given number
  * of continuous elements to run some kind of algorithm or to produce
  * statistics. However the function is much faster than dictGetRandomKey()
- * at producing N elements, and the elements are guaranteed to be non
- * repeating. */
-unsigned int dictGetRandomKeys(dict *d, dictEntry **des, unsigned int count) {
-    int j; /* internal hash table id, 0 or 1. */
-    unsigned int stored = 0;
+ * at producing N elements. */
+unsigned int dictGetSomeKeys(dict *d, dictEntry **des, unsigned int count) {
+    unsigned int j; /* internal hash table id, 0 or 1. */
+    unsigned int tables; /* 1 or 2 tables? */
+    unsigned int stored = 0, maxsizemask;
+    unsigned int maxsteps;
 
     if (dictSize(d) < count) count = dictSize(d);
-    while(stored < count) {
-        for (j = 0; j < 2; j++) {
-            /* Pick a random point inside the hash table 0 or 1. */
-            unsigned int i = random() & d->ht[j].sizemask;
-            int size = d->ht[j].size;
+    maxsteps = count*10;
 
-            /* Make sure to visit every bucket by iterating 'size' times. */
-            while(size--) {
-                dictEntry *he = d->ht[j].table[i];
+    /* Try to do a rehashing work proportional to 'count'. */
+    for (j = 0; j < count; j++) {
+        if (dictIsRehashing(d))
+            _dictRehashStep(d);
+        else
+            break;
+    }
+
+    tables = dictIsRehashing(d) ? 2 : 1;
+    maxsizemask = d->ht[0].sizemask;
+    if (tables > 1 && maxsizemask < d->ht[1].sizemask)
+        maxsizemask = d->ht[1].sizemask;
+
+    /* Pick a random point inside the larger table. */
+    unsigned int i = random() & maxsizemask;
+    unsigned int emptylen = 0; /* Continuous empty entries so far. */
+    while(stored < count && maxsteps--) {
+        for (j = 0; j < tables; j++) {
+            /* Invariant of the dict.c rehashing: up to the indexes already
+             * visited in ht[0] during the rehashing, there are no populated
+             * buckets, so we can skip ht[0] for indexes between 0 and idx-1. */
+            if (tables == 2 && j == 0 && i < d->rehashidx) {
+                /* Moreover, if we are currently out of range in the second
+                 * table, there will be no elements in both tables up to
+                 * the current rehashing index, so we jump if possible.
+                 * (this happens when going from big to small table). */
+                if (i >= d->ht[1].size) i = d->rehashidx;
+                continue;
+            }
+            if (i >= d->ht[j].size) continue; /* Out of range for this table. */
+            dictEntry *he = d->ht[j].table[i];
+
+            /* Count contiguous empty buckets, and jump to other
+             * locations if they reach 'count' (with a minimum of 5). */
+            if (he == NULL) {
+                emptylen++;
+                if (emptylen >= 5 && emptylen > count) {
+                    i = random() & maxsizemask;
+                    emptylen = 0;
+                }
+            } else {
+                emptylen = 0;
                 while (he) {
                     /* Collect all the elements of the buckets found non
                      * empty while iterating. */
@@ -688,14 +746,11 @@ unsigned int dictGetRandomKeys(dict *d, dictEntry **des, unsigned int count) {
                     stored++;
                     if (stored == count) return stored;
                 }
-                i = (i+1) & d->ht[j].sizemask;
             }
-            /* If there is only one table and we iterated it all, we should
-             * already have 'count' elements. Assert this condition. */
-            assert(dictIsRehashing(d) != 0);
         }
+        i = (i+1) & maxsizemask;
     }
-    return stored; /* Never reached. */
+    return stored;
 }
 
 /* Function to reverse bits. Algorithm from:

src/dict.h

@@ -164,7 +164,7 @@ dictIterator *dictGetSafeIterator(dict *d);
 dictEntry *dictNext(dictIterator *iter);
 void dictReleaseIterator(dictIterator *iter);
 dictEntry *dictGetRandomKey(dict *d);
-unsigned int dictGetRandomKeys(dict *d, dictEntry **des, unsigned int count);
+unsigned int dictGetSomeKeys(dict *d, dictEntry **des, unsigned int count);
 void dictPrintStats(dict *d);
 unsigned int dictGenHashFunction(const void *key, int len);
 unsigned int dictGenCaseHashFunction(const unsigned char *buf, int len);

src/intset.c

@@ -261,90 +261,6 @@ int64_t intsetRandom(intset *is) {
     return _intsetGet(is,rand()%intrev32ifbe(is->length));
 }
 
-/* How many times bigger should the set length be compared to the requested
- * count of members for us to use the Floyd algorithm instead of
- * the Knuth algorithm */
-#define RANDOMMEMBERS_ALGORITHM_SELECTION_RATIO (2)
-
-/* Copies 'count' random members from the set into the 'values' array.
- * 'values' must be an array of int64_t values, of length 'count'.
- * Returns the amount of items returned. If this amount is less than 'count',
- * then the remaining 'values' are left uninitialized. */
-int intsetRandomMembers(intset *is, int64_t* values, int count) {
-
-    /* We don't check that is and values are non-NULL - the caller must
-     * play nice. */
-
-    int length = intsetLen(is);
-
-    if (count > length) {
-        /* Return everything in the set */
-        count = length;
-    }
-
-    /* Choose between the Knuth shuffle algorithm, O(1) space, O(length) time,
-     * and the Floyd algorithm, O(length) space, O(count) time. */
-    if ((RANDOMMEMBERS_ALGORITHM_SELECTION_RATIO * count) > length) {
-
-        /* If the count of members requested is almost the length of the set,
-         * use the Knuth shuffle algorithm, O(1) space, O(length) time. */
-
-        /* First, fill the values array with unique random indexes inside
-         * the set. */
-        int in, im, rn, rm;
-        im = 0;
-        for (in = 0; in < length && im < count; in++) {
-
-            rn = length - in;
-            rm = count - im;
-            if (rand() % rn < rm) {
-                values[im++] = in;
-            }
-        }
-
-    } else {
-
-        /* If the length is considerably more than the count of members
-         * requested, use Robert Floyd's algorithm, O(length) space,
-         * O(count) time.
-         * Based on Jon Bentley's Programming Pearls */
-
-        int64_t *is_used = zcalloc(sizeof(int64_t) * length);
-        int in, im, r;
-
-        r = 0;
-        im = 0;
-
-        for (in = length - count; in < length && im < count; in++) {
-
-            /* Generate a random number r */
-            r = rand() % (in + 1);
-
-            /* Do we already have the value in r? */
-            if (is_used[r]) {
-                /* Use in instead of the generated number */
-                r = in;
-            }
-
-            values[im++] = r ;
-
-            /* Mark it as used */
-            is_used[r] = 1;
-        }
-
-        zfree(is_used);
-    }
-
-    /* Replace each random index with the value stored there in the intset */
-    uint8_t encoding = intrev32ifbe(is->encoding);
-    for (int currentValue = 0; currentValue < count; currentValue++) {
-        values[currentValue] =
-            _intsetGetEncoded(is, values[currentValue], encoding);
-    }
-
-    return count;
-}
-
 /* Sets the value to the value at the given position. When this position is
  * out of range the function returns 0, when in range it returns 1. */
 uint8_t intsetGet(intset *is, uint32_t pos, int64_t *value) {

src/intset.h

@@ -43,7 +43,6 @@ intset *intsetAdd(intset *is, int64_t value, uint8_t *success);
 intset *intsetRemove(intset *is, int64_t value, int *success);
 uint8_t intsetFind(intset *is, int64_t value);
 int64_t intsetRandom(intset *is);
-int intsetRandomMembers(intset *is, int64_t* value, int count);
 uint8_t intsetGet(intset *is, uint32_t pos, int64_t *value);
 uint32_t intsetLen(intset *is);
 size_t intsetBlobLen(intset *is);

src/latency.c

@@ -228,6 +228,7 @@ sds createLatencyReport(void) {
     int advise_write_load_info = 0; /* Print info about AOF and write load. */
     int advise_hz = 0;              /* Use higher HZ. */
     int advise_large_objects = 0;   /* Deletion of large objects. */
+    int advise_mass_eviction = 0;   /* Avoid mass eviction of keys. */
     int advise_relax_fsync_policy = 0; /* appendfsync always is slow. */
     int advise_disable_thp = 0;     /* AnonHugePages detected. */
     int advices = 0;
@@ -364,11 +365,16 @@ sds createLatencyReport(void) {
         }
 
         /* Eviction cycle. */
-        if (!strcasecmp(event,"eviction-cycle")) {
+        if (!strcasecmp(event,"eviction-del")) {
             advise_large_objects = 1;
             advices++;
         }
 
+        if (!strcasecmp(event,"eviction-cycle")) {
+            advise_mass_eviction = 1;
+            advices++;
+        }
+
         report = sdscatlen(report,"\n",1);
     }
     dictReleaseIterator(di);
@@ -452,6 +458,10 @@ sds createLatencyReport(void) {
             report = sdscat(report,"- Deleting, expiring or evicting (because of maxmemory policy) large objects is a blocking operation. If you have very large objects that are often deleted, expired, or evicted, try to fragment those objects into multiple smaller objects.\n");
         }
 
+        if (advise_mass_eviction) {
+            report = sdscat(report,"- Sudden changes to the 'maxmemory' setting via 'CONFIG SET', or allocation of large objects via sets or sorted sets intersections, STORE option of SORT, Redis Cluster large keys migrations (RESTORE command), may create sudden memory pressure forcing the server to block trying to evict keys. \n");
+        }
+
         if (advise_disable_thp) {
             report = sdscat(report,"- I detected a non zero amount of anonymous huge pages used by your process. This creates very serious latency events in different conditions, especially when Redis is persisting on disk. To disable THP support use the command 'echo never > /sys/kernel/mm/transparent_hugepage/enabled', make sure to also add it into /etc/rc.local so that the command will be executed again after a reboot. Note that even if you have already disabled THP, you still need to restart the Redis process to get rid of the huge pages already created.\n");
         }

src/latency.h

@@ -86,4 +86,8 @@ int THPIsEnabled(void);
         (var) >= server.latency_monitor_threshold) \
           latencyAddSample((event),(var));
 
+/* Remove time from a nested event. */
+#define latencyRemoveNestedEvent(event_var,nested_var) \
+    event_var += nested_var;
+
 #endif /* __LATENCY_H */

src/redis-cli.c

@@ -1935,6 +1935,7 @@ static void statMode(void) {
         /* Children */
         aux = getLongInfoField(reply->str,"bgsave_in_progress");
         aux |= getLongInfoField(reply->str,"aof_rewrite_in_progress") << 1;
+        aux |= getLongInfoField(reply->str,"loading") << 2;
         switch(aux) {
         case 0: break;
         case 1:
@@ -1946,6 +1947,9 @@ static void statMode(void) {
         case 3:
             printf("SAVE+AOF");
             break;
+        case 4:
+            printf("LOAD");
+            break;
         }
 
         printf("\n");

src/redis.c

@@ -1529,6 +1529,7 @@ void initServerConfig(void) {
     server.lpushCommand = lookupCommandByCString("lpush");
     server.lpopCommand = lookupCommandByCString("lpop");
     server.rpopCommand = lookupCommandByCString("rpop");
+    server.sremCommand = lookupCommandByCString("srem");
 
     /* Slow log */
     server.slowlog_log_slower_than = REDIS_SLOWLOG_LOG_SLOWER_THAN;
@@ -2001,6 +2002,9 @@ struct redisCommand *lookupCommandOrOriginal(sds name) {
  * + REDIS_PROPAGATE_NONE (no propagation of command at all)
  * + REDIS_PROPAGATE_AOF (propagate into the AOF file if is enabled)
  * + REDIS_PROPAGATE_REPL (propagate into the replication link)
+ *
+ * This should not be used inside commands implementation. Use instead
+ * alsoPropagate(), preventCommandPropagation(), forceCommandPropagation().
  */
 void propagate(struct redisCommand *cmd, int dbid, robj **argv, int argc,
                int flags)
@@ -2012,11 +2016,31 @@ void propagate(struct redisCommand *cmd, int dbid, robj **argv, int argc,
 }
 
 /* Used inside commands to schedule the propagation of additional commands
- * after the current command is propagated to AOF / Replication. */
+ * after the current command is propagated to AOF / Replication.
+ *
+ * 'cmd' must be a pointer to the Redis command to replicate, dbid is the
+ * database ID the command should be propagated into.
+ * Arguments of the command to propagte are passed as an array of redis
+ * objects pointers of len 'argc', using the 'argv' vector.
+ *
+ * The function does not take a reference to the passed 'argv' vector,
+ * so it is up to the caller to release the passed argv (but it is usually
+ * stack allocated).  The function autoamtically increments ref count of
+ * passed objects, so the caller does not need to. */
 void alsoPropagate(struct redisCommand *cmd, int dbid, robj **argv, int argc,
                    int target)
 {
-    redisOpArrayAppend(&server.also_propagate,cmd,dbid,argv,argc,target);
+    robj **argvcopy;
+    int j;
+
+    if (server.loading) return; /* No propagation during loading. */
+
+    argvcopy = zmalloc(sizeof(robj*)*argc);
+    for (j = 0; j < argc; j++) {
+        argvcopy[j] = argv[j];
+        incrRefCount(argv[j]);
+    }
+    redisOpArrayAppend(&server.also_propagate,cmd,dbid,argvcopy,argc,target);
 }
 
 /* It is possible to call the function forceCommandPropagation() inside a
@@ -2027,6 +2051,13 @@ void forceCommandPropagation(redisClient *c, int flags) {
     if (flags & REDIS_PROPAGATE_AOF) c->flags |= REDIS_FORCE_AOF;
 }
 
+/* Avoid that the executed command is propagated at all. This way we
+ * are free to just propagate what we want using the alsoPropagate()
+ * API. */
+void preventCommandPropagation(redisClient *c) {
+    c->flags |= REDIS_PREVENT_PROP;
+}
+
 /* Call() is the core of Redis execution of a command */
 void call(redisClient *c, int flags) {
     long long dirty, start, duration;
@@ -2080,7 +2111,7 @@ void call(redisClient *c, int flags) {
     }
 
     /* Propagate the command into the AOF and replication link */
-    if (flags & REDIS_CALL_PROPAGATE) {
+    if (flags & REDIS_CALL_PROPAGATE && (c->flags & REDIS_PREVENT_PROP) == 0) {
         int flags = REDIS_PROPAGATE_NONE;
 
         if (c->flags & REDIS_FORCE_REPL) flags |= REDIS_PROPAGATE_REPL;
@@ -2091,20 +2122,24 @@ void call(redisClient *c, int flags) {
             propagate(c->cmd,c->db->id,c->argv,c->argc,flags);
     }
 
-    /* Restore the old FORCE_AOF/REPL flags, since call can be executed
+    /* Restore the old replication flags, since call can be executed
      * recursively. */
-    c->flags &= ~(REDIS_FORCE_AOF|REDIS_FORCE_REPL);
-    c->flags |= client_old_flags & (REDIS_FORCE_AOF|REDIS_FORCE_REPL);
+    c->flags &= ~(REDIS_FORCE_AOF|REDIS_FORCE_REPL|REDIS_PREVENT_PROP);
+    c->flags |= client_old_flags &
+        (REDIS_FORCE_AOF|REDIS_FORCE_REPL|REDIS_PREVENT_PROP);
 
     /* Handle the alsoPropagate() API to handle commands that want to propagate
-     * multiple separated commands. */
+     * multiple separated commands. Note that alsoPropagate() is not affected
+     * by REDIS_PREVENT_PROP flag. */
     if (server.also_propagate.numops) {
         int j;
         redisOp *rop;
 
-        for (j = 0; j < server.also_propagate.numops; j++) {
-            rop = &server.also_propagate.ops[j];
-            propagate(rop->cmd, rop->dbid, rop->argv, rop->argc, rop->target);
+        if (flags & REDIS_CALL_PROPAGATE) {
+            for (j = 0; j < server.also_propagate.numops; j++) {
+                rop = &server.also_propagate.ops[j];
+                propagate(rop->cmd,rop->dbid,rop->argv,rop->argc,rop->target);
+            }
         }
         redisOpArrayFree(&server.also_propagate);
     }
@@ -3147,13 +3182,7 @@ void evictionPoolPopulate(dict *sampledict, dict *keydict, struct evictionPoolEn
         samples = zmalloc(sizeof(samples[0])*server.maxmemory_samples);
     }
 
-#if 1 /* Use bulk get by default. */
-    count = dictGetRandomKeys(sampledict,samples,server.maxmemory_samples);
-#else
-    count = server.maxmemory_samples;
-    for (j = 0; j < count; j++) samples[j] = dictGetRandomKey(sampledict);
-#endif
-
+    count = dictGetSomeKeys(sampledict,samples,server.maxmemory_samples);
     for (j = 0; j < count; j++) {
         unsigned long long idle;
         sds key;
@@ -3208,7 +3237,7 @@ void evictionPoolPopulate(dict *sampledict, dict *keydict, struct evictionPoolEn
 int freeMemoryIfNeeded(void) {
     size_t mem_used, mem_tofree, mem_freed;
     int slaves = listLength(server.slaves);
-    mstime_t latency;
+    mstime_t latency, eviction_latency;
 
     /* Remove the size of slaves output buffers and AOF buffer from the
      * count of used memory. */
@@ -3339,7 +3368,11 @@ int freeMemoryIfNeeded(void) {
                  * AOF and Output buffer memory will be freed eventually so
                  * we only care about memory used by the key space. */
                 delta = (long long) zmalloc_used_memory();
+                latencyStartMonitor(eviction_latency);
                 dbDelete(db,keyobj);
+                latencyEndMonitor(eviction_latency);
+                latencyAddSampleIfNeeded("eviction-del",eviction_latency);
+                latencyRemoveNestedEvent(latency,eviction_latency);
                 delta -= (long long) zmalloc_used_memory();
                 mem_freed += delta;
                 server.stat_evictedkeys++;

src/redis.h

@@ -257,6 +257,7 @@ typedef long long mstime_t; /* millisecond time type. */
 #define REDIS_PRE_PSYNC (1<<16)   /* Instance don't understand PSYNC. */
 #define REDIS_READONLY (1<<17)    /* Cluster client is in read-only state. */
 #define REDIS_PUBSUB (1<<18)      /* Client is in Pub/Sub mode. */
+#define REDIS_PREVENT_PROP (1<<19)  /* Don't propagate to AOF / Slaves. */
 
 /* Client block type (btype field in client structure)
  * if REDIS_BLOCKED flag is set. */
@@ -708,7 +709,7 @@ struct redisServer {
     off_t loading_process_events_interval_bytes;
     /* Fast pointers to often looked up command */
     struct redisCommand *delCommand, *multiCommand, *lpushCommand, *lpopCommand,
-                        *rpopCommand;
+                        *rpopCommand, *sremCommand;
     /* Fields used only for stats */
     time_t stat_starttime;          /* Server start time */
     long long stat_numcommands;     /* Number of processed commands */
@@ -1252,6 +1253,7 @@ void call(redisClient *c, int flags);
 void propagate(struct redisCommand *cmd, int dbid, robj **argv, int argc, int flags);
 void alsoPropagate(struct redisCommand *cmd, int dbid, robj **argv, int argc, int target);
 void forceCommandPropagation(redisClient *c, int flags);
+void preventCommandPropagation(redisClient *c);
 int prepareForShutdown();
 #ifdef __GNUC__
 void redisLog(int level, const char *fmt, ...)

src/t_set.c

@@ -45,6 +45,11 @@ robj *setTypeCreate(robj *value) {
     return createSetObject();
 }
 
+/* Add the specified value into a set. The function takes care of incrementing
+ * the reference count of the object if needed in order to retain a copy.
+ *
+ * If the value was already member of the set, nothing is done and 0 is
+ * returned, otherwise the new element is added and 1 is returned. */
 int setTypeAdd(robj *subject, robj *value) {
     long long llval;
     if (subject->encoding == REDIS_ENCODING_HT) {
@@ -207,106 +212,6 @@ int setTypeRandomElement(robj *setobj, robj **objele, int64_t *llele) {
     return setobj->encoding;
 }
 
-/* Return a number of random elements from a non empty set.
- *
- * This is a version of setTypeRandomElement() that is modified in order to
- * return multiple entries, using dictGetRandomKeys() and intsetRandomMembers().
- *
- * The elements are stored into 'aux_set' which should be of a set type.
- *
- * The function returns the number of items stored into 'aux_set', that may
- * be less than 'count' if the hash table has less than 'count' elements
- * inside.
- *
- * Note that this function is not suitable when you need a good distribution
- * of the returned items, but only when you need to "sample" a given number
- * of continuous elements to run some kind of algorithm or to produce
- * statistics. However the function is much faster than setTypeRandomElement()
- * at producing N elements, and the elements are guaranteed to be non
- * repeating.
- */
-unsigned long setTypeRandomElements(robj *set, unsigned long count,
-                                    robj *aux_set) {
-    unsigned long set_size;
-    unsigned long elements_to_return = count;
-    unsigned long elements_copied = 0;
-    unsigned long current_element = 0;
-
-    /* Like all setType* functions, we assume good behavior on part of the
-     * caller, so no extra parameter checks are made. */
-
-    /* If the number of elements in the the set is less than the count
-     * requested, just return all of them. */
-    set_size = setTypeSize(set);
-    if (set_size < count) {
-        elements_to_return = set_size;
-    }
-
-    /* TODO: It is definitely faster adding items to the set by directly
-     * handling the Dict or intset inside it, avoiding the constant encoding
-     * checks inside setTypeAdd(). However, We don't want to touch the set
-     * internals in non setType* functions. So, we just call setTypeAdd()
-     * multiple times, but this isn't an optimal solution.
-     * Another option would be to create a bulk-add function:
-     * setTypeAddBulk(). */
-    if (set->encoding == REDIS_ENCODING_HT) {
-        /* Allocate result array */
-        dictEntry **random_elements =
-            zmalloc(sizeof(dictEntry*) * elements_to_return);
-
-        /* Get the random elements */
-        elements_copied =
-            dictGetRandomKeys(set->ptr, random_elements, elements_to_return);
-        redisAssert(elements_copied == elements_to_return);
-
-        /* Put them into the set */
-        for (current_element = 0; current_element < elements_copied;
-             current_element++) {
-
-            /* We get the key and duplicate it, as we know it is a string */
-            setTypeAdd(aux_set,
-                dictGetKey(random_elements[current_element]));
-        }
-
-        zfree(random_elements);
-
-    } else if (set->encoding == REDIS_ENCODING_INTSET) {
-        /* Allocate result array */
-        int64_t *random_elements =
-            zmalloc(sizeof(int64_t) * elements_to_return);
-        robj* element_as_str = NULL;
-
-        elements_copied =
-            intsetRandomMembers((intset*) set->ptr,
-                                random_elements,
-                                elements_to_return);
-
-        redisAssert(elements_copied == elements_to_return);
-
-        /* Put them into the set */
-        for (current_element = 0; current_element < elements_copied;
-             current_element++) {
-
-            element_as_str = createStringObjectFromLongLong(
-                    random_elements[current_element]);
-
-            /* Put the values in the set */
-            setTypeAdd(aux_set,
-                       element_as_str);
-
-            decrRefCount(element_as_str);
-        }
-
-        zfree(random_elements);
-    } else {
-        redisPanic("Unknown set encoding");
-    }
-
-    /* We have a set with random elements. Return the actual elements in
-       the aux_set. */
-    return elements_copied;
-}
-
 unsigned long setTypeSize(robj *subject) {
     if (subject->encoding == REDIS_ENCODING_HT) {
         return dictSize((dict*)subject->ptr);
@@ -480,15 +385,18 @@ void scardCommand(redisClient *c) {
     addReplyLongLong(c,setTypeSize(o));
 }
 
-/* handle the "SPOP key <count>" variant. The normal version of the
+/* Handle the "SPOP key <count>" variant. The normal version of the
  * command is handled by the spopCommand() function itself. */
 
+/* How many times bigger should be the set compared to the remaining size
+ * for us to use the "create new set" strategy? Read later in the
+ * implementation for more info. */
+#define SPOP_MOVE_STRATEGY_MUL 5
+
 void spopWithCountCommand(redisClient *c) {
     long l;
     unsigned long count, size;
-    unsigned long elements_returned;
-    robj *set, *aux, *aux_set;
-    int64_t llele;
+    robj *set;
 
     /* Get the count argument */
     if (getLongFromObjectOrReply(c,c->argv[2],&l,NULL) != REDIS_OK) return;
@@ -511,18 +419,16 @@ void spopWithCountCommand(redisClient *c) {
         return;
     }
 
-    /* Get the size of the set. It is always > 0, as empty sets get
-     * deleted.  */
     size = setTypeSize(set);
 
     /* Generate an SPOP keyspace notification */
     notifyKeyspaceEvent(REDIS_NOTIFY_SET,"spop",c->argv[1],c->db->id);
+    server.dirty += count;
 
     /* CASE 1:
      * The number of requested elements is greater than or equal to
      * the number of elements inside the set: simply return the whole set. */
     if (count >= size) {
-
         /* We just return the entire set */
         sunionDiffGenericCommand(c,c->argv+1,1,NULL,REDIS_OP_UNION);
 
@@ -530,73 +436,110 @@ void spopWithCountCommand(redisClient *c) {
         dbDelete(c->db,c->argv[1]);
         notifyKeyspaceEvent(REDIS_NOTIFY_GENERIC,"del",c->argv[1],c->db->id);
 
-        /* Replicate/AOF this command as an SREM operation */
-        aux = createStringObject("DEL",3);
-        rewriteClientCommandVector(c,2,aux,c->argv[1]);
-        decrRefCount(aux);
-
+        /* Propagate this command as an DEL operation */
+        rewriteClientCommandVector(c,2,shared.del,c->argv[1]);
+        signalModifiedKey(c->db,c->argv[1]);
+        server.dirty++;
         return;
     }
 
-    /* CASE 2:
-     * The number of requested elements is less than the number
-     * of elements inside the set. */
+    /* Case 2 and 3 require to replicate SPOP as a set of SERM commands.
+     * Prepare our replication argument vector. Also send the array length
+     * which is common to both the code paths. */
+    robj *propargv[3];
+    propargv[0] = createStringObject("SREM",4);
+    propargv[1] = c->argv[1];
+    addReplyMultiBulkLen(c,count);
 
-    /* We need an auxiliary set. Optimistically, we create a set using an
-     * Intset internally. */
-    aux = createStringObjectFromLongLong(0);
-    aux_set = setTypeCreate(aux);
-    decrRefCount(aux);
+    /* Common iteration vars. */
+    robj *objele;
+    int encoding;
+    int64_t llele;
+    unsigned long remaining = size-count; /* Elements left after SPOP. */
 
-    /* Get the count requested of random elements from the set into our
-     * auxiliary set. */
-    elements_returned = setTypeRandomElements(set, count, aux_set);
-    redisAssert(elements_returned == count);
-
-    {
-        setTypeIterator *si;
-        robj *objele;
-        int element_encoding;
-
-        addReplyMultiBulkLen(c, elements_returned);
-
-        /* Replicate/AOF this command as an SREM operation */
-        aux = createStringObject("SREM",4);
-
-        si = setTypeInitIterator(aux_set);
-        while ((element_encoding = setTypeNext(si, &objele, &llele)) != -1) {
-            if (element_encoding == REDIS_ENCODING_HT) {
-
-                addReplyBulk(c, objele);
-
-                /* Replicate/AOF this command as an SREM commands */
-                rewriteClientCommandVector(c, 3, aux, c->argv[1], objele);
-                setTypeRemove(set, objele);
-            }
-            else if (element_encoding == REDIS_ENCODING_INTSET) {
-                /* TODO: setTypeRemove() forces us to convert all of the ints
-                 *       to string... isn't there a nicer way to do this? */
+    /* If we are here, the number of requested elements is less than the
+     * number of elements inside the set. Also we are sure that count < size.
+     * Use two different strategies.
+     *
+     * CASE 2: The number of elements to return is small compared to the
+     * set size. We can just extract random elements and return them to
+     * the set. */
+    if (remaining*SPOP_MOVE_STRATEGY_MUL > count) {
+        while(count--) {
+            encoding = setTypeRandomElement(set,&objele,&llele);
+            if (encoding == REDIS_ENCODING_INTSET) {
                 objele = createStringObjectFromLongLong(llele);
-                addReplyBulk(c, objele);
-
-                /* Replicate/AOF this command as an SREM commands */
-                rewriteClientCommandVector(c, 3, aux, c->argv[1], objele);
-                setTypeRemove(set, objele);
-
-                /* We created it, we kill it. */
-                decrRefCount(objele);
+            } else {
+                incrRefCount(objele);
             }
-            else {
-                redisPanic("Unknown set encoding");
+
+            /* Return the element to the client and remove from the set. */
+            addReplyBulk(c,objele);
+            setTypeRemove(set,objele);
+
+            /* Replicate/AOF this command as an SREM operation */
+            propargv[2] = objele;
+            alsoPropagate(server.sremCommand,c->db->id,propargv,3,
+                REDIS_PROPAGATE_AOF|REDIS_PROPAGATE_REPL);
+            decrRefCount(objele);
+        }
+    } else {
+    /* CASE 3: The number of elements to return is very big, approaching
+     * the size of the set itself. After some time extracting random elements
+     * from such a set becomes computationally expensive, so we use
+     * a different strategy, we extract random elements that we don't
+     * want to return (the elements that will remain part of the set),
+     * creating a new set as we do this (that will be stored as the original
+     * set). Then we return the elements left in the original set and
+     * release it. */
+        robj *newset = NULL;
+
+        /* Create a new set with just the remaining elements. */
+        while(remaining--) {
+            encoding = setTypeRandomElement(set,&objele,&llele);
+            if (encoding == REDIS_ENCODING_INTSET) {
+                objele = createStringObjectFromLongLong(llele);
+            } else {
+                incrRefCount(objele);
             }
+            if (!newset) newset = setTypeCreate(objele);
+            setTypeAdd(newset,objele);
+            setTypeRemove(set,objele);
+            decrRefCount(objele);
+        }
+
+        /* Assign the new set as the key value. */
+        incrRefCount(set); /* Protect the old set value. */
+        dbOverwrite(c->db,c->argv[1],newset);
+
+        /* Tranfer the old set to the client and release it. */
+        setTypeIterator *si;
+        si = setTypeInitIterator(set);
+        while((encoding = setTypeNext(si,&objele,&llele)) != -1) {
+            if (encoding == REDIS_ENCODING_INTSET) {
+                objele = createStringObjectFromLongLong(llele);
+            } else {
+                incrRefCount(objele);
+            }
+            addReplyBulk(c,objele);
+
+            /* Replicate/AOF this command as an SREM operation */
+            propargv[2] = objele;
+            alsoPropagate(server.sremCommand,c->db->id,propargv,3,
+                REDIS_PROPAGATE_AOF|REDIS_PROPAGATE_REPL);
+
+            decrRefCount(objele);
         }
         setTypeReleaseIterator(si);
-
-        decrRefCount(aux);
+        decrRefCount(set);
     }
 
-    /* Free the auxiliary set - we need it no more. */
-    decrRefCount(aux_set);
+    /* Don't propagate the command itself even if we incremented the
+     * dirty counter. We don't want to propagate an SPOP command since
+     * we propagated the command as a set of SREMs operations using
+     * the alsoPropagate() API. */
+    decrRefCount(propargv[0]);
+    preventCommandPropagation(c);
 }
 
 void spopCommand(redisClient *c) {

tests/integration/aof.tcl

@@ -204,7 +204,7 @@ tags {"aof"} {
         }
     }
 
-    ## Test that SPOP with <count> (that modifies the client's argc/argv) is correctly free'd
+    ## Uses the alsoPropagate() API.
     create_aof {
         append_to_aof [formatCommand sadd set foo]
         append_to_aof [formatCommand sadd set bar]

tests/integration/replication-4.tcl

@@ -132,5 +132,24 @@ start_server {tags {"repl"}} {
             }
             assert {[$master dbsize] > 0}
         }
+
+        test {Replication of SPOP command -- alsoPropagate() API} {
+            $master del myset
+            set size [randomInt 100]
+            set content {}
+            for {set j 0} {$j < $size} {incr j} {
+                lappend content [randomValue]
+            }
+            $master sadd myset {*}$content
+
+            set count [randomInt 100]
+            set result [$master spop myset $count]
+
+            wait_for_condition 50 100 {
+                [$master debug digest] eq [$slave debug digest]
+            } else {
+                fail "SPOP replication inconsistency"
+            }
+        }
     }
 }

tests/unit/type/set.tcl

@@ -346,6 +346,33 @@ start_server {
         r spop nonexisting_key 100
     } {}
 
+    test "SPOP new implementation: code path #1" {
+        set content {1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20}
+        create_set myset $content
+        set res [r spop myset 30]
+        assert {[lsort $content] eq [lsort $res]}
+    }
+
+    test "SPOP new implementation: code path #2" {
+        set content {1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20}
+        create_set myset $content
+        set res [r spop myset 2]
+        assert {[llength $res] == 2}
+        assert {[r scard myset] == 18}
+        set union [concat [r smembers myset] $res]
+        assert {[lsort $union] eq [lsort $content]}
+    }
+
+    test "SPOP new implementation: code path #3" {
+        set content {1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20}
+        create_set myset $content
+        set res [r spop myset 18]
+        assert {[llength $res] == 18}
+        assert {[r scard myset] == 2}
+        set union [concat [r smembers myset] $res]
+        assert {[lsort $union] eq [lsort $content]}
+    }
+
     test "SRANDMEMBER with <count> against non existing key" {
         r srandmember nonexisting_key 100
     } {}

utils/hashtable/README

@@ -0,0 +1,13 @@
+Hash table implementation related utilities.
+
+rehashing.c
+---
+
+Visually show buckets in the two hash tables between rehashings. Also stress
+test getRandomKeys() implementation, that may actually disappear from
+Redis soon, however visualizaiton some code is reusable in new bugs
+investigation.
+
+Compile with:
+
+    cc -I ../../src/ rehashing.c ../../src/zmalloc.c ../../src/dict.c -o rehashing_test

utils/hashtable/rehashing.c

@@ -0,0 +1,132 @@
+#include "redis.h"
+#include "dict.h"
+
+void _redisAssert(char *x, char *y, int l) {
+    printf("ASSERT: %s %s %d\n",x,y,l);
+    exit(1);
+}
+
+unsigned int dictKeyHash(const void *keyp) {
+    unsigned long key = (unsigned long)keyp;
+    key = dictGenHashFunction(&key,sizeof(key));
+    key += ~(key << 15);
+    key ^=  (key >> 10);
+    key +=  (key << 3);
+    key ^=  (key >> 6);
+    key += ~(key << 11);
+    key ^=  (key >> 16);
+    return key;
+}
+
+int dictKeyCompare(void *privdata, const void *key1, const void *key2) {
+    unsigned long k1 = (unsigned long)key1;
+    unsigned long k2 = (unsigned long)key2;
+    return k1 == k2;
+}
+
+dictType dictTypeTest = {
+    dictKeyHash,                   /* hash function */
+    NULL,                          /* key dup */
+    NULL,                          /* val dup */
+    dictKeyCompare,                /* key compare */
+    NULL,                          /* key destructor */
+    NULL                           /* val destructor */
+};
+
+void showBuckets(dictht ht) {
+    if (ht.table == NULL) {
+        printf("NULL\n");
+    } else {
+        int j;
+        for (j = 0; j < ht.size; j++) {
+            printf("%c", ht.table[j] ? '1' : '0');
+        }
+        printf("\n");
+    }
+}
+
+void show(dict *d) {
+    int j;
+    if (d->rehashidx != -1) {
+        printf("rhidx: ");
+        for (j = 0; j < d->rehashidx; j++)
+            printf(".");
+        printf("|\n");
+    }
+    printf("ht[0]: ");
+    showBuckets(d->ht[0]);
+    printf("ht[1]: ");
+    showBuckets(d->ht[1]);
+    printf("\n");
+}
+
+int sortPointers(const void *a, const void *b) {
+    unsigned long la, lb;
+
+    la = (long) (*((dictEntry**)a));
+    lb = (long) (*((dictEntry**)b));
+    return la-lb;
+}
+
+void stressGetKeys(dict *d, int times) {
+    int j;
+    dictEntry **des = zmalloc(sizeof(dictEntry*)*dictSize(d));
+    for (j = 0; j < times; j++) {
+        int requested = rand() % (dictSize(d)+1);
+        int returned = dictGetRandomKeys(d, des, requested);
+        if (requested != returned) {
+            printf("*** ERROR! Req: %d, Ret: %d\n", requested, returned);
+            exit(1);
+        }
+        qsort(des,returned,sizeof(dictEntry*),sortPointers);
+        if (returned > 1) {
+            int i;
+            for (i = 0; i < returned-1; i++) {
+                if (des[i] == des[i+1]) {
+                    printf("*** ERROR! Duplicated element detected\n");
+                    exit(1);
+                }
+            }
+        }
+    }
+    zfree(des);
+}
+
+#define MAX1 120
+#define MAX2 1000
+int main(void) {
+    dict *d = dictCreate(&dictTypeTest,NULL);
+    unsigned long i;
+    srand(time(NULL));
+
+    for (i = 0; i < MAX1; i++) {
+        dictAdd(d,(void*)i,NULL);
+        show(d);
+    }
+    printf("Size: %d\n", (int)dictSize(d));
+
+    for (i = 0; i < MAX1; i++) {
+        dictDelete(d,(void*)i);
+        dictResize(d);
+        show(d);
+    }
+    dictRelease(d);
+
+    d = dictCreate(&dictTypeTest,NULL);
+    printf("Getkeys stress test\n");
+
+    for (i = 0; i < MAX2; i++) {
+        dictAdd(d,(void*)i,NULL);
+        stressGetKeys(d,100);
+    }
+
+    for (i = 0; i < MAX2; i++) {
+        dictDelete(d,(void*)i);
+        dictResize(d);
+        stressGetKeys(d,100);
+    }
+    dictRelease(d);
+
+    printf("TEST PASSED!\n");
+    return 0;
+}