Merge pull request #2198 from mathieui/netplay-threading-fix

Fix netplay threading issues

Source/Core/Common/CMakeLists.txt

@@ -1,6 +1,7 @@
 set(SRCS BreakPoints.cpp
          CDUtils.cpp
          ColorUtil.cpp
+         ENetUtil.cpp
          FileSearch.cpp
          FileUtil.cpp
          GekkoDisassembler.cpp

Source/Core/Common/Common.vcxproj

@@ -51,6 +51,7 @@
     <ClInclude Include="CommonTypes.h" />
     <ClInclude Include="CPUDetect.h" />
     <ClInclude Include="DebugInterface.h" />
+    <ClInclude Include="ENetUtil.h" />
     <ClInclude Include="Event.h" />
     <ClInclude Include="FifoQueue.h" />
     <ClInclude Include="FileSearch.h" />
@@ -94,6 +95,7 @@
     <ClCompile Include="BreakPoints.cpp" />
     <ClCompile Include="CDUtils.cpp" />
     <ClCompile Include="ColorUtil.cpp" />
+    <ClCompile Include="ENetUtil.cpp" />
     <ClCompile Include="FileSearch.cpp" />
     <ClCompile Include="FileUtil.cpp" />
     <ClCompile Include="GekkoDisassembler.cpp" />

Source/Core/Common/Common.vcxproj.filters

@@ -25,6 +25,7 @@
     <ClInclude Include="CommonTypes.h" />
     <ClInclude Include="CPUDetect.h" />
     <ClInclude Include="DebugInterface.h" />
+    <ClInclude Include="ENetUtil.h" />
     <ClInclude Include="FifoQueue.h" />
     <ClInclude Include="FileSearch.h" />
     <ClInclude Include="FileUtil.h" />
@@ -78,6 +79,7 @@
     <ClCompile Include="BreakPoints.cpp" />
     <ClCompile Include="CDUtils.cpp" />
     <ClCompile Include="ColorUtil.cpp" />
+    <ClCompile Include="ENetUtil.cpp" />
     <ClCompile Include="FileSearch.cpp" />
     <ClCompile Include="FileUtil.cpp" />
     <ClCompile Include="Hash.cpp" />

Source/Core/Common/ENetUtil.cpp

@@ -0,0 +1,40 @@
+// Copyright 2015 Dolphin Emulator Project
+// Licensed under GPLv2
+// Refer to the license.txt file included.
+
+#include "ENetUtil.h"
+
+namespace ENetUtil
+{
+
+void WakeupThread(ENetHost* host)
+{
+	// Send ourselves a spurious message.  This is hackier than it should be.
+	// comex reported this as https://github.com/lsalzman/enet/issues/23, so
+	// hopefully there will be a better way to do it in the future.
+	ENetAddress address;
+	if (host->address.port != 0)
+		address.port = host->address.port;
+	else
+		enet_socket_get_address(host->socket, &address);
+	address.host = 0x0100007f; // localhost
+	u8 byte = 0;
+	ENetBuffer buf;
+	buf.data = &byte;
+	buf.dataLength = 1;
+	enet_socket_send(host->socket, &address, &buf, 1);
+}
+
+int ENET_CALLBACK InterceptCallback(ENetHost* host, ENetEvent* event)
+{
+	// wakeup packet received
+	if (host->receivedDataLength == 1 && host->receivedData[0] == 0)
+	{
+		event->type = (ENetEventType) 42;
+		return 1;
+	}
+	return 0;
+}
+
+
+}

Source/Core/Common/ENetUtil.h

@@ -0,0 +1,16 @@
+// Copyright 2015 Dolphin Emulator Project
+// Licensed under GPLv2
+// Refer to the license.txt file included.
+//
+#pragma once
+
+#include <enet/enet.h>
+#include "Common.h"
+
+namespace ENetUtil
+{
+
+void WakeupThread(ENetHost* host);
+int ENET_CALLBACK InterceptCallback(ENetHost* host, ENetEvent* event);
+
+}

Source/Core/Common/TraversalClient.cpp

@@ -7,9 +7,10 @@ static void GetRandomishBytes(u8* buf, size_t size)
 {
 	// We don't need high quality random numbers (which might not be available),
 	// just non-repeating numbers!
-	srand(enet_time_get());
+	static std::mt19937 prng(enet_time_get());
+	static std::uniform_int_distribution<unsigned int> u8_distribution(0, 255);
 	for (size_t i = 0; i < size; i++)
-		buf[i] = rand() & 0xff;
+		buf[i] = u8_distribution(prng);
 }
 
 TraversalClient::TraversalClient(ENetHost* netHost, const std::string& server, const u16 port)
@@ -301,7 +302,8 @@ void TraversalClient::Reset()
 int ENET_CALLBACK TraversalClient::InterceptCallback(ENetHost* host, ENetEvent* event)
 {
 	auto traversalClient = g_TraversalClient.get();
-	if (traversalClient->TestPacket(host->receivedData, host->receivedDataLength, &host->receivedAddress))
+	if (traversalClient->TestPacket(host->receivedData, host->receivedDataLength, &host->receivedAddress)
+			|| (host->receivedDataLength == 1 && host->receivedData[0] == 0))
 	{
 		event->type = (ENetEventType)42;
 		return 1;

Source/Core/Common/TraversalClient.h

@@ -4,6 +4,7 @@
 #include <functional>
 #include <list>
 #include <memory>
+#include <random>
 #include <enet/enet.h>
 #include "Common/Common.h"
 #include "Common/Thread.h"

Source/Core/Core/NetPlayClient.cpp

@@ -103,8 +103,11 @@ NetPlayClient::NetPlayClient(const std::string& address, const u16 port, NetPlay
 		if (net > 0 && netEvent.type == ENET_EVENT_TYPE_CONNECT)
 		{
 			if (Connect())
+			{
+				m_client->intercept = ENetUtil::InterceptCallback;
 				m_thread = std::thread(&NetPlayClient::ThreadFunc, this);
 			}
+		}
 		else
 		{
 			PanicAlertT("Failed to Connect!");
@@ -416,7 +419,6 @@ unsigned int NetPlayClient::OnData(sf::Packet& packet)
 		spac << (MessageId)NP_MSG_PONG;
 		spac << ping_key;
 
-		std::lock_guard<std::recursive_mutex> lks(m_crit.send);
 		Send(spac);
 	}
 	break;
@@ -480,6 +482,15 @@ void NetPlayClient::Disconnect()
 	m_server = nullptr;
 }
 
+void NetPlayClient::SendAsync(sf::Packet* packet)
+{
+	{
+		std::lock_guard<std::recursive_mutex> lkq(m_crit.async_queue_write);
+		m_async_queue.Push(std::unique_ptr<sf::Packet>(packet));
+	}
+	ENetUtil::WakeupThread(m_client);
+}
+
 // called from ---NETPLAY--- thread
 void NetPlayClient::ThreadFunc()
 {
@@ -487,11 +498,13 @@ void NetPlayClient::ThreadFunc()
 	{
 		ENetEvent netEvent;
 		int net;
-		{
-			std::lock_guard<std::recursive_mutex> lks(m_crit.send);
 		if (m_traversal_client)
 			m_traversal_client->HandleResends();
-			net = enet_host_service(m_client, &netEvent, 4);
+		net = enet_host_service(m_client, &netEvent, 250);
+		while (!m_async_queue.Empty())
+		{
+			Send(*(m_async_queue.Front().get()));
+			m_async_queue.Pop();
 		}
 		if (net > 0)
 		{
@@ -517,7 +530,6 @@ void NetPlayClient::ThreadFunc()
 				break;
 			}
 		}
-
 	}
 
 	Disconnect();
@@ -577,57 +589,47 @@ void NetPlayClient::GetPlayers(std::vector<const Player *> &player_list)
 // called from ---GUI--- thread
 void NetPlayClient::SendChatMessage(const std::string& msg)
 {
-	sf::Packet spac;
+	sf::Packet* spac = new sf::Packet;
-	spac << (MessageId)NP_MSG_CHAT_MESSAGE;
+	*spac << (MessageId)NP_MSG_CHAT_MESSAGE;
-	spac << msg;
+	*spac << msg;
-
+	SendAsync(spac);
-	std::lock_guard<std::recursive_mutex> lks(m_crit.send);
-	Send(spac);
 }
 
 // called from ---CPU--- thread
 void NetPlayClient::SendPadState(const PadMapping in_game_pad, const GCPadStatus& pad)
 {
-	// send to server
+	sf::Packet* spac = new sf::Packet;
-	sf::Packet spac;
+	*spac << (MessageId)NP_MSG_PAD_DATA;
-	spac << (MessageId)NP_MSG_PAD_DATA;
+	*spac << in_game_pad;
-	spac << in_game_pad;
+	*spac << pad.button << pad.analogA << pad.analogB << pad.stickX << pad.stickY << pad.substickX << pad.substickY << pad.triggerLeft << pad.triggerRight;
-	spac << pad.button << pad.analogA << pad.analogB << pad.stickX << pad.stickY << pad.substickX << pad.substickY << pad.triggerLeft << pad.triggerRight;
 
-	std::lock_guard<std::recursive_mutex> lks(m_crit.send);
+	SendAsync(spac);
-	Send(spac);
 }
 
 // called from ---CPU--- thread
 void NetPlayClient::SendWiimoteState(const PadMapping in_game_pad, const NetWiimote& nw)
 {
-	// send to server
+	sf::Packet* spac = new sf::Packet;
-	sf::Packet spac;
+	*spac << (MessageId)NP_MSG_WIIMOTE_DATA;
-	spac << (MessageId)NP_MSG_WIIMOTE_DATA;
+	*spac << in_game_pad;
-	spac << in_game_pad;
+	*spac << (u8)nw.size();
-	spac << (u8)nw.size();
 	for (auto it : nw)
 	{
-		spac << it;
+		*spac << it;
 	}
-
+	SendAsync(spac);
-	std::lock_guard<std::recursive_mutex> lks(m_crit.send);
-	Send(spac);
 }
 
 // called from ---GUI--- thread
 bool NetPlayClient::StartGame(const std::string &path)
 {
 	std::lock_guard<std::recursive_mutex> lkg(m_crit.game);
-
 	// tell server i started the game
-	sf::Packet spac;
+	sf::Packet* spac = new sf::Packet;
-	spac << (MessageId)NP_MSG_START_GAME;
+	*spac << (MessageId)NP_MSG_START_GAME;
-	spac << m_current_game;
+	*spac << m_current_game;
-	spac << (char *)&g_NetPlaySettings;
+	*spac << (char *)&g_NetPlaySettings;
-
+	SendAsync(spac);
-	std::lock_guard<std::recursive_mutex> lks(m_crit.send);
-	Send(spac);
 
 	if (m_is_running)
 	{
@@ -954,6 +956,7 @@ bool NetPlayClient::StopGame()
 	return true;
 }
 
+// called from ---GUI--- thread
 void NetPlayClient::Stop()
 {
 	if (m_is_running == false)
@@ -976,9 +979,9 @@ void NetPlayClient::Stop()
 	// tell the server to stop if we have a pad mapped in game.
 	if (isPadMapped)
 	{
-		sf::Packet spac;
+		sf::Packet* spac = new sf::Packet;
-		spac << (MessageId)NP_MSG_STOP_GAME;
+		*spac << (MessageId)NP_MSG_STOP_GAME;
-		Send(spac);
+		SendAsync(spac);
 	}
 }
 

Source/Core/Core/NetPlayClient.h

@@ -9,6 +9,7 @@
 #include <sstream>
 #include <SFML/Network/Packet.hpp>
 #include "Common/CommonTypes.h"
+#include "Common/ENetUtil.h"
 #include "Common/FifoQueue.h"
 #include "Common/Thread.h"
 #include "Common/Timer.h"
@@ -47,6 +48,7 @@ class NetPlayClient : public TraversalClientClient
 {
 public:
 	void ThreadFunc();
+	void SendAsync(sf::Packet* packet);
 
 	NetPlayClient(const std::string& address, const u16 port, NetPlayUI* dialog, const std::string& name, bool traversal, std::string centralServer, u16 centralPort);
 	~NetPlayClient();
@@ -92,9 +94,12 @@ protected:
 	{
 		std::recursive_mutex game;
 		// lock order
-		std::recursive_mutex players, send;
+		std::recursive_mutex players;
+		std::recursive_mutex async_queue_write;
 	} m_crit;
 
+	Common::FifoQueue<std::unique_ptr<sf::Packet>, false> m_async_queue;
+
 	Common::FifoQueue<GCPadStatus> m_pad_buffer[4];
 	Common::FifoQueue<NetWiimote>  m_wiimote_buffer[4];
 

Source/Core/Core/NetPlayServer.cpp

@@ -91,8 +91,9 @@ NetPlayServer::NetPlayServer(const u16 port, bool traversal, std::string central
 		serverAddr.host = ENET_HOST_ANY;
 		serverAddr.port = port;
 		m_server = enet_host_create(&serverAddr, 10, 3, 0, 0);
+		if (m_server != nullptr)
+			m_server->intercept = ENetUtil::InterceptCallback;
 	}
-
 	if (m_server != nullptr)
 	{
 		is_connected = true;
@@ -117,7 +118,6 @@ void NetPlayServer::ThreadFunc()
 			spac << (MessageId)NP_MSG_PING;
 			spac << m_ping_key;
 
-			std::lock_guard<std::recursive_mutex> lks(m_crit.send);
 			m_ping_timer.Start();
 			SendToClients(spac);
 			m_update_pings = false;
@@ -125,11 +125,16 @@ void NetPlayServer::ThreadFunc()
 
 		ENetEvent netEvent;
 		int net;
-		{
-			std::lock_guard<std::recursive_mutex> lks(m_crit.send);
 		if (m_traversal_client)
 			m_traversal_client->HandleResends();
-			net = enet_host_service(m_server, &netEvent, 4);
+		net = enet_host_service(m_server, &netEvent, 1000);
+		while (!m_async_queue.Empty())
+		{
+			{
+				std::lock_guard<std::recursive_mutex> lkp(m_crit.players);
+				SendToClients(*(m_async_queue.Front().get()));
+			}
+			m_async_queue.Pop();
 		}
 		if (net > 0)
 		{
@@ -149,7 +154,6 @@ void NetPlayServer::ThreadFunc()
 					sf::Packet spac;
 					spac << (MessageId)error;
 					// don't need to lock, this client isn't in the client map
-					std::lock_guard<std::recursive_mutex> lks(m_crit.send);
 					Send(accept_peer, spac);
 					if (netEvent.peer->data)
 					{
@@ -272,9 +276,6 @@ unsigned int NetPlayServer::OnConnect(ENetPeer* socket)
 		}
 	}
 
-	{
-		std::lock_guard<std::recursive_mutex> lks(m_crit.send);
-
 	// send join message to already connected clients
 	sf::Packet spac;
 	spac << (MessageId)NP_MSG_PLAYER_JOIN;
@@ -311,13 +312,10 @@ unsigned int NetPlayServer::OnConnect(ENetPeer* socket)
 		Send(player.socket, spac);
 	}
 
-	} // unlock send
-
 	// add client to the player list
 	{
 		std::lock_guard<std::recursive_mutex> lkp(m_crit.players);
 		m_players.insert(std::pair<PlayerId, Client>(*(PlayerId *)player.socket->data, player));
-		std::lock_guard<std::recursive_mutex> lks(m_crit.send);
 		UpdatePadMapping(); // sync pad mappings with everyone
 		UpdateWiimoteMapping();
 	}
@@ -343,7 +341,6 @@ unsigned int NetPlayServer::OnDisconnect(Client& player)
 				sf::Packet spac;
 				spac << (MessageId)NP_MSG_DISABLE_GAME;
 				// this thread doesn't need players lock
-				std::lock_guard<std::recursive_mutex> lks(m_crit.send);
 				SendToClients(spac, 1);
 				break;
 			}
@@ -362,7 +359,6 @@ unsigned int NetPlayServer::OnDisconnect(Client& player)
 		m_players.erase(it);
 
 	// alert other players of disconnect
-	std::lock_guard<std::recursive_mutex> lks(m_crit.send);
 	SendToClients(spac);
 
 	for (PadMapping& mapping : m_pad_map)
@@ -447,13 +443,20 @@ void NetPlayServer::AdjustPadBufferSize(unsigned int size)
 	m_target_buffer_size = size;
 
 	// tell clients to change buffer size
-	sf::Packet spac;
+	sf::Packet* spac = new sf::Packet;
-	spac << (MessageId)NP_MSG_PAD_BUFFER;
+	*spac << (MessageId)NP_MSG_PAD_BUFFER;
-	spac << (u32)m_target_buffer_size;
+	*spac << (u32)m_target_buffer_size;
 
-	std::lock_guard<std::recursive_mutex> lkp(m_crit.players);
+	SendAsyncToClients(spac);
-	std::lock_guard<std::recursive_mutex> lks(m_crit.send);
+}
-	SendToClients(spac);
+
+void NetPlayServer::SendAsyncToClients(sf::Packet* packet)
+{
+	{
+		std::lock_guard<std::recursive_mutex> lkq(m_crit.async_queue_write);
+		m_async_queue.Push(std::unique_ptr<sf::Packet>(packet));
+	}
+	ENetUtil::WakeupThread(m_server);
 }
 
 // called from ---NETPLAY--- thread
@@ -478,11 +481,8 @@ unsigned int NetPlayServer::OnData(sf::Packet& packet, Client& player)
 		spac << player.pid;
 		spac << msg;
 
-		{
-			std::lock_guard<std::recursive_mutex> lks(m_crit.send);
 		SendToClients(spac, player.pid);
 	}
-	}
 	break;
 
 	case NP_MSG_PAD_DATA:
@@ -505,7 +505,6 @@ unsigned int NetPlayServer::OnData(sf::Packet& packet, Client& player)
 		spac << (MessageId)NP_MSG_PAD_DATA;
 		spac << map << pad.button << pad.analogA << pad.analogB << pad.stickX << pad.stickY << pad.substickX << pad.substickY << pad.triggerLeft << pad.triggerRight;
 
-		std::lock_guard<std::recursive_mutex> lks(m_crit.send);
 		SendToClients(spac, player.pid);
 	}
 	break;
@@ -538,7 +537,6 @@ unsigned int NetPlayServer::OnData(sf::Packet& packet, Client& player)
 		for (const u8& byte : data)
 			spac << byte;
 
-		std::lock_guard<std::recursive_mutex> lks(m_crit.send);
 		SendToClients(spac, player.pid);
 	}
 	break;
@@ -559,7 +557,6 @@ unsigned int NetPlayServer::OnData(sf::Packet& packet, Client& player)
 		spac << player.pid;
 		spac << player.ping;
 
-		std::lock_guard<std::recursive_mutex> lks(m_crit.send);
 		SendToClients(spac);
 	}
 	break;
@@ -577,7 +574,6 @@ unsigned int NetPlayServer::OnData(sf::Packet& packet, Client& player)
 		spac << (MessageId)NP_MSG_STOP_GAME;
 
 		std::lock_guard<std::recursive_mutex> lkp(m_crit.players);
-		std::lock_guard<std::recursive_mutex> lks(m_crit.send);
 		SendToClients(spac);
 
 		m_is_running = false;
@@ -601,17 +597,15 @@ void NetPlayServer::OnTraversalStateChanged()
 		m_dialog->Update();
 }
 
