首頁 > 軟體

C++中高效能記憶體池的實現詳解

2022-10-26 14:02:58

一、概述

在 C/C++ 中,記憶體管理是一個非常棘手的問題,我們在編寫一個程式的時候幾乎不可避免的要遇到記憶體的分配邏輯,這時候隨之而來的有這樣一些問題:是否有足夠的記憶體可供分配?分配失敗了怎麼辦? 如何管理自身的記憶體使用情況? 等等一系列問題。在一個高可用的軟體中,如果我們僅僅單純的向作業系統去申請記憶體,當出現記憶體不足時就退出軟體,是明顯不合理的。正確的思路應該是在記憶體不足的時,考慮如何管理並優化自身已經使用的記憶體,這樣才能使得軟體變得更加可用。本次專案我們將實現一個記憶體池,並使用一個棧結構來測試我們的記憶體池提供的分配效能。最終,我們要實現的記憶體池在棧結構中的效能,要遠高於使用 std::allocator 和 std::vector,如下圖所示:

專案涉及的知識點

C++ 中的記憶體分配器 std::allocator

記憶體池技術

手動實現模板鏈式棧

鏈式棧和列表棧的效能比較

記憶體池簡介

記憶體池是池化技術中的一種形式。通常我們在編寫程式的時候回使用 new delete 這些關鍵字來向作業系統申請記憶體,而這樣造成的後果就是每次申請記憶體和釋放記憶體的時候,都需要和作業系統的系統呼叫打交道,從堆中分配所需的記憶體。如果這樣的操作太過頻繁,就會找成大量的記憶體碎片進而降低記憶體的分配效能,甚至出現記憶體分配失敗的情況。

而記憶體池就是為了解決這個問題而產生的一種技術。從記憶體分配的概念上看,記憶體申請無非就是向記憶體分配方索要一個指標,當向作業系統申請記憶體時,

作業系統需要進行復雜的記憶體管理排程之後,才能正確的分配出一個相應的指標。而這個分配的過程中,我們還面臨著分配失敗的風險。

所以,每一次進行記憶體分配,就會消耗一次分配記憶體的時間,設這個時間為 T,那麼進行 n 次分配總共消耗的時間就是 nT;如果我們一開始就確定好我們可能需要多少記憶體,那麼在最初的時候就分配好這樣的一塊記憶體區域,當我們需要記憶體的時候,直接從這塊已經分配好的記憶體中使用即可,那麼總共需要的分配時間僅僅只有 T。當 n 越大時,節約的時間就越多。

二、主函數設計

我們要設計實現一個高效能的記憶體池,那麼自然避免不了需要對比已有的記憶體,而比較記憶體池對記憶體的分配效能,就需要實現一個需要對記憶體進行動態分配的結構(比如:連結串列棧),為此,可以寫出如下的程式碼:

#include <iostream>   // std::cout, std::endl
#include <cassert>    // assert()
#include <ctime>      // clock()
#include <vector>     // std::vector
#include "MemoryPool.hpp"  // MemoryPool<T>
#include "StackAlloc.hpp"  // StackAlloc<T, Alloc>
// 插入元素個數
#define ELEMS 10000000
// 重複次數
#define REPS 100
int main()
{
    clock_t start;
    // 使用 STL 預設分配器
    StackAlloc<int, std::allocator<int> > stackDefault;
    start = clock();
    for (int j = 0; j < REPS; j++) {
        assert(stackDefault.empty());
        for (int i = 0; i < ELEMS; i++)
          stackDefault.push(i);
        for (int i = 0; i < ELEMS; i++)
          stackDefault.pop();
    }
    std::cout << "Default Allocator Time: ";
    std::cout << (((double)clock() - start) / CLOCKS_PER_SEC) << "nn";
    // 使用記憶體池
    StackAlloc<int, MemoryPool<int> > stackPool;
    start = clock();
    for (int j = 0; j < REPS; j++) {
        assert(stackPool.empty());
        for (int i = 0; i < ELEMS; i++)
          stackPool.push(i);
        for (int i = 0; i < ELEMS; i++)
          stackPool.pop();
    }
    std::cout << "MemoryPool Allocator Time: ";
    std::cout << (((double)clock() - start) / CLOCKS_PER_SEC) << "nn";
    return 0;
}

