feat(voice): relay voice frames on the network thread

Source/include/mafianet/peer.h

@@ -42,6 +42,7 @@
 #include "mafianet/lockless_types.h"
 #include "mafianet/ds_queue.h"
 #include <atomic>
+#include <thread>
 
 namespace MafiaNet {
 /// Forward declarations
@@ -994,6 +995,19 @@ class RAK_DLL_EXPORT RakPeer : public RakPeerInterface, public RNS2EventHandler
 	DataStructures::List<BanStruct*> banList;
 	// Threadsafe, and not thread safe
 	DataStructures::List<PluginInterface2*> pluginListTS, pluginListNTS;
+	// Every attached plugin again, for OnReceiveNetworkThread. The mutex lets AttachPlugin and
+	// DetachPlugin run while the update thread is dispatching.
+	DataStructures::List<PluginInterface2*> pluginListNetworkThread;
+	SimpleMutex pluginListNetworkThreadMutex;
+	// The thread inside DispatchNetworkThreadReceive. AttachPlugin/DetachPlugin called from a hook
+	// on it are ignored: they would deadlock on the lock that thread holds, and they also change
+	// the plugin lists Receive() walks on the user thread.
+	std::atomic<std::thread::id> networkThreadDispatcher;
+	// Set once Shutdown() starts telling plugins, cleared again once its update thread has exited.
+	// Starts false, so the first Startup() offers messages. Guarded by the mutex above.
+	bool networkThreadReceiveStopped;
+	bool IsInsideNetworkThreadReceive(void) const;
+	void SetNetworkThreadReceiveStopped(bool stopped);
 
 	DataStructures::Queue<RequestedConnectionStruct*> requestedConnectionQueue;
 	SimpleMutex requestedConnectionQueueMutex;
@@ -1151,6 +1165,8 @@ class RAK_DLL_EXPORT RakPeer : public RakPeerInterface, public RNS2EventHandler
 	void ResetSendReceipt(void);
 	void OnConnectedPong(MafiaNet::Time sendPingTime, MafiaNet::Time sendPongTime, RemoteSystemStruct *remoteSystem);
 	void CallPluginCallbacks(DataStructures::List<PluginInterface2*> &pluginList, Packet *packet);
+	// Offers a message to the network-thread plugins; true if one of them took it.
+	bool DispatchNetworkThreadReceive(Packet *packet);
 
 #if LIBCAT_SECURITY==1
 	// Encryption and security

Source/include/mafianet/plugin_interface2.h

@@ -139,6 +139,22 @@ class RAK_DLL_EXPORT PluginInterface2
 	/// If true, then you cannot call RakPeer::AttachPlugin() or RakPeer::DetachPlugin() for this plugin, while RakPeer is active
 	virtual bool UsesReliabilityLayer(void) const {return false;}
 
+	/// Called on RakPeer's network update thread for each message from a connected system, before
+	/// it is queued for RakPeer::Receive(). Use it for traffic that must not wait for the
+	/// application to call Receive(), such as voice. Connection notifications are not routed here.
+	/// Runs concurrently with the user thread, so anything the plugin shares with its other
+	/// callbacks must be synchronized. RakPeer::Send() is safe to call from here; AttachPlugin() and
+	/// DetachPlugin() do nothing when called from here, and Shutdown() must not be called from here.
+	/// Not called once Shutdown() has begun calling OnRakPeerShutdown(). ID_TIMESTAMP payloads arrive
+	/// unshifted. A sender's messages can arrive here before the user thread has processed its
+	/// connection notification, so do not assume the application already knows the sender.
+	/// DetachPlugin() and Shutdown() wait for a call in progress, so this must never wait on a
+	/// thread that may be inside either of them.
+	/// \param[in] packet The message. RR_STOP_PROCESSING_AND_DEALLOCATE frees it, RR_STOP_PROCESSING
+	///            hands it to the plugin (free it with RakPeer::DeallocatePacket), and
+	///            RR_CONTINUE_PROCESSING queues it for Receive() as usual.
+	virtual PluginReceiveResult OnReceiveNetworkThread(Packet *packet) {(void) packet; return RR_CONTINUE_PROCESSING;}
+
 	/// Called on a send to the socket, per datagram, that does not go through the reliability layer
 	/// \pre To be called, UsesReliabilityLayer() must return true
 	/// \param[in] data The data being sent

Source/include/mafianet/voice.h

@@ -24,6 +24,11 @@
 #include "mafianet/plugin_interface2.h"
 #include "mafianet/ds_ordered_list.h"
 #include "mafianet/native_types.h"
+#include "mafianet/simple_mutex.h"
+
+#include <atomic>
+#include <functional>
+#include <thread>
 
 // Forward declarations for Opus and RNNoise
 struct OpusEncoder;
@@ -306,11 +311,20 @@ class RAK_DLL_EXPORT RakVoice : public PluginInterface2
 	/// \return false, leaving both unchanged, if either value is outside 0..NUMBER_OF_ORDERED_STREAMS-1.
 	bool SetOrderingChannels(char frameChannel, char controlChannel);
 
-	/// \brief Marks this peer as the relay host (the server).
-	/// A relay host never decodes: incoming ID_RAKVOICE_RELAY_DATA packets are passed
-	/// through to the application loop so they can be handed back to RelayFrame().
+	/// \brief Makes this peer the relay host (the server), relaying with \a relay.
+	/// A relay host never decodes. Each incoming ID_RAKVOICE_RELAY_DATA packet is handed to
+	/// \a relay on RakPeer's network update thread and freed afterwards, so frames are forwarded
+	/// even while the application's main thread is busy, and never reach Receive().
+	/// \a relay typically picks recipients and calls RelayFrame(), which is safe there; it must
+	/// not keep the packet, and whatever it reads must be safe to read from that thread. It can see
+	/// a sender before the application has processed that sender's connection, and must never
+	/// wait on a thread that may be inside SetRelayHost(), DetachPlugin() or Shutdown(), which
+	/// wait for it.
+	/// Pass nullptr to stop being a relay host. Once this returns, the previous \a relay is not
+	/// running and will not be called again. Called from inside \a relay, the change takes
+	/// effect as soon as that call returns.
 	/// Distinct from SetRelayMode(), which is what a talking/listening client enables.
-	void SetRelayHost(bool enable);
+	void SetRelayHost(std::function<void(Packet *packet)> relay);
 
 	/// \brief Forwards a received relay frame to a set of recipients without decoding it.
 	/// Costs one Send per recipient; no codec state is touched.
@@ -352,6 +366,7 @@ class RAK_DLL_EXPORT RakVoice : public PluginInterface2
 	virtual void OnShutdown(void);
 	virtual void Update(void);
 	virtual PluginReceiveResult OnReceive(Packet *packet);
+	virtual PluginReceiveResult OnReceiveNetworkThread(Packet *packet);
 	virtual void OnClosedConnection(const SystemAddress &systemAddress, RakNetGUID rakNetGUID, PI2_LostConnectionReason lostConnectionReason );
 
 protected:
@@ -405,7 +420,15 @@ class RAK_DLL_EXPORT RakVoice : public PluginInterface2
 	int defaultSignalType;
 	bool loopbackMode;
 	bool relayMode;
-	bool relayHost;
+	// Set together with relayFrameHandler; read on the network thread by OnReceiveNetworkThread.
+	std::atomic<bool> relayHost;
+	std::function<void(Packet *packet)> relayFrameHandler;
+	MafiaNet::SimpleMutex relayFrameHandlerMutex;
+	// The thread running relayFrameHandler, and a SetRelayHost() it made from inside: applied once
+	// the handler returns, since replacing a std::function while it runs destroys it mid-call.
+	std::atomic<std::thread::id> relayThread;
+	std::function<void(Packet *packet)> pendingRelay;
+	bool hasPendingRelay;
 	bool perSpeakerOutput;
 	/// Concurrent relay decoders the application allows; 0 means only the memory backstop.
 	unsigned maxDecodedSpeakers;

Source/src/peer.cpp

@@ -245,6 +245,7 @@ RakPeer::RakPeer()
 	remoteSystemLookup=0;
 	bytesSentPerSecond = bytesReceivedPerSecond = 0;
 	endThreads = true;
+	networkThreadReceiveStopped = false;
 	isMainLoopThreadActive = false;
 	socketFactory = nullptr;
 	manualUpdate = false;
@@ -1055,6 +1056,10 @@ void RakPeer::Shutdown( unsigned int blockDuration, unsigned char orderingChanne
 			time = MafiaNet::GetTimeMS();
 		}
 	}