-// called from ---GUI--- thread / and ---NETPLAY--- thread
+// called from ---GUI--- thread
 void NetPlayServer::SendChatMessage(const std::string& msg)
 {
-	sf::Packet spac;
+	sf::Packet* spac = new sf::Packet;
-	spac << (MessageId)NP_MSG_CHAT_MESSAGE;
+	*spac << (MessageId)NP_MSG_CHAT_MESSAGE;
-	spac << (PlayerId)0; // server id always 0
+	*spac << (PlayerId)0; // server id always 0
-	spac << msg;
+	*spac << msg;
 
-	std::lock_guard<std::recursive_mutex> lkp(m_crit.players);
+	SendAsyncToClients(spac);
-	std::lock_guard<std::recursive_mutex> lks(m_crit.send);
-	SendToClients(spac);
 }
 
 // called from ---GUI--- thread
@@ -622,13 +616,11 @@ bool NetPlayServer::ChangeGame(const std::string &game)
 	m_selected_game = game;
 
 	// send changed game to clients
-	sf::Packet spac;
+	sf::Packet* spac = new sf::Packet;
-	spac << (MessageId)NP_MSG_CHANGE_GAME;
+	*spac << (MessageId)NP_MSG_CHANGE_GAME;
-	spac << game;
+	*spac << game;
 
