Boost lock-free Queue 성능 테스트

Boost 1.55 버전 기준으로 작성했습니다.

#pragma once

#include <Windows.h>

#include <boost/lockfree/queue.hpp>

#include <boost/lockfree/spsc_queue.hpp>

#include <boost/pool/pool_alloc.hpp>

#include <boost/pool/object_pool.hpp>

#include <boost/lockfree/policies.hpp>

#include <boost/thread.hpp>

#include <boost/timer.hpp>

#include <boost/progress.hpp>

#pragma pack(push,4)

#define PACKET_TOTAL_SIZE 2048

#define PACKET_EOL_LENGTH 10

#define PACKET_MAX_BUFFER 2010

#define PACKET_HDR_SIZE 28//sizeof( UINT64 )

class CPacket

{

public:

union

{

// Restrict: ACK (Server to Client)

struct

{

UCHAR m_ucOwner;

UCHAR m_ucErr;

};

// Restrict: REQ (Client to Server)

USHORT m_usSerial; // Security: Record & Replay

};

USHORT m_usCmd; // internal command protocol

USHORT m_usPacketID; // packet id

USHORT m_usSize; // Stream data size (+PACKETHDR)

INT m_nHostID; //  Host ID

DWORD64 m_dw64DBID; // DB ID

DWORD m_dwPacketOption; // packet option

DWORD m_dwDataCRC; // data crc

// EOL 사이즈를 포함한 크기

BYTE m_lpData [ 2020 ];

public:

CPacket()

{

setEmpty();

}

CPacket( const CPacket& _Right )

{

this->m_nHostID = _Right.m_nHostID;

//memcpy( this, _Right.segment(), _Right.size() );

memcpy( this->m_lpData, _Right.m_lpData, sizeof( BYTE ) * 2020 );

}

//// 튜플 대입 연산자 선언

//CPacket& operator=( const CPacket& _Right ) _THROW0()

//{

// memcpy( this->m_lpData, _Right.m_lpData, sizeof( BYTE ) * 2020 );

// return (*this);

//}

/*inline

VOID copy_of ( CPacket& _Right )

{ memcpy( this, _Right, _Right.size() ); }*/

/*~CPacket()

{

}*/

//BYTE* segment () const { return reinterpret_cast<BYTE*>(this); }

VOID CPacket::setEmpty()

{

ZeroMemory( this, PACKET_TOTAL_SIZE );

m_usSize = PACKET_TOTAL_SIZE;

}

// 스트림 총 크기

//USHORT& size () const { return m_usSize; }

};

#define CAPACITY_MAX 50000

//

// que 1

// 

typedef boost::fast_pool_allocator< CPacket, boost::default_user_allocator_new_delete, boost::details::pool::null_mutex, 32, 0 > fast_pool_alloc_null_mutex;

fast_pool_alloc_null_mutex alloc1;

boost::lockfree::queue< CPacket, boost::lockfree::capacity<10000>, boost::lockfree::allocator< fast_pool_alloc_null_mutex > > que1( alloc1 );

//

// que 2

// 

typedef boost::fast_pool_allocator< CPacket > fast_pool_alloc;

fast_pool_alloc alloc2;

boost::lockfree::queue< CPacket, boost::lockfree::capacity<10000>, boost::lockfree::allocator< fast_pool_alloc > > que2( alloc2 );

//

// que 3

// 

boost::lockfree::spsc_queue< CPacket, boost::lockfree::capacity<10000> > que3;

//

// que 4

// 

//boost::lockfree::queue< CPacket > que4;

int sum1 = 0;

int sum2 = 0;

int sum3 = 0;

void produce()

{

int i = 0;

for( ; i <= CAPACITY_MAX; ++i )

{

CPacket p;

p.m_nHostID = i;

//que1.bounded_push( p );

while( !que1.bounded_push( p ) );

}

}

void consume()

{

int i = 0;

for( ; i <= CAPACITY_MAX; )

{

CPacket p;

while( que1.pop(p))

{

++i;

sum1 += p.m_nHostID;

}

}

}

void produce2()

{

int i = 0;

for( ; i <= CAPACITY_MAX; ++i )

{

CPacket p;

p.m_nHostID = i;

while( !que2.bounded_push( p ) );

}

}

void consume2()

{

int i = 0;

for( ; i <= CAPACITY_MAX; )

{

CPacket p;

while( que2.pop(p))

{

++i;

sum2 += p.m_nHostID;

}

}

}

void produce3()

{

int i = 0;

for( ; i <= CAPACITY_MAX; ++i )

{

CPacket p;

p.m_nHostID = i;

while( !que3.push( p ) );

}

}

void consume3()

{

int i = 0;

for( ; i <= CAPACITY_MAX; )

{

CPacket p;

while( que3.pop(p))

{

++i;

sum3 += p.m_nHostID;

}

}

}

void lock_free_queue_test()

{

{

// 프로그레스 타이머

boost::progress_timer progress_t;

sum1 = 0;

boost::thread t1(produce);

boost::thread t2(consume);

t1.join();

t2.join();

//consume();

std::cout << "boost::lockfree::queue< CPacket, boost::lockfree::capacity<60000>, boost::lockfree::allocator< fast_pool_alloc_null_mutex > >" << std::endl;

std::cout << sum1 << "\n";

}

Sleep(10);

{

// 프로그레스 타이머

boost::progress_timer progress_t;

sum1 = 0;

boost::thread t1(produce2);

boost::thread t2(consume2);

t1.join();

t2.join();

//consume2();

std::cout << "boost::lockfree::queue< CPacket, boost::lockfree::capacity<60000>, boost::lockfree::allocator< fast_pool_alloc > >" << std::endl;

std::cout << sum2 << "\n";

}

Sleep(10);

{

// 프로그레스 타이머

boost::progress_timer progress_t;

sum1 = 0;

boost::thread t1(produce3);

boost::thread t2(consume3);

t1.join();

t2.join();

//consume3();

std::cout << "boost::lockfree::queue< CPacket, boost::lockfree::capacity<60000> >" << std::endl;

std::cout << sum3 << "\n";

}

//typedef boost::lockfree::spsc_queue< ENK::CPacket, boost::lockfree::capacity<65535>, boost::lockfree::allocator >

}