redis源码解析（十）hyperloglog之稀疏存储和密集存储

谁践踏了优雅 2022-01-21 05:59 131阅读 0赞

### 1. 前言 ###

本文主要介绍hyperloglog的稀疏存储和密集存储结构及其实现。

### 2. 基本介绍 ###

hyperloglog是一种不精确的去重统计算法，最基本的点在于先存储数据后进行计算。因此先研究存储数据的结构体很有必要性。hyperloglog采用密集存储和稀疏存储两种存储方式。当数据较多时采用密集存储，而数据较少时，使用稀疏存储来保存大量连续的0、少量相同或不同的基数。

##### 2.1 密集存储的取值、设置 #####

密集存储使用6位存储一个数据，用连续的16384个6位数据组成了整个基数集合。因为仅用6位存储因此需要考虑不在一个字节的偏移问题。 假设密集存储中某一片段如下所示：

|11000000|22221111|33333322|55444444

这里由于存储格式为LSB到MSB，因此直观看起来仿佛是分割开的，起始是连续的6个1， 6个2，诸如此类。假设我们想要获取pos = 1，即“111111”的位置，第一个含有该数据的字节b0计算公式为

fb = 6 * pos % 8 -> 6
      此处有 b0 = 6 * pos / 8 = 0

也就是

|11000000|  <- Our byte at b0

b0右移6字节、b1左移2字节，则结果如下

* Right shift b0 of 'fb' bits.
     *
     *   +--------+
     *   |11000000|  <- Initial value of b0
     *   |00000011|  <- After right shift of 6 pos.
     *   +--------+
     *
     * Left shift b1 of bits 8-fb bits (2 bits)
     *
     *   +--------+
     *   |22221111|  <- Initial value of b1
     *   |22111100|  <- After left shift of 2 bits.
     *   +--------+
     *

二者进行或运算之后以111111做掩码进行与运算，得到

* OR the two bits, and finally AND with 111111 (63 in decimal) to
     * clean the higher order bits we are not interested in:
     *
     *   +--------+
     *   |00000011|  <- b0 right shifted
     *   |22111100|  <- b1 left shifted
     *   |22111111|  <- b0 OR b1
     *   |  111111|  <- (b0 OR b1) AND 63, our value.
     *   +--------+

--------------------

下面再看看基数的设置。设置基数首先需要清空已有值，然后通过或运算存储进入基数集合类。同样以pos = 1作为示例。首先如上计算出fb = 6。

*   +--------+
     *   |11000000|  <- Our byte at b0
     *   +--------+

为了清空数据，我们需要用一个掩码，保证不改变其他数据的前提下仅仅去掉该位置的数据。因此我们先制造一个如上文所示的掩码，之后进行反转，用0进行与运算除去即可。

*   +--------+
     *   |00111111|  <- "mask" starts at 63
     *   |11000000|  <- "mask" after left shift of "ls" bits.
     *   |00111111|  <- "mask" after invert.
     *   +--------+
     *

使用反转后的掩码进行与操作清空旧数据并设置新数据即可。然后再看看b1字节的剩余数据如何处理。

*   +--------+
     *   |22221111|  <- Initial value of b1
     *   +--------+

和上述操作类似的创造出反转的掩码

*   +--------+
     *   |00111111|  <- "mask" set at 2&6-1
     *   |00001111|  <- "mask" after the right shift by 8-fb = 2 bits
     *   |11110000|  <- "mask" after bitwise not.
     *   +--------+
     *

进行清除和设置即可。

##### 2.2 稀疏存储的设置 #####

稀疏存储 使用ZERO,XZERO和VAL来表示稀疏存储，ZERO和VAL长度为1字节，XZERO长度为2字节。ZERO和XZERO分别表示连续64个以内和连续16384个以内的连续的0， VAL表示连续几个以内的值，值超过32则改用密集存储。

ZERO 格式为 00xxxxxx，x有6位可以表示64以内的数字。  
   XZERO格式为 01xxxxxx yyyyyyyy。x为高位y为低位，最多表示连续16384个0  
   VAL 格式为 1vvvvvxx，v表示值，x表示有几个相同值，最多只能4个

