Redis使用及原始碼剖析-6.Redis整數集合-2021-1-20
文章目錄
前言
整數集合(intset)是集合鍵的底層實現之一: 當一個集合只包含整數值元素, 並且這個集合的元素數量不多時, Redis 就會使用整數集合作為集合鍵的底層實現。整數集合涉及的檔案是intset.h和intset.c
一、整數集合實現
整數集合(intset)是 Redis 用於儲存整數值的集合抽象資料結構, 它可以儲存型別為 int16_t 、 int32_t 或者 int64_t 的整數值, 並且保證集合中不會出現重複元素。實現程式碼如下:
typedef struct intset {
// 編碼方式
uint32_t encoding;
// 集合包含的元素數量
uint32_t length;
// 儲存元素的陣列
int8_t contents[];
} intset;
contents 陣列是整數集合的底層實現: 整數集合的每個元素都是 contents 陣列的一個數組項(item), 各個項在陣列中按值的大小從小到大有序地排列, 並且陣列中不包含任何重複項。
length 屬性記錄了整數集合包含的元素數量, 也即是 contents 陣列的長度。
雖然 intset 結構將 contents 屬性宣告為 int8_t 型別的陣列, 但實際上 contents 陣列並不儲存任何 int8_t 型別的值 —— contents 陣列的真正型別取決於 encoding 屬性的值,共有下面三種:
/* Note that these encodings are ordered, so:
* INTSET_ENC_INT16 < INTSET_ENC_INT32 < INTSET_ENC_INT64. */
/*
* intset 的編碼方式
*/
#define INTSET_ENC_INT16 (sizeof(int16_t))
#define INTSET_ENC_INT32 (sizeof(int32_t))
#define INTSET_ENC_INT64 (sizeof(int64_t))
如果 encoding 屬性的值為 INTSET_ENC_INT16 , 那麼 contents 就是一個 int16_t 型別的陣列, 數組裡的每個項都是一個 int16_t 型別的整數值 (最小值為 -32,768 ,最大值為 32,767 )。
如果 encoding 屬性的值為 INTSET_ENC_INT64 , 那麼 contents 就是一個 int64_t 型別的陣列, 數組裡的每個項都是一個 int64_t 型別的整數值 (最小值為 -9,223,372,036,854,775,808 ,最大值為 9,223,372,036,854,775,807 )。
陣列的encoding 屬性是取數組裡每一個元素編碼方式的最大值,如下圖所示:
雖然 contents 陣列儲存的四個整數值中, 只有 -2675256175807981027 是真正需要用 int64_t 型別來儲存的, 而其他的 1 、 3 、 5 三個值都可以用 int16_t 型別來儲存, 不過根據整數集合的規則, 編碼方式仍然是按照 int64_t型別儲存。
二、intset api
1.根據值確認編碼方式
根據值確定編碼方式的函式如下所示:
/* Return the required encoding for the provided value.
*
* 返回適用於傳入值 v 的編碼方式
*
* T = O(1)
*/
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;
else
return INTSET_ENC_INT16;
}
2.根據值確認編碼方式
根據值確定編碼方式的函式如下所示:
/* Return the required encoding for the provided value.
*
* 返回適用於傳入值 v 的編碼方式
*
* T = O(1)
*/
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;
else
return INTSET_ENC_INT16;
}
3.獲取指定索引的元素
根據給定的編碼方式 enc ,返回集合的底層陣列在 pos 索引上的元素:
static int64_t _intsetGetEncoded(intset *is, int pos, uint8_t enc) {
int64_t v64;
int32_t v32;
int16_t v16;
// ((ENCODING*)is->contents) 首先將陣列轉換回被編碼的型別
// 然後 ((ENCODING*)is->contents)+pos 計算出元素在陣列中的正確位置
// 之後 member(&vEnc, ..., sizeof(vEnc)) 再從陣列中拷貝出正確數量的位元組
// 如果有需要的話, memrevEncifbe(&vEnc) 會對拷貝出的位元組進行大小端轉換
// 最後將值返回
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;
}
}
4.設定指定索引的元素
根據集合的編碼方式,將底層陣列在 pos 位置上的值設為 value,程式碼如下:
static void _intsetSet(intset *is, int pos, int64_t value) {
// 取出集合的編碼方式
uint32_t encoding = intrev32ifbe(is->encoding);
// 根據編碼 ((Enc_t*)is->contents) 將陣列轉換回正確的型別
// 然後 ((Enc_t*)is->contents)[pos] 定位到陣列索引上
// 接著 ((Enc_t*)is->contents)[pos] = value 將值賦給陣列
// 最後, ((Enc_t*)is->contents)+pos 定位到剛剛設定的新值上
// 如果有需要的話, memrevEncifbe 將對值進行大小端轉換
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);
}
}
5.在集合中查詢指定元素的位置
在集合 is 的底層陣列中查詢值 value 所在的索引,成功找到 value 時,函式返回 1 ,並將 *pos 的值設為 value 所在的索引。當在陣列中沒找到 value 時,返回 0 。並將 *pos 的值設為 value 可以插入到陣列中的位置,程式碼如下:
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 */
// 處理 is 為空時的情況
if (intrev32ifbe(is->length) == 0) {
if (pos) *pos = 0;
return 0;
} else {
/* Check for the case where we know we cannot find the value,
* but do know the insert position. */
// 因為底層陣列是有序的,如果 value 比陣列中最後一個值都要大
// 那麼 value 肯定不存在於集合中,
// 並且應該將 value 新增到底層陣列的最末端
if (value > _intsetGet(is,intrev32ifbe(is->length)-1)) {
if (pos) *pos = intrev32ifbe(is->length);
return 0;
// 因為底層陣列是有序的,如果 value 比陣列中最前一個值都要小
// 那麼 value 肯定不存在於集合中,
// 並且應該將它新增到底層陣列的最前端
} else if (value < _intsetGet(is,0)) {
if (pos) *pos = 0;
return 0;
}
}
// 在有序陣列中進行二分查詢
// T = O(log N)
while(max >= min) {
mid = (min+max)/2;
cur = _intsetGet(is,mid);
if (value > cur) {
min = mid+1;
} else if (value < cur) {
max = mid-1;
} else {
break;
}
}
// 檢查是否已經找到了 value
if (value == cur) {
if (pos) *pos = mid;
return 1;
} else {
if (pos) *pos = min;
return 0;
}
}
6.集合編碼方式升級
集合中升級編碼方式程式碼如下:
/* Upgrades the intset to a larger encoding and inserts the given integer.
*
* 根據值 value 所使用的編碼方式,對整數集合的編碼進行升級,
* 並將值 value 新增到升級後的整數集合中。
*
* 返回值:新增新元素之後的整數集合
*
* T = O(N)
*/
static intset *intsetUpgradeAndAdd(intset *is, int64_t value) {
// 當前的編碼方式
uint8_t curenc = intrev32ifbe(is->encoding);
// 新值所需的編碼方式
uint8_t newenc = _intsetValueEncoding(value);
// 當前集合的元素數量
int length = intrev32ifbe(is->length);
// 根據 value 的值,決定是將它新增到底層陣列的最前端還是最後端
// 注意,因為 value 的編碼比集合原有的其他元素的編碼都要大
// 所以 value 要麼大於集合中的所有元素,要麼小於集合中的所有元素
// 因此,value 只能新增到底層陣列的最前端或最後端
int prepend = value < 0 ? 1 : 0;
/* First set new encoding and resize */
// 更新集合的編碼方式
is->encoding = intrev32ifbe(newenc);
// 根據新編碼對集合(的底層陣列)進行空間調整
// T = O(N)
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. */
// 根據集合原來的編碼方式,從底層陣列中取出集合元素
// 然後再將元素以新編碼的方式新增到集合中
// 當完成了這個步驟之後,集合中所有原有的元素就完成了從舊編碼到新編碼的轉換
// 因為新分配的空間都放在陣列的後端,所以程式先從後端向前端移動元素
// 舉個例子,假設原來有 curenc 編碼的三個元素,它們在陣列中排列如下:
// | x | y | z |
// 當程式對陣列進行重分配之後,陣列就被擴容了(符號 ? 表示未使用的記憶體):
// | x | y | z | ? | ? | ? |
// 這時程式從陣列後端開始,重新插入元素:
// | x | y | z | ? | z | ? |
// | x | y | y | z | ? |
// | x | y | z | ? |
// 最後,程式可以將新元素新增到最後 ? 號標示的位置中:
// | x | y | z | new |
// 上面演示的是新元素比原來的所有元素都大的情況,也即是 prepend == 0
// 當新元素比原來的所有元素都小時(prepend == 1),調整的過程如下:
// | x | y | z | ? | ? | ? |
// | x | y | z | ? | ? | z |
// | x | y | z | ? | y | z |
// | x | y | x | y | z |
// 當新增新值時,原本的 | x | y | 的資料將被新值代替
// | new | x | y | z |
// T = O(N)
while(length--)
_intsetSet(is,length+prepend,_intsetGetEncoded(is,length,curenc));
/* Set the value at the beginning or the end. */
// 設定新值,根據 prepend 的值來決定是新增到陣列頭還是陣列尾
if (prepend)
_intsetSet(is,0,value);
else
_intsetSet(is,intrev32ifbe(is->length),value);
// 更新整數集合的元素數量
is->length = intrev32ifbe(intrev32ifbe(is->length)+1);
return is;
}
7.集合元素移動
集合元素移動程式碼如下:
/*
* 向前或先後移動指定索引範圍內的陣列元素
*
* 函式名中的 MoveTail 其實是一個有誤導性的名字,
* 這個函式可以向前或向後移動元素,
* 而不僅僅是向後
*
* 在新增新元素到陣列時,就需要進行向後移動,
* 如果陣列表示如下(?表示一個未設定新值的空間):
* | x | y | z | ? |
* |<----->|
* 而新元素 n 的 pos 為 1 ,那麼陣列將移動 y 和 z 兩個元素
* | x | y | y | z |
* |<----->|
* 接著就可以將新元素 n 設定到 pos 上了:
* | x | n | y | z |
*
* 當從陣列中刪除元素時,就需要進行向前移動,
* 如果陣列表示如下,並且 b 為要刪除的目標:
* | a | b | c | d |
* |<----->|
* 那麼程式就會移動 b 後的所有元素向前一個元素的位置,
* 從而覆蓋 b 的資料:
* | a | c | d | d |
* |<----->|
* 最後,程式再從陣列末尾刪除一個元素的空間:
* | a | c | d |
* 這樣就完成了刪除操作。
*
* T = O(N)
*/
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);
// 根據不同的編碼
// src = (Enc_t*)is->contents+from 記錄移動開始的位置
// dst = (Enc_t*)is_.contents+to 記錄移動結束的位置
// bytes *= sizeof(Enc_t) 計算一共要移動多少位元組
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);
}
// 進行移動
// T = O(N)
memmove(dst,src,bytes);
}
8.集合元素增加刪除
集合元素增加和刪除的程式碼如下:
/* Insert an integer in the intset
*
* 嘗試將元素 value 新增到整數集合中。
*
* *success 的值指示新增是否成功:
* - 如果新增成功,那麼將 *success 的值設為 1 。
* - 因為元素已存在而造成新增失敗時,將 *success 的值設為 0 。
*
* T = O(N)
*/
intset *intsetAdd(intset *is, int64_t value, uint8_t *success) {
// 計算編碼 value 所需的長度
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. */
// 如果 value 的編碼比整數集合現在的編碼要大
// 那麼表示 value 必然可以新增到整數集合中
// 並且整數集合需要對自身進行升級,才能滿足 value 所需的編碼
if (valenc > intrev32ifbe(is->encoding)) {
/* This always succeeds, so we don't need to curry *success. */
// T = O(N)
return intsetUpgradeAndAdd(is,value);
} else {
// 執行到這裡,表示整數集合現有的編碼方式適用於 value
/* 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. */
// 在整數集合中查詢 value ,看他是否存在:
// - 如果存在,那麼將 *success 設定為 0 ,並返回未經改動的整數集合
// - 如果不存在,那麼可以插入 value 的位置將被儲存到 pos 指標中
// 等待後續程式使用
if (intsetSearch(is,value,&pos)) {
if (success) *success = 0;
return is;
}
// 執行到這裡,表示 value 不存在於集合中
// 程式需要將 value 新增到整數集合中
// 為 value 在集合中分配空間
is = intsetResize(is,intrev32ifbe(is->length)+1);
// 如果新元素不是被新增到底層陣列的末尾
// 那麼需要對現有元素的資料進行移動,空出 pos 上的位置,用於設定新值
// 舉個例子
// 如果陣列為:
// | x | y | z | ? |
// |<----->|
// 而新元素 n 的 pos 為 1 ,那麼陣列將移動 y 和 z 兩個元素
// | x | y | y | z |
// |<----->|
// 這樣就可以將新元素設定到 pos 上了:
// | x | n | y | z |
// T = O(N)
if (pos < intrev32ifbe(is->length)) intsetMoveTail(is,pos,pos+1);
}
// 將新值設定到底層陣列的指定位置中
_intsetSet(is,pos,value);
// 增一集合元素數量的計數器
is->length = intrev32ifbe(intrev32ifbe(is->length)+1);
// 返回新增新元素後的整數集合
return is;
/* p.s. 上面的程式碼可以重構成以下更簡單的形式:
if (valenc > intrev32ifbe(is->encoding)) {
return intsetUpgradeAndAdd(is,value);
}
if (intsetSearch(is,value,&pos)) {
if (success) *success = 0;
return is;
} else {
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;
}
*/
}
/* Delete integer from intset
*
* 從整數集合中刪除值 value 。
*
* *success 的值指示刪除是否成功:
* - 因值不存在而造成刪除失敗時該值為 0 。
* - 刪除成功時該值為 1 。
*
* T = O(N)
*/
intset *intsetRemove(intset *is, int64_t value, int *success) {
// 計算 value 的編碼方式
uint8_t valenc = _intsetValueEncoding(value);
uint32_t pos;
// 預設設定標識值為刪除失敗
if (success) *success = 0;
// 當 value 的編碼大小小於或等於集合的當前編碼方式(說明 value 有可能存在於集合)
// 並且 intsetSearch 的結果為真,那麼執行刪除
// T = O(log N)
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 */
// 如果 value 不是位於陣列的末尾
// 那麼需要對原本位於 value 之後的元素進行移動
//
// 舉個例子,如果陣列表示如下,而 b 為刪除的目標
// | a | b | c | d |
// 那麼 intsetMoveTail 將 b 之後的所有資料向前移動一個元素的空間,
// 覆蓋 b 原來的資料
// | a | c | d | d |
// 之後 intsetResize 縮小記憶體大小時,
// 陣列末尾多出來的一個元素的空間將被移除
// | a | c | d |
if (pos < (len-1)) intsetMoveTail(is,pos+1,pos);
// 縮小陣列的大小,移除被刪除元素佔用的空間
// T = O(N)
is = intsetResize(is,len-1);
// 更新集合的元素數量
is->length = intrev32ifbe(len-1);
}
return is;
}
總結
整數集合的API設計的特別優雅,特別是其中的升級和元素移動的程式碼,值得好好觀摩學習。