EXI: Refactor BBA-TAP interface for Windows to use a read thread

This also has the added benefit of not crashing under most circumstances.
Includes a few other changes, including replacing the atomic<bool> with a
Flag, as well as adding a flag for indicating read thread shutdown.

Source/Core/Core/HW/BBA-TAP/TAP_Apple.cpp

@@ -23,10 +23,8 @@ bool CEXIETHERNET::Activate()
 		return false;
 	}
 
-	readEnabled.store(false);
-
 	INFO_LOG(SP1, "BBA initialized.");
-	return true;
+	return RecvInit();
 }
 
 void CEXIETHERNET::Deactivate()
@@ -34,7 +32,8 @@ void CEXIETHERNET::Deactivate()
 	close(fd);
 	fd = -1;
 
-	readEnabled.store(false);
+	readEnabled.Clear();
+	readThreadShutdown.Set();
 	if (readThread.joinable())
 		readThread.join();
 }
@@ -64,11 +63,8 @@ bool CEXIETHERNET::SendFrame(u8* frame, u32 size)
 
 static void ReadThreadHandler(CEXIETHERNET* self)
 {
-	while (true)
+	while (!self->readThreadShutdown.IsSet())
 	{
-		if (self->fd < 0)
-			return;
-
 		fd_set rfds;
 		FD_ZERO(&rfds);
 		FD_SET(self->fd, &rfds);
@@ -84,7 +80,7 @@ static void ReadThreadHandler(CEXIETHERNET* self)
 		{
 			ERROR_LOG(SP1, "Failed to read from BBA, err=%d", readBytes);
 		}
-		else if (self->readEnabled.load())
+		else if (self->readEnabled.IsSet())
 		{
 			INFO_LOG(SP1, "Read data: %s", ArrayToString(self->mRecvBuffer, readBytes, 0x10).c_str());
 			self->mRecvBufferLength = readBytes;
@@ -99,16 +95,12 @@ bool CEXIETHERNET::RecvInit()
 	return true;
 }
 
-bool CEXIETHERNET::RecvStart()
+void CEXIETHERNET::RecvStart()
 {
-	if (!readThread.joinable())
+	readEnabled.Set();
-		RecvInit();
-
-	readEnabled.store(true);
-	return true;
 }
 
 void CEXIETHERNET::RecvStop()
 {
-	readEnabled.store(false);
+	readEnabled.Clear();
 }

Source/Core/Core/HW/BBA-TAP/TAP_Unix.cpp

@@ -68,10 +68,8 @@ bool CEXIETHERNET::Activate()
 	}
 	ioctl(fd, TUNSETNOCSUM, 1);
 
-	readEnabled.store(false);
-
 	INFO_LOG(SP1, "BBA initialized with associated tap %s", ifr.ifr_name);
-	return true;
+	return RecvInit();
 #else
 	NOTIMPLEMENTED("Activate");
 	return false;
@@ -84,7 +82,8 @@ void CEXIETHERNET::Deactivate()
 	close(fd);
 	fd = -1;
 
-	readEnabled.store(false);
+	readEnabled.Clear();
+	readThreadShutdown.Set();
 	if (readThread.joinable())
 		readThread.join();
 #else
@@ -126,11 +125,8 @@ bool CEXIETHERNET::SendFrame(u8* frame, u32 size)
 
 static void ReadThreadHandler(CEXIETHERNET* self)
 {
-	while (true)
+	while (!self->readThreadShutdown.IsSet())
 	{
-		if (self->fd < 0)
-			return;
-
 		fd_set rfds;
 		FD_ZERO(&rfds);
 		FD_SET(self->fd, &rfds);
@@ -146,7 +142,7 @@ static void ReadThreadHandler(CEXIETHERNET* self)
 		{
 			ERROR_LOG(SP1, "Failed to read from BBA, err=%d", readBytes);
 		}
-		else if (self->readEnabled.load())
+		else if (self->readEnabled.IsSet())
 		{
 			INFO_LOG(SP1, "Read data: %s", ArrayToString(self->mRecvBuffer, readBytes, 0x10).c_str());
 			self->mRecvBufferLength = readBytes;
@@ -166,24 +162,19 @@ bool CEXIETHERNET::RecvInit()
 #endif
 }
 
-bool CEXIETHERNET::RecvStart()
+void CEXIETHERNET::RecvStart()
 {
 #ifdef __linux__
-	if (!readThread.joinable())
+	readEnabled.Set();
-		RecvInit();
-
-	readEnabled.store(true);
-	return true;
 #else
 	NOTIMPLEMENTED("RecvStart");
-	return false;
 #endif
 }
 
 void CEXIETHERNET::RecvStop()
 {
 #ifdef __linux__
-	readEnabled.store(false);
+	readEnabled.Clear();
 #else
 	NOTIMPLEMENTED("RecvStop");
 #endif

Source/Core/Core/HW/BBA-TAP/TAP_Win32.cpp

@@ -2,6 +2,7 @@
 // Licensed under GPLv2+
 // Refer to the license.txt file included.
 
+#include "Common/Assert.h"
 #include "Common/MsgHandler.h"
 #include "Common/StringUtil.h"
 #include "Common/Logging/Log.h"
@@ -222,7 +223,14 @@ bool CEXIETHERNET::Activate()
 		return false;
 	}
 
-	return true;
+	/* initialize read/write events */
+	mReadOverlapped.hEvent = CreateEvent(nullptr, TRUE, FALSE, nullptr);
+	mWriteOverlapped.hEvent = CreateEvent(nullptr, TRUE, FALSE, nullptr);
+	if (mReadOverlapped.hEvent == nullptr || mWriteOverlapped.hEvent == nullptr)
+		return false;
+
+	mWriteBuffer.reserve(1518);
+	return RecvInit();
 }
 
 void CEXIETHERNET::Deactivate()
@@ -230,10 +238,24 @@ void CEXIETHERNET::Deactivate()
 	if (!IsActivated())
 		return;
 
-	RecvStop();
+	// Signal read thread to exit.
+	readEnabled.Clear();
+	readThreadShutdown.Set();
 
+	// Cancel any outstanding requests from both this thread (writes), and the read thread.
+	CancelIoEx(mHAdapter, nullptr);
+
+	// Wait for read thread to exit.
+	if (readThread.joinable())
+		readThread.join();
+
+	// Clean-up handles
+	CloseHandle(mReadOverlapped.hEvent);
+	CloseHandle(mWriteOverlapped.hEvent);
 	CloseHandle(mHAdapter);
 	mHAdapter = INVALID_HANDLE_VALUE;
+	memset(&mReadOverlapped, 0, sizeof(mReadOverlapped));
+	memset(&mWriteOverlapped, 0, sizeof(mWriteOverlapped));
 }
 
 bool CEXIETHERNET::IsActivated()
@@ -241,101 +263,103 @@ bool CEXIETHERNET::IsActivated()
 	return mHAdapter != INVALID_HANDLE_VALUE;
 }
 
