Loading src/event/iocp.cpp +416 −286 Original line number Diff line number Diff line /******************************************************************************* Copyright (c) 2014, Jan Koester jan.koester@gmx.net All rights reserved. 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. 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. *******************************************************************************/ * iocp.cpp (Windows backend for event) * * Implements netplus::event using IOCP: * - CreateIoCompletionPort * - AcceptEx for incoming connections * - WSARecv / WSASend for async I/O * * Works with http.cpp usage: * class MyServer : public event { ... RequestEvent(...) override ... } * ******************************************************************************/ #ifdef Windows #define WIN32_LEAN_AND_MEAN #include <Windows.h> #include <WinSock2.h> #include <mswsock.h> #include <ws2tcpip.h> #pragma comment(lib, "Ws2_32.lib") #include <iostream> #include <mutex> #include <vector> #include <atomic> #include <thread> #include <vector> #include <memory> #include <atomic> #include <chrono> #include <cstring> #include <mutex> #include <iostream> #include "socket.h" #include "exception.h" #include "eventapi.h" #include "connection.h" #include "eventapi.h" #include "exception.h" #include <cstdio> #include <cinttypes> #include <windows.h> static inline uint64_t now_ms() { return GetTickCount64(); } #define IOLOG(fmt, ...) \ do { \ std::fprintf(stderr, "[%10" PRIu64 " ms][tid=%lu] " fmt "\n", \ now_ms(), GetCurrentThreadId(), __VA_ARGS__); \ std::fflush(stderr); \ } while (0) using namespace netplus; namespace netplus { std::atomic<bool> event::Running(true); std::atomic<bool> event::Restart(false); std::vector<socket*> SSOCKETS; std::map<int,std::shared_ptr<con>> CONNECTIONS; std::mutex POLL_HANDLER_MUTEX; std::atomic<size_t> g_con_id(0); // ------------------------------------------------------------ // Client object (per accepted connection context) // ------------------------------------------------------------ class client { public: explicit client(eventapi* eapi) : api(eapi), readCtx(BLOCKSIZE), writeCtx(BLOCKSIZE) { api->CreateConnection(CurCon); readCtx.owner = this; writeCtx.owner = this; } std::shared_ptr<con> CurCon; eventapi* api; std::mutex cltmtx; buffer readCtx; // Accept + Read buffer writeCtx; // Write }; struct EventWorkerArgs { int timeout; HANDLE iocp; eventapi* ev; socket* listenSock; LPFN_ACCEPTEX lpfnAcceptEx; void* userArgs; }; static const char* op_name(int op) { switch (op) { case OP_ACCEPT: return "ACCEPT"; case OP_READ: return "READ"; case OP_WRITE: return "WRITE"; default: return "UNKNOWN"; static LPFN_ACCEPTEX LoadAcceptEx(SOCKET listenSock) { GUID guid = WSAID_ACCEPTEX; LPFN_ACCEPTEX fn = nullptr; DWORD bytes = 0; int rc = WSAIoctl( listenSock, SIO_GET_EXTENSION_FUNCTION_POINTER, &guid, sizeof(guid), &fn, sizeof(fn), &bytes, nullptr, nullptr ); if (rc == SOCKET_ERROR || !fn) { NetException e; e[NetException::Critical] << "LoadAcceptEx failed: " << WSAGetLastError(); throw e; } return fn; } static void post_accept(EventWorkerArgs* a) { client* c = new client(a->ev); if (a->listenSock->_Type == sockettype::TCP) { c->CurCon->csock = std::make_unique<tcp>(); // ---------------------------------------------------------------------------- // Small helper: create a new connection object, assign socket, bind to IOCP. // ---------------------------------------------------------------------------- static std::unique_ptr<con> CreateConnectionFromAccepted(event* ev, socket* serverSock, SOCKET acceptedSock, HANDLE iocpHandle) { // Your connection type "con" is defined in connection.h // Typically you use something like: std::make_unique<con>(...) // But since I don't see con constructors here, I use the default approach: std::unique_ptr<con> c = std::make_unique<con>(); // Create a tcp socket wrapper for accepted socket. // Your socket class likely has setSocket(SOCKET) and takes ownership. auto clientSock = std::make_unique<tcp>(); clientSock->setSocket(acceptedSock); // Bind accepted socket to IOCP: completion key = clientSock pointer HANDLE h = CreateIoCompletionPort( (HANDLE)acceptedSock, iocpHandle, (ULONG_PTR)clientSock.get(), 0 ); if (!h) { NetException e; e[NetException::Critical] << "CreateIoCompletionPort(accepted) failed: " << GetLastError(); throw e; } // Bind the socket to IOCP CreateIoCompletionPort((HANDLE)c->CurCon->csock->fd(), a->iocp, (ULONG_PTR)c, 0); // attach socket to con c->csocket = std::move(clientSock); c->ssocket = serverSock; // CRITICAL: Set the owner so the Worker knows who this is c->readCtx.owner = c; c->readCtx.operation = OP_ACCEPT; std::memset(&c->readCtx.overlapped, 0, sizeof(WSAOVERLAPPED)); // mark connected c->connected = true; // Post the actual AcceptEx a->listenSock->accept(a->lpfnAcceptEx, c->CurCon->csock, c->readCtx); return c; } class EventWorker { // ---------------------------------------------------------------------------- // IOCP Worker thread // ---------------------------------------------------------------------------- class IocpWorker { public: EventWorker(int tid, ULONG_PTR args, EventWorkerArgs* eargs) { DWORD bytesTransferred = 0; ULONG_PTR completionKey = 0; LPOVERLAPPED overlapped = nullptr; IocpWorker(event* ev, socket* serverSock, HANDLE iocp, LPFN_ACCEPTEX fnAcceptEx, int timeout) : _ev(ev), _serverSock(serverSock), _iocp(iocp), _fnAcceptEx(fnAcceptEx), _timeout(timeout) {} void run() { while (event::Running) { // 1) Warte auf Abschluss einer I/O Operation BOOL res = GetQueuedCompletionStatus( eargs->iocp, &bytesTransferred, &completionKey, &overlapped, eargs->timeout ); DWORD bytes = 0; ULONG_PTR key = 0; OVERLAPPED* pov = nullptr; if (!res && !overlapped) { if (GetLastError() == WAIT_TIMEOUT) continue; break; } BOOL ok = GetQueuedCompletionStatus(_iocp, &bytes, &key, &pov, INFINITE); int fd = (int)completionKey; std::shared_ptr<con> curcon; if (!event::Running) break; if (!pov && key == 0) continue; { std::lock_guard<std::mutex> lock(POLL_HANDLER_MUTEX); auto it = CONNECTIONS.find(fd); if (it != CONNECTIONS.end()) curcon = it->second; // AcceptEx completions use AcceptContext.ov (not buffer.overlapped) if (AcceptContext* actx = FindAcceptContext(pov)) { HandleAcceptCompletion(ok, actx); continue; } if (!curcon || !curcon->csock) continue; // Otherwise: must be buffer completion (WSARecv / WSASend) socket* sockObj = reinterpret_cast<socket*>(key); if (!sockObj) continue; std::lock_guard<std::mutex> lk(curcon->event_mutex); buffer* buf = reinterpret_cast<buffer*>( reinterpret_cast<char*>(pov) - offsetof(buffer, overlapped) ); try { // 2) Handshake-Logik (Analog zu epoll.cpp) if (!curcon->csock->getHandshakeDone()) { // Da iocp.cpp als Template für SSL dient, casten wir sicherheitshalber if (static_cast<ssl*>(curcon->csock.get())->handleHandshakeAsynch()) { // Handshake gerade fertig -> Wir könnten sofort lesen } // Wir müssen immer einen neuen Recv-Request absetzen _triggerNextRecv(curcon); if (!ok) { // I/O error => close _ev->CloseEvent(sockObj); continue; } if (buf->operation == OP_READ) { if (bytes == 0) { _ev->CloseEvent(sockObj); continue; } // 3) Datenverarbeitung if (bytesTransferred > 0) { ReadEventHandler(curcon, tid, eargs->userArgs, eargs); } else if (bytesTransferred == 0) { _closeConnection(fd, eargs); // Find connection for this socket con* c = _ev->FindConnection(sockObj); if (!c) { _ev->CloseEvent(sockObj); continue; } } catch (NetException& e) { if (e.getErrorType() != NetException::Note) { _closeConnection(fd, eargs); // Let user code handle it _ev->RequestEvent(*c, _tid, (ULONG_PTR)bytes); // Re-post receive try { c->csocket->recvDataWSA(c->rbuf, 0); } catch (...) { _ev->CloseEvent(sockObj); } } else if (buf->operation == OP_WRITE) { // send completed => user callback con* c = _ev->FindConnection(sockObj); if (!c) { _ev->CloseEvent(sockObj); continue; } _ev->SendEvent(*c, _tid, (ULONG_PTR)bytes); } } void ReadEventHandler(std::shared_ptr<con> rcon, int tid, void* args, EventWorkerArgs* eargs) { try { for (;;) { buffer buf(BLOCKSIZE); size_t rcv = 0; try { // Nutzt WSARecv intern via socket.cpp/tcp.cpp rcv = rcon->csock->RecvDataWSA(buf, 0); } catch (NetException& e) { if (e.getErrorType() == NetException::Note) return; // WSA_IO_PENDING throw; } if (rcv == 0) break; rcon->RecvData.append(buf.data.buf, rcv); rcon->lasteventime = time(nullptr); void setThreadId(int tid) { _tid = tid; } // Protokoll-Event triggern (z.B. HTTP Parser) eargs->ev->RequestEvent(*rcon, tid, args); // AcceptContext tracking void RegisterAccept(AcceptContext* a) { std::lock_guard<std::mutex> lk(_accMutex); _accepts.push_back(a); } // Nach RequestEvent abbrechen, um Senden zu priorisieren (wie epoll) break; private: AcceptContext* FindAcceptContext(OVERLAPPED* pov) { std::lock_guard<std::mutex> lk(_accMutex); for (auto* a : _accepts) { if ((OVERLAPPED*)&a->ov == pov) return a; } return nullptr; } if (!rcon->SendData.empty()) { WriteEventHandler(rcon, tid, args, eargs); void RemoveAccept(AcceptContext* a) { std::lock_guard<std::mutex> lk(_accMutex); for (size_t i = 0; i < _accepts.size(); ++i) { if (_accepts[i] == a) { _accepts.erase(_accepts.begin() + i); return; } } catch (NetException& e) { throw; } } void WriteEventHandler(std::shared_ptr<con> wcon, int tid, void* args, EventWorkerArgs* eargs) { try { while (!wcon->SendData.empty()) { size_t sendlen = (std::min)(wcon->SendData.size(), (size_t)BLOCKSIZE); buffer out(wcon->SendData.data(), sendlen); void PostAccept() { auto* actx = new AcceptContext(); std::memset(&actx->ov, 0, sizeof(WSAOVERLAPPED)); size_t consumed = 0; try { // Nutzt WSASend intern consumed = wcon->csock->SendDataWSA(out, 0); } catch (NetException& e) { if (e.getErrorType() == NetException::Note) return; throw; SOCKET as = WSASocket(AF_INET, SOCK_STREAM, IPPROTO_TCP, nullptr, 0, WSA_FLAG_OVERLAPPED); if (as == INVALID_SOCKET) { delete actx; return; } actx->acceptSock = as; if (consumed == 0) return; wcon->SendData.erase(wcon->SendData.begin(), wcon->SendData.begin() + consumed); } DWORD bytes = 0; BOOL ok = _fnAcceptEx( (SOCKET)_serverSock->getSocket(), as, actx->addrBuf, 0, sizeof(SOCKADDR_STORAGE) + 16, sizeof(SOCKADDR_STORAGE) + 16, &bytes, &actx->ov ); if (wcon->SendData.empty()) { eargs->ev->ResponseEvent(*wcon, tid, args); if (!ok) { int err = WSAGetLastError(); if (err != WSA_IO_PENDING) { closesocket(as); delete actx; return; } } catch (NetException& e) { throw; } RegisterAccept(actx); } private: void _triggerNextRecv(std::shared_ptr<con> c) { buffer dummy(0); try { c->csock->RecvDataWSA(dummy, 0); } catch(...) {} void HandleAcceptCompletion(BOOL ok, AcceptContext* actx) { RemoveAccept(actx); if (!ok) { closesocket(actx->acceptSock); delete actx; PostAccept(); // keep accepting return; } void _closeConnection(int fd, EventWorkerArgs* eargs) { std::lock_guard<std::mutex> lock(POLL_HANDLER_MUTEX); auto it = CONNECTIONS.find(fd); if (it != CONNECTIONS.end()) { CONNECTIONS.erase(it); // must update accept context SOCKET ls = (SOCKET)_serverSock->getSocket(); setsockopt(actx->acceptSock, SOL_SOCKET, SO_UPDATE_ACCEPT_CONTEXT, (char*)&ls, sizeof(ls)); try { // create connection std::unique_ptr<con> newcon = CreateConnectionFromAccepted(_ev, _serverSock, actx->acceptSock, _iocp); // register connection in event _ev->AddConnection(std::move(newcon)); // Start first read con* c = _ev->FindConnectionBySocket((SOCKET)actx->acceptSock); if (c) { c->csocket->recvDataWSA(c->rbuf, 0); } } catch (...) { closesocket(actx->acceptSock); } closesocket(fd); delete actx; // Repost accept immediately PostAccept(); } private: event* _ev = nullptr; socket* _serverSock = nullptr; HANDLE _iocp = nullptr; LPFN_ACCEPTEX _fnAcceptEx = nullptr; int _timeout = 1000; int _tid = 0; std::mutex _accMutex; std::vector<AcceptContext*> _accepts; }; // ------------------------------------------------------------ // event // ------------------------------------------------------------ event::event(socket* serversocket, int timeout) { _Timeout = timeout; _ServerSocket = serversocket; _ServerSocket->bind(); _ServerSocket->listen(); // ---------------------------------------------------------------------------- // event class implementation (Windows/IOCP) // ---------------------------------------------------------------------------- event::event(socket* serversocket, int timeout) : _ServerSocket(serversocket), _Timeout(timeout) { if (!_ServerSocket) { NetException e; e[NetException::Critical] << "server socket empty!"; throw e; } event::~event() {} // Ensure server socket is bound/listening _ServerSocket->bind(); _ServerSocket->listen(); void event::runEventloop(ULONG_PTR args) { NetException exception; // Create IOCP _Iocp = CreateIoCompletionPort(INVALID_HANDLE_VALUE, nullptr, 0, 0); if (!_Iocp) { NetException e; e[NetException::Critical] << "CreateIoCompletionPort failed: " << GetLastError(); throw e; } HANDLE iocp = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 0); if (!iocp) { exception[NetException::Critical] << "IOCP: CreateIoCompletionPort failed"; throw exception; // Bind server socket to IOCP (key = server socket pointer) HANDLE h = CreateIoCompletionPort( (HANDLE)_ServerSocket->getSocket(), _Iocp, (ULONG_PTR)_ServerSocket, 0 ); if (!h) { NetException e; e[NetException::Critical] << "CreateIoCompletionPort(bind server) failed: " << GetLastError(); throw e; } _fnAcceptEx = LoadAcceptEx((SOCKET)_ServerSocket->getSocket()); } event::~event() { Stop(); } // start worker threads + initial AcceptEx posts void runEventloop(ULONG_PTR args=0){ Running = true; const int nthreads = (int)std::max<size_t>(1, std::thread::hardware_concurrency()); _Workers.reserve(nthreads); for (int i = 0; i < nthreads; ++i) { _Workers.emplace_back([&, i]() { IocpWorker w(this, _ServerSocket, _Iocp, _fnAcceptEx, _Timeout); w.setThreadId(i); // Pre-post accepts per worker (you can tune this) for (int k = 0; k < 4; ++k) { w.RegisterAccept(new AcceptContext()); // placeholder } // Actually post accepts for (int k = 0; k < 4; ++k) { // internal call // (we need to post on this worker instance) // so we recreate minimal pattern: // easiest: call PostAccept by starting run() which posts first } // Post initial accepts now: // (just call run() which will post from accept completions) // Here we do 8 accepts total as baseline: for (int k = 0; k < 8; ++k) { // hack: call private PostAccept by creating a local lambda // Instead: do it in constructor? => simplest: call accept via server socket method // We'll just post directly here: AcceptContext* actx = new AcceptContext(); std::memset(&actx->ov, 0, sizeof(WSAOVERLAPPED)); SOCKET as = WSASocket(AF_INET, SOCK_STREAM, IPPROTO_TCP, nullptr, 0, WSA_FLAG_OVERLAPPED); if (as == INVALID_SOCKET) { delete actx; continue; } actx->acceptSock = as; DWORD bytes = 0; BOOL ok = _fnAcceptEx( (SOCKET)_ServerSocket->getSocket(), as, actx->addrBuf, 0, sizeof(SOCKADDR_STORAGE) + 16, sizeof(SOCKADDR_STORAGE) + 16, &bytes, &actx->ov ); if (!ok) { int err = WSAGetLastError(); if (err != WSA_IO_PENDING) { closesocket(as); delete actx; continue; } } w.RegisterAccept(actx); } // Bind listening socket to IOCP (so AcceptEx completions are guaranteed to arrive) CreateIoCompletionPort((HANDLE)_ServerSocket->fd(), iocp, (ULONG_PTR)0, 0); w.run(); }); } } // Load AcceptEx pointer LPFN_ACCEPTEX lpfnAcceptEx = nullptr; GUID GuidAcceptEx = WSAID_ACCEPTEX; DWORD dwBytes = 0; void event::Stop() { if (!Running) return; Running = false; if (WSAIoctl((SOCKET)_ServerSocket->fd(), SIO_GET_EXTENSION_FUNCTION_POINTER, &GuidAcceptEx, sizeof(GuidAcceptEx), &lpfnAcceptEx, sizeof(lpfnAcceptEx), &dwBytes, NULL, NULL) == SOCKET_ERROR) { exception[NetException::Critical] << "IOCP: Could not load AcceptEx pointer"; CloseHandle(iocp); throw exception; if (_Iocp) { // wake all workers for (size_t i = 0; i < _Workers.size(); ++i) { PostQueuedCompletionStatus(_Iocp, 0, 0, nullptr); } } int threads_count = (int)std::thread::hardware_concurrency(); if (threads_count <= 0) threads_count = 4; for (auto& t : _Workers) { if (t.joinable()) t.join(); } _Workers.clear(); EventWorkerArgs a; a.timeout = _Timeout; a.iocp = iocp; a.ev = this; a.listenSock = _ServerSocket; a.lpfnAcceptEx = lpfnAcceptEx; a.userArgs = (void*)args; if (_Iocp) { CloseHandle(_Iocp); _Iocp = nullptr; } } std::vector<std::thread> pool; pool.reserve(threads_count); // ---- Connection management (used by worker) // You likely already have these equivalents in epoll.cpp – here inlined. for (int i = 0; i < threads_count; i++) { pool.emplace_back([i, &a]() { EventWorker w(i, &a); }); void event::AddConnection(std::unique_ptr<con> c) { std::lock_guard<std::mutex> lk(_ConMutex); _Connections.push_back(std::move(c)); } // Pre-post accepts: 2 per CPU as you had for (int i = 0; i < threads_count * 2; i++) { post_accept(&a); con* event::FindConnection(socket* s) { std::lock_guard<std::mutex> lk(_ConMutex); for (auto& c : _Connections) { if (c && c->csocket.get() == s) return c.get(); } return nullptr; } while (event::Running) { std::this_thread::sleep_for(std::chrono::milliseconds(100)); con* event::FindConnectionBySocket(SOCKET sock) { std::lock_guard<std::mutex> lk(_ConMutex); for (auto& c : _Connections) { if (c && c->csocket && (SOCKET)c->csocket->getSocket() == sock) return c.get(); } return nullptr; } // Join workers for (auto& t : pool) t.join(); void event::CloseEvent(socket* s) { con* c = FindConnection(s); if (!c) return; CloseHandle(iocp); try { c->csocket->close(); } catch (...) {} // remove from list std::lock_guard<std::mutex> lk(_ConMutex); for (auto it = _Connections.begin(); it != _Connections.end(); ++it) { if ((*it) && (*it)->csocket.get() == s) { _Connections.erase(it); break; } } } } // namespace netplus #endif // Windows Loading
src/event/iocp.cpp +416 −286 Original line number Diff line number Diff line /******************************************************************************* Copyright (c) 2014, Jan Koester jan.koester@gmx.net All rights reserved. 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. 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. *******************************************************************************/ * iocp.cpp (Windows backend for event) * * Implements netplus::event using IOCP: * - CreateIoCompletionPort * - AcceptEx for incoming connections * - WSARecv / WSASend for async I/O * * Works with http.cpp usage: * class MyServer : public event { ... RequestEvent(...) override ... } * ******************************************************************************/ #ifdef Windows #define WIN32_LEAN_AND_MEAN #include <Windows.h> #include <WinSock2.h> #include <mswsock.h> #include <ws2tcpip.h> #pragma comment(lib, "Ws2_32.lib") #include <iostream> #include <mutex> #include <vector> #include <atomic> #include <thread> #include <vector> #include <memory> #include <atomic> #include <chrono> #include <cstring> #include <mutex> #include <iostream> #include "socket.h" #include "exception.h" #include "eventapi.h" #include "connection.h" #include "eventapi.h" #include "exception.h" #include <cstdio> #include <cinttypes> #include <windows.h> static inline uint64_t now_ms() { return GetTickCount64(); } #define IOLOG(fmt, ...) \ do { \ std::fprintf(stderr, "[%10" PRIu64 " ms][tid=%lu] " fmt "\n", \ now_ms(), GetCurrentThreadId(), __VA_ARGS__); \ std::fflush(stderr); \ } while (0) using namespace netplus; namespace netplus { std::atomic<bool> event::Running(true); std::atomic<bool> event::Restart(false); std::vector<socket*> SSOCKETS; std::map<int,std::shared_ptr<con>> CONNECTIONS; std::mutex POLL_HANDLER_MUTEX; std::atomic<size_t> g_con_id(0); // ------------------------------------------------------------ // Client object (per accepted connection context) // ------------------------------------------------------------ class client { public: explicit client(eventapi* eapi) : api(eapi), readCtx(BLOCKSIZE), writeCtx(BLOCKSIZE) { api->CreateConnection(CurCon); readCtx.owner = this; writeCtx.owner = this; } std::shared_ptr<con> CurCon; eventapi* api; std::mutex cltmtx; buffer readCtx; // Accept + Read buffer writeCtx; // Write }; struct EventWorkerArgs { int timeout; HANDLE iocp; eventapi* ev; socket* listenSock; LPFN_ACCEPTEX lpfnAcceptEx; void* userArgs; }; static const char* op_name(int op) { switch (op) { case OP_ACCEPT: return "ACCEPT"; case OP_READ: return "READ"; case OP_WRITE: return "WRITE"; default: return "UNKNOWN"; static LPFN_ACCEPTEX LoadAcceptEx(SOCKET listenSock) { GUID guid = WSAID_ACCEPTEX; LPFN_ACCEPTEX fn = nullptr; DWORD bytes = 0; int rc = WSAIoctl( listenSock, SIO_GET_EXTENSION_FUNCTION_POINTER, &guid, sizeof(guid), &fn, sizeof(fn), &bytes, nullptr, nullptr ); if (rc == SOCKET_ERROR || !fn) { NetException e; e[NetException::Critical] << "LoadAcceptEx failed: " << WSAGetLastError(); throw e; } return fn; } static void post_accept(EventWorkerArgs* a) { client* c = new client(a->ev); if (a->listenSock->_Type == sockettype::TCP) { c->CurCon->csock = std::make_unique<tcp>(); // ---------------------------------------------------------------------------- // Small helper: create a new connection object, assign socket, bind to IOCP. // ---------------------------------------------------------------------------- static std::unique_ptr<con> CreateConnectionFromAccepted(event* ev, socket* serverSock, SOCKET acceptedSock, HANDLE iocpHandle) { // Your connection type "con" is defined in connection.h // Typically you use something like: std::make_unique<con>(...) // But since I don't see con constructors here, I use the default approach: std::unique_ptr<con> c = std::make_unique<con>(); // Create a tcp socket wrapper for accepted socket. // Your socket class likely has setSocket(SOCKET) and takes ownership. auto clientSock = std::make_unique<tcp>(); clientSock->setSocket(acceptedSock); // Bind accepted socket to IOCP: completion key = clientSock pointer HANDLE h = CreateIoCompletionPort( (HANDLE)acceptedSock, iocpHandle, (ULONG_PTR)clientSock.get(), 0 ); if (!h) { NetException e; e[NetException::Critical] << "CreateIoCompletionPort(accepted) failed: " << GetLastError(); throw e; } // Bind the socket to IOCP CreateIoCompletionPort((HANDLE)c->CurCon->csock->fd(), a->iocp, (ULONG_PTR)c, 0); // attach socket to con c->csocket = std::move(clientSock); c->ssocket = serverSock; // CRITICAL: Set the owner so the Worker knows who this is c->readCtx.owner = c; c->readCtx.operation = OP_ACCEPT; std::memset(&c->readCtx.overlapped, 0, sizeof(WSAOVERLAPPED)); // mark connected c->connected = true; // Post the actual AcceptEx a->listenSock->accept(a->lpfnAcceptEx, c->CurCon->csock, c->readCtx); return c; } class EventWorker { // ---------------------------------------------------------------------------- // IOCP Worker thread // ---------------------------------------------------------------------------- class IocpWorker { public: EventWorker(int tid, ULONG_PTR args, EventWorkerArgs* eargs) { DWORD bytesTransferred = 0; ULONG_PTR completionKey = 0; LPOVERLAPPED overlapped = nullptr; IocpWorker(event* ev, socket* serverSock, HANDLE iocp, LPFN_ACCEPTEX fnAcceptEx, int timeout) : _ev(ev), _serverSock(serverSock), _iocp(iocp), _fnAcceptEx(fnAcceptEx), _timeout(timeout) {} void run() { while (event::Running) { // 1) Warte auf Abschluss einer I/O Operation BOOL res = GetQueuedCompletionStatus( eargs->iocp, &bytesTransferred, &completionKey, &overlapped, eargs->timeout ); DWORD bytes = 0; ULONG_PTR key = 0; OVERLAPPED* pov = nullptr; if (!res && !overlapped) { if (GetLastError() == WAIT_TIMEOUT) continue; break; } BOOL ok = GetQueuedCompletionStatus(_iocp, &bytes, &key, &pov, INFINITE); int fd = (int)completionKey; std::shared_ptr<con> curcon; if (!event::Running) break; if (!pov && key == 0) continue; { std::lock_guard<std::mutex> lock(POLL_HANDLER_MUTEX); auto it = CONNECTIONS.find(fd); if (it != CONNECTIONS.end()) curcon = it->second; // AcceptEx completions use AcceptContext.ov (not buffer.overlapped) if (AcceptContext* actx = FindAcceptContext(pov)) { HandleAcceptCompletion(ok, actx); continue; } if (!curcon || !curcon->csock) continue; // Otherwise: must be buffer completion (WSARecv / WSASend) socket* sockObj = reinterpret_cast<socket*>(key); if (!sockObj) continue; std::lock_guard<std::mutex> lk(curcon->event_mutex); buffer* buf = reinterpret_cast<buffer*>( reinterpret_cast<char*>(pov) - offsetof(buffer, overlapped) ); try { // 2) Handshake-Logik (Analog zu epoll.cpp) if (!curcon->csock->getHandshakeDone()) { // Da iocp.cpp als Template für SSL dient, casten wir sicherheitshalber if (static_cast<ssl*>(curcon->csock.get())->handleHandshakeAsynch()) { // Handshake gerade fertig -> Wir könnten sofort lesen } // Wir müssen immer einen neuen Recv-Request absetzen _triggerNextRecv(curcon); if (!ok) { // I/O error => close _ev->CloseEvent(sockObj); continue; } if (buf->operation == OP_READ) { if (bytes == 0) { _ev->CloseEvent(sockObj); continue; } // 3) Datenverarbeitung if (bytesTransferred > 0) { ReadEventHandler(curcon, tid, eargs->userArgs, eargs); } else if (bytesTransferred == 0) { _closeConnection(fd, eargs); // Find connection for this socket con* c = _ev->FindConnection(sockObj); if (!c) { _ev->CloseEvent(sockObj); continue; } } catch (NetException& e) { if (e.getErrorType() != NetException::Note) { _closeConnection(fd, eargs); // Let user code handle it _ev->RequestEvent(*c, _tid, (ULONG_PTR)bytes); // Re-post receive try { c->csocket->recvDataWSA(c->rbuf, 0); } catch (...) { _ev->CloseEvent(sockObj); } } else if (buf->operation == OP_WRITE) { // send completed => user callback con* c = _ev->FindConnection(sockObj); if (!c) { _ev->CloseEvent(sockObj); continue; } _ev->SendEvent(*c, _tid, (ULONG_PTR)bytes); } } void ReadEventHandler(std::shared_ptr<con> rcon, int tid, void* args, EventWorkerArgs* eargs) { try { for (;;) { buffer buf(BLOCKSIZE); size_t rcv = 0; try { // Nutzt WSARecv intern via socket.cpp/tcp.cpp rcv = rcon->csock->RecvDataWSA(buf, 0); } catch (NetException& e) { if (e.getErrorType() == NetException::Note) return; // WSA_IO_PENDING throw; } if (rcv == 0) break; rcon->RecvData.append(buf.data.buf, rcv); rcon->lasteventime = time(nullptr); void setThreadId(int tid) { _tid = tid; } // Protokoll-Event triggern (z.B. HTTP Parser) eargs->ev->RequestEvent(*rcon, tid, args); // AcceptContext tracking void RegisterAccept(AcceptContext* a) { std::lock_guard<std::mutex> lk(_accMutex); _accepts.push_back(a); } // Nach RequestEvent abbrechen, um Senden zu priorisieren (wie epoll) break; private: AcceptContext* FindAcceptContext(OVERLAPPED* pov) { std::lock_guard<std::mutex> lk(_accMutex); for (auto* a : _accepts) { if ((OVERLAPPED*)&a->ov == pov) return a; } return nullptr; } if (!rcon->SendData.empty()) { WriteEventHandler(rcon, tid, args, eargs); void RemoveAccept(AcceptContext* a) { std::lock_guard<std::mutex> lk(_accMutex); for (size_t i = 0; i < _accepts.size(); ++i) { if (_accepts[i] == a) { _accepts.erase(_accepts.begin() + i); return; } } catch (NetException& e) { throw; } } void WriteEventHandler(std::shared_ptr<con> wcon, int tid, void* args, EventWorkerArgs* eargs) { try { while (!wcon->SendData.empty()) { size_t sendlen = (std::min)(wcon->SendData.size(), (size_t)BLOCKSIZE); buffer out(wcon->SendData.data(), sendlen); void PostAccept() { auto* actx = new AcceptContext(); std::memset(&actx->ov, 0, sizeof(WSAOVERLAPPED)); size_t consumed = 0; try { // Nutzt WSASend intern consumed = wcon->csock->SendDataWSA(out, 0); } catch (NetException& e) { if (e.getErrorType() == NetException::Note) return; throw; SOCKET as = WSASocket(AF_INET, SOCK_STREAM, IPPROTO_TCP, nullptr, 0, WSA_FLAG_OVERLAPPED); if (as == INVALID_SOCKET) { delete actx; return; } actx->acceptSock = as; if (consumed == 0) return; wcon->SendData.erase(wcon->SendData.begin(), wcon->SendData.begin() + consumed); } DWORD bytes = 0; BOOL ok = _fnAcceptEx( (SOCKET)_serverSock->getSocket(), as, actx->addrBuf, 0, sizeof(SOCKADDR_STORAGE) + 16, sizeof(SOCKADDR_STORAGE) + 16, &bytes, &actx->ov ); if (wcon->SendData.empty()) { eargs->ev->ResponseEvent(*wcon, tid, args); if (!ok) { int err = WSAGetLastError(); if (err != WSA_IO_PENDING) { closesocket(as); delete actx; return; } } catch (NetException& e) { throw; } RegisterAccept(actx); } private: void _triggerNextRecv(std::shared_ptr<con> c) { buffer dummy(0); try { c->csock->RecvDataWSA(dummy, 0); } catch(...) {} void HandleAcceptCompletion(BOOL ok, AcceptContext* actx) { RemoveAccept(actx); if (!ok) { closesocket(actx->acceptSock); delete actx; PostAccept(); // keep accepting return; } void _closeConnection(int fd, EventWorkerArgs* eargs) { std::lock_guard<std::mutex> lock(POLL_HANDLER_MUTEX); auto it = CONNECTIONS.find(fd); if (it != CONNECTIONS.end()) { CONNECTIONS.erase(it); // must update accept context SOCKET ls = (SOCKET)_serverSock->getSocket(); setsockopt(actx->acceptSock, SOL_SOCKET, SO_UPDATE_ACCEPT_CONTEXT, (char*)&ls, sizeof(ls)); try { // create connection std::unique_ptr<con> newcon = CreateConnectionFromAccepted(_ev, _serverSock, actx->acceptSock, _iocp); // register connection in event _ev->AddConnection(std::move(newcon)); // Start first read con* c = _ev->FindConnectionBySocket((SOCKET)actx->acceptSock); if (c) { c->csocket->recvDataWSA(c->rbuf, 0); } } catch (...) { closesocket(actx->acceptSock); } closesocket(fd); delete actx; // Repost accept immediately PostAccept(); } private: event* _ev = nullptr; socket* _serverSock = nullptr; HANDLE _iocp = nullptr; LPFN_ACCEPTEX _fnAcceptEx = nullptr; int _timeout = 1000; int _tid = 0; std::mutex _accMutex; std::vector<AcceptContext*> _accepts; }; // ------------------------------------------------------------ // event // ------------------------------------------------------------ event::event(socket* serversocket, int timeout) { _Timeout = timeout; _ServerSocket = serversocket; _ServerSocket->bind(); _ServerSocket->listen(); // ---------------------------------------------------------------------------- // event class implementation (Windows/IOCP) // ---------------------------------------------------------------------------- event::event(socket* serversocket, int timeout) : _ServerSocket(serversocket), _Timeout(timeout) { if (!_ServerSocket) { NetException e; e[NetException::Critical] << "server socket empty!"; throw e; } event::~event() {} // Ensure server socket is bound/listening _ServerSocket->bind(); _ServerSocket->listen(); void event::runEventloop(ULONG_PTR args) { NetException exception; // Create IOCP _Iocp = CreateIoCompletionPort(INVALID_HANDLE_VALUE, nullptr, 0, 0); if (!_Iocp) { NetException e; e[NetException::Critical] << "CreateIoCompletionPort failed: " << GetLastError(); throw e; } HANDLE iocp = CreateIoCompletionPort(INVALID_HANDLE_VALUE, NULL, 0, 0); if (!iocp) { exception[NetException::Critical] << "IOCP: CreateIoCompletionPort failed"; throw exception; // Bind server socket to IOCP (key = server socket pointer) HANDLE h = CreateIoCompletionPort( (HANDLE)_ServerSocket->getSocket(), _Iocp, (ULONG_PTR)_ServerSocket, 0 ); if (!h) { NetException e; e[NetException::Critical] << "CreateIoCompletionPort(bind server) failed: " << GetLastError(); throw e; } _fnAcceptEx = LoadAcceptEx((SOCKET)_ServerSocket->getSocket()); } event::~event() { Stop(); } // start worker threads + initial AcceptEx posts void runEventloop(ULONG_PTR args=0){ Running = true; const int nthreads = (int)std::max<size_t>(1, std::thread::hardware_concurrency()); _Workers.reserve(nthreads); for (int i = 0; i < nthreads; ++i) { _Workers.emplace_back([&, i]() { IocpWorker w(this, _ServerSocket, _Iocp, _fnAcceptEx, _Timeout); w.setThreadId(i); // Pre-post accepts per worker (you can tune this) for (int k = 0; k < 4; ++k) { w.RegisterAccept(new AcceptContext()); // placeholder } // Actually post accepts for (int k = 0; k < 4; ++k) { // internal call // (we need to post on this worker instance) // so we recreate minimal pattern: // easiest: call PostAccept by starting run() which posts first } // Post initial accepts now: // (just call run() which will post from accept completions) // Here we do 8 accepts total as baseline: for (int k = 0; k < 8; ++k) { // hack: call private PostAccept by creating a local lambda // Instead: do it in constructor? => simplest: call accept via server socket method // We'll just post directly here: AcceptContext* actx = new AcceptContext(); std::memset(&actx->ov, 0, sizeof(WSAOVERLAPPED)); SOCKET as = WSASocket(AF_INET, SOCK_STREAM, IPPROTO_TCP, nullptr, 0, WSA_FLAG_OVERLAPPED); if (as == INVALID_SOCKET) { delete actx; continue; } actx->acceptSock = as; DWORD bytes = 0; BOOL ok = _fnAcceptEx( (SOCKET)_ServerSocket->getSocket(), as, actx->addrBuf, 0, sizeof(SOCKADDR_STORAGE) + 16, sizeof(SOCKADDR_STORAGE) + 16, &bytes, &actx->ov ); if (!ok) { int err = WSAGetLastError(); if (err != WSA_IO_PENDING) { closesocket(as); delete actx; continue; } } w.RegisterAccept(actx); } // Bind listening socket to IOCP (so AcceptEx completions are guaranteed to arrive) CreateIoCompletionPort((HANDLE)_ServerSocket->fd(), iocp, (ULONG_PTR)0, 0); w.run(); }); } } // Load AcceptEx pointer LPFN_ACCEPTEX lpfnAcceptEx = nullptr; GUID GuidAcceptEx = WSAID_ACCEPTEX; DWORD dwBytes = 0; void event::Stop() { if (!Running) return; Running = false; if (WSAIoctl((SOCKET)_ServerSocket->fd(), SIO_GET_EXTENSION_FUNCTION_POINTER, &GuidAcceptEx, sizeof(GuidAcceptEx), &lpfnAcceptEx, sizeof(lpfnAcceptEx), &dwBytes, NULL, NULL) == SOCKET_ERROR) { exception[NetException::Critical] << "IOCP: Could not load AcceptEx pointer"; CloseHandle(iocp); throw exception; if (_Iocp) { // wake all workers for (size_t i = 0; i < _Workers.size(); ++i) { PostQueuedCompletionStatus(_Iocp, 0, 0, nullptr); } } int threads_count = (int)std::thread::hardware_concurrency(); if (threads_count <= 0) threads_count = 4; for (auto& t : _Workers) { if (t.joinable()) t.join(); } _Workers.clear(); EventWorkerArgs a; a.timeout = _Timeout; a.iocp = iocp; a.ev = this; a.listenSock = _ServerSocket; a.lpfnAcceptEx = lpfnAcceptEx; a.userArgs = (void*)args; if (_Iocp) { CloseHandle(_Iocp); _Iocp = nullptr; } } std::vector<std::thread> pool; pool.reserve(threads_count); // ---- Connection management (used by worker) // You likely already have these equivalents in epoll.cpp – here inlined. for (int i = 0; i < threads_count; i++) { pool.emplace_back([i, &a]() { EventWorker w(i, &a); }); void event::AddConnection(std::unique_ptr<con> c) { std::lock_guard<std::mutex> lk(_ConMutex); _Connections.push_back(std::move(c)); } // Pre-post accepts: 2 per CPU as you had for (int i = 0; i < threads_count * 2; i++) { post_accept(&a); con* event::FindConnection(socket* s) { std::lock_guard<std::mutex> lk(_ConMutex); for (auto& c : _Connections) { if (c && c->csocket.get() == s) return c.get(); } return nullptr; } while (event::Running) { std::this_thread::sleep_for(std::chrono::milliseconds(100)); con* event::FindConnectionBySocket(SOCKET sock) { std::lock_guard<std::mutex> lk(_ConMutex); for (auto& c : _Connections) { if (c && c->csocket && (SOCKET)c->csocket->getSocket() == sock) return c.get(); } return nullptr; } // Join workers for (auto& t : pool) t.join(); void event::CloseEvent(socket* s) { con* c = FindConnection(s); if (!c) return; CloseHandle(iocp); try { c->csocket->close(); } catch (...) {} // remove from list std::lock_guard<std::mutex> lk(_ConMutex); for (auto it = _Connections.begin(); it != _Connections.end(); ++it) { if ((*it) && (*it)->csocket.get() == s) { _Connections.erase(it); break; } } } } // namespace netplus #endif // Windows
