1. 程式人生 > 其它 >Redis使用及原始碼剖析-6.Redis整數集合-2021-1-20

Redis使用及原始碼剖析-6.Redis整數集合-2021-1-20

技術標籤:redisredis資料庫

文章目錄


前言

整數集合(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_INT32 , 那麼 contents 就是一個 int32_t 型別的陣列, 數組裡的每個項都是一個 int32_t 型別的整數值 (最小值為 -2,147,483,648 ,最大值為 2,147,483,647 )。
如果 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設計的特別優雅,特別是其中的升級和元素移動的程式碼,值得好好觀摩學習。