有些時候需要給資源分配一個唯一id(32bit or 64bit or CHAR[N])，這里主要說下分配方法問題。

首先我們有個基本前提，如果是單線程分配，那么我們無需下面的方法，直接++value即可(CHAR型無論幾線程都可使用GUID)順序產(chǎn)生不重復(fù)序列，下面討論的方法都是多線程下的分配策略：

方法1、 win下做簡單的方法莫過于使用InterlocckedIncrement(or InterlockedIncrement64)了，這個調(diào)用也很簡單，每次遞增一個，多線程間保證順序遞增絕無重復(fù)。此方法只可在單一進程上使用。

 

方法2、區(qū)間法，每個線程一次申請一個id區(qū)間[m, n]，用完了再申請下一個區(qū)段，申請的時候鎖一次，其他時間都不用鎖，效率比3略低，比1高。此方法也只可在一個進程上使用，當(dāng)然如果申請的策略修改一下也可實現(xiàn)多個進程甚至不同機器上的進程之間獨立分配id。

 

方法3、方法1雖然簡單但畢竟InterlockedXXX系列函數(shù)調(diào)用還是有些耗時的，大概50cpu周期級別，更簡單的方法可以使用線程切分原理，如有3個線程參與id分配，我們這樣分配：

線程1 base=1, step =3，序列1,4,7,10,…

線程2 base=2, step=3，序列2,5,8,11,…

線程3 base=3, step=3，序列3,6,9,12,…

絕無重復(fù)，調(diào)用非常簡單每個線程id = base; base += step;即可。

此方法在單進程上使用很簡單，如果要拓展到多個進程上使用要通過配置來實現(xiàn)，但也是不難的。

 

方法4、如果id可用GUID表示那么方法要簡單一點，生成id直接調(diào)用guid生成算法，這個id生成算法即使在多個進程之間甚至不同機器之間也可以保證唯一，也有其價值。

執(zhí)行100w次的結(jié)果（單位：微妙）
------------------------------------------------------------------------------------
線程 volatile讀 volatile寫 InterlockedInc  CS SRWLock共享 SRWLock獨占 互斥量
  1     8            8               35                 66      66                       67       1060
  2     8           76             153                268    134                   148      11082
  4     9           145           361                768    244                   307      23785

volatile讀 long lvalue = gv_value;

volatile寫 gv_value = 0;

InterlockedIncrement(&gv_value);

EnterCriticalSection(&cs);
gv_value = 0;
LeaveCriticalSection(&cs);

AcquireSRWLockShared/Exclusive(&g_srwlock);
gv_value = 0;
ReleaseSRWLockShared/Exclusive(&g_srwlock);

WaitForSingleObject(&g_hMutex, INFINITE);
gv_value = 0;
RleaseMutex(g_hMutex);

如果希望應(yīng)用程序獲得最佳性能，那么首先應(yīng)該嘗試不要共享數(shù)據(jù)，然后依次使用
volatile讀取，volatile寫入，Interlocked API, SRWLock以及關(guān)鍵段，當(dāng)前僅當(dāng)
所有這些都不滿足要求的時候再使用內(nèi)核對象。

來自Jeffrey Riched《windows核心編程（5）》的數(shù)據(jù)，錄在這里備查。

從2004年使用iocp開始，嘗試過很多種client生存期管理，最初使用
一個鎖userlock，在sendcomp和recvcomp以及senddata等都用一個鎖，而且是
進去的地方就開始鎖，這種方式最簡單，也產(chǎn)生了最初的版本，之后為
追求更高效率，慢慢演變?yōu)槭褂脙蓚€鎖，readlock和writelock，再之后
演變?yōu)閟end使用一個鎖，recv使用interlockedxxx，再之后演變?yōu)?br>send使用一個cs鎖和一個sending，recv使用一個ref，經(jīng)歷了如此多種演變。

1、一個鎖userlock, sendcomp recvcomp senddata都用這個鎖。
2、兩個鎖，userlock, sendlock
3、一個鎖userlock+socket查找，鎖范圍比1小，多一次全局查找操作。
4、一個鎖sendlock，一個ref（InterlockedXXX）
   wsasend wsarecv 對象生存期等都用ref計數(shù)
