From: Pieter Noordhuis <pcnoordhuis@gmail.com>
Date: Fri, 2 Jul 2010 17:57:12 +0000 (+0200)
Subject: merged intset code into the split files
X-Git-Url: https://git.saurik.com/redis.git/commitdiff_plain/96ffb2fe97c3e77879e7a4f6f7457397a18bf233

merged intset code into the split files
---

96ffb2fe97c3e77879e7a4f6f7457397a18bf233
diff --cc src/Makefile
index a7cc6b9a,00000000..fb343e80
mode 100644,000000..100644
--- a/src/Makefile
+++ b/src/Makefile
@@@ -1,111 -1,0 +1,112 @@@
 +# Redis Makefile
 +# Copyright (C) 2009 Salvatore Sanfilippo <antirez at gmail dot com>
 +# This file is released under the BSD license, see the COPYING file
 +
 +release_hdr := $(shell sh -c './mkreleasehdr.sh')
 +uname_S := $(shell sh -c 'uname -s 2>/dev/null || echo not')
 +OPTIMIZATION?=-O2
 +ifeq ($(uname_S),SunOS)
 +  CFLAGS?= -std=c99 -pedantic $(OPTIMIZATION) -Wall -W -D__EXTENSIONS__ -D_XPG6
 +  CCLINK?= -ldl -lnsl -lsocket -lm -lpthread
 +else
 +  CFLAGS?= -std=c99 -pedantic $(OPTIMIZATION) -Wall -W $(ARCH) $(PROF)
 +  CCLINK?= -lm -pthread
 +endif
 +CCOPT= $(CFLAGS) $(CCLINK) $(ARCH) $(PROF)
 +DEBUG?= -g -rdynamic -ggdb 
 +
- OBJ = adlist.o ae.o anet.o dict.o redis.o sds.o zmalloc.o lzf_c.o lzf_d.o pqsort.o zipmap.o sha1.o ziplist.o release.o networking.o util.o object.o db.o replication.o rdb.o t_string.o t_list.o t_set.o t_zset.o t_hash.o config.o aof.o vm.o pubsub.o multi.o debug.o sort.o
++OBJ = adlist.o ae.o anet.o dict.o redis.o sds.o zmalloc.o lzf_c.o lzf_d.o pqsort.o zipmap.o sha1.o ziplist.o release.o networking.o util.o object.o db.o replication.o rdb.o t_string.o t_list.o t_set.o t_zset.o t_hash.o config.o aof.o vm.o pubsub.o multi.o debug.o sort.o intset.o
 +BENCHOBJ = ae.o anet.o redis-benchmark.o sds.o adlist.o zmalloc.o
 +CLIOBJ = anet.o sds.o adlist.o redis-cli.o zmalloc.o linenoise.o
 +CHECKDUMPOBJ = redis-check-dump.o lzf_c.o lzf_d.o
 +CHECKAOFOBJ = redis-check-aof.o
 +
 +PRGNAME = redis-server
 +BENCHPRGNAME = redis-benchmark
 +CLIPRGNAME = redis-cli
 +CHECKDUMPPRGNAME = redis-check-dump
 +CHECKAOFPRGNAME = redis-check-aof
 +
 +all: redis-server redis-benchmark redis-cli redis-check-dump redis-check-aof
 +
 +# Deps (use make dep to generate this)
 +adlist.o: adlist.c adlist.h zmalloc.h
 +ae.o: ae.c ae.h zmalloc.h config.h ae_kqueue.c
 +ae_epoll.o: ae_epoll.c
 +ae_kqueue.o: ae_kqueue.c
 +ae_select.o: ae_select.c
 +anet.o: anet.c fmacros.h anet.h
 +dict.o: dict.c fmacros.h dict.h zmalloc.h
 +linenoise.o: linenoise.c fmacros.h
 +lzf_c.o: lzf_c.c lzfP.h
 +lzf_d.o: lzf_d.c lzfP.h
 +pqsort.o: pqsort.c
 +redis-benchmark.o: redis-benchmark.c fmacros.h ae.h anet.h sds.h adlist.h \
 +  zmalloc.h
 +redis-check-aof.o: redis-check-aof.c fmacros.h config.h
 +redis-check-dump.o: redis-check-dump.c lzf.h
 +redis-cli.o: redis-cli.c fmacros.h anet.h sds.h adlist.h zmalloc.h \
 +  linenoise.h
 +redis.o: redis.c fmacros.h config.h redis.h ae.h sds.h anet.h dict.h \
 +  adlist.h zmalloc.h lzf.h pqsort.h zipmap.h ziplist.h sha1.h
 +release.o: release.c release.h
 +sds.o: sds.c sds.h zmalloc.h
 +sha1.o: sha1.c sha1.h
 +ziplist.o: ziplist.c zmalloc.h ziplist.h
 +zipmap.o: zipmap.c zmalloc.h
++intset.o: intset.c zmalloc.h
 +zmalloc.o: zmalloc.c config.h
 +
 +redis-server: $(OBJ)
 +	$(CC) -o $(PRGNAME) $(CCOPT) $(DEBUG) $(OBJ)
 +	@echo ""
 +	@echo "Hint: To run 'make test' is a good idea ;)"
 +	@echo ""
 +
 +redis-benchmark: $(BENCHOBJ)
 +	$(CC) -o $(BENCHPRGNAME) $(CCOPT) $(DEBUG) $(BENCHOBJ)
 +
 +redis-cli: $(CLIOBJ)
 +	$(CC) -o $(CLIPRGNAME) $(CCOPT) $(DEBUG) $(CLIOBJ)
 +
 +redis-check-dump: $(CHECKDUMPOBJ)
 +	$(CC) -o $(CHECKDUMPPRGNAME) $(CCOPT) $(DEBUG) $(CHECKDUMPOBJ)
 +
 +redis-check-aof: $(CHECKAOFOBJ)
 +	$(CC) -o $(CHECKAOFPRGNAME) $(CCOPT) $(DEBUG) $(CHECKAOFOBJ)
 +
 +.c.o:
 +	$(CC) -c $(CFLAGS) $(DEBUG) $(COMPILE_TIME) $<
 +
 +clean:
 +	rm -rf $(PRGNAME) $(BENCHPRGNAME) $(CLIPRGNAME) $(CHECKDUMPPRGNAME) $(CHECKAOFPRGNAME) *.o *.gcda *.gcno *.gcov
 +
 +dep:
 +	$(CC) -MM *.c
 +
 +test:
 +	(cd ..; tclsh8.5 tests/test_helper.tcl --tags "${TAGS}")
 +
 +bench:
 +	./redis-benchmark
 +
 +log:
 +	git log '--pretty=format:%ad %s (%cn)' --date=short > ../Changelog
 +
 +32bit:
 +	@echo ""
 +	@echo "WARNING: if it fails under Linux you probably need to install libc6-dev-i386"
 +	@echo ""
 +	make ARCH="-m32"
 +
 +gprof:
 +	make PROF="-pg"
 +
 +gcov:
 +	make PROF="-fprofile-arcs -ftest-coverage"
 +
 +noopt:
 +	make OPTIMIZATION=""
 +
 +32bitgprof:
 +	make PROF="-pg" ARCH="-arch i386"
diff --cc src/config.c
index 6d946ee0,00000000..e1b743db
mode 100644,000000..100644
--- a/src/config.c
+++ b/src/config.c
@@@ -1,438 -1,0 +1,440 @@@
 +#include "redis.h"
 +
 +/*-----------------------------------------------------------------------------
 + * Config file parsing
 + *----------------------------------------------------------------------------*/
 +
 +int yesnotoi(char *s) {
 +    if (!strcasecmp(s,"yes")) return 1;
 +    else if (!strcasecmp(s,"no")) return 0;
 +    else return -1;
 +}
 +
 +void appendServerSaveParams(time_t seconds, int changes) {
 +    server.saveparams = zrealloc(server.saveparams,sizeof(struct saveparam)*(server.saveparamslen+1));
 +    server.saveparams[server.saveparamslen].seconds = seconds;
 +    server.saveparams[server.saveparamslen].changes = changes;
 +    server.saveparamslen++;
 +}
 +
 +void resetServerSaveParams() {
 +    zfree(server.saveparams);
 +    server.saveparams = NULL;
 +    server.saveparamslen = 0;
 +}
 +
 +/* I agree, this is a very rudimental way to load a configuration...
 +   will improve later if the config gets more complex */
 +void loadServerConfig(char *filename) {
 +    FILE *fp;
 +    char buf[REDIS_CONFIGLINE_MAX+1], *err = NULL;
 +    int linenum = 0;
 +    sds line = NULL;
 +
 +    if (filename[0] == '-' && filename[1] == '\0')
 +        fp = stdin;
 +    else {
 +        if ((fp = fopen(filename,"r")) == NULL) {
 +            redisLog(REDIS_WARNING, "Fatal error, can't open config file '%s'", filename);
 +            exit(1);
 +        }
 +    }
 +
 +    while(fgets(buf,REDIS_CONFIGLINE_MAX+1,fp) != NULL) {
 +        sds *argv;
 +        int argc, j;
 +
 +        linenum++;
 +        line = sdsnew(buf);
 +        line = sdstrim(line," \t\r\n");
 +
 +        /* Skip comments and blank lines*/
 +        if (line[0] == '#' || line[0] == '\0') {
 +            sdsfree(line);
 +            continue;
 +        }
 +
 +        /* Split into arguments */
 +        argv = sdssplitlen(line,sdslen(line)," ",1,&argc);
 +        sdstolower(argv[0]);
 +
 +        /* Execute config directives */
 +        if (!strcasecmp(argv[0],"timeout") && argc == 2) {
 +            server.maxidletime = atoi(argv[1]);
 +            if (server.maxidletime < 0) {
 +                err = "Invalid timeout value"; goto loaderr;
 +            }
 +        } else if (!strcasecmp(argv[0],"port") && argc == 2) {
 +            server.port = atoi(argv[1]);
 +            if (server.port < 1 || server.port > 65535) {
 +                err = "Invalid port"; goto loaderr;
 +            }
 +        } else if (!strcasecmp(argv[0],"bind") && argc == 2) {
 +            server.bindaddr = zstrdup(argv[1]);
 +        } else if (!strcasecmp(argv[0],"save") && argc == 3) {
 +            int seconds = atoi(argv[1]);
 +            int changes = atoi(argv[2]);
 +            if (seconds < 1 || changes < 0) {
 +                err = "Invalid save parameters"; goto loaderr;
 +            }
 +            appendServerSaveParams(seconds,changes);
 +        } else if (!strcasecmp(argv[0],"dir") && argc == 2) {
 +            if (chdir(argv[1]) == -1) {
 +                redisLog(REDIS_WARNING,"Can't chdir to '%s': %s",
 +                    argv[1], strerror(errno));
 +                exit(1);
 +            }
 +        } else if (!strcasecmp(argv[0],"loglevel") && argc == 2) {
 +            if (!strcasecmp(argv[1],"debug")) server.verbosity = REDIS_DEBUG;
 +            else if (!strcasecmp(argv[1],"verbose")) server.verbosity = REDIS_VERBOSE;
 +            else if (!strcasecmp(argv[1],"notice")) server.verbosity = REDIS_NOTICE;
 +            else if (!strcasecmp(argv[1],"warning")) server.verbosity = REDIS_WARNING;
 +            else {
 +                err = "Invalid log level. Must be one of debug, notice, warning";
 +                goto loaderr;
 +            }
 +        } else if (!strcasecmp(argv[0],"logfile") && argc == 2) {
 +            FILE *logfp;
 +
 +            server.logfile = zstrdup(argv[1]);
 +            if (!strcasecmp(server.logfile,"stdout")) {
 +                zfree(server.logfile);
 +                server.logfile = NULL;
 +            }
 +            if (server.logfile) {
 +                /* Test if we are able to open the file. The server will not
 +                 * be able to abort just for this problem later... */
 +                logfp = fopen(server.logfile,"a");
 +                if (logfp == NULL) {
 +                    err = sdscatprintf(sdsempty(),
 +                        "Can't open the log file: %s", strerror(errno));
 +                    goto loaderr;
 +                }
 +                fclose(logfp);
 +            }
 +        } else if (!strcasecmp(argv[0],"databases") && argc == 2) {
 +            server.dbnum = atoi(argv[1]);
 +            if (server.dbnum < 1) {
 +                err = "Invalid number of databases"; goto loaderr;
 +            }
 +        } else if (!strcasecmp(argv[0],"include") && argc == 2) {
 +            loadServerConfig(argv[1]);
 +        } else if (!strcasecmp(argv[0],"maxclients") && argc == 2) {
 +            server.maxclients = atoi(argv[1]);
 +        } else if (!strcasecmp(argv[0],"maxmemory") && argc == 2) {
 +            server.maxmemory = memtoll(argv[1],NULL);
 +        } else if (!strcasecmp(argv[0],"slaveof") && argc == 3) {
 +            server.masterhost = sdsnew(argv[1]);
 +            server.masterport = atoi(argv[2]);
 +            server.replstate = REDIS_REPL_CONNECT;
 +        } else if (!strcasecmp(argv[0],"masterauth") && argc == 2) {
 +        	server.masterauth = zstrdup(argv[1]);
 +        } else if (!strcasecmp(argv[0],"glueoutputbuf") && argc == 2) {
 +            if ((server.glueoutputbuf = yesnotoi(argv[1])) == -1) {
 +                err = "argument must be 'yes' or 'no'"; goto loaderr;
 +            }
 +        } else if (!strcasecmp(argv[0],"rdbcompression") && argc == 2) {
 +            if ((server.rdbcompression = yesnotoi(argv[1])) == -1) {
 +                err = "argument must be 'yes' or 'no'"; goto loaderr;
 +            }
 +        } else if (!strcasecmp(argv[0],"activerehashing") && argc == 2) {
 +            if ((server.activerehashing = yesnotoi(argv[1])) == -1) {
 +                err = "argument must be 'yes' or 'no'"; goto loaderr;
 +            }
 +        } else if (!strcasecmp(argv[0],"daemonize") && argc == 2) {
 +            if ((server.daemonize = yesnotoi(argv[1])) == -1) {
 +                err = "argument must be 'yes' or 'no'"; goto loaderr;
 +            }
 +        } else if (!strcasecmp(argv[0],"appendonly") && argc == 2) {
 +            if ((server.appendonly = yesnotoi(argv[1])) == -1) {
 +                err = "argument must be 'yes' or 'no'"; goto loaderr;
 +            }
 +        } else if (!strcasecmp(argv[0],"appendfilename") && argc == 2) {
 +            zfree(server.appendfilename);
 +            server.appendfilename = zstrdup(argv[1]);
 +        } else if (!strcasecmp(argv[0],"no-appendfsync-on-rewrite")
 +                   && argc == 2) {
 +            if ((server.no_appendfsync_on_rewrite= yesnotoi(argv[1])) == -1) {
 +                err = "argument must be 'yes' or 'no'"; goto loaderr;
 +            }
 +        } else if (!strcasecmp(argv[0],"appendfsync") && argc == 2) {
 +            if (!strcasecmp(argv[1],"no")) {
 +                server.appendfsync = APPENDFSYNC_NO;
 +            } else if (!strcasecmp(argv[1],"always")) {
 +                server.appendfsync = APPENDFSYNC_ALWAYS;
 +            } else if (!strcasecmp(argv[1],"everysec")) {
 +                server.appendfsync = APPENDFSYNC_EVERYSEC;
 +            } else {
 +                err = "argument must be 'no', 'always' or 'everysec'";
 +                goto loaderr;
 +            }
 +        } else if (!strcasecmp(argv[0],"requirepass") && argc == 2) {
 +            server.requirepass = zstrdup(argv[1]);
 +        } else if (!strcasecmp(argv[0],"pidfile") && argc == 2) {
 +            zfree(server.pidfile);
 +            server.pidfile = zstrdup(argv[1]);
 +        } else if (!strcasecmp(argv[0],"dbfilename") && argc == 2) {
 +            zfree(server.dbfilename);
 +            server.dbfilename = zstrdup(argv[1]);
 +        } else if (!strcasecmp(argv[0],"vm-enabled") && argc == 2) {
 +            if ((server.vm_enabled = yesnotoi(argv[1])) == -1) {
 +                err = "argument must be 'yes' or 'no'"; goto loaderr;
 +            }
 +        } else if (!strcasecmp(argv[0],"vm-swap-file") && argc == 2) {
 +            zfree(server.vm_swap_file);
 +            server.vm_swap_file = zstrdup(argv[1]);
 +        } else if (!strcasecmp(argv[0],"vm-max-memory") && argc == 2) {
 +            server.vm_max_memory = memtoll(argv[1],NULL);
 +        } else if (!strcasecmp(argv[0],"vm-page-size") && argc == 2) {
 +            server.vm_page_size = memtoll(argv[1], NULL);
 +        } else if (!strcasecmp(argv[0],"vm-pages") && argc == 2) {
 +            server.vm_pages = memtoll(argv[1], NULL);
 +        } else if (!strcasecmp(argv[0],"vm-max-threads") && argc == 2) {
 +            server.vm_max_threads = strtoll(argv[1], NULL, 10);
 +        } else if (!strcasecmp(argv[0],"hash-max-zipmap-entries") && argc == 2){
 +            server.hash_max_zipmap_entries = memtoll(argv[1], NULL);
 +        } else if (!strcasecmp(argv[0],"hash-max-zipmap-value") && argc == 2){
 +            server.hash_max_zipmap_value = memtoll(argv[1], NULL);
 +        } else if (!strcasecmp(argv[0],"list-max-ziplist-entries") && argc == 2){
 +            server.list_max_ziplist_entries = memtoll(argv[1], NULL);
 +        } else if (!strcasecmp(argv[0],"list-max-ziplist-value") && argc == 2){
 +            server.list_max_ziplist_value = memtoll(argv[1], NULL);
++        } else if (!strcasecmp(argv[0],"set-max-intset-entries") && argc == 2){
++            server.set_max_intset_entries = memtoll(argv[1], NULL);
 +        } else {
 +            err = "Bad directive or wrong number of arguments"; goto loaderr;
 +        }
 +        for (j = 0; j < argc; j++)
 +            sdsfree(argv[j]);
 +        zfree(argv);
 +        sdsfree(line);
 +    }
 +    if (fp != stdin) fclose(fp);
 +    return;
 +
 +loaderr:
 +    fprintf(stderr, "\n*** FATAL CONFIG FILE ERROR ***\n");
 +    fprintf(stderr, "Reading the configuration file, at line %d\n", linenum);
 +    fprintf(stderr, ">>> '%s'\n", line);
 +    fprintf(stderr, "%s\n", err);
 +    exit(1);
 +}
 +
 +/*-----------------------------------------------------------------------------
 + * CONFIG command for remote configuration
 + *----------------------------------------------------------------------------*/
 +
 +void configSetCommand(redisClient *c) {
 +    robj *o = getDecodedObject(c->argv[3]);
 +    long long ll;
 +
 +    if (!strcasecmp(c->argv[2]->ptr,"dbfilename")) {
 +        zfree(server.dbfilename);
 +        server.dbfilename = zstrdup(o->ptr);
 +    } else if (!strcasecmp(c->argv[2]->ptr,"requirepass")) {
 +        zfree(server.requirepass);
 +        server.requirepass = zstrdup(o->ptr);
 +    } else if (!strcasecmp(c->argv[2]->ptr,"masterauth")) {
 +        zfree(server.masterauth);
 +        server.masterauth = zstrdup(o->ptr);
 +    } else if (!strcasecmp(c->argv[2]->ptr,"maxmemory")) {
 +        if (getLongLongFromObject(o,&ll) == REDIS_ERR ||
 +            ll < 0) goto badfmt;
 +        server.maxmemory = ll;
 +    } else if (!strcasecmp(c->argv[2]->ptr,"timeout")) {
 +        if (getLongLongFromObject(o,&ll) == REDIS_ERR ||
 +            ll < 0 || ll > LONG_MAX) goto badfmt;
 +        server.maxidletime = ll;
 +    } else if (!strcasecmp(c->argv[2]->ptr,"appendfsync")) {
 +        if (!strcasecmp(o->ptr,"no")) {
 +            server.appendfsync = APPENDFSYNC_NO;
 +        } else if (!strcasecmp(o->ptr,"everysec")) {
 +            server.appendfsync = APPENDFSYNC_EVERYSEC;
 +        } else if (!strcasecmp(o->ptr,"always")) {
 +            server.appendfsync = APPENDFSYNC_ALWAYS;
 +        } else {
 +            goto badfmt;
 +        }
 +    } else if (!strcasecmp(c->argv[2]->ptr,"no-appendfsync-on-rewrite")) {
 +        int yn = yesnotoi(o->ptr);
 +
 +        if (yn == -1) goto badfmt;
 +        server.no_appendfsync_on_rewrite = yn;
 +    } else if (!strcasecmp(c->argv[2]->ptr,"appendonly")) {
 +        int old = server.appendonly;
 +        int new = yesnotoi(o->ptr);
 +
 +        if (new == -1) goto badfmt;
 +        if (old != new) {
 +            if (new == 0) {
 +                stopAppendOnly();
 +            } else {
 +                if (startAppendOnly() == REDIS_ERR) {
 +                    addReplySds(c,sdscatprintf(sdsempty(),
 +                        "-ERR Unable to turn on AOF. Check server logs.\r\n"));
 +                    decrRefCount(o);
 +                    return;
 +                }
 +            }
 +        }
 +    } else if (!strcasecmp(c->argv[2]->ptr,"save")) {
 +        int vlen, j;
 +        sds *v = sdssplitlen(o->ptr,sdslen(o->ptr)," ",1,&vlen);
 +
 +        /* Perform sanity check before setting the new config:
 +         * - Even number of args
 +         * - Seconds >= 1, changes >= 0 */
 +        if (vlen & 1) {
 +            sdsfreesplitres(v,vlen);
 +            goto badfmt;
 +        }
 +        for (j = 0; j < vlen; j++) {
 +            char *eptr;
 +            long val;
 +
 +            val = strtoll(v[j], &eptr, 10);
 +            if (eptr[0] != '\0' ||
 +                ((j & 1) == 0 && val < 1) ||
 +                ((j & 1) == 1 && val < 0)) {
 +                sdsfreesplitres(v,vlen);
 +                goto badfmt;
 +            }
 +        }
 +        /* Finally set the new config */
 +        resetServerSaveParams();
 +        for (j = 0; j < vlen; j += 2) {
 +            time_t seconds;
 +            int changes;
 +
 +            seconds = strtoll(v[j],NULL,10);
 +            changes = strtoll(v[j+1],NULL,10);
 +            appendServerSaveParams(seconds, changes);
 +        }
 +        sdsfreesplitres(v,vlen);
 +    } else {
 +        addReplySds(c,sdscatprintf(sdsempty(),
 +            "-ERR not supported CONFIG parameter %s\r\n",
 +            (char*)c->argv[2]->ptr));
 +        decrRefCount(o);
 +        return;
 +    }
 +    decrRefCount(o);
 +    addReply(c,shared.ok);
 +    return;
 +
 +badfmt: /* Bad format errors */
 +    addReplySds(c,sdscatprintf(sdsempty(),
 +        "-ERR invalid argument '%s' for CONFIG SET '%s'\r\n",
 +            (char*)o->ptr,
 +            (char*)c->argv[2]->ptr));
 +    decrRefCount(o);
 +}
 +
 +void configGetCommand(redisClient *c) {
 +    robj *o = getDecodedObject(c->argv[2]);
 +    robj *lenobj = createObject(REDIS_STRING,NULL);
 +    char *pattern = o->ptr;
 +    int matches = 0;
 +
 +    addReply(c,lenobj);
 +    decrRefCount(lenobj);
 +
 +    if (stringmatch(pattern,"dbfilename",0)) {
 +        addReplyBulkCString(c,"dbfilename");
 +        addReplyBulkCString(c,server.dbfilename);
 +        matches++;
 +    }
 +    if (stringmatch(pattern,"requirepass",0)) {
 +        addReplyBulkCString(c,"requirepass");
 +        addReplyBulkCString(c,server.requirepass);
 +        matches++;
 +    }
 +    if (stringmatch(pattern,"masterauth",0)) {
 +        addReplyBulkCString(c,"masterauth");
 +        addReplyBulkCString(c,server.masterauth);
 +        matches++;
 +    }
 +    if (stringmatch(pattern,"maxmemory",0)) {
 +        char buf[128];
 +
 +        ll2string(buf,128,server.maxmemory);
 +        addReplyBulkCString(c,"maxmemory");
 +        addReplyBulkCString(c,buf);
 +        matches++;
 +    }
 +    if (stringmatch(pattern,"timeout",0)) {
 +        char buf[128];
 +
 +        ll2string(buf,128,server.maxidletime);
 +        addReplyBulkCString(c,"timeout");
 +        addReplyBulkCString(c,buf);
 +        matches++;
 +    }
 +    if (stringmatch(pattern,"appendonly",0)) {
 +        addReplyBulkCString(c,"appendonly");
 +        addReplyBulkCString(c,server.appendonly ? "yes" : "no");
 +        matches++;
 +    }
 +    if (stringmatch(pattern,"no-appendfsync-on-rewrite",0)) {
 +        addReplyBulkCString(c,"no-appendfsync-on-rewrite");
 +        addReplyBulkCString(c,server.no_appendfsync_on_rewrite ? "yes" : "no");
 +        matches++;
 +    }
 +    if (stringmatch(pattern,"appendfsync",0)) {
 +        char *policy;
 +
 +        switch(server.appendfsync) {
 +        case APPENDFSYNC_NO: policy = "no"; break;
 +        case APPENDFSYNC_EVERYSEC: policy = "everysec"; break;
 +        case APPENDFSYNC_ALWAYS: policy = "always"; break;
 +        default: policy = "unknown"; break; /* too harmless to panic */
 +        }
 +        addReplyBulkCString(c,"appendfsync");
 +        addReplyBulkCString(c,policy);
 +        matches++;
 +    }
 +    if (stringmatch(pattern,"save",0)) {
 +        sds buf = sdsempty();
 +        int j;
 +
 +        for (j = 0; j < server.saveparamslen; j++) {
 +            buf = sdscatprintf(buf,"%ld %d",
 +                    server.saveparams[j].seconds,
 +                    server.saveparams[j].changes);
 +            if (j != server.saveparamslen-1)
 +                buf = sdscatlen(buf," ",1);
 +        }
 +        addReplyBulkCString(c,"save");
 +        addReplyBulkCString(c,buf);
 +        sdsfree(buf);
 +        matches++;
 +    }
 +    decrRefCount(o);
 +    lenobj->ptr = sdscatprintf(sdsempty(),"*%d\r\n",matches*2);
 +}
 +
 +void configCommand(redisClient *c) {
 +    if (!strcasecmp(c->argv[1]->ptr,"set")) {
 +        if (c->argc != 4) goto badarity;
 +        configSetCommand(c);
 +    } else if (!strcasecmp(c->argv[1]->ptr,"get")) {
 +        if (c->argc != 3) goto badarity;
 +        configGetCommand(c);
 +    } else if (!strcasecmp(c->argv[1]->ptr,"resetstat")) {
 +        if (c->argc != 2) goto badarity;
 +        server.stat_numcommands = 0;
 +        server.stat_numconnections = 0;
 +        server.stat_expiredkeys = 0;
 +        server.stat_starttime = time(NULL);
 +        addReply(c,shared.ok);
 +    } else {
 +        addReplySds(c,sdscatprintf(sdsempty(),
 +            "-ERR CONFIG subcommand must be one of GET, SET, RESETSTAT\r\n"));
 +    }
 +    return;
 +
 +badarity:
 +    addReplySds(c,sdscatprintf(sdsempty(),
 +        "-ERR Wrong number of arguments for CONFIG %s\r\n",
 +        (char*) c->argv[1]->ptr));
 +}
