高性能對象池實現

作者：jaeyang，騰訊 WXG 後臺研發工程師

| 導語內存池用於對頻繁申請的內存進行管理進而提升分配效率，但缺乏對一些創建和銷燬開銷比較大的對象的複用手段，因此對象池應運而生。而當系統中存在大量對象需要頻繁創建和銷燬時，如何減少大量的耗時開銷是對象池構建的關鍵點之一，本文以此出發，與大家共同探討高性能對象池的實現。文章作者：楊哲，騰訊 WXG 後臺研發工程師。

一、背景

內存池用於對頻繁申請的內存進行管理，通過合理的分配策略和內存佈局來減少內存的碎片化以及提高內存的分配效率。但是對於一些創建和銷燬開銷大的對象，內存池缺乏對這些對象進行復用的手段，因此出現了對象池。

從內存分配的角度來看，相對於內存池，對象池管理的是定長內存，所以無需考慮內存碎片的問題，在內存管理策略上也更加的簡單。我們的系統中存在的大量對象需要頻繁地創建和銷燬，產生了大量的耗時開銷，因此需要對象池提供對象複用的方式來避免構造析構產生的開銷，或者是通過對象的重置來減少創建銷燬對象的開銷。

另一方面相對於目前的內存分配器，對象內存的管理更加簡單，因此相對於現有的內存分配器在內存分配和釋放的效率有一定的提升空間。

二、目標

對象可複用：通過複用對象來避免頻繁地調用 malloc 和 free 函數，或者是減少構造析構產生的開銷，從而提升性能；
高性能：高性能是設計這個對象池最主要的目標，這裏的高性能指的是內存分配和釋放的開銷足夠低；
線程安全：對象池可能會被多個線程同時訪問，因此要保證對象池的線程安全；
對象的容量支持動態擴展；
優先分配使用過的對象。

三、方案調研

1. 對象池

（1）brpc object pool

brpc object pool 通過批量分配和歸還內存以避免全局競爭，從而降低單次分配、釋放的開銷。brpc object pool 每個線程的分配流程如下：

查看 thread-local free block 或者空閒對象數組。如果還有 free 的對象，返回。沒有的話步驟 2。
嘗試從全局資源池取一塊空閒的空間，若取到的話回到步驟 1，否則步驟 3。
從全局資源池從系統申請一大塊內存，返回其中第一個對象。

釋放流程爲將對象回收到 thread-local 的空閒對象數組中，攢夠數量後回收到全局資源池。

（2）go 對象池

Pool 會爲每個協程維護一個本地池，本地池分爲私有池 private 和共享池 shared。私有池中的元素只能本地協程使用，共享池中的元素可能會被其他協程偷走，所以使用私有池 private 時不用加鎖，而使用共享池 shared 時需加鎖。

通過對象池獲取對象時會優先查找本地 private 池，再查找本地 shared 池，最後查找其他協程的 shared 池，如果以上全部沒有可用元素，最後會調用 New 函數獲取新元素。詳細的分配過程如下圖所示：

回收對象時優先把元素放在 private 池中。如果 private 不爲空，則放在 shared 池中。

（3）Netty recycler

每個線程都擁有 thread-local 的 Stack, 在 Stack 中維護對象數組以及 WeakOrderQueue 的相關指針。對於全局資源分配機制，當本線程 thread1 回收本線程產生的對象時, 會將對象以 DefaultHandle 的形式存放在 Stack 中。其它線程 thread2 也可以回收 thread1 產生的對象，thread2 回收的對象不會立即放回 thread1 的 Stack 中，而是保存在 thread2 內部的一個 WeakOrderQueue 中。這些外部線程的 WeakOrderQueue 以鏈表的方式和 Stack 關聯起來。

默認情況下一個線程最多持有 2 * 核數個 WeakOrderQueue，也就是說一個線程最多可以幫 2 * 核數個外部線程的對象池回收對象。WeakOrderQueue 內部有以 Link 來管理對象。每個 Link 存放的對象是有限的，一個 Link 最多存放 16 個對象。如果滿了則會再產生一個 Link 繼續存放。

