TCmalloc google开源项目核心部分模拟实现(学习记录)

TCmalloc google开源项目核心部分学习实现（学习记录）

• • 1什么是[内存池](https://baike.baidu.com/item/%E5%86%85%E5%AD%98%E6%B1%A0/8577153?fr=aladdin)
  • • 1.1池化技术
    • 1.2内存池
• 2.开胃菜-设计一个定长内存池
• • 代码实现：
  • • ObjectPool.h
    • ObjectPool.cpp测试
  • 3.[TCmalloc](https://baike.baidu.com/item/tcmalloc/2832982?fr=aladdin)(高并发内存池)整体框架介绍：
  • • ThreadCache
    • CentralCache
    • page cache
  • 4.代码实现：
  • • Common.h
    • ThreadCache.h
    • ThreadCache.cpp
    • CentralCache.h
    • CentralCache.cpp
    • PageMap.h
    • PageCache.h
    • PageCache.cpp
    • ConcurrentAlloc.h
  • 5.测试文件：
  • • Test.cpp
    • malloc与tcmallo测试对比结果
  • 总结
  • 项目源代码Github链接

1什么是内存池 1.1池化技术

所谓“池化技术”，就是程序先向系统申请过量的资源，然后自己管理，以备不时之需。之所以要申请过 量的资源，是因为每次申请该资源都有较大的开销，不如提前申请好了，这样使用时就会变得非常快 捷，大大提高程序运行效率。 在计算机中，有很多使用“池”这种技术的地方，除了内存池，还有连接池、线程池、对象池等。以服务 器上的线程池为例，它的主要思想是：先启动若干数量的线程，让它们处于睡眠状态，当接收到客户端 的请求时，唤醒池中某个睡眠的线程，让它来处理客户端的请求，当处理完这个请求，线程又进入睡眠 状态。

1.2内存池

内存池是指程序预先从操作系统申请一块足够大内存，此后，当程序中需要申请内存的时候，不是直接 向操作系统申请，而是直接从内存池中获取；同理，当程序释放内存的时候，并不真正将内存返回给操 作系统，而是返回内存池。当程序退出(或者特定时间)时，内存池才将之前申请的内存真正释放。减少用户层向系统层的内存申请调用，实现高效;

2.开胃菜-设计一个定长内存池

此次设计的定长内存池ObjectPool结构如下：(其单位小内存块的大小为T类型的大小）

• 需要申请内存时，一次性申请一个大块空间，记录起始位置_memory;
• 需要使用size大小时，将大块内存抽象切割成小块，将起始位置_memory向后移动size大小;
• 释放内存时，将需要释放的空间挂入自由链表_freeList，可供下次申请使用;注意：这个链表不是通过内部指针连接下一个，而是通过前一块空间的前4or8个字节记录后一个小空间的起始地址抽象连接的;

这样做的目的是循环利用预先开辟的一大块空间，减少用户层申请内存时与系统层的交互，提高效率;

代码实现： ObjectPool.h

#pragma once
#define _CRT_SECURE_NO_WARNINGS 1
#include
using namespace std;


#ifdef _WIN32
#include
#else
// 包Linux相关头文件，增加代码的可移植性;
#endif

// 直接去堆上按页申请空间 摆脱malloc;
inline static void* SystemAlloc(size_t kpage)
{
#ifdef _WIN32
	void* ptr = VirtualAlloc(0, kpage << 13, MEM_COMMIT | MEM_RESERVE, PAGE_READWRITE);
#else
	// linux下brk mmap等
#endif

	if (ptr == nullptr)
		throw std::bad_alloc();

	return ptr;
}
	

template
class ObjectPool
{
	char* _memory = nullptr; // 指向大块内存的指针
	size_t _remainBytes = 0; // 大块内存在切分过程中剩余字节数
	void* _freeList = nullptr; // 还回来过程中链接的自由链表的头指针

public:
	T* New()
	{
		T* obj = nullptr;
		//如果之前有申请的空间被释放，那就先从free了的空间拿;(内存池技术)
		if (_freeList != nullptr)	
		{
			//头删
            //注意！*((void**)_freeList)相当访取_freeList前4or8个字节操作(由32位系统->指针4byte and 64位系统->指针8byte决定);
			void* next = *((void**)_freeList);
			obj = (T*)_freeList;
			_freeList = next;
		}
		else
		{
			if (_remainBytes < sizeof(T))
			{	
				// 剩余内存不够一个对象大小时，则重新开大块空间
				_remainBytes = 128 * 1024;
				//_memory = (char*)malloc(_remainBytes);
                //使用SystemAlloc直接向堆申请内存，脱离malloc，方便体现malloc和该ObjectPool的差别;
				_memory = (char*)SystemAlloc(_remainBytes >> 13);//char*类型内存可以使其+1or-1的单位操作为1字节，方便内存管理;

				if (_memory == nullptr)
				{
					throw std::bad_alloc();
				}
				
			}
			//防止某个T类还没当前系统下一个指针大小大 那么就装不下后一个的地址了，这里做特殊处理;至少保证一个对象内足够装的下一个指针大小;
			int Objsize = sizeof(T) < sizeof(void*) ? sizeof(void*): sizeof(T);

			obj = (T*)_memory;
			_memory += Objsize;
			_remainBytes -= Objsize;

		}
		// 定位new，内置类型不处理，自定义类型调用其构造函数;
		new(obj)T;
		return obj;
	}

	void Delete(T* obj)
	{
        //内置类型不处理，自定义类型调用其构析构函数;
		obj->~T();
		//头插(此处不是真正的删除，而是标志为未使用，挂入自由链表)
		*(void**)obj = _freeList;
		_freeList = obj;
	}
}; 

ObjectPool.cpp测试

#include"ObjectPool.h"
#include
#include "ConcurrentAlloc.h"

//用于测试的类型树节点;
struct TreeNode
{
	int _val;
	TreeNode* _left;
	TreeNode* _right;

	TreeNode()
		:_val(0)
		, _left(nullptr)
		, _right(nullptr)
	{}
};
void TestObjectPool()
{
	// 申请释放的轮次
	const size_t Rounds = 5;

	// 每轮申请释放多少次
	const size_t N = 100000;

	std::vector v1;
	v1.reserve(N);

	size_t begin1 = clock();
	for (size_t j = 0; j < Rounds; ++j)
	{
		for (int i = 0; i < N; ++i)
		{
			v1.push_back(new TreeNode);
		}
		for (int i = 0; i < N; ++i)
		{
			delete v1[i];
		}
		v1.clear();
	}

	size_t end1 = clock();

	std::vector v2;
	v2.reserve(N);

	ObjectPool TNPool;
	size_t begin2 = clock();
	for (size_t j = 0; j < Rounds; ++j)
	{
		for (int i = 0; i < N; ++i)
		{
			v2.push_back(TNPool.New());
		}
		for (int i = 0; i < N; ++i)
		{
			TNPool.Delete(v2[i]);
		}
		v2.clear();
	}
	size_t end2 = clock();

	cout << "new cost time:" << end1 - begin1 << endl;
	cout << "object pool cost time:" << end2 - begin2 << endl;
}
int mian()
{
    TestObjectPool();
    return 0;
}

运行结果：

显然，定长内存池的New空间比new(底层封装的是malloc申请空间)更高效

3.TCmalloc(高并发内存池)整体框架介绍：

谷歌开源项目：TCMalloc (google-perftools) 是用于优化C++写的多线程应用，比glibc 2.3的malloc快。这个模块可以用来让MySQL在高并发下内存占用更加稳定。

ThreadCache

thread cache：线程缓存是每个线程独有的，用于小于256KB的内存的分配（设计规定），线程从这里申请内存不需要加锁，每个线程独享一个cache，这也就是这个并发线程池高效的地方。核心结构是FreeList _freeLists[NFREELIST];eg：_freeLists[2]代表该size对应哈希桶中有n个未使用的obj size大小的小内存块，当需要申请时优先从_freeList中拿无人使用的被挂起的obj空间;

CentralCache

central cache：中心缓存是所有线程所共享，thread cache是按需从central cache中获取的对 象。central cache合适的时机回收thread cache中的对象，避免一个线程占用了太多的内存，而 其他线程的内存吃紧，达到内存分配在多个线程中更均衡的按需调度的目的。central cache是存在竞争的，所以从这里取内存对象是需要加锁，首先这里用的是桶锁，其次只有thread cache的 没有内存对象时才会找central cache，所以这里竞争不会很激烈。其核心结构是SpanList _spanLists[NFREELIST];eg：_spanLists[2]代表该size对应链桶中的n个span，每个span下面挂了n个大小为size的obj小内存块；(size大小是指通过测试和计算，将所需内存大小x字节依据某种对齐规则对其为size字节，使span的链式桶结构数量合适，大小统一)

page cache

page cache：页缓存是在central cache缓存上面的一层缓存，存储的内存是以页为单位存储及分 配的，central cache没有内存对象时，从page cache分配出一定数量的page，并切割成定长大小 的小块内存，分配给central cache。当一个span的几个跨度页的对象都回收以后，page cache 会回收central cache满足条件的span对象，并且合并相邻的页，组成更大的页，缓解内存碎片 的问题。Page中的核心结构是SpanList _spanLists[NPAGES];eg:_spanLists[2]代表该链桶含有n个大小为2页(2*8字节)的span;如果从page[1]到page[128]都没有可用span，那么只能从系统堆直接申请一块大空间放进去，再给central thread层用了;

本文TCmalloc整体项目围绕上述三层结构实现，抽取原项目的小部分核心内容学习;这个项目会用到C/C++、数据结构（链表、哈希桶）、操作系统内存管理、单例模式、多线程、互斥锁 等等方面的知识。

4.代码实现： Common.h

这是一个通用头文件，包含各种所需要的库文件和三层结构需要的对其数和size-申请数量转换的函数;

#pragma once

#include 
#include 
#include 
#include 

ThreadCache.h

thread cache是哈希桶结构，每个桶是一个按桶位置映射大小的内存块对象的自由链表。每个线程都会 有一个thread cache对象，这样每个线程在这里获取对象和释放对象时是无锁的。

申请内存：

1. 当内存申请size<=256KB，先获取到线程本地存储的thread cache对象，计算size映射的哈希桶自 由链表下标i。

2. 如果自由链表_freeLists[i]中有对象，则直接Pop一个内存对象返回。

3. 如果_freeLists[i]中没有对象时，则批量从central cache中获取一定数量的对象，插入到自由链表 并返回一个对象。

释放内存：

  1. 当释放内存小于256k时将内存释放回thread cache，计算size映射自由链表桶位置i，将对象Push 到_freeLists[i]。 
  2. 当链表的长度过长，则回收一部分内存对象到central cache。

// thread cache本质是由一个哈希映射的对象自由链表构成
class ThreadCache
{
public:
	// 申请和释放内存对象
	void* Allocate(size_t size);
	void Deallocate(void* ptr, size_t size);
	// 从中心缓存获取对象
	void* FetchFromCentralCache(size_t index, size_t size);
	// 释放对象时，链表过长时，回收内存回到中心缓存
	void ListTooLong(FreeList& list, size_t size);
private:
	FreeList _freeLists[NFREELIST];
};
// TLS thread local storage TLS技术，使每个线程里的ThreadCache*独享，互不影响，实现高并发;
static _declspec(thread) ThreadCache* pTLSThreadCache = nullptr;

// 管理切分好的小对象的自由链表
class FreeList
{
public:
	void Push(void* obj)
	{
		assert(obj);
		// 头插
		/
	auto ret = (Span*)_idSpanMap.get(id);
	assert(ret != nullptr);
	return ret;
}

//从central向page归还span大块内存（这个Span的 usecount==0了即分出去的小obj都还回来了）,归还Span并尝试合并！
void PageCache::ReleaseSpanToPageCache(Span* span)
{
	// 大于128 page的直接还给堆
	if (span->_n > NPAGES - 1)
	{
		void* ptr = (void*)(span->_pageId << PAGE_SHIFT);
		SystemFree(ptr);
		//delete span;
		_spanPool.Delete(span);

		return;
	}

	// 对span前后的页，尝试进行合并，缓解内存碎片问题
	while (1)
	{
		PAGE_ID prevId = span->_pageId - 1;
		
		// 前面的页号没有，不合并了
		auto ret = (Span*)_idSpanMap.get(prevId);
		if (ret == nullptr)
		{
			break;
		}

		// 前面相邻页的span在使用，不合并了
		Span* prevSpan = ret;
		if (prevSpan->_isUse == true)
		{
			break;
		}

		// 合并出超过128页的span没办法管理，不合并了
		if (prevSpan->_n + span->_n > NPAGES - 1)
		{
			break;
		}

		span->_pageId = prevSpan->_pageId;
		span->_n += prevSpan->_n;

		_spanLists[prevSpan->_n].Erase(prevSpan);
		//delete prevSpan;
		_spanPool.Delete(prevSpan);
	}

	// 向后合并
	while (1)
	{
		PAGE_ID nextId = span->_pageId + span->_n;
		

		auto ret = (Span*)_idSpanMap.get(nextId);
		if (ret == nullptr)
		{
			break;
		}

		Span* nextSpan = ret;
		if (nextSpan->_isUse == true)
		{
			break;
		}

		if (nextSpan->_n + span->_n > NPAGES - 1)
		{
			break;
		}

		span->_n += nextSpan->_n;

		_spanLists[nextSpan->_n].Erase(nextSpan);
		//delete nextSpan;
		_spanPool.Delete(nextSpan);
	}
	//合并的大内存块 插回Pagecache中 设置为未使用(未分到central中)状态;
	_spanLists[span->_n].PushFront(span);
	span->_isUse = false;
	//_idSpanMap[span->_pageId] = span;
	//_idSpanMap[span->_pageId+span->_n-1] = span;

	//记录前后页号方便下次合并;
	_idSpanMap.set(span->_pageId, span);
	_idSpanMap.set(span->_pageId + span->_n - 1, span);
}

ConcurrentAlloc.h

#pragma once

#include "Common.h"
#include "ThreadCache.h"
#include "PageCache.h"
#include "ObjectPool.h"

//tcmallo
static void* ConcurrentAlloc(size_t size)
{
	//一次申请内存大于thread最大的256kb，则直接向page层按页直接申请，不需要经过Thread层;
	if (size > MAX_BYTES)
	{
		size_t alignSize = SizeClass::RoundUp(size);
		size_t kpage = alignSize >> PAGE_SHIFT;

		PageCache::GetInstance()->_pageMtx.lock();
		Span* span = PageCache::GetInstance()->NewSpan(kpage);
		span->_objSize = size;//这个span就是为了_objsize单位内存而申请的
		PageCache::GetInstance()->_pageMtx.unlock();

		void* ptr = (void*)(span->_pageId << PAGE_SHIFT);
		return ptr;
	}
	else
	{
		// 通过TLS 每个线程无锁的获取自己的专属的ThreadCache对象
		if (pTLSThreadCache == nullptr)
		{
			static ObjectPool tcPool;//只需要一个tcpool，用于调用New，所以设置成静态;
			//pTLSThreadCache = new ThreadCache;
			pTLSThreadCache = tcPool.New();
		}

		return pTLSThreadCache->Allocate(size);
	}
}

//tcfree
static void ConcurrentFree(void* ptr)
{
	Span* span = PageCache::GetInstance()->MapObjectToSpan(ptr);
	size_t size = span->_objSize;

	if (size > MAX_BYTES)//证明这个Span直接是按页从page申请的_objsize>MAX_SIZE，没经过thread，那么直接ReleaseSpanToPageCache
	{
		PageCache::GetInstance()->_pageMtx.lock();
		PageCache::GetInstance()->ReleaseSpanToPageCache(span);
		PageCache::GetInstance()->_pageMtx.unlock();
	}
	else
	{
		assert(pTLSThreadCache);
		pTLSThreadCache->Deallocate(ptr, size);
	}
}

5.测试文件： Test.cpp

#include"ConcurrentAlloc.h"

// ntimes 一轮申请和释放内存的次数
// rounds 轮次
void BenchmarkMalloc(size_t ntimes, size_t nworks, size_t rounds)
{
	std::vector vthread(nworks);
	std::atomic malloc_costtime = 0;
	std::atomic free_costtime = 0;

	for (size_t k = 0; k < nworks; ++k)
	{
		vthread[k] = std::thread([&, k]() {
			std::vector v;
			v.reserve(ntimes);

			for (size_t j = 0; j < rounds; ++j)
			{
				size_t begin1 = clock();
				for (size_t i = 0; i < ntimes; i++)
				{
					//v.push_back(malloc(16));
					v.push_back(malloc((16 + i) % 8192 + 1));
				}
				size_t end1 = clock();

				size_t begin2 = clock();
				for (size_t i = 0; i < ntimes; i++)
				{
					free(v[i]);
				}
				size_t end2 = clock();
				v.clear();

				malloc_costtime += (end1 - begin1);
				free_costtime += (end2 - begin2);
			}
		});
	}

	for (auto& t : vthread)
	{
		t.join();
	}

	printf("%u个线程并发执行%u轮次，每轮次malloc %u次: 花费：%u msn",
		nworks, rounds, ntimes, malloc_costtime.load());

	printf("%u个线程并发执行%u轮次，每轮次free %u次: 花费：%u msn",
		nworks, rounds, ntimes, free_costtime.load());

	printf("%u个线程并发malloc&free %u次，总计花费：%u msn",
		nworks, nworks*rounds*ntimes, malloc_costtime.load() + free_costtime.load());
}


// 单轮次申请释放次数 线程数 轮次
void BenchmarkConcurrentMalloc(size_t ntimes, size_t nworks, size_t rounds)
{
	std::vector vthread(nworks);
	std::atomic malloc_costtime = 0;
	std::atomic free_costtime = 0;

	for (size_t k = 0; k < nworks; ++k)
	{
		vthread[k] = std::thread([&]() {
			std::vector v;
			v.reserve(ntimes);

			for (size_t j = 0; j < rounds; ++j)
			{
				size_t begin1 = clock();
				for (size_t i = 0; i < ntimes; i++)
				{
					//v.push_back(ConcurrentAlloc(16));
					v.push_back(ConcurrentAlloc((16 + i) % 8192 + 1));
				}
				size_t end1 = clock();

				size_t begin2 = clock();
				for (size_t i = 0; i < ntimes; i++)
				{
					ConcurrentFree(v[i]);
				}
				size_t end2 = clock();
				v.clear();

				malloc_costtime += (end1 - begin1);
				free_costtime += (end2 - begin2);
			}
		});
	}

	for (auto& t : vthread)
	{
		t.join();
	}

	printf("%u个线程并发执行%u轮次，每轮次concurrent alloc %u次: 花费：%u msn",
		nworks, rounds, ntimes, malloc_costtime.load());

	printf("%u个线程并发执行%u轮次，每轮次concurrent dealloc %u次: 花费：%u msn",
		nworks, rounds, ntimes, free_costtime.load());

	printf("%u个线程并发concurrent alloc&dealloc %u次，总计花费：%u msn",
		nworks, nworks*rounds*ntimes.load(), malloc_costtime + free_costtime.load());
}

int main()
{
	size_t n = 10000;
	cout << "==========================================================" << endl;
	BenchmarkConcurrentMalloc(n, 4, 10);
	cout << endl << endl;

	BenchmarkMalloc(n, 4, 10);
	cout << "==========================================================" << endl;

	return 0;
}

malloc与tcmallo测试对比结果

总结

在整体框架编写完毕后，进行测试对比并不能体现tcmalloc比malloc效率高，如下图;但是在tcmalloc进行 map(Span*，PAGE_ID)，这个映射采用基数树的优化后效率大大提高，超过了malloc。

不高效存在性能瓶颈的原因

根据测试，map的find越慢，会导致其lock（）越慢，因此map中的find占了整体的15%，加锁解锁的过程占据了程序运行的20%times，这就是未优化的性能瓶颈；优化前Span*和page_id的映射使用stl标准库中的unorder_map，是非线程安全的，需要加锁保证线程安全; 优化了后使用PAGE_MAP 设计原理是直接开满page_id所有范围的空间，采取直接映射;这样即便是多线程也不需要加锁;

不加锁的原因

1.两个线程对map中不同的位置读写：因为两个线程在对基数树map一个读，一个写的时候，相应空间早都开好了，不会改变结构，因此读写互不影响,一个在自己的地方读，一个在自己的地方写，而STL容器中map红黑树写入会旋转，unorder_map哈希写入也涉及增容，结构有可能会改变，因此两个位置的读写有可能互相影响导致线程不安全;

2.两个线程对map中相同的位置写：因为写只会在NewSpan和ReleaseSpanToPageCache中，一个是central没有span了从pagecache申请，一个是central 把没使用的usecount==0的span大块内存还给paghecache，因为page只有有span和无span两种状态这两个函数不可能在两个线程中同时执行

3.两个线程对map中相同的位置读：因为读只会在ConcurrentFree和ReleaseListToSpans中，对于同一个位置index，肯定是先ConcurrentFree（void*obj），之后objfree到一定的量（_freeLists[index].Size() >= _freeLists[index].MaxSize(慢开始的maxsize)）时，再调用ReleaseListToSpans将freelist[index]从threadcache还给centralcache的，也不会两个线程同时发生;

• stl的map中的find会依次遍历O（n），而直接映射的时间效率是O（1），相当于以部分空间换取时间的一种举措；

优化了stl中map.find()慢的问题 同时不需要加锁了，两个主要问题在牺牲点空间的情况下完美解决,因此性能瓶颈得到优化，tcmalloc高效与malloc;

项目源代码Github链接