X-Git-Url: https://git.saurik.com/redis.git/blobdiff_plain/a5eb649bfad4b66168feddb6729b75e266b04f44..9fe33a0e1245a0c74628894a3a174786bf23a323:/redis.c

diff --git a/redis.c b/redis.c
index 65571f32..43a86ca9 100644
--- a/redis.c
+++ b/redis.c
@@ -27,7 +27,7 @@
  * POSSIBILITY OF SUCH DAMAGE.
  */
 
-#define REDIS_VERSION "1.07"
+#define REDIS_VERSION "1.3.0"
 
 #include "fmacros.h"
 #include "config.h"
@@ -157,6 +157,8 @@
 #define REDIS_SLAVE 2       /* This client is a slave server */
 #define REDIS_MASTER 4      /* This client is a master server */
 #define REDIS_MONITOR 8      /* This client is a slave monitor, see MONITOR */
+#define REDIS_MULTI 16      /* This client is in a MULTI context */
+#define REDIS_BLOCKED 32    /* The client is waiting in a blocking operation */
 
 /* Slave replication state - slave side */
 #define REDIS_REPL_NONE 0   /* No active replication */
@@ -198,6 +200,10 @@
 #define APPENDFSYNC_ALWAYS 1
 #define APPENDFSYNC_EVERYSEC 2
 
+/* We can print the stacktrace, so our assert is defined this way: */
+#define redisAssert(_e) ((_e)?(void)0 : (_redisAssert(#_e),exit(1)))
+static void _redisAssert(char *estr);
+
 /*================================= Data types ============================== */
 
 /* A redis object, that is a type able to hold a string / list / set */
@@ -209,12 +215,36 @@ typedef struct redisObject {
     int refcount;
 } robj;
 
