packetquemanager.cpp

一个实现可以在进程间或者线程间通信的高效环形队列类

// PacketQueManager.cpp: implementation of the CPacketQueManager class.
//
//////////////////////////////////////////////////////////////////////

#include "stdafx.h"

#include "PacketQueManager.h"
#include <string.h>

#ifdef _DEBUG
#define DEBUG_DO(x)	(x)
#else
#define DEBUG_DO(x)	
#endif//_DEBUG

//////////////////////////////////////////////////////////////////////
// Construction/Destruction
//////////////////////////////////////////////////////////////////////



CPacketQueManager::CPacketQueManager()
:m_dwUserType(0),m_pShareBufferHeader(NULL),
m_pInfoBuffer(NULL),m_dwEndPos(0),m_dwLastPackLen(0),
m_bCreateBuffer(false),m_dwNowWritePos((ACE_UINT32)-1)
{
}

CPacketQueManager::~CPacketQueManager()
{
	Destroy();
}

CPacketQueManager::CPacketQueManager(const CPacketQueManager& pqm)
{
	*this = pqm;
}
const CPacketQueManager& CPacketQueManager::operator = (const CPacketQueManager& pqm)
{
	return *this;
}

void * CPacketQueManager::Create(ACE_UINT32 dwUserType, void * pBuffer, ACE_UINT32 dwSizeBuffer )
{	
	m_dwUserType = dwUserType;

	if( (dwUserType&WRITER) != 0 )
	{		
		//如果是开辟内存者,传入的内存大小至少是100.
		if( dwSizeBuffer < 100 )
			return NULL;
		
		m_bCreateBuffer = (pBuffer==NULL);
		
		//+sizeof(PACKET_HEADER)的目的是为了当前空间结尾内存不够放下一个包的时候，
		//后面内存足够放下一个PACKET_HEADER结构的数据置为无效。
		if( m_bCreateBuffer )
		{
			pBuffer = new BYTE[dwSizeBuffer];
		}
		
		m_pShareBufferHeader = (SHARE_BUFFER_HEADER *)pBuffer;
		m_pInfoBuffer		 = (BYTE *)pBuffer + sizeof(SHARE_BUFFER_HEADER);
		
		m_pShareBufferHeader->dwReadPos = 0;
		m_pShareBufferHeader->dwWritePos = 0;
		
		m_dwEndPos = dwSizeBuffer - (sizeof(SHARE_BUFFER_HEADER)+sizeof(PACKET_HEADER));
		
		m_pShareBufferHeader->dwSizeBuffer = m_dwEndPos;
		
		m_dwLastPackLen = 0;
	}
	else
	{
		//注意：m_pShareBufferHeader中内容不能初始化，因为Writer已经初始化了	
		m_bCreateBuffer = false;

		m_pShareBufferHeader = (SHARE_BUFFER_HEADER *)pBuffer;
		m_pInfoBuffer		 = (BYTE *)pBuffer + sizeof(SHARE_BUFFER_HEADER);

		m_dwEndPos = m_pShareBufferHeader->dwSizeBuffer;
	}
 		
	return pBuffer;
}


void CPacketQueManager::Destroy()
{
	if( m_bCreateBuffer )
	{
		delete []((BYTE *)m_pShareBufferHeader);
		m_bCreateBuffer = false;
	}
}

//只有在m_dwUserType&WRITERER != 0时才能调用
BYTE *  CPacketQueManager::ReserveBufferWritePacket(ACE_UINT32 dwPacketSize)
{
//	ASSERT( m_pShareBufferHeader && (m_dwUserType&WRITER)!=0 && dwPacketSize > 0 );
	
	m_dwNowWritePos=(ACE_UINT32)-1;	

	ACE_UINT32 dwRequiredMinFreeLen = GetFourByteAlign(dwPacketSize + sizeof(PACKET_HEADER));

	if (m_pShareBufferHeader->dwWritePos < m_pShareBufferHeader->dwReadPos)
	{
		//计算头部指针与尾部指针之间的空闲长度
		ACE_UINT32 dwSpaceLeft = m_pShareBufferHeader->dwReadPos - m_pShareBufferHeader->dwWritePos;
		
		//注意:不能用dwSpaceLeft>=dwRequiredMinFreeLen
		if (dwSpaceLeft > dwRequiredMinFreeLen)//头部指针与尾部指针之间的空闲缓冲区可以分配
		{
			m_dwNowWritePos = m_pShareBufferHeader->dwWritePos;
		}
	}
	else
	{
		ACE_UINT32 dwSpaceLeft = m_dwEndPos - m_pShareBufferHeader->dwWritePos;
		
		//注意:能(可以)用dwSpaceLeft>=dwRequiredMinFreeLen
		if (dwSpaceLeft >= dwRequiredMinFreeLen)//尾部的空闲缓冲区可以直接分配
		{
			m_dwNowWritePos = m_pShareBufferHeader->dwWritePos;
		}
		else
		{			
			//注意:不能用m_pShareBufferHeader->dwReadPos>=dwRequiredMinFreeLen
			if (m_pShareBufferHeader->dwReadPos > dwRequiredMinFreeLen)//前面的空闲缓冲区可以分配
			{
				//将尾部空闲的缓冲区填充成为一个无效帧
				//实际长度为0，表示一个填空的无效帧
				((PACKET_HEADER *)(m_pInfoBuffer + m_pShareBufferHeader->dwWritePos))->dwOrgSize = 0;
				
				//跳过无效帧
				m_dwNowWritePos = 0;
			}
		}
	}
	
	if (m_dwNowWritePos != (ACE_UINT32)-1)
	{
		m_dwLastPackLen = dwPacketSize;

		return (m_pInfoBuffer+m_dwNowWritePos+sizeof(PACKET_HEADER));
	}
	
	return NULL;
	
}