diff --cc src/intset.c
index 00000000,00000000..2532582e
new file mode 100644
--- /dev/null
+++ b/src/intset.c
@@@ -1,0 -1,0 +1,394 @@@
++#include <stdio.h>
++#include <stdlib.h>
++#include <string.h>
++#include "intset.h"
++#include "zmalloc.h"
++
++/* Note that these encodings are ordered, so:
++ * INTSET_ENC_INT16 < INTSET_ENC_INT32 < INTSET_ENC_INT64. */
++#define INTSET_ENC_INT16 (sizeof(int16_t))
++#define INTSET_ENC_INT32 (sizeof(int32_t))
++#define INTSET_ENC_INT64 (sizeof(int64_t))
++
++/* Accessors for each type of encoding */
++#define INTSET_VALUE_ENCODING(__val) (((__val) < INT32_MIN || (__val) > INT32_MAX) ? \
++    INTSET_ENC_INT64 : (((__val) < INT16_MIN || (__val) > INT16_MAX) ? \
++    INTSET_ENC_INT32 : INTSET_ENC_INT16))
++#define INTSET_GET_ENCODED(__is,__pos,__enc) ((__enc == INTSET_ENC_INT64) ? \
++    ((int64_t*)(__is)->contents)[__pos] : ((__enc == INTSET_ENC_INT32) ? \
++    ((int32_t*)(__is)->contents)[__pos] : ((int16_t*)(__is)->contents)[__pos]))
++#define INTSET_GET(__is,__pos) (INTSET_GET_ENCODED(__is,__pos,(__is)->encoding))
++#define INTSET_SET(__is,__pos,__val) { \
++    if ((__is)->encoding == INTSET_ENC_INT64) \
++        ((int64_t*)(__is)->contents)[__pos] = (__val); \
++    else if ((__is)->encoding == INTSET_ENC_INT32) \
++        ((int32_t*)(__is)->contents)[__pos] = (__val); \
++    else \
++        ((int16_t*)(__is)->contents)[__pos] = (__val); }
++
++/* Create an empty intset. */
++intset *intsetNew(void) {
++    intset *is = zmalloc(sizeof(intset));
++    is->encoding = INTSET_ENC_INT16;
++    is->length = 0;
++    return is;
++}
++
++/* Resize the intset */
++static intset *intsetResize(intset *is, uint32_t len) {
++    uint32_t size = len*is->encoding;
++    is = zrealloc(is,sizeof(intset)+size);
++    return is;
++}
++
++static intset *intsetUpgrade(intset *is, uint8_t newenc, uint8_t extra, uint8_t offset) {
++    uint8_t curenc = is->encoding;
++    int length = is->length;
++
++    /* First set new encoding and resize */
++    is->encoding = newenc;
++    is = intsetResize(is,is->length+extra);
++
++    /* Upgrade back-to-front so we don't overwrite values */
++    while(length--)
++        INTSET_SET(is,length+offset,INTSET_GET_ENCODED(is,length,curenc));
++    return is;
++}
++
++/* Search for the position of "value". Return 1 when the value was found and
++ * sets "pos" to the position of the value within the intset. Return 0 when
++ * the value is not present in the intset and sets "pos" to the position
++ * where "value" can be inserted. */
++static uint8_t intsetSearch(intset *is, int64_t value, uint32_t *pos) {
++    int min = 0, max = is->length-1, mid = -1;
++    int64_t cur = -1;
++
++    /* The value can never be found when the set is empty */
++    if (is->length == 0) {
++        if (pos) *pos = 0;
++        return 0;
++    } else {
++        /* Check for the case where we know we cannot find the value,
++         * but do know the insert position. */
++        if (value > INTSET_GET(is,is->length-1)) {
++            if (pos) *pos = is->length;
++            return 0;
++        } else if (value < INTSET_GET(is,0)) {
++            if (pos) *pos = 0;
++            return 0;
++        }
++    }
++
++    while(max >= min) {
++        mid = (min+max)/2;
++        cur = INTSET_GET(is,mid);
++        if (value > cur) {
++            min = mid+1;
++        } else if (value < cur) {
++            max = mid-1;
++        } else {
++            break;
++        }
++    }
++
++    if (value == cur) {
++        if (pos) *pos = mid;
++        return 1;
++    } else {
++        if (pos) *pos = min;
++        return 0;
++    }
++}
++
++static void intsetMoveTail(intset *is, uint32_t from, uint32_t to) {
++    void *src, *dst;
++    uint32_t bytes = is->length-from;
++    if (is->encoding == INTSET_ENC_INT64) {
++        src = (int64_t*)is->contents+from;
++        dst = (int64_t*)is->contents+to;
++        bytes *= sizeof(int64_t);
++    } else if (is->encoding == INTSET_ENC_INT32) {
++        src = (int32_t*)is->contents+from;
++        dst = (int32_t*)is->contents+to;
++        bytes *= sizeof(int32_t);
++    } else {
++        src = (int16_t*)is->contents+from;
++        dst = (int16_t*)is->contents+to;
++        bytes *= sizeof(int16_t);
++    }
++    memmove(dst,src,bytes);
++}
++
++/* Insert an integer in the intset */
++intset *intsetAdd(intset *is, int64_t value, uint8_t *success) {
++    uint8_t valenc = INTSET_VALUE_ENCODING(value);
++    uint32_t pos, offset;
++    if (success) *success = 1;
++
++    /* Upgrade encoding if necessary. If we need to upgrade, we know that
++     * this value should be either appended (if > 0) or prepended (if < 0),
++     * because it lies outside the range of existing values. */
++    if (valenc > is->encoding) {
++        offset = value < 0 ? 1 : 0;
++        is = intsetUpgrade(is,valenc,1,offset);
++        pos = (value < 0) ? 0 : is->length;
++    } else {
++        /* Abort if the value is already present in the set.
++         * This call will populate "pos" with the right position to insert
++         * the value when it cannot be found. */
++        if (intsetSearch(is,value,&pos)) {
++            if (success) *success = 0;
++            return is;
++        }
++
++        is = intsetResize(is,is->length+1);
++        if (pos < is->length) intsetMoveTail(is,pos,pos+1);
++    }
++
++    INTSET_SET(is,pos,value);
++    is->length++;
++    return is;
++}
++
++/* Delete integer from intset */
++intset *intsetRemove(intset *is, int64_t value, uint8_t *success) {
++    uint8_t valenc = INTSET_VALUE_ENCODING(value);
++    uint32_t pos;
++    if (success) *success = 0;
++
++    if (valenc <= is->encoding && intsetSearch(is,value,&pos)) {
++        /* We know we can delete */
++        if (success) *success = 1;
++
++        /* Overwrite value with tail and update length */
++        if (pos < (is->length-1)) intsetMoveTail(is,pos+1,pos);
++        is = intsetResize(is,is->length-1);
++        is->length--;
++    }
++    return is;
++}
++
++/* Determine whether a value belongs to this set */
++uint8_t intsetFind(intset *is, int64_t value) {
++    uint8_t valenc = INTSET_VALUE_ENCODING(value);
++    return valenc <= is->encoding && intsetSearch(is,value,NULL);
++}
++
++/* Return random member */
++int64_t intsetRandom(intset *is) {
++    return INTSET_GET(is,rand()%is->length);
++}
++
++/* 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) {
++    if (pos < is->length) {
++        *value = INTSET_GET(is,pos);
++        return 1;
++    }
++    return 0;
++}
++
++/* Return intset length */
++uint32_t intsetLen(intset *is) {
++    return is->length;
++}
++
++#ifdef INTSET_TEST_MAIN
++#include <sys/time.h>
++
++void intsetRepr(intset *is) {
++    int i;
++    for (i = 0; i < is->length; i++) {
++        printf("%lld\n", (uint64_t)INTSET_GET(is,i));
++    }
++    printf("\n");
++}
++
++void error(char *err) {
++    printf("%s\n", err);
++    exit(1);
++}
++
++void ok(void) {
++    printf("OK\n");
++}
++
++long long usec(void) {
++    struct timeval tv;
++    gettimeofday(&tv,NULL);
++    return (((long long)tv.tv_sec)*1000000)+tv.tv_usec;
++}
++
++#define assert(_e) ((_e)?(void)0:(_assert(#_e,__FILE__,__LINE__),exit(1)))
++void _assert(char *estr, char *file, int line) {
++    printf("\n\n=== ASSERTION FAILED ===\n");
++    printf("==> %s:%d '%s' is not true\n",file,line,estr);
++}
++
++intset *createSet(int bits, int size) {
++    uint64_t mask = (1<<bits)-1;
++    uint64_t i, value;
++    intset *is = intsetNew();
++
++    for (i = 0; i < size; i++) {
++        if (bits > 32) {
++            value = (rand()*rand()) & mask;
++        } else {
++            value = rand() & mask;
++        }
++        is = intsetAdd(is,value,NULL);
++    }
++    return is;
++}
++
++void checkConsistency(intset *is) {
++    int i;
++
++    for (i = 0; i < (is->length-1); i++) {
++        if (is->encoding == INTSET_ENC_INT16) {
++            int16_t *i16 = (int16_t*)is->contents;
++            assert(i16[i] < i16[i+1]);
++        } else if (is->encoding == INTSET_ENC_INT32) {
++            int32_t *i32 = (int32_t*)is->contents;
++            assert(i32[i] < i32[i+1]);
++        } else {
++            int64_t *i64 = (int64_t*)is->contents;
++            assert(i64[i] < i64[i+1]);
++        }
++    }
++}
++
++int main(int argc, char **argv) {
++    uint8_t success;
++    int i;
++    intset *is;
++    sranddev();
++
++    printf("Value encodings: "); {
++        assert(INTSET_VALUE_ENCODING(-32768) == INTSET_ENC_INT16);
++        assert(INTSET_VALUE_ENCODING(+32767) == INTSET_ENC_INT16);
++        assert(INTSET_VALUE_ENCODING(-32769) == INTSET_ENC_INT32);
++        assert(INTSET_VALUE_ENCODING(+32768) == INTSET_ENC_INT32);
++        assert(INTSET_VALUE_ENCODING(-2147483648) == INTSET_ENC_INT32);
++        assert(INTSET_VALUE_ENCODING(+2147483647) == INTSET_ENC_INT32);
++        assert(INTSET_VALUE_ENCODING(-2147483649) == INTSET_ENC_INT64);
++        assert(INTSET_VALUE_ENCODING(+2147483648) == INTSET_ENC_INT64);
++        assert(INTSET_VALUE_ENCODING(-9223372036854775808ull) == INTSET_ENC_INT64);
++        assert(INTSET_VALUE_ENCODING(+9223372036854775807ull) == INTSET_ENC_INT64);
++        ok();
++    }
++
++    printf("Basic adding: "); {
++        is = intsetNew();
++        is = intsetAdd(is,5,&success); assert(success);
++        is = intsetAdd(is,6,&success); assert(success);
++        is = intsetAdd(is,4,&success); assert(success);
++        is = intsetAdd(is,4,&success); assert(!success);
++        ok();
++    }
++
++    printf("Large number of random adds: "); {
++        int inserts = 0;
++        is = intsetNew();
++        for (i = 0; i < 1024; i++) {
++            is = intsetAdd(is,rand()%0x800,&success);
++            if (success) inserts++;
++        }
++        assert(is->length == inserts);
++        checkConsistency(is);
++        ok();
++    }
++
++    printf("Upgrade from int16 to int32: "); {
++        is = intsetNew();
++        is = intsetAdd(is,32,NULL);
++        assert(is->encoding == INTSET_ENC_INT16);
++        is = intsetAdd(is,65535,NULL);
++        assert(is->encoding == INTSET_ENC_INT32);
++        assert(intsetFind(is,32));
++        assert(intsetFind(is,65535));
++        checkConsistency(is);
++
++        is = intsetNew();
++        is = intsetAdd(is,32,NULL);
++        assert(is->encoding == INTSET_ENC_INT16);
++        is = intsetAdd(is,-65535,NULL);
++        assert(is->encoding == INTSET_ENC_INT32);
++        assert(intsetFind(is,32));
++        assert(intsetFind(is,-65535));
++        checkConsistency(is);
++        ok();
++    }
++
++    printf("Upgrade from int16 to int64: "); {
++        is = intsetNew();
++        is = intsetAdd(is,32,NULL);
++        assert(is->encoding == INTSET_ENC_INT16);
++        is = intsetAdd(is,4294967295,NULL);
++        assert(is->encoding == INTSET_ENC_INT64);
++        assert(intsetFind(is,32));
++        assert(intsetFind(is,4294967295));
++        checkConsistency(is);
++
++        is = intsetNew();
++        is = intsetAdd(is,32,NULL);
++        assert(is->encoding == INTSET_ENC_INT16);
++        is = intsetAdd(is,-4294967295,NULL);
++        assert(is->encoding == INTSET_ENC_INT64);
++        assert(intsetFind(is,32));
++        assert(intsetFind(is,-4294967295));
++        checkConsistency(is);
++        ok();
++    }
++
++    printf("Upgrade from int32 to int64: "); {
++        is = intsetNew();
++        is = intsetAdd(is,65535,NULL);
++        assert(is->encoding == INTSET_ENC_INT32);
++        is = intsetAdd(is,4294967295,NULL);
++        assert(is->encoding == INTSET_ENC_INT64);
++        assert(intsetFind(is,65535));
++        assert(intsetFind(is,4294967295));
++        checkConsistency(is);
++
++        is = intsetNew();
++        is = intsetAdd(is,65535,NULL);
++        assert(is->encoding == INTSET_ENC_INT32);
++        is = intsetAdd(is,-4294967295,NULL);
++        assert(is->encoding == INTSET_ENC_INT64);
++        assert(intsetFind(is,65535));
++        assert(intsetFind(is,-4294967295));
++        checkConsistency(is);
++        ok();
++    }
++
++    printf("Stress lookups: "); {
++        long num = 100000, size = 10000;
++        int i, bits = 20;
++        long long start;
++        is = createSet(bits,size);
++        checkConsistency(is);
++
++        start = usec();
++        for (i = 0; i < num; i++) intsetSearch(is,rand() % ((1<<bits)-1),NULL);
++        printf("%ld lookups, %ld element set, %lldusec\n",num,size,usec()-start);
++    }
++
++    printf("Stress add+delete: "); {
++        int i, v1, v2;
++        is = intsetNew();
++        for (i = 0; i < 0xffff; i++) {
++            v1 = rand() % 0xfff;
++            is = intsetAdd(is,v1,NULL);
++            assert(intsetFind(is,v1));
++
++            v2 = rand() % 0xfff;
++            is = intsetRemove(is,v2,NULL);
++            assert(!intsetFind(is,v2));
++        }
++        checkConsistency(is);
++        ok();
++    }
++}
++#endif
diff --cc src/intset.h
index 00000000,00000000..25afc18d
new file mode 100644
--- /dev/null
+++ b/src/intset.h
@@@ -1,0 -1,0 +1,19 @@@
++#ifndef __INTSET_H
++#define __INTSET_H
++#include <stdint.h>
++
++typedef struct intset {
++    uint32_t encoding;
++    uint32_t length;
++    int8_t contents[];
++} intset;
++
++intset *intsetNew(void);
++intset *intsetAdd(intset *is, int64_t value, uint8_t *success);
++intset *intsetRemove(intset *is, int64_t value, uint8_t *success);
++uint8_t intsetFind(intset *is, int64_t value);
++int64_t intsetRandom(intset *is);
++uint8_t intsetGet(intset *is, uint32_t pos, int64_t *value);
++uint32_t intsetLen(intset *is);
++
++#endif // __INTSET_H
diff --cc src/object.c
index 4854909e,00000000..16c4d74c
mode 100644,000000..100644
--- a/src/object.c
+++ b/src/object.c
@@@ -1,405 -1,0 +1,424 @@@
 +#include "redis.h"
 +#include <pthread.h>
 +
 +robj *createObject(int type, void *ptr) {
 +    robj *o;
 +
 +    if (server.vm_enabled) pthread_mutex_lock(&server.obj_freelist_mutex);
 +    if (listLength(server.objfreelist)) {
 +        listNode *head = listFirst(server.objfreelist);
 +        o = listNodeValue(head);
 +        listDelNode(server.objfreelist,head);
 +        if (server.vm_enabled) pthread_mutex_unlock(&server.obj_freelist_mutex);
 +    } else {
 +        if (server.vm_enabled)
 +            pthread_mutex_unlock(&server.obj_freelist_mutex);
 +        o = zmalloc(sizeof(*o));
 +    }
 +    o->type = type;
 +    o->encoding = REDIS_ENCODING_RAW;
 +    o->ptr = ptr;
 +    o->refcount = 1;
 +    if (server.vm_enabled) {
 +        /* Note that this code may run in the context of an I/O thread
 +         * and accessing server.lruclock in theory is an error
 +         * (no locks). But in practice this is safe, and even if we read
 +         * garbage Redis will not fail. */
 +        o->lru = server.lruclock;
 +        o->storage = REDIS_VM_MEMORY;
 +    }
 +    return o;
 +}
 +
 +robj *createStringObject(char *ptr, size_t len) {
 +    return createObject(REDIS_STRING,sdsnewlen(ptr,len));
 +}
 +
 +robj *createStringObjectFromLongLong(long long value) {
 +    robj *o;
 +    if (value >= 0 && value < REDIS_SHARED_INTEGERS) {
 +        incrRefCount(shared.integers[value]);
 +        o = shared.integers[value];
 +    } else {
 +        if (value >= LONG_MIN && value <= LONG_MAX) {
 +            o = createObject(REDIS_STRING, NULL);
 +            o->encoding = REDIS_ENCODING_INT;
 +            o->ptr = (void*)((long)value);
 +        } else {
 +            o = createObject(REDIS_STRING,sdsfromlonglong(value));
 +        }
 +    }
 +    return o;
 +}
 +
 +robj *dupStringObject(robj *o) {
 +    redisAssert(o->encoding == REDIS_ENCODING_RAW);
 +    return createStringObject(o->ptr,sdslen(o->ptr));
 +}
 +
 +robj *createListObject(void) {
 +    list *l = listCreate();
 +    robj *o = createObject(REDIS_LIST,l);
 +    listSetFreeMethod(l,decrRefCount);
 +    o->encoding = REDIS_ENCODING_LINKEDLIST;
 +    return o;
 +}
 +
 +robj *createZiplistObject(void) {
 +    unsigned char *zl = ziplistNew();
 +    robj *o = createObject(REDIS_LIST,zl);
 +    o->encoding = REDIS_ENCODING_ZIPLIST;
 +    return o;
 +}
 +
 +robj *createSetObject(void) {
 +    dict *d = dictCreate(&setDictType,NULL);
-     return createObject(REDIS_SET,d);
++    robj *o = createObject(REDIS_SET,d);
++    o->encoding = REDIS_ENCODING_HT;
++    return o;
++}
++
++robj *createIntsetObject(void) {
++    intset *is = intsetNew();
++    robj *o = createObject(REDIS_SET,is);
++    o->encoding = REDIS_ENCODING_INTSET;
++    return o;
 +}
 +
 +robj *createHashObject(void) {
 +    /* All the Hashes start as zipmaps. Will be automatically converted
 +     * into hash tables if there are enough elements or big elements
 +     * inside. */
 +    unsigned char *zm = zipmapNew();
 +    robj *o = createObject(REDIS_HASH,zm);
 +    o->encoding = REDIS_ENCODING_ZIPMAP;
 +    return o;
 +}
 +
 +robj *createZsetObject(void) {
 +    zset *zs = zmalloc(sizeof(*zs));
 +
 +    zs->dict = dictCreate(&zsetDictType,NULL);
 +    zs->zsl = zslCreate();
 +    return createObject(REDIS_ZSET,zs);
 +}
 +
 +void freeStringObject(robj *o) {
 +    if (o->encoding == REDIS_ENCODING_RAW) {
 +        sdsfree(o->ptr);
 +    }
 +}
 +
 +void freeListObject(robj *o) {
 +    switch (o->encoding) {
 +    case REDIS_ENCODING_LINKEDLIST:
 +        listRelease((list*) o->ptr);
 +        break;
 +    case REDIS_ENCODING_ZIPLIST:
 +        zfree(o->ptr);
 +        break;
 +    default:
 +        redisPanic("Unknown list encoding type");
 +    }
 +}
 +
 +void freeSetObject(robj *o) {
-     dictRelease((dict*) o->ptr);
++    switch (o->encoding) {
++    case REDIS_ENCODING_HT:
++        dictRelease((dict*) o->ptr);
++        break;
++    case REDIS_ENCODING_INTSET:
++        zfree(o->ptr);
++        break;
++    default:
++        redisPanic("Unknown set encoding type");
++    }
 +}
 +
 +void freeZsetObject(robj *o) {
 +    zset *zs = o->ptr;
 +
 +    dictRelease(zs->dict);
 +    zslFree(zs->zsl);
 +    zfree(zs);
 +}
 +
 +void freeHashObject(robj *o) {
 +    switch (o->encoding) {
 +    case REDIS_ENCODING_HT:
 +        dictRelease((dict*) o->ptr);
 +        break;
 +    case REDIS_ENCODING_ZIPMAP:
 +        zfree(o->ptr);
 +        break;
 +    default:
 +        redisPanic("Unknown hash encoding type");
 +        break;
 +    }
 +}
 +
 +void incrRefCount(robj *o) {
 +    o->refcount++;
 +}
 +
 +void decrRefCount(void *obj) {
 +    robj *o = obj;
 +
 +    /* Object is a swapped out value, or in the process of being loaded. */
 +    if (server.vm_enabled &&
 +        (o->storage == REDIS_VM_SWAPPED || o->storage == REDIS_VM_LOADING))
 +    {
 +        vmpointer *vp = obj;
 +        if (o->storage == REDIS_VM_LOADING) vmCancelThreadedIOJob(o);
 +        vmMarkPagesFree(vp->page,vp->usedpages);
 +        server.vm_stats_swapped_objects--;
 +        zfree(vp);
 +        return;
 +    }
 +
 +    if (o->refcount <= 0) redisPanic("decrRefCount against refcount <= 0");
 +    /* Object is in memory, or in the process of being swapped out.
 +     *
 +     * If the object is being swapped out, abort the operation on
 +     * decrRefCount even if the refcount does not drop to 0: the object
 +     * is referenced at least two times, as value of the key AND as
 +     * job->val in the iojob. So if we don't invalidate the iojob, when it is
 +     * done but the relevant key was removed in the meantime, the
 +     * complete jobs handler will not find the key about the job and the
 +     * assert will fail. */
 +    if (server.vm_enabled && o->storage == REDIS_VM_SWAPPING)
 +        vmCancelThreadedIOJob(o);
 +    if (--(o->refcount) == 0) {
 +        switch(o->type) {
 +        case REDIS_STRING: freeStringObject(o); break;
 +        case REDIS_LIST: freeListObject(o); break;
 +        case REDIS_SET: freeSetObject(o); break;
 +        case REDIS_ZSET: freeZsetObject(o); break;
 +        case REDIS_HASH: freeHashObject(o); break;
 +        default: redisPanic("Unknown object type"); break;
 +        }
 +        if (server.vm_enabled) pthread_mutex_lock(&server.obj_freelist_mutex);
 +        if (listLength(server.objfreelist) > REDIS_OBJFREELIST_MAX ||
 +            !listAddNodeHead(server.objfreelist,o))
 +            zfree(o);
 +        if (server.vm_enabled) pthread_mutex_unlock(&server.obj_freelist_mutex);
 +    }
 +}
 +
 +int checkType(redisClient *c, robj *o, int type) {
 +    if (o->type != type) {
 +        addReply(c,shared.wrongtypeerr);
 +        return 1;
 +    }
 +    return 0;
 +}
 +
 +/* Try to encode a string object in order to save space */
 +robj *tryObjectEncoding(robj *o) {
 +    long value;
 +    sds s = o->ptr;
 +
 +    if (o->encoding != REDIS_ENCODING_RAW)
 +        return o; /* Already encoded */
 +
 +    /* It's not safe to encode shared objects: shared objects can be shared
 +     * everywhere in the "object space" of Redis. Encoded objects can only
 +     * appear as "values" (and not, for instance, as keys) */
 +     if (o->refcount > 1) return o;
 +
 +    /* Currently we try to encode only strings */
 +    redisAssert(o->type == REDIS_STRING);
 +
 +    /* Check if we can represent this string as a long integer */
 +    if (isStringRepresentableAsLong(s,&value) == REDIS_ERR) return o;
 +
 +    /* Ok, this object can be encoded */
 +    if (value >= 0 && value < REDIS_SHARED_INTEGERS) {
 +        decrRefCount(o);
 +        incrRefCount(shared.integers[value]);
 +        return shared.integers[value];
 +    } else {
 +        o->encoding = REDIS_ENCODING_INT;
 +        sdsfree(o->ptr);
 +        o->ptr = (void*) value;
 +        return o;
 +    }
 +}
 +
 +/* Get a decoded version of an encoded object (returned as a new object).
 + * If the object is already raw-encoded just increment the ref count. */
 +robj *getDecodedObject(robj *o) {
 +    robj *dec;
 +
 +    if (o->encoding == REDIS_ENCODING_RAW) {
 +        incrRefCount(o);
 +        return o;
 +    }
 +    if (o->type == REDIS_STRING && o->encoding == REDIS_ENCODING_INT) {
 +        char buf[32];
 +
 +        ll2string(buf,32,(long)o->ptr);
 +        dec = createStringObject(buf,strlen(buf));
 +        return dec;
 +    } else {
 +        redisPanic("Unknown encoding type");
 +    }
 +}
 +
 +/* Compare two string objects via strcmp() or alike.
 + * Note that the objects may be integer-encoded. In such a case we
 + * use ll2string() to get a string representation of the numbers on the stack
 + * and compare the strings, it's much faster than calling getDecodedObject().
 + *
 + * Important note: if objects are not integer encoded, but binary-safe strings,
 + * sdscmp() from sds.c will apply memcmp() so this function ca be considered
 + * binary safe. */
 +int compareStringObjects(robj *a, robj *b) {
 +    redisAssert(a->type == REDIS_STRING && b->type == REDIS_STRING);
 +    char bufa[128], bufb[128], *astr, *bstr;
 +    int bothsds = 1;
 +
 +    if (a == b) return 0;
 +    if (a->encoding != REDIS_ENCODING_RAW) {
 +        ll2string(bufa,sizeof(bufa),(long) a->ptr);
 +        astr = bufa;
 +        bothsds = 0;
 +    } else {
 +        astr = a->ptr;
 +    }
 +    if (b->encoding != REDIS_ENCODING_RAW) {
 +        ll2string(bufb,sizeof(bufb),(long) b->ptr);
 +        bstr = bufb;
 +        bothsds = 0;
 +    } else {
 +        bstr = b->ptr;
 +    }
 +    return bothsds ? sdscmp(astr,bstr) : strcmp(astr,bstr);
 +}
 +
 +/* Equal string objects return 1 if the two objects are the same from the
 + * point of view of a string comparison, otherwise 0 is returned. Note that
 + * this function is faster then checking for (compareStringObject(a,b) == 0)
 + * because it can perform some more optimization. */
 +int equalStringObjects(robj *a, robj *b) {
 +    if (a->encoding != REDIS_ENCODING_RAW && b->encoding != REDIS_ENCODING_RAW){
 +        return a->ptr == b->ptr;
 +    } else {
 +        return compareStringObjects(a,b) == 0;
 +    }
 +}
 +
 +size_t stringObjectLen(robj *o) {
 +    redisAssert(o->type == REDIS_STRING);
 +    if (o->encoding == REDIS_ENCODING_RAW) {
 +        return sdslen(o->ptr);
 +    } else {
 +        char buf[32];
 +
 +        return ll2string(buf,32,(long)o->ptr);
 +    }
 +}
 +
 +int getDoubleFromObject(robj *o, double *target) {
 +    double value;
 +    char *eptr;
 +
 +    if (o == NULL) {
 +        value = 0;
 +    } else {
 +        redisAssert(o->type == REDIS_STRING);
 +        if (o->encoding == REDIS_ENCODING_RAW) {
 +            value = strtod(o->ptr, &eptr);
 +            if (eptr[0] != '\0') return REDIS_ERR;
 +        } else if (o->encoding == REDIS_ENCODING_INT) {
 +            value = (long)o->ptr;
 +        } else {
 +            redisPanic("Unknown string encoding");
 +        }
 +    }
 +
 +    *target = value;
 +    return REDIS_OK;
 +}
 +
 +int getDoubleFromObjectOrReply(redisClient *c, robj *o, double *target, const char *msg) {
 +    double value;
 +    if (getDoubleFromObject(o, &value) != REDIS_OK) {
 +        if (msg != NULL) {
 +            addReplySds(c, sdscatprintf(sdsempty(), "-ERR %s\r\n", msg));
 +        } else {
 +            addReplySds(c, sdsnew("-ERR value is not a double\r\n"));
 +        }
 +        return REDIS_ERR;
 +    }
 +
 +    *target = value;
 +    return REDIS_OK;
 +}
 +
 +int getLongLongFromObject(robj *o, long long *target) {
 +    long long value;
 +    char *eptr;
 +
 +    if (o == NULL) {
 +        value = 0;
 +    } else {
 +        redisAssert(o->type == REDIS_STRING);
 +        if (o->encoding == REDIS_ENCODING_RAW) {
 +            value = strtoll(o->ptr, &eptr, 10);
 +            if (eptr[0] != '\0') return REDIS_ERR;
 +        } else if (o->encoding == REDIS_ENCODING_INT) {
 +            value = (long)o->ptr;
 +        } else {
 +            redisPanic("Unknown string encoding");
 +        }
 +    }
 +
-     *target = value;
++    if (target) *target = value;
 +    return REDIS_OK;
 +}
 +
 +int getLongLongFromObjectOrReply(redisClient *c, robj *o, long long *target, const char *msg) {
 +    long long value;
 +    if (getLongLongFromObject(o, &value) != REDIS_OK) {
 +        if (msg != NULL) {
 +            addReplySds(c, sdscatprintf(sdsempty(), "-ERR %s\r\n", msg));
 +        } else {
 +            addReplySds(c, sdsnew("-ERR value is not an integer\r\n"));
 +        }
 +        return REDIS_ERR;
 +    }
 +
 +    *target = value;
 +    return REDIS_OK;
 +}
 +
 +int getLongFromObjectOrReply(redisClient *c, robj *o, long *target, const char *msg) {
 +    long long value;
 +
 +    if (getLongLongFromObjectOrReply(c, o, &value, msg) != REDIS_OK) return REDIS_ERR;
 +    if (value < LONG_MIN || value > LONG_MAX) {
 +        if (msg != NULL) {
 +            addReplySds(c, sdscatprintf(sdsempty(), "-ERR %s\r\n", msg));
 +        } else {
 +            addReplySds(c, sdsnew("-ERR value is out of range\r\n"));
 +        }
 +        return REDIS_ERR;
 +    }
 +
 +    *target = value;
 +    return REDIS_OK;
 +}
 +
 +char *strEncoding(int encoding) {
 +    switch(encoding) {
 +    case REDIS_ENCODING_RAW: return "raw";
 +    case REDIS_ENCODING_INT: return "int";
 +    case REDIS_ENCODING_HT: return "hashtable";
 +    case REDIS_ENCODING_ZIPMAP: return "zipmap";
 +    case REDIS_ENCODING_LINKEDLIST: return "linkedlist";
 +    case REDIS_ENCODING_ZIPLIST: return "ziplist";
++    case REDIS_ENCODING_INTSET: return "intset";
 +    default: return "unknown";
 +    }
 +}