当计数值达到一定程度后，稀疏存储将会不可逆一次性转换为密集存储。转换的条件有两个，任意一个满足就会立即发生转换，也就是任意一个计数值从 32 变成 33，因为VAL指令已经无法容纳，它能表示的计数值最大为 32稀疏存储占用的总字节数超过 3000 字节，这个阈值可以通过 hll\_sparse\_max\_bytes 参数进行调整。

### 3. 源码解析 ###

/* ================== Dense representation implementation ================== */
    //密集存储实现
    
    /* 重新设置桶类计数 * Low level function to set the dense HLL register at 'index' to the * specified value if the current value is smaller than 'count'. * * 'registers' is expected to have room for HLL_REGISTERS plus an * additional byte on the right. This requirement is met by sds strings * automatically since they are implicitly null terminated. * * The function always succeed, however if as a result of the operation * the approximated cardinality changed, 1 is returned. Otherwise 0 * is returned. */
    int hllDenseSet(uint8_t *registers, long index, uint8_t count) { 
        uint8_t oldcount;
    
        HLL_DENSE_GET_REGISTER(oldcount,registers,index);
        if (count > oldcount) { 
            HLL_DENSE_SET_REGISTER(registers,index,count);
            return 1;
        } else { 
            return 0;
        }
    }
    
    /* 调用hllDenseSet对新加入元素的判断：是否需要改变桶大小 * "Add" the element in the dense hyperloglog data structure. * Actually nothing is added, but the max 0 pattern counter of the subset * the element belongs to is incremented if needed. * * This is just a wrapper to hllDenseSet(), performing the hashing of the * element in order to retrieve the index and zero-run count. */
    int hllDenseAdd(uint8_t *registers, unsigned char *ele, size_t elesize) { 
        long index;
        uint8_t count = hllPatLen(ele,elesize,&index);
    	
        /* 如果比现有的最大值还大，则添加该值到数据部分 * Update the register if this element produced a longer run of zeroes. */
        return hllDenseSet(registers,index,count);
    }
    
    /* 计算和 * Compute SUM(2^-reg) in the dense representation. * PE is an array with a pre-computer table of values 2^-reg indexed by reg. * As a side effect the integer pointed by 'ezp' is set to the number * of zero registers. */
    double hllDenseSum(uint8_t *registers, double *PE, int *ezp) { 
        double E = 0;
        int j, ez = 0;
    
        /* Redis default is to use 16384 registers 6 bits each. The code works * with other values by modifying the defines, but for our target value * we take a faster path with unrolled loops. */
        if (HLL_REGISTERS == 16384 && HLL_BITS == 6) { 
            uint8_t *r = registers;
            unsigned long r0, r1, r2, r3, r4, r5, r6, r7, r8, r9,
                          r10, r11, r12, r13, r14, r15;
            for (j = 0; j < 1024; j++) { 
                /* Handle 16 registers per iteration. */
                r0 = r[0] & 63; if (r0 == 0) ez++;
                r1 = (r[0] >> 6 | r[1] << 2) & 63; if (r1 == 0) ez++;
                r2 = (r[1] >> 4 | r[2] << 4) & 63; if (r2 == 0) ez++;
                r3 = (r[2] >> 2) & 63; if (r3 == 0) ez++;
                r4 = r[3] & 63; if (r4 == 0) ez++;
                r5 = (r[3] >> 6 | r[4] << 2) & 63; if (r5 == 0) ez++;
                r6 = (r[4] >> 4 | r[5] << 4) & 63; if (r6 == 0) ez++;
                r7 = (r[5] >> 2) & 63; if (r7 == 0) ez++;
                r8 = r[6] & 63; if (r8 == 0) ez++;
                r9 = (r[6] >> 6 | r[7] << 2) & 63; if (r9 == 0) ez++;
                r10 = (r[7] >> 4 | r[8] << 4) & 63; if (r10 == 0) ez++;
                r11 = (r[8] >> 2) & 63; if (r11 == 0) ez++;
                r12 = r[9] & 63; if (r12 == 0) ez++;
                r13 = (r[9] >> 6 | r[10] << 2) & 63; if (r13 == 0) ez++;
                r14 = (r[10] >> 4 | r[11] << 4) & 63; if (r14 == 0) ez++;
                r15 = (r[11] >> 2) & 63; if (r15 == 0) ez++;
    
                /* Additional parens will allow the compiler to optimize the * code more with a loss of precision that is not very relevant * here (floating point math is not commutative!). */
                E += (PE[r0] + PE[r1]) + (PE[r2] + PE[r3]) + (PE[r4] + PE[r5]) +
                     (PE[r6] + PE[r7]) + (PE[r8] + PE[r9]) + (PE[r10] + PE[r11]) +
                     (PE[r12] + PE[r13]) + (PE[r14] + PE[r15]);
                r += 12;
            }
        } else { 
            for (j = 0; j < HLL_REGISTERS; j++) { 
                unsigned long reg;
    
                HLL_DENSE_GET_REGISTER(reg,registers,j);
                if (reg == 0) { 
                    ez++;
                    /* Increment E at the end of the loop. */
                } else { 
                    E += PE[reg]; /* Precomputed 2^(-reg[j]). */
                }
            }
            E += ez; /* Add 2^0 'ez' times. */
        }
        *ezp = ez;
        return E;
    }
    
    /* ================== Sparse representation implementation ================= */
    
    /* 将稀疏存储改变为密集存储 * Convert the HLL with sparse representation given as input in its dense * representation. Both representations are represented by SDS strings, and * the input representation is freed as a side effect. * * The function returns C_OK if the sparse representation was valid, * otherwise C_ERR is returned if the representation was corrupted. */
    int hllSparseToDense(robj *o) { 
        sds sparse = o->ptr, dense;
        struct hllhdr *hdr, *oldhdr = (struct hllhdr*)sparse;
        int idx = 0, runlen, regval;
        uint8_t *p = (uint8_t*)sparse, *end = p+sdslen(sparse);
    
        /* If the representation is already the right one return ASAP. */
        hdr = (struct hllhdr*) sparse;
        if (hdr->encoding == HLL_DENSE) return C_OK;
    
        /* 创建新的基数为0的dense * Create a string of the right size filled with zero bytes. * Note that the cached cardinality is set to 0 as a side effect * that is exactly the cardinality of an empty HLL. */
        dense = sdsnewlen(NULL,HLL_DENSE_SIZE);
        hdr = (struct hllhdr*) dense;
        *hdr = *oldhdr; /* This will copy the magic and cached cardinality. */
        hdr->encoding = HLL_DENSE;
    
        /* 读取稀疏存储并赋值给密集存储 * Now read the sparse representation and set non-zero registers * accordingly. */
        p += HLL_HDR_SIZE;
        while(p < end) { 
            if (HLL_SPARSE_IS_ZERO(p)) { 
                runlen = HLL_SPARSE_ZERO_LEN(p);
                idx += runlen;
                p++;
            } else if (HLL_SPARSE_IS_XZERO(p)) { 
                runlen = HLL_SPARSE_XZERO_LEN(p);
                idx += runlen;
                p += 2;
            } else { 
                runlen = HLL_SPARSE_VAL_LEN(p);
                regval = HLL_SPARSE_VAL_VALUE(p);
                while(runlen--) { 
                    HLL_DENSE_SET_REGISTER(hdr->registers,idx,regval);
                    idx++;
                }
                p++;
            }
        }
    
        /* If the sparse representation was valid, we expect to find idx * set to HLL_REGISTERS. */
        if (idx != HLL_REGISTERS) { 
            sdsfree(dense);
            return C_ERR;
        }
    
        /* Free the old representation and set the new one. */
        sdsfree(o->ptr);
        o->ptr = dense;
        return C_OK;
    }
    
    /* 稀疏存储赋值，可能会改变稀疏存储转为密集存储 * 还需要考虑ZERO, XZERO和VAL的变化 * * Low level function to set the sparse HLL register at 'index' to the * specified value if the current value is smaller than 'count'. * * The object 'o' is the String object holding the HLL. The function requires * a reference to the object in order to be able to enlarge the string if * needed. * * On success, the function returns 1 if the cardinality changed, or 0 * if the register for this element was not updated. * On error (if the representation is invalid) -1 is returned. * * As a side effect the function may promote the HLL representation from * sparse to dense: this happens when a register requires to be set to a value * not representable with the sparse representation, or when the resulting * size would be greater than server.hll_sparse_max_bytes. */
    int hllSparseSet(robj *o, long index, uint8_t count) { 
        struct hllhdr *hdr;
        uint8_t oldcount, *sparse, *end, *p, *prev, *next;
        long first, span;
        long is_zero = 0, is_xzero = 0, is_val = 0, runlen = 0;
    
        /* 数据过多则改用密集存储 * If the count is too big to be representable by the sparse representation * switch to dense representation. */
        if (count > HLL_SPARSE_VAL_MAX_VALUE) goto promote;
    
        /* 扩展空间 * When updating a sparse representation, sometimes we may need to * enlarge the buffer for up to 3 bytes in the worst case (XZERO split * into XZERO-VAL-XZERO). Make sure there is enough space right now * so that the pointers we take during the execution of the function * will be valid all the time. */
        o->ptr = sdsMakeRoomFor(o->ptr,3);
    
        /* 第一步：检测是否需要修改桶 * Step 1: we need to locate the opcode we need to modify to check * if a value update is actually needed. */
        sparse = p = ((uint8_t*)o->ptr) + HLL_HDR_SIZE;
        end = p + sdslen(o->ptr) - HLL_HDR_SIZE;
    
        first = 0;
        prev = NULL; /* Points to previos opcode at the end of the loop. */
        next = NULL; /* Points to the next opcode at the end of the loop. */
        span = 0;
        while(p < end) { 
            long oplen;
    
            /* Set span to the number of registers covered by this opcode. * * This is the most performance critical loop of the sparse * representation. Sorting the conditionals from the most to the * least frequent opcode in many-bytes sparse HLLs is faster. */
            oplen = 1;
            if (HLL_SPARSE_IS_ZERO(p)) { 
                span = HLL_SPARSE_ZERO_LEN(p);
            } else if (HLL_SPARSE_IS_VAL(p)) { 
                span = HLL_SPARSE_VAL_LEN(p);
            } else {  /* XZERO. */
                span = HLL_SPARSE_XZERO_LEN(p);
                oplen = 2;
            }
            /* Break if this opcode covers the register as 'index'. */
            if (index <= first+span-1) break;
            prev = p;
            p += oplen;
            first += span;
        }
        if (span == 0) return -1; /* Invalid format. */
    
        next = HLL_SPARSE_IS_XZERO(p) ? p+2 : p+1;
        if (next >= end) next = NULL;
    
        /* 存储当前类型 * Cache current opcode type to avoid using the macro again and * again for something that will not change. * Also cache the run-length of the opcode. */
        if (HLL_SPARSE_IS_ZERO(p)) { 
            is_zero = 1;
            runlen = HLL_SPARSE_ZERO_LEN(p);
        } else if (HLL_SPARSE_IS_XZERO(p)) { 
            is_xzero = 1;
            runlen = HLL_SPARSE_XZERO_LEN(p);
        } else { 
            is_val = 1;
            runlen = HLL_SPARSE_VAL_LEN(p);
        }
    
        /* 第二步，根据不同情况选择 * * 1. len = 1的时候 * 原值为VAL：小于现值则不变，大于则调用API修改 * 原值为ZERO则直接替换为VAL * * 2. len > 1的时候 * 无论原值为哪种都需要调整为XZERO - VAL - XZERO的结合体 * * Step 2: After the loop: * * 'first' stores to the index of the first register covered * by the current opcode, which is pointed by 'p'. * * 'next' ad 'prev' store respectively the next and previous opcode, * or NULL if the opcode at 'p' is respectively the last or first. * * 'span' is set to the number of registers covered by the current * opcode. * * There are different cases in order to update the data structure * in place without generating it from scratch: * * A) If it is a VAL opcode already set to a value >= our 'count' * no update is needed, regardless of the VAL run-length field. * In this case PFADD returns 0 since no changes are performed. * * B) If it is a VAL opcode with len = 1 (representing only our * register) and the value is less than 'count', we just update it * since this is a trivial case. */
        if (is_val) { 
            oldcount = HLL_SPARSE_VAL_VALUE(p);
            /* Case A. */
            if (oldcount >= count) return 0;
    
            /* Case B. */
            if (runlen == 1) { 
                HLL_SPARSE_VAL_SET(p,count,1);
                goto updated;
            }
        }
    
        /* C) Another trivial to handle case is a ZERO opcode with a len of 1. * We can just replace it with a VAL opcode with our value and len of 1. */
        if (is_zero && runlen == 1) { 
            HLL_SPARSE_VAL_SET(p,count,1);
            goto updated;
        }
    
        /* D) General case. * * The other cases are more complex: our register requires to be updated * and is either currently represented by a VAL opcode with len > 1, * by a ZERO opcode with len > 1, or by an XZERO opcode. * * In those cases the original opcode must be split into muliple * opcodes. The worst case is an XZERO split in the middle resuling into * XZERO - VAL - XZERO, so the resulting sequence max length is * 5 bytes. * * We perform the split writing the new sequence into the 'new' buffer * with 'newlen' as length. Later the new sequence is inserted in place * of the old one, possibly moving what is on the right a few bytes * if the new sequence is longer than the older one. */
        uint8_t seq[5], *n = seq;
        int last = first+span-1; /* Last register covered by the sequence. */
        int len;
    
        if (is_zero || is_xzero) { 
            /* Handle splitting of ZERO / XZERO. */
            if (index != first) { 
                len = index-first;
                if (len > HLL_SPARSE_ZERO_MAX_LEN) { 
                    HLL_SPARSE_XZERO_SET(n,len);
                    n += 2;
                } else { 
                    HLL_SPARSE_ZERO_SET(n,len);
                    n++;
                }
            }
            HLL_SPARSE_VAL_SET(n,count,1);
            n++;
            if (index != last) { 
                len = last-index;
                if (len > HLL_SPARSE_ZERO_MAX_LEN) { 
                    HLL_SPARSE_XZERO_SET(n,len);
                    n += 2;
                } else { 
                    HLL_SPARSE_ZERO_SET(n,len);
                    n++;
                }
            }
        } else { 
            /* Handle splitting of VAL. */
            int curval = HLL_SPARSE_VAL_VALUE(p);
    
            if (index != first) { 
                len = index-first;
                HLL_SPARSE_VAL_SET(n,curval,len);
                n++;
            }
            HLL_SPARSE_VAL_SET(n,count,1);
            n++;
            if (index != last) { 
                len = last-index;
                HLL_SPARSE_VAL_SET(n,curval,len);
                n++;
            }
        }
    
        /* 第三步：替换旧的序列 * Step 3: substitute the new sequence with the old one. * * Note that we already allocated space on the sds string * calling sdsMakeRoomFor(). */
         int seqlen = n-seq;
         int oldlen = is_xzero ? 2 : 1;
         int deltalen = seqlen-oldlen;
    
         if (deltalen > 0 &&
             sdslen(o->ptr)+deltalen > server.hll_sparse_max_bytes) goto promote;
         if (deltalen && next) memmove(next+deltalen,next,end-next);
         sdsIncrLen(o->ptr,deltalen);
         memcpy(p,seq,seqlen);
         end += deltalen;
    
    updated:
        /* 第四步：优化。相邻的VAL有时候可以合并 * Step 4: Merge adjacent values if possible. * * The representation was updated, however the resulting representation * may not be optimal: adjacent VAL opcodes can sometimes be merged into * a single one. */
        p = prev ? prev : sparse;
        int scanlen = 5; /* Scan up to 5 upcodes starting from prev. */
        while (p < end && scanlen--) { 
            if (HLL_SPARSE_IS_XZERO(p)) { 
                p += 2;
                continue;
            } else if (HLL_SPARSE_IS_ZERO(p)) { 
                p++;
                continue;
            }
            /* We need two adjacent VAL opcodes to try a merge, having * the same value, and a len that fits the VAL opcode max len. */
            if (p+1 < end && HLL_SPARSE_IS_VAL(p+1)) { 
                int v1 = HLL_SPARSE_VAL_VALUE(p);
                int v2 = HLL_SPARSE_VAL_VALUE(p+1);
                if (v1 == v2) { 
                    int len = HLL_SPARSE_VAL_LEN(p)+HLL_SPARSE_VAL_LEN(p+1);
                    if (len <= HLL_SPARSE_VAL_MAX_LEN) { 
                        HLL_SPARSE_VAL_SET(p+1,v1,len);
                        memmove(p,p+1,end-p);
                        sdsIncrLen(o->ptr,-1);
                        end--;
                        /* After a merge we reiterate without incrementing 'p' * in order to try to merge the just merged value with * a value on its right. */
                        continue;
                    }
                }
            }
            p++;
        }
    
        /* Invalidate the cached cardinality. */
        hdr = o->ptr;
        HLL_INVALIDATE_CACHE(hdr);
        return 1;
    
    promote: /* Promote to dense representation. */
        if (hllSparseToDense(o) == C_ERR) return -1; /* Corrupted HLL. */
        hdr = o->ptr;
    
        /* 重新设置hllDense * We need to call hllDenseAdd() to perform the operation after the * conversion. However the result must be 1, since if we need to * convert from sparse to dense a register requires to be updated. * * Note that this in turn means that PFADD will make sure the command * is propagated to slaves / AOF, so if there is a sparse -> dense * convertion, it will be performed in all the slaves as well. */
        int dense_retval = hllDenseSet(hdr->registers,index,count);
        serverAssert(dense_retval == 1);
        return dense_retval;
    }
    
    /* 对hllSparseSet的封装函数 * "Add" the element in the sparse hyperloglog data structure. * Actually nothing is added, but the max 0 pattern counter of the subset * the element belongs to is incremented if needed. * * This function is actually a wrapper for hllSparseSet(), it only performs * the hashshing of the elmenet to obtain the index and zeros run length. */
    int hllSparseAdd(robj *o, unsigned char *ele, size_t elesize) { 
        long index;
        uint8_t count = hllPatLen(ele,elesize,&index);
        /* Update the register if this element produced a longer run of zeroes. */
        return hllSparseSet(o,index,count);
    }
    
    /* 计算稀疏矩阵的和 * Compute SUM(2^-reg) in the sparse representation. * PE is an array with a pre-computer table of values 2^-reg indexed by reg. * As a side effect the integer pointed by 'ezp' is set to the number * of zero registers. */
    double hllSparseSum(uint8_t *sparse, int sparselen, double *PE, int *ezp, int *invalid) { 
        double E = 0;
        int ez = 0, idx = 0, runlen, regval;
        uint8_t *end = sparse+sparselen, *p = sparse;
    
        while(p < end) { 
            if (HLL_SPARSE_IS_ZERO(p)) { 
                runlen = HLL_SPARSE_ZERO_LEN(p);
                idx += runlen;
                ez += runlen;
                /* Increment E at the end of the loop. */
                p++;
            } else if (HLL_SPARSE_IS_XZERO(p)) { 
                runlen = HLL_SPARSE_XZERO_LEN(p);
                idx += runlen;
                ez += runlen;
                /* Increment E at the end of the loop. */
                p += 2;
            } else { 
                runlen = HLL_SPARSE_VAL_LEN(p);
                regval = HLL_SPARSE_VAL_VALUE(p);
                idx += runlen;
                E += PE[regval]*runlen;
                p++;
            }
        }
        if (idx != HLL_REGISTERS && invalid) *invalid = 1;
        E += ez; /* Add 2^0 'ez' times. */
        *ezp = ez;
        return E;
    }

### 4. 总结 ###

本文主要分析hyperloglog的存储结构体，具体hyperloglog计算算法在下一篇中进行分析