terst (151494e1) · Commits · jan.koester / libnetplus

src/event/iocp.cpp

+144 −77

Original line number	Diff line number	Diff line
		@@ -166,43 +166,41 @@ namespace netplus {

		class EventWorker {
		public:
		// Post RAW recv on socket (TCP or SSL-underlying-TCP) into ctx->readCtx.buffer
		static void start_read(client* ctx) {
		con& c = *ctx->CurCon;
		// Point the Overlapped buffer to our persistent context memory
		buffer buf(ctx->readCtx.buffer, BLOCKSIZE);

		// Debug: report read post
		if (c.csock) {
		std::cerr << "[IOCP] start_read posting WSARecv for fd=" << c.csock->fd() << std::endl;
		} else {
		std::cerr << "[IOCP] start_read: csock == nullptr" << std::endl;
		if (!c.csock) {
		std::cerr << "[IOCP] start_read: csock == nullptr\n";
		return;
		}

		// This triggers the raw WSARecv. For SSL, it fills the internal _rx_netbuf.
		std::cerr << "[IOCP] start_read posting WSARecv for fd=" << c.csock->fd() << "\n";

		// IMPORTANT:
		// For SSL we must receive ciphertext using the underlying TCP recvDataWSA.
		// ssl::recvDataWSA() is a decoder and (in your version) returns 0 until handshakeDone.
		if (c.csock->_Type == sockettype::SSL) {
		static_cast<ssl*>(c.csock.get())->recvDataWSA(buf, &ctx->readCtx.overlapped, 0);
		static_cast<ssl*>(c.csock.get())->tcp::recvDataWSA(buf, &ctx->readCtx.overlapped, 0);
		}
		else {
		static_cast<tcp*>(c.csock.get())->recvDataWSA(buf, &ctx->readCtx.overlapped, 0);
		}
		}

		static void start_write(client* ctx) {
		// Send plaintext application data (TCP) or encrypt+send (SSL) using ctx->writeCtx
		static void start_write_app(client* ctx) {
		con& c = *ctx->CurCon;
		if (!c.csock) return;
		if (c.SendData.empty()) return;

		// Debug: report write post
		if (c.csock) {
		std::cerr << "[IOCP] start_write posting WSASend for fd=" << c.csock->fd() << " bytes=" << c.SendData.size() << std::endl;
		}
		std::cerr << "[IOCP] start_write_app WSASend for fd=" << c.csock->fd()
		<< " bytes=" << c.SendData.size() << "\n";

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

		// Ensure the socket's internal overlapped is NOT used implicitly;
		// EventWorker will pass the client's persistent write overlapped to the socket.
		size_t consumed = 0;
		if (c.csock->_Type == sockettype::SSL) {
		consumed = static_cast<ssl*>(c.csock.get())->sendDataWSA(out, &ctx->writeCtx.overlapped, 0);
		@@ -211,25 +209,46 @@ namespace netplus {
		consumed = static_cast<tcp*>(c.csock.get())->sendDataWSA(out, &ctx->writeCtx.overlapped, 0);
		}

		// Remove the plaintext from the queue that was actually "consumed" into a TLS record
		if (consumed > 0)
		// Remove plaintext that was consumed into a TLS record (SSL) or into the send buffer (TCP).
		if (consumed > 0) {
		c.SendData.erase(c.SendData.begin(), c.SendData.begin() + consumed);
		}
		}

		// Send queued handshake bytes (_hs_tx) RAW over TCP using ctx->writeCtx
		static void start_write_handshake(client* ctx, ssl* s) {
		if (!s) return;
		if (s->_hs_tx.empty() \|\| s->_hs_tx_off >= s->_hs_tx.size()) return;

		const size_t left = s->_hs_tx.size() - s->_hs_tx_off;

		std::cerr << "[IOCP] start_write_handshake WSASend fd=" << ctx->CurCon->csock->fd()
		<< " hs_left=" << left << "\n";

		buffer out((const char*)s->_hs_tx.data() + s->_hs_tx_off, left);
		s->tcp::sendDataWSA(out, &ctx->writeCtx.overlapped, 0);
		}

		EventWorker(int tid, ULONG_PTR args, EventWorkerArgs* eargs) {
		EventWorker(int tid, ULONG_PTR /args/, EventWorkerArgs* eargs) {
		while (event::Running) {
		DWORD dwBytesTransfered = 0;
		ULONG_PTR lpContext = 0;
		OVERLAPPED* pOverlapped = nullptr;

		BOOL bReturn = GetQueuedCompletionStatus(eargs->eviocp, &dwBytesTransfered,
		&lpContext, &pOverlapped, eargs->timeout);
		BOOL bReturn = GetQueuedCompletionStatus(
		eargs->eviocp,
		&dwBytesTransfered,
		&lpContext,
		&pOverlapped,
		eargs->timeout
		);

		if (!pOverlapped) continue;

		// Listener/AcceptEx completion path
		// --- Listener / AcceptEx completion ---
		if (lpContext == eargs->listenerKey) {
		std::unique_ptr<socket> accepted;

		{
		std::lock_guard<std::mutex> lk(ACCEPT_MTX);
		auto it = ACCEPT_PENDING.find(pOverlapped);
		@@ -240,14 +259,14 @@ namespace netplus {
		}

		if (!accepted) {
		// Unknown overlapped - ignore
		std::cerr << "[IOCP] AcceptEx completion: no pending entry for overlapped=" << pOverlapped << std::endl;
		std::cerr << "[IOCP] AcceptEx completion: no pending entry for overlapped=" << pOverlapped << "\n";
		continue;
		}

		// Update accept context
		SOCKET accSock = (SOCKET)accepted->fd();
		std::cerr << "[IOCP] AcceptEx completion: accepted fd=" << accSock << " overlapped=" << pOverlapped << std::endl;
		std::cerr << "[IOCP] AcceptEx completion: accepted fd=" << accSock
		<< " overlapped=" << pOverlapped << "\n";

		setsockopt(accSock, SOL_SOCKET, SO_UPDATE_ACCEPT_CONTEXT,
		(char*)&eargs->listenSock, sizeof(eargs->listenSock));

		@@ -255,53 +274,51 @@ namespace netplus {
		client* pClient = new client(eargs->event);
		pClient->CurCon->csock = std::move(accepted);

		// If SSL: run blocking handshake now (no IOCP recv posted yet)
		if (pClient->CurCon->csock->_Type == sockettype::SSL) {
		SOCKET s = (SOCKET)pClient->CurCon->csock->fd();
		u_long nb = 0;
		ioctlsocket(s, FIONBIO, &nb); // blocking during handshake
		static_cast<ssl*>(pClient->CurCon->csock.get())->handshake_after_accept();
		nb = 1;
		ioctlsocket(s, FIONBIO, &nb);
		}

		// Associate accepted socket with IOCP
		HANDLE h = CreateIoCompletionPort((HANDLE)(uintptr_t)pClient->CurCon->csock->fd(),
		eargs->eviocp, (ULONG_PTR)pClient, 0);
		HANDLE h = CreateIoCompletionPort(
		(HANDLE)(uintptr_t)pClient->CurCon->csock->fd(),
		eargs->eviocp,
		(ULONG_PTR)pClient,
		0
		);

		if (!h) {
		std::cerr << "[IOCP] CreateIoCompletionPort associate accepted failed for fd=" << pClient->CurCon->csock->fd() << std::endl;
		std::cerr << "[IOCP] CreateIoCompletionPort associate accepted failed for fd="
		<< pClient->CurCon->csock->fd() << "\n";
		eargs->event->DisconnectEvent(*pClient->CurCon, tid, (ULONG_PTR)eargs->args);
		delete pClient;
		}
		else {
		std::cerr << "[IOCP] calling ConnectEvent for fd=" << pClient->CurCon->csock->fd() << std::endl;
		std::cerr << "[IOCP] calling ConnectEvent for fd=" << pClient->CurCon->csock->fd() << "\n";
		eargs->event->ConnectEvent(*pClient->CurCon, tid, (ULONG_PTR)eargs->args);
		start_read(pClient);
		}

		// Re-post another accept
		// Re-post another AcceptEx
		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() << std::endl;
		std::cerr << "[IOCP] Re-posted AcceptEx; pending_count=" << ACCEPT_PENDING.size() << "\n";
		}
		}
		catch (NetException& e) {
		std::cerr << "AcceptEx repost error: " << e.what() << std::endl;
		std::cerr << "AcceptEx repost error: " << e.what() << "\n";
		}

		continue;
		}

		// Normal client completion path
		// --- Normal client completion ---
		client* pClientContext = reinterpret_cast<client*>(lpContext);
		IO_CONTEXT* pIoCtx = CONTAINING_RECORD(pOverlapped, IO_CONTEXT, overlapped);
		con& c = *pClientContext->CurCon;

		if (!bReturn \|\| (bReturn && dwBytesTransfered == 0)) {
		std::cerr << "[IOCP] connection closed or error on fd=" << c.csock->fd() << " bReturn=" << bReturn << " bytes=" << dwBytesTransfered << std::endl;
		if (!bReturn \|\| dwBytesTransfered == 0) {
		std::cerr << "[IOCP] connection closed/error fd=" << (c.csock ? c.csock->fd() : -1)
		<< " bReturn=" << bReturn << " bytes=" << dwBytesTransfered << "\n";
		eargs->event->DisconnectEvent(c, tid, (ULONG_PTR)eargs->args);
		delete pClientContext;
		continue;
		@@ -311,63 +328,103 @@ namespace netplus {
		std::lock_guard<std::mutex> guard(pClientContext->cltmtx);

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

		// 1. Move raw ciphertext from IOCP buffer to SSL's rx_netbuf
		sslSocket->_rx_netbuf.insert(sslSocket->_rx_netbuf.end(),
		if (c.csock && c.csock->_Type == sockettype::SSL) {
		ssl* s = static_cast<ssl*>(c.csock.get());

		// 0) FIRST: append ciphertext bytes into SSL net buffer
		s->_rx_netbuf.insert(s->_rx_netbuf.end(),
		pIoCtx->buffer,
		pIoCtx->buffer + dwBytesTransfered);

		// 2. Process all complete TLS records currently in the buffer
		// 1) Handshake not done -> drive handshake state machine
		if (!s->_handshakeDone) {
		while (s->handshakeStepIOCP()) {
		// keep stepping while full records exist and state advances
		}

		// if handshake produced outbound bytes, send them raw now
		if (!s->_hs_tx.empty() && s->_hs_tx_off < s->_hs_tx.size()) {
		start_write_handshake(pClientContext, s);
		}

		// keep receiving more ciphertext
		start_read(pClientContext);
		continue;
		}

		// 2) Handshake done -> decrypt appdata from buffered ciphertext
		buffer plain(BLOCKSIZE);
		size_t decrypted = 0;
		while ((decrypted = sslSocket->recvDataWSA(plain, nullptr, 0)) > 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() << " appending " << decrypted << " bytes total_recv=" << c.RecvData.size() << std::endl;
		std::cerr << "[IOCP] RequestEvent (SSL) fd=" << c.csock->fd()
		<< " +=" << decrypted << " total=" << c.RecvData.size() << "\n";
		eargs->event->RequestEvent(c, tid, (ULONG_PTR)eargs->args);
		}

		}
		else {
		// Plain TCP
		c.RecvData.append(pIoCtx->buffer, dwBytesTransfered);
		std::cerr << "[IOCP] RequestEvent (TCP) fd=" << c.csock->fd() << " appended bytes=" << dwBytesTransfered << " total_recv=" << c.RecvData.size() << std::endl;
		std::cerr << "[IOCP] RequestEvent (TCP) fd=" << c.csock->fd()
		<< " +=" << dwBytesTransfered << " total=" << c.RecvData.size() << "\n";
		eargs->event->RequestEvent(c, tid, (ULONG_PTR)eargs->args);
		}

		// Continue the loop
		if (!c.SendData.empty()) start_write(pClientContext);
		// Continue loop
		if (!c.SendData.empty()) start_write_app(pClientContext);
		else start_read(pClientContext);

		}

		else if (pIoCtx->operation == OP_WRITE) {
		// For SSL: dwBytesTransfered is the size of the ENCRYPTED record sent.
		// For TCP: It is the size of the plaintext sent.

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

		// --- handshake flight sending completion ---
		if (!s->_handshakeDone && !s->_hs_tx.empty()) {
		s->_hs_tx_off += dwBytesTransfered;

		// If the full TLS record is gone, we can send the next one
		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
		if (s->_hs_tx_off >= s->_hs_tx.size()) {
		// finished sending the queued handshake bytes
		s->_hs_tx.clear();
		s->_hs_tx_off = 0;
		}
		else {
		// still more handshake bytes to send
		start_write_handshake(pClientContext, s);
		continue;
		}

		// After handshake bytes were sent, keep receiving (client will respond with next flight)
		start_read(pClientContext);
		continue;
		}

		// --- normal SSL record sending completion ---
		s->_send_off += dwBytesTransfered;

		if (!s->_send_record.empty() && s->_send_off >= s->_send_record.size()) {
		s->_send_record.clear();
		s->_send_off = 0;
		}
		}

		// For TCP, the plaintext queue was already removed in start_write_app by "consumed".
		if (c.SendData.empty()) {
		std::cerr << "[IOCP] ResponseEvent fd=" << c.csock->fd() << " send_queue_empty" << std::endl;
		std::cerr << "[IOCP] ResponseEvent fd=" << c.csock->fd() << " send_queue_empty\n";
		eargs->event->ResponseEvent(c, tid, (ULONG_PTR)eargs->args);
		}

		if (!c.SendData.empty()) start_write(pClientContext);
		if (!c.SendData.empty()) start_write_app(pClientContext);
		else start_read(pClientContext);
		}
		}
		catch (NetException& e) {
		if (e.getErrorType() != NetException::Note) {
		std::cerr << "[IOCP] NetException during IO processing for fd=" << c.csock->fd() << ": " << e.what() << std::endl;
		std::cerr << "[IOCP] NetException fd=" << (c.csock ? c.csock->fd() : -1)
		<< ": " << e.what() << "\n";
		eargs->event->DisconnectEvent(c, tid, (ULONG_PTR)eargs->args);
		delete pClientContext;
		}
		@@ -491,8 +548,18 @@ namespace netplus {

		std::cerr << "[IOCP] accept() returned, client fd=" << pClient->CurCon->csock->fd() << std::endl;

		HANDLE h = CreateIoCompletionPort((HANDLE)(uintptr_t)pClient->CurCon->csock->fd(), iocp, (ULONG_PTR)pClient, 0);
		SOCKET s = INVALID_SOCKET;

		if (pClient->CurCon->csock->_Type == sockettype::SSL) {
		s = (SOCKET)static_cast<netplus::ssl*>(pClient->CurCon->csock.get())->_Socket;
		}
		else {
		s = (SOCKET)static_cast<netplus::tcp*>(pClient->CurCon->csock.get())->_Socket;
		}

		HANDLE h = CreateIoCompletionPort((HANDLE)s, iocp, (ULONG_PTR)pClient, 0);
		if (!h) {
		std::cerr << "[IOCP] CreateIoCompletionPort failed err=" << GetLastError() << "\n";
		delete pClient;
		continue;
		}

src/socket.h

+18 −0

Original line number	Diff line number	Diff line
		@@ -239,6 +239,7 @@ namespace netplus {
		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);
		bool handshakeStepIOCP();
		#endif
		bool loadServerPrivateKeyDer(const std::string& keyDerPath);

		@@ -275,6 +276,23 @@ namespace netplus {
		#ifdef Windows
		AcceptContext _acpt;
		std::array<char, 2 * (sizeof(SOCKADDR_STORAGE) + 16)> _acceptBuf{};

		std::vector<uint8_t> _hs_tx;
		size_t _hs_tx_off = 0;

		enum class HsState {
		WAIT_CH,
		SEND_SERVER_FLIGHT,
		WAIT_CKE,
		WAIT_CCS,
		WAIT_FIN,
		SEND_CCS_FIN,
		DONE,
		FAIL
		};

		HsState _hs_state = HsState::WAIT_CH;
		std::vector<uint8_t> _masterSecret;
		#endif
		size_t _recv_off = 0;

src/ssl.cpp

+349 −0

File changed.

Preview size limit exceeded, changes collapsed.