[redis.git] / src / bitops.c

#include "redis.h"

/* -----------------------------------------------------------------------------
 * Helpers and low level bit functions.
 * -------------------------------------------------------------------------- */

/* This helper function used by GETBIT / SETBIT parses the bit offset arguemnt
 * making sure an error is returned if it is negative or if it overflows
 * Redis 512 MB limit for the string value. */
static int getBitOffsetFromArgument(redisClient *c, robj *o, size_t *offset) {
    long long loffset;
    char *err = "bit offset is not an integer or out of range";

    if (getLongLongFromObjectOrReply(c,o,&loffset,err) != REDIS_OK)
        return REDIS_ERR;

    /* Limit offset to 512MB in bytes */
    if ((loffset < 0) || ((unsigned long long)loffset >> 3) >= (512*1024*1024))
    {
        addReplyError(c,err);
        return REDIS_ERR;
    }

    *offset = (size_t)loffset;
    return REDIS_OK;
}

/* Count number of bits set in the binary array pointed by 's' and long
 * 'count' bytes. The implementation of this function is required to
 * work with a input string length up to 512 MB. */
long popcount(void *s, long count) {
    long bits = 0;
    unsigned char *p;
    uint32_t *p4 = s;
    static const unsigned char bitsinbyte[256] = {0,1,1,2,1,2,2,3,1,2,2,3,2,3,3,4,1,2,2,3,2,3,3,4,2,3,3,4,3,4,4,5,1,2,2,3,2,3,3,4,2,3,3,4,3,4,4,5,2,3,3,4,3,4,4,5,3,4,4,5,4,5,5,6,1,2,2,3,2,3,3,4,2,3,3,4,3,4,4,5,2,3,3,4,3,4,4,5,3,4,4,5,4,5,5,6,2,3,3,4,3,4,4,5,3,4,4,5,4,5,5,6,3,4,4,5,4,5,5,6,4,5,5,6,5,6,6,7,1,2,2,3,2,3,3,4,2,3,3,4,3,4,4,5,2,3,3,4,3,4,4,5,3,4,4,5,4,5,5,6,2,3,3,4,3,4,4,5,3,4,4,5,4,5,5,6,3,4,4,5,4,5,5,6,4,5,5,6,5,6,6,7,2,3,3,4,3,4,4,5,3,4,4,5,4,5,5,6,3,4,4,5,4,5,5,6,4,5,5,6,5,6,6,7,3,4,4,5,4,5,5,6,4,5,5,6,5,6,6,7,4,5,5,6,5,6,6,7,5,6,6,7,6,7,7,8};

    /* Count bits 16 bytes at a time */
    while(count>=16) {
        uint32_t aux1, aux2, aux3, aux4;

        aux1 = *p4++;
        aux2 = *p4++;
        aux3 = *p4++;
        aux4 = *p4++;
        count -= 16;

        aux1 = aux1 - ((aux1 >> 1) & 0x55555555);
        aux1 = (aux1 & 0x33333333) + ((aux1 >> 2) & 0x33333333);
        aux2 = aux2 - ((aux2 >> 1) & 0x55555555);
        aux2 = (aux2 & 0x33333333) + ((aux2 >> 2) & 0x33333333);
        aux3 = aux3 - ((aux3 >> 1) & 0x55555555);
        aux3 = (aux3 & 0x33333333) + ((aux3 >> 2) & 0x33333333);
        aux4 = aux4 - ((aux4 >> 1) & 0x55555555);
        aux4 = (aux4 & 0x33333333) + ((aux4 >> 2) & 0x33333333);
        bits += ((((aux1 + (aux1 >> 4)) & 0x0F0F0F0F) * 0x01010101) >> 24) +
                ((((aux2 + (aux2 >> 4)) & 0x0F0F0F0F) * 0x01010101) >> 24) +
                ((((aux3 + (aux3 >> 4)) & 0x0F0F0F0F) * 0x01010101) >> 24) +
                ((((aux4 + (aux4 >> 4)) & 0x0F0F0F0F) * 0x01010101) >> 24);
    }
    /* Count the remaining bytes */
    p = (unsigned char*)p4;
    while(count--) bits += bitsinbyte[*p++];
    return bits;
}

