test (d8e2ba79) · Commits · jan.koester / libnetplus

src/event/iocp.cpp

+184 −239

Original line number	Diff line number	Diff line
		@@ -58,12 +58,10 @@ namespace netplus {
		std::atomic<bool> event::Running(true);

		// -------------------------------------------------------------------------
		// AcceptContext matches socket.h concept: keep overlapped + accept socket + buffer
		// AcceptContext matches socket.h concept: keep accept socket + buffer
		// -------------------------------------------------------------------------
		struct AcceptCtx {
		WSAOVERLAPPED ov{};
		SOCKET acceptSock = INVALID_SOCKET;
		char addrBuf[2 * (sizeof(SOCKADDR_STORAGE) + 16)];
		std::unique_ptr<socket> csock;
		};

		// -------------------------------------------------------------------------
		@@ -76,8 +74,9 @@ namespace netplus {
		std::mutex conMutex;
		std::unordered_map<SOCKET, std::shared_ptr<con>> sockToCon;

		std::mutex bufMutex;
		std::unordered_map<con*, std::unique_ptr<buffer>> recvBuf; // persistent read buffer per con
		// Recv buffers are now in sockets
		// std::mutex bufMutex;
		// std::unordered_map<con*, std::unique_ptr<buffer>> recvBuf;

		std::mutex accMutex;
		std::unordered_map<OVERLAPPED, AcceptCtx> acceptByOv;
		@@ -148,6 +147,7 @@ namespace netplus {
		}

		st->fnAcceptEx = loadAcceptEx((SOCKET)serverSock->fd());
		st->threads = ev->threads;

		g_states[ev] = std::move(st);
		}
		@@ -157,37 +157,22 @@ namespace netplus {
		// -------------------------------------------------------------------------
		static void post_accept(EventState* st, socket* serverSock) {
		auto* ctx = new AcceptCtx();
		std::memset(&ctx->ov, 0, sizeof(WSAOVERLAPPED));

		ctx->acceptSock = WSASocket(AF_INET, SOCK_STREAM, IPPROTO_TCP, nullptr, 0, WSA_FLAG_OVERLAPPED);
		if (ctx->acceptSock == INVALID_SOCKET) {
		delete ctx;
		return;
		}

		DWORD bytes = 0;
		BOOL ok = st->fnAcceptEx((SOCKET)serverSock->fd(),
		ctx->acceptSock,
		ctx->addrBuf,
		0,
		sizeof(SOCKADDR_STORAGE) + 16,
		sizeof(SOCKADDR_STORAGE) + 16,
		&bytes,
		&ctx->ov);

		if (!ok) {
		int err = WSAGetLastError();
		if (err != WSA_IO_PENDING) {
		closesocket(ctx->acceptSock);
		try {
		// Uses the virtual accept which uses internal buffer
		serverSock->accept(st->fnAcceptEx, ctx->csock);
		} catch (...) {
		delete ctx;
		return;
		}
		}

		// remember accept ctx by overlapped*
		{
		if (ctx->csock && ctx->csock->_ReadBuffer) {
		WSAOVERLAPPED* ov = &ctx->csock->_ReadBuffer->overlapped;
		std::lock_guard<std::mutex> lk(st->accMutex);
		st->acceptByOv[(OVERLAPPED*)&ctx->ov] = ctx;
		st->acceptByOv[(OVERLAPPED*)ov] = ctx;
		} else {
		delete ctx;
		}
		}

		@@ -231,40 +216,33 @@ namespace netplus {
		}
		// Also remove the recv buffer for this connection
		if (cptr) {
		std::lock_guard<std::mutex> lk(st->bufMutex);
		st->recvBuf.erase(cptr);
		// Recv buffer is now owned by socket, so no need to erase from EventState
		}
		}

		// -------------------------------------------------------------------------
		// Ensure persistent recv buffer exists for a connection
		// -------------------------------------------------------------------------
		static buffer& ensure_recv_buffer(EventState* st, con& c) {
		std::lock_guard<std::mutex> lk(st->bufMutex);
		auto it = st->recvBuf.find(&c);
		if (it == st->recvBuf.end()) {
		auto b = std::make_unique<buffer>(BLOCKSIZE);
		static buffer* ensure_recv_buffer(EventState* st, con& c) {
		if (!c.csock->_ReadBuffer) {
		c.csock->_ReadBuffer = std::make_unique<buffer>(BLOCKSIZE);
		}
		buffer* b = c.csock->_ReadBuffer.get();
		b->operation = OP_READ;
		b->owner = &c;
		auto* ptr = b.get();
		st->recvBuf[&c] = std::move(b);
		return *ptr;
		}
		it->second->operation = OP_READ;
		it->second->owner = &c;
		return *(it->second);
		return b;
		}

		// -------------------------------------------------------------------------
		// Post receive for a connection
		// -------------------------------------------------------------------------
		static void post_recv(EventState* st, con& c) {
		buffer& b = ensure_recv_buffer(st, c);
		std::memset(&b.overlapped, 0, sizeof(WSAOVERLAPPED));
		b.operation = OP_READ;
		b.owner = &c;
		buffer* b = ensure_recv_buffer(st, c);
		std::memset(&b->overlapped, 0, sizeof(WSAOVERLAPPED));
		b->operation = OP_READ;
		b->owner = &c;

		c.csock->prime_read(b);
		c.csock->prime_read();
		}

		// -------------------------------------------------------------------------
		@@ -280,7 +258,7 @@ namespace netplus {
		const size_t remaining = avail - c.SendOff;
		const char* ptr = c.SendData.data() + c.SendOff;

		bool isSSL = (c.csock->getSocketType() == (int)sockettype::SSL);
		bool isSSL = (c.csock->getSocketType() == sockettype::SSL);

		if (isSSL) {
		// For SSL: use synchronous encrypted send via sendData
		@@ -305,20 +283,25 @@ namespace netplus {
		return false;
		}
		} else {
		// For plain TCP: use async WSASend
		auto* sbuf = new buffer(ptr, remaining);
		sbuf->operation = OP_WRITE;
		sbuf->owner = &c;
		std::memset(&sbuf->overlapped, 0, sizeof(WSAOVERLAPPED));

		// For plain TCP: use async WSASend (via flush_out)
		c.WritePending.store(true);

		try {
		c.csock->sendDataWSA(*sbuf, 0);
		// Ensure _SendBuffer is large enough
		if (!c.csock->_SendBuffer) {
		c.csock->_SendBuffer = std::make_unique<buffer>(remaining);
		}
		if (c.csock->_SendBuffer->size < remaining) {
		c.csock->_SendBuffer = std::make_unique<buffer>(remaining);
		}

		std::memcpy(c.csock->_SendBuffer->data.buf, ptr, remaining);
		c.csock->_SendBuffer->data.len = (ULONG)remaining;

		c.csock->flush_out();
		return false; // Async - completion will come later
		} catch (...) {
		c.WritePending.store(false);
		delete sbuf;
		throw;
		}
		}
		@@ -327,7 +310,9 @@ namespace netplus {
		// -------------------------------------------------------------------------
		// Worker loop
		// -------------------------------------------------------------------------
		static void worker_loop(event* ev, socket* serverSock, int tid) {
		class EventWorker {
		public:
		void operator()(event* ev, socket* serverSock, int tid) {
		EventState* st = ST(ev);

		while (event::Running) {
		@@ -349,10 +334,12 @@ namespace netplus {
		if (AcceptCtx* actx = pop_accept(st, pov)) {
		std::cerr << "[IOCP] AcceptEx completion detected!" << std::endl;

		SOCKET accepted = actx->acceptSock;
		// Take ownership of the accepted socket
		std::unique_ptr<socket> acceptedSock = std::move(actx->csock);
		SOCKET accepted = acceptedSock->fd();

		if (!ok) {
		closesocket(accepted);
		// socket closed by destructor
		delete actx;
		post_accept(st, serverSock);
		continue;
		@@ -363,25 +350,17 @@ namespace netplus {
		setsockopt(accepted, SOL_SOCKET, SO_UPDATE_ACCEPT_CONTEXT, (char*)&ls, sizeof(ls));

		// Create connection via user allocator
		std::shared_ptr<con> c;
		ev->CreateConnection(c);
		std::shared_ptr<con> c; // create empty shared_ptr
		ev->CreateConnection(c); // populate it

		if (!c) {
		closesocket(accepted);
		delete actx;
		post_accept(st, serverSock);
		continue;
		}

		// attach accepted socket to con::csock
		if(serverSock->_Type==sockettype::TCP)
		c->csock = std::make_unique<tcp>(accepted);
		else if(serverSock->_Type==sockettype::SSL) {
		ssl* sslServer = static_cast<ssl*>(serverSock);
		c->csock = std::make_unique<ssl>(sslServer->getCert(), (int)accepted);
		}
		else
		continue;
		c->csock = std::move(acceptedSock); // Transfers type (tcp/ssl) and state

		// bind accepted socket to IOCP (key = socket*)
		HANDLE h = CreateIoCompletionPort((HANDLE)accepted,
		@@ -389,7 +368,6 @@ namespace netplus {
		(ULONG_PTR)c->csock.get(),
		0);
		if (!h) {
		closesocket(accepted);
		delete actx;
		post_accept(st, serverSock);
		continue;
		@@ -408,7 +386,7 @@ namespace netplus {
		} catch (...) {
		ev->DisconnectEvent(*c, tid, 0);
		remove_con(st, accepted);
		c->csock->close();
		try { c->csock->close(); } catch (...) {}
		}

		delete actx;
		@@ -437,7 +415,8 @@ namespace netplus {
		// Connection was already removed, ignore this completion
		std::cerr << "[IOCP] Connection not found for fd=" << cs << ", ignoring" << std::endl;
		if (buf->operation == OP_WRITE) {
		delete buf; // send buffers are dynamically allocated
		// Internal send buffer (owned by socket)
		// Do not delete!
		}
		continue;
		}
		@@ -450,7 +429,7 @@ namespace netplus {
		remove_con(st, cs);
		try { owner->csock->close(); } catch (...) {}
		if (buf->operation == OP_WRITE) {
		delete buf;
		// Internal send buffe, no delete
		}
		continue;
		}
		@@ -465,20 +444,11 @@ namespace netplus {
		}

		// For SSL: need to handle handshake and decryption
		bool isSSL = (owner->csock->getSocketType() == (int)sockettype::SSL);
		std::cerr << "[IOCP] isSSL=" << isSSL << " handshakeDone=" << owner->csock->getHandshakeDone() << std::endl;

		if (isSSL) {
		// Cast to ssl* to access feedRawBytes
		ssl* sslSock = static_cast<ssl*>(owner->csock.get());

		// Feed the raw bytes we received into the SSL internal buffer
		std::cerr << "[IOCP] Feeding " << bytes << " raw bytes to SSL" << std::endl;
		sslSock->feedRawBytes(buf->data.buf, bytes);
		std::cerr << "[IOCP] handshakeDone=" << owner->csock->getHandshakeDone() << std::endl;

		// Check if handshake is done
		if (!owner->csock->getHandshakeDone()) {
		std::cerr << "[IOCP] Starting handshake_after_accept..." << std::endl;
		std::cerr << "[IOCP] Starting handshake_after_accept (fd=" << owner->csock->fd() << ")..." << std::endl;
		std::cerr.flush();
		try {
		// Process handshake from internal buffer
		@@ -494,18 +464,13 @@ namespace netplus {
		if (owner->csock->hasPendingWrite()) {
		std::cerr << "[IOCP] Note but hasPendingWrite - flushing first..." << std::endl;
		try {
		sslSock->flush_out();
		buffer* sendBuf = sslSock->getIocpBuf();
		if (sendBuf) {
		sendBuf->owner = owner;
		std::cerr << "[IOCP] flush_out posted " << sendBuf->size << " bytes" << std::endl;
		}
		owner->csock->flush_out();
		} catch (...) {
		std::cerr << "[IOCP] flush_out failed during Note handling" << std::endl;
		}
		}
		// Repost recv for more handshake data
		std::cerr << "[IOCP] Reposting recv for more handshake data" << std::endl;
		std::cerr << "[IOCP] Reposting recv for more handshake data (fd=" << owner->csock->fd() << ")" << std::endl;
		try {
		post_recv(st, *owner);
		} catch (...) {}
		@@ -534,12 +499,7 @@ namespace netplus {
		// If there's pending data to send (handshake response)
		if (owner->csock->hasPendingWrite()) {
		std::cerr << "[IOCP] Flushing handshake response via IOCP..." << std::endl;
		sslSock->flush_out();
		buffer* sendBuf = sslSock->getIocpBuf();
		if (sendBuf) {
		sendBuf->owner = owner;
		std::cerr << "[IOCP] flush_out posted " << sendBuf->size << " bytes" << std::endl;
		}
		owner->csock->flush_out();
		}

		// If handshake still not done, wait for more data
		@@ -552,12 +512,14 @@ namespace netplus {
		}
		}

		// Handshake done - try to decrypt application data from internal buffer
		// Handshake done - retrieve application data
		try {
		while (true) {
		buffer decrypted(BLOCKSIZE);
		size_t decLen = owner->csock->recvData(decrypted, 0);

		if (decLen > 0) {
		if (decLen == 0) break;

		owner->RecvData.append(decrypted.data.buf, decLen);
		ev->RequestEvent(*owner, tid, (ULONG_PTR)decLen);
		}
		@@ -570,11 +532,6 @@ namespace netplus {
		}
		// Note = no complete record yet, just continue
		}
		} else {
		// Plain TCP: append directly
		owner->RecvData.append(buf->data.buf, (size_t)bytes);
		ev->RequestEvent(*owner, tid, (ULONG_PTR)bytes);
		}

		// if user filled SendData -> try send
		try {
		@@ -599,22 +556,27 @@ namespace netplus {
		else if (buf->operation == OP_WRITE) {
		// For SSL, the send encrypts all pending data at once
		// So we clear SendData entirely on completion
		bool isSSL = (owner->csock->getSocketType() == (int)sockettype::SSL);
		bool isSSL = (owner->csock->getSocketType() == sockettype::SSL);

		if (isSSL) {
		// SSL: all data was encrypted and sent
		owner->SendData.clear();
		owner->SendOff = 0;
		} else {
		// Plain TCP: track bytes sent
		owner->SendOff += (size_t)bytes;

		// finished sending? clear buffer
		if (owner->SendOff >= owner->SendData.size()) {
		owner->SendData.clear();
		owner->SendOff = 0;
		}
		}

		// If handshake not done, continue handshake after send completes
		if (!owner->csock->getHandshakeDone()) {
		ssl* sslSock = static_cast<ssl*>(owner->csock.get());
		std::cerr << "[IOCP] Write completed during handshake, continuing handshake..." << std::endl;

		// Delete the send buffer first
		delete buf;
		buf = nullptr;

		try {
		owner->csock->handshake_after_accept();
		std::cerr << "[IOCP] handshake_after_accept returned, handshakeDone=" << owner->csock->getHandshakeDone() << std::endl;
		@@ -622,11 +584,7 @@ namespace netplus {
		// If there's more data to send, flush it
		if (owner->csock->hasPendingWrite()) {
		std::cerr << "[IOCP] More data to send, flushing..." << std::endl;
		sslSock->flush_out();
		buffer* sendBuf = sslSock->getIocpBuf();
		if (sendBuf) {
		sendBuf->owner = owner;
		}
		owner->csock->flush_out();
		}
		} catch (NetException& e) {
		if (e.getErrorType() == NetException::Note) {
		@@ -634,11 +592,7 @@ namespace netplus {
		// Need more data - but first check if we have data to send
		if (owner->csock->hasPendingWrite()) {
		try {
		sslSock->flush_out();
		buffer* sendBuf = sslSock->getIocpBuf();
		if (sendBuf) {
		sendBuf->owner = owner;
		}
		owner->csock->flush_out();
		} catch (...) {}
		}
		// Post recv for more handshake data
		@@ -672,16 +626,6 @@ namespace netplus {
		owner->WritePending.store(false);
		continue;
		}
		} else {
		// Plain TCP: track bytes sent
		owner->SendOff += (size_t)bytes;

		// finished sending? clear buffer
		if (owner->SendOff >= owner->SendData.size()) {
		owner->SendData.clear();
		owner->SendOff = 0;
		}
		}

		owner->WritePending.store(false);

		@@ -697,11 +641,12 @@ namespace netplus {
		try { owner->csock->close(); } catch (...) {}
		}

		// send buffer was dynamically allocated
		if (buf) delete buf;
		// send buffer is internal - no delete
		// if (buf) delete buf;
		}
		}
		}
		};

		// -------------------------------------------------------------------------
		// event implementation
		@@ -770,7 +715,7 @@ namespace netplus {
		// start workers
		for (int i = 0; i < st->threads; ++i) {
		st->workers.emplace_back([=]() {
		worker_loop(this, _ServerSocket, i);
		EventWorker()(this, _ServerSocket, i);
		});
		}

src/socket.h

+14 −23

Original line number	Diff line number	Diff line
		@@ -28,6 +28,7 @@
		#pragma once

		#include <array>
		#include <iostream>
		#include <atomic>
		#include <string>
		#include <memory>
		@@ -161,10 +162,10 @@ namespace netplus {
		virtual void handshake_after_connect(){};

		#ifdef Windows
		virtual void accept(LPFN_ACCEPTEX lpfnAcceptEx, std::unique_ptr<socket>& csock, buffer& data) = 0;
		virtual void sendDataWSA(buffer& data, int flags) = 0;
		virtual void recvDataWSA(buffer& data, int flags) = 0;
		virtual void prime_read(buffer& data) = 0;
		virtual void accept(LPFN_ACCEPTEX lpfnAcceptEx, std::unique_ptr<socket>& csock) = 0;
		virtual void prime_read() = 0;
		std::unique_ptr<buffer> _ReadBuffer;
		std::unique_ptr<buffer> _SendBuffer;
		#endif

		virtual void bind() = 0;
		@@ -188,7 +189,7 @@ namespace netplus {

		virtual socket& operator=(SOCKET sock) = 0;
		virtual void close();
		virtual void flush_out(){};
		virtual void flush_out() = 0;

		// ===================================================
		// ✅ NEW ADDRESS STORAGE (Linux + Windows also usable)
		@@ -238,12 +239,11 @@ namespace netplus {
		tcp(const std::string& uxsocket, int maxconnections, int sockopts);
		tcp(const std::string& addr, int port, int maxconnections, int sockopts);
		~tcp();
		void flush_out() override;

		#ifdef Windows
		void accept(LPFN_ACCEPTEX lpfnAcceptEx, std::unique_ptr<socket>& csock, buffer& data);
		void sendDataWSA(buffer& data, int flags);
		void recvDataWSA(buffer& data, int flags);
		void prime_read(buffer& data);
		void accept(LPFN_ACCEPTEX lpfnAcceptEx, std::unique_ptr<socket>& csock);
		void prime_read();
		#endif

		void accept(std::unique_ptr<socket>& csock,bool nonblock) override;
		@@ -276,14 +276,13 @@ namespace netplus {
		udp(const std::string& uxsocket, int maxconnections,int sockopts);
		udp(const std::string& addr, int port, int maxconnections,int sockopts);
		~udp();
		void flush_out() override;

		void accept(std::unique_ptr<socket>& csock,bool nonblock) override;

		#ifdef Windows
		void accept(LPFN_ACCEPTEX lpfnAcceptEx, std::unique_ptr<socket>& csock, buffer& data);
		void sendDataWSA(buffer& data, int flags);
		void recvDataWSA(buffer& data, int flags);
		void prime_read(buffer& data);
		void accept(LPFN_ACCEPTEX lpfnAcceptEx, std::unique_ptr<socket>& csock);
		void prime_read();
		#endif

		void bind() override;
		@@ -327,11 +326,8 @@ namespace netplus {
		size_t recvData(buffer& data, int flags = 0) override;

		#ifdef Windows
		void accept(LPFN_ACCEPTEX lpfnAcceptEx, std::unique_ptr<socket>& csock, buffer& data) override;
		void sendDataWSA(buffer& data, int flags);
		void recvDataWSA(buffer& data, int flags);
		void prime_read(buffer& data);
		buffer* getIocpBuf() { return _iocpBuf; }
		void accept(LPFN_ACCEPTEX lpfnAcceptEx, std::unique_ptr<socket>& csock) override;
		void prime_read();
		#endif

		bool loadServerPrivateKeyDer(const std::string& keyDerPath);
		@@ -352,11 +348,6 @@ namespace netplus {

		void resetTLS();

		// Feed raw bytes received from async I/O (IOCP) into internal buffer
		void feedRawBytes(const char* data, size_t len) {
		_rx_tcp_buf.insert(_rx_tcp_buf.end(), data, data + len);
		}

		// Public accessor for certificate (used by event loop for SSL accept)
		std::shared_ptr<netplus::x509cert> getCert() const { return _cert; }

src/ssl.cpp

+78 −64

File changed.

Preview size limit exceeded, changes collapsed.

src/windows/tcp.cpp

+61 −24

File changed.

Preview size limit exceeded, changes collapsed.

src/windows/udp.cpp

+46 −21

File changed.

Preview size limit exceeded, changes collapsed.