test (1cc5c7a1) · Commits · jan.koester / libnetplus

src/event/iocp.cpp

+51 −206

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

		@@ -166,21 +166,10 @@ 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;
		buffer buf(ctx->readCtx.buffer, BLOCKSIZE);

		if (!c.csock) {
		std::cerr << "[IOCP] start_read: csock == nullptr\n";
		return;
		}

		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())->tcp::recvDataWSA(buf, &ctx->readCtx.overlapped, 0);
		}
		@@ -189,245 +178,101 @@ namespace netplus {
		}
		}

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

		std::cerr << "[IOCP] start_write_app WSASend for fd=" << c.csock->fd()
		<< " bytes=" << c.SendData.size() << "\n";

		size_t toSend = (std::min)((size_t)BLOCKSIZE, c.SendData.size());
		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);
		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);
		consumed = static_cast<tcp*>(c.csock.get())
		->sendDataWSA(out, &ctx->writeCtx.overlapped, 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);
		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
		);

		if (!pOverlapped) continue;

		// --- 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);
		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 overlapped=" << pOverlapped << "\n";
		continue;
		}

		SOCKET accSock = (SOCKET)accepted->fd();
		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));
		DWORD bytes = 0;
		ULONG_PTR key = 0;
		OVERLAPPED* ov = nullptr;

		// Build client connection
		client* pClient = new client(eargs->event);
		pClient->CurCon->csock = std::move(accepted);
		BOOL ok = GetQueuedCompletionStatus(
		eargs->eviocp, &bytes, &key, &ov, eargs->timeout);

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

		if (!h) {
		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() << "\n";
		eargs->event->ConnectEvent(*pClient->CurCon, tid, (ULONG_PTR)eargs->args);
		start_read(pClient);
		}

		// 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() << "\n";
		}
		}
		catch (NetException& e) {
		std::cerr << "AcceptEx repost error: " << e.what() << "\n";
		}

		continue;
		}

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

		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;
		if (!ok \|\| bytes == 0) {
		eargs->event->DisconnectEvent(c, tid, args);
		delete ctx;
		continue;
		}

		try {
		std::lock_guard<std::mutex> guard(pClientContext->cltmtx);
		std::lock_guard<std::mutex> lock(ctx->cltmtx);

		if (pIoCtx->operation == OP_READ) {
		if (io->operation == OP_READ) {

		if (c.csock && c.csock->_Type == sockettype::SSL) {
		if (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);
		// 🔴 FIX: append bytes FIRST
		s->_rx_netbuf.insert(
		s->_rx_netbuf.end(),
		io->buffer,
		io->buffer + bytes
		);

		// 1) Handshake not done -> drive handshake state machine
		// 🔴 THEN handshake
		if (!s->_handshakeDone) {
		while (s->handshakeStepIOCP()) {
		// keep stepping while full records exist and state advances
		}
		while (s->handshakeStepIOCP()) {}

		// 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);
		}
		if (!s->_hs_tx.empty())
		start_write(ctx);

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

		// 2) Handshake done -> decrypt appdata from buffered ciphertext
		// decrypt application data
		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";
		eargs->event->RequestEvent(c, tid, (ULONG_PTR)eargs->args);
		size_t n = 0;
		while ((n = s->recvDataWSA(plain, nullptr, 0)) > 0) {
		c.RecvData.append(plain.data.buf, n);
		eargs->event->RequestEvent(c, tid, args);
		}

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

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

		else if (pIoCtx->operation == OP_WRITE) {

		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 (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\n";
		eargs->event->ResponseEvent(c, tid, (ULONG_PTR)eargs->args);
		}

		if (!c.SendData.empty()) start_write_app(pClientContext);
		else start_read(pClientContext);
		else if (io->operation == OP_WRITE) {
		if (!c.SendData.empty()) start_write(ctx);
		else start_read(ctx);
		}
		}
		catch (NetException& e) {
		if (e.getErrorType() != NetException::Note) {
		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;
		}
		eargs->event->DisconnectEvent(c, tid, args);
		delete ctx;
		}
		}
		}