當前線程從對象池中拿對象時, 首先從 Stack 中獲取，若沒有的話，將嘗試從 cursor 指向的 WeakOrderQueue 中回收一個 Link 的對象,。如果回收到了就繼續從 Stack 中獲取對象；如果沒有回收到就創建對象。一個對象池中最多存放 4K 個對象 , Link 節點中每個 DefaultHandle 數組默認長度 16，這兩個參數可以控制。

2. 內存池

雖然內存池使用的場景和對象池有區別，除了分配的速度外內存池還需要考慮內存碎片的問題，但是內存池在應對多線程訪問時的減少鎖競爭思路是可以借鑑的。

（1）tcmalloc

在 tcmalloc 中每個線程都有一個線程局部的 ThreadCache，按照對象的大小進行分類維護對象的鏈表。如果 ThreadCache 的對象不夠了，就從 CentralCache 進行批量分配。如果 CentralCache 沒有可分配的對象，就從 PageHeap 申請 Span。如果 PageHeap 沒有合適的 Page，就從操作系統申請。

在釋放內存的時候，ThreadCache 依然遵循批量釋放的策略，對象積累到一定程度就釋放給 CentralCache。CentralCache 發現一個 Span 的內存完全釋放了，就可以把這個 Span 歸還給 PageHeap；PageHeap 發現一批連續的 Page 都釋放了，就可以歸還給操作系統。

（2）jemalloc

虛擬內存被邏輯上分割成 chunks（默認是 4MB，1024 個 4k 頁），訪問線程通過 round-robin 算法在第一次 malloc 的時候分配 arena，每個 arena 都是相互獨立的，維護自己的 chunks， chunk 切割 pages 到 small/large 對象。free() 的內存總是返回到所屬的 arena 中，而不管是哪個線程調用 free()。通過 arena 分配的時候需要對 arena bin（每個 small size-class 一個，細粒度）加鎖，或 arena 本身加鎖。並且線程 cache 對象也會通過垃圾回收指數退讓算法返回到 arena 中。

四、整體設計

從上面的對內存分配系統的調研來看，在應對多線程訪問時爲了減少鎖競爭的方式大體上一致，都是通過分區減小鎖的粒度以及使用 TLS 來實現每個線程獨享的資源池來避免大部分的鎖競爭。所以本文中對象池在保存空閒對象時使用 freelist + TLS + 多資源池的組合，使用 freelist 可以節省指針部分的內存，而且在交換資源時只需對隊頭指針進行修改，速度非常快而且減少了在臨界區中的耗時，緩解了公共資源池中的鎖競爭。對象池的整體結構圖如下：

一個 Object Pool 中主要有兩個部分：

1. Local Pool

每個訪問對象池的線程都會獨自擁有一個 Local Pool，使用 TLS(Thread-Local Storage) 實現，Object Pool 中使用了一個 thread_local 指針指向一個 Local Pool，訪問對象池的線程與 Local Pool 直接交互，申請的對象直接從 Local Pool 中獲取，釋放的對象也直接歸還到 Local Pool 中。

Local Pool 維護一個 Block 指針和空閒隊列 (FreeSlots)，其中 Block 只負責分配對象，對象只會回收到空閒隊列中而不會回收到 Block 中，Local Pool 中的空閒隊列達到一定長度就會回收到 Global Pool 中。當 Local Pool 對象不足時就會從 Global Pool 中申請對象資源，Global Pool 會把空閒的鏈表或者 Block 給 Local Pool。

2. Global Pool

負責整體的內存資源申請。Global Pool 中維護了 BlockManager 和 FreeslotsManager 兩個數據結構，其中 BlockManager 用於管理 BlockChunk，每個 BlockChunk 中包含多個 Block，如果當前 BlockChunk 耗盡，那麼會 Global Pool 會 new 一個新的 BlockChunk。