/* -----------------------------------------------------------------------------
 * Bits related string commands: GETBIT, SETBIT, BITCOUNT, BITOP.
 * -------------------------------------------------------------------------- */

#define BITOP_AND   0
#define BITOP_OR    1
#define BITOP_XOR   2
#define BITOP_NOT   3

/* SETBIT key offset bitvalue */
void setbitCommand(redisClient *c) {
    robj *o;
    char *err = "bit is not an integer or out of range";
    size_t bitoffset;
    int byte, bit;
    int byteval, bitval;
    long on;

    if (getBitOffsetFromArgument(c,c->argv[2],&bitoffset) != REDIS_OK)
        return;

    if (getLongFromObjectOrReply(c,c->argv[3],&on,err) != REDIS_OK)
        return;

    /* Bits can only be set or cleared... */
    if (on & ~1) {
        addReplyError(c,err);
        return;
    }

    o = lookupKeyWrite(c->db,c->argv[1]);
    if (o == NULL) {
        o = createObject(REDIS_STRING,sdsempty());
        dbAdd(c->db,c->argv[1],o);
    } else {
        if (checkType(c,o,REDIS_STRING)) return;

        /* 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));
            decrRefCount(decoded);
            dbOverwrite(c->db,c->argv[1],o);
        }
    }

    /* Grow sds value to the right length if necessary */
    byte = bitoffset >> 3;
    o->ptr = sdsgrowzero(o->ptr,byte+1);

    /* Get current values */
    byteval = ((char*)o->ptr)[byte];
    bit = 7 - (bitoffset & 0x7);
    bitval = byteval & (1 << bit);

    /* Update byte with new bit value and return original value */
    byteval &= ~(1 << bit);
    byteval |= ((on & 0x1) << bit);
    ((char*)o->ptr)[byte] = byteval;
    signalModifiedKey(c->db,c->argv[1]);
    server.dirty++;
    addReply(c, bitval ? shared.cone : shared.czero);
}

/* GETBIT key offset */
void getbitCommand(redisClient *c) {
    robj *o;
    char llbuf[32];
    size_t bitoffset;
    size_t byte, bit;
    size_t bitval = 0;

    if (getBitOffsetFromArgument(c,c->argv[2],&bitoffset) != REDIS_OK)
        return;

    if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL ||
        checkType(c,o,REDIS_STRING)) return;

    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 (byte < sdslen(o->ptr))
            bitval = ((char*)o->ptr)[byte] & (1 << bit);
    }

    addReply(c, bitval ? shared.cone : shared.czero);
}