在上面的兩段程式碼中,StackAlloc 是一個連結串列棧,接受兩個模板引數,第一個引數是棧中的元素型別,第二個引數就是棧使用的記憶體分配器。

因此,這個記憶體分配器的模板引數就是整個比較過程中唯一的變數,使用預設分配器的模板引數為 std::allocator,而使用記憶體池的模板引數為 MemoryPool。

std::allocator 是 C++標準庫中提供的預設分配器,他的特點就在於我們在 使用 new 來申請記憶體構造新物件的時候,勢必要呼叫類物件的預設建構函式,而使用 std::allocator 則可以將記憶體分配和物件的構造這兩部分邏輯給分離開來,使得分配的記憶體是原始、未構造的。

下面我們來實現這個連結串列棧。

三、模板連結串列棧

棧的結構非常的簡單,沒有什麼複雜的邏輯操作,其成員函數只需要考慮兩個基本的操作:入棧、出棧。為了操作上的方便,我們可能還需要這樣一些方法:判斷棧是否空、清空棧、獲得棧頂元素。

#include <memory>
template <typename T>
struct StackNode_
{
  T data;
  StackNode_* prev;
};
// T 為儲存的物件型別, Alloc 為使用的分配器, 並預設使用 std::allocator 作為物件的分配器
template <typename T, typename Alloc = std::allocator<T> >
class StackAlloc
{
  public:
    // 使用 typedef 簡化型別名
    typedef StackNode_<T> Node;
    typedef typename Alloc::template rebind<Node>::other allocator;
    // 預設構造
    StackAlloc() { head_ = 0; }
    // 預設解構
    ~StackAlloc() { clear(); }
    // 當棧中元素為空時返回 true
    bool empty() {return (head_ == 0);}
    // 釋放棧中元素的所有記憶體
    void clear();
    // 壓棧
    void push(T element);
    // 出棧
    T pop();
    // 返回棧頂元素
    T top() { return (head_->data); }
  private:
    // 
    allocator allocator_;
    // 棧頂
    Node* head_;
};

簡單的邏輯諸如構造、解構、判斷棧是否空、返回棧頂元素的邏輯都非常簡單,直接在上面的定義中實現了,下面我們來實現 clear(), push() 和 pop() 這三個重要的邏輯:

// 釋放棧中元素的所有記憶體
void clear() {
  Node* curr = head_;
  // 依次出棧
  while (curr != 0)
  {
    Node* tmp = curr->prev;
    // 先解構, 再回收記憶體
    allocator_.destroy(curr);
    allocator_.deallocate(curr, 1);
    curr = tmp;
  }
  head_ = 0;
}
// 入棧
void push(T element) {
  // 為一個節點分配記憶體
  Node* newNode = allocator_.allocate(1);
  // 呼叫節點的建構函式
  allocator_.construct(newNode, Node());
  // 入棧操作
  newNode->data = element;
  newNode->prev = head_;
  head_ = newNode;
}
// 出棧
T pop() {
  // 出棧操作 返回出棧元素
  T result = head_->data;
  Node* tmp = head_->prev;
  allocator_.destroy(head_);
  allocator_.deallocate(head_, 1);
  head_ = tmp;
  return result;
}

至此,我們完成了整個模板連結串列棧,現在我們可以先註釋掉 main() 函數中使用記憶體池部分的程式碼來測試這個連表棧的記憶體分配情況,我們就能夠得到這樣的結果:

在使用 std::allocator 的預設記憶體分配器中,在

#define ELEMS 10000000
#define REPS 100

的條件下,總共花費了近一分鐘的時間。

如果覺得花費的時間較長,不願等待,則你嘗試可以減小這兩個值

總結

本節我們實現了一個用於測試效能比較的模板連結串列棧,目前的程式碼如下。在下一節中,我們開始詳細實現我們的高效能記憶體池。

