詳談記憶體管理技術（二）、記憶體池

嗯，這篇講可用的多線程記憶體池。零、上期彩蛋：不要重載全局new 或許，是一次很不愉快的經歷，所以在會有這麼一個“認識”。反正，大概就是：即使你足夠聰明，也不要自作聰明；在這就是不要重載全局new，無論你有著怎樣的目的和智商。因為：這是一個不對稱的介面：只告訴了我們如何創建一個【堆】對象，但是釋放 ...

　　嗯，這篇講可用的多線程記憶體池。

零、上期彩蛋：不要重載全局new

　　或許，是一次很不愉快的經歷，所以在會有這麼一個“認識”。反正，大概就是：即使你足夠聰明，也不要自作聰明；在這就是不要重載全局new，無論你有著怎樣的目的和智商。因為：

class XXX{
public:
    XXX* createInstance();
};

　　這是一個不對稱的介面：只告訴了我們如何創建一個【堆】對象，但是釋放呢？？！　　很無奈，只能假設其使用全局預設的delete來刪除（除此之外，沒有其他選擇）；這時，我為了某些目的，重載了全局delete（或許是為了監視、為了優化、為了...）：

void operator delete(void* addr)
{
    ...
    //一些事情發生了
    ...
    std::free(addr);
}

　　這是一種很自然的做法；但是，但是會崩的，在未來或現在的某日；其名為：堆錯誤。也就是崩在了堆上，原因也很簡單：代碼中有誰並不是使用std::malloc來分配記憶體的——比如說前面的那個【XXX】，我們誰也不知道它是分配在那個堆上面的：是預設的系統堆，還是VS-debug中的調試堆（此為坑點）。

　　當然，我們可以足夠小心；比如仔細考察每個對象的分配方式，對於非我們自己new出來的，給予特別的關懷。或者，也可以這樣：

void operator delete(void* addr)
{
    ...
    //一些事情又發生了
    ...
    ::operator delete(add);
}

　　我們最後使用了原來的全局釋放方式；嗯，這是一種安全的方式。當然，這樣的話，你可以自定義的部分，只有一種：監視。你不能夠通過自定義的記憶體分配方式（比如將要講的記憶體池），來優化。當然，如果沒有那個【XXX】來攪局就好了；但，作為【全局】的操作，一旦修改了，你必須給予極其健壯的支持和保證。

　　最後，因為是全局而隱式調用，你並不能夠完全地控制，該操作什麼時候是一定會被調用，什麼時候卻沒有被調用（當你的代碼作為lib/dll庫被調用，而new重載沒有被導出）；假如，有同樣一個聰明的人也重載了，那麼當需要混用代碼時（如lib/dll），你會覺得整個世界都不好了.......

　　當然，這是個人見解；如果你執意用，建議使用巨集的方式，去調用非全局版本的等價物。

一、什麼是記憶體池？

　　嗯，就是下麵一坨代碼：

struct BlockNode{
    BlockNode* next;
};

//創建一堆BlockNode
BlockNode* allocate(size_t index, size_t count);

//對外的分配記憶體介面
void* alloc(size_t size)
{
    size_t index = (size - 1)/8 + 1;
    BlockNode* data = freeList[index];
    if(data){
        freeList[index] = data->next;
        return data;
    }
    else{
        freeList[index] = allocate(index, /*一個合理的大數*/);
        return alloc(size);
    }
}

//對外的釋放記憶體介面
void dealloc(void* addr, size_t size)
{
    //與alloc相反的操作（我懶）
}

　　以上就是記憶體池本身的所有細節；至於allocate和dealloc，不難想象出來。

二、什麼是多線程記憶體池？

　　直白的翻譯就是：支持多線程安全操作的記憶體池。當然，我們不能夠通過加鎖的方式來獲得安全；否則，我們只會做的更糟（會比系統的慢....可能）。

　　所以，這裡我們需要用到下一篇將要講到的技術之一：TLS（線程本地存儲）。是的，我們將在每一個線程里創建這麼一個記憶體池；這樣，便不需要鎖就能夠自然地獲得記憶體分配時的線程安全。那麼，釋放時呢？

//發生於線程A
void* addr = alloc(23);
...
...
//這裡面發生很多很多的事情
...
...
dealloc(addr, 23);//這是哪裡?是線程A嗎?