/* BITOP op_name target_key src_key1 src_key2 src_key3 ... src_keyN */
void bitopCommand(redisClient *c) {
    char *opname = c->argv[1]->ptr;
    robj *o, *targetkey = c->argv[2];
    long op, j, numkeys;
    robj **objects;      /* Array of soruce objects. */
    unsigned char **src; /* Array of source strings pointers. */
    long *len, maxlen = 0; /* Array of length of src strings, and max len. */
    long minlen = 0;    /* Min len among the input keys. */
    unsigned char *res = NULL; /* Resulting string. */

    /* Parse the operation name. */
    if ((opname[0] == 'a' || opname[0] == 'A') && !strcasecmp(opname,"and"))
        op = BITOP_AND;
    else if((opname[0] == 'o' || opname[0] == 'O') && !strcasecmp(opname,"or"))
        op = BITOP_OR;
    else if((opname[0] == 'x' || opname[0] == 'X') && !strcasecmp(opname,"xor"))
        op = BITOP_XOR;
    else if((opname[0] == 'n' || opname[0] == 'N') && !strcasecmp(opname,"not"))
        op = BITOP_NOT;
    else {
        addReply(c,shared.syntaxerr);
        return;
    }

    /* Sanity check: NOT accepts only a single key argument. */
    if (op == BITOP_NOT && c->argc != 4) {
        addReplyError(c,"BITOP NOT must be called with a single source key.");
        return;
    }

    /* Lookup keys, and store pointers to the string objects into an array. */
    numkeys = c->argc - 3;
    src = zmalloc(sizeof(unsigned char*) * numkeys);
    len = zmalloc(sizeof(long) * numkeys);
    objects = zmalloc(sizeof(robj*) * numkeys);
    for (j = 0; j < numkeys; j++) {
        o = lookupKeyRead(c->db,c->argv[j+3]);
        /* Handle non-existing keys as empty strings. */
        if (o == NULL) {
            objects[j] = NULL;
            src[j] = NULL;
            len[j] = 0;
            minlen = 0;
            continue;
        }
        /* Return an error if one of the keys is not a string. */
        if (checkType(c,o,REDIS_STRING)) {
            for (j = j-1; j >= 0; j--) {
                if (objects[j])
                    decrRefCount(objects[j]);
            }
            zfree(src);
            zfree(len);
            zfree(objects);
            return;
        }
        objects[j] = getDecodedObject(o);
        src[j] = objects[j]->ptr;
        len[j] = sdslen(objects[j]->ptr);
        if (len[j] > maxlen) maxlen = len[j];
        if (j == 0 || len[j] < minlen) minlen = len[j];
    }

    /* Compute the bit operation, if at least one string is not empty. */
    if (maxlen) {
        res = (unsigned char*) sdsnewlen(NULL,maxlen);
        unsigned char output, byte;
        long i;

        /* Fast path: as far as we have data for all the input bitmaps we
         * can take a fast path that performs much better than the
         * vanilla algorithm. */
        j = 0;
        if (minlen && numkeys <= 16) {
            unsigned long *lp[16];
            unsigned long *lres = (unsigned long*) res;

            /* Note: sds pointer is always aligned to 8 byte boundary. */
            memcpy(lp,src,sizeof(unsigned long*)*numkeys);
            memcpy(res,src[0],minlen);

            /* Different branches per different operations for speed (sorry). */
            if (op == BITOP_AND) {
                while(minlen >= sizeof(unsigned long)*4) {
                    for (i = 1; i < numkeys; i++) {
                        lres[0] &= lp[i][0];
                        lres[1] &= lp[i][1];
                        lres[2] &= lp[i][2];
                        lres[3] &= lp[i][3];
                        lp[i]+=4;
                    }
                    lres+=4;
                    j += sizeof(unsigned long)*4;
                    minlen -= sizeof(unsigned long)*4;
                }
            } else if (op == BITOP_OR) {
                while(minlen >= sizeof(unsigned long)*4) {
                    for (i = 1; i < numkeys; i++) {
                        lres[0] |= lp[i][0];
                        lres[1] |= lp[i][1];
                        lres[2] |= lp[i][2];
                        lres[3] |= lp[i][3];
                        lp[i]+=4;
                    }
                    lres+=4;
                    j += sizeof(unsigned long)*4;
                    minlen -= sizeof(unsigned long)*4;
                }
            } else if (op == BITOP_XOR) {
                while(minlen >= sizeof(unsigned long)*4) {
                    for (i = 1; i < numkeys; i++) {
                        lres[0] ^= lp[i][0];
                        lres[1] ^= lp[i][1];
                        lres[2] ^= lp[i][2];
                        lres[3] ^= lp[i][3];
                        lp[i]+=4;
                    }
                    lres+=4;
                    j += sizeof(unsigned long)*4;
                    minlen -= sizeof(unsigned long)*4;
                }
            } else if (op == BITOP_NOT) {
                while(minlen >= sizeof(unsigned long)*4) {
                    lres[0] = ~lres[0];
                    lres[1] = ~lres[1];
                    lres[2] = ~lres[2];
                    lres[3] = ~lres[3];
                    lres+=4;
                    j += sizeof(unsigned long)*4;
                    minlen -= sizeof(unsigned long)*4;
                }
            }
        }

        /* j is set to the next byte to process by the previous loop. */
        for (; j < maxlen; j++) {
            output = (len[0] <= j) ? 0 : src[0][j];
            if (op == BITOP_NOT) output = ~output;
            for (i = 1; i < numkeys; i++) {
                byte = (len[i] <= j) ? 0 : src[i][j];
                switch(op) {
                case BITOP_AND: output &= byte; break;
                case BITOP_OR:  output |= byte; break;
                case BITOP_XOR: output ^= byte; break;
                }
            }
            res[j] = output;
        }
    }
    for (j = 0; j < numkeys; j++) {
        if (objects[j])
            decrRefCount(objects[j]);
    }
    zfree(src);
    zfree(len);
    zfree(objects);

    /* Store the computed value into the target key */
    if (maxlen) {
        o = createObject(REDIS_STRING,res);
        setKey(c->db,targetkey,o);
        decrRefCount(o);
    } else if (dbDelete(c->db,targetkey)) {
        signalModifiedKey(c->db,targetkey);
    }
    server.dirty++;
    addReplyLongLong(c,maxlen); /* Return the output string length in bytes. */
}

