FSBII（五）IOCP

港控/mmm° 2022-06-13 06:44 152阅读 0赞

/********************************************************************
    	created:	2003/02/14
    	file base:	IOBuffer
    	file ext:	h
    	author:		liupeng
    	
    	purpose:	Header file for CIOBuffer class
    *********************************************************************/
    #ifndef __INCLUDE_IOBUFFER_H__
    #define __INCLUDE_IOBUFFER_H__
    
    #if defined (_MSC_VER) && (_MSC_VER >= 1020)
    	#pragma once
    #endif
    
    #ifndef _WINDOWS_
    	#define WIN32_LEAN_AND_MEAN
    		#include <windows.h>
    	#undef WIN32_LEAN_AND_MEAN
    #endif
    
    /*
     * Identifier was truncated to '255' characters 
     * in the debug information
     */
    #pragma warning(disable : 4786)                               
    
    #include <winsock2.h>
    
    #include "CriticalSection.h" 
    #include "tstring.h"
    
    #include "NodeList.h"
    #include "OpaqueUserData.h"
    #include "Macro.h"
    
    #include <map>
    
    /*
     * Nonstandard extension used : zero-sized array in struct/union
     */
    #pragma warning(disable: 4200)
    
    /*
     * namespace OnlineGameLib::Win32
     */
    
    namespace OnlineGameLib {
    namespace Win32 {
    
    /*
     * CIOBuffer
     */
    class CIOBuffer : 
    	public OVERLAPPED, 
    	public CNodeList::Node, 
    	public COpaqueUserData
    {
    public:
    	
    	class Allocator;
    	friend class Allocator;
    	
    	class InOrderBufferList;
    	
    	WSABUF *GetWSABUF() const { return const_cast< WSABUF * >( &m_wsabuf ); };
    	
    	size_t GetUsed() const { return m_used; };
    	size_t GetSize() const { return m_size; };
    	
    	const BYTE *GetBuffer() const { return m_buffer_ptr; };
    	
    	void SetupZeroByteRead();
    	void SetupRead();
    	void SetupWrite();
    	
    	void AddData( const char * const pData, size_t dataLength );
    	void AddData( const BYTE * const pData, size_t dataLength );
    	void AddData( BYTE data );
    	
    	void Use( size_t dataUsed ) { m_used += dataUsed; };
    	
    	CIOBuffer *SplitBuffer( size_t bytesToRemove );
    	void RemoveBuffer( size_t bytesToRemove );	
    	CIOBuffer *AllocateNewBuffer() const;
    	
    	void ConsumeAndRemove( size_t bytesToRemove );
    	
    	void Empty();
    	
    	void AddRef() { ::InterlockedIncrement( &m_ref ); };
    	void Release();
    	
    	size_t GetOperation() const { return m_operation; };
    	void SetOperation( size_t operation ) { m_operation = operation; };
    	
    	size_t GetSequenceNumber() const { return m_sequenceNumber; };
    	void SetSequenceNumber( size_t sequenceNumber ) { m_sequenceNumber = sequenceNumber; };
    
    private:
    
    	size_t m_operation;
    	size_t m_sequenceNumber;
    	
    	WSABUF m_wsabuf;
    	
    	Allocator &m_allocator;
    	
    	long m_ref;
    	const size_t m_size;
    	size_t m_used;
    
    	/*
    	 * Start of the actual buffer, must remain the last
    	 * data member in the class.
    	 */
    	BYTE *m_buffer_ptr;			// four bytes aligned
    	
    	//BYTE m_buffer_base_addr[0];
    
    private:
    
    	static void *operator new( size_t objSize, size_t bufferSize );
    	static void operator delete( void *pObj, size_t bufferSize );
    	
    	CIOBuffer( Allocator &m_allocator, size_t size );
    	~CIOBuffer();
    	
    	/*
         * No copies do not implement
         */
    	CIOBuffer( const CIOBuffer &rhs );
    	CIOBuffer &operator=( const CIOBuffer &rhs );
    };
    
    /*
     * CIOBuffer::Allocator
     */
    
    class CIOBuffer::Allocator
    {
    public:
    
          friend class CIOBuffer;
    
          explicit Allocator( size_t bufferSize, size_t maxFreeBuffers );
    
          virtual ~Allocator();
    
          CIOBuffer *Allocate();
    
          size_t GetBufferSize() const { return m_bufferSize; };
    
    protected:
    
          void Flush();
    
    private:
    
          void Release( CIOBuffer *pBuffer );
    
          virtual void OnBufferCreated() {}
          virtual void OnBufferAllocated() {}
          virtual void OnBufferReleased() {}
    
          void DestroyBuffer( CIOBuffer *pBuffer );
    
          const size_t m_bufferSize;
    
          typedef TNodeList< CIOBuffer > BufferList;
          
          BufferList m_freeList;
          BufferList m_activeList;
          
          const size_t m_maxFreeBuffers;
    
          CCriticalSection m_criticalSection;
    
          /*
    	   * No copies do not implement
    	   */
          Allocator( const Allocator &rhs );
          Allocator &operator=( const Allocator &rhs );
    };
    
    /*
     * CIOBuffer::InOrderBufferList
     */
    
    class CIOBuffer::InOrderBufferList
    {
    public:
    
          explicit InOrderBufferList( CCriticalSection &lock );
    
          void AddBuffer( CIOBuffer *pBuffer );
    
          void ProcessBuffer();
    
          CIOBuffer *ProcessAndGetNext();
    
          CIOBuffer *GetNext();
          CIOBuffer *GetNext( CIOBuffer *pBuffer );
    
          void Reset();
    
          bool Empty() const { return m_list.empty(); };
    
    private:
    
          size_t m_next;
       
          typedef std::map< size_t, CIOBuffer * > BufferSequence;
    
          BufferSequence m_list;
    
          CCriticalSection &m_criticalSection;
    };
    
    inline void CIOBuffer::ConsumeAndRemove( size_t bytesToRemove )
    {
    	m_used -= bytesToRemove;
    
    	memmove( m_buffer_ptr, m_buffer_ptr + bytesToRemove, m_used );
    }
    
    inline void CIOBuffer::SetupZeroByteRead()
    {
    	m_wsabuf.buf = reinterpret_cast< char * >( m_buffer_ptr );
    
    	m_wsabuf.len = 0; 
    }
    
    inline void CIOBuffer::SetupRead()
    {
    	if ( m_used == 0 )
    	{
    		m_wsabuf.buf = reinterpret_cast< char * >( m_buffer_ptr );
    		m_wsabuf.len = m_size; 
    	}
    	else
    	{
    		m_wsabuf.buf = reinterpret_cast< char * >( m_buffer_ptr ) + m_used;
    		m_wsabuf.len = m_size - m_used;
       }
    }
    
    inline void CIOBuffer::SetupWrite()
    {
    	m_wsabuf.buf = reinterpret_cast< char * >( m_buffer_ptr );
    	m_wsabuf.len = m_used;
    
    	m_used = 0;
    }
    
    } // End of namespace OnlineGameLib
    } // End of namespace Win32
    
    
    #endif //__INCLUDE_IOBUFFER_H__

