? ? ? ? 有序整數(shù)集是Redis源碼中一個以大尾（big endian）形式存儲，由小到大排列且無重復(fù)的整型集合。它存儲的類型包括16位、32位和64位的整型數(shù)。在介紹這個庫的實現(xiàn)前，我們還需要先熟悉下大小尾內(nèi)存存儲機制。（轉(zhuǎn)載請指明出于breaksoftware的csdn博客）

大小尾（Big Endian/Little Endian）

? ? ? ? 第一次接觸這個概念還是在大學(xué)時上“計算機原理”時，當(dāng)時只是簡單的知道它的特點，但是絲毫沒有深究它們產(chǎn)生的原因和特點。這次借著解析Redis源碼，重新學(xué)習(xí)了一下大小尾。

? ? ? ? 如果進行過逆向的同學(xué)一般都會知道ESP指針的作用，它指向當(dāng)前函數(shù)棧Frame的棧頂。相應(yīng)的，EBP寄存器指向當(dāng)前函數(shù)棧Frame的棧底。在逆向的匯編代碼中，我們會發(fā)現(xiàn)一般使用EBP結(jié)合偏移的方法去表示棧上的一個變量，如：

mov    eax, dword ptr  [ebp+0ch]
mov    ebx, dword ptr  [ebp+08h]

? ? ? ? 但是為什么沒有使用ESP表示變量呢？這是因為ESP指針在當(dāng)前調(diào)用堆棧中也不是穩(wěn)定的，比如我們進行函數(shù)調(diào)用，會將一些信息進行入棧處理，這個時候棧頂指針就會減小以擴大有效棧的區(qū)域。為什么是棧頂值減小以擴大棧區(qū)域呢？這是因為棧結(jié)構(gòu)的特點：棧底地址比棧頂?shù)刂反蟆?/p>

?

? ? ? ? 我們再看下大小尾數(shù)據(jù)在棧空間的布局。

? ? ? ? 大尾結(jié)構(gòu)將數(shù)據(jù)的高位放在地址低處，而小尾結(jié)構(gòu)將數(shù)據(jù)的高位放在地址的高位。于是我們看下0x00123456在不同結(jié)構(gòu)中的布局

? ? ? ? 如果我們直接在內(nèi)存中查看，則是這樣展現(xiàn)的

? ? ? ? 可以發(fā)現(xiàn)大尾的結(jié)構(gòu)比較正常，它展現(xiàn)的數(shù)據(jù)的形式和人類的認(rèn)知方式是相同的——高位在前，低位在后。但是小尾的展現(xiàn)形式則比較反人類——需要倒著看。但是存在即合理，那么小尾結(jié)構(gòu)那么反人類為什么它還存在呢？

? ? ? ? 這就要從CPU的歷史講起來。歷史上關(guān)于選擇大尾還是小尾有著很多爭論，各派都有自己的理論依據(jù)。我在網(wǎng)上找到一篇關(guān)于這段歷史的文檔——《ON HOLY WARS AND A PLEA FOR PEACE》，鑒于文檔稍長，我沒有仔細(xì)閱讀，有興趣的讀者可以去了解下。我在這兒簡單講一下我個人的認(rèn)識：大尾結(jié)構(gòu)便于人類理解，小尾結(jié)構(gòu)便于計算機計算。

? ? ? ? 大尾結(jié)構(gòu)便于人類理解，這點我們在上面的圖中已經(jīng)發(fā)現(xiàn)了?，F(xiàn)行的網(wǎng)絡(luò)傳輸協(xié)議也是采用大尾結(jié)構(gòu)，很明顯作為協(xié)議，其重點是協(xié)議的可被理解性，而非其參與計算的能力。