5、一個鎖sendlock和一個sending（InterlockedXXX），一個ref（InterlockedXXX）
   wsasend用sending計數(shù)，wsarecv等用ref計數(shù)

歷經(jīng)如此幾個版本的修改，鎖的范圍越來越小，interlocked次數(shù)越來越少，反應(yīng)自然
越來越快。
今天為了寫總結(jié)文章，拿1 4 5進行了一次比較，1除了大量連接的時候比4 5 反應(yīng)
慢了很多之外在cpu mem等占用方面幾乎和4 5 相當(dāng)，也就是說除了實時性方面差一些
其他方面沒有太大差別。
但4 5都是用到了線程級別的內(nèi)存池，1的測試?yán)踢€是用老的內(nèi)存池，這個差別也
很大，所以綜合地看，雖然做了很多優(yōu)化卻是花費80%的時間，做了20%的鳥事。

5、線程關(guān)聯(lián)內(nèi)存池再提速

 

 

上一節(jié)已經(jīng)提到問題，解決辦法是這樣的

struct tm_bufunit

{

        tm_pool *pool;                        //pool指針

        union

        {

                tm_bufunit *next;   //下一個塊指針

                char data[4];           //數(shù)據(jù)區(qū)域

        };

};

 

static void *tm_malloc(size_t size, size_t *osize=NULL)

{

        tm_bufunit *p = (tm_bufunit *)malloc(sizeof(tm_bufunit)-offsetof(tm_bufunit, data)+size);

        if(p)

        {

                p->pool = NULL;

                if(osize) *osize = size;

                return p->data;

        }

        return      NULL;

}

看上面的代碼應(yīng)該很容易明白，就是將由該池malloc的內(nèi)存塊也打上統(tǒng)一的標(biāo)記，這樣由該池分配的任何內(nèi)存塊都可用最簡單的判斷釋放，省去了查找線程查找目標(biāo)池的兩次查詢，不光提速了而且解決了上一節(jié)提到的那個bug。

最終實現(xiàn)的線程關(guān)聯(lián)內(nèi)存池通用分配函數(shù)tm_new大概相當(dāng)于malloc 15倍左右的速度，定位到pool之后的newobj相當(dāng)于malloc 45倍左右的速度。通用函數(shù)大致相當(dāng)于nedmalloc速度的2.6-3倍，直接定位到pool的分配速度大概相當(dāng)于dlmalloc 的2倍。

 

關(guān)于線程關(guān)聯(lián)的內(nèi)存池還有一些細(xì)節(jié)問題我沒有展開討論，如free表是每個線程保留一份還是全局保留一份，如果是全局保留一份則涉及到復(fù)用的時候如何分配，還有就是tls系列函數(shù)我看nedmalloc也在用，我第一版也在用，但后來實測發(fā)現(xiàn)這些函數(shù)貌似效率不高，后面的版本沒有采用tls系列函數(shù)。

關(guān)于線程關(guān)聯(lián)的內(nèi)存池我寫了5個版本，當(dāng)然最重要的還是第一個版本，后面的版本除了這一節(jié)提到的重要改進之外變化不是很大，最后的第五版增了一些和我的私有lib相關(guān)的功能。

 

以前寫文章太少，總是看別人的文章，在網(wǎng)絡(luò)時代覺得自己挺自私，這次一鼓作氣，一口氣寫了出來，可能寫得很粗略，不知道有多少人能看明白，如能給讀者一點啟示我將感到很欣慰。

4、線程關(guān)聯(lián)的內(nèi)存池

 

每每想到單線程下內(nèi)存池飛一般的速度和多線程下蝸牛一般的速度我就不能原諒自己，為什么差這么多，就不能讓多線程下內(nèi)存分配更快一點嗎？解決方法有了，那就是讓緩存線程化，各個線程有自己私有的緩存，分配的時候預(yù)先從當(dāng)前線程私有緩存分配，分配空了的時候去全局free表取一組freeunit或直接向系統(tǒng)申請一大塊緩存（各個線程緩存完全獨立），不管具體采用什么方式，速度都大幅度的提高了，雖然還是比單線程下內(nèi)存池慢了許多，不過比前面提到的多線程內(nèi)存池以及nedmalloc都要快很多，我的實現(xiàn)大概比nedmalloc快1.6 ~ 2倍，離單線程下內(nèi)存池速度也很近了，只是多了些查找線程id比較線程id等動作而已，基本上達(dá)到了自己的目標(biāo)。

 