-bool CEXIETHERNET::SendFrame(u8 *frame, u32 size)
+static void ReadThreadHandler(CEXIETHERNET* self)
 {
-	DEBUG_LOG(SP1, "SendFrame %x\n%s",
+	while (!self->readThreadShutdown.IsSet())
-		size, ArrayToString(frame, size, 0x10).c_str());
-
-	OVERLAPPED overlap;
-	ZeroMemory(&overlap, sizeof(overlap));
-
-	// WriteFile will always return false because the TAP handle is async
-	WriteFile(mHAdapter, frame, size, nullptr, &overlap);
-
-	DWORD res = GetLastError();
-	if (res != ERROR_IO_PENDING)
 	{
-		ERROR_LOG(SP1, "Failed to send packet with error 0x%X", res);
+		DWORD transferred;
+
+		// Read from TAP into internal buffer.
+		if (ReadFile(self->mHAdapter, self->mRecvBuffer, BBA_RECV_SIZE, &transferred, &self->mReadOverlapped))
+		{
+			// Returning immediately is not likely to happen, but if so, reset the event state manually.
+			ResetEvent(self->mReadOverlapped.hEvent);
+		}
+		else
+		{
+			// IO should be pending.
+			if (GetLastError() != ERROR_IO_PENDING)
+			{
+				ERROR_LOG(SP1, "ReadFile failed (err=0x%X)", GetLastError());
+				continue;
+			}
+
+			// Block until the read completes.
+			if (!GetOverlappedResult(self->mHAdapter, &self->mReadOverlapped, &transferred, TRUE))
+			{
+				// If CancelIO was called, we should exit (the flag will be set).
+				if (GetLastError() == ERROR_OPERATION_ABORTED)
+					continue;
+
+				// Something else went wrong.
+				ERROR_LOG(SP1, "GetOverlappedResult failed (err=0x%X)", GetLastError());
+				continue;
+			}
+		}
+
+		// Copy to BBA buffer, and fire interrupt if enabled.
+		DEBUG_LOG(SP1, "Received %u bytes\n: %s", transferred, ArrayToString(self->mRecvBuffer, transferred, 0x10).c_str());
+		if (self->readEnabled.IsSet())
+		{
+			self->mRecvBufferLength = transferred;
+			self->RecvHandlePacket();
+		}
+	}
+}
+
+bool CEXIETHERNET::SendFrame(u8* frame, u32 size)
+{
+	DEBUG_LOG(SP1, "SendFrame %u bytes:\n%s", size, ArrayToString(frame, size, 0x10).c_str());
+
+	// Check for a background write. We can't issue another one until this one has completed.
+	DWORD transferred;
+	if (mWritePending)
+	{
+		// Wait for previous write to complete.
+		if (!GetOverlappedResult(mHAdapter, &mWriteOverlapped, &transferred, TRUE))
+			ERROR_LOG(SP1, "GetOverlappedResult failed (err=0x%X)", GetLastError());
+	}
+
+	// Copy to write buffer.
+	mWriteBuffer.resize(size);
+	memcpy(mWriteBuffer.data(), frame, size);
+	mWritePending = true;
+
+	// Queue async write.
+	if (WriteFile(mHAdapter, mWriteBuffer.data(), size, &transferred, &mWriteOverlapped))
+	{
+		// Returning immediately is not likely to happen, but if so, reset the event state manually.
+		ResetEvent(mWriteOverlapped.hEvent);
+	}
+	else
+	{
+		// IO should be pending.
+		if (GetLastError() != ERROR_IO_PENDING)
+		{
+			ERROR_LOG(SP1, "WriteFile failed (err=0x%X)", GetLastError());
+			ResetEvent(mWriteOverlapped.hEvent);
+			mWritePending = false;
+			return false;
+		}
 	}
 
 	// Always report the packet as being sent successfully, even though it might be a lie
 	SendComplete();
-
 	return true;
 }
 