? ? ? ? 小尾結(jié)構(gòu)便于計算機計算怎么理解呢？舉個例子，比如我們要對上面數(shù)據(jù)（假設(shè)為a）執(zhí)行加法操作，操作數(shù)是0x12efcdab。那么計算機取到a的地址是0xFF000000，然后用操作數(shù)中的0xab與0xFF000000地址的數(shù)據(jù)進行add操作，操作結(jié)果還保存在0xFF000000中；如果有進位，則參與到操作數(shù)0xcd與0xFF000001地址的數(shù)據(jù)相加中。CPU只要對操作數(shù)和被操作數(shù)的地址向后移動取值相加就可以了。我們再想像下大尾數(shù)據(jù)的處理方法，如果也是從地址低位開始計算——即是數(shù)據(jù)高位，則可能產(chǎn)生回溯的問題——數(shù)據(jù)低位計算有進位則要求改之前計算的值——甚至還要改之前的之前計算的值。如果從地址高位開始計算——即數(shù)據(jù)低位，則有一次通過a地址（地址低位）跳轉(zhuǎn)到地址高位的過程?？赡苡腥藭f你為什么不拿減法例子呢？人類的減法操作都是從高位向低位進行的。但是計算機沒有減法器——它是通過加法操作進行減法運算的。

? ? ? ? 大小尾雖然是一個非常古老的問題，但是我們在進行數(shù)據(jù)跨網(wǎng)絡(luò)交互時要考慮，因為網(wǎng)絡(luò)字節(jié)序和本機字節(jié)序可能不一樣?？缯Z言傳輸數(shù)據(jù)時也要考慮，像Java的數(shù)據(jù)就是大尾結(jié)構(gòu)的。

? ? ? ? Redis在源碼的endianconv.c提供了一系列小尾結(jié)構(gòu)向大尾結(jié)構(gòu)轉(zhuǎn)換的方法。我們看一下64位數(shù)據(jù)的處理函數(shù)：

/* Toggle the 64 bit unsigned integer pointed by *p from little endian to* big endian */
void memrev64(void *p) {unsigned char *x = p, t;t = x[0];x[0] = x[7];x[7] = t;t = x[1];x[1] = x[6];x[6] = t;t = x[2];x[2] = x[5];x[5] = t;t = x[3];x[3] = x[4];x[4] = t;
}uint64_t intrev64(uint64_t v) {memrev64(&v);return v;
}

? ? ? ? Redis使用大尾也印證了我之前的觀點，因為Redis是重要的功能是存儲和網(wǎng)絡(luò)交互，而非進行數(shù)值計算。

? ? ? ? 接下來我們看看Redis的有序整數(shù)集的保存結(jié)構(gòu)。

基礎(chǔ)結(jié)構(gòu)

typedef struct intset {uint32_t encoding;uint32_t length;int8_t contents[];
} intset;

? ? ? ? encoding是表示該結(jié)構(gòu)保存的數(shù)據(jù)類型。它可以是下列類型值：

#define INTSET_ENC_INT16 (sizeof(int16_t))
#define INTSET_ENC_INT32 (sizeof(int32_t))
#define INTSET_ENC_INT64 (sizeof(int64_t))

? ? ? ? 至于結(jié)構(gòu)中encoding最終設(shè)置為何種類型，則要視其存儲的最大數(shù)據(jù)的類型來決定。比如該結(jié)構(gòu)最初始的內(nèi)容只有0x08，則類型是INTSET_ENC_INT16。而如果加入了0x12345678，則類型將變成INTSET_ENC_INT32。具體的判斷方法是看數(shù)值的范圍：

static uint8_t _intsetValueEncoding(int64_t v) {if (v < INT32_MIN || v > INT32_MAX)return INTSET_ENC_INT64;else if (v < INT16_MIN || v > INT16_MAX)return INTSET_ENC_INT32;elsereturn INTSET_ENC_INT16;
}

? ? ? 不要被上面的區(qū)間名給欺騙了，它們的定義如下。

#define INT8_MIN     ((int8_t)_I8_MIN)
#define INT8_MAX     _I8_MAX
#define INT16_MIN    ((int16_t)_I16_MIN)
#define INT16_MAX    _I16_MAX
#define INT32_MIN    ((int32_t)_I32_MIN)
#define INT32_MAX    _I32_MAX

? ? ? ??_IXX_MIN都是負(fù)數(shù)，其最高位為1，這樣使用無符號類型強轉(zhuǎn)之后便是更高區(qū)間的最大值。? ? ? ??

