在2004年Maged M. Michael那片开创性的文章“Scalable Lock-Free Dynamic Memory Allocation”,采用了Lock-Free机制的malloc实现击败了目前所有现存的malloc库,包括ptmalloc3,hoard等,Lock-Free一时间成为主流的研究重点。我们也打算采用这种方法来提高Alloc()的性能,不过由于实现的复杂性,本文所要介绍的Lock-Free实现,主要是用于避免在读写空闲链表头部的Lock/Unlock带来的性能损失,MemoryPool采用的lib_malloc仍然是Linux/GNU环境下的基于Lock/Unlock的ptmalloc实现。尽管如此,采用Lock-Free的实现比mutex/spin lock有明显的性能提高。
严格讲,Lock-Free是一套以原子操作为基础,采用事务->提交->提交失败->重试这样特定编程手法的机制,它使得正在访问共享资源的线程不依赖于任何其它线程的调度和执行,并且能够在有限的步骤内完成。
Lock-Free的实现往往都是通过CAS-Compare And Swap原子操作完成,声明如下:
CAS(addr,expval,newval) atomically do
if (*addr == expval) {
*addr = newval;
return true;
} else
return false;
Linux上一种典型的实现是:
inline int CAS(unsigned long *mem,unsigned long newval,unsigned long oldval)
{
__typeof (*mem) ret;
__asm __volatile ("lock; cmpxchgl %2, %1"
: "=a" (ret), "=m" (*mem)
: "r" (newval), "m" (*mem), "0" (oldval));
return (int) ret;
}
以MemoryPool < T > :: alloc (size_t)为例:
….
//更新next指针的内容
do {
oldheader = next;
if (oldheader==NULL)
expandTheFreeList();
else
target=oldheader->next;
}
while (CAS(&next,target,oldheader)!=oldheader)
实际执行中,如果next 的值为oldheader,那么就会更新next的值为target,并返回oldheader,这一过程是原子操作;如果在执行CAS指令的时候,next的内容已经发生变化,那么CAS的返回值就不会等于oldheader,while中的条件为false,程序会返回重新执行oldheader = next,直到next 的值被成功更新。同时,expandTheFreeList也要作类似的Lock-Free修改,下面的例子中把原先expandTheFreeList的代码放置到alloc中。
newbuf_head = NULL //note1
do
{
oldheader =(MemoryPool*) next; //note2
if (!oldheader)
{
if (!newbuf_head)
{
size_t size = (m_sizeofClass > sizeof(MemoryPool *))?m_sizeofClass:sizeof(MemoryPool*);
MemoryPool *runner = reinterpret_cast<MemoryPool*>(new char[size]);
newbuf_head = runner;
for (int i = 0 ; i < EXPANSION_SIZE; i++)
{
runner->next = reinterpret_cast<MemoryPool *>(new char [size]);
runner =(MemoryPool*) runner->next;
}
runner->next = 0;
newbuf_tail = runner;
}
}
if (newbuf_head) //note3
{
newbuf_tail->next = (MemoryPool*)next;
target = newbuf_head;
}
else if (oldheader)
{
target = oldheader;//note4
}
//note5
}while ( CAS((unsigned long*)&next,/*m_freeheader,*/(unsigned long)target->next,(unsigned long)oldheader) !=(int) oldheader);
如果oldheader 为NULL,那么就会分配一段新的内存,头部为newbuf_head,尾部为newbuf_tail,并且把原先的FreeList头部挂接到新内存的尾部,以新内存的头部作为新的FreeList的头部去更新next的值。
如果分配内存后,CAS执行失败了,这没关系,此时newbuf_head已被置值,因此并不会再次分配内存,程序仍然试图执行上一次执行过的路径。
考虑一种极端情况,起初next指向一个有效地址(oldheader也指向一个有效值),因此程序跳过了分配新内存那段(斜体字),不幸在执行到note3处时,next却由于另一个线程分配了内存的关系置为NULL,此时怎么办?
答案,CAS操作时next的内容与oldheader不一致,因此CAS失败,程序再次返回note2处执行。
<4>Lock-Free中的ABA问题
承接上面的那一个极端问题,
线程A在执行note2时的空闲链表状态如下: A-B-C-D-E-NULL,在执行到note5时,期间发生了如下事情:
线程B从FreeList中分配了一个A和B两个内存块,FreeList变为C-D-E-NULL,但随后线程B归还了内存块A,从而FreeList再次变为A-C-D-E-NULL。
线程A在执行到note5,进入CAS临界操作时,next的内容仍然和note2时的一致,均指向A,但CAS的第二个关键参数,target->next 的却已不是B了!(在note4时target被设置为oldheader即A,从而target->next为B),如果把next更新为B,那么就会造成程序错误。
这个就是Lock-Free与生俱来的ABA问题,CAS无法判断目标内容从A变为B,然后又变为A这种情况,解决的办法通常是使用一个额外的tag来记录这种情况,并且使用CAS2,来同时检查tag和目标内存两个值又没有发生变化,声明如下:
bool CAS2 (volatile void * ptr, uint32_t old1, uint32_t old2, uint32_t new1, uint32_t new2)
{
bool ret;
__asm__ __volatile__ ("lock cmpxchg8b (%1) /n/t sete %%al"
: "=a"(ret)
: "D"(ptr), "a"(old1), "d"(old2), "b"(new1), "c"(new2)
: "memory");
return ret;
}
CAS2可以用来检查一个64 bit长度的内存指并原子交换他们的内容。
在某些情况下也可以采用thread specified Pool这种方案,就是每个线程有其专用的内存池,不会出现两个线程同时在一个Pool上分配内存的情况,这需要额外的空间在分配的对象中嵌入线程的信息。
<5>Lock-Free的性能
可见Lock-Free的实现并不算一件太新鲜复杂的事情,Windows平台上早就有了InterlockedCompareExchange之类的函数,Windows中提供的Slist也是一个Lock-Free 的linked-list。那么在何种情况下使用Lock-Free能达到最理想的效果?
Lock-Free机制在single core上会取得明显的性能优势。所有的CAS操作实质上在CPU微观世界里是串行执行的,由于没有别的线程干扰,一个处于运行状态的线程永远不会发生CAS操作失败。
2 cpu core的情况下,一个线程alloc对象另一个线程free对象内存,可能会导致一个线程里有少量的CAS失败,但是与pthread_spin_lock机制相比,耗时仍然非常少。究其原因,pthread_spin_lock会受到锁粒度的的影响,假如Alloc需要耗时0.00023秒的话,那么pthread_spin_lock就可能需要消耗同样的时间。而CAS是一种尽力处理的机制,即使在某个线程里发生了一个CAS失败,一方面程序会以一个很细的粒度马上重试,另一方面也表明与此同时另一个线程成功执行了一次CAS操作。
从下表看出,CAS的耗时仅仅是pthread_mutex_lock的1/10。而整个alloc的耗时比较,Lock-Free也只有pthread_mutex版本的1/4。
如果增加alloc/free的线程以及cpu 的数量,Lock-Free机制会由于CAS操作失败而不断重复执行do {}while(CAS) 中的代码,从而导致一定的性能下降。因此,应该尽量让do{}中的代码写得简短,以保证较高的CAS成功率,其次是针对特定场景开发Lock-Free应用,例如在多核程序中进行pipeline处理,在同一个地方分配内存,在同一个地方回收内存。