1 /* hyperloglog.c - Redis HyperLogLog probabilistic cardinality approximation. 2 * This file implements the algorithm and the exported Redis commands. 3 * 4 * Copyright (c) 2014, Salvatore Sanfilippo <antirez at gmail dot com> 5 * All rights reserved. 6 * 7 * Redistribution and use in source and binary forms, with or without 8 * modification, are permitted provided that the following conditions are met: 9 * 10 * * Redistributions of source code must retain the above copyright notice, 11 * this list of conditions and the following disclaimer. 12 * * Redistributions in binary form must reproduce the above copyright 13 * notice, this list of conditions and the following disclaimer in the 14 * documentation and/or other materials provided with the distribution. 15 * * Neither the name of Redis nor the names of its contributors may be used 16 * to endorse or promote products derived from this software without 17 * specific prior written permission. 18 * 19 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" 20 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE 21 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE 22 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE 23 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR 24 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF 25 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS 26 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN 27 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) 28 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE 29 * POSSIBILITY OF SUCH DAMAGE. 30 */ 31 32 #include "server.h" 33 34 #include <stdint.h> 35 #include <math.h> 36 37 /* The Redis HyperLogLog implementation is based on the following ideas: 38 * 39 * * The use of a 64 bit hash function as proposed in [1], in order to don't 40 * limited to cardinalities up to 10^9, at the cost of just 1 additional 41 * bit per register. 42 * * The use of 16384 6-bit registers for a great level of accuracy, using 43 * a total of 12k per key. 44 * * The use of the Redis string data type. No new type is introduced. 45 * * No attempt is made to compress the data structure as in [1]. Also the 46 * algorithm used is the original HyperLogLog Algorithm as in [2], with 47 * the only difference that a 64 bit hash function is used, so no correction 48 * is performed for values near 2^32 as in [1]. 49 * 50 * [1] Heule, Nunkesser, Hall: HyperLogLog in Practice: Algorithmic 51 * Engineering of a State of The Art Cardinality Estimation Algorithm. 52 * 53 * [2] P. Flajolet, Éric Fusy, O. Gandouet, and F. Meunier. Hyperloglog: The 54 * analysis of a near-optimal cardinality estimation algorithm. 55 * 56 * Redis uses two representations: 57 * 58 * 1) A "dense" representation where every entry is represented by 59 * a 6-bit integer. 60 * 2) A "sparse" representation using run length compression suitable 61 * for representing HyperLogLogs with many registers set to 0 in 62 * a memory efficient way. 63 * 64 * 65 * HLL header 66 * === 67 * 68 * Both the dense and sparse representation have a 16 byte header as follows: 69 * 70 * +------+---+-----+----------+ 71 * | HYLL | E | N/U | Cardin. | 72 * +------+---+-----+----------+ 73 * 74 * The first 4 bytes are a magic string set to the bytes "HYLL". 75 * "E" is one byte encoding, currently set to HLL_DENSE or 76 * HLL_SPARSE. N/U are three not used bytes. 77 * 78 * The "Cardin." field is a 64 bit integer stored in little endian format 79 * with the latest cardinality computed that can be reused if the data 80 * structure was not modified since the last computation (this is useful 81 * because there are high probabilities that HLLADD operations don't 82 * modify the actual data structure and hence the approximated cardinality). 83 * 84 * When the most significant bit in the most significant byte of the cached 85 * cardinality is set, it means that the data structure was modified and 86 * we can't reuse the cached value that must be recomputed. 87 * 88 * Dense representation 89 * === 90 * 91 * The dense representation used by Redis is the following: 92 * 93 * +--------+--------+--------+------// //--+ 94 * |11000000|22221111|33333322|55444444 .... | 95 * +--------+--------+--------+------// //--+ 96 * 97 * The 6 bits counters are encoded one after the other starting from the 98 * LSB to the MSB, and using the next bytes as needed. 99 * 100 * Sparse representation 101 * === 102 * 103 * The sparse representation encodes registers using a run length 104 * encoding composed of three opcodes, two using one byte, and one using 105 * of two bytes. The opcodes are called ZERO, XZERO and VAL. 106 * 107 * ZERO opcode is represented as 00xxxxxx. The 6-bit integer represented 108 * by the six bits 'xxxxxx', plus 1, means that there are N registers set 109 * to 0. This opcode can represent from 1 to 64 contiguous registers set 110 * to the value of 0. 111 * 112 * XZERO opcode is represented by two bytes 01xxxxxx yyyyyyyy. The 14-bit 113 * integer represented by the bits 'xxxxxx' as most significant bits and 114 * 'yyyyyyyy' as least significant bits, plus 1, means that there are N 115 * registers set to 0. This opcode can represent from 0 to 16384 contiguous 116 * registers set to the value of 0. 117 * 118 * VAL opcode is represented as 1vvvvvxx. It contains a 5-bit integer 119 * representing the value of a register, and a 2-bit integer representing 120 * the number of contiguous registers set to that value 'vvvvv'. 121 * To obtain the value and run length, the integers vvvvv and xx must be 122 * incremented by one. This opcode can represent values from 1 to 32, 123 * repeated from 1 to 4 times. 124 * 125 * The sparse representation can't represent registers with a value greater 126 * than 32, however it is very unlikely that we find such a register in an 127 * HLL with a cardinality where the sparse representation is still more 128 * memory efficient than the dense representation. When this happens the 129 * HLL is converted to the dense representation. 130 * 131 * The sparse representation is purely positional. For example a sparse 132 * representation of an empty HLL is just: XZERO:16384. 133 * 134 * An HLL having only 3 non-zero registers at position 1000, 1020, 1021 135 * respectively set to 2, 3, 3, is represented by the following three 136 * opcodes: 137 * 138 * XZERO:1000 (Registers 0-999 are set to 0) 139 * VAL:2,1 (1 register set to value 2, that is register 1000) 140 * ZERO:19 (Registers 1001-1019 set to 0) 141 * VAL:3,2 (2 registers set to value 3, that is registers 1020,1021) 142 * XZERO:15362 (Registers 1022-16383 set to 0) 143 * 144 * In the example the sparse representation used just 7 bytes instead 145 * of 12k in order to represent the HLL registers. In general for low 146 * cardinality there is a big win in terms of space efficiency, traded 147 * with CPU time since the sparse representation is slower to access: 148 * 149 * The following table shows average cardinality vs bytes used, 100 150 * samples per cardinality (when the set was not representable because 151 * of registers with too big value, the dense representation size was used 152 * as a sample). 153 * 154 * 100 267 155 * 200 485 156 * 300 678 157 * 400 859 158 * 500 1033 159 * 600 1205 160 * 700 1375 161 * 800 1544 162 * 900 1713 163 * 1000 1882 164 * 2000 3480 165 * 3000 4879 166 * 4000 6089 167 * 5000 7138 168 * 6000 8042 169 * 7000 8823 170 * 8000 9500 171 * 9000 10088 172 * 10000 10591 173 * 174 * The dense representation uses 12288 bytes, so there is a big win up to 175 * a cardinality of ~2000-3000. For bigger cardinalities the constant times 176 * involved in updating the sparse representation is not justified by the 177 * memory savings. The exact maximum length of the sparse representation 178 * when this implementation switches to the dense representation is 179 * configured via the define server.hll_sparse_max_bytes. 180 */ 181 182 struct hllhdr { 183 char magic[4]; /* "HYLL" */ 184 uint8_t encoding; /* HLL_DENSE or HLL_SPARSE. */ 185 uint8_t notused[3]; /* Reserved for future use, must be zero. */ 186 uint8_t card[8]; /* Cached cardinality, little endian. */ 187 uint8_t registers[]; /* Data bytes. */ 188 }; 189 190 /* The cached cardinality MSB is used to signal validity of the cached value. */ 191 #define HLL_INVALIDATE_CACHE(hdr) (hdr)->card[7] |= (1<<7) 192 #define HLL_VALID_CACHE(hdr) (((hdr)->card[7] & (1<<7)) == 0) 193 194 #define HLL_P 14 /* The greater is P, the smaller the error. */ 195 #define HLL_Q (64-HLL_P) /* The number of bits of the hash value used for 196 determining the number of leading zeros. */ 197 #define HLL_REGISTERS (1<<HLL_P) /* With P=14, 16384 registers. */ 198 #define HLL_P_MASK (HLL_REGISTERS-1) /* Mask to index register. */ 199 #define HLL_BITS 6 /* Enough to count up to 63 leading zeroes. */ 200 #define HLL_REGISTER_MAX ((1<<HLL_BITS)-1) 201 #define HLL_HDR_SIZE sizeof(struct hllhdr) 202 #define HLL_DENSE_SIZE (HLL_HDR_SIZE+((HLL_REGISTERS*HLL_BITS+7)/8)) 203 #define HLL_DENSE 0 /* Dense encoding. */ 204 #define HLL_SPARSE 1 /* Sparse encoding. */ 205 #define HLL_RAW 255 /* Only used internally, never exposed. */ 206 #define HLL_MAX_ENCODING 1 207 208 static char *invalid_hll_err = "-INVALIDOBJ Corrupted HLL object detected\r\n"; 209 210 /* =========================== Low level bit macros ========================= */ 211 212 /* Macros to access the dense representation. 213 * 214 * We need to get and set 6 bit counters in an array of 8 bit bytes. 215 * We use macros to make sure the code is inlined since speed is critical 216 * especially in order to compute the approximated cardinality in 217 * HLLCOUNT where we need to access all the registers at once. 218 * For the same reason we also want to avoid conditionals in this code path. 219 * 220 * +--------+--------+--------+------// 221 * |11000000|22221111|33333322|55444444 222 * +--------+--------+--------+------// 223 * 224 * Note: in the above representation the most significant bit (MSB) 225 * of every byte is on the left. We start using bits from the LSB to MSB, 226 * and so forth passing to the next byte. 227 * 228 * Example, we want to access to counter at pos = 1 ("111111" in the 229 * illustration above). 230 * 231 * The index of the first byte b0 containing our data is: 232 * 233 * b0 = 6 * pos / 8 = 0 234 * 235 * +--------+ 236 * |11000000| <- Our byte at b0 237 * +--------+ 238 * 239 * The position of the first bit (counting from the LSB = 0) in the byte 240 * is given by: 241 * 242 * fb = 6 * pos % 8 -> 6 243 * 244 * Right shift b0 of 'fb' bits. 245 * 246 * +--------+ 247 * |11000000| <- Initial value of b0 248 * |00000011| <- After right shift of 6 pos. 249 * +--------+ 250 * 251 * Left shift b1 of bits 8-fb bits (2 bits) 252 * 253 * +--------+ 254 * |22221111| <- Initial value of b1 255 * |22111100| <- After left shift of 2 bits. 256 * +--------+ 257 * 258 * OR the two bits, and finally AND with 111111 (63 in decimal) to 259 * clean the higher order bits we are not interested in: 260 * 261 * +--------+ 262 * |00000011| <- b0 right shifted 263 * |22111100| <- b1 left shifted 264 * |22111111| <- b0 OR b1 265 * | 111111| <- (b0 OR b1) AND 63, our value. 266 * +--------+ 267 * 268 * We can try with a different example, like pos = 0. In this case 269 * the 6-bit counter is actually contained in a single byte. 270 * 271 * b0 = 6 * pos / 8 = 0 272 * 273 * +--------+ 274 * |11000000| <- Our byte at b0 275 * +--------+ 276 * 277 * fb = 6 * pos % 8 = 0 278 * 279 * So we right shift of 0 bits (no shift in practice) and 280 * left shift the next byte of 8 bits, even if we don't use it, 281 * but this has the effect of clearing the bits so the result 282 * will not be affacted after the OR. 283 * 284 * ------------------------------------------------------------------------- 285 * 286 * Setting the register is a bit more complex, let's assume that 'val' 287 * is the value we want to set, already in the right range. 288 * 289 * We need two steps, in one we need to clear the bits, and in the other 290 * we need to bitwise-OR the new bits. 291 * 292 * Let's try with 'pos' = 1, so our first byte at 'b' is 0, 293 * 294 * "fb" is 6 in this case. 295 * 296 * +--------+ 297 * |11000000| <- Our byte at b0 298 * +--------+ 299 * 300 * To create a AND-mask to clear the bits about this position, we just 301 * initialize the mask with the value 63, left shift it of "fs" bits, 302 * and finally invert the result. 303 * 304 * +--------+ 305 * |00111111| <- "mask" starts at 63 306 * |11000000| <- "mask" after left shift of "ls" bits. 307 * |00111111| <- "mask" after invert. 308 * +--------+ 309 * 310 * Now we can bitwise-AND the byte at "b" with the mask, and bitwise-OR 311 * it with "val" left-shifted of "ls" bits to set the new bits. 312 * 313 * Now let's focus on the next byte b1: 314 * 315 * +--------+ 316 * |22221111| <- Initial value of b1 317 * +--------+ 318 * 319 * To build the AND mask we start again with the 63 value, right shift 320 * it by 8-fb bits, and invert it. 321 * 322 * +--------+ 323 * |00111111| <- "mask" set at 2&6-1 324 * |00001111| <- "mask" after the right shift by 8-fb = 2 bits 325 * |11110000| <- "mask" after bitwise not. 326 * +--------+ 327 * 328 * Now we can mask it with b+1 to clear the old bits, and bitwise-OR 329 * with "val" left-shifted by "rs" bits to set the new value. 330 */ 331 332 /* Note: if we access the last counter, we will also access the b+1 byte 333 * that is out of the array, but sds strings always have an implicit null 334 * term, so the byte exists, and we can skip the conditional (or the need 335 * to allocate 1 byte more explicitly). */ 336 337 /* Store the value of the register at position 'regnum' into variable 'target'. 338 * 'p' is an array of unsigned bytes. */ 339 #define HLL_DENSE_GET_REGISTER(target,p,regnum) do { \ 340 uint8_t *_p = (uint8_t*) p; \ 341 unsigned long _byte = regnum*HLL_BITS/8; \ 342 unsigned long _fb = regnum*HLL_BITS&7; \ 343 unsigned long _fb8 = 8 - _fb; \ 344 unsigned long b0 = _p[_byte]; \ 345 unsigned long b1 = _p[_byte+1]; \ 346 target = ((b0 >> _fb) | (b1 << _fb8)) & HLL_REGISTER_MAX; \ 347 } while(0) 348 349 /* Set the value of the register at position 'regnum' to 'val'. 350 * 'p' is an array of unsigned bytes. */ 351 #define HLL_DENSE_SET_REGISTER(p,regnum,val) do { \ 352 uint8_t *_p = (uint8_t*) p; \ 353 unsigned long _byte = regnum*HLL_BITS/8; \ 354 unsigned long _fb = regnum*HLL_BITS&7; \ 355 unsigned long _fb8 = 8 - _fb; \ 356 unsigned long _v = val; \ 357 _p[_byte] &= ~(HLL_REGISTER_MAX << _fb); \ 358 _p[_byte] |= _v << _fb; \ 359 _p[_byte+1] &= ~(HLL_REGISTER_MAX >> _fb8); \ 360 _p[_byte+1] |= _v >> _fb8; \ 361 } while(0) 362 363 /* Macros to access the sparse representation. 364 * The macros parameter is expected to be an uint8_t pointer. */ 365 #define HLL_SPARSE_XZERO_BIT 0x40 /* 01xxxxxx */ 366 #define HLL_SPARSE_VAL_BIT 0x80 /* 1vvvvvxx */ 367 #define HLL_SPARSE_IS_ZERO(p) (((*(p)) & 0xc0) == 0) /* 00xxxxxx */ 368 #define HLL_SPARSE_IS_XZERO(p) (((*(p)) & 0xc0) == HLL_SPARSE_XZERO_BIT) 369 #define HLL_SPARSE_IS_VAL(p) ((*(p)) & HLL_SPARSE_VAL_BIT) 370 #define HLL_SPARSE_ZERO_LEN(p) (((*(p)) & 0x3f)+1) 371 #define HLL_SPARSE_XZERO_LEN(p) (((((*(p)) & 0x3f) << 8) | (*((p)+1)))+1) 372 #define HLL_SPARSE_VAL_VALUE(p) ((((*(p)) >> 2) & 0x1f)+1) 373 #define HLL_SPARSE_VAL_LEN(p) (((*(p)) & 0x3)+1) 374 #define HLL_SPARSE_VAL_MAX_VALUE 32 375 #define HLL_SPARSE_VAL_MAX_LEN 4 376 #define HLL_SPARSE_ZERO_MAX_LEN 64 377 #define HLL_SPARSE_XZERO_MAX_LEN 16384 378 #define HLL_SPARSE_VAL_SET(p,val,len) do { \ 379 *(p) = (((val)-1)<<2|((len)-1))|HLL_SPARSE_VAL_BIT; \ 380 } while(0) 381 #define HLL_SPARSE_ZERO_SET(p,len) do { \ 382 *(p) = (len)-1; \ 383 } while(0) 384 #define HLL_SPARSE_XZERO_SET(p,len) do { \ 385 int _l = (len)-1; \ 386 *(p) = (_l>>8) | HLL_SPARSE_XZERO_BIT; \ 387 *((p)+1) = (_l&0xff); \ 388 } while(0) 389 #define HLL_ALPHA_INF 0.721347520444481703680 /* constant for 0.5/ln(2) */ 390 391 /* ========================= HyperLogLog algorithm ========================= */ 392 393 /* Our hash function is MurmurHash2, 64 bit version. 394 * It was modified for Redis in order to provide the same result in 395 * big and little endian archs (endian neutral). */ 396 uint64_t MurmurHash64A (const void * key, int len, unsigned int seed) { 397 const uint64_t m = 0xc6a4a7935bd1e995; 398 const int r = 47; 399 uint64_t h = seed ^ (len * m); 400 const uint8_t *data = (const uint8_t *)key; 401 const uint8_t *end = data + (len-(len&7)); 402 403 while(data != end) { 404 uint64_t k; 405 406 #if (BYTE_ORDER == LITTLE_ENDIAN) 407 #ifdef USE_ALIGNED_ACCESS 408 memcpy(&k,data,sizeof(uint64_t)); 409 #else 410 k = *((uint64_t*)data); 411 #endif 412 #else 413 k = (uint64_t) data[0]; 414 k |= (uint64_t) data[1] << 8; 415 k |= (uint64_t) data[2] << 16; 416 k |= (uint64_t) data[3] << 24; 417 k |= (uint64_t) data[4] << 32; 418 k |= (uint64_t) data[5] << 40; 419 k |= (uint64_t) data[6] << 48; 420 k |= (uint64_t) data[7] << 56; 421 #endif 422 423 k *= m; 424 k ^= k >> r; 425 k *= m; 426 h ^= k; 427 h *= m; 428 data += 8; 429 } 430 431 switch(len & 7) { 432 case 7: h ^= (uint64_t)data[6] << 48; /* fall-thru */ 433 case 6: h ^= (uint64_t)data[5] << 40; /* fall-thru */ 434 case 5: h ^= (uint64_t)data[4] << 32; /* fall-thru */ 435 case 4: h ^= (uint64_t)data[3] << 24; /* fall-thru */ 436 case 3: h ^= (uint64_t)data[2] << 16; /* fall-thru */ 437 case 2: h ^= (uint64_t)data[1] << 8; /* fall-thru */ 438 case 1: h ^= (uint64_t)data[0]; 439 h *= m; /* fall-thru */ 440 }; 441 442 h ^= h >> r; 443 h *= m; 444 h ^= h >> r; 445 return h; 446 } 447 448 /* Given a string element to add to the HyperLogLog, returns the length 449 * of the pattern 000..1 of the element hash. As a side effect 'regp' is 450 * set to the register index this element hashes to. */ 451 int hllPatLen(unsigned char *ele, size_t elesize, long *regp) { 452 uint64_t hash, bit, index; 453 int count; 454 455 /* Count the number of zeroes starting from bit HLL_REGISTERS 456 * (that is a power of two corresponding to the first bit we don't use 457 * as index). The max run can be 64-P+1 = Q+1 bits. 458 * 459 * Note that the final "1" ending the sequence of zeroes must be 460 * included in the count, so if we find "001" the count is 3, and 461 * the smallest count possible is no zeroes at all, just a 1 bit 462 * at the first position, that is a count of 1. 463 * 464 * This may sound like inefficient, but actually in the average case 465 * there are high probabilities to find a 1 after a few iterations. */ 466 hash = MurmurHash64A(ele,elesize,0xadc83b19ULL); 467 index = hash & HLL_P_MASK; /* Register index. */ 468 hash >>= HLL_P; /* Remove bits used to address the register. */ 469 hash |= ((uint64_t)1<<HLL_Q); /* Make sure the loop terminates 470 and count will be <= Q+1. */ 471 bit = 1; 472 count = 1; /* Initialized to 1 since we count the "00000...1" pattern. */ 473 while((hash & bit) == 0) { 474 count++; 475 bit <<= 1; 476 } 477 *regp = (int) index; 478 return count; 479 } 480 481 /* ================== Dense representation implementation ================== */ 482 483 /* Low level function to set the dense HLL register at 'index' to the 484 * specified value if the current value is smaller than 'count'. 485 * 486 * 'registers' is expected to have room for HLL_REGISTERS plus an 487 * additional byte on the right. This requirement is met by sds strings 488 * automatically since they are implicitly null terminated. 489 * 490 * The function always succeed, however if as a result of the operation 491 * the approximated cardinality changed, 1 is returned. Otherwise 0 492 * is returned. */ 493 int hllDenseSet(uint8_t *registers, long index, uint8_t count) { 494 uint8_t oldcount; 495 496 HLL_DENSE_GET_REGISTER(oldcount,registers,index); 497 if (count > oldcount) { 498 HLL_DENSE_SET_REGISTER(registers,index,count); 499 return 1; 500 } else { 501 return 0; 502 } 503 } 504 505 /* "Add" the element in the dense hyperloglog data structure. 506 * Actually nothing is added, but the max 0 pattern counter of the subset 507 * the element belongs to is incremented if needed. 508 * 509 * This is just a wrapper to hllDenseSet(), performing the hashing of the 510 * element in order to retrieve the index and zero-run count. */ 511 int hllDenseAdd(uint8_t *registers, unsigned char *ele, size_t elesize) { 512 long index; 513 uint8_t count = hllPatLen(ele,elesize,&index); 514 /* Update the register if this element produced a longer run of zeroes. */ 515 return hllDenseSet(registers,index,count); 516 } 517 518 /* Compute the register histogram in the dense representation. */ 519 void hllDenseRegHisto(uint8_t *registers, int* reghisto) { 520 int j; 521 522 /* Redis default is to use 16384 registers 6 bits each. The code works 523 * with other values by modifying the defines, but for our target value 524 * we take a faster path with unrolled loops. */ 525 if (HLL_REGISTERS == 16384 && HLL_BITS == 6) { 526 uint8_t *r = registers; 527 unsigned long r0, r1, r2, r3, r4, r5, r6, r7, r8, r9, 528 r10, r11, r12, r13, r14, r15; 529 for (j = 0; j < 1024; j++) { 530 /* Handle 16 registers per iteration. */ 531 r0 = r[0] & 63; 532 r1 = (r[0] >> 6 | r[1] << 2) & 63; 533 r2 = (r[1] >> 4 | r[2] << 4) & 63; 534 r3 = (r[2] >> 2) & 63; 535 r4 = r[3] & 63; 536 r5 = (r[3] >> 6 | r[4] << 2) & 63; 537 r6 = (r[4] >> 4 | r[5] << 4) & 63; 538 r7 = (r[5] >> 2) & 63; 539 r8 = r[6] & 63; 540 r9 = (r[6] >> 6 | r[7] << 2) & 63; 541 r10 = (r[7] >> 4 | r[8] << 4) & 63; 542 r11 = (r[8] >> 2) & 63; 543 r12 = r[9] & 63; 544 r13 = (r[9] >> 6 | r[10] << 2) & 63; 545 r14 = (r[10] >> 4 | r[11] << 4) & 63; 546 r15 = (r[11] >> 2) & 63; 547 548 reghisto[r0]++; 549 reghisto[r1]++; 550 reghisto[r2]++; 551 reghisto[r3]++; 552 reghisto[r4]++; 553 reghisto[r5]++; 554 reghisto[r6]++; 555 reghisto[r7]++; 556 reghisto[r8]++; 557 reghisto[r9]++; 558 reghisto[r10]++; 559 reghisto[r11]++; 560 reghisto[r12]++; 561 reghisto[r13]++; 562 reghisto[r14]++; 563 reghisto[r15]++; 564 565 r += 12; 566 } 567 } else { 568 for(j = 0; j < HLL_REGISTERS; j++) { 569 unsigned long reg; 570 HLL_DENSE_GET_REGISTER(reg,registers,j); 571 reghisto[reg]++; 572 } 573 } 574 } 575 576 /* ================== Sparse representation implementation ================= */ 577 578 /* Convert the HLL with sparse representation given as input in its dense 579 * representation. Both representations are represented by SDS strings, and 580 * the input representation is freed as a side effect. 581 * 582 * The function returns C_OK if the sparse representation was valid, 583 * otherwise C_ERR is returned if the representation was corrupted. */ 584 int hllSparseToDense(robj *o) { 585 sds sparse = o->ptr, dense; 586 struct hllhdr *hdr, *oldhdr = (struct hllhdr*)sparse; 587 int idx = 0, runlen, regval; 588 uint8_t *p = (uint8_t*)sparse, *end = p+sdslen(sparse); 589 590 /* If the representation is already the right one return ASAP. */ 591 hdr = (struct hllhdr*) sparse; 592 if (hdr->encoding == HLL_DENSE) return C_OK; 593 594 /* Create a string of the right size filled with zero bytes. 595 * Note that the cached cardinality is set to 0 as a side effect 596 * that is exactly the cardinality of an empty HLL. */ 597 dense = sdsnewlen(NULL,HLL_DENSE_SIZE); 598 hdr = (struct hllhdr*) dense; 599 *hdr = *oldhdr; /* This will copy the magic and cached cardinality. */ 600 hdr->encoding = HLL_DENSE; 601 602 /* Now read the sparse representation and set non-zero registers 603 * accordingly. */ 604 p += HLL_HDR_SIZE; 605 while(p < end) { 606 if (HLL_SPARSE_IS_ZERO(p)) { 607 runlen = HLL_SPARSE_ZERO_LEN(p); 608 idx += runlen; 609 p++; 610 } else if (HLL_SPARSE_IS_XZERO(p)) { 611 runlen = HLL_SPARSE_XZERO_LEN(p); 612 idx += runlen; 613 p += 2; 614 } else { 615 runlen = HLL_SPARSE_VAL_LEN(p); 616 regval = HLL_SPARSE_VAL_VALUE(p); 617 if ((runlen + idx) > HLL_REGISTERS) break; /* Overflow. */ 618 while(runlen--) { 619 HLL_DENSE_SET_REGISTER(hdr->registers,idx,regval); 620 idx++; 621 } 622 p++; 623 } 624 } 625 626 /* If the sparse representation was valid, we expect to find idx 627 * set to HLL_REGISTERS. */ 628 if (idx != HLL_REGISTERS) { 629 sdsfree(dense); 630 return C_ERR; 631 } 632 633 /* Free the old representation and set the new one. */ 634 sdsfree(o->ptr); 635 o->ptr = dense; 636 return C_OK; 637 } 638 639 /* Low level function to set the sparse HLL register at 'index' to the 640 * specified value if the current value is smaller than 'count'. 641 * 642 * The object 'o' is the String object holding the HLL. The function requires 643 * a reference to the object in order to be able to enlarge the string if 644 * needed. 645 * 646 * On success, the function returns 1 if the cardinality changed, or 0 647 * if the register for this element was not updated. 648 * On error (if the representation is invalid) -1 is returned. 649 * 650 * As a side effect the function may promote the HLL representation from 651 * sparse to dense: this happens when a register requires to be set to a value 652 * not representable with the sparse representation, or when the resulting 653 * size would be greater than server.hll_sparse_max_bytes. */ 654 int hllSparseSet(robj *o, long index, uint8_t count) { 655 struct hllhdr *hdr; 656 uint8_t oldcount, *sparse, *end, *p, *prev, *next; 657 long first, span; 658 long is_zero = 0, is_xzero = 0, is_val = 0, runlen = 0; 659 660 /* If the count is too big to be representable by the sparse representation 661 * switch to dense representation. */ 662 if (count > HLL_SPARSE_VAL_MAX_VALUE) goto promote; 663 664 /* When updating a sparse representation, sometimes we may need to 665 * enlarge the buffer for up to 3 bytes in the worst case (XZERO split 666 * into XZERO-VAL-XZERO). Make sure there is enough space right now 667 * so that the pointers we take during the execution of the function 668 * will be valid all the time. */ 669 o->ptr = sdsMakeRoomFor(o->ptr,3); 670 671 /* Step 1: we need to locate the opcode we need to modify to check 672 * if a value update is actually needed. */ 673 sparse = p = ((uint8_t*)o->ptr) + HLL_HDR_SIZE; 674 end = p + sdslen(o->ptr) - HLL_HDR_SIZE; 675 676 first = 0; 677 prev = NULL; /* Points to previous opcode at the end of the loop. */ 678 next = NULL; /* Points to the next opcode at the end of the loop. */ 679 span = 0; 680 while(p < end) { 681 long oplen; 682 683 /* Set span to the number of registers covered by this opcode. 684 * 685 * This is the most performance critical loop of the sparse 686 * representation. Sorting the conditionals from the most to the 687 * least frequent opcode in many-bytes sparse HLLs is faster. */ 688 oplen = 1; 689 if (HLL_SPARSE_IS_ZERO(p)) { 690 span = HLL_SPARSE_ZERO_LEN(p); 691 } else if (HLL_SPARSE_IS_VAL(p)) { 692 span = HLL_SPARSE_VAL_LEN(p); 693 } else { /* XZERO. */ 694 span = HLL_SPARSE_XZERO_LEN(p); 695 oplen = 2; 696 } 697 /* Break if this opcode covers the register as 'index'. */ 698 if (index <= first+span-1) break; 699 prev = p; 700 p += oplen; 701 first += span; 702 } 703 if (span == 0) return -1; /* Invalid format. */ 704 705 next = HLL_SPARSE_IS_XZERO(p) ? p+2 : p+1; 706 if (next >= end) next = NULL; 707 708 /* Cache current opcode type to avoid using the macro again and 709 * again for something that will not change. 710 * Also cache the run-length of the opcode. */ 711 if (HLL_SPARSE_IS_ZERO(p)) { 712 is_zero = 1; 713 runlen = HLL_SPARSE_ZERO_LEN(p); 714 } else if (HLL_SPARSE_IS_XZERO(p)) { 715 is_xzero = 1; 716 runlen = HLL_SPARSE_XZERO_LEN(p); 717 } else { 718 is_val = 1; 719 runlen = HLL_SPARSE_VAL_LEN(p); 720 } 721 722 /* Step 2: After the loop: 723 * 724 * 'first' stores to the index of the first register covered 725 * by the current opcode, which is pointed by 'p'. 726 * 727 * 'next' ad 'prev' store respectively the next and previous opcode, 728 * or NULL if the opcode at 'p' is respectively the last or first. 729 * 730 * 'span' is set to the number of registers covered by the current 731 * opcode. 732 * 733 * There are different cases in order to update the data structure 734 * in place without generating it from scratch: 735 * 736 * A) If it is a VAL opcode already set to a value >= our 'count' 737 * no update is needed, regardless of the VAL run-length field. 738 * In this case PFADD returns 0 since no changes are performed. 739 * 740 * B) If it is a VAL opcode with len = 1 (representing only our 741 * register) and the value is less than 'count', we just update it 742 * since this is a trivial case. */ 743 if (is_val) { 744 oldcount = HLL_SPARSE_VAL_VALUE(p); 745 /* Case A. */ 746 if (oldcount >= count) return 0; 747 748 /* Case B. */ 749 if (runlen == 1) { 750 HLL_SPARSE_VAL_SET(p,count,1); 751 goto updated; 752 } 753 } 754 755 /* C) Another trivial to handle case is a ZERO opcode with a len of 1. 756 * We can just replace it with a VAL opcode with our value and len of 1. */ 757 if (is_zero && runlen == 1) { 758 HLL_SPARSE_VAL_SET(p,count,1); 759 goto updated; 760 } 761 762 /* D) General case. 763 * 764 * The other cases are more complex: our register requires to be updated 765 * and is either currently represented by a VAL opcode with len > 1, 766 * by a ZERO opcode with len > 1, or by an XZERO opcode. 767 * 768 * In those cases the original opcode must be split into multiple 769 * opcodes. The worst case is an XZERO split in the middle resuling into 770 * XZERO - VAL - XZERO, so the resulting sequence max length is 771 * 5 bytes. 772 * 773 * We perform the split writing the new sequence into the 'new' buffer 774 * with 'newlen' as length. Later the new sequence is inserted in place 775 * of the old one, possibly moving what is on the right a few bytes 776 * if the new sequence is longer than the older one. */ 777 uint8_t seq[5], *n = seq; 778 int last = first+span-1; /* Last register covered by the sequence. */ 779 int len; 780 781 if (is_zero || is_xzero) { 782 /* Handle splitting of ZERO / XZERO. */ 783 if (index != first) { 784 len = index-first; 785 if (len > HLL_SPARSE_ZERO_MAX_LEN) { 786 HLL_SPARSE_XZERO_SET(n,len); 787 n += 2; 788 } else { 789 HLL_SPARSE_ZERO_SET(n,len); 790 n++; 791 } 792 } 793 HLL_SPARSE_VAL_SET(n,count,1); 794 n++; 795 if (index != last) { 796 len = last-index; 797 if (len > HLL_SPARSE_ZERO_MAX_LEN) { 798 HLL_SPARSE_XZERO_SET(n,len); 799 n += 2; 800 } else { 801 HLL_SPARSE_ZERO_SET(n,len); 802 n++; 803 } 804 } 805 } else { 806 /* Handle splitting of VAL. */ 807 int curval = HLL_SPARSE_VAL_VALUE(p); 808 809 if (index != first) { 810 len = index-first; 811 HLL_SPARSE_VAL_SET(n,curval,len); 812 n++; 813 } 814 HLL_SPARSE_VAL_SET(n,count,1); 815 n++; 816 if (index != last) { 817 len = last-index; 818 HLL_SPARSE_VAL_SET(n,curval,len); 819 n++; 820 } 821 } 822 823 /* Step 3: substitute the new sequence with the old one. 824 * 825 * Note that we already allocated space on the sds string 826 * calling sdsMakeRoomFor(). */ 827 int seqlen = n-seq; 828 int oldlen = is_xzero ? 2 : 1; 829 int deltalen = seqlen-oldlen; 830 831 if (deltalen > 0 && 832 sdslen(o->ptr)+deltalen > server.hll_sparse_max_bytes) goto promote; 833 if (deltalen && next) memmove(next+deltalen,next,end-next); 834 sdsIncrLen(o->ptr,deltalen); 835 memcpy(p,seq,seqlen); 836 end += deltalen; 837 838 updated: 839 /* Step 4: Merge adjacent values if possible. 840 * 841 * The representation was updated, however the resulting representation 842 * may not be optimal: adjacent VAL opcodes can sometimes be merged into 843 * a single one. */ 844 p = prev ? prev : sparse; 845 int scanlen = 5; /* Scan up to 5 upcodes starting from prev. */ 846 while (p < end && scanlen--) { 847 if (HLL_SPARSE_IS_XZERO(p)) { 848 p += 2; 849 continue; 850 } else if (HLL_SPARSE_IS_ZERO(p)) { 851 p++; 852 continue; 853 } 854 /* We need two adjacent VAL opcodes to try a merge, having 855 * the same value, and a len that fits the VAL opcode max len. */ 856 if (p+1 < end && HLL_SPARSE_IS_VAL(p+1)) { 857 int v1 = HLL_SPARSE_VAL_VALUE(p); 858 int v2 = HLL_SPARSE_VAL_VALUE(p+1); 859 if (v1 == v2) { 860 int len = HLL_SPARSE_VAL_LEN(p)+HLL_SPARSE_VAL_LEN(p+1); 861 if (len <= HLL_SPARSE_VAL_MAX_LEN) { 862 HLL_SPARSE_VAL_SET(p+1,v1,len); 863 memmove(p,p+1,end-p); 864 sdsIncrLen(o->ptr,-1); 865 end--; 866 /* After a merge we reiterate without incrementing 'p' 867 * in order to try to merge the just merged value with 868 * a value on its right. */ 869 continue; 870 } 871 } 872 } 873 p++; 874 } 875 876 /* Invalidate the cached cardinality. */ 877 hdr = o->ptr; 878 HLL_INVALIDATE_CACHE(hdr); 879 return 1; 880 881 promote: /* Promote to dense representation. */ 882 if (hllSparseToDense(o) == C_ERR) return -1; /* Corrupted HLL. */ 883 hdr = o->ptr; 884 885 /* We need to call hllDenseAdd() to perform the operation after the 886 * conversion. However the result must be 1, since if we need to 887 * convert from sparse to dense a register requires to be updated. 888 * 889 * Note that this in turn means that PFADD will make sure the command 890 * is propagated to slaves / AOF, so if there is a sparse -> dense 891 * conversion, it will be performed in all the slaves as well. */ 892 int dense_retval = hllDenseSet(hdr->registers,index,count); 893 serverAssert(dense_retval == 1); 894 return dense_retval; 895 } 896 897 /* "Add" the element in the sparse hyperloglog data structure. 898 * Actually nothing is added, but the max 0 pattern counter of the subset 899 * the element belongs to is incremented if needed. 900 * 901 * This function is actually a wrapper for hllSparseSet(), it only performs 902 * the hashshing of the elmenet to obtain the index and zeros run length. */ 903 int hllSparseAdd(robj *o, unsigned char *ele, size_t elesize) { 904 long index; 905 uint8_t count = hllPatLen(ele,elesize,&index); 906 /* Update the register if this element produced a longer run of zeroes. */ 907 return hllSparseSet(o,index,count); 908 } 909 910 /* Compute the register histogram in the sparse representation. */ 911 void hllSparseRegHisto(uint8_t *sparse, int sparselen, int *invalid, int* reghisto) { 912 int idx = 0, runlen, regval; 913 uint8_t *end = sparse+sparselen, *p = sparse; 914 915 while(p < end) { 916 if (HLL_SPARSE_IS_ZERO(p)) { 917 runlen = HLL_SPARSE_ZERO_LEN(p); 918 idx += runlen; 919 reghisto[0] += runlen; 920 p++; 921 } else if (HLL_SPARSE_IS_XZERO(p)) { 922 runlen = HLL_SPARSE_XZERO_LEN(p); 923 idx += runlen; 924 reghisto[0] += runlen; 925 p += 2; 926 } else { 927 runlen = HLL_SPARSE_VAL_LEN(p); 928 regval = HLL_SPARSE_VAL_VALUE(p); 929 idx += runlen; 930 reghisto[regval] += runlen; 931 p++; 932 } 933 } 934 if (idx != HLL_REGISTERS && invalid) *invalid = 1; 935 } 936 937 /* ========================= HyperLogLog Count ============================== 938 * This is the core of the algorithm where the approximated count is computed. 939 * The function uses the lower level hllDenseRegHisto() and hllSparseRegHisto() 940 * functions as helpers to compute histogram of register values part of the 941 * computation, which is representation-specific, while all the rest is common. */ 942 943 /* Implements the register histogram calculation for uint8_t data type 944 * which is only used internally as speedup for PFCOUNT with multiple keys. */ 945 void hllRawRegHisto(uint8_t *registers, int* reghisto) { 946 uint64_t *word = (uint64_t*) registers; 947 uint8_t *bytes; 948 int j; 949 950 for (j = 0; j < HLL_REGISTERS/8; j++) { 951 if (*word == 0) { 952 reghisto[0] += 8; 953 } else { 954 bytes = (uint8_t*) word; 955 reghisto[bytes[0]]++; 956 reghisto[bytes[1]]++; 957 reghisto[bytes[2]]++; 958 reghisto[bytes[3]]++; 959 reghisto[bytes[4]]++; 960 reghisto[bytes[5]]++; 961 reghisto[bytes[6]]++; 962 reghisto[bytes[7]]++; 963 } 964 word++; 965 } 966 } 967 968 /* Helper function sigma as defined in 969 * "New cardinality estimation algorithms for HyperLogLog sketches" 970 * Otmar Ertl, arXiv:1702.01284 */ 971 double hllSigma(double x) { 972 if (x == 1.) return INFINITY; 973 double zPrime; 974 double y = 1; 975 double z = x; 976 do { 977 x *= x; 978 zPrime = z; 979 z += x * y; 980 y += y; 981 } while(zPrime != z); 982 return z; 983 } 984 985 /* Helper function tau as defined in 986 * "New cardinality estimation algorithms for HyperLogLog sketches" 987 * Otmar Ertl, arXiv:1702.01284 */ 988 double hllTau(double x) { 989 if (x == 0. || x == 1.) return 0.; 990 double zPrime; 991 double y = 1.0; 992 double z = 1 - x; 993 do { 994 x = sqrt(x); 995 zPrime = z; 996 y *= 0.5; 997 z -= pow(1 - x, 2)*y; 998 } while(zPrime != z); 999 return z / 3; 1000 } 1001 1002 /* Return the approximated cardinality of the set based on the harmonic 1003 * mean of the registers values. 'hdr' points to the start of the SDS 1004 * representing the String object holding the HLL representation. 1005 * 1006 * If the sparse representation of the HLL object is not valid, the integer 1007 * pointed by 'invalid' is set to non-zero, otherwise it is left untouched. 1008 * 1009 * hllCount() supports a special internal-only encoding of HLL_RAW, that 1010 * is, hdr->registers will point to an uint8_t array of HLL_REGISTERS element. 1011 * This is useful in order to speedup PFCOUNT when called against multiple 1012 * keys (no need to work with 6-bit integers encoding). */ 1013 uint64_t hllCount(struct hllhdr *hdr, int *invalid) { 1014 double m = HLL_REGISTERS; 1015 double E; 1016 int j; 1017 /* Note that reghisto size could be just HLL_Q+2, becuase HLL_Q+1 is 1018 * the maximum frequency of the "000...1" sequence the hash function is 1019 * able to return. However it is slow to check for sanity of the 1020 * input: instead we history array at a safe size: overflows will 1021 * just write data to wrong, but correctly allocated, places. */ 1022 int reghisto[64] = {0}; 1023 1024 /* Compute register histogram */ 1025 if (hdr->encoding == HLL_DENSE) { 1026 hllDenseRegHisto(hdr->registers,reghisto); 1027 } else if (hdr->encoding == HLL_SPARSE) { 1028 hllSparseRegHisto(hdr->registers, 1029 sdslen((sds)hdr)-HLL_HDR_SIZE,invalid,reghisto); 1030 } else if (hdr->encoding == HLL_RAW) { 1031 hllRawRegHisto(hdr->registers,reghisto); 1032 } else { 1033 serverPanic("Unknown HyperLogLog encoding in hllCount()"); 1034 } 1035 1036 /* Estimate cardinality form register histogram. See: 1037 * "New cardinality estimation algorithms for HyperLogLog sketches" 1038 * Otmar Ertl, arXiv:1702.01284 */ 1039 double z = m * hllTau((m-reghisto[HLL_Q+1])/(double)m); 1040 for (j = HLL_Q; j >= 1; --j) { 1041 z += reghisto[j]; 1042 z *= 0.5; 1043 } 1044 z += m * hllSigma(reghisto[0]/(double)m); 1045 E = llroundl(HLL_ALPHA_INF*m*m/z); 1046 1047 return (uint64_t) E; 1048 } 1049 1050 /* Call hllDenseAdd() or hllSparseAdd() according to the HLL encoding. */ 1051 int hllAdd(robj *o, unsigned char *ele, size_t elesize) { 1052 struct hllhdr *hdr = o->ptr; 1053 switch(hdr->encoding) { 1054 case HLL_DENSE: return hllDenseAdd(hdr->registers,ele,elesize); 1055 case HLL_SPARSE: return hllSparseAdd(o,ele,elesize); 1056 default: return -1; /* Invalid representation. */ 1057 } 1058 } 1059 1060 /* Merge by computing MAX(registers[i],hll[i]) the HyperLogLog 'hll' 1061 * with an array of uint8_t HLL_REGISTERS registers pointed by 'max'. 1062 * 1063 * The hll object must be already validated via isHLLObjectOrReply() 1064 * or in some other way. 1065 * 1066 * If the HyperLogLog is sparse and is found to be invalid, C_ERR 1067 * is returned, otherwise the function always succeeds. */ 1068 int hllMerge(uint8_t *max, robj *hll) { 1069 struct hllhdr *hdr = hll->ptr; 1070 int i; 1071 1072 if (hdr->encoding == HLL_DENSE) { 1073 uint8_t val; 1074 1075 for (i = 0; i < HLL_REGISTERS; i++) { 1076 HLL_DENSE_GET_REGISTER(val,hdr->registers,i); 1077 if (val > max[i]) max[i] = val; 1078 } 1079 } else { 1080 uint8_t *p = hll->ptr, *end = p + sdslen(hll->ptr); 1081 long runlen, regval; 1082 1083 p += HLL_HDR_SIZE; 1084 i = 0; 1085 while(p < end) { 1086 if (HLL_SPARSE_IS_ZERO(p)) { 1087 runlen = HLL_SPARSE_ZERO_LEN(p); 1088 i += runlen; 1089 p++; 1090 } else if (HLL_SPARSE_IS_XZERO(p)) { 1091 runlen = HLL_SPARSE_XZERO_LEN(p); 1092 i += runlen; 1093 p += 2; 1094 } else { 1095 runlen = HLL_SPARSE_VAL_LEN(p); 1096 regval = HLL_SPARSE_VAL_VALUE(p); 1097 if ((runlen + i) > HLL_REGISTERS) break; /* Overflow. */ 1098 while(runlen--) { 1099 if (regval > max[i]) max[i] = regval; 1100 i++; 1101 } 1102 p++; 1103 } 1104 } 1105 if (i != HLL_REGISTERS) return C_ERR; 1106 } 1107 return C_OK; 1108 } 1109 1110 /* ========================== HyperLogLog commands ========================== */ 1111 1112 /* Create an HLL object. We always create the HLL using sparse encoding. 1113 * This will be upgraded to the dense representation as needed. */ 1114 robj *createHLLObject(void) { 1115 robj *o; 1116 struct hllhdr *hdr; 1117 sds s; 1118 uint8_t *p; 1119 int sparselen = HLL_HDR_SIZE + 1120 (((HLL_REGISTERS+(HLL_SPARSE_XZERO_MAX_LEN-1)) / 1121 HLL_SPARSE_XZERO_MAX_LEN)*2); 1122 int aux; 1123 1124 /* Populate the sparse representation with as many XZERO opcodes as 1125 * needed to represent all the registers. */ 1126 aux = HLL_REGISTERS; 1127 s = sdsnewlen(NULL,sparselen); 1128 p = (uint8_t*)s + HLL_HDR_SIZE; 1129 while(aux) { 1130 int xzero = HLL_SPARSE_XZERO_MAX_LEN; 1131 if (xzero > aux) xzero = aux; 1132 HLL_SPARSE_XZERO_SET(p,xzero); 1133 p += 2; 1134 aux -= xzero; 1135 } 1136 serverAssert((p-(uint8_t*)s) == sparselen); 1137 1138 /* Create the actual object. */ 1139 o = createObject(OBJ_STRING,s); 1140 hdr = o->ptr; 1141 memcpy(hdr->magic,"HYLL",4); 1142 hdr->encoding = HLL_SPARSE; 1143 return o; 1144 } 1145 1146 /* Check if the object is a String with a valid HLL representation. 1147 * Return C_OK if this is true, otherwise reply to the client 1148 * with an error and return C_ERR. */ 1149 int isHLLObjectOrReply(client *c, robj *o) { 1150 struct hllhdr *hdr; 1151 1152 /* Key exists, check type */ 1153 if (checkType(c,o,OBJ_STRING)) 1154 return C_ERR; /* Error already sent. */ 1155 1156 if (!sdsEncodedObject(o)) goto invalid; 1157 if (stringObjectLen(o) < sizeof(*hdr)) goto invalid; 1158 hdr = o->ptr; 1159 1160 /* Magic should be "HYLL". */ 1161 if (hdr->magic[0] != 'H' || hdr->magic[1] != 'Y' || 1162 hdr->magic[2] != 'L' || hdr->magic[3] != 'L') goto invalid; 1163 1164 if (hdr->encoding > HLL_MAX_ENCODING) goto invalid; 1165 1166 /* Dense representation string length should match exactly. */ 1167 if (hdr->encoding == HLL_DENSE && 1168 stringObjectLen(o) != HLL_DENSE_SIZE) goto invalid; 1169 1170 /* All tests passed. */ 1171 return C_OK; 1172 1173 invalid: 1174 addReplySds(c, 1175 sdsnew("-WRONGTYPE Key is not a valid " 1176 "HyperLogLog string value.\r\n")); 1177 return C_ERR; 1178 } 1179 1180 /* PFADD var ele ele ele ... ele => :0 or :1 */ 1181 void pfaddCommand(client *c) { 1182 robj *o = lookupKeyWrite(c->db,c->argv[1]); 1183 struct hllhdr *hdr; 1184 int updated = 0, j; 1185 1186 if (o == NULL) { 1187 /* Create the key with a string value of the exact length to 1188 * hold our HLL data structure. sdsnewlen() when NULL is passed 1189 * is guaranteed to return bytes initialized to zero. */ 1190 o = createHLLObject(); 1191 dbAdd(c->db,c->argv[1],o); 1192 updated++; 1193 } else { 1194 if (isHLLObjectOrReply(c,o) != C_OK) return; 1195 o = dbUnshareStringValue(c->db,c->argv[1],o); 1196 } 1197 /* Perform the low level ADD operation for every element. */ 1198 for (j = 2; j < c->argc; j++) { 1199 int retval = hllAdd(o, (unsigned char*)c->argv[j]->ptr, 1200 sdslen(c->argv[j]->ptr)); 1201 switch(retval) { 1202 case 1: 1203 updated++; 1204 break; 1205 case -1: 1206 addReplySds(c,sdsnew(invalid_hll_err)); 1207 return; 1208 } 1209 } 1210 hdr = o->ptr; 1211 if (updated) { 1212 signalModifiedKey(c->db,c->argv[1]); 1213 notifyKeyspaceEvent(NOTIFY_STRING,"pfadd",c->argv[1],c->db->id); 1214 server.dirty++; 1215 HLL_INVALIDATE_CACHE(hdr); 1216 } 1217 addReply(c, updated ? shared.cone : shared.czero); 1218 } 1219 1220 /* PFCOUNT var -> approximated cardinality of set. */ 1221 void pfcountCommand(client *c) { 1222 robj *o; 1223 struct hllhdr *hdr; 1224 uint64_t card; 1225 1226 /* Case 1: multi-key keys, cardinality of the union. 1227 * 1228 * When multiple keys are specified, PFCOUNT actually computes 1229 * the cardinality of the merge of the N HLLs specified. */ 1230 if (c->argc > 2) { 1231 uint8_t max[HLL_HDR_SIZE+HLL_REGISTERS], *registers; 1232 int j; 1233 1234 /* Compute an HLL with M[i] = MAX(M[i]_j). */ 1235 memset(max,0,sizeof(max)); 1236 hdr = (struct hllhdr*) max; 1237 hdr->encoding = HLL_RAW; /* Special internal-only encoding. */ 1238 registers = max + HLL_HDR_SIZE; 1239 for (j = 1; j < c->argc; j++) { 1240 /* Check type and size. */ 1241 robj *o = lookupKeyRead(c->db,c->argv[j]); 1242 if (o == NULL) continue; /* Assume empty HLL for non existing var.*/ 1243 if (isHLLObjectOrReply(c,o) != C_OK) return; 1244 1245 /* Merge with this HLL with our 'max' HHL by setting max[i] 1246 * to MAX(max[i],hll[i]). */ 1247 if (hllMerge(registers,o) == C_ERR) { 1248 addReplySds(c,sdsnew(invalid_hll_err)); 1249 return; 1250 } 1251 } 1252 1253 /* Compute cardinality of the resulting set. */ 1254 addReplyLongLong(c,hllCount(hdr,NULL)); 1255 return; 1256 } 1257 1258 /* Case 2: cardinality of the single HLL. 1259 * 1260 * The user specified a single key. Either return the cached value 1261 * or compute one and update the cache. */ 1262 o = lookupKeyWrite(c->db,c->argv[1]); 1263 if (o == NULL) { 1264 /* No key? Cardinality is zero since no element was added, otherwise 1265 * we would have a key as HLLADD creates it as a side effect. */ 1266 addReply(c,shared.czero); 1267 } else { 1268 if (isHLLObjectOrReply(c,o) != C_OK) return; 1269 o = dbUnshareStringValue(c->db,c->argv[1],o); 1270 1271 /* Check if the cached cardinality is valid. */ 1272 hdr = o->ptr; 1273 if (HLL_VALID_CACHE(hdr)) { 1274 /* Just return the cached value. */ 1275 card = (uint64_t)hdr->card[0]; 1276 card |= (uint64_t)hdr->card[1] << 8; 1277 card |= (uint64_t)hdr->card[2] << 16; 1278 card |= (uint64_t)hdr->card[3] << 24; 1279 card |= (uint64_t)hdr->card[4] << 32; 1280 card |= (uint64_t)hdr->card[5] << 40; 1281 card |= (uint64_t)hdr->card[6] << 48; 1282 card |= (uint64_t)hdr->card[7] << 56; 1283 } else { 1284 int invalid = 0; 1285 /* Recompute it and update the cached value. */ 1286 card = hllCount(hdr,&invalid); 1287 if (invalid) { 1288 addReplySds(c,sdsnew(invalid_hll_err)); 1289 return; 1290 } 1291 hdr->card[0] = card & 0xff; 1292 hdr->card[1] = (card >> 8) & 0xff; 1293 hdr->card[2] = (card >> 16) & 0xff; 1294 hdr->card[3] = (card >> 24) & 0xff; 1295 hdr->card[4] = (card >> 32) & 0xff; 1296 hdr->card[5] = (card >> 40) & 0xff; 1297 hdr->card[6] = (card >> 48) & 0xff; 1298 hdr->card[7] = (card >> 56) & 0xff; 1299 /* This is not considered a read-only command even if the 1300 * data structure is not modified, since the cached value 1301 * may be modified and given that the HLL is a Redis string 1302 * we need to propagate the change. */ 1303 signalModifiedKey(c->db,c->argv[1]); 1304 server.dirty++; 1305 } 1306 addReplyLongLong(c,card); 1307 } 1308 } 1309 1310 /* PFMERGE dest src1 src2 src3 ... srcN => OK */ 1311 void pfmergeCommand(client *c) { 1312 uint8_t max[HLL_REGISTERS]; 1313 struct hllhdr *hdr; 1314 int j; 1315 int use_dense = 0; /* Use dense representation as target? */ 1316 1317 /* Compute an HLL with M[i] = MAX(M[i]_j). 1318 * We store the maximum into the max array of registers. We'll write 1319 * it to the target variable later. */ 1320 memset(max,0,sizeof(max)); 1321 for (j = 1; j < c->argc; j++) { 1322 /* Check type and size. */ 1323 robj *o = lookupKeyRead(c->db,c->argv[j]); 1324 if (o == NULL) continue; /* Assume empty HLL for non existing var. */ 1325 if (isHLLObjectOrReply(c,o) != C_OK) return; 1326 1327 /* If at least one involved HLL is dense, use the dense representation 1328 * as target ASAP to save time and avoid the conversion step. */ 1329 hdr = o->ptr; 1330 if (hdr->encoding == HLL_DENSE) use_dense = 1; 1331 1332 /* Merge with this HLL with our 'max' HHL by setting max[i] 1333 * to MAX(max[i],hll[i]). */ 1334 if (hllMerge(max,o) == C_ERR) { 1335 addReplySds(c,sdsnew(invalid_hll_err)); 1336 return; 1337 } 1338 } 1339 1340 /* Create / unshare the destination key's value if needed. */ 1341 robj *o = lookupKeyWrite(c->db,c->argv[1]); 1342 if (o == NULL) { 1343 /* Create the key with a string value of the exact length to 1344 * hold our HLL data structure. sdsnewlen() when NULL is passed 1345 * is guaranteed to return bytes initialized to zero. */ 1346 o = createHLLObject(); 1347 dbAdd(c->db,c->argv[1],o); 1348 } else { 1349 /* If key exists we are sure it's of the right type/size 1350 * since we checked when merging the different HLLs, so we 1351 * don't check again. */ 1352 o = dbUnshareStringValue(c->db,c->argv[1],o); 1353 } 1354 1355 /* Convert the destination object to dense representation if at least 1356 * one of the inputs was dense. */ 1357 if (use_dense && hllSparseToDense(o) == C_ERR) { 1358 addReplySds(c,sdsnew(invalid_hll_err)); 1359 return; 1360 } 1361 1362 /* Write the resulting HLL to the destination HLL registers and 1363 * invalidate the cached value. */ 1364 for (j = 0; j < HLL_REGISTERS; j++) { 1365 if (max[j] == 0) continue; 1366 hdr = o->ptr; 1367 switch(hdr->encoding) { 1368 case HLL_DENSE: hllDenseSet(hdr->registers,j,max[j]); break; 1369 case HLL_SPARSE: hllSparseSet(o,j,max[j]); break; 1370 } 1371 } 1372 hdr = o->ptr; /* o->ptr may be different now, as a side effect of 1373 last hllSparseSet() call. */ 1374 HLL_INVALIDATE_CACHE(hdr); 1375 1376 signalModifiedKey(c->db,c->argv[1]); 1377 /* We generate a PFADD event for PFMERGE for semantical simplicity 1378 * since in theory this is a mass-add of elements. */ 1379 notifyKeyspaceEvent(NOTIFY_STRING,"pfadd",c->argv[1],c->db->id); 1380 server.dirty++; 1381 addReply(c,shared.ok); 1382 } 1383 1384 /* ========================== Testing / Debugging ========================== */ 1385 1386 /* PFSELFTEST 1387 * This command performs a self-test of the HLL registers implementation. 1388 * Something that is not easy to test from within the outside. */ 1389 #define HLL_TEST_CYCLES 1000 1390 void pfselftestCommand(client *c) { 1391 unsigned int j, i; 1392 sds bitcounters = sdsnewlen(NULL,HLL_DENSE_SIZE); 1393 struct hllhdr *hdr = (struct hllhdr*) bitcounters, *hdr2; 1394 robj *o = NULL; 1395 uint8_t bytecounters[HLL_REGISTERS]; 1396 1397 /* Test 1: access registers. 1398 * The test is conceived to test that the different counters of our data 1399 * structure are accessible and that setting their values both result in 1400 * the correct value to be retained and not affect adjacent values. */ 1401 for (j = 0; j < HLL_TEST_CYCLES; j++) { 1402 /* Set the HLL counters and an array of unsigned byes of the 1403 * same size to the same set of random values. */ 1404 for (i = 0; i < HLL_REGISTERS; i++) { 1405 unsigned int r = rand() & HLL_REGISTER_MAX; 1406 1407 bytecounters[i] = r; 1408 HLL_DENSE_SET_REGISTER(hdr->registers,i,r); 1409 } 1410 /* Check that we are able to retrieve the same values. */ 1411 for (i = 0; i < HLL_REGISTERS; i++) { 1412 unsigned int val; 1413 1414 HLL_DENSE_GET_REGISTER(val,hdr->registers,i); 1415 if (val != bytecounters[i]) { 1416 addReplyErrorFormat(c, 1417 "TESTFAILED Register %d should be %d but is %d", 1418 i, (int) bytecounters[i], (int) val); 1419 goto cleanup; 1420 } 1421 } 1422 } 1423 1424 /* Test 2: approximation error. 1425 * The test adds unique elements and check that the estimated value 1426 * is always reasonable bounds. 1427 * 1428 * We check that the error is smaller than a few times than the expected 1429 * standard error, to make it very unlikely for the test to fail because 1430 * of a "bad" run. 1431 * 1432 * The test is performed with both dense and sparse HLLs at the same 1433 * time also verifying that the computed cardinality is the same. */ 1434 memset(hdr->registers,0,HLL_DENSE_SIZE-HLL_HDR_SIZE); 1435 o = createHLLObject(); 1436 double relerr = 1.04/sqrt(HLL_REGISTERS); 1437 int64_t checkpoint = 1; 1438 uint64_t seed = (uint64_t)rand() | (uint64_t)rand() << 32; 1439 uint64_t ele; 1440 for (j = 1; j <= 10000000; j++) { 1441 ele = j ^ seed; 1442 hllDenseAdd(hdr->registers,(unsigned char*)&ele,sizeof(ele)); 1443 hllAdd(o,(unsigned char*)&ele,sizeof(ele)); 1444 1445 /* Make sure that for small cardinalities we use sparse 1446 * encoding. */ 1447 if (j == checkpoint && j < server.hll_sparse_max_bytes/2) { 1448 hdr2 = o->ptr; 1449 if (hdr2->encoding != HLL_SPARSE) { 1450 addReplyError(c, "TESTFAILED sparse encoding not used"); 1451 goto cleanup; 1452 } 1453 } 1454 1455 /* Check that dense and sparse representations agree. */ 1456 if (j == checkpoint && hllCount(hdr,NULL) != hllCount(o->ptr,NULL)) { 1457 addReplyError(c, "TESTFAILED dense/sparse disagree"); 1458 goto cleanup; 1459 } 1460 1461 /* Check error. */ 1462 if (j == checkpoint) { 1463 int64_t abserr = checkpoint - (int64_t)hllCount(hdr,NULL); 1464 uint64_t maxerr = ceil(relerr*6*checkpoint); 1465 1466 /* Adjust the max error we expect for cardinality 10 1467 * since from time to time it is statistically likely to get 1468 * much higher error due to collision, resulting into a false 1469 * positive. */ 1470 if (j == 10) maxerr = 1; 1471 1472 if (abserr < 0) abserr = -abserr; 1473 if (abserr > (int64_t)maxerr) { 1474 addReplyErrorFormat(c, 1475 "TESTFAILED Too big error. card:%llu abserr:%llu", 1476 (unsigned long long) checkpoint, 1477 (unsigned long long) abserr); 1478 goto cleanup; 1479 } 1480 checkpoint *= 10; 1481 } 1482 } 1483 1484 /* Success! */ 1485 addReply(c,shared.ok); 1486 1487 cleanup: 1488 sdsfree(bitcounters); 1489 if (o) decrRefCount(o); 1490 } 1491 1492 /* PFDEBUG <subcommand> <key> ... args ... 1493 * Different debugging related operations about the HLL implementation. */ 1494 void pfdebugCommand(client *c) { 1495 char *cmd = c->argv[1]->ptr; 1496 struct hllhdr *hdr; 1497 robj *o; 1498 int j; 1499 1500 o = lookupKeyWrite(c->db,c->argv[2]); 1501 if (o == NULL) { 1502 addReplyError(c,"The specified key does not exist"); 1503 return; 1504 } 1505 if (isHLLObjectOrReply(c,o) != C_OK) return; 1506 o = dbUnshareStringValue(c->db,c->argv[2],o); 1507 hdr = o->ptr; 1508 1509 /* PFDEBUG GETREG <key> */ 1510 if (!strcasecmp(cmd,"getreg")) { 1511 if (c->argc != 3) goto arityerr; 1512 1513 if (hdr->encoding == HLL_SPARSE) { 1514 if (hllSparseToDense(o) == C_ERR) { 1515 addReplySds(c,sdsnew(invalid_hll_err)); 1516 return; 1517 } 1518 server.dirty++; /* Force propagation on encoding change. */ 1519 } 1520 1521 hdr = o->ptr; 1522 addReplyMultiBulkLen(c,HLL_REGISTERS); 1523 for (j = 0; j < HLL_REGISTERS; j++) { 1524 uint8_t val; 1525 1526 HLL_DENSE_GET_REGISTER(val,hdr->registers,j); 1527 addReplyLongLong(c,val); 1528 } 1529 } 1530 /* PFDEBUG DECODE <key> */ 1531 else if (!strcasecmp(cmd,"decode")) { 1532 if (c->argc != 3) goto arityerr; 1533 1534 uint8_t *p = o->ptr, *end = p+sdslen(o->ptr); 1535 sds decoded = sdsempty(); 1536 1537 if (hdr->encoding != HLL_SPARSE) { 1538 addReplyError(c,"HLL encoding is not sparse"); 1539 return; 1540 } 1541 1542 p += HLL_HDR_SIZE; 1543 while(p < end) { 1544 int runlen, regval; 1545 1546 if (HLL_SPARSE_IS_ZERO(p)) { 1547 runlen = HLL_SPARSE_ZERO_LEN(p); 1548 p++; 1549 decoded = sdscatprintf(decoded,"z:%d ",runlen); 1550 } else if (HLL_SPARSE_IS_XZERO(p)) { 1551 runlen = HLL_SPARSE_XZERO_LEN(p); 1552 p += 2; 1553 decoded = sdscatprintf(decoded,"Z:%d ",runlen); 1554 } else { 1555 runlen = HLL_SPARSE_VAL_LEN(p); 1556 regval = HLL_SPARSE_VAL_VALUE(p); 1557 p++; 1558 decoded = sdscatprintf(decoded,"v:%d,%d ",regval,runlen); 1559 } 1560 } 1561 decoded = sdstrim(decoded," "); 1562 addReplyBulkCBuffer(c,decoded,sdslen(decoded)); 1563 sdsfree(decoded); 1564 } 1565 /* PFDEBUG ENCODING <key> */ 1566 else if (!strcasecmp(cmd,"encoding")) { 1567 char *encodingstr[2] = {"dense","sparse"}; 1568 if (c->argc != 3) goto arityerr; 1569 1570 addReplyStatus(c,encodingstr[hdr->encoding]); 1571 } 1572 /* PFDEBUG TODENSE <key> */ 1573 else if (!strcasecmp(cmd,"todense")) { 1574 int conv = 0; 1575 if (c->argc != 3) goto arityerr; 1576 1577 if (hdr->encoding == HLL_SPARSE) { 1578 if (hllSparseToDense(o) == C_ERR) { 1579 addReplySds(c,sdsnew(invalid_hll_err)); 1580 return; 1581 } 1582 conv = 1; 1583 server.dirty++; /* Force propagation on encoding change. */ 1584 } 1585 addReply(c,conv ? shared.cone : shared.czero); 1586 } else { 1587 addReplyErrorFormat(c,"Unknown PFDEBUG subcommand '%s'", cmd); 1588 } 1589 return; 1590 1591 arityerr: 1592 addReplyErrorFormat(c, 1593 "Wrong number of arguments for the '%s' subcommand",cmd); 1594 } 1595 1596