-VOID CALLBACK CEXIETHERNET::ReadWaitCallback(PVOID lpParameter, BOOLEAN TimerFired)
-{
-	CEXIETHERNET* self = (CEXIETHERNET*)lpParameter;
-
-	GetOverlappedResult(self->mHAdapter, &self->mReadOverlapped,
-		(LPDWORD)&self->mRecvBufferLength, false);
-
-	self->RecvHandlePacket();
-}
-
 bool CEXIETHERNET::RecvInit()
 {
-	// Set up recv event
+	readThread = std::thread(ReadThreadHandler, this);
-
-	if ((mHRecvEvent = CreateEvent(nullptr, false, false, nullptr)) == nullptr)
-	{
-		ERROR_LOG(SP1, "Failed to create recv event:%x", GetLastError());
-		return false;
-	}
-
-	ZeroMemory(&mReadOverlapped, sizeof(mReadOverlapped));
-
-	RegisterWaitForSingleObject(&mHReadWait, mHRecvEvent, ReadWaitCallback,
-		this, INFINITE, WT_EXECUTEDEFAULT);
-
-	mReadOverlapped.hEvent = mHRecvEvent;
-
 	return true;
 }
 
-bool CEXIETHERNET::RecvStart()
+void CEXIETHERNET::RecvStart()
 {
-	if (!IsActivated())
+	readEnabled.Set();
-		return false;
-
-	if (mHRecvEvent == INVALID_HANDLE_VALUE)
-		RecvInit();
-
-	DWORD res = ReadFile(mHAdapter, mRecvBuffer, BBA_RECV_SIZE,
-		(LPDWORD)&mRecvBufferLength, &mReadOverlapped);
-
-	if (res)
-	{
-		// Since the read is synchronous here, complete immediately
-		RecvHandlePacket();
-		return true;
-	}
-	else
-	{
-		DWORD err = GetLastError();
-		if (err == ERROR_IO_PENDING)
-		{
-			return true;
-		}
-
-		// Unexpected error
-		ERROR_LOG(SP1, "Failed to recieve packet with error 0x%X", err);
-		return false;
-	}
-
 }
 
 void CEXIETHERNET::RecvStop()
 {
-	if (!IsActivated())
+	readEnabled.Clear();
-		return;
-
-	UnregisterWaitEx(mHReadWait, INVALID_HANDLE_VALUE);
-
-	if (mHRecvEvent != INVALID_HANDLE_VALUE)
-	{
-		CloseHandle(mHRecvEvent);
-		mHRecvEvent = INVALID_HANDLE_VALUE;
-	}
 }

