Implement left and right shift BITOP operators

Former-commit-id: ba365298ed37a76f0a8630e0ec6c86393293aebe

src/bitops.cpp

@@ -396,6 +396,8 @@ void printBits(unsigned char *p, unsigned long count) {
 #define BITOP_OR    1
 #define BITOP_XOR   2
 #define BITOP_NOT   3
+#define BITOP_LSHIFT 4
+#define BITOP_RSHIFT 5
 
 #define BITFIELDOP_GET 0
 #define BITFIELDOP_SET 1
@@ -592,7 +594,8 @@ void bitopCommand(client *c) {
     char *opname = szFromObj(c->argv[1]);
     robj *targetkey = c->argv[2];
     robj_roptr o;
-    unsigned long op, j, numkeys;
+    int op;
+    unsigned long j, numkeys;
     robj_roptr *objects;      /* Array of source objects. */
     unsigned char **src; /* Array of source strings pointers. */
     unsigned long *len, maxlen = 0; /* Array of length of src strings,
@@ -609,6 +612,10 @@ void bitopCommand(client *c) {
         op = BITOP_XOR;
     else if((opname[0] == 'n' || opname[0] == 'N') && !strcasecmp(opname,"not"))
         op = BITOP_NOT;
+    else if (!strcasecmp(opname, "lshift"))
+        op = BITOP_LSHIFT;
+    else if (!strcasecmp(opname, "rshift"))
+        op = BITOP_RSHIFT;
     else {
         addReply(c,shared.syntaxerr);
         return;
@@ -620,8 +627,25 @@ void bitopCommand(client *c) {
         return;
     }
 
+    bool fShiftOp = (op == BITOP_LSHIFT) || (op == BITOP_RSHIFT);
+    long long shift = 0;
+    
+    /* Sanity check: SHIFTS only accept a single arg and an integer */
+    if (fShiftOp) {
+        if (c->argc != 5) {
+            addReplyError(c,"BITOP SHIFT must be called with a single source key and an integer shift.");
+            return;
+        }
+        if (getLongLongFromObject(c->argv[4], &shift) != C_OK) {
+            addReplyError(c, "BITOP SHIFT's last parameter must be an integer");
+            return;
+        }
+        if (op == BITOP_RSHIFT)
+            shift = -shift;
+    }
+
     /* Lookup keys, and store pointers to the string objects into an array. */
-    numkeys = c->argc - 3;
+    numkeys = c->argc - (fShiftOp ? 4 : 3);
     src = (unsigned char**)zmalloc(sizeof(unsigned char*) * numkeys, MALLOC_LOCAL);
     len = (unsigned long*)zmalloc(sizeof(long) * numkeys, MALLOC_LOCAL);
     objects = (robj_roptr*)zmalloc(sizeof(robj_roptr) * numkeys, MALLOC_LOCAL);
@@ -654,6 +678,54 @@ void bitopCommand(client *c) {
         if (j == 0 || len[j] < minlen) minlen = len[j];
     }
 
+    if (fShiftOp)
+    {
+        long newlen = (long)maxlen + shift/CHAR_BIT;
+        if (shift > 0 && (shift % CHAR_BIT) != 0)
+            newlen++;
+
+        if (newlen < 0)
+            newlen = 0;
+        
+        if (newlen)
+        {
+            res = (unsigned char*) sdsnewlen(NULL,newlen);
+            if (shift >= 0)
+            {   // left shift
+                long byteoffset = shift/CHAR_BIT;
+                memset(res, 0, byteoffset);
+                long srcLen = newlen - byteoffset - ((shift % CHAR_BIT) ? 1 : 0);
+
+                // now the bitshift+copy
+                unsigned bitshift = shift % CHAR_BIT;
+                unsigned char carry = 0;
+                for (long iSrc = 0; iSrc < srcLen; ++iSrc)
+                {
+                    res[byteoffset+iSrc] = (src[0][iSrc] << bitshift) | carry;
+                    carry = src[0][iSrc] >> (CHAR_BIT - bitshift);
+                }
+                if (bitshift)
+                    res[newlen-1] = carry;
+            } 
+            else 
+            {   // right shift
+                long byteoffset = -shift/CHAR_BIT;
+                unsigned bitshift = -shift % CHAR_BIT;
+                if (bitshift)
+                    ++byteoffset;
+                res[0] = (src[0][byteoffset] << (CHAR_BIT-bitshift));
+                if (byteoffset > 0)
+                    res[0] |= (src[0][byteoffset-1] >> bitshift);
+                for (long idx = 1; idx < newlen; ++idx)
+                {
+                    res[idx] = (src[0][byteoffset+idx] << (CHAR_BIT-bitshift)) | (src[0][byteoffset+idx-1] >> bitshift);
+                }
+            }
+        }
+        maxlen = newlen;    // this is to ensure we DEL below if newlen was 0
+    }
+    else
+    {
         /* Compute the bit operation, if at least one string is not empty. */
         if (maxlen) {
             res = (unsigned char*) sdsnewlen(NULL,maxlen);
@@ -744,6 +816,7 @@ void bitopCommand(client *c) {
                 res[j] = output;
             }
         }
+    }
     for (j = 0; j < numkeys; j++) {
         if (objects[j])
             decrRefCount(objects[j]);

src/server.h

@@ -729,7 +729,7 @@ typedef struct redisObject {
                             * LFU data (least significant 8 bits frequency
                             * and most significant 16 bits access time). */
 private:
-    mutable std::atomic<unsigned> refcount;
+    mutable std::atomic<unsigned> refcount {0};
 public:
     uint64_t mvcc_tstamp;
     void *m_ptr;

tests/unit/bitops.tcl

@@ -214,6 +214,64 @@ start_server {tags {"bitops"}} {
         r bitop or x a b
     } {32}
 
+    test {BITOP lshift size} {
+        r set a " "
+        r bitop lshift x a 1
+    } {2}
+
+    test {BITOP rshift size} {
+        r set a " "
+        r bitop rshift x a 1
+    } {1}
+
+    test {BITOP rshift 0 byte} {
+        r set a " "
+        r bitop rshift x a 8
+    } {0}
+
+    test {BITOP rshift underflow} {
+        r set a " "
+        r bitop rshift x a 65
+    } {0}
+
+    test {BITOP lshift string} {
+        r set a "abcdefg"
+        r bitop lshift x a 8
+        r get x
+    } "\x00abcdefg"
+
+    test {BITOP lshift char} {
+        r set a "\xAA"
+        r bitop lshift x a 4
+        r get x
+    } "\xA0\x0A"
+
+    test {BITOP rshift char} {
+        r set a "\xAA"
+        r bitop rshift x a 3
+        r get x
+    } "\x15"
+
+    test {BITOP lshift carry} {
+        r set a "\xFF"
+        r bitop lshift x a 1
+        r get x
+    } "\xFE\x01"
+
+    test {BITOP rshift carry} {
+        r set a "\x00\xFF"
+        r bitop rshift x a 1
+        r get x
+    } "\x80\x7F"
+
+    test {BITOP rshift reciprocal} {
+        r flushdb
+        r set a "abcdefg"
+        r bitop lshift b a 14
+        r bitop rshift res b 14
+        r get res
+    } "abcdefg\x00"
+
     test {BITPOS bit=0 with empty key returns 0} {
         r del str
         r bitpos str 0