FreeSlotsManager 用於管理 Global Pool 中的空閒鏈表，如果 FreeSlots 中的空閒鏈表都回收自 Local Pool，當 Local Pool 中的空閒鏈表的長度到達 kFreeslotsSize 時，就會將該空閒鏈表回收到 Global Pool 的 FreeSlotsManager 中，所以 FreeSlotsManager 中每條空閒鏈表的長度都是 kFreeSlotsSize。

Global Pool 的數量可以是多個，這個參數是可以設置的，設置多個 Global Pool 可以緩解 Global Pool 的鎖競爭問題從而減少耗時，但可能會帶來一定的內存膨脹，可以根據訪問線程個數等因素來通過合理設置 Global Pool 的數量在速度與內存之間進行平衡。

這種將對象池分離成 Local Pool 和 Global Pool 的設計有利於避免激烈的鎖競爭，只有涉及到 Global Pool 與 Local Pool 的資源交換時纔會出現鎖競爭，大部分情況下線程只和 Local Pool 進行交互就可以完成資源的申請、釋放，所以大大地提高了性能。

五、實現

下面是對象池中的數據結構：（設對象池中需要分配的對象爲 T）

1. Slot

union Slot {
  Slot *next_ = NULL;
  T val_;
};

Slot 都是一個對象的內存單位，使用 union 的原因是：Slot 有兩種狀態，一種是作爲分配出去的對象使用，另一種是未分配時作爲空閒鏈表中的一個結點。

2. Block

struct Block {
  Slot slots_[kBlockSize];
  size_t idx_ = 0;
};

Local Pool 向 Global Pool 申請資源的基本單位，每個 Local Pool 中會維護一個 Block，當空閒鏈表用光時，對象內存從 Block 中獲取。

3. BlockChunk

struct BlockChunk {
  Block blocks_[kBlockChunkSize];
  size_t idx_ = 0;
};

Global Pool 申請資源的基本單位，每個 BlockChunk 中包含了 kBlockChunkSize 個 Block。

4. FreeSlots

struct FreeSlots {
  Slot *head_ = NULL;
  size_t length_ = 0;
};

空閒鏈表，這裏維護了空閒鏈表的隊頭指針，以及目前鏈表的長度。

5. BlockManager

struct BlockManager {
  std::vector<BlockChunk*> block_chunks_;
};

Global Pool 中管理 BlockChunk 的數據結構，便於最後釋放整個對象池的資源。

6. FreeSlotsManager

struct FreeSlotsManager {
  size_t free_num_ = 0;
  std::vector<Slot*> freeslots_ptrs;
};

Global Pool 中管理空閒鏈表的數據結構，freeslots_ptrs 中的每個非空指針對應一條長度爲 kFreeSlotsSize 的空閒鏈表。

7. Local Pool

LocalPool 的數據結構定義如下：主要有 3 個成員變量，global_pool_ 用於在要和 Global Pool 進行資源交換時調用 Global Pool 的對應接口，block_ 用於維護一塊可用的 Block，作爲當 Local Pool 和 Global Pool 中所有的空閒鏈表都消耗完時的備用內存，freeslots_ 爲 Local Pool 中維護的空閒鏈表。

class GlobalPool {
    GlobalPool<T> *global_pool_;
    Block<T> *block_;
    FreeSlots<T> freeslots_;
}

（1）分配對象流程

如果 Local Pool 的 freeslots 中存在空閒對象，直接從 freeslots 中分配一個 Slot 出去；
如果 Local Pool 的 freeslots 分配完了，那麼從 Global Pool 的 FreeSlotsManager 中查看是否有可用的空閒鏈表，如果有那麼分配一條空閒鏈表到 Local Pool 的 freeslots 中，然後從 freeslots 中分配一個 Slot 出去；
如果 Global Pool 的 FreeSlotsManager 中的空閒鏈表分配完了，那麼從 Local Pool 的 block_ 分配一個 Slot 出去；
如果 Local Pool 中的 block_ 用光了，從 Global Pool 中申請一個 Block，然後從這個 Block 中分配一個 Slot 出去；
如果 Global Pool 中的 Block 用光了，new 一個新的 BlockChunk，然後將其中一個 Block 分配給 Local Pool。