// StackAlloc.hpp
#ifndef STACK_ALLOC_H
#define STACK_ALLOC_H
#include <memory>
template <typename T>
struct StackNode_
{
  T data;
  StackNode_* prev;
};
// T 為儲存的物件型別, Alloc 為使用的分配器,
// 並預設使用 std::allocator 作為物件的分配器
template <class T, class Alloc = std::allocator<T> >
class StackAlloc
{
  public:
    // 使用 typedef 簡化型別名
    typedef StackNode_<T> Node;
    typedef typename Alloc::template rebind<Node>::other allocator;
    // 預設構造
    StackAlloc() { head_ = 0; }
    // 預設解構
    ~StackAlloc() { clear(); }
    // 當棧中元素為空時返回 true
    bool empty() {return (head_ == 0);}
    // 釋放棧中元素的所有記憶體
    void clear() {
      Node* curr = head_;
      while (curr != 0)
      {
        Node* tmp = curr->prev;
        allocator_.destroy(curr);
        allocator_.deallocate(curr, 1);
        curr = tmp;
      }
      head_ = 0;
    }
    // 入棧
    void push(T element) {
      // 為一個節點分配記憶體
      Node* newNode = allocator_.allocate(1);
      // 呼叫節點的建構函式
      allocator_.construct(newNode, Node());
      // 入棧操作
      newNode->data = element;
      newNode->prev = head_;
      head_ = newNode;
    }
    // 出棧
    T pop() {
      // 出棧操作 返回出棧結果
      T result = head_->data;
      Node* tmp = head_->prev;
      allocator_.destroy(head_);
      allocator_.deallocate(head_, 1);
      head_ = tmp;
      return result;
    }
    // 返回棧頂元素
    T top() { return (head_->data); }
  private:
    allocator allocator_;
    Node* head_;
};
#endif // STACK_ALLOC_H
// main.cpp
#include <iostream>
#include <cassert>
#include <ctime>
#include <vector>
// #include "MemoryPool.hpp"
#include "StackAlloc.hpp"
// 根據電腦效能調整這些值
// 插入元素個數
#define ELEMS 25000000
// 重複次數
#define REPS 50
int main()
{
    clock_t start;
   // 使用預設分配器
    StackAlloc<int, std::allocator<int> > stackDefault;
    start = clock();
    for (int j = 0; j < REPS; j++) {
        assert(stackDefault.empty());
        for (int i = 0; i < ELEMS; i++)
          stackDefault.push(i);
        for (int i = 0; i < ELEMS; i++)
          stackDefault.pop();
    }
    std::cout << "Default Allocator Time: ";
    std::cout << (((double)clock() - start) / CLOCKS_PER_SEC) << "nn";
    // 使用記憶體池
    // StackAlloc<int, MemoryPool<int> > stackPool;
    // start = clock();
    // for (int j = 0; j < REPS; j++) {
    //     assert(stackPool.empty());
    //     for (int i = 0; i < ELEMS; i++)
    //       stackPool.push(i);
    //     for (int i = 0; i < ELEMS; i++)
    //       stackPool.pop();
    // }
    // std::cout << "MemoryPool Allocator Time: ";
    // std::cout << (((double)clock() - start) / CLOCKS_PER_SEC) << "nn";
    return 0;
}

四、設計記憶體池

在上一節實驗中,我們在模板連結串列棧中使用了預設構造器來管理棧操作中的元素記憶體,一共涉及到了 rebind::other, allocate(), dealocate(), construct(), destroy()這些關鍵性的介面。所以為了讓程式碼直接可用,我們同樣應該在記憶體池中設計同樣的介面:

#ifndef MEMORY_POOL_HPP
#define MEMORY_POOL_HPP

#include <climits>
#include <cstddef>

template <typename T, size_t BlockSize = 4096>
class MemoryPool
{
  public:
    // 使用 typedef 簡化型別書寫
    typedef T*              pointer;