? ? ? ? intset結(jié)構(gòu)中l(wèi)ength字段表示contents數(shù)組元素的個數(shù)；contents則是整型數(shù)數(shù)組的首地址，但是不要被它類型int8_t欺騙了，它實際存儲的類型可能是int32_t或者int64_t。

創(chuàng)建集合

? ? ? ? 集合創(chuàng)建通過下面方法實現(xiàn)

intset *intsetNew(void) {intset *is = zmalloc(sizeof(intset));is->encoding = intrev32ifbe(INTSET_ENC_INT16);is->length = 0;return is;
}

? ? ? ? 可見集合結(jié)構(gòu)是在堆上分配的，初始類型是INTSET_ENC_INT16，元素個數(shù)是0。此時contents指針還是無效的，這說明該結(jié)構(gòu)可能沒有采用預(yù)分配空間的設(shè)計，而是實時分配。之后的代碼也印證了這點。

重分配集合空間

? ? ? ? 因為inset結(jié)構(gòu)是個可變長度結(jié)構(gòu)，其可變部分就是contents數(shù)組的長度，所以重分配集合空間主要是根據(jù)集合保存的數(shù)據(jù)類型和數(shù)組元素個數(shù)重新分配空間。

static intset *intsetResize(intset *is, uint32_t len) {uint32_t size = len*intrev32ifbe(is->encoding);is = zrealloc(is,sizeof(intset)+size);return is;
}

獲取集合長度

? ? ? ? 即返回intset的length字段，它表示其保存的數(shù)字個數(shù)

/* Return intset length */
uint32_t intsetLen(intset *is) {return intrev32ifbe(is->length);
}

獲取集合占用空間大小

? ? ? ? 集合結(jié)構(gòu)的設(shè)計說明其是一個可變長結(jié)構(gòu)，所以計算空間大小要把結(jié)構(gòu)的頭大小和可變長度數(shù)組長度相加

/* Return intset blob size in bytes. */
size_t intsetBlobLen(intset *is) {return sizeof(intset)+intrev32ifbe(is->length)*intrev32ifbe(is->encoding);
}

通過位置設(shè)置值

? ? ? ? 因為contents保存的數(shù)值長度要視intset的encoding類型決定，所以通過位置定位元素時，需要將contents強轉(zhuǎn)為相應(yīng)類型的指針。這樣通過加法操作，可以讓指針步進的長度為元素類型的長度

static void _intsetSet(intset *is, int pos, int64_t value) {uint32_t encoding = intrev32ifbe(is->encoding);if (encoding == INTSET_ENC_INT64) {((int64_t*)is->contents)[pos] = value;memrev64ifbe(((int64_t*)is->contents)+pos);} else if (encoding == INTSET_ENC_INT32) {((int32_t*)is->contents)[pos] = value;memrev32ifbe(((int32_t*)is->contents)+pos);} else {((int16_t*)is->contents)[pos] = value;memrev16ifbe(((int16_t*)is->contents)+pos);}
}

通過位置獲取值

? ? ? ? 獲取值時同樣要根據(jù)intset保存的數(shù)據(jù)類型決定對contents進行加法操作時步進的長度

/* Return the value at pos, given an encoding. */
static int64_t _intsetGetEncoded(intset *is, int pos, uint8_t enc) {int64_t v64;int32_t v32;int16_t v16;if (enc == INTSET_ENC_INT64) {memcpy(&v64,((int64_t*)is->contents)+pos,sizeof(v64));memrev64ifbe(&v64);return v64;} else if (enc == INTSET_ENC_INT32) {memcpy(&v32,((int32_t*)is->contents)+pos,sizeof(v32));memrev32ifbe(&v32);return v32;} else {memcpy(&v16,((int16_t*)is->contents)+pos,sizeof(v16));memrev16ifbe(&v16);return v16;}
}/* Return the value at pos, using the configured encoding. */
static int64_t _intsetGet(intset *is, int pos) {return _intsetGetEncoded(is,pos,intrev32ifbe(is->encoding));
}

查找元素

? ? ? ? 查找元素時，先看待查找的元素數(shù)值是否在該集合可以表達(dá)的數(shù)值空間之內(nèi)。如果不在則直接認(rèn)為找不到元素，這樣可以免去查找操作

