Loading src/event/iocp.cpp +268 −440 Original line number Diff line number Diff line Loading @@ -53,298 +53,198 @@ SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. #define BLOCKSIZE 16384 namespace netplus { std::atomic<bool> event::Running(true); std::atomic<bool> event::Restart(false); BOOL WINAPI ConsoleHandler(DWORD event) { switch (event) { case CTRL_C_EVENT: event::Running=false; WSACleanup(); return TRUE; default: return FALSE; } } enum IOState { IO_READ_STATE = 0, IO_WRITE_STATE = 1 }; class client { public: client(eventapi *eapi) : api(eapi), State(IO_READ_STATE), RecvBuf(BLOCKSIZE) { api->CreateConnection(CurCon); } ~client() { } int State; eventapi *api; std::unique_ptr<con> CurCon; std::mutex cltmtx; buffer RecvBuf; }; class poll { public: poll(HANDLE iocp, eventapi *api,socket *ssock) : g_iocp(iocp), g_eventapi(api), g_serversocket(ssock) { } /******************************************************************************* Copyright (c) 2014, Jan Koester jan.koester@gmx.net All rights reserved. void AssociateWithIOCP(SOCKET socket,client *pClientContext) { //Associate the socket with IOCP HANDLE giocp2 = CreateIoCompletionPort((HANDLE)socket, g_iocp,reinterpret_cast<ULONG_PTR>(pClientContext), 0); Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met: * Redistributions of source code must retain the above copyright notice, this list of conditions and the following disclaimer. * Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the following disclaimer in the documentation and/or other materials provided with the distribution. * Neither the name of the <organization> nor the names of its contributors may be used to endorse or promote products derived from this software without specific prior written permission. if (!giocp2 || g_iocp != giocp2) { DWORD err = GetLastError(); NetException exp; exp[NetException::Error] << "AssociateWithIOCP failed: failcode " << err; throw exp; } } THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL <COPYRIGHT HOLDER> BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. *******************************************************************************/ void AcceptConnection(LPFN_ACCEPTEX lpfnAcceptEx, int tid, ULONG_PTR args) { client* pClientContext = new client(g_eventapi); #include <iostream> #include <algorithm> #include <chrono> #include <mutex> #include <cstring> #include <vector> #include <thread> #include <atomic> #include <memory> if (g_serversocket->_Type == TCP) { pClientContext->CurCon->csock = std::make_unique<tcp>(); }else if (g_serversocket->_Type == SSL) { netplus::ssl *srv = static_cast<netplus::ssl*>(g_serversocket); pClientContext->CurCon->csock =std::make_unique<ssl>( srv->_cert); }else { NetException exp; exp[NetException::Error] << "Wrong Protocol only tcp supported at the moment !"; throw exp; } #include <winsock2.h> #include <ws2ipdef.h> #include <mswsock.h> try { g_serversocket->accept(lpfnAcceptEx, pClientContext->CurCon->csock); #include "socket.h" #include "exception.h" #include "eventapi.h" #include "connection.h" pClientContext->CurCon->csock->setNonBlock(); #define BLOCKSIZE 16384 AssociateWithIOCP(pClientContext->CurCon->csock->fd(), pClientContext); } catch (NetException& e) { delete pClientContext; throw e; } namespace netplus { std::atomic<bool> event::Running(true); std::atomic<bool> event::Restart(false); g_eventapi->ConnectEvent(*pClientContext->CurCon, tid, args); // Structure to track specific Overlapped I/O operations enum IO_OPERATION { OP_READ = 0, OP_WRITE = 1 }; size_t ret = pClientContext->CurCon->csock->recvData(pClientContext->RecvBuf, 0); } private: HANDLE g_iocp; socket *g_serversocket; eventapi *g_eventapi; friend class client; friend class EventWorker; struct IO_CONTEXT { WSAOVERLAPPED overlapped; WSABUF wsaBuf; char buffer[BLOCKSIZE]; IO_OPERATION operation; }; event::event(netplus::socket *serversocket, int timeout) : _ServerSocket(serversocket) { if (!serversocket) { NetException exp; exp[NetException::Critical] << "server socket empty!"; throw exp; } _pollFD = -1; _Timeout = timeout; _ServerSocket->setNonBlock(); _ServerSocket->bind(); _ServerSocket->listen(); SYSTEM_INFO sysinfo; GetSystemInfo(&sysinfo); threads = sysinfo.dwNumberOfProcessors; } class client { public: client(eventapi *eapi) : api(eapi) { api->CreateConnection(CurCon); event::event(const event &src) : _ServerSocket(src._ServerSocket) { _Timeout = src._Timeout; _pollFD = src._pollFD; threads = src.threads; } // Initialize Read Context memset(&readCtx, 0, sizeof(IO_CONTEXT)); readCtx.operation = OP_READ; readCtx.wsaBuf.buf = readCtx.buffer; readCtx.wsaBuf.len = BLOCKSIZE; event::~event() { // Initialize Write Context memset(&writeCtx, 0, sizeof(IO_CONTEXT)); writeCtx.operation = OP_WRITE; writeCtx.wsaBuf.buf = writeCtx.buffer; writeCtx.wsaBuf.len = 0; } void eventapi::RequestEvent(con &curcon, const int tid, ULONG_PTR args) { //dummy }; void eventapi::ResponseEvent(con &curcon, const int tid, ULONG_PTR args) { //dummy }; ~client() {} void eventapi::ConnectEvent(con &curcon, const int tid, ULONG_PTR args) { //dummy }; std::unique_ptr<con> CurCon; eventapi* api; std::mutex cltmtx; void eventapi::DisconnectEvent(con &curcon, const int tid, ULONG_PTR args) { //dummy IO_CONTEXT readCtx; IO_CONTEXT writeCtx; }; class EventWorkerArgs { public: EventWorkerArgs() { event = nullptr; eviocp = nullptr; evpoll = nullptr; ssocket = 0; timeout = 0; args = nullptr; } EventWorkerArgs(const EventWorkerArgs& eargs) { event = eargs.event; eviocp = eargs.eviocp; evpoll = eargs.evpoll; ssocket = eargs.ssocket; timeout = eargs.timeout; args = eargs.args; } int timeout; HANDLE eviocp; poll *evpoll; eventapi *event; socket *ssocket; void *args; }; class EventWorker { private: public: static void start_read(client* ctx) { if (!ctx) return; con& c = *ctx->CurCon; if (!c.csock) return; DWORD flags = 0; DWORD bytesRecv = 0; memset(&ctx->readCtx.overlapped, 0, sizeof(WSAOVERLAPPED)); // For IOCP, recvData() should post WSARecv and return quickly (or block if _Wait true). // Use your existing RecvBuf as the target. c.csock->recvData(ctx->RecvBuf, 0); int ret = WSARecv(ctx->CurCon->csock->fd(), &ctx->readCtx.wsaBuf, 1, &bytesRecv, &flags, &ctx->readCtx.overlapped, NULL); if (ret == SOCKET_ERROR && WSAGetLastError() != WSA_IO_PENDING) { NetException exp; exp[NetException::Error] << "WSARecv failed: " << WSAGetLastError(); throw exp; } } static void start_write(client* ctx) { if (!ctx) return; con& c = *ctx->CurCon; if (!c.csock) return; if (c.SendData.empty()) return; if (c.SendData.empty()) { ctx->last_plain_consumed = 0; return; } // Prepare buffer for WSASend size_t toSend = std::min<size_t>(BLOCKSIZE, c.SendData.size()); memcpy(ctx->writeCtx.buffer, c.SendData.data(), toSend); ctx->writeCtx.wsaBuf.len = (ULONG)toSend; const size_t toSend = std::min<size_t>(BLOCKSIZE, c.SendData.size()); buffer out(c.SendData.data(), toSend); memset(&ctx->writeCtx.overlapped, 0, sizeof(WSAOVERLAPPED)); DWORD bytesSent = 0; // IMPORTANT: returns PLAINTEXT CONSUMED (for TCP==socket bytes, for SSL==plaintext bytes) ctx->last_plain_consumed = c.csock->sendData(out, 0); int ret = WSASend(c.csock->fd(), &ctx->writeCtx.wsaBuf, 1, &bytesSent, 0, &ctx->writeCtx.overlapped, NULL); // If it consumed 0, we still posted work (TLS may be flushing a record), // or TCP would-block. Next completion/wakeup will retry. if (ret == SOCKET_ERROR && WSAGetLastError() != WSA_IO_PENDING) { NetException exp; exp[NetException::Error] << "WSASend failed: " << WSAGetLastError(); throw exp; } } public: EventWorker(int tid, ULONG_PTR args, EventWorkerArgs* eargs) { while (event::Running) { OVERLAPPED* pOverlapped = nullptr; ULONG_PTR lpContext = 0; DWORD dwBytesTransfered = 0; ULONG_PTR lpContext = 0; OVERLAPPED* pOverlapped = nullptr; bool bReturn = GetQueuedCompletionStatus( eargs->eviocp, &dwBytesTransfered, &lpContext, &pOverlapped, eargs->timeout); BOOL bReturn = GetQueuedCompletionStatus(eargs->eviocp, &dwBytesTransfered, &lpContext, &pOverlapped, eargs->timeout); if (lpContext == 0) { continue; } if (lpContext == 0) continue; client* pClientContext = reinterpret_cast<client*>(lpContext); IO_CONTEXT* pIoCtx = CONTAINING_RECORD(pOverlapped, IO_CONTEXT, overlapped); con& c = *pClientContext->CurCon; std::unique_ptr<con> &curCon = pClientContext->CurCon; // Check for I/O completion errors or graceful disconnects // Handle Disconnection or Errors if (!bReturn || (bReturn && dwBytesTransfered == 0)) { DWORD dwError = GetLastError(); if (dwError != WAIT_TIMEOUT && dwError != ERROR_OPERATION_ABORTED) { NetException exp; exp[NetException::Error] << "I/O operation failed or client disconnected. Error code: " << dwError; std::cerr << exp.what() << std::endl; } eargs->event->DisconnectEvent(*curCon, tid, args); eargs->event->DisconnectEvent(c, tid, args); delete pClientContext; continue; } try { std::unique_lock<std::mutex> guard(pClientContext->cltmtx); switch (pClientContext->State) { case IO_READ_STATE: { // IOCP completed a read: bytes are in csock->_wsaBuf.buf (your tcp/ssl layer owns that buffer) // For TCP: these bytes are plaintext // For SSL: these bytes are *ciphertext* and must be fed into SSL record decoder. // The clean way: have ssl expose a "feedEncrypted()" method and then pull plaintext via ssl::recvData(). // If you don't have that yet, do NOT treat _wsaBuf as plaintext for SSL. con& c = *curCon; if (c.csock->_Type == SSL) { // PREFERRED FIX: // static_cast<ssl*>(c.csock.get())->feedEncrypted((uint8_t*)c.csock->_wsaBuf.buf, dwBytesTransfered); // // then pull plaintext: // for(;;){ buffer plain(BLOCKSIZE); size_t n = c.csock->recvData(plain,0); ... } // // If you can't do that today, you *must* not append ciphertext to RecvData. // Return to posting another recv. start_read(pClientContext); break; std::lock_guard<std::mutex> guard(pClientContext->cltmtx); if (pIoCtx->operation == OP_READ) { // SSL vs Plaintext Decryption Logic if (c.csock->_Type == sockettype::SSL) { // Feed encrypted data to SSL engine; recvData should return plaintext buffer enc(pIoCtx->buffer, dwBytesTransfered); c.csock->recvData(enc, 0); } else { c.RecvData.append(pIoCtx->buffer, dwBytesTransfered); } // TCP case: c.RecvData.append(c.csock->_wsaBuf.buf, dwBytesTransfered); // Notify Protocol Layer eargs->event->RequestEvent(c, tid, args); // Decide next step: Write if data queued, else keep reading if (!c.SendData.empty()) { pClientContext->State = IO_WRITE_STATE; start_write(pClientContext); } else { start_read(pClientContext); } } break; case IO_WRITE_STATE: { con& c = *curCon; const size_t consumed = pClientContext->last_plain_consumed; // consumed==0 means: would-block or TLS flush pending record; retry send if (consumed == 0) { start_write(pClientContext); break; } if (consumed > c.SendData.size()) { NetException e; e[NetException::Error] << "IOCP Write: consumed > SendData.size()"; throw e; } else if (pIoCtx->operation == OP_WRITE) { // For SSL, we track plaintext consumed, but IOCP tells us bytes sent on wire. // We erase based on the application layer's queue management. if (c.csock->_Type == sockettype::SSL) { // SSL layers handle SendData erasure internally or via return codes } else { c.SendData.erase(c.SendData.begin(), c.SendData.begin() + dwBytesTransfered); } // Erase PLAINTEXT, not dwBytesTransfered c.SendData.erase(c.SendData.begin(), c.SendData.begin() + consumed); // Only after fully flushing current queue, allow protocol to enqueue more. if (c.SendData.empty()) { eargs->event->ResponseEvent(c, tid, args); } Loading @@ -352,148 +252,76 @@ namespace netplus { if (!c.SendData.empty()) { start_write(pClientContext); } else { pClientContext->State = IO_READ_STATE; start_read(pClientContext); } } break; default: { NetException exp; exp[NetException::Error] << "Unknown IO state: " << pClientContext->State; throw exp; }break; } } catch (NetException& e) { std::cerr << e.what() << std::endl; eargs->event->DisconnectEvent(*(pClientContext->CurCon), tid, args); closesocket(curCon->csock->fd()); } catch (NetException& e) { std::cerr << "IOCP Worker Error: " << e.what() << std::endl; eargs->event->DisconnectEvent(c, tid, args); delete pClientContext; } catch (...) { eargs->event->DisconnectEvent(*(pClientContext->CurCon), tid, args); closesocket(curCon->csock->fd()); delete pClientContext; } } // for } }; event::event(socket* serversocket, int timeout) : _ServerSocket(serversocket) { if (!serversocket) throw NetException("server socket empty!"); _Timeout = timeout; _ServerSocket->bind(); _ServerSocket->listen(); void event::runEventloop(ULONG_PTR args) { NetException exception; EventWorkerArgs eargs; if (!SetConsoleCtrlHandler(ConsoleHandler, TRUE)) { NetException exp; exp[NetException::Critical] << "event: runEventloop couldn't create ConsoleHandler !"; throw exp; } HANDLE iocp = CreateIoCompletionPort(INVALID_HANDLE_VALUE, nullptr, 0, 0); if (!iocp) { NetException exp; exp[NetException::Critical] << "event: runEventloop couldn't create iocp !"; throw exp; SYSTEM_INFO sysinfo; GetSystemInfo(&sysinfo); threads = sysinfo.dwNumberOfProcessors; } LPFN_ACCEPTEX lpfnAcceptEx = NULL; GUID GuidAcceptEx = WSAID_ACCEPTEX; WSAOVERLAPPED olOverlap; DWORD dwBytes; event::~event() {} memset(&olOverlap, 0, sizeof(olOverlap)); poll evpoll(iocp,this,_ServerSocket); void event::runEventloop(ULONG_PTR args) { HANDLE iocp = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 0); if (!iocp) throw NetException("couldn't create iocp!"); EventWorkerArgs eargs; eargs.ssocket = _ServerSocket; eargs.event = this; eargs.evpoll = &evpoll; eargs.timeout = _Timeout; eargs.eviocp = iocp; eargs.args = (void*)args; std::vector<std::thread> threadpool; for (int i = 0; i < threads; i++) { try { threadpool.push_back(std::thread([&eargs, args, i] { EventWorker(i, args, &eargs); })); } catch (NetException& e) { throw e; } } int iResult = WSAIoctl(_ServerSocket->fd(), SIO_GET_EXTENSION_FUNCTION_POINTER, &GuidAcceptEx, sizeof(GuidAcceptEx), &lpfnAcceptEx, sizeof(lpfnAcceptEx), &dwBytes, NULL, NULL); if (iResult == SOCKET_ERROR) { NetException exp; exp[NetException::Critical] << "WSAIoctl failed with error: " << WSAGetLastError(); closesocket(_ServerSocket->fd()); WSACleanup(); throw exp; } WSAEVENT g_hAcceptEvent = WSACreateEvent(); WSANETWORKEVENTS WSAEvents; if (WSA_INVALID_EVENT == g_hAcceptEvent) { NetException exp; exp[NetException::Error] << "Error occurred while WSACreateEvent()."; throw exp; } if (SOCKET_ERROR == WSAEventSelect(_ServerSocket->fd(), g_hAcceptEvent, FD_ACCEPT)) { NetException exp; exp[NetException::Error] << "Error occurred while WSAEventSelect()."; WSACloseEvent(g_hAcceptEvent); throw exp; } // Listener Loop (Standard Accept for IOCP) while (event::Running) { // This is the correct blocking call. It waits for the g_hAcceptEvent to be signaled. DWORD dwResult = WSAWaitForMultipleEvents(1, &g_hAcceptEvent, FALSE, _Timeout, FALSE); if (WSA_WAIT_FAILED == dwResult) { // Handle wait failure NetException exp; exp[NetException::Error] << "WSAWaitForMultipleEvents failed."; // ... cleanup and exit ... throw exp; std::unique_ptr<socket> cltSrv; if (_ServerSocket->_Type == sockettype::TCP) { cltSrv = std::make_unique<tcp>(); } else if (_ServerSocket->_Type == sockettype::SSL) { netplus::ssl* srv = static_cast<netplus::ssl*>(_ServerSocket); cltSrv = std::make_unique<ssl>(srv->_cert); } // Now, call WSAEnumNetworkEvents to get the details and reset the event. WSAEnumNetworkEvents(_ServerSocket->fd(), g_hAcceptEvent, &WSAEvents); // Check for FD_ACCEPT event and its error code. if ((WSAEvents.lNetworkEvents & FD_ACCEPT) && (0 == WSAEvents.iErrorCode[FD_ACCEPT_BIT])) { // Handle the successful connection. try { evpoll.AcceptConnection(lpfnAcceptEx, 0, args); } catch (NetException& e) { std::cerr << "AcceptConnection error: " << e.what() << std::endl; } } } _ServerSocket->accept(cltSrv); client* pClient = new client(this); pClient->CurCon->csock = std::move(cltSrv); // Associate with IOCP CreateIoCompletionPort((HANDLE)pClient->CurCon->csock->fd(), iocp, (ULONG_PTR)pClient, 0); // Cleanup for (int i = 0; i < threads; ++i) { PostQueuedCompletionStatus( iocp, // The IOCP Handle 0, // dwNumberOfBytesTransferred (0 for exit) 0, // lpCompletionKey (0 is safe for exit signal) NULL // lpOverlapped (NULL for exit signal) ); this->ConnectEvent(*pClient->CurCon, 0, args); EventWorker::start_read(pClient); } catch (NetException& e) { if (event::Running) std::cerr << "Accept error: " << e.what() << std::endl; } } for (auto i = threadpool.begin(); i != threadpool.end(); ++i) { i->join(); for (int i = 0; i < threads; i++) PostQueuedCompletionStatus(iocp, 0, 0, NULL); for (auto& t : threadpool) t.join(); CloseHandle(iocp); } } }; Loading
src/event/iocp.cpp +268 −440 Original line number Diff line number Diff line Loading @@ -53,298 +53,198 @@ SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. #define BLOCKSIZE 16384 namespace netplus { std::atomic<bool> event::Running(true); std::atomic<bool> event::Restart(false); BOOL WINAPI ConsoleHandler(DWORD event) { switch (event) { case CTRL_C_EVENT: event::Running=false; WSACleanup(); return TRUE; default: return FALSE; } } enum IOState { IO_READ_STATE = 0, IO_WRITE_STATE = 1 }; class client { public: client(eventapi *eapi) : api(eapi), State(IO_READ_STATE), RecvBuf(BLOCKSIZE) { api->CreateConnection(CurCon); } ~client() { } int State; eventapi *api; std::unique_ptr<con> CurCon; std::mutex cltmtx; buffer RecvBuf; }; class poll { public: poll(HANDLE iocp, eventapi *api,socket *ssock) : g_iocp(iocp), g_eventapi(api), g_serversocket(ssock) { } /******************************************************************************* Copyright (c) 2014, Jan Koester jan.koester@gmx.net All rights reserved. void AssociateWithIOCP(SOCKET socket,client *pClientContext) { //Associate the socket with IOCP HANDLE giocp2 = CreateIoCompletionPort((HANDLE)socket, g_iocp,reinterpret_cast<ULONG_PTR>(pClientContext), 0); Redistribution and use in source and binary forms, with or without modification, are permitted provided that the following conditions are met: * Redistributions of source code must retain the above copyright notice, this list of conditions and the following disclaimer. * Redistributions in binary form must reproduce the above copyright notice, this list of conditions and the following disclaimer in the documentation and/or other materials provided with the distribution. * Neither the name of the <organization> nor the names of its contributors may be used to endorse or promote products derived from this software without specific prior written permission. if (!giocp2 || g_iocp != giocp2) { DWORD err = GetLastError(); NetException exp; exp[NetException::Error] << "AssociateWithIOCP failed: failcode " << err; throw exp; } } THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL <COPYRIGHT HOLDER> BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. *******************************************************************************/ void AcceptConnection(LPFN_ACCEPTEX lpfnAcceptEx, int tid, ULONG_PTR args) { client* pClientContext = new client(g_eventapi); #include <iostream> #include <algorithm> #include <chrono> #include <mutex> #include <cstring> #include <vector> #include <thread> #include <atomic> #include <memory> if (g_serversocket->_Type == TCP) { pClientContext->CurCon->csock = std::make_unique<tcp>(); }else if (g_serversocket->_Type == SSL) { netplus::ssl *srv = static_cast<netplus::ssl*>(g_serversocket); pClientContext->CurCon->csock =std::make_unique<ssl>( srv->_cert); }else { NetException exp; exp[NetException::Error] << "Wrong Protocol only tcp supported at the moment !"; throw exp; } #include <winsock2.h> #include <ws2ipdef.h> #include <mswsock.h> try { g_serversocket->accept(lpfnAcceptEx, pClientContext->CurCon->csock); #include "socket.h" #include "exception.h" #include "eventapi.h" #include "connection.h" pClientContext->CurCon->csock->setNonBlock(); #define BLOCKSIZE 16384 AssociateWithIOCP(pClientContext->CurCon->csock->fd(), pClientContext); } catch (NetException& e) { delete pClientContext; throw e; } namespace netplus { std::atomic<bool> event::Running(true); std::atomic<bool> event::Restart(false); g_eventapi->ConnectEvent(*pClientContext->CurCon, tid, args); // Structure to track specific Overlapped I/O operations enum IO_OPERATION { OP_READ = 0, OP_WRITE = 1 }; size_t ret = pClientContext->CurCon->csock->recvData(pClientContext->RecvBuf, 0); } private: HANDLE g_iocp; socket *g_serversocket; eventapi *g_eventapi; friend class client; friend class EventWorker; struct IO_CONTEXT { WSAOVERLAPPED overlapped; WSABUF wsaBuf; char buffer[BLOCKSIZE]; IO_OPERATION operation; }; event::event(netplus::socket *serversocket, int timeout) : _ServerSocket(serversocket) { if (!serversocket) { NetException exp; exp[NetException::Critical] << "server socket empty!"; throw exp; } _pollFD = -1; _Timeout = timeout; _ServerSocket->setNonBlock(); _ServerSocket->bind(); _ServerSocket->listen(); SYSTEM_INFO sysinfo; GetSystemInfo(&sysinfo); threads = sysinfo.dwNumberOfProcessors; } class client { public: client(eventapi *eapi) : api(eapi) { api->CreateConnection(CurCon); event::event(const event &src) : _ServerSocket(src._ServerSocket) { _Timeout = src._Timeout; _pollFD = src._pollFD; threads = src.threads; } // Initialize Read Context memset(&readCtx, 0, sizeof(IO_CONTEXT)); readCtx.operation = OP_READ; readCtx.wsaBuf.buf = readCtx.buffer; readCtx.wsaBuf.len = BLOCKSIZE; event::~event() { // Initialize Write Context memset(&writeCtx, 0, sizeof(IO_CONTEXT)); writeCtx.operation = OP_WRITE; writeCtx.wsaBuf.buf = writeCtx.buffer; writeCtx.wsaBuf.len = 0; } void eventapi::RequestEvent(con &curcon, const int tid, ULONG_PTR args) { //dummy }; void eventapi::ResponseEvent(con &curcon, const int tid, ULONG_PTR args) { //dummy }; ~client() {} void eventapi::ConnectEvent(con &curcon, const int tid, ULONG_PTR args) { //dummy }; std::unique_ptr<con> CurCon; eventapi* api; std::mutex cltmtx; void eventapi::DisconnectEvent(con &curcon, const int tid, ULONG_PTR args) { //dummy IO_CONTEXT readCtx; IO_CONTEXT writeCtx; }; class EventWorkerArgs { public: EventWorkerArgs() { event = nullptr; eviocp = nullptr; evpoll = nullptr; ssocket = 0; timeout = 0; args = nullptr; } EventWorkerArgs(const EventWorkerArgs& eargs) { event = eargs.event; eviocp = eargs.eviocp; evpoll = eargs.evpoll; ssocket = eargs.ssocket; timeout = eargs.timeout; args = eargs.args; } int timeout; HANDLE eviocp; poll *evpoll; eventapi *event; socket *ssocket; void *args; }; class EventWorker { private: public: static void start_read(client* ctx) { if (!ctx) return; con& c = *ctx->CurCon; if (!c.csock) return; DWORD flags = 0; DWORD bytesRecv = 0; memset(&ctx->readCtx.overlapped, 0, sizeof(WSAOVERLAPPED)); // For IOCP, recvData() should post WSARecv and return quickly (or block if _Wait true). // Use your existing RecvBuf as the target. c.csock->recvData(ctx->RecvBuf, 0); int ret = WSARecv(ctx->CurCon->csock->fd(), &ctx->readCtx.wsaBuf, 1, &bytesRecv, &flags, &ctx->readCtx.overlapped, NULL); if (ret == SOCKET_ERROR && WSAGetLastError() != WSA_IO_PENDING) { NetException exp; exp[NetException::Error] << "WSARecv failed: " << WSAGetLastError(); throw exp; } } static void start_write(client* ctx) { if (!ctx) return; con& c = *ctx->CurCon; if (!c.csock) return; if (c.SendData.empty()) return; if (c.SendData.empty()) { ctx->last_plain_consumed = 0; return; } // Prepare buffer for WSASend size_t toSend = std::min<size_t>(BLOCKSIZE, c.SendData.size()); memcpy(ctx->writeCtx.buffer, c.SendData.data(), toSend); ctx->writeCtx.wsaBuf.len = (ULONG)toSend; const size_t toSend = std::min<size_t>(BLOCKSIZE, c.SendData.size()); buffer out(c.SendData.data(), toSend); memset(&ctx->writeCtx.overlapped, 0, sizeof(WSAOVERLAPPED)); DWORD bytesSent = 0; // IMPORTANT: returns PLAINTEXT CONSUMED (for TCP==socket bytes, for SSL==plaintext bytes) ctx->last_plain_consumed = c.csock->sendData(out, 0); int ret = WSASend(c.csock->fd(), &ctx->writeCtx.wsaBuf, 1, &bytesSent, 0, &ctx->writeCtx.overlapped, NULL); // If it consumed 0, we still posted work (TLS may be flushing a record), // or TCP would-block. Next completion/wakeup will retry. if (ret == SOCKET_ERROR && WSAGetLastError() != WSA_IO_PENDING) { NetException exp; exp[NetException::Error] << "WSASend failed: " << WSAGetLastError(); throw exp; } } public: EventWorker(int tid, ULONG_PTR args, EventWorkerArgs* eargs) { while (event::Running) { OVERLAPPED* pOverlapped = nullptr; ULONG_PTR lpContext = 0; DWORD dwBytesTransfered = 0; ULONG_PTR lpContext = 0; OVERLAPPED* pOverlapped = nullptr; bool bReturn = GetQueuedCompletionStatus( eargs->eviocp, &dwBytesTransfered, &lpContext, &pOverlapped, eargs->timeout); BOOL bReturn = GetQueuedCompletionStatus(eargs->eviocp, &dwBytesTransfered, &lpContext, &pOverlapped, eargs->timeout); if (lpContext == 0) { continue; } if (lpContext == 0) continue; client* pClientContext = reinterpret_cast<client*>(lpContext); IO_CONTEXT* pIoCtx = CONTAINING_RECORD(pOverlapped, IO_CONTEXT, overlapped); con& c = *pClientContext->CurCon; std::unique_ptr<con> &curCon = pClientContext->CurCon; // Check for I/O completion errors or graceful disconnects // Handle Disconnection or Errors if (!bReturn || (bReturn && dwBytesTransfered == 0)) { DWORD dwError = GetLastError(); if (dwError != WAIT_TIMEOUT && dwError != ERROR_OPERATION_ABORTED) { NetException exp; exp[NetException::Error] << "I/O operation failed or client disconnected. Error code: " << dwError; std::cerr << exp.what() << std::endl; } eargs->event->DisconnectEvent(*curCon, tid, args); eargs->event->DisconnectEvent(c, tid, args); delete pClientContext; continue; } try { std::unique_lock<std::mutex> guard(pClientContext->cltmtx); switch (pClientContext->State) { case IO_READ_STATE: { // IOCP completed a read: bytes are in csock->_wsaBuf.buf (your tcp/ssl layer owns that buffer) // For TCP: these bytes are plaintext // For SSL: these bytes are *ciphertext* and must be fed into SSL record decoder. // The clean way: have ssl expose a "feedEncrypted()" method and then pull plaintext via ssl::recvData(). // If you don't have that yet, do NOT treat _wsaBuf as plaintext for SSL. con& c = *curCon; if (c.csock->_Type == SSL) { // PREFERRED FIX: // static_cast<ssl*>(c.csock.get())->feedEncrypted((uint8_t*)c.csock->_wsaBuf.buf, dwBytesTransfered); // // then pull plaintext: // for(;;){ buffer plain(BLOCKSIZE); size_t n = c.csock->recvData(plain,0); ... } // // If you can't do that today, you *must* not append ciphertext to RecvData. // Return to posting another recv. start_read(pClientContext); break; std::lock_guard<std::mutex> guard(pClientContext->cltmtx); if (pIoCtx->operation == OP_READ) { // SSL vs Plaintext Decryption Logic if (c.csock->_Type == sockettype::SSL) { // Feed encrypted data to SSL engine; recvData should return plaintext buffer enc(pIoCtx->buffer, dwBytesTransfered); c.csock->recvData(enc, 0); } else { c.RecvData.append(pIoCtx->buffer, dwBytesTransfered); } // TCP case: c.RecvData.append(c.csock->_wsaBuf.buf, dwBytesTransfered); // Notify Protocol Layer eargs->event->RequestEvent(c, tid, args); // Decide next step: Write if data queued, else keep reading if (!c.SendData.empty()) { pClientContext->State = IO_WRITE_STATE; start_write(pClientContext); } else { start_read(pClientContext); } } break; case IO_WRITE_STATE: { con& c = *curCon; const size_t consumed = pClientContext->last_plain_consumed; // consumed==0 means: would-block or TLS flush pending record; retry send if (consumed == 0) { start_write(pClientContext); break; } if (consumed > c.SendData.size()) { NetException e; e[NetException::Error] << "IOCP Write: consumed > SendData.size()"; throw e; } else if (pIoCtx->operation == OP_WRITE) { // For SSL, we track plaintext consumed, but IOCP tells us bytes sent on wire. // We erase based on the application layer's queue management. if (c.csock->_Type == sockettype::SSL) { // SSL layers handle SendData erasure internally or via return codes } else { c.SendData.erase(c.SendData.begin(), c.SendData.begin() + dwBytesTransfered); } // Erase PLAINTEXT, not dwBytesTransfered c.SendData.erase(c.SendData.begin(), c.SendData.begin() + consumed); // Only after fully flushing current queue, allow protocol to enqueue more. if (c.SendData.empty()) { eargs->event->ResponseEvent(c, tid, args); } Loading @@ -352,148 +252,76 @@ namespace netplus { if (!c.SendData.empty()) { start_write(pClientContext); } else { pClientContext->State = IO_READ_STATE; start_read(pClientContext); } } break; default: { NetException exp; exp[NetException::Error] << "Unknown IO state: " << pClientContext->State; throw exp; }break; } } catch (NetException& e) { std::cerr << e.what() << std::endl; eargs->event->DisconnectEvent(*(pClientContext->CurCon), tid, args); closesocket(curCon->csock->fd()); } catch (NetException& e) { std::cerr << "IOCP Worker Error: " << e.what() << std::endl; eargs->event->DisconnectEvent(c, tid, args); delete pClientContext; } catch (...) { eargs->event->DisconnectEvent(*(pClientContext->CurCon), tid, args); closesocket(curCon->csock->fd()); delete pClientContext; } } // for } }; event::event(socket* serversocket, int timeout) : _ServerSocket(serversocket) { if (!serversocket) throw NetException("server socket empty!"); _Timeout = timeout; _ServerSocket->bind(); _ServerSocket->listen(); void event::runEventloop(ULONG_PTR args) { NetException exception; EventWorkerArgs eargs; if (!SetConsoleCtrlHandler(ConsoleHandler, TRUE)) { NetException exp; exp[NetException::Critical] << "event: runEventloop couldn't create ConsoleHandler !"; throw exp; } HANDLE iocp = CreateIoCompletionPort(INVALID_HANDLE_VALUE, nullptr, 0, 0); if (!iocp) { NetException exp; exp[NetException::Critical] << "event: runEventloop couldn't create iocp !"; throw exp; SYSTEM_INFO sysinfo; GetSystemInfo(&sysinfo); threads = sysinfo.dwNumberOfProcessors; } LPFN_ACCEPTEX lpfnAcceptEx = NULL; GUID GuidAcceptEx = WSAID_ACCEPTEX; WSAOVERLAPPED olOverlap; DWORD dwBytes; event::~event() {} memset(&olOverlap, 0, sizeof(olOverlap)); poll evpoll(iocp,this,_ServerSocket); void event::runEventloop(ULONG_PTR args) { HANDLE iocp = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 0); if (!iocp) throw NetException("couldn't create iocp!"); EventWorkerArgs eargs; eargs.ssocket = _ServerSocket; eargs.event = this; eargs.evpoll = &evpoll; eargs.timeout = _Timeout; eargs.eviocp = iocp; eargs.args = (void*)args; std::vector<std::thread> threadpool; for (int i = 0; i < threads; i++) { try { threadpool.push_back(std::thread([&eargs, args, i] { EventWorker(i, args, &eargs); })); } catch (NetException& e) { throw e; } } int iResult = WSAIoctl(_ServerSocket->fd(), SIO_GET_EXTENSION_FUNCTION_POINTER, &GuidAcceptEx, sizeof(GuidAcceptEx), &lpfnAcceptEx, sizeof(lpfnAcceptEx), &dwBytes, NULL, NULL); if (iResult == SOCKET_ERROR) { NetException exp; exp[NetException::Critical] << "WSAIoctl failed with error: " << WSAGetLastError(); closesocket(_ServerSocket->fd()); WSACleanup(); throw exp; } WSAEVENT g_hAcceptEvent = WSACreateEvent(); WSANETWORKEVENTS WSAEvents; if (WSA_INVALID_EVENT == g_hAcceptEvent) { NetException exp; exp[NetException::Error] << "Error occurred while WSACreateEvent()."; throw exp; } if (SOCKET_ERROR == WSAEventSelect(_ServerSocket->fd(), g_hAcceptEvent, FD_ACCEPT)) { NetException exp; exp[NetException::Error] << "Error occurred while WSAEventSelect()."; WSACloseEvent(g_hAcceptEvent); throw exp; } // Listener Loop (Standard Accept for IOCP) while (event::Running) { // This is the correct blocking call. It waits for the g_hAcceptEvent to be signaled. DWORD dwResult = WSAWaitForMultipleEvents(1, &g_hAcceptEvent, FALSE, _Timeout, FALSE); if (WSA_WAIT_FAILED == dwResult) { // Handle wait failure NetException exp; exp[NetException::Error] << "WSAWaitForMultipleEvents failed."; // ... cleanup and exit ... throw exp; std::unique_ptr<socket> cltSrv; if (_ServerSocket->_Type == sockettype::TCP) { cltSrv = std::make_unique<tcp>(); } else if (_ServerSocket->_Type == sockettype::SSL) { netplus::ssl* srv = static_cast<netplus::ssl*>(_ServerSocket); cltSrv = std::make_unique<ssl>(srv->_cert); } // Now, call WSAEnumNetworkEvents to get the details and reset the event. WSAEnumNetworkEvents(_ServerSocket->fd(), g_hAcceptEvent, &WSAEvents); // Check for FD_ACCEPT event and its error code. if ((WSAEvents.lNetworkEvents & FD_ACCEPT) && (0 == WSAEvents.iErrorCode[FD_ACCEPT_BIT])) { // Handle the successful connection. try { evpoll.AcceptConnection(lpfnAcceptEx, 0, args); } catch (NetException& e) { std::cerr << "AcceptConnection error: " << e.what() << std::endl; } } } _ServerSocket->accept(cltSrv); client* pClient = new client(this); pClient->CurCon->csock = std::move(cltSrv); // Associate with IOCP CreateIoCompletionPort((HANDLE)pClient->CurCon->csock->fd(), iocp, (ULONG_PTR)pClient, 0); // Cleanup for (int i = 0; i < threads; ++i) { PostQueuedCompletionStatus( iocp, // The IOCP Handle 0, // dwNumberOfBytesTransferred (0 for exit) 0, // lpCompletionKey (0 is safe for exit signal) NULL // lpOverlapped (NULL for exit signal) ); this->ConnectEvent(*pClient->CurCon, 0, args); EventWorker::start_read(pClient); } catch (NetException& e) { if (event::Running) std::cerr << "Accept error: " << e.what() << std::endl; } } for (auto i = threadpool.begin(); i != threadpool.end(); ++i) { i->join(); for (int i = 0; i < threads; i++) PostQueuedCompletionStatus(iocp, 0, 0, NULL); for (auto& t : threadpool) t.join(); CloseHandle(iocp); } } };
