kylin源码分析-内存池

Posted 2021-04-05 猿马

一、内存池类图

CBufPoolV：pool基类

CTranBufPool：内存分配器

TBufPool：通用线程安全固定大小内存分配器

TObjectPool：继承自TBufPool的模板类

CBufHandlePool：继承自TObjectPool<BufHandle>，内部包含两个CTranBufPool，通过BufHandle操作管理分配出的内存。

二、代码拆解

接下来依次分析几个类。

1.CBufPoolV：主要定义一个统一的基类，继承类用到的变量主要为m_nUnitSize

class CBufPoolV{public: DLINK link;char m_name[32];int m_nCategory; uint32 m_nUnitSize;  CBufPoolV(const char* pName, int nCategory);virtual ~CBufPoolV(); virtual int GetAllocCount() { return 0; }virtual int GetFreeCount() { return 0; }virtual void GarbageCollect() { } virtual void Print() { ... }};

2.TBufPool：带锁简单内存分配器实现，定义如下

template <class TLock=CNoLock>class TBufPool : public CBufPoolV{ uint32 m_nAlloc/*设置每次分配多少个单位块*/, m_nCur/*当前分配了几次*/, m_nMax/*最多分配几次*/, m_nStart/*设置起始分配次数*/; uint32 m_nSize; /*设置单位分配块大小*/ TLock m_Lock;  #pragma pack(push, r1, 1) struct AllocUnit { SLINK link; char buf[0]; }; #pragma pack(pop, r1)  TSingleList<AllocUnit> m_FreeList; TSingleList<AllocUnit> m_Pointers;  bool AllocOnce() { ... } public: TBufPool(char* name, int nCategory, uint32 nSize) : CBufPoolV(name, nCategory) { m_nAlloc = m_nCur = 0; m_nMax = (uint32)-1; m_nSize = nSize; m_nStart = 0; } TBufPool(char* name, int nCategory, uint32 nSize, uint32 nAlloc, uint32 nStart, uint32 nMax=(uint32)-1) : CBufPoolV(name, nCategory) { m_nSize = nSize; Create(nAlloc, nStart, nMax); } virtual ~TBufPool() { Destroy(false); }  bool Create(uint32 nAlloc, uint32 nStart, uint32 nMax=(uint32)-1) { ... }  bool Destroy(bool /*bCheck*/) { m_FreeList.Init();  AllocUnit* pUnit; while (NULL != (pUnit=m_Pointers.pop())) { free(pUnit); } m_nCur = 0; return true; }  void* Allocate() { ... }  void* ZeroAllocate() { void* pBuf = Allocate(); if (pBuf) { ZeroMemory(pBuf, m_nSize); } return pBuf; }  void Free(void* pBuf) { ... }  int GetAllocCount() { return m_nAlloc * m_nCur; }  int GetFreeCount() { int m;  m_Lock.Lock(); m = m_FreeList.size(); m_Lock.Unlock(); return m; }  void GarbageCollect() { m_Lock.Lock(); if (m_nCur!=m_nStart && (uint32)m_FreeList.size()==m_nAlloc*m_nCur) { Destroy(false); Create(m_nAlloc, m_nStart, m_nMax); } m_Lock.Unlock(); }};

AllocOnce：私有函数，被其他成员函数调用进行单次分配。每一个内存分配单元为AllocUnit，可见每个内存块前都隐含分配一个单链结构SLINK用于进行内存块的管理。AllocUnit中通过定义一个0size的char数组实现可变数组（C-Hack，参考https://stackoverflow.com/questions/295027/array-of-zero-length）。

bool AllocOnce() { SHARE_ASSERT(m_nAlloc != 0); SHARE_ASSERT(m_nCur <= m_nMax); AllocUnit* pUnit = (AllocUnit*)malloc(m_nAlloc * m_nUnitSize + sizeof(SLINK));  if (pUnit) { m_Pointers.push(pUnit); m_nCur ++;  pUnit = MAKE_PTR(AllocUnit*, pUnit, sizeof(SLINK)); m_FreeList.pushn(m_nAlloc, pUnit, m_nUnitSize); return true; } PERROR("malloc"); return false;}

‍Create：主要做参数设置，如果设置了nStart非零，则需调用AllocOnce nStart次，先进行nStart次分配。如果构造函数未传递赋值m_nAlloc，则使用BufPool前需先调用Create赋值m_nAlloc。

bool Create(uint32 nAlloc, uint32 nStart, uint32 nMax=(uint32)-1) { SHARE_ASSERT2(nStart, <=, nMax); SHARE_ASSERT2(0, !=, nAlloc); m_nAlloc = nAlloc; m_nMax = nMax;  m_nStart = nStart; m_nUnitSize = m_nSize + sizeof(SLINK);  for (m_nCur=0; m_nCur<m_nStart; ) { if (!AllocOnce()) { return false; } } return true;}

Allocate：带一次重试分配。第一次如果从m_FreeList pop非空则直接返回，否则在当前分配次数低于最大分配次数时调用AllocOnce再进行一次分配（分配成功进行到重试，失败直接返回NULL）；第二次仍尝试从m_FreeList pop，非空直接返回，否则返回NULL。

void* Allocate() { SHARE_ASSERT_NOT_EQUAL(0, m_nAlloc); m_Lock.Lock(); for (int i=0; ; i++) { AllocUnit* pUnit = m_FreeList.pop(); if (NULL != pUnit) { m_Lock.Unlock(); return &pUnit->buf; } if (m_nCur>=m_nMax || 0!=i || !AllocOnce()) { break; } } m_Lock.Unlock(); return NULL;}

ZeroAllocate：分配同Allocate，额外进行内存赋0操作。

void* ZeroAllocate() { void* pBuf = Allocate(); if (pBuf) { ZeroMemory(pBuf, m_nSize); } return pBuf;}

Free：内存归还操作，直接将内存单元插入m_FreeList中。

void Free(void* pBuf) { AllocUnit* pUnit = CONTAINING_RECORD(pBuf, AllocUnit, buf);// CONTAINING_RECORD为宏定义，#define CONTAINING_RECORD(address, type, field) ((type *)((char*)(address) - offsetof(type, field)))  m_Lock.Lock(); m_FreeList.push(pUnit); m_Lock.Unlock();}

3.TObjectPool：继承自TBufPool的对象池模板类。重写了Allocate及ZeroAllocate两个方法。

template <typename TObj, class TLock=CNoLock>class TObjectPool : public TBufPool<TLock>{public: TObjectPool(char* name, int nCategory) : TBufPool<TLock>(name, nCategory, sizeof(TObj)) {} TObjectPool(char* name, int nCategory, uint32 nAlloc, uint32 nStart, uint32 nMax=(uint32)-1) : TBufPool<TLock>(name, nCategory, sizeof(TObj), nAlloc, nStart, nMax) { } ~TObjectPool() { }   TObj* Allocate() { return (TObj*)TBufPool<TLock>::Allocate(); }  TObj* ZeroAllocate() { return (TObj*)TBufPool<TLock>::ZeroAllocate(); }};

4.CTranBufPool：更加复杂化的内存分配器实现，相较TBufPool主要支持连续内存块的分配。

class CTranBufPool : public CBufPoolV{private: struct Handle { DLINK link; char* pBuffer; Handle* pRealHdl; //单内存块pRealHdl=this，连续内存块pRealHdl=首内存块handle地址 int nRef; int nConsBuf; }; typedef TLinkedList<Handle> FreeList; typedef std::map<char*, Handle*> BufferMap;  FreeList m_FreeList; BufferMap m_BufferMap; int m_nBlockSize/*单个内存块大小*/, m_nBufferSize/*单次分配内存大小=m_nBlockSize*m_nAlloc*/, m_nBlockBase/*单个内存块大小的对数*/; int m_nAlloc/*单次分配内存块数目*/, m_nMaxBuffers/*最大内存块数=m_nMax*m_nAlloc*/, m_nBuffers/*当前内存块数*/, m_nWaterMarks[3]/*三个不同优先级watermark*/; int m_nMin/*最小分配次数*/, m_nMax/*最大分配次数*/; double m_fRatio1, m_fRatio2;  bool AllocOnce() { ... }  Handle* GetHandle(char* pBuffer) { ... }  void Destroy() { m_FreeList.Init(); m_nBuffers = 0;  BufferMap::iterator it; for (it=m_BufferMap.begin(); it!=m_BufferMap.end(); it++) { free(it->first); free(it->second); } m_BufferMap.clear(); } public: CTranBufPool(const char* name, int nCategory) : CBufPoolV(name, nCategory) { m_nBuffers = 0; m_nBlockSize = m_nBufferSize = m_nAlloc = m_nMaxBuffers = m_nMin = 0; m_nWaterMarks[0] = m_nWaterMarks[1] = m_nWaterMarks[2] = 0; m_fRatio1 = m_fRatio2 = 0; }  bool Create(int nBlockSize, int nAlloc, int nMin, int nMax, double fRatio1, double fRatio2) { ... }  ~CTranBufPool() { Destroy(); }  char* Allocate(uint32 nPriority, int count=1) { ... }  int AddRef(char* p, bool bCheck=false) { Handle* pHdl = GetHandle(p); if (NULL == pHdl) { if (!bCheck) { return -1; } RaiseError(Invalid_Block); }  int n = ++ pHdl->pRealHdl->nRef; ASSERT(2 <= n); return n; }  int Free(char* p, bool bCheck=false) { ... }  void Enlarge(int nNewTranBuf) { ... }  int GetFreePercent() const { int n = (int)m_FreeList.size(); int n1 = m_nMaxBuffers-m_nBuffers+n; return (n1*100) / m_nMaxBuffers; }  int GetAllocCount() { return m_nBuffers; } int GetFreeCount() { return m_FreeList.size(); } void GarbageCollect() {} void _GarbageCollect() { if (m_nBuffers!=m_nMin*m_nAlloc && (int)m_FreeList.size()==m_nBuffers) { Destroy(); Create(m_nBlockSize, m_nAlloc, m_nMin, m_nMax, 0, 0); } }  int GetBlockSize() const { return m_nBlockSize; }  /* return the maximum capacity of the pool. */ int GetTotalBufNum() const { return m_nMaxBuffers; }  /* return the allocated buffers' number, including FREED or be USING. * !!NOTICE: Make sure you know the meaning of 'm_nBuffers' when use. */ int GetAllocatedBufNum() const { return m_nBuffers; }  /* return the number of buffers free to use. */ int GetFreeBufNum() const { return m_nMaxBuffers-m_nBuffers+m_FreeList.size(); }};

‍AllocOnce：私有成员函数，被其他成员函数调用，进行一次内存分配。分配m_nBufferSize大小的内存块及m_nAlloc个Handle用于管理分配出来的m_nAlloc个内存块。

bool AllocOnce() { char* pBuffer = (char*)AlignAlloc(m_nBlockSize, m_nBufferSize); Handle* pHdl = (Handle*)ZeroAlloc(m_nAlloc * sizeof(Handle)); if (pBuffer && pHdl) { m_BufferMap.insert(BufferMap::value_type(pBuffer, pHdl)); m_nBuffers += m_nAlloc; pBuffer += m_nBufferSize - m_nBlockSize; pHdl += m_nAlloc - 1;  for (int i=0; i<m_nAlloc; i++) { pHdl->pBuffer = pBuffer; pHdl->nRef = 0; pHdl->nConsBuf = i+1; pHdl->pRealHdl = pHdl; m_FreeList.push_back(pHdl);  pBuffer -= m_nBlockSize; pHdl --; } return true; } if (pBuffer) free(pBuffer); if (pHdl) free(pHdl); return false;}

GetHandle：获取对应内存块的Handle

Handle* GetHandle(char* pBuffer) { BufferMap::iterator it = m_BufferMap.upper_bound(pBuffer); if (it != m_BufferMap.begin()) { it --;  char* pHead = it->first; ASSERT(pHead <= pBuffer); if (pBuffer < pHead + m_nBufferSize) { int n = (pBuffer-pHead) >> m_nBlockBase; Handle* pHdl = it->second + n; ASSERT(pHdl->pBuffer == pHead + (((uint32)n) << m_nBlockBase)); return pHdl; } } return NULL;}

Create：主要进行各参数初始化，如果设置了最小分配次数m_nMin，还需调用m_nMin次AllocOnce，进行初始分配。

bool Create(int nBlockSize, int nAlloc, int nMin, int nMax, double fRatio1, double fRatio2) { m_nUnitSize = nBlockSize; m_nBlockSize = nBlockSize; m_nBlockBase = Log_2(nBlockSize); if (-1 == m_nBlockBase) { return false; } m_nAlloc = nAlloc; m_nMaxBuffers = nMax * nAlloc; m_nBufferSize = m_nBlockSize * m_nAlloc; m_nBuffers = 0; m_nMax = nMax; m_fRatio1 = fRatio1; m_fRatio2 = fRatio2; m_nMin = nMin; if (0!=fRatio1 && 0!=fRatio2) { m_nWaterMarks[0] = (int)((double)m_nMaxBuffers * fRatio1); m_nWaterMarks[1] = (int)((double)m_nMaxBuffers * fRatio2); m_nWaterMarks[2] = m_nMaxBuffers-1; } for (int i=0; i<m_nMin; i++) { if (!AllocOnce()) return false; } return true;}

Allocate：内存分配入口，带一次重试

#define _ALLOC_TRAN_BUF(p, how) \ p = m_FreeList.how(); \ ASSERT(DLINK_IS_STANDALONE(&p->link)); \ ASSERT(0 == p->nRef); \ ASSERT(p->pRealHdl == p); \ p->nRef = 1 char* Allocate(uint32 nPriority, int count=1) { int n; ASSERT(0 != count); for (int i=0; i<2; i++) { n = (int)m_FreeList.size(); if (m_nBuffers-n > m_nWaterMarks[nPriority]) { // 超出对应waterMark直接返回NULL return NULL; } if (n >= count) { Handle *pHdl, *pTmp; if (1 == count) { _ALLOC_TRAN_BUF(pHdl, pop_front); return pHdl->pBuffer; } // Big block are formed by multiple consecutive blocks. // We try from the tail of free list, which brings higher probability. _ALLOC_TRAN_BUF(pHdl, pop_back); int i = 1; if (pHdl->nConsBuf >= count) { for ( ; i<count; i++) { pTmp = pHdl + i; UNLIKELY_IF (0 != pTmp->nRef) { break; } m_FreeList.remove(pTmp); DLINK_INSERT_PREV(&pHdl->link, &pTmp->link); pTmp->pRealHdl = pHdl; pTmp->nRef = 1; } } if (i == count) { // 找到连续内存块，返回首地址。 return pHdl->pBuffer; } else { // 连续内存块校验失败，回退已分配操作 for (int j=0; j<i; j++) { pTmp = pHdl + j; DLINK_INITIALIZE(&pTmp->link); pTmp->pRealHdl = pTmp; pTmp->nRef = 0; m_FreeList.push_front(pTmp); } } } if (m_nBuffers >= m_nMaxBuffers || !AllocOnce()) { // 尚未超出最大内存分配块数的情况下，尝试进行一次新分配 return NULL; } } return NULL;}

Free：归还内存块

#define _FREE_TRAN_BUF(p, how) \ m_FreeList.how(p)#endif  int Free(char* p, bool bCheck=false) { Handle* pHdl = GetHandle(p); if (NULL == pHdl) { if (bCheck) { RaiseError(Invalid_Block); } return -1; }  pHdl = pHdl->pRealHdl; int n = -- pHdl->nRef; if (0 == n) { Handle* pTmp = dlink_get_prev(pHdl); if (pTmp == pHdl) { ASSERT_EQUAL(pHdl->pRealHdl, pHdl); ASSERT_EQUAL(0, pHdl->nRef); _FREE_TRAN_BUF(pHdl, push_front); return 0; } // here comes big block Handle* p = pHdl; do { pHdl = pTmp; pTmp = dlink_get_prev(pTmp); ASSERT_EQUAL(1, pHdl->nRef); ASSERT_EQUAL(p, pHdl->pRealHdl); pHdl->pRealHdl = pHdl; pHdl->nRef = 0; DLINK_INITIALIZE(&pHdl->link); _FREE_TRAN_BUF(pHdl, push_back); } while (p != pTmp); ASSERT_EQUAL(p, p->pRealHdl); ASSERT_EQUAL(0, p->nRef); DLINK_INITIALIZE(&p->link); _FREE_TRAN_BUF(p, push_back); return 0; } return n;}

Enlarge：调大pool的参数，使其能分配更多内存块

void Enlarge(int nNewTranBuf) { UNLIKELY_IF (nNewTranBuf < m_nMaxBuffers) { return; } int nDeltaTranBuf = nNewTranBuf - m_nMaxBuffers; int nDeltaAllocNum = nDeltaTranBuf / m_nAlloc; m_nMax += nDeltaAllocNum; m_nMaxBuffers = m_nMax * m_nAlloc; if (0!=m_fRatio1 && 0!=m_fRatio2) { m_nWaterMarks[0] = (int)((double)m_nMaxBuffers * m_fRatio1); m_nWaterMarks[1] = (int)((double)m_nMaxBuffers * m_fRatio2); m_nWaterMarks[2] = m_nMaxBuffers-1; }}

5.CBufHandlePool：基于CTranBufPool封装的应用层内存分配器。

class CBufHandlePool : public TObjectPool<BufHandle>{ volatile int m_lock; CTranBufPool m_TranBufPool; CTranBufPool m_BigBufPool;  BufHandle* DoAllocate(CTranBufPool* pPool, int nRetry) { BufHandle* pHdl;  for (int i = 0; i < nRetry; i++) { LOCK; pHdl = TObjectPool<BufHandle>::Allocate(); pHdl->pBuf = pPool->Allocate(i > 0 ? 2 : 1); if (NULL == pHdl->pBuf) { TObjectPool<BufHandle>::Free(pHdl); pHdl = NULL; } UNLOCK; if (NULL != pHdl) return pHdl; if (i > 1) { TRACE0("No enough memory, sleep %d\n", i+1); } sleep(1); } RaiseError(TODO_NO_ENOUGH_MEMORY); return NULL; }  BufHandle* DoAllocateCanFail(CTranBufPool* pPool, int nSize) { BufHandle* pHdl; int nBlockSize = pPool->GetBlockSize(); ASSERT(0 != nSize);  LOCK; pHdl = TObjectPool<BufHandle>::Allocate(); if (nSize == nBlockSize) { pHdl->pBuf = pPool->Allocate(0); } else { pHdl->pBuf = pPool->Allocate(0, (nSize+nBlockSize-1) / nBlockSize); } if (NULL == pHdl->pBuf) { TObjectPool<BufHandle>::Free(pHdl); pHdl = NULL; } UNLOCK; return pHdl; }  BufHandle* _DoAddRef(BufHandle* pHdl, BufHandle* pNext, BufHandle*** pppLast) { if (-1!=m_TranBufPool.AddRef(pHdl->pBuf) || -1!=m_BigBufPool.AddRef(pHdl->pBuf)) { BufHandle* pTmp = TObjectPool<BufHandle>::Allocate(); pTmp->_next = pNext; pTmp->pBuf = pHdl->pBuf; pTmp->nBufLen = pHdl->nDataLen; pTmp->nDataLen = pHdl->nDataLen; *pppLast = &pTmp->_next; return pTmp; } return NULL; }  void _DoFree(BufHandle* pHdl) { if (-1 == m_TranBufPool.Free(pHdl->pBuf)) m_BigBufPool.Free(pHdl->pBuf); TObjectPool<BufHandle>::Free(pHdl); } public: CBufHandlePool() : TObjectPool<BufHandle>("BufHandle", BUFPOOL_C2), m_TranBufPool("TranBuffer", BUFPOOL_C1), m_BigBufPool("BigBuffer", BUFPOOL_C1) { m_lock = 0; Create(1024, 1);  int nAlloc = s_nTranBuf; int nMax = 1; while ((s_nBufSize/1024) * nAlloc > 524288) { /* Max alloc: 512M */ nAlloc >>= 1; nMax <<= 1; } m_TranBufPool.Create(s_nBufSize, nAlloc, 1, nMax, s_fLowMark, s_fHighMark); m_BigBufPool.Create(SZ_BIG_BUF, s_nBigTranBuf, 0, 10, 0.9, 0.9); } ~CBufHandlePool() { } BufHandle* AllocateBig(bool bCanFail) { BufHandle* pHdl;  pHdl = bCanFail ? DoAllocateCanFail(&m_BigBufPool, SZ_BIG_BUF) : DoAllocate(&m_BigBufPool, 60); if (pHdl) { pHdl->_next = NULL; pHdl->nBufLen = SZ_BIG_BUF; pHdl->nDataLen = 0; } return pHdl; }  BufHandle* AllocateCanFail(int nSize) { BufHandle* pHdl = DoAllocateCanFail(&m_TranBufPool, nSize); if (pHdl) { pHdl->_next = NULL; pHdl->nBufLen = nSize; pHdl->nDataLen = 0; } return pHdl; }  BufHandle* AllocForBuf(char* pBuf, int nLen, BufHandle* pNext, BufHandle*** pppLast) { BufHandle *pFirst, *pHdl, **ppLast;  pFirst = NULL; ppLast = &pFirst; while (nLen > 0) { pHdl = DoAllocate(&m_TranBufPool, 120);  pHdl->nBufLen = s_nBufSize; pHdl->nDataLen = nLen>s_nBufSize ? s_nBufSize : nLen; memcpy(pHdl->pBuf, pBuf, pHdl->nDataLen); pBuf += pHdl->nDataLen; nLen -= pHdl->nDataLen;  pHdl->_next = pNext; *ppLast = pHdl; ppLast = &pHdl->_next; } if (pppLast) { *pppLast = ppLast; } return pFirst; }  BufHandle* Allocate(bool bInPool=false, char* pBuf=NULL, int nLen=0, BufHandle* pNext=NULL ) { BufHandle* pHdl;  UNLIKELY_IF (false == bInPool) { LOCK; pHdl = TObjectPool<BufHandle>::Allocate(); if (-1 == m_TranBufPool.AddRef(pBuf)) m_BigBufPool.AddRef(pBuf); UNLOCK;  pHdl->_next = pNext; pHdl->pBuf = pBuf; pHdl->nBufLen = nLen; pHdl->nDataLen = nLen; return pHdl; } if (pBuf==NULL || nLen==0) { pHdl = DoAllocate(&m_TranBufPool, 120);  pHdl->_next = pNext; pHdl->nBufLen = s_nBufSize; pHdl->nDataLen = nLen; return pHdl; }  return AllocForBuf(pBuf, nLen, pNext, NULL); }  int GetFreePercent() { int n; LOCK; n = m_TranBufPool.GetFreePercent(); UNLOCK; return n; }  void Free(BufHandle* pHdl) { ASSERT(NULL != pHdl); LOCK; _DoFree(pHdl); UNLOCK; }  void EnlargeTranBuf(int nNewSmallNum) { ASSERT(nNewSmallNum > 0); LOCK; m_TranBufPool.Enlarge(nNewSmallNum); UNLOCK; }  // pNext is not included void ChainFree(BufHandle* pHdl, BufHandle* pNext) { BufHandle *pTmp; LOCK; for ( ; pHdl!=pNext; pHdl=pTmp) { ASSERT(NULL != pHdl); pTmp = pHdl->_next; _DoFree(pHdl); } UNLOCK; }  BufHandle* CloneAndTerminate(BufHandle* pHdl, BufHandle* pNext, int* pnLen, bool bCopyNonTranBuf ) { BufHandle *pFirst, *pTmp, **ppLast, **ppLastTmp; int nLen = 0;  pFirst = NULL; ppLast = &pFirst; LOCK; for ( ; pHdl!=pNext; pHdl=pHdl->_next) { pTmp = _DoAddRef(pHdl, NULL, &ppLastTmp); if (NULL == pTmp) { if (bCopyNonTranBuf) { UNLOCK; pTmp = AllocForBuf(pHdl->pBuf, pHdl->nDataLen, NULL, &ppLastTmp); LOCK; } else { pTmp = TObjectPool<BufHandle>::Allocate(); pTmp->pBuf = pHdl->pBuf; pTmp->nDataLen = pTmp->nBufLen = pHdl->nDataLen; pTmp->_next = NULL; ppLastTmp = &pTmp->_next; } } nLen += pHdl->nDataLen; *ppLast = pTmp; ppLast = ppLastTmp; } UNLOCK;  if (pnLen) { *pnLen = nLen; } if (nLen) { return pFirst; } ChainFreeHdl(pFirst, NULL); return NULL; }  void GarbageCollect() { LOCK; TObjectPool<BufHandle>::GarbageCollect(); m_TranBufPool._GarbageCollect(); m_BigBufPool._GarbageCollect(); UNLOCK; }  CTranBufPool* GetTranBufPool() { return &m_TranBufPool; } CTranBufPool* GetBigBufPool() { return &m_BigBufPool; }};