static uint8_t intsetSearch(intset *is, int64_t value, uint32_t *pos) {int min = 0, max = intrev32ifbe(is->length)-1, mid = -1;int64_t cur = -1;/* The value can never be found when the set is empty */if (intrev32ifbe(is->length) == 0) {if (pos) *pos = 0;return 0;} 

? ? ? ? 由于intset保存的是有序數(shù)字，且數(shù)字從小到大排列。這樣如果元素數(shù)值比第一個元素小，或者比最后一個元素大，則說明待查元素也不在數(shù)組中。

      else {/* Check for the case where we know we cannot find the value,* but do know the insert position. */if (value > _intsetGet(is,intrev32ifbe(is->length)-1)) {if (pos) *pos = intrev32ifbe(is->length);return 0;} else if (value < _intsetGet(is,0)) {if (pos) *pos = 0;return 0;}}

? ? ? ? 其他情況則說明待查元素，這個時候就采用二分查找的方式進行

    while(max >= min) {mid = ((unsigned int)min + (unsigned int)max) >> 1;cur = _intsetGet(is,mid);if (value > cur) {min = mid+1;} else if (value < cur) {max = mid-1;} else {break;}}if (value == cur) {if (pos) *pos = mid;return 1;} else {if (pos) *pos = min;return 0;}
}

檢測元素是否在集合中

? ? ? ? 檢測操作非常簡單，只是簡單的調(diào)用intsetSearch方法

/* Determine whether a value belongs to this set */
uint8_t intsetFind(intset *is, int64_t value) {uint8_t valenc = _intsetValueEncoding(value);return valenc <= intrev32ifbe(is->encoding) && intsetSearch(is,value,NULL);
}

新增一個更大數(shù)值類型元素

? ? ? ? 如果一開始時，集合中保存的元素只有0x01，那么集合的類型是INTSET_ENC_INT16。contents數(shù)組的長度也是INTSET_ENC_INT16的長度?，F(xiàn)在要往集合中新增一個元素0x12345678，這個時候INTSET_ENC_INT16類型長度的空間已經(jīng)不能保存該數(shù)據(jù)了。于是需要對整個結(jié)構(gòu)進行升級

static intset *intsetUpgradeAndAdd(intset *is, int64_t value) {uint8_t curenc = intrev32ifbe(is->encoding);uint8_t newenc = _intsetValueEncoding(value);int length = intrev32ifbe(is->length);

? ? ? ? 先要獲取以前集合類型和長度，還要計算新集合為何種類型

    int prepend = value < 0 ? 1 : 0;

? ? ? ? 然后檢查新增的值是否為負(fù)數(shù)。因為該數(shù)值的絕對值比之前數(shù)組中所有元素都要大，所以如果該數(shù)如果是負(fù)數(shù)，則它比之前任何元素都小，這樣它就要插在頭部。相反，如果它是正數(shù)，則可能是插在尾部。

    /* First set new encoding and resize */is->encoding = intrev32ifbe(newenc);is = intsetResize(is,intrev32ifbe(is->length)+1);/* Upgrade back-to-front so we don't overwrite values.* Note that the "prepend" variable is used to make sure we have an empty* space at either the beginning or the end of the intset. */while(length--)_intsetSet(is,length+prepend,_intsetGetEncoded(is,length,curenc));

? ? ? ? 使用新的類型更新集合encoding字段，再通過intsetResize重新分配集合的內(nèi)存。因為當(dāng)前內(nèi)存數(shù)據(jù)分布和之前的一致（除了變長了），所以還要通過之后的while循環(huán)將之前的值轉(zhuǎn)移到其現(xiàn)在應(yīng)該在的位置。

    /* Set the value at the beginning or the end. */if (prepend)_intsetSet(is,0,value);else_intsetSet(is,intrev32ifbe(is->length),value);is->length = intrev32ifbe(intrev32ifbe(is->length)+1);return is;
}

? ? ? ? 最后視新增數(shù)據(jù)的正負(fù)情況插入到新結(jié)構(gòu)的不同位置。

數(shù)組尾部空間平移

