test (1758d5bc) · Commits · jan.koester / libnetplus

src/event/iocp.cpp

+112 −22

Original line number	Diff line number	Diff line
		@@ -80,6 +80,8 @@ namespace netplus {
		std::mutex cltmtx;
		IO_CONTEXT readCtx;
		IO_CONTEXT writeCtx;
		std::atomic<bool> recvPending{false};
		std::atomic<bool> sendPending{false};
		};

		class EventWorkerArgs {
		@@ -94,39 +96,116 @@ namespace netplus {
		class EventWorker {
		public:
		static void start_read(client* ctx) {
		// Ensure only one outstanding recv per connection
		if (ctx->recvPending.exchange(true)) return;

		con& c = *ctx->CurCon;
		// Point the Overlapped buffer to our persistent context memory
		buffer buf(ctx->readCtx.buffer, BLOCKSIZE);

		// This triggers the raw WSARecv. For SSL, it fills the internal _rx_netbuf.
		if (c.csock->_Type == sockettype::SSL) {
		static_cast<ssl*>(c.csock.get())->recvDataWSA(buf, 0);
		} else {
		static_cast<tcp*>(c.csock.get())->recvDataWSA(buf, 0);
		// Reset OVERLAPPED each time before (re)posting
		std::memset(&ctx->readCtx.overlapped, 0, sizeof(WSAOVERLAPPED));
		ctx->readCtx.operation = OP_READ;
		ctx->readCtx.wsaBuf.buf = ctx->readCtx.buffer;
		ctx->readCtx.wsaBuf.len = BLOCKSIZE;

		DWORD flags = 0;
		DWORD bytes = 0;

		const SOCKET s = (SOCKET)c.csock->fd();
		const int rc = WSARecv(s, &ctx->readCtx.wsaBuf, 1, &bytes, &flags, &ctx->readCtx.overlapped, nullptr);
		if (rc == SOCKET_ERROR) {
		const int err = WSAGetLastError();
		if (err != WSA_IO_PENDING) {
		ctx->recvPending.store(false);
		NetException e;
		e[NetException::Error] << "start_read: WSARecv failed: " << err;
		throw e;
		}
		}
		}

		static void start_write(client* ctx) {
		con& c = *ctx->CurCon;
		if (c.SendData.empty()) return;
		// Ensure only one outstanding send per connection
		if (ctx->sendPending.exchange(true)) return;

		// Take plaintext from SendData and pass it to the WSA method
		// For SSL, this will perform the encryption before calling WSASend
		size_t toSend = (std::min)((size_t)BLOCKSIZE, c.SendData.size());
		buffer out(c.SendData.data(), toSend);
		con& c = *ctx->CurCon;

		size_t consumed = 0;
		// If SSL still has a partially-sent record, continue sending it.
		if (c.csock->_Type == sockettype::SSL) {
		consumed = static_cast<ssl*>(c.csock.get())->sendDataWSA(out, 0);
		} else {
		consumed = static_cast<tcp*>(c.csock.get())->sendDataWSA(out, 0);
		ssl* sslSocket = static_cast<ssl*>(c.csock.get());

		// If no pending encrypted record, build one from queued plaintext.
		if (sslSocket->_send_record.empty() \|\| sslSocket->_send_off >= sslSocket->_send_record.size()) {
		sslSocket->_send_record.clear();
		sslSocket->_send_off = 0;

		if (c.SendData.empty()) {
		ctx->sendPending.store(false);
		return;
		}

		// Remove the plaintext from the queue that was actually "consumed" into a TLS record
		if (consumed > 0)
		const size_t toSend = (std::min)((size_t)BLOCKSIZE, c.SendData.size());
		buffer plain(c.SendData.data(), toSend);

		// Builds sslSocket->_send_record and returns plaintext bytes consumed
		const size_t consumed = sslSocket->sendDataWSA(plain, 0);
		if (consumed == 0) {
		ctx->sendPending.store(false);
		return;
		}
		c.SendData.erase(c.SendData.begin(), c.SendData.begin() + consumed);
		}

		const size_t remaining = sslSocket->_send_record.size() - sslSocket->_send_off;
		const size_t chunk = (std::min)((size_t)BLOCKSIZE, remaining);

		std::memset(&ctx->writeCtx.overlapped, 0, sizeof(WSAOVERLAPPED));
		ctx->writeCtx.operation = OP_WRITE;
		ctx->writeCtx.wsaBuf.buf = (CHAR*)sslSocket->_send_record.data() + sslSocket->_send_off;
		ctx->writeCtx.wsaBuf.len = (ULONG)chunk;

		DWORD bytes = 0;
		const SOCKET s = (SOCKET)c.csock->fd();
		const int rc = WSASend(s, &ctx->writeCtx.wsaBuf, 1, &bytes, 0, &ctx->writeCtx.overlapped, nullptr);
		if (rc == SOCKET_ERROR) {
		const int err = WSAGetLastError();
		if (err != WSA_IO_PENDING) {
		ctx->sendPending.store(false);
		NetException e;
		e[NetException::Error] << "start_write(ssl): WSASend failed: " << err;
		throw e;
		}
		}
		return;
		}

		// Plain TCP send
		if (c.SendData.empty()) {
		ctx->sendPending.store(false);
		return;
		}

		const size_t toSend = (std::min)((size_t)BLOCKSIZE, c.SendData.size());

		std::memset(&ctx->writeCtx.overlapped, 0, sizeof(WSAOVERLAPPED));
		ctx->writeCtx.operation = OP_WRITE;
		ctx->writeCtx.wsaBuf.buf = (CHAR*)c.SendData.data();
		ctx->writeCtx.wsaBuf.len = (ULONG)toSend;

		DWORD bytes = 0;
		const SOCKET s = (SOCKET)c.csock->fd();
		const int rc = WSASend(s, &ctx->writeCtx.wsaBuf, 1, &bytes, 0, &ctx->writeCtx.overlapped, nullptr);
		if (rc == SOCKET_ERROR) {
		const int err = WSAGetLastError();
		if (err != WSA_IO_PENDING) {
		ctx->sendPending.store(false);
		NetException e;
		e[NetException::Error] << "start_write(tcp): WSASend failed: " << err;
		throw e;
		}
		}
		}


		EventWorker(int tid, ULONG_PTR args, EventWorkerArgs* eargs) {
		while (event::Running) {
		DWORD dwBytesTransfered = 0;
		@@ -152,6 +231,8 @@ namespace netplus {
		std::lock_guard<std::mutex> guard(pClientContext->cltmtx);

		if (pIoCtx->operation == OP_READ) {
		pClientContext->recvPending.store(false);

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

		@@ -178,6 +259,8 @@ namespace netplus {
		else start_read(pClientContext);

		} else if (pIoCtx->operation == OP_WRITE) {
		pClientContext->sendPending.store(false);

		// For SSL: dwBytesTransfered is the size of the ENCRYPTED record sent.
		// For TCP: It is the size of the plaintext sent.

		@@ -189,7 +272,14 @@ namespace netplus {
		if (sslSocket->_send_off >= sslSocket->_send_record.size()) {
		sslSocket->_send_record.clear();
		sslSocket->_send_off = 0;
		// Note: We increment seq in sendDataWSA or here depending on logic
		sslSocket->_send_seq++; // advance TLS sequence after full record sent
		}
		} else {
		// Plain TCP: remove actually sent plaintext from queue
		if (dwBytesTransfered > 0 && dwBytesTransfered <= c.SendData.size()) {
		c.SendData.erase(c.SendData.begin(), c.SendData.begin() + dwBytesTransfered);
		} else if (dwBytesTransfered > c.SendData.size()) {
		c.SendData.clear();
		}
		}

src/ssl.cpp

+10 −3

Original line number	Diff line number	Diff line
		@@ -1777,9 +1777,16 @@ bool loadServerPrivateKeyDer(const std::string& keyDerPath);

		#ifdef Windows
		size_t netplus::ssl::sendDataWSA(buffer& data, int flags) {
		// 1) During handshake, we send raw TCP (usually for ServerHello/Certificates)
		// 1) During handshake we may still need to write bytes, but under IOCP we
		// must NOT call tcp::sendDataWSA here (it would post its own OVERLAPPED).
		// Instead, queue raw bytes into _send_record and let the IOCP layer WSASend it.
		if (!_handshakeDone) {
		return tcp::sendDataWSA(data, flags);
		if (!_send_record.empty() && _send_off < _send_record.size()) return 0;
		if (data.size == 0) return 0;
		const size_t take = (std::min)((size_t)data.size, (size_t)16384);
		_send_record.assign((const uint8_t)data.data.buf, (const uint8_t)data.data.buf + take);
		_send_off = 0;
		return take;
		}

		auto throwSSL = [&](int etype, const std::string& msg) -> void {
		@@ -1855,7 +1862,7 @@ size_t netplus::ssl::sendDataWSA(buffer& data, int flags) {
		buffer out((const char*)_send_record.data(), _send_record.size());

		// This call triggers WSASend with the Overlapped structure.
		tcp::sendDataWSA(out, flags);
		// NOTE: Do not send here. IOCP EventWorker will WSASend _send_record.

		// Return 'take' to inform the application how many plaintext bytes were consumed
		return take;