void CPacketQueManager::WriteReservedBuffer()
{
	WriteReservedBufferUseRealSize(m_dwLastPackLen);
}

void CPacketQueManager::WriteReservedBufferUseRealSize(ACE_UINT32 dwRealPacketSize)
{
	//ASSERT(m_dwNowWritePos!=-1);
	//ASSERT( dwRealPacketSize <= m_dwLastPackLen && m_dwLastPackLen > 0 );
		
	((PACKET_HEADER *)(m_pInfoBuffer + m_dwNowWritePos))->dwOrgSize = dwRealPacketSize;
	
	//调整大小
	dwRealPacketSize = GetFourByteAlign(dwRealPacketSize+sizeof(PACKET_HEADER));	

	m_dwNowWritePos += dwRealPacketSize;	
	
	m_pShareBufferHeader->dwWritePos = m_dwNowWritePos;	
	
	DEBUG_DO(m_dwNowWritePos=-1);	
}


bool CPacketQueManager::WritePacket(const void * packet,ACE_UINT32 dwPacketSize)
{
	BYTE * pBuffer = ReserveBufferWritePacket(dwPacketSize);
	
	if( pBuffer != NULL )
	{
		memcpy( pBuffer, packet, dwPacketSize );

		WriteReservedBufferUseRealSize(dwPacketSize);

		return true;		
	}

	return false;
}


/////////////////////////////////////////////////////////////////////////////////////

//
//从缓冲区中取走当前头部指示的包的信息，但不拷贝该包，头指针不移动
//
//返回值：0--没有取到数据信息
//        数据包长度--取到当前包信息，如果len为0，该包是无效的
//
//
ACE_UINT32 CPacketQueManager::ReadPacket(void** ppPakcet)
{
	//ASSERT( m_pShareBufferHeader && (m_dwUserType&READER)!=0 );	

	if( m_pShareBufferHeader->dwWritePos == m_pShareBufferHeader->dwReadPos )
		return 0;
	
	ACE_UINT32 dwPacketLen = ((PACKET_HEADER *)(m_pInfoBuffer + m_pShareBufferHeader->dwReadPos))->dwOrgSize;
	
	if (dwPacketLen == 0)
	{	//是一个添写空白缓冲区的空包，应该跳过去,继续读
		m_pShareBufferHeader->dwReadPos = 0;
		
		return ReadPacket(ppPakcet);
	}
	else
	{
		*ppPakcet = m_pInfoBuffer + m_pShareBufferHeader->dwReadPos + sizeof(PACKET_HEADER);			
	}		

	return dwPacketLen;	
}

void CPacketQueManager::SkipToNextPacket()
{
	//ASSERT( m_pShareBufferHeader && (m_dwUserType&READER)!=0 );

	//ASSERT(m_pShareBufferHeader->dwReadPos != m_pShareBufferHeader->dwWritePos );

	ACE_UINT32 dwPacketLen = ((PACKET_HEADER *)(m_pInfoBuffer + m_pShareBufferHeader->dwReadPos))->dwOrgSize;

	//ASSERT(dwPacketLen!=0);

	dwPacketLen = GetFourByteAlign(dwPacketLen);

	ACE_UINT32 dwNextReadPos = m_pShareBufferHeader->dwReadPos + dwPacketLen + sizeof(PACKET_HEADER);
	
	if( dwNextReadPos >= m_dwEndPos )
		dwNextReadPos = 0;

	m_pShareBufferHeader->dwReadPos = dwNextReadPos;
}