T* GetObject() {
        // 如果freeslots還有可用空間
        if (freeslots_.head_ != NULL) {
            Slot<T> *res = freeslots_.head_;
            freeslots_.head_ = res->next_;
            freeslots_.length_--;
            return (T*)res;
        }
        // 如果global pool中有可用的freeslots
        else if (global_pool_->PopFreeSlots(freeslots_)) {
            Slot<T> *res = freeslots_.head_;
            freeslots_.head_ = res->next_;
            freeslots_.length_--;
            return (T*)res;
        }
        // 如果local pool的block還有可用空間
        else if (block_->idx_ < kBlockSize) {
            return (T*)&block_->slots_[block_->idx_++];
        }
        // 如果global pool還有可用的block
        else if (block_ = global_pool_->PopBlock()) {
            return (T*)&block_->slots_[block_->idx_++];
        }
        // 沒有可用的空間
        return NULL;
    }

（2）回收對象流程

所有回收的內存都是回收到 Local Pool 的 freeslots 中，回收的內存插入到空閒鏈表的頭部，如果插入後 freeslots 的長度達到 kFreeSlotsSize，那麼將這條 Local Pool 中的空閒鏈表回收到 Global Pool 中。

void ReturnObject(T *obj) {
        // 如果freeslots還剩最後一個slot的回收空間
        ((Slot<T>*)obj)->next_ = freeslots_.head_;
        freeslots_.head_ = (Slot<T>*)obj;
        freeslots_.length_++;
        // 如果freeslots_中的長度滿足條件，回收到global pool中
        if (freeslots_.length_ == kFreeSlotsSize) {
            global_pool_->PushFreeSlots(freeslots_);
        }
}

8. Global Pool

GlobalPool 的數據結構定義如下：主要有 4 個成員變量，block_manager_ 爲 BlockChunk 管理單元，freeslots_manager_ 爲空閒鏈表，其中可以維護多條空閒鏈表。由於在 Global Pool 中會存在多個線程同時進行資源交換，因此需要對 block_manager_ 和 freeslots_manager_ 進行操作時需要上鎖，爲了減少鎖的粒度分別對兩個資源管理單元使用不同的鎖，其中 freeslots_lck_ 對應 freeslots_manager_，block_mtx_ 對應 block_manager。

class GlobalPool {
    BlockManager<T> block_manager_;
    FreeSlotsManager<T> freeslots_manager_;
    pthread_spinlock_t freeslots_lck_;
    pthread_mutex_t block_mtx_;
}

（1）Global Pool 中的資源操作

從 FreeSlotsManager 中取出空閒鏈表，如果 freeslots_manager_ 存在可用的空閒鏈表，取出空閒鏈表給 Local Pool。

bool PopFreeSlots(FreeSlots<T> &freeslots) {
        pthread_spin_lock(&freeslots_lck_);
        // 如果Global Pool中有可用的空閒鏈表
        if (freeslots_manager_.free_num_ > 0) {
            freeslots.head_ = freeslots_manager_.freeslots_ptrs[--freeslots_manager_.free_num_];
            pthread_spin_unlock(&freeslots_lck_);
            // Global Pool中每條空閒鏈表的長度都爲kFreeSlotsSize
            freeslots.length_ = kFreeSlotsSize;
            return true;
        }
        pthread_spin_unlock(&freeslots_lck_);
        return false;
    }

將 LocalPool 中的空閒鏈表回收到 Global Pool。

bool PushFreeSlots(FreeSlots<T> &freeslots) {
        pthread_spin_lock(&freeslots_lck_);
        // 如果freeslots_manager中存儲的空閒鏈表的指針位置不夠用，增加1000個位置
        if (freeslots_manager_.free_num_ >= freeslots_manager_.freeslots_ptrs.size()) {
            freeslots_manager_.freeslots_ptrs.resize(freeslots_manager_.freeslots_ptrs.size() + 1000);
        }
  // 將Local Pool的空閒鏈表的隊頭指針存儲到freeslots_manager中
        freeslots_manager_.freeslots_ptrs[freeslots_manager_.free_num_++] = freeslots.head_;
        pthread_spin_unlock(freeslots_lck_);
        // 重置Local Pool中空閒鏈表的信息
        freeslots.head_ = NULL;
        freeslots.length_ = 0;
        return true;
    }