#include "stdafx.h"
    #include "KWin32.h"
    #include "IOBuffer.h"
    
    #include "Exception.h"
    #include "Utils.h"
    
    /*
     * namespace OnlineGameLib::Win32
     */
    
    namespace OnlineGameLib {
    namespace Win32 {
    
    CIOBuffer::CIOBuffer( Allocator &allocator, size_t size )
    	:  m_operation( 0 )
          , m_sequenceNumber( 0 )
          , m_allocator( allocator )
          , m_ref( 1 )
          , m_size( size )
          , m_used( 0 )
    {
    	//( ( BYTE * )( ( DWORD )( pMemory + 3 ) & ( ~3 ) ) )
    	m_buffer_ptr = ::new BYTE[ size ];
    
    	memset( this, 0, sizeof( OVERLAPPED ) );
    
    	Empty();
    }
    
    CIOBuffer::~CIOBuffer()
    {
        if (m_buffer_ptr)
        {
            ::delete []m_buffer_ptr;
            m_buffer_ptr = NULL;
        }
    }
    
    void CIOBuffer::Empty()
    {
    	m_wsabuf.buf = reinterpret_cast< char * >( m_buffer_ptr );
    	m_wsabuf.len = m_size;
    
    	m_used = 0;
    }
    
    void *CIOBuffer::operator new( size_t objectSize, size_t /* bufferSize */ )
    {
    	/*
    	 * ASSERT( sizeof( DWORD ) == 4 );
    	 *
    	 * For four bytes aligned base on win32 system
    	 */
    
    	void *pMem = ::new char[ objectSize ]; // + bufferSize + 4 ];
    
    	return pMem;
    }
    
    void CIOBuffer::operator delete( void *pObject, size_t /* bufferSize*/ )
    {
        if (pObject)
        {
            ::delete [](char *)pObject;
            pObject = NULL;
        }
    }
    
    CIOBuffer *CIOBuffer::SplitBuffer( size_t bytesToRemove )
    {
    	CIOBuffer *pNewBuffer = m_allocator.Allocate();
    
    	pNewBuffer->AddData( m_buffer_ptr, bytesToRemove );
    
    	m_used -= bytesToRemove;
    
    	memmove( m_buffer_ptr, m_buffer_ptr + bytesToRemove, m_used );
    
    	return pNewBuffer;
    }
    
    VOID CIOBuffer::RemoveBuffer( size_t bytesToRemove )
    {
    	if ( m_used < bytesToRemove )
    		return;
    
    	m_used -= bytesToRemove;
    
    	memmove( m_buffer_ptr, m_buffer_ptr + bytesToRemove, m_used );
    }
    
    CIOBuffer *CIOBuffer::AllocateNewBuffer() const
    {
    	return m_allocator.Allocate();
    }
    
    void CIOBuffer::AddData( const char * const pData, size_t dataLength )
    {
    	if (dataLength > m_size - m_used)
    	{
    		DEBUG_ONLY( Message( "CIOBuffer::AddData - Not enough space in buffer!" ) );
    		
    		throw CException( _T("CIOBuffer::AddData"), _T("Not enough space in buffer") );
    	}
    
    	memcpy( m_buffer_ptr + m_used, pData, dataLength );
    
    	m_used += dataLength;
    }
    
    void CIOBuffer::AddData( const BYTE * const pData, size_t dataLength )
    {
    	AddData( reinterpret_cast< const char * >( pData ), dataLength );
    }
    
    void CIOBuffer::AddData( BYTE data )
    {
    	AddData( &data, 1 );
    }
    
    void CIOBuffer::Release()
    {
    	if (m_ref == 0)
    	{
    		/*
    		 * Error! double release
    		 */
    		throw CException( _T("CIOBuffer::Release()"), _T("m_ref is already zero") );
    	}
    
    	if ( 0 == ::InterlockedDecrement( &m_ref ) )
    	{
    		m_sequenceNumber = 0;
    		m_operation = 0;
    		m_used = 0;
    
    		m_allocator.Release( this );
    	}
    }
    
    /*
     * CIOBuffer::Allocator
     */
    
    CIOBuffer::Allocator::Allocator( size_t bufferSize, size_t maxFreeBuffers )
       :  m_bufferSize( bufferSize ),
          m_maxFreeBuffers( maxFreeBuffers )
    {
    
    }
    
    CIOBuffer::Allocator::~Allocator()
    {
    	try
    	{
    		Flush();
    	}
    	catch(...)
    	{
    		TRACE( "CIOBuffer::Allocator::~Allocator exception!" );
    	}
    }
    
    CIOBuffer *CIOBuffer::Allocator::Allocate()
    {
    	CCriticalSection::Owner lock( m_criticalSection );
    
    	CIOBuffer *pBuffer = 0;
    
    	if ( !m_freeList.Empty() )
    	{
    		pBuffer = m_freeList.PopNode();
    
    		pBuffer->AddRef();
    	}
    	else
    	{
    		pBuffer = new( m_bufferSize )CIOBuffer( *this, m_bufferSize );
    		
    		if ( !pBuffer )
    		{
    			throw CException( _T("CIOBuffer::Allocator::Allocate()"), _T("Out of memory") );
    		}
    		
    		OnBufferCreated();
    	}
    
    	m_activeList.PushNode( pBuffer );
    	
    	OnBufferAllocated();
    	
    	return pBuffer;
    }
    
    void CIOBuffer::Allocator::Release( CIOBuffer *pBuffer )
    {
    	if ( !pBuffer )
    	{
    		throw CException( _T("CIOBuffer::Allocator::Release()"), _T("pBuffer is null") );
    	}
    	
    	CCriticalSection::Owner lock( m_criticalSection );
    	
    	OnBufferReleased();
    	
    	/*
    	 * unlink from the in use list
    	 */	
    	pBuffer->RemoveFromList();
    	
    	if ( m_maxFreeBuffers == 0 || m_freeList.Count() < m_maxFreeBuffers )
    	{
    		pBuffer->Empty();           
    		
    		/*
    		 * add to the free list
    		 */
    		
    		m_freeList.PushNode( pBuffer );
    	}
    	else
    	{
    		DestroyBuffer( pBuffer );
    	}
    }
    
    void CIOBuffer::Allocator::DestroyBuffer( CIOBuffer *pBuffer )
    {
    	SAFE_DELETE( pBuffer );
    }
    
    void CIOBuffer::Allocator::Flush()
    {
    	CCriticalSection::Owner lock( m_criticalSection );
    	
    	while ( !m_activeList.Empty() )
    	{
    		OnBufferReleased();
    		
    		DestroyBuffer( m_activeList.PopNode() );
    	}
    	
    	while ( !m_freeList.Empty() )
    	{
    		DestroyBuffer( m_freeList.PopNode() );
    	}
    }
    
    /*
     * CIOBuffer::InOrderBufferList
     */
    
    CIOBuffer::InOrderBufferList::InOrderBufferList(
    				CCriticalSection &criticalSection)
    			:   m_next(0),
    				m_criticalSection( criticalSection )
    {
    }
    
    void CIOBuffer::InOrderBufferList::AddBuffer( CIOBuffer *pBuffer )
    {
    	CCriticalSection::Owner lock( m_criticalSection );
    	
    	std::pair< BufferSequence::iterator, bool > result = 
    		m_list.insert( BufferSequence::value_type( pBuffer->GetSequenceNumber(), pBuffer ) );
    	
    	if ( !result.second )
    	{
    		DEBUG_ONLY( Output( _T("UNEXPECTED! element already in map!") ) );
    	}
    }
    
    CIOBuffer *CIOBuffer::InOrderBufferList::ProcessAndGetNext()
    {
    	CCriticalSection::Owner lock( m_criticalSection );
    	
    	m_next ++;
    	
    	CIOBuffer *pNext = 0;
    	
    	BufferSequence::iterator it;
    	
    	it = m_list.begin();
    	
    	if ( it != m_list.end() )
    	{
    		if ( it->first == m_next )
    		{
    			pNext = it->second;
    			
    			m_list.erase( it );
    		}
    		else
    		{
    			DEBUG_ONLY( Output( ToString( this ) +
    				_T(" Got buffer : ") + 
    				ToString( it->first ) +
    				_T("Want buffer : ") +
    				ToString( m_next ) ) );
    		}
    	}
    	
    	return pNext;
    }
    
    CIOBuffer *CIOBuffer::InOrderBufferList::GetNext()
    {
    	CCriticalSection::Owner lock( m_criticalSection );
    	
    	CIOBuffer *pNext = 0;
    	
    	BufferSequence::iterator it;
    	
    	it = m_list.begin();
    	
    	if ( it != m_list.end() )
    	{
    		if ( it->first == m_next )
    		{
    			pNext = it->second;
    			
    			m_list.erase(it);
    		}
    		else
    		{
    			DEBUG_ONLY( Output( ToString( this ) + 
    				_T(" Got buffer  : ") + 
    				ToString( it->first ) + 
    				_T("Want buffer : ") + 
    				ToString( m_next ) ) );
    		}
    	}
    	
    	return pNext;
    }
    
    CIOBuffer *CIOBuffer::InOrderBufferList::GetNext( CIOBuffer *pBuffer )
    {
    	CCriticalSection::Owner lock( m_criticalSection );
    	
    	if ( m_next == pBuffer->GetSequenceNumber() )
    	{
    		return pBuffer;
    	}
    	
    	std::pair< BufferSequence::iterator, bool > result = 
    		m_list.insert( BufferSequence::value_type( pBuffer->GetSequenceNumber(), pBuffer ) );
    	
    	if ( !result.second )
    	{
    		DEBUG_ONLY( Output( _T("UNEXPECTED! element already in map!") ) );
    	}
    	
    	CIOBuffer *pNext = 0;
    	
    	BufferSequence::iterator it;
    	
    	it = m_list.begin();
    	
    	if (it != m_list.end()) 
    	{
    		if (it->first == m_next)
    		{
    			pNext = it->second;
    			
    			m_list.erase(it);
    		}
    		else
    		{
    			DEBUG_ONLY( Output( ToString( this ) +
    				_T(" Got buffer  : ") + 
    				ToString( it->first ) + 
    				_T("Want buffer : ") + 
    				ToString( m_next ) ) );
    		}
    	}
    	
    	return pNext;
    }
    
    void CIOBuffer::InOrderBufferList::ProcessBuffer()
    {
    	CCriticalSection::Owner lock( m_criticalSection );
    
    	DEBUG_ONLY( Output( ToString( this ) +
    		_T(" Processed : ") + 
    		ToString( m_next ) ) );
    
    	m_next ++;
    }
    
    void CIOBuffer::InOrderBufferList::Reset()
    {
    	m_next = 0;
    
    	if ( !m_list.empty() )
    	{
    		DEBUG_ONLY( Output( _T("List not empty when reset !") ) );
    	}
    }
    
    } // End of namespace Win32
    } // End of namespace OnlineGameLib