-	std::lock_guard<std::recursive_mutex> lkp(m_crit.players);
+	SendAsyncToClients(spac);
-	std::lock_guard<std::recursive_mutex> lks(m_crit.send);
-	SendToClients(spac);
 
 	return true;
 }
@@ -651,24 +643,22 @@ bool NetPlayServer::StartGame()
 	g_netplay_initial_gctime = Common::Timer::GetLocalTimeSinceJan1970();
 
 	// tell clients to start game
-	sf::Packet spac;
+	sf::Packet* spac = new sf::Packet;
-	spac << (MessageId)NP_MSG_START_GAME;
+	*spac << (MessageId)NP_MSG_START_GAME;
-	spac << m_current_game;
+	*spac << m_current_game;
-	spac << m_settings.m_CPUthread;
+	*spac << m_settings.m_CPUthread;
-	spac << m_settings.m_CPUcore;
+	*spac << m_settings.m_CPUcore;
-	spac << m_settings.m_DSPEnableJIT;
+	*spac << m_settings.m_DSPEnableJIT;
-	spac << m_settings.m_DSPHLE;
+	*spac << m_settings.m_DSPHLE;
-	spac << m_settings.m_WriteToMemcard;
+	*spac << m_settings.m_WriteToMemcard;
-	spac << m_settings.m_OCEnable;
+	*spac << m_settings.m_OCEnable;
-	spac << m_settings.m_OCFactor;
+	*spac << m_settings.m_OCFactor;
-	spac << m_settings.m_EXIDevice[0];
+	*spac << m_settings.m_EXIDevice[0];
-	spac << m_settings.m_EXIDevice[1];
+	*spac << m_settings.m_EXIDevice[1];
-	spac << (u32)g_netplay_initial_gctime;
+	*spac << (u32)g_netplay_initial_gctime;
-	spac << (u32)g_netplay_initial_gctime << 32;
+	*spac << (u32)g_netplay_initial_gctime << 32;
 