　　如果，你的代碼支持多線程；那麼，記憶體釋放的時候，其絕對不會一定在原來分配時的線程！當然，我們可以將每段記憶體打上標記，來指明其出生在那個線程。嗯這是一個不錯的失敗的嘗試；首先，其接下來的複雜度就會將你打垮（在不同的線程釋放時，如何回到分配線程），其次，如果分配線程死了呢？？？（我相信聰明如斯的你，總會有辦法的....）

　　所以，我們需要一個合理且有效的模型：線程間記憶體池交換記憶體的模型——使用一個全局的共用記憶體池，然後各個線程內部的記憶體池，向其發起分配和釋放的請求。這樣，我們也就不在擔心上面的問題了；我們可以通過這個全局池，來完成跨線程間的記憶體操作。

　　當然，全局池需要加鎖，這點毋庸置疑。為了減少加鎖的消耗，我們可以縮短線程內部池的訪問頻率，比如：內部池的分配/釋放頻率與全局池的訪問頻率，比例在：10000:1，或更高。這樣，通過均攤，最後加鎖的消耗，幾乎完全沒有了（即使消耗1ms，現在均攤後也只有0.1us）。

　　所以，現在的挑戰就是：如何維持這個均攤比例？（因為，在畸形的分配中，會變成1:1甚至更低）

三、我們需要性能！

　　沒有更好的性能，那我們還造毛？？所以，這裡，最大的目的是保持住，我們預定的均攤比例。或者說，控制住線程內部池向全局池的訪問頻率。

　　對於向全局池的分配申請策略；我們可以使用一個足夠大的申請值：比如100000，或者10MB。

BlockNode* ThreadMemoryPool::allocate(size_t index, size_t size)
{
    //我們直接申請一個夠大的
    BlockNode* result = globalPool.allocate(100000*8*(index + 1));
    return result;
}

　　那麼，在什麼時候釋放呢？比如，現線上程A申請了10MB的記憶體，那麼在怎樣的情況下才釋放？釋放什麼？釋放多少？這一直以來是一個盲區（對於我來說，數年來都沒完全解決）。當然，聰明的你或許，馬上就有了各種的方案。

　　我們，為什麼要釋放？？？因為，其他線程沒有記憶體可用了；而你，線程A正持有著100TB的記憶體。

四、我們需要均衡...

　　我們不能夠容忍，任何一個線程持有超過10MB的可用記憶體！！！所以，有瞭如下的方案：

void ThreadMemoryPool::dealloc(void* addr, size_t size)
{
    .....
    //我們就不要在意釋放過程了
    ...
    if(listSize[index] > 1024*1024*10){
        globalPool.deallocte(freeList[index], listsize[index]);
    }
}

　　線上程內部池的釋放操作時，檢測當前池是否有超過10MB的記憶體；如果有，那麼我們就墮掉它！這時，便會有一個畸形的分配情形：

//線程A向全局池申請10MB
threadMemoryPool.allocate(index, 10MB);
//並內部消耗一個單位（當前持有10MB - 一個單位）
threadMemoryPool.alloc(...);
...
//線程A內部釋放一個單位（當前持有10MB）
threadMemoryPool.dealloc(...);
...
//線程A內部釋放一個單位（當前持有10MB + 一個單位 > 10MB）
threadMemoryPool.dealloc(...);

　　總共進行了3次分配/釋放操作，便向全局池返回了所申請的記憶體。這時均攤比例為3:2（內部操作3次，全局操作2次：申請+釋放），其次全局池本身的任何操作都是消耗巨大的（比如那10MB記憶體是從何而來的，從系統），那麼這個實際的比例會變成1:100甚至更低。

　　當然，我們可以錯開分配和釋放的全局操作閾值，比如：分配1MB、釋放10MB。這樣，我們就有了10-1=9MB的餘地，不會發生上面的情形。（當然，反過來絕對不行：分配10MB、釋放1MB，可以自己想象。）

五、我們還需要什麼？

　　如果分配值和釋放閾值不相等，那麼，我們就有可能永遠也沒有機會回收小於釋放閾值，但大於等於分配值的那部分記憶體。在最常見的情況下：線程A的所有分配釋放操作，都在本地進行。

//線程A
for(i = 0 : 10000){
    data[i] = threadMemoryPool.alloc(size);
}
....
...
//線程A
for(i = 0 : 10000){
    threadMemoryPool.dealloc(data[i], size);
}
//沒了

