并发内存分配问题以及TBB的解决方案

在多线程程序中，普通的内存分配将成为严重的性能瓶颈。本文介绍了怎样使用Threading Building Blocks的可扩展内存分配器来避免内存分配竞争和假共享问题。

内存分配不仅是编程的基本任务，也是在多核编程时影响效率的一大挑战。在C++里我们可以用自定义内存分配器代替std:: allocator，Threading Building Blocks就提供了一个与std::allocator兼容的可扩展内存分配器。

每个内存分配器都有自己的特点。TBB的可扩展分配器致力于可扩展性和速度。在某些情况下，这是有代价的：浪费虚拟空间。具体来说，当 分配9k到12k的块可它会浪费较大的空间。

内存分配问题

在多线程程序中，普通的内存分配将成为严重的性能瓶颈。这是因为普通的内存分配器用一个全局锁从一个单一的全局堆中分配和释放内存块， 每个线程分配和释放内存时就会引发竞争。正是由于这种竞争，频繁内存分配的程序可能降低多核带来的优势。使用C++标准库（STL）的程序可能会更为严 重，因为它们的内存分配往往隐藏在背后。

这里先放一段代码，它并发地执行一个有内存申请和释放操作的函数：

#include <iostream>
#include <tbb/parallel_for.h>
#include <tbb/tick_count.h>
#include <vector>

using namespace tbb;

void alloctask( int )
{ 
 //vector构造和析构时会申请和释放空间
 std::vector<int> data(100);
}

int main()
{
 tick_count t1 = tick_count::now(); //用于记录花费的时间
 parallel_for(0,100000,1,alloctask); //十万次执行alloctask(并发)
 tick_count t2 = tick_count::now();

 std::cout << (t2-t1).seconds() << std::endl;
 return 0;
}

你可以运行这段代码查看所花费的时间，我们可以很容易地把上面的代码修改成使用TBB可扩展内存分配器，这样速度会加快将近一倍（在偶 双核CPU上，预计更多内核的CPU会有更好的表现）。

注：如果之前看过本站关于TBB循环的 文章的朋友可能会奇怪，这里没有task_scheduler_init，而且parallel_for的参数也不一样。其实这段代码是基于最新的TBB2.2版 本的，这个版本已经不再强制要求task_scheduler_init。parallel_for也多了几个重载以方便使用。

“假共享”是并发程序中另一个严重问题，它常发生于当多个线程使用的内存块紧靠在一起时。在处理器内核中有一个称为“cache lines”的高速缓存区，它只能由同一个线程存取同一个缓存区，否则就会引发缓存切换，这可以轻易地造成上百时钟周期的浪费。

为了说明为什么假共享有这样的性能损失，我们可以看看当两个线程访问紧靠在一起的内存时引起的额外的开销。假设这个缓存区有64字节， 两个线程共享这个缓存。

首先，程序定义了两个数组，包含1000个float(4字节)：

float A_array [1000];
float B_array [1000];

由于编译器的顺序分配，这两个数组很可能是紧靠在一起的。考虑下面的动作：

1. 线程A写入A_array[999];
   处理器将包含A_array[999]这个元素的64个字节放入缓存
2. 线程B写入B_array [0];
   额外的开销：处理器必须刷新缓存，把A_array[999]保存到内存中。把包含B_array[0]的64个字节载入缓存并设置线程A的缓存 标记为无效。
3. 继续工作，线程A写入A_array[1];
   额外的开销：处理器必须刷新缓存，以便把B_array [0]保存到内存中。重新为线程A加载缓存并设置线程B的缓存标记为无效。

看，即使线程A和线程B使用各自的内存还是会造成极大的开销。解决方法假共享的办法是把数组按缓存边界对齐。

内存分配器

TBB的可扩展内存分配器可以用来解决上面所述的问题，TBB提供了两个分配器：scalable_allocator和cache_aligned_allocator，它们分别定义于tbb/scalable_allocator.h和 tbb/cache_aligned_allocator.h里。

• scalable_allocator 解决了分配竞争的情况，它并没有完全防止假共享。不过每个线程从不同的内存池中取得内存，这也可以从一定程 序上避免假共享的发生。
• cache_aligned_allocator 解决了分配竞争和假共享问题。由于分配的内存是缓存大小的倍数所以要花费更多的空间，尤其是分配大量小空 间时。我们应该在确定假共享已成为性能瓶颈时才使用cache_aligned_allocator。在你的程序中分别使用两种分配器来测试性能以确定最 终使用哪一个是个好主意。

在STL容器中使用分配器

scalable_allocator和cache_aligned_allocator与std::allocator是兼容的，我们可以和使用 std::allocator一样使用它们。下面的例子演示了使用cache_aligned_allocator作为std::vector的分配器。

std::vector< int, cache_aligned_allocator<int> >;

现在我们可以把前面的代码修改一下了：

void alloctask( int )
{  
    //vector构造和析构时会申请和释放空间
    std::vector<int, scalable_allocator<int> > data(100);
}

对比一下在你的电脑上效率提升了多少吧^_^

代替malloc,free,realloc和calloc

TBB为malloc,free,realloc和calloc提供了对应的可扩展版本：

#include "tbb\scalable_allocator.h"

void * scalable_malloc (size_t size);
void   scalable_free (void* ptr);
void * scalable_realloc (void* ptr, size_t size);
void * scalable_calloc (size_t nobj, size_t size);

代替new和delete

要完整地重载C++中的new和delete，我们要实现下面这四对new/delete操作：

void* operator new(std::size_t size) throw(std::bad_alloc);
void* operator new(std::size_t size, const std::nothrow_t&) throw( );
void* operator new[](std::size_t size) throw(std::bad_alloc);
void* operator new[](std::size_t size, const std::nothrow_t&) throw( );
void  operator delete(void* ptr) throw( );
void  operator delete(void* ptr, const std::nothrow_t&) throw( );
void  operator delete[](void* ptr) throw( );
void  operator delete[](void* ptr, const std::nothrow_t&) throw( );

我们可以利用前面说到的scalable_malloc()和scalable_free()来实现这些操作：

#include "tbb\scalable_allocator.h"

void* operator new (size_t size) throw (std::bad_alloc)
{
    if (size == 0) size = 1;
    if (void* ptr = scalable_malloc (size))
        return ptr;
    throw std::bad_alloc ( );
}
void* operator new[] (size_t size) throw (std::bad_alloc)
{
    return operator new (size);
}
void* operator new (size_t size, const std::nothrow_t&) throw ( )
{
    if (size == 0) size = 1;
    if (void* ptr = scalable_malloc (size))
        return ptr;
    return NULL;
}
void* operator new[] (size_t size, const std::nothrow_t&) throw ( )
{
    return operator new (size, std::nothrow);
}
void operator delete (void* ptr) throw ( )
{
    if (ptr != 0) scalable_free (ptr);
}
void operator delete[] (void* ptr) throw ( )
{
    operator delete (ptr);
}
void operator delete (void* ptr, const std::nothrow_t&) throw ( )
{
    if (ptr != 0) scalable_free (ptr);
}
void operator delete[] (void* ptr, const std::nothrow_t&) throw ( )
{
    operator delete (ptr, std::nothrow);
}

<<完>>