-	std::lock_guard<std::recursive_mutex> lkp(m_crit.players);
+	SendAsyncToClients(spac);
-	std::lock_guard<std::recursive_mutex> lks(m_crit.send);
-	SendToClients(spac);
 
 	m_is_running = true;
 

Source/Core/Core/NetPlayServer.h

@@ -9,6 +9,7 @@
 #include <sstream>
 #include <unordered_set>
 #include <SFML/Network/Packet.hpp>
+#include "Common/ENetUtil.h"
 #include "Common/Thread.h"
 #include "Common/Timer.h"
 #include "Common/TraversalClient.h"
@@ -20,6 +21,7 @@ class NetPlayServer : public TraversalClientClient
 {
 public:
 	void ThreadFunc();
+	void SendAsyncToClients(sf::Packet* packet);
 
 	NetPlayServer(const u16 port, bool traversal, std::string centralServer, u16 centralPort);
 	~NetPlayServer();
@@ -101,11 +103,13 @@ private:
 	{
 		std::recursive_mutex game;
 		// lock order
-		std::recursive_mutex players, send;
+		std::recursive_mutex players;
+		std::recursive_mutex async_queue_write;
 	} m_crit;
 
 	std::string m_selected_game;
 	std::thread m_thread;
+	Common::FifoQueue<std::unique_ptr<sf::Packet>, false> m_async_queue;
 
 	ENetHost*        m_server;
 	TraversalClient* m_traversal_client;

Source/Core/DolphinWX/NetWindow.cpp

@@ -360,10 +360,10 @@ void NetPlaySetupDiag::OnHost(wxCommandEvent&)
 	unsigned long centralPort = 0;
 	m_traversal_port->GetValue().ToULong(&centralPort);
 	netplay_server = new NetPlayServer(u16(port), trav, WxStrToStr(m_traversal_server->GetValue()), u16(centralPort));
-	netplay_server->ChangeGame(game);
-	netplay_server->AdjustPadBufferSize(INITIAL_PAD_BUFFER_SIZE);
 	if (netplay_server->is_connected)
 	{
+		netplay_server->ChangeGame(game);
+		netplay_server->AdjustPadBufferSize(INITIAL_PAD_BUFFER_SIZE);
 #ifdef USE_UPNP
 		if (m_upnp_chk->GetValue())
 			netplay_server->TryPortmapping(port);