1 files changed, 93 insertions, 18 deletions

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";
     }
 }