+/* 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; \
+} while(0);
+
 typedef struct redisDb {
-    dict *dict;
-    dict *expires;
+    dict *dict;                 /* The keyspace for this DB */
+    dict *expires;              /* Timeout of keys with a timeout set */
+    dict *blockingkeys;         /* Keys with clients waiting for data (BLPOP) */
     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 {
@@ -230,12 +260,18 @@ typedef struct redisClient {
     int sentlen;
     time_t lastinteraction; /* time of the last interaction, used for timeout */
     int flags;              /* REDIS_CLOSE | 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 */
+    long repldboff;         /* replication DB file offset */
     off_t repldbsize;       /* replication DB file size */
+    multiState mstate;      /* MULTI/EXEC state */
+    robj *blockingkey;      /* The key we waiting to terminate a blocking
+                             * operation such as BLPOP. Otherwise NULL. */
+    time_t blockingto;      /* Blocking operation timeout. If UNIX current time
+                             * is >= blockingto then the operation timed out. */
 } redisClient;
 
 struct saveparam {
@@ -248,7 +284,7 @@ struct redisServer {
     int port;
     int fd;
     redisDb *db;
-    dict *sharingpool;
+    dict *sharingpool;          /* Poll used for object sharing */
     unsigned int sharingpoolsize;
     long long dirty;            /* changes to DB from the last save */
     list *clients;
@@ -286,6 +322,7 @@ struct redisServer {
     char *appendfilename;
     char *requirepass;
     int shareobjects;
+    int rdbcompression;
     /* Replication related */
     int isslave;
     char *masterauth;
@@ -352,7 +389,7 @@ typedef struct zset {
 
 struct sharedObjectsStruct {
     robj *crlf, *ok, *err, *emptybulk, *czero, *cone, *pong, *space,
-    *colon, *nullbulk, *nullmultibulk,
+    *colon, *nullbulk, *nullmultibulk, *queued,
     *emptymultibulk, *wrongtypeerr, *nokeyerr, *syntaxerr, *sameobjecterr,
     *outofrangeerr, *plus,
     *select0, *select1, *select2, *select3, *select4,
@@ -403,6 +440,11 @@ static zskiplist *zslCreate(void);
 static void zslFree(zskiplist *zsl);
 static void zslInsert(zskiplist *zsl, double score, robj *obj);
 static void sendReplyToClientWritev(aeEventLoop *el, int fd, void *privdata, int mask);
+static void initClientMultiState(redisClient *c);
+static void freeClientMultiState(redisClient *c);
+static void queueMultiCommand(redisClient *c, struct redisCommand *cmd);
+static void unblockClient(redisClient *c);
+static int handleClientsWaitingListPush(redisClient *c, robj *key, robj *ele);
 
 static void authCommand(redisClient *c);
 static void pingCommand(redisClient *c);
@@ -477,6 +519,10 @@ static void zcardCommand(redisClient *c);
 static void zremCommand(redisClient *c);
 static void zscoreCommand(redisClient *c);
 static void zremrangebyscoreCommand(redisClient *c);
+static void multiCommand(redisClient *c);
+static void execCommand(redisClient *c);
+static void blpopCommand(redisClient *c);
+static void brpopCommand(redisClient *c);
 
 /*================================= Globals ================================= */
 
@@ -495,13 +541,15 @@ static struct redisCommand cmdTable[] = {
     {"lpush",lpushCommand,3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM},
     {"rpop",rpopCommand,2,REDIS_CMD_INLINE},
     {"lpop",lpopCommand,2,REDIS_CMD_INLINE},
+    {"brpop",brpopCommand,3,REDIS_CMD_INLINE},
+    {"blpop",blpopCommand,3,REDIS_CMD_INLINE},
     {"llen",llenCommand,2,REDIS_CMD_INLINE},
     {"lindex",lindexCommand,3,REDIS_CMD_INLINE},
     {"lset",lsetCommand,4,REDIS_CMD_BULK|REDIS_CMD_DENYOOM},
     {"lrange",lrangeCommand,4,REDIS_CMD_INLINE},
     {"ltrim",ltrimCommand,4,REDIS_CMD_INLINE},
     {"lrem",lremCommand,4,REDIS_CMD_BULK},
-    {"rpoplpush",rpoplpushcommand,3,REDIS_CMD_BULK},
+    {"rpoplpush",rpoplpushcommand,3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM},
     {"sadd",saddCommand,3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM},
     {"srem",sremCommand,3,REDIS_CMD_BULK},
     {"smove",smoveCommand,4,REDIS_CMD_BULK},
@@ -520,9 +568,9 @@ static struct redisCommand cmdTable[] = {
     {"zincrby",zincrbyCommand,4,REDIS_CMD_BULK|REDIS_CMD_DENYOOM},
     {"zrem",zremCommand,3,REDIS_CMD_BULK},
     {"zremrangebyscore",zremrangebyscoreCommand,4,REDIS_CMD_INLINE},
-    {"zrange",zrangeCommand,4,REDIS_CMD_INLINE},
+    {"zrange",zrangeCommand,-4,REDIS_CMD_INLINE},
     {"zrangebyscore",zrangebyscoreCommand,-4,REDIS_CMD_INLINE},
-    {"zrevrange",zrevrangeCommand,4,REDIS_CMD_INLINE},
+    {"zrevrange",zrevrangeCommand,-4,REDIS_CMD_INLINE},
     {"zcard",zcardCommand,2,REDIS_CMD_INLINE},
     {"zscore",zscoreCommand,3,REDIS_CMD_BULK|REDIS_CMD_DENYOOM},
     {"incrby",incrbyCommand,3,REDIS_CMD_INLINE|REDIS_CMD_DENYOOM},
@@ -548,6 +596,8 @@ static struct redisCommand cmdTable[] = {
     {"shutdown",shutdownCommand,1,REDIS_CMD_INLINE},
     {"lastsave",lastsaveCommand,1,REDIS_CMD_INLINE},
     {"type",typeCommand,2,REDIS_CMD_INLINE},
+    {"multi",multiCommand,1,REDIS_CMD_INLINE},
+    {"exec",execCommand,1,REDIS_CMD_INLINE},
     {"sync",syncCommand,1,REDIS_CMD_INLINE},
     {"flushdb",flushdbCommand,1,REDIS_CMD_INLINE},
     {"flushall",flushallCommand,1,REDIS_CMD_INLINE},
@@ -722,6 +772,12 @@ static void dictVanillaFree(void *privdata, void *val)
     zfree(val);
 }
 
+static void dictListDestructor(void *privdata, void *val)
+{
+    DICT_NOTUSED(privdata);
+    listRelease((list*)val);
+}
+
 static int sdsDictKeyCompare(void *privdata, const void *key1,
         const void *key2)
 {
@@ -791,7 +847,7 @@ static dictType zsetDictType = {
     NULL,                      /* val dup */
     dictEncObjKeyCompare,      /* key compare */
     dictRedisObjectDestructor, /* key destructor */
-    dictVanillaFree            /* val destructor */
+    dictVanillaFree            /* val destructor of malloc(sizeof(double)) */
 };
 
 static dictType hashDictType = {
@@ -803,6 +859,17 @@ static dictType hashDictType = {
     dictRedisObjectDestructor   /* val destructor */
 };
 
+/* Keylist hash table type has unencoded redis objects as keys and
+ * lists as values. It's used for blocking operations (BLPOP) */
+static dictType keylistDictType = {
+    dictObjHash,                /* hash function */
+    NULL,                       /* key dup */
+    NULL,                       /* val dup */
+    dictObjKeyCompare,          /* key compare */
+    dictRedisObjectDestructor,  /* key destructor */
+    dictListDestructor          /* val destructor */
+};
+
 /* ========================= Random utility functions ======================= */
 
 /* Redis generally does not try to recover from out of memory conditions
@@ -811,8 +878,7 @@ static dictType hashDictType = {
  * is based on heap allocation for send buffers, so we simply abort.
  * At least the code will be simpler to read... */
 static void oom(const char *msg) {
-    fprintf(stderr, "%s: Out of memory\n",msg);
-    fflush(stderr);
+    redisLog(REDIS_WARNING, "%s: Out of memory\n",msg);
     sleep(1);
     abort();
 }
@@ -909,7 +975,7 @@ void backgroundRewriteDoneHandler(int statloc) {
             close(fd);
             goto cleanup;
         }
-        redisLog(REDIS_WARNING,"Parent diff flushed into the new append log file with success");
+        redisLog(REDIS_NOTICE,"Parent diff flushed into the new append log file with success (%lu bytes)",sdslen(server.bgrewritebuf));
         /* Now our work is to rename the temp file into the stable file. And
          * switch the file descriptor used by the server for append only. */
         if (rename(tmpfile,server.appendfilename) == -1) {
@@ -1067,8 +1133,8 @@ static void createSharedObjects(void) {
     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"));
-    /* no such key */
     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(
@@ -1127,6 +1193,7 @@ static void initServerConfig() {
     server.appendfilename = "appendonly.aof";
     server.requirepass = NULL;
     server.shareobjects = 0;
+    server.rdbcompression = 1;
     server.sharingpoolsize = 1024;
     server.maxclients = 0;
     server.maxmemory = 0;
@@ -1173,6 +1240,7 @@ static void initServer() {
     for (j = 0; j < server.dbnum; j++) {
         server.db[j].dict = dictCreate(&hashDictType,NULL);
         server.db[j].expires = dictCreate(&setDictType,NULL);
+        server.db[j].blockingkeys = dictCreate(&keylistDictType,NULL);
         server.db[j].id = j;
     }
     server.cronloops = 0;
@@ -1327,6 +1395,10 @@ static void loadServerConfig(char *filename) {
             if ((server.shareobjects = 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],"shareobjectspoolsize") && argc == 2) {
             server.sharingpoolsize = atoi(argv[1]);
             if (server.sharingpoolsize < 1) {
@@ -1390,21 +1462,30 @@ static void freeClientArgv(redisClient *c) {
 static void freeClient(redisClient *c) {
     listNode *ln;
 
+    /* Note that if the client we are freeing is blocked into a blocking
+     * call, we have to set querybuf to NULL *before* to call unblockClient()
+     * to avoid processInputBuffer() will get called. Also it is important
+     * to remove the file events after this, because this call adds
+     * the READABLE event. */
+    sdsfree(c->querybuf);
+    c->querybuf = NULL;
+    if (c->flags & REDIS_BLOCKED)
+        unblockClient(c);
+
     aeDeleteFileEvent(server.el,c->fd,AE_READABLE);
     aeDeleteFileEvent(server.el,c->fd,AE_WRITABLE);
-    sdsfree(c->querybuf);
     listRelease(c->reply);
     freeClientArgv(c);
     close(c->fd);
     ln = listSearchKey(server.clients,c);
-    assert(ln != NULL);
+    redisAssert(ln != NULL);
     listDelNode(server.clients,ln);
     if (c->flags & REDIS_SLAVE) {
         if (c->replstate == REDIS_REPL_SEND_BULK && c->repldbfd != -1)
             close(c->repldbfd);
         list *l = (c->flags & REDIS_MONITOR) ? server.monitors : server.slaves;
         ln = listSearchKey(l,c);
-        assert(ln != NULL);
+        redisAssert(ln != NULL);
         listDelNode(l,ln);
     }
     if (c->flags & REDIS_MASTER) {
@@ -1413,6 +1494,7 @@ static void freeClient(redisClient *c) {
     }
     zfree(c->argv);
     zfree(c->mbargv);
+    freeClientMultiState(c);
     zfree(c);
 }
 
@@ -1603,6 +1685,21 @@ static void resetClient(redisClient *c) {
     c->multibulk = 0;
 }
 
+/* Call() is the core of Redis execution of a command */
+static void call(redisClient *c, struct redisCommand *cmd) {
+    long long dirty;
+
+    dirty = server.dirty;
+    cmd->proc(c);
+    if (server.appendonly && server.dirty-dirty)
+        feedAppendOnlyFile(cmd,c->db->id,c->argv,c->argc);
+    if (server.dirty-dirty && listLength(server.slaves))
+        replicationFeedSlaves(server.slaves,cmd,c->db->id,c->argv,c->argc);
+    if (listLength(server.monitors))
+        replicationFeedSlaves(server.monitors,cmd,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
@@ -1613,7 +1710,6 @@ static void resetClient(redisClient *c) {
  * if 0 is returned the client was destroied (i.e. after QUIT). */
 static int processCommand(redisClient *c) {
     struct redisCommand *cmd;
-    long long dirty;
 
     /* Free some memory if needed (maxmemory setting) */
     if (server.maxmemory) freeMemoryIfNeeded();
@@ -1694,12 +1790,17 @@ static int processCommand(redisClient *c) {
     }
     cmd = lookupCommand(c->argv[0]->ptr);
     if (!cmd) {
-        addReplySds(c,sdsnew("-ERR unknown command\r\n"));
+        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,sdsnew("-ERR wrong number of arguments\r\n"));
+        addReplySds(c,
+            sdscatprintf(sdsempty(),
+                "-ERR wrong number of arguments for '%s' command\r\n",
+                cmd->name));
         resetClient(c);
         return 1;
     } else if (server.maxmemory && cmd->flags & REDIS_CMD_DENYOOM && zmalloc_used_memory() > server.maxmemory) {
@@ -1749,15 +1850,12 @@ static int processCommand(redisClient *c) {
     }
 
     /* Exec the command */
-    dirty = server.dirty;
-    cmd->proc(c);
-    if (server.appendonly && server.dirty-dirty)
-        feedAppendOnlyFile(cmd,c->db->id,c->argv,c->argc);
-    if (server.dirty-dirty && listLength(server.slaves))
-        replicationFeedSlaves(server.slaves,cmd,c->db->id,c->argv,c->argc);
-    if (listLength(server.monitors))
-        replicationFeedSlaves(server.monitors,cmd,c->db->id,c->argv,c->argc);
-    server.stat_numcommands++;
+    if (c->flags & REDIS_MULTI && cmd->proc != execCommand) {
+        queueMultiCommand(c,cmd);
+        addReply(c,shared.queued);
+    } else {
+        call(c,cmd);
+    }
 
     /* Prepare the client for the next command */
     if (c->flags & REDIS_CLOSE) {
@@ -1787,8 +1885,8 @@ static void replicationFeedSlaves(list *slaves, struct redisCommand *cmd, int di
             robj *lenobj;
 
             lenobj = createObject(REDIS_STRING,
-                sdscatprintf(sdsempty(),"%d\r\n",
-                    stringObjectLen(argv[j])));
+                sdscatprintf(sdsempty(),"%lu\r\n",
+                    (unsigned long) stringObjectLen(argv[j])));
             lenobj->refcount = 0;
             outv[outc++] = lenobj;
         }
@@ -1839,6 +1937,13 @@ static void replicationFeedSlaves(list *slaves, struct redisCommand *cmd, int di
 
 static void processInputBuffer(redisClient *c) {
 again:
+    /* Before to process the input buffer, make sure the client is not
+     * waitig for a blocking operation such as BLPOP. Note that the first
+     * iteration the client is never blocked, otherwise the processInputBuffer
+     * would not be called at all, but after the execution of the first commands
+     * in the input buffer the client may be blocked, and the "goto again"
+     * will try to reiterate. The following line will make it return asap. */
+    if (c->flags & REDIS_BLOCKED) return;
     if (c->bulklen == -1) {
         /* Read the first line of the query */
         char *p = strchr(c->querybuf,'\n');
@@ -1860,11 +1965,6 @@ again:
             sdsupdatelen(query);
 
             /* Now we can split the query in arguments */
-            if (sdslen(query) == 0) {
-                /* Ignore empty query */
-                sdsfree(query);
-                return;
-            }
             argv = sdssplitlen(query,sdslen(query)," ",1,&argc);
             sdsfree(query);
 
@@ -1880,10 +1980,16 @@ again:
                 }
             }
             zfree(argv);
-            /* Execute the command. If the client is still valid
-             * after processCommand() return and there is something
-             * on the query buffer try to process the next command. */
-            if (c->argc && processCommand(c) && sdslen(c->querybuf)) goto again;
+            if (c->argc) {
+                /* Execute the command. If the client is still valid
+                 * after processCommand() return and there is something
+                 * on the query buffer try to process the next command. */
+                if (processCommand(c) && sdslen(c->querybuf)) goto again;
+            } else {
+                /* Nothing to process, argc == 0. Just process the query
+                 * buffer if it's not empty or return to the caller */
+                if (sdslen(c->querybuf)) goto again;
+            }
             return;
         } else if (sdslen(c->querybuf) >= REDIS_REQUEST_MAX_SIZE) {
             redisLog(REDIS_DEBUG, "Client protocol error");
@@ -1974,6 +2080,7 @@ static redisClient *createClient(int fd) {
     c->authenticated = 0;
     c->replstate = REDIS_REPL_NONE;
     c->reply = listCreate();
+    c->blockingkey = NULL;
     listSetFreeMethod(c->reply,decrRefCount);
     listSetDupMethod(c->reply,dupClientReplyValue);
     if (aeCreateFileEvent(server.el, c->fd, AE_READABLE,
@@ -1982,6 +2089,7 @@ static redisClient *createClient(int fd) {
         return NULL;
     }
     listAddNodeTail(server.clients,c);
+    initClientMultiState(c);
     return c;
 }
 
@@ -2004,8 +2112,8 @@ static void addReplyDouble(redisClient *c, double d) {
     char buf[128];
 
     snprintf(buf,sizeof(buf),"%.17g",d);
-    addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n%s\r\n",
-        strlen(buf),buf));
+    addReplySds(c,sdscatprintf(sdsempty(),"$%lu\r\n%s\r\n",
+        (unsigned long) strlen(buf),buf));
 }
 
 static void addReplyBulkLen(redisClient *c, robj *obj) {
@@ -2016,6 +2124,7 @@ static void addReplyBulkLen(redisClient *c, robj *obj) {
     } else {
         long n = (long)obj->ptr;
 
+        /* Compute how many bytes will take this integer as a radix 10 string */
         len = 1;
         if (n < 0) {
             len++;
@@ -2025,7 +2134,7 @@ static void addReplyBulkLen(redisClient *c, robj *obj) {
             len++;
         }
     }
-    addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",len));
+    addReplySds(c,sdscatprintf(sdsempty(),"$%lu\r\n",(unsigned long)len));
 }
 
 static void acceptHandler(aeEventLoop *el, int fd, void *privdata, int mask) {
@@ -2155,7 +2264,7 @@ static void decrRefCount(void *obj) {
         case REDIS_SET: freeSetObject(o); break;
         case REDIS_ZSET: freeZsetObject(o); break;
         case REDIS_HASH: freeHashObject(o); break;
-        default: assert(0 != 0); break;
+        default: redisAssert(0 != 0); break;
         }
         if (listLength(server.objfreelist) > REDIS_OBJFREELIST_MAX ||
             !listAddNodeHead(server.objfreelist,o))
@@ -2200,7 +2309,7 @@ static robj *tryObjectSharing(robj *o) {
 
     if (o == NULL || server.shareobjects == 0) return o;
 
-    assert(o->type == REDIS_STRING);
+    redisAssert(o->type == REDIS_STRING);
     de = dictFind(server.sharingpool,o);
     if (de) {
         robj *shared = dictGetEntryKey(de);
@@ -2218,7 +2327,7 @@ static robj *tryObjectSharing(robj *o) {
         if (dictSize(server.sharingpool) >=
                 server.sharingpoolsize) {
             de = dictGetRandomKey(server.sharingpool);
-            assert(de != NULL);
+            redisAssert(de != NULL);
             c = ((unsigned long) dictGetEntryVal(de))-1;
             dictGetEntryVal(de) = (void*) c;
             if (c == 0) {
@@ -2231,7 +2340,7 @@ static robj *tryObjectSharing(robj *o) {
             int retval;
 
             retval = dictAdd(server.sharingpool,o,(void*)1);
-            assert(retval == DICT_OK);
+            redisAssert(retval == DICT_OK);
             incrRefCount(o);
         }
         return o;
@@ -2274,7 +2383,7 @@ static int tryObjectEncoding(robj *o) {
      if (o->refcount > 1) return REDIS_ERR;
 
     /* Currently we try to encode only strings */
-    assert(o->type == REDIS_STRING);
+    redisAssert(o->type == REDIS_STRING);
 
     /* Check if we can represent this string as a long integer */
     if (isStringRepresentableAsLong(s,&value) == REDIS_ERR) return REDIS_ERR;
@@ -2302,7 +2411,7 @@ static robj *getDecodedObject(robj *o) {
         dec = createStringObject(buf,strlen(buf));
         return dec;
     } else {
-        assert(1 != 1);
+        redisAssert(1 != 1);
     }
 }
 
@@ -2315,7 +2424,7 @@ static robj *getDecodedObject(robj *o) {
  * sdscmp() from sds.c will apply memcmp() so this function ca be considered
  * binary safe. */
 static int compareStringObjects(robj *a, robj *b) {
-    assert(a->type == REDIS_STRING && b->type == REDIS_STRING);
+    redisAssert(a->type == REDIS_STRING && b->type == REDIS_STRING);
     char bufa[128], bufb[128], *astr, *bstr;
     int bothsds = 1;
 
@@ -2338,7 +2447,7 @@ static int compareStringObjects(robj *a, robj *b) {
 }
 
 static size_t stringObjectLen(robj *o) {
-    assert(o->type == REDIS_STRING);
+    redisAssert(o->type == REDIS_STRING);
     if (o->encoding == REDIS_ENCODING_RAW) {
         return sdslen(o->ptr);
     } else {
@@ -2469,7 +2578,7 @@ static int rdbSaveStringObjectRaw(FILE *fp, robj *obj) {
 
     /* Try LZF compression - under 20 bytes it's unable to compress even
      * aaaaaaaaaaaaaaaaaa so skip it */
-    if (len > 20) {
+    if (server.rdbcompression && len > 20) {
         int retval;
 
         retval = rdbSaveLzfStringObject(fp,obj);
@@ -2611,7 +2720,7 @@ static int rdbSave(char *filename) {
                 }
                 dictReleaseIterator(di);
             } else {
-                assert(0 != 0);
+                redisAssert(0 != 0);
             }
         }
         dictReleaseIterator(di);
@@ -2745,7 +2854,7 @@ static robj *rdbLoadIntegerObject(FILE *fp, int enctype) {
         val = (int32_t)v;
     } else {
         val = 0; /* anti-warning */
-        assert(0!=0);
+        redisAssert(0!=0);
     }
     return createObject(REDIS_STRING,sdscatprintf(sdsempty(),"%lld",val));
 }
@@ -2784,7 +2893,7 @@ static robj *rdbLoadStringObject(FILE*fp, int rdbver) {
         case REDIS_RDB_ENC_LZF:
             return tryObjectSharing(rdbLoadLzfStringObject(fp,rdbver));
         default:
-            assert(0!=0);
+            redisAssert(0!=0);
         }
     }
 
@@ -2809,6 +2918,7 @@ static int rdbLoadDoubleValue(FILE *fp, double *val) {
     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;
     }
@@ -2910,7 +3020,7 @@ static int rdbLoad(char *filename) {
                 incrRefCount(ele); /* added to skiplist */
             }
         } else {
-            assert(0 != 0);
+            redisAssert(0 != 0);
         }
         /* Add the new object in the hash table */
         retval = dictAdd(d,keyobj,o);
@@ -2964,6 +3074,7 @@ static void echoCommand(redisClient *c) {
 static void setGenericCommand(redisClient *c, int nx) {
     int retval;
 
+    if (nx) deleteIfVolatile(c->db,c->argv[1]);
     retval = dictAdd(c->db->dict,c->argv[1],c->argv[2]);
     if (retval == DICT_ERR) {
         if (!nx) {
@@ -2990,24 +3101,31 @@ static void setnxCommand(redisClient *c) {
     setGenericCommand(c,1);
 }
 
-static void getCommand(redisClient *c) {
+static int getGenericCommand(redisClient *c) {
     robj *o = lookupKeyRead(c->db,c->argv[1]);
 
     if (o == NULL) {
         addReply(c,shared.nullbulk);
+        return REDIS_OK;
     } else {
         if (o->type != REDIS_STRING) {
             addReply(c,shared.wrongtypeerr);
+            return REDIS_ERR;
         } else {
             addReplyBulkLen(c,o);
             addReply(c,o);
             addReply(c,shared.crlf);
+            return REDIS_OK;
         }
     }
 }
 
+static void getCommand(redisClient *c) {
+    getGenericCommand(c);
+}
+
 static void getsetCommand(redisClient *c) {
-    getCommand(c);
+    if (getGenericCommand(c) == REDIS_ERR) return;
     if (dictAdd(c->db->dict,c->argv[1],c->argv[2]) == DICT_ERR) {
         dictReplace(c->db->dict,c->argv[1],c->argv[2]);
     } else {
@@ -3039,22 +3157,25 @@ static void mgetCommand(redisClient *c) {
 }
 
 static void msetGenericCommand(redisClient *c, int nx) {
-    int j;
+    int j, busykeys = 0;
 
     if ((c->argc % 2) == 0) {
-        addReplySds(c,sdsnew("-ERR wrong number of arguments\r\n"));
+        addReplySds(c,sdsnew("-ERR wrong number of arguments for MSET\r\n"));
         return;
     }
     /* Handle the NX flag. The MSETNX semantic is to return zero and don't
      * set nothing at all if at least one already key exists. */
     if (nx) {
         for (j = 1; j < c->argc; j += 2) {
-            if (dictFind(c->db->dict,c->argv[j]) != NULL) {
-                addReply(c, shared.czero);
-                return;
+            if (lookupKeyWrite(c->db,c->argv[j]) != NULL) {
+                busykeys++;
             }
         }
     }
+    if (busykeys) {
+        addReply(c, shared.czero);
+        return;
+    }
 
     for (j = 1; j < c->argc; j += 2) {
         int retval;
@@ -3101,7 +3222,7 @@ static void incrDecrCommand(redisClient *c, long long incr) {
             else if (o->encoding == REDIS_ENCODING_INT)
                 value = (long)o->ptr;
             else
-                assert(1 != 1);
+                redisAssert(1 != 1);
         }
     }
 
@@ -3199,7 +3320,7 @@ static void keysCommand(redisClient *c) {
     dictEntry *de;
     sds pattern = c->argv[1]->ptr;
     int plen = sdslen(pattern);
-    int numkeys = 0, keyslen = 0;
+    unsigned long numkeys = 0, keyslen = 0;
     robj *lenobj = createObject(REDIS_STRING,NULL);
 
     di = dictGetIterator(c->db->dict);
@@ -3273,7 +3394,8 @@ static void bgsaveCommand(redisClient *c) {
         return;
     }
     if (rdbSaveBackground(server.dbfilename) == REDIS_OK) {
-        addReply(c,shared.ok);
+        char *status = "+Background saving started\r\n";
+        addReplySds(c,sdsnew(status));
     } else {
         addReply(c,shared.err);
     }
@@ -3289,20 +3411,26 @@ static void shutdownCommand(redisClient *c) {
         kill(server.bgsavechildpid,SIGKILL);
         rdbRemoveTempFile(server.bgsavechildpid);
     }
-    /* SYNC SAVE */
-    if (rdbSave(server.dbfilename) == REDIS_OK) {
-        if (server.daemonize)
-            unlink(server.pidfile);
-        redisLog(REDIS_WARNING,"%zu bytes used at exit",zmalloc_used_memory());
-        redisLog(REDIS_WARNING,"Server exit now, bye bye...");
-        exit(1);
+    if (server.appendonly) {
+        /* Append only file: fsync() the AOF and exit */
+        fsync(server.appendfd);
+        exit(0);
     } 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"); 
-        addReplySds(c,sdsnew("-ERR can't quit, problems saving the DB\r\n"));
+        /* 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());
+            redisLog(REDIS_WARNING,"Server exit now, bye bye...");
+            exit(0);
+        } 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"); 
+            addReplySds(c,sdsnew("-ERR can't quit, problems saving the DB\r\n"));
+        }
     }
 }
 
@@ -3396,6 +3524,7 @@ static void pushGenericCommand(redisClient *c, int where) {
 
     lobj = lookupKeyWrite(c->db,c->argv[1]);
     if (lobj == NULL) {
+        if (handleClientsWaitingListPush(c,c->argv[1],c->argv[2])) return;
         lobj = createListObject();
         list = lobj->ptr;
         if (where == REDIS_HEAD) {
@@ -3411,6 +3540,7 @@ static void pushGenericCommand(redisClient *c, int where) {
             addReply(c,shared.wrongtypeerr);
             return;
         }
+        if (handleClientsWaitingListPush(c,c->argv[1],c->argv[2])) return;
         list = lobj->ptr;
         if (where == REDIS_HEAD) {
             listAddNodeHead(list,c->argv[2]);
@@ -3600,7 +3730,7 @@ static void ltrimCommand(redisClient *c) {
     
     o = lookupKeyWrite(c->db,c->argv[1]);
     if (o == NULL) {
-        addReply(c,shared.nokeyerr);
+        addReply(c,shared.ok);
     } else {
         if (o->type != REDIS_LIST) {
             addReply(c,shared.wrongtypeerr);
@@ -3856,7 +3986,7 @@ static void scardCommand(redisClient *c) {
             addReply(c,shared.wrongtypeerr);
         } else {
             s = o->ptr;
-            addReplySds(c,sdscatprintf(sdsempty(),":%d\r\n",
+            addReplySds(c,sdscatprintf(sdsempty(),":%lu\r\n",
                 dictSize(s)));
         }
     }
@@ -3921,12 +4051,12 @@ static int qsortCompareSetsByCardinality(const void *s1, const void *s2) {
     return dictSize(*d1)-dictSize(*d2);
 }
 
-static void sinterGenericCommand(redisClient *c, robj **setskeys, int setsnum, robj *dstkey) {
+static 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;
-    int j, cardinality = 0;
+    unsigned long j, cardinality = 0;
 
     for (j = 0; j < setsnum; j++) {
         robj *setobj;
@@ -3937,8 +4067,9 @@ static void sinterGenericCommand(redisClient *c, robj **setskeys, int setsnum, r
         if (!setobj) {
             zfree(dv);
             if (dstkey) {
-                deleteKey(c->db,dstkey);
-                addReply(c,shared.ok);
+                if (deleteKey(c->db,dstkey))
+                    server.dirty++;
+                addReply(c,shared.czero);
             } else {
                 addReply(c,shared.nullmultibulk);
             }
@@ -4003,9 +4134,9 @@ static void sinterGenericCommand(redisClient *c, robj **setskeys, int setsnum, r
     }
 
     if (!dstkey) {
-        lenobj->ptr = sdscatprintf(sdsempty(),"*%d\r\n",cardinality);
+        lenobj->ptr = sdscatprintf(sdsempty(),"*%lu\r\n",cardinality);
     } else {
-        addReplySds(c,sdscatprintf(sdsempty(),":%d\r\n",
+        addReplySds(c,sdscatprintf(sdsempty(),":%lu\r\n",
             dictSize((dict*)dstset->ptr)));
         server.dirty++;
     }
@@ -4107,7 +4238,7 @@ static void sunionDiffGenericCommand(redisClient *c, robj **setskeys, int setsnu
     if (!dstkey) {
         decrRefCount(dstset);
     } else {
-        addReplySds(c,sdscatprintf(sdsempty(),":%d\r\n",
+        addReplySds(c,sdscatprintf(sdsempty(),":%lu\r\n",
             dictSize((dict*)dstset->ptr)));
         server.dirty++;
     }
@@ -4393,14 +4524,14 @@ static void zaddGenericCommand(redisClient *c, robj *key, robj *ele, double scor
         
         /* case 2: Score update operation */
         de = dictFind(zs->dict,ele);
-        assert(de != NULL);
+        redisAssert(de != NULL);
         oldscore = dictGetEntryVal(de);
         if (*score != *oldscore) {
             int deleted;
 
             /* Remove and insert the element in the skip list with new score */
             deleted = zslDelete(zs->zsl,*oldscore,ele);
-            assert(deleted != 0);
+            redisAssert(deleted != 0);
             zslInsert(zs->zsl,*score,ele);
             incrRefCount(ele);
             /* Update the score in the hash table */
@@ -4455,7 +4586,7 @@ static void zremCommand(redisClient *c) {
         /* Delete from the skiplist */
         oldscore = dictGetEntryVal(de);
         deleted = zslDelete(zs->zsl,*oldscore,c->argv[2]);
-        assert(deleted != 0);
+        redisAssert(deleted != 0);
 
         /* Delete from the hash table */
         dictDelete(zs->dict,c->argv[2]);
@@ -4493,6 +4624,14 @@ static void zrangeGenericCommand(redisClient *c, int reverse) {
     robj *o;
     int start = atoi(c->argv[2]->ptr);
     int end = atoi(c->argv[3]->ptr);
+    int withscores = 0;
+
+    if (c->argc == 5 && !strcasecmp(c->argv[4]->ptr,"withscores")) {
+        withscores = 1;
+    } else if (c->argc >= 5) {
+        addReply(c,shared.syntaxerr);
+        return;
+    }
 
     o = lookupKeyRead(c->db,c->argv[1]);
     if (o == NULL) {
@@ -4535,12 +4674,15 @@ static void zrangeGenericCommand(redisClient *c, int reverse) {
                     ln = ln->forward[0];
             }
 
-            addReplySds(c,sdscatprintf(sdsempty(),"*%d\r\n",rangelen));
+            addReplySds(c,sdscatprintf(sdsempty(),"*%d\r\n",
+                withscores ? (rangelen*2) : rangelen));
             for (j = 0; j < rangelen; j++) {
                 ele = ln->obj;
                 addReplyBulkLen(c,ele);
                 addReply(c,ele);
                 addReply(c,shared.crlf);
+                if (withscores)
+                    addReplyDouble(c,ln->score);
                 ln = reverse ? ln->backward : ln->forward[0];
             }
         }
@@ -4562,7 +4704,8 @@ static void zrangebyscoreCommand(redisClient *c) {
     int offset = 0, limit = -1;
 
     if (c->argc != 4 && c->argc != 7) {
-        addReplySds(c,sdsnew("-ERR wrong number of arguments\r\n"));
+        addReplySds(c,
+            sdsnew("-ERR wrong number of arguments for ZRANGEBYSCORE\r\n"));
         return;
     } else if (c->argc == 7 && strcasecmp(c->argv[4]->ptr,"limit")) {
         addReply(c,shared.syntaxerr);
@@ -4570,6 +4713,7 @@ static void zrangebyscoreCommand(redisClient *c) {
     } else if (c->argc == 7) {
         offset = atoi(c->argv[5]->ptr);
         limit = atoi(c->argv[6]->ptr);
+        if (offset < 0) offset = 0;
     }
 
     o = lookupKeyRead(c->db,c->argv[1]);
@@ -4634,7 +4778,7 @@ static void zcardCommand(redisClient *c) {
             addReply(c,shared.wrongtypeerr);
         } else {
             zs = o->ptr;
-            addReplySds(c,sdscatprintf(sdsempty(),":%d\r\n",zs->zsl->length));
+            addReplySds(c,sdscatprintf(sdsempty(),":%lu\r\n",zs->zsl->length));
         }
     }
 }
@@ -4732,10 +4876,7 @@ static robj *lookupKeyByPattern(redisDb *db, robj *pattern, robj *subst) {
     keyname.buf[prefixlen+sublen+postfixlen] = '\0';
     keyname.len = prefixlen+sublen+postfixlen;
 
-    keyobj.refcount = 1;
-    keyobj.type = REDIS_STRING;
-    keyobj.ptr = ((char*)&keyname)+(sizeof(long)*2);
-
+    initStaticStringObject(keyobj,((char*)&keyname)+(sizeof(long)*2))
     decrRefCount(subst);
 
     /* printf("lookup '%s' => %p\n", keyname.buf,de); */
@@ -4802,7 +4943,7 @@ static void sortCommand(redisClient *c) {
     /* Lookup the key to sort. It must be of the right types */
     sortval = lookupKeyRead(c->db,c->argv[1]);
     if (sortval == NULL) {
-        addReply(c,shared.nokeyerr);
+        addReply(c,shared.nullmultibulk);
         return;
     }
     if (sortval->type != REDIS_SET && sortval->type != REDIS_LIST &&
@@ -4864,7 +5005,7 @@ static void sortCommand(redisClient *c) {
     case REDIS_LIST: vectorlen = listLength((list*)sortval->ptr); break;
     case REDIS_SET: vectorlen =  dictSize((dict*)sortval->ptr); break;
     case REDIS_ZSET: vectorlen = dictSize(((zset*)sortval->ptr)->dict); break;
-    default: vectorlen = 0; assert(0); /* Avoid GCC warning */
+    default: vectorlen = 0; redisAssert(0); /* Avoid GCC warning */
     }
     vector = zmalloc(sizeof(redisSortObject)*vectorlen);
     j = 0;
@@ -4902,7 +5043,7 @@ static void sortCommand(redisClient *c) {
         }
         dictReleaseIterator(di);
     }
-    assert(j == vectorlen);
+    redisAssert(j == vectorlen);
 
     /* Now it's time to load the right scores in the sorting vector */
     if (dontsort == 0) {
@@ -4924,7 +5065,7 @@ static void sortCommand(redisClient *c) {
                         if (byval->encoding == REDIS_ENCODING_INT) {
                             vector[j].u.score = (long)byval->ptr;
                         } else
-                            assert(1 != 1);
+                            redisAssert(1 != 1);
                     }
                 }
             } else {
@@ -4935,7 +5076,7 @@ static void sortCommand(redisClient *c) {
                         if (vector[j].obj->encoding == REDIS_ENCODING_INT)
                             vector[j].u.score = (long) vector[j].obj->ptr;
                         else
-                            assert(1 != 1);
+                            redisAssert(1 != 1);
                     }
                 }
             }
@@ -4990,7 +5131,7 @@ static void sortCommand(redisClient *c) {
                         addReply(c,shared.crlf);
                     }
                 } else {
-                    assert(sop->type == REDIS_SORT_GET); /* always fails */
+                    redisAssert(sop->type == REDIS_SORT_GET); /* always fails */
                 }
             }
         }
@@ -5019,7 +5160,7 @@ static void sortCommand(redisClient *c) {
                         incrRefCount(val);
                     }
                 } else {
-                    assert(sop->type == REDIS_SORT_GET); /* always fails */
+                    redisAssert(sop->type == REDIS_SORT_GET); /* always fails */
                 }
             }
         }
@@ -5055,14 +5196,15 @@ static sds genRedisInfoString(void) {
         "redis_version:%s\r\n"
         "arch_bits:%s\r\n"
         "multiplexing_api:%s\r\n"
-        "uptime_in_seconds:%d\r\n"
-        "uptime_in_days:%d\r\n"
+        "uptime_in_seconds:%ld\r\n"
+        "uptime_in_days:%ld\r\n"
         "connected_clients:%d\r\n"
         "connected_slaves:%d\r\n"
         "used_memory:%zu\r\n"
         "changes_since_last_save:%lld\r\n"
         "bgsave_in_progress:%d\r\n"
-        "last_save_time:%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"
         "role:%s\r\n"
@@ -5077,6 +5219,7 @@ static sds genRedisInfoString(void) {
         server.dirty,
         server.bgsavechildpid != -1,
         server.lastsave,
+        server.bgrewritechildpid != -1,
         server.stat_numconnections,
         server.stat_numcommands,
         server.masterhost == NULL ? "master" : "slave"
@@ -5109,7 +5252,8 @@ static sds genRedisInfoString(void) {
 
 static void infoCommand(redisClient *c) {
     sds info = genRedisInfoString();
-    addReplySds(c,sdscatprintf(sdsempty(),"$%d\r\n",sdslen(info)));
+    addReplySds(c,sdscatprintf(sdsempty(),"$%lu\r\n",
+        (unsigned long)sdslen(info)));
     addReplySds(c,info);
     addReply(c,shared.crlf);
 }
@@ -5228,6 +5372,233 @@ static void ttlCommand(redisClient *c) {
     addReplySds(c,sdscatprintf(sdsempty(),":%d\r\n",ttl));
 }
 
+/* ================================ MULTI/EXEC ============================== */
+
+/* Client state initialization for MULTI/EXEC */
+static void initClientMultiState(redisClient *c) {
+    c->mstate.commands = NULL;
+    c->mstate.count = 0;
+}
+
+/* Release all the resources associated with MULTI/EXEC state */
+static void freeClientMultiState(redisClient *c) {
+    int j;
+
+    for (j = 0; j < c->mstate.count; j++) {
+        int i;
+        multiCmd *mc = c->mstate.commands+j;
+
+        for (i = 0; i < mc->argc; i++)
+            decrRefCount(mc->argv[i]);
+        zfree(mc->argv);
+    }
+    zfree(c->mstate.commands);
+}
+
+/* Add a new command into the MULTI commands queue */
+static void queueMultiCommand(redisClient *c, struct redisCommand *cmd) {
+    multiCmd *mc;
+    int j;
+
+    c->mstate.commands = zrealloc(c->mstate.commands,
+            sizeof(multiCmd)*(c->mstate.count+1));
+    mc = c->mstate.commands+c->mstate.count;
+    mc->cmd = cmd;
+    mc->argc = c->argc;
+    mc->argv = zmalloc(sizeof(robj*)*c->argc);
+    memcpy(mc->argv,c->argv,sizeof(robj*)*c->argc);
+    for (j = 0; j < c->argc; j++)
+        incrRefCount(mc->argv[j]);
+    c->mstate.count++;
+}
+
+static void multiCommand(redisClient *c) {
+    c->flags |= REDIS_MULTI;
+    addReply(c,shared.ok);
+}
+
+static void execCommand(redisClient *c) {
+    int j;
+    robj **orig_argv;
+    int orig_argc;
+
+    if (!(c->flags & REDIS_MULTI)) {
+        addReplySds(c,sdsnew("-ERR EXEC without MULTI\r\n"));
+        return;
+    }
+
+    orig_argv = c->argv;
+    orig_argc = c->argc;
+    addReplySds(c,sdscatprintf(sdsempty(),"*%d\r\n",c->mstate.count));
+    for (j = 0; j < c->mstate.count; j++) {
+        c->argc = c->mstate.commands[j].argc;
+        c->argv = c->mstate.commands[j].argv;
+        call(c,c->mstate.commands[j].cmd);
+    }
+    c->argv = orig_argv;
+    c->argc = orig_argc;
+    freeClientMultiState(c);
+    initClientMultiState(c);
+    c->flags &= (~REDIS_MULTI);
+}
+
+/* =========================== Blocking Operations  ========================= */
+
+/* Currently Redis blocking operations support is limited to list POP ops,
+ * so the current implementation is not fully generic, but it is also not
+ * completely specific so it will not require a rewrite to support new
+ * kind of blocking operations in the future.
+ *
+ * Still it's important to note that list blocking operations can be already
+ * used as a notification mechanism in order to implement other blocking
+ * operations at application level, so there must be a very strong evidence
+ * of usefulness and generality before new blocking operations are implemented.
+ *
+ * This is how the current blocking POP works, we use BLPOP as example:
+ * - If the user calls BLPOP and the key exists and contains a non empty list
+ *   then LPOP is called instead. So BLPOP is semantically the same as LPOP
+ *   if there is not to block.
+ * - If instead BLPOP is called and the key does not exists or the list is
+ *   empty we need to block. In order to do so we remove the notification for
+ *   new data to read in the client socket (so that we'll not serve new
+ *   requests if the blocking request is not served). Also we put the client
+ *   in a dictionary (server.blockingkeys) mapping keys to a list of clients
+ *   blocking for this keys.
+ * - If a PUSH operation against a key with blocked clients waiting is
+ *   performed, we serve the first in the list: basically instead to push
+ *   the new element inside the list we return it to the (first / oldest)
+ *   blocking client, unblock the client, and remove it form the list.
+ *
+ * The above comment and the source code should be enough in order to understand
+ * the implementation and modify / fix it later.
+ */
+
+/* Set a client in blocking mode for the specified key, with the specified
+ * timeout */
+static void blockForKey(redisClient *c, robj *key, time_t timeout) {
+    dictEntry *de;
+    list *l;
+
+    printf("blockForKey %p %s\n", c, key->ptr);
+    c->blockingkey = key;
+    incrRefCount(key);
+    c->blockingto = timeout;
+    de = dictFind(c->db->blockingkeys,key);
+    if (de == NULL) {
+        int retval;
+
+        l = listCreate();
+        retval = dictAdd(c->db->blockingkeys,key,l);
+        incrRefCount(key);
+        assert(retval == DICT_OK);
+    } else {
+        l = dictGetEntryVal(de);
+    }
+    listAddNodeTail(l,c);
+    c->flags |= REDIS_BLOCKED;
+    aeDeleteFileEvent(server.el,c->fd,AE_READABLE);
+}
+
+/* Unblock a client that's waiting in a blocking operation such as BLPOP */
+static void unblockClient(redisClient *c) {
+    dictEntry *de;
+    list *l;
+
+    printf("unblockClient %p %s\n", c, c->blockingkey->ptr);
+    /* Remove this client from the list of clients waiting for this key. */
+    assert(c->blockingkey != NULL);
+    de = dictFind(c->db->blockingkeys,c->blockingkey);
+    assert(de != NULL);
+    l = dictGetEntryVal(de);
+    listDelNode(l,listSearchKey(l,c));
+    /* If the list is empty we need to remove it to avoid wasting memory */
+    if (listLength(l) == 0)
+        dictDelete(c->db->blockingkeys,c->blockingkey);
+    /* Finally set the right flags in the client structure */
+    decrRefCount(c->blockingkey);
+    c->blockingkey = NULL;
+    c->flags &= (~REDIS_BLOCKED);
+    /* Ok now we are ready to get read events from socket, note that we
+     * can't trap errors here as it's possible that unblockClients() is
+     * called from freeClient() itself, and the only thing we can do
+     * if we failed to register the READABLE event is to kill the client.
+     * Still the following function should never fail in the real world as
+     * we are sure the file descriptor is sane, and we exit on out of mem. */
+    aeCreateFileEvent(server.el, c->fd, AE_READABLE, readQueryFromClient, c);
+    /* As a final step we want to process data if there is some command waiting
+     * in the input buffer. Note that this is safe even if unblockClient()
+     * gets called from freeClient() because freeClient() will be smart
+     * enough to call this function *after* c->querybuf was set to NULL. */
+    if (c->querybuf && sdslen(c->querybuf) > 0) processInputBuffer(c);
+}
+
+/* This should be called from any function PUSHing into lists.
+ * 'c' is the "pushing client", 'key' is the key it is pushing data against,
+ * 'ele' is the element pushed.
+ *
+ * If the function returns 0 there was no client waiting for a list push
+ * against this key.
+ *
+ * If the function returns 1 there was a client waiting for a list push
+ * against this key, the element was passed to this client thus it's not
+ * needed to actually add it to the list and the caller should return asap. */
+static int handleClientsWaitingListPush(redisClient *c, robj *key, robj *ele) {
+    struct dictEntry *de;
+    redisClient *receiver;
+    list *l;
+    listNode *ln;
+
+    de = dictFind(c->db->blockingkeys,key);
+    if (de == NULL) return 0;
+    l = dictGetEntryVal(de);
+    ln = listFirst(l);
+    assert(ln != NULL);
+    receiver = ln->value;
+    listDelNode(l,ln);
+    if (listLength(l) == 0)
+        dictDelete(c->db->blockingkeys,key);
+
+    addReplyBulkLen(receiver,ele);
+    addReply(receiver,ele);
+    addReply(receiver,shared.crlf);
+    unblockClient(receiver);
+    return 1;
+}
+
+/* Blocking RPOP/LPOP */
+static void blockingPopGenericCommand(redisClient *c, int where) {
+    robj *o;
+    time_t timeout;
+
+    o = lookupKeyWrite(c->db,c->argv[1]);
+    if (o != NULL) {
+        if (o->type != REDIS_LIST) {
+            popGenericCommand(c,where);
+            return;
+        } else {
+            list *list = o->ptr;
+            if (listLength(list) != 0) {
+                /* If the list contains elements fall back to the usual
+                 * non-blocking POP operation */
+                popGenericCommand(c,where);
+                return;
+            }
+        }
+    }
+    /* If the list is empty or the key does not exists we must block */
+    timeout = strtol(c->argv[2]->ptr,NULL,10);
+    if (timeout > 0) timeout += time(NULL);
+    blockForKey(c,c->argv[1],timeout);
+}
+
+static void blpopCommand(redisClient *c) {
+    blockingPopGenericCommand(c,REDIS_HEAD);
+}
+
+static void brpopCommand(redisClient *c) {
+    blockingPopGenericCommand(c,REDIS_TAIL);
+}
+
 /* =============================== Replication  ============================= */
 
 static int syncWrite(int fd, char *ptr, ssize_t size, int timeout) {
@@ -5558,6 +5929,7 @@ static int syncWithMaster(void) {
     }
     server.master = createClient(fd);
     server.master->flags |= REDIS_MASTER;
+    server.master->authenticated = 1;
     server.replstate = REDIS_REPL_CONNECTED;
     return REDIS_OK;
 }
@@ -5651,8 +6023,8 @@ static void feedAppendOnlyFile(struct redisCommand *cmd, int dictid, robj **argv
         char seldb[64];
 
         snprintf(seldb,sizeof(seldb),"%d",dictid);
-        buf = sdscatprintf(buf,"*2\r\n$6\r\nSELECT\r\n$%d\r\n%s\r\n",
-            strlen(seldb),seldb);
+        buf = sdscatprintf(buf,"*2\r\n$6\r\nSELECT\r\n$%lu\r\n%s\r\n",
+            (unsigned long)strlen(seldb),seldb);
         server.appendseldb = dictid;
     }
 
@@ -5676,7 +6048,7 @@ static void feedAppendOnlyFile(struct redisCommand *cmd, int dictid, robj **argv
         robj *o = argv[j];
 
         o = getDecodedObject(o);
-        buf = sdscatprintf(buf,"$%d\r\n",sdslen(o->ptr));
+        buf = sdscatprintf(buf,"$%lu\r\n",(unsigned long)sdslen(o->ptr));
         buf = sdscatlen(buf,o->ptr,sdslen(o->ptr));
         buf = sdscatlen(buf,"\r\n",2);
         decrRefCount(o);
@@ -5840,7 +6212,8 @@ static int fwriteBulk(FILE *fp, robj *obj) {
     obj = getDecodedObject(obj);
     snprintf(buf,sizeof(buf),"$%ld\r\n",(long)sdslen(obj->ptr));
     if (fwrite(buf,strlen(buf),1,fp) == 0) goto err;
-    if (fwrite(obj->ptr,sdslen(obj->ptr),1,fp) == 0) goto err;
+    if (sdslen(obj->ptr) && fwrite(obj->ptr,sdslen(obj->ptr),1,fp) == 0)
+        goto err;
     if (fwrite("\r\n",2,1,fp) == 0) goto err;
     decrRefCount(obj);
     return 1;
@@ -5965,11 +6338,11 @@ static int rewriteAppendOnlyFile(char *filename) {
                 }
                 dictReleaseIterator(di);
             } else {
-                assert(0 != 0);
+                redisAssert(0 != 0);
             }
             /* Save the expire time */
             if (expiretime != -1) {
-                char cmd[]="*3\r\n$6\r\nEXPIRE\r\n";
+                char cmd[]="*3\r\n$8\r\nEXPIREAT\r\n";
                 /* If this key is already expired skip it */
                 if (expiretime < now) continue;
                 if (fwrite(cmd,sizeof(cmd)-1,1,fp) == 0) goto werr;
@@ -5998,7 +6371,7 @@ static int rewriteAppendOnlyFile(char *filename) {
 werr:
     fclose(fp);
     unlink(tmpfile);
-    redisLog(REDIS_WARNING,"Write error writing append only fileon disk: %s", strerror(errno));
+    redisLog(REDIS_WARNING,"Write error writing append only file on disk: %s", strerror(errno));
     if (di) dictReleaseIterator(di);
     return REDIS_ERR;
 }
@@ -6057,7 +6430,8 @@ static void bgrewriteaofCommand(redisClient *c) {
         return;
     }
     if (rewriteAppendOnlyFileBackground() == REDIS_OK) {
-        addReply(c,shared.ok);
+        char *status = "+Background append only file rewriting started\r\n";
+        addReplySds(c,sdsnew(status));
     } else {
         addReply(c,shared.err);
     }
@@ -6087,6 +6461,14 @@ static void debugCommand(redisClient *c) {
         }
         redisLog(REDIS_WARNING,"DB reloaded by DEBUG RELOAD");
         addReply(c,shared.ok);
+    } else if (!strcasecmp(c->argv[1]->ptr,"loadaof")) {
+        emptyDb();
+        if (loadAppendOnlyFile(server.appendfilename) != REDIS_OK) {
+            addReply(c,shared.err);
+            return;
+        }
+        redisLog(REDIS_WARNING,"Append Only File loaded by DEBUG LOADAOF");
+        addReply(c,shared.ok);
     } else if (!strcasecmp(c->argv[1]->ptr,"object") && c->argc == 3) {
         dictEntry *de = dictFind(c->db->dict,c->argv[2]);
         robj *key, *val;
@@ -6099,13 +6481,23 @@ static void debugCommand(redisClient *c) {
         val = dictGetEntryVal(de);
         addReplySds(c,sdscatprintf(sdsempty(),
             "+Key at:%p refcount:%d, value at:%p refcount:%d encoding:%d\r\n",
-                key, key->refcount, val, val->refcount, val->encoding));
+                (void*)key, key->refcount, (void*)val, val->refcount,
+                val->encoding));
     } else {
         addReplySds(c,sdsnew(
             "-ERR Syntax error, try DEBUG [SEGFAULT|OBJECT <key>|RELOAD]\r\n"));
     }
 }
 
+static void _redisAssert(char *estr) {
+    redisLog(REDIS_WARNING,"=== ASSERTION FAILED ===");
+    redisLog(REDIS_WARNING,"==> %s\n",estr);
+#ifdef HAVE_BACKTRACE
+    redisLog(REDIS_WARNING,"(forcing SIGSEGV in order to print the stack trace)");
+    *((char*)-1) = 'x';
+#endif
+}
+
 /* =================================== Main! ================================ */
 
 #ifdef __linux__
@@ -6135,6 +6527,7 @@ static void daemonize(void) {
     FILE *fp;
 
     if (fork() != 0) exit(0); /* parent exits */
+    printf("New pid: %d\n", getpid());
     setsid(); /* create a new session */
 
     /* Every output goes to /dev/null. If Redis is daemonized but
@@ -6165,8 +6558,8 @@ int main(int argc, char **argv) {
     } 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'");
     }
-    initServer();
     if (server.daemonize) daemonize();
+    initServer();
     redisLog(REDIS_NOTICE,"Server started, Redis version " REDIS_VERSION);
 #ifdef __linux__
     linuxOvercommitMemoryWarning();
@@ -6197,15 +6590,19 @@ static void *getMcontextEip(ucontext_t *uc) {
 #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__) || defined(__X86_64__) /* Linux x86 */
-    return (void*) uc->uc_mcontext.gregs[REG_EIP];
+#elif defined(__i386__) || defined(__X86_64__)  || defined(__x86_64__)
+    return (void*) uc->uc_mcontext.gregs[REG_EIP]; /* Linux 32/64 bit */
 #elif defined(__ia64__) /* Linux IA64 */
     return (void*) uc->uc_mcontext.sc_ip;
 #else