    // 定義 rebind<U>::other 介面
    template <typename U> struct rebind {
      typedef MemoryPool<U> other;
    };

    // 預設構造, 初始化所有的槽指標
    // C++11 使用了 noexcept 來顯式的宣告此函數不會丟擲異常
    MemoryPool() noexcept {
      currentBlock_ = nullptr;
      currentSlot_ = nullptr;
      lastSlot_ = nullptr;
      freeSlots_ = nullptr;
    }

    // 銷燬一個現有的記憶體池
    ~MemoryPool() noexcept;

    // 同一時間只能分配一個物件, n 和 hint 會被忽略
    pointer allocate(size_t n = 1, const T* hint = 0);

    // 銷燬指標 p 指向的記憶體區塊
    void deallocate(pointer p, size_t n = 1);

    // 呼叫建構函式
    template <typename U, typename... Args>
    void construct(U* p, Args&&... args);

    // 銷燬記憶體池中的物件, 即呼叫物件的解構函式
    template <typename U>
    void destroy(U* p) {
      p->~U();
    }

​​​​​​​  private:
    // 用於儲存記憶體池中的物件槽, 
    // 要麼被範例化為一個存放物件的槽, 
    // 要麼被範例化為一個指向存放物件槽的槽指標
    union Slot_ {
      T element;
      Slot_* next;
    };

    // 資料指標
    typedef char* data_pointer_;
    // 物件槽
    typedef Slot_ slot_type_;
    // 物件槽指標
    typedef Slot_* slot_pointer_;

    // 指向當前記憶體區塊
    slot_pointer_ currentBlock_;
    // 指向當前記憶體區塊的一個物件槽
    slot_pointer_ currentSlot_;
    // 指向當前記憶體區塊的最後一個物件槽
    slot_pointer_ lastSlot_;
    // 指向當前記憶體區塊中的空閒物件槽
    slot_pointer_ freeSlots_;

    // 檢查定義的記憶體池大小是否過小
    static_assert(BlockSize >= 2 * sizeof(slot_type_), "BlockSize too small.");
};

#endif // MEMORY_POOL_HPP

在上面的類設計中可以看到,在這個記憶體池中,其實是使用連結串列來管理整個記憶體池的記憶體區塊的。記憶體池首先會定義固定大小的基本記憶體區塊(Block),然後在其中定義了一個可以範例化為存放物件記憶體槽的物件槽(Slot_)和物件槽指標的一個聯合。然後在區塊中,定義了四個關鍵性質的指標,它們的作用分別是:

currentBlock_: 指向當前記憶體區塊的指標

currentSlot_: 指向當前記憶體區塊中的物件槽

lastSlot_: 指向當前記憶體區塊中的最後一個物件槽

freeSlots_: 指向當前記憶體區塊中所有空閒的物件槽

梳理好整個記憶體池的設計結構之後,我們就可以開始實現關鍵性的邏輯了。

五、實現

MemoryPool::construct() 實現

MemoryPool::construct() 的邏輯是最簡單的,我們需要實現的,僅僅只是呼叫信件物件的建構函式即可,因此:

// 呼叫建構函式, 使用 std::forward 轉發變參模板
template <typename U, typename... Args>
void construct(U* p, Args&&... args) {
    new (p) U (std::forward<Args>(args)...);
}

MemoryPool::deallocate() 實現

MemoryPool::deallocate() 是在物件槽中的物件被解構後才會被呼叫的,主要目的是銷燬記憶體槽。其邏輯也不復雜:

// 銷燬指標 p 指向的記憶體區塊
void deallocate(pointer p, size_t n = 1) {
  if (p != nullptr) {
    // reinterpret_cast 是強制型別轉換符
    // 要存取 next 必須強制將 p 轉成 slot_pointer_
    reinterpret_cast<slot_pointer_>(p)->next = freeSlots_;
    freeSlots_ = reinterpret_cast<slot_pointer_>(p);
  }
}

MemoryPool::~MemoryPool() 實現

解構函式負責銷燬整個記憶體池,因此我們需要逐個刪除掉最初向作業系統申請的記憶體塊:

​​​​// 銷燬一個現有的記憶體池
~MemoryPool() noexcept {
  // 迴圈銷燬記憶體池中分配的記憶體區塊
  slot_pointer_ curr = currentBlock_;
  while (curr != nullptr) {
    slot_pointer_ prev = curr->next;
    operator delete(reinterpret_cast<void*>(curr));
    curr = prev;
  }
}

MemoryPool::allocate() 實現

MemoryPool::allocate() 毫無疑問是整個記憶體池的關鍵所在,但實際上理清了整個記憶體池的設計之後,其實現並不複雜。具體實現如下:

​​​​// 同一時間只能分配一個物件, n 和 hint 會被忽略
pointer allocate(size_t n = 1, const T* hint = 0) {
  // 如果有空閒的物件槽,那麼直接將空閒區域交付出去
  if (freeSlots_ != nullptr) {
    pointer result = reinterpret_cast<pointer>(freeSlots_);
    freeSlots_ = freeSlots_->next;
    return result;
  } else {
    // 如果物件槽不夠用了,則分配一個新的記憶體區塊
    if (currentSlot_ >= lastSlot_) {
      // 分配一個新的記憶體區塊,並指向前一個記憶體區塊
      data_pointer_ newBlock = reinterpret_cast<data_pointer_>(operator new(BlockSize));
      reinterpret_cast<slot_pointer_>(newBlock)->next = currentBlock_;
      currentBlock_ = reinterpret_cast<slot_pointer_>(newBlock);
      // 填補整個區塊來滿足元素記憶體區域的對齊要求
      data_pointer_ body = newBlock + sizeof(slot_pointer_);
      uintptr_t result = reinterpret_cast<uintptr_t>(body);
      size_t bodyPadding = (alignof(slot_type_) - result) % alignof(slot_type_);
      currentSlot_ = reinterpret_cast<slot_pointer_>(body + bodyPadding);
      lastSlot_ = reinterpret_cast<slot_pointer_>(newBlock + BlockSize - sizeof(slot_type_) + 1);
    }
    return reinterpret_cast<pointer>(currentSlot_++);
  }
}

六、與 std::vector 的效能對比

我們知道,對於棧來說,鏈棧其實並不是最好的實現方式,因為這種結構的棧不可避免的會涉及到指標相關的操作,同時,還會消耗一定量的空間來存放節點之間的指標。事實上,我們可以使用 std::vector 中的 push_back() 和 pop_back() 這兩個操作來模擬一個棧,我們不妨來對比一下這個 std::vector 與我們所實現的記憶體池在效能上誰高誰低,我們在 主函數中加入如下程式碼:

// 比較記憶體池和 std::vector 之間的效能
    std::vector<int> stackVector;
    start = clock();
    for (int j = 0; j < REPS; j++) {
        assert(stackVector.empty());
        for (int i = 0; i < ELEMS; i++)
          stackVector.push_back(i);
        for (int i = 0; i < ELEMS; i++)
          stackVector.pop_back();
    }
    std::cout << "Vector Time: ";
    std::cout << (((double)clock() - start) / CLOCKS_PER_SEC) << "nn";

這時候,我們重新編譯程式碼,就能夠看出這裡面的差距了:

首先是使用預設分配器的連結串列棧速度最慢,其次是使用 std::vector 模擬的棧結構,在連結串列棧的基礎上大幅度削減了時間。

std::vector 的實現方式其實和記憶體池較為類似,在 std::vector 空間不夠用時,會拋棄現在的記憶體區域重新申請一塊更大的區域,並將現在記憶體區域中的資料整體拷貝一份到新區域中。

最後,對於我們實現的記憶體池,消耗的時間最少,即記憶體分配效能最佳,完成了本專案。

總結

本節中,我們實現了我們上節實驗中未實現的記憶體池,完成了整個專案的目標。這個記憶體池不僅精簡而且高效,整個記憶體池的完整程式碼如下:

#ifndef MEMORY_POOL_HPP
#define MEMORY_POOL_HPP

