fix: use SPSCQueue for thread safety

project-slippi · Oct 30, 2023 · 6907106 · 6907106
1 parent 5c7502f
commit 6907106
Show file tree

Hide file tree

Showing 4 changed files with 15 additions and 11 deletions.
diff --git a/Source/Core/Core/HW/EXI/EXI_DeviceSlippi.cpp b/Source/Core/Core/HW/EXI/EXI_DeviceSlippi.cpp
@@ -475,19 +475,19 @@ void CEXISlippi::writeToFileAsync(u8* payload, u32 length, std::string file_opti
   write_msg->data = payload_data;
   write_msg->operation = file_option;
 
-  file_write_queue.push(std::move(write_msg));
+  file_write_queue.Push(std::move(write_msg));
 }
 
 void CEXISlippi::FileWriteThread(void)
 {
   Common::SetCurrentThreadName("Slippi File Write");
-  while (write_thread_running || !file_write_queue.empty())
+  while (write_thread_running || !file_write_queue.Empty())
   {
     // Process all messages
-    while (!file_write_queue.empty())
+    while (!file_write_queue.Empty())
     {
-      writeToFile(std::move(file_write_queue.front()));
-      file_write_queue.pop();
+      writeToFile(std::move(file_write_queue.Front()));
+      file_write_queue.Pop();
 
       Common::SleepCurrentThread(0);
     }

diff --git a/Source/Core/Core/HW/EXI/EXI_DeviceSlippi.h b/Source/Core/Core/HW/EXI/EXI_DeviceSlippi.h
@@ -7,6 +7,7 @@
 #include "Common/CommonTypes.h"
 #include "Common/FileUtil.h"
 #include "Common/IOFile.h"
+#include "Common/SPSCQueue.h"
 #include "Core/Slippi/SlippiDirectCodes.h"
 #include "Core/Slippi/SlippiExiTypes.h"
 #include "Core/Slippi/SlippiGame.h"
@@ -247,7 +248,7 @@ class CEXISlippi : public IEXIDevice
 
   void FileWriteThread(void);
 
-  std::queue<std::unique_ptr<WriteMessage>> file_write_queue;
+  Common::SPSCQueue<std::unique_ptr<WriteMessage>, false> file_write_queue;
   bool write_thread_running = false;
   std::thread m_file_write_thread;
 

diff --git a/Source/Core/Core/Slippi/SlippiNetplay.cpp b/Source/Core/Core/Slippi/SlippiNetplay.cpp
@@ -724,7 +724,7 @@ void SlippiNetplayClient::SendAsync(std::unique_ptr<sf::Packet> packet)
 {
   {
     std::lock_guard<std::recursive_mutex> lkq(m_crit.async_queue_write);
-    m_async_queue.emplace(std::move(packet));
+    m_async_queue.Push(std::move(packet));
   }
   ENetUtil::WakeupThread(m_client);
 }
@@ -951,10 +951,10 @@ void SlippiNetplayClient::ThreadFunc()
     ENetEvent net_event;
     int net;
     net = enet_host_service(m_client, &net_event, 250);
-    while (!m_async_queue.empty())
+    while (!m_async_queue.Empty())
     {
-      Send(*(m_async_queue.front().get()));
-      m_async_queue.pop();
+      Send(*(m_async_queue.Front().get()));
+      m_async_queue.Pop();
     }
     if (net > 0)
     {

diff --git a/Source/Core/Core/Slippi/SlippiNetplay.h b/Source/Core/Core/Slippi/SlippiNetplay.h
@@ -18,11 +18,13 @@
 
 #include "Common/CommonTypes.h"
 #include "Common/Event.h"
+#include "Common/SPSCQueue.h"
 #include "Common/Timer.h"
 #include "Common/TraversalClient.h"
 #include "Core/NetPlayProto.h"
 #include "Core/Slippi/SlippiPad.h"
 #include "InputCommon/GCPadStatus.h"
+
 #ifdef _WIN32
 #include <Qos2.h>
 #endif
@@ -207,7 +209,8 @@ class SlippiNetplayClient
     std::recursive_mutex async_queue_write;
   } m_crit;
 
-  std::queue<std::unique_ptr<sf::Packet>> m_async_queue;
+  // SLIPPITODO: consider using AsyncQueueEntry like in NetPlayServer.h
+  Common::SPSCQueue<std::unique_ptr<sf::Packet>, false> m_async_queue;
 
   ENetHost* m_client = nullptr;
   std::vector<ENetPeer*> m_server;