开通博客赚积分
发布资源赚积分
分类

源码开发语言/平台
当前位置：首页 > 源码/资料 > 游戏 > 模拟服务器 > 查看源码
SocketServer.cpp
资源名称：SwordOnline.rar　[点击查看]
上传用户：dzyhzl
上传日期：2019-04-29
资源大小：56270k
文件大小：26k
源码类别：

模拟服务器

开发平台：

C/C++

SocketServer.cpp：源码内容
  1. #include "stdafx.h"
  2. #include "SocketServer.h"
  3. #include "IOCompletionPort.h"
  4. #include "Win32Exception.h"
  5. #include "Utils.h"
  6. #include "SystemInfo.h"
  7. #include "Socket.h"
  8. #include "Macro.h"
  9. #include "..heavenserverstage.h"
  11. #include <vector>
  13. #pragma comment(lib, "ws2_32.lib")
  15. /*
  16.  * Link options and warning
  17.  */
  18. #pragma message( "NOTE : --------------------OnlineGame [Server] : Announcement--------------------" )
  19. #pragma message( "NOTE : Developing a truly scalable Winsock Server using IO Completion Ports(IOCP)" )
  20. #pragma message( "NOTE : But this module depend on the microsoft platform" )
  21. #pragma message( "NOTE : Requirements :" )
  22. #pragma message( "NOTE :    * Windows NT/2000/XP: Included in Windows NT 4.0 and later." )
  23. #pragma message( "NOTE :    * Windows 95/98/Me: Unsupported." )
  24. #pragma message( "NOTE :" )
  25. #pragma message( "NOTE : liupeng xishanju.zhuhai.china 2003.1" )
  26. #pragma message( "NOTE : -----------------------------------------------------------------------------" )
  28. using std::vector;
  30. /*
  31.  * namespace OnlineGameLib::Win32
  32.  */
  34. namespace OnlineGameLib {
  35. namespace Win32 {
  37. /*
  38.  * Static helper methods
  39.  */
  40. static size_t CalculateNumberOfThreads( size_t numThreads );
  42. static size_t CalculateNumberOfThreads( size_t numThreads )
  43. {
  44. if ( numThreads == 0 )
  45. {
  46. CSystemInfo systemInfo;
  48. numThreads = systemInfo.dwNumberOfProcessors * 2 + 2;
  49. }
  51. return numThreads;
  52. }
  54. /*
  55.  * CSocketServer
  56.  */
  57. CSocketServer::CSocketServer(
  58.    size_t maxFreeSockets,
  59.    size_t maxFreeBuffers,
  60.    size_t bufferSize /* = 1024 */,
  61.    size_t numThreads /* = 0 */,
  62.    bool useSequenceNumbers /* = true */,
  63.    bool postZeroByteReads /* = false */)
  64.    :  CIOBuffer::Allocator( bufferSize, maxFreeBuffers ),
  65.   m_numThreads( CalculateNumberOfThreads( numThreads ) ),
  66.   m_listeningSocket( INVALID_SOCKET ),
  67.   m_iocp( 0 ),
  68.   m_maxFreeSockets( maxFreeSockets ),
  69.   m_useSequenceNumbers( useSequenceNumbers ),
  70.   m_postZeroByteReads( postZeroByteReads )
  71. {
  72. }
  74. CSocketServer::~CSocketServer()
  75. {
  76. }
  78. void CSocketServer::Open( unsigned long addressToListenOn, 
  79. unsigned short portToListenOn )
  80. {
  81. m_address = addressToListenOn;
  82. m_port = portToListenOn;
  83. }
  85. void CSocketServer::ReleaseSockets()
  86. {
  87. CCriticalSection::Owner lock( m_listManipulationSection );
  89. while ( m_activeList.Head() )
  90. {
  91. ReleaseSocket( m_activeList.Head() );
  92. }
  94. while ( m_freeList.Head() )
  95. {
  96. DestroySocket( m_freeList.PopNode() );
  97. }
  99. if ( m_activeList.Count() + m_freeList.Count() != 0 )
  100. {
  101. OnError( _T("CSocketServer::ReleaseSockets() - Leaked sockets") );
  102. }
  103. }
  105. void CSocketServer::StartAcceptingConnections()
  106. {
  107. if ( m_listeningSocket == INVALID_SOCKET )
  108. {
  109. OnStartAcceptingConnections();
  111. m_listeningSocket = CreateListeningSocket( m_address, m_port );
  113. m_acceptConnectionsEvent.Set();
  114. }
  115. }
  117. void CSocketServer::StopAcceptingConnections()
  118. {
  119. if ( m_listeningSocket != INVALID_SOCKET )
  120. {
  121. m_acceptConnectionsEvent.Reset();
  123. if ( 0 != ::closesocket( m_listeningSocket ) )
  124. {
  125. OnError( _T("CSocketServer::StopAcceptingConnections() - closesocket - ") +
  126. GetLastErrorMessage( ::WSAGetLastError() ) );
  127. }
  129. m_listeningSocket = INVALID_SOCKET;
  131. OnStopAcceptingConnections();
  132. }
  133. }
  135. void CSocketServer::InitiateShutdown()
  136. {
  137. StopAcceptingConnections();
  139.    /*
  140.     * enter m_listManipulationSection
  141. */
  142.    {
  143.    CCriticalSection::Owner lock( m_listManipulationSection );
  144.    
  145.    Socket *pSocket = m_activeList.Head();
  146.    
  147.    while ( pSocket )
  148.    {
  149.    Socket *pNext = SocketList::Next( pSocket );
  150.    
  151.    pSocket->AbortiveClose();
  152.    
  153.    pSocket = pNext;
  154.    }
  155.    }
  156.    /*
  157.     * leave m_listManipulationSection
  158. */
  160.    /*
  161.     * signal that the dispatch thread should shut down all worker threads and then exit
  162. */
  163.    m_shutdownEvent.Set();
  164.    
  165.    OnShutdownInitiated();
  166. }
  168. void CSocketServer::WaitForShutdownToComplete()
  169. {
  170. /*
  171.  * If we havent already started a shut down, do so...
  172.  */
  173. InitiateShutdown();
  175. Wait();
  177. ReleaseSockets();
  178. Flush();
  179. }
  181. SOCKET CSocketServer::CreateListeningSocket( unsigned long address, unsigned short port )
  182. {
  183. SOCKET s = ::WSASocket( AF_INET, 
  184. SOCK_STREAM, 
  185. IPPROTO_IP, 
  186. NULL, 
  187. 0, 
  188. WSA_FLAG_OVERLAPPED /* for IOCP */ );
  190. if ( s == INVALID_SOCKET )
  191. {
  192. throw CWin32Exception( _T("CSocket::CreateListeningSocket()"), 
  193. ::WSAGetLastError() );
  194. }
  196. CSocket listeningSocket( s );
  198. CSocket::InternetAddress localAddress( address, port );
  200. listeningSocket.Bind( localAddress );
  202. listeningSocket.Listen( 5 );
  204. return listeningSocket.Detatch();
  205. }
  207. CSocketServer::WorkerThread *CSocketServer::CreateWorkerThread(
  208. CIOCompletionPort &iocp )
  209. {
  210. return new WorkerThread( *this, iocp );
  211. }
  213. int CSocketServer::Run()
  214. {
  215. try
  216. {
  217. vector< WorkerThread * > workers;
  219. workers.reserve( m_numThreads );
  221. for ( size_t i = 0; i < m_numThreads; ++i )
  222. {
  223. WorkerThread *pThread = CreateWorkerThread( m_iocp ); 
  225. workers.push_back( pThread );
  227. pThread->Start();
  228. }
  230. HANDLE handlesToWaitFor[2];
  232. handlesToWaitFor[0] = m_shutdownEvent.GetEvent();
  233. handlesToWaitFor[1] = m_acceptConnectionsEvent.GetEvent();
  235. while ( !m_shutdownEvent.Wait( 0 ) )
  236. {
  237. DWORD waitResult = ::WaitForMultipleObjects( 2,
  238. handlesToWaitFor, 
  239. false, 
  240. INFINITE );
  242. if ( waitResult == WAIT_OBJECT_0 )
  243. {
  244. /*
  245.  * Time to shutdown
  246.  */
  247. break;
  248. }
  249. else if (waitResult == WAIT_OBJECT_0 + 1)
  250. {
  251. /*
  252.  * accept connections
  253.  */
  254. while ( !m_shutdownEvent.Wait(0) && 
  255. m_acceptConnectionsEvent.Wait( 0 ) )
  256. {
  257. CIOBuffer *pAddress = Allocate();
  259. int addressSize = (int)pAddress->GetSize();
  261. SOCKET acceptedSocket = ::WSAAccept(
  262. m_listeningSocket, 
  263. reinterpret_cast< sockaddr * >( const_cast< BYTE * >( pAddress->GetBuffer() ) ), 
  264. &addressSize, 
  265. 0, 
  266. 0);
  268. pAddress->Use( addressSize );
  270. if ( acceptedSocket != INVALID_SOCKET )
  271. {
  272. Socket *pSocket = AllocateSocket( acceptedSocket );
  274. OnConnectionEstablished( pSocket, pAddress );
  276. pSocket->Release();
  277. }
  278. else if ( m_acceptConnectionsEvent.Wait( 0 ) )
  279. {
  280. OnError( _T("CSocketServer::Run() - WSAAccept:") + 
  281. GetLastErrorMessage( ::WSAGetLastError() ) );
  282. }
  284. pAddress->Release();
  285. }
  286. }
  287. else
  288. {
  289. OnError( _T("CSocketServer::Run() - WaitForMultipleObjects: ") +
  290. GetLastErrorMessage( ::GetLastError() ) );
  291. }
  292. }
  294. for ( i = 0; i < m_numThreads; ++i )
  295. {
  296. workers[i]->InitiateShutdown();
  297. }  
  299. for ( i = 0; i < m_numThreads; ++i )
  300. {
  301. workers[i]->WaitForShutdownToComplete();
  303. delete workers[i];
  304. workers[i] = 0;
  305. }  
  306. }
  307. catch( const CException &e )
  308. {
  309. OnError( _T("CSocketServer::Run() - Exception: ") + 
  310. e.GetWhere() + 
  311. _T(" - ") + 
  312. e.GetMessage() );
  313. }
  314. catch(...)
  315. {
  316. OnError( _T("CSocketServer::Run() - Unexpected exception") );
  317. }
  319. OnShutdownComplete();
  321. return 0;
  322. }
  324. CSocketServer::Socket *CSocketServer::AllocateSocket( SOCKET theSocket )
  325. {
  326. CCriticalSection::Owner lock( m_listManipulationSection );
  328. Socket *pSocket = 0;
  330. if ( !m_freeList.Empty() )
  331. {
  332. pSocket = m_freeList.PopNode();
  334. pSocket->Attach( theSocket );
  336. pSocket->AddRef();
  337. }
  338. else
  339. {
  340. pSocket = new Socket( *this, theSocket, m_useSequenceNumbers );
  342. OnConnectionCreated();
  343. }
  345. m_activeList.PushNode( pSocket );
  347. m_iocp.AssociateDevice( reinterpret_cast< HANDLE >( theSocket ), 
  348. ( ULONG_PTR )pSocket);
  350. return pSocket;
  351. }
  353. void CSocketServer::ReleaseSocket( Socket *pSocket )
  354. {
  355. if ( !pSocket )
  356. {
  357. throw CException( _T("CSocketServer::ReleaseSocket()"),
  358. _T("pSocket is null") );
  359. }
  361. CCriticalSection::Owner lock( m_listManipulationSection );
  363. pSocket->SetMessageCallback(NULL);
  364. pSocket->RemoveFromList();
  366. if ( m_maxFreeSockets == 0 || 
  367. m_freeList.Count() < m_maxFreeSockets )
  368. {
  369. m_freeList.PushNode( pSocket );
  370. }
  371. else
  372. {
  373. DestroySocket( pSocket );
  374. }
  375. }
  377. void CSocketServer::DestroySocket( Socket *pSocket )
  378. {
  379. delete pSocket;
  381. OnConnectionDestroyed();
  382. }
  384. void CSocketServer::OnError( const _tstring &message )
  385. {
  386. DEBUG_ONLY( Output( message ) );
  387. }
  388.   
  389. void CSocketServer::WriteCompleted( Socket * /*pSocket*/, CIOBuffer *pBuffer )
  390. {
  391. if ( pBuffer->GetUsed() != pBuffer->GetWSABUF()->len )
  392. {
  393. OnError( _T("CSocketServer::WorkerThread::WriteCompleted - Socket write where not all data was written") );
  394. }
  395. }
  397. void CSocketServer::SetServerDataPtr( Socket *pSocket, void *pData )
  398. {
  399. pSocket->SetServerDataPtr( pData );
  400. }
  402. void *CSocketServer::GetServerDataPtr( const Socket *pSocket )
  403. {
  404. return pSocket->GetServerDataPtr();
  405. }
  407. void CSocketServer::WriteCompleted( Socket *pSocket )
  408. {
  409. pSocket->WriteCompleted();
  410. }
  412. bool CSocketServer::FilterSocketShutdown( Socket * /*pSocket*/, int /*how*/ )
  413. {
  414. return true;
  415. }
  417. bool CSocketServer::FilterSocketClose( Socket * /*pSocket*/ )
  418. {
  419. return true;
  420. }
  422. void CSocketServer::PostIoOperation( Socket *pSocket,
  423. CIOBuffer *pBuffer,
  424. enumIO_Operation operation)
  425. {
  426. pBuffer->SetOperation( operation );
  427. pBuffer->AddRef();
  429. pSocket->AddRef();
  431. m_iocp.PostStatus( ( ULONG_PTR )pSocket, 0, pBuffer );
  432. }
  434. CSocketServer &CSocketServer::GetServer( Socket *pSocket )
  435. {
  436. return pSocket->m_server;
  437. }
  439. void CSocketServer::OnConnectionError( enumConnectionErrorSource /*source*/,
  440. Socket * /*pSocket*/,
  441. CIOBuffer * /*pBuffer*/,
  442. DWORD lastError)
  443. {
  444. _tstring sErrorInfo = GetLastErrorMessage( lastError );
  446. DEBUG_ONLY( Output( _T("CSocketServer::OnConnectionError() - : ") + sErrorInfo ) );
  448. //DEBUG_ONLY( Message( sErrorInfo.c_str() ) );
  449. }
  451. /*
  452.  * CSocketServer::Socket
  453.  */
  454. CSocketServer::Socket::Socket( CSocketServer &server,                                 
  455. SOCKET theSocket,
  456. bool useSequenceNumbers )
  457. : m_server( server ),
  458.  m_socket( theSocket ),
  459.  m_ref( 1 ),
  460.  m_outstandingWrites( 0 ),
  461.  m_readShutdown( false ),
  462.  m_writeShutdown( false ),
  463.  m_closing( false ),
  464.  m_clientClosed( false ),
  465.  m_pSequenceData( 0 ),
  466.  m_pIM (0 )
  467. {
  468. if ( !IsValid() )
  469. {
  470. throw CException( _T("CSocketServer::Socket::Socket()"), 
  471. _T("Invalid socket") );
  472. }
  474. if ( useSequenceNumbers )
  475. {
  476. m_pSequenceData = new SequenceData( m_crit );
  477. }
  478. }
  480. CSocketServer::Socket::~Socket()
  481. {
  482. try
  483. {
  484. delete m_pSequenceData;
  485. }
  486. catch(...)
  487. {
  488. TRACE( "CSocketServer::Socket::~Socket exception!" );
  489. }
  490. }
  492. void CSocketServer::Socket::Attach( SOCKET theSocket )
  493. {
  494. if ( IsValid() )
  495. {
  496. throw CException( _T("CSocketServer::Socket::Attach()"), 
  497. _T("Socket already attached"));
  498. }
  500. m_socket = theSocket;
  502. SetUserData( 0 );
  504. m_readShutdown = false;
  505. m_writeShutdown = false;
  506. m_outstandingWrites = 0;
  507. m_closing = false;
  508. m_clientClosed = false;
  510. if ( m_pSequenceData )
  511. {
  512. m_pSequenceData->Reset();
  513. }
  514. }
  516. void CSocketServer::Socket::AddRef()
  517. {
  518. ::InterlockedIncrement( &m_ref );
  519. }
  521. void CSocketServer::Socket::Release()
  522. {
  523. if ( 0 == ::InterlockedDecrement( &m_ref ) )
  524. {
  525. if ( IsValid() )
  526. {
  527. AddRef();
  529. if ( !m_closing )
  530. {
  531. m_closing = true;
  533. if ( !m_server.OnConnectionClosing( this ) )
  534. {
  535. AbortiveClose();
  536. }
  537. }
  538. else
  539. {
  540. AbortiveClose();
  541. }
  543. Release();
  544. return;
  545. }
  547. m_server.ReleaseSocket( this );
  548. }
  549. }
  551. void CSocketServer::Socket::Shutdown( int how /* = SD_BOTH */ )
  552. {
  553. DEBUG_ONLY( Output( _T("CSocketServer::Socket::Shutdown() ") + ToString( how ) ) );
  555. if ( m_server.FilterSocketShutdown( this, how ) )
  556. {
  557. if ( how == SD_RECEIVE || how == SD_BOTH )
  558. {
  559. m_readShutdown = true;
  560. }
  562. if ( how == SD_SEND || how == SD_BOTH )
  563. {
  564. m_writeShutdown = true;
  565. }
  567. if ( ::InterlockedExchange( &m_outstandingWrites, m_outstandingWrites ) > 0 )
  568. {
  569. /*
  570.  * Send side will be shut down when last pending write completes...
  571.  */
  572. if ( how == SD_BOTH )
  573. {
  574. how = SD_RECEIVE;      
  575. }
  576. else if ( how == SD_SEND )
  577. {
  578. return;
  579. }
  580. }
  582. if ( IsValid() )
  583. {
  584. if ( 0 != ::shutdown( m_socket, how ) )
  585. {
  586. m_server.OnError( _T("CSocketServer::Server::Shutdown() - ") +
  587. GetLastErrorMessage( ::WSAGetLastError() ) );
  588. }
  590. DEBUG_ONLY( Output( _T("shutdown initiated") ) );
  591. }
  592. }
  593. }
  595. bool CSocketServer::Socket::IsConnected( int how /*= SD_BOTH*/ ) const
  596. {
  597. if ( how == SD_RECEIVE )
  598. {
  599. return !m_readShutdown;
  600. }
  602. if ( how == SD_SEND )
  603. {
  604. return !m_writeShutdown;
  605. }
  607. if ( how == SD_BOTH )
  608. {
  609. return ( !m_writeShutdown && !m_readShutdown );
  610. }
  612. return false;
  613. }
  615. void CSocketServer::Socket::Close()
  616. {
  617. CCriticalSection::Owner lock( m_crit );
  619. if ( IsValid() )
  620. {
  621. if ( m_server.FilterSocketClose( this ) )
  622. {
  623. InternalClose();
  624. }
  625. }
  626. }
  628. bool CSocketServer::Socket::WritePending()
  629. {
  630. if ( m_writeShutdown )
  631. {
  632. DEBUG_ONLY( Output( _T("CSocketServer::Socket::WritePending() - Attempt to write after write shutdown") ) );
  634. return false;
  635. }
  637. ::InterlockedIncrement( &m_outstandingWrites );
  639. return true;
  640. }
  642. void CSocketServer::Socket::WriteCompleted()
  643. {
  644. if ( ::InterlockedDecrement( &m_outstandingWrites ) == 0 )
  645. {
  646. if ( m_writeShutdown )
  647. {
  648. /*
  649.  * The final pending write has been completed so we can now shutdown
  650.  * the send side of the connection.  
  651.  */
  653. Shutdown( SD_SEND );
  654. }
  655. }
  656. }
  658. void CSocketServer::Socket::AbortiveClose()
  659. {
  660. /*
  661.  * Force an abortive close.
  662.  */
  663. LINGER lingerStruct;
  665. lingerStruct.l_onoff = 1;
  666. lingerStruct.l_linger = 0;
  668. if ( SOCKET_ERROR == ::setsockopt( m_socket, 
  669. SOL_SOCKET, 
  670. SO_LINGER, 
  671. ( char * )&lingerStruct, 
  672. sizeof( lingerStruct ) ) )
  673. {
  674. m_server.OnError( _T("CSocketServer::Socket::AbortiveClose() - setsockopt( SO_LINGER ) - ") + 
  675. GetLastErrorMessage(::WSAGetLastError() ) );
  676. }
  678. InternalClose();
  679. }
  681. void CSocketServer::Socket::OnClientClose()
  682. {
  683. if ( 0 == ::InterlockedExchange( &m_clientClosed, 1 ) )
  684. {
  685. Shutdown( SD_RECEIVE );
  687. m_server.OnConnectionClientClose( this );
  688. }
  689. }
  691. void CSocketServer::Socket::OnConnectionReset()
  692. {
  693. CCriticalSection::Owner lock( m_crit );
  695. if ( IsValid() )
  696. {
  697. m_server.OnConnectionReset( this );
  699. InternalClose();
  700. }
  701. }
  703. void CSocketServer::Socket::InternalClose()
  704. {
  705. CCriticalSection::Owner lock( m_crit );
  707. if ( IsValid() )
  708. {
  709. if ( 0 != ::closesocket( m_socket ) )
  710. {
  711. m_server.OnError( _T("CSocketServer::Socket::InternalClose() - closesocket - ") + 
  712. GetLastErrorMessage( ::WSAGetLastError() ) );
  713. }
  715. m_socket = INVALID_SOCKET;
  717. m_readShutdown = true;
  718. m_writeShutdown = true;
  720. m_server.OnConnectionClosed( this );
  721. }
  722. }
  724. bool CSocketServer::Socket::Read( CIOBuffer *pBuffer /* = 0 */, bool throwOnFailure /* = false*/ )
  725. {
  726. if ( !IsValid() )
  727. {
  728. if ( throwOnFailure )
  729. {
  730. throw CException( _T("CSocketServer::Socket::Read()"), _T("Socket is closed") );
  731. }
  732. else
  733. {
  734. return false;
  735. }
  736. }
  738. /*
  739.  * Post a read request to the iocp so that the actual socket read gets performed by
  740.  * one of the server's IO threads...
  741.  */
  743. if ( !pBuffer )
  744. {
  745. pBuffer = m_server.Allocate();
  746. }
  747. else
  748. {
  749. pBuffer->AddRef();
  750. }
  752. m_server.PostIoOperation( this, pBuffer, m_server.m_postZeroByteReads ? enumIO_Zero_Byte_Read_Request : enumIO_Read_Request );
  754. pBuffer->Release();
  756. return true;
  757. }
  759. bool CSocketServer::Socket::Write( const char *pData, size_t dataLength, bool throwOnFailure /* = false*/ )
  760. {
  761. return Write( reinterpret_cast< const BYTE * >( pData ), dataLength, throwOnFailure );
  762. }
  764. bool CSocketServer::Socket::Write( const BYTE *pData, size_t dataLength, bool throwOnFailure /* = false*/ )
  765. {
  766. if ( !IsValid() )
  767. {
  768. if ( throwOnFailure )
  769. {
  770. /*
  771.  * Todo throw SocketClosedException();
  772.  */
  773. throw CException( _T("CSocketServer::Socket::Write()"), _T("Socket is closed") );
  774. }
  775. else
  776. {
  777. return false;
  778. }
  779. }
  781. if ( !WritePending() )
  782. {
  783. if ( throwOnFailure )
  784. {
  785. /*
  786.  * Todo throw SocketClosedException();
  787.  */
  788. throw CException(_T("CSocketServer::Socket::Write()"), _T("Socket is shutdown"));
  789. }
  790. else
  791. {
  792. return false;
  793. }
  794. }
  796. CIOBuffer *pBuffer = m_server.Allocate();
  798. pBuffer->AddData( pData, dataLength );
  800. pBuffer->SetSequenceNumber( GetSequenceNumber( enumWriteSequenceNo ) );
  802. m_server.PostIoOperation( this, pBuffer, enumIO_Write_Request );
  804. pBuffer->Release();
  806. return true;
  807. }
  809. bool CSocketServer::Socket::Write( CIOBuffer *pBuffer, bool throwOnFailure /* = false*/ )
  810. {
  811. if ( !IsValid() )
  812. {
  813. if ( throwOnFailure )
  814. {
  815. throw CException( _T("CSocketServer::Socket::Write()"), _T("Socket is closed") );
  816. }
  817. else
  818. {
  819. return false;
  820. }
  821. }
  823. if ( !WritePending() )
  824. {
  825. if ( throwOnFailure )
  826. {
  827. throw CException( _T("CSocketServer::Socket::Write()"), _T("Socket is shutdown") );
  828. }
  829. else
  830. {
  831. return false;
  832. }
  833. }
  835. pBuffer->SetSequenceNumber( GetSequenceNumber( enumWriteSequenceNo ) );
  837. m_server.PostIoOperation( this, pBuffer, enumIO_Write_Request );
  839. return true;
  840. }
  842. long CSocketServer::Socket::GetSequenceNumber( enumSequenceType type )
  843. {
  844. if ( m_pSequenceData )
  845. {
  846. return m_pSequenceData->m_numbers[type]++;
  847. }
  849. return 0;
  850. }
  852. CIOBuffer *CSocketServer::Socket::GetNextBuffer( CIOBuffer *pBuffer /* = 0 */ )
  853. {
  854. if ( m_pSequenceData )
  855. {
  856. if ( pBuffer )
  857. {
  858. return m_pSequenceData->m_outOfSequenceWrites.GetNext( pBuffer );
  859. }
  860. else
  861. {
  862. return m_pSequenceData->m_outOfSequenceWrites.ProcessAndGetNext();
  863. }
  864. }
  866. return pBuffer;
  867. }
  869. bool CSocketServer::Socket::IsValid()
  870. {
  871. CCriticalSection::Owner lock(m_crit);
  873. return (INVALID_SOCKET != m_socket);
  874. }
  876. void CSocketServer::Socket::OnConnectionError( CSocketServer::enumConnectionErrorSource source,
  877.    CIOBuffer *pBuffer,
  878.    DWORD lastError)
  879. {
  880. if ( WSAESHUTDOWN == lastError )
  881. {
  882. OnClientClose();
  883. }
  884. else if ( WSAECONNRESET == lastError || WSAECONNABORTED == lastError )
  885. {
  886. OnConnectionReset();
  887. }
  888. else if ( !IsValid() && WSAENOTSOCK == lastError )
  889. {
  890. /*
  891.  * Swallow this error as we expect it...
  892.  */
  893. }
  894. else
  895. {
  896. m_server.OnConnectionError( source, this, pBuffer, lastError );
  897. }
  898. }
  900. /*
  901.  * CSocketServer::Socket::SequenceData
  902.  */
  904. CSocketServer::Socket::SequenceData::SequenceData( CCriticalSection &section )
  905. : m_outOfSequenceWrites( section )
  906. {
  907. memset( m_numbers, 0, sizeof( m_numbers ) );
  908. }
  910. void CSocketServer::Socket::SequenceData::Reset()
  911. {
  912.    memset( m_numbers, 0, sizeof( m_numbers ) );
  914.    m_outOfSequenceWrites.Reset();
  915. }
  918. /*
  919.  * CSocketServer::WorkerThread
  920.  */
  922. CSocketServer::WorkerThread::WorkerThread( CSocketServer &server, CIOCompletionPort &iocp )
  923. : m_server( server ),
  924.   m_iocp( iocp )
  925. {
  926. m_server.OnThreadCreated();
  927. }
  929. CSocketServer::WorkerThread::~WorkerThread()
  930. {
  931. try
  932. {
  933. m_server.OnThreadDestroyed();
  934. }
  935. catch(...)
  936. {
  937. TRACE( "CSocketServer::WorkerThread::~WorkerThread exception!" );
  938. }
  939. }
  941. int CSocketServer::WorkerThread::Run()
  942. {
  943. try
  944. {
  945. while ( true )
  946. {
  947. /*
  948.  * continually loop to service io completion packets
  949.  */
  951. DWORD dwIoSize = 0;
  952. Socket *pSocket = 0;
  953. CIOBuffer *pBuffer = 0;
  955. bool weClosedSocket = false;
  957. DWORD dwResult = m_iocp.GetStatus( ( PDWORD_PTR )&pSocket, &dwIoSize, ( OVERLAPPED ** )&pBuffer );
  959. if ( S_OK != dwResult )
  960. {
  961. if ( ERROR_NETNAME_DELETED == dwResult )
  962. {
  963. weClosedSocket = true;
  964. }
  965. else if ( WSA_OPERATION_ABORTED != dwResult )
  966. {
  967. throw CWin32Exception( _T("CIOCompletionPort::GetStatus() - GetQueuedCompletionStatus"), dwResult );
  968. }
  970. DEBUG_ONLY( Output( _T("IOCP error - client connection dropped") ) );
  971. }
  973. if ( !pSocket )
  974. {
  975. /*
  976.  * A completion key of 0 is posted to the iocp to request us to shut down...
  977.  */
  979. break;
  980. }
  982. /*
  983.  * Thread Begin Processing
  984.  */
  986. HandleOperation( pSocket, pBuffer, dwIoSize, weClosedSocket );
  988. /*
  989.  * Thread End Processing
  990.  */
  992. }// while ( true )
  993. }
  994. catch( const CException &e )
  995. {
  996. m_server.OnError( _T("CSocketServer::WorkerThread::Run() - Exception: ") +
  997. e.GetWhere() +
  998. _T(" - ") + 
  999. e.GetMessage() );
  1000. }
  1001. catch(...)
  1002. {
  1003. m_server.OnError( _T("CSocketServer::WorkerThread::Run() - Unexpected exception") );
  1004. }
  1006. return 0;
  1007. }
  1009. void CSocketServer::WorkerThread::InitiateShutdown()
  1010. {
  1011. m_iocp.PostStatus( 0 );
  1012. }
  1014. void CSocketServer::WorkerThread::WaitForShutdownToComplete()
  1015. {
  1016. /*
  1017.      * if we havent already started a shut down, do so...
  1018.  */
  1020. InitiateShutdown();
  1022. Wait();
  1023. }
  1025. void CSocketServer::WorkerThread::HandleOperation( Socket *pSocket,
  1026.    CIOBuffer *pBuffer,
  1027.    DWORD dwIoSize,
  1028.    bool weClosedSocket )
  1029. {
  1030. if ( pBuffer )
  1031. {
  1032. const enumIO_Operation operation = static_cast< enumIO_Operation >( pBuffer->GetOperation() );
  1034. switch ( operation )
  1035. {
  1036. case enumIO_Zero_Byte_Read_Request:
  1038.             DEBUG_ONLY( Output( _T("ZeroByteRead:") + ToString( pBuffer ) ) );
  1040.             ZeroByteRead( pSocket, pBuffer );
  1042.             pSocket->Release();
  1043.             pBuffer->Release();
  1045. break ;
  1047. case enumIO_Zero_Byte_Read_Completed :
  1048. case enumIO_Read_Request :
  1050.             DEBUG_ONLY( Output( _T("enumIO_Zero_Byte_Read_Completed | enumIO_Read_Request:") + ToString( pBuffer ) ) );
  1052. Read( pSocket, pBuffer );
  1053.             
  1054.             pSocket->Release();
  1055.             pBuffer->Release();
  1057. break;
  1059. case enumIO_Read_Completed :
  1061.             DEBUG_ONLY( Output( _T("enumIO_Read_Completed:") + ToString( pBuffer ) ) );
  1063.             pBuffer->Use( dwIoSize );            
  1064.            
  1065. if ( 0 != dwIoSize )
  1066. {
  1067. m_server.ReadCompleted( pSocket, pBuffer );
  1068. }
  1070.             if ( 0 == dwIoSize && !weClosedSocket )
  1071.             {
  1072. /*
  1073.  * client connection dropped?
  1074.  */
  1076. DEBUG_ONLY( Output( _T("ReadCompleted - 0 bytes - client connection dropped") ) );
  1078. pSocket->OnClientClose();
  1079.             }
  1081.             pSocket->Release();
  1082.             pBuffer->Release();
  1084. break;
  1086. case enumIO_Write_Request :
  1088.             DEBUG_ONLY( Output( _T("enumIO_Write_Request:") +
  1089. ToString( pBuffer ) ) );
  1091. Write( pSocket, pBuffer );
  1093.             pSocket->Release();
  1094.             pBuffer->Release();
  1096. break;
  1098. case enumIO_Write_Completed :
  1100.             DEBUG_ONLY( Output( _T("enumIO_Write_Completed:") + ToString( pBuffer ) ) );
  1102.             pBuffer->Use( dwIoSize );
  1104. m_server.WriteCompleted( pSocket, pBuffer );
  1106. pSocket->WriteCompleted();
  1108.             pSocket->Release();
  1109.             pBuffer->Release();
  1111. break;
  1113. default :
  1114.             m_server.OnError( _T("CSocketServer::WorkerThread::Run() - Unexpected operation") );
  1115. break;
  1116.       } 
  1117.    }
  1118.    else
  1119.    {
  1120.    m_server.OnError( _T("CSocketServer::WorkerThread::Run() - Unexpected - pBuffer is 0") );
  1121.    }
  1122. }
  1124. void CSocketServer::WorkerThread::ZeroByteRead( Socket *pSocket, CIOBuffer *pBuffer ) const
  1125. {
  1126. pSocket->AddRef();
  1128. pBuffer->SetOperation( enumIO_Zero_Byte_Read_Completed );
  1129. pBuffer->SetupZeroByteRead();
  1130. pBuffer->AddRef();
  1132. DWORD dwNumBytes = 0;
  1133. DWORD dwFlags = 0;
  1135. if ( SOCKET_ERROR == ::WSARecv(
  1136. pSocket->m_socket, 
  1137. pBuffer->GetWSABUF(), 
  1138. 1, 
  1139. &dwNumBytes,
  1140. &dwFlags,
  1141. pBuffer, 
  1142. NULL ) )
  1143. {
  1144. DWORD lastError = ::WSAGetLastError();
  1146. if ( ERROR_IO_PENDING != lastError )
  1147. {
  1148. pSocket->OnConnectionError( enumZeroByteReadError, pBuffer, lastError );
  1150. pSocket->Release();
  1151. pBuffer->Release();
  1152. }
  1153. }
  1154. }
  1156. void CSocketServer::WorkerThread::Read( Socket *pSocket, CIOBuffer *pBuffer ) const
  1157. {
  1158. pSocket->AddRef();
  1160. pBuffer->SetOperation( enumIO_Read_Completed );
  1161. pBuffer->SetupRead();
  1162. pBuffer->AddRef();
  1164. CCriticalSection::Owner lock( pSocket->m_crit );
  1166. pBuffer->SetSequenceNumber( pSocket->GetSequenceNumber( Socket::enumReadSequenceNo ) );
  1168. DWORD dwNumBytes = 0;
  1169. DWORD dwFlags = 0;
  1171. if ( SOCKET_ERROR == ::WSARecv(
  1172. pSocket->m_socket, 
  1173. pBuffer->GetWSABUF(), 
  1174. 1, 
  1175. &dwNumBytes,
  1176. &dwFlags,
  1177. pBuffer, 
  1178. NULL ) )
  1179. {
  1180. DWORD lastError = ::WSAGetLastError();
  1182. if ( ERROR_IO_PENDING != lastError )
  1183. {
  1184. pSocket->OnConnectionError( enumReadError, pBuffer, lastError );
  1186. pSocket->Release();
  1187. pBuffer->Release();
  1188. }
  1189. }
  1190. }
  1192. void CSocketServer::WorkerThread::Write( Socket *pSocket, CIOBuffer *pBuffer ) const
  1193. {
  1194. pSocket->AddRef();
  1196. {
  1197. CCriticalSection::Owner lock( pSocket->m_critWriteQueue );
  1199. pBuffer->SetOperation( enumIO_Write_Completed );
  1200. pBuffer->SetupWrite();
  1201. pBuffer->AddRef();
  1203. pBuffer = pSocket->GetNextBuffer( pBuffer );
  1205. // DWORD dwID = ::GetCurrentThreadId();
  1206. // printf( "Begin ID:%d - Socket:%dn", dwID, ( DWORD )pSocket );
  1207. //
  1208. // ::Sleep( 10 );
  1210. while ( pBuffer )
  1211. {
  1212. DWORD dwFlags = 0;
  1213. DWORD dwSendNumBytes = 0;
  1215. // ::Sleep( 10 );
  1217. if ( SOCKET_ERROR == ::WSASend(
  1218. pSocket->m_socket,
  1219. pBuffer->GetWSABUF(), 
  1220. 1, 
  1221. &dwSendNumBytes,
  1222. dwFlags,
  1223. pBuffer, 
  1224. NULL ) )
  1225. {
  1226. DWORD lastError = ::WSAGetLastError();
  1228. if ( ERROR_IO_PENDING != lastError )
  1229. {
  1230. pSocket->OnConnectionError( enumWriteError, pBuffer, lastError );
  1232. /*
  1233. * this pending write will never complete...
  1234. */
  1235. pSocket->WriteCompleted();
  1237. pSocket->Release();
  1238. pBuffer->Release();
  1239. }
  1240. }
  1242. pBuffer = pSocket->GetNextBuffer();
  1243. }
  1245. // printf( "End ID:%d - Socket:%dn", dwID, ( DWORD )pSocket );
  1246. }
  1247. }
  1249. } // End of namespace OnlineGameLib
  1250. } // End of namespace Win32