+	// Plugins tear down in OnRakPeerShutdown while the update thread may still be running; stop
+	// offering them messages there first, so no hook is mid-call or starts after this point.
+	SetNetworkThreadReceiveStopped(true);
+
 	for (i=0; i < pluginListTS.Size(); i++)
 	{
 		pluginListTS[i]->OnRakPeerShutdown();
@@ -1093,6 +1098,11 @@ void RakPeer::Shutdown( unsigned int blockDuration, unsigned char orderingChanne
 		RakSleep(15);
 	}
 
+	// The update thread has exited, so nothing can be in or enter the hook: re-arm it for the next
+	// Startup(). Re-arming here rather than in Startup() means the constructor's value is the one
+	// the first Startup() runs with.
+	SetNetworkThreadReceiveStopped(false);
+
 	activeSystemListSize = 0;
 
 	/*
@@ -2910,6 +2920,9 @@ void RakPeer::SendTTL( const char* host, unsigned short remotePort, int ttl, uns
 // --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
 void RakPeer::AttachPlugin( PluginInterface2 *plugin )
 {
+	if (IsInsideNetworkThreadReceive())
+		return;
+
 	bool isNotThreadsafe = plugin->UsesReliabilityLayer();
 	if (isNotThreadsafe)
 	{
@@ -2929,6 +2942,11 @@ void RakPeer::AttachPlugin( PluginInterface2 *plugin )
 			pluginListTS.Insert(plugin, _FILE_AND_LINE_);
 		}
 	}
+
+	pluginListNetworkThreadMutex.Lock();
+	if (pluginListNetworkThread.GetIndexOf(plugin)==MAX_UNSIGNED_LONG)
+		pluginListNetworkThread.Insert(plugin, _FILE_AND_LINE_);
+	pluginListNetworkThreadMutex.Unlock();
 }
 
 // --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
@@ -2938,7 +2956,7 @@ void RakPeer::AttachPlugin( PluginInterface2 *plugin )
 // --------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
 void RakPeer::DetachPlugin( PluginInterface2 *plugin )
 {
-	if (plugin==0)
+	if (plugin==0 || IsInsideNetworkThreadReceive())
 		return;
 
 	unsigned int index;
@@ -2964,6 +2982,17 @@ void RakPeer::DetachPlugin( PluginInterface2 *plugin )
 			pluginListTS.RemoveFromEnd();
 		}
 	}
+	// Removed under the lock the update thread dispatches under, so once this returns the plugin
+	// is not inside OnReceiveNetworkThread and will not be called again.
+	pluginListNetworkThreadMutex.Lock();
+	index = pluginListNetworkThread.GetIndexOf(plugin);
+	if (index!=MAX_UNSIGNED_LONG)
+	{
+		pluginListNetworkThread[index]=pluginListNetworkThread[pluginListNetworkThread.Size()-1];
+		pluginListNetworkThread.RemoveFromEnd();
+	}
+	pluginListNetworkThreadMutex.Unlock();
+
 	plugin->OnDetach();
 	plugin->SetRakPeerInterface(0);
 }
@@ -4835,6 +4864,39 @@ void RakPeer::ClearRequestedConnectionList(void)
 		MafiaNet::OP_DELETE(freeQueue[i], _FILE_AND_LINE_ );
 	}
 }
+bool RakPeer::IsInsideNetworkThreadReceive(void) const
+{
+	return networkThreadDispatcher.load()==std::this_thread::get_id();
+}
+
+void RakPeer::SetNetworkThreadReceiveStopped(bool stopped)
+{
+	pluginListNetworkThreadMutex.Lock();
+	networkThreadReceiveStopped = stopped;
+	pluginListNetworkThreadMutex.Unlock();
+}
+
+bool RakPeer::DispatchNetworkThreadReceive(Packet *packet)
+{
+	pluginListNetworkThreadMutex.Lock();
+	if (networkThreadReceiveStopped)
+	{
+		pluginListNetworkThreadMutex.Unlock();
+		return false;
+	}
+
+	networkThreadDispatcher = std::this_thread::get_id();
+	PluginReceiveResult result = RR_CONTINUE_PROCESSING;
+	for (unsigned int i=0; i < pluginListNetworkThread.Size() && result==RR_CONTINUE_PROCESSING; i++)
+		result = pluginListNetworkThread[i]->OnReceiveNetworkThread(packet);
+	networkThreadDispatcher = std::thread::id();
+	pluginListNetworkThreadMutex.Unlock();
+
+	if (result==RR_STOP_PROCESSING_AND_DEALLOCATE)
+		DeallocatePacket(packet);
+	return result!=RR_CONTINUE_PROCESSING;
+}
+
 inline void RakPeer::AddPacketToProducer(MafiaNet::Packet *p)
 {
 	packetReturnMutex.Lock();
@@ -6935,7 +6997,8 @@ bool RakPeer::RunUpdateCycle(BitStream &updateBitStream )
 							packet->systemAddress.systemIndex = remoteSystem->remoteSystemIndex;
 							packet->guid = remoteSystem->guid;
 							packet->guid.systemIndex=packet->systemAddress.systemIndex;
-							AddPacketToProducer(packet);
+							if (!DispatchNetworkThreadReceive(packet))
+								AddPacketToProducer(packet);
 						}
 						else
 						{

Source/src/voice.cpp

@@ -66,6 +66,7 @@ RakVoice::RakVoice()
 	bufferSizeBytes = 0;
 	relayMode = false;
 	relayHost = false;
+	hasPendingRelay = false;
 	perSpeakerOutput = false;
 	maxDecodedSpeakers = 0;
 	relayTarget = UNASSIGNED_RAKNET_GUID;
@@ -158,9 +159,22 @@ void RakVoice::SetRelayTarget(RakNetGUID server)
 	relayTarget = server;
 }
 
-void RakVoice::SetRelayHost(bool enable)
+void RakVoice::SetRelayHost(std::function<void(Packet *packet)> relay)
 {
-	relayHost = enable;
+	// Swapped under the lock OnReceiveNetworkThread calls it under, so a relay being replaced is
+	// never mid-call when this returns.
+	if (relayThread.load() == std::this_thread::get_id())
+	{
+		// From inside the running relay, which holds the lock: swap once it returns.
+		pendingRelay = std::move(relay);
+		hasPendingRelay = true;
+		return;
+	}
+
+	relayFrameHandlerMutex.Lock();
+	relayHost = static_cast<bool>(relay);
+	relayFrameHandler = std::move(relay);
+	relayFrameHandlerMutex.Unlock();
 }
 
 bool RakVoice::SetOrderingChannels(char frameChannel, char controlChannel)
@@ -741,18 +755,43 @@ PluginReceiveResult RakVoice::OnReceive(Packet *packet)
 		OnVoiceData(packet);
 		return RR_STOP_PROCESSING_AND_DEALLOCATE;
 	case ID_RAKVOICE_RELAY_DATA:
-		// A relay host never decodes. Plugin OnReceive runs inside RakPeer::Receive, before
-		// the packet reaches the application loop, so swallowing it here would leave the
-		// server with nothing to hand to RelayFrame(). Pass it through instead.
+		// A relay host never decodes. OnReceiveNetworkThread relays these before they are
+		// queued; one only gets here if it was queued just before SetRelayHost() took effect.
 		if (relayHost)
-			return RR_CONTINUE_PROCESSING;
+			return RR_STOP_PROCESSING_AND_DEALLOCATE;
 		OnRelayVoiceData(packet);
 		return RR_STOP_PROCESSING_AND_DEALLOCATE;
 	}
 
 	return RR_CONTINUE_PROCESSING;
 }
 
+PluginReceiveResult RakVoice::OnReceiveNetworkThread(Packet *packet)
+{
+	if (!relayHost || packet == nullptr || packet->length == 0 || packet->data[0] != ID_RAKVOICE_RELAY_DATA)
+		return RR_CONTINUE_PROCESSING;
+
+	relayFrameHandlerMutex.Lock();
+	// Re-checked under the lock: SetRelayHost(nullptr) may have run since the test above.
+	if (!relayFrameHandler)
+	{
+		relayFrameHandlerMutex.Unlock();
+		return RR_CONTINUE_PROCESSING;
+	}
+	relayThread = std::this_thread::get_id();
+	relayFrameHandler(packet);
+	relayThread = std::thread::id();
+	if (hasPendingRelay)
+	{
+		relayHost = static_cast<bool>(pendingRelay);
+		relayFrameHandler = std::move(pendingRelay);
+		pendingRelay = nullptr;
+		hasPendingRelay = false;
+	}
+	relayFrameHandlerMutex.Unlock();
+	return RR_STOP_PROCESSING_AND_DEALLOCATE;
+}
+
 void RakVoice::OnClosedConnection(const SystemAddress &systemAddress, RakNetGUID rakNetGUID, PI2_LostConnectionReason lostConnectionReason)
 {
 	(void)systemAddress;