我正在实现Anthony Williams 在“C++ Concurrency in Action”中描述的无锁队列。我将它作为libcds的新容器进行测试。弹出和推送测试工作正常。但是多个生产者，多个消费者测试有时会失败。VLD（或 Intel Inspector XE 或 ASan）显示内存泄漏。我通过添加节点析构函数来修复它，但所有元素存在的问题仍然存在。我怎样才能解决这个问题？谢谢。
威廉姆斯无锁队列：
#include <memory>
template <class T>
class williams_queue
{
public:
williams_queue()
{
counted_node_ptr counted_node;
counted_node.ptr = new node;
counted_node.external_count = 1;
head_.store(counted_node);
tail_.store(head_);
}
williams_queue(const lock_free_queue_mpmc& other) = delete;
williams_queue& operator=(const lock_free_queue_mpmc& other) = delete;
~williams_queue()
{
counted_node_ptr old_head = head_.load();
while (node* const old_node = old_head.ptr)
{
head_.store(old_node->next);
delete old_node;
old_head = head_.load();
}
}
void push(const T& new_value)
{
std::unique_ptr<T> new_data(new T(new_value));
counted_node_ptr new_next;
new_next.ptr = new node;
new_next.external_count = 1;
counted_node_ptr old_tail = tail_.load();
while (true)
{
increase_external_count(tail_, old_tail);
T* old_data = nullptr;
if (old_tail.ptr->data.compare_exchange_strong(old_data, new_data.get()))
{
counted_node_ptr old_next = {0};
if (!old_tail.ptr->next.compare_exchange_strong(old_next, new_next))
{
delete new_next.ptr;
new_next = old_next;
}
set_new_tail(old_tail, new_next);
new_data.release();
break;
}
else
{
counted_node_ptr old_next = {0};
if(old_tail.ptr->next.compare_exchange_strong(old_next, new_next))
{
old_next = new_next;
new_next.ptr = new node;
}
set_new_tail(old_tail, old_next);
}
}
}
bool pop(Func f)
{
counted_node_ptr old_head = head_.load(std::memory_order_relaxed);
while (true)
{
increase_external_count(head_, old_head);
node* const ptr = old_head.ptr;
if(ptr == tail_.load().ptr)
{
release_ref( p );
return false;
}
counted_node_ptr next = ptr->next.load();
if (head_.compare_exchange_strong(old_head,next))
{
T* const res = ptr->data.exchange(nullptr);
free_external_counter(old_head);
f(res.get());
return true;
}
ptr->release_ref();
}
}
private:
struct node;
struct counted_node_ptr
{
int external_count;
node* ptr;
};
struct node_counter
{
unsigned internal_count : 30;
unsigned external_counters : 2;
};
struct node
{
std::atomic<T*> data;
std::atomic<node_counter> count;
std::atomic<counted_node_ptr> next;
node()
{
node_counter new_count;
new_count.internal_count = 0;
new_count.external_counters = 2;
count.store(new_count);
counted_node_ptr new_next;
new_next.ptr = nullptr;
new_next.external_count = 0;
next.store(new_next);
}
};
static void release_ref(node * p)
{
node_counter old_counter = p->count.load(std::memory_order_relaxed);
node_counter new_counter;
do
{
new_counter=old_counter;
--new_counter.internal_count;
}
while(!p->count.compare_exchange_strong(old_counter, new_counter,
std::memory_order_acquire,
std::memory_order_relaxed));
if(!new_counter.internal_count && !new_counter.external_counters)
{
delete p;
}
}
private:
void set_new_tail(counted_node_ptr& old_tail,
const counted_node_ptr& new_tail)
{
node* const current_tail_ptr = old_tail.ptr;
while (!tail_.compare_exchange_weak(old_tail, new_tail) &&
old_tail.ptr == current_tail_ptr);
if(old_tail.ptr == current_tail_ptr)
{
free_external_counter(old_tail);
}
else
{
release_ref(current_tail_ptr);
}
}
static void increase_external_count(std::atomic<counted_node_ptr>& counter,
counted_node_ptr& old_counter)
{
counted_node_ptr new_counter;
do
{
new_counter = old_counter;
++new_counter.external_count;
}
while(!counter.compare_exchange_strong(old_counter, new_counter,
std::memory_order_acquire,
std::memory_order_relaxed));
old_counter.external_count = new_counter.external_count;
}
static void free_external_counter(counted_node_ptr& old_node_ptr)
{
node* const ptr = old_node_ptr.ptr;
const int count_increase = old_node_ptr.external_count - 2;
node_counter old_counter= ptr->count.load(std::memory_order_relaxed);
node_counter new_counter;
do
{
new_counter = old_counter;
--new_counter.external_counters;
new_counter.internal_count += count_increase;
}
while(!ptr->count.compare_exchange_strong(old_counter, new_counter,
std::memory_order_acquire,
std::memory_order_relaxed));
if(!new_counter.internal_count && !new_counter.external_counters)
{
delete ptr;
}
}
private:
std::atomic<counted_node_ptr> head_;
std::atomic<counted_node_ptr> tail_;
};
测试结果：
Visual Leak Detector 从以下位置读取设置：D:\Development\COMMON_UTILS\Visual Leak Detector\vld.ini Visual Leak Detector 版本 2.5 已安装。libcds 版本 2.1.0 测试开始于 2017 年 1 月 31 日 01:19:03 使用测试配置文件：test-debug.conf 系统拓扑：逻辑处理器数：4
Queue_ReaderWriter::WilliamsQueue_default reader count=3 writer count=3 item count=99999... Item count: 0 Item count: 0 Item count: 0 Post pops: 0 Reader 0 popped count=35822 Reader 1 popped count=32755 Reader 2 popped count=31420 Readers: duration=0.893811, success pop=99997, failed pops=261140 Writers: duration=0.841302, failed push=0d:\development\libcds\tests\unit\queue\queue_reader_writer.cpp(253) : CPPUNIT_CH ECK(nTotalPops + nPostTestPops == nQueueSize: popped=99997 必须是 99999); 测试弹出序列的一致性...
警告：Visual Leak Detector 检测到内存泄漏！---------- Block 116955 at 0x00DB33D0: 8 bytes ---------- Leak Hash: 0xD835B211, Count: 1, Total 8 bytes Call Stack (TID 2836): ucrtbased.dll！ malloc() f:\dd\vctools\crt\vcstartup\src\heap\new_scalar.cpp (19): unit-queue_d.exe!operator new() + 0x9 bytes d:\development\libcds\cds\container\ williams_queue.h (297): unit-queue_d.exe !cds::container::WilliamsQueue::push() d:\development\libcds\tests\unit\queue\queue_reader_writer.cpp (85): unit-que eue_d。 exe!queue::Queue_ReaderWriter::WriterThread >::test() + 0xF 字节
然后我通过添加带有数据删除的节点析构函数来修复内存泄漏。但测试失败仍然存在。
试运行代码
namespace {
static size_t s_nReaderThreadCount = 4;
static size_t s_nWriterThreadCount = 4;
static size_t s_nQueueSize = 100000; // by default 4000000;
struct Value {
size_t nNo;
size_t nWriterNo;
};
}
class Queue_ReaderWriter: public CppUnitMini::TestCase
{
template <class Queue>
class WriterThread: public CppUnitMini::TestThread
{
public:
Queue& m_Queue;
double m_fTime;
size_t m_nPushFailed;
virtual void test()
{
size_t nPushCount = getTest().m_nThreadPushCount;
Value v;
v.nWriterNo = m_nThreadNo;
v.nNo = 0;
m_nPushFailed = 0;
m_fTime = m_Timer.duration();
while ( v.nNo < nPushCount ) {
if (m_Queue.push(v)) {
++v.nNo;
}
else
++m_nPushFailed;
}
m_fTime = m_Timer.duration() - m_fTime;
getTest().m_nWriterDone.fetch_add( 1 );
}
};
template <class Queue>
class ReaderThread: public CppUnitMini::TestThread
{
public:
Queue& m_Queue;
double m_fTime;
size_t m_nPopEmpty;
size_t m_nPopped;
size_t m_nBadWriter;
typedef std::vector<size_t> TPoppedData;
std::vector<TPoppedData> m_WriterData;
virtual void test()
{
m_nPopEmpty = 0;
m_nPopped = 0;
m_nBadWriter = 0;
const size_t nTotalWriters = s_nWriterThreadCount;
Value v;
m_fTime = m_Timer.duration();
while ( true ) {
if ( m_Queue.pop( v ) ) {
++m_nPopped;
if ( /*v.nWriterNo >= 0 &&*/ v.nWriterNo < nTotalWriters )
m_WriterData[ v.nWriterNo ].push_back( v.nNo );
else
++m_nBadWriter;
}
else
++m_nPopEmpty;
if ( m_Queue.empty() ) {
if ( getTest().m_nWriterDone.load() >= nTotalWriters ) {
CPPUNIT_MSG(" Item count: " << m_Queue.size());
if ( m_Queue.empty() )
break;
}
}
}
m_fTime = m_Timer.duration() - m_fTime;
}
};
protected:
size_t m_nThreadPushCount;
atomics::atomic<size_t> m_nWriterDone;
protected:
template <class Queue>
void analyze( CppUnitMini::ThreadPool& pool, Queue& testQueue, size_t /*nLeftOffset*/ = 0, size_t nRightOffset = 0 )
{
typedef ReaderThread<Queue> Reader;
typedef WriterThread<Queue> Writer;
size_t nPostTestPops = 0;
{
Value v;
while ( testQueue.pop( v ))
++nPostTestPops;
}
CPPUNIT_MSG(" Post pops: " << nPostTestPops);
double fTimeWriter = 0;
double fTimeReader = 0;
size_t nTotalPops = 0;
size_t nPopFalse = 0;
size_t nPoppedItems = 0;
size_t nPushFailed = 0;
std::vector< Reader * > arrReaders;
for ( CppUnitMini::ThreadPool::iterator it = pool.begin(); it != pool.end(); ++it ) {
Reader * pReader = dynamic_cast<Reader *>( *it );
if ( pReader ) {
fTimeReader += pReader->m_fTime;
nTotalPops += pReader->m_nPopped;
nPopFalse += pReader->m_nPopEmpty;
arrReaders.push_back( pReader );
CPPUNIT_CHECK_EX( pReader->m_nBadWriter == 0, "reader " << pReader->m_nThreadNo << " bad writer event count=" << pReader->m_nBadWriter );
size_t nPopped = 0;
for ( size_t n = 0; n < s_nWriterThreadCount; ++n )
nPopped += pReader->m_WriterData[n].size();
CPPUNIT_MSG( " Reader " << pReader->m_nThreadNo - s_nWriterThreadCount << " popped count=" << nPopped );
nPoppedItems += nPopped;
}
else {
Writer * pWriter = dynamic_cast<Writer *>( *it );
CPPUNIT_ASSERT( pWriter != nullptr );
fTimeWriter += pWriter->m_fTime;
nPushFailed += pWriter->m_nPushFailed;
if ( !boost::is_base_of<cds::bounded_container, Queue>::value ) {
CPPUNIT_CHECK_EX( pWriter->m_nPushFailed == 0,
"writer " << pWriter->m_nThreadNo << " push failed count=" << pWriter->m_nPushFailed );
}
}
}
CPPUNIT_CHECK_EX( nTotalPops == nPoppedItems, "nTotalPops=" << nTotalPops << ", nPoppedItems=" << nPoppedItems );
CPPUNIT_MSG( " Readers: duration=" << fTimeReader / s_nReaderThreadCount << ", success pop=" << nTotalPops << ", failed pops=" << nPopFalse );
CPPUNIT_MSG( " Writers: duration=" << fTimeWriter / s_nWriterThreadCount << ", failed push=" << nPushFailed );
size_t nQueueSize = m_nThreadPushCount * s_nWriterThreadCount;
CPPUNIT_CHECK_EX( nTotalPops + nPostTestPops == nQueueSize, "popped=" << nTotalPops + nPostTestPops << " must be " << nQueueSize );
CPPUNIT_CHECK( testQueue.empty() );
}
template <class Queue>
void test()
{
m_nThreadPushCount = s_nQueueSize / s_nWriterThreadCount;
CPPUNIT_MSG( " reader count=" << s_nReaderThreadCount << " writer count=" << s_nWriterThreadCount
<< " item count=" << m_nThreadPushCount * s_nWriterThreadCount << "..." );
Queue testQueue;
CppUnitMini::ThreadPool pool( *this );
m_nWriterDone.store( 0 );
// Writers must be first
pool.add( new WriterThread<Queue>( pool, testQueue ), s_nWriterThreadCount );
pool.add( new ReaderThread<Queue>( pool, testQueue ), s_nReaderThreadCount );
pool.run();
analyze( pool, testQueue );
CPPUNIT_MSG( testQueue.statistics() );
}