oldworm

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

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

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

 

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

 

方法3、方法1雖然簡單但畢竟InterlockedXXX系列函數調用還是有些耗時的，大概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,…

絕無重復，調用非常簡單每個線程id = base; base += step;即可。

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

 

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

執行100w次的結果（單位：微妙）
------------------------------------------------------------------------------------
線程 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);

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

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

從2004年使用iocp開始，嘗試過很多種client生存期管理，最初使用
一個鎖userlock，在sendcomp和recvcomp以及senddata等都用一個鎖，而且是
進去的地方就開始鎖，這種方式最簡單，也產生了最初的版本，之后為
追求更高效率，慢慢演變為使用兩個鎖，readlock和writelock，再之后
演變為send使用一個鎖，recv使用interlockedxxx，再之后演變為
send使用一個cs鎖和一個sending，recv使用一個ref，經歷了如此多種演變。

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

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

5、線程關聯內存池再提速

 

 

上一節已經提到問題，解決辦法是這樣的

struct tm_bufunit

{

        tm_pool *pool;                        //pool指針

        union

        {

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

                char data[4];           //數據區域

        };

};

 

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;

}

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

最終實現的線程關聯內存池通用分配函數tm_new大概相當于malloc 15倍左右的速度，定位到pool之后的newobj相當于malloc 45倍左右的速度。通用函數大致相當于nedmalloc速度的2.6-3倍，直接定位到pool的分配速度大概相當于dlmalloc 的2倍。

 

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

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

 

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

4、線程關聯的內存池

 

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

 

看看第一版線程關聯內存池的一些代碼：

 

struct tm_bufunit

{

        tm_pool *pool;                        //pool指針

        union

        {

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

                char data[4];           //數據區域

        };

};

 

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;

};

 

//公開的數據及函數

extern DWORD tm_tlsindex; //tls索引

 

//app初始化，分配index

void tm_init();

void tm_free();

 

//關聯到該線程

void tm_attach();

void tm_detach();

 

tm_poolset *tm_getpoolset();

//添加trunk

bool tm_addtrunk(size_t size, size_t allocnum);

//tls相關分配

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

//tls相關釋放

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;

}

 

…

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

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

1、單線程內存池

 

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

 

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;

}

詳細實現建議看云風的內存池，我也不過是學習了它的實現而已。

不要小看了單線程內存池，它是我們走向更復雜應用的基礎，它是我們后面提及的多線程內存池以及線程關聯內存池的基礎。

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

 

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

0、內存池之引言

 

 

這是關于內存池的一系列簡短文章，當然它不是短期的研究結果，而是長期使用經驗的總結，介紹得可能不會很詳細，一些別人介紹得很細節的東西我就基本掠過。

轉載請署名作者：袁斌

 

內容如下：

1、 單線程內存池。

2、 多線程內存池。

3、 Dlmalloc nedmalloc

4、 實現線程關聯的內存池。

5、 線程關聯內存池再提速。

 

關于內存池有一些經典資源可參考，如早期侯杰《池內春秋》，云風（網易核心開發人員）《我的編程感悟》中提及的內存池，以及許世偉（原金山CTO，現盛大創新院技術專家）關于內存池的系列文章。

 

可以通過QQ：345585946和我討論

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

前些天無意中在網上看到一篇文章，寫的是幾種腳本語言以及c的速度比較，大意是測試下來lua的速度很慢，只相當于c的1/60左右，看了一下測試函數，如下（命名版本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)

 

這是個迭代調用，理論上這種調用速度不快，所以改寫了一下，命名版本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循環再改寫了一個版本，版本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跟直接執行.lua耗時差不多，沒明顯變化。

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

 

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

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

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

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

 

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

 

 

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