　　在這之後在沒有任何操作，那麼，直到線程死亡；我們都不可能回收這段可用的記憶體！

　　所以，我們需要分配值，足夠的合理；也需要釋放閾值足夠的小，且能夠維持均攤比例。當然，我們可以辦到！我們只需要完全隔離當前內部池的持有的【分配】記憶體值，和【已釋放】記憶體值。

void ThreadMemoryPool::dealloc(void* addr, size_t size)
{
    .....
    //我們依舊不要在意釋放過程
    ...
    deadSize[index] += (index + 1)*8;//使用了一個額外的死亡記憶體值
    if(deadSize[index] > 1024*1024*10){
        globalPool.deallocate(freeList[index], listsize[index]);
    }
}

　　如此簡單，卻困擾了我如此之久.....這時，我們就可以隨意地操作分配值和釋放閾值；以維持一個我們所認為的合理的均攤比例。

六、我們還需要什麼？？

　　可能有註意到了，我們維持了一種假象：我們的記憶體池可以回收。

　　1、我們不能夠向系統釋放我們可能不會再用到的記憶體。（這對沒有使用我們的記憶體池的部分代碼和系統本身而言，就是記憶體泄漏！）

　　2、可能大家有註意到了這個【index】，每個內部池，我們維護了數個不同大小節點鏈。而這些不同大小的鏈之間，我們是沒有辦法重覆使用的。（這是記憶體池內部的泄露）

　　是的，我們可能正在製造最大規模的記憶體泄漏；還是我們以一種不可避免的方式造成的（我們要用性能更高的記憶體分配）。從理論上來說，這是不可避免的；我們唯一能夠做到的是，儘量避免上面的情況，演化成最糟糕的局面。

　　所以，我們可能需要更加精細的模型了；我們要改造【全局池】！！使其能夠支持一定程度上的：記憶體整理。

　　所以，我們需要做一下兩個改進：

　　1、我們需要保存每一塊從系統分配得到的大記憶體的地址（以可以釋放）。

　　2、我們需要一種演算法，能夠整理我們所有不用的記憶體；讓其恢復到從系統分配時的狀態（完整的大塊記憶體）。

　　這時，我們便可以完成之前的2個目標：向系統釋放、節點鏈間的復用（通過釋放給系統，而後再次從系統獲取）。在我個人的記憶體池中是實現了類似的功能，所以，我相信，做到這點並不困難。

七、我們還需要什麼？？？

　　重要的事情說3遍......

　　回到最開始的問題：為什麼我們需要記憶體池？我們需要性能。所以，還有那些地方，值得我們關註；以獲得更高的性能？

　　有，還有很多！之前的向全局池的釋放部分，有一個關鍵的細節，沒有提到：我們釋放的記憶體存在哪裡？又怎麼復用？有兩種方案：

　　1、如同線程內部池一樣，維護一個鏈，將釋放的部分，加入到鏈中（需要O(n)）；在分配時，從鏈中獲取（同樣需要O(n)）。

　　2、將該節點鏈整個打包，作為一個單獨的鏈保存；在向內部池分配時，直接返回該鏈。（只有O(1)）

　　直觀地，我們會選擇第二種方案（即使，最初我們可能只會想到第一種）。但是，一旦使用第二種，我們將不能夠控制每次線程所申請的記憶體大小：我們只能夠返回一個可用的節點鏈，而並不能夠保證是否是其所期望的大小。（我們最多只能夠儘可能保證返回大於其期望值；而不能保證和期望值一致；否則會破壞O(1)的複雜度）

　　嗯，我們喪失了一小部分控制均攤比例的能力。但，只要我們足夠小心安排釋放閾值，是不會發生什麼畸形的情形。

　　其次，分配時，我們可以再小心一些：不是直接申請1MB（可能只會用到很小一部分），而是按照某種增長策略來申請（比如：100、200、400....）。在能夠維持均攤比例的前提下，我們可以做很多想做的事情。

　　總結一下：我們需要足夠大的分配值和釋放閾值，以維持合理的均攤比例；而，我們又想要保留足夠小的記憶體，以避免任何可能的記憶體泄漏。　　嗯，這正是矛盾之處，也是我們所追求的。而剩下的，就只有一個：權衡。

PS：使用記憶體池後，一旦發生記憶體越界；其後果將是災難性的，對於調試。