socket wrapper

#include "stdafx.h" #include "SocketServer.h" #include "IOCompletionPort.h" #include "Win32Exception.h" #include "Utils.h" #include "SystemInfo.h" #include "Socket.h" #include <vector> #pragma comment(lib, "ws2_32.lib") using std::vector; namespace NvNetwork{ /////////////////////////////////////////////////////////////////////////////// // Static helper methods /////////////////////////////////////////////////////////////////////////////// static size_t CalculateNumberOfThreads( size_t numThreads); /////////////////////////////////////////////////////////////////////////////// // CSocketServer /////////////////////////////////////////////////////////////////////////////// CSocketServer::CSocketServer( unsigned long addressToListenOn, unsigned short portToListenOn, size_t maxFreeSockets, size_t maxFreeBuffers, size_t bufferSize /* = 1024 */, size_t numThreads /* = 0 */, bool useSequenceNumbers /* = true */, bool postZeroByteReads /* = false */) : CIOBuffer::Allocator(bufferSize, maxFreeBuffers), m_numThreads(CalculateNumberOfThreads(numThreads)), m_listeningSocket(INVALID_SOCKET), m_iocp(0), m_address(addressToListenOn), m_port(portToListenOn), m_maxFreeSockets(maxFreeSockets), m_useSequenceNumbers(useSequenceNumbers), m_postZeroByteReads(postZeroByteReads) { } CSocketServer::~CSocketServer() { } void CSocketServer::ReleaseSockets() { CCriticalSection::Owner lock(m_listManipulationSection); while (m_activeList.Head()) { ReleaseSocket(m_activeList.Head()); } while (m_freeList.Head()) { DestroySocket(m_freeList.PopNode()); } if (m_freeList.Count() + m_freeList.Count() != 0) { //lint -e{1933} call to unqualified virtual function OnError(_T("CSocketServer::ReleaseSockets() - Leaked sockets")); } } void CSocketServer::StartAcceptingConnections() { if (m_listeningSocket == INVALID_SOCKET) { //lint -e{1933} call to unqualified virtual function OnStartAcceptingConnections(); //lint -e{1933} call to unqualified virtual function m_listeningSocket = CreateListeningSocket(m_address, m_port); m_acceptConnectionsEvent.Set(); } } void CSocketServer::StopAcceptingConnections() { if (m_listeningSocket != INVALID_SOCKET) { m_acceptConnectionsEvent.Reset(); if (0 != ::closesocket(m_listeningSocket)) { //lint -e{1933} call to unqualified virtual function OnError(_T("CSocketServer::StopAcceptingConnections() - closesocket - ") + GetLastErrorMessage(::WSAGetLastError())); } m_listeningSocket = INVALID_SOCKET; //lint -e{1933} call to unqualified virtual function OnStopAcceptingConnections(); } } void CSocketServer::InitiateShutdown() { // signal that the dispatch thread should shut down all worker threads and then exit StopAcceptingConnections(); { CCriticalSection::Owner lock(m_listManipulationSection); Socket *pSocket = m_activeList.Head(); while (pSocket) { Socket *pNext = SocketList::Next(pSocket); pSocket->AbortiveClose(); pSocket = pNext; } } m_shutdownEvent.Set(); //lint -e{1933} call to unqualified virtual function OnShutdownInitiated(); } void CSocketServer::WaitForShutdownToComplete() { // if we havent already started a shut down, do so... InitiateShutdown(); Wait(); ReleaseSockets(); Flush(); } SOCKET CSocketServer::CreateListeningSocket( unsigned long address, unsigned short port) { SOCKET s = ::WSASocket(AF_INET, SOCK_STREAM, IPPROTO_IP, NULL, 0, WSA_FLAG_OVERLAPPED); if (s == INVALID_SOCKET) { throw CWin32Exception(_T("CSocket::CreateListeningSocket()"), ::WSAGetLastError()); } CSocket listeningSocket(s); CSocket::InternetAddress localAddress(address, port); listeningSocket.Bind(localAddress); listeningSocket.Listen(5); return listeningSocket.Detatch(); } CSocketServer::WorkerThread *CSocketServer::CreateWorkerThread( CIOCompletionPort &iocp) { return new WorkerThread(*this, iocp); } int CSocketServer::Run() { try { vector<WorkerThread *> workers; workers.reserve(m_numThreads); for (size_t i = 0; i < m_numThreads; ++i) { //lint -e{1933} call to unqualified virtual function WorkerThread *pThread = CreateWorkerThread(m_iocp); workers.push_back(pThread); pThread->Start(); //lint -e{429} custodial pointer neither freed nor returned } HANDLE handlesToWaitFor[2]; handlesToWaitFor[0] = m_shutdownEvent.GetEvent(); handlesToWaitFor[1] = m_acceptConnectionsEvent.GetEvent(); while (!m_shutdownEvent.Wait(0)) { DWORD waitResult = ::WaitForMultipleObjects(2, handlesToWaitFor, false, INFINITE); if (waitResult == WAIT_OBJECT_0) { // Time to shutdown break; } else if (waitResult == WAIT_OBJECT_0 + 1) { // accept connections while (!m_shutdownEvent.Wait(0) && m_acceptConnectionsEvent.Wait(0)) { CIOBuffer *pAddress = Allocate(); int addressSize = (int)pAddress->GetSize(); //lint -e{826} suspicious pointer conversion SOCKET acceptedSocket = ::WSAAccept( m_listeningSocket, reinterpret_cast<sockaddr*>(const_cast<BYTE*>(pAddress->GetBuffer())), &addressSize, 0, 0); pAddress->Use(addressSize); if (acceptedSocket != INVALID_SOCKET) { Socket *pSocket = AllocateSocket(acceptedSocket); //lint -e{1933} call to unqualified virtual function OnConnectionEstablished(pSocket, pAddress); pSocket->Release(); } else if (m_acceptConnectionsEvent.Wait(0)) { //lint -e{1933} call to unqualified virtual function OnError(_T("CSocketServer::Run() - WSAAccept:") + GetLastErrorMessage(::WSAGetLastError())); } pAddress->Release(); } } else { //lint -e{1933} call to unqualified virtual function OnError(_T("CSocketServer::Run() - WaitForMultipleObjects: ") + GetLastErrorMessage(::GetLastError())); } } for (int i = 0; i < m_numThreads; ++i) { workers[i]->InitiateShutdown(); } for (int i = 0; i < m_numThreads; ++i) { workers[i]->WaitForShutdownToComplete(); delete workers[i]; workers[i] = 0; } } catch(const CException &e) { //lint -e{1933} call to unqualified virtual function OnError(_T("CSocketServer::Run() - Exception: ") + e.GetWhere() + _T(" - ") + e.GetMessage()); } catch(...) { //lint -e{1933} call to unqualified virtual function OnError(_T("CSocketServer::Run() - Unexpected exception")); } //lint -e{1933} call to unqualified virtual function OnShutdownComplete(); return 0; } CSocketServer::Socket *CSocketServer::AllocateSocket( SOCKET theSocket) { CCriticalSection::Owner lock(m_listManipulationSection); Socket *pSocket = 0; if (!m_freeList.Empty()) { pSocket = m_freeList.PopNode(); pSocket->Attach(theSocket); pSocket->AddRef(); } else { pSocket = new Socket(*this, theSocket, m_useSe