/********************************************************************
    	created:	2003/02/14
    	file base:	IOCompletionPort
    	file ext:	h
    	author:		liupeng
    	
    	purpose:	Header file for CIOCompletionPort routines
    *********************************************************************/
    #ifndef __INCLUDE_IOCOMPLETIONPORT_H__
    #define __INCLUDE_IOCOMPLETIONPORT_H__
    
    #ifndef _WINDOWS_
    	#define WIN32_LEAN_AND_MEAN
    		#include <windows.h>
    	#undef WIN32_LEAN_AND_MEAN
    #endif
    
    #include "Macro.h"
    
    /*
     * namespace OnlineGameLib::Win32
     */
    
    namespace OnlineGameLib {
    namespace Win32 {
    
    /*
     * CIOCompletionPort
     */
    class CIOCompletionPort
    {
    public:
       
    	explicit CIOCompletionPort( size_t maxConcurrency );
    
    	~CIOCompletionPort();
    
    	void AssociateDevice( HANDLE hDevice, ULONG_PTR completionKey );
    
    	void PostStatus( ULONG_PTR completionKey, 
    		DWORD dwNumBytes = 0, 
    		OVERLAPPED *pOverlapped = 0 );
    
    	DWORD GetStatus( ULONG_PTR *pCompletionKey, 
    		PDWORD pdwNumBytes,
    		OVERLAPPED **ppOverlapped );
    
    	DWORD GetStatus( ULONG_PTR *pCompletionKey, 
    		PDWORD pdwNumBytes,
    		OVERLAPPED **ppOverlapped, 
    		DWORD dwMilliseconds );
    
    private:
          
    	HANDLE m_iocp;
    
    	/*
    	 * No copies do not implement
    	 */
    	CIOCompletionPort( const CIOCompletionPort &rhs );
    	CIOCompletionPort &operator=( const CIOCompletionPort &rhs );
    
    };
    
    } // End of namespace OnlineGameLib
    } // End of namespace Win32
    
    #endif //__INCLUDE_IOCOMPLETIONPORT_H__

