九、idSpanMap使用基数树代替原本的unordered_map 十、使用基数树前后性能对比

九、idSpanMap使用基数树代替原本的unordered_map

我们原本的idSpanMap用的是STL容器中的unordered_map哈希桶，因为STL的容器本身是不保证线程安全的，所以我们在访问时需要加锁保证线程安全，这也就是我们写的内存池的性能的瓶颈点。因为我做的这个内存池的项目是参照谷歌的开源项目tc-malloc，然后自主实现的mini版本，我查看tc-malloc的源码的优化策略是利用一颗基数树来代替stl中的unordered_map的，因为我们的需求是要保存页号和Span的映射关系，Span是指针本质也是一个整数，也就是说我们要保存的是<int,int>的映射关系，而基数树正好可以满足我们的需求，所以基数树就成为了优化我们的内存池的不二人选，并且用这个基数树定义idSpanMap最大的好处是访问idSpanMap时不需要加锁也能保证线程安全的。那它是怎么做到的呢？接下来我们就来学习一下。

9.1 什么是基数树？

基数树说白了也是一种哈希结构，基数树分为一层基数树，两层基数树，三层基数树。

9.1.1 一层基数树

一层基数树采用的是直接映射的方法

32位机器下才可以用一层基数树，64位机器下不可以，因为2^62次方太大了，早已超出了我们机器的内存的大小了，所以64位机器要用两层基数树或者三层基数树。

9.1.2 两层基数树

两层基数树是把哈希映射的关系分成两层，第一层是利用低19位的高5位判断该id在第一层的哪一个位置，再按照低14位判断在第二层的哪个位置。

三层基数树也是按照两层基数树的模式继续延伸下去即可，在这也就不过多叙述了，详情可以看后面的代码实现。

9.2 tcmalloc中基数树的源码实现