diff --cc src/rdb.c
index 509c70c3,00000000..019aa9a0
mode 100644,000000..100644
--- a/src/rdb.c
+++ b/src/rdb.c
@@@ -1,891 -1,0 +1,927 @@@
 +#include "redis.h"
 +#include "lzf.h"    /* LZF compression library */
 +
 +#include <math.h>
 +#include <sys/types.h>
 +#include <sys/time.h>
 +#include <sys/resource.h>
 +#include <sys/wait.h>
 +#include <arpa/inet.h>
 +
 +int rdbSaveType(FILE *fp, unsigned char type) {
 +    if (fwrite(&type,1,1,fp) == 0) return -1;
 +    return 0;
 +}
 +
 +int rdbSaveTime(FILE *fp, time_t t) {
 +    int32_t t32 = (int32_t) t;
 +    if (fwrite(&t32,4,1,fp) == 0) return -1;
 +    return 0;
 +}
 +
 +/* check rdbLoadLen() comments for more info */
 +int rdbSaveLen(FILE *fp, uint32_t len) {
 +    unsigned char buf[2];
 +
 +    if (len < (1<<6)) {
 +        /* Save a 6 bit len */
 +        buf[0] = (len&0xFF)|(REDIS_RDB_6BITLEN<<6);
 +        if (fwrite(buf,1,1,fp) == 0) return -1;
 +    } else if (len < (1<<14)) {
 +        /* Save a 14 bit len */
 +        buf[0] = ((len>>8)&0xFF)|(REDIS_RDB_14BITLEN<<6);
 +        buf[1] = len&0xFF;
 +        if (fwrite(buf,2,1,fp) == 0) return -1;
 +    } else {
 +        /* Save a 32 bit len */
 +        buf[0] = (REDIS_RDB_32BITLEN<<6);
 +        if (fwrite(buf,1,1,fp) == 0) return -1;
 +        len = htonl(len);
 +        if (fwrite(&len,4,1,fp) == 0) return -1;
 +    }
 +    return 0;
 +}
 +
 +/* Encode 'value' as an integer if possible (if integer will fit the
 + * supported range). If the function sucessful encoded the integer
 + * then the (up to 5 bytes) encoded representation is written in the
 + * string pointed by 'enc' and the length is returned. Otherwise
 + * 0 is returned. */
 +int rdbEncodeInteger(long long value, unsigned char *enc) {
 +    /* Finally check if it fits in our ranges */
 +    if (value >= -(1<<7) && value <= (1<<7)-1) {
 +        enc[0] = (REDIS_RDB_ENCVAL<<6)|REDIS_RDB_ENC_INT8;
 +        enc[1] = value&0xFF;
 +        return 2;
 +    } else if (value >= -(1<<15) && value <= (1<<15)-1) {
 +        enc[0] = (REDIS_RDB_ENCVAL<<6)|REDIS_RDB_ENC_INT16;
 +        enc[1] = value&0xFF;
 +        enc[2] = (value>>8)&0xFF;
 +        return 3;
 +    } else if (value >= -((long long)1<<31) && value <= ((long long)1<<31)-1) {
 +        enc[0] = (REDIS_RDB_ENCVAL<<6)|REDIS_RDB_ENC_INT32;
 +        enc[1] = value&0xFF;
 +        enc[2] = (value>>8)&0xFF;
 +        enc[3] = (value>>16)&0xFF;
 +        enc[4] = (value>>24)&0xFF;
 +        return 5;
 +    } else {
 +        return 0;
 +    }
 +}
 +
 +/* String objects in the form "2391" "-100" without any space and with a
 + * range of values that can fit in an 8, 16 or 32 bit signed value can be
 + * encoded as integers to save space */
 +int rdbTryIntegerEncoding(char *s, size_t len, unsigned char *enc) {
 +    long long value;
 +    char *endptr, buf[32];
 +
 +    /* Check if it's possible to encode this value as a number */
 +    value = strtoll(s, &endptr, 10);
 +    if (endptr[0] != '\0') return 0;
 +    ll2string(buf,32,value);
 +
 +    /* If the number converted back into a string is not identical
 +     * then it's not possible to encode the string as integer */
 +    if (strlen(buf) != len || memcmp(buf,s,len)) return 0;
 +
 +    return rdbEncodeInteger(value,enc);
 +}
 +
 +int rdbSaveLzfStringObject(FILE *fp, unsigned char *s, size_t len) {
 +    size_t comprlen, outlen;
 +    unsigned char byte;
 +    void *out;
 +
 +    /* We require at least four bytes compression for this to be worth it */
 +    if (len <= 4) return 0;
 +    outlen = len-4;
 +    if ((out = zmalloc(outlen+1)) == NULL) return 0;
 +    comprlen = lzf_compress(s, len, out, outlen);
 +    if (comprlen == 0) {
 +        zfree(out);
 +        return 0;
 +    }
 +    /* Data compressed! Let's save it on disk */
 +    byte = (REDIS_RDB_ENCVAL<<6)|REDIS_RDB_ENC_LZF;
 +    if (fwrite(&byte,1,1,fp) == 0) goto writeerr;
 +    if (rdbSaveLen(fp,comprlen) == -1) goto writeerr;
 +    if (rdbSaveLen(fp,len) == -1) goto writeerr;
 +    if (fwrite(out,comprlen,1,fp) == 0) goto writeerr;
 +    zfree(out);
 +    return comprlen;
 +
 +writeerr:
 +    zfree(out);
 +    return -1;
 +}
 +
 +/* Save a string objet as [len][data] on disk. If the object is a string
 + * representation of an integer value we try to safe it in a special form */
 +int rdbSaveRawString(FILE *fp, unsigned char *s, size_t len) {
 +    int enclen;
 +
 +    /* Try integer encoding */
 +    if (len <= 11) {
 +        unsigned char buf[5];
 +        if ((enclen = rdbTryIntegerEncoding((char*)s,len,buf)) > 0) {
 +            if (fwrite(buf,enclen,1,fp) == 0) return -1;
 +            return 0;
 +        }
 +    }
 +
 +    /* Try LZF compression - under 20 bytes it's unable to compress even
 +     * aaaaaaaaaaaaaaaaaa so skip it */
 +    if (server.rdbcompression && len > 20) {
 +        int retval;
 +
 +        retval = rdbSaveLzfStringObject(fp,s,len);
 +        if (retval == -1) return -1;
 +        if (retval > 0) return 0;
 +        /* retval == 0 means data can't be compressed, save the old way */
 +    }
 +
 +    /* Store verbatim */
 +    if (rdbSaveLen(fp,len) == -1) return -1;
 +    if (len && fwrite(s,len,1,fp) == 0) return -1;
 +    return 0;
 +}
 +
 +/* Save a long long value as either an encoded string or a string. */
 +int rdbSaveLongLongAsStringObject(FILE *fp, long long value) {
 +    unsigned char buf[32];
 +    int enclen = rdbEncodeInteger(value,buf);
 +    if (enclen > 0) {
 +        if (fwrite(buf,enclen,1,fp) == 0) return -1;
 +    } else {
 +        /* Encode as string */
 +        enclen = ll2string((char*)buf,32,value);
 +        redisAssert(enclen < 32);
 +        if (rdbSaveLen(fp,enclen) == -1) return -1;
 +        if (fwrite(buf,enclen,1,fp) == 0) return -1;
 +    }
 +    return 0;
 +}
 +
 +/* Like rdbSaveStringObjectRaw() but handle encoded objects */
 +int rdbSaveStringObject(FILE *fp, robj *obj) {
 +    /* Avoid to decode the object, then encode it again, if the
 +     * object is alrady integer encoded. */
 +    if (obj->encoding == REDIS_ENCODING_INT) {
 +        return rdbSaveLongLongAsStringObject(fp,(long)obj->ptr);
 +    } else {
 +        redisAssert(obj->encoding == REDIS_ENCODING_RAW);
 +        return rdbSaveRawString(fp,obj->ptr,sdslen(obj->ptr));
 +    }
 +}
 +
 +/* Save a double value. Doubles are saved as strings prefixed by an unsigned
 + * 8 bit integer specifing the length of the representation.
 + * This 8 bit integer has special values in order to specify the following
 + * conditions:
 + * 253: not a number
 + * 254: + inf
 + * 255: - inf
 + */
 +int rdbSaveDoubleValue(FILE *fp, double val) {
 +    unsigned char buf[128];
 +    int len;
 +
 +    if (isnan(val)) {
 +        buf[0] = 253;
 +        len = 1;
 +    } else if (!isfinite(val)) {
 +        len = 1;
 +        buf[0] = (val < 0) ? 255 : 254;
 +    } else {
 +#if (DBL_MANT_DIG >= 52) && (LLONG_MAX == 0x7fffffffffffffffLL)
 +        /* Check if the float is in a safe range to be casted into a
 +         * long long. We are assuming that long long is 64 bit here.
 +         * Also we are assuming that there are no implementations around where
 +         * double has precision < 52 bit.
 +         *
 +         * Under this assumptions we test if a double is inside an interval
 +         * where casting to long long is safe. Then using two castings we
 +         * make sure the decimal part is zero. If all this is true we use
 +         * integer printing function that is much faster. */
 +        double min = -4503599627370495; /* (2^52)-1 */
 +        double max = 4503599627370496; /* -(2^52) */
 +        if (val > min && val < max && val == ((double)((long long)val)))
 +            ll2string((char*)buf+1,sizeof(buf),(long long)val);
 +        else
 +#endif
 +            snprintf((char*)buf+1,sizeof(buf)-1,"%.17g",val);
 +        buf[0] = strlen((char*)buf+1);
 +        len = buf[0]+1;
 +    }
 +    if (fwrite(buf,len,1,fp) == 0) return -1;
 +    return 0;
 +}
 +
 +/* Save a Redis object. */
 +int rdbSaveObject(FILE *fp, robj *o) {
 +    if (o->type == REDIS_STRING) {
 +        /* Save a string value */
 +        if (rdbSaveStringObject(fp,o) == -1) return -1;
 +    } else if (o->type == REDIS_LIST) {
 +        /* Save a list value */
 +        if (o->encoding == REDIS_ENCODING_ZIPLIST) {
 +            unsigned char *p;
 +            unsigned char *vstr;
 +            unsigned int vlen;
 +            long long vlong;
 +
 +            if (rdbSaveLen(fp,ziplistLen(o->ptr)) == -1) return -1;
 +            p = ziplistIndex(o->ptr,0);
 +            while(ziplistGet(p,&vstr,&vlen,&vlong)) {
 +                if (vstr) {
 +                    if (rdbSaveRawString(fp,vstr,vlen) == -1)
 +                        return -1;
 +                } else {
 +                    if (rdbSaveLongLongAsStringObject(fp,vlong) == -1)
 +                        return -1;
 +                }
 +                p = ziplistNext(o->ptr,p);
 +            }
 +        } else if (o->encoding == REDIS_ENCODING_LINKEDLIST) {
 +            list *list = o->ptr;
 +            listIter li;
 +            listNode *ln;
 +
 +            if (rdbSaveLen(fp,listLength(list)) == -1) return -1;
 +            listRewind(list,&li);
 +            while((ln = listNext(&li))) {
 +                robj *eleobj = listNodeValue(ln);
 +                if (rdbSaveStringObject(fp,eleobj) == -1) return -1;
 +            }
 +        } else {
 +            redisPanic("Unknown list encoding");
 +        }
 +    } else if (o->type == REDIS_SET) {
 +        /* Save a set value */
-         dict *set = o->ptr;
-         dictIterator *di = dictGetIterator(set);
-         dictEntry *de;
- 
-         if (rdbSaveLen(fp,dictSize(set)) == -1) return -1;
-         while((de = dictNext(di)) != NULL) {
-             robj *eleobj = dictGetEntryKey(de);
++        if (o->encoding == REDIS_ENCODING_HT) {
++            dict *set = o->ptr;
++            dictIterator *di = dictGetIterator(set);
++            dictEntry *de;
 +
-             if (rdbSaveStringObject(fp,eleobj) == -1) return -1;
++            if (rdbSaveLen(fp,dictSize(set)) == -1) return -1;
++            while((de = dictNext(di)) != NULL) {
++                robj *eleobj = dictGetEntryKey(de);
++                if (rdbSaveStringObject(fp,eleobj) == -1) return -1;
++            }
++            dictReleaseIterator(di);
++        } else if (o->encoding == REDIS_ENCODING_INTSET) {
++            intset *is = o->ptr;
++            long long llval;
++            int i = 0;
++
++            if (rdbSaveLen(fp,intsetLen(is)) == -1) return -1;
++            while(intsetGet(is,i++,&llval)) {
++                if (rdbSaveLongLongAsStringObject(fp,llval) == -1) return -1;
++            }
++        } else {
++            redisPanic("Unknown set encoding");
 +        }
-         dictReleaseIterator(di);
 +    } else if (o->type == REDIS_ZSET) {
 +        /* Save a set value */
 +        zset *zs = o->ptr;
 +        dictIterator *di = dictGetIterator(zs->dict);
 +        dictEntry *de;
 +
 +        if (rdbSaveLen(fp,dictSize(zs->dict)) == -1) return -1;
 +        while((de = dictNext(di)) != NULL) {
 +            robj *eleobj = dictGetEntryKey(de);
 +            double *score = dictGetEntryVal(de);
 +
 +            if (rdbSaveStringObject(fp,eleobj) == -1) return -1;
 +            if (rdbSaveDoubleValue(fp,*score) == -1) return -1;
 +        }
 +        dictReleaseIterator(di);
 +    } else if (o->type == REDIS_HASH) {
 +        /* Save a hash value */
 +        if (o->encoding == REDIS_ENCODING_ZIPMAP) {
 +            unsigned char *p = zipmapRewind(o->ptr);
 +            unsigned int count = zipmapLen(o->ptr);
 +            unsigned char *key, *val;
 +            unsigned int klen, vlen;
 +
 +            if (rdbSaveLen(fp,count) == -1) return -1;
 +            while((p = zipmapNext(p,&key,&klen,&val,&vlen)) != NULL) {
 +                if (rdbSaveRawString(fp,key,klen) == -1) return -1;
 +                if (rdbSaveRawString(fp,val,vlen) == -1) return -1;
 +            }
 +        } else {
 +            dictIterator *di = dictGetIterator(o->ptr);
 +            dictEntry *de;
 +
 +            if (rdbSaveLen(fp,dictSize((dict*)o->ptr)) == -1) return -1;
 +            while((de = dictNext(di)) != NULL) {
 +                robj *key = dictGetEntryKey(de);
 +                robj *val = dictGetEntryVal(de);
 +
 +                if (rdbSaveStringObject(fp,key) == -1) return -1;
 +                if (rdbSaveStringObject(fp,val) == -1) return -1;
 +            }
 +            dictReleaseIterator(di);
 +        }
 +    } else {
 +        redisPanic("Unknown object type");
 +    }
 +    return 0;
 +}
 +
 +/* Return the length the object will have on disk if saved with
 + * the rdbSaveObject() function. Currently we use a trick to get
 + * this length with very little changes to the code. In the future
 + * we could switch to a faster solution. */
 +off_t rdbSavedObjectLen(robj *o, FILE *fp) {
 +    if (fp == NULL) fp = server.devnull;
 +    rewind(fp);
 +    redisAssert(rdbSaveObject(fp,o) != 1);
 +    return ftello(fp);
 +}
 +
 +/* Return the number of pages required to save this object in the swap file */
 +off_t rdbSavedObjectPages(robj *o, FILE *fp) {
 +    off_t bytes = rdbSavedObjectLen(o,fp);
 +
 +    return (bytes+(server.vm_page_size-1))/server.vm_page_size;
 +}
 +
 +/* Save the DB on disk. Return REDIS_ERR on error, REDIS_OK on success */
 +int rdbSave(char *filename) {
 +    dictIterator *di = NULL;
 +    dictEntry *de;
 +    FILE *fp;
 +    char tmpfile[256];
 +    int j;
 +    time_t now = time(NULL);
 +
 +    /* Wait for I/O therads to terminate, just in case this is a
 +     * foreground-saving, to avoid seeking the swap file descriptor at the
 +     * same time. */
 +    if (server.vm_enabled)
 +        waitEmptyIOJobsQueue();
 +
 +    snprintf(tmpfile,256,"temp-%d.rdb", (int) getpid());
 +    fp = fopen(tmpfile,"w");
 +    if (!fp) {
 +        redisLog(REDIS_WARNING, "Failed saving the DB: %s", strerror(errno));
 +        return REDIS_ERR;
 +    }
 +    if (fwrite("REDIS0001",9,1,fp) == 0) goto werr;
 +    for (j = 0; j < server.dbnum; j++) {
 +        redisDb *db = server.db+j;
 +        dict *d = db->dict;
 +        if (dictSize(d) == 0) continue;
 +        di = dictGetIterator(d);
 +        if (!di) {
 +            fclose(fp);
 +            return REDIS_ERR;
 +        }
 +
 +        /* Write the SELECT DB opcode */
 +        if (rdbSaveType(fp,REDIS_SELECTDB) == -1) goto werr;
 +        if (rdbSaveLen(fp,j) == -1) goto werr;
 +
 +        /* Iterate this DB writing every entry */
 +        while((de = dictNext(di)) != NULL) {
 +            sds keystr = dictGetEntryKey(de);
 +            robj key, *o = dictGetEntryVal(de);
 +            time_t expiretime;
 +            
 +            initStaticStringObject(key,keystr);
 +            expiretime = getExpire(db,&key);
 +
 +            /* Save the expire time */
 +            if (expiretime != -1) {
 +                /* If this key is already expired skip it */
 +                if (expiretime < now) continue;
 +                if (rdbSaveType(fp,REDIS_EXPIRETIME) == -1) goto werr;
 +                if (rdbSaveTime(fp,expiretime) == -1) goto werr;
 +            }
 +            /* Save the key and associated value. This requires special
 +             * handling if the value is swapped out. */
 +            if (!server.vm_enabled || o->storage == REDIS_VM_MEMORY ||
 +                                      o->storage == REDIS_VM_SWAPPING) {
 +                /* Save type, key, value */
 +                if (rdbSaveType(fp,o->type) == -1) goto werr;
 +                if (rdbSaveStringObject(fp,&key) == -1) goto werr;
 +                if (rdbSaveObject(fp,o) == -1) goto werr;
 +            } else {
 +                /* REDIS_VM_SWAPPED or REDIS_VM_LOADING */
 +                robj *po;
 +                /* Get a preview of the object in memory */
 +                po = vmPreviewObject(o);
 +                /* Save type, key, value */
 +                if (rdbSaveType(fp,po->type) == -1) goto werr;
 +                if (rdbSaveStringObject(fp,&key) == -1) goto werr;
 +                if (rdbSaveObject(fp,po) == -1) goto werr;
 +                /* Remove the loaded object from memory */
 +                decrRefCount(po);
 +            }
 +        }
 +        dictReleaseIterator(di);
 +    }
 +    /* EOF opcode */
 +    if (rdbSaveType(fp,REDIS_EOF) == -1) goto werr;
 +
 +    /* Make sure data will not remain on the OS's output buffers */
 +    fflush(fp);
 +    fsync(fileno(fp));
 +    fclose(fp);
 +
 +    /* Use RENAME to make sure the DB file is changed atomically only
 +     * if the generate DB file is ok. */
 +    if (rename(tmpfile,filename) == -1) {
 +        redisLog(REDIS_WARNING,"Error moving temp DB file on the final destination: %s", strerror(errno));
 +        unlink(tmpfile);
 +        return REDIS_ERR;
 +    }
 +    redisLog(REDIS_NOTICE,"DB saved on disk");
 +    server.dirty = 0;
 +    server.lastsave = time(NULL);
 +    return REDIS_OK;
 +
 +werr:
 +    fclose(fp);
 +    unlink(tmpfile);
 +    redisLog(REDIS_WARNING,"Write error saving DB on disk: %s", strerror(errno));
 +    if (di) dictReleaseIterator(di);
 +    return REDIS_ERR;
 +}
 +
 +int rdbSaveBackground(char *filename) {
 +    pid_t childpid;
 +
 +    if (server.bgsavechildpid != -1) return REDIS_ERR;
 +    if (server.vm_enabled) waitEmptyIOJobsQueue();
 +    if ((childpid = fork()) == 0) {
 +        /* Child */
 +        if (server.vm_enabled) vmReopenSwapFile();
 +        close(server.fd);
 +        if (rdbSave(filename) == REDIS_OK) {
 +            _exit(0);
 +        } else {
 +            _exit(1);
 +        }
 +    } else {
 +        /* Parent */
 +        if (childpid == -1) {
 +            redisLog(REDIS_WARNING,"Can't save in background: fork: %s",
 +                strerror(errno));
 +            return REDIS_ERR;
 +        }
 +        redisLog(REDIS_NOTICE,"Background saving started by pid %d",childpid);
 +        server.bgsavechildpid = childpid;
 +        updateDictResizePolicy();
 +        return REDIS_OK;
 +    }
 +    return REDIS_OK; /* unreached */
 +}
 +
 +void rdbRemoveTempFile(pid_t childpid) {
 +    char tmpfile[256];
 +
 +    snprintf(tmpfile,256,"temp-%d.rdb", (int) childpid);
 +    unlink(tmpfile);
 +}
 +
 +int rdbLoadType(FILE *fp) {
 +    unsigned char type;
 +    if (fread(&type,1,1,fp) == 0) return -1;
 +    return type;
 +}
 +
 +time_t rdbLoadTime(FILE *fp) {
 +    int32_t t32;
 +    if (fread(&t32,4,1,fp) == 0) return -1;
 +    return (time_t) t32;
 +}
 +
 +/* Load an encoded length from the DB, see the REDIS_RDB_* defines on the top
 + * of this file for a description of how this are stored on disk.
 + *
 + * isencoded is set to 1 if the readed length is not actually a length but
 + * an "encoding type", check the above comments for more info */
 +uint32_t rdbLoadLen(FILE *fp, int *isencoded) {
 +    unsigned char buf[2];
 +    uint32_t len;
 +    int type;
 +
 +    if (isencoded) *isencoded = 0;
 +    if (fread(buf,1,1,fp) == 0) return REDIS_RDB_LENERR;
 +    type = (buf[0]&0xC0)>>6;
 +    if (type == REDIS_RDB_6BITLEN) {
 +        /* Read a 6 bit len */
 +        return buf[0]&0x3F;
 +    } else if (type == REDIS_RDB_ENCVAL) {
 +        /* Read a 6 bit len encoding type */
 +        if (isencoded) *isencoded = 1;
 +        return buf[0]&0x3F;
 +    } else if (type == REDIS_RDB_14BITLEN) {
 +        /* Read a 14 bit len */
 +        if (fread(buf+1,1,1,fp) == 0) return REDIS_RDB_LENERR;
 +        return ((buf[0]&0x3F)<<8)|buf[1];
 +    } else {
 +        /* Read a 32 bit len */
 +        if (fread(&len,4,1,fp) == 0) return REDIS_RDB_LENERR;
 +        return ntohl(len);
 +    }
 +}
 +
 +/* Load an integer-encoded object from file 'fp', with the specified
 + * encoding type 'enctype'. If encode is true the function may return
 + * an integer-encoded object as reply, otherwise the returned object
 + * will always be encoded as a raw string. */
 +robj *rdbLoadIntegerObject(FILE *fp, int enctype, int encode) {
 +    unsigned char enc[4];
 +    long long val;
 +
 +    if (enctype == REDIS_RDB_ENC_INT8) {
 +        if (fread(enc,1,1,fp) == 0) return NULL;
 +        val = (signed char)enc[0];
 +    } else if (enctype == REDIS_RDB_ENC_INT16) {
 +        uint16_t v;
 +        if (fread(enc,2,1,fp) == 0) return NULL;
 +        v = enc[0]|(enc[1]<<8);
 +        val = (int16_t)v;
 +    } else if (enctype == REDIS_RDB_ENC_INT32) {
 +        uint32_t v;
 +        if (fread(enc,4,1,fp) == 0) return NULL;
 +        v = enc[0]|(enc[1]<<8)|(enc[2]<<16)|(enc[3]<<24);
 +        val = (int32_t)v;
 +    } else {
 +        val = 0; /* anti-warning */
 +        redisPanic("Unknown RDB integer encoding type");
 +    }
 +    if (encode)
 +        return createStringObjectFromLongLong(val);
 +    else
 +        return createObject(REDIS_STRING,sdsfromlonglong(val));
 +}
 +
 +robj *rdbLoadLzfStringObject(FILE*fp) {
 +    unsigned int len, clen;
 +    unsigned char *c = NULL;
 +    sds val = NULL;
 +
 +    if ((clen = rdbLoadLen(fp,NULL)) == REDIS_RDB_LENERR) return NULL;
 +    if ((len = rdbLoadLen(fp,NULL)) == REDIS_RDB_LENERR) return NULL;
 +    if ((c = zmalloc(clen)) == NULL) goto err;
 +    if ((val = sdsnewlen(NULL,len)) == NULL) goto err;
 +    if (fread(c,clen,1,fp) == 0) goto err;
 +    if (lzf_decompress(c,clen,val,len) == 0) goto err;
 +    zfree(c);
 +    return createObject(REDIS_STRING,val);
 +err:
 +    zfree(c);
 +    sdsfree(val);
 +    return NULL;
 +}
 +
 +robj *rdbGenericLoadStringObject(FILE*fp, int encode) {
 +    int isencoded;
 +    uint32_t len;
 +    sds val;
 +
 +    len = rdbLoadLen(fp,&isencoded);
 +    if (isencoded) {
 +        switch(len) {
 +        case REDIS_RDB_ENC_INT8:
 +        case REDIS_RDB_ENC_INT16:
 +        case REDIS_RDB_ENC_INT32:
 +            return rdbLoadIntegerObject(fp,len,encode);
 +        case REDIS_RDB_ENC_LZF:
 +            return rdbLoadLzfStringObject(fp);
 +        default:
 +            redisPanic("Unknown RDB encoding type");
 +        }
 +    }
 +
 +    if (len == REDIS_RDB_LENERR) return NULL;
 +    val = sdsnewlen(NULL,len);
 +    if (len && fread(val,len,1,fp) == 0) {
 +        sdsfree(val);
 +        return NULL;
 +    }
 +    return createObject(REDIS_STRING,val);
 +}
 +
 +robj *rdbLoadStringObject(FILE *fp) {
 +    return rdbGenericLoadStringObject(fp,0);
 +}
 +
 +robj *rdbLoadEncodedStringObject(FILE *fp) {
 +    return rdbGenericLoadStringObject(fp,1);
 +}
 +
 +/* For information about double serialization check rdbSaveDoubleValue() */
 +int rdbLoadDoubleValue(FILE *fp, double *val) {
 +    char buf[128];
 +    unsigned char len;
 +
 +    if (fread(&len,1,1,fp) == 0) return -1;
 +    switch(len) {
 +    case 255: *val = R_NegInf; return 0;
 +    case 254: *val = R_PosInf; return 0;
 +    case 253: *val = R_Nan; return 0;
 +    default:
 +        if (fread(buf,len,1,fp) == 0) return -1;
 +        buf[len] = '\0';
 +        sscanf(buf, "%lg", val);
 +        return 0;
 +    }
 +}
 +
 +/* Load a Redis object of the specified type from the specified file.
 + * On success a newly allocated object is returned, otherwise NULL. */
 +robj *rdbLoadObject(int type, FILE *fp) {
 +    robj *o, *ele, *dec;
 +    size_t len;
++    unsigned int i;
 +
 +    redisLog(REDIS_DEBUG,"LOADING OBJECT %d (at %d)\n",type,ftell(fp));
 +    if (type == REDIS_STRING) {
 +        /* Read string value */
 +        if ((o = rdbLoadEncodedStringObject(fp)) == NULL) return NULL;
 +        o = tryObjectEncoding(o);
 +    } else if (type == REDIS_LIST) {
 +        /* Read list value */
 +        if ((len = rdbLoadLen(fp,NULL)) == REDIS_RDB_LENERR) return NULL;
 +
 +        /* Use a real list when there are too many entries */
 +        if (len > server.list_max_ziplist_entries) {
 +            o = createListObject();
 +        } else {
 +            o = createZiplistObject();
 +        }
 +
 +        /* Load every single element of the list */
 +        while(len--) {
 +            if ((ele = rdbLoadEncodedStringObject(fp)) == NULL) return NULL;
 +
 +            /* If we are using a ziplist and the value is too big, convert
 +             * the object to a real list. */
 +            if (o->encoding == REDIS_ENCODING_ZIPLIST &&
 +                ele->encoding == REDIS_ENCODING_RAW &&
 +                sdslen(ele->ptr) > server.list_max_ziplist_value)
 +                    listTypeConvert(o,REDIS_ENCODING_LINKEDLIST);
 +
 +            if (o->encoding == REDIS_ENCODING_ZIPLIST) {
 +                dec = getDecodedObject(ele);
 +                o->ptr = ziplistPush(o->ptr,dec->ptr,sdslen(dec->ptr),REDIS_TAIL);
 +                decrRefCount(dec);
 +                decrRefCount(ele);
 +            } else {
 +                ele = tryObjectEncoding(ele);
 +                listAddNodeTail(o->ptr,ele);
 +            }
 +        }
 +    } else if (type == REDIS_SET) {
 +        /* Read list/set value */
 +        if ((len = rdbLoadLen(fp,NULL)) == REDIS_RDB_LENERR) return NULL;
-         o = createSetObject();
-         /* It's faster to expand the dict to the right size asap in order
-          * to avoid rehashing */
-         if (len > DICT_HT_INITIAL_SIZE)
-             dictExpand(o->ptr,len);
++
++        /* Use a regular set when there are too many entries. */
++        if (len > server.set_max_intset_entries) {
++            o = createSetObject();
++            /* It's faster to expand the dict to the right size asap in order
++             * to avoid rehashing */
++            if (len > DICT_HT_INITIAL_SIZE)
++                dictExpand(o->ptr,len);
++        } else {
++            o = createIntsetObject();
++        }
++
 +        /* Load every single element of the list/set */
-         while(len--) {
++        for (i = 0; i < len; i++) {
++            long long llval;
 +            if ((ele = rdbLoadEncodedStringObject(fp)) == NULL) return NULL;
 +            ele = tryObjectEncoding(ele);
-             dictAdd((dict*)o->ptr,ele,NULL);
++
++            if (o->encoding == REDIS_ENCODING_INTSET) {
++                /* Fetch integer value from element */
++                if (getLongLongFromObject(ele,&llval) == REDIS_OK) {
++                    o->ptr = intsetAdd(o->ptr,llval,NULL);
++                } else {
++                    setTypeConvert(o,REDIS_ENCODING_HT);
++                    dictExpand(o->ptr,len);
++                }
++            }
++
++            /* This will also be called when the set was just converted
++             * to regular hashtable encoded set */
++            if (o->encoding == REDIS_ENCODING_HT) {
++                dictAdd((dict*)o->ptr,ele,NULL);
++            }
 +        }
 +    } else if (type == REDIS_ZSET) {
 +        /* Read list/set value */
 +        size_t zsetlen;
 +        zset *zs;
 +
 +        if ((zsetlen = rdbLoadLen(fp,NULL)) == REDIS_RDB_LENERR) return NULL;
 +        o = createZsetObject();
 +        zs = o->ptr;
 +        /* Load every single element of the list/set */
 +        while(zsetlen--) {
 +            robj *ele;
 +            double *score = zmalloc(sizeof(double));
 +
 +            if ((ele = rdbLoadEncodedStringObject(fp)) == NULL) return NULL;
 +            ele = tryObjectEncoding(ele);
 +            if (rdbLoadDoubleValue(fp,score) == -1) return NULL;
 +            dictAdd(zs->dict,ele,score);
 +            zslInsert(zs->zsl,*score,ele);
 +            incrRefCount(ele); /* added to skiplist */
 +        }
 +    } else if (type == REDIS_HASH) {
 +        size_t hashlen;
 +
 +        if ((hashlen = rdbLoadLen(fp,NULL)) == REDIS_RDB_LENERR) return NULL;
 +        o = createHashObject();
 +        /* Too many entries? Use an hash table. */
 +        if (hashlen > server.hash_max_zipmap_entries)
 +            convertToRealHash(o);
 +        /* Load every key/value, then set it into the zipmap or hash
 +         * table, as needed. */
 +        while(hashlen--) {
 +            robj *key, *val;
 +
 +            if ((key = rdbLoadEncodedStringObject(fp)) == NULL) return NULL;
 +            if ((val = rdbLoadEncodedStringObject(fp)) == NULL) return NULL;
 +            /* If we are using a zipmap and there are too big values
 +             * the object is converted to real hash table encoding. */
 +            if (o->encoding != REDIS_ENCODING_HT &&
 +               ((key->encoding == REDIS_ENCODING_RAW &&
 +                sdslen(key->ptr) > server.hash_max_zipmap_value) ||
 +                (val->encoding == REDIS_ENCODING_RAW &&
 +                sdslen(val->ptr) > server.hash_max_zipmap_value)))
 +            {
 +                    convertToRealHash(o);
 +            }
 +
 +            if (o->encoding == REDIS_ENCODING_ZIPMAP) {
 +                unsigned char *zm = o->ptr;
 +                robj *deckey, *decval;
 +
 +                /* We need raw string objects to add them to the zipmap */
 +                deckey = getDecodedObject(key);
 +                decval = getDecodedObject(val);
 +                zm = zipmapSet(zm,deckey->ptr,sdslen(deckey->ptr),
 +                                  decval->ptr,sdslen(decval->ptr),NULL);
 +                o->ptr = zm;
 +                decrRefCount(deckey);
 +                decrRefCount(decval);
 +                decrRefCount(key);
 +                decrRefCount(val);
 +            } else {
 +                key = tryObjectEncoding(key);
 +                val = tryObjectEncoding(val);
 +                dictAdd((dict*)o->ptr,key,val);
 +            }
 +        }
 +    } else {
 +        redisPanic("Unknown object type");
 +    }
 +    return o;
 +}
 +
 +int rdbLoad(char *filename) {
 +    FILE *fp;
 +    uint32_t dbid;
 +    int type, retval, rdbver;
 +    int swap_all_values = 0;
 +    redisDb *db = server.db+0;
 +    char buf[1024];
 +    time_t expiretime, now = time(NULL);
 +
 +    fp = fopen(filename,"r");
 +    if (!fp) return REDIS_ERR;
 +    if (fread(buf,9,1,fp) == 0) goto eoferr;
 +    buf[9] = '\0';
 +    if (memcmp(buf,"REDIS",5) != 0) {
 +        fclose(fp);
 +        redisLog(REDIS_WARNING,"Wrong signature trying to load DB from file");
 +        return REDIS_ERR;
 +    }
 +    rdbver = atoi(buf+5);
 +    if (rdbver != 1) {
 +        fclose(fp);
 +        redisLog(REDIS_WARNING,"Can't handle RDB format version %d",rdbver);
 +        return REDIS_ERR;
 +    }
 +    while(1) {
 +        robj *key, *val;
 +        int force_swapout;
 +
 +        expiretime = -1;
 +        /* Read type. */
 +        if ((type = rdbLoadType(fp)) == -1) goto eoferr;
 +        if (type == REDIS_EXPIRETIME) {
 +            if ((expiretime = rdbLoadTime(fp)) == -1) goto eoferr;
 +            /* We read the time so we need to read the object type again */
 +            if ((type = rdbLoadType(fp)) == -1) goto eoferr;
 +        }
 +        if (type == REDIS_EOF) break;
 +        /* Handle SELECT DB opcode as a special case */
 +        if (type == REDIS_SELECTDB) {
 +            if ((dbid = rdbLoadLen(fp,NULL)) == REDIS_RDB_LENERR)
 +                goto eoferr;
 +            if (dbid >= (unsigned)server.dbnum) {
 +                redisLog(REDIS_WARNING,"FATAL: Data file was created with a Redis server configured to handle more than %d databases. Exiting\n", server.dbnum);
 +                exit(1);
 +            }
 +            db = server.db+dbid;
 +            continue;
 +        }
 +        /* Read key */
 +        if ((key = rdbLoadStringObject(fp)) == NULL) goto eoferr;
 +        /* Read value */
 +        if ((val = rdbLoadObject(type,fp)) == NULL) goto eoferr;
 +        /* Check if the key already expired */
 +        if (expiretime != -1 && expiretime < now) {
 +            decrRefCount(key);
 +            decrRefCount(val);
 +            continue;
 +        }
 +        /* Add the new object in the hash table */
 +        retval = dbAdd(db,key,val);
 +        if (retval == REDIS_ERR) {
 +            redisLog(REDIS_WARNING,"Loading DB, duplicated key (%s) found! Unrecoverable error, exiting now.", key->ptr);
 +            exit(1);
 +        }
 +        /* Set the expire time if needed */
 +        if (expiretime != -1) setExpire(db,key,expiretime);
 +
 +        /* Handle swapping while loading big datasets when VM is on */
 +
 +        /* If we detecter we are hopeless about fitting something in memory
 +         * we just swap every new key on disk. Directly...
 +         * Note that's important to check for this condition before resorting
 +         * to random sampling, otherwise we may try to swap already
 +         * swapped keys. */
 +        if (swap_all_values) {
 +            dictEntry *de = dictFind(db->dict,key->ptr);
 +
 +            /* de may be NULL since the key already expired */
 +            if (de) {
 +                vmpointer *vp;
 +                val = dictGetEntryVal(de);
 +
 +                if (val->refcount == 1 &&
 +                    (vp = vmSwapObjectBlocking(val)) != NULL)
 +                    dictGetEntryVal(de) = vp;
 +            }
 +            decrRefCount(key);
 +            continue;
 +        }
 +        decrRefCount(key);
 +
 +        /* Flush data on disk once 32 MB of additional RAM are used... */
 +        force_swapout = 0;
 +        if ((zmalloc_used_memory() - server.vm_max_memory) > 1024*1024*32)
 +            force_swapout = 1;
 +
 +        /* If we have still some hope of having some value fitting memory
 +         * then we try random sampling. */
 +        if (!swap_all_values && server.vm_enabled && force_swapout) {
 +            while (zmalloc_used_memory() > server.vm_max_memory) {
 +                if (vmSwapOneObjectBlocking() == REDIS_ERR) break;
 +            }
 +            if (zmalloc_used_memory() > server.vm_max_memory)
 +                swap_all_values = 1; /* We are already using too much mem */
 +        }
 +    }
 +    fclose(fp);
 +    return REDIS_OK;
 +
 +eoferr: /* unexpected end of file is handled here with a fatal exit */
 +    redisLog(REDIS_WARNING,"Short read or OOM loading DB. Unrecoverable error, aborting now.");
 +    exit(1);
 +    return REDIS_ERR; /* Just to avoid warning */
 +}
 +
 +/* A background saving child (BGSAVE) terminated its work. Handle this. */
 +void backgroundSaveDoneHandler(int statloc) {
 +    int exitcode = WEXITSTATUS(statloc);
 +    int bysignal = WIFSIGNALED(statloc);
 +
 +    if (!bysignal && exitcode == 0) {
 +        redisLog(REDIS_NOTICE,
 +            "Background saving terminated with success");
 +        server.dirty = 0;
 +        server.lastsave = time(NULL);
 +    } else if (!bysignal && exitcode != 0) {
 +        redisLog(REDIS_WARNING, "Background saving error");
 +    } else {
 +        redisLog(REDIS_WARNING,
 +            "Background saving terminated by signal %d", WTERMSIG(statloc));
 +        rdbRemoveTempFile(server.bgsavechildpid);
 +    }
 +    server.bgsavechildpid = -1;
 +    /* Possibly there are slaves waiting for a BGSAVE in order to be served
 +     * (the first stage of SYNC is a bulk transfer of dump.rdb) */
 +    updateSlavesWaitingBgsave(exitcode == 0 ? REDIS_OK : REDIS_ERR);
 +}
diff --cc src/redis.c
index 86536b43,00000000..7809d057
mode 100644,000000..100644
--- a/src/redis.c
+++ b/src/redis.c
@@@ -1,1517 -1,0 +1,1518 @@@
 +/*
 + * Copyright (c) 2009-2010, Salvatore Sanfilippo <antirez at gmail dot com>
 + * 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 "redis.h"
 +
 +#ifdef HAVE_BACKTRACE
 +#include <execinfo.h>
 +#include <ucontext.h>
 +#endif /* HAVE_BACKTRACE */
 +
 +#include <time.h>
 +#include <signal.h>
 +#include <sys/wait.h>
 +#include <errno.h>
 +#include <assert.h>
 +#include <ctype.h>
 +#include <stdarg.h>
 +#include <arpa/inet.h>
 +#include <sys/stat.h>
 +#include <fcntl.h>
 +#include <sys/time.h>
 +#include <sys/resource.h>
 +#include <sys/uio.h>
 +#include <limits.h>
 +#include <float.h>
 +#include <math.h>
 +#include <pthread.h>
 +
 +/* Our shared "common" objects */
 +
 +struct sharedObjectsStruct shared;
 +
 +/* Global vars that are actally used as constants. The following double
 + * values are used for double on-disk serialization, and are initialized
 + * at runtime to avoid strange compiler optimizations. */
 +
 +double R_Zero, R_PosInf, R_NegInf, R_Nan;
 +
 +/*================================= Globals ================================= */
 +
 +/* Global vars */
 +struct redisServer server; /* server global state */
 +struct redisCommand *commandTable;
 +struct redisCommand readonlyCommandTable[] = {
 +    {"get",getCommand,2,REDIS_CMD_INLINE,NULL,1,1,1},
 +    {"set",setCommand,3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM,NULL,0,0,0},
 +    {"setnx",setnxCommand,3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM,NULL,0,0,0},
 +    {"setex",setexCommand,4,REDIS_CMD_BULK|REDIS_CMD_DENYOOM,NULL,0,0,0},
 +    {"append",appendCommand,3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM,NULL,1,1,1},
 +    {"substr",substrCommand,4,REDIS_CMD_INLINE,NULL,1,1,1},
 +    {"del",delCommand,-2,REDIS_CMD_INLINE,NULL,0,0,0},
 +    {"exists",existsCommand,2,REDIS_CMD_INLINE,NULL,1,1,1},
 +    {"incr",incrCommand,2,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM,NULL,1,1,1},
 +    {"decr",decrCommand,2,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM,NULL,1,1,1},
 +    {"mget",mgetCommand,-2,REDIS_CMD_INLINE,NULL,1,-1,1},
 +    {"rpush",rpushCommand,3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM,NULL,1,1,1},
 +    {"lpush",lpushCommand,3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM,NULL,1,1,1},
 +    {"rpushx",rpushxCommand,3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM,NULL,1,1,1},
 +    {"lpushx",lpushxCommand,3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM,NULL,1,1,1},
 +    {"linsert",linsertCommand,5,REDIS_CMD_BULK|REDIS_CMD_DENYOOM,NULL,1,1,1},
 +    {"rpop",rpopCommand,2,REDIS_CMD_INLINE,NULL,1,1,1},
 +    {"lpop",lpopCommand,2,REDIS_CMD_INLINE,NULL,1,1,1},
 +    {"brpop",brpopCommand,-3,REDIS_CMD_INLINE,NULL,1,1,1},
 +    {"blpop",blpopCommand,-3,REDIS_CMD_INLINE,NULL,1,1,1},
 +    {"llen",llenCommand,2,REDIS_CMD_INLINE,NULL,1,1,1},
 +    {"lindex",lindexCommand,3,REDIS_CMD_INLINE,NULL,1,1,1},
 +    {"lset",lsetCommand,4,REDIS_CMD_BULK|REDIS_CMD_DENYOOM,NULL,1,1,1},
 +    {"lrange",lrangeCommand,4,REDIS_CMD_INLINE,NULL,1,1,1},
 +    {"ltrim",ltrimCommand,4,REDIS_CMD_INLINE,NULL,1,1,1},
 +    {"lrem",lremCommand,4,REDIS_CMD_BULK,NULL,1,1,1},
 +    {"rpoplpush",rpoplpushcommand,3,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM,NULL,1,2,1},
 +    {"sadd",saddCommand,3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM,NULL,1,1,1},
 +    {"srem",sremCommand,3,REDIS_CMD_BULK,NULL,1,1,1},
 +    {"smove",smoveCommand,4,REDIS_CMD_BULK,NULL,1,2,1},
 +    {"sismember",sismemberCommand,3,REDIS_CMD_BULK,NULL,1,1,1},
 +    {"scard",scardCommand,2,REDIS_CMD_INLINE,NULL,1,1,1},
 +    {"spop",spopCommand,2,REDIS_CMD_INLINE,NULL,1,1,1},
 +    {"srandmember",srandmemberCommand,2,REDIS_CMD_INLINE,NULL,1,1,1},
 +    {"sinter",sinterCommand,-2,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM,NULL,1,-1,1},
 +    {"sinterstore",sinterstoreCommand,-3,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM,NULL,2,-1,1},
 +    {"sunion",sunionCommand,-2,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM,NULL,1,-1,1},
 +    {"sunionstore",sunionstoreCommand,-3,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM,NULL,2,-1,1},
 +    {"sdiff",sdiffCommand,-2,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM,NULL,1,-1,1},
 +    {"sdiffstore",sdiffstoreCommand,-3,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM,NULL,2,-1,1},
 +    {"smembers",sinterCommand,2,REDIS_CMD_INLINE,NULL,1,1,1},
 +    {"zadd",zaddCommand,4,REDIS_CMD_BULK|REDIS_CMD_DENYOOM,NULL,1,1,1},
 +    {"zincrby",zincrbyCommand,4,REDIS_CMD_BULK|REDIS_CMD_DENYOOM,NULL,1,1,1},
 +    {"zrem",zremCommand,3,REDIS_CMD_BULK,NULL,1,1,1},
 +    {"zremrangebyscore",zremrangebyscoreCommand,4,REDIS_CMD_INLINE,NULL,1,1,1},
 +    {"zremrangebyrank",zremrangebyrankCommand,4,REDIS_CMD_INLINE,NULL,1,1,1},
 +    {"zunionstore",zunionstoreCommand,-4,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM,zunionInterBlockClientOnSwappedKeys,0,0,0},
 +    {"zinterstore",zinterstoreCommand,-4,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM,zunionInterBlockClientOnSwappedKeys,0,0,0},
 +    {"zrange",zrangeCommand,-4,REDIS_CMD_INLINE,NULL,1,1,1},
 +    {"zrangebyscore",zrangebyscoreCommand,-4,REDIS_CMD_INLINE,NULL,1,1,1},
 +    {"zcount",zcountCommand,4,REDIS_CMD_INLINE,NULL,1,1,1},
 +    {"zrevrange",zrevrangeCommand,-4,REDIS_CMD_INLINE,NULL,1,1,1},
 +    {"zcard",zcardCommand,2,REDIS_CMD_INLINE,NULL,1,1,1},
 +    {"zscore",zscoreCommand,3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM,NULL,1,1,1},
 +    {"zrank",zrankCommand,3,REDIS_CMD_BULK,NULL,1,1,1},
 +    {"zrevrank",zrevrankCommand,3,REDIS_CMD_BULK,NULL,1,1,1},
 +    {"hset",hsetCommand,4,REDIS_CMD_BULK|REDIS_CMD_DENYOOM,NULL,1,1,1},
 +    {"hsetnx",hsetnxCommand,4,REDIS_CMD_BULK|REDIS_CMD_DENYOOM,NULL,1,1,1},
 +    {"hget",hgetCommand,3,REDIS_CMD_BULK,NULL,1,1,1},
 +    {"hmset",hmsetCommand,-4,REDIS_CMD_BULK|REDIS_CMD_DENYOOM,NULL,1,1,1},
 +    {"hmget",hmgetCommand,-3,REDIS_CMD_BULK,NULL,1,1,1},
 +    {"hincrby",hincrbyCommand,4,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM,NULL,1,1,1},
 +    {"hdel",hdelCommand,3,REDIS_CMD_BULK,NULL,1,1,1},
 +    {"hlen",hlenCommand,2,REDIS_CMD_INLINE,NULL,1,1,1},
 +    {"hkeys",hkeysCommand,2,REDIS_CMD_INLINE,NULL,1,1,1},
 +    {"hvals",hvalsCommand,2,REDIS_CMD_INLINE,NULL,1,1,1},
 +    {"hgetall",hgetallCommand,2,REDIS_CMD_INLINE,NULL,1,1,1},
 +    {"hexists",hexistsCommand,3,REDIS_CMD_BULK,NULL,1,1,1},
 +    {"incrby",incrbyCommand,3,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM,NULL,1,1,1},
 +    {"decrby",decrbyCommand,3,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM,NULL,1,1,1},
 +    {"getset",getsetCommand,3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM,NULL,1,1,1},
 +    {"mset",msetCommand,-3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM,NULL,1,-1,2},
 +    {"msetnx",msetnxCommand,-3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM,NULL,1,-1,2},
 +    {"randomkey",randomkeyCommand,1,REDIS_CMD_INLINE,NULL,0,0,0},
 +    {"select",selectCommand,2,REDIS_CMD_INLINE,NULL,0,0,0},
 +    {"move",moveCommand,3,REDIS_CMD_INLINE,NULL,1,1,1},
 +    {"rename",renameCommand,3,REDIS_CMD_INLINE,NULL,1,1,1},
 +    {"renamenx",renamenxCommand,3,REDIS_CMD_INLINE,NULL,1,1,1},
 +    {"expire",expireCommand,3,REDIS_CMD_INLINE,NULL,0,0,0},
 +    {"expireat",expireatCommand,3,REDIS_CMD_INLINE,NULL,0,0,0},
 +    {"keys",keysCommand,2,REDIS_CMD_INLINE,NULL,0,0,0},
 +    {"dbsize",dbsizeCommand,1,REDIS_CMD_INLINE,NULL,0,0,0},
 +    {"auth",authCommand,2,REDIS_CMD_INLINE,NULL,0,0,0},
 +    {"ping",pingCommand,1,REDIS_CMD_INLINE,NULL,0,0,0},
 +    {"echo",echoCommand,2,REDIS_CMD_BULK,NULL,0,0,0},
 +    {"save",saveCommand,1,REDIS_CMD_INLINE,NULL,0,0,0},
 +    {"bgsave",bgsaveCommand,1,REDIS_CMD_INLINE,NULL,0,0,0},
 +    {"bgrewriteaof",bgrewriteaofCommand,1,REDIS_CMD_INLINE,NULL,0,0,0},
 +    {"shutdown",shutdownCommand,1,REDIS_CMD_INLINE,NULL,0,0,0},
 +    {"lastsave",lastsaveCommand,1,REDIS_CMD_INLINE,NULL,0,0,0},
 +    {"type",typeCommand,2,REDIS_CMD_INLINE,NULL,1,1,1},
 +    {"multi",multiCommand,1,REDIS_CMD_INLINE,NULL,0,0,0},
 +    {"exec",execCommand,1,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM,execBlockClientOnSwappedKeys,0,0,0},
 +    {"discard",discardCommand,1,REDIS_CMD_INLINE,NULL,0,0,0},
 +    {"sync",syncCommand,1,REDIS_CMD_INLINE,NULL,0,0,0},
 +    {"flushdb",flushdbCommand,1,REDIS_CMD_INLINE,NULL,0,0,0},
 +    {"flushall",flushallCommand,1,REDIS_CMD_INLINE,NULL,0,0,0},
 +    {"sort",sortCommand,-2,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM,NULL,1,1,1},
 +    {"info",infoCommand,1,REDIS_CMD_INLINE,NULL,0,0,0},
 +    {"monitor",monitorCommand,1,REDIS_CMD_INLINE,NULL,0,0,0},
 +    {"ttl",ttlCommand,2,REDIS_CMD_INLINE,NULL,1,1,1},
 +    {"slaveof",slaveofCommand,3,REDIS_CMD_INLINE,NULL,0,0,0},
 +    {"debug",debugCommand,-2,REDIS_CMD_INLINE,NULL,0,0,0},
 +    {"config",configCommand,-2,REDIS_CMD_BULK,NULL,0,0,0},
 +    {"subscribe",subscribeCommand,-2,REDIS_CMD_INLINE,NULL,0,0,0},
 +    {"unsubscribe",unsubscribeCommand,-1,REDIS_CMD_INLINE,NULL,0,0,0},
 +    {"psubscribe",psubscribeCommand,-2,REDIS_CMD_INLINE,NULL,0,0,0},
 +    {"punsubscribe",punsubscribeCommand,-1,REDIS_CMD_INLINE,NULL,0,0,0},
 +    {"publish",publishCommand,3,REDIS_CMD_BULK|REDIS_CMD_FORCE_REPLICATION,NULL,0,0,0},
 +    {"watch",watchCommand,-2,REDIS_CMD_INLINE,NULL,0,0,0},
 +    {"unwatch",unwatchCommand,1,REDIS_CMD_INLINE,NULL,0,0,0}
 +};
 +
 +/*============================ Utility functions ============================ */
 +
 +void redisLog(int level, const char *fmt, ...) {
 +    va_list ap;
 +    FILE *fp;
 +
 +    fp = (server.logfile == NULL) ? stdout : fopen(server.logfile,"a");
 +    if (!fp) return;
 +
 +    va_start(ap, fmt);
 +    if (level >= server.verbosity) {
 +        char *c = ".-*#";
 +        char buf[64];
 +        time_t now;
 +
 +        now = time(NULL);
 +        strftime(buf,64,"%d %b %H:%M:%S",localtime(&now));
 +        fprintf(fp,"[%d] %s %c ",(int)getpid(),buf,c[level]);
 +        vfprintf(fp, fmt, ap);
 +        fprintf(fp,"\n");
 +        fflush(fp);
 +    }
 +    va_end(ap);
 +
 +    if (server.logfile) fclose(fp);
 +}
 +
 +/* Redis generally does not try to recover from out of memory conditions
 + * when allocating objects or strings, it is not clear if it will be possible
 + * to report this condition to the client since the networking layer itself
 + * is based on heap allocation for send buffers, so we simply abort.
 + * At least the code will be simpler to read... */
 +void oom(const char *msg) {
 +    redisLog(REDIS_WARNING, "%s: Out of memory\n",msg);
 +    sleep(1);
 +    abort();
 +}
 +
 +/*====================== Hash table type implementation  ==================== */
 +
 +/* This is an hash table type that uses the SDS dynamic strings libary as
 + * keys and radis objects as values (objects can hold SDS strings,
 + * lists, sets). */
 +
 +void dictVanillaFree(void *privdata, void *val)
 +{
 +    DICT_NOTUSED(privdata);
 +    zfree(val);
 +}
 +
 +void dictListDestructor(void *privdata, void *val)
 +{
 +    DICT_NOTUSED(privdata);
 +    listRelease((list*)val);
 +}
 +
 +int dictSdsKeyCompare(void *privdata, const void *key1,
 +        const void *key2)
 +{
 +    int l1,l2;
 +    DICT_NOTUSED(privdata);
 +
 +    l1 = sdslen((sds)key1);
 +    l2 = sdslen((sds)key2);
 +    if (l1 != l2) return 0;
 +    return memcmp(key1, key2, l1) == 0;
 +}
 +
 +void dictRedisObjectDestructor(void *privdata, void *val)
 +{
 +    DICT_NOTUSED(privdata);
 +
 +    if (val == NULL) return; /* Values of swapped out keys as set to NULL */
 +    decrRefCount(val);
 +}
 +
 +void dictSdsDestructor(void *privdata, void *val)
 +{
 +    DICT_NOTUSED(privdata);
 +
 +    sdsfree(val);
 +}
 +
 +int dictObjKeyCompare(void *privdata, const void *key1,
 +        const void *key2)
 +{
 +    const robj *o1 = key1, *o2 = key2;
 +    return dictSdsKeyCompare(privdata,o1->ptr,o2->ptr);
 +}
 +
 +unsigned int dictObjHash(const void *key) {
 +    const robj *o = key;
 +    return dictGenHashFunction(o->ptr, sdslen((sds)o->ptr));
 +}
 +
 +unsigned int dictSdsHash(const void *key) {
 +    return dictGenHashFunction((unsigned char*)key, sdslen((char*)key));
 +}
 +
 +int dictEncObjKeyCompare(void *privdata, const void *key1,
 +        const void *key2)
 +{
 +    robj *o1 = (robj*) key1, *o2 = (robj*) key2;
 +    int cmp;
 +
 +    if (o1->encoding == REDIS_ENCODING_INT &&
 +        o2->encoding == REDIS_ENCODING_INT)
 +            return o1->ptr == o2->ptr;
 +
 +    o1 = getDecodedObject(o1);
 +    o2 = getDecodedObject(o2);
 +    cmp = dictSdsKeyCompare(privdata,o1->ptr,o2->ptr);
 +    decrRefCount(o1);
 +    decrRefCount(o2);
 +    return cmp;
 +}
 +
 +unsigned int dictEncObjHash(const void *key) {
 +    robj *o = (robj*) key;
 +
 +    if (o->encoding == REDIS_ENCODING_RAW) {
 +        return dictGenHashFunction(o->ptr, sdslen((sds)o->ptr));
 +    } else {
 +        if (o->encoding == REDIS_ENCODING_INT) {
 +            char buf[32];
 +            int len;
 +
 +            len = ll2string(buf,32,(long)o->ptr);
 +            return dictGenHashFunction((unsigned char*)buf, len);
 +        } else {
 +            unsigned int hash;
 +
 +            o = getDecodedObject(o);
 +            hash = dictGenHashFunction(o->ptr, sdslen((sds)o->ptr));
 +            decrRefCount(o);
 +            return hash;
 +        }
 +    }
 +}
 +
 +/* Sets type */
 +dictType setDictType = {
 +    dictEncObjHash,            /* hash function */
 +    NULL,                      /* key dup */
 +    NULL,                      /* val dup */
 +    dictEncObjKeyCompare,      /* key compare */
 +    dictRedisObjectDestructor, /* key destructor */
 +    NULL                       /* val destructor */
 +};
 +
 +/* Sorted sets hash (note: a skiplist is used in addition to the hash table) */
 +dictType zsetDictType = {
 +    dictEncObjHash,            /* hash function */
 +    NULL,                      /* key dup */
 +    NULL,                      /* val dup */
 +    dictEncObjKeyCompare,      /* key compare */
 +    dictRedisObjectDestructor, /* key destructor */
 +    dictVanillaFree            /* val destructor of malloc(sizeof(double)) */
 +};
 +
 +/* Db->dict, keys are sds strings, vals are Redis objects. */
 +dictType dbDictType = {
 +    dictSdsHash,                /* hash function */
 +    NULL,                       /* key dup */
 +    NULL,                       /* val dup */
 +    dictSdsKeyCompare,          /* key compare */
 +    dictSdsDestructor,          /* key destructor */
 +    dictRedisObjectDestructor   /* val destructor */
 +};
 +
 +/* Db->expires */
 +dictType keyptrDictType = {
 +    dictSdsHash,               /* hash function */
 +    NULL,                      /* key dup */
 +    NULL,                      /* val dup */
 +    dictSdsKeyCompare,         /* key compare */
 +    NULL,                      /* key destructor */
 +    NULL                       /* val destructor */
 +};
 +
 +/* Hash type hash table (note that small hashes are represented with zimpaps) */
 +dictType hashDictType = {
 +    dictEncObjHash,             /* hash function */
 +    NULL,                       /* key dup */
 +    NULL,                       /* val dup */
 +    dictEncObjKeyCompare,       /* key compare */
 +    dictRedisObjectDestructor,  /* key destructor */
 +    dictRedisObjectDestructor   /* val destructor */
 +};
 +
 +/* Keylist hash table type has unencoded redis objects as keys and
 + * lists as values. It's used for blocking operations (BLPOP) and to
 + * map swapped keys to a list of clients waiting for this keys to be loaded. */
 +dictType keylistDictType = {
 +    dictObjHash,                /* hash function */
 +    NULL,                       /* key dup */
 +    NULL,                       /* val dup */
 +    dictObjKeyCompare,          /* key compare */
 +    dictRedisObjectDestructor,  /* key destructor */
 +    dictListDestructor          /* val destructor */
 +};
 +
 +int htNeedsResize(dict *dict) {
 +    long long size, used;
 +
 +    size = dictSlots(dict);
 +    used = dictSize(dict);
 +    return (size && used && size > DICT_HT_INITIAL_SIZE &&
 +            (used*100/size < REDIS_HT_MINFILL));
 +}
 +
 +/* If the percentage of used slots in the HT reaches REDIS_HT_MINFILL
 + * we resize the hash table to save memory */
 +void tryResizeHashTables(void) {
 +    int j;
 +
 +    for (j = 0; j < server.dbnum; j++) {
 +        if (htNeedsResize(server.db[j].dict))
 +            dictResize(server.db[j].dict);
 +        if (htNeedsResize(server.db[j].expires))
 +            dictResize(server.db[j].expires);
 +    }
 +}
 +
 +/* Our hash table implementation performs rehashing incrementally while
 + * we write/read from the hash table. Still if the server is idle, the hash
 + * table will use two tables for a long time. So we try to use 1 millisecond
 + * of CPU time at every serverCron() loop in order to rehash some key. */
 +void incrementallyRehash(void) {
 +    int j;
 +
 +    for (j = 0; j < server.dbnum; j++) {
 +        if (dictIsRehashing(server.db[j].dict)) {
 +            dictRehashMilliseconds(server.db[j].dict,1);
 +            break; /* already used our millisecond for this loop... */
 +        }
 +    }
 +}
 +
 +/* This function is called once a background process of some kind terminates,
 + * as we want to avoid resizing the hash tables when there is a child in order
 + * to play well with copy-on-write (otherwise when a resize happens lots of
 + * memory pages are copied). The goal of this function is to update the ability
 + * for dict.c to resize the hash tables accordingly to the fact we have o not
 + * running childs. */
 +void updateDictResizePolicy(void) {
 +    if (server.bgsavechildpid == -1 && server.bgrewritechildpid == -1)
 +        dictEnableResize();
 +    else
 +        dictDisableResize();
 +}
 +
 +/* ======================= Cron: called every 100 ms ======================== */
 +
 +int serverCron(struct aeEventLoop *eventLoop, long long id, void *clientData) {
 +    int j, loops = server.cronloops++;
 +    REDIS_NOTUSED(eventLoop);
 +    REDIS_NOTUSED(id);
 +    REDIS_NOTUSED(clientData);
 +
 +    /* We take a cached value of the unix time in the global state because
 +     * with virtual memory and aging there is to store the current time
 +     * in objects at every object access, and accuracy is not needed.
 +     * To access a global var is faster than calling time(NULL) */
 +    server.unixtime = time(NULL);
 +    /* We have just 21 bits per object for LRU information.
 +     * So we use an (eventually wrapping) LRU clock with minutes resolution.
 +     *
 +     * When we need to select what object to swap, we compute the minimum
 +     * time distance between the current lruclock and the object last access
 +     * lruclock info. Even if clocks will wrap on overflow, there is
 +     * the interesting property that we are sure that at least
 +     * ABS(A-B) minutes passed between current time and timestamp B.
 +     *
 +     * This is not precise but we don't need at all precision, but just
 +     * something statistically reasonable.
 +     */
 +    server.lruclock = (time(NULL)/60)&((1<<21)-1);
 +
 +    /* We received a SIGTERM, shutting down here in a safe way, as it is
 +     * not ok doing so inside the signal handler. */
 +    if (server.shutdown_asap) {
 +        if (prepareForShutdown() == REDIS_OK) exit(0);
 +        redisLog(REDIS_WARNING,"SIGTERM received but errors trying to shut down the server, check the logs for more information");
 +    }
 +
 +    /* Show some info about non-empty databases */
 +    for (j = 0; j < server.dbnum; j++) {
 +        long long size, used, vkeys;
 +
 +        size = dictSlots(server.db[j].dict);
 +        used = dictSize(server.db[j].dict);
 +        vkeys = dictSize(server.db[j].expires);
 +        if (!(loops % 50) && (used || vkeys)) {
 +            redisLog(REDIS_VERBOSE,"DB %d: %lld keys (%lld volatile) in %lld slots HT.",j,used,vkeys,size);
 +            /* dictPrintStats(server.dict); */
 +        }
 +    }
 +
 +    /* We don't want to resize the hash tables while a bacground saving
 +     * is in progress: the saving child is created using fork() that is
 +     * implemented with a copy-on-write semantic in most modern systems, so
 +     * if we resize the HT while there is the saving child at work actually
 +     * a lot of memory movements in the parent will cause a lot of pages
 +     * copied. */
 +    if (server.bgsavechildpid == -1 && server.bgrewritechildpid == -1) {
 +        if (!(loops % 10)) tryResizeHashTables();
 +        if (server.activerehashing) incrementallyRehash();
 +    }
 +
 +    /* Show information about connected clients */
 +    if (!(loops % 50)) {
 +        redisLog(REDIS_VERBOSE,"%d clients connected (%d slaves), %zu bytes in use",
 +            listLength(server.clients)-listLength(server.slaves),
 +            listLength(server.slaves),
 +            zmalloc_used_memory());
 +    }
 +
 +    /* Close connections of timedout clients */
 +    if ((server.maxidletime && !(loops % 100)) || server.blpop_blocked_clients)
 +        closeTimedoutClients();
 +
 +    /* Check if a background saving or AOF rewrite in progress terminated */
 +    if (server.bgsavechildpid != -1 || server.bgrewritechildpid != -1) {
 +        int statloc;
 +        pid_t pid;
 +
 +        if ((pid = wait3(&statloc,WNOHANG,NULL)) != 0) {
 +            if (pid == server.bgsavechildpid) {
 +                backgroundSaveDoneHandler(statloc);
 +            } else {
 +                backgroundRewriteDoneHandler(statloc);
 +            }
 +            updateDictResizePolicy();
 +        }
 +    } else {
 +        /* If there is not a background saving in progress check if
 +         * we have to save now */
 +         time_t now = time(NULL);
 +         for (j = 0; j < server.saveparamslen; j++) {
 +            struct saveparam *sp = server.saveparams+j;
 +
 +            if (server.dirty >= sp->changes &&
 +                now-server.lastsave > sp->seconds) {
 +                redisLog(REDIS_NOTICE,"%d changes in %d seconds. Saving...",
 +                    sp->changes, sp->seconds);
 +                rdbSaveBackground(server.dbfilename);
 +                break;
 +            }
 +         }
 +    }
 +
 +    /* Try to expire a few timed out keys. The algorithm used is adaptive and
 +     * will use few CPU cycles if there are few expiring keys, otherwise
 +     * it will get more aggressive to avoid that too much memory is used by
 +     * keys that can be removed from the keyspace. */
 +    for (j = 0; j < server.dbnum; j++) {
 +        int expired;
 +        redisDb *db = server.db+j;
 +
 +        /* Continue to expire if at the end of the cycle more than 25%
 +         * of the keys were expired. */
 +        do {
 +            long num = dictSize(db->expires);
 +            time_t now = time(NULL);
 +
 +            expired = 0;
 +            if (num > REDIS_EXPIRELOOKUPS_PER_CRON)
 +                num = REDIS_EXPIRELOOKUPS_PER_CRON;
 +            while (num--) {
 +                dictEntry *de;
 +                time_t t;
 +
 +                if ((de = dictGetRandomKey(db->expires)) == NULL) break;
 +                t = (time_t) dictGetEntryVal(de);
 +                if (now > t) {
 +                    sds key = dictGetEntryKey(de);
 +                    robj *keyobj = createStringObject(key,sdslen(key));
 +
 +                    dbDelete(db,keyobj);
 +                    decrRefCount(keyobj);
 +                    expired++;
 +                    server.stat_expiredkeys++;
 +                }
 +            }
 +        } while (expired > REDIS_EXPIRELOOKUPS_PER_CRON/4);
 +    }
 +
 +    /* Swap a few keys on disk if we are over the memory limit and VM
 +     * is enbled. Try to free objects from the free list first. */
 +    if (vmCanSwapOut()) {
 +        while (server.vm_enabled && zmalloc_used_memory() >
 +                server.vm_max_memory)
 +        {
 +            int retval;
 +
 +            if (tryFreeOneObjectFromFreelist() == REDIS_OK) continue;
 +            retval = (server.vm_max_threads == 0) ?
 +                        vmSwapOneObjectBlocking() :
 +                        vmSwapOneObjectThreaded();
 +            if (retval == REDIS_ERR && !(loops % 300) &&
 +                zmalloc_used_memory() >
 +                (server.vm_max_memory+server.vm_max_memory/10))
 +            {
 +                redisLog(REDIS_WARNING,"WARNING: vm-max-memory limit exceeded by more than 10%% but unable to swap more objects out!");
 +            }
 +            /* Note that when using threade I/O we free just one object,
 +             * because anyway when the I/O thread in charge to swap this
 +             * object out will finish, the handler of completed jobs
 +             * will try to swap more objects if we are still out of memory. */
 +            if (retval == REDIS_ERR || server.vm_max_threads > 0) break;
 +        }
 +    }
 +
 +    /* Check if we should connect to a MASTER */
 +    if (server.replstate == REDIS_REPL_CONNECT && !(loops % 10)) {
 +        redisLog(REDIS_NOTICE,"Connecting to MASTER...");
 +        if (syncWithMaster() == REDIS_OK) {
 +            redisLog(REDIS_NOTICE,"MASTER <-> SLAVE sync succeeded");
 +            if (server.appendonly) rewriteAppendOnlyFileBackground();
 +        }
 +    }
 +    return 100;
 +}
 +
 +/* This function gets called every time Redis is entering the
 + * main loop of the event driven library, that is, before to sleep
 + * for ready file descriptors. */
 +void beforeSleep(struct aeEventLoop *eventLoop) {
 +    REDIS_NOTUSED(eventLoop);
 +
 +    /* Awake clients that got all the swapped keys they requested */
 +    if (server.vm_enabled && listLength(server.io_ready_clients)) {
 +        listIter li;
 +        listNode *ln;
 +
 +        listRewind(server.io_ready_clients,&li);
 +        while((ln = listNext(&li))) {
 +            redisClient *c = ln->value;
 +            struct redisCommand *cmd;
 +
 +            /* Resume the client. */
 +            listDelNode(server.io_ready_clients,ln);
 +            c->flags &= (~REDIS_IO_WAIT);
 +            server.vm_blocked_clients--;
 +            aeCreateFileEvent(server.el, c->fd, AE_READABLE,
 +                readQueryFromClient, c);
 +            cmd = lookupCommand(c->argv[0]->ptr);
 +            redisAssert(cmd != NULL);
 +            call(c,cmd);
 +            resetClient(c);
 +            /* There may be more data to process in the input buffer. */
 +            if (c->querybuf && sdslen(c->querybuf) > 0)
 +                processInputBuffer(c);
 +        }
 +    }
 +    /* Write the AOF buffer on disk */
 +    flushAppendOnlyFile();
 +}
 +
 +/* =========================== Server initialization ======================== */
 +
 +void createSharedObjects(void) {
 +    int j;
 +
 +    shared.crlf = createObject(REDIS_STRING,sdsnew("\r\n"));
 +    shared.ok = createObject(REDIS_STRING,sdsnew("+OK\r\n"));
 +    shared.err = createObject(REDIS_STRING,sdsnew("-ERR\r\n"));
 +    shared.emptybulk = createObject(REDIS_STRING,sdsnew("$0\r\n\r\n"));
 +    shared.czero = createObject(REDIS_STRING,sdsnew(":0\r\n"));
 +    shared.cone = createObject(REDIS_STRING,sdsnew(":1\r\n"));
 +    shared.cnegone = createObject(REDIS_STRING,sdsnew(":-1\r\n"));
 +    shared.nullbulk = createObject(REDIS_STRING,sdsnew("$-1\r\n"));
 +    shared.nullmultibulk = createObject(REDIS_STRING,sdsnew("*-1\r\n"));
 +    shared.emptymultibulk = createObject(REDIS_STRING,sdsnew("*0\r\n"));
 +    shared.pong = createObject(REDIS_STRING,sdsnew("+PONG\r\n"));
 +    shared.queued = createObject(REDIS_STRING,sdsnew("+QUEUED\r\n"));
 +    shared.wrongtypeerr = createObject(REDIS_STRING,sdsnew(
 +        "-ERR Operation against a key holding the wrong kind of value\r\n"));
 +    shared.nokeyerr = createObject(REDIS_STRING,sdsnew(
 +        "-ERR no such key\r\n"));
 +    shared.syntaxerr = createObject(REDIS_STRING,sdsnew(
 +        "-ERR syntax error\r\n"));
 +    shared.sameobjecterr = createObject(REDIS_STRING,sdsnew(
 +        "-ERR source and destination objects are the same\r\n"));
 +    shared.outofrangeerr = createObject(REDIS_STRING,sdsnew(
 +        "-ERR index out of range\r\n"));
 +    shared.space = createObject(REDIS_STRING,sdsnew(" "));
 +    shared.colon = createObject(REDIS_STRING,sdsnew(":"));
 +    shared.plus = createObject(REDIS_STRING,sdsnew("+"));
 +    shared.select0 = createStringObject("select 0\r\n",10);
 +    shared.select1 = createStringObject("select 1\r\n",10);
 +    shared.select2 = createStringObject("select 2\r\n",10);
 +    shared.select3 = createStringObject("select 3\r\n",10);
 +    shared.select4 = createStringObject("select 4\r\n",10);
 +    shared.select5 = createStringObject("select 5\r\n",10);
 +    shared.select6 = createStringObject("select 6\r\n",10);
 +    shared.select7 = createStringObject("select 7\r\n",10);
 +    shared.select8 = createStringObject("select 8\r\n",10);
 +    shared.select9 = createStringObject("select 9\r\n",10);
 +    shared.messagebulk = createStringObject("$7\r\nmessage\r\n",13);
 +    shared.pmessagebulk = createStringObject("$8\r\npmessage\r\n",14);
 +    shared.subscribebulk = createStringObject("$9\r\nsubscribe\r\n",15);
 +    shared.unsubscribebulk = createStringObject("$11\r\nunsubscribe\r\n",18);
 +    shared.psubscribebulk = createStringObject("$10\r\npsubscribe\r\n",17);
 +    shared.punsubscribebulk = createStringObject("$12\r\npunsubscribe\r\n",19);
 +    shared.mbulk3 = createStringObject("*3\r\n",4);
 +    shared.mbulk4 = createStringObject("*4\r\n",4);
 +    for (j = 0; j < REDIS_SHARED_INTEGERS; j++) {
 +        shared.integers[j] = createObject(REDIS_STRING,(void*)(long)j);
 +        shared.integers[j]->encoding = REDIS_ENCODING_INT;
 +    }
 +}
 +
 +void initServerConfig() {
 +    server.dbnum = REDIS_DEFAULT_DBNUM;
 +    server.port = REDIS_SERVERPORT;
 +    server.verbosity = REDIS_VERBOSE;
 +    server.maxidletime = REDIS_MAXIDLETIME;
 +    server.saveparams = NULL;
 +    server.logfile = NULL; /* NULL = log on standard output */
 +    server.bindaddr = NULL;
 +    server.glueoutputbuf = 1;
 +    server.daemonize = 0;
 +    server.appendonly = 0;
 +    server.appendfsync = APPENDFSYNC_EVERYSEC;
 +    server.no_appendfsync_on_rewrite = 0;
 +    server.lastfsync = time(NULL);
 +    server.appendfd = -1;
 +    server.appendseldb = -1; /* Make sure the first time will not match */
 +    server.pidfile = zstrdup("/var/run/redis.pid");
 +    server.dbfilename = zstrdup("dump.rdb");
 +    server.appendfilename = zstrdup("appendonly.aof");
 +    server.requirepass = NULL;
 +    server.rdbcompression = 1;
 +    server.activerehashing = 1;
 +    server.maxclients = 0;
 +    server.blpop_blocked_clients = 0;
 +    server.maxmemory = 0;
 +    server.vm_enabled = 0;
 +    server.vm_swap_file = zstrdup("/tmp/redis-%p.vm");
 +    server.vm_page_size = 256;          /* 256 bytes per page */
 +    server.vm_pages = 1024*1024*100;    /* 104 millions of pages */
 +    server.vm_max_memory = 1024LL*1024*1024*1; /* 1 GB of RAM */
 +    server.vm_max_threads = 4;
 +    server.vm_blocked_clients = 0;
 +    server.hash_max_zipmap_entries = REDIS_HASH_MAX_ZIPMAP_ENTRIES;
 +    server.hash_max_zipmap_value = REDIS_HASH_MAX_ZIPMAP_VALUE;
 +    server.list_max_ziplist_entries = REDIS_LIST_MAX_ZIPLIST_ENTRIES;
 +    server.list_max_ziplist_value = REDIS_LIST_MAX_ZIPLIST_VALUE;
++    server.set_max_intset_entries = REDIS_SET_MAX_INTSET_ENTRIES;
 +    server.shutdown_asap = 0;
 +
 +    resetServerSaveParams();
 +
 +    appendServerSaveParams(60*60,1);  /* save after 1 hour and 1 change */
 +    appendServerSaveParams(300,100);  /* save after 5 minutes and 100 changes */
 +    appendServerSaveParams(60,10000); /* save after 1 minute and 10000 changes */
 +    /* Replication related */
 +    server.isslave = 0;
 +    server.masterauth = NULL;
 +    server.masterhost = NULL;
 +    server.masterport = 6379;
 +    server.master = NULL;
 +    server.replstate = REDIS_REPL_NONE;
 +
 +    /* Double constants initialization */
 +    R_Zero = 0.0;
 +    R_PosInf = 1.0/R_Zero;
 +    R_NegInf = -1.0/R_Zero;
 +    R_Nan = R_Zero/R_Zero;
 +}
 +
 +void initServer() {
 +    int j;
 +
 +    signal(SIGHUP, SIG_IGN);
 +    signal(SIGPIPE, SIG_IGN);
 +    setupSigSegvAction();
 +
 +    server.devnull = fopen("/dev/null","w");
 +    if (server.devnull == NULL) {
 +        redisLog(REDIS_WARNING, "Can't open /dev/null: %s", server.neterr);
 +        exit(1);
 +    }
 +    server.clients = listCreate();
 +    server.slaves = listCreate();
 +    server.monitors = listCreate();
 +    server.objfreelist = listCreate();
 +    createSharedObjects();
 +    server.el = aeCreateEventLoop();
 +    server.db = zmalloc(sizeof(redisDb)*server.dbnum);
 +    server.fd = anetTcpServer(server.neterr, server.port, server.bindaddr);
 +    if (server.fd == -1) {
 +        redisLog(REDIS_WARNING, "Opening TCP port: %s", server.neterr);
 +        exit(1);
 +    }
 +    for (j = 0; j < server.dbnum; j++) {
 +        server.db[j].dict = dictCreate(&dbDictType,NULL);
 +        server.db[j].expires = dictCreate(&keyptrDictType,NULL);
 +        server.db[j].blocking_keys = dictCreate(&keylistDictType,NULL);
 +        server.db[j].watched_keys = dictCreate(&keylistDictType,NULL);
 +        if (server.vm_enabled)
 +            server.db[j].io_keys = dictCreate(&keylistDictType,NULL);
 +        server.db[j].id = j;
 +    }
 +    server.pubsub_channels = dictCreate(&keylistDictType,NULL);
 +    server.pubsub_patterns = listCreate();
 +    listSetFreeMethod(server.pubsub_patterns,freePubsubPattern);
 +    listSetMatchMethod(server.pubsub_patterns,listMatchPubsubPattern);
 +    server.cronloops = 0;
 +    server.bgsavechildpid = -1;
 +    server.bgrewritechildpid = -1;
 +    server.bgrewritebuf = sdsempty();
 +    server.aofbuf = sdsempty();
 +    server.lastsave = time(NULL);
 +    server.dirty = 0;
 +    server.stat_numcommands = 0;
 +    server.stat_numconnections = 0;
 +    server.stat_expiredkeys = 0;
 +    server.stat_starttime = time(NULL);
 +    server.unixtime = time(NULL);
 +    aeCreateTimeEvent(server.el, 1, serverCron, NULL, NULL);
 +    if (aeCreateFileEvent(server.el, server.fd, AE_READABLE,
 +        acceptHandler, NULL) == AE_ERR) oom("creating file event");
 +
 +    if (server.appendonly) {
 +        server.appendfd = open(server.appendfilename,O_WRONLY|O_APPEND|O_CREAT,0644);
 +        if (server.appendfd == -1) {
 +            redisLog(REDIS_WARNING, "Can't open the append-only file: %s",
 +                strerror(errno));
 +            exit(1);
 +        }
 +    }
 +
 +    if (server.vm_enabled) vmInit();
 +}
 +
 +int qsortRedisCommands(const void *r1, const void *r2) {
 +    return strcasecmp(
 +        ((struct redisCommand*)r1)->name,
 +        ((struct redisCommand*)r2)->name);
 +}
 +
 +void sortCommandTable() {
 +    /* Copy and sort the read-only version of the command table */
 +    commandTable = (struct redisCommand*)malloc(sizeof(readonlyCommandTable));
 +    memcpy(commandTable,readonlyCommandTable,sizeof(readonlyCommandTable));
 +    qsort(commandTable,
 +        sizeof(readonlyCommandTable)/sizeof(struct redisCommand),
 +        sizeof(struct redisCommand),qsortRedisCommands);
 +}
 +
 +/* ====================== Commands lookup and execution ===================== */
 +
 +struct redisCommand *lookupCommand(char *name) {
 +    struct redisCommand tmp = {name,NULL,0,0,NULL,0,0,0};
 +    return bsearch(
 +        &tmp,
 +        commandTable,
 +        sizeof(readonlyCommandTable)/sizeof(struct redisCommand),
 +        sizeof(struct redisCommand),
 +        qsortRedisCommands);
 +}
 +
 +/* Call() is the core of Redis execution of a command */
 +void call(redisClient *c, struct redisCommand *cmd) {
 +    long long dirty;
 +
 +    dirty = server.dirty;
 +    cmd->proc(c);
 +    dirty = server.dirty-dirty;
 +
 +    if (server.appendonly && dirty)
 +        feedAppendOnlyFile(cmd,c->db->id,c->argv,c->argc);
 +    if ((dirty || cmd->flags & REDIS_CMD_FORCE_REPLICATION) &&
 +        listLength(server.slaves))
 +        replicationFeedSlaves(server.slaves,c->db->id,c->argv,c->argc);
 +    if (listLength(server.monitors))
 +        replicationFeedMonitors(server.monitors,c->db->id,c->argv,c->argc);
 +    server.stat_numcommands++;
 +}
 +
 +/* If this function gets called we already read a whole
 + * command, argments are in the client argv/argc fields.
 + * processCommand() execute the command or prepare the
 + * server for a bulk read from the client.
 + *
 + * If 1 is returned the client is still alive and valid and
 + * and other operations can be performed by the caller. Otherwise
 + * if 0 is returned the client was destroied (i.e. after QUIT). */
 +int processCommand(redisClient *c) {
 +    struct redisCommand *cmd;
 +
 +    /* Free some memory if needed (maxmemory setting) */
 +    if (server.maxmemory) freeMemoryIfNeeded();
 +
 +    /* Handle the multi bulk command type. This is an alternative protocol
 +     * supported by Redis in order to receive commands that are composed of
 +     * multiple binary-safe "bulk" arguments. The latency of processing is
 +     * a bit higher but this allows things like multi-sets, so if this
 +     * protocol is used only for MSET and similar commands this is a big win. */
 +    if (c->multibulk == 0 && c->argc == 1 && ((char*)(c->argv[0]->ptr))[0] == '*') {
 +        c->multibulk = atoi(((char*)c->argv[0]->ptr)+1);
 +        if (c->multibulk <= 0) {
 +            resetClient(c);
 +            return 1;
 +        } else {
 +            decrRefCount(c->argv[c->argc-1]);
 +            c->argc--;
 +            return 1;
 +        }
 +    } else if (c->multibulk) {
 +        if (c->bulklen == -1) {
 +            if (((char*)c->argv[0]->ptr)[0] != '$') {
 +                addReplySds(c,sdsnew("-ERR multi bulk protocol error\r\n"));
 +                resetClient(c);
 +                return 1;
 +            } else {
 +                int bulklen = atoi(((char*)c->argv[0]->ptr)+1);
 +                decrRefCount(c->argv[0]);
 +                if (bulklen < 0 || bulklen > 1024*1024*1024) {
 +                    c->argc--;
 +                    addReplySds(c,sdsnew("-ERR invalid bulk write count\r\n"));
 +                    resetClient(c);
 +                    return 1;
 +                }
 +                c->argc--;
 +                c->bulklen = bulklen+2; /* add two bytes for CR+LF */
 +                return 1;
 +            }
 +        } else {
 +            c->mbargv = zrealloc(c->mbargv,(sizeof(robj*))*(c->mbargc+1));
 +            c->mbargv[c->mbargc] = c->argv[0];
 +            c->mbargc++;
 +            c->argc--;
 +            c->multibulk--;
 +            if (c->multibulk == 0) {
 +                robj **auxargv;
 +                int auxargc;
 +
 +                /* Here we need to swap the multi-bulk argc/argv with the
 +                 * normal argc/argv of the client structure. */
 +                auxargv = c->argv;
 +                c->argv = c->mbargv;
 +                c->mbargv = auxargv;
 +
 +                auxargc = c->argc;
 +                c->argc = c->mbargc;
 +                c->mbargc = auxargc;
 +
 +                /* We need to set bulklen to something different than -1
 +                 * in order for the code below to process the command without
 +                 * to try to read the last argument of a bulk command as
 +                 * a special argument. */
 +                c->bulklen = 0;
 +                /* continue below and process the command */
 +            } else {
 +                c->bulklen = -1;
 +                return 1;
 +            }
 +        }
 +    }
 +    /* -- end of multi bulk commands processing -- */
 +
 +    /* The QUIT command is handled as a special case. Normal command
 +     * procs are unable to close the client connection safely */
 +    if (!strcasecmp(c->argv[0]->ptr,"quit")) {
 +        freeClient(c);
 +        return 0;
 +    }
 +
 +    /* Now lookup the command and check ASAP about trivial error conditions
 +     * such wrong arity, bad command name and so forth. */
 +    cmd = lookupCommand(c->argv[0]->ptr);
 +    if (!cmd) {
 +        addReplySds(c,
 +            sdscatprintf(sdsempty(), "-ERR unknown command '%s'\r\n",
 +                (char*)c->argv[0]->ptr));
 +        resetClient(c);
 +        return 1;
 +    } else if ((cmd->arity > 0 && cmd->arity != c->argc) ||
 +               (c->argc < -cmd->arity)) {
 +        addReplySds(c,
 +            sdscatprintf(sdsempty(),
 +                "-ERR wrong number of arguments for '%s' command\r\n",
 +                cmd->name));
 +        resetClient(c);
 +        return 1;
 +    } else if (cmd->flags & REDIS_CMD_BULK && c->bulklen == -1) {
 +        /* This is a bulk command, we have to read the last argument yet. */
 +        int bulklen = atoi(c->argv[c->argc-1]->ptr);
 +
 +        decrRefCount(c->argv[c->argc-1]);
 +        if (bulklen < 0 || bulklen > 1024*1024*1024) {
 +            c->argc--;
 +            addReplySds(c,sdsnew("-ERR invalid bulk write count\r\n"));
 +            resetClient(c);
 +            return 1;
 +        }
 +        c->argc--;
 +        c->bulklen = bulklen+2; /* add two bytes for CR+LF */
 +        /* It is possible that the bulk read is already in the
 +         * buffer. Check this condition and handle it accordingly.
 +         * This is just a fast path, alternative to call processInputBuffer().
 +         * It's a good idea since the code is small and this condition
 +         * happens most of the times. */
 +        if ((signed)sdslen(c->querybuf) >= c->bulklen) {
 +            c->argv[c->argc] = createStringObject(c->querybuf,c->bulklen-2);
 +            c->argc++;
 +            c->querybuf = sdsrange(c->querybuf,c->bulklen,-1);
 +        } else {
 +            /* Otherwise return... there is to read the last argument
 +             * from the socket. */
 +            return 1;
 +        }
 +    }
 +    /* Let's try to encode the bulk object to save space. */
 +    if (cmd->flags & REDIS_CMD_BULK)
 +        c->argv[c->argc-1] = tryObjectEncoding(c->argv[c->argc-1]);
 +
 +    /* Check if the user is authenticated */
 +    if (server.requirepass && !c->authenticated && cmd->proc != authCommand) {
 +        addReplySds(c,sdsnew("-ERR operation not permitted\r\n"));
 +        resetClient(c);
 +        return 1;
 +    }
 +
 +    /* Handle the maxmemory directive */
 +    if (server.maxmemory && (cmd->flags & REDIS_CMD_DENYOOM) &&
 +        zmalloc_used_memory() > server.maxmemory)
 +    {
 +        addReplySds(c,sdsnew("-ERR command not allowed when used memory > 'maxmemory'\r\n"));
 +        resetClient(c);
 +        return 1;
 +    }
 +
 +    /* Only allow SUBSCRIBE and UNSUBSCRIBE in the context of Pub/Sub */
 +    if ((dictSize(c->pubsub_channels) > 0 || listLength(c->pubsub_patterns) > 0)
 +        &&
 +        cmd->proc != subscribeCommand && cmd->proc != unsubscribeCommand &&
 +        cmd->proc != psubscribeCommand && cmd->proc != punsubscribeCommand) {
 +        addReplySds(c,sdsnew("-ERR only (P)SUBSCRIBE / (P)UNSUBSCRIBE / QUIT allowed in this context\r\n"));
 +        resetClient(c);
 +        return 1;
 +    }
 +
 +    /* Exec the command */
 +    if (c->flags & REDIS_MULTI &&
 +        cmd->proc != execCommand && cmd->proc != discardCommand &&
 +        cmd->proc != multiCommand && cmd->proc != watchCommand)
 +    {
 +        queueMultiCommand(c,cmd);
 +        addReply(c,shared.queued);
 +    } else {
 +        if (server.vm_enabled && server.vm_max_threads > 0 &&
 +            blockClientOnSwappedKeys(c,cmd)) return 1;
 +        call(c,cmd);
 +    }
 +
 +    /* Prepare the client for the next command */
 +    resetClient(c);
 +    return 1;
 +}
 +
 +/*================================== Shutdown =============================== */
 +
 +int prepareForShutdown() {
 +    redisLog(REDIS_WARNING,"User requested shutdown, saving DB...");
 +    /* Kill the saving child if there is a background saving in progress.
 +       We want to avoid race conditions, for instance our saving child may
 +       overwrite the synchronous saving did by SHUTDOWN. */
 +    if (server.bgsavechildpid != -1) {
 +        redisLog(REDIS_WARNING,"There is a live saving child. Killing it!");
 +        kill(server.bgsavechildpid,SIGKILL);
 +        rdbRemoveTempFile(server.bgsavechildpid);
 +    }
 +    if (server.appendonly) {
 +        /* Append only file: fsync() the AOF and exit */
 +        aof_fsync(server.appendfd);
 +        if (server.vm_enabled) unlink(server.vm_swap_file);
 +    } else {
 +        /* Snapshotting. Perform a SYNC SAVE and exit */
 +        if (rdbSave(server.dbfilename) == REDIS_OK) {
 +            if (server.daemonize)
 +                unlink(server.pidfile);
 +            redisLog(REDIS_WARNING,"%zu bytes used at exit",zmalloc_used_memory());
 +        } else {
 +            /* Ooops.. error saving! The best we can do is to continue
 +             * operating. Note that if there was a background saving process,
 +             * in the next cron() Redis will be notified that the background
 +             * saving aborted, handling special stuff like slaves pending for
 +             * synchronization... */
 +            redisLog(REDIS_WARNING,"Error trying to save the DB, can't exit");
 +            return REDIS_ERR;
 +        }
 +    }
 +    redisLog(REDIS_WARNING,"Server exit now, bye bye...");
 +    return REDIS_OK;
 +}
 +
 +/*================================== Commands =============================== */
 +
 +void authCommand(redisClient *c) {
 +    if (!server.requirepass || !strcmp(c->argv[1]->ptr, server.requirepass)) {
 +      c->authenticated = 1;
 +      addReply(c,shared.ok);
 +    } else {
 +      c->authenticated = 0;
 +      addReplySds(c,sdscatprintf(sdsempty(),"-ERR invalid password\r\n"));
 +    }
 +}
 +
 +void pingCommand(redisClient *c) {
 +    addReply(c,shared.pong);
 +}
 +
 +void echoCommand(redisClient *c) {
 +    addReplyBulk(c,c->argv[1]);
 +}
 +
 +/* Convert an amount of bytes into a human readable string in the form
 + * of 100B, 2G, 100M, 4K, and so forth. */
 +void bytesToHuman(char *s, unsigned long long n) {
 +    double d;
 +
 +    if (n < 1024) {
 +        /* Bytes */
 +        sprintf(s,"%lluB",n);
 +        return;
 +    } else if (n < (1024*1024)) {
 +        d = (double)n/(1024);
 +        sprintf(s,"%.2fK",d);
 +    } else if (n < (1024LL*1024*1024)) {
 +        d = (double)n/(1024*1024);
 +        sprintf(s,"%.2fM",d);
 +    } else if (n < (1024LL*1024*1024*1024)) {
 +        d = (double)n/(1024LL*1024*1024);
 +        sprintf(s,"%.2fG",d);
 +    }
 +}
 +
 +/* Create the string returned by the INFO command. This is decoupled
 + * by the INFO command itself as we need to report the same information
 + * on memory corruption problems. */
 +sds genRedisInfoString(void) {
 +    sds info;
 +    time_t uptime = time(NULL)-server.stat_starttime;
 +    int j;
 +    char hmem[64];
 +
 +    bytesToHuman(hmem,zmalloc_used_memory());
 +    info = sdscatprintf(sdsempty(),
 +        "redis_version:%s\r\n"
 +        "redis_git_sha1:%s\r\n"
 +        "redis_git_dirty:%d\r\n"
 +        "arch_bits:%s\r\n"
 +        "multiplexing_api:%s\r\n"
 +        "process_id:%ld\r\n"
 +        "uptime_in_seconds:%ld\r\n"
 +        "uptime_in_days:%ld\r\n"
 +        "connected_clients:%d\r\n"
 +        "connected_slaves:%d\r\n"
 +        "blocked_clients:%d\r\n"
 +        "used_memory:%zu\r\n"
 +        "used_memory_human:%s\r\n"
 +        "changes_since_last_save:%lld\r\n"
 +        "bgsave_in_progress:%d\r\n"
 +        "last_save_time:%ld\r\n"
 +        "bgrewriteaof_in_progress:%d\r\n"
 +        "total_connections_received:%lld\r\n"
 +        "total_commands_processed:%lld\r\n"
 +        "expired_keys:%lld\r\n"
 +        "hash_max_zipmap_entries:%zu\r\n"
 +        "hash_max_zipmap_value:%zu\r\n"
 +        "pubsub_channels:%ld\r\n"
 +        "pubsub_patterns:%u\r\n"
 +        "vm_enabled:%d\r\n"
 +        "role:%s\r\n"
 +        ,REDIS_VERSION,
 +        redisGitSHA1(),
 +        strtol(redisGitDirty(),NULL,10) > 0,
 +        (sizeof(long) == 8) ? "64" : "32",
 +        aeGetApiName(),
 +        (long) getpid(),
 +        uptime,
 +        uptime/(3600*24),
 +        listLength(server.clients)-listLength(server.slaves),
 +        listLength(server.slaves),
 +        server.blpop_blocked_clients,
 +        zmalloc_used_memory(),
 +        hmem,
 +        server.dirty,
 +        server.bgsavechildpid != -1,
 +        server.lastsave,
 +        server.bgrewritechildpid != -1,
 +        server.stat_numconnections,
 +        server.stat_numcommands,
 +        server.stat_expiredkeys,
 +        server.hash_max_zipmap_entries,
 +        server.hash_max_zipmap_value,
 +        dictSize(server.pubsub_channels),
 +        listLength(server.pubsub_patterns),
 +        server.vm_enabled != 0,
 +        server.masterhost == NULL ? "master" : "slave"
 +    );
 +    if (server.masterhost) {
 +        info = sdscatprintf(info,
 +            "master_host:%s\r\n"
 +            "master_port:%d\r\n"
 +            "master_link_status:%s\r\n"
 +            "master_last_io_seconds_ago:%d\r\n"
 +            ,server.masterhost,
 +            server.masterport,
 +            (server.replstate == REDIS_REPL_CONNECTED) ?
 +                "up" : "down",
 +            server.master ? ((int)(time(NULL)-server.master->lastinteraction)) : -1
 +        );
 +    }
 +    if (server.vm_enabled) {
 +        lockThreadedIO();
 +        info = sdscatprintf(info,
 +            "vm_conf_max_memory:%llu\r\n"
 +            "vm_conf_page_size:%llu\r\n"
 +            "vm_conf_pages:%llu\r\n"
 +            "vm_stats_used_pages:%llu\r\n"
 +            "vm_stats_swapped_objects:%llu\r\n"
 +            "vm_stats_swappin_count:%llu\r\n"
 +            "vm_stats_swappout_count:%llu\r\n"
 +            "vm_stats_io_newjobs_len:%lu\r\n"
 +            "vm_stats_io_processing_len:%lu\r\n"
 +            "vm_stats_io_processed_len:%lu\r\n"
 +            "vm_stats_io_active_threads:%lu\r\n"
 +            "vm_stats_blocked_clients:%lu\r\n"
 +            ,(unsigned long long) server.vm_max_memory,
 +            (unsigned long long) server.vm_page_size,
 +            (unsigned long long) server.vm_pages,
 +            (unsigned long long) server.vm_stats_used_pages,
 +            (unsigned long long) server.vm_stats_swapped_objects,
 +            (unsigned long long) server.vm_stats_swapins,
 +            (unsigned long long) server.vm_stats_swapouts,
 +            (unsigned long) listLength(server.io_newjobs),
 +            (unsigned long) listLength(server.io_processing),
 +            (unsigned long) listLength(server.io_processed),
 +            (unsigned long) server.io_active_threads,
 +            (unsigned long) server.vm_blocked_clients
 +        );
 +        unlockThreadedIO();
 +    }
 +    for (j = 0; j < server.dbnum; j++) {
 +        long long keys, vkeys;
 +
 +        keys = dictSize(server.db[j].dict);
 +        vkeys = dictSize(server.db[j].expires);
 +        if (keys || vkeys) {
 +            info = sdscatprintf(info, "db%d:keys=%lld,expires=%lld\r\n",
 +                j, keys, vkeys);
 +        }
 +    }
 +    return info;
 +}
 +
 +void infoCommand(redisClient *c) {
 +    sds info = genRedisInfoString();
 +    addReplySds(c,sdscatprintf(sdsempty(),"$%lu\r\n",
 +        (unsigned long)sdslen(info)));
 +    addReplySds(c,info);
 +    addReply(c,shared.crlf);
 +}
 +
 +void monitorCommand(redisClient *c) {
 +    /* ignore MONITOR if aleady slave or in monitor mode */
 +    if (c->flags & REDIS_SLAVE) return;
 +
 +    c->flags |= (REDIS_SLAVE|REDIS_MONITOR);
 +    c->slaveseldb = 0;
 +    listAddNodeTail(server.monitors,c);
 +    addReply(c,shared.ok);
 +}
 +
 +/* ============================ Maxmemory directive  ======================== */
 +
 +/* Try to free one object form the pre-allocated objects free list.
 + * This is useful under low mem conditions as by default we take 1 million
 + * free objects allocated. On success REDIS_OK is returned, otherwise
 + * REDIS_ERR. */
 +int tryFreeOneObjectFromFreelist(void) {
 +    robj *o;
 +
 +    if (server.vm_enabled) pthread_mutex_lock(&server.obj_freelist_mutex);
 +    if (listLength(server.objfreelist)) {
 +        listNode *head = listFirst(server.objfreelist);
 +        o = listNodeValue(head);
 +        listDelNode(server.objfreelist,head);
 +        if (server.vm_enabled) pthread_mutex_unlock(&server.obj_freelist_mutex);
 +        zfree(o);
 +        return REDIS_OK;
 +    } else {
 +        if (server.vm_enabled) pthread_mutex_unlock(&server.obj_freelist_mutex);
 +        return REDIS_ERR;
 +    }
 +}
 +
 +/* This function gets called when 'maxmemory' is set on the config file to limit
 + * the max memory used by the server, and we are out of memory.
 + * This function will try to, in order:
 + *
 + * - Free objects from the free list
 + * - Try to remove keys with an EXPIRE set
 + *
 + * It is not possible to free enough memory to reach used-memory < maxmemory
 + * the server will start refusing commands that will enlarge even more the
 + * memory usage.
 + */
 +void freeMemoryIfNeeded(void) {
 +    while (server.maxmemory && zmalloc_used_memory() > server.maxmemory) {
 +        int j, k, freed = 0;
 +
 +        if (tryFreeOneObjectFromFreelist() == REDIS_OK) continue;
 +        for (j = 0; j < server.dbnum; j++) {
 +            int minttl = -1;
 +            robj *minkey = NULL;
 +            struct dictEntry *de;
 +
 +            if (dictSize(server.db[j].expires)) {
 +                freed = 1;
 +                /* From a sample of three keys drop the one nearest to
 +                 * the natural expire */
 +                for (k = 0; k < 3; k++) {
 +                    time_t t;
 +
 +                    de = dictGetRandomKey(server.db[j].expires);
 +                    t = (time_t) dictGetEntryVal(de);
 +                    if (minttl == -1 || t < minttl) {
 +                        minkey = dictGetEntryKey(de);
 +                        minttl = t;
 +                    }
 +                }
 +                dbDelete(server.db+j,minkey);
 +            }
 +        }
 +        if (!freed) return; /* nothing to free... */
 +    }
 +}
 +
 +/* =================================== Main! ================================ */
 +
 +#ifdef __linux__
 +int linuxOvercommitMemoryValue(void) {
 +    FILE *fp = fopen("/proc/sys/vm/overcommit_memory","r");
 +    char buf[64];
 +
 +    if (!fp) return -1;
 +    if (fgets(buf,64,fp) == NULL) {
 +        fclose(fp);
 +        return -1;
 +    }
 +    fclose(fp);
 +
 +    return atoi(buf);
 +}
 +
 +void linuxOvercommitMemoryWarning(void) {
 +    if (linuxOvercommitMemoryValue() == 0) {
 +        redisLog(REDIS_WARNING,"WARNING overcommit_memory is set to 0! Background save may fail under low memory condition. To fix this issue add 'vm.overcommit_memory = 1' to /etc/sysctl.conf and then reboot or run the command 'sysctl vm.overcommit_memory=1' for this to take effect.");
 +    }
 +}
 +#endif /* __linux__ */
 +
 +void daemonize(void) {
 +    int fd;
 +    FILE *fp;
 +
 +    if (fork() != 0) exit(0); /* parent exits */
 +    setsid(); /* create a new session */
 +
 +    /* Every output goes to /dev/null. If Redis is daemonized but
 +     * the 'logfile' is set to 'stdout' in the configuration file
 +     * it will not log at all. */
 +    if ((fd = open("/dev/null", O_RDWR, 0)) != -1) {
 +        dup2(fd, STDIN_FILENO);
 +        dup2(fd, STDOUT_FILENO);
 +        dup2(fd, STDERR_FILENO);
 +        if (fd > STDERR_FILENO) close(fd);
 +    }
 +    /* Try to write the pid file */
 +    fp = fopen(server.pidfile,"w");
 +    if (fp) {
 +        fprintf(fp,"%d\n",getpid());
 +        fclose(fp);
 +    }
 +}
 +
 +void version() {
 +    printf("Redis server version %s (%s:%d)\n", REDIS_VERSION,
 +        redisGitSHA1(), atoi(redisGitDirty()) > 0);
 +    exit(0);
 +}
 +
 +void usage() {
 +    fprintf(stderr,"Usage: ./redis-server [/path/to/redis.conf]\n");
 +    fprintf(stderr,"       ./redis-server - (read config from stdin)\n");
 +    exit(1);
 +}
 +
 +int main(int argc, char **argv) {
 +    time_t start;
 +
 +    initServerConfig();
 +    sortCommandTable();
 +    if (argc == 2) {
 +        if (strcmp(argv[1], "-v") == 0 ||
 +            strcmp(argv[1], "--version") == 0) version();
 +        if (strcmp(argv[1], "--help") == 0) usage();
 +        resetServerSaveParams();
 +        loadServerConfig(argv[1]);
 +    } else if ((argc > 2)) {
 +        usage();
 +    } else {
 +        redisLog(REDIS_WARNING,"Warning: no config file specified, using the default config. In order to specify a config file use 'redis-server /path/to/redis.conf'");
 +    }
 +    if (server.daemonize) daemonize();
 +    initServer();
 +    redisLog(REDIS_NOTICE,"Server started, Redis version " REDIS_VERSION);
 +#ifdef __linux__
 +    linuxOvercommitMemoryWarning();
 +#endif
 +    start = time(NULL);
 +    if (server.appendonly) {
 +        if (loadAppendOnlyFile(server.appendfilename) == REDIS_OK)
 +            redisLog(REDIS_NOTICE,"DB loaded from append only file: %ld seconds",time(NULL)-start);
 +    } else {
 +        if (rdbLoad(server.dbfilename) == REDIS_OK)
 +            redisLog(REDIS_NOTICE,"DB loaded from disk: %ld seconds",time(NULL)-start);
 +    }
 +    redisLog(REDIS_NOTICE,"The server is now ready to accept connections on port %d", server.port);
 +    aeSetBeforeSleepProc(server.el,beforeSleep);
 +    aeMain(server.el);
 +    aeDeleteEventLoop(server.el);
 +    return 0;
 +}
 +
 +/* ============================= Backtrace support ========================= */
 +
 +#ifdef HAVE_BACKTRACE
 +void *getMcontextEip(ucontext_t *uc) {
 +#if defined(__FreeBSD__)
 +    return (void*) uc->uc_mcontext.mc_eip;
 +#elif defined(__dietlibc__)
 +    return (void*) uc->uc_mcontext.eip;
 +#elif defined(__APPLE__) && !defined(MAC_OS_X_VERSION_10_6)
 +  #if __x86_64__
 +    return (void*) uc->uc_mcontext->__ss.__rip;
 +  #else
 +    return (void*) uc->uc_mcontext->__ss.__eip;
 +  #endif
 +#elif defined(__APPLE__) && defined(MAC_OS_X_VERSION_10_6)
 +  #if defined(_STRUCT_X86_THREAD_STATE64) && !defined(__i386__)
 +    return (void*) uc->uc_mcontext->__ss.__rip;
 +  #else
 +    return (void*) uc->uc_mcontext->__ss.__eip;
 +  #endif
 +#elif defined(__i386__)
 +    return (void*) uc->uc_mcontext.gregs[14]; /* Linux 32 */
 +#elif defined(__X86_64__) || defined(__x86_64__)
 +    return (void*) uc->uc_mcontext.gregs[16]; /* Linux 64 */
 +#elif defined(__ia64__) /* Linux IA64 */
 +    return (void*) uc->uc_mcontext.sc_ip;
 +#else
 +    return NULL;
 +#endif
 +}
 +
 +void segvHandler(int sig, siginfo_t *info, void *secret) {
 +    void *trace[100];
 +    char **messages = NULL;
 +    int i, trace_size = 0;
 +    ucontext_t *uc = (ucontext_t*) secret;
 +    sds infostring;
 +    REDIS_NOTUSED(info);
 +
 +    redisLog(REDIS_WARNING,
 +        "======= Ooops! Redis %s got signal: -%d- =======", REDIS_VERSION, sig);
 +    infostring = genRedisInfoString();
 +    redisLog(REDIS_WARNING, "%s",infostring);
 +    /* It's not safe to sdsfree() the returned string under memory
 +     * corruption conditions. Let it leak as we are going to abort */
 +
 +    trace_size = backtrace(trace, 100);
 +    /* overwrite sigaction with caller's address */
 +    if (getMcontextEip(uc) != NULL) {
 +        trace[1] = getMcontextEip(uc);
 +    }
 +    messages = backtrace_symbols(trace, trace_size);
 +
 +    for (i=1; i<trace_size; ++i)
 +        redisLog(REDIS_WARNING,"%s", messages[i]);
 +
 +    /* free(messages); Don't call free() with possibly corrupted memory. */
 +    _exit(0);
 +}
 +
 +void sigtermHandler(int sig) {
 +    REDIS_NOTUSED(sig);
 +
 +    redisLog(REDIS_WARNING,"SIGTERM received, scheduling shutting down...");
 +    server.shutdown_asap = 1;
 +}
 +
 +void setupSigSegvAction(void) {
 +    struct sigaction act;
 +
 +    sigemptyset (&act.sa_mask);
 +    /* When the SA_SIGINFO flag is set in sa_flags then sa_sigaction
 +     * is used. Otherwise, sa_handler is used */
 +    act.sa_flags = SA_NODEFER | SA_ONSTACK | SA_RESETHAND | SA_SIGINFO;
 +    act.sa_sigaction = segvHandler;
 +    sigaction (SIGSEGV, &act, NULL);
 +    sigaction (SIGBUS, &act, NULL);
 +    sigaction (SIGFPE, &act, NULL);
 +    sigaction (SIGILL, &act, NULL);
 +    sigaction (SIGBUS, &act, NULL);
 +
 +    act.sa_flags = SA_NODEFER | SA_ONSTACK | SA_RESETHAND;
 +    act.sa_handler = sigtermHandler;
 +    sigaction (SIGTERM, &act, NULL);
 +    return;
 +}
 +
 +#else /* HAVE_BACKTRACE */
 +void setupSigSegvAction(void) {
 +}
 +#endif /* HAVE_BACKTRACE */
 +
 +/* The End */
diff --cc src/redis.h
index 545a3e0f,00000000..463db704
mode 100644,000000..100644
--- a/src/redis.h
+++ b/src/redis.h
@@@ -1,886 -1,0 +1,911 @@@
 +#ifndef __REDIS_H
 +#define __REDIS_H
 +
 +#include "fmacros.h"
 +#include "config.h"
 +
 +#if defined(__sun)
 +#include "solarisfixes.h"
 +#endif
 +
 +#include <stdio.h>
 +#include <stdlib.h>
 +#include <string.h>
 +#include <time.h>
 +#include <limits.h>
 +#include <unistd.h>
 +#include <errno.h>
 +#include <inttypes.h>
 +
 +#include "ae.h"     /* Event driven programming library */
 +#include "sds.h"    /* Dynamic safe strings */
 +#include "dict.h"   /* Hash tables */
 +#include "adlist.h" /* Linked lists */
 +#include "zmalloc.h" /* total memory usage aware version of malloc/free */
 +#include "anet.h"   /* Networking the easy way */
 +#include "zipmap.h" /* Compact string -> string data structure */
 +#include "ziplist.h" /* Compact list data structure */
++#include "intset.h" /* Compact integer set structure */
 +#include "version.h"
 +
 +/* Error codes */
 +#define REDIS_OK                0
 +#define REDIS_ERR               -1
 +
 +/* Static server configuration */
 +#define REDIS_SERVERPORT        6379    /* TCP port */
 +#define REDIS_MAXIDLETIME       (60*5)  /* default client timeout */
 +#define REDIS_IOBUF_LEN         1024
 +#define REDIS_LOADBUF_LEN       1024
 +#define REDIS_STATIC_ARGS       8
 +#define REDIS_DEFAULT_DBNUM     16
 +#define REDIS_CONFIGLINE_MAX    1024
 +#define REDIS_OBJFREELIST_MAX   1000000 /* Max number of objects to cache */
 +#define REDIS_MAX_SYNC_TIME     60      /* Slave can't take more to sync */
 +#define REDIS_EXPIRELOOKUPS_PER_CRON    10 /* lookup 10 expires per loop */
 +#define REDIS_MAX_WRITE_PER_EVENT (1024*64)
 +#define REDIS_REQUEST_MAX_SIZE (1024*1024*256) /* max bytes in inline command */
 +#define REDIS_SHARED_INTEGERS 10000
 +
 +/* If more then REDIS_WRITEV_THRESHOLD write packets are pending use writev */
 +#define REDIS_WRITEV_THRESHOLD      3
 +/* Max number of iovecs used for each writev call */
 +#define REDIS_WRITEV_IOVEC_COUNT    256
 +
 +/* Hash table parameters */
 +#define REDIS_HT_MINFILL        10      /* Minimal hash table fill 10% */
 +
 +/* Command flags */
 +#define REDIS_CMD_BULK          1       /* Bulk write command */
 +#define REDIS_CMD_INLINE        2       /* Inline command */
 +/* REDIS_CMD_DENYOOM reserves a longer comment: all the commands marked with
 +   this flags will return an error when the 'maxmemory' option is set in the
 +   config file and the server is using more than maxmemory bytes of memory.
 +   In short this commands are denied on low memory conditions. */
 +#define REDIS_CMD_DENYOOM       4
 +#define REDIS_CMD_FORCE_REPLICATION 8 /* Force replication even if dirty is 0 */
 +
 +/* Object types */
 +#define REDIS_STRING 0
 +#define REDIS_LIST 1
 +#define REDIS_SET 2
 +#define REDIS_ZSET 3
 +#define REDIS_HASH 4
 +#define REDIS_VMPOINTER 8
 +
 +/* Objects encoding. Some kind of objects like Strings and Hashes can be
 + * internally represented in multiple ways. The 'encoding' field of the object
 + * is set to one of this fields for this object. */
 +#define REDIS_ENCODING_RAW 0     /* Raw representation */
 +#define REDIS_ENCODING_INT 1     /* Encoded as integer */
 +#define REDIS_ENCODING_HT 2      /* Encoded as hash table */
 +#define REDIS_ENCODING_ZIPMAP 3  /* Encoded as zipmap */
 +#define REDIS_ENCODING_LINKEDLIST 4 /* Encoded as regular linked list */
 +#define REDIS_ENCODING_ZIPLIST 5 /* Encoded as ziplist */
++#define REDIS_ENCODING_INTSET 6  /* Encoded as intset */
 +
 +/* Object types only used for dumping to disk */
 +#define REDIS_EXPIRETIME 253
 +#define REDIS_SELECTDB 254
 +#define REDIS_EOF 255
 +
 +/* Defines related to the dump file format. To store 32 bits lengths for short
 + * keys requires a lot of space, so we check the most significant 2 bits of
 + * the first byte to interpreter the length:
 + *
 + * 00|000000 => if the two MSB are 00 the len is the 6 bits of this byte
 + * 01|000000 00000000 =>  01, the len is 14 byes, 6 bits + 8 bits of next byte
 + * 10|000000 [32 bit integer] => if it's 01, a full 32 bit len will follow
 + * 11|000000 this means: specially encoded object will follow. The six bits
 + *           number specify the kind of object that follows.
 + *           See the REDIS_RDB_ENC_* defines.
 + *
 + * Lenghts up to 63 are stored using a single byte, most DB keys, and may
 + * values, will fit inside. */
 +#define REDIS_RDB_6BITLEN 0
 +#define REDIS_RDB_14BITLEN 1
 +#define REDIS_RDB_32BITLEN 2
 +#define REDIS_RDB_ENCVAL 3
 +#define REDIS_RDB_LENERR UINT_MAX
 +
 +/* When a length of a string object stored on disk has the first two bits
 + * set, the remaining two bits specify a special encoding for the object
 + * accordingly to the following defines: */
 +#define REDIS_RDB_ENC_INT8 0        /* 8 bit signed integer */
 +#define REDIS_RDB_ENC_INT16 1       /* 16 bit signed integer */
 +#define REDIS_RDB_ENC_INT32 2       /* 32 bit signed integer */
 +#define REDIS_RDB_ENC_LZF 3         /* string compressed with FASTLZ */
 +
 +/* Virtual memory object->where field. */
 +#define REDIS_VM_MEMORY 0       /* The object is on memory */
 +#define REDIS_VM_SWAPPED 1      /* The object is on disk */
 +#define REDIS_VM_SWAPPING 2     /* Redis is swapping this object on disk */
 +#define REDIS_VM_LOADING 3      /* Redis is loading this object from disk */
 +
 +/* Virtual memory static configuration stuff.
 + * Check vmFindContiguousPages() to know more about this magic numbers. */
 +#define REDIS_VM_MAX_NEAR_PAGES 65536
 +#define REDIS_VM_MAX_RANDOM_JUMP 4096
 +#define REDIS_VM_MAX_THREADS 32
 +#define REDIS_THREAD_STACK_SIZE (1024*1024*4)
 +/* The following is the *percentage* of completed I/O jobs to process when the
 + * handelr is called. While Virtual Memory I/O operations are performed by
 + * threads, this operations must be processed by the main thread when completed
 + * in order to take effect. */
 +#define REDIS_MAX_COMPLETED_JOBS_PROCESSED 1
 +
 +/* Client flags */
 +#define REDIS_SLAVE 1       /* This client is a slave server */
 +#define REDIS_MASTER 2      /* This client is a master server */
 +#define REDIS_MONITOR 4     /* This client is a slave monitor, see MONITOR */
 +#define REDIS_MULTI 8       /* This client is in a MULTI context */
 +#define REDIS_BLOCKED 16    /* The client is waiting in a blocking operation */
 +#define REDIS_IO_WAIT 32    /* The client is waiting for Virtual Memory I/O */
 +#define REDIS_DIRTY_CAS 64  /* Watched keys modified. EXEC will fail. */
 +
 +/* Slave replication state - slave side */
 +#define REDIS_REPL_NONE 0   /* No active replication */
 +#define REDIS_REPL_CONNECT 1    /* Must connect to master */
 +#define REDIS_REPL_CONNECTED 2  /* Connected to master */
 +
 +/* Slave replication state - from the point of view of master
 + * Note that in SEND_BULK and ONLINE state the slave receives new updates
 + * in its output queue. In the WAIT_BGSAVE state instead the server is waiting
 + * to start the next background saving in order to send updates to it. */
 +#define REDIS_REPL_WAIT_BGSAVE_START 3 /* master waits bgsave to start feeding it */
 +#define REDIS_REPL_WAIT_BGSAVE_END 4 /* master waits bgsave to start bulk DB transmission */
 +#define REDIS_REPL_SEND_BULK 5 /* master is sending the bulk DB */
 +#define REDIS_REPL_ONLINE 6 /* bulk DB already transmitted, receive updates */
 +
 +/* List related stuff */
 +#define REDIS_HEAD 0
 +#define REDIS_TAIL 1
 +
 +/* Sort operations */
 +#define REDIS_SORT_GET 0
 +#define REDIS_SORT_ASC 1
 +#define REDIS_SORT_DESC 2
 +#define REDIS_SORTKEY_MAX 1024
 +
 +/* Log levels */
 +#define REDIS_DEBUG 0
 +#define REDIS_VERBOSE 1
 +#define REDIS_NOTICE 2
 +#define REDIS_WARNING 3
 +
 +/* Anti-warning macro... */
 +#define REDIS_NOTUSED(V) ((void) V)
 +
 +#define ZSKIPLIST_MAXLEVEL 32 /* Should be enough for 2^32 elements */
 +#define ZSKIPLIST_P 0.25      /* Skiplist P = 1/4 */
 +
 +/* Append only defines */
 +#define APPENDFSYNC_NO 0
 +#define APPENDFSYNC_ALWAYS 1
 +#define APPENDFSYNC_EVERYSEC 2
 +
 +/* Zip structure related defaults */
 +#define REDIS_HASH_MAX_ZIPMAP_ENTRIES 64
 +#define REDIS_HASH_MAX_ZIPMAP_VALUE 512
 +#define REDIS_LIST_MAX_ZIPLIST_ENTRIES 1024
 +#define REDIS_LIST_MAX_ZIPLIST_VALUE 32
++#define REDIS_SET_MAX_INTSET_ENTRIES 4096
 +
 +/* Sets operations codes */
 +#define REDIS_OP_UNION 0
 +#define REDIS_OP_DIFF 1
 +#define REDIS_OP_INTER 2
 +
 +/* We can print the stacktrace, so our assert is defined this way: */
 +#define redisAssert(_e) ((_e)?(void)0 : (_redisAssert(#_e,__FILE__,__LINE__),_exit(1)))
 +#define redisPanic(_e) _redisPanic(#_e,__FILE__,__LINE__),_exit(1)
 +void _redisAssert(char *estr, char *file, int line);
 +void _redisPanic(char *msg, char *file, int line);
 +
 +/*-----------------------------------------------------------------------------
 + * Data types
 + *----------------------------------------------------------------------------*/
 +
 +/* A redis object, that is a type able to hold a string / list / set */
 +
 +/* The actual Redis Object */
 +typedef struct redisObject {
 +    unsigned type:4;
 +    unsigned storage:2;     /* REDIS_VM_MEMORY or REDIS_VM_SWAPPING */
 +    unsigned encoding:4;
 +    unsigned lru:22;        /* lru time (relative to server.lruclock) */
 +    int refcount;
 +    void *ptr;
 +    /* VM fields are only allocated if VM is active, otherwise the
 +     * object allocation function will just allocate
 +     * sizeof(redisObjct) minus sizeof(redisObjectVM), so using
 +     * Redis without VM active will not have any overhead. */
 +} robj;
 +
 +/* The VM pointer structure - identifies an object in the swap file.
 + *
 + * This object is stored in place of the value
 + * object in the main key->value hash table representing a database.
 + * Note that the first fields (type, storage) are the same as the redisObject
 + * structure so that vmPointer strucuters can be accessed even when casted
 + * as redisObject structures.
 + *
 + * This is useful as we don't know if a value object is or not on disk, but we
 + * are always able to read obj->storage to check this. For vmPointer
 + * structures "type" is set to REDIS_VMPOINTER (even if without this field
 + * is still possible to check the kind of object from the value of 'storage').*/
 +typedef struct vmPointer {
 +    unsigned type:4;
 +    unsigned storage:2; /* REDIS_VM_SWAPPED or REDIS_VM_LOADING */
 +    unsigned notused:26;
 +    unsigned int vtype; /* type of the object stored in the swap file */
 +    off_t page;         /* the page at witch the object is stored on disk */
 +    off_t usedpages;    /* number of pages used on disk */
 +} vmpointer;
 +
 +/* Macro used to initalize a Redis object allocated on the stack.
 + * Note that this macro is taken near the structure definition to make sure
 + * we'll update it when the structure is changed, to avoid bugs like
 + * bug #85 introduced exactly in this way. */
 +#define initStaticStringObject(_var,_ptr) do { \
 +    _var.refcount = 1; \
 +    _var.type = REDIS_STRING; \
 +    _var.encoding = REDIS_ENCODING_RAW; \
 +    _var.ptr = _ptr; \
 +    _var.storage = REDIS_VM_MEMORY; \
 +} while(0);
 +
 +typedef struct redisDb {
 +    dict *dict;                 /* The keyspace for this DB */
 +    dict *expires;              /* Timeout of keys with a timeout set */
 +    dict *blocking_keys;        /* Keys with clients waiting for data (BLPOP) */
 +    dict *io_keys;              /* Keys with clients waiting for VM I/O */
 +    dict *watched_keys;         /* WATCHED keys for MULTI/EXEC CAS */
 +    int id;
 +} redisDb;
 +
 +/* Client MULTI/EXEC state */
 +typedef struct multiCmd {
 +    robj **argv;
 +    int argc;
 +    struct redisCommand *cmd;
 +} multiCmd;
 +
 +typedef struct multiState {
 +    multiCmd *commands;     /* Array of MULTI commands */
 +    int count;              /* Total number of MULTI commands */
 +} multiState;
 +
 +/* With multiplexing we need to take per-clinet state.
 + * Clients are taken in a liked list. */
 +typedef struct redisClient {
 +    int fd;
 +    redisDb *db;
 +    int dictid;
 +    sds querybuf;
 +    robj **argv, **mbargv;
 +    int argc, mbargc;
 +    int bulklen;            /* bulk read len. -1 if not in bulk read mode */
 +    int multibulk;          /* multi bulk command format active */
 +    list *reply;
 +    int sentlen;
 +    time_t lastinteraction; /* time of the last interaction, used for timeout */
 +    int flags;              /* REDIS_SLAVE | REDIS_MONITOR | REDIS_MULTI ... */
 +    int slaveseldb;         /* slave selected db, if this client is a slave */
 +    int authenticated;      /* when requirepass is non-NULL */
 +    int replstate;          /* replication state if this is a slave */
 +    int repldbfd;           /* replication DB file descriptor */
 +    long repldboff;         /* replication DB file offset */
 +    off_t repldbsize;       /* replication DB file size */
 +    multiState mstate;      /* MULTI/EXEC state */
 +    robj **blocking_keys;   /* The key we are waiting to terminate a blocking
 +                             * operation such as BLPOP. Otherwise NULL. */
 +    int blocking_keys_num;  /* Number of blocking keys */
 +    time_t blockingto;      /* Blocking operation timeout. If UNIX current time
 +                             * is >= blockingto then the operation timed out. */
 +    list *io_keys;          /* Keys this client is waiting to be loaded from the
 +                             * swap file in order to continue. */
 +    list *watched_keys;     /* Keys WATCHED for MULTI/EXEC CAS */
 +    dict *pubsub_channels;  /* channels a client is interested in (SUBSCRIBE) */
 +    list *pubsub_patterns;  /* patterns a client is interested in (SUBSCRIBE) */
 +} redisClient;
 +
 +struct saveparam {
 +    time_t seconds;
 +    int changes;
 +};
 +
 +struct sharedObjectsStruct {
 +    robj *crlf, *ok, *err, *emptybulk, *czero, *cone, *cnegone, *pong, *space,
 +    *colon, *nullbulk, *nullmultibulk, *queued,
 +    *emptymultibulk, *wrongtypeerr, *nokeyerr, *syntaxerr, *sameobjecterr,
 +    *outofrangeerr, *plus,
 +    *select0, *select1, *select2, *select3, *select4,
 +    *select5, *select6, *select7, *select8, *select9,
 +    *messagebulk, *pmessagebulk, *subscribebulk, *unsubscribebulk, *mbulk3,
 +    *mbulk4, *psubscribebulk, *punsubscribebulk,
 +    *integers[REDIS_SHARED_INTEGERS];
 +};
 +
 +/* Global server state structure */
 +struct redisServer {
 +    int port;
 +    int fd;
 +    redisDb *db;
 +    long long dirty;            /* changes to DB from the last save */
 +    list *clients;
 +    list *slaves, *monitors;
 +    char neterr[ANET_ERR_LEN];
 +    aeEventLoop *el;
 +    int cronloops;              /* number of times the cron function run */
 +    list *objfreelist;          /* A list of freed objects to avoid malloc() */
 +    time_t lastsave;            /* Unix time of last save succeeede */
 +    /* Fields used only for stats */
 +    time_t stat_starttime;         /* server start time */
 +    long long stat_numcommands;    /* number of processed commands */
 +    long long stat_numconnections; /* number of connections received */
 +    long long stat_expiredkeys;   /* number of expired keys */
 +    /* Configuration */
 +    int verbosity;
 +    int glueoutputbuf;
 +    int maxidletime;
 +    int dbnum;
 +    int daemonize;
 +    int appendonly;
 +    int appendfsync;
 +    int no_appendfsync_on_rewrite;
 +    int shutdown_asap;
 +    time_t lastfsync;
 +    int appendfd;
 +    int appendseldb;
 +    char *pidfile;
 +    pid_t bgsavechildpid;
 +    pid_t bgrewritechildpid;
 +    sds bgrewritebuf; /* buffer taken by parent during oppend only rewrite */
 +    sds aofbuf;       /* AOF buffer, written before entering the event loop */
 +    struct saveparam *saveparams;
 +    int saveparamslen;
 +    char *logfile;
 +    char *bindaddr;
 +    char *dbfilename;
 +    char *appendfilename;
 +    char *requirepass;
 +    int rdbcompression;
 +    int activerehashing;
 +    /* Replication related */
 +    int isslave;
 +    char *masterauth;
 +    char *masterhost;
 +    int masterport;
 +    redisClient *master;    /* client that is master for this slave */
 +    int replstate;
 +    unsigned int maxclients;
 +    unsigned long long maxmemory;
 +    unsigned int blpop_blocked_clients;
 +    unsigned int vm_blocked_clients;
 +    /* Sort parameters - qsort_r() is only available under BSD so we
 +     * have to take this state global, in order to pass it to sortCompare() */
 +    int sort_desc;
 +    int sort_alpha;
 +    int sort_bypattern;
 +    /* Virtual memory configuration */
 +    int vm_enabled;
 +    char *vm_swap_file;
 +    off_t vm_page_size;
 +    off_t vm_pages;
 +    unsigned long long vm_max_memory;
 +    /* Zip structure config */
 +    size_t hash_max_zipmap_entries;
 +    size_t hash_max_zipmap_value;
 +    size_t list_max_ziplist_entries;
 +    size_t list_max_ziplist_value;
++    size_t set_max_intset_entries;
 +    /* Virtual memory state */
 +    FILE *vm_fp;
 +    int vm_fd;
 +    off_t vm_next_page; /* Next probably empty page */
 +    off_t vm_near_pages; /* Number of pages allocated sequentially */
 +    unsigned char *vm_bitmap; /* Bitmap of free/used pages */
 +    time_t unixtime;    /* Unix time sampled every second. */
 +    /* Virtual memory I/O threads stuff */
 +    /* An I/O thread process an element taken from the io_jobs queue and
 +     * put the result of the operation in the io_done list. While the
 +     * job is being processed, it's put on io_processing queue. */
 +    list *io_newjobs; /* List of VM I/O jobs yet to be processed */
 +    list *io_processing; /* List of VM I/O jobs being processed */
 +    list *io_processed; /* List of VM I/O jobs already processed */
 +    list *io_ready_clients; /* Clients ready to be unblocked. All keys loaded */
 +    pthread_mutex_t io_mutex; /* lock to access io_jobs/io_done/io_thread_job */
 +    pthread_mutex_t obj_freelist_mutex; /* safe redis objects creation/free */
 +    pthread_mutex_t io_swapfile_mutex; /* So we can lseek + write */
 +    pthread_attr_t io_threads_attr; /* attributes for threads creation */
 +    int io_active_threads; /* Number of running I/O threads */
 +    int vm_max_threads; /* Max number of I/O threads running at the same time */
 +    /* Our main thread is blocked on the event loop, locking for sockets ready
 +     * to be read or written, so when a threaded I/O operation is ready to be
 +     * processed by the main thread, the I/O thread will use a unix pipe to
 +     * awake the main thread. The followings are the two pipe FDs. */
 +    int io_ready_pipe_read;
 +    int io_ready_pipe_write;
 +    /* Virtual memory stats */
 +    unsigned long long vm_stats_used_pages;
 +    unsigned long long vm_stats_swapped_objects;
 +    unsigned long long vm_stats_swapouts;
 +    unsigned long long vm_stats_swapins;
 +    /* Pubsub */
 +    dict *pubsub_channels; /* Map channels to list of subscribed clients */
 +    list *pubsub_patterns; /* A list of pubsub_patterns */
 +    /* Misc */
 +    FILE *devnull;
 +    unsigned lruclock:22;        /* clock incrementing every minute, for LRU */
 +    unsigned lruclock_padding:10;
 +};
 +
 +typedef struct pubsubPattern {
 +    redisClient *client;
 +    robj *pattern;
 +} pubsubPattern;
 +
 +typedef void redisCommandProc(redisClient *c);
 +typedef void redisVmPreloadProc(redisClient *c, struct redisCommand *cmd, int argc, robj **argv);
 +struct redisCommand {
 +    char *name;
 +    redisCommandProc *proc;
 +    int arity;
 +    int flags;
 +    /* Use a function to determine which keys need to be loaded
 +     * in the background prior to executing this command. Takes precedence
 +     * over vm_firstkey and others, ignored when NULL */
 +    redisVmPreloadProc *vm_preload_proc;
 +    /* What keys should be loaded in background when calling this command? */
 +    int vm_firstkey; /* The first argument that's a key (0 = no keys) */
 +    int vm_lastkey;  /* THe last argument that's a key */
 +    int vm_keystep;  /* The step between first and last key */
 +};
 +
 +struct redisFunctionSym {
 +    char *name;
 +    unsigned long pointer;
 +};
 +
 +typedef struct _redisSortObject {
 +    robj *obj;
 +    union {
 +        double score;
 +        robj *cmpobj;
 +    } u;
 +} redisSortObject;
 +
 +typedef struct _redisSortOperation {
 +    int type;
 +    robj *pattern;
 +} redisSortOperation;
 +
 +/* ZSETs use a specialized version of Skiplists */
 +
 +typedef struct zskiplistNode {
 +    struct zskiplistNode **forward;
 +    struct zskiplistNode *backward;
 +    unsigned int *span;
 +    double score;
 +    robj *obj;
 +} zskiplistNode;
 +
 +typedef struct zskiplist {
 +    struct zskiplistNode *header, *tail;
 +    unsigned long length;
 +    int level;
 +} zskiplist;
 +
 +typedef struct zset {
 +    dict *dict;
 +    zskiplist *zsl;
 +} zset;
 +
 +/* VM threaded I/O request message */
 +#define REDIS_IOJOB_LOAD 0          /* Load from disk to memory */
 +#define REDIS_IOJOB_PREPARE_SWAP 1  /* Compute needed pages */
 +#define REDIS_IOJOB_DO_SWAP 2       /* Swap from memory to disk */
 +typedef struct iojob {
 +    int type;   /* Request type, REDIS_IOJOB_* */
 +    redisDb *db;/* Redis database */
 +    robj *key;  /* This I/O request is about swapping this key */
 +    robj *id;   /* Unique identifier of this job:
 +                   this is the object to swap for REDIS_IOREQ_*_SWAP, or the
 +                   vmpointer objct for REDIS_IOREQ_LOAD. */
 +    robj *val;  /* the value to swap for REDIS_IOREQ_*_SWAP, otherwise this
 +                 * field is populated by the I/O thread for REDIS_IOREQ_LOAD. */
 +    off_t page; /* Swap page where to read/write the object */
 +    off_t pages; /* Swap pages needed to save object. PREPARE_SWAP return val */
 +    int canceled; /* True if this command was canceled by blocking side of VM */
 +    pthread_t thread; /* ID of the thread processing this entry */
 +} iojob;
 +
 +/* Structure to hold list iteration abstraction. */
 +typedef struct {
 +    robj *subject;
 +    unsigned char encoding;
 +    unsigned char direction; /* Iteration direction */
 +    unsigned char *zi;
 +    listNode *ln;
 +} listTypeIterator;
 +
 +/* Structure for an entry while iterating over a list. */
 +typedef struct {
 +    listTypeIterator *li;
 +    unsigned char *zi;  /* Entry in ziplist */
 +    listNode *ln;       /* Entry in linked list */
 +} listTypeEntry;
 +
++/* Structure to hold set iteration abstraction. */
++typedef struct {
++    robj *subject;
++    int encoding;
++    int ii; /* intset iterator */
++    dictIterator *di;
++} setIterator;
++
 +/* Structure to hold hash iteration abstration. Note that iteration over
 + * hashes involves both fields and values. Because it is possible that
 + * not both are required, store pointers in the iterator to avoid
 + * unnecessary memory allocation for fields/values. */
 +typedef struct {
 +    int encoding;
 +    unsigned char *zi;
 +    unsigned char *zk, *zv;
 +    unsigned int zklen, zvlen;
 +
 +    dictIterator *di;
 +    dictEntry *de;
 +} hashTypeIterator;
 +
 +#define REDIS_HASH_KEY 1
 +#define REDIS_HASH_VALUE 2
 +
 +/*-----------------------------------------------------------------------------
 + * Extern declarations
 + *----------------------------------------------------------------------------*/
 +
 +extern struct redisServer server;
 +extern struct sharedObjectsStruct shared;
 +extern dictType setDictType;
 +extern dictType zsetDictType;
 +extern double R_Zero, R_PosInf, R_NegInf, R_Nan;
 +dictType hashDictType;
 +
 +/*-----------------------------------------------------------------------------
 + * Functions prototypes
 + *----------------------------------------------------------------------------*/
 +
 +/* networking.c -- Networking and Client related operations */
 +redisClient *createClient(int fd);
 +void closeTimedoutClients(void);
 +void freeClient(redisClient *c);
 +void resetClient(redisClient *c);
 +void sendReplyToClient(aeEventLoop *el, int fd, void *privdata, int mask);
 +void sendReplyToClientWritev(aeEventLoop *el, int fd, void *privdata, int mask);
 +void addReply(redisClient *c, robj *obj);
 +void addReplySds(redisClient *c, sds s);
 +void processInputBuffer(redisClient *c);
 +void acceptHandler(aeEventLoop *el, int fd, void *privdata, int mask);
 +void readQueryFromClient(aeEventLoop *el, int fd, void *privdata, int mask);
 +void addReplyBulk(redisClient *c, robj *obj);
 +void addReplyBulkCString(redisClient *c, char *s);
 +void acceptHandler(aeEventLoop *el, int fd, void *privdata, int mask);
 +void addReply(redisClient *c, robj *obj);
 +void addReplySds(redisClient *c, sds s);
 +void addReplyDouble(redisClient *c, double d);
 +void addReplyLongLong(redisClient *c, long long ll);
 +void addReplyUlong(redisClient *c, unsigned long ul);
 +void *dupClientReplyValue(void *o);
 +
 +/* List data type */
 +void listTypeTryConversion(robj *subject, robj *value);
 +void listTypePush(robj *subject, robj *value, int where);
 +robj *listTypePop(robj *subject, int where);
 +unsigned long listTypeLength(robj *subject);
 +listTypeIterator *listTypeInitIterator(robj *subject, int index, unsigned char direction);
 +void listTypeReleaseIterator(listTypeIterator *li);
 +int listTypeNext(listTypeIterator *li, listTypeEntry *entry);
 +robj *listTypeGet(listTypeEntry *entry);
 +void listTypeInsert(listTypeEntry *entry, robj *value, int where);
 +int listTypeEqual(listTypeEntry *entry, robj *o);
 +void listTypeDelete(listTypeEntry *entry);
 +void listTypeConvert(robj *subject, int enc);
 +void unblockClientWaitingData(redisClient *c);
 +int handleClientsWaitingListPush(redisClient *c, robj *key, robj *ele);
 +void popGenericCommand(redisClient *c, int where);
 +
 +/* MULTI/EXEC/WATCH... */
 +void unwatchAllKeys(redisClient *c);
 +void initClientMultiState(redisClient *c);
 +void freeClientMultiState(redisClient *c);
 +void queueMultiCommand(redisClient *c, struct redisCommand *cmd);
 +void touchWatchedKey(redisDb *db, robj *key);
 +void touchWatchedKeysOnFlush(int dbid);
 +
 +/* Redis object implementation */
 +void decrRefCount(void *o);
 +void incrRefCount(robj *o);
 +void freeStringObject(robj *o);
 +void freeListObject(robj *o);
 +void freeSetObject(robj *o);
 +void freeZsetObject(robj *o);
 +void freeHashObject(robj *o);
 +robj *createObject(int type, void *ptr);
 +robj *createStringObject(char *ptr, size_t len);
 +robj *dupStringObject(robj *o);
 +robj *tryObjectEncoding(robj *o);
 +robj *getDecodedObject(robj *o);
 +size_t stringObjectLen(robj *o);
 +int tryFreeOneObjectFromFreelist(void);
 +robj *createStringObjectFromLongLong(long long value);
 +robj *createListObject(void);
 +robj *createZiplistObject(void);
 +robj *createSetObject(void);
++robj *createIntsetObject(void);
 +robj *createHashObject(void);
 +robj *createZsetObject(void);
 +int getLongFromObjectOrReply(redisClient *c, robj *o, long *target, const char *msg);
 +int checkType(redisClient *c, robj *o, int type);
 +int getLongLongFromObjectOrReply(redisClient *c, robj *o, long long *target, const char *msg);
 +int getDoubleFromObjectOrReply(redisClient *c, robj *o, double *target, const char *msg);
 +int getLongLongFromObject(robj *o, long long *target);
 +char *strEncoding(int encoding);
 +int compareStringObjects(robj *a, robj *b);
 +int equalStringObjects(robj *a, robj *b);
 +
 +/* Replication */
 +void replicationFeedSlaves(list *slaves, int dictid, robj **argv, int argc);
 +void replicationFeedMonitors(list *monitors, int dictid, robj **argv, int argc);
 +int syncWithMaster(void);
 +void updateSlavesWaitingBgsave(int bgsaveerr);
 +
 +/* RDB persistence */
 +int rdbLoad(char *filename);
 +int rdbSaveBackground(char *filename);
 +void rdbRemoveTempFile(pid_t childpid);
 +int rdbSave(char *filename);
 +int rdbSaveObject(FILE *fp, robj *o);
 +off_t rdbSavedObjectPages(robj *o, FILE *fp);
 +off_t rdbSavedObjectLen(robj *o, FILE *fp);
 +robj *rdbLoadObject(int type, FILE *fp);
 +void backgroundSaveDoneHandler(int statloc);
 +
 +/* AOF persistence */
 +void flushAppendOnlyFile(void);
 +void feedAppendOnlyFile(struct redisCommand *cmd, int dictid, robj **argv, int argc);
 +void aofRemoveTempFile(pid_t childpid);
 +int rewriteAppendOnlyFileBackground(void);
 +int loadAppendOnlyFile(char *filename);
 +void stopAppendOnly(void);
 +int startAppendOnly(void);
 +void backgroundRewriteDoneHandler(int statloc);
 +
 +/* Sorted sets data type */
 +zskiplist *zslCreate(void);
 +void zslFree(zskiplist *zsl);
 +void zslInsert(zskiplist *zsl, double score, robj *obj);
 +
 +/* Core functions */
 +void freeMemoryIfNeeded(void);
 +int processCommand(redisClient *c);
 +void setupSigSegvAction(void);
 +struct redisCommand *lookupCommand(char *name);
 +void call(redisClient *c, struct redisCommand *cmd);
 +int prepareForShutdown();
 +void redisLog(int level, const char *fmt, ...);
 +void usage();
 +void updateDictResizePolicy(void);
 +int htNeedsResize(dict *dict);
 +void oom(const char *msg);
 +
 +/* Virtual Memory */
 +void vmInit(void);
 +void vmMarkPagesFree(off_t page, off_t count);
 +robj *vmLoadObject(robj *o);
 +robj *vmPreviewObject(robj *o);
 +int vmSwapOneObjectBlocking(void);
 +int vmSwapOneObjectThreaded(void);
 +int vmCanSwapOut(void);
 +void vmThreadedIOCompletedJob(aeEventLoop *el, int fd, void *privdata, int mask);
 +void vmCancelThreadedIOJob(robj *o);
 +void lockThreadedIO(void);
 +void unlockThreadedIO(void);
 +int vmSwapObjectThreaded(robj *key, robj *val, redisDb *db);
 +void freeIOJob(iojob *j);
 +void queueIOJob(iojob *j);
 +int vmWriteObjectOnSwap(robj *o, off_t page);
 +robj *vmReadObjectFromSwap(off_t page, int type);
 +void waitEmptyIOJobsQueue(void);
 +void vmReopenSwapFile(void);
 +int vmFreePage(off_t page);
 +void zunionInterBlockClientOnSwappedKeys(redisClient *c, struct redisCommand *cmd, int argc, robj **argv);
 +void execBlockClientOnSwappedKeys(redisClient *c, struct redisCommand *cmd, int argc, robj **argv);
 +int blockClientOnSwappedKeys(redisClient *c, struct redisCommand *cmd);
 +int dontWaitForSwappedKey(redisClient *c, robj *key);
 +void handleClientsBlockedOnSwappedKey(redisDb *db, robj *key);
 +vmpointer *vmSwapObjectBlocking(robj *val);
 +
++/* Set data type */
++robj *setTypeCreate(robj *value);
++int setTypeAdd(robj *subject, robj *value);
++int setTypeRemove(robj *subject, robj *value);
++int setTypeIsMember(robj *subject, robj *value);
++setIterator *setTypeInitIterator(robj *subject);
++void setTypeReleaseIterator(setIterator *si);
++robj *setTypeNext(setIterator *si);
++robj *setTypeRandomElement(robj *subject);
++unsigned long setTypeSize(robj *subject);
++void setTypeConvert(robj *subject, int enc);
++
 +/* Hash data type */
 +void convertToRealHash(robj *o);
 +void hashTypeTryConversion(robj *subject, robj **argv, int start, int end);
 +void hashTypeTryObjectEncoding(robj *subject, robj **o1, robj **o2);
 +robj *hashTypeGet(robj *o, robj *key);
 +int hashTypeExists(robj *o, robj *key);
 +int hashTypeSet(robj *o, robj *key, robj *value);
 +int hashTypeDelete(robj *o, robj *key);
 +unsigned long hashTypeLength(robj *o);
 +hashTypeIterator *hashTypeInitIterator(robj *subject);
 +void hashTypeReleaseIterator(hashTypeIterator *hi);
 +int hashTypeNext(hashTypeIterator *hi);
 +robj *hashTypeCurrent(hashTypeIterator *hi, int what);
 +robj *hashTypeLookupWriteOrCreate(redisClient *c, robj *key);
 +
 +/* Pub / Sub */
 +int pubsubUnsubscribeAllChannels(redisClient *c, int notify);
 +int pubsubUnsubscribeAllPatterns(redisClient *c, int notify);
 +void freePubsubPattern(void *p);
 +int listMatchPubsubPattern(void *a, void *b);
 +
 +/* Utility functions */
 +int stringmatchlen(const char *pattern, int patternLen,
 +        const char *string, int stringLen, int nocase);
 +int stringmatch(const char *pattern, const char *string, int nocase);
 +long long memtoll(const char *p, int *err);
 +int ll2string(char *s, size_t len, long long value);
 +int isStringRepresentableAsLong(sds s, long *longval);
 +
 +/* Configuration */
 +void loadServerConfig(char *filename);
 +void appendServerSaveParams(time_t seconds, int changes);
 +void resetServerSaveParams();
 +
 +/* db.c -- Keyspace access API */
 +int removeExpire(redisDb *db, robj *key);
 +int expireIfNeeded(redisDb *db, robj *key);
 +int deleteIfVolatile(redisDb *db, robj *key);
 +time_t getExpire(redisDb *db, robj *key);
 +int setExpire(redisDb *db, robj *key, time_t when);
 +robj *lookupKey(redisDb *db, robj *key);
 +robj *lookupKeyRead(redisDb *db, robj *key);
 +robj *lookupKeyWrite(redisDb *db, robj *key);
 +robj *lookupKeyReadOrReply(redisClient *c, robj *key, robj *reply);
 +robj *lookupKeyWriteOrReply(redisClient *c, robj *key, robj *reply);
 +int dbAdd(redisDb *db, robj *key, robj *val);
 +int dbReplace(redisDb *db, robj *key, robj *val);
 +int dbExists(redisDb *db, robj *key);
 +robj *dbRandomKey(redisDb *db);
 +int dbDelete(redisDb *db, robj *key);
 +long long emptyDb();
 +int selectDb(redisClient *c, int id);
 +
 +/* Git SHA1 */
 +char *redisGitSHA1(void);
 +char *redisGitDirty(void);
 +
 +/* Commands prototypes */
 +void authCommand(redisClient *c);
 +void pingCommand(redisClient *c);
 +void echoCommand(redisClient *c);
 +void setCommand(redisClient *c);
 +void setnxCommand(redisClient *c);
 +void setexCommand(redisClient *c);
 +void getCommand(redisClient *c);
 +void delCommand(redisClient *c);
 +void existsCommand(redisClient *c);
 +void incrCommand(redisClient *c);
 +void decrCommand(redisClient *c);
 +void incrbyCommand(redisClient *c);
 +void decrbyCommand(redisClient *c);
 +void selectCommand(redisClient *c);
 +void randomkeyCommand(redisClient *c);
 +void keysCommand(redisClient *c);
 +void dbsizeCommand(redisClient *c);
 +void lastsaveCommand(redisClient *c);
 +void saveCommand(redisClient *c);
 +void bgsaveCommand(redisClient *c);
 +void bgrewriteaofCommand(redisClient *c);
 +void shutdownCommand(redisClient *c);
 +void moveCommand(redisClient *c);
 +void renameCommand(redisClient *c);
 +void renamenxCommand(redisClient *c);
 +void lpushCommand(redisClient *c);
 +void rpushCommand(redisClient *c);
 +void lpushxCommand(redisClient *c);
 +void rpushxCommand(redisClient *c);
 +void linsertCommand(redisClient *c);
 +void lpopCommand(redisClient *c);
 +void rpopCommand(redisClient *c);
 +void llenCommand(redisClient *c);
 +void lindexCommand(redisClient *c);
 +void lrangeCommand(redisClient *c);
 +void ltrimCommand(redisClient *c);
 +void typeCommand(redisClient *c);
 +void lsetCommand(redisClient *c);
 +void saddCommand(redisClient *c);
 +void sremCommand(redisClient *c);
 +void smoveCommand(redisClient *c);
 +void sismemberCommand(redisClient *c);
 +void scardCommand(redisClient *c);
 +void spopCommand(redisClient *c);
 +void srandmemberCommand(redisClient *c);
 +void sinterCommand(redisClient *c);
 +void sinterstoreCommand(redisClient *c);
 +void sunionCommand(redisClient *c);
 +void sunionstoreCommand(redisClient *c);
 +void sdiffCommand(redisClient *c);
 +void sdiffstoreCommand(redisClient *c);
 +void syncCommand(redisClient *c);
 +void flushdbCommand(redisClient *c);
 +void flushallCommand(redisClient *c);
 +void sortCommand(redisClient *c);
 +void lremCommand(redisClient *c);
 +void rpoplpushcommand(redisClient *c);
 +void infoCommand(redisClient *c);
 +void mgetCommand(redisClient *c);
 +void monitorCommand(redisClient *c);
 +void expireCommand(redisClient *c);
 +void expireatCommand(redisClient *c);
 +void getsetCommand(redisClient *c);
 +void ttlCommand(redisClient *c);
 +void slaveofCommand(redisClient *c);
 +void debugCommand(redisClient *c);
 +void msetCommand(redisClient *c);
 +void msetnxCommand(redisClient *c);
 +void zaddCommand(redisClient *c);
 +void zincrbyCommand(redisClient *c);
 +void zrangeCommand(redisClient *c);
 +void zrangebyscoreCommand(redisClient *c);
 +void zcountCommand(redisClient *c);
 +void zrevrangeCommand(redisClient *c);
 +void zcardCommand(redisClient *c);
 +void zremCommand(redisClient *c);
 +void zscoreCommand(redisClient *c);
 +void zremrangebyscoreCommand(redisClient *c);
 +void multiCommand(redisClient *c);
 +void execCommand(redisClient *c);
 +void discardCommand(redisClient *c);
 +void blpopCommand(redisClient *c);
 +void brpopCommand(redisClient *c);
 +void appendCommand(redisClient *c);
 +void substrCommand(redisClient *c);
 +void zrankCommand(redisClient *c);
 +void zrevrankCommand(redisClient *c);
 +void hsetCommand(redisClient *c);
 +void hsetnxCommand(redisClient *c);
 +void hgetCommand(redisClient *c);
 +void hmsetCommand(redisClient *c);
 +void hmgetCommand(redisClient *c);
 +void hdelCommand(redisClient *c);
 +void hlenCommand(redisClient *c);
 +void zremrangebyrankCommand(redisClient *c);
 +void zunionstoreCommand(redisClient *c);
 +void zinterstoreCommand(redisClient *c);
 +void hkeysCommand(redisClient *c);
 +void hvalsCommand(redisClient *c);
 +void hgetallCommand(redisClient *c);
 +void hexistsCommand(redisClient *c);
 +void configCommand(redisClient *c);
 +void hincrbyCommand(redisClient *c);
 +void subscribeCommand(redisClient *c);
 +void unsubscribeCommand(redisClient *c);
 +void psubscribeCommand(redisClient *c);
 +void punsubscribeCommand(redisClient *c);
 +void publishCommand(redisClient *c);
 +void watchCommand(redisClient *c);
 +void unwatchCommand(redisClient *c);
 +
 +#endif
diff --cc src/t_set.c
index 808ef268,00000000..3fbf13a3
mode 100644,000000..100644
--- a/src/t_set.c
+++ b/src/t_set.c
@@@ -1,349 -1,0 +1,503 @@@
 +#include "redis.h"
 +
 +/*-----------------------------------------------------------------------------
 + * Set Commands
 + *----------------------------------------------------------------------------*/
 +
++/* Factory method to return a set that *can* hold "value". When the object has
++ * an integer-encodable value, an intset will be returned. Otherwise a regular
++ * hash table. */
++robj *setTypeCreate(robj *value) {
++    if (getLongLongFromObject(value,NULL) == REDIS_OK)
++        return createIntsetObject();
++    return createSetObject();
++}
++
++int setTypeAdd(robj *subject, robj *value) {
++    long long llval;
++    if (subject->encoding == REDIS_ENCODING_HT) {
++        if (dictAdd(subject->ptr,value,NULL) == DICT_OK) {
++            incrRefCount(value);
++            return 1;
++        }
++    } else if (subject->encoding == REDIS_ENCODING_INTSET) {
++        if (getLongLongFromObject(value,&llval) == REDIS_OK) {
++            uint8_t success = 0;
++            subject->ptr = intsetAdd(subject->ptr,llval,&success);
++            if (success) {
++                /* Convert to regular set when the intset contains
++                 * too many entries. */
++                if (intsetLen(subject->ptr) > server.set_max_intset_entries)
++                    setTypeConvert(subject,REDIS_ENCODING_HT);
++                return 1;
++            }
++        } else {
++            /* Failed to get integer from object, convert to regular set. */
++            setTypeConvert(subject,REDIS_ENCODING_HT);
++
++            /* The set *was* an intset and this value is not integer
++             * encodable, so dictAdd should always work. */
++            redisAssert(dictAdd(subject->ptr,value,NULL) == DICT_OK);
++            incrRefCount(value);
++            return 1;
++        }
++    } else {
++        redisPanic("Unknown set encoding");
++    }
++    return 0;
++}
++
++int setTypeRemove(robj *subject, robj *value) {
++    long long llval;
++    if (subject->encoding == REDIS_ENCODING_HT) {
++        if (dictDelete(subject->ptr,value) == DICT_OK) {
++            if (htNeedsResize(subject->ptr)) dictResize(subject->ptr);
++            return 1;
++        }
++    } else if (subject->encoding == REDIS_ENCODING_INTSET) {
++        if (getLongLongFromObject(value,&llval) == REDIS_OK) {
++            uint8_t success;
++            subject->ptr = intsetRemove(subject->ptr,llval,&success);
++            if (success) return 1;
++        }
++    } else {
++        redisPanic("Unknown set encoding");
++    }
++    return 0;
++}
++
++int setTypeIsMember(robj *subject, robj *value) {
++    long long llval;
++    if (subject->encoding == REDIS_ENCODING_HT) {
++        return dictFind((dict*)subject->ptr,value) != NULL;
++    } else if (subject->encoding == REDIS_ENCODING_INTSET) {
++        if (getLongLongFromObject(value,&llval) == REDIS_OK) {
++            return intsetFind((intset*)subject->ptr,llval);
++        }
++    } else {
++        redisPanic("Unknown set encoding");
++    }
++    return 0;
++}
++
++setIterator *setTypeInitIterator(robj *subject) {
++    setIterator *si = zmalloc(sizeof(setIterator));
++    si->subject = subject;
++    si->encoding = subject->encoding;
++    if (si->encoding == REDIS_ENCODING_HT) {
++        si->di = dictGetIterator(subject->ptr);
++    } else if (si->encoding == REDIS_ENCODING_INTSET) {
++        si->ii = 0;
++    } else {
++        redisPanic("Unknown set encoding");
++    }
++    return si;
++}
++
++void setTypeReleaseIterator(setIterator *si) {
++    if (si->encoding == REDIS_ENCODING_HT)
++        dictReleaseIterator(si->di);
++    zfree(si);
++}
++
++/* Move to the next entry in the set. Returns the object at the current
++ * position, or NULL when the end is reached. This object will have its
++ * refcount incremented, so the caller needs to take care of this. */
++robj *setTypeNext(setIterator *si) {
++    robj *ret = NULL;
++    if (si->encoding == REDIS_ENCODING_HT) {
++        dictEntry *de = dictNext(si->di);
++        if (de != NULL) {
++            ret = dictGetEntryKey(de);
++            incrRefCount(ret);
++        }
++    } else if (si->encoding == REDIS_ENCODING_INTSET) {
++        long long llval;
++        if (intsetGet(si->subject->ptr,si->ii++,&llval))
++            ret = createStringObjectFromLongLong(llval);
++    }
++    return ret;
++}
++
++
++/* Return random element from set. The returned object will always have
++ * an incremented refcount. */
++robj *setTypeRandomElement(robj *subject) {
++    robj *ret = NULL;
++    if (subject->encoding == REDIS_ENCODING_HT) {
++        dictEntry *de = dictGetRandomKey(subject->ptr);
++        ret = dictGetEntryKey(de);
++        incrRefCount(ret);
++    } else if (subject->encoding == REDIS_ENCODING_INTSET) {
++        long long llval = intsetRandom(subject->ptr);
++        ret = createStringObjectFromLongLong(llval);
++    } else {
++        redisPanic("Unknown set encoding");
++    }
++    return ret;
++}
++
++unsigned long setTypeSize(robj *subject) {
++    if (subject->encoding == REDIS_ENCODING_HT) {
++        return dictSize((dict*)subject->ptr);
++    } else if (subject->encoding == REDIS_ENCODING_INTSET) {
++        return intsetLen((intset*)subject->ptr);
++    } else {
++        redisPanic("Unknown set encoding");
++    }
++}
++
++/* Convert the set to specified encoding. The resulting dict (when converting
++ * to a hashtable) is presized to hold the number of elements in the original
++ * set. */
++void setTypeConvert(robj *subject, int enc) {
++    setIterator *si;
++    robj *element;
++    redisAssert(subject->type == REDIS_SET);
++
++    if (enc == REDIS_ENCODING_HT) {
++        dict *d = dictCreate(&setDictType,NULL);
++        /* Presize the dict to avoid rehashing */
++        dictExpand(d,intsetLen(subject->ptr));
++
++        /* setTypeGet returns a robj with incremented refcount */
++        si = setTypeInitIterator(subject);
++        while ((element = setTypeNext(si)) != NULL)
++            redisAssert(dictAdd(d,element,NULL) == DICT_OK);
++        setTypeReleaseIterator(si);
++
++        subject->encoding = REDIS_ENCODING_HT;
++        zfree(subject->ptr);
++        subject->ptr = d;
++    } else {
++        redisPanic("Unsupported set conversion");
++    }
++}
++
 +void saddCommand(redisClient *c) {
 +    robj *set;
 +
 +    set = lookupKeyWrite(c->db,c->argv[1]);
 +    if (set == NULL) {
-         set = createSetObject();
++        set = setTypeCreate(c->argv[2]);
 +        dbAdd(c->db,c->argv[1],set);
 +    } else {
 +        if (set->type != REDIS_SET) {
 +            addReply(c,shared.wrongtypeerr);
 +            return;
 +        }
 +    }
-     if (dictAdd(set->ptr,c->argv[2],NULL) == DICT_OK) {
-         incrRefCount(c->argv[2]);
++    if (setTypeAdd(set,c->argv[2])) {
 +        server.dirty++;
 +        addReply(c,shared.cone);
 +    } else {
 +        addReply(c,shared.czero);
 +    }
 +}
 +
 +void sremCommand(redisClient *c) {
 +    robj *set;
 +
 +    if ((set = lookupKeyWriteOrReply(c,c->argv[1],shared.czero)) == NULL ||
 +        checkType(c,set,REDIS_SET)) return;
 +
-     if (dictDelete(set->ptr,c->argv[2]) == DICT_OK) {
++    if (setTypeRemove(set,c->argv[2])) {
++        if (setTypeSize(set) == 0) dbDelete(c->db,c->argv[1]);
 +        server.dirty++;
-         if (htNeedsResize(set->ptr)) dictResize(set->ptr);
-         if (dictSize((dict*)set->ptr) == 0) dbDelete(c->db,c->argv[1]);
 +        addReply(c,shared.cone);
 +    } else {
 +        addReply(c,shared.czero);
 +    }
 +}
 +
 +void smoveCommand(redisClient *c) {
-     robj *srcset, *dstset;
- 
++    robj *srcset, *dstset, *ele;
 +    srcset = lookupKeyWrite(c->db,c->argv[1]);
 +    dstset = lookupKeyWrite(c->db,c->argv[2]);
++    ele = c->argv[3];
 +
-     /* If the source key does not exist return 0, if it's of the wrong type
-      * raise an error */
-     if (srcset == NULL || srcset->type != REDIS_SET) {
-         addReply(c, srcset ? shared.wrongtypeerr : shared.czero);
++    /* If the source key does not exist return 0 */
++    if (srcset == NULL) {
++        addReply(c,shared.czero);
 +        return;
 +    }
-     /* Error if the destination key is not a set as well */
-     if (dstset && dstset->type != REDIS_SET) {
-         addReply(c,shared.wrongtypeerr);
++
++    /* If the source key has the wrong type, or the destination key
++     * is set and has the wrong type, return with an error. */
++    if (checkType(c,srcset,REDIS_SET) ||
++        (dstset && checkType(c,dstset,REDIS_SET))) return;
++
++    /* If srcset and dstset are equal, SMOVE is a no-op */
++    if (srcset == dstset) {
++        addReply(c,shared.cone);
 +        return;
 +    }
-     /* Remove the element from the source set */
-     if (dictDelete(srcset->ptr,c->argv[3]) == DICT_ERR) {
-         /* Key not found in the src set! return zero */
++
++    /* If the element cannot be removed from the src set, return 0. */
++    if (!setTypeRemove(srcset,ele)) {
 +        addReply(c,shared.czero);
 +        return;
 +    }
-     if (dictSize((dict*)srcset->ptr) == 0 && srcset != dstset)
-         dbDelete(c->db,c->argv[1]);
++
++    /* Remove the src set from the database when empty */
++    if (setTypeSize(srcset) == 0) dbDelete(c->db,c->argv[1]);
 +    server.dirty++;
-     /* Add the element to the destination set */
++
++    /* Create the destination set when it doesn't exist */
 +    if (!dstset) {
-         dstset = createSetObject();
++        dstset = setTypeCreate(ele);
 +        dbAdd(c->db,c->argv[2],dstset);
 +    }
-     if (dictAdd(dstset->ptr,c->argv[3],NULL) == DICT_OK)
-         incrRefCount(c->argv[3]);
++
++    /* An extra key has changed when ele was successfully added to dstset */
++    if (setTypeAdd(dstset,ele)) server.dirty++;
 +    addReply(c,shared.cone);
 +}
 +
 +void sismemberCommand(redisClient *c) {
 +    robj *set;
 +
 +    if ((set = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL ||
 +        checkType(c,set,REDIS_SET)) return;
 +
-     if (dictFind(set->ptr,c->argv[2]))
++    if (setTypeIsMember(set,c->argv[2]))
 +        addReply(c,shared.cone);
 +    else
 +        addReply(c,shared.czero);
 +}
 +
 +void scardCommand(redisClient *c) {
 +    robj *o;
-     dict *s;
 +
 +    if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL ||
 +        checkType(c,o,REDIS_SET)) return;
 +
-     s = o->ptr;
-     addReplyUlong(c,dictSize(s));
++    addReplyUlong(c,setTypeSize(o));
 +}
 +
 +void spopCommand(redisClient *c) {
-     robj *set;
-     dictEntry *de;
++    robj *set, *ele;
 +
 +    if ((set = lookupKeyWriteOrReply(c,c->argv[1],shared.nullbulk)) == NULL ||
 +        checkType(c,set,REDIS_SET)) return;
 +
-     de = dictGetRandomKey(set->ptr);
-     if (de == NULL) {
++    ele = setTypeRandomElement(set);
++    if (ele == NULL) {
 +        addReply(c,shared.nullbulk);
 +    } else {
-         robj *ele = dictGetEntryKey(de);
- 
++        setTypeRemove(set,ele);
 +        addReplyBulk(c,ele);
-         dictDelete(set->ptr,ele);
-         if (htNeedsResize(set->ptr)) dictResize(set->ptr);
-         if (dictSize((dict*)set->ptr) == 0) dbDelete(c->db,c->argv[1]);
++        decrRefCount(ele);
++        if (setTypeSize(set) == 0) dbDelete(c->db,c->argv[1]);
 +        server.dirty++;
 +    }
 +}
 +
 +void srandmemberCommand(redisClient *c) {
-     robj *set;
-     dictEntry *de;
++    robj *set, *ele;
 +
 +    if ((set = lookupKeyReadOrReply(c,c->argv[1],shared.nullbulk)) == NULL ||
 +        checkType(c,set,REDIS_SET)) return;
 +
-     de = dictGetRandomKey(set->ptr);
-     if (de == NULL) {
++    ele = setTypeRandomElement(set);
++    if (ele == NULL) {
 +        addReply(c,shared.nullbulk);
 +    } else {
-         robj *ele = dictGetEntryKey(de);
- 
 +        addReplyBulk(c,ele);
++        decrRefCount(ele);
 +    }
 +}
 +
 +int qsortCompareSetsByCardinality(const void *s1, const void *s2) {
-     dict **d1 = (void*) s1, **d2 = (void*) s2;
- 
-     return dictSize(*d1)-dictSize(*d2);
++    return setTypeSize(*(robj**)s1)-setTypeSize(*(robj**)s2);
 +}
 +
- void sinterGenericCommand(redisClient *c, robj **setskeys, unsigned long setsnum, robj *dstkey) {
-     dict **dv = zmalloc(sizeof(dict*)*setsnum);
-     dictIterator *di;
-     dictEntry *de;
-     robj *lenobj = NULL, *dstset = NULL;
++void sinterGenericCommand(redisClient *c, robj **setkeys, unsigned long setnum, robj *dstkey) {
++    robj **sets = zmalloc(sizeof(robj*)*setnum);
++    setIterator *si;
++    robj *ele, *lenobj = NULL, *dstset = NULL;
 +    unsigned long j, cardinality = 0;
 +
-     for (j = 0; j < setsnum; j++) {
-         robj *setobj;
- 
-         setobj = dstkey ?
-                     lookupKeyWrite(c->db,setskeys[j]) :
-                     lookupKeyRead(c->db,setskeys[j]);
++    for (j = 0; j < setnum; j++) {
++        robj *setobj = dstkey ?
++            lookupKeyWrite(c->db,setkeys[j]) :
++            lookupKeyRead(c->db,setkeys[j]);
 +        if (!setobj) {
-             zfree(dv);
++            zfree(sets);
 +            if (dstkey) {
 +                if (dbDelete(c->db,dstkey))
 +                    server.dirty++;
 +                addReply(c,shared.czero);
 +            } else {
 +                addReply(c,shared.emptymultibulk);
 +            }
 +            return;
 +        }
-         if (setobj->type != REDIS_SET) {
-             zfree(dv);
-             addReply(c,shared.wrongtypeerr);
++        if (checkType(c,setobj,REDIS_SET)) {
++            zfree(sets);
 +            return;
 +        }
-         dv[j] = setobj->ptr;
++        sets[j] = setobj;
 +    }
 +    /* Sort sets from the smallest to largest, this will improve our
 +     * algorithm's performace */
-     qsort(dv,setsnum,sizeof(dict*),qsortCompareSetsByCardinality);
++    qsort(sets,setnum,sizeof(robj*),qsortCompareSetsByCardinality);
 +
 +    /* The first thing we should output is the total number of elements...
 +     * since this is a multi-bulk write, but at this stage we don't know
 +     * the intersection set size, so we use a trick, append an empty object
 +     * to the output list and save the pointer to later modify it with the
 +     * right length */
 +    if (!dstkey) {
 +        lenobj = createObject(REDIS_STRING,NULL);
 +        addReply(c,lenobj);
 +        decrRefCount(lenobj);
 +    } else {
 +        /* If we have a target key where to store the resulting set
 +         * create this key with an empty set inside */
-         dstset = createSetObject();
++        dstset = createIntsetObject();
 +    }
 +
 +    /* Iterate all the elements of the first (smallest) set, and test
 +     * the element against all the other sets, if at least one set does
 +     * not include the element it is discarded */
-     di = dictGetIterator(dv[0]);
- 
-     while((de = dictNext(di)) != NULL) {
-         robj *ele;
- 
-         for (j = 1; j < setsnum; j++)
-             if (dictFind(dv[j],dictGetEntryKey(de)) == NULL) break;
-         if (j != setsnum)
-             continue; /* at least one set does not contain the member */
-         ele = dictGetEntryKey(de);
-         if (!dstkey) {
-             addReplyBulk(c,ele);
-             cardinality++;
-         } else {
-             dictAdd(dstset->ptr,ele,NULL);
-             incrRefCount(ele);
++    si = setTypeInitIterator(sets[0]);
++    while((ele = setTypeNext(si)) != NULL) {
++        for (j = 1; j < setnum; j++)
++            if (!setTypeIsMember(sets[j],ele)) break;
++
++        /* Only take action when all sets contain the member */
++        if (j == setnum) {
++            if (!dstkey) {
++                addReplyBulk(c,ele);
++                cardinality++;
++            } else {
++                setTypeAdd(dstset,ele);
++            }
 +        }
++        decrRefCount(ele);
 +    }
-     dictReleaseIterator(di);
++    setTypeReleaseIterator(si);
 +
 +    if (dstkey) {
 +        /* Store the resulting set into the target, if the intersection
 +         * is not an empty set. */
 +        dbDelete(c->db,dstkey);
-         if (dictSize((dict*)dstset->ptr) > 0) {
++        if (setTypeSize(dstset) > 0) {
 +            dbAdd(c->db,dstkey,dstset);
-             addReplyLongLong(c,dictSize((dict*)dstset->ptr));
++            addReplyLongLong(c,setTypeSize(dstset));
 +        } else {
 +            decrRefCount(dstset);
 +            addReply(c,shared.czero);
 +        }
 +        server.dirty++;
 +    } else {
 +        lenobj->ptr = sdscatprintf(sdsempty(),"*%lu\r\n",cardinality);
 +    }
-     zfree(dv);
++    zfree(sets);
 +}
 +
 +void sinterCommand(redisClient *c) {
 +    sinterGenericCommand(c,c->argv+1,c->argc-1,NULL);
 +}
 +
 +void sinterstoreCommand(redisClient *c) {
 +    sinterGenericCommand(c,c->argv+2,c->argc-2,c->argv[1]);
 +}
 +
- void sunionDiffGenericCommand(redisClient *c, robj **setskeys, int setsnum, robj *dstkey, int op) {
-     dict **dv = zmalloc(sizeof(dict*)*setsnum);
-     dictIterator *di;
-     dictEntry *de;
-     robj *dstset = NULL;
-     int j, cardinality = 0;
++#define REDIS_OP_UNION 0
++#define REDIS_OP_DIFF 1
++#define REDIS_OP_INTER 2
 +
-     for (j = 0; j < setsnum; j++) {
-         robj *setobj;
++void sunionDiffGenericCommand(redisClient *c, robj **setkeys, int setnum, robj *dstkey, int op) {
++    robj **sets = zmalloc(sizeof(robj*)*setnum);
++    setIterator *si;
++    robj *ele, *dstset = NULL;
++    int j, cardinality = 0;
 +
-         setobj = dstkey ?
-                     lookupKeyWrite(c->db,setskeys[j]) :
-                     lookupKeyRead(c->db,setskeys[j]);
++    for (j = 0; j < setnum; j++) {
++        robj *setobj = dstkey ?
++            lookupKeyWrite(c->db,setkeys[j]) :
++            lookupKeyRead(c->db,setkeys[j]);
 +        if (!setobj) {
-             dv[j] = NULL;
++            sets[j] = NULL;
 +            continue;
 +        }
-         if (setobj->type != REDIS_SET) {
-             zfree(dv);
-             addReply(c,shared.wrongtypeerr);
++        if (checkType(c,setobj,REDIS_SET)) {
++            zfree(sets);
 +            return;
 +        }
-         dv[j] = setobj->ptr;
++        sets[j] = setobj;
 +    }
 +
 +    /* We need a temp set object to store our union. If the dstkey
 +     * is not NULL (that is, we are inside an SUNIONSTORE operation) then
 +     * this set object will be the resulting object to set into the target key*/
-     dstset = createSetObject();
++    dstset = createIntsetObject();
 +
 +    /* Iterate all the elements of all the sets, add every element a single
 +     * time to the result set */
-     for (j = 0; j < setsnum; j++) {
-         if (op == REDIS_OP_DIFF && j == 0 && !dv[j]) break; /* result set is empty */
-         if (!dv[j]) continue; /* non existing keys are like empty sets */
- 
-         di = dictGetIterator(dv[j]);
++    for (j = 0; j < setnum; j++) {
++        if (op == REDIS_OP_DIFF && j == 0 && !sets[j]) break; /* result set is empty */
++        if (!sets[j]) continue; /* non existing keys are like empty sets */
 +
-         while((de = dictNext(di)) != NULL) {
-             robj *ele;
- 
-             /* dictAdd will not add the same element multiple times */
-             ele = dictGetEntryKey(de);
++        si = setTypeInitIterator(sets[j]);
++        while((ele = setTypeNext(si)) != NULL) {
 +            if (op == REDIS_OP_UNION || j == 0) {
-                 if (dictAdd(dstset->ptr,ele,NULL) == DICT_OK) {
-                     incrRefCount(ele);
++                if (setTypeAdd(dstset,ele)) {
 +                    cardinality++;
 +                }
 +            } else if (op == REDIS_OP_DIFF) {
-                 if (dictDelete(dstset->ptr,ele) == DICT_OK) {
++                if (setTypeRemove(dstset,ele)) {
 +                    cardinality--;
 +                }
 +            }
++            decrRefCount(ele);
 +        }
-         dictReleaseIterator(di);
++        setTypeReleaseIterator(si);
 +
-         /* result set is empty? Exit asap. */
++        /* Exit when result set is empty. */
 +        if (op == REDIS_OP_DIFF && cardinality == 0) break;
 +    }
 +
 +    /* Output the content of the resulting set, if not in STORE mode */
 +    if (!dstkey) {
 +        addReplySds(c,sdscatprintf(sdsempty(),"*%d\r\n",cardinality));
-         di = dictGetIterator(dstset->ptr);
-         while((de = dictNext(di)) != NULL) {
-             robj *ele;
- 
-             ele = dictGetEntryKey(de);
++        si = setTypeInitIterator(dstset);
++        while((ele = setTypeNext(si)) != NULL) {
 +            addReplyBulk(c,ele);
++            decrRefCount(ele);
 +        }
-         dictReleaseIterator(di);
++        setTypeReleaseIterator(si);
 +        decrRefCount(dstset);
 +    } else {
 +        /* If we have a target key where to store the resulting set
 +         * create this key with the result set inside */
 +        dbDelete(c->db,dstkey);
-         if (dictSize((dict*)dstset->ptr) > 0) {
++        if (setTypeSize(dstset) > 0) {
 +            dbAdd(c->db,dstkey,dstset);
-             addReplyLongLong(c,dictSize((dict*)dstset->ptr));
++            addReplyLongLong(c,setTypeSize(dstset));
 +        } else {
 +            decrRefCount(dstset);
 +            addReply(c,shared.czero);
 +        }
 +        server.dirty++;
 +    }
-     zfree(dv);
++    zfree(sets);
 +}
 +
 +void sunionCommand(redisClient *c) {
 +    sunionDiffGenericCommand(c,c->argv+1,c->argc-1,NULL,REDIS_OP_UNION);
 +}
 +
 +void sunionstoreCommand(redisClient *c) {
 +    sunionDiffGenericCommand(c,c->argv+2,c->argc-2,c->argv[1],REDIS_OP_UNION);
 +}
 +
 +void sdiffCommand(redisClient *c) {
 +    sunionDiffGenericCommand(c,c->argv+1,c->argc-1,NULL,REDIS_OP_DIFF);
 +}
 +
 +void sdiffstoreCommand(redisClient *c) {
 +    sunionDiffGenericCommand(c,c->argv+2,c->argc-2,c->argv[1],REDIS_OP_DIFF);
 +}