Merge remote-tracking branch 'refs/remotes/antirez/unstable' into unstable

50 files changed, 2668 insertions, 420 deletions

diff --git a/CONTRIBUTING b/CONTRIBUTING
index b33aacb3e..f57de3fd9 100644
--- a/CONTRIBUTING
+++ b/CONTRIBUTING
@@ -12,7 +12,7 @@ each source file that you contribute.
 
   PLEASE DO NOT POST GENERAL QUESTIONS that are not about bugs or suspected
   bugs in the Github issues system. We'll be very happy to help you and provide
-  all the support Reddit sub:
+  all the support at the Reddit sub:
 
       http://reddit.com/r/redis
 
@@ -24,7 +24,7 @@ each source file that you contribute.
 
 1. If it is a major feature or a semantical change, please post it as a new submission in r/redis on Reddit at http://reddit.com/r/redis. Try to be passionate about why the feature is needed, make users upvote your proposal to gain traction and so forth. Read feedbacks about the community. But in this first step **please don't write code yet**.
 
-2. If in step 1 you get an acknowledge from the project leaders, use the
+2. If in step 1 you get an acknowledgment from the project leaders, use the
    following procedure to submit a patch:
 
     a. Fork Redis on github ( http://help.github.com/fork-a-repo/ )
diff --git a/README.md b/README.md
index c6d46e6e2..70a15790f 100644
--- a/README.md
+++ b/README.md
@@ -39,7 +39,7 @@ You can run a 32 bit Redis binary using:
 
     % make 32bit
 
-After building Redis is a good idea to test it, using:
+After building Redis, it is a good idea to test it using:
 
     % make test
 
@@ -47,8 +47,8 @@ Fixing build problems with dependencies or cached build options
 ---------
 
 Redis has some dependencies which are included into the `deps` directory.
-`make` does not rebuild dependencies automatically, even if something in the
-source code of dependencies is changed.
+`make` does not automatically rebuild dependencies even if something in
+the source code of dependencies changes.
 
 When you update the source code with `git pull` or when code inside the
 dependencies tree is modified in any other way, make sure to use the following
@@ -109,14 +109,14 @@ To run Redis with the default configuration just type:
 
     % cd src
     % ./redis-server
-    
+
 If you want to provide your redis.conf, you have to run it using an additional
 parameter (the path of the configuration file):
 
     % cd src
     % ./redis-server /path/to/redis.conf
 
-It is possible to alter the Redis configuration passing parameters directly
+It is possible to alter the Redis configuration by passing parameters directly
 as options using the command line. Examples:
 
     % ./redis-server --port 9999 --slaveof 127.0.0.1 6379
@@ -174,7 +174,7 @@ You'll be able to stop and start Redis using the script named
 `/etc/init.d/redis_<portnumber>`, for instance `/etc/init.d/redis_6379`.
 
 Code contributions
----
+-----------------
 
 Note: by contributing code to the Redis project in any form, including sending
 a pull request via Github, a code fragment or patch via private email or
@@ -196,8 +196,8 @@ or you just untarred the Redis distribution tar ball. In both the cases
 you are basically one step away from the source code, so here we explain
 the Redis source code layout, what is in each file as a general idea, the
 most important functions and structures inside the Redis server and so forth.
-We keep all the discussion at an high level without digging into the details
-since this document would be huge otherwise, and our code base changes
+We keep all the discussion at a high level without digging into the details
+since this document would be huge otherwise and our code base changes
 continuously, but a general idea should be a good starting point to
 understand more. Moreover most of the code is heavily commented and easy
 to follow.
@@ -206,17 +206,17 @@ Source code layout
 ---
 
 The Redis root directory just contains this README, the Makefile which
-actually calls the real Makefile inside the `src` directory, an example
-configuration for Redis and Sentinel. Finally you can find a few shell
+calls the real Makefile inside the `src` directory and an example
+configuration for Redis and Sentinel. You can find a few shell
 scripts that are used in order to execute the Redis, Redis Cluster and
 Redis Sentinel unit tests, which are implemented inside the `tests`
 directory.
 
-Inside the root directory the are the following important directories:
+Inside the root are the following important directories:
 
 * `src`: contains the Redis implementation, written in C.
 * `tests`: contains the unit tests, implemented in Tcl.
-* `deps`: contains libraries Redis uses. Everything needed to compile Redis is inside this directory, your system needs to provide just the `libc`, a POSIX compatible interface, and a C compiler. Notably `deps` contains a copy of `jemalloc`, which is the default allocator of Redis under Linux. Note that under `deps` there are also things which started with the Redis project, but for which the main repository is not `anitrez/redis`. an exception to this rule is `deps/geohash-int` which is the low level geocoding library used by Redis: it originated from a different project, but at this point it diverged so much that it is developed as a separated entity directly inside the Redis repository.
+* `deps`: contains libraries Redis uses. Everything needed to compile Redis is inside this directory; your system just needs to provide `libc`, a POSIX compatible interface and a C compiler. Notably `deps` contains a copy of `jemalloc`, which is the default allocator of Redis under Linux. Note that under `deps` there are also things which started with the Redis project, but for which the main repository is not `anitrez/redis`. An exception to this rule is `deps/geohash-int` which is the low level geocoding library used by Redis: it originated from a different project, but at this point it diverged so much that it is developed as a separated entity directly inside the Redis repository.
 
 There are a few more directories but they are not very important for our goals
 here. We'll focus mostly on `src`, where the Redis implementation is contained,
@@ -225,34 +225,34 @@ exposed is the logical one to follow in order to disclose different layers
 of complexity incrementally.
 
 Note: lately Redis was refactored quite a bit. Function names and file
-names changed, so you may find that this documentation reflects the
+names have been changed, so you may find that this documentation reflects the
 `unstable` branch more closely. For instance in Redis 3.0 the `server.c`
-and `server.h` files were renamed `redis.c` and `redis.h`. However the overall
+and `server.h` files were named to `redis.c` and `redis.h`. However the overall
 structure is the same. Keep in mind that all the new developments and pull
 requests should be performed against the `unstable` branch.
 
 server.h
 ---
 
-The simplest way to understand how a program works, is to understand the
+The simplest way to understand how a program works is to understand the
 data structures it uses. So we'll start from the main header file of
 Redis, which is `server.h`.
 
 All the server configuration and in general all the shared state is
 defined in a global structure called `server`, of type `struct redisServer`.
-A few important fields in this structure:
+A few important fields in this structure are:
 
 * `server.db` is an array of Redis databases, where data is stored.
 * `server.commands` is the command table.
 * `server.clients` is a linked list of clients connected to the server.
 * `server.master` is a special client, the master, if the instance is a slave.
 
-There are tons of other fields, most fields are commented directly inside
+There are tons of other fields. Most fields are commented directly inside
 the structure definition.
 
 Another important Redis data structure is the one defining a client.
 In the past it was called `redisClient`, now just `client`. The structure
-has many fields, here we'll show just the main ones:
+has many fields, here we'll just show the main ones:
 
     struct client {
         int fd;
@@ -270,7 +270,7 @@ The client structure defines a *connected client*:
 
 * The `fd` field is the client socket file descriptor.
 * `argc` and `argv` are populated with the command the client is executing, so that functions implementing a given Redis command can read the arguments.
-* `querybuf` accumulates the requests from the client, which are parsed by the Redis server according to the Redis protocol, and executed calling the implementations of the commands the client is executing.
+* `querybuf` accumulates the requests from the client, which are parsed by the Redis server according to the Redis protocol and executed by calling the implementations of the commands the client is executing.
 * `reply` and `buf` are dynamic and static buffers that accumulate the replies the server sends to the client. These buffers are incrementally written to the socket as soon as the file descriptor is writable.
 
 As you can see in the client structure above, arguments in a command
@@ -288,16 +288,16 @@ structure, which defines a *Redis object*:
 Basically this structure can represent all the basic Redis data types like
 strings, lists, sets, sorted sets and so forth. The interesting thing is that
 it has a `type` field, so that it is possible to know what type a given
-object is, and a `refcount`, so that the same object can be referenced
+object has, and a `refcount`, so that the same object can be referenced
 in multiple places without allocating it multiple times. Finally the `ptr`
-field points to the actual representation of the object, that may vary
+field points to the actual representation of the object, which might vary
 even for the same type, depending on the `encoding` used.
 
 Redis objects are used extensively in the Redis internals, however in order
 to avoid the overhead of indirect accesses, recently in many places
 we just use plain dynamic strings not wrapped inside a Redis object.
 
-sever.c
+server.c
 ---
 
 This is the entry point of the Redis server, where the `main()` function
@@ -306,7 +306,7 @@ the Redis server.
 
 * `initServerConfig()` setups the default values of the `server` structure.
 * `initServer()` allocates the data structures needed to operate, setup the listening socket, and so forth.
-* `aeMain()` enters the event loop listening for new connections.
+* `aeMain()` starts the event loop which listens for new connections.
 
 There are two special functions called periodically by the event loop:
 
@@ -328,7 +328,7 @@ This file defines all the I/O functions with clients, masters and slaves
 
 * `createClient()` allocates and initializes a new client.
 * the `addReply*()` family of functions are used by commands implementations in order to append data to the client structure, that will be transmitted to the client as a reply for a given command executed.
-* `writeToClient()` transmits the data pending in the output buffers to the client, and is called by the *writable event handler* `sendReplyToClient()`.
+* `writeToClient()` transmits the data pending in the output buffers to the client and is called by the *writable event handler* `sendReplyToClient()`.
 * `readQueryFromClient()` is the *readable event handler* and accumulates data from read from the client into the query buffer.
 * `processInputBuffer()` is the entry point in order to parse the client query buffer according to the Redis protocol. Once commands are ready to be processed, it calls `processCommand()` which is defined inside `server.c` in order to actually execute the command.
 * `freeClient()` deallocates, disconnects and removes a client.
@@ -439,9 +439,8 @@ There are tons of commands implementations inside th Redis source code
 that can serve as examples of actual commands implementations. To write
 a few toy commands can be a good exercise to familiarize with the code base.
 
-There are also many other files not described here,  but it is useless to
-cover everything, we want just to help you with the first steps,
-eventually you'll find your way inside the Redis code base :-)
+There are also many other files not described here, but it is useless to
+cover everything. We want to just help you with the first steps.
+Eventually you'll find your way inside the Redis code base :-)
 
 Enjoy!
-
diff --git a/deps/Makefile b/deps/Makefile
index 10ae6e790..1c10bce9e 100644
--- a/deps/Makefile
+++ b/deps/Makefile
@@ -78,7 +78,7 @@ JEMALLOC_LDFLAGS= $(LDFLAGS)
 
 jemalloc: .make-prerequisites
 	@printf '%b %b\n' $(MAKECOLOR)MAKE$(ENDCOLOR) $(BINCOLOR)$@$(ENDCOLOR)
-	cd jemalloc && ./configure --with-jemalloc-prefix=je_ --enable-cc-silence CFLAGS="$(JEMALLOC_CFLAGS)" LDFLAGS="$(JEMALLOC_LDFLAGS)"
+	cd jemalloc && ./configure --with-lg-quantum=3 --with-jemalloc-prefix=je_ --enable-cc-silence CFLAGS="$(JEMALLOC_CFLAGS)" LDFLAGS="$(JEMALLOC_LDFLAGS)"
 	cd jemalloc && $(MAKE) CFLAGS="$(JEMALLOC_CFLAGS)" LDFLAGS="$(JEMALLOC_LDFLAGS)" lib/libjemalloc.a
 
 .PHONY: jemalloc
diff --git a/deps/geohash-int/geohash_helper.c b/deps/geohash-int/geohash_helper.c
index 4c3762faf..4b8894676 100644
--- a/deps/geohash-int/geohash_helper.c
+++ b/deps/geohash-int/geohash_helper.c
@@ -72,7 +72,7 @@ uint8_t geohashEstimateStepsByRadius(double range_meters, double lat) {
 
     /* Frame to valid range. */
     if (step < 1) step = 1;
-    if (step > 26) step = 25;
+    if (step > 26) step = 26;
     return step;
 }
 
@@ -89,6 +89,8 @@ int geohashBoundingBox(double longitude, double latitude, double radius_meters,
     lonr = deg_rad(longitude);
     latr = deg_rad(latitude);
 
+    if (radius_meters > EARTH_RADIUS_IN_METERS)
+        radius_meters = EARTH_RADIUS_IN_METERS;
     double distance = radius_meters / EARTH_RADIUS_IN_METERS;
     double min_latitude = latr - distance;
     double max_latitude = latr + distance;
diff --git a/redis.conf b/redis.conf
index a39a643ab..67cd50245 100644
--- a/redis.conf
+++ b/redis.conf
@@ -125,8 +125,9 @@ timeout 0
 # Note that to close the connection the double of the time is needed.
 # On other kernels the period depends on the kernel configuration.
 #
-# A reasonable value for this option is 60 seconds.
-tcp-keepalive 0
+# A reasonable value for this option is 300 seconds, which is the new
+# Redis default starting with Redis 3.2.1.
+tcp-keepalive 300
 
 ################################# GENERAL #####################################
 
@@ -154,7 +155,7 @@ supervised no
 #
 # Creating a pid file is best effort: if Redis is not able to create it
 # nothing bad happens, the server will start and run normally.
-pidfile /var/run/redis.pid
+pidfile /var/run/redis_6379.pid
 
 # Specify the server verbosity level.
 # This can be one of:
diff --git a/src/Makefile b/src/Makefile
index c390d3f2e..89355984c 100644
--- a/src/Makefile
+++ b/src/Makefile
@@ -65,17 +65,27 @@ ifeq ($(uname_S),SunOS)
 	FINAL_LIBS+= -ldl -lnsl -lsocket -lresolv -lpthread -lrt
 else
 ifeq ($(uname_S),Darwin)
-	# Darwin (nothing to do)
+	# Darwin
+	FINAL_LIBS+= -ldl
 else
 ifeq ($(uname_S),AIX)
         # AIX
         FINAL_LDFLAGS+= -Wl,-bexpall
-        FINAL_LIBS+= -pthread -lcrypt -lbsd
-
+        FINAL_LIBS+=-ldl -pthread -lcrypt -lbsd
+else
+ifeq ($(uname_S),OpenBSD)
+	# OpenBSD
+	FINAL_LIBS+= -lpthread
+else
+ifeq ($(uname_S),FreeBSD)
+	# FreeBSD
+	FINAL_LIBS+= -lpthread
 else
 	# All the other OSes (notably Linux)
 	FINAL_LDFLAGS+= -rdynamic
-	FINAL_LIBS+= -pthread
+	FINAL_LIBS+=-ldl -pthread
+endif
+endif
 endif
 endif
 endif
@@ -95,7 +105,7 @@ endif
 ifeq ($(MALLOC),jemalloc)
 	DEPENDENCY_TARGETS+= jemalloc
 	FINAL_CFLAGS+= -DUSE_JEMALLOC -I../deps/jemalloc/include
-	FINAL_LIBS+= ../deps/jemalloc/lib/libjemalloc.a -ldl
+	FINAL_LIBS+= ../deps/jemalloc/lib/libjemalloc.a
 endif
 
 REDIS_CC=$(QUIET_CC)$(CC) $(FINAL_CFLAGS)
diff --git a/src/ae_select.c b/src/ae_select.c
index e2b7a9e8a..c039a8ea3 100644
--- a/src/ae_select.c
+++ b/src/ae_select.c
@@ -29,6 +29,7 @@
  */
 
 
+#include <sys/select.h>
 #include <string.h>
 
 typedef struct aeApiState {
diff --git a/src/aof.c b/src/aof.c
index 9df1e9b9e..aa726d33b 100644
--- a/src/aof.c
+++ b/src/aof.c
@@ -693,6 +693,7 @@ int loadAppendOnlyFile(char *filename) {
         }
 
         /* Run the command in the context of a fake client */
+        fakeClient->cmd = cmd;
         cmd->proc(fakeClient);
 
         /* The fake client should not have a reply */
@@ -703,6 +704,7 @@ int loadAppendOnlyFile(char *filename) {
         /* Clean up. Command code may have changed argv/argc so we use the
          * argv/argc of the client instead of the local variables. */
         freeFakeClientArgv(fakeClient);
+        fakeClient->cmd = NULL;
         if (server.aof_load_truncated) valid_up_to = ftello(fp);
     }
 
@@ -983,6 +985,18 @@ int rewriteHashObject(rio *r, robj *key, robj *o) {
     return 1;
 }
 
+/* Call the module type callback in order to rewrite a data type
+ * taht is exported by a module and is not handled by Redis itself.
+ * The function returns 0 on error, 1 on success. */
+int rewriteModuleObject(rio *r, robj *key, robj *o) {
+    RedisModuleIO io;
+    moduleValue *mv = o->ptr;
+    moduleType *mt = mv->type;
+    moduleInitIOContext(io,mt,r);
+    mt->aof_rewrite(&io,key,mv->value);
+    return io.error ? 0 : 1;
+}
+
 /* This function is called by the child rewriting the AOF file to read
  * the difference accumulated from the parent into a buffer, that is
  * concatenated at the end of the rewrite. */
@@ -1075,6 +1089,8 @@ int rewriteAppendOnlyFile(char *filename) {
                 if (rewriteSortedSetObject(&aof,&key,o) == 0) goto werr;
             } else if (o->type == OBJ_HASH) {
                 if (rewriteHashObject(&aof,&key,o) == 0) goto werr;
+            } else if (o->type == OBJ_MODULE) {
+                if (rewriteModuleObject(&aof,&key,o) == 0) goto werr;
             } else {
                 serverPanic("Unknown object type");
             }
diff --git a/src/bitops.c b/src/bitops.c
index ed7e384a0..302e811d2 100644
--- a/src/bitops.c
+++ b/src/bitops.c
@@ -215,12 +215,7 @@ void setUnsignedBitfield(unsigned char *p, uint64_t offset, uint64_t bits, uint6
 }
 
 void setSignedBitfield(unsigned char *p, uint64_t offset, uint64_t bits, int64_t value) {
-    uint64_t uv;
-
-    if (value >= 0)
-        uv = value;
-    else
-        uv = UINT64_MAX + value + 1;
+    uint64_t uv = value; /* Casting will add UINT64_MAX + 1 if v is negative. */
     setUnsignedBitfield(p,offset,bits,uv);
 }
 
@@ -239,9 +234,21 @@ uint64_t getUnsignedBitfield(unsigned char *p, uint64_t offset, uint64_t bits) {
 }
 
 int64_t getSignedBitfield(unsigned char *p, uint64_t offset, uint64_t bits) {
-    int64_t value = getUnsignedBitfield(p,offset,bits);
+    int64_t value;
+    union {uint64_t u; int64_t i;} conv;
+
+    /* Converting from unsigned to signed is undefined when the value does
+     * not fit, however here we assume two's complement and the original value
+     * was obtained from signed -> unsigned conversion, so we'll find the
+     * most significant bit set if the original value was negative.
+     *
+     * Note that two's complement is mandatory for exact-width types
+     * according to the C99 standard. */
+    conv.u = getUnsignedBitfield(p,offset,bits);
+    value = conv.i;
+
     /* If the top significant bit is 1, propagate it to all the
-     * higher bits for two complement representation of signed
+     * higher bits for two's complement representation of signed
      * integers. */
     if (value & ((uint64_t)1 << (bits-1)))
         value |= ((uint64_t)-1) << bits;
@@ -299,7 +306,7 @@ int checkUnsignedBitfieldOverflow(uint64_t value, int64_t incr, uint64_t bits, i
 
 handle_wrap:
     {
-        uint64_t mask = ((int64_t)-1) << bits;
+        uint64_t mask = ((uint64_t)-1) << bits;
         uint64_t res = value+incr;
 
         res &= ~mask;
@@ -342,7 +349,7 @@ int checkSignedBitfieldOverflow(int64_t value, int64_t incr, uint64_t bits, int
 
 handle_wrap:
     {
-        uint64_t mask = ((int64_t)-1) << bits;
+        uint64_t mask = ((uint64_t)-1) << bits;
         uint64_t msb = (uint64_t)1 << (bits-1);
         uint64_t a = value, b = incr, c;
         c = a+b; /* Perform addition as unsigned so that's defined. */
@@ -476,6 +483,37 @@ robj *lookupStringForBitCommand(client *c, size_t maxbit) {
     return o;
 }
 
+/* Return a pointer to the string object content, and stores its length
+ * in 'len'. The user is required to pass (likely stack allocated) buffer
+ * 'llbuf' of at least LONG_STR_SIZE bytes. Such a buffer is used in the case
+ * the object is integer encoded in order to provide the representation
+ * without usign heap allocation.
+ *
+ * The function returns the pointer to the object array of bytes representing
+ * the string it contains, that may be a pointer to 'llbuf' or to the
+ * internal object representation. As a side effect 'len' is filled with
+ * the length of such buffer.
+ *
+ * If the source object is NULL the function is guaranteed to return NULL
+ * and set 'len' to 0. */
+unsigned char *getObjectReadOnlyString(robj *o, long *len, char *llbuf) {
+    serverAssert(o->type == OBJ_STRING);
+    unsigned char *p = NULL;
+
+    /* Set the 'p' pointer to the string, that can be just a stack allocated
+     * array if our string was integer encoded. */
+    if (o && o->encoding == OBJ_ENCODING_INT) {
+        p = (unsigned char*) llbuf;
+        if (len) *len = ll2string(llbuf,LONG_STR_SIZE,(long)o->ptr);
+    } else if (o) {
+        p = (unsigned char*) o->ptr;
+        if (len) *len = sdslen(o->ptr);
+    } else {
+        if (len) *len = 0;
+    }
+    return p;
+}
+
 /* SETBIT key offset bitvalue */
 void setbitCommand(client *c) {
     robj *o;
@@ -721,21 +759,12 @@ void bitcountCommand(client *c) {
     robj *o;
     long start, end, strlen;
     unsigned char *p;
-    char llbuf[32];
+    char llbuf[LONG_STR_SIZE];
 
     /* Lookup, check for type, and return 0 for non existing keys. */
     if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL ||
         checkType(c,o,OBJ_STRING)) return;
-
-    /* Set the 'p' pointer to the string, that can be just a stack allocated
-     * array if our string was integer encoded. */
-    if (o->encoding == OBJ_ENCODING_INT) {
-        p = (unsigned char*) llbuf;
-        strlen = ll2string(llbuf,sizeof(llbuf),(long)o->ptr);
-    } else {
-        p = (unsigned char*) o->ptr;
-        strlen = sdslen(o->ptr);
-    }
+    p = getObjectReadOnlyString(o,&strlen,llbuf);
 
     /* Parse start/end range if any. */
     if (c->argc == 4) {
@@ -744,6 +773,10 @@ void bitcountCommand(client *c) {
         if (getLongFromObjectOrReply(c,c->argv[3],&end,NULL) != C_OK)
             return;
         /* Convert negative indexes */
+        if (start < 0 && end < 0 && start > end) {
+            addReply(c,shared.czero);
+            return;
+        }
         if (start < 0) start = strlen+start;
         if (end < 0) end = strlen+end;
         if (start < 0) start = 0;
@@ -775,7 +808,7 @@ void bitposCommand(client *c) {
     robj *o;
     long bit, start, end, strlen;
     unsigned char *p;
-    char llbuf[32];
+    char llbuf[LONG_STR_SIZE];
     int end_given = 0;
 
     /* Parse the bit argument to understand what we are looking for, set
@@ -795,16 +828,7 @@ void bitposCommand(client *c) {
         return;
     }
     if (checkType(c,o,OBJ_STRING)) return;
-
-    /* Set the 'p' pointer to the string, that can be just a stack allocated
-     * array if our string was integer encoded. */
-    if (o->encoding == OBJ_ENCODING_INT) {
-        p = (unsigned char*) llbuf;
-        strlen = ll2string(llbuf,sizeof(llbuf),(long)o->ptr);
-    } else {
-        p = (unsigned char*) o->ptr;
-        strlen = sdslen(o->ptr);
-    }
+    p = getObjectReadOnlyString(o,&strlen,llbuf);
 
     /* Parse start/end range if any. */
     if (c->argc == 4 || c->argc == 5) {
@@ -882,6 +906,8 @@ void bitfieldCommand(client *c) {
     int j, numops = 0, changes = 0;
     struct bitfieldOp *ops = NULL; /* Array of ops to execute at end. */
     int owtype = BFOVERFLOW_WRAP; /* Overflow type. */
+    int readonly = 1;
+    long higest_write_offset = 0;
 
     for (j = 2; j < c->argc; j++) {
         int remargs = c->argc-j-1; /* Remaining args other than current. */
@@ -929,8 +955,10 @@ void bitfieldCommand(client *c) {
             return;
         }
 
-        /* INCRBY and SET require another argument. */
         if (opcode != BITFIELDOP_GET) {
+            readonly = 0;
+            higest_write_offset = bitoffset + bits - 1;
+            /* INCRBY and SET require another argument. */
             if (getLongLongFromObjectOrReply(c,c->argv[j+3],&i64,NULL) != C_OK){
                 zfree(ops);
                 return;
@@ -950,6 +978,18 @@ void bitfieldCommand(client *c) {
         j += 3 - (opcode == BITFIELDOP_GET);
     }
 
+    if (readonly) {
+        /* Lookup for read is ok if key doesn't exit, but errors
+         * if it's not a string. */
+        o = lookupKeyRead(c->db,c->argv[1]);
+        if (o != NULL && checkType(c,o,OBJ_STRING)) return;
+    } else {
+        /* Lookup by making room up to the farest bit reached by
+         * this operation. */
+        if ((o = lookupStringForBitCommand(c,
+            higest_write_offset)) == NULL) return;
+    }
+
     addReplyMultiBulkLen(c,numops);
 
     /* Actually process the operations. */
@@ -964,11 +1004,6 @@ void bitfieldCommand(client *c) {
              * for simplicity. SET return value is the previous value so
              * we need fetch & store as well. */
 
-            /* Lookup by making room up to the farest bit reached by
-             * this operation. */
-            if ((o = lookupStringForBitCommand(c,
-                thisop->offset + (thisop->bits-1))) == NULL) return;
-
             /* We need two different but very similar code paths for signed
              * and unsigned operations, since the set of functions to get/set
              * the integers and the used variables types are different. */
@@ -1035,20 +1070,23 @@ void bitfieldCommand(client *c) {
             changes++;
         } else {
             /* GET */
-            o = lookupKeyRead(c->db,c->argv[1]);
-            size_t olen = (o == NULL) ? 0 : sdslen(o->ptr);
             unsigned char buf[9];
+            long strlen = 0;
+            unsigned char *src = NULL;
+            char llbuf[LONG_STR_SIZE];
+
+            if (o != NULL)
+                src = getObjectReadOnlyString(o,&strlen,llbuf);
 
             /* For GET we use a trick: before executing the operation
              * copy up to 9 bytes to a local buffer, so that we can easily
              * execute up to 64 bit operations that are at actual string
              * object boundaries. */
             memset(buf,0,9);
-            unsigned char *src = o ? o->ptr : NULL;
             int i;
             size_t byte = thisop->offset >> 3;
             for (i = 0; i < 9; i++) {
-                if (src == NULL || i+byte >= olen) break;
+                if (src == NULL || i+byte >= (size_t)strlen) break;
                 buf[i] = src[i+byte];
             }
 
diff --git a/src/cluster.c b/src/cluster.c
index 1f19db3e4..9289f6782 100644
--- a/src/cluster.c
+++ b/src/cluster.c
@@ -4535,7 +4535,7 @@ int verifyDumpPayload(unsigned char *p, size_t len) {
 
     /* Verify RDB version */
     rdbver = (footer[1] << 8) | footer[0];
-    if (rdbver != RDB_VERSION) return C_ERR;
+    if (rdbver > RDB_VERSION) return C_ERR;
 
     /* Verify CRC64 */
     crc = crc64(0,p,len-8);
diff --git a/src/config.c b/src/config.c
index c72f0aeb2..77029b934 100644
--- a/src/config.c
+++ b/src/config.c
@@ -153,6 +153,20 @@ void resetServerSaveParams(void) {
     server.saveparamslen = 0;
 }
 
+void queueLoadModule(sds path, sds *argv, int argc) {
+    int i;
+    struct moduleLoadQueueEntry *loadmod;
+
+    loadmod = zmalloc(sizeof(struct moduleLoadQueueEntry));
+    loadmod->argv = zmalloc(sizeof(robj*)*argc);
+    loadmod->path = sdsnew(path);
+    loadmod->argc = argc;
+    for (i = 0; i < argc; i++) {
+        loadmod->argv[i] = createRawStringObject(argv[i],sdslen(argv[i]));
+    }
+    listAddNodeTail(server.loadmodule_queue,loadmod);
+}
+
 void loadServerConfigFromString(char *config) {
     char *err = NULL;
     int linenum = 0, totlines, i;
@@ -632,8 +646,8 @@ void loadServerConfigFromString(char *config) {
                     "Allowed values: 'upstart', 'systemd', 'auto', or 'no'";
                 goto loaderr;
             }
-        } else if (!strcasecmp(argv[0],"loadmodule") && argc == 2) {
-            listAddNodeTail(server.loadmodule_queue,sdsnew(argv[1]));
+        } else if (!strcasecmp(argv[0],"loadmodule") && argc >= 2) {
+            queueLoadModule(argv[1],&argv[2],argc-2);
         } else if (!strcasecmp(argv[0],"sentinel")) {
             /* argc == 1 is handled by main() as we need to enter the sentinel
              * mode ASAP. */
@@ -719,7 +733,7 @@ void loadServerConfig(char *filename, char *options) {
 
 #define config_set_numerical_field(_name,_var,min,max) \
     } else if (!strcasecmp(c->argv[2]->ptr,_name)) { \
-        if (getLongLongFromObject(o,&ll) == C_ERR || ll < 0) goto badfmt; \
+        if (getLongLongFromObject(o,&ll) == C_ERR) goto badfmt; \
         if (min != LLONG_MIN && ll < min) goto badfmt; \
         if (max != LLONG_MAX && ll > max) goto badfmt; \
         _var = ll;
@@ -950,9 +964,9 @@ void configSetCommand(client *c) {
     } config_set_numerical_field(
       "hash-max-ziplist-value",server.hash_max_ziplist_value,0,LLONG_MAX) {
     } config_set_numerical_field(
-      "list-max-ziplist-size",server.list_max_ziplist_size,0,LLONG_MAX) {
+      "list-max-ziplist-size",server.list_max_ziplist_size,INT_MIN,INT_MAX) {
     } config_set_numerical_field(
-      "list-compress-depth",server.list_compress_depth,0,LLONG_MAX) {
+      "list-compress-depth",server.list_compress_depth,0,INT_MAX) {
     } config_set_numerical_field(
       "set-max-intset-entries",server.set_max_intset_entries,0,LLONG_MAX) {
     } config_set_numerical_field(
diff --git a/src/db.c b/src/db.c
index 6f70a5383..4db7d890f 100644
--- a/src/db.c
+++ b/src/db.c
@@ -38,7 +38,10 @@
  * C-level DB API
  *----------------------------------------------------------------------------*/
 
-robj *lookupKey(redisDb *db, robj *key) {
+/* Low level key lookup API, not actually called directly from commands
+ * implementations that should instead rely on lookupKeyRead(),
+ * lookupKeyWrite() and lookupKeyReadWithFlags(). */
+robj *lookupKey(redisDb *db, robj *key, int flags) {
     dictEntry *de = dictFind(db->dict,key->ptr);
     if (de) {
         robj *val = dictGetVal(de);
@@ -46,15 +49,40 @@ robj *lookupKey(redisDb *db, robj *key) {
         /* Update the access time for the ageing algorithm.
          * Don't do it if we have a saving child, as this will trigger
          * a copy on write madness. */
-        if (server.rdb_child_pid == -1 && server.aof_child_pid == -1)
+        if (server.rdb_child_pid == -1 &&
+            server.aof_child_pid == -1 &&
+            !(flags & LOOKUP_NOTOUCH))
+        {
             val->lru = LRU_CLOCK();
+        }
         return val;
     } else {
         return NULL;
     }
 }
 
-robj *lookupKeyRead(redisDb *db, robj *key) {
+/* Lookup a key for read operations, or return NULL if the key is not found
+ * in the specified DB.
+ *
+ * As a side effect of calling this function:
+ * 1. A key gets expired if it reached it's TTL.
+ * 2. The key last access time is updated.
+ * 3. The global keys hits/misses stats are updated (reported in INFO).
+ *
+ * This API should not be used when we write to the key after obtaining
+ * the object linked to the key, but only for read only operations.
+ *
+ * Flags change the behavior of this command:
+ *
+ *  LOOKUP_NONE (or zero): no special flags are passed.
+ *  LOOKUP_NOTOUCH: don't alter the last access time of the key.
+ *
+ * Note: this function also returns NULL is the key is logically expired
+ * but still existing, in case this is a slave, since this API is called only
+ * for read operations. Even if the key expiry is master-driven, we can
+ * correctly report a key is expired on slaves even if the master is lagging
+ * expiring our key via DELs in the replication link. */
+robj *lookupKeyReadWithFlags(redisDb *db, robj *key, int flags) {
     robj *val;
 
     if (expireIfNeeded(db,key) == 1) {
@@ -83,7 +111,7 @@ robj *lookupKeyRead(redisDb *db, robj *key) {
             return NULL;
         }
     }
-    val = lookupKey(db,key);
+    val = lookupKey(db,key,flags);
     if (val == NULL)
         server.stat_keyspace_misses++;
     else
@@ -91,9 +119,20 @@ robj *lookupKeyRead(redisDb *db, robj *key) {
     return val;
 }
 
+/* Like lookupKeyReadWithFlags(), but does not use any flag, which is the
+ * common case. */
+robj *lookupKeyRead(redisDb *db, robj *key) {
+    return lookupKeyReadWithFlags(db,key,LOOKUP_NONE);
+}
+
+/* Lookup a key for write operations, and as a side effect, if needed, expires
+ * the key if its TTL is reached.
+ *
+ * Returns the linked value object if the key exists or NULL if the key
+ * does not exist in the specified DB. */
 robj *lookupKeyWrite(redisDb *db, robj *key) {
     expireIfNeeded(db,key);
-    return lookupKey(db,key);
+    return lookupKey(db,key,LOOKUP_NONE);
 }
 
 robj *lookupKeyReadOrReply(client *c, robj *key, robj *reply) {
@@ -721,7 +760,7 @@ void typeCommand(client *c) {
     robj *o;
     char *type;
 
-    o = lookupKeyRead(c->db,c->argv[1]);
+    o = lookupKeyReadWithFlags(c->db,c->argv[1],LOOKUP_NOTOUCH);
     if (o == NULL) {
         type = "none";
     } else {
@@ -731,6 +770,10 @@ void typeCommand(client *c) {
         case OBJ_SET: type = "set"; break;
         case OBJ_ZSET: type = "zset"; break;
         case OBJ_HASH: type = "hash"; break;
+        case OBJ_MODULE: {
+            moduleValue *mv = o->ptr;
+            type = mv->type->name;
+        }; break;
         default: type = "unknown"; break;
         }
     }
@@ -1045,7 +1088,7 @@ void ttlGenericCommand(client *c, int output_ms) {
     long long expire, ttl = -1;
 
     /* If the key does not exist at all, return -2 */
-    if (lookupKeyRead(c->db,c->argv[1]) == NULL) {
+    if (lookupKeyReadWithFlags(c->db,c->argv[1],LOOKUP_NOTOUCH) == NULL) {
         addReplyLongLong(c,-2);
         return;
     }
@@ -1087,6 +1130,14 @@ void persistCommand(client *c) {
     }
 }
 
+/* TOUCH key1 [key2 key3 ... keyN] */
+void touchCommand(client *c) {
+    int touched = 0;
+    for (int j = 1; j < c->argc; j++)
+        if (lookupKeyRead(c->db,c->argv[j]) != NULL) touched++;
+    addReplyLongLong(c,touched);
+}
+
 /* -----------------------------------------------------------------------------
  * API to get key arguments from commands
  * ---------------------------------------------------------------------------*/
diff --git a/src/debug.c b/src/debug.c
index 1e179caff..f3e109479 100644
--- a/src/debug.c
+++ b/src/debug.c
@@ -550,7 +550,7 @@ void debugCommand(client *c) {
 
 /* =========================== Crash handling  ============================== */
 
-void _serverAssert(char *estr, char *file, int line) {
+void _serverAssert(const char *estr, const char *file, int line) {
     bugReportStart();
     serverLog(LL_WARNING,"=== ASSERTION FAILED ===");
     serverLog(LL_WARNING,"==> %s:%d '%s' is not true",file,line,estr);
@@ -563,7 +563,7 @@ void _serverAssert(char *estr, char *file, int line) {
     *((char*)-1) = 'x';
 }
 
-void _serverAssertPrintClientInfo(client *c) {
+void _serverAssertPrintClientInfo(const client *c) {
     int j;
 
     bugReportStart();
@@ -587,7 +587,7 @@ void _serverAssertPrintClientInfo(client *c) {
     }
 }
 
-void serverLogObjectDebugInfo(robj *o) {
+void serverLogObjectDebugInfo(const robj *o) {
     serverLog(LL_WARNING,"Object type: %d", o->type);
     serverLog(LL_WARNING,"Object encoding: %d", o->encoding);
     serverLog(LL_WARNING,"Object refcount: %d", o->refcount);
@@ -607,23 +607,23 @@ void serverLogObjectDebugInfo(robj *o) {
     } else if (o->type == OBJ_ZSET) {
         serverLog(LL_WARNING,"Sorted set size: %d", (int) zsetLength(o));
         if (o->encoding == OBJ_ENCODING_SKIPLIST)
-            serverLog(LL_WARNING,"Skiplist level: %d", (int) ((zset*)o->ptr)->zsl->level);
+            serverLog(LL_WARNING,"Skiplist level: %d", (int) ((const zset*)o->ptr)->zsl->level);
     }
 }
 
-void _serverAssertPrintObject(robj *o) {
+void _serverAssertPrintObject(const robj *o) {
     bugReportStart();
     serverLog(LL_WARNING,"=== ASSERTION FAILED OBJECT CONTEXT ===");
     serverLogObjectDebugInfo(o);
 }
 
-void _serverAssertWithInfo(client *c, robj *o, char *estr, char *file, int line) {
+void _serverAssertWithInfo(const client *c, const robj *o, const char *estr, const char *file, int line) {
     if (c) _serverAssertPrintClientInfo(c);
     if (o) _serverAssertPrintObject(o);
     _serverAssert(estr,file,line);
 }
 
-void _serverPanic(char *msg, char *file, int line) {
+void _serverPanic(const char *msg, const char *file, int line) {
     bugReportStart();
     serverLog(LL_WARNING,"------------------------------------------------");
     serverLog(LL_WARNING,"!!! Software Failure. Press left mouse button to continue");
diff --git a/src/geo.c b/src/geo.c
index 2d351d8e0..28cb433dc 100644
--- a/src/geo.c
+++ b/src/geo.c
@@ -156,9 +156,13 @@ double extractDistanceOrReply(client *c, robj **argv,
         return -1;
     }
 
+    if (distance < 0) {
+        addReplyError(c,"radius cannot be negative");
+        return -1;
+    }
+    
     double to_meters = extractUnitOrReply(c,argv[1]);
     if (to_meters < 0) {
-        addReplyError(c,"radius cannot be negative");
         return -1;
     }
 
diff --git a/src/help.h b/src/help.h
index 673b71155..5f927c303 100644
--- a/src/help.h
+++ b/src/help.h
@@ -52,6 +52,11 @@ struct commandHelp {
     "Count set bits in a string",
     1,
     "2.6.0" },
+    { "BITFIELD",
+    "key [GET type offset] [SET type offset value] [INCRBY type offset increment] [OVERFLOW WRAP|SAT|FAIL]",
+    "Perform arbitrary bitfield integer operations on strings",
+    1,
+    "3.2.0" },
     { "BITOP",
     "operation destkey key [key ...]",
     "Perform bitwise operations between strings",
@@ -326,32 +331,32 @@ struct commandHelp {
     "key longitude latitude member [longitude latitude member ...]",
     "Add one or more geospatial items in the geospatial index represented using a sorted set",
     13,
-    "" },
+    "3.2.0" },
     { "GEODIST",
     "key member1 member2 [unit]",
     "Returns the distance between two members of a geospatial index",
     13,
-    "" },
+    "3.2.0" },
     { "GEOHASH",
     "key member [member ...]",
     "Returns members of a geospatial index as standard geohash strings",
     13,
-    "" },
+    "3.2.0" },
     { "GEOPOS",
     "key member [member ...]",
     "Returns longitude and latitude of members of a geospatial index",
     13,
-    "" },
+    "3.2.0" },
     { "GEORADIUS",
-    "key longitude latitude radius m|km|ft|mi [WITHCOORD] [WITHDIST] [WITHHASH] [COUNT count] [ASC|DESC]",
+    "key longitude latitude radius m|km|ft|mi [WITHCOORD] [WITHDIST] [WITHHASH] [COUNT count] [ASC|DESC] [STORE key] [STOREDIST key]",
     "Query a sorted set representing a geospatial index to fetch members matching a given maximum distance from a point",
     13,
-    "" },
+    "3.2.0" },
     { "GEORADIUSBYMEMBER",
-    "key member radius m|km|ft|mi [WITHCOORD] [WITHDIST] [WITHHASH] [COUNT count] [ASC|DESC]",
+    "key member radius m|km|ft|mi [WITHCOORD] [WITHDIST] [WITHHASH] [COUNT count] [ASC|DESC] [STORE key] [STOREDIST key]",
     "Query a sorted set representing a geospatial index to fetch members matching a given maximum distance from a member",
     13,
-    "" },
+    "3.2.0" },
     { "GET",
     "key",
     "Get the value of a key",
diff --git a/src/intset.c b/src/intset.c
index b0a597fc7..30ea85344 100644
--- a/src/intset.c
+++ b/src/intset.c
@@ -272,7 +272,7 @@ uint8_t intsetGet(intset *is, uint32_t pos, int64_t *value) {
 }
 
 /* Return intset length */
-uint32_t intsetLen(intset *is) {
+uint32_t intsetLen(const intset *is) {
     return intrev32ifbe(is->length);
 }
 
diff --git a/src/intset.h b/src/intset.h
index 30a854f89..8119e6636 100644
--- a/src/intset.h
+++ b/src/intset.h
@@ -44,7 +44,7 @@ intset *intsetRemove(intset *is, int64_t value, int *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);
+uint32_t intsetLen(const intset *is);
 size_t intsetBlobLen(intset *is);
 
 #ifdef REDIS_TEST
diff --git a/src/module.c b/src/module.c
index 72d6aeb0f..e3603e1d7 100644
--- a/src/module.c
+++ b/src/module.c
@@ -17,6 +17,7 @@ struct RedisModule {
     char *name;     /* Module name. */
     int ver;        /* Module version. We use just progressive integers. */
     int apiver;     /* Module API version as requested during initialization.*/
+    list *types;    /* Module data types. */
 };
 typedef struct RedisModule RedisModule;
 
@@ -35,6 +36,29 @@ struct AutoMemEntry {
 #define REDISMODULE_AM_REPLY 2
 #define REDISMODULE_AM_FREED 3 /* Explicitly freed by user already. */
 
+/* The pool allocator block. Redis Modules can allocate memory via this special
+ * allocator that will automatically release it all once the callback returns.
+ * This means that it can only be used for ephemeral allocations. However
+ * there are two advantages for modules to use this API:
+ *
+ * 1) The memory is automatically released when the callback returns.
+ * 2) This allocator is faster for many small allocations since whole blocks
+ *    are allocated, and small pieces returned to the caller just advancing
+ *    the index of the allocation.
+ *
+ * Allocations are always rounded to the size of the void pointer in order
+ * to always return aligned memory chunks. */
+
+#define REDISMODULE_POOL_ALLOC_MIN_SIZE (1024*8)
+#define REDISMODULE_POOL_ALLOC_ALIGN (sizeof(void*))
+
+typedef struct RedisModulePoolAllocBlock {
+    uint32_t size;
+    uint32_t used;
+    struct RedisModulePoolAllocBlock *next;
+    char memory[];
+} RedisModulePoolAllocBlock;
+
 /* This structure represents the context in which Redis modules operate.
  * Most APIs module can access, get a pointer to the context, so that the API
  * implementation can hold state across calls, or remember what to free after
@@ -56,10 +80,12 @@ struct RedisModuleCtx {
     /* Used if there is the REDISMODULE_CTX_KEYS_POS_REQUEST flag set. */
     int *keys_pos;
     int keys_count;
+
+    struct RedisModulePoolAllocBlock *pa_head;
 };
 typedef struct RedisModuleCtx RedisModuleCtx;
 
-#define REDISMODULE_CTX_INIT {(void*)(unsigned long)&RM_GetApi, NULL, NULL, NULL, 0, 0, 0, NULL, 0, NULL, 0}
+#define REDISMODULE_CTX_INIT {(void*)(unsigned long)&RM_GetApi, NULL, NULL, NULL, 0, 0, 0, NULL, 0, NULL, 0, NULL}
 #define REDISMODULE_CTX_MULTI_EMITTED (1<<0)
 #define REDISMODULE_CTX_AUTO_MEMORY (1<<1)
 #define REDISMODULE_CTX_KEYS_POS_REQUEST (1<<2)
@@ -111,7 +137,7 @@ typedef struct RedisModuleCommandProxy RedisModuleCommandProxy;
 /* Reply of RM_Call() function. The function is filled in a lazy
  * way depending on the function called on the reply structure. By default
  * only the type, proto and protolen are filled. */
-struct RedisModuleCallReply {
+typedef struct RedisModuleCallReply {
     RedisModuleCtx *ctx;
     int type;       /* REDISMODULE_REPLY_... */
     int flags;      /* REDISMODULE_REPLYFLAG_...  */
@@ -126,8 +152,7 @@ struct RedisModuleCallReply {
         long long ll;    /* Reply value for integer reply. */
         struct RedisModuleCallReply *array; /* Array of sub-reply elements. */
     } val;
-};
-typedef struct RedisModuleCallReply RedisModuleCallReply;
+} RedisModuleCallReply;
 
 /* --------------------------------------------------------------------------
  * Prototypes
@@ -138,7 +163,103 @@ void RM_CloseKey(RedisModuleKey *key);
 void autoMemoryCollect(RedisModuleCtx *ctx);
 robj **moduleCreateArgvFromUserFormat(const char *cmdname, const char *fmt, int *argcp, int *flags, va_list ap);
 void moduleReplicateMultiIfNeeded(RedisModuleCtx *ctx);
-void RM_ZsetRangeStop(RedisModuleKey *key);
+void RM_ZsetRangeStop(RedisModuleKey *kp);
+static void zsetKeyReset(RedisModuleKey *key);
+
+/* --------------------------------------------------------------------------
+ * Heap allocation raw functions
+ * -------------------------------------------------------------------------- */
+
+/* Use like malloc(). Memory allocated with this function is reported in
+ * Redis INFO memory, used for keys eviction according to maxmemory settings
+ * and in general is taken into account as memory allocated by Redis.
+ * You should avoid using malloc(). */
+void *RM_Alloc(size_t bytes) {
+    return zmalloc(bytes);
+}
+
+/* Use like calloc(). Memory allocated with this function is reported in
+ * Redis INFO memory, used for keys eviction according to maxmemory settings
+ * and in general is taken into account as memory allocated by Redis.
+ * You should avoid using calloc() directly. */
+void *RM_Calloc(size_t nmemb, size_t size) {
+    return zcalloc(nmemb*size);
+}
+
+/* Use like realloc() for memory obtained with RedisModule_Alloc(). */
+void* RM_Realloc(void *ptr, size_t bytes) {
+    return zrealloc(ptr,bytes);
+}
+
+/* Use like free() for memory obtained by RedisModule_Alloc() and
+ * RedisModule_Realloc(). However you should never try to free with
+ * RedisModule_Free() memory allocated with malloc() inside your module. */
+void RM_Free(void *ptr) {
+    zfree(ptr);
+}
+
+/* Like strdup() but returns memory allocated with RedisModule_Alloc(). */
+char *RM_Strdup(const char *str) {
+    return zstrdup(str);
+}
+
+/* --------------------------------------------------------------------------
+ * Pool allocator
+ * -------------------------------------------------------------------------- */
+
+/* Release the chain of blocks used for pool allocations. */
+void poolAllocRelease(RedisModuleCtx *ctx) {
+    RedisModulePoolAllocBlock *head = ctx->pa_head, *next;
+
+    while(head != NULL) {
+        next = head->next;
+        zfree(head);
+        head = next;
+    }
+    ctx->pa_head = NULL;
+}
+
+/* Return heap allocated memory that will be freed automatically when the
+ * module callback function returns. Mostly suitable for small allocations
+ * that are short living and must be released when the callback returns
+ * anyway. The returned memory is aligned to the architecture word size
+ * if at least word size bytes are requested, otherwise it is just
+ * aligned to the next power of two, so for example a 3 bytes request is
+ * 4 bytes aligned while a 2 bytes request is 2 bytes aligned.
+ *
+ * There is no realloc style function since when this is needed to use the
+ * pool allocator is not a good idea.
+ *
+ * The function returns NULL if `bytes` is 0. */
+void *RM_PoolAlloc(RedisModuleCtx *ctx, size_t bytes) {
+    if (bytes == 0) return NULL;
+    RedisModulePoolAllocBlock *b = ctx->pa_head;
+    size_t left = b ? b->size - b->used : 0;
+
+    /* Fix alignment. */
+    if (left >= bytes) {
+        size_t alignment = REDISMODULE_POOL_ALLOC_ALIGN;
+        while (bytes < alignment && alignment/2 >= bytes) alignment /= 2;
+        if (b->used % alignment)
+            b->used += alignment - (b->used % alignment);
+        left = (b->used > b->size) ? 0 : b->size - b->used;
+    }
+
+    /* Create a new block if needed. */
+    if (left < bytes) {
+        size_t blocksize = REDISMODULE_POOL_ALLOC_MIN_SIZE;
+        if (blocksize < bytes) blocksize = bytes;
+        b = zmalloc(sizeof(*b) + blocksize);
+        b->size = blocksize;
+        b->used = 0;
+        b->next = ctx->pa_head;
+        ctx->pa_head = b;
+    }
+
+    char *retval = b->memory + b->used;
+    b->used += bytes;
+    return retval;
+}
 
 /* --------------------------------------------------------------------------
  * Helpers for modules API implementation
@@ -240,6 +361,7 @@ int RM_GetApi(const char *funcname, void **targetPtrPtr) {
 /* Free the context after the user function was called. */
 void moduleFreeContext(RedisModuleCtx *ctx) {
     autoMemoryCollect(ctx);
+    poolAllocRelease(ctx);
     if (ctx->postponed_arrays) {
         zfree(ctx->postponed_arrays);
         ctx->postponed_arrays_count = 0;
@@ -463,6 +585,7 @@ void RM_SetModuleAttribs(RedisModuleCtx *ctx, const char *name, int ver, int api
     module->name = sdsnew((char*)name);
     module->ver = ver;
     module->apiver = apiver;
+    module->types = listCreate();
     ctx->module = module;
 }
 
@@ -496,15 +619,28 @@ void autoMemoryAdd(RedisModuleCtx *ctx, int type, void *ptr) {
 void autoMemoryFreed(RedisModuleCtx *ctx, int type, void *ptr) {
     if (!(ctx->flags & REDISMODULE_CTX_AUTO_MEMORY)) return;
 
-    int j;
-    for (j = 0; j < ctx->amqueue_used; j++) {
-        if (ctx->amqueue[j].type == type &&
-            ctx->amqueue[j].ptr == ptr)
-        {
-            ctx->amqueue[j].type = REDISMODULE_AM_FREED;
-            /* Optimization: if this is the last element, we can
-             * reuse it. */
-            if (j == ctx->amqueue_used-1) ctx->amqueue_used--;
+    int count = (ctx->amqueue_used+1)/2;
+    for (int j = 0; j < count; j++) {
+        for (int side = 0; side < 2; side++) {
+            /* For side = 0 check right side of the array, for
+             * side = 1 check the left side instead (zig-zag scanning). */
+            int i = (side == 0) ? (ctx->amqueue_used - 1 - j) : j;
+            if (ctx->amqueue[i].type == type &&
+                ctx->amqueue[i].ptr == ptr)
+            {
+                ctx->amqueue[i].type = REDISMODULE_AM_FREED;
+
+                /* Switch the freed element and the last element, to avoid growing
+                 * the queue unnecessarily if we allocate/free in a loop */
+                if (i != ctx->amqueue_used-1) {
+                    ctx->amqueue[i] = ctx->amqueue[ctx->amqueue_used-1];
+                }
+
+                /* Reduce the size of the queue because we either moved the top
+                 * element elsewhere or freed it */
+                ctx->amqueue_used--;
+                return;
+            }
         }
     }
 }
@@ -559,6 +695,17 @@ RedisModuleString *RM_CreateStringFromLongLong(RedisModuleCtx *ctx, long long ll
     return RM_CreateString(ctx,buf,len);
 }
 
+/* Like RedisModule_CreatString(), but creates a string starting from another
+ * RedisModuleString.
+ *
+ * The returned string must be released with RedisModule_FreeString() or by
+ * enabling automatic memory management. */
+RedisModuleString *RM_CreateStringFromString(RedisModuleCtx *ctx, const RedisModuleString *str) {
+    RedisModuleString *o = dupStringObject(str);
+    autoMemoryAdd(ctx,REDISMODULE_AM_STRING,o);
+    return o;
+}
+
 /* Free a module string object obtained with one of the Redis modules API calls
  * that return new string objects.
  *
@@ -573,7 +720,7 @@ void RM_FreeString(RedisModuleCtx *ctx, RedisModuleString *str) {
 /* Given a string module object, this function returns the string pointer
  * and length of the string. The returned pointer and length should only
  * be used for read only accesses and never modified. */
-const char *RM_StringPtrLen(RedisModuleString *str, size_t *len) {
+const char *RM_StringPtrLen(const RedisModuleString *str, size_t *len) {
     if (len) *len = sdslen(str->ptr);
     return str->ptr;
 }
@@ -582,7 +729,7 @@ const char *RM_StringPtrLen(RedisModuleString *str, size_t *len) {
  * Returns REDISMODULE_OK on success. If the string can't be parsed
  * as a valid, strict long long (no spaces before/after), REDISMODULE_ERR
  * is returned. */
-int RM_StringToLongLong(RedisModuleString *str, long long *ll) {
+int RM_StringToLongLong(const RedisModuleString *str, long long *ll) {
     return string2ll(str->ptr,sdslen(str->ptr),ll) ? REDISMODULE_OK :
                                                      REDISMODULE_ERR;
 }
@@ -590,7 +737,7 @@ int RM_StringToLongLong(RedisModuleString *str, long long *ll) {
 /* Convert the string into a double, storing it at `*d`.
  * Returns REDISMODULE_OK on success or REDISMODULE_ERR if the string is
  * not a valid string representation of a double value. */
-int RM_StringToDouble(RedisModuleString *str, double *d) {
+int RM_StringToDouble(const RedisModuleString *str, double *d) {
     int retval = getDoubleFromObject(str,d);
     return (retval == C_OK) ? REDISMODULE_OK : REDISMODULE_ERR;
 }
@@ -933,7 +1080,7 @@ void *RM_OpenKey(RedisModuleCtx *ctx, robj *keyname, int mode) {
     kp->value = value;
     kp->iter = NULL;
     kp->mode = mode;
-    RM_ZsetRangeStop(kp);
+    zsetKeyReset(kp);
     autoMemoryAdd(ctx,REDISMODULE_AM_KEY,kp);
     return (void*)kp;
 }
@@ -961,6 +1108,7 @@ int RM_KeyType(RedisModuleKey *key) {
     case OBJ_SET: return REDISMODULE_KEYTYPE_SET;
     case OBJ_ZSET: return REDISMODULE_KEYTYPE_ZSET;
     case OBJ_HASH: return REDISMODULE_KEYTYPE_HASH;
+    case OBJ_MODULE: return REDISMODULE_KEYTYPE_MODULE;
     default: return 0;
     }
 }
@@ -1318,6 +1466,12 @@ int RM_ZsetScore(RedisModuleKey *key, RedisModuleString *ele, double *score) {
  * Key API for Sorted Set iterator
  * -------------------------------------------------------------------------- */
 
+void zsetKeyReset(RedisModuleKey *key) {
+    key->ztype = REDISMODULE_ZSET_RANGE_NONE;
+    key->zcurrent = NULL;
+    key->zer = 1;
+}
+
 /* Stop a sorted set iteration. */
 void RM_ZsetRangeStop(RedisModuleKey *key) {
     /* Free resources if needed. */
@@ -1326,9 +1480,7 @@ void RM_ZsetRangeStop(RedisModuleKey *key) {
     /* Setup sensible values so that misused iteration API calls when an
      * iterator is not active will result into something more sensible
      * than crashing. */
-    key->ztype = REDISMODULE_ZSET_RANGE_NONE;
-    key->zcurrent = NULL;
-    key->zer = 1;
+    zsetKeyReset(key);
 }
 
 /* Return the "End of range" flag value to signal the end of the iteration. */
@@ -1705,17 +1857,20 @@ int RM_HashSet(RedisModuleKey *key, int flags, ...) {
             continue;
         }
 
+        int low_flags = HASH_SET_COPY;
         /* If CFIELDS is active, we can pass the ownership of the
          * SDS object to the low level function that sets the field
          * to avoid a useless copy. */
-        int low_flags = HASH_SET_COPY;
         if (flags & REDISMODULE_HASH_CFIELDS)
             low_flags |= HASH_SET_TAKE_FIELD;
         updated += hashTypeSet(key->value, field->ptr, value->ptr, low_flags);
-        field->ptr = NULL; /* Ownership is now of hashTypeSet() */
 
-        /* Cleanup */
-        if (flags & REDISMODULE_HASH_CFIELDS) decrRefCount(field);
+        /* If CFIELDS is active, SDS string ownership is now of hashTypeSet(),
+         * however we still have to release the 'field' object shell. */
+        if (flags & REDISMODULE_HASH_CFIELDS) {
+           field->ptr = NULL; /* Prevent the SDS string from being freed. */
+           decrRefCount(field);
+        }
     }
     va_end(ap);
     moduleDelKeyIfEmpty(key);
@@ -2195,6 +2350,490 @@ const char *RM_CallReplyProto(RedisModuleCallReply *reply, size_t *len) {
 }
 
 /* --------------------------------------------------------------------------
+ * Modules data types
+ *
+ * When String DMA or using existing data structures is not enough, it is
+ * possible to create new data types from scratch and export them to
+ * Redis. The module must provide a set of callbacks for handling the
+ * new values exported (for example in order to provide RDB saving/loading,
+ * AOF rewrite, and so forth). In this section we define this API.
+ * -------------------------------------------------------------------------- */
+
+/* Turn a 9 chars name in the specified charset and a 10 bit encver into
+ * a single 64 bit unsigned integer that represents this exact module name
+ * and version. This final number is called a "type ID" and is used when
+ * writing module exported values to RDB files, in order to re-associate the
+ * value to the right module to load them during RDB loading.
+ *
+ * If the string is not of the right length or the charset is wrong, or
+ * if encver is outside the unsigned 10 bit integer range, 0 is returned,
+ * otherwise the function returns the right type ID.
+ *
+ * The resulting 64 bit integer is composed as follows:
+ *
+ * (high order bits) 6|6|6|6|6|6|6|6|6|10 (low order bits)
+ *
+ * The first 6 bits value is the first character, name[0], while the last
+ * 6 bits value, immediately before the 10 bits integer, is name[8].
+ * The last 10 bits are the encoding version.
+ *
+ * Note that a name and encver combo of "AAAAAAAAA" and 0, will produce
+ * zero as return value, that is the same we use to signal errors, thus
+ * this combination is invalid, and also useless since type names should
+ * try to be vary to avoid collisions. */
+
+const char *ModuleTypeNameCharSet =
+             "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
+             "abcdefghijklmnopqrstuvwxyz"
+             "0123456789-_";
+
+uint64_t moduleTypeEncodeId(const char *name, int encver) {
+    /* We use 64 symbols so that we can map each character into 6 bits
+     * of the final output. */
+    const char *cset = ModuleTypeNameCharSet;
+    if (strlen(name) != 9) return 0;
+    if (encver < 0 || encver > 1023) return 0;
+
+    uint64_t id = 0;
+    for (int j = 0; j < 9; j++) {
+        char *p = strchr(cset,name[j]);
+        if (!p) return 0;
+        unsigned long pos = p-cset;
+        id = (id << 6) | pos;
+    }
+    id = (id << 10) | encver;
+    return id;
+}
+
+/* Search, in the list of exported data types of all the modules registered,
+ * a type with the same name as the one given. Returns the moduleType
+ * structure pointer if such a module is found, or NULL otherwise. */
+moduleType *moduleTypeLookupModuleByName(const char *name) {
+    dictIterator *di = dictGetIterator(modules);
+    dictEntry *de;
+
+    while ((de = dictNext(di)) != NULL) {
+        struct RedisModule *module = dictGetVal(de);
+        listIter li;
+        listNode *ln;
+
+        listRewind(module->types,&li);
+        while((ln = listNext(&li))) {
+            moduleType *mt = ln->value;
+            if (memcmp(name,mt->name,sizeof(mt->name)) == 0) {
+                dictReleaseIterator(di);
+                return mt;
+            }
+        }
+    }
+    dictReleaseIterator(di);
+    return NULL;
+}
+
+/* Lookup a module by ID, with caching. This function is used during RDB
+ * loading. Modules exporting data types should never be able to unload, so
+ * our cache does not need to expire. */
+#define MODULE_LOOKUP_CACHE_SIZE 3
+
+moduleType *moduleTypeLookupModuleByID(uint64_t id) {
+    static struct {
+        uint64_t id;
+        moduleType *mt;
+    } cache[MODULE_LOOKUP_CACHE_SIZE];
+
+    /* Search in cache to start. */
+    int j;
+    for (j = 0; j < MODULE_LOOKUP_CACHE_SIZE; j++)
+        if (cache[j].id == id) return cache[j].mt;
+
+    /* Slow module by module lookup. */
+    moduleType *mt = NULL;
+    dictIterator *di = dictGetIterator(modules);
+    dictEntry *de;
+
+    while ((de = dictNext(di)) != NULL) {
+        struct RedisModule *module = dictGetVal(de);
+        listIter li;
+        listNode *ln;
+
+        listRewind(module->types,&li);
+        while((ln = listNext(&li))) {
+            mt = ln->value;
+            /* Compare only the 54 bit module identifier and not the
+             * encoding version. */
+            if (mt->id >> 10 == id >> 10) break;
+        }
+    }
+    dictReleaseIterator(di);
+
+    /* Add to cache if possible. */
+    if (mt && j < MODULE_LOOKUP_CACHE_SIZE) {
+        cache[j].id = id;
+        cache[j].mt = mt;
+    }
+    return mt;
+}
+
+/* Turn an (unresolved) module ID into a type name, to show the user an
+ * error when RDB files contain module data we can't load. */
+void moduleTypeNameByID(char *name, uint64_t moduleid) {
+    const char *cset = ModuleTypeNameCharSet;
+
+    name[0] = '\0';
+    char *p = name+8;
+    moduleid >>= 10;
+    for (int j = 0; j < 9; j++) {
+        *p-- = cset[moduleid & 63];
+        moduleid >>= 6;
+    }
+}
+
+/* Register a new data type exported by the module. The parameters are the
+ * following. Please for in depth documentation check the modules API
+ * documentation, especially the INTRO.md file.
+ *
+ * * **name**: A 9 characters data type name that MUST be unique in the Redis
+ *   Modules ecosystem. Be creative... and there will be no collisions. Use
+ *   the charset A-Z a-z 9-0, plus the two "-_" characters. A good
+ *   idea is to use, for example `<typename>-<vendor>`. For example
+ *   "tree-AntZ" may mean "Tree data structure by @antirez". To use both
+ *   lower case and upper case letters helps in order to prevent collisions.
+ * * **encver**: Encoding version, which is, the version of the serialization
+ *   that a module used in order to persist data. As long as the "name"
+ *   matches, the RDB loading will be dispatched to the type callbacks
+ *   whatever 'encver' is used, however the module can understand if
+ *   the encoding it must load are of an older version of the module.
+ *   For example the module "tree-AntZ" initially used encver=0. Later
+ *   after an upgrade, it started to serialize data in a different format
+ *   and to register the type with encver=1. However this module may
+ *   still load old data produced by an older version if the rdb_load
+ *   callback is able to check the encver value and act accordingly.
+ *   The encver must be a positive value between 0 and 1023.
+ * * **rdb_load**: A callback function pointer that loads data from RDB files.
+ * * **rdb_save**: A callback function pointer that saves data to RDB files.
+ * * **aof_rewrite**: A callback function pointer that rewrites data as commands.
+ * * **digest**: A callback function pointer that is used for `DEBUG DIGEST`.
+ * * **free**: A callback function pointer that can free a type value.
+ *
+ * Note: the module name "AAAAAAAAA" is reserved and produces an error, it
+ * happens to be pretty lame as well.
+ *
+ * If there is already a module registering a type with the same name,
+ * and if the module name or encver is invalid, NULL is returned.
+ * Otherwise the new type is registered into Redis, and a reference of
+ * type RedisModuleType is returned: the caller of the function should store
+ * this reference into a gobal variable to make future use of it in the
+ * modules type API, since a single module may register multiple types.
+ * Example code fragment:
+ *
+ *      static RedisModuleType *BalancedTreeType;
+ *
+ *      int RedisModule_OnLoad(RedisModuleCtx *ctx) {
+ *          // some code here ...
+ *          BalancedTreeType = RM_CreateDataType(...);
+ *      }
+ */
+moduleType *RM_CreateDataType(RedisModuleCtx *ctx, const char *name, int encver, moduleTypeLoadFunc rdb_load, moduleTypeSaveFunc rdb_save, moduleTypeRewriteFunc aof_rewrite, moduleTypeDigestFunc digest, moduleTypeFreeFunc free) {
+    uint64_t id = moduleTypeEncodeId(name,encver);
+    if (id == 0) return NULL;
+    if (moduleTypeLookupModuleByName(name) != NULL) return NULL;
+
+    moduleType *mt = zmalloc(sizeof(*mt));
+    mt->id = id;
+    mt->module = ctx->module;
+    mt->rdb_load = rdb_load;
+    mt->rdb_save = rdb_save;
+    mt->aof_rewrite = aof_rewrite;
+    mt->digest = digest;
+    mt->free = free;
+    memcpy(mt->name,name,sizeof(mt->name));
+    listAddNodeTail(ctx->module->types,mt);
+    return mt;
+}
+
+/* If the key is open for writing, set the specified module type object
+ * as the value of the key, deleting the old value if any.
+ * On success REDISMODULE_OK is returned. If the key is not open for
+ * writing or there is an active iterator, REDISMODULE_ERR is returned. */
+int RM_ModuleTypeSetValue(RedisModuleKey *key, moduleType *mt, void *value) {
+    if (!(key->mode & REDISMODULE_WRITE) || key->iter) return REDISMODULE_ERR;
+    RM_DeleteKey(key);
+    robj *o = createModuleObject(mt,value);
+    setKey(key->db,key->key,o);
+    decrRefCount(o);
+    key->value = o;
+    return REDISMODULE_OK;
+}
+
+/* Assuming RedisModule_KeyType() returned REDISMODULE_KEYTYPE_MODULE on
+ * the key, returns the moduel type pointer of the value stored at key.
+ *
+ * If the key is NULL, is not associated with a module type, or is empty,
+ * then NULL is returned instead. */
+moduleType *RM_ModuleTypeGetType(RedisModuleKey *key) {
+    if (key == NULL ||
+        key->value == NULL ||
+        RM_KeyType(key) != REDISMODULE_KEYTYPE_MODULE) return NULL;
+    moduleValue *mv = key->value->ptr;
+    return mv->type;
+}
+
+/* Assuming RedisModule_KeyType() returned REDISMODULE_KEYTYPE_MODULE on
+ * the key, returns the module type low-level value stored at key, as
+ * it was set by the user via RedisModule_ModuleTypeSet().
+ *
+ * If the key is NULL, is not associated with a module type, or is empty,
+ * then NULL is returned instead. */
+void *RM_ModuleTypeGetValue(RedisModuleKey *key) {
+    if (key == NULL ||
+        key->value == NULL ||
+        RM_KeyType(key) != REDISMODULE_KEYTYPE_MODULE) return NULL;
+    moduleValue *mv = key->value->ptr;
+    return mv->value;
+}
+
+/* --------------------------------------------------------------------------
+ * RDB loading and saving functions
+ * -------------------------------------------------------------------------- */
+
+/* Called when there is a load error in the context of a module. This cannot
+ * be recovered like for the built-in types. */
+void moduleRDBLoadError(RedisModuleIO *io) {
+    serverLog(LL_WARNING,
+        "Error loading data from RDB (short read or EOF). "
+        "Read performed by module '%s' about type '%s' "
+        "after reading '%llu' bytes of a value.",
+        io->type->module->name,
+        io->type->name,
+        (unsigned long long)io->bytes);
+    exit(1);
+}
+
+/* Save an unsigned 64 bit value into the RDB file. This function should only
+ * be called in the context of the rdb_save method of modules implementing new
+ * data types. */
+void RM_SaveUnsigned(RedisModuleIO *io, uint64_t value) {
+    if (io->error) return;
+    int retval = rdbSaveLen(io->rio, value);
+    if (retval == -1) {
+        io->error = 1;
+    } else {
+        io->bytes += retval;
+    }
+}
+
+/* Load an unsigned 64 bit value from the RDB file. This function should only
+ * be called in the context of the rdb_load method of modules implementing
+ * new data types. */
+uint64_t RM_LoadUnsigned(RedisModuleIO *io) {
+    uint64_t value;
+    int retval = rdbLoadLenByRef(io->rio, NULL, &value);
+    if (retval == -1) {
+        moduleRDBLoadError(io);
+        return 0; /* Never reached. */
+    }
+    return value;
+}
+
+/* Like RedisModule_SaveUnsigned() but for signed 64 bit values. */
+void RM_SaveSigned(RedisModuleIO *io, int64_t value) {
+    union {uint64_t u; int64_t i;} conv;
+    conv.i = value;
+    RM_SaveUnsigned(io,conv.u);
+}
+
+/* Like RedisModule_LoadUnsigned() but for signed 64 bit values. */
+int64_t RM_LoadSigned(RedisModuleIO *io) {
+    union {uint64_t u; int64_t i;} conv;
+    conv.u = RM_LoadUnsigned(io);
+    return conv.i;
+}
+
+/* In the context of the rdb_save method of a module type, saves a
+ * string into the RDB file taking as input a RedisModuleString.
+ *
+ * The string can be later loaded with RedisModule_LoadString() or
+ * other Load family functions expecting a serialized string inside
+ * the RDB file. */
+void RM_SaveString(RedisModuleIO *io, RedisModuleString *s) {
+    if (io->error) return;
+    int retval = rdbSaveStringObject(io->rio,s);
+    if (retval == -1) {
+        io->error = 1;
+    } else {
+        io->bytes += retval;
+    }
+}
+
+/* Like RedisModule_SaveString() but takes a raw C pointer and length
+ * as input. */
+void RM_SaveStringBuffer(RedisModuleIO *io, const char *str, size_t len) {
+    if (io->error) return;
+    int retval = rdbSaveRawString(io->rio,(unsigned char*)str,len);
+    if (retval == -1) {
+        io->error = 1;
+    } else {
+        io->bytes += retval;
+    }
+}
+
+/* Implements RM_LoadString() and RM_LoadStringBuffer() */
+void *moduleLoadString(RedisModuleIO *io, int plain, size_t *lenptr) {
+    void *s = rdbGenericLoadStringObject(io->rio,
+              plain ? RDB_LOAD_PLAIN : RDB_LOAD_NONE, lenptr);
+    if (s == NULL) {
+        moduleRDBLoadError(io);
+        return NULL; /* Never reached. */
+    }
+    return s;
+}
+
+/* In the context of the rdb_load method of a module data type, loads a string
+ * from the RDB file, that was previously saved with RedisModule_SaveString()
+ * functions family.
+ *
+ * The returned string is a newly allocated RedisModuleString object, and
+ * the user should at some point free it with a call to RedisModule_FreeString().
+ *
+ * If the data structure does not store strings as RedisModuleString objects,
+ * the similar function RedisModule_LoadStringBuffer() could be used instead. */
+RedisModuleString *RM_LoadString(RedisModuleIO *io) {
+    return moduleLoadString(io,0,NULL);
+}
+
+/* Like RedisModule_LoadString() but returns an heap allocated string that
+ * was allocated with RedisModule_Alloc(), and can be resized or freed with
+ * RedisModule_Realloc() or RedisModule_Free().
+ *
+ * The size of the string is stored at '*lenptr' if not NULL.
+ * The returned string is not automatically NULL termianted, it is loaded
+ * exactly as it was stored inisde the RDB file. */
+char *RM_LoadStringBuffer(RedisModuleIO *io, size_t *lenptr) {
+    return moduleLoadString(io,1,lenptr);
+}
+
+/* In the context of the rdb_save method of a module data type, saves a double
+ * value to the RDB file. The double can be a valid number, a NaN or infinity.
+ * It is possible to load back the value with RedisModule_LoadDouble(). */
+void RM_SaveDouble(RedisModuleIO *io, double value) {
+    if (io->error) return;
+    int retval = rdbSaveBinaryDoubleValue(io->rio, value);
+    if (retval == -1) {
+        io->error = 1;
+    } else {
+        io->bytes += retval;
+    }
+}
+
+/* In the context of the rdb_save method of a module data type, loads back the
+ * double value saved by RedisModule_SaveDouble(). */
+double RM_LoadDouble(RedisModuleIO *io) {
+    double value;
+    int retval = rdbLoadBinaryDoubleValue(io->rio, &value);
+    if (retval == -1) {
+        moduleRDBLoadError(io);
+        return 0; /* Never reached. */
+    }
+    return value;
+}
+
+/* --------------------------------------------------------------------------
+ * AOF API for modules data types
+ * -------------------------------------------------------------------------- */
+
+/* Emits a command into the AOF during the AOF rewriting process. This function
+ * is only called in the context of the aof_rewrite method of data types exported
+ * by a module. The command works exactly like RedisModule_Call() in the way
+ * the parameters are passed, but it does not return anything as the error
+ * handling is performed by Redis itself. */
+void RM_EmitAOF(RedisModuleIO *io, const char *cmdname, const char *fmt, ...) {
+    if (io->error) return;
+    struct redisCommand *cmd;
+    robj **argv = NULL;
+    int argc = 0, flags = 0, j;
+    va_list ap;
+
+    cmd = lookupCommandByCString((char*)cmdname);
+    if (!cmd) {
+        serverLog(LL_WARNING,
+            "Fatal: AOF method for module data type '%s' tried to "
+            "emit unknown command '%s'",
+            io->type->name, cmdname);
+        io->error = 1;
+        errno = EINVAL;
+        return;
+    }
+
+    /* Emit the arguments into the AOF in Redis protocol format. */
+    va_start(ap, fmt);
+    argv = moduleCreateArgvFromUserFormat(cmdname,fmt,&argc,&flags,ap);
+    va_end(ap);
+    if (argv == NULL) {
+        serverLog(LL_WARNING,
+            "Fatal: AOF method for module data type '%s' tried to "
+            "call RedisModule_EmitAOF() with wrong format specifiers '%s'",
+            io->type->name, fmt);
+        io->error = 1;
+        errno = EINVAL;
+        return;
+    }
+
+    /* Bulk count. */
+    if (!io->error && rioWriteBulkCount(io->rio,'*',argc) == 0)
+        io->error = 1;
+
+    /* Arguments. */
+    for (j = 0; j < argc; j++) {
+        if (!io->error && rioWriteBulkObject(io->rio,argv[j]) == 0)
+            io->error = 1;
+        decrRefCount(argv[j]);
+    }
+    zfree(argv);
+    return;
+}
+
+/* --------------------------------------------------------------------------
+ * Logging
+ * -------------------------------------------------------------------------- */
+
+/* Produces a log message to the standard Redis log, the format accepts
+ * printf-alike specifiers, while level is a string describing the log
+ * level to use when emitting the log, and must be one of the following:
+ *
+ * * "debug"
+ * * "verbose"
+ * * "notice"
+ * * "warning"
+ *
+ * If the specified log level is invalid, verbose is used by default.
+ * There is a fixed limit to the length of the log line this function is able
+ * to emit, this limti is not specified but is guaranteed to be more than
+ * a few lines of text.
+ */
+void RM_Log(RedisModuleCtx *ctx, const char *levelstr, const char *fmt, ...) {
+    va_list ap;
+    char msg[LOG_MAX_LEN];
+    size_t name_len;
+    int level;
+
+    if (!ctx->module) return;   /* Can only log if module is initialized */
+
+    if (!strcasecmp(levelstr,"debug")) level = LL_DEBUG;
+    else if (!strcasecmp(levelstr,"verbose")) level = LL_VERBOSE;
+    else if (!strcasecmp(levelstr,"notice")) level = LL_NOTICE;
+    else if (!strcasecmp(levelstr,"warning")) level = LL_WARNING;
+    else level = LL_VERBOSE; /* Default. */
+
+    name_len = snprintf(msg, sizeof(msg),"<%s> ", ctx->module->name);
+
+    va_start(ap, fmt);
+    vsnprintf(msg + name_len, sizeof(msg) - name_len, fmt, ap);
+    va_end(ap);
+
+    serverLogRaw(level,msg);
+}
+
+/* --------------------------------------------------------------------------
  * Modules API internals
  * -------------------------------------------------------------------------- */
 
@@ -2229,6 +2868,11 @@ int moduleRegisterApi(const char *funcname, void *funcptr) {
 /* Register all the APIs we export. */
 void moduleRegisterCoreAPI(void) {
     server.moduleapi = dictCreate(&moduleAPIDictType,NULL);
+    REGISTER_API(Alloc);
+    REGISTER_API(Calloc);
+    REGISTER_API(Realloc);
+    REGISTER_API(Free);
+    REGISTER_API(Strdup);
     REGISTER_API(CreateCommand);
     REGISTER_API(SetModuleAttribs);
     REGISTER_API(WrongArity);
@@ -2263,6 +2907,7 @@ void moduleRegisterCoreAPI(void) {
     REGISTER_API(CreateStringFromCallReply);
     REGISTER_API(CreateString);
     REGISTER_API(CreateStringFromLongLong);
+    REGISTER_API(CreateStringFromString);
     REGISTER_API(FreeString);
     REGISTER_API(StringPtrLen);
     REGISTER_API(AutoMemory);
@@ -2292,6 +2937,23 @@ void moduleRegisterCoreAPI(void) {
     REGISTER_API(IsKeysPositionRequest);
     REGISTER_API(KeyAtPos);
     REGISTER_API(GetClientId);
+    REGISTER_API(PoolAlloc);
+    REGISTER_API(CreateDataType);
+    REGISTER_API(ModuleTypeSetValue);
+    REGISTER_API(ModuleTypeGetType);
+    REGISTER_API(ModuleTypeGetValue);
+    REGISTER_API(SaveUnsigned);
+    REGISTER_API(LoadUnsigned);
+    REGISTER_API(SaveSigned);
+    REGISTER_API(LoadSigned);
+    REGISTER_API(SaveString);
+    REGISTER_API(SaveStringBuffer);
+    REGISTER_API(LoadString);
+    REGISTER_API(LoadStringBuffer);
+    REGISTER_API(SaveDouble);
+    REGISTER_API(LoadDouble);
+    REGISTER_API(EmitAOF);
+    REGISTER_API(Log);
 }
 
 /* Global initialization at Redis startup. */
@@ -2316,25 +2978,28 @@ void moduleLoadFromQueue(void) {
 
     listRewind(server.loadmodule_queue,&li);
     while((ln = listNext(&li))) {
-        sds modulepath = ln->value;
-        if (moduleLoad(modulepath) == C_ERR) {
+        struct moduleLoadQueueEntry *loadmod = ln->value;
+        if (moduleLoad(loadmod->path,(void **)loadmod->argv,loadmod->argc)
+            == C_ERR)
+        {
             serverLog(LL_WARNING,
                 "Can't load module from %s: server aborting",
-                modulepath);
+                loadmod->path);
             exit(1);
         }
     }
 }
 
 void moduleFreeModuleStructure(struct RedisModule *module) {
+    listRelease(module->types);
     sdsfree(module->name);
     zfree(module);
 }
 
 /* Load a module and initialize it. On success C_OK is returned, otherwise
  * C_ERR is returned. */
-int moduleLoad(const char *path) {
-    int (*onload)(void *);
+int moduleLoad(const char *path, void **module_argv, int module_argc) {
+    int (*onload)(void *, void **, int);
     void *handle;
     RedisModuleCtx ctx = REDISMODULE_CTX_INIT;
 
@@ -2343,14 +3008,14 @@ int moduleLoad(const char *path) {
         serverLog(LL_WARNING, "Module %s failed to load: %s", path, dlerror());
         return C_ERR;
     }
-    onload = (int (*)(void *))(unsigned long) dlsym(handle,"RedisModule_OnLoad");
+    onload = (int (*)(void *, void **, int))(unsigned long) dlsym(handle,"RedisModule_OnLoad");
     if (onload == NULL) {
         serverLog(LL_WARNING,
             "Module %s does not export RedisModule_OnLoad() "
             "symbol. Module not loaded.",path);
         return C_ERR;
     }
-    if (onload((void*)&ctx) == REDISMODULE_ERR) {
+    if (onload((void*)&ctx,module_argv,module_argc) == REDISMODULE_ERR) {
         if (ctx.module) moduleFreeModuleStructure(ctx.module);
         dlclose(handle);
         serverLog(LL_WARNING,
@@ -2362,6 +3027,7 @@ int moduleLoad(const char *path) {
     dictAdd(modules,ctx.module->name,ctx.module);
     ctx.module->handle = handle;
     serverLog(LL_NOTICE,"Module '%s' loaded from %s",ctx.module->name,path);
+    moduleFreeContext(&ctx);
     return C_OK;
 }
 
@@ -2369,14 +3035,21 @@ int moduleLoad(const char *path) {
  * C_OK is returned, otherwise C_ERR is returned and errno is set
  * to the following values depending on the type of error:
  *
- * ENONET: No such module having the specified name. */
+ * ENONET: No such module having the specified name.
+ * EBUSY: The module exports a new data type and can only be reloaded. */
 int moduleUnload(sds name) {
     struct RedisModule *module = dictFetchValue(modules,name);
+
     if (module == NULL) {
         errno = ENOENT;
         return REDISMODULE_ERR;
     }
 
+    if (listLength(module->types)) {
+        errno = EBUSY;
+        return REDISMODULE_ERR;
+    }
+
     /* Unregister all the commands registered by this module. */
     dictIterator *di = dictGetSafeIterator(server.commands);
     dictEntry *de;
@@ -2410,21 +3083,27 @@ int moduleUnload(sds name) {
     /* Remove from list of modules. */
     serverLog(LL_NOTICE,"Module %s unloaded",module->name);
     dictDelete(modules,module->name);
-
-    /* Free the module structure. */
-    zfree(module);
+    moduleFreeModuleStructure(module);
 
     return REDISMODULE_OK;
 }
 
 /* Redis MODULE command.
  *
- * MODULE LOAD <path> */
+ * MODULE LOAD <path> [args...] */
 void moduleCommand(client *c) {
     char *subcmd = c->argv[1]->ptr;
 
-    if (!strcasecmp(subcmd,"load") && c->argc == 3) {
-        if (moduleLoad(c->argv[2]->ptr) == C_OK)
+    if (!strcasecmp(subcmd,"load") && c->argc >= 3) {
+        robj **argv = NULL;
+        int argc = 0;
+
+        if (c->argc > 3) {
+            argc = c->argc - 3;
+            argv = &c->argv[3];
+        }
+
+        if (moduleLoad(c->argv[2]->ptr,(void **)argv,argc) == C_OK)
             addReply(c,shared.ok);
         else
             addReplyError(c,
@@ -2433,9 +3112,17 @@ void moduleCommand(client *c) {
         if (moduleUnload(c->argv[2]->ptr) == C_OK)
             addReply(c,shared.ok);
         else {
-            char *errmsg = "operation not possible.";
+            char *errmsg;
             switch(errno) {
-            case ENOENT: errmsg = "no such module with that name";
+            case ENOENT:
+                errmsg = "no such module with that name";
+                break;
+            case EBUSY:
+                errmsg = "the module exports one or more module-side data types, can't unload";
+                break;
+            default:
+                errmsg = "operation not possible.";
+                break;
             }
             addReplyErrorFormat(c,"Error unloading module: %s",errmsg);
         }
diff --git a/src/modules/API.md b/src/modules/API.md
index 0ff303d0e..9c7ada9dd 100644
--- a/src/modules/API.md
+++ b/src/modules/API.md
@@ -1,5 +1,51 @@
 # Modules API reference
 
+## `RM_Alloc`
+
+    void *RM_Alloc(size_t bytes);
+
+Use like malloc(). Memory allocated with this function is reported in
+Redis INFO memory, used for keys eviction according to maxmemory settings
+and in general is taken into account as memory allocated by Redis.
+You should avoid to use malloc().
+
+## `RM_Realloc`
+
+    void* RM_Realloc(void *ptr, size_t bytes);
+
+Use like realloc() for memory obtained with `RedisModule_Alloc()`.
+
+## `RM_Free`
+
+    void RM_Free(void *ptr);
+
+Use like free() for memory obtained by `RedisModule_Alloc()` and
+`RedisModule_Realloc()`. However you should never try to free with
+`RedisModule_Free()` memory allocated with malloc() inside your module.
+
+## `RM_Strdup`
+
+    char *RM_Strdup(const char *str);
+
+Like strdup() but returns memory allocated with `RedisModule_Alloc()`.
+
+## `RM_PoolAlloc`
+
+    void *RM_PoolAlloc(RedisModuleCtx *ctx, size_t bytes);
+
+Return heap allocated memory that will be freed automatically when the
+module callback function returns. Mostly suitable for small allocations
+that are short living and must be released when the callback returns
+anyway. The returned memory is aligned to the architecture word size
+if at least word size bytes are requested, otherwise it is just
+aligned to the next power of two, so for example a 3 bytes request is
+4 bytes aligned while a 2 bytes request is 2 bytes aligned.
+
+There is no realloc style function since when this is needed to use the
+pool allocator is not a good idea.
+
+The function returns NULL if `bytes` is 0.
+
 ## `RM_GetApi`
 
     int RM_GetApi(const char *funcname, void **targetPtrPtr);
@@ -133,6 +179,16 @@ integer instead of taking a buffer and its length.
 The returned string must be released with `RedisModule_FreeString()` or by
 enabling automatic memory management.
 
+## `RM_CreateStringFromString`
+
+    RedisModuleString *RM_CreateStringFromString(RedisModuleCtx *ctx, const RedisModuleString *str);
+
+Like `RedisModule_CreatString()`, but creates a string starting from an existing
+RedisModuleString.
+
+The returned string must be released with `RedisModule_FreeString()` or by
+enabling automatic memory management.
+
 ## `RM_FreeString`
 
     void RM_FreeString(RedisModuleCtx *ctx, RedisModuleString *str);
@@ -579,9 +635,9 @@ The output flags are:
 On success the function returns `REDISMODULE_OK`. On the following errors
 `REDISMODULE_ERR` is returned:
 
-- The key was not opened for writing.
-- The key is of the wrong type.
-- 'score' double value is not a number (NaN).
+* The key was not opened for writing.
+* The key is of the wrong type.
+* 'score' double value is not a number (NaN).
 
 ## `RM_ZsetIncrby`
 
@@ -609,8 +665,8 @@ Remove the specified element from the sorted set.
 The function returns `REDISMODULE_OK` on success, and `REDISMODULE_ERR`
 on one of the following conditions:
 
-- The key was not opened for writing.
-- The key is of the wrong type.
+* The key was not opened for writing.
+* The key is of the wrong type.
 
 The return value does NOT indicate the fact the element was really
 removed (since it existed) or not, just if the function was executed
@@ -632,9 +688,9 @@ On success retrieve the double score associated at the sorted set element
 'ele' and returns `REDISMODULE_OK`. Otherwise `REDISMODULE_ERR` is returned
 to signal one of the following conditions:
 
-- There is no such element 'ele' in the sorted set.
-- The key is not a sorted set.
-- The key is an open empty key.
+* There is no such element 'ele' in the sorted set.
+* The key is not a sorted set.
+* The key is an open empty key.
 
 ## `RM_ZsetRangeStop`
 
@@ -774,8 +830,8 @@ specified because of the XX or NX options).
 
 In the following case the return value is always zero:
 
-- The key was not open for writing.
-- The key was associated with a non Hash value.
+* The key was not open for writing.
+* The key was associated with a non Hash value.
 
 ## `RM_HashGet`
 
@@ -893,3 +949,197 @@ EPERM:  operation in Cluster instance with key in non local slot.
 Return a pointer, and a length, to the protocol returned by the command
 that returned the reply object.
 
+## `RM_CreateDataType`
+
+    moduleType *RM_CreateDataType(RedisModuleCtx *ctx, const char *name, int encver, moduleTypeLoadFunc rdb_load, moduleTypeSaveFunc rdb_save, moduleTypeRewriteFunc aof_rewrite, moduleTypeDigestFunc digest, moduleTypeFreeFunc free);
+
+Register a new data type exported by the module. The parameters are the
+following. Please for in depth documentation check the modules API
+documentation, especially the INTRO.md file.
+
+* **name**: A 9 characters data type name that MUST be unique in the Redis
+  Modules ecosystem. Be creative... and there will be no collisions. Use
+  the charset A-Z a-z 9-0, plus the two "-_" characters. A good
+  idea is to use, for example `<typename>-<vendor>`. For example
+  "tree-AntZ" may mean "Tree data structure by @antirez". To use both
+  lower case and upper case letters helps in order to prevent collisions.
+* **encver**: Encoding version, which is, the version of the serialization
+  that a module used in order to persist data. As long as the "name"
+  matches, the RDB loading will be dispatched to the type callbacks
+  whatever 'encver' is used, however the module can understand if
+  the encoding it must load are of an older version of the module.
+  For example the module "tree-AntZ" initially used encver=0. Later
+  after an upgrade, it started to serialize data in a different format
+  and to register the type with encver=1. However this module may
+  still load old data produced by an older version if the rdb_load
+  callback is able to check the encver value and act accordingly.
+  The encver must be a positive value between 0 and 1023.
+* **rdb_load**: A callback function pointer that loads data from RDB files.
+* **rdb_save**: A callback function pointer that saves data to RDB files.
+* **aof_rewrite**: A callback function pointer that rewrites data as commands.
+* **digest**: A callback function pointer that is used for `DEBUG DIGEST`.
+* **free**: A callback function pointer that can free a type value.
+
+Note: the module name "AAAAAAAAA" is reserved and produces an error, it
+happens to be pretty lame as well.
+
+If there is already a module registering a type with the same name,
+and if the module name or encver is invalid, NULL is returned.
+Otherwise the new type is registered into Redis, and a reference of
+type RedisModuleType is returned: the caller of the function should store
+this reference into a gobal variable to make future use of it in the
+modules type API, since a single module may register multiple types.
+Example code fragment:
+
+     static RedisModuleType *BalancedTreeType;
+
+     int `RedisModule_OnLoad(RedisModuleCtx` *ctx) {
+         // some code here ...
+         BalancedTreeType = `RM_CreateDataType(`...);
+     }
+
+## `RM_ModuleTypeSetValue`
+
+    int RM_ModuleTypeSetValue(RedisModuleKey *key, moduleType *mt, void *value);
+
+If the key is open for writing, set the specified module type object
+as the value of the key, deleting the old value if any.
+On success `REDISMODULE_OK` is returned. If the key is not open for
+writing or there is an active iterator, `REDISMODULE_ERR` is returned.
+
+## `RM_ModuleTypeGetType`
+
+    moduleType *RM_ModuleTypeGetType(RedisModuleKey *key);
+
+Assuming `RedisModule_KeyType()` returned `REDISMODULE_KEYTYPE_MODULE` on
+the key, returns the moduel type pointer of the value stored at key.
+
+If the key is NULL, is not associated with a module type, or is empty,
+then NULL is returned instead.
+
+## `RM_ModuleTypeGetValue`
+
+    void *RM_ModuleTypeGetValue(RedisModuleKey *key);
+
+Assuming `RedisModule_KeyType()` returned `REDISMODULE_KEYTYPE_MODULE` on
+the key, returns the module type low-level value stored at key, as
+it was set by the user via `RedisModule_ModuleTypeSet()`.
+
+If the key is NULL, is not associated with a module type, or is empty,
+then NULL is returned instead.
+
+## `RM_SaveUnsigned`
+
+    void RM_SaveUnsigned(RedisModuleIO *io, uint64_t value);
+
+Save an unsigned 64 bit value into the RDB file. This function should only
+be called in the context of the rdb_save method of modules implementing new
+data types.
+
+## `RM_LoadUnsigned`
+
+    uint64_t RM_LoadUnsigned(RedisModuleIO *io);
+
+Load an unsigned 64 bit value from the RDB file. This function should only
+be called in the context of the rdb_load method of modules implementing
+new data types.
+
+## `RM_SaveSigned`
+
+    void RM_SaveSigned(RedisModuleIO *io, int64_t value);
+
+Like `RedisModule_SaveUnsigned()` but for signed 64 bit values.
+
+## `RM_LoadSigned`
+
+    int64_t RM_LoadSigned(RedisModuleIO *io);
+
+Like `RedisModule_LoadUnsigned()` but for signed 64 bit values.
+
+## `RM_SaveString`
+
+    void RM_SaveString(RedisModuleIO *io, RedisModuleString *s);
+
+In the context of the rdb_save method of a module type, saves a
+string into the RDB file taking as input a RedisModuleString.
+
+The string can be later loaded with `RedisModule_LoadString()` or
+other Load family functions expecting a serialized string inside
+the RDB file.
+
+## `RM_SaveStringBuffer`
+
+    void RM_SaveStringBuffer(RedisModuleIO *io, const char *str, size_t len);
+
+Like `RedisModule_SaveString()` but takes a raw C pointer and length
+as input.
+
+## `RM_LoadString`
+
+    RedisModuleString *RM_LoadString(RedisModuleIO *io);
+
+In the context of the rdb_load method of a module data type, loads a string
+from the RDB file, that was previously saved with `RedisModule_SaveString()`
+functions family.
+
+The returned string is a newly allocated RedisModuleString object, and
+the user should at some point free it with a call to `RedisModule_FreeString()`.
+
+If the data structure does not store strings as RedisModuleString objects,
+the similar function `RedisModule_LoadStringBuffer()` could be used instead.
+
+## `RM_LoadStringBuffer`
+
+    char *RM_LoadStringBuffer(RedisModuleIO *io, size_t *lenptr);
+
+Like `RedisModule_LoadString()` but returns an heap allocated string that
+was allocated with `RedisModule_Alloc()`, and can be resized or freed with
+`RedisModule_Realloc()` or `RedisModule_Free()`.
+
+The size of the string is stored at '*lenptr' if not NULL.
+The returned string is not automatically NULL termianted, it is loaded
+exactly as it was stored inisde the RDB file.
+
+## `RM_SaveDouble`
+
+    void RM_SaveDouble(RedisModuleIO *io, double value);
+
+In the context of the rdb_save method of a module data type, saves a double
+value to the RDB file. The double can be a valid number, a NaN or infinity.
+It is possible to load back the value with `RedisModule_LoadDouble()`.
+
+## `RM_LoadDouble`
+
+    double RM_LoadDouble(RedisModuleIO *io);
+
+In the context of the rdb_save method of a module data type, loads back the
+double value saved by `RedisModule_SaveDouble()`.
+
+## `RM_EmitAOF`
+
+    void RM_EmitAOF(RedisModuleIO *io, const char *cmdname, const char *fmt, ...);
+
+Emits a command into the AOF during the AOF rewriting process. This function
+is only called in the context of the aof_rewrite method of data types exported
+by a module. The command works exactly like `RedisModule_Call()` in the way
+the parameters are passed, but it does not return anything as the error
+handling is performed by Redis itself.
+
+## `RM_Log`
+
+    void RM_Log(RedisModuleCtx *ctx, const char *levelstr, const char *fmt, ...);
+
+Produces a log message to the standard Redis log, the format accepts
+printf-alike specifiers, while level is a string describing the log
+level to use when emitting the log, and must be one of the following:
+
+* "debug"
+* "verbose"
+* "notice"
+* "warning"
+
+If the specified log level is invalid, verbose is used by default.
+There is a fixed limit to the length of the log line this function is able
+to emit, this limti is not specified but is guaranteed to be more than
+a few lines of text.
+
diff --git a/src/modules/INTRO.md b/src/modules/INTRO.md
index c64a50078..e5576b7fc 100644
--- a/src/modules/INTRO.md
+++ b/src/modules/INTRO.md
@@ -1,6 +1,12 @@
-Redis Modules API reference manual
+Redis Modules: an introduction to the API
 ===
 
+The modules documentation is composed of the following files:
+
+* `INTRO.md` (this file). An overview about Redis Modules system and API. It's a good idea to start your reading here.
+* `API.md` is generated from module.c top comments of RedisMoule functions. It is a good reference in order to understand how each function works.
+* `TYPES.md` covers the implementation of native data types into modules.
+
 Redis modules make possible to extend Redis functionality using external
 modules, implementing new Redis commands at a speed and with features
 similar to what can be done inside the core itself.
@@ -59,7 +65,7 @@ simple module that implements a command that outputs a random number.
         return REDISMODULE_OK;
     }
 
-    int RedisModule_OnLoad(RedisModuleCtx *ctx) {
+    int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
         if (RedisModule_Init(ctx,"helloworld",1,REDISMODULE_APIVER_1)
             == REDISMODULE_ERR) return REDISMODULE_ERR;
 
@@ -150,6 +156,24 @@ exported.
 
 The module will be able to load into different versions of Redis.
 
+# Passing configuration parameters to Redis modules
+
+When the module is loaded with the `MODULE LOAD` command, or using the
+`loadmodule` directive in the `redis.conf` file, the user is able to pass
+configuration parameters to the module by adding arguments after the module
+file name:
+
+    loadmodule mymodule.so foo bar 1234
+
+In the above example the strings `foo`, `bar` and `123` will be passed
+to the module `OnLoad()` function in the `argv` argument as an array
+of RedisModuleString pointers. The number of arguments passed is into `argc`.
+
+The way you can access those strings will be explained in the rest of this
+document. Normally the module will store the module configuration parameters
+in some `static` global variable that can be accessed module wide, so that
+the configuration can change the behavior of different commands.
+
 # Working with RedisModuleString objects
 
 The command argument vector `argv` passed to module commands, and the
@@ -162,7 +186,7 @@ There are a few functions in order to work with string objects:
 
     const char *RedisModule_StringPtrLen(RedisModuleString *string, size_t *len);
 
-The above function accesses a string by returning its pointer and setting its 
+The above function accesses a string by returning its pointer and setting its
 length in `len`.
 You should never write to a string object pointer, as you can see from the
 `const` pointer qualifier.
@@ -344,7 +368,7 @@ section).
 
 # Releasing call reply objects
 
-Reply objects must be freed using `RedisModule_FreeCallRelpy`. For arrays,
+Reply objects must be freed using `RedisModule_FreeCallReply`. For arrays,
 you need to free only the top level reply, not the nested replies.
 Currently the module implementation provides a protection in order to avoid
 crashing if you free a nested reply object for error, however this feature
@@ -623,7 +647,7 @@ access) for speed. The API will return a pointer and a length, so that's
 possible to access and, if needed, modify the string directly.
 
     size_t len, j;
-    char *myptr = RedisModule_StringDMA(key,REDISMODULE_WRITE,&len);
+    char *myptr = RedisModule_StringDMA(key,&len,REDISMODULE_WRITE);
     for (j = 0; j < len; j++) myptr[j] = 'A';
 
 In the above example we write directly on the string. Note that if you want
@@ -777,10 +801,56 @@ Automatic memory management is usually the way to go, however experienced
 C programmers may not use it in order to gain some speed and memory usage
 benefit.
 
+# Allocating memory into modules
+
+Normal C programs use `malloc()` and `free()` in order to allocate and
+release memory dynamically. While in Redis modules the use of malloc is
+not technically forbidden, it is a lot better to use the Redis Modules
+specific functions, that are exact replacements for `malloc`, `free`,
+`realloc` and `strdup`. These functions are:
+
+    void *RedisModule_Alloc(size_t bytes);
+    void* RedisModule_Realloc(void *ptr, size_t bytes);
+    void RedisModule_Free(void *ptr);
+    void RedisModule_Calloc(size_t nmemb, size_t size);
+    char *RedisModule_Strdup(const char *str);
+
+They work exactly like their `libc` equivalent calls, however they use
+the same allocator Redis uses, and the memory allocated using these
+functions is reported by the `INFO` command in the memory section, is
+accounted when enforcing the `maxmemory` policy, and in general is
+a first citizen of the Redis executable. On the contrar, the method
+allocated inside modules with libc `malloc()` is transparent to Redis.
+
+Another reason to use the modules functions in order to allocate memory
+is that, when creating native data types inside modules, the RDB loading
+functions can return deserialized strings (from the RDB file) directly
+as `RedisModule_Alloc()` allocations, so they can be used directly to
+populate data structures after loading, instead of having to copy them
+to the data structure.
+
+## Pool allocator
+
+Sometimes in commands implementations, it is required to perform many
+small allocations that will be not retained at the end of the command
+execution, but are just functional to execute the command itself.
+
+This work can be more easily accomplished using the Redis pool allocator:
+
+    void *RedisModule_PoolAlloc(RedisModuleCtx *ctx, size_t bytes);
+
+It works similarly to `malloc()`, and returns memory aligned to the
+next power of two of greater or equal to `bytes` (for a maximum alignment
+of 8 bytes). However it allocates memory in blocks, so it the overhead
+of the allocations is small, and more important, the memory allocated
+is automatically released when the command returns.
+
+So in general short living allocations are a good candidates for the pool
+allocator.
+
 # Writing commands compatible with Redis Cluster
 
 Documentation missing, please check the following functions inside `module.c`:
 
     RedisModule_IsKeysPositionRequest(ctx);
     RedisModule_KeyAtPos(ctx,pos);
-
diff --git a/src/modules/Makefile b/src/modules/Makefile
index 0c91361a1..ecac4683f 100644
--- a/src/modules/Makefile
+++ b/src/modules/Makefile
@@ -13,7 +13,7 @@ endif
 
 .SUFFIXES: .c .so .xo .o
 
-all: helloworld.so
+all: helloworld.so hellotype.so
 
 .c.xo:
 	$(CC) -I. $(CFLAGS) $(SHOBJ_CFLAGS) -fPIC -c $< -o $@
@@ -23,5 +23,10 @@ helloworld.xo: ../redismodule.h
 helloworld.so: helloworld.xo
 	$(LD) -o $@ $< $(SHOBJ_LDFLAGS) $(LIBS) -lc
 
+hellotype.xo: ../redismodule.h
+
+hellotype.so: hellotype.xo
+	$(LD) -o $@ $< $(SHOBJ_LDFLAGS) $(LIBS) -lc
+
 clean:
 	rm -rf *.xo *.so
diff --git a/src/modules/TYPES.md b/src/modules/TYPES.md
new file mode 100644
index 000000000..1c31950fa
--- /dev/null
+++ b/src/modules/TYPES.md
@@ -0,0 +1,371 @@
+Native types in Redis modules
+===
+
+Redis modules can access Redis built-in data structures both at high level,
+by calling Redis commands, and at low level, by manipulating the data structures
+directly.
+
+By using these capabilities in order to build new abstractions on top of existing
+Redis data structures, or by using strings DMA in order to encode modules
+data structures into Redis strings, it is possible to create modules that
+*feel like* they are exporting new data types. However, for more complex
+problems, this is not enough, and the implementation of new data structures
+inside the module is needed.
+
+We call the ability of Redis modules to implement new data structures that
+feel like native Redis ones **native types support**. This document describes
+the API exported by the Redis modules system in order to create new data
+structures and handle the serialization in RDB files, the rewriting process
+in AOF, the type reporting via the `TYPE` command, and so forth.
+
+Overview of native types
+---
+
+A module exporting a native type is composed of the following main parts:
+
+* The implementation of some kind of new data structure and of commands operating on the new data structure.
+* A set of callbacks that handle: RDB saving, RDB loading, AOF rewriting, releasing of a value associated with a key, calculation of a value digest (hash) to be used with the `DEBUG DIGEST` command.
+* A 9 characters name that is unique to each module native data type.
+* An encoding version, used to persist into RDB files a module-specific data version, so that a module will be able to load older representations from RDB files.
+
+While to handle RDB loading, saving and AOF rewriting may look complex as a first glance, the modules API provide very high level function for handling all this, without requiring the user to handle read/write errors, so in practical terms, writing a new data structure for Redis is a simple task.
+
+A **very easy** to understand but complete example of native type implementation
+is available inside the Redis distribution in the `/modules/hellotype.c` file.
+The reader is encouraged to read the documentation by looking at this example
+implementation to see how things are applied in the practice.
+
+Registering a new data type
+===
+
+In order to register a new native type into the Redis core, the module needs
+to declare a global variable that will hold a reference to the data type.
+The API to register the data type will return a data type reference that will
+be stored in the global variable.
+
+    static RedisModuleType *MyType;
+    #define MYTYPE_ENCODING_VERSION 0
+
+    int RedisModule_OnLoad(RedisModuleCtx *ctx) {
+        MyType = RedisModule_CreateDataType("MyType-AZ", MYTYPE_ENCODING_VERSION,
+            MyTypeRDBLoad, MyTypeRDBSave, MyTypeAOFRewrite, MyTypeDigest,
+            MyTypeFree);
+        if (MyType == NULL) return REDISMODULE_ERR;
+    }
+
+As you can see from the example above, a single API call is needed in order to
+register the new type. However a number of function pointers are passed as
+arguments. The prototype of `RedisModule_CreateDataType` is the following:
+
+    moduleType *RedisModule_CreateDataType(RedisModuleCtx *ctx,
+               const char *name, int encver,
+               moduleTypeLoadFunc rdb_load,
+               moduleTypeSaveFunc rdb_save,
+               moduleTypeRewriteFunc aof_rewrite,
+               moduleTypeDigestFunc digest,
+               moduleTypeFreeFunc free);
+
+The `ctx` argument is the context that we receive in the `OnLoad` function.
+The type `name` is a 9 character name in the character set that includes
+from `A-Z`, `a-z`, `0-9`, plus the underscore `_` and minus `-` characters.
+
+Note that **this name must be unique** for each data type in the Redis
+ecosystem, so be creative, use both lower-case and upper case if it makes
+sense, and try to use the convention of mixing the type name with the name
+of the author of the module, to create a 9 character unique name.
+
+For example if I'm building a *b-tree* data structure and my name is *antirez*
+I'll call my type **btree1-az**. The name, converted to a 64 bit integer,
+is stored inside the RDB file when saving the type, and will be used when the
+RDB data is loaded in order to resolve what module can load the data. If Redis
+finds no matching module, the integer is converted back to a name in order to
+provide some clue to the user about what module is missing in order to load
+the data.
+
+The type name is also used as a reply for the `TYPE` command when called
+with a key holding the registered type.
+
+The `encver` argument is the encoding version used by the module to store data
+inside the RDB file. For example I can start with an encoding version of 0,
+but later when I release version 2.0 of my module, I can switch encoding to
+something better. The new module will register with an encoding version of 1,
+so when it saves new RDB files, the new version will be stored on disk. However
+when loading RDB files, the module `rdb_load` method will be called even if
+there is data found for a different encoding version (and the encoding version
+is passed as argument to `rdb_load`), so that the module can still load old
+RDB files.
+
+The remaining arguments `rdb_load`, `rdb_save`, `aof_rewrite`, `digest` and
+`free` are all callbacks with the following prototypes and uses:
+
+    typedef void *(*RedisModuleTypeLoadFunc)(RedisModuleIO *rdb, int encver);
+    typedef void (*RedisModuleTypeSaveFunc)(RedisModuleIO *rdb, void *value);
+    typedef void (*RedisModuleTypeRewriteFunc)(RedisModuleIO *aof, RedisModuleString *key, void *value);
+    typedef void (*RedisModuleTypeDigestFunc)(RedisModuleDigest *digest, void *value);
+    typedef void (*RedisModuleTypeFreeFunc)(void *value);
+
+* `rdb_load` is called when loading data from the RDB file. It loads data in the same format as `rdb_save` produces.
+* `rdb_save` is called when saving data to the RDB file.
+* `aof_rewrite` is called when the AOF is being rewritten, and the module needs to tell Redis what is the sequence of commands to recreate the content of a given key.
+* `digest` is called when `DEBUG DIGEST` is executed and a key holding this module type is found. Currently this is not yet implemented so the function ca be left empty.
+* `free` is called when a key with the module native type is deleted via `DEL` or in any other mean, in order to let the module reclaim the memory associated with such a value.
+
+Ok, but *why* modules types require a 9 characters name?
+---
+
+Oh, I understand you need to understand this, so here is a very specific
+explanation.
+
+When Redis persists to RDB files, modules specific data types require to
+be persisted as well. Now RDB files are sequences of key-value pairs
+like the following:
+
+    [1 byte type] [key] [a type specific value]
+
+The 1 byte type identifies strings, lists, sets, and so forth. In the case
+of modules data, it is set to a special value of `module data`, but of
+course this is not enough, we need the information needed to link a specific
+value with a specific module type that is able to load and handle it.
+
+So when we save a `type specific value` about a module, we prefix it with
+a 64 bit integer. 64 bits is large enough to store the informations needed
+in order to lookup the module that can handle that specific type, but is
+short enough that we can prefix each module value we store inside the RDB
+without making the final RDB file too big. At the same time, this solution
+of prefixing the value with a 64 bit *signature* does not require to do
+strange things like defining in the RDB header a list of modules specific
+types. Everything is pretty simple.
+
+So, what you can store in 64 bits in order to identify a given module in
+a reliable way? Well if you build a character set of 64 symbols, you can
+easily store 9 characters of 6 bits, and you are left with 10 bits, that
+are used in order to store the *encoding version* of the type, so that
+the same type can evolve in the future and provide a different and more
+efficient or updated serialization format for RDB files.
+
+So the 64 bit prefix stored before each module value is like the following:
+
+    6|6|6|6|6|6|6|6|6|10
+
+The first 9 elements are 6-bits characters, the final 10 bits is the
+encoding version.
+
+When the RDB file is loaded back, it reads the 64 bit value, masks the final
+10 bits, and searches for a matching module in the modules types cache.
+When a matching one is found, the method to load the RDB file value is called
+with the 10 bits encoding version as argument, so that the module knows
+what version of the data layout to load, if it can support multiple versions.
+
+Now the interesting thing about all this is that, if instead the module type
+cannot be resolved, since there is no loaded module having this signature,
+we can convert back the 64 bit value into a 9 characters name, and print
+an error to the user that includes the module type name! So that she or he
+immediately realizes what's wrong.
+
+Setting and getting keys
+---
+
+After registering our new data type in the `RedisModule_OnLoad()` function,
+we also need to be able to set Redis keys having as value our native type.
+
+This normally happens in the context of commands that write data to a key.
+The native types API allow to set and get keys to module native data types,
+and to test if a given key is already associated to a value of a specific data
+type.
+
+The API uses the normal modules `RedisModule_OpenKey()` low level key access
+interface in order to deal with this. This is an eaxmple of setting a
+native type private data structure to a Redis key:
+
+    RedisModuleKey *key = RedisModule_OpenKey(ctx,keyname,REDISMODULE_WRITE);
+    struct some_private_struct *data = createMyDataStructure();
+    RedisModule_ModuleTypeSetValue(key,MyType,data);
+
+The function `RedisModule_ModuleTypeSetValue()` is used with a key handle open
+for writing, and gets three arguments: the key handle, the reference to the
+native type, as obtained during the type registration, and finally a `void*`
+pointer that contains the private data implementing the module native type.
+
+Note that Redis has no clues at all about what your data contains. It will
+just call the callbacks you provided during the method registration in order
+to perform operations on the type.
+
+Similarly we can retrieve the private data from a key using this function:
+
+    struct some_private_struct *data;
+    data = RedisModule_ModuleTypeGetValue(key);
+
+We can also test for a key to have our native type as value:
+
+    if (RedisModule_ModuleTypeGetType(key) == MyType) {
+        /* ... do something ... */
+    }
+
+However for the calls to do the right thing, we need to check if the key
+is empty, if it contains a value of the right kind, and so forth. So
+the idiomatic code to implement a command writing to our native type
+is along these lines:
+
+    RedisModuleKey *key = RedisModule_OpenKey(ctx,argv[1],
+        REDISMODULE_READ|REDISMODULE_WRITE);
+    int type = RedisModule_KeyType(key);
+    if (type != REDISMODULE_KEYTYPE_EMPTY &&
+        RedisModule_ModuleTypeGetType(key) != MyType)
+    {
+        return RedisModule_ReplyWithError(ctx,REDISMODULE_ERRORMSG_WRONGTYPE);
+    }
+
+Then if we successfully verified the key is not of the wrong type, and
+we are going to write to it, we usually want to create a new data structure if
+the key is empty, or retrieve the reference to the value associated to the
+key if there is already one:
+
+    /* Create an empty value object if the key is currently empty. */
+    struct some_private_struct *data;
+    if (type == REDISMODULE_KEYTYPE_EMPTY) {
+        data = createMyDataStructure();
+        RedisModule_ModuleTypeSetValue(key,MyTyke,data);
+    } else {
+        data = RedisModule_ModuleTypeGetValue(key);
+    }
+    /* Do something with 'data'... */
+
+Free method
+---
+
+As already mentioned, when Redis needs to free a key holding a native type
+value, it needs help from the module in order to release the memory. This
+is the reason why we pass a `free` callback during the type registration:
+
+    typedef void (*RedisModuleTypeFreeFunc)(void *value);
+
+A trivial implementation of the free method can be something like this,
+assuming our data structure is composed of a single allocation:
+
+    void MyTypeFreeCallback(void *value) {
+        RedisModule_Free(value);
+    }
+
+However a more real world one will call some function that performs a more
+complex memory reclaiming, by casting the void pointer to some structure
+and freeing all the resources composing the value.
+
+RDB load and save methods
+---
+
+The RDB saving and loading callbacks need to create (and load back) a
+representation of the data type on disk. Redis offers an high level API
+that can automatically store inside the RDB file the following types:
+
+* Unsigned 64 bit integers.
+* Signed 64 bit integers.
+* Doubles.
+* Strings.
+
+It is up to the module to find a viable representation using the above base
+types. However note that while the integer and double values are stored
+and loaded in an architecture and *endianess* agnostic way, if you use
+the raw string saving API to, for example, save a structure on disk, you
+have to care those details yourself.
+
+This is the list of functions performing RDB saving and loading:
+
+    void RedisModule_SaveUnsigned(RedisModuleIO *io, uint64_t value);
+    uint64_t RedisModule_LoadUnsigned(RedisModuleIO *io);
+    void RedisModule_SaveSigned(RedisModuleIO *io, int64_t value);
+    int64_t RedisModule_LoadSigned(RedisModuleIO *io);
+    void RedisModule_SaveString(RedisModuleIO *io, RedisModuleString *s);
+    void RedisModule_SaveStringBuffer(RedisModuleIO *io, const char *str, size_t len);
+    RedisModuleString *RedisModule_LoadString(RedisModuleIO *io);
+    char *RedisModule_LoadStringBuffer(RedisModuleIO *io, size_t *lenptr);
+    void RedisModule_SaveDouble(RedisModuleIO *io, double value);
+    double RedisModule_LoadDouble(RedisModuleIO *io);
+
+The functions don't require any error checking from the module, that can
+always assume calls succeed.
+
+As an example, imagine I've a native type that implements an array of
+double values, with the following structure:
+
+    struct double_array {
+        size_t count;
+        double *values;
+    };
+
+My `rdb_save` method may look like the following:
+
+    void DoubleArrayRDBSave(RedisModuleIO *io, void *ptr) {
+        struct dobule_array *da = ptr;
+        RedisModule_SaveUnsigned(io,da->count);
+        for (size_t j = 0; j < da->count; j++)
+            RedisModule_SaveDouble(io,da->values[j]);
+    }
+
+What we did was to store the number of elements followed by each double
+value. So when later we'll have to load the structure in the `rdb_load`
+method we'll do something like this:
+
+    void *DoubleArrayRDBLoad(RedisModuleIO *io, int encver) {
+        if (encver != DOUBLE_ARRAY_ENC_VER) {
+            /* We should actually log an error here, or try to implement
+               the ability to load older versions of our data structure. */
+            return NULL;
+        }
+
+        struct double_array *da;
+        da = RedisModule_Alloc(sizeof(*da));
+        da->count = RedisModule_LoadUnsigned(io);
+        da->values = RedisModule_Alloc(da->count * sizeof(double));
+        for (size_t j = 0; j < da->count; j++)
+            da->values = RedisModule_LoadDouble(io);
+        return da;
+    }
+
+The load callback just reconstruct back the data structure from the data
+we stored in the RDB file.
+
+Note that while there is no error handling on the API that writes and reads
+from disk, still the load callback can return NULL on errors in case what
+it reads does not look correct. Redis will just panic in that case.
+
+AOF rewriting
+---
+
+    void RedisModule_EmitAOF(RedisModuleIO *io, const char *cmdname, const char *fmt, ...);
+
+Handling multiple encodings
+---
+
+    WORK IN PROGRESS
+
+Allocating memory
+---
+
+Modules data types should try to use `RedisModule_Alloc()` functions family
+in order to allocate, reallocate and release heap memory used to implement the native data structures (see the other Redis Modules documentation for detailed information).
+
+This is not just useful in order for Redis to be able to account for the memory used by the module, but there are also more advantages:
+
+* Redis uses the `jemalloc` allcator, that often prevents fragmentation problems that could be caused by using the libc allocator.
+* When loading strings from the RDB file, the native types API is able to return strings allocated directly with `RedisModule_Alloc()`, so that the module can directly link this memory into the data structure representation, avoiding an useless copy of the data.
+
+Even if you are using external libraries implementing your data structures, the
+allocation functions provided by the module API is exactly compatible with
+`malloc()`, `realloc()`, `free()` and `strdup()`, so converting the libraries
+in order to use these functions should be trivial.
+
+In case you have an external library that uses libc `malloc()`, and you want
+to avoid replacing manually all the calls with the Redis Modules API calls,
+an approach could be to use simple macros in order to replace the libc calls
+with the Redis API calls. Something like this could work:
+
+    #define malloc RedisModule_Alloc
+    #define realloc RedisModule_Realloc
+    #define free RedisModule_Free
+    #define strdup RedisModule_Strdup
+
+However take in mind that mixing libc calls with Redis API calls will result
+into troubles and crashes, so if you replace calls using macros, you need to
+make sure that all the calls are correctly replaced, and that the code with
+the substituted calls will never, for example, attempt to call
+`RedisModule_Free()` with a pointer allocated using libc `malloc()`.
diff --git a/src/modules/hellotype.c b/src/modules/hellotype.c
new file mode 100644
index 000000000..a9c2d20fc
--- /dev/null
+++ b/src/modules/hellotype.c
@@ -0,0 +1,259 @@
+/* This file implements a new module native data type called "HELLOTYPE".
+ * The data structure implemented is a very simple ordered linked list of
+ * 64 bit integers, in order to have something that is real world enough, but
+ * at the same time, extremely simple to understand, to show how the API
+ * works, how a new data type is created, and how to write basic methods
+ * for RDB loading, saving and AOF rewriting.
+ *
+ * ------------------------------------------------------------------------------
+ *
+ * Copyright (c) 2016, 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 "../redismodule.h"
+#include <stdio.h>
+#include <stdlib.h>
+#include <ctype.h>
+#include <string.h>
+#include <stdint.h>
+
+static RedisModuleType *HelloType;
+
+/* ========================== Internal data structure  =======================
+ * This is just a linked list of 64 bit integers where elements are inserted
+ * in-place, so it's ordered. There is no pop/push operation but just insert
+ * because it is enough to show the implementation of new data types without
+ * making things complex. */
+
+struct HelloTypeNode {
+    int64_t value;
+    struct HelloTypeNode *next;
+};
+
+struct HelloTypeObject {
+    struct HelloTypeNode *head;
+    size_t len; /* Number of elements added. */
+};
+
+struct HelloTypeObject *createHelloTypeObject(void) {
+    struct HelloTypeObject *o;
+    o = RedisModule_Alloc(sizeof(*o));
+    o->head = NULL;
+    o->len = 0;
+    return o;
+}
+
+void HelloTypeInsert(struct HelloTypeObject *o, int64_t ele) {
+    struct HelloTypeNode *next = o->head, *newnode, *prev = NULL;
+
+    while(next && next->value < ele) {
+        prev = next;
+        next = next->next;
+    }
+    newnode = RedisModule_Alloc(sizeof(*newnode));
+    newnode->value = ele;
+    newnode->next = next;
+    if (prev) {
+        prev->next = newnode;
+    } else {
+        o->head = newnode;
+    }
+    o->len++;
+}
+
+void HelloTypeReleaseObject(struct HelloTypeObject *o) {
+    struct HelloTypeNode *cur, *next;
+    cur = o->head;
+    while(cur) {
+        next = cur->next;
+        RedisModule_Free(cur);
+        cur = next;
+    }
+    RedisModule_Free(o);
+}
+
+/* ========================= "hellotype" type commands ======================= */
+
+/* HELLOTYPE.INSERT key value */
+int HelloTypeInsert_RedisCommand(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
+    RedisModule_AutoMemory(ctx); /* Use automatic memory management. */
+
+    if (argc != 3) return RedisModule_WrongArity(ctx);
+    RedisModuleKey *key = RedisModule_OpenKey(ctx,argv[1],
+        REDISMODULE_READ|REDISMODULE_WRITE);
+    int type = RedisModule_KeyType(key);
+    if (type != REDISMODULE_KEYTYPE_EMPTY &&
+        RedisModule_ModuleTypeGetType(key) != HelloType)
+    {
+        return RedisModule_ReplyWithError(ctx,REDISMODULE_ERRORMSG_WRONGTYPE);
+    }
+
+    long long value;
+    if ((RedisModule_StringToLongLong(argv[2],&value) != REDISMODULE_OK)) {
+        return RedisModule_ReplyWithError(ctx,"ERR invalid value: must be a signed 64 bit integer");
+    }
+
+    /* Create an empty value object if the key is currently empty. */
+    struct HelloTypeObject *hto;
+    if (type == REDISMODULE_KEYTYPE_EMPTY) {
+        hto = createHelloTypeObject();
+        RedisModule_ModuleTypeSetValue(key,HelloType,hto);
+    } else {
+        hto = RedisModule_ModuleTypeGetValue(key);
+    }
+
+    /* Insert the new element. */
+    HelloTypeInsert(hto,value);
+
+    RedisModule_ReplyWithLongLong(ctx,hto->len);
+    RedisModule_ReplicateVerbatim(ctx);
+    return REDISMODULE_OK;
+}
+
+/* HELLOTYPE.RANGE key first count */
+int HelloTypeRange_RedisCommand(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
+    RedisModule_AutoMemory(ctx); /* Use automatic memory management. */
+
+    if (argc != 4) return RedisModule_WrongArity(ctx);
+    RedisModuleKey *key = RedisModule_OpenKey(ctx,argv[1],
+        REDISMODULE_READ|REDISMODULE_WRITE);
+    int type = RedisModule_KeyType(key);
+    if (type != REDISMODULE_KEYTYPE_EMPTY &&
+        RedisModule_ModuleTypeGetType(key) != HelloType)
+    {
+        return RedisModule_ReplyWithError(ctx,REDISMODULE_ERRORMSG_WRONGTYPE);
+    }
+
+    long long first, count;
+    if (RedisModule_StringToLongLong(argv[2],&first) != REDISMODULE_OK ||
+        RedisModule_StringToLongLong(argv[3],&count) != REDISMODULE_OK ||
+        first < 0 || count < 0)
+    {
+        return RedisModule_ReplyWithError(ctx,
+            "ERR invalid first or count parameters");
+    }
+
+    struct HelloTypeObject *hto = RedisModule_ModuleTypeGetValue(key);
+    struct HelloTypeNode *node = hto ? hto->head : NULL;
+    RedisModule_ReplyWithArray(ctx,REDISMODULE_POSTPONED_ARRAY_LEN);
+    long long arraylen = 0;
+    while(node && count--) {
+        RedisModule_ReplyWithLongLong(ctx,node->value);
+        arraylen++;
+        node = node->next;
+    }
+    RedisModule_ReplySetArrayLength(ctx,arraylen);
+    return REDISMODULE_OK;
+}
+
+/* HELLOTYPE.LEN key */
+int HelloTypeLen_RedisCommand(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
+    RedisModule_AutoMemory(ctx); /* Use automatic memory management. */
+
+    if (argc != 2) return RedisModule_WrongArity(ctx);
+    RedisModuleKey *key = RedisModule_OpenKey(ctx,argv[1],
+        REDISMODULE_READ|REDISMODULE_WRITE);
+    int type = RedisModule_KeyType(key);
+    if (type != REDISMODULE_KEYTYPE_EMPTY &&
+        RedisModule_ModuleTypeGetType(key) != HelloType)
+    {
+        return RedisModule_ReplyWithError(ctx,REDISMODULE_ERRORMSG_WRONGTYPE);
+    }
+
+    struct HelloTypeObject *hto = RedisModule_ModuleTypeGetValue(key);
+    RedisModule_ReplyWithLongLong(ctx,hto ? hto->len : 0);
+    return REDISMODULE_OK;
+}
+
+
+/* ========================== "hellotype" type methods ======================= */
+
+void *HelloTypeRdbLoad(RedisModuleIO *rdb, int encver) {
+    if (encver != 0) {
+        /* RedisModule_Log("warning","Can't load data with version %d", encver);*/
+        return NULL;
+    }
+    uint64_t elements = RedisModule_LoadUnsigned(rdb);
+    struct HelloTypeObject *hto = createHelloTypeObject();
+    while(elements--) {
+        int64_t ele = RedisModule_LoadSigned(rdb);
+        HelloTypeInsert(hto,ele);
+    }
+    return hto;
+}
+
+void HelloTypeRdbSave(RedisModuleIO *rdb, void *value) {
+    struct HelloTypeObject *hto = value;
+    struct HelloTypeNode *node = hto->head;
+    RedisModule_SaveUnsigned(rdb,hto->len);
+    while(node) {
+        RedisModule_SaveSigned(rdb,node->value);
+        node = node->next;
+    }
+}
+
+void HelloTypeAofRewrite(RedisModuleIO *aof, RedisModuleString *key, void *value) {
+    struct HelloTypeObject *hto = value;
+    struct HelloTypeNode *node = hto->head;
+    while(node) {
+        RedisModule_EmitAOF(aof,"HELLOTYPE.INSERT","sl",key,node->value);
+        node = node->next;
+    }
+}
+
+void HelloTypeDigest(RedisModuleDigest *digest, void *value) {
+    /* TODO: The DIGEST module interface is yet not implemented. */
+}
+
+void HelloTypeFree(void *value) {
+    HelloTypeReleaseObject(value);
+}
+
+/* This function must be present on each Redis module. It is used in order to
+ * register the commands into the Redis server. */
+int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
+    if (RedisModule_Init(ctx,"hellotype",1,REDISMODULE_APIVER_1)
+        == REDISMODULE_ERR) return REDISMODULE_ERR;
+
+    HelloType = RedisModule_CreateDataType(ctx,"hellotype",0,HelloTypeRdbLoad,HelloTypeRdbSave,HelloTypeAofRewrite,HelloTypeDigest,HelloTypeFree);
+    if (HelloType == NULL) return REDISMODULE_ERR;
+
+    if (RedisModule_CreateCommand(ctx,"hellotype.insert",
+        HelloTypeInsert_RedisCommand,"write deny-oom",1,1,1) == REDISMODULE_ERR)
+        return REDISMODULE_ERR;
+
+    if (RedisModule_CreateCommand(ctx,"hellotype.range",
+        HelloTypeRange_RedisCommand,"readonly",1,1,1) == REDISMODULE_ERR)
+        return REDISMODULE_ERR;
+
+    if (RedisModule_CreateCommand(ctx,"hellotype.len",
+        HelloTypeLen_RedisCommand,"readonly",1,1,1) == REDISMODULE_ERR)
+        return REDISMODULE_ERR;
+
+    return REDISMODULE_OK;
+}
diff --git a/src/modules/helloworld.c b/src/modules/helloworld.c
index 3734503bb..8d657a52b 100644
--- a/src/modules/helloworld.c
+++ b/src/modules/helloworld.c
@@ -1,7 +1,44 @@
+/* Helloworld module -- A few examples of the Redis Modules API in the form
+ * of commands showing how to accomplish common tasks.
+ *
+ * This module does not do anything useful, if not for a few commands. The
+ * examples are designed in order to show the API.
+ *
+ * ------------------------------------------------------------------------------
+ *
+ * Copyright (c) 2016, 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 "../redismodule.h"
 #include <stdio.h>
 #include <stdlib.h>
 #include <ctype.h>
+#include <string.h>
 
 /* HELLO.SIMPLE is among the simplest commands you can implement.
  * It just returns the currently selected DB id, a functionality which is
@@ -448,12 +485,70 @@ int HelloHCopy_RedisCommand(RedisModuleCtx *ctx, RedisModuleString **argv, int a
     return REDISMODULE_OK;
 }
 
+/* HELLO.LEFTPAD str len ch
+ * This is an implementation of the infamous LEFTPAD function, that
+ * was at the center of an issue with the npm modules system in March 2016.
+ *
+ * LEFTPAD is a good example of using a Redis Modules API called
+ * "pool allocator", that was a famous way to allocate memory in yet another
+ * open source project, the Apache web server.
+ *
+ * The concept is very simple: there is memory that is useful to allocate
+ * only in the context of serving a request, and must be freed anyway when
+ * the callback implementing the command returns. So in that case the module
+ * does not need to retain a reference to these allocations, it is just
+ * required to free the memory before returning. When this is the case the
+ * module can call RedisModule_PoolAlloc() instead, that works like malloc()
+ * but will automatically free the memory when the module callback returns.
+ *
+ * Note that PoolAlloc() does not necessarily require AutoMemory to be
+ * active. */
+int HelloLeftPad_RedisCommand(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
+    RedisModule_AutoMemory(ctx); /* Use automatic memory management. */
+    long long padlen;
+
+    if (argc != 4) return RedisModule_WrongArity(ctx);
+
+    if ((RedisModule_StringToLongLong(argv[2],&padlen) != REDISMODULE_OK) ||
+        (padlen< 0)) {
+        return RedisModule_ReplyWithError(ctx,"ERR invalid padding length");
+    }
+    size_t strlen, chlen;
+    const char *str = RedisModule_StringPtrLen(argv[1], &strlen);
+    const char *ch = RedisModule_StringPtrLen(argv[3], &chlen);
+
+    /* If the string is already larger than the target len, just return
+     * the string itself. */
+    if (strlen >= padlen)
+        return RedisModule_ReplyWithString(ctx,argv[1]);
+
+    /* Padding must be a single character in this simple implementation. */
+    if (chlen != 1)
+        return RedisModule_ReplyWithError(ctx,
+            "ERR padding must be a single char");
+
+    /* Here we use our pool allocator, for our throw-away allocation. */
+    padlen -= strlen;
+    char *buf = RedisModule_PoolAlloc(ctx,padlen+strlen);
+    for (size_t j = 0; j < padlen; j++) buf[j] = *ch;
+    memcpy(buf+padlen,str,strlen);
+
+    RedisModule_ReplyWithStringBuffer(ctx,buf,padlen+strlen);
+    return REDISMODULE_OK;
+}
+
 /* This function must be present on each Redis module. It is used in order to
  * register the commands into the Redis server. */
-int RedisModule_OnLoad(RedisModuleCtx *ctx) {
+int RedisModule_OnLoad(RedisModuleCtx *ctx, RedisModuleString **argv, int argc) {
     if (RedisModule_Init(ctx,"helloworld",1,REDISMODULE_APIVER_1)
         == REDISMODULE_ERR) return REDISMODULE_ERR;
 
+    /* Log the list of parameters passing loading the module. */
+    for (int j = 0; j < argc; j++) {
+        const char *s = RedisModule_StringPtrLen(argv[j],NULL);
+        printf("Module loaded with ARGV[%d] = %s\n", j, s);
+    }
+
     if (RedisModule_CreateCommand(ctx,"hello.simple",
         HelloSimple_RedisCommand,"readonly",0,0,0) == REDISMODULE_ERR)
         return REDISMODULE_ERR;
@@ -515,5 +610,9 @@ int RedisModule_OnLoad(RedisModuleCtx *ctx) {
         HelloHCopy_RedisCommand,"write deny-oom",1,1,1) == REDISMODULE_ERR)
         return REDISMODULE_ERR;
 
+    if (RedisModule_CreateCommand(ctx,"hello.leftpad",
+        HelloLeftPad_RedisCommand,"",1,1,1) == REDISMODULE_ERR)
+        return REDISMODULE_ERR;
+
     return REDISMODULE_OK;
 }
diff --git a/src/networking.c b/src/networking.c
index d50d2c852..242022a03 100644
--- a/src/networking.c
+++ b/src/networking.c
@@ -1605,7 +1605,7 @@ void clientCommand(client *c) {
         pauseClients(duration);
         addReply(c,shared.ok);
     } else {
-        addReplyError(c, "Syntax error, try CLIENT (LIST | KILL ip:port | GETNAME | SETNAME connection-name)");
+        addReplyError(c, "Syntax error, try CLIENT (LIST | KILL | GETNAME | SETNAME | PAUSE | REPLY)");
     }
 }
 
diff --git a/src/object.c b/src/object.c
index 167290d49..447e5fc30 100644
--- a/src/object.c
+++ b/src/object.c
@@ -97,7 +97,7 @@ robj *createEmbeddedStringObject(const char *ptr, size_t len) {
 }
 
 /* Create a string object with EMBSTR encoding if it is smaller than
- * REIDS_ENCODING_EMBSTR_SIZE_LIMIT, otherwise the RAW encoding is
+ * OBJ_ENCODING_EMBSTR_SIZE_LIMIT, otherwise the RAW encoding is
  * used.
  *
  * The current limit of 39 is chosen so that the biggest string object
@@ -147,7 +147,7 @@ robj *createStringObjectFromLongDouble(long double value, int humanfriendly) {
  * will always result in a fresh object that is unshared (refcount == 1).
  *
  * The resulting object always has refcount set to 1. */
-robj *dupStringObject(robj *o) {
+robj *dupStringObject(const robj *o) {
     robj *d;
 
     serverAssert(o->type == OBJ_STRING);
@@ -221,6 +221,13 @@ robj *createZsetZiplistObject(void) {
     return o;
 }
 
+robj *createModuleObject(moduleType *mt, void *value) {
+    moduleValue *mv = zmalloc(sizeof(*mv));
+    mv->type = mt;
+    mv->value = value;
+    return createObject(OBJ_MODULE,mv);
+}
+
 void freeStringObject(robj *o) {
     if (o->encoding == OBJ_ENCODING_RAW) {
         sdsfree(o->ptr);
@@ -281,6 +288,12 @@ void freeHashObject(robj *o) {
     }
 }
 
+void freeModuleObject(robj *o) {
+    moduleValue *mv = o->ptr;
+    mv->type->free(mv->value);
+    zfree(mv);
+}
+
 void incrRefCount(robj *o) {
     if (o->refcount != OBJ_SHARED_REFCOUNT) o->refcount++;
 }
@@ -293,6 +306,7 @@ void decrRefCount(robj *o) {
         case OBJ_SET: freeSetObject(o); break;
         case OBJ_ZSET: freeZsetObject(o); break;
         case OBJ_HASH: freeHashObject(o); break;
+        case OBJ_MODULE: freeModuleObject(o); break;
         default: serverPanic("Unknown object type"); break;
         }
         zfree(o);
@@ -371,10 +385,10 @@ robj *tryObjectEncoding(robj *o) {
      if (o->refcount > 1) return o;
 
     /* Check if we can represent this string as a long integer.
-     * Note that we are sure that a string larger than 21 chars is not
+     * Note that we are sure that a string larger than 20 chars is not
      * representable as a 32 nor 64 bit integer. */
     len = sdslen(s);
-    if (len <= 21 && string2l(s,len,&value)) {
+    if (len <= 20 && string2l(s,len,&value)) {
         /* This object is encodable as a long. Try to use a shared object.
          * Note that we avoid using shared integers when maxmemory is used
          * because every object needs to have a private LRU field for the LRU
@@ -525,7 +539,7 @@ size_t stringObjectLen(robj *o) {
     }
 }
 
-int getDoubleFromObject(robj *o, double *target) {
+int getDoubleFromObject(const robj *o, double *target) {
     double value;
     char *eptr;
 
@@ -536,7 +550,7 @@ int getDoubleFromObject(robj *o, double *target) {
         if (sdsEncodedObject(o)) {
             errno = 0;
             value = strtod(o->ptr, &eptr);
-            if (isspace(((char*)o->ptr)[0]) ||
+            if (isspace(((const char*)o->ptr)[0]) ||
                 eptr[0] != '\0' ||
                 (errno == ERANGE &&
                     (value == HUGE_VAL || value == -HUGE_VAL || value == 0)) ||
diff --git a/src/quicklist.c b/src/quicklist.c
index be02e3276..adf9ba1de 100644
--- a/src/quicklist.c
+++ b/src/quicklist.c
@@ -149,7 +149,7 @@ REDIS_STATIC quicklistNode *quicklistCreateNode(void) {
 }
 
 /* Return cached quicklist count */
-unsigned int quicklistCount(quicklist *ql) { return ql->count; }
+unsigned int quicklistCount(const quicklist *ql) { return ql->count; }
 
 /* Free entire quicklist. */
 void quicklistRelease(quicklist *quicklist) {
diff --git a/src/quicklist.h b/src/quicklist.h
index 5c9530ccd..8f3875900 100644
--- a/src/quicklist.h
+++ b/src/quicklist.h
@@ -92,8 +92,8 @@ typedef struct quicklistEntry {
     quicklistNode *node;
     unsigned char *zi;
     unsigned char *value;
-    unsigned int sz;
     long long longval;
+    unsigned int sz;
     int offset;
 } quicklistEntry;
 
@@ -154,7 +154,7 @@ int quicklistPopCustom(quicklist *quicklist, int where, unsigned char **data,
                        void *(*saver)(unsigned char *data, unsigned int sz));
 int quicklistPop(quicklist *quicklist, int where, unsigned char **data,
                  unsigned int *sz, long long *slong);
-unsigned int quicklistCount(quicklist *ql);
+unsigned int quicklistCount(const quicklist *ql);
 int quicklistCompare(unsigned char *p1, unsigned char *p2, int p2_len);
 size_t quicklistGetLzf(const quicklistNode *node, void **data);
 
diff --git a/src/rdb.c b/src/rdb.c
index 57b759278..6d29f80ce 100644
--- a/src/rdb.c
+++ b/src/rdb.c
@@ -41,11 +41,6 @@
 #include <sys/stat.h>
 #include <sys/param.h>
 
-#define RDB_LOAD_NONE   0
-#define RDB_LOAD_ENC    (1<<0)
-#define RDB_LOAD_PLAIN  (1<<1)
-#define RDB_LOAD_SDS    (1<<2)
-
 #define rdbExitReportCorruptRDB(reason) rdbCheckThenExit(reason, __LINE__);
 
 void rdbCheckThenExit(char *reason, int where) {
@@ -95,7 +90,7 @@ long long rdbLoadMillisecondTime(rio *rdb) {
 /* Saves an encoded length. The first two bits in the first byte are used to
  * hold the encoding type. See the RDB_* definitions for more information
  * on the types of encoding. */
-int rdbSaveLen(rio *rdb, uint32_t len) {
+int rdbSaveLen(rio *rdb, uint64_t len) {
     unsigned char buf[2];
     size_t nwritten;
 
@@ -110,44 +105,78 @@ int rdbSaveLen(rio *rdb, uint32_t len) {
         buf[1] = len&0xFF;
         if (rdbWriteRaw(rdb,buf,2) == -1) return -1;
         nwritten = 2;
-    } else {
+    } else if (len <= UINT32_MAX) {
         /* Save a 32 bit len */
-        buf[0] = (RDB_32BITLEN<<6);
+        buf[0] = RDB_32BITLEN;
         if (rdbWriteRaw(rdb,buf,1) == -1) return -1;
-        len = htonl(len);
-        if (rdbWriteRaw(rdb,&len,4) == -1) return -1;
+        uint32_t len32 = htonl(len);
+        if (rdbWriteRaw(rdb,&len32,4) == -1) return -1;
         nwritten = 1+4;
+    } else {
+        /* Save a 64 bit len */
+        buf[0] = RDB_64BITLEN;
+        if (rdbWriteRaw(rdb,buf,1) == -1) return -1;
+        len = htonu64(len);
+        if (rdbWriteRaw(rdb,&len,8) == -1) return -1;
+        nwritten = 1+8;
     }
     return nwritten;
 }
 
-/* Load an encoded length. The "isencoded" argument is set to 1 if the length
- * is not actually a length but an "encoding type". See the RDB_ENC_*
- * definitions in rdb.h for more information. */
-uint32_t rdbLoadLen(rio *rdb, int *isencoded) {
+
+/* Load an encoded length. If the loaded length is a normal length as stored
+ * with rdbSaveLen(), the read length is set to '*lenptr'. If instead the
+ * loaded length describes a special encoding that follows, then '*isencoded'
+ * is set to 1 and the encoding format is stored at '*lenptr'.
+ *
+ * See the RDB_ENC_* definitions in rdb.h for more information on special
+ * encodings.
+ *
+ * The function returns -1 on error, 0 on success. */
+int rdbLoadLenByRef(rio *rdb, int *isencoded, uint64_t *lenptr) {
     unsigned char buf[2];
-    uint32_t len;
     int type;
 
     if (isencoded) *isencoded = 0;
-    if (rioRead(rdb,buf,1) == 0) return RDB_LENERR;
+    if (rioRead(rdb,buf,1) == 0) return -1;
     type = (buf[0]&0xC0)>>6;
     if (type == RDB_ENCVAL) {
         /* Read a 6 bit encoding type. */
         if (isencoded) *isencoded = 1;
-        return buf[0]&0x3F;
+        *lenptr = buf[0]&0x3F;
     } else if (type == RDB_6BITLEN) {
         /* Read a 6 bit len. */
-        return buf[0]&0x3F;
+        *lenptr = buf[0]&0x3F;
     } else if (type == RDB_14BITLEN) {
         /* Read a 14 bit len. */
-        if (rioRead(rdb,buf+1,1) == 0) return RDB_LENERR;
-        return ((buf[0]&0x3F)<<8)|buf[1];
-    } else {
+        if (rioRead(rdb,buf+1,1) == 0) return -1;
+        *lenptr = ((buf[0]&0x3F)<<8)|buf[1];
+    } else if (buf[0] == RDB_32BITLEN) {
         /* Read a 32 bit len. */
-        if (rioRead(rdb,&len,4) == 0) return RDB_LENERR;
-        return ntohl(len);
+        uint32_t len;
+        if (rioRead(rdb,&len,4) == 0) return -1;
+        *lenptr = ntohl(len);
+    } else if (buf[0] == RDB_64BITLEN) {
+        /* Read a 64 bit len. */
+        uint64_t len;
+        if (rioRead(rdb,&len,8) == 0) return -1;
+        *lenptr = ntohu64(len);
+    } else {
+        rdbExitReportCorruptRDB("Unknown length encoding in rdbLoadLen()");
+        return -1; /* Never reached. */
     }
+    return 0;
+}
+
+/* This is like rdbLoadLenByRef() but directly returns the value read
+ * from the RDB stream, signaling an error by returning RDB_LENERR
+ * (since it is a too large count to be applicable in any Redis data
+ * structure). */
+uint64_t rdbLoadLen(rio *rdb, int *isencoded) {
+    uint64_t len;
+
+    if (rdbLoadLenByRef(rdb,isencoded,&len) == -1) return RDB_LENERR;
+    return len;
 }
 
 /* Encodes the "value" argument as integer when it fits in the supported ranges
@@ -179,7 +208,7 @@ int rdbEncodeInteger(long long value, unsigned char *enc) {
 /* Loads an integer-encoded object with the specified encoding type "enctype".
  * The returned value changes according to the flags, see
  * rdbGenerincLoadStringObject() for more info. */
-void *rdbLoadIntegerObject(rio *rdb, int enctype, int flags) {
+void *rdbLoadIntegerObject(rio *rdb, int enctype, int flags, size_t *lenptr) {
     int plain = flags & RDB_LOAD_PLAIN;
     int sds = flags & RDB_LOAD_SDS;
     int encode = flags & RDB_LOAD_ENC;
@@ -206,6 +235,7 @@ void *rdbLoadIntegerObject(rio *rdb, int enctype, int flags) {
     if (plain || sds) {
         char buf[LONG_STR_SIZE], *p;
         int len = ll2string(buf,sizeof(buf),val);
+        if (lenptr) *lenptr = len;
         p = plain ? zmalloc(len) : sdsnewlen(NULL,len);
         memcpy(p,buf,len);
         return p;
@@ -281,10 +311,10 @@ ssize_t rdbSaveLzfStringObject(rio *rdb, unsigned char *s, size_t len) {
 /* Load an LZF compressed string in RDB format. The returned value
  * changes according to 'flags'. For more info check the
  * rdbGenericLoadStringObject() function. */
-void *rdbLoadLzfStringObject(rio *rdb, int flags) {
+void *rdbLoadLzfStringObject(rio *rdb, int flags, size_t *lenptr) {
     int plain = flags & RDB_LOAD_PLAIN;
     int sds = flags & RDB_LOAD_SDS;
-    unsigned int len, clen;
+    uint64_t len, clen;
     unsigned char *c = NULL;
     char *val = NULL;
 
@@ -295,6 +325,7 @@ void *rdbLoadLzfStringObject(rio *rdb, int flags) {
     /* Allocate our target according to the uncompressed size. */
     if (plain) {
         val = zmalloc(len);
+        if (lenptr) *lenptr = len;
     } else {
         val = sdsnewlen(NULL,len);
     }
@@ -393,13 +424,15 @@ int rdbSaveStringObject(rio *rdb, robj *obj) {
  * RDB_LOAD_PLAIN: Return a plain string allocated with zmalloc()
  *                 instead of a Redis object with an sds in it.
  * RDB_LOAD_SDS: Return an SDS string instead of a Redis object.
-*/
-void *rdbGenericLoadStringObject(rio *rdb, int flags) {
+ *
+ * On I/O error NULL is returned.
+ */
+void *rdbGenericLoadStringObject(rio *rdb, int flags, size_t *lenptr) {
     int encode = flags & RDB_LOAD_ENC;
     int plain = flags & RDB_LOAD_PLAIN;
     int sds = flags & RDB_LOAD_SDS;
     int isencoded;
-    uint32_t len;
+    uint64_t len;
 
     len = rdbLoadLen(rdb,&isencoded);
     if (isencoded) {
@@ -407,9 +440,9 @@ void *rdbGenericLoadStringObject(rio *rdb, int flags) {
         case RDB_ENC_INT8:
         case RDB_ENC_INT16:
         case RDB_ENC_INT32:
-            return rdbLoadIntegerObject(rdb,len,flags);
+            return rdbLoadIntegerObject(rdb,len,flags,lenptr);
         case RDB_ENC_LZF:
-            return rdbLoadLzfStringObject(rdb,flags);
+            return rdbLoadLzfStringObject(rdb,flags,lenptr);
         default:
             rdbExitReportCorruptRDB("Unknown RDB encoding type");
         }
@@ -418,6 +451,7 @@ void *rdbGenericLoadStringObject(rio *rdb, int flags) {
     if (len == RDB_LENERR) return NULL;
     if (plain || sds) {
         void *buf = plain ? zmalloc(len) : sdsnewlen(NULL,len);
+        if (lenptr) *lenptr = len;
         if (len && rioRead(rdb,buf,len) == 0) {
             if (plain)
                 zfree(buf);
@@ -438,11 +472,11 @@ void *rdbGenericLoadStringObject(rio *rdb, int flags) {
 }
 
 robj *rdbLoadStringObject(rio *rdb) {
-    return rdbGenericLoadStringObject(rdb,RDB_LOAD_NONE);
+    return rdbGenericLoadStringObject(rdb,RDB_LOAD_NONE,NULL);
 }
 
 robj *rdbLoadEncodedStringObject(rio *rdb) {
-    return rdbGenericLoadStringObject(rdb,RDB_LOAD_ENC);
+    return rdbGenericLoadStringObject(rdb,RDB_LOAD_ENC,NULL);
 }
 
 /* Save a double value. Doubles are saved as strings prefixed by an unsigned
@@ -505,6 +539,24 @@ int rdbLoadDoubleValue(rio *rdb, double *val) {
     }
 }
 
+/* Saves a double for RDB 8 or greater, where IE754 binary64 format is assumed.
+ * We just make sure the integer is always stored in little endian, otherwise
+ * the value is copied verbatim from memory to disk.
+ *
+ * Return -1 on error, the size of the serialized value on success. */
+int rdbSaveBinaryDoubleValue(rio *rdb, double val) {
+    memrev64ifbe(&val);
+    return rdbWriteRaw(rdb,&val,8);
+}
+
+/* Loads a double from RDB 8 or greater. See rdbSaveBinaryDoubleValue() for
+ * more info. On error -1 is returned, otherwise 0. */
+int rdbLoadBinaryDoubleValue(rio *rdb, double *val) {
+    if (rioRead(rdb,val,8) == 0) return -1;
+    memrev64ifbe(val);
+    return 0;
+}
+
 /* Save the object type of object "o". */
 int rdbSaveObjectType(rio *rdb, robj *o) {
     switch (o->type) {
@@ -526,7 +578,7 @@ int rdbSaveObjectType(rio *rdb, robj *o) {
         if (o->encoding == OBJ_ENCODING_ZIPLIST)
             return rdbSaveType(rdb,RDB_TYPE_ZSET_ZIPLIST);
         else if (o->encoding == OBJ_ENCODING_SKIPLIST)
-            return rdbSaveType(rdb,RDB_TYPE_ZSET);
+            return rdbSaveType(rdb,RDB_TYPE_ZSET_2);
         else
             serverPanic("Unknown sorted set encoding");
     case OBJ_HASH:
@@ -536,6 +588,8 @@ int rdbSaveObjectType(rio *rdb, robj *o) {
             return rdbSaveType(rdb,RDB_TYPE_HASH);
         else
             serverPanic("Unknown hash encoding");
+    case OBJ_MODULE:
+        return rdbSaveType(rdb,RDB_TYPE_MODULE);
     default:
         serverPanic("Unknown object type");
     }
@@ -629,7 +683,7 @@ ssize_t rdbSaveObject(rio *rdb, robj *o) {
                 if ((n = rdbSaveRawString(rdb,(unsigned char*)ele,sdslen(ele)))
                     == -1) return -1;
                 nwritten += n;
-                if ((n = rdbSaveDoubleValue(rdb,*score)) == -1) return -1;
+                if ((n = rdbSaveBinaryDoubleValue(rdb,*score)) == -1) return -1;
                 nwritten += n;
             }
             dictReleaseIterator(di);
@@ -667,6 +721,22 @@ ssize_t rdbSaveObject(rio *rdb, robj *o) {
             serverPanic("Unknown hash encoding");
         }
 
+    } else if (o->type == OBJ_MODULE) {
+        /* Save a module-specific value. */
+        RedisModuleIO io;
+        moduleValue *mv = o->ptr;
+        moduleType *mt = mv->type;
+        moduleInitIOContext(io,mt,rdb);
+
+        /* Write the "module" identifier as prefix, so that we'll be able
+         * to call the right module during loading. */
+        int retval = rdbSaveLen(rdb,mt->id);
+        if (retval == -1) return -1;
+        io.bytes += retval;
+
+        /* Then write the module-specific representation. */
+        mt->rdb_save(&io,mv->value);
+        return io.error ? -1 : (ssize_t)io.bytes;
     } else {
         serverPanic("Unknown object type");
     }
@@ -1005,8 +1075,8 @@ robj *rdbLoadObject(int rdbtype, rio *rdb) {
             long long llval;
             sds sdsele;
 
-            if ((sdsele = rdbGenericLoadStringObject(rdb,RDB_LOAD_SDS)) == NULL)
-                return NULL;
+            if ((sdsele = rdbGenericLoadStringObject(rdb,RDB_LOAD_SDS,NULL))
+                == NULL) return NULL;
 
             if (o->encoding == OBJ_ENCODING_INTSET) {
                 /* Fetch integer value from element. */
@@ -1026,7 +1096,7 @@ robj *rdbLoadObject(int rdbtype, rio *rdb) {
                 sdsfree(sdsele);
             }
         }
-    } else if (rdbtype == RDB_TYPE_ZSET) {
+    } else if (rdbtype == RDB_TYPE_ZSET_2 || rdbtype == RDB_TYPE_ZSET) {
         /* Read list/set value. */
         size_t zsetlen;
         size_t maxelelen = 0;
@@ -1042,9 +1112,14 @@ robj *rdbLoadObject(int rdbtype, rio *rdb) {
             double score;
             zskiplistNode *znode;
 
-            if ((sdsele = rdbGenericLoadStringObject(rdb,RDB_LOAD_SDS)) == NULL)
-                return NULL;
-            if (rdbLoadDoubleValue(rdb,&score) == -1) return NULL;
+            if ((sdsele = rdbGenericLoadStringObject(rdb,RDB_LOAD_SDS,NULL))
+                == NULL) return NULL;
+
+            if (rdbtype == RDB_TYPE_ZSET_2) {
+                if (rdbLoadBinaryDoubleValue(rdb,&score) == -1) return NULL;
+            } else {
+                if (rdbLoadDoubleValue(rdb,&score) == -1) return NULL;
+            }
 
             /* Don't care about integer-encoded strings. */
             if (sdslen(sdsele) > maxelelen) maxelelen = sdslen(sdsele);
@@ -1075,10 +1150,10 @@ robj *rdbLoadObject(int rdbtype, rio *rdb) {
         while (o->encoding == OBJ_ENCODING_ZIPLIST && len > 0) {
             len--;
             /* Load raw strings */
-            if ((field = rdbGenericLoadStringObject(rdb,RDB_LOAD_SDS)) == NULL)
-                return NULL;
-            if ((value = rdbGenericLoadStringObject(rdb,RDB_LOAD_SDS)) == NULL)
-                return NULL;
+            if ((field = rdbGenericLoadStringObject(rdb,RDB_LOAD_SDS,NULL))
+                == NULL) return NULL;
+            if ((value = rdbGenericLoadStringObject(rdb,RDB_LOAD_SDS,NULL))
+                == NULL) return NULL;
 
             /* Add pair to ziplist */
             o->ptr = ziplistPush(o->ptr, (unsigned char*)field,
@@ -1103,10 +1178,10 @@ robj *rdbLoadObject(int rdbtype, rio *rdb) {
         while (o->encoding == OBJ_ENCODING_HT && len > 0) {
             len--;
             /* Load encoded strings */
-            if ((field = rdbGenericLoadStringObject(rdb,RDB_LOAD_SDS)) == NULL)
-                return NULL;
-            if ((value = rdbGenericLoadStringObject(rdb,RDB_LOAD_SDS)) == NULL)
-                return NULL;
+            if ((field = rdbGenericLoadStringObject(rdb,RDB_LOAD_SDS,NULL))
+                == NULL) return NULL;
+            if ((value = rdbGenericLoadStringObject(rdb,RDB_LOAD_SDS,NULL))
+                == NULL) return NULL;
 
             /* Add pair to hash table */
             ret = dictAdd((dict*)o->ptr, field, value);
@@ -1124,7 +1199,8 @@ robj *rdbLoadObject(int rdbtype, rio *rdb) {
                             server.list_compress_depth);
 
         while (len--) {
-            unsigned char *zl = rdbGenericLoadStringObject(rdb,RDB_LOAD_PLAIN);
+            unsigned char *zl =
+                rdbGenericLoadStringObject(rdb,RDB_LOAD_PLAIN,NULL);
             if (zl == NULL) return NULL;
             quicklistAppendZiplist(o->ptr, zl);
         }
@@ -1134,7 +1210,8 @@ robj *rdbLoadObject(int rdbtype, rio *rdb) {
                rdbtype == RDB_TYPE_ZSET_ZIPLIST ||
                rdbtype == RDB_TYPE_HASH_ZIPLIST)
     {
-        unsigned char *encoded = rdbGenericLoadStringObject(rdb,RDB_LOAD_PLAIN);
+        unsigned char *encoded =
+            rdbGenericLoadStringObject(rdb,RDB_LOAD_PLAIN,NULL);
         if (encoded == NULL) return NULL;
         o = createObject(OBJ_STRING,encoded); /* Obj type fixed below. */
 
@@ -1201,6 +1278,27 @@ robj *rdbLoadObject(int rdbtype, rio *rdb) {
                 rdbExitReportCorruptRDB("Unknown encoding");
                 break;
         }
+    } else if (rdbtype == RDB_TYPE_MODULE) {
+        uint64_t moduleid = rdbLoadLen(rdb,NULL);
+        moduleType *mt = moduleTypeLookupModuleByID(moduleid);
+        char name[10];
+
+        if (mt == NULL) {
+            moduleTypeNameByID(name,moduleid);
+            serverLog(LL_WARNING,"The RDB file contains module data I can't load: no matching module '%s'", name);
+            exit(1);
+        }
+        RedisModuleIO io;
+        moduleInitIOContext(io,mt,rdb);
+        /* Call the rdb_load method of the module providing the 10 bit
+         * encoding version in the lower 10 bits of the module ID. */
+        void *ptr = mt->rdb_load(&io,moduleid&1023);
+        if (ptr == NULL) {
+            moduleTypeNameByID(name,moduleid);
+            serverLog(LL_WARNING,"The RDB file contains module data for the module type '%s', that the responsible module is not able to load. Check for modules log above for additional clues.", name);
+            exit(1);
+        }
+        o = createModuleObject(mt,ptr);
     } else {
         rdbExitReportCorruptRDB("Unknown object type");
     }
@@ -1255,7 +1353,7 @@ void rdbLoadProgressCallback(rio *r, const void *buf, size_t len) {
 }
 
 int rdbLoad(char *filename) {
-    uint32_t dbid;
+    uint64_t dbid;
     int type, rdbver;
     redisDb *db = server.db+0;
     char buf[1024];
@@ -1328,7 +1426,7 @@ int rdbLoad(char *filename) {
         } else if (type == RDB_OPCODE_RESIZEDB) {
             /* RESIZEDB: Hint about the size of the keys in the currently
              * selected data base, in order to avoid useless rehashing. */
-            uint32_t db_size, expires_size;
+            uint64_t db_size, expires_size;
             if ((db_size = rdbLoadLen(&rdb,NULL)) == RDB_LENERR)
                 goto eoferr;
             if ((expires_size = rdbLoadLen(&rdb,NULL)) == RDB_LENERR)
diff --git a/src/rdb.h b/src/rdb.h
index 48a064a19..a71ecb16e 100644
--- a/src/rdb.h
+++ b/src/rdb.h
@@ -38,16 +38,17 @@
 
 /* The current RDB version. When the format changes in a way that is no longer
  * backward compatible this number gets incremented. */
-#define RDB_VERSION 7
+#define RDB_VERSION 8
 
 /* 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
+ * 00|XXXXXX => if the two MSB are 00 the len is the 6 bits of this byte
+ * 01|XXXXXX XXXXXXXX =>  01, the len is 14 byes, 6 bits + 8 bits of next byte
+ * 10|000000 [32 bit integer] => A full 32 bit len in net byte order will follow
+ * 10|000001 [64 bit integer] => A full 64 bit len in net byte order will follow
+ * 11|OBKIND this means: specially encoded object will follow. The six bits
  *           number specify the kind of object that follows.
  *           See the RDB_ENC_* defines.
  *
@@ -55,12 +56,13 @@
  * values, will fit inside. */
 #define RDB_6BITLEN 0
 #define RDB_14BITLEN 1
-#define RDB_32BITLEN 2
+#define RDB_32BITLEN 0x80
+#define RDB_64BITLEN 0x81
 #define RDB_ENCVAL 3
-#define RDB_LENERR UINT_MAX
+#define RDB_LENERR UINT64_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
+ * set, the remaining six bits specify a special encoding for the object
  * accordingly to the following defines: */
 #define RDB_ENC_INT8 0        /* 8 bit signed integer */
 #define RDB_ENC_INT16 1       /* 16 bit signed integer */
@@ -74,6 +76,8 @@
 #define RDB_TYPE_SET    2
 #define RDB_TYPE_ZSET   3
 #define RDB_TYPE_HASH   4
+#define RDB_TYPE_ZSET_2 5 /* ZSET version 2 with doubles stored in binary. */
+#define RDB_TYPE_MODULE 6
 /* NOTE: WHEN ADDING NEW RDB TYPE, UPDATE rdbIsObjectType() BELOW */
 
 /* Object types for encoded objects. */
@@ -86,7 +90,7 @@
 /* NOTE: WHEN ADDING NEW RDB TYPE, UPDATE rdbIsObjectType() BELOW */
 
 /* Test if a type is an object type. */
-#define rdbIsObjectType(t) ((t >= 0 && t <= 4) || (t >= 9 && t <= 14))
+#define rdbIsObjectType(t) ((t >= 0 && t <= 6) || (t >= 9 && t <= 14))
 
 /* Special RDB opcodes (saved/loaded with rdbSaveType/rdbLoadType). */
 #define RDB_OPCODE_AUX        250
@@ -96,12 +100,19 @@
 #define RDB_OPCODE_SELECTDB   254
 #define RDB_OPCODE_EOF        255
 
+/* rdbLoad...() functions flags. */
+#define RDB_LOAD_NONE   0
+#define RDB_LOAD_ENC    (1<<0)
+#define RDB_LOAD_PLAIN  (1<<1)
+#define RDB_LOAD_SDS    (1<<2)
+
 int rdbSaveType(rio *rdb, unsigned char type);
 int rdbLoadType(rio *rdb);
 int rdbSaveTime(rio *rdb, time_t t);
 time_t rdbLoadTime(rio *rdb);
-int rdbSaveLen(rio *rdb, uint32_t len);
-uint32_t rdbLoadLen(rio *rdb, int *isencoded);
+int rdbSaveLen(rio *rdb, uint64_t len);
+uint64_t rdbLoadLen(rio *rdb, int *isencoded);
+int rdbLoadLenByRef(rio *rdb, int *isencoded, uint64_t *lenptr);
 int rdbSaveObjectType(rio *rdb, robj *o);
 int rdbLoadObjectType(rio *rdb);
 int rdbLoad(char *filename);
@@ -115,5 +126,10 @@ robj *rdbLoadObject(int type, rio *rdb);
 void backgroundSaveDoneHandler(int exitcode, int bysignal);
 int rdbSaveKeyValuePair(rio *rdb, robj *key, robj *val, long long expiretime, long long now);
 robj *rdbLoadStringObject(rio *rdb);
+int rdbSaveStringObject(rio *rdb, robj *obj);
+ssize_t rdbSaveRawString(rio *rdb, unsigned char *s, size_t len);
+void *rdbGenericLoadStringObject(rio *rdb, int flags, size_t *lenptr);
+int rdbSaveBinaryDoubleValue(rio *rdb, double val);
+int rdbLoadBinaryDoubleValue(rio *rdb, double *val);
 
 #endif
diff --git a/src/redis-check-rdb.c b/src/redis-check-rdb.c
index 7bb93b60b..0723d2af4 100644
--- a/src/redis-check-rdb.c
+++ b/src/redis-check-rdb.c
@@ -171,7 +171,7 @@ static int processTime(int type) {
     return 0;
 }
 
-static uint32_t loadLength(int *isencoded) {
+static uint64_t loadLength(int *isencoded) {
     unsigned char buf[2];
     uint32_t len;
     int type;
@@ -190,10 +190,16 @@ static uint32_t loadLength(int *isencoded) {
         /* Read a 14 bit len */
         if (!readBytes(buf+1,1)) return RDB_LENERR;
         return ((buf[0] & 0x3F) << 8) | buf[1];
-    } else {
+    } else if (buf[0] == RDB_32BITLEN) {
         /* Read a 32 bit len */
         if (!readBytes(&len, 4)) return RDB_LENERR;
-        return (unsigned int)ntohl(len);
+        return ntohl(len);
+    } else if (buf[0] == RDB_64BITLEN) {
+        /* Read a 64 bit len */
+        if (!readBytes(&len, 8)) return RDB_LENERR;
+        return ntohu64(len);
+    } else {
+        return RDB_LENERR;
     }
 }
 
@@ -230,7 +236,7 @@ static char *loadIntegerObject(int enctype) {
 }
 
 static char* loadLzfStringObject() {
-    unsigned int slen, clen;
+    uint64_t slen, clen;
     char *c, *s;
 
     if ((clen = loadLength(NULL)) == RDB_LENERR) return NULL;
@@ -254,9 +260,9 @@ static char* loadLzfStringObject() {
 
 /* returns NULL when not processable, char* when valid */
 static char* loadStringObject() {
-    uint32_t offset = CURR_OFFSET;
+    uint64_t offset = CURR_OFFSET;
+    uint64_t len;
     int isencoded;
-    uint32_t len;
 
     len = loadLength(&isencoded);
     if (isencoded) {
@@ -269,7 +275,8 @@ static char* loadStringObject() {
             return loadLzfStringObject();
         default:
             /* unknown encoding */
-            SHIFT_ERROR(offset, "Unknown string encoding (0x%02x)", len);
+            SHIFT_ERROR(offset, "Unknown string encoding (0x%02llx)",
+                (unsigned long long) len);
             return NULL;
         }
     }
@@ -344,8 +351,8 @@ static int processDoubleValue(double** store) {
 }
 
 static int loadPair(entry *e) {
-    uint32_t offset = CURR_OFFSET;
-    uint32_t i;
+    uint64_t offset = CURR_OFFSET;
+    uint64_t i;
 
     /* read key first */
     char *key;
@@ -356,7 +363,7 @@ static int loadPair(entry *e) {
         return 0;
     }
 
-    uint32_t length = 0;
+    uint64_t length = 0;
     if (e->type == RDB_TYPE_LIST ||
         e->type == RDB_TYPE_SET  ||
         e->type == RDB_TYPE_ZSET ||
@@ -384,7 +391,8 @@ static int loadPair(entry *e) {
         for (i = 0; i < length; i++) {
             offset = CURR_OFFSET;
             if (!processStringObject(NULL)) {
-                SHIFT_ERROR(offset, "Error reading element at index %d (length: %d)", i, length);
+                SHIFT_ERROR(offset, "Error reading element at index %llu (length: %llu)",
+                    (unsigned long long) i, (unsigned long long) length);
                 return 0;
             }
         }
@@ -393,12 +401,14 @@ static int loadPair(entry *e) {
         for (i = 0; i < length; i++) {
             offset = CURR_OFFSET;
             if (!processStringObject(NULL)) {
-                SHIFT_ERROR(offset, "Error reading element key at index %d (length: %d)", i, length);
+                SHIFT_ERROR(offset, "Error reading element key at index %llu (length: %llu)",
+                    (unsigned long long) i, (unsigned long long) length);
                 return 0;
             }
             offset = CURR_OFFSET;
             if (!processDoubleValue(NULL)) {
-                SHIFT_ERROR(offset, "Error reading element value at index %d (length: %d)", i, length);
+                SHIFT_ERROR(offset, "Error reading element value at index %llu (length: %llu)",
+                    (unsigned long long) i, (unsigned long long) length);
                 return 0;
             }
         }
@@ -407,12 +417,14 @@ static int loadPair(entry *e) {
         for (i = 0; i < length; i++) {
             offset = CURR_OFFSET;
             if (!processStringObject(NULL)) {
-                SHIFT_ERROR(offset, "Error reading element key at index %d (length: %d)", i, length);
+                SHIFT_ERROR(offset, "Error reading element key at index %llu (length: %llu)",
+                    (unsigned long long) i, (unsigned long long) length);
                 return 0;
             }
             offset = CURR_OFFSET;
             if (!processStringObject(NULL)) {
-                SHIFT_ERROR(offset, "Error reading element value at index %d (length: %d)", i, length);
+                SHIFT_ERROR(offset, "Error reading element value at index %llu (length: %llu)",
+                    (unsigned long long) i, (unsigned long long) length);
                 return 0;
             }
         }
@@ -428,7 +440,7 @@ static int loadPair(entry *e) {
 
 static entry loadEntry() {
     entry e = { NULL, -1, 0 };
-    uint32_t length, offset[4];
+    uint64_t length, offset[4];
 
     /* reset error container */
     errors.level = 0;
@@ -445,7 +457,8 @@ static entry loadEntry() {
             return e;
         }
         if (length > 63) {
-            SHIFT_ERROR(offset[1], "Database number out of range (%d)", length);
+            SHIFT_ERROR(offset[1], "Database number out of range (%llu)",
+                (unsigned long long) length);
             return e;
         }
     } else if (e.type == RDB_OPCODE_EOF) {
diff --git a/src/redis-cli.c b/src/redis-cli.c
index cf939c8c9..17fb53394 100644
--- a/src/redis-cli.c
+++ b/src/redis-cli.c
@@ -284,7 +284,6 @@ static void cliIntegrateHelp(void) {
                 break;
         }
         if (i != helpEntriesLen) continue;
-        printf("%s\n", cmdname);
 
         helpEntriesLen++;
         helpEntries = zrealloc(helpEntries,sizeof(helpEntry)*helpEntriesLen);
@@ -314,8 +313,6 @@ static void cliIntegrateHelp(void) {
         new->org = ch;
     }
     freeReplyObject(reply);
-
-    printf("%s\n", helpEntries[80].full);
 }
 
 /* Output command help to stdout. */
@@ -2594,13 +2591,16 @@ int main(int argc, char **argv) {
     else
         config.output = OUTPUT_STANDARD;
     config.mb_delim = sdsnew("\n");
-    cliInitHelp();
-    cliIntegrateHelp();
 
     firstarg = parseOptions(argc,argv);
     argc -= firstarg;
     argv += firstarg;
 
+    /* Initialize the help and, if possible, use the COMMAND command in order
+     * to retrieve missing entries. */
+    cliInitHelp();
+    cliIntegrateHelp();
+
     /* Latency mode */
     if (config.latency_mode) {
         if (cliConnect(0) == REDIS_ERR) exit(1);
diff --git a/src/redismodule.h b/src/redismodule.h
index e54825dc2..f1aaea49b 100644
--- a/src/redismodule.h
+++ b/src/redismodule.h
@@ -28,6 +28,7 @@
 #define REDISMODULE_KEYTYPE_HASH 3
 #define REDISMODULE_KEYTYPE_SET 4
 #define REDISMODULE_KEYTYPE_ZSET 5
+#define REDISMODULE_KEYTYPE_MODULE 6
 
 /* Reply types. */
 #define REDISMODULE_REPLY_UNKNOWN -1
@@ -78,14 +79,29 @@ typedef struct RedisModuleCtx RedisModuleCtx;
 typedef struct RedisModuleKey RedisModuleKey;
 typedef struct RedisModuleString RedisModuleString;
 typedef struct RedisModuleCallReply RedisModuleCallReply;
+typedef struct RedisModuleIO RedisModuleIO;
+typedef struct RedisModuleType RedisModuleType;
+typedef struct RedisModuleDigest RedisModuleDigest;
 
 typedef int (*RedisModuleCmdFunc) (RedisModuleCtx *ctx, RedisModuleString **argv, int argc);
 
+typedef void *(*RedisModuleTypeLoadFunc)(RedisModuleIO *rdb, int encver);
+typedef void (*RedisModuleTypeSaveFunc)(RedisModuleIO *rdb, void *value);
+typedef void (*RedisModuleTypeRewriteFunc)(RedisModuleIO *aof, RedisModuleString *key, void *value);
+typedef void (*RedisModuleTypeDigestFunc)(RedisModuleDigest *digest, void *value);
+typedef void (*RedisModuleTypeFreeFunc)(void *value);
+
 #define REDISMODULE_GET_API(name) \
     RedisModule_GetApi("RedisModule_" #name, ((void **)&RedisModule_ ## name))
 
 #define REDISMODULE_API_FUNC(x) (*x)
 
+
+void *REDISMODULE_API_FUNC(RedisModule_Alloc)(size_t bytes);
+void *REDISMODULE_API_FUNC(RedisModule_Realloc)(void *ptr, size_t bytes);
+void REDISMODULE_API_FUNC(RedisModule_Free)(void *ptr);
+void REDISMODULE_API_FUNC(RedisModule_Calloc)(size_t nmemb, size_t size);
+char *REDISMODULE_API_FUNC(RedisModule_Strdup)(const char *str);
 int REDISMODULE_API_FUNC(RedisModule_GetApi)(const char *, void *);
 int REDISMODULE_API_FUNC(RedisModule_CreateCommand)(RedisModuleCtx *ctx, const char *name, RedisModuleCmdFunc cmdfunc, const char *strflags, int firstkey, int lastkey, int keystep);
 int REDISMODULE_API_FUNC(RedisModule_SetModuleAttribs)(RedisModuleCtx *ctx, const char *name, int ver, int apiver);
@@ -108,8 +124,9 @@ size_t REDISMODULE_API_FUNC(RedisModule_CallReplyLength)(RedisModuleCallReply *r
 RedisModuleCallReply *REDISMODULE_API_FUNC(RedisModule_CallReplyArrayElement)(RedisModuleCallReply *reply, size_t idx);
 RedisModuleString *REDISMODULE_API_FUNC(RedisModule_CreateString)(RedisModuleCtx *ctx, const char *ptr, size_t len);
 RedisModuleString *REDISMODULE_API_FUNC(RedisModule_CreateStringFromLongLong)(RedisModuleCtx *ctx, long long ll);
+RedisModuleString *REDISMODULE_API_FUNC(RedisModule_CreateStringFromString)(RedisModuleCtx *ctx, const RedisModuleString *str);
 void REDISMODULE_API_FUNC(RedisModule_FreeString)(RedisModuleCtx *ctx, RedisModuleString *str);
-const char *REDISMODULE_API_FUNC(RedisModule_StringPtrLen)(RedisModuleString *str, size_t *len);
+const char *REDISMODULE_API_FUNC(RedisModule_StringPtrLen)(const RedisModuleString *str, size_t *len);
 int REDISMODULE_API_FUNC(RedisModule_ReplyWithError)(RedisModuleCtx *ctx, const char *err);
 int REDISMODULE_API_FUNC(RedisModule_ReplyWithSimpleString)(RedisModuleCtx *ctx, const char *msg);
 int REDISMODULE_API_FUNC(RedisModule_ReplyWithArray)(RedisModuleCtx *ctx, long len);
@@ -119,8 +136,8 @@ int REDISMODULE_API_FUNC(RedisModule_ReplyWithString)(RedisModuleCtx *ctx, Redis
 int REDISMODULE_API_FUNC(RedisModule_ReplyWithNull)(RedisModuleCtx *ctx);
 int REDISMODULE_API_FUNC(RedisModule_ReplyWithDouble)(RedisModuleCtx *ctx, double d);
 int REDISMODULE_API_FUNC(RedisModule_ReplyWithCallReply)(RedisModuleCtx *ctx, RedisModuleCallReply *reply);
-int REDISMODULE_API_FUNC(RedisModule_StringToLongLong)(RedisModuleString *str, long long *ll);
-int REDISMODULE_API_FUNC(RedisModule_StringToDouble)(RedisModuleString *str, double *d);
+int REDISMODULE_API_FUNC(RedisModule_StringToLongLong)(const RedisModuleString *str, long long *ll);
+int REDISMODULE_API_FUNC(RedisModule_StringToDouble)(const RedisModuleString *str, double *d);
 void REDISMODULE_API_FUNC(RedisModule_AutoMemory)(RedisModuleCtx *ctx);
 int REDISMODULE_API_FUNC(RedisModule_Replicate)(RedisModuleCtx *ctx, const char *cmdname, const char *fmt, ...);
 int REDISMODULE_API_FUNC(RedisModule_ReplicateVerbatim)(RedisModuleCtx *ctx);
@@ -150,11 +167,34 @@ int REDISMODULE_API_FUNC(RedisModule_HashGet)(RedisModuleKey *key, int flags, ..
 int REDISMODULE_API_FUNC(RedisModule_IsKeysPositionRequest)(RedisModuleCtx *ctx);
 void REDISMODULE_API_FUNC(RedisModule_KeyAtPos)(RedisModuleCtx *ctx, int pos);
 unsigned long long REDISMODULE_API_FUNC(RedisModule_GetClientId)(RedisModuleCtx *ctx);
+void *REDISMODULE_API_FUNC(RedisModule_PoolAlloc)(RedisModuleCtx *ctx, size_t bytes);
+RedisModuleType *REDISMODULE_API_FUNC(RedisModule_CreateDataType)(RedisModuleCtx *ctx, const char *name, int encver, RedisModuleTypeLoadFunc rdb_load, RedisModuleTypeSaveFunc rdb_save, RedisModuleTypeRewriteFunc aof_rewrite, RedisModuleTypeDigestFunc digest, RedisModuleTypeFreeFunc free);
+int REDISMODULE_API_FUNC(RedisModule_ModuleTypeSetValue)(RedisModuleKey *key, RedisModuleType *mt, void *value);
+RedisModuleType *REDISMODULE_API_FUNC(RedisModule_ModuleTypeGetType)(RedisModuleKey *key);
+void *REDISMODULE_API_FUNC(RedisModule_ModuleTypeGetValue)(RedisModuleKey *key);
+void REDISMODULE_API_FUNC(RedisModule_SaveUnsigned)(RedisModuleIO *io, uint64_t value);
+uint64_t REDISMODULE_API_FUNC(RedisModule_LoadUnsigned)(RedisModuleIO *io);
+void REDISMODULE_API_FUNC(RedisModule_SaveSigned)(RedisModuleIO *io, int64_t value);
+int64_t REDISMODULE_API_FUNC(RedisModule_LoadSigned)(RedisModuleIO *io);
+void REDISMODULE_API_FUNC(RedisModule_EmitAOF)(RedisModuleIO *io, const char *cmdname, const char *fmt, ...);
+void REDISMODULE_API_FUNC(RedisModule_SaveString)(RedisModuleIO *io, RedisModuleString *s);
+void REDISMODULE_API_FUNC(RedisModule_SaveStringBuffer)(RedisModuleIO *io, const char *str, size_t len);
+RedisModuleString *REDISMODULE_API_FUNC(RedisModule_LoadString)(RedisModuleIO *io);
+char *REDISMODULE_API_FUNC(RedisModule_LoadStringBuffer)(RedisModuleIO *io, size_t *lenptr);
+void REDISMODULE_API_FUNC(RedisModule_SaveDouble)(RedisModuleIO *io, double value);
+double REDISMODULE_API_FUNC(RedisModule_LoadDouble)(RedisModuleIO *io);
+void REDISMODULE_API_FUNC(RedisModule_Log)(RedisModuleCtx *ctx, const char *level, const char *fmt, ...);
 
 /* This is included inline inside each Redis module. */
+static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int apiver) __attribute__((unused));
 static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int apiver) {
     void *getapifuncptr = ((void**)ctx)[0];
     RedisModule_GetApi = (int (*)(const char *, void *)) (unsigned long)getapifuncptr;
+    REDISMODULE_GET_API(Alloc);
+    REDISMODULE_GET_API(Calloc);
+    REDISMODULE_GET_API(Free);
+    REDISMODULE_GET_API(Realloc);
+    REDISMODULE_GET_API(Strdup);
     REDISMODULE_GET_API(CreateCommand);
     REDISMODULE_GET_API(SetModuleAttribs);
     REDISMODULE_GET_API(WrongArity);
@@ -190,6 +230,7 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int
     REDISMODULE_GET_API(CreateStringFromCallReply);
     REDISMODULE_GET_API(CreateString);
     REDISMODULE_GET_API(CreateStringFromLongLong);
+    REDISMODULE_GET_API(CreateStringFromString);
     REDISMODULE_GET_API(FreeString);
     REDISMODULE_GET_API(StringPtrLen);
     REDISMODULE_GET_API(AutoMemory);
@@ -219,6 +260,23 @@ static int RedisModule_Init(RedisModuleCtx *ctx, const char *name, int ver, int
     REDISMODULE_GET_API(IsKeysPositionRequest);
     REDISMODULE_GET_API(KeyAtPos);
     REDISMODULE_GET_API(GetClientId);
+    REDISMODULE_GET_API(PoolAlloc);
+    REDISMODULE_GET_API(CreateDataType);
+    REDISMODULE_GET_API(ModuleTypeSetValue);
+    REDISMODULE_GET_API(ModuleTypeGetType);
+    REDISMODULE_GET_API(ModuleTypeGetValue);
+    REDISMODULE_GET_API(SaveUnsigned);
+    REDISMODULE_GET_API(LoadUnsigned);
+    REDISMODULE_GET_API(SaveSigned);
+    REDISMODULE_GET_API(LoadSigned);
+    REDISMODULE_GET_API(SaveString);
+    REDISMODULE_GET_API(SaveStringBuffer);
+    REDISMODULE_GET_API(LoadString);
+    REDISMODULE_GET_API(LoadStringBuffer);
+    REDISMODULE_GET_API(SaveDouble);
+    REDISMODULE_GET_API(LoadDouble);
+    REDISMODULE_GET_API(EmitAOF);
+    REDISMODULE_GET_API(Log);
 
     RedisModule_SetModuleAttribs(ctx,name,ver,apiver);
     return REDISMODULE_OK;
diff --git a/src/rio.h b/src/rio.h
index 711308ce6..6749723d2 100644
--- a/src/rio.h
+++ b/src/rio.h
@@ -135,6 +135,9 @@ size_t rioWriteBulkString(rio *r, const char *buf, size_t len);
 size_t rioWriteBulkLongLong(rio *r, long long l);
 size_t rioWriteBulkDouble(rio *r, double d);
 
+struct redisObject;
+int rioWriteBulkObject(rio *r, struct redisObject *obj);
+
 void rioGenericUpdateChecksum(rio *r, const void *buf, size_t len);
 void rioSetAutoSync(rio *r, off_t bytes);
 
diff --git a/src/sds.c b/src/sds.c
index e3dd67352..26e90a6db 100644
--- a/src/sds.c
+++ b/src/sds.c
@@ -55,13 +55,13 @@ static inline int sdsHdrSize(char type) {
 }
 
 static inline char sdsReqType(size_t string_size) {
-    if (string_size < 32)
+    if (string_size < 1<<5)
         return SDS_TYPE_5;
-    if (string_size < 0xff)
+    if (string_size < 1<<8)
         return SDS_TYPE_8;
-    if (string_size < 0xffff)
+    if (string_size < 1<<16)
         return SDS_TYPE_16;
-    if (string_size < 0xffffffff)
+    if (string_size < 1ll<<32)
         return SDS_TYPE_32;
     return SDS_TYPE_64;
 }
diff --git a/src/sentinel.c b/src/sentinel.c
index 0d1eb78aa..f8ebd0c6f 100644
--- a/src/sentinel.c
+++ b/src/sentinel.c
@@ -1910,6 +1910,7 @@ void sentinelReconnectInstance(sentinelRedisInstance *ri) {
                 link->cc->errstr);
             instanceLinkCloseConnection(link,link->cc);
         } else {
+            link->pending_commands = 0;
             link->cc_conn_time = mstime();
             link->cc->data = link;
             redisAeAttach(server.el,link->cc);
@@ -3874,11 +3875,11 @@ int compareSlavesForPromotion(const void *a, const void *b) {
         return (*sa)->slave_priority - (*sb)->slave_priority;
 
     /* If priority is the same, select the slave with greater replication
-     * offset (processed more data frmo the master). */
+     * offset (processed more data from the master). */
     if ((*sa)->slave_repl_offset > (*sb)->slave_repl_offset) {
         return -1; /* a < b */
     } else if ((*sa)->slave_repl_offset < (*sb)->slave_repl_offset) {
-        return 1; /* b > a */
+        return 1; /* a > b */
     }
 
     /* If the replication offset is the same select the slave with that has
@@ -3996,7 +3997,7 @@ void sentinelFailoverSendSlaveOfNoOne(sentinelRedisInstance *ri) {
     /* We can't send the command to the promoted slave if it is now
      * disconnected. Retry again and again with this state until the timeout
      * is reached, then abort the failover. */
-    if (ri->link->disconnected) {
+    if (ri->promoted_slave->link->disconnected) {
         if (mstime() - ri->failover_state_change_time > ri->failover_timeout) {
             sentinelEvent(LL_WARNING,"-failover-abort-slave-timeout",ri,"%@");
             sentinelAbortFailover(ri);
diff --git a/src/server.c b/src/server.c
index f070865cb..06244081f 100644
--- a/src/server.c
+++ b/src/server.c
@@ -145,8 +145,8 @@ struct redisCommand redisCommandTable[] = {
     {"mget",mgetCommand,-2,"r",0,NULL,1,-1,1,0,0},
     {"rpush",rpushCommand,-3,"wmF",0,NULL,1,1,1,0,0},
     {"lpush",lpushCommand,-3,"wmF",0,NULL,1,1,1,0,0},
-    {"rpushx",rpushxCommand,3,"wmF",0,NULL,1,1,1,0,0},
-    {"lpushx",lpushxCommand,3,"wmF",0,NULL,1,1,1,0,0},
+    {"rpushx",rpushxCommand,-3,"wmF",0,NULL,1,1,1,0,0},
+    {"lpushx",lpushxCommand,-3,"wmF",0,NULL,1,1,1,0,0},
     {"linsert",linsertCommand,5,"wm",0,NULL,1,1,1,0,0},
     {"rpop",rpopCommand,2,"wF",0,NULL,1,1,1,0,0},
     {"lpop",lpopCommand,2,"wF",0,NULL,1,1,1,0,0},
@@ -165,7 +165,7 @@ struct redisCommand redisCommandTable[] = {
     {"smove",smoveCommand,4,"wF",0,NULL,1,2,1,0,0},
     {"sismember",sismemberCommand,3,"rF",0,NULL,1,1,1,0,0},
     {"scard",scardCommand,2,"rF",0,NULL,1,1,1,0,0},
-    {"spop",spopCommand,-2,"wRsF",0,NULL,1,1,1,0,0},
+    {"spop",spopCommand,-2,"wRF",0,NULL,1,1,1,0,0},
     {"srandmember",srandmemberCommand,-2,"rR",0,NULL,1,1,1,0,0},
     {"sinter",sinterCommand,-2,"rS",0,NULL,1,-1,1,0,0},
     {"sinterstore",sinterstoreCommand,-3,"wm",0,NULL,1,-1,1,0,0},
@@ -250,6 +250,7 @@ struct redisCommand redisCommandTable[] = {
     {"info",infoCommand,-1,"lt",0,NULL,0,0,0,0,0},
     {"monitor",monitorCommand,1,"as",0,NULL,0,0,0,0,0},
     {"ttl",ttlCommand,2,"rF",0,NULL,1,1,1,0,0},
+    {"touch",touchCommand,-2,"rF",0,NULL,1,1,1,0,0},
     {"pttl",pttlCommand,2,"rF",0,NULL,1,1,1,0,0},
     {"persist",persistCommand,2,"wF",0,NULL,1,1,1,0,0},
     {"slaveof",slaveofCommand,3,"ast",0,NULL,0,0,0,0,0},
@@ -688,7 +689,7 @@ int htNeedsResize(dict *dict) {
 
     size = dictSlots(dict);
     used = dictSize(dict);
-    return (size && used && size > DICT_HT_INITIAL_SIZE &&
+    return (size > DICT_HT_INITIAL_SIZE &&
             (used*100/size < HASHTABLE_MIN_FILL));
 }
 
@@ -1959,8 +1960,9 @@ void initServer(void) {
     server.repl_good_slaves_count = 0;
     updateCachedTime();
 
-    /* Create out timers, that's our main way to process background
-     * operations. */
+    /* Create the timer callback, this is our way to process many background
+     * operations incrementally, like clients timeout, eviction of unaccessed
+     * expired keys and so forth. */
     if (aeCreateTimeEvent(server.el, 1, serverCron, NULL, NULL) == AE_ERR) {
         serverPanic("Can't create event loop timers.");
         exit(1);
@@ -3823,7 +3825,7 @@ void setupSignalHandlers(void) {
 void memtest(size_t megabytes, int passes);
 
 /* Returns 1 if there is --sentinel among the arguments or if
- * argv[0] is exactly "redis-sentinel". */
+ * argv[0] contains "redis-sentinel". */
 int checkForSentinelMode(int argc, char **argv) {
     int j;
 
diff --git a/src/server.h b/src/server.h
index 28a8bebf6..c1963bf21 100644
--- a/src/server.h
+++ b/src/server.h
@@ -33,6 +33,7 @@
 #include "fmacros.h"
 #include "config.h"
 #include "solarisfixes.h"
+#include "rio.h"
 
 #include <stdio.h>
 #include <stdlib.h>
@@ -164,7 +165,7 @@ typedef long long mstime_t; /* millisecond time type. */
 #define PROTO_REPLY_CHUNK_BYTES (16*1024) /* 16k output buffer */
 #define PROTO_INLINE_MAX_SIZE   (1024*64) /* Max size of inline reads */
 #define PROTO_MBULK_BIG_ARG     (1024*32)
-#define LONG_STR_SIZE      21          /* Bytes needed for long -> str */
+#define LONG_STR_SIZE      21          /* Bytes needed for long -> str + '\0' */
 #define AOF_AUTOSYNC_BYTES (1024*1024*32) /* fdatasync every 32MB */
 
 /* When configuring the server eventloop, we setup it so that the total number
@@ -195,33 +196,6 @@ typedef long long mstime_t; /* millisecond time type. */
 #define CMD_MODULE_GETKEYS (1<<14)  /* Use the modules getkeys interface. */
 #define CMD_MODULE_NO_CLUSTER (1<<15) /* Deny on Redis Cluster. */
 
-/* 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 10, 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 RDB_ENC_* defines.
- *
- * Lengths up to 63 are stored using a single byte, most DB keys, and may
- * values, will fit inside. */
-#define RDB_6BITLEN 0
-#define RDB_14BITLEN 1
-#define RDB_32BITLEN 2
-#define RDB_ENCVAL 3
-#define 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 RDB_ENC_INT8 0        /* 8 bit signed integer */
-#define RDB_ENC_INT16 1       /* 16 bit signed integer */
-#define RDB_ENC_INT32 2       /* 32 bit signed integer */
-#define RDB_ENC_LZF 3         /* string compressed with FASTLZ */
-
 /* AOF states */
 #define AOF_OFF 0             /* AOF is off */
 #define AOF_ON 1              /* AOF is on */
@@ -448,6 +422,90 @@ typedef long long mstime_t; /* millisecond time type. */
 #define OBJ_ZSET 3
 #define OBJ_HASH 4
 
+/* The "module" object type is a special one that signals that the object
+ * is one directly managed by a Redis module. In this case the value points
+ * to a moduleValue struct, which contains the object value (which is only
+ * handled by the module itself) and the RedisModuleType struct which lists
+ * function pointers in order to serialize, deserialize, AOF-rewrite and
+ * free the object.
+ *
+ * Inside the RDB file, module types are encoded as OBJ_MODULE followed
+ * by a 64 bit module type ID, which has a 54 bits module-specific signature
+ * in order to dispatch the loading to the right module, plus a 10 bits
+ * encoding version. */
+#define OBJ_MODULE 5
+
+/* Extract encver / signature from a module type ID. */
+#define REDISMODULE_TYPE_ENCVER_BITS 10
+#define REDISMODULE_TYPE_ENCVER_MASK ((1<<REDISMODULE_TYPE_ENCVER_BITS)-1)
+#define REDISMODULE_TYPE_ENCVER(id) (id & REDISMODULE_TYPE_ENCVER_MASK)
+#define REDISMODULE_TYPE_SIGN(id) ((id & ~((uint64_t)REDISMODULE_TYPE_ENCVER_MASK)) >>REDISMODULE_TYPE_ENCVER_BITS)
+
+struct RedisModule;
+struct RedisModuleIO;
+struct RedisModuleDigest;
+struct redisObject;
+
+/* Each module type implementation should export a set of methods in order
+ * to serialize and deserialize the value in the RDB file, rewrite the AOF
+ * log, create the digest for "DEBUG DIGEST", and free the value when a key
+ * is deleted. */
+typedef void *(*moduleTypeLoadFunc)(struct RedisModuleIO *io, int encver);
+typedef void (*moduleTypeSaveFunc)(struct RedisModuleIO *io, void *value);
+typedef void (*moduleTypeRewriteFunc)(struct RedisModuleIO *io, struct redisObject *key, void *value);
+typedef void (*moduleTypeDigestFunc)(struct RedisModuleDigest *digest, void *value);
+typedef void (*moduleTypeFreeFunc)(void *value);
+
+/* The module type, which is referenced in each value of a given type, defines
+ * the methods and links to the module exporting the type. */
+typedef struct RedisModuleType {
+    uint64_t id; /* Higher 54 bits of type ID + 10 lower bits of encoding ver. */
+    struct RedisModule *module;
+    moduleTypeLoadFunc rdb_load;
+    moduleTypeSaveFunc rdb_save;
+    moduleTypeRewriteFunc aof_rewrite;
+    moduleTypeDigestFunc digest;
+    moduleTypeFreeFunc free;
+    char name[10]; /* 9 bytes name + null term. Charset: A-Z a-z 0-9 _- */
+} moduleType;
+
+/* In Redis objects 'robj' structures of type OBJ_MODULE, the value pointer
+ * is set to the following structure, referencing the moduleType structure
+ * in order to work with the value, and at the same time providing a raw
+ * pointer to the value, as created by the module commands operating with
+ * the module type.
+ *
+ * So for example in order to free such a value, it is possible to use
+ * the following code:
+ *
+ *  if (robj->type == OBJ_MODULE) {
+ *      moduleValue *mt = robj->ptr;
+ *      mt->type->free(mt->value);
+ *      zfree(mt); // We need to release this in-the-middle struct as well.
+ *  }
+ */
+typedef struct moduleValue {
+    moduleType *type;
+    void *value;
+} moduleValue;
+
+/* This is a wrapper for the 'rio' streams used inside rdb.c in Redis, so that
+ * the user does not have to take the total count of the written bytes nor
+ * to care about error conditions. */
+typedef struct RedisModuleIO {
+    size_t bytes;       /* Bytes read / written so far. */
+    rio *rio;           /* Rio stream. */
+    moduleType *type;   /* Module type doing the operation. */
+    int error;          /* True if error condition happened. */
+} RedisModuleIO;
+
+#define moduleInitIOContext(iovar,mtype,rioptr) do { \
+    iovar.rio = rioptr; \
+    iovar.type = mtype; \
+    iovar.bytes = 0; \
+    iovar.error = 0; \
+} while(0);
+
 /* 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. */
@@ -478,7 +536,7 @@ typedef struct redisObject {
 /* Macro used to obtain the current LRU clock.
  * If the current resolution is lower than the frequency we refresh the
  * LRU clock (as it should be in production servers) we return the
- * precomputed value, otherwise we need to resort to a function call. */
+ * precomputed value, otherwise we need to resort to a system call. */
 #define LRU_CLOCK() ((1000/server.hz <= LRU_CLOCK_RESOLUTION) ? server.lruclock : getLRUClock())
 
 /* Macro used to initialize a Redis object allocated on the stack.
@@ -625,6 +683,12 @@ struct saveparam {
     int changes;
 };
 
+struct moduleLoadQueueEntry {
+    sds path;
+    int argc;
+    robj **argv;
+};
+
 struct sharedObjectsStruct {
     robj *crlf, *ok, *err, *emptybulk, *czero, *cone, *cnegone, *pong, *space,
     *colon, *nullbulk, *nullmultibulk, *queued,
@@ -984,8 +1048,8 @@ struct redisServer {
     long long latency_monitor_threshold;
     dict *latency_events;
     /* Assert & bug reporting */
-    char *assert_failed;
-    char *assert_file;
+    const char *assert_failed;
+    const char *assert_file;
     int assert_line;
     int bug_report_start; /* True if bug report header was already logged. */
     int watchdog_period;  /* Software watchdog period in ms. 0 = off */
@@ -1098,9 +1162,11 @@ extern dictType modulesDictType;
 
 /* Modules */
 void moduleInitModulesSystem(void);
-int moduleLoad(const char *path);
+int moduleLoad(const char *path, void **argv, int argc);
 void moduleLoadFromQueue(void);
 int *moduleGetCommandKeysViaAPI(struct redisCommand *cmd, robj **argv, int argc, int *numkeys);
+moduleType *moduleTypeLookupModuleByID(uint64_t id);
+void moduleTypeNameByID(char *name, uint64_t moduleid);
 
 /* Utils */
 long long ustime(void);
@@ -1179,7 +1245,7 @@ void addReplyStatusFormat(client *c, const char *fmt, ...);
 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);
+unsigned long listTypeLength(const robj *subject);
 listTypeIterator *listTypeInitIterator(robj *subject, long index, unsigned char direction);
 void listTypeReleaseIterator(listTypeIterator *li);
 int listTypeNext(listTypeIterator *li, listTypeEntry *entry);
@@ -1219,7 +1285,7 @@ robj *createObject(int type, void *ptr);
 robj *createStringObject(const char *ptr, size_t len);
 robj *createRawStringObject(const char *ptr, size_t len);
 robj *createEmbeddedStringObject(const char *ptr, size_t len);
-robj *dupStringObject(robj *o);
+robj *dupStringObject(const robj *o);
 int isSdsRepresentableAsLongLong(sds s, long long *llval);
 int isObjectRepresentableAsLongLong(robj *o, long long *llongval);
 robj *tryObjectEncoding(robj *o);
@@ -1234,11 +1300,12 @@ robj *createIntsetObject(void);
 robj *createHashObject(void);
 robj *createZsetObject(void);
 robj *createZsetZiplistObject(void);
+robj *createModuleObject(moduleType *mt, void *value);
 int getLongFromObjectOrReply(client *c, robj *o, long *target, const char *msg);
 int checkType(client *c, robj *o, int type);
 int getLongLongFromObjectOrReply(client *c, robj *o, long long *target, const char *msg);
 int getDoubleFromObjectOrReply(client *c, robj *o, double *target, const char *msg);
-int getDoubleFromObject(robj *o, double *target);
+int getDoubleFromObject(const robj *o, double *target);
 int getLongLongFromObject(robj *o, long long *target);
 int getLongDoubleFromObject(robj *o, long double *target);
 int getLongDoubleFromObjectOrReply(client *c, robj *o, long double *target, const char *msg);
@@ -1339,7 +1406,7 @@ void zzlNext(unsigned char *zl, unsigned char **eptr, unsigned char **sptr);
 void zzlPrev(unsigned char *zl, unsigned char **eptr, unsigned char **sptr);
 unsigned char *zzlFirstInRange(unsigned char *zl, zrangespec *range);
 unsigned char *zzlLastInRange(unsigned char *zl, zrangespec *range);
-unsigned int zsetLength(robj *zobj);
+unsigned int zsetLength(const robj *zobj);
 void zsetConvert(robj *zobj, int encoding);
 void zsetConvertToZiplistIfNeeded(robj *zobj, size_t maxelelen);
 int zsetScore(robj *zobj, sds member, double *score);
@@ -1387,7 +1454,6 @@ void serverLogFromHandler(int level, const char *msg);
 void usage(void);
 void updateDictResizePolicy(void);
 int htNeedsResize(dict *dict);
-void oom(const char *msg);
 void populateCommandTable(void);
 void resetCommandTableStats(void);
 void adjustOpenFilesLimit(void);
@@ -1413,7 +1479,7 @@ int setTypeNext(setTypeIterator *si, sds *sdsele, int64_t *llele);
 sds setTypeNextObject(setTypeIterator *si);
 int setTypeRandomElement(robj *setobj, sds *sdsele, int64_t *llele);
 unsigned long setTypeRandomElements(robj *set, unsigned long count, robj *aux_set);
-unsigned long setTypeSize(robj *subject);
+unsigned long setTypeSize(const robj *subject);
 void setTypeConvert(robj *subject, int enc);
 
 /* Hash data type */
@@ -1426,7 +1492,7 @@ void hashTypeTryConversion(robj *subject, robj **argv, int start, int end);
 void hashTypeTryObjectEncoding(robj *subject, robj **o1, robj **o2);
 int hashTypeExists(robj *o, sds key);
 int hashTypeDelete(robj *o, sds key);
-unsigned long hashTypeLength(robj *o);
+unsigned long hashTypeLength(const robj *o);
 hashTypeIterator *hashTypeInitIterator(robj *subject);
 void hashTypeReleaseIterator(hashTypeIterator *hi);
 int hashTypeNext(hashTypeIterator *hi);
@@ -1467,11 +1533,14 @@ void propagateExpire(redisDb *db, robj *key, int lazy);
 int expireIfNeeded(redisDb *db, robj *key);
 long long getExpire(redisDb *db, robj *key);
 void setExpire(redisDb *db, robj *key, long long when);
-robj *lookupKey(redisDb *db, robj *key);
+robj *lookupKey(redisDb *db, robj *key, int flags);
 robj *lookupKeyRead(redisDb *db, robj *key);
 robj *lookupKeyWrite(redisDb *db, robj *key);
 robj *lookupKeyReadOrReply(client *c, robj *key, robj *reply);
 robj *lookupKeyWriteOrReply(client *c, robj *key, robj *reply);
+robj *lookupKeyReadWithFlags(redisDb *db, robj *key, int flags);
+#define LOOKUP_NONE 0
+#define LOOKUP_NOTOUCH (1<<0)
 void dbAdd(redisDb *db, robj *key, robj *val);
 void dbOverwrite(redisDb *db, robj *key, robj *val);
 void setKey(redisDb *db, robj *key, robj *val);
@@ -1627,6 +1696,7 @@ void pexpireCommand(client *c);
 void pexpireatCommand(client *c);
 void getsetCommand(client *c);
 void ttlCommand(client *c);
+void touchCommand(client *c);
 void pttlCommand(client *c);
 void persistCommand(client *c);
 void slaveofCommand(client *c);
@@ -1729,11 +1799,11 @@ void *realloc(void *ptr, size_t size) __attribute__ ((deprecated));
 #endif
 
 /* Debugging stuff */
-void _serverAssertWithInfo(client *c, robj *o, char *estr, char *file, int line);
-void _serverAssert(char *estr, char *file, int line);
-void _serverPanic(char *msg, char *file, int line);
+void _serverAssertWithInfo(const client *c, const robj *o, const char *estr, const char *file, int line);
+void _serverAssert(const char *estr, const char *file, int line);
+void _serverPanic(const char *msg, const char *file, int line);
 void bugReportStart(void);
-void serverLogObjectDebugInfo(robj *o);
+void serverLogObjectDebugInfo(const robj *o);
 void sigsegvHandler(int sig, siginfo_t *info, void *secret);
 sds genRedisInfoString(char *section);
 void enableWatchdog(int period);
diff --git a/src/t_hash.c b/src/t_hash.c
index c75b391d7..a49559336 100644
--- a/src/t_hash.c
+++ b/src/t_hash.c
@@ -308,13 +308,13 @@ int hashTypeDelete(robj *o, sds field) {
 }
 
 /* Return the number of elements in a hash. */
-unsigned long hashTypeLength(robj *o) {
+unsigned long hashTypeLength(const robj *o) {
     unsigned long length = ULONG_MAX;
 
     if (o->encoding == OBJ_ENCODING_ZIPLIST) {
         length = ziplistLen(o->ptr) / 2;
     } else if (o->encoding == OBJ_ENCODING_HT) {
-        length = dictSize((dict*)o->ptr);
+        length = dictSize((const dict*)o->ptr);
     } else {
         serverPanic("Unknown hash encoding");
     }
diff --git a/src/t_list.c b/src/t_list.c
index 7d5be11af..a0a30998d 100644
--- a/src/t_list.c
+++ b/src/t_list.c
@@ -71,7 +71,7 @@ robj *listTypePop(robj *subject, int where) {
     return value;
 }
 
-unsigned long listTypeLength(robj *subject) {
+unsigned long listTypeLength(const robj *subject) {
     if (subject->encoding == OBJ_ENCODING_QUICKLIST) {
         return quicklistCount(subject->ptr);
     } else {
@@ -195,7 +195,7 @@ void listTypeConvert(robj *subject, int enc) {
  *----------------------------------------------------------------------------*/
 
 void pushGenericCommand(client *c, int where) {
-    int j, waiting = 0, pushed = 0;
+    int j, pushed = 0;
     robj *lobj = lookupKeyWrite(c->db,c->argv[1]);
 
     if (lobj && lobj->type != OBJ_LIST) {
@@ -204,7 +204,6 @@ void pushGenericCommand(client *c, int where) {
     }
 
     for (j = 2; j < c->argc; j++) {
-        c->argv[j] = tryObjectEncoding(c->argv[j]);
         if (!lobj) {
             lobj = createQuicklistObject();
             quicklistSetOptions(lobj->ptr, server.list_max_ziplist_size,
@@ -214,7 +213,7 @@ void pushGenericCommand(client *c, int where) {
         listTypePush(lobj,c->argv[j],where);
         pushed++;
     }
-    addReplyLongLong(c, waiting + (lobj ? listTypeLength(lobj) : 0));
+    addReplyLongLong(c, (lobj ? listTypeLength(lobj) : 0));
     if (pushed) {
         char *event = (where == LIST_HEAD) ? "lpush" : "rpush";
 
@@ -232,68 +231,78 @@ void rpushCommand(client *c) {
     pushGenericCommand(c,LIST_TAIL);
 }
 
-void pushxGenericCommand(client *c, robj *refval, robj *val, int where) {
+void pushxGenericCommand(client *c, int where) {
+    int j, pushed = 0;
     robj *subject;
-    listTypeIterator *iter;
-    listTypeEntry entry;
-    int inserted = 0;
 
     if ((subject = lookupKeyWriteOrReply(c,c->argv[1],shared.czero)) == NULL ||
         checkType(c,subject,OBJ_LIST)) return;
 
-    if (refval != NULL) {
-        /* Seek refval from head to tail */
-        iter = listTypeInitIterator(subject,0,LIST_TAIL);
-        while (listTypeNext(iter,&entry)) {
-            if (listTypeEqual(&entry,refval)) {
-                listTypeInsert(&entry,val,where);
-                inserted = 1;
-                break;
-            }
-        }
-        listTypeReleaseIterator(iter);
+    for (j = 2; j < c->argc; j++) {
+        listTypePush(subject,c->argv[j],where);
+        pushed++;
+    }
 
-        if (inserted) {
-            signalModifiedKey(c->db,c->argv[1]);
-            notifyKeyspaceEvent(NOTIFY_LIST,"linsert",
-                                c->argv[1],c->db->id);
-            server.dirty++;
-        } else {
-            /* Notify client of a failed insert */
-            addReply(c,shared.cnegone);
-            return;
-        }
-    } else {
-        char *event = (where == LIST_HEAD) ? "lpush" : "rpush";
+    addReplyLongLong(c,listTypeLength(subject));
 
-        listTypePush(subject,val,where);
+    if (pushed) {
+        char *event = (where == LIST_HEAD) ? "lpush" : "rpush";
         signalModifiedKey(c->db,c->argv[1]);
         notifyKeyspaceEvent(NOTIFY_LIST,event,c->argv[1],c->db->id);
-        server.dirty++;
     }
-
-    addReplyLongLong(c,listTypeLength(subject));
+    server.dirty += pushed;
 }
 
 void lpushxCommand(client *c) {
-    c->argv[2] = tryObjectEncoding(c->argv[2]);
-    pushxGenericCommand(c,NULL,c->argv[2],LIST_HEAD);
+    pushxGenericCommand(c,LIST_HEAD);
 }
 
 void rpushxCommand(client *c) {
-    c->argv[2] = tryObjectEncoding(c->argv[2]);
-    pushxGenericCommand(c,NULL,c->argv[2],LIST_TAIL);
+    pushxGenericCommand(c,LIST_TAIL);
 }
 
 void linsertCommand(client *c) {
-    c->argv[4] = tryObjectEncoding(c->argv[4]);
+    int where;
+    robj *subject;
+    listTypeIterator *iter;
+    listTypeEntry entry;
+    int inserted = 0;
+
     if (strcasecmp(c->argv[2]->ptr,"after") == 0) {
-        pushxGenericCommand(c,c->argv[3],c->argv[4],LIST_TAIL);
+        where = LIST_TAIL;
     } else if (strcasecmp(c->argv[2]->ptr,"before") == 0) {
-        pushxGenericCommand(c,c->argv[3],c->argv[4],LIST_HEAD);
+        where = LIST_HEAD;
     } else {
         addReply(c,shared.syntaxerr);
+        return;
     }
+
+    if ((subject = lookupKeyWriteOrReply(c,c->argv[1],shared.czero)) == NULL ||
+        checkType(c,subject,OBJ_LIST)) return;
+
+    /* Seek pivot from head to tail */
+    iter = listTypeInitIterator(subject,0,LIST_TAIL);
+    while (listTypeNext(iter,&entry)) {
+        if (listTypeEqual(&entry,c->argv[3])) {
+            listTypeInsert(&entry,c->argv[4],where);
+            inserted = 1;
+            break;
+        }
+    }
+    listTypeReleaseIterator(iter);
+
+    if (inserted) {
+        signalModifiedKey(c->db,c->argv[1]);
+        notifyKeyspaceEvent(NOTIFY_LIST,"linsert",
+                            c->argv[1],c->db->id);
+        server.dirty++;
+    } else {
+        /* Notify client of a failed insert */
+        addReply(c,shared.cnegone);
+        return;
+    }
+
+    addReplyLongLong(c,listTypeLength(subject));
 }
 
 void llenCommand(client *c) {
diff --git a/src/t_set.c b/src/t_set.c
index 7a2a77ff6..ddd82b8b0 100644
--- a/src/t_set.c
+++ b/src/t_set.c
@@ -219,11 +219,11 @@ int setTypeRandomElement(robj *setobj, sds *sdsele, int64_t *llele) {
     return setobj->encoding;
 }
 
-unsigned long setTypeSize(robj *subject) {
+unsigned long setTypeSize(const robj *subject) {
     if (subject->encoding == OBJ_ENCODING_HT) {
-        return dictSize((dict*)subject->ptr);
+        return dictSize((const dict*)subject->ptr);
     } else if (subject->encoding == OBJ_ENCODING_INTSET) {
-        return intsetLen((intset*)subject->ptr);
+        return intsetLen((const intset*)subject->ptr);
     } else {
         serverPanic("Unknown set encoding");
     }
@@ -351,9 +351,6 @@ void smoveCommand(client *c) {
         dbDelete(c->db,c->argv[1]);
         notifyKeyspaceEvent(NOTIFY_GENERIC,"del",c->argv[1],c->db->id);
     }
-    signalModifiedKey(c->db,c->argv[1]);
-    signalModifiedKey(c->db,c->argv[2]);
-    server.dirty++;
 
     /* Create the destination set when it doesn't exist */
     if (!dstset) {
@@ -361,6 +358,10 @@ void smoveCommand(client *c) {
         dbAdd(c->db,c->argv[2],dstset);
     }
 
+    signalModifiedKey(c->db,c->argv[1]);
+    signalModifiedKey(c->db,c->argv[2]);
+    server.dirty++;
+
     /* An extra key has changed when ele was successfully added to dstset */
     if (setTypeAdd(dstset,ele->ptr)) {
         server.dirty++;
@@ -547,6 +548,8 @@ void spopWithCountCommand(client *c) {
      * the alsoPropagate() API. */
     decrRefCount(propargv[0]);
     preventCommandPropagation(c);
+    signalModifiedKey(c->db,c->argv[1]);
+    server.dirty++;
 }
 
 void spopCommand(client *c) {
diff --git a/src/t_string.c b/src/t_string.c
index 35eb9d7c1..8c737c4e3 100644
--- a/src/t_string.c
+++ b/src/t_string.c
@@ -263,6 +263,10 @@ void getrangeCommand(client *c) {
     }
 
     /* Convert negative indexes */
+    if (start < 0 && end < 0 && start > end) {
+        addReply(c,shared.emptybulk);
+        return;
+    }
     if (start < 0) start = strlen+start;
     if (end < 0) end = strlen+end;
     if (start < 0) start = 0;
diff --git a/src/t_zset.c b/src/t_zset.c
index 86aa5510d..c61ba8089 100644
--- a/src/t_zset.c
+++ b/src/t_zset.c
@@ -1100,12 +1100,12 @@ unsigned char *zzlDeleteRangeByRank(unsigned char *zl, unsigned int start, unsig
  * Common sorted set API
  *----------------------------------------------------------------------------*/
 
-unsigned int zsetLength(robj *zobj) {
+unsigned int zsetLength(const robj *zobj) {
     int length = -1;
     if (zobj->encoding == OBJ_ENCODING_ZIPLIST) {
         length = zzlLength(zobj->ptr);
     } else if (zobj->encoding == OBJ_ENCODING_SKIPLIST) {
-        length = ((zset*)zobj->ptr)->zsl->length;
+        length = ((const zset*)zobj->ptr)->zsl->length;
     } else {
         serverPanic("Unknown sorted set encoding");
     }
@@ -2327,16 +2327,13 @@ void zunionInterGenericCommand(client *c, robj *dstkey, int op) {
         serverPanic("Unknown operator");
     }
 
-    if (dbDelete(c->db,dstkey)) {
-        signalModifiedKey(c->db,dstkey);
+    if (dbDelete(c->db,dstkey))
         touched = 1;
-        server.dirty++;
-    }
     if (dstzset->zsl->length) {
         zsetConvertToZiplistIfNeeded(dstobj,maxelelen);
         dbAdd(c->db,dstkey,dstobj);
         addReplyLongLong(c,zsetLength(dstobj));
-        if (!touched) signalModifiedKey(c->db,dstkey);
+        signalModifiedKey(c->db,dstkey);
         notifyKeyspaceEvent(NOTIFY_ZSET,
             (op == SET_OP_UNION) ? "zunionstore" : "zinterstore",
             dstkey,c->db->id);
@@ -2344,8 +2341,11 @@ void zunionInterGenericCommand(client *c, robj *dstkey, int op) {
     } else {
         decrRefCount(dstobj);
         addReply(c,shared.czero);
-        if (touched)
+        if (touched) {
+            signalModifiedKey(c->db,dstkey);
             notifyKeyspaceEvent(NOTIFY_GENERIC,"del",dstkey,c->db->id);
+            server.dirty++;
+        }
     }
     zfree(src);
 }
diff --git a/tests/test_helper.tcl b/tests/test_helper.tcl
index 45cecfdde..5f114c5dc 100644
--- a/tests/test_helper.tcl
+++ b/tests/test_helper.tcl
@@ -46,10 +46,12 @@ set ::all_tests {
     unit/scripting
     unit/maxmemory
     unit/introspection
+    unit/introspection-2
     unit/limits
     unit/obuf-limits
     unit/bitops
     unit/bitfield
+    unit/geo
     unit/memefficiency
     unit/hyperloglog
     unit/lazyfree
diff --git a/tests/unit/bitfield.tcl b/tests/unit/bitfield.tcl
index 368ff9fc6..26e47db0f 100644
--- a/tests/unit/bitfield.tcl
+++ b/tests/unit/bitfield.tcl
@@ -184,4 +184,9 @@ start_server {tags {"bitops"}} {
             }
         }
     }
+
+    test {BITFIELD regression for #3221} {
+        r set bits 1
+        r bitfield bits get u1 0
+    } {0}
 }
diff --git a/tests/unit/bitops.tcl b/tests/unit/bitops.tcl
index 30aa832c7..926f38295 100644
--- a/tests/unit/bitops.tcl
+++ b/tests/unit/bitops.tcl
@@ -43,6 +43,16 @@ start_server {tags {"bitops"}} {
         r bitcount no-key
     } 0
 
+    test {BITCOUNT returns 0 with out of range indexes} {
+        r set str "xxxx"
+        r bitcount str 4 10
+    } 0
+
+    test {BITCOUNT returns 0 with negative indexes where start > end} {
+        r set str "xxxx"
+        r bitcount str -6 -7
+    } 0
+
     catch {unset num}
     foreach vec [list "" "\xaa" "\x00\x00\xff" "foobar" "123"] {
         incr num
diff --git a/tests/unit/introspection-2.tcl b/tests/unit/introspection-2.tcl
new file mode 100644
index 000000000..350a8a016
--- /dev/null
+++ b/tests/unit/introspection-2.tcl
@@ -0,0 +1,23 @@
+start_server {tags {"introspection"}} {
+    test {TTL and TYPYE do not alter the last access time of a key} {
+        r set foo bar
+        after 3000
+        r ttl foo
+        r type foo
+        assert {[r object idletime foo] >= 2}
+    }
+
+    test {TOUCH alters the last access time of a key} {
+        r set foo bar
+        after 3000
+        r touch foo
+        assert {[r object idletime foo] < 2}
+    }
+
+    test {TOUCH returns the number of existing keys specified} {
+        r flushdb
+        r set key1 1
+        r set key2 2
+        r touch key0 key1 key2 key3
+    } 2
+}
diff --git a/tests/unit/scripting.tcl b/tests/unit/scripting.tcl
index 825a73ed3..be82e1559 100644
--- a/tests/unit/scripting.tcl
+++ b/tests/unit/scripting.tcl
@@ -142,7 +142,7 @@ start_server {tags {"scripting"}} {
 
     test {EVAL - Scripts can't run certain commands} {
         set e {}
-        catch {r eval {return redis.pcall('spop','x')} 0} e
+        catch {r eval {return redis.pcall('blpop','x',0)} 0} e
         set e
     } {*not allowed*}
 
diff --git a/tests/unit/type/list.tcl b/tests/unit/type/list.tcl
index e4d568cf1..1557082a2 100644
--- a/tests/unit/type/list.tcl
+++ b/tests/unit/type/list.tcl
@@ -507,7 +507,9 @@ start_server {
             create_list xlist "$large c"
             assert_equal 3 [r rpushx xlist d]
             assert_equal 4 [r lpushx xlist a]
-            assert_equal "a $large c d" [r lrange xlist 0 -1]
+            assert_equal 6 [r rpushx xlist 42 x]
+            assert_equal 9 [r lpushx xlist y3 y2 y1]
+            assert_equal "y1 y2 y3 a $large c d 42 x" [r lrange xlist 0 -1]
         }
 
         test "LINSERT - $type" {
diff --git a/utils/install_server.sh b/utils/install_server.sh
index 98e047e3d..3d920a125 100755
--- a/utils/install_server.sh
+++ b/utils/install_server.sh
@@ -25,9 +25,25 @@
 #
 ################################################################################
 #
-# Interactive service installer for redis server
-# this generates a redis config file and an /etc/init.d script, and installs them
-# this scripts should be run as root
+# Service installer for redis server, runs interactively by default.
+#
+# To run this script non-interactively (for automation/provisioning purposes),
+# feed the variables into the script. Any missing variables will be prompted!
+# Tip: Environment variables also support command substitution (see REDIS_EXECUTABLE)
+#
+# Example:
+#
+# sudo REDIS_PORT=1234 \
+# 		 REDIS_CONFIG_FILE=/etc/redis/1234.conf \
+# 		 REDIS_LOG_FILE=/var/log/redis_1234.log \
+# 		 REDIS_DATA_DIR=/var/lib/redis/1234 \
+# 		 REDIS_EXECUTABLE=`command -v redis-server` ./utils/install_server.sh
+#
+# This generates a redis config file and an /etc/init.d script, and installs them.
+#
+# /!\ This script should be run as root
+#
+################################################################################
 
 die () {
 	echo "ERROR: $1. Aborting!"
@@ -42,6 +58,7 @@ SCRIPTPATH=$(dirname $SCRIPT)
 
 #Initial defaults
 _REDIS_PORT=6379
+_MANUAL_EXECUTION=false
 
 echo "Welcome to the redis service installer"
 echo "This script will help you easily set up a running redis server"
@@ -53,47 +70,61 @@ if [ "$(id -u)" -ne 0 ] ; then
 	exit 1
 fi
 
-#Read the redis port
-read  -p "Please select the redis port for this instance: [$_REDIS_PORT] " REDIS_PORT
 if ! echo $REDIS_PORT | egrep -q '^[0-9]+$' ; then
-	echo "Selecting default: $_REDIS_PORT"
-	REDIS_PORT=$_REDIS_PORT
+	_MANUAL_EXECUTION=true
+	#Read the redis port
+	read  -p "Please select the redis port for this instance: [$_REDIS_PORT] " REDIS_PORT
+	if ! echo $REDIS_PORT | egrep -q '^[0-9]+$' ; then
+		echo "Selecting default: $_REDIS_PORT"
+		REDIS_PORT=$_REDIS_PORT
+	fi
 fi
 
-#read the redis config file
-_REDIS_CONFIG_FILE="/etc/redis/$REDIS_PORT.conf"
-read -p "Please select the redis config file name [$_REDIS_CONFIG_FILE] " REDIS_CONFIG_FILE
 if [ -z "$REDIS_CONFIG_FILE" ] ; then
-	REDIS_CONFIG_FILE=$_REDIS_CONFIG_FILE
-	echo "Selected default - $REDIS_CONFIG_FILE"
+	_MANUAL_EXECUTION=true
+	#read the redis config file
+	_REDIS_CONFIG_FILE="/etc/redis/$REDIS_PORT.conf"
+	read -p "Please select the redis config file name [$_REDIS_CONFIG_FILE] " REDIS_CONFIG_FILE
+	if [ -z "$REDIS_CONFIG_FILE" ] ; then
+		REDIS_CONFIG_FILE=$_REDIS_CONFIG_FILE
+		echo "Selected default - $REDIS_CONFIG_FILE"
+	fi
 fi
 
-#read the redis log file path
-_REDIS_LOG_FILE="/var/log/redis_$REDIS_PORT.log"
-read -p "Please select the redis log file name [$_REDIS_LOG_FILE] " REDIS_LOG_FILE
 if [ -z "$REDIS_LOG_FILE" ] ; then
-	REDIS_LOG_FILE=$_REDIS_LOG_FILE
-	echo "Selected default - $REDIS_LOG_FILE"
+	_MANUAL_EXECUTION=true
+	#read the redis log file path
+	_REDIS_LOG_FILE="/var/log/redis_$REDIS_PORT.log"
+	read -p "Please select the redis log file name [$_REDIS_LOG_FILE] " REDIS_LOG_FILE
+	if [ -z "$REDIS_LOG_FILE" ] ; then
+		REDIS_LOG_FILE=$_REDIS_LOG_FILE
+		echo "Selected default - $REDIS_LOG_FILE"
+	fi
 fi
 
-
-#get the redis data directory
-_REDIS_DATA_DIR="/var/lib/redis/$REDIS_PORT"
-read -p "Please select the data directory for this instance [$_REDIS_DATA_DIR] " REDIS_DATA_DIR
 if [ -z "$REDIS_DATA_DIR" ] ; then
-	REDIS_DATA_DIR=$_REDIS_DATA_DIR
-	echo "Selected default - $REDIS_DATA_DIR"
+	_MANUAL_EXECUTION=true
+	#get the redis data directory
+	_REDIS_DATA_DIR="/var/lib/redis/$REDIS_PORT"
+	read -p "Please select the data directory for this instance [$_REDIS_DATA_DIR] " REDIS_DATA_DIR
+	if [ -z "$REDIS_DATA_DIR" ] ; then
+		REDIS_DATA_DIR=$_REDIS_DATA_DIR
+		echo "Selected default - $REDIS_DATA_DIR"
+	fi
 fi
 
-#get the redis executable path
-_REDIS_EXECUTABLE=`command -v redis-server`
-read -p "Please select the redis executable path [$_REDIS_EXECUTABLE] " REDIS_EXECUTABLE
 if [ ! -x "$REDIS_EXECUTABLE" ] ; then
-	REDIS_EXECUTABLE=$_REDIS_EXECUTABLE
-
+	_MANUAL_EXECUTION=true
+	#get the redis executable path
+	_REDIS_EXECUTABLE=`command -v redis-server`
+	read -p "Please select the redis executable path [$_REDIS_EXECUTABLE] " REDIS_EXECUTABLE
 	if [ ! -x "$REDIS_EXECUTABLE" ] ; then
-		echo "Mmmmm...  it seems like you don't have a redis executable. Did you run make install yet?"
-		exit 1
+		REDIS_EXECUTABLE=$_REDIS_EXECUTABLE
+
+		if [ ! -x "$REDIS_EXECUTABLE" ] ; then
+			echo "Mmmmm...  it seems like you don't have a redis executable. Did you run make install yet?"
+			exit 1
+		fi
 	fi
 fi
 
@@ -112,7 +143,9 @@ echo "Data dir       : $REDIS_DATA_DIR"
 echo "Executable     : $REDIS_EXECUTABLE"
 echo "Cli Executable : $CLI_EXEC"
 
-read -p "Is this ok? Then press ENTER to go on or Ctrl-C to abort." _UNUSED_
+if $_MANUAL_EXECUTION == true ; then
+	read -p "Is this ok? Then press ENTER to go on or Ctrl-C to abort." _UNUSED_
+fi
 
 mkdir -p `dirname "$REDIS_CONFIG_FILE"` || die "Could not create redis config directory"
 mkdir -p `dirname "$REDIS_LOG_FILE"` || die "Could not create redis log dir"