Introduction of a new string encoding: EMBSTRmemopt

Previously two string encodings were used for string objects:

1) REDIS_ENCODING_RAW: a string object with obj->ptr pointing to an sds
stirng.

2) REDIS_ENCODING_INT: a string object where the obj->ptr void pointer
is casted to a long.

This commit introduces a experimental new encoding called
REDIS_ENCODING_EMBSTR that implements an object represented by an sds
string that is not modifiable but allocated in the same memory chunk as
the robj structure itself.

The chunk looks like the following:

+--------------+-----------+------------+--------+----+
| robj data... | robj->ptr | sds header | string | \0 |
+--------------+-----+-----+------------+--------+----+
                     |                       ^
                     +-----------------------+

The robj->ptr points to the contiguous sds string data, so the object
can be manipulated with the same functions used to manipulate plan
string objects, however we need just on malloc and one free in order to
allocate or release this kind of objects. Moreover it has better cache
locality.

This new allocation strategy should benefit both the memory usage and
the performances. A performance gain between 60 and 70% was observed
during micro-benchmarks, however there is more work to do to evaluate
the performance impact and the memory usage behavior.

17 files changed, 156 insertions, 70 deletions

diff --git a/src/aof.c b/src/aof.c
index 441ccaf18..b8cd40ef5 100644
--- a/src/aof.c
+++ b/src/aof.c
@@ -561,7 +561,7 @@ int rioWriteBulkObject(rio *r, robj *obj) {
      * in a child process when this function is called). */
     if (obj->encoding == REDIS_ENCODING_INT) {
         return rioWriteBulkLongLong(r,(long)obj->ptr);
-    } else if (obj->encoding == REDIS_ENCODING_RAW) {
+    } else if (sdsEncodedObject(obj)) {
         return rioWriteBulkString(r,obj->ptr,sdslen(obj->ptr));
     } else {
         redisPanic("Unknown string encoding");
diff --git a/src/bitops.c b/src/bitops.c
index 39d24ab7d..0925a3b4e 100644
--- a/src/bitops.c
+++ b/src/bitops.c
@@ -103,7 +103,7 @@ void setbitCommand(redisClient *c) {
         /* Create a copy when the object is shared or encoded. */
         if (o->refcount != 1 || o->encoding != REDIS_ENCODING_RAW) {
             robj *decoded = getDecodedObject(o);
-            o = createStringObject(decoded->ptr, sdslen(decoded->ptr));
+            o = createRawStringObject(decoded->ptr, sdslen(decoded->ptr));
             decrRefCount(decoded);
             dbOverwrite(c->db,c->argv[1],o);
         }
@@ -143,12 +143,12 @@ void getbitCommand(redisClient *c) {
 
     byte = bitoffset >> 3;
     bit = 7 - (bitoffset & 0x7);
-    if (o->encoding != REDIS_ENCODING_RAW) {
-        if (byte < (size_t)ll2string(llbuf,sizeof(llbuf),(long)o->ptr))
-            bitval = llbuf[byte] & (1 << bit);
-    } else {
+    if (sdsEncodedObject(o)) {
         if (byte < sdslen(o->ptr))
             bitval = ((uint8_t*)o->ptr)[byte] & (1 << bit);
+    } else {
+        if (byte < (size_t)ll2string(llbuf,sizeof(llbuf),(long)o->ptr))
+            bitval = llbuf[byte] & (1 << bit);
     }
 
     addReply(c, bitval ? shared.cone : shared.czero);
diff --git a/src/cluster.c b/src/cluster.c
index 57243132d..0384860de 100644
--- a/src/cluster.c
+++ b/src/cluster.c
@@ -1632,7 +1632,7 @@ void migrateCommand(redisClient *c) {
     }
     redisAssertWithInfo(c,NULL,rioWriteBulkCount(&cmd,'*',4));
     redisAssertWithInfo(c,NULL,rioWriteBulkString(&cmd,"RESTORE",7));
-    redisAssertWithInfo(c,NULL,c->argv[3]->encoding == REDIS_ENCODING_RAW);
+    redisAssertWithInfo(c,NULL,sdsEncodedObject(c->argv[3]));
     redisAssertWithInfo(c,NULL,rioWriteBulkString(&cmd,c->argv[3]->ptr,sdslen(c->argv[3]->ptr)));
     redisAssertWithInfo(c,NULL,rioWriteBulkLongLong(&cmd,ttl));
 
diff --git a/src/config.c b/src/config.c
index 2470f374b..ee4b5f9bb 100644
--- a/src/config.c
+++ b/src/config.c
@@ -427,8 +427,8 @@ void loadServerConfig(char *filename, char *options) {
 void configSetCommand(redisClient *c) {
     robj *o;
     long long ll;
-    redisAssertWithInfo(c,c->argv[2],c->argv[2]->encoding == REDIS_ENCODING_RAW);
-    redisAssertWithInfo(c,c->argv[2],c->argv[3]->encoding == REDIS_ENCODING_RAW);
+    redisAssertWithInfo(c,c->argv[2],sdsEncodedObject(c->argv[2]));
+    redisAssertWithInfo(c,c->argv[3],sdsEncodedObject(c->argv[3]));
     o = c->argv[3];
 
     if (!strcasecmp(c->argv[2]->ptr,"dbfilename")) {
@@ -717,7 +717,7 @@ void configGetCommand(redisClient *c) {
     char *pattern = o->ptr;
     char buf[128];
     int matches = 0;
-    redisAssertWithInfo(c,o,o->encoding == REDIS_ENCODING_RAW);
+    redisAssertWithInfo(c,o,sdsEncodedObject(o));
 
     /* String values */
     config_get_string_field("dbfilename",server.rdb_filename);
diff --git a/src/debug.c b/src/debug.c
index 566b2b959..6951df149 100644
--- a/src/debug.c
+++ b/src/debug.c
@@ -334,9 +334,7 @@ void _redisAssertPrintClientInfo(redisClient *c) {
         char buf[128];
         char *arg;
 
-        if (c->argv[j]->type == REDIS_STRING &&
-            c->argv[j]->encoding == REDIS_ENCODING_RAW)
-        {
+        if (c->argv[j]->type == REDIS_STRING && sdsEncodedObject(c->argv[j])) {
             arg = (char*) c->argv[j]->ptr;
         } else {
             snprintf(buf,sizeof(buf),"Object type: %d, encoding: %d",
@@ -352,7 +350,7 @@ void redisLogObjectDebugInfo(robj *o) {
     redisLog(REDIS_WARNING,"Object type: %d", o->type);
     redisLog(REDIS_WARNING,"Object encoding: %d", o->encoding);
     redisLog(REDIS_WARNING,"Object refcount: %d", o->refcount);
-    if (o->type == REDIS_STRING && o->encoding == REDIS_ENCODING_RAW) {
+    if (o->type == REDIS_STRING && sdsEncodedObject(o)) {
         redisLog(REDIS_WARNING,"Object raw string len: %d", sdslen(o->ptr));
         if (sdslen(o->ptr) < 4096)
             redisLog(REDIS_WARNING,"Object raw string content: \"%s\"", (char*)o->ptr);
diff --git a/src/networking.c b/src/networking.c
index 3bc084f7d..631a09248 100644
--- a/src/networking.c
+++ b/src/networking.c
@@ -147,12 +147,15 @@ void _addReplyObjectToList(redisClient *c, robj *o) {
     if (listLength(c->reply) == 0) {
         incrRefCount(o);
         listAddNodeTail(c->reply,o);
-        c->reply_bytes += zmalloc_size_sds(o->ptr);
+        c->reply_bytes += (o->encoding == REDIS_ENCODING_RAW) ?
+                          zmalloc_size_sds(o->ptr) :
+                          sdslen(o->ptr);
     } else {
         tail = listNodeValue(listLast(c->reply));
 
         /* Append to this object when possible. */
         if (tail->ptr != NULL &&
+            tail->encoding == REDIS_ENCODING_RAW &&
             sdslen(tail->ptr)+sdslen(o->ptr) <= REDIS_REPLY_CHUNK_BYTES)
         {
             c->reply_bytes -= zmalloc_size_sds(tail->ptr);
@@ -162,7 +165,9 @@ void _addReplyObjectToList(redisClient *c, robj *o) {
         } else {
             incrRefCount(o);
             listAddNodeTail(c->reply,o);
-            c->reply_bytes += zmalloc_size_sds(o->ptr);
+            c->reply_bytes += (o->encoding == REDIS_ENCODING_RAW) ?
+                              zmalloc_size_sds(o->ptr) :
+                              sdslen(o->ptr);
         }
     }
     asyncCloseClientOnOutputBufferLimitReached(c);
@@ -185,7 +190,7 @@ void _addReplySdsToList(redisClient *c, sds s) {
         tail = listNodeValue(listLast(c->reply));
 
         /* Append to this object when possible. */
-        if (tail->ptr != NULL &&
+        if (tail->ptr != NULL && tail->encoding == REDIS_ENCODING_RAW &&
             sdslen(tail->ptr)+sdslen(s) <= REDIS_REPLY_CHUNK_BYTES)
         {
             c->reply_bytes -= zmalloc_size_sds(tail->ptr);
@@ -210,12 +215,14 @@ void _addReplyStringToList(redisClient *c, char *s, size_t len) {
         robj *o = createStringObject(s,len);
 
         listAddNodeTail(c->reply,o);
-        c->reply_bytes += zmalloc_size_sds(o->ptr);
+        c->reply_bytes += (o->encoding == REDIS_ENCODING_RAW) ?
+                          zmalloc_size_sds(o->ptr) :
+                          sdslen(o->ptr);
     } else {
         tail = listNodeValue(listLast(c->reply));
 
         /* Append to this object when possible. */
-        if (tail->ptr != NULL &&
+        if (tail->ptr != NULL && tail->encoding == REDIS_ENCODING_RAW &&
             sdslen(tail->ptr)+len <= REDIS_REPLY_CHUNK_BYTES)
         {
             c->reply_bytes -= zmalloc_size_sds(tail->ptr);
@@ -226,7 +233,9 @@ void _addReplyStringToList(redisClient *c, char *s, size_t len) {
             robj *o = createStringObject(s,len);
 
             listAddNodeTail(c->reply,o);
-            c->reply_bytes += zmalloc_size_sds(o->ptr);
+            c->reply_bytes += (o->encoding == REDIS_ENCODING_RAW) ?
+                              zmalloc_size_sds(o->ptr) :
+                              sdslen(o->ptr);
         }
     }
     asyncCloseClientOnOutputBufferLimitReached(c);
@@ -247,7 +256,7 @@ void addReply(redisClient *c, robj *obj) {
      * If the encoding is RAW and there is room in the static buffer
      * we'll be able to send the object to the client without
      * messing with its page. */
-    if (obj->encoding == REDIS_ENCODING_RAW) {
+    if (sdsEncodedObject(obj)) {
         if (_addReplyToBuffer(c,obj->ptr,sdslen(obj->ptr)) != REDIS_OK)
             _addReplyObjectToList(c,obj);
     } else if (obj->encoding == REDIS_ENCODING_INT) {
@@ -359,6 +368,7 @@ void setDeferredMultiBulkLength(redisClient *c, void *node, long length) {
 
     len = listNodeValue(ln);
     len->ptr = sdscatprintf(sdsempty(),"*%ld\r\n",length);
+    len->encoding = REDIS_ENCODING_RAW; /* in case it was an EMBSTR. */
     c->reply_bytes += zmalloc_size_sds(len->ptr);
     if (ln->next != NULL) {
         next = listNodeValue(ln->next);
@@ -425,7 +435,7 @@ void addReplyMultiBulkLen(redisClient *c, long length) {
 void addReplyBulkLen(redisClient *c, robj *obj) {
     size_t len;
 
-    if (obj->encoding == REDIS_ENCODING_RAW) {
+    if (sdsEncodedObject(obj)) {
         len = sdslen(obj->ptr);
     } else {
         long n = (long)obj->ptr;
@@ -691,7 +701,9 @@ void sendReplyToClient(aeEventLoop *el, int fd, void *privdata, int mask) {
         } else {
             o = listNodeValue(listFirst(c->reply));
             objlen = sdslen(o->ptr);
-            objmem = zmalloc_size_sds(o->ptr);
+            objmem = (o->encoding == REDIS_ENCODING_RAW) ?
+                        zmalloc_size_sds(o->ptr) :
+                        sdslen(o->ptr);
 
             if (objlen == 0) {
                 listDelNode(c->reply,listFirst(c->reply));
diff --git a/src/object.c b/src/object.c
index ba7ea323a..6c6d3d863 100644
--- a/src/object.c
+++ b/src/object.c
@@ -14,10 +14,47 @@ robj *createObject(int type, void *ptr) {
     return o;
 }
 
-robj *createStringObject(char *ptr, size_t len) {
+/* Create a string object with encoding REDIS_ENCODING_RAW, that is a plain
+ * string object where o->ptr points to a proper sds string. */
+robj *createRawStringObject(char *ptr, size_t len) {
     return createObject(REDIS_STRING,sdsnewlen(ptr,len));
 }
 
+/* Create a string object with encoding REDIS_ENCODING_EMBSTR, that is
+ * an object where the sds string is actually an unmodifiable string
+ * allocated in the same chunk as the object itself. */
+robj *createEmbeddedStringObject(char *ptr, size_t len) {
+    robj *o = zmalloc(sizeof(robj)+sizeof(struct sdshdr)+len+1);
+    struct sdshdr *sh = (void*)(o+1);
+
+    o->type = REDIS_STRING;
+    o->encoding = REDIS_ENCODING_EMBSTR;
+    o->ptr = sh+1;
+    o->refcount = 1;
+    o->lru = server.lruclock;
+
+    sh->len = len;
+    sh->free = 0;
+    if (ptr) {
+        memcpy(sh->buf,ptr,len);
+        sh->buf[len] = '\0';
+    } else {
+        memset(sh->buf,0,len+1);
+    }
+    return o;
+}
+
+/* Create a string object with EMBSTR encoding if it is smaller than
+ * REIDS_ENCODING_EMBSTR_SIZE_LIMIT, otherwise the RAW encoding is
+ * used. */
+#define REDIS_ENCODING_EMBSTR_SIZE_LIMIT 32
+robj *createStringObject(char *ptr, size_t len) {
+    if (len <= REDIS_ENCODING_EMBSTR_SIZE_LIMIT)
+        return createEmbeddedStringObject(ptr,len);
+    else
+        return createRawStringObject(ptr,len);
+}
+
 robj *createStringObjectFromLongLong(long long value) {
     robj *o;
     if (value >= 0 && value < REDIS_SHARED_INTEGERS) {
@@ -59,9 +96,33 @@ robj *createStringObjectFromLongDouble(long double value) {
     return createStringObject(buf,len);
 }
 
+/* Duplicate a string object, with the guarantee that the returned object
+ * has the same encoding as the original one.
+ *
+ * This function also guarantees that duplicating a small integere object
+ * (or a string object that contains a representation of a small integer)
+ * 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) {
-    redisAssertWithInfo(NULL,o,o->encoding == REDIS_ENCODING_RAW);
-    return createStringObject(o->ptr,sdslen(o->ptr));
+    robj *d;
+
+    redisAssert(o->type == REDIS_STRING);
+
+    switch(o->encoding) {
+    case REDIS_ENCODING_RAW:
+        return createRawStringObject(o->ptr,sdslen(o->ptr));
+    case REDIS_ENCODING_EMBSTR:
+        return createEmbeddedStringObject(o->ptr,sdslen(o->ptr));
+    case REDIS_ENCODING_INT:
+        d = createObject(REDIS_STRING, NULL);
+        d->encoding = REDIS_ENCODING_INT;
+        d->ptr = o->ptr;
+        return d;
+    default:
+        redisPanic("Wrong encoding.");
+        break;
+    }
 }
 
 robj *createListObject(void) {
@@ -244,7 +305,7 @@ robj *tryObjectEncoding(robj *o) {
     long value;
     sds s = o->ptr;
 
-    if (o->encoding != REDIS_ENCODING_RAW)
+    if (o->encoding == REDIS_ENCODING_INT)
         return o; /* Already encoded */
 
     /* It's not safe to encode shared objects: shared objects can be shared
@@ -256,7 +317,17 @@ robj *tryObjectEncoding(robj *o) {
     redisAssertWithInfo(NULL,o,o->type == REDIS_STRING);
 
     /* Check if we can represent this string as a long integer */
-    if (!string2l(s,sdslen(s),&value)) return o;
+    if (!string2l(s,sdslen(s),&value)) {
+        /* Integer encoding not possible. Check if we can use EMBSTR. */
+        if (sdslen(s) <= REDIS_ENCODING_EMBSTR_SIZE_LIMIT) {
+            robj *emb = createEmbeddedStringObject(s,sdslen(s));
+            decrRefCount(o);
+            return emb;
+        } else {
+            /* Otherwise return the original object. */
+            return o;
+        }
+    }
 
     /* Ok, this object can be encoded...
      *
@@ -270,8 +341,8 @@ robj *tryObjectEncoding(robj *o) {
         incrRefCount(shared.integers[value]);
         return shared.integers[value];
     } else {
+        if (o->encoding == REDIS_ENCODING_RAW) sdsfree(o->ptr);
         o->encoding = REDIS_ENCODING_INT;
-        sdsfree(o->ptr);
         o->ptr = (void*) value;
         return o;
     }
@@ -282,7 +353,7 @@ robj *tryObjectEncoding(robj *o) {
 robj *getDecodedObject(robj *o) {
     robj *dec;
 
-    if (o->encoding == REDIS_ENCODING_RAW) {
+    if (sdsEncodedObject(o)) {
         incrRefCount(o);
         return o;
     }
@@ -311,19 +382,19 @@ int compareStringObjects(robj *a, robj *b) {
     int bothsds = 1;
 
     if (a == b) return 0;
-    if (a->encoding != REDIS_ENCODING_RAW) {
+    if (sdsEncodedObject(a)) {
+        astr = a->ptr;
+    } else {
         ll2string(bufa,sizeof(bufa),(long) a->ptr);
         astr = bufa;
         bothsds = 0;
-    } else {
-        astr = a->ptr;
     }
-    if (b->encoding != REDIS_ENCODING_RAW) {
+    if (sdsEncodedObject(b)) {
+        bstr = b->ptr;
+    } else {
         ll2string(bufb,sizeof(bufb),(long) b->ptr);
         bstr = bufb;
         bothsds = 0;
-    } else {
-        bstr = b->ptr;
     }
     return bothsds ? sdscmp(astr,bstr) : strcmp(astr,bstr);
 }
@@ -333,7 +404,10 @@ int compareStringObjects(robj *a, robj *b) {
  * this function is faster then checking for (compareStringObject(a,b) == 0)
  * because it can perform some more optimization. */
 int equalStringObjects(robj *a, robj *b) {
-    if (a->encoding != REDIS_ENCODING_RAW && b->encoding != REDIS_ENCODING_RAW){
+    if (a->encoding == REDIS_ENCODING_INT &&
+        b->encoding == REDIS_ENCODING_INT){
+        /* If both strings are integer encoded just check if the stored
+         * long is the same. */
         return a->ptr == b->ptr;
     } else {
         return compareStringObjects(a,b) == 0;
@@ -342,7 +416,7 @@ int equalStringObjects(robj *a, robj *b) {
 
 size_t stringObjectLen(robj *o) {
     redisAssertWithInfo(NULL,o,o->type == REDIS_STRING);
-    if (o->encoding == REDIS_ENCODING_RAW) {
+    if (sdsEncodedObject(o)) {
         return sdslen(o->ptr);
     } else {
         char buf[32];
@@ -359,7 +433,7 @@ int getDoubleFromObject(robj *o, double *target) {
         value = 0;
     } else {
         redisAssertWithInfo(NULL,o,o->type == REDIS_STRING);
-        if (o->encoding == REDIS_ENCODING_RAW) {
+        if (sdsEncodedObject(o)) {
             errno = 0;
             value = strtod(o->ptr, &eptr);
             if (isspace(((char*)o->ptr)[0]) || eptr[0] != '\0' ||
@@ -397,7 +471,7 @@ int getLongDoubleFromObject(robj *o, long double *target) {
         value = 0;
     } else {
         redisAssertWithInfo(NULL,o,o->type == REDIS_STRING);
-        if (o->encoding == REDIS_ENCODING_RAW) {
+        if (sdsEncodedObject(o)) {
             errno = 0;
             value = strtold(o->ptr, &eptr);
             if (isspace(((char*)o->ptr)[0]) || eptr[0] != '\0' ||
@@ -435,7 +509,7 @@ int getLongLongFromObject(robj *o, long long *target) {
         value = 0;
     } else {
         redisAssertWithInfo(NULL,o,o->type == REDIS_STRING);
-        if (o->encoding == REDIS_ENCODING_RAW) {
+        if (sdsEncodedObject(o)) {
             errno = 0;
             value = strtoll(o->ptr, &eptr, 10);
             if (isspace(((char*)o->ptr)[0]) || eptr[0] != '\0' ||
@@ -490,6 +564,7 @@ char *strEncoding(int encoding) {
     case REDIS_ENCODING_ZIPLIST: return "ziplist";
     case REDIS_ENCODING_INTSET: return "intset";
     case REDIS_ENCODING_SKIPLIST: return "skiplist";
+    case REDIS_ENCODING_EMBSTR: return "embstr";
     default: return "unknown";
     }
 }
diff --git a/src/rdb.c b/src/rdb.c
index 9190ea514..9db9af0e1 100644
--- a/src/rdb.c
+++ b/src/rdb.c
@@ -296,7 +296,7 @@ int rdbSaveStringObject(rio *rdb, robj *obj) {
     if (obj->encoding == REDIS_ENCODING_INT) {
         return rdbSaveLongLongAsStringObject(rdb,(long)obj->ptr);
     } else {
-        redisAssertWithInfo(NULL,obj,obj->encoding == REDIS_ENCODING_RAW);
+        redisAssertWithInfo(NULL,obj,sdsEncodedObject(obj));
         return rdbSaveRawString(rdb,obj->ptr,sdslen(obj->ptr));
     }
 }
@@ -757,7 +757,7 @@ robj *rdbLoadObject(int rdbtype, rio *rdb) {
             /* If we are using a ziplist and the value is too big, convert
              * the object to a real list. */
             if (o->encoding == REDIS_ENCODING_ZIPLIST &&
-                ele->encoding == REDIS_ENCODING_RAW &&
+                sdsEncodedObject(ele) &&
                 sdslen(ele->ptr) > server.list_max_ziplist_value)
                     listTypeConvert(o,REDIS_ENCODING_LINKEDLIST);
 
@@ -831,9 +831,8 @@ robj *rdbLoadObject(int rdbtype, rio *rdb) {
             if (rdbLoadDoubleValue(rdb,&score) == -1) return NULL;
 
             /* Don't care about integer-encoded strings. */
-            if (ele->encoding == REDIS_ENCODING_RAW &&
-                sdslen(ele->ptr) > maxelelen)
-                    maxelelen = sdslen(ele->ptr);
+            if (sdsEncodedObject(ele) && sdslen(ele->ptr) > maxelelen)
+                maxelelen = sdslen(ele->ptr);
 
             znode = zslInsert(zs->zsl,score,ele);
             dictAdd(zs->dict,ele,&znode->score);
@@ -865,10 +864,10 @@ robj *rdbLoadObject(int rdbtype, rio *rdb) {
             /* Load raw strings */
             field = rdbLoadStringObject(rdb);
             if (field == NULL) return NULL;
-            redisAssert(field->encoding == REDIS_ENCODING_RAW);
+            redisAssert(sdsEncodedObject(field));
             value = rdbLoadStringObject(rdb);
             if (value == NULL) return NULL;
-            redisAssert(field->encoding == REDIS_ENCODING_RAW);
+            redisAssert(sdsEncodedObject(value));
 
             /* Add pair to ziplist */
             o->ptr = ziplistPush(o->ptr, field->ptr, sdslen(field->ptr), ZIPLIST_TAIL);
diff --git a/src/redis.c b/src/redis.c
index 070aeb095..8f6eab6b2 100644
--- a/src/redis.c
+++ b/src/redis.c
@@ -457,7 +457,7 @@ int dictEncObjKeyCompare(void *privdata, const void *key1,
 unsigned int dictEncObjHash(const void *key) {
     robj *o = (robj*) key;
 
-    if (o->encoding == REDIS_ENCODING_RAW) {
+    if (sdsEncodedObject(o)) {
         return dictGenHashFunction(o->ptr, sdslen((sds)o->ptr));
     } else {
         if (o->encoding == REDIS_ENCODING_INT) {
diff --git a/src/redis.h b/src/redis.h
index 24efef57b..795e67aac 100644
--- a/src/redis.h
+++ b/src/redis.h
@@ -105,6 +105,7 @@
 #define REDIS_ENCODING_ZIPLIST 5 /* Encoded as ziplist */
 #define REDIS_ENCODING_INTSET 6  /* Encoded as intset */
 #define REDIS_ENCODING_SKIPLIST 7  /* Encoded as skiplist */
+#define REDIS_ENCODING_EMBSTR 8  /* Embedded sds string encoding */
 
 /* 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
@@ -960,6 +961,8 @@ void freeZsetObject(robj *o);
 void freeHashObject(robj *o);
 robj *createObject(int type, void *ptr);
 robj *createStringObject(char *ptr, size_t len);
+robj *createRawStringObject(char *ptr, size_t len);
+robj *createEmbeddedStringObject(char *ptr, size_t len);
 robj *dupStringObject(robj *o);
 int isObjectRepresentableAsLongLong(robj *o, long long *llongval);
 robj *tryObjectEncoding(robj *o);
@@ -985,6 +988,7 @@ char *strEncoding(int encoding);
 int compareStringObjects(robj *a, robj *b);
 int equalStringObjects(robj *a, robj *b);
 unsigned long estimateObjectIdleTime(robj *o);
+#define sdsEncodedObject(objptr) (objptr->encoding == REDIS_ENCODING_RAW || objptr->encoding == REDIS_ENCODING_EMBSTR)
 
 /* Synchronous I/O with timeout */
 ssize_t syncWrite(int fd, char *ptr, ssize_t size, long long timeout);
diff --git a/src/slowlog.c b/src/slowlog.c
index 53c44a017..fb2b48008 100644
--- a/src/slowlog.c
+++ b/src/slowlog.c
@@ -32,7 +32,7 @@ slowlogEntry *slowlogCreateEntry(robj **argv, int argc, long long duration) {
         } else {
             /* Trim too long strings as well... */
             if (argv[j]->type == REDIS_STRING &&
-                argv[j]->encoding == REDIS_ENCODING_RAW &&
+                sdsEncodedObject(argv[j]) &&
                 sdslen(argv[j]->ptr) > SLOWLOG_ENTRY_MAX_STRING)
             {
                 sds s = sdsnewlen(argv[j]->ptr, SLOWLOG_ENTRY_MAX_STRING);
diff --git a/src/sort.c b/src/sort.c
index e5178cd0d..afc79de95 100644
--- a/src/sort.c
+++ b/src/sort.c
@@ -292,7 +292,7 @@ void sortCommand(redisClient *c) {
             if (alpha) {
                 if (sortby) vector[j].u.cmpobj = getDecodedObject(byval);
             } else {
-                if (byval->encoding == REDIS_ENCODING_RAW) {
+                if (sdsEncodedObject(byval)) {
                     char *eptr;
 
                     vector[j].u.score = strtod(byval->ptr,&eptr);
diff --git a/src/t_hash.c b/src/t_hash.c
index aa021b038..e21ee1f1d 100644
--- a/src/t_hash.c
+++ b/src/t_hash.c
@@ -14,7 +14,7 @@ void hashTypeTryConversion(robj *o, robj **argv, int start, int end) {
     if (o->encoding != REDIS_ENCODING_ZIPLIST) return;
 
     for (i = start; i <= end; i++) {
-        if (argv[i]->encoding == REDIS_ENCODING_RAW &&
+        if (sdsEncodedObject(argv[i]) &&
             sdslen(argv[i]->ptr) > server.hash_max_ziplist_value)
         {
             hashTypeConvert(o, REDIS_ENCODING_HT);
diff --git a/src/t_list.c b/src/t_list.c
index 77e40eb6b..e962bda5d 100644
--- a/src/t_list.c
+++ b/src/t_list.c
@@ -11,7 +11,7 @@ void signalListAsReady(redisClient *c, robj *key);
  * objects are never too long. */
 void listTypeTryConversion(robj *subject, robj *value) {
     if (subject->encoding != REDIS_ENCODING_ZIPLIST) return;
-    if (value->encoding == REDIS_ENCODING_RAW &&
+    if (sdsEncodedObject(value) &&
         sdslen(value->ptr) > server.list_max_ziplist_value)
             listTypeConvert(subject,REDIS_ENCODING_LINKEDLIST);
 }
@@ -205,7 +205,7 @@ void listTypeInsert(listTypeEntry *entry, robj *value, int where) {
 int listTypeEqual(listTypeEntry *entry, robj *o) {
     listTypeIterator *li = entry->li;
     if (li->encoding == REDIS_ENCODING_ZIPLIST) {
-        redisAssertWithInfo(NULL,o,o->encoding == REDIS_ENCODING_RAW);
+        redisAssertWithInfo(NULL,o,sdsEncodedObject(o));
         return ziplistCompare(entry->zi,o->ptr,sdslen(o->ptr));
     } else if (li->encoding == REDIS_ENCODING_LINKEDLIST) {
         return equalStringObjects(o,listNodeValue(entry->ln));
@@ -311,7 +311,7 @@ void pushxGenericCommand(redisClient *c, robj *refval, robj *val, int where) {
     if (refval != NULL) {
         /* Note: we expect refval to be string-encoded because it is *not* the
          * last argument of the multi-bulk LINSERT. */
-        redisAssertWithInfo(c,refval,refval->encoding == REDIS_ENCODING_RAW);
+        redisAssertWithInfo(c,refval,sdsEncodedObject(refval));
 
         /* We're not sure if this value can be inserted yet, but we cannot
          * convert the list inside the iterator. We don't want to loop over
diff --git a/src/t_set.c b/src/t_set.c
index fada1095a..d4afead3e 100644
--- a/src/t_set.c
+++ b/src/t_set.c
@@ -447,11 +447,8 @@ void srandmemberWithCountCommand(redisClient *c) {
 
             if (encoding == REDIS_ENCODING_INTSET) {
                 retval = dictAdd(d,createStringObjectFromLongLong(llele),NULL);
-            } else if (ele->encoding == REDIS_ENCODING_RAW) {
+            } else {
                 retval = dictAdd(d,dupStringObject(ele),NULL);
-            } else if (ele->encoding == REDIS_ENCODING_INT) {
-                retval = dictAdd(d,
-                    createStringObjectFromLongLong((long)ele->ptr),NULL);
             }
             redisAssert(retval == DICT_OK);
         }
@@ -479,10 +476,8 @@ void srandmemberWithCountCommand(redisClient *c) {
             encoding = setTypeRandomElement(set,&ele,&llele);
             if (encoding == REDIS_ENCODING_INTSET) {
                 ele = createStringObjectFromLongLong(llele);
-            } else if (ele->encoding == REDIS_ENCODING_RAW) {
+            } else {
                 ele = dupStringObject(ele);
-            } else if (ele->encoding == REDIS_ENCODING_INT) {
-                ele = createStringObjectFromLongLong((long)ele->ptr);
             }
             /* Try to add the object to the dictionary. If it already exists
              * free it, otherwise increment the number of objects we have
diff --git a/src/t_string.c b/src/t_string.c
index 1e29a6133..c20bf754b 100644
--- a/src/t_string.c
+++ b/src/t_string.c
@@ -130,7 +130,7 @@ void setrangeCommand(redisClient *c) {
         /* Create a copy when the object is shared or encoded. */
         if (o->refcount != 1 || o->encoding != REDIS_ENCODING_RAW) {
             robj *decoded = getDecodedObject(o);
-            o = createStringObject(decoded->ptr, sdslen(decoded->ptr));
+            o = createRawStringObject(decoded->ptr, sdslen(decoded->ptr));
             decrRefCount(decoded);
             dbOverwrite(c->db,c->argv[1],o);
         }
@@ -344,7 +344,7 @@ void appendCommand(redisClient *c) {
         /* If the object is shared or encoded, we have to make a copy */
         if (o->refcount != 1 || o->encoding != REDIS_ENCODING_RAW) {
             robj *decoded = getDecodedObject(o);
-            o = createStringObject(decoded->ptr, sdslen(decoded->ptr));
+            o = createRawStringObject(decoded->ptr, sdslen(decoded->ptr));
             decrRefCount(decoded);
             dbOverwrite(c->db,c->argv[1],o);
         }
diff --git a/src/t_zset.c b/src/t_zset.c
index 4812709e1..3653af23a 100644
--- a/src/t_zset.c
+++ b/src/t_zset.c
@@ -617,7 +617,7 @@ unsigned char *zzlInsertAt(unsigned char *zl, unsigned char *eptr, robj *ele, do
     int scorelen;
     size_t offset;
 
-    redisAssertWithInfo(NULL,ele,ele->encoding == REDIS_ENCODING_RAW);
+    redisAssertWithInfo(NULL,ele,sdsEncodedObject(ele));
     scorelen = d2string(scorebuf,sizeof(scorebuf),score);
     if (eptr == NULL) {
         zl = ziplistPush(zl,ele->ptr,sdslen(ele->ptr),ZIPLIST_TAIL);
@@ -1315,7 +1315,7 @@ int zuiLongLongFromValue(zsetopval *val) {
             if (val->ele->encoding == REDIS_ENCODING_INT) {
                 val->ell = (long)val->ele->ptr;
                 val->flags |= OPVAL_VALID_LL;
-            } else if (val->ele->encoding == REDIS_ENCODING_RAW) {
+            } else if (sdsEncodedObject(val->ele)) {
                 if (string2ll(val->ele->ptr,sdslen(val->ele->ptr),&val->ell))
                     val->flags |= OPVAL_VALID_LL;
             } else {
@@ -1350,7 +1350,7 @@ int zuiBufferFromValue(zsetopval *val) {
             if (val->ele->encoding == REDIS_ENCODING_INT) {
                 val->elen = ll2string((char*)val->_buf,sizeof(val->_buf),(long)val->ele->ptr);
                 val->estr = val->_buf;
-            } else if (val->ele->encoding == REDIS_ENCODING_RAW) {
+            } else if (sdsEncodedObject(val->ele)) {
                 val->elen = sdslen(val->ele->ptr);
                 val->estr = val->ele->ptr;
             } else {
@@ -1576,9 +1576,10 @@ void zunionInterGenericCommand(redisClient *c, robj *dstkey, int op) {
                     dictAdd(dstzset->dict,tmp,&znode->score);
                     incrRefCount(tmp); /* added to dictionary */
 
-                    if (tmp->encoding == REDIS_ENCODING_RAW)
+                    if (sdsEncodedObject(tmp)) {
                         if (sdslen(tmp->ptr) > maxelelen)
                             maxelelen = sdslen(tmp->ptr);
+                    }
                 }
             }
         }
@@ -1618,9 +1619,10 @@ void zunionInterGenericCommand(redisClient *c, robj *dstkey, int op) {
                 dictAdd(dstzset->dict,tmp,&znode->score);
                 incrRefCount(zval.ele); /* added to dictionary */
 
-                if (tmp->encoding == REDIS_ENCODING_RAW)
+                if (sdsEncodedObject(tmp)) {
                     if (sdslen(tmp->ptr) > maxelelen)
                         maxelelen = sdslen(tmp->ptr);
+                }
             }
         }
     } else {
@@ -2093,7 +2095,8 @@ void zrankGenericCommand(redisClient *c, int reverse) {
         checkType(c,zobj,REDIS_ZSET)) return;
     llen = zsetLength(zobj);
 
-    redisAssertWithInfo(c,ele,ele->encoding == REDIS_ENCODING_RAW);
+    redisAssertWithInfo(c,ele,sdsEncodedObject(ele));
+
     if (zobj->encoding == REDIS_ENCODING_ZIPLIST) {
         unsigned char *zl = zobj->ptr;
         unsigned char *eptr, *sptr;