看看第一版線程關(guān)聯(lián)內(nèi)存池的一些代碼：

 

struct tm_bufunit

{

        tm_pool *pool;                        //pool指針

        union

        {

                tm_bufunit *next;   //下一個塊指針

                char data[4];           //數(shù)據(jù)區(qū)域

        };

};

 

struct tm_gcontrol

{

        tm_bufunit *gfree;

        CRITICAL_SECTION gcs;

 

        tm_gcontrol() : gfree(NULL) { InitializeCriticalSection(&gcs); }

        ~tm_gcontrol()        { DeleteCriticalSection(&gcs); }

        Inline void lock()     { EnterCriticalSection(&gcs); }

        Inline void unlock() { LeaveCriticalSection(&gcs); }

        void free(tm_bufunit *buf)

        {

                lock();

                buf->next = gfree;

                gfree = buf;

                unlock();

        }

};

 

struct tm_memblock

{

        tm_memblock *next;

};

 

class tm_pool

{

private:

        size_t bksize;                   //一個分配塊大小

        size_t onebknum;            //一次分配多少個bksize

        DWORD thid;                          //線程id

        tm_bufunit *next;           //pool中自由塊鏈

        tm_memblock *mbk;              //trunk表

        tm_gcontrol gcontrol;     //全局free表

       

        friend tm_poolset;

private:

        void expand();

 

public:

        tm_pool(size_t size, size_t bknum);

        ~tm_pool();

 

        void destroy();

        void *newobj();

        static void delobj(void *pbuf);

};

 

class tm_poolset

{

public:

        tm_poolset();

        virtual ~tm_poolset();

 

        //添加分配池

        bool addpool(size_t size, size_t allocnum);

        void *newobj(size_t size, size_t *osize=NULL);

        void delobj(void *pbuf, size_t size);

        void destroy();

 

        tm_pool *findpool(size_t size)

        {

                TMPOOLS::iterator it = tmpools.lower_bound(size);

                if(it != tmpools.end())

                        return it->second;

                return NULL;

        }

protected:

        typedef std::map<size_t, tm_pool *> TMPOOLS;

        TMPOOLS tmpools;

};

 

//公開的數(shù)據(jù)及函數(shù)

extern DWORD tm_tlsindex; //tls索引

 

//app初始化，分配index

void tm_init();

void tm_free();

 

//關(guān)聯(lián)到該線程

void tm_attach();

void tm_detach();

 

tm_poolset *tm_getpoolset();

//添加trunk

bool tm_addtrunk(size_t size, size_t allocnum);

//tls相關(guān)分配

void *tm_new(size_t size, size_t *osize=NULL);

//tls相關(guān)釋放

void tm_del(void *buf, size_t size);

 

 

 

.cpp代碼如下：

tm_pool::tm_pool(size_t size, size_t bknum) :

        next(NULL), mbk(NULL),

        bksize(size), onebknum(bknum)

{

        thid = GetCurrentThreadId();

}

 

tm_pool::~tm_pool()

{

        destroy();

}

 

void tm_pool::destroy()

{

        for(tm_memblock *p = mbk; p; )

        {

                tm_memblock *q = p->next;

                free((char *)p);

                p = q;

        }

        mbk = NULL;

        next = NULL;

}

 

void *tm_pool::newobj()

{

        if(! next)

        {

                gcontrol.lock();

                if(gcontrol.gfree)

                {

                        next = gcontrol.gfree;

                        gcontrol.gfree = NULL;

                }

                gcontrol.unlock();

        }

        if(! next)

        {

                expand();

        }

        tm_bufunit *head = next;

        next = head->next;

//     return (void *)head;

        return (void *)head->data;

}

 

void tm_pool::delobj(void *pbuf)