/* BITCOUNT key [start end] */
void bitcountCommand(redisClient *c) {
    robj *o;
    long start, end;
    unsigned char *p;
    char llbuf[32];
    size_t strlen;

    /* Lookup, check for type, and return 0 for non existing keys. */
    if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL ||
        checkType(c,o,REDIS_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 == REDIS_ENCODING_INT) {
        p = (unsigned char*) llbuf;
        strlen = ll2string(llbuf,sizeof(llbuf),(long)o->ptr);
    } else {
        p = (unsigned char*) o->ptr;
        strlen = sdslen(o->ptr);
    }

    /* Parse start/end range if any. */
    if (c->argc == 4) {
        if (getLongFromObjectOrReply(c,c->argv[2],&start,NULL) != REDIS_OK)
            return;
        if (getLongFromObjectOrReply(c,c->argv[3],&end,NULL) != REDIS_OK)
            return;
        /* Convert negative indexes */
        if (start < 0) start = strlen+start;
        if (end < 0) end = strlen+end;
        if (start < 0) start = 0;
        if (end < 0) end = 0;
        if ((unsigned)end >= strlen) end = strlen-1;
    } else if (c->argc == 2) {
        /* The whole string. */
        start = 0;
        end = strlen-1;
    } else {
        /* Syntax error. */
        addReply(c,shared.syntaxerr);
        return;
    }

    /* Precondition: end >= 0 && end < strlen, so the only condition where
     * zero can be returned is: start > end. */
    if (start > end) {
        addReply(c,shared.czero);
    } else {
        long bytes = end-start+1;

        addReplyLongLong(c,popcount(p+start,bytes));
    }
}
Commit	Line	Data
760e7765	1	#include "redis.h"
	2
	3	/* -----------------------------------------------------------------------------
dbbbe49e	4	* Helpers and low level bit functions.
760e7765	5	* -------------------------------------------------------------------------- */
760e7765	6
760e7765	7	/* This helper function used by GETBIT / SETBIT parses the bit offset arguemnt
	8	* making sure an error is returned if it is negative or if it overflows
	9	* Redis 512 MB limit for the string value. */
	10	static int getBitOffsetFromArgument(redisClient c, robj o, size_t *offset) {
	11	long long loffset;
	12	char *err = "bit offset is not an integer or out of range";
	13
	14	if (getLongLongFromObjectOrReply(c,o,&loffset,err) != REDIS_OK)
	15	return REDIS_ERR;
	16
	17	/* Limit offset to 512MB in bytes */
	18	if ((loffset < 0) \|\| ((unsigned long long)loffset >> 3) >= (51210241024))
	19	{
	20	addReplyError(c,err);
	21	return REDIS_ERR;
	22	}
	23
	24	*offset = (size_t)loffset;
	25	return REDIS_OK;
	26	}
	27