#include "stdafx.h"
    #include "IOCompletionPort.h"
    #include "Win32Exception.h"
    #include "Macro.h"
    
    /*
     * namespace OnlineGameLib::Win32
     */
    
    namespace OnlineGameLib {
    namespace Win32 {
    
    CIOCompletionPort::CIOCompletionPort( size_t maxConcurrency )
    		: m_iocp( ::CreateIoCompletionPort( INVALID_HANDLE_VALUE, NULL, 0, maxConcurrency ) )
    {
    	if ( m_iocp == 0 )
    	{
    		throw CWin32Exception( _T("CIOCompletionPort::CIOCompletionPort() - CreateIoCompletionPort"), ::GetLastError() );
    	}
    }
    
    CIOCompletionPort::~CIOCompletionPort() 
    {
    	SAFE_CLOSEHANDLE( m_iocp );
    }
    
    void CIOCompletionPort::AssociateDevice( HANDLE hDevice, ULONG_PTR completionKey )
    {
    	if ( m_iocp != ::CreateIoCompletionPort( hDevice,
    									m_iocp, 
    									completionKey, 
    									0 ) )
    	{
    		throw CWin32Exception( _T("CIOCompletionPort::AssociateDevice() - CreateIoCompletionPort"), ::GetLastError() );
    	}
    }
    
    void CIOCompletionPort::PostStatus( ULONG_PTR completionKey, 
    			DWORD dwNumBytes /* = 0 */, 
    			OVERLAPPED *pOverlapped /* = 0 */) 
    {
    	if ( 0 == ::PostQueuedCompletionStatus( m_iocp, 
    						dwNumBytes, 
    						completionKey, 
    						pOverlapped ) )
    	{
    		throw CWin32Exception( _T("CIOCompletionPort::PostStatus() - PostQueuedCompletionStatus"), ::GetLastError() );
    	}
    }
    
    DWORD CIOCompletionPort::GetStatus( ULONG_PTR *pCompletionKey, 
    				PDWORD pdwNumBytes,
    				OVERLAPPED **ppOverlapped )
    {
    	if ( 0 == ::GetQueuedCompletionStatus( m_iocp, 
    						pdwNumBytes, 
    						pCompletionKey, 
    						ppOverlapped, 
    						INFINITE ) )
    	{
    		return ::GetLastError();
    	}
    
    	return S_OK;
    }
    
    DWORD CIOCompletionPort::GetStatus( ULONG_PTR *pCompletionKey, 
    				PDWORD pdwNumBytes,
    				OVERLAPPED **ppOverlapped, 
    				DWORD dwMilliseconds)
    {
    	if ( 0 == ::GetQueuedCompletionStatus( m_iocp, 
    						pdwNumBytes, 
    						pCompletionKey, 
    						ppOverlapped, 
    						dwMilliseconds ) )
    	{
    		DWORD lastError = ::GetLastError();
    
    		if ( lastError != WAIT_TIMEOUT )
    		{
    			return lastError;
    		}
    
    		return S_FALSE;
    	}
    
    	return S_OK;
    }
    
    } // End of namespace OnlineGameLib
    } // End of namespace Win32