分配 Block 給 Local Pool，如果 BlockManager 中當前 BlockChunk 存在空閒的 Block，那麼直接分配給 Local Pool；如果 BlockManager 中的 Block 已經用完那麼申請一個新的 BlockChunk，並從 BlockChunk 中劃分出 Block 給 Local Pool。

// 申請空間
bool NewBlockChunk() {
    BlockChunk<T> *new_block_chunk = new (std::nothrow) BlockChunk<T>;
    if (unlikely(new_block_chunk == NULL))
        return false;
    block_manager_.block_chunks_.push_back(new_block_chunk);
    return true;
}
Block<T>* PopBlock() {
    pthread_mutex_lock(&block_mtx_);
    BlockChunk<T>* block_chunk = block_manager_.block_chunks_.back();
    // 如果當前BlockChunk已耗盡，申請一個新的BlockChunk
    if (block_chunk == NULL || block_chunk->idx_ >= kBlockChunkSize) {
        if (NewBlockChunk()) {
            block_chunk = block_manager_.block_chunks_.back();
            size_t res_idx = block_chunk->idx_;
            block_chunk->idx_++;
            pthread_mutex_unlock(&block_mtx_);
            return &block_chunk->blocks_[res_idx];
        }
        else {
            pthread_mutex_unlock(&block_mtx_);
            return NULL; 
        }
    }
    // 如果有空閒的Block那麼直接分配
    else {
        size_t res_idx = block_chunk->idx_;
        block_chunk->idx_++;
        pthread_mutex_unlock(&block_mtx_);
        return &block_chunk->blocks_[res_idx];
    }
    pthread_mutex_unlock(&block_mtx_);
    return NULL;
}

（2）對象池釋放資源流程

在 Global Pool 的析構函數中，遍歷 BlockChunk 數組，將所有的 BlockChunk 釋放掉，這樣做的優點是對象池中的資源統一管理不會出現內存泄露的問題，即便存在沒有回收的對象。缺點是在整個過程中對象池所佔用的內存都沒有釋放，如果出現分配對象數量峯值高但後面並不需要這麼多對象時會出現較多的內存浪費。

~GlobalPool() {
        pthread_spin_destroy(&freeslots_mtx_);
        pthread_mutex_destroy(&block_mtx_);
        for (int i = 0; i < block_manager_.block_chunks_.size(); i++) {
            delete block_manager_.block_chunks_[i];
        }
    }

9. Object Pool

在 Object Pool 中提供了訪問對象池的接口，其中維護了 Global Pool 和 Local Pool，在新建 Local Pool 時使用 round-robin 算法給 Local Pool 分配對應的 Global Pool。