dbbbe49e	28	/* Count number of bits set in the binary array pointed by 's' and long
	29	* 'count' bytes. The implementation of this function is required to
	30	* work with a input string length up to 512 MB. */
	31	long popcount(void *s, long count) {
	32	long bits = 0;
343d3bd2	33	unsigned char *p;
343d3bd2	34	uint32_t *p4 = s;
dbbbe49e	35	static const unsigned char bitsinbyte[256] = {0,1,1,2,1,2,2,3,1,2,2,3,2,3,3,4,1,2,2,3,2,3,3,4,2,3,3,4,3,4,4,5,1,2,2,3,2,3,3,4,2,3,3,4,3,4,4,5,2,3,3,4,3,4,4,5,3,4,4,5,4,5,5,6,1,2,2,3,2,3,3,4,2,3,3,4,3,4,4,5,2,3,3,4,3,4,4,5,3,4,4,5,4,5,5,6,2,3,3,4,3,4,4,5,3,4,4,5,4,5,5,6,3,4,4,5,4,5,5,6,4,5,5,6,5,6,6,7,1,2,2,3,2,3,3,4,2,3,3,4,3,4,4,5,2,3,3,4,3,4,4,5,3,4,4,5,4,5,5,6,2,3,3,4,3,4,4,5,3,4,4,5,4,5,5,6,3,4,4,5,4,5,5,6,4,5,5,6,5,6,6,7,2,3,3,4,3,4,4,5,3,4,4,5,4,5,5,6,3,4,4,5,4,5,5,6,4,5,5,6,5,6,6,7,3,4,4,5,4,5,5,6,4,5,5,6,5,6,6,7,4,5,5,6,5,6,6,7,5,6,6,7,6,7,7,8};
dbbbe49e	36
7c34643f	37	/* Count bits 16 bytes at a time */
	38	while(count>=16) {
	39	uint32_t aux1, aux2, aux3, aux4;
	40
	41	aux1 = *p4++;
	42	aux2 = *p4++;
	43	aux3 = *p4++;
	44	aux4 = *p4++;
	45	count -= 16;
	46
	47	aux1 = aux1 - ((aux1 >> 1) & 0x55555555);
	48	aux1 = (aux1 & 0x33333333) + ((aux1 >> 2) & 0x33333333);
	49	aux2 = aux2 - ((aux2 >> 1) & 0x55555555);
	50	aux2 = (aux2 & 0x33333333) + ((aux2 >> 2) & 0x33333333);
	51	aux3 = aux3 - ((aux3 >> 1) & 0x55555555);
	52	aux3 = (aux3 & 0x33333333) + ((aux3 >> 2) & 0x33333333);
	53	aux4 = aux4 - ((aux4 >> 1) & 0x55555555);
	54	aux4 = (aux4 & 0x33333333) + ((aux4 >> 2) & 0x33333333);
	55	bits += ((((aux1 + (aux1 >> 4)) & 0x0F0F0F0F) * 0x01010101) >> 24) +
	56	((((aux2 + (aux2 >> 4)) & 0x0F0F0F0F) * 0x01010101) >> 24) +
	57	((((aux3 + (aux3 >> 4)) & 0x0F0F0F0F) * 0x01010101) >> 24) +
	58	((((aux4 + (aux4 >> 4)) & 0x0F0F0F0F) * 0x01010101) >> 24);
343d3bd2	59	}
	60	/* Count the remaining bytes */
	61	p = (unsigned char*)p4;
dbbbe49e	62	while(count--) bits += bitsinbyte[*p++];
	63	return bits;
	64	}
	65
	66	/* -----------------------------------------------------------------------------
	67	* Bits related string commands: GETBIT, SETBIT, BITCOUNT, BITOP.
	68	* -------------------------------------------------------------------------- */
	69
	70	#define BITOP_AND 0
	71	#define BITOP_OR 1
	72	#define BITOP_XOR 2
	73	#define BITOP_NOT 3
	74