//基数树，用来代替 idSpanMap，访问时可以不用加锁// Single-level array
template <size_t BITS>
class TCMalloc_PageMap1 {
private:static const int LENGTH = 1 << BITS;void** array_;public:typedef uintptr_t Number;//explicit TCMalloc_PageMap1(void* (*allocator)(size_t)) {explicit TCMalloc_PageMap1() {//array_ = reinterpret_cast<void**>((*allocator)(sizeof(void*) << BITS));size_t size = sizeof(void*) << BITS;size_t alignSize = SizeClass::_RoundUp(size, 1 << PAGE_SHIFT);array_ = (void**)SystemAlloc(alignSize >> PAGE_SHIFT);memset(array_, 0, sizeof(void*) << BITS);}// Return the current value for KEY.  Returns NULL if not yet set,// or if k is out of range.void* get(Number k) const {if ((k >> BITS) > 0) {return NULL;}return array_[k];}// REQUIRES "k" is in range "[0,2^BITS-1]".// REQUIRES "k" has been ensured before.//// Sets the value 'v' for key 'k'.void set(Number k, void* v) {array_[k] = v;}
};// Two-level radix tree
template <int BITS>
class TCMalloc_PageMap2 {
private:// Put 32 entries in the root and (2^BITS)/32 entries in each leaf.static const int ROOT_BITS = 5;static const int ROOT_LENGTH = 1 << ROOT_BITS;static const int LEAF_BITS = BITS - ROOT_BITS;static const int LEAF_LENGTH = 1 << LEAF_BITS;// Leaf nodestruct Leaf {void* values[LEAF_LENGTH];};Leaf* root_[ROOT_LENGTH];             // Pointers to 32 child nodesvoid* (*allocator_)(size_t);          // Memory allocatorpublic:typedef uintptr_t Number;//explicit TCMalloc_PageMap2(void* (*allocator)(size_t)) {explicit TCMalloc_PageMap2() {//allocator_ = allocator;memset(root_, 0, sizeof(root_));PreallocateMoreMemory();}void* get(Number k) const {const Number i1 = k >> LEAF_BITS;const Number i2 = k & (LEAF_LENGTH - 1);if ((k >> BITS) > 0 || root_[i1] == NULL) {return NULL;}return root_[i1]->values[i2];}void set(Number k, void* v) {const Number i1 = k >> LEAF_BITS;const Number i2 = k & (LEAF_LENGTH - 1);ASSERT(i1 < ROOT_LENGTH);root_[i1]->values[i2] = v;}bool Ensure(Number start, size_t n) {for (Number key = start; key <= start + n - 1;) {const Number i1 = key >> LEAF_BITS;// Check for overflowif (i1 >= ROOT_LENGTH)return false;// Make 2nd level node if necessaryif (root_[i1] == NULL) {//Leaf* leaf = reinterpret_cast<Leaf*>((*allocator_)(sizeof(Leaf)));//if (leaf == NULL) return false;static ObjectPool<Leaf>	leafPool;Leaf* leaf = (Leaf*)leafPool.New();memset(leaf, 0, sizeof(*leaf));root_[i1] = leaf;}// Advance key past whatever is covered by this leaf nodekey = ((key >> LEAF_BITS) + 1) << LEAF_BITS;}return true;}void PreallocateMoreMemory() {// Allocate enough to keep track of all possible pagesEnsure(0, 1 << BITS);}
};// Three-level radix tree
template <int BITS>
class TCMalloc_PageMap3 {
private:// How many bits should we consume at each interior levelstatic const int INTERIOR_BITS = (BITS + 2) / 3; // Round-upstatic const int INTERIOR_LENGTH = 1 << INTERIOR_BITS;// How many bits should we consume at leaf levelstatic const int LEAF_BITS = BITS - 2 * INTERIOR_BITS;static const int LEAF_LENGTH = 1 << LEAF_BITS;// Interior nodestruct Node {Node* ptrs[INTERIOR_LENGTH];};// Leaf nodestruct Leaf {void* values[LEAF_LENGTH];};Node* root_;                          // Root of radix treevoid* (*allocator_)(size_t);          // Memory allocatorNode* NewNode() {Node* result = reinterpret_cast<Node*>((*allocator_)(sizeof(Node)));if (result != NULL) {memset(result, 0, sizeof(*result));}return result;}public:typedef uintptr_t Number;explicit TCMalloc_PageMap3(void* (*allocator)(size_t)) {allocator_ = allocator;root_ = NewNode();}void* get(Number k) const {const Number i1 = k >> (LEAF_BITS + INTERIOR_BITS);const Number i2 = (k >> LEAF_BITS) & (INTERIOR_LENGTH - 1);const Number i3 = k & (LEAF_LENGTH - 1);if ((k >> BITS) > 0 ||root_->ptrs[i1] == NULL || root_->ptrs[i1]->ptrs[i2] == NULL) {return NULL;}return reinterpret_cast<Leaf*>(root_->ptrs[i1]->ptrs[i2])->values[i3];}void set(Number k, void* v) {ASSERT(k >> BITS == 0);const Number i1 = k >> (LEAF_BITS + INTERIOR_BITS);const Number i2 = (k >> LEAF_BITS) & (INTERIOR_LENGTH - 1);const Number i3 = k & (LEAF_LENGTH - 1);reinterpret_cast<Leaf*>(root_->ptrs[i1]->ptrs[i2])->values[i3] = v;}bool Ensure(Number start, size_t n) {for (Number key = start; key <= start + n - 1;) {const Number i1 = key >> (LEAF_BITS + INTERIOR_BITS);const Number i2 = (key >> LEAF_BITS) & (INTERIOR_LENGTH - 1);// Check for overflowif (i1 >= INTERIOR_LENGTH || i2 >= INTERIOR_LENGTH)return false;// Make 2nd level node if necessaryif (root_->ptrs[i1] == NULL) {Node* n = NewNode();if (n == NULL) return false;root_->ptrs[i1] = n;}// Make leaf node if necessaryif (root_->ptrs[i1]->ptrs[i2] == NULL) {Leaf* leaf = reinterpret_cast<Leaf*>((*allocator_)(sizeof(Leaf)));if (leaf == NULL) return false;memset(leaf, 0, sizeof(*leaf));root_->ptrs[i1]->ptrs[i2] = reinterpret_cast<Node*>(leaf);}// Advance key past whatever is covered by this leaf nodekey = ((key >> LEAF_BITS) + 1) << LEAF_BITS;}return true;}void PreallocateMoreMemory() {}
};

由于上面的基数树是更改idSpanMap的结构的，所以需要把idSpanMap的访问操作修改成对应的基数树的set和get
idSpanMap主要是在PageCache中使用，修改后如下：