class ObjectPool {
   GlobalPool<T> global_pool_[kGlobalPoolNum];
    thread_local static LocalPool<T> *local_pool_;
    std::atomic pool_idx_;
｝

實現過程中涉及到的一些問題：

**（1）內存對齊
**

使用 __attribute ((aligned(64))) 與 cacheline 進行對齊，內存對齊可以避免 cacheline 的僞共享。僞共享是什麼？現在的 CPU 一般有三級緩存，其中在 CPU 中 L1 cache 和 L2 cache 爲每個核獨有，L3 則所有核共享，因此產生了 cache 的一致性問題。爲了解決 cache 的一致性問題，一個核在寫入自己的 L1 cache 後，另一個核對在同一個 cacheline 上的變量進行訪問 / 修改時需要根據 MESI 協議把對應的 cacheline 同步到其他核，從而保證 cache 的一致性。但是在多線程程序中有可能發生下圖中的現象：被不同線程訪問、修改的變量被加載到同一 cacheline 中。

當多核要操作的不同變量處於同一 cacheline，其中一個核心更新緩存行中的某個變量時，這個 cacheline 會被標爲失效，如果其他核心需要訪問這個 cacheline 時需要從內存中重新加載，這種現象被稱爲僞共享。

如果在僞共享的情況下對該 cacheline 上的變量頻繁讀寫，會產生大量的 cache 同步開銷。爲了避免僞共享，可以通過 cacheline 填充使得該 cacheline 是專屬於某個核的。在對象池中的數據結構類如 Local Pool、Global Pool 都使用了 cacheline 對齊，防止在訪問這些數據時被其他的變量所影響，這是一種用空間換時間的方法。下面是對象池中對 Local Pool 和 Global Pool 進行內存對齊的例子：

struct __attribute__((aligned(64))) LocalPool {
    GlobalPool<T> *global_pool_;
    Block<T> *block_;
    FreeSlots<T> freeslots_;
};

class __attribute__((aligned(64))) GlobalPool {
    BlockManager<T> block_manager_;
    FreeSlotsManager<T> freeslots_manager_;
    pthread_spinlock_t freeslots_mtx_;
    pthread_mutex_t block_mtx_;
};

由於 Local Pool 和 Global Pool 中的成員變量在對象池進行分配和釋放的過程中會被頻繁訪問，如果不進行內存對齊有可能會發生僞共享產生較大的性能損失，因此這裏通過內存對齊來避免僞共享。

進行內存對齊後耗時減少約 5%。

（2）鎖優化

鎖優化的手段一般有這幾種：

**減少鎖持有的時間 **
減少鎖粒度

除了這兩種手段外當然最好就是能避開鎖，thread local 的資源池就是比較典型的例子。減少鎖持有的時間就是縮短臨界區，儘量將可以不在臨界區中進行操作的語句移到上鎖的區域之外。減少鎖的粒度就是一把大鎖劃分爲多個小鎖，這樣就可以使得加鎖的成功率得到提高，達到優化性能的目的。

在實現的對象池中主要是將對 Global Pool 的鎖劃分爲對 FreeSlotsManager 以及 BlockManager 這兩把鎖，但是具體用什麼鎖呢？這需要根據 FreeSlotsManager 和 BlockManager 中操作的臨界區特點來決定，首先看下 Mutex 和 SpinLock 的區別：

對 Mutex：嘗試獲取鎖，如果可得到就佔有，如果不能就進入睡眠等待，缺點是會產生 context switch 和 scheduling 開銷；
Spin Lock: 嘗試獲取鎖. 如果可得到就佔有，如果不能持續嘗試直到獲取，spin lock 的 lock/unlock 性能更好 (花費更少的 CPU 指令)，缺點是在獲取到鎖前線程不做任何事情，相當於 CPU 一直在空轉，造成了算力的浪費。

由上面的特點可知：Spin Lock 適用於臨界區運行時間很短的場景，Mutex 適合運用於臨界區運行時間較長的場景，對於 Global Pool 中的 FreeSlotsManager 更適合使用 Spin Lock，因爲在發生資源交換時臨界區操作比較輕量級，只涉及到簡單整型數值的比較以及加法。而 BlockManager 部分有可能需要申請一大片內存，臨界區消耗大使用 Mutex 比較合適，根據臨界區耗時使用不同類型鎖後耗時降低 9%。

（3）分支預測優化

使用 __builtin_expect 控制分支預測結果，__builtin_expect() 是 GCC 提供給開發人員使用的一種將分支轉移的信息提供給編譯器的手段，這樣編譯器可以根據所提供的分支轉移信息可以對代碼進行優化，以減少指令跳轉帶來的性能下降。

下面是 __builtin_expect 的使用方法：

__builtin_expect((x),1) 表示 x 的值爲真的可能性更大；
__builtin_expect((x),0) 表示 x 的值爲假的可能性更大。

因此可以在一些 if 語句中嵌入 __builtin_expect 來對分支預測進行優化，如在 Global Pool 中，當 Block 用光時需要 new 一個新的 BlockChunk，分配失敗的概率是非常小的，因此可以這樣寫：