760e7765	75	/* SETBIT key offset bitvalue */
	76	void setbitCommand(redisClient *c) {
	77	robj *o;
	78	char *err = "bit is not an integer or out of range";
	79	size_t bitoffset;
	80	int byte, bit;
	81	int byteval, bitval;
	82	long on;
	83
	84	if (getBitOffsetFromArgument(c,c->argv[2],&bitoffset) != REDIS_OK)
	85	return;
	86
	87	if (getLongFromObjectOrReply(c,c->argv[3],&on,err) != REDIS_OK)
	88	return;
	89
	90	/* Bits can only be set or cleared... */
	91	if (on & ~1) {
	92	addReplyError(c,err);
	93	return;
	94	}
	95
	96	o = lookupKeyWrite(c->db,c->argv[1]);
	97	if (o == NULL) {
	98	o = createObject(REDIS_STRING,sdsempty());
	99	dbAdd(c->db,c->argv[1],o);
	100	} else {
	101	if (checkType(c,o,REDIS_STRING)) return;
	102
	103	/* Create a copy when the object is shared or encoded. */
	104	if (o->refcount != 1 \|\| o->encoding != REDIS_ENCODING_RAW) {
	105	robj *decoded = getDecodedObject(o);
	106	o = createStringObject(decoded->ptr, sdslen(decoded->ptr));
	107	decrRefCount(decoded);
	108	dbOverwrite(c->db,c->argv[1],o);
	109	}
	110	}
	111
	112	/* Grow sds value to the right length if necessary */
	113	byte = bitoffset >> 3;
	114	o->ptr = sdsgrowzero(o->ptr,byte+1);
	115
	116	/* Get current values */
	117	byteval = ((char*)o->ptr)[byte];
	118	bit = 7 - (bitoffset & 0x7);
	119	bitval = byteval & (1 << bit);
	120
	121	/* Update byte with new bit value and return original value */
	122	byteval &= ~(1 << bit);
	123	byteval \|= ((on & 0x1) << bit);
	124	((char*)o->ptr)[byte] = byteval;
	125	signalModifiedKey(c->db,c->argv[1]);
	126	server.dirty++;
	127	addReply(c, bitval ? shared.cone : shared.czero);
	128	}
	129
	130	/* GETBIT key offset */
	131	void getbitCommand(redisClient *c) {
	132	robj *o;
	133	char llbuf[32];
	134	size_t bitoffset;
	135	size_t byte, bit;
	136	size_t bitval = 0;
	137
	138	if (getBitOffsetFromArgument(c,c->argv[2],&bitoffset) != REDIS_OK)
139	return;
140
141	if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL \|\|
142	checkType(c,o,REDIS_STRING)) return;
143
144	byte = bitoffset >> 3;
145	bit = 7 - (bitoffset & 0x7);
146	if (o->encoding != REDIS_ENCODING_RAW) {
147	if (byte < (size_t)ll2string(llbuf,sizeof(llbuf),(long)o->ptr))
148	bitval = llbuf[byte] & (1 << bit);
149	} else {
150	if (byte < sdslen(o->ptr))
151	bitval = ((char*)o->ptr)[byte] & (1 << bit);
152	}
153
154	addReply(c, bitval ? shared.cone : shared.czero);
155	}
156
157	/* BITOP op_name target_key src_key1 src_key2 src_key3 ... src_keyN */
158	void bitopCommand(redisClient *c) {
159	char *opname = c->argv[1]->ptr;
160	robj o, targetkey = c->argv[2];
161	long op, j, numkeys;
fa4a5d59	162	robj *objects; / Array of soruce objects. */
760e7765	163	unsigned char *src; / Array of source strings pointers. */
760e7765	164	long len, maxlen = 0; / Array of length of src strings, and max len. */
d8668038	165	long minlen = 0; /* Min len among the input keys. */
760e7765	166	unsigned char res = NULL; / Resulting string. */
	167
	168	/* Parse the operation name. */
	169	if ((opname[0] == 'a' \|\| opname[0] == 'A') && !strcasecmp(opname,"and"))
	170	op = BITOP_AND;
	171	else if((opname[0] == 'o' \|\| opname[0] == 'O') && !strcasecmp(opname,"or"))
	172	op = BITOP_OR;
	173	else if((opname[0] == 'x' \|\| opname[0] == 'X') && !strcasecmp(opname,"xor"))
	174	op = BITOP_XOR;
	175	else if((opname[0] == 'n' \|\| opname[0] == 'N') && !strcasecmp(opname,"not"))
	176	op = BITOP_NOT;
	177	else {
	178	addReply(c,shared.syntaxerr);
	179	return;
	180	}
	181
	182	/* Sanity check: NOT accepts only a single key argument. */
	183	if (op == BITOP_NOT && c->argc != 4) {
	184	addReplyError(c,"BITOP NOT must be called with a single source key.");
	185	return;
	186	}
	187
	188	/* Lookup keys, and store pointers to the string objects into an array. */
	189	numkeys = c->argc - 3;
	190	src = zmalloc(sizeof(unsigned char) numkeys);
	191	len = zmalloc(sizeof(long) * numkeys);
fa4a5d59	192	objects = zmalloc(sizeof(robj) numkeys);
760e7765	193	for (j = 0; j < numkeys; j++) {
	194	o = lookupKeyRead(c->db,c->argv[j+3]);
	195	/* Handle non-existing keys as empty strings. */
	196	if (o == NULL) {
fa4a5d59	197	objects[j] = NULL;
760e7765	198	src[j] = NULL;
760e7765	199	len[j] = 0;
1419406e	200	minlen = 0;
760e7765	201	continue;
	202	}
	203	/* Return an error if one of the keys is not a string. */
	204	if (checkType(c,o,REDIS_STRING)) {
fa4a5d59	205	for (j = j-1; j >= 0; j--) {
	206	if (objects[j])
	207	decrRefCount(objects[j]);
	208	}
760e7765	209	zfree(src);
760e7765	210	zfree(len);
fa4a5d59	211	zfree(objects);
760e7765	212	return;
760e7765	213	}
fa4a5d59	214	objects[j] = getDecodedObject(o);
	215	src[j] = objects[j]->ptr;
	216	len[j] = sdslen(objects[j]->ptr);
760e7765	217	if (len[j] > maxlen) maxlen = len[j];
d8668038	218	if (j == 0 \|\| len[j] < minlen) minlen = len[j];
760e7765	219	}
	220
	221	/* Compute the bit operation, if at least one string is not empty. */
	222	if (maxlen) {
	223	res = (unsigned char*) sdsnewlen(NULL,maxlen);
	224	unsigned char output, byte;
	225	long i;
	226
d8668038	227	/* Fast path: as far as we have data for all the input bitmaps we
	228	* can take a fast path that performs much better than the
	229	* vanilla algorithm. */
	230	j = 0;
	231	if (minlen && numkeys <= 16) {
	232	unsigned long *lp[16];
	233	unsigned long lres = (unsigned long) res;
	234
	235	/* Note: sds pointer is always aligned to 8 byte boundary. */
	236	memcpy(lp,src,sizeof(unsigned long)numkeys);
	237	memcpy(res,src[0],minlen);
	238
	239	/* Different branches per different operations for speed (sorry). */
	240	if (op == BITOP_AND) {
	241	while(minlen >= sizeof(unsigned long)*4) {
	242	for (i = 1; i < numkeys; i++) {
	243	lres[0] &= lp[i][0];
	244	lres[1] &= lp[i][1];
	245	lres[2] &= lp[i][2];
	246	lres[3] &= lp[i][3];
	247	lp[i]+=4;
	248	}
	249	lres+=4;
	250	j += sizeof(unsigned long)*4;
	251	minlen -= sizeof(unsigned long)*4;
	252	}
	253	} else if (op == BITOP_OR) {
	254	while(minlen >= sizeof(unsigned long)*4) {
	255	for (i = 1; i < numkeys; i++) {
	256	lres[0] \|= lp[i][0];
	257	lres[1] \|= lp[i][1];
	258	lres[2] \|= lp[i][2];
	259	lres[3] \|= lp[i][3];
	260	lp[i]+=4;
	261	}
	262	lres+=4;
	263	j += sizeof(unsigned long)*4;
	264	minlen -= sizeof(unsigned long)*4;
	265	}
	266	} else if (op == BITOP_XOR) {
	267	while(minlen >= sizeof(unsigned long)*4) {
	268	for (i = 1; i < numkeys; i++) {
	269	lres[0] ^= lp[i][0];
	270	lres[1] ^= lp[i][1];
	271	lres[2] ^= lp[i][2];
	272	lres[3] ^= lp[i][3];
	273	lp[i]+=4;
	274	}
	275	lres+=4;
	276	j += sizeof(unsigned long)*4;
	277	minlen -= sizeof(unsigned long)*4;
	278	}
	279	} else if (op == BITOP_NOT) {
	280	while(minlen >= sizeof(unsigned long)*4) {
	281	lres[0] = ~lres[0];
	282	lres[1] = ~lres[1];
	283	lres[2] = ~lres[2];
	284	lres[3] = ~lres[3];
	285	lres+=4;
	286	j += sizeof(unsigned long)*4;
	287	minlen -= sizeof(unsigned long)*4;
	288	}
	289	}
	290	}
291
292	/* j is set to the next byte to process by the previous loop. */
293	for (; j < maxlen; j++) {
760e7765	294	output = (len[0] <= j) ? 0 : src[0][j];
	295	if (op == BITOP_NOT) output = ~output;
	296	for (i = 1; i < numkeys; i++) {
	297	byte = (len[i] <= j) ? 0 : src[i][j];
	298	switch(op) {
	299	case BITOP_AND: output &= byte; break;
	300	case BITOP_OR: output \|= byte; break;
	301	case BITOP_XOR: output ^= byte; break;
	302	}
	303	}
	304	res[j] = output;
	305	}
	306	}
fa4a5d59	307	for (j = 0; j < numkeys; j++) {
	308	if (objects[j])
	309	decrRefCount(objects[j]);
	310	}
760e7765	311	zfree(src);
760e7765	312	zfree(len);
fa4a5d59	313	zfree(objects);
760e7765	314
	315	/* Store the computed value into the target key */
	316	if (maxlen) {
	317	o = createObject(REDIS_STRING,res);
	318	setKey(c->db,targetkey,o);
	319	decrRefCount(o);
	320	} else if (dbDelete(c->db,targetkey)) {
	321	signalModifiedKey(c->db,targetkey);
	322	}
	323	server.dirty++;
	324	addReplyLongLong(c,maxlen); /* Return the output string length in bytes. */
	325	}
	326
	327	/* BITCOUNT key [start end] */
	328	void bitcountCommand(redisClient *c) {
760e7765	329	robj *o;
	330	long start, end;
	331	unsigned char *p;
	332	char llbuf[32];
	333	size_t strlen;
	334
	335	/* Lookup, check for type, and return 0 for non existing keys. */
	336	if ((o = lookupKeyReadOrReply(c,c->argv[1],shared.czero)) == NULL \|\|
	337	checkType(c,o,REDIS_STRING)) return;
	338
	339	/* Set the 'p' pointer to the string, that can be just a stack allocated
	340	* array if our string was integer encoded. */
	341	if (o->encoding == REDIS_ENCODING_INT) {
	342	p = (unsigned char*) llbuf;
	343	strlen = ll2string(llbuf,sizeof(llbuf),(long)o->ptr);
	344	} else {
	345	p = (unsigned char*) o->ptr;
	346	strlen = sdslen(o->ptr);
	347	}
	348
	349	/* Parse start/end range if any. */
	350	if (c->argc == 4) {
	351	if (getLongFromObjectOrReply(c,c->argv[2],&start,NULL) != REDIS_OK)
	352	return;
	353	if (getLongFromObjectOrReply(c,c->argv[3],&end,NULL) != REDIS_OK)
	354	return;
	355	/* Convert negative indexes */
	356	if (start < 0) start = strlen+start;
	357	if (end < 0) end = strlen+end;
	358	if (start < 0) start = 0;
	359	if (end < 0) end = 0;
	360	if ((unsigned)end >= strlen) end = strlen-1;
	361	} else if (c->argc == 2) {
	362	/* The whole string. */
	363	start = 0;
	364	end = strlen-1;
	365	} else {
	366	/* Syntax error. */
	367	addReply(c,shared.syntaxerr);
	368	return;
	369	}
	370
	371	/* Precondition: end >= 0 && end < strlen, so the only condition where
	372	* zero can be returned is: start > end. */
	373	if (start > end) {
	374	addReply(c,shared.czero);
	375	} else {
dbbbe49e	376	long bytes = end-start+1;
760e7765	377
dbbbe49e	378	addReplyLongLong(c,popcount(p+start,bytes));
760e7765	379	}
760e7765	380	}