Source/Core/Core/HW/EXI_DeviceEthernet.cpp

@@ -47,8 +47,9 @@ CEXIETHERNET::CEXIETHERNET()
 
 #if defined(_WIN32)
 	mHAdapter = INVALID_HANDLE_VALUE;
-	mHRecvEvent = INVALID_HANDLE_VALUE;
+	memset(&mReadOverlapped, 0, sizeof(mReadOverlapped));
-	mHReadWait = INVALID_HANDLE_VALUE;
+	memset(&mWriteOverlapped, 0, sizeof(mWriteOverlapped));
+	mWritePending = false;
 #elif defined(__linux__) || defined(__APPLE__)
 	fd = -1;
 #endif

Source/Core/Core/HW/EXI_DeviceEthernet.h

@@ -6,11 +6,13 @@
 
 #include <atomic>
 #include <thread>
+#include <vector>
 
 #ifdef _WIN32
 #include <Windows.h>
 #endif
 
+#include "Common/Flag.h"
 #include "Core/HW/EXI_Device.h"
 
 class PointerWrap;
@@ -316,21 +318,26 @@ public:
 	bool IsActivated();
 	bool SendFrame(u8 *frame, u32 size);
 	bool RecvInit();
-	bool RecvStart();
+	void RecvStart();
 	void RecvStop();
 
 	u8 *mRecvBuffer;
 	u32 mRecvBufferLength;
 
 #if defined(_WIN32)
-	HANDLE mHAdapter, mHRecvEvent, mHReadWait;
+	HANDLE mHAdapter;
-	DWORD mMtu;
 	OVERLAPPED mReadOverlapped;
-	static VOID CALLBACK ReadWaitCallback(PVOID lpParameter, BOOLEAN TimerFired);
+	OVERLAPPED mWriteOverlapped;
+	std::vector<u8> mWriteBuffer;
+	bool mWritePending;
 #elif defined(__linux__) || defined(__APPLE__) || defined(__FreeBSD__)
 	int fd;
+#endif
+
+#if defined(WIN32) || defined(__linux__) || defined(__APPLE__) || defined(__FreeBSD__)
 	std::thread readThread;
-	std::atomic<bool> readEnabled;
+	Common::Flag readEnabled;
+	Common::Flag readThreadShutdown;
 #endif
 
 };