 BlockChunk<T> *new_block_chunk = new (std::nothrow) BlockChunk<T>;
 if (unlikely(new_block_chunk == NULL))
     return false;

分支預測優化後耗時減少約 2%，效果一般，原因可能是分支預測器已經做得很好了，所以手動提供分支轉移的信息提升的也不多。

（4）如何在複用的內存上調用對象的構造和析構函數

可以結合可變模板參數、placement new、std::forward 這幾個特性來實現在複用的內存上調用對象的構造和析構函數，通過可變模板參數和 std::forward 來傳遞參數，通過 placement new 在固定的內存上調用構造函數，析構函數直接通過指針直接調用即可，代碼如下：

    template<class... Args>
    void Construct(T *p, Args&&... args) {
        new (p) T (std::forward<Args>(args)...);
    }
    void Destory(T *p) {
        p->~T();
    }
    template<class... Args>
    T* New(Args&&... args) {
        T *p = Allocate();
        Construct(p, std::forward<Args>(args)...);
        return p;
    }
    void Delete(T *p) {
        if (p) {
            p->~T();
            Free(p);
        }
    }

六、測試

1. 如何證明對象的內存被有效分配了？

對所分配出來的對象進行讀寫，最後在所有對象分配結束後驗證對象的值是否發生改變。

2. 如何證明對象被成功複用？

先批量分配 n 個對象，然後回收這 n 個對象，檢查此時的進程使用的內存使用量，下一次繼續分配 n 個對象，如果此時的進程內存使用量沒有改變，那麼說明這些對象是成功複用的。

3. 內存泄露測試

使用 valgrind 工具進行內存泄露測試：

valgrind --tool=memcheck --leak-check=full ./object_pool_test

4. 操作開銷定位

使用 perf 工具進行記錄，函數開銷圖如下：

5. 耗時測試

耗時測試是從內存分配效率的角度來進行測試，測試的對象是 POD，因爲相同類型的對象的構造析構的成本在不同的對象池中是相同的，在測試過程中需要降低對象的構造和析構對內存分配結果的影響。耗時測試主要與 brpc obejct pool、glibc malloc/free、jemalloc 的內存分配、釋放效率進行對比。

測試場景：每個對象的大小爲 64 Byte，使用 thread_num 個線程訪問對象池，每個線程每輪分配 10w 個對象，打亂對象後進行回收，重複 50 輪。比較的對象爲 glibc malloc/free, jemalloc, brpc object pool，設置的 global pool 的數量爲 4。

下圖是線程數量爲 1-4 時各個分配器的耗時對比圖，其中在線程的數量爲 (1-4) 時較其他三個分配器都有較大的優勢，相對於其他分配器都有超過 50% 的耗時減少。

下圖是訪問線程數量爲 1-16 時各個分配器的耗時曲線圖，在線程數較多的情況下 object pool 與 jemalloc 較 glibc malloc/free 以及 brpc object pool 都有較爲明顯的優勢，其中 object pool 相對 glibc malloc/free 耗時減少 60%，相對 brpc object pool 耗時減少 69%。object pool 在線程較少時相較於 jemalloc 有明顯優勢，但是隨着訪問的線程逐漸變多這種差距逐漸縮小了。

另外還需要說明的是在 brpc 官方文檔中稱：brpc object pool 穩定好於 glibc malloc/free，根據實測在分配的輪次較少的情況下的確是這樣的，但是在複用輪次變多時性能變差，個人認爲的原因是：brpc object pool 在 thread local 的資源池的實現中，對於那些空閒的對象使用了一個指針數組來保存，在進行資源交換時使用 memcpy 來拷貝空閒對象的指針使得效率非常低，這種劣勢在複用輪次變多時被放大了。

在內存上相對於與 brpc object pool 的內存消耗大致在同一水平，使用 16 個線程訪問對象池，每個線程分配 10w 個對象，然後進行回收，然後查看進程的 VmRSS， object pool 使用的內存爲 120M，brpc object pool 使用 132M。

本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/EOKk5rfUD0T-49XNL2jCLA

猜你喜歡