test (631e821d) · Commits · jan.koester / libnetplus

.gitignore

+3 −1

Original line number	Diff line number	Diff line
		@@ -6,3 +6,5 @@ build/
		.kdevelop/
		.cache/
		.vscode/
		test/
		build_ninja/
		No newline at end of file

src/event/iocp.cpp

+153 −44

Original line number	Diff line number	Diff line
		/*******************************************************************************
		/*******************************************************************************
		Copyright (c) 2014, Jan Koester jan.koester@gmx.net
		All rights reserved.

		@@ -193,6 +193,10 @@ namespace netplus {

		static AcceptCtx* pop_accept(EventState* st, OVERLAPPED* pov) {
		std::lock_guard<std::mutex> lk(st->accMutex);
		std::cerr << "[IOCP] pop_accept: looking for pov=" << pov << " in " << st->acceptByOv.size() << " entries" << std::endl;
		for (auto& kv : st->acceptByOv) {
		std::cerr << "[IOCP] stored ov=" << kv.first << std::endl;
		}
		auto it = st->acceptByOv.find(pov);
		if (it == st->acceptByOv.end()) return nullptr;
		AcceptCtx* ctx = it->second;
		@@ -216,8 +220,20 @@ namespace netplus {
		}

		static void remove_con(EventState* st, SOCKET s) {
		con* cptr = nullptr;
		{
		std::lock_guard<std::mutex> lk(st->conMutex);
		st->sockToCon.erase(s);
		auto it = st->sockToCon.find(s);
		if (it != st->sockToCon.end()) {
		cptr = it->second.get();
		st->sockToCon.erase(it);
		}
		}
		// Also remove the recv buffer for this connection
		if (cptr) {
		std::lock_guard<std::mutex> lk(st->bufMutex);
		st->recvBuf.erase(cptr);
		}
		}

		// -------------------------------------------------------------------------
		@@ -253,18 +269,43 @@ namespace netplus {

		// -------------------------------------------------------------------------
		// Post send (only if not already pending)
		// Returns true if data was sent (for ResponseEvent purposes)
		// -------------------------------------------------------------------------
		static void try_post_send(EventState* /st/, con& c) {
		if (c.WritePending.load()) return;
		static bool try_post_send(event* ev, EventState* /st/, con& c, int tid) {
		if (c.WritePending.load()) return false;

		const size_t avail = c.SendData.size();
		if (avail <= c.SendOff) return;
		if (avail <= c.SendOff) return false;

		const size_t remaining = avail - c.SendOff;
		const char* ptr = c.SendData.data() + c.SendOff;

		// Create buffer that references SendData memory
		// buffer(pointer,len) exists in socket.h and sets WSABUF to that memory
		bool isSSL = (c.csock->getSocketType() == (int)sockettype::SSL);

		if (isSSL) {
		// For SSL: use synchronous encrypted send via sendData
		// The ssl::sendData encrypts and sends via flush_out
		c.WritePending.store(true);
		try {
		buffer sbuf(ptr, remaining);
		size_t sent = c.csock->sendData(sbuf, 0);
		// All data sent for SSL (it encrypts and sends in one go)
		c.SendData.clear();
		c.SendOff = 0;
		c.WritePending.store(false);
		// Fire ResponseEvent for SSL since we completed synchronously
		ev->ResponseEvent(c, tid, (ULONG_PTR)sent);
		return true;
		} catch (NetException& e) {
		c.WritePending.store(false);
		if (e.getErrorType() != NetException::Note) {
		throw;
		}
		// Note = would block, will retry later
		return false;
		}
		} else {
		// For plain TCP: use async WSASend
		auto* sbuf = new buffer(ptr, remaining);
		sbuf->operation = OP_WRITE;
		sbuf->owner = &c;
		@@ -274,12 +315,14 @@ namespace netplus {

		try {
		c.csock->sendDataWSA(*sbuf, 0);
		return false; // Async - completion will come later
		} catch (...) {
		c.WritePending.store(false);
		delete sbuf;
		throw;
		}
		}
		}

		// -------------------------------------------------------------------------
		// Worker loop
		@@ -288,19 +331,23 @@ namespace netplus {
		EventState* st = ST(ev);

		while (event::Running) {
		std::cerr << "[IOCP] Worker " << tid << " waiting..." << std::endl;

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

		BOOL ok = GetQueuedCompletionStatus(st->iocp, &bytes, &key, &pov, INFINITE);

		DWORD lastErr = GetLastError();
		std::cerr << "[IOCP] Worker " << tid << " got completion: ok=" << ok << " bytes=" << bytes << " key=" << key << " pov=" << pov << " err=" << lastErr << std::endl;
		if (!event::Running) break;

		if (!pov && key == 0) continue; // wake packet

		// 1) Accept completion?
		std::cerr << "[IOCP] Checking if AcceptEx completion, pov=" << pov << std::endl;
		if (AcceptCtx* actx = pop_accept(st, pov)) {
		std::cerr << "[IOCP] AcceptEx completion detected!" << std::endl;

		SOCKET accepted = actx->acceptSock;

		@@ -331,7 +378,7 @@ namespace netplus {
		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(), accepted);
		c->csock = std::make_unique<ssl>(sslServer->getCert(), (int)accepted);
		}
		else
		continue;
		@@ -350,6 +397,7 @@ namespace netplus {

		// register mapping
		register_con(st, accepted, c);
		std::cerr << "[IOCP] Registered connection: socket=" << accepted << " csock->fd()=" << c->csock->fd() << std::endl;

		// connect callback
		ev->ConnectEvent(*c, tid, 0);
		@@ -372,22 +420,38 @@ namespace netplus {

		// 2) Normal I/O completion (recv/send)
		socket* sockObj = reinterpret_cast<socket*>(key);
		if (!sockObj) continue;
		if (!sockObj) {
		std::cerr << "[IOCP] sockObj is null, skipping" << std::endl;
		continue;
		}

		buffer* buf = reinterpret_cast<buffer*>(
		reinterpret_cast<char*>(pov) - offsetof(buffer, overlapped)
		);

		con* owner = reinterpret_cast<con*>(buf->owner);
		if (!owner) continue;

		SOCKET cs = (SOCKET)owner->csock->fd();
		// Look up connection by socket fd (key is the socket*)
		SOCKET cs = (SOCKET)sockObj->fd();
		std::cerr << "[IOCP] Looking up connection for socket fd=" << cs << std::endl;
		std::shared_ptr<con> conPtr = find_con(st, cs);
		if (!conPtr) {
		// 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
		}
		continue;
		}
		con* owner = conPtr.get();
		std::cerr << "[IOCP] Found connection, operation=" << (buf->operation == OP_READ ? "READ" : "WRITE") << " bytes=" << bytes << std::endl;

		if (!ok) {
		// error -> disconnect
		ev->DisconnectEvent(*owner, tid, 0);
		remove_con(st, cs);
		try { owner->csock->close(); } catch (...) {}
		if (buf->operation == OP_WRITE) {
		delete buf;
		}
		continue;
		}

		@@ -402,39 +466,85 @@ 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) {
		// Create buffer with received raw bytes for SSL processing
		buffer rawBuf(buf->data.buf, bytes);
		// 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);

		// Check if handshake is done
		if (!owner->csock->getHandshakeDone()) {
		std::cerr << "[IOCP] Starting handshake_after_accept..." << std::endl;
		std::cerr.flush();
		try {
		// recvData for SSL will process raw bytes from buffer
		owner->csock->recvData(rawBuf, 0);
		// Process handshake from internal buffer
		owner->csock->handshake_after_accept();
		std::cerr << "[IOCP] handshake_after_accept returned OK, hasPendingWrite=" << owner->csock->hasPendingWrite() << std::endl;
		std::cerr.flush();

		// If there's pending data to send (handshake response)
		if (owner->csock->hasPendingWrite()) {
		owner->csock->flush_out();
		}
		} catch (NetException& e) {
		std::cerr << "[IOCP] handshake_after_accept threw NetException type=" << e.getErrorType() << " msg=" << e.what() << std::endl;
		std::cerr.flush();
		if (e.getErrorType() == NetException::Note) {
		// Need more data - repost recv
		// Need more data - but first check if we have data to send
		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;
		}
		} 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;
		try {
		post_recv(st, *owner);
		} catch (...) {}
		continue;
		}
		// Real error - disconnect
		std::cerr << "[IOCP] Non-Note error - disconnecting" << std::endl;
		ev->DisconnectEvent(*owner, tid, 0);
		remove_con(st, cs);
		try { owner->csock->close(); } catch (...) {}
		continue;
		} catch (std::exception& e) {
		std::cerr << "[IOCP] handshake_after_accept threw std::exception: " << e.what() << std::endl;
		ev->DisconnectEvent(*owner, tid, 0);
		remove_con(st, cs);
		try { owner->csock->close(); } catch (...) {}
		continue;
		} catch (...) {
		std::cerr << "[IOCP] handshake_after_accept threw unknown exception" << std::endl;
		ev->DisconnectEvent(*owner, tid, 0);
		remove_con(st, cs);
		try { owner->csock->close(); } catch (...) {}
		continue;
		}

		// 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;
		}
		}

		// If handshake still not done, wait for more data
		if (!owner->csock->getHandshakeDone()) {
		std::cerr << "[IOCP] Handshake not done yet, reposting recv" << std::endl;
		try {
		post_recv(st, *owner);
		} catch (...) {}
		@@ -442,9 +552,9 @@ namespace netplus {
		}
		}

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

		if (decLen > 0) {
		@@ -468,7 +578,7 @@ namespace netplus {

		// if user filled SendData -> try send
		try {
		try_post_send(st, *owner);
		try_post_send(ev, st, *owner, tid);
		} catch (...) {
		ev->DisconnectEvent(*owner, tid, 0);
		remove_con(st, cs);
		@@ -511,7 +621,7 @@ namespace netplus {

		// continue sending if more data is pending
		try {
		try_post_send(st, *owner);
		try_post_send(ev, st, *owner, tid);
		} catch (...) {
		ev->DisconnectEvent(*owner, tid, 0);
		remove_con(st, cs);
		@@ -552,8 +662,6 @@ namespace netplus {
		EventState* st = ST(this);
		if (!st) return;

		st->threads = threads;

		// wake all workers
		for (size_t i = 0; i < st->workers.size(); ++i) {
		PostQueuedCompletionStatus(st->iocp, 0, 0, nullptr);
		@@ -582,6 +690,7 @@ namespace netplus {
		_pollFD = (int)(intptr_t)st->iocp;

		Running = true;
		std::cerr << "[IOCP] XXXXX Starting event loop, Running=" << Running.load() << " threads=" << st->threads << std::endl; std::cerr.flush();

		// Pre-post accepts (like "listen ready")
		// recommended: 2*threads AcceptEx in flight

src/socket.h

+7 −0

Original line number	Diff line number	Diff line
		@@ -331,6 +331,7 @@ namespace netplus {
		void sendDataWSA(buffer& data, int flags);
		void recvDataWSA(buffer& data, int flags);
		void prime_read(buffer& data);
		buffer* getIocpBuf() { return _iocpBuf; }
		#endif

		bool loadServerPrivateKeyDer(const std::string& keyDerPath);
		@@ -351,6 +352,11 @@ 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; }

		@@ -540,6 +546,7 @@ namespace netplus {
		std::array<char, 2 * (sizeof(SOCKADDR_STORAGE) + 16)> _acceptBuf{};
		std::vector<uint8_t> _hs_tx;
		size_t _hs_tx_off = 0;
		buffer* _iocpBuf = nullptr;
		#endif
		std::vector<uint8_t> _client_session_id;

src/ssl.cpp

+73 −23

Original line number	Diff line number	Diff line
		@@ -360,16 +360,22 @@ std::vector<uint8_t> netplus::ssl::readTlsRecordAsync()
		// ------------------------------------------------------------
		// 1) ensure we have at least TLS record header (5 bytes)
		// ------------------------------------------------------------
		#ifdef Windows
		// IOCP mode: data is pre-fed into _rx_tcp_buf via feedRawBytes()
		if (_rx_tcp_buf.size() < 5) {
		NetException n;
		n[NetException::Note] << "ssl: record incomplete (need header)";
		throw n;
		}
		#else
		while (_rx_tcp_buf.size() < 5) {
		buffer tmpBuf(4096);

		try {
		size_t r = tcp::recvData(tmpBuf, 0);
		if (r > 0) {
		_rx_tcp_buf.insert(_rx_tcp_buf.end(), tmpBuf.data.buf, tmpBuf.data.buf + r);
		continue;
		}
		// r == 0 means peer closed
		return {};
		} catch (NetException& e) {
		if (e.getErrorType() == NetException::Note) {
		@@ -380,9 +386,10 @@ std::vector<uint8_t> netplus::ssl::readTlsRecordAsync()
		throw;
		}
		}
		#endif

		// ------------------------------------------------------------
		// 2) parse record length (header already in buffer)
		// 2) parse record length
		// ------------------------------------------------------------
		uint16_t recLen = (uint16_t(_rx_tcp_buf[3]) << 8) \| uint16_t(_rx_tcp_buf[4]);
		size_t total = 5 + recLen;
		@@ -390,16 +397,21 @@ std::vector<uint8_t> netplus::ssl::readTlsRecordAsync()
		// ------------------------------------------------------------
		// 3) read until full record present
		// ------------------------------------------------------------
		#ifdef Windows
		if (_rx_tcp_buf.size() < total) {
		NetException n;
		n[NetException::Note] << "ssl: record incomplete";
		throw n;
		}
		#else
		while (_rx_tcp_buf.size() < total) {
		buffer tmpBuf(4096);

		try {
		size_t r = tcp::recvData(tmpBuf, 0);
		if (r > 0) {
		_rx_tcp_buf.insert(_rx_tcp_buf.end(), tmpBuf.data.buf, tmpBuf.data.buf + r);
		continue;
		}
		// peer closed mid-record
		return {};
		} catch (NetException& e) {
		if (e.getErrorType() == NetException::Note) {
		@@ -410,6 +422,7 @@ std::vector<uint8_t> netplus::ssl::readTlsRecordAsync()
		throw;
		}
		}
		#endif

		// ------------------------------------------------------------
		// 4) extract full record
		@@ -2441,7 +2454,7 @@ void netplus::ssl::handshake_after_accept(){
		flush_out();

		_hs_state = HsState::TLS13_SEND_ENCRYPTED_FLIGHT;
		continue;
		return;
		}

		default:
		@@ -2490,6 +2503,43 @@ void netplus::ssl::queueRaw(const uint8_t* p, size_t n) {
		}

		void netplus::ssl::flush_out(){
		#ifdef Windows
		// IOCP mode: collect all pending records into one buffer for async send
		std::vector<uint8_t> combined;

		// Add current record if any
		if (!_send_record.empty()) {
		combined.insert(combined.end(),
		_send_record.begin() + _send_off,
		_send_record.end());
		_send_record.clear();
		_send_off = 0;
		}

		// Add all queued records
		while (!_send_queue.empty()) {
		auto& rec = _send_queue.front();
		combined.insert(combined.end(), rec.begin(), rec.end());
		_send_queue.pop_front();
		}

		if (combined.empty()) {
		return;
		}

		// Clean up previous buffer if any
		if (_iocpBuf) {
		delete _iocpBuf;
		_iocpBuf = nullptr;
		}

		// Allocate new buffer with the combined data
		_iocpBuf = new buffer(reinterpret_cast<const char*>(combined.data()), combined.size());
		_iocpBuf->operation = OP_WRITE;

		// Post async send via WSASend
		tcp::sendDataWSA(*_iocpBuf, 0);
		#else
		// load next record if none active
		if (_send_record.empty() && !_send_queue.empty()) {
		_send_record = std::move(_send_queue.front());
		@@ -2536,6 +2586,7 @@ void netplus::ssl::flush_out(){
		}
		}
		}
		#endif
		}

		#ifdef Windows
		@@ -4151,9 +4202,11 @@ void netplus::ssl::sendDataWSA(buffer& data, int flags) {
		// The data in buffer is plaintext - encrypt it and queue for sending

		if (!_handshakeDone) {
		// During handshake, send raw TLS records from _hs_tx
		// (handshake data is already queued via queueRaw)
		} else {
		// During handshake, flush any pending TLS records
		flush_out();
		return;
		}

		// After handshake, encrypt application data from the buffer
		const char* p = data.data.buf;
		std::vector<uint8_t> plain(p, p + data.size);
		@@ -4165,13 +4218,10 @@ void netplus::ssl::sendDataWSA(buffer& data, int flags) {
		// TLS 1.2: Use CBC encryption
		_sendTLS12Record(0x17, plain);
		}
		}

		// Now post the actual send using tcp method
		// Flush the encrypted data - this will send the TLS records
		// using tcp::sendDataWSA internally
		flush_out();

		// Use tcp::sendDataWSA for the underlying send
		tcp::sendDataWSA(data, flags);
		}

		void netplus::ssl::recvDataWSA(buffer& data, int flags) {