DEV9: Move ATA write queue logic into its own class

pcsx2/CMakeLists.txt

@@ -344,6 +344,7 @@ set(pcsx2DEV9Headers
 	DEV9/DEV9.h
 	DEV9/net.h
 	DEV9/pcap_io.h
+	DEV9/SimpleQueue.h
 	DEV9/smap.h
 	${pcsx2DEV9UIHeaders}
 	)

pcsx2/DEV9/ATA/ATA.h

@@ -26,6 +26,8 @@
 #include "PS2Edefs.h"
 #include "PS2Eext.h"
 
+#include "../SimpleQueue.h"
+
 class ATA
 {
 public:
@@ -106,15 +108,11 @@ private:
 
 	struct WriteQueueEntry
 	{
-		std::atomic_bool ready{false};
-		WriteQueueEntry* next;
 		u8* data;
 		u32 length;
 		u64 sector;
 	};
-
-	WriteQueueEntry* head = nullptr;
-	WriteQueueEntry* tail = nullptr;
+	SimpleQueue<WriteQueueEntry> writeQueue;
 
 	std::thread ioThread;
 	bool ioRunning = false;
@@ -213,10 +211,6 @@ private:
 	bool HDD_CanAssessOrSetError();
 	void HDD_SetErrorAtTransferEnd();
 
-	void QueueWrite(u64 sector, u8* data, u32 length);
-	bool DequeueWrite(u64* sector, u8** data, u32* length);
-	bool IsQueueEmpty();
-
 	//Commands
 	void IDE_ExecCmd(u16 value);
 

pcsx2/DEV9/ATA/ATA_State.cpp

@@ -53,10 +53,6 @@ int ATA::Open(ghc::filesystem::path hddPath)
 	hddImage.seekg(0, std::ios::end);
 	hddImageSize = hddImage.tellg();
 
-	//Setup Write queue
-	tail = new WriteQueueEntry();
-	head = tail;
-
 	{
 		std::lock_guard ioSignallock(ioMutex);
 		ioRead = false;
@@ -84,20 +80,15 @@ void ATA::Close()
 		ioThread.join();
 		ioRunning = false;
 	}
-	//Delete queue
-	if (head != nullptr)
-	{
-		if (!IsQueueEmpty())
-		{
-			Console.Error("DEV9: ATA: Write queue not empty");
-			pxAssert(false);
-			abort(); //All data must be written at this point
-		}
 
-		delete head;
-		head = nullptr;
-		tail = nullptr;
+	//verify queue
+	if (!writeQueue.IsQueueEmpty())
+	{
+		Console.Error("DEV9: ATA: Write queue not empty, possible data loss");
+		pxAssert(false);
+		abort(); //All data must be written at this point
 	}
+
 	//Close File Handle
 	if (hddImage.is_open())
 		hddImage.close();
@@ -314,7 +305,7 @@ void ATA::Async(uint cycles)
 			waitingCmd = nullptr;
 			(this->*cmd)();
 		}
-		else if (!IsQueueEmpty()) //Flush cache
+		else if (!writeQueue.IsQueueEmpty()) //Flush cache
 		{
 			//Log_Info("Starting async write");
 			{

pcsx2/DEV9/ATA/ATA_Transfer.cpp

@@ -89,18 +89,16 @@ void ATA::IO_Read()
 
 bool ATA::IO_Write()
 {
-	u64 sector = 0;
-	u8* data = nullptr;
-	u32 len = 0;
-	if (!DequeueWrite(&sector, &data, &len))
+	WriteQueueEntry entry;
+	if (!writeQueue.Dequeue(&entry))
 	{
 		std::lock_guard ioSignallock(ioMutex);
 		ioWrite = false;
 		return false;
 	}
 
-	hddImage.seekp(sector * 512, std::ios::beg);
-	hddImage.write((char*)data, len);
+	hddImage.seekp(entry.sector * 512, std::ios::beg);
+	hddImage.write((char*)entry.data, entry.length);
 	if (hddImage.fail())
 	{
 		Console.Error("DEV9: ATA: File write error");
@@ -108,7 +106,7 @@ bool ATA::IO_Write()
 		abort();
 	}
 	hddImage.flush();
-	delete[] data;
+	delete[] entry.data;
 	return true;
 }
 
@@ -211,40 +209,3 @@ void ATA::HDD_SetErrorAtTransferEnd()
 		HDD_SetLBA(currSect);
 	}
 }
-
-//Used by EE thread only
-void ATA::QueueWrite(u64 sector, u8* data, u32 length)
-{
-	WriteQueueEntry* newEntry = head;
-	newEntry->data = data;
-	newEntry->length = length;
-	newEntry->sector = sector;
-
-	//Allocate Next entry
-	newEntry->next = new WriteQueueEntry();
-	head = newEntry->next;
-	//Set ready
-	newEntry->ready.store(true);
-}
-
-//Used by IO thread only
-bool ATA::DequeueWrite(u64* sector, u8** data, u32* length)
-{
-	if (!tail->ready.load())
-		return false;
-
-	WriteQueueEntry* entry = tail;
-	tail = entry->next;
-
-	*sector = entry->sector;
-	*data = entry->data;
-	*length = entry->length;
-	delete entry;
-	return true;
-}
-
-//Used by EE thread only
-bool ATA::IsQueueEmpty()
-{
-	return !tail->ready.load();
-}

pcsx2/DEV9/ATA/Commands/ATA_CmdDMA.cpp

@@ -62,7 +62,11 @@ void ATA::DRQCmdDMADataFromHost()
 }
 void ATA::PostCmdDMADataFromHost()
 {
-	QueueWrite(currentWriteSectors, currentWrite, currentWriteLength);
+	WriteQueueEntry entry{0};
+	entry.data = currentWrite;
+	entry.length = currentWriteLength;
+	entry.sector = currentWriteSectors;
+	writeQueue.Enqueue(entry);
 	currentWrite = nullptr;
 	currentWriteLength = 0;
 	currentWriteSectors = 0;

pcsx2/DEV9/SimpleQueue.h

@@ -0,0 +1,110 @@
+/*  PCSX2 - PS2 Emulator for PCs
+ *  Copyright (C) 2002-2021  PCSX2 Dev Team
+ *
+ *  PCSX2 is free software: you can redistribute it and/or modify it under the terms
+ *  of the GNU Lesser General Public License as published by the Free Software Found-
+ *  ation, either version 3 of the License, or (at your option) any later version.
+ *
+ *  PCSX2 is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY;
+ *  without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR
+ *  PURPOSE.  See the GNU General Public License for more details.
+ *
+ *  You should have received a copy of the GNU General Public License along with PCSX2.
+ *  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+#pragma once
+
+//Designed to allow one thread to queue data to another thread
+template <class T>
+class SimpleQueue
+{
+private:
+	struct SimpleQueueEntry
+	{
+		std::atomic_bool ready{false};
+		SimpleQueueEntry* next;
+		T value;
+	};
+
+	std::atomic<SimpleQueueEntry*> head{nullptr};
+	SimpleQueueEntry* tail = nullptr;
+
+public:
+	SimpleQueue();
+
+	//Used by single queue thread (i.e. EE)
+	void Enqueue(T entry);
+	//Used by single worker thread (i.e. IO)
+	bool Dequeue(T* entry);
+	//May return false negative when another thread is mid Queue()
+	//Intended to only be used from queue thread
+	bool IsQueueEmpty();
+
+	~SimpleQueue();
+};
+
+template <class T>
+SimpleQueue<T>::SimpleQueue()
+{
+	tail = new SimpleQueueEntry();
+	head.store(tail);
+}
+
+template <class T>
+void SimpleQueue<T>::Enqueue(T entry)
+{
+	//Allocate Next entry, and assign to head
+	SimpleQueueEntry* newHead = new SimpleQueueEntry();
+	SimpleQueueEntry* newEntry = head.exchange(newHead);
+
+	//Fill in
+	newEntry->value = entry;
+	newEntry->next = newHead;
+
+	//Set ready (can be dequeued)
+	newEntry->ready.store(true);
+}
+
+template <class T>
+bool SimpleQueue<T>::Dequeue(T* entry)
+{
+	if (!tail->ready.load())
+		return false;
+
+	SimpleQueueEntry* retEntry = tail;
+	tail = retEntry->next;
+
+	*entry = retEntry->value;
+	delete retEntry;
+	return true;
+}
+
+//Note, next entry may not be ready to dequeue
+template <class T>
+bool SimpleQueue<T>::IsQueueEmpty()
+{
+	return head.load() == tail;
+}
+
+template <class T>
+SimpleQueue<T>::~SimpleQueue()
+{
+	if (head != nullptr)
+	{
+		if (!IsQueueEmpty())
+		{
+			Console.Error("DEV9: Queue not empty");
+			pxAssert(false);
+
+			//Empty Queue
+			T entry;
+			while (!IsQueueEmpty())
+				Dequeue(&entry);
+		}
+
+		delete head;
+		head = nullptr;
+		tail = nullptr;
+	}
+}

pcsx2/windows/VCprojects/pcsx2.vcxproj

@@ -650,6 +650,7 @@
     <ClInclude Include="..\..\DEV9\DEV9.h" />
     <ClInclude Include="..\..\DEV9\net.h" />
     <ClInclude Include="..\..\DEV9\pcap_io.h" />
+    <ClInclude Include="..\..\DEV9\SimpleQueue.h" />
     <ClInclude Include="..\..\DEV9\smap.h" />
     <ClInclude Include="..\..\DEV9\Win32\pcap_io_win32_funcs.h" />
     <ClInclude Include="..\..\DEV9\Win32\resource.h" />

pcsx2/windows/VCprojects/pcsx2.vcxproj.filters

@@ -1851,6 +1851,9 @@
     <ClInclude Include="..\..\DEV9\pcap_io.h">
       <Filter>System\Ps2\DEV9</Filter>
     </ClInclude>
+    <ClInclude Include="..\..\DEV9\SimpleQueue.h">
+      <Filter>System\Ps2\DEV9</Filter>
+    </ClInclude>
     <ClInclude Include="..\..\DEV9\smap.h">
       <Filter>System\Ps2\DEV9</Filter>
     </ClInclude>