PageCache PageCache::_sInst;//k代表的是这个span的大小k页
Span* PageCache::NewSpan(size_t k)
{assert(k > 0);//如果申请的span的大小大于128页，则需要直接向堆申请if (k > NPAGES - 1){//向堆申请k页内存void* ptr = SystemAlloc(k);Span* kSpan = _spanPool.New();//地址转化成页号kSpan->_pageId = (PAGE_ID)ptr >> PAGE_SHIFT;kSpan->_n = k;//把页号和Kspan的映射关系放进Map中//_idSpanMap[kSpan->_pageId] = kSpan;_idSpanMap.set(kSpan->_pageId, kSpan);return kSpan;}else{//如果PageCache第k个位置的哈希桶上有k页大小的span，则直接返回一个spanif (!_spanLists[k].Empty()){Span* kSpan = _spanLists[k].PopFront();//把kSpan的页号和对应的Span*的映射关系存放到哈希桶中去,方便//CentralCache回收小块内存时，查找对应的span//kSpan代表的是一个k页大小的Span的大块内存，kSpan->_pageId//代表这个大块内存的起始地址，有k页，所以这k页映射到的都是这个Spanfor (PAGE_ID i = 0; i < kSpan->_n; i++){//_idSpanMap[kSpan->_pageId + i] = kSpan;_idSpanMap.set(kSpan->_pageId + i, kSpan);}return kSpan;}//走到这里说明PageCache第k个位置的哈希桶没有k页大小的span，则需要遍历//后面的大于k页的哈希桶，找到了一个n页大小的span就把这个span切分成一个// k页大小的span和一个n-k页大小的span，k页的返回，n-k页的挂到对应的哈希桶中//遍历后面的哈希桶for (size_t i = k + 1; i < NPAGES; i++){//找到了一个不为空的i页的哈希桶，就对它进行切分if (!_spanLists[i].Empty()){//k页的spanSpan* kSpan = _spanPool.New();//n页的spanSpan* nSpan = _spanLists[i].PopFront();//开始把一个n页的span切分成一个k页的span和一个n-k页的span// //从nSpan的头上切k页给kSpan，所以kSpan的页号就是nSpan的页号kSpan->_pageId = nSpan->_pageId;kSpan->_n = k;//kSpan的页数是k//被切分以后nSpan的页号需要+=k页，因为头nSpan的头k页已经切分给了kSpannSpan->_pageId += k;nSpan->_n -= k;//nSpan的页数要-=k页，因为nSpan被切走了k页//把kSpan的页号和对应的Span*的映射关系存放到哈希桶中去,方便// CentralCache回收小块内存时，查找对应的span//kSpan代表的是一个k页大小的Span的大块内存，kSpan->_pageId//代表这个大块内存的起始地址，有k页，所以这k页映射到的都是这个Spanfor (PAGE_ID i = 0; i < kSpan->_n; i++){//_idSpanMap[kSpan->_pageId + i] = kSpan;_idSpanMap.set(kSpan->_pageId + i, kSpan);}//nSpan被切分后的首页和尾页的页号和nspan的映射关系也需要保存起来//以便后续合并，因为合并的方式是前后页合并，往前找肯定找到的是一个span的//最后一页，往后找一定找的是一个span的第一页，所以挂在PageCache对应哈希桶//的span的第一页和最后一页与span的关系也需要保存起来// //_idSpanMap[nSpan->_pageId] = nSpan;//_idSpanMap[nSpan->_pageId + nSpan->_n - 1] = nSpan;_idSpanMap.set(nSpan->_pageId, nSpan);_idSpanMap.set(nSpan->_pageId + nSpan->_n - 1, nSpan);if (nSpan->_pageId == 0){int x = 0;}//把剩余的n-k页的span头插到对应下标的哈希桶中_spanLists[nSpan->_n].PushFront(nSpan);return kSpan;}}//走到这里说明前面的NPAGES个哈希桶中都没有Span，(例如第一次申请内存时)//则需要向堆申请一个128页大小的span大块内存，挂到对应的哈希桶中void* ptr = SystemAlloc(NPAGES - 1);Span* bigSpan = _spanPool.New();bigSpan->_pageId = (PAGE_ID)ptr >> PAGE_SHIFT;//内存的地址需要转换成页号映射到对应的哈希桶中bigSpan->_n = NPAGES - 1;//把NPAGES-1页大小的span头插到对应NPAGES-1号桶中去_spanLists[bigSpan->_n].PushFront(bigSpan);//本质是运用了复用的设计，避免代码中出现重复的逻辑return NewSpan(k);}}Span* PageCache::MapObjectToSpan(void* obj)
{//计算出obj对应的页号PAGE_ID id = (PAGE_ID)obj >> PAGE_SHIFT;访问_idSpanMap的时候需要加锁，避免线程安全的问题这里使用C++11的RAII锁，出了这个函数这把锁会自动解掉//std::unique_lock<std::mutex> lock(_pageMtx);通过页号查找该内存块对应的是哪一个span//auto ret = _idSpanMap.find(id);//if (ret != _idSpanMap.end())//{//	return ret->second;//}//else//{//	assert(false);//	return nullptr;//}//换成了基数树作为map存放页号和Span*的映射关系之后，访问的时候是不需要再加锁的，原因有以下几点：//1、只有在NewSpan函数和ReleaseSpanToPageCache函数会去写基数树。//2、基数树在写之前就会开好空间，写的过程中是不会影响到基数树的结构的，也就是说在两个线程在访问// 不同位置时互相是不会受到影响的。//3、基数树对同一个位置的读写是分离的，线程1对一个位置进行读写的时候，线程2不可能也在对同一个位置进行读写//写是在申请span和释放span的时候，是在没人用的时候做的；而读是在有人用这个span的时候做的，所以读写是分离的// 4、另一方面就是NewSpan函数和ReleaseSpanToPageCache函数在调用之前本身就已经加锁了，所以这里就不用加锁了//但是为什么用stl下的map要加锁呢？本质是红黑树在插入的时候会改变树的结构，一个线程在插入节点改变红黑树的结构，//一个线程在遍历就会有线程安全的问题，而基数树插入的时候不会影响结构，而且基数树读取的时候并不是遍历，而是直接通过下标//就访问到了对应的位置的//通过页号查找该内存块对应的是哪一个spanvoid* ret = _idSpanMap.get(id);if (ret != nullptr){return (Span*)ret;}else{assert(false);return nullptr;}
}//CentralCache把span还回来给PageCache
void PageCache::ReleaseSpanToPageCache(Span* span)
{//如果span的页数大于128页，则说明这个span是从堆上直接申请的，//直接释放给堆即可，不能挂到PageCache的哈希桶中，因为PageCache一个只有128个桶if (span->_n > NPAGES - 1){void* ptr = (void*)(span->_pageId << PAGE_SHIFT);SystemFree(ptr);_spanPool.Delete(span);return;}while (1){//找前一页的span，看是否能够和当前页合并，如果能，则循环向前合并，直到不能合并为止PAGE_ID prevId = span->_pageId - 1;void* ret = _idSpanMap.get(prevId);//_idSpanMap中没找到前一页和对应span，说明前一页的内存没有被申请，结束合并if (ret == nullptr){break;}Span* prevSpan = (Span*)ret;//如果前一页对应的span在CentralCache中正在被使用，结束合并if (prevSpan->_isUse == true){break;}//如果和前一页合并之后会超过哈希桶的最大的映射返回，结束合并if (prevSpan->_n + span->_n > NPAGES - 1){break;}//合并span和prevSpanspan->_pageId = prevSpan->_pageId;span->_n = prevSpan->_n + span->_n;//合并之后需要把prevSpan在对应的哈希桶中删除掉_spanLists[prevSpan->_n].Erase(prevSpan);//因为prevSpan已经被合并到了span中，所以prevSpan对应的内存可以delete掉了_spanPool.Delete(prevSpan);}while (1){//找span的下一个span的起始页号PAGE_ID nextId = span->_pageId + span->_n;void* ret = _idSpanMap.get(nextId);if (ret == nullptr){break;}Span* nextSpan = (Span*)ret;if (nextSpan->_isUse == true){break;}if (nextSpan->_n + span->_n > NPAGES - 1)//曾经写成NPAGES+1了{break;}//span的起始页号不变，页数相加span->_n = span->_n + nextSpan->_n;//合并之后需要把prevSpan在对应的哈希桶中删除掉_spanLists[nextSpan->_n].Erase(nextSpan);_spanPool.Delete(nextSpan);}//合并得到的新的span需要挂到对应页数的哈希桶中_spanLists[span->_n].PushFront(span);//在PageCache中的span要设置为false，好让后面相邻的span来合并span->_isUse = false;//为了方便后续的合并，需要把span的起始页号和尾页号和span建立映射关系//_idSpanMap[span->_pageId] = span;//_idSpanMap[span->_pageId + span->_n - 1] = span;_idSpanMap.set(span->_pageId, span);_idSpanMap.set(span->_pageId + span->_n - 1, span);}

代码中用的是一层基数树：

9.3 为什么使用基数树在访问时不需要加锁？

十、使用基数树前后性能对比

使用基数树优化前：我们的内存池比malloc还要稍微慢一些。

使用基数树优化后：我们的内存池的效率大概是malloc的10倍左右的样子，可见，基数树优化后的效率还是提高了不少的。