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

config.h.in

+1 −0

Original line number	Diff line number	Diff line
		#define ${CMAKE_SYSTEM_NAME}
		#define ARCH_${CMAKE_SYSTEM_PROCESSOR}
		#define THREAD_STACK_SIZE (4 * 1024 * 1024)
		#define BLOCKSIZE 16384
		No newline at end of file

src/event/iocp.cpp

+297 −546

Original line number	Diff line number	Diff line
		@@ -26,9 +26,6 @@ SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
		*******************************************************************************/

		#include <iostream>
		#include <iomanip>
		#include <algorithm>
		#include <chrono>
		#include <mutex>
		#include <cstring>
		#include <vector>
		@@ -38,7 +35,6 @@ SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
		#include <unordered_map>

		#include <winsock2.h>
		#include <ws2ipdef.h>
		#include <mswsock.h>

		#include "socket.h"
		@@ -46,46 +42,25 @@ SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
		#include "eventapi.h"
		#include "connection.h"

		#define BLOCKSIZE 16384

		namespace netplus {
		std::atomic<bool> event::Running(true);
		std::atomic<bool> event::Restart(false);

		// AcceptEx tracking (OVERLAPPED* -> pending accepted socket)
		// Verwaltung ausstehender Accept-Operationen
		static std::mutex ACCEPT_MTX;
		static std::unordered_map<OVERLAPPED*, std::unique_ptr<socket>> ACCEPT_PENDING;

		enum IO_OPERATION { OP_READ = 0, OP_WRITE = 1 };

		struct IO_CONTEXT {
		WSAOVERLAPPED overlapped;
		WSABUF wsaBuf;
		char buffer[BLOCKSIZE];
		IO_OPERATION operation;
		};
		static std::unordered_map<WSAOVERLAPPED*, std::unique_ptr<socket>> ACCEPT_PENDING;

		class client {
		public:
		client(eventapi* eapi) : api(eapi) {
		client(eventapi* eapi) : api(eapi), readCtx(BLOCKSIZE), writeCtx(BLOCKSIZE) {
		api->CreateConnection(CurCon);

		memset(&readCtx, 0, sizeof(IO_CONTEXT));
		readCtx.operation = OP_READ;
		readCtx.wsaBuf.buf = readCtx.buffer;
		readCtx.wsaBuf.len = BLOCKSIZE;

		memset(&writeCtx, 0, sizeof(IO_CONTEXT));
		writeCtx.operation = OP_WRITE;
		writeCtx.wsaBuf.buf = writeCtx.buffer;
		writeCtx.wsaBuf.len = 0;
		}

		std::shared_ptr<con> CurCon;
		eventapi* api;
		std::mutex cltmtx;
		IO_CONTEXT readCtx;
		IO_CONTEXT writeCtx;
		buffer readCtx; // Nutzt die neue buffer-Klasse mit integriertem OVERLAPPED
		buffer writeCtx;
		};

		class EventWorkerArgs {
		@@ -100,207 +75,82 @@ namespace netplus {
		LPFN_ACCEPTEX lpfnAcceptEx;
		};

		// ---- AcceptEx integration (epoll-like accept without a blocking listener thread) ----
		struct PendingAccept {
		std::unique_ptr<socket> sock;
		};

		static std::mutex g_accept_mtx;
		static std::unordered_map<OVERLAPPED, PendingAccept> g_accept_by_ov;

		static void free_accept_buffer(socket& s) {
		#ifdef Windows
		if (s._Extension) {
		auto* p = reinterpret_cast<char*>(s._Extension);
		delete[] p;
		s._Extension = 0;
		}
		#endif
		}

		static LPFN_ACCEPTEX load_acceptex(SOCKET listenSock) {
		GUID guidAcceptEx = WSAID_ACCEPTEX;
		LPFN_ACCEPTEX fn = nullptr;
		DWORD bytes = 0;
		int rc = WSAIoctl(listenSock,
		SIO_GET_EXTENSION_FUNCTION_POINTER,
		&guidAcceptEx,
		sizeof(guidAcceptEx),
		&fn,
		sizeof(fn),
		&bytes,
		nullptr,
		nullptr);
		if (rc == SOCKET_ERROR \|\| !fn) {
		NetException e;
		e[NetException::Error] << "WSAIoctl(SIO_GET_EXTENSION_FUNCTION_POINTER, AcceptEx) failed: " << WSAGetLastError();
		throw e;
		}
		return fn;
		}

		static void post_accept(EventWorkerArgs* eargs, LPFN_ACCEPTEX fnAcceptEx) {
		// Allocate a pending accept slot that owns the accept socket object.
		auto* pa = new PendingAccept;

		if (eargs->ssocket->_Type == sockettype::TCP) {
		pa->sock = std::make_unique<tcp>();
		}
		else if (eargs->ssocket->_Type == sockettype::SSL) {
		auto* srv = static_cast<ssl*>(eargs->ssocket);
		pa->sock = std::make_unique<ssl>(srv->_cert);
		}
		else {
		delete pa;
		return;
		}

		// Post AcceptEx via the socket implementation.
		eargs->ssocket->accept(fnAcceptEx, pa->sock);

		// Track by OVERLAPPED pointer used for AcceptEx.
		{
		std::lock_guard<std::mutex> lk(g_accept_mtx);
		g_accept_by_ov[reinterpret_cast<OVERLAPPED*>(&pa->sock->_Overlapped)] = pa;
		}
		}

		class EventWorker {
		public:
		// Startet den nächsten Lese-Vorgang (Agnostisch für TCP/SSL)
		static void start_read(client* ctx) {
		con& c = *ctx->CurCon;
		buffer buf(ctx->readCtx.buffer, BLOCKSIZE);

		if (c.csock->_Type == sockettype::SSL) {
		static_cast<ssl*>(c.csock.get())->tcp::recvDataWSA(buf, &ctx->readCtx.overlapped, 0);
		if (!ctx->CurCon->csock) return;
		// Die Socket-Implementierung (TCP/SSL) kümmert sich um die Details
		ctx->CurCon->csock->prime_read(ctx->readCtx);
		}
		else {
		static_cast<tcp*>(c.csock.get())->recvDataWSA(buf, &ctx->readCtx.overlapped, 0);
		}
		}

		static void start_write(client* ctx) {
		con& c = *ctx->CurCon;
		if (c.SendData.empty()) return;

		size_t toSend = std::min<size_t>(BLOCKSIZE, c.SendData.size());
		buffer out(c.SendData.data(), toSend);

		size_t consumed = 0;
		if (c.csock->_Type == sockettype::SSL) {
		consumed = static_cast<ssl*>(c.csock.get())
		->sendDataWSA(out, &ctx->writeCtx.overlapped, 0);
		}
		else {
		consumed = static_cast<tcp*>(c.csock.get())
		->sendDataWSA(out, &ctx->writeCtx.overlapped, 0);
		}

		if (consumed > 0)
		c.SendData.erase(c.SendData.begin(),
		c.SendData.begin() + consumed);
		}

		EventWorker(int tid, ULONG_PTR args, EventWorkerArgs* eargs) {
		EventWorker(int tid, ULONG_PTR /args/, EventWorkerArgs* eargs) {
		while (event::Running) {

		DWORD bytes = 0;
		ULONG_PTR key = 0;
		OVERLAPPED* ov = nullptr;

		BOOL ok = GetQueuedCompletionStatus(
		eargs->eviocp, &bytes, &key, &ov, eargs->timeout);

		// Warte auf Abschluss einer asynchronen Operation
		BOOL ok = GetQueuedCompletionStatus(eargs->eviocp, &bytes, &key, &ov, eargs->timeout);
		if (!ov) continue;

		// --------------------------------------------------------------------
		// 1) AcceptEx completions arrive on the IOCP with key == listenerKey
		// DO NOT cast key to client* in that case.
		// --------------------------------------------------------------------
		// --- FALL 1: Neue Verbindung (AcceptEx Abschluss) ---
		if (key == eargs->listenerKey) {
		std::unique_ptr<socket> accepted;
		{
		std::lock_guard<std::mutex> lk(ACCEPT_MTX);
		auto it = ACCEPT_PENDING.find(ov);
		auto it = ACCEPT_PENDING.find(reinterpret_cast<WSAOVERLAPPED*>(ov));
		if (it != ACCEPT_PENDING.end()) {
		accepted = std::move(it->second);
		ACCEPT_PENDING.erase(it);
		}
		}

		if (!accepted) {
		std::cerr << "[IOCP] AcceptEx completion: no pending entry for ov=" << ov << "\n";
		continue;
		}

		SOCKET accSock = (SOCKET)accepted->fd();
		std::cerr << "[IOCP] AcceptEx completion: accepted fd=" << accSock
		<< " ov=" << ov << " bytes=" << bytes << "\n";
		if (!accepted) continue;

		setsockopt(accSock, SOL_SOCKET, SO_UPDATE_ACCEPT_CONTEXT,
		// Kontext für Peer-Informationen aktualisieren
		setsockopt((SOCKET)accepted->fd(), SOL_SOCKET, SO_UPDATE_ACCEPT_CONTEXT,
		(char*)&eargs->listenSock, sizeof(eargs->listenSock));

		// Build client connection
		client* pClient = new client(eargs->event);
		pClient->CurCon->csock = std::move(accepted);

		// Associate accepted socket with IOCP using key = pClient
		HANDLE h = CreateIoCompletionPort(
		(HANDLE)(uintptr_t)pClient->CurCon->csock->fd(),
		eargs->eviocp,
		(ULONG_PTR)pClient,
		0);
		client* ctx = new client(eargs->event);
		ctx->CurCon->csock = std::move(accepted);
		con& c = *ctx->CurCon;

		if (!h) {
		std::cerr << "[IOCP] CreateIoCompletionPort(accepted) failed fd="
		<< pClient->CurCon->csock->fd()
		<< " err=" << GetLastError() << "\n";
		eargs->event->DisconnectEvent(*pClient->CurCon, tid, (ULONG_PTR)eargs->args);
		delete pClient;
		// Socket an IOCP binden
		if (CreateIoCompletionPort((HANDLE)(uintptr_t)c.csock->fd(), eargs->eviocp, (ULONG_PTR)ctx, 0)) {
		eargs->event->ConnectEvent(c, tid, (ULONG_PTR)eargs->args);
		start_read(ctx);
		}
		else {
		std::cerr << "[IOCP] calling ConnectEvent for fd="
		<< pClient->CurCon->csock->fd() << "\n";
		eargs->event->ConnectEvent(*pClient->CurCon, tid, (ULONG_PTR)eargs->args);

		// Start reading raw bytes immediately (TLS handshake will be stepped from OP_READ)
		start_read(pClient);
		delete ctx;
		}

		// Re-post AcceptEx
		// Nächsten Accept vorbereiten
		try {
		std::unique_ptr<socket> nextSock;
		static_cast<ssl*>(eargs->ssocket)->accept(eargs->lpfnAcceptEx, nextSock);
		{
		std::lock_guard<std::mutex> lk(ACCEPT_MTX);
		ACCEPT_PENDING.emplace(&nextSock->_Overlapped, std::move(nextSock));
		std::cerr << "[IOCP] Re-posted AcceptEx; pending_count=" << ACCEPT_PENDING.size() << "\n";
		}
		std::unique_ptr<socket> next;
		if (eargs->ssocket->_Type == sockettype::TCP) {
		next = std::make_unique<tcp>();
		}
		catch (NetException& e) {
		std::cerr << "AcceptEx repost error: " << e.what() << "\n";
		else {
		next = std::make_unique<ssl>(static_cast<ssl*>(eargs->ssocket)->_cert);
		}

		continue;
		eargs->ssocket->accept(eargs->lpfnAcceptEx, next);
		std::lock_guard<std::mutex> lk(ACCEPT_MTX);
		ACCEPT_PENDING.emplace(&next->_Overlapped, std::move(next));
		}

		// --------------------------------------------------------------------
		// 2) Normal client completions: key is client*
		// --------------------------------------------------------------------
		client* ctx = reinterpret_cast<client*>(key);
		if (!ctx \|\| !ctx->CurCon \|\| !ctx->CurCon->csock) {
		std::cerr << "[IOCP] ERROR: invalid client key=" << (void*)key
		<< " ov=" << ov << "\n";
		catch (...) {}
		continue;
		}

		IO_CONTEXT* io = CONTAINING_RECORD(ov, IO_CONTEXT, overlapped);
		// --- FALL 2: Daten-Transfer (Read/Write Abschluss) ---
		client* ctx = reinterpret_cast<client*>(key);
		con& c = *ctx->CurCon;

		// Berechne die buffer-Objektadresse basierend auf dem OVERLAPPED-Pointer
		buffer* pBuf = CONTAINING_RECORD(ov, buffer, overlapped);

		if (!ok \|\| bytes == 0) {
		std::cerr << "[IOCP] closed/error fd=" << c.csock->fd()
		<< " ok=" << ok << " bytes=" << bytes
		<< " err=" << GetLastError() << "\n";
		eargs->event->DisconnectEvent(c, tid, (ULONG_PTR)eargs->args);
		delete ctx;
		continue;
		@@ -309,123 +159,25 @@ namespace netplus {
		try {
		std::lock_guard<std::mutex> guard(ctx->cltmtx);

		if (io->operation == OP_READ) {

		if (c.csock->_Type == sockettype::SSL) {
		ssl* s = static_cast<ssl*>(c.csock.get());

		// Debug header for "bad TLS version"
		if (bytes >= 5) {
		const unsigned char* p = (const unsigned char*)io->buffer;
		std::cerr << "[IOCP][SSL] RX bytes=" << bytes
		<< " hdr: "
		<< std::hex
		<< (int)p[0] << " " << (int)p[1] << " " << (int)p[2] << " "
		<< (int)p[3] << " " << (int)p[4]
		<< std::dec << "\n";
		}
		else {
		std::cerr << "[IOCP][SSL] RX bytes=" << bytes << " (<5)\n";
		}

		// Append ciphertext into SSL net buffer first (handshakeStepIOCP expects it there)
		s->_rx_netbuf.insert(s->_rx_netbuf.end(),
		(unsigned char*)io->buffer,
		(unsigned char*)io->buffer + bytes);

		// --- Handshake phase (IOCP stepped) ---
		if (!s->_handshakeDone) {
		bool progressed = false;
		for (;;) {
		bool step = s->handshakeStepIOCP(); // consumes at most one record per call
		if (!step) break;
		progressed = true;
		}
		if (pBuf->operation == OP_READ) {
		// Verarbeite empfangene Daten (TCP: direkt; SSL: Entschlüsselung)
		size_t processed = c.csock->recvDataWSA(*pBuf, 0);

		// If handshake produced outbound bytes, send them using overlapped write
		if (!s->_hs_tx.empty() && s->_hs_tx_off < s->_hs_tx.size()) {
		// Kick a WSASend of handshake bytes.
		// Use ctx->writeCtx.overlapped, not some undefined "ctx".
		buffer out(
		(const char*)s->_hs_tx.data() + s->_hs_tx_off,
		(int)(s->_hs_tx.size() - s->_hs_tx_off)
		);
		s->tcp::sendDataWSA(out, &ctx->writeCtx.overlapped, 0);
		}

		// Always post next read while handshaking
		start_read(ctx);
		continue;
		}

		// --- Application phase ---
		buffer plain(BLOCKSIZE);
		size_t decrypted = 0;
		while ((decrypted = s->recvDataWSA(plain, nullptr, 0)) > 0) {
		c.RecvData.append(plain.data.buf, decrypted);
		std::cerr << "[IOCP] RequestEvent (SSL) fd=" << c.csock->fd()
		<< " +" << decrypted << " total=" << c.RecvData.size() << "\n";
		if (processed > 0) {
		c.RecvData.append(pBuf->data.buf, (int)processed);
		eargs->event->RequestEvent(c, tid, (ULONG_PTR)eargs->args);
		}
		}
		else {
		// TCP
		c.RecvData.append(io->buffer, bytes);
		std::cerr << "[IOCP] RequestEvent (TCP) fd=" << c.csock->fd()
		<< " +" << bytes << " total=" << c.RecvData.size() << "\n";
		eargs->event->RequestEvent(c, tid, (ULONG_PTR)eargs->args);
		}

		// Continue I/O
		if (!c.SendData.empty()) start_write(ctx);
		else start_read(ctx);
		}
		else if (io->operation == OP_WRITE) {

		if (c.csock->_Type == sockettype::SSL) {
		ssl* s = static_cast<ssl*>(c.csock.get());

		// Handshake TX in progress?
		if (!s->_handshakeDone && !s->_hs_tx.empty()) {
		s->_hs_tx_off += bytes;

		if (s->_hs_tx_off < s->_hs_tx.size()) {
		buffer out(
		(const char*)s->_hs_tx.data() + s->_hs_tx_off,
		(int)(s->_hs_tx.size() - s->_hs_tx_off)
		);
		s->tcp::sendDataWSA(out, &ctx->writeCtx.overlapped, 0);
		continue;
		}

		// handshake flight fully sent
		s->_hs_tx.clear();
		s->_hs_tx_off = 0;

		// keep reading to continue handshake
		// Sofort wieder auf Empfang gehen
		start_read(ctx);
		continue;
		}

		// Normal encrypted record send bookkeeping
		s->_send_off += bytes;
		if (s->_send_off >= s->_send_record.size()) {
		s->_send_record.clear();
		s->_send_off = 0;
		}
		}

		if (c.SendData.empty()) {
		std::cerr << "[IOCP] ResponseEvent fd=" << c.csock->fd() << " send_queue_empty\n";
		else if (pBuf->operation == OP_WRITE) {
		// Sende-Bestätigung verarbeiten
		eargs->event->ResponseEvent(c, tid, (ULONG_PTR)eargs->args);
		}

		if (!c.SendData.empty()) start_write(ctx);
		else start_read(ctx);
		}
		}
		catch (NetException& e) {
		std::cerr << "[IOCP] NetException fd=" << c.csock->fd() << ": " << e.what() << "\n";
		if (e.getErrorType() != NetException::Note) {
		eargs->event->DisconnectEvent(c, tid, (ULONG_PTR)eargs->args);
		delete ctx;
		@@ -434,7 +186,6 @@ namespace netplus {
		}
		}
		};

		void eventapi::CreateConnection(std::shared_ptr<con>& res) {
		res = std::make_shared<con>(this);
		}

src/posix/udp.cpp

+1 −2

Original line number	Diff line number	Diff line
		@@ -265,7 +265,6 @@ void netplus::udp::connect(std::unique_ptr<socket> &csock){
		}
		}


		void netplus::udp::getAddress(std::string &addr){
		char buf[INET6_ADDRSTRLEN]={0};

src/socket.h

+328 −278

Original line number	Diff line number	Diff line
		@@ -50,6 +50,8 @@ typedef unsigned long ULONG_PTR;
		namespace netplus {
		enum sockettype { TCP = 0, UDP = 1, SSL = 2 };
		#ifdef Windows
		enum IO_OPERATION { OP_READ = 0, OP_WRITE = 1 };

		struct AcceptContext {
		WSAOVERLAPPED ov{};
		SOCKET acceptSock = INVALID_SOCKET;
		@@ -61,7 +63,50 @@ namespace netplus {

		alignas(void*) char addrBuf[ADDR_BUF_SZ]{};
		};

		class buffer {
		public:
		buffer(size_t s = BLOCKSIZE) {
		size = s;
		// Initialisiere die WSABUF Struktur
		data.buf = new char[size];
		data.len = (ULONG)size;

		// IOCP Metadaten nullen
		std::memset(&overlapped, 0, sizeof(WSAOVERLAPPED));
		operation = OP_READ;
		}

		buffer(const char *pointer,size_t s) {
		size = s;
		// Initialisiere die WSABUF Struktur
		data.buf = new char[size];
		data.len = (ULONG)size;

		std::copy(pointer, pointer + size, data.buf);

		// IOCP Metadaten nullen
		std::memset(&overlapped, 0, sizeof(WSAOVERLAPPED));
		operation = OP_READ;
		}

		~buffer() {
		delete[] data.buf;
		}

		bool ptr=false;

		size_t size;

		WSAOVERLAPPED overlapped;
		WSABUF data;
		IO_OPERATION operation;

		buffer(const buffer&) = delete;
		buffer& operator=(const buffer&) = delete;
		};
		#endif
		#ifndef Windows
		class buffer {
		public:
		buffer(size_t bsize) {
		@@ -88,8 +133,8 @@ namespace netplus {
		char* buf;
		const char* ptr;
		} data;

		};
		#endif

		class socket {
		public:
		@@ -107,8 +152,9 @@ namespace netplus {
		virtual void accept(std::unique_ptr<socket>& csock) = 0;
		#ifdef Windows
		virtual void accept(LPFN_ACCEPTEX lpfnAcceptEx, std::unique_ptr<socket>& csock) = 0;
		virtual size_t sendDataWSA(buffer& data, WSAOVERLAPPED* pOv, int flags)=0;
		virtual size_t recvDataWSA(buffer& data, WSAOVERLAPPED* pOv, int flags)=0;
		virtual size_t sendDataWSA(buffer& data, int flags) = 0;
		virtual size_t recvDataWSA(buffer& data, int flags) = 0;
		virtual void prime_read(buffer& data) = 0;
		#endif
		virtual void bind() = 0;
		virtual void listen() = 0;
		@@ -160,8 +206,9 @@ namespace netplus {
		~tcp();
		#ifdef Windows
		void accept(LPFN_ACCEPTEX lpfnAcceptEx, std::unique_ptr<socket>& csock);
		size_t sendDataWSA(buffer& data, WSAOVERLAPPED* pOv, int flags);
		size_t recvDataWSA(buffer& data, WSAOVERLAPPED* pOv, int flags);
		size_t sendDataWSA(buffer& data, int flags);
		size_t recvDataWSA(buffer& data, int flags);
		void prime_read(buffer& data);
		#endif
		void accept(std::unique_ptr <socket>& csock);
		void bind();
		@@ -196,8 +243,9 @@ namespace netplus {
		void accept(std::unique_ptr<socket>& csock);
		#ifdef Windows
		void accept(LPFN_ACCEPTEX lpfnAcceptEx, std::unique_ptr<socket>& csock);
		size_t sendDataWSA(buffer& data, WSAOVERLAPPED* pOv, int flags);
		size_t recvDataWSA(buffer& data, WSAOVERLAPPED* pOv, int flags);
		size_t sendDataWSA(buffer& data, int flags);
		size_t recvDataWSA(buffer& data, int flags);
		void prime_read(buffer& data);
		#endif
		void bind();
		void listen();
		@@ -235,10 +283,12 @@ namespace netplus {

		size_t sendData(buffer& data, int flags = 0) override;
		size_t recvData(buffer& data, int flags = 0) override;

		#ifdef Windows
		void accept(LPFN_ACCEPTEX lpfnAcceptEx, std::unique_ptr<socket>& csock) override;
		size_t sendDataWSA(buffer& data, WSAOVERLAPPED* pOv, int flags);
		size_t recvDataWSA(buffer& data, WSAOVERLAPPED* pOv, int flags);
		size_t sendDataWSA(buffer& data, int flags);
		size_t recvDataWSA(buffer& data, int flags);
		void prime_read(buffer& data);
		bool handshakeStepIOCP();
		#endif
		bool loadServerPrivateKeyDer(const std::string& keyDerPath);

src/ssl.cpp

+41 −27

File changed.

Preview size limit exceeded, changes collapsed.