{

//     tm_bufunit *head = (tm_bufunit*)(pbuf);

        tm_bufunit *head = (tm_bufunit *)((char *)pbuf-offsetof(tm_bufunit, data));

        tm_pool *pool = head->pool;

        if(pool->thid == GetCurrentThreadId())

        {

                head->next = pool->next;

                pool->next = head;

        }

        else

        {

                pool->gcontrol.free(head);

        }

}

 

void tm_pool::expand()

{

        size_t unitsize = offsetof(tm_bufunit, data) + bksize;

        size_t size = (unitsize * onebknum + sizeof(tm_memblock));

        tm_memblock *pbk = (tm_memblock *)malloc(size);

        pbk->next = mbk;

        mbk = pbk;

        tm_bufunit *p = (tm_bufunit*)((char *)pbk+sizeof(tm_memblock));

        p->pool = this;

        next = p;

        for(size_t i=0; i<onebknum-1; ++i)

        {

                p->next = (tm_bufunit *)((char *)p+unitsize);

                p = p->next;

                p->pool = this;

        }

        p->next = NULL;

}

 

…

這一版基本實現(xiàn)了第一步的提速目標(biāo)，并且每個分配塊還記錄了來自哪個pool，這樣free的時候就省去了查找pool的動作，只是還有一些問題，如何判斷一個內(nèi)存是來源于malloc的分配還是來源于pool的分配沒有做終結(jié)的判斷，而且還留下了一個bug，對于a線程來說，可能只有256，512兩個塊的緩存，b線程可能多一個塊1024，這樣a線程分配的1024字節(jié)的內(nèi)存是用malloc分配，到b線程釋放的時候會調(diào)用pool釋放，這個bug將在下一章解決。

     摘要: 2、多線程內(nèi)存池   上一節(jié)很簡略的說了下單線程內(nèi)存池，單線程內(nèi)存池如果要放在多線程環(huán)境下使用是不安全的，我們需要進行保護，如何保護，最簡單的方法就是加臨界區(qū)，云風(fēng)的實現(xiàn)里面是用原子操作模擬一個臨界區(qū)，我實測跟臨界區(qū)性能非常接近，甚至很多時候不如臨界區(qū)，所以我就不追求用原子操作模擬臨界區(qū)了，還是直接用臨界區(qū)最簡單。   class CMemPool { publi...  閱讀全文

1、單線程內(nèi)存池

 

內(nèi)存池的基本思想是大塊向系統(tǒng)申請內(nèi)存，內(nèi)部切割為小塊，內(nèi)部cache之后有選擇的分配，不夠的時候繼續(xù)向系統(tǒng)大塊申請內(nèi)存，示例代碼如下：

 

struct tm_memblock

{

        tm_memblock *next;

};

class tm_pool

{

…

        tm_bufunit *next;           //pool中自由塊鏈

        tm_memblock *mbk;              //trunk表

…

};

 

void *tm_pool::newobj()

{

        if(! next)

        {

                expand();

        }

        tm_bufunit *head = next;

        next = head->next;

        return (void *)head;

}

 

void tm_pool::delobj(void *pbuf)

{

        tm_bufunit *head = (tm_bufunit*)(pbuf);

        head->next = next;

next = head;

}

詳細(xì)實現(xiàn)建議看云風(fēng)的內(nèi)存池，我也不過是學(xué)習(xí)了它的實現(xiàn)而已。

不要小看了單線程內(nèi)存池，它是我們走向更復(fù)雜應(yīng)用的基礎(chǔ)，它是我們后面提及的多線程內(nèi)存池以及線程關(guān)聯(lián)內(nèi)存池的基礎(chǔ)。

這種單線程的內(nèi)存池分配釋放速度是很快的，比dlmalloc更快近1倍，大約相當(dāng)于malloc/free的50-100倍(具體倍率視分配的大小而不同，分配小塊倍率小，分配大塊倍率大)。

 

有的朋友可能會考慮使用std::list之類的東西來構(gòu)建內(nèi)存池，我奉勸有這種想法的人打住，std::list是效率很低的，此外用一個高層的東西構(gòu)建底層模塊，總體上屬于本末倒置。

0、內(nèi)存池之引言

 

 

這是關(guān)于內(nèi)存池的一系列簡短文章，當(dāng)然它不是短期的研究結(jié)果，而是長期使用經(jīng)驗的總結(jié)，介紹得可能不會很詳細(xì)，一些別人介紹得很細(xì)節(jié)的東西我就基本掠過。

轉(zhuǎn)載請署名作者：袁斌

 

內(nèi)容如下：

1、 單線程內(nèi)存池。

2、 多線程內(nèi)存池。

3、 Dlmalloc nedmalloc

4、 實現(xiàn)線程關(guān)聯(lián)的內(nèi)存池。

5、 線程關(guān)聯(lián)內(nèi)存池再提速。

 

關(guān)于內(nèi)存池有一些經(jīng)典資源可參考，如早期侯杰《池內(nèi)春秋》，云風(fēng)（網(wǎng)易核心開發(fā)人員）《我的編程感悟》中提及的內(nèi)存池，以及許世偉（原金山CTO，現(xiàn)盛大創(chuàng)新院技術(shù)專家）關(guān)于內(nèi)存池的系列文章。

 

可以通過QQ：345585946和我討論

也可加入我的群：41947449 一起討論游戲開發(fā)。

前些天無意中在網(wǎng)上看到一篇文章，寫的是幾種腳本語言以及c的速度比較，大意是測試下來lua的速度很慢，只相當(dāng)于c的1/60左右，看了一下測試函數(shù)，如下（命名版本1）：

 

tkstart = os.clock()

 

local count = 10000000

local accumv = 34.5;

local function iterate (times,accumv)

    if 0 == times then

        return accumv

    else

        return iterate (times - 1, accumv + 2)

    end

end

print(iterate(count,accumv))

 

print(os.clock() - tkstart)

 

這是個迭代調(diào)用，理論上這種調(diào)用速度不快，所以改寫了一下，命名版本2

 

tkstart = os.clock()

 

local i=0.0;

local function process (ret)

 while(i < 10000000.0) do

 i = i+1.0

 ret = ret + 2.0

 end

 return ret

 

end

 

process(34.5)

print(os.clock() - tkstart)

 

再改寫了一下，版本3，如下:

 

tkstart = os.clock()

 

local i=0.0;

local function process (ret)

 local t = ret

 while(i < 10000000.0) do

 i = i+1.0

 t = t + 2.0

 end

 return ret

 

end

 

process(34.5)

print(os.clock() - tkstart)

 

用for循環(huán)再改寫了一個版本，版本4，如下：

 

tkstart = os.clock()

 

local count = 10000000

local accumv = 34.5;

local function calc (times,accumv)

        local result = accumv

        for i=1,times do

                result = result + 2.0

        end

        return result

end

print(calc(count,accumv))

 

print(os.clock() - tkstart)

 

簡單測試耗時如下

版本1 0.945

版本2 0.484

版本3 0.475

版本4 0.138   編譯成.out跟直接執(zhí)行.lua耗時差不多，沒明顯變化。

雖然是一個功能很簡單實現(xiàn)也很簡單的lua函數(shù)，但速度差別很大，由此可見不同寫法對執(zhí)行效率影響很大，跟版本4差不多的c程序執(zhí)行耗時大約為0.016秒, lua大致相當(dāng)于c的1/8 - 1/9速度，而不是那篇文章的作者認(rèn)為的只相當(dāng)于c的1/60。

 

從上面四個不同代碼大致可以看出以下問題：

1、 迭代算法是很慢的（版本1）。

2、 While的效率比for慢很多（版本23和4的比較）。

3、 臨時變量比參數(shù)效率高（版本23）。

 

測試次數(shù)	傳統(tǒng)線程池耗時	系統(tǒng)線程池耗時	完成端口線程池耗時
1	49250	72234	20391
2	50906	71266	20281
3	50156	73000	20297
4	50437	71157	20000
5	49250	73078	19547
6	49844	73156	19469

 

 

線程池模型	適用操作系統(tǒng)	效率	易用性
傳統(tǒng)線程池	任意	中	較麻煩，大多數(shù)人寫不出好的線程池
系統(tǒng)線程池	Win2000之后	低	最簡單
完成端口線程池	Win2000之后	高	較簡單