Redis 位数组实现

Redis 提供了 SETBIT，GETBIT，BITCOUNT，BITOP 4个命令用于处理二进制位数组。

• SETBIT 用于为为数组指定偏移量上的二进制位设置值

  setbit {key} {offset} {value}
  eg: 
  127.0.0.1:6379> SETBIT bit 30 1
  (integer) 0
  

• GETBIT 用于获取位数组指定偏移量上的二进制位设置值

  getbit {key} {offset}
  eg: 
  127.0.0.1:6379> GETBIT tmp 30
  (integer) 1
  

• BITCOUNT 用于统计二进制位数组中，值为1的元素数量

  bitcount {key} [start end]
  eg:  
  

• BITOP 对多个位数组进行按位与，按位或，按位异或，取反运算

• bitmaps间的运算

  • 交集
  • 并集
  • 非
  • 异或

位数组相关源码在 bitops.c 文件中。

位数组表示

Redis 使用字符串对象来表示位数组，因为字符串对象使用的SDS是二进制安全的。

image.png

SETBIT命令实现

调用栈：

(gdb) bt
#0  setbitCommand (c=0x5555558dddc8) at bitops.c:213
#1  0x0000555555577528 in call (c=0x5555558dddc8, flags=7) at redis.c:2441
#2  0x0000555555578044 in processCommand (c=0x5555558dddc8) at redis.c:2766
#3  0x00005555555865c4 in processInputBuffer (c=0x5555558dddc8) at networking.c:1539
#4  0x00005555555868b2 in readQueryFromClient (el=0x555555898158, fd=6, privdata=0x5555558dddc8, mask=1)
    at networking.c:1631
#5  0x00005555555701bf in aeProcessEvents (eventLoop=0x555555898158, flags=3) at ae.c:576
#6  0x0000555555570380 in aeMain (eventLoop=0x555555898158) at ae.c:635
#7  0x000055555557b30a in main (argc=1, argv=0x7fffffffe4d8) at redis.c:4079

setbitCommand 实现：

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;

    // 获取 offset 参数
    if (getBitOffsetFromArgument(c,c->argv[2],&bitoffset) != REDIS_OK)
        return;

    // 获取 value 参数
    if (getLongFromObjectOrReply(c,c->argv[3],&on,err) != REDIS_OK)
        return;

    /* Bits can only be set or cleared... */
    // value 参数的值只能是 0 或者 1 ，否则返回错误
    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;

        o = dbUnshareStringValue(c->db,c->argv[1],o);
    }

    /* Grow sds value to the right length if necessary */
    // 计算容纳 offset 参数所指定的偏移量所需的字节数
    // 如果 o 对象的字节不够长的话，就扩展它
    // 长度的计算公式是 bitoffset >> 3 + 1
    // 比如 30 >> 3 + 1 = 4 ，也即是为了设置 offset 30 ，
    // 我们需要创建一个 4 字节（32 位长的 SDS）
    byte = bitoffset >> 3;
    o->ptr = sdsgrowzero(o->ptr,byte+1);

    // 将指针定位到要设置的位所在的字节上
    byteval = ((uint8_t*)o->ptr)[byte];
    // 定位到要设置的位上面
    bit = 7 - (bitoffset & 0x7);
    // 记录位现在的值
    bitval = byteval & (1 << bit);

    // 更新字节中的位，设置它的值为 on 参数的值
    byteval &= ~(1 << bit);
    byteval |= ((on & 0x1) << bit);
    ((uint8_t*)o->ptr)[byte] = byteval;
    
    ...
}

GETBIT 命令实现

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

    // 读取 offset 参数
    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;

    // 计算出 offset 所指定的位所在的字节
    byte = bitoffset >> 3;
    // 计算出位所在的位置
    bit = 7 - (bitoffset & 0x7);

    // 取出位
    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);
}

BITCOUNT命令实现

BITCOUNT 命令用于统计给定位数组中，值为1的二进制位的数量，这个如果要高效的实现这个命令，需要一些精巧的算法。

• 遍历算法
  最简单直接的算法就是逐个遍历二进制位。

• 查表算法
  我们可以以8位二进制所表示的数包含多少个1，建立一个查找表，每次就可以8位一次的遍历。

  编码	1的个数
  0000 0000	0
  0000 0001	1
  0000 0010	1
  0000 0011	2
  ...	...
  1111 11110	7
  1111 11111	8

  查表算法，典型的是以空间换时间。我们还可以16位来建立查找表，不过查表需要的内存也会增加。同时，越大的表也会带来CPU缓存不命中出现的概率也会越高的问题。

• SWAR算法

• Redis 中的实现
  Redis中的实现用到了 查表 和 SWAR 2种算法。

  • 查表算法使用了8位的表
  • SWAR 算法，BITCOUNT 命令每次循环中载入 128 个二进制位，然后调用四次32位 SWAR算法 来计算这128个二进制位的汉明重量

  在执命令时，程序会根据未处理的二进制位的数量来决定使用哪种算法：

  • 如果未处理的二进制位的数量大于等于128位，那么程序使用
  • 如果未处理的二进制位的数量小于128位，那么程序使用查表算法来计算二进制位的汉明重量

BITOP命令实现

BITOP 所有操作都是用C语言内置的位操作符来实现。