? ? ? ? 這步操作在要往數(shù)組中間插入或者刪除元素時發(fā)生。如果插入元素，則需要將插入位置的元素及之后的元素一起向后平移。如果刪除元素，則要將被刪除元素之后的元素向前平移

static void intsetMoveTail(intset *is, uint32_t from, uint32_t to) {void *src, *dst;uint32_t bytes = intrev32ifbe(is->length)-from;uint32_t encoding = intrev32ifbe(is->encoding);if (encoding == INTSET_ENC_INT64) {src = (int64_t*)is->contents+from;dst = (int64_t*)is->contents+to;bytes *= sizeof(int64_t);} else if (encoding == INTSET_ENC_INT32) {src = (int32_t*)is->contents+from;dst = (int32_t*)is->contents+to;bytes *= sizeof(int32_t);} else {src = (int16_t*)is->contents+from;dst = (int16_t*)is->contents+to;bytes *= sizeof(int16_t);}memmove(dst,src,bytes);
}

增加元素

? ? ? ? 增加元素時，要先判斷待添加的元素是否比現(xiàn)在的集合類型大。如果是，則要重新分配和更新整個集合內(nèi)存空間

intset *intsetAdd(intset *is, int64_t value, uint8_t *success) {uint8_t valenc = _intsetValueEncoding(value);uint32_t pos;if (success) *success = 1;/* Upgrade encoding if necessary. If we need to upgrade, we know that* this value should be either appended (if > 0) or prepended (if < 0),* because it lies outside the range of existing values. */if (valenc > intrev32ifbe(is->encoding)) {/* This always succeeds, so we don't need to curry *success. */return intsetUpgradeAndAdd(is,value);}

? ? ? ? 如果之前集合的類型可以承載待添加的元素，則先去檢查元素是否已經(jīng)在數(shù)組中。如果已經(jīng)存在，則不再進行添加操作，直接認(rèn)為操作成功

      else {/* Abort if the value is already present in the set.* This call will populate "pos" with the right position to insert* the value when it cannot be found. */if (intsetSearch(is,value,&pos)) {if (success) *success = 0;return is;}

? ? ? ? 如果不在數(shù)組中，則上面的intsetSearch方法將計算出待添加的數(shù)據(jù)需要被插入到數(shù)組中的的位置。這個時候就需要重新分配集合長度，并將要插入的位置及之后的數(shù)據(jù)向后平移，并把待添加數(shù)據(jù)設(shè)置到數(shù)組的相應(yīng)位置。

        is = intsetResize(is,intrev32ifbe(is->length)+1);if (pos < intrev32ifbe(is->length)) intsetMoveTail(is,pos,pos+1);}_intsetSet(is,pos,value);is->length = intrev32ifbe(intrev32ifbe(is->length)+1);return is;
}

刪除元素

? ? ? ? 刪除元素比較簡單，只要通過intsetSearch找到元素的位置，將該位置之后的元素向前平移就行了

/* Delete integer from intset */
intset *intsetRemove(intset *is, int64_t value, int *success) {uint8_t valenc = _intsetValueEncoding(value);uint32_t pos;if (success) *success = 0;if (valenc <= intrev32ifbe(is->encoding) && intsetSearch(is,value,&pos)) {uint32_t len = intrev32ifbe(is->length);/* We know we can delete */if (success) *success = 1;/* Overwrite value with tail and update length */if (pos < (len-1)) intsetMoveTail(is,pos+1,pos);is = intsetResize(is,len-1);is->length = intrev32ifbe(len-1);}return is;
}

? ? ? ? 通過上面的函數(shù)，我們發(fā)現(xiàn)Redis的整數(shù)集在增刪改元素時要自動調(diào)整元素排序。在新增絕對值超過當(dāng)前集合可以表達(dá)的數(shù)據(jù)時，升級當(dāng)前集合。但是如果刪除元素時，即使現(xiàn)存的數(shù)字都比當(dāng)前集合表達(dá)的區(qū)間的最小值還要小，也不會發(fā)生降級的操作。

總結(jié)

以上是生活随笔為你收集整理的Redis源码解析——有序整数集的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。