#include <climits>
#include <cstddef>

template <typename T, size_t BlockSize = 4096>
class MemoryPool
{
  public:
    // 使用 typedef 簡化型別書寫
    typedef T*              pointer;

    // 定義 rebind<U>::other 介面
    template <typename U> struct rebind {
      typedef MemoryPool<U> other;
    };

    // 預設構造
    // C++11 使用了 noexcept 來顯式的宣告此函數不會丟擲異常
    MemoryPool() noexcept {
      currentBlock_ = nullptr;
      currentSlot_ = nullptr;
      lastSlot_ = nullptr;
      freeSlots_ = nullptr;
    }

    // 銷燬一個現有的記憶體池
    ~MemoryPool() noexcept {
      // 迴圈銷燬記憶體池中分配的記憶體區塊
      slot_pointer_ curr = currentBlock_;
      while (curr != nullptr) {
        slot_pointer_ prev = curr->next;
        operator delete(reinterpret_cast<void*>(curr));
        curr = prev;
      }
    }

    // 同一時間只能分配一個物件, n 和 hint 會被忽略
    pointer allocate(size_t n = 1, const T* hint = 0) {
      if (freeSlots_ != nullptr) {
        pointer result = reinterpret_cast<pointer>(freeSlots_);
        freeSlots_ = freeSlots_->next;
        return result;
      }
      else {
        if (currentSlot_ >= lastSlot_) {
          // 分配一個記憶體區塊
          data_pointer_ newBlock = reinterpret_cast<data_pointer_>(operator new(BlockSize));
          reinterpret_cast<slot_pointer_>(newBlock)->next = currentBlock_;
          currentBlock_ = reinterpret_cast<slot_pointer_>(newBlock);
          data_pointer_ body = newBlock + sizeof(slot_pointer_);
          uintptr_t result = reinterpret_cast<uintptr_t>(body);
          size_t bodyPadding = (alignof(slot_type_) - result) % alignof(slot_type_);
          currentSlot_ = reinterpret_cast<slot_pointer_>(body + bodyPadding);
          lastSlot_ = reinterpret_cast<slot_pointer_>(newBlock + BlockSize - sizeof(slot_type_) + 1);
        }
        return reinterpret_cast<pointer>(currentSlot_++);
      }
    }

    // 銷燬指標 p 指向的記憶體區塊
    void deallocate(pointer p, size_t n = 1) {
      if (p != nullptr) {
        reinterpret_cast<slot_pointer_>(p)->next = freeSlots_;
        freeSlots_ = reinterpret_cast<slot_pointer_>(p);
      }
    }

    // 呼叫建構函式, 使用 std::forward 轉發變參模板
    template <typename U, typename... Args>
    void construct(U* p, Args&&... args) {
      new (p) U (std::forward<Args>(args)...);
    }

    // 銷燬記憶體池中的物件, 即呼叫物件的解構函式
    template <typename U>
    void destroy(U* p) {
      p->~U();
    }

  private:
    // 用於儲存記憶體池中的物件槽
    union Slot_ {
      T element;
      Slot_* next;
    };

    // 資料指標
    typedef char* data_pointer_;
    // 物件槽
    typedef Slot_ slot_type_;
    // 物件槽指標
    typedef Slot_* slot_pointer_;

    // 指向當前記憶體區塊
    slot_pointer_ currentBlock_;
    // 指向當前記憶體區塊的一個物件槽
    slot_pointer_ currentSlot_;
    // 指向當前記憶體區塊的最後一個物件槽
    slot_pointer_ lastSlot_;
    // 指向當前記憶體區塊中的空閒物件槽
    slot_pointer_ freeSlots_;
    // 檢查定義的記憶體池大小是否過小
    static_assert(BlockSize >= 2 * sizeof(slot_type_), "BlockSize too small.");
};

#endif // MEMORY_POOL_HPP

以上就是C++中高效能記憶體池的實現詳解的詳細內容,更多關於C++高效能記憶體池的資料請關注it145.com其它相關文章!


IT145.com E-mail:sddin#qq.com