fix(peer): join internal threads on shutdown instead of abandoning them

Source/include/mafianet/DS_ThreadsafeAllocatingQueue.h

@@ -70,15 +70,16 @@ void ThreadsafeAllocatingQueue<structureType>::Push(structureType *s)
 template <class structureType>
 structureType *ThreadsafeAllocatingQueue<structureType>::PopInaccurate(void)
 {
+	// Historically this probed queue.IsEmpty() without the mutex as a fast path
+	// ("inaccurate"); that unlocked read raced with Push from other threads. An
+	// uncontended lock is cheap enough that the fast path isn't worth the UB.
 	structureType *s;
-	if (queue.IsEmpty())
-		return 0;
 	queueMutex.Lock();
 	if (queue.IsEmpty()==false)
 		s=queue.Pop();
 	else
 		s=0;
-	queueMutex.Unlock();	
+	queueMutex.Unlock();
 	return s;
 }
 

Source/include/mafianet/LocklessTypes.h

@@ -18,11 +18,10 @@
 
 #include "Export.h"
 #include "NativeTypes.h"
+// Kept for transitive users even though the atomics below no longer need it.
 #include "WindowsIncludes.h"
-#if defined(ANDROID) || defined(__S3E__) || defined(__APPLE__)
-// __sync_fetch_and_add not supported apparently
-#include "SimpleMutex.h"
-#endif
+
+#include <atomic>
 
 namespace MafiaNet
 {
@@ -36,18 +35,10 @@ class RAK_DLL_EXPORT LocklessUint32_t
 	uint32_t Increment(void);
 	// Returns variable value after changing it
 	uint32_t Decrement(void);
-	uint32_t GetValue(void) const {return value;}
+	uint32_t GetValue(void) const {return value.load();}
 
 protected:
-#ifdef _WIN32
-	volatile LONG value;
-#elif defined(ANDROID) || defined(__S3E__) || defined(__APPLE__)
-	// __sync_fetch_and_add not supported apparently
-	SimpleMutex mutex;
-	uint32_t value;
-#else
-	volatile uint32_t value;
-#endif
+	std::atomic<uint32_t> value;
 };
 
 }

Source/include/mafianet/peer.h

@@ -42,6 +42,8 @@
 #include "LocklessTypes.h"
 #include "DS_Queue.h"
 
+#include <atomic>
+
 namespace MafiaNet {
 /// Forward declarations
 class HuffmanEncodingTree;
@@ -761,11 +763,16 @@ class RAK_DLL_EXPORT RakPeer : public RakPeerInterface, public RNS2EventHandler
 	bool IsLoopbackAddress(const AddressOrGUID &systemIdentifier, bool matchPort) const;
 	SystemAddress GetLoopbackAddress(void) const;
 
-	///Set this to true to terminate the Peer thread execution 
-	volatile bool endThreads;
-	///true if the peer thread is active. 
-	volatile bool isMainLoopThreadActive;
-
+	///Set this to true to terminate the Peer thread execution
+	std::atomic<bool> endThreads;
+	///true if the peer thread is active.
+	std::atomic<bool> isMainLoopThreadActive;
+	/// Joinable handle for the update/network thread. Shutdown() joins it so the
+	/// thread has fully exited -- with all its writes visible -- before any
+	/// connection state is torn down (issue #7).
+	RakThread::ThreadHandle updateThread;
+	bool updateThreadJoinable;
+
 	// MafiaNet::LocklessUint32_t isRecvFromLoopThreadActive;
 
 

Source/include/mafianet/socket2.h

@@ -23,6 +23,8 @@
 #include "DS_ThreadsafeAllocatingQueue.h"
 #include "Export.h"
 
+#include <atomic>
+
 // Batched datagram I/O (recvmmsg / sendmmsg) is guarded by a plain
 // `#if defined(__linux__)` wherever it appears. There is no macro and no build
 // option for it: the syscalls exist on Linux and nowhere else MafiaNet targets,
@@ -240,7 +242,13 @@ class RNS2_Berkley : public IRNS2_Berkley
 	// same guard, and only defined on that platform.
 	void RecvFromBatchedLoop(void);
 	MafiaNet::LocklessUint32_t isRecvFromLoopThreadActive;
-	volatile bool endThreads;
+	std::atomic<bool> endThreads;
+	// The recv polling thread is joinable so teardown can wait for it to fully
+	// exit before the socket (and the RakPeer it calls back into) are freed.
+	// A detached thread with a bounded wait allowed a leaked thread to keep
+	// dereferencing both after Shutdown (issue #7).
+	RakThread::ThreadHandle recvThread;
+	bool recvThreadJoinable;
 	// Constructor not called!
 
 #if defined(__APPLE__)

Source/include/mafianet/thread.h

@@ -18,6 +18,10 @@
 
 #include "Export.h"
 
+#if !defined(_WIN32)
+#include <pthread.h>
+#endif
+
 namespace MafiaNet
 {
 /// To define a thread, use RAK_THREAD_DECLARATION(functionName);
@@ -50,6 +54,30 @@ class RAK_DLL_EXPORT RakThread
 #else
 	static int Create( void* start_address( void* ), void *arglist, int priority=0);
 #endif
+
+	/// Handle to a joinable thread created with CreateJoinable(). Must be
+	/// reaped with Join() exactly once, or the thread's resources leak.
+#if defined(_WIN32)
+	typedef void *ThreadHandle; // HANDLE
+#else
+	typedef pthread_t ThreadHandle;
+#endif
+
+	/// Like Create(), but the thread is joinable: the caller receives a handle
+	/// and MUST call Join() on it. Join() blocks until the thread function has
+	/// returned and establishes a happens-before edge with all of the thread's
+	/// writes -- use this for threads whose owner frees state the thread uses.
+	/// \param[out] handle Receives the thread handle on success; unchanged on failure.
+	/// \return 0=success. >0 = error code
+#if defined(_WIN32)
+	static int CreateJoinable( unsigned __stdcall start_address( void* ), void *arglist, ThreadHandle *handle, int priority=0);
+#else
+	static int CreateJoinable( void* start_address( void* ), void *arglist, ThreadHandle *handle, int priority=0);
+#endif
+
+	/// Block until the thread behind \a handle has fully exited, then release
+	/// the handle. Call exactly once per CreateJoinable().
+	static void Join( ThreadHandle handle );
 };
 
 }

Source/src/GetTime.cpp

@@ -46,10 +46,8 @@
 #else
 #include <sys/time.h>
 #include <unistd.h>
-MafiaNet::TimeUS initialTime;
 #endif
 
-static bool initialized=false;
 
 #if defined(GET_TIME_SPIKE_LIMIT) && GET_TIME_SPIKE_LIMIT>0
 #include "mafianet/SimpleMutex.h"
@@ -141,21 +139,9 @@ MafiaNet::TimeMS MafiaNet::GetTimeMS( void )
 #if   defined(_WIN32)
 MafiaNet::TimeUS GetTimeUS_Windows( void )
 {
-	if ( initialized == false)
-	{
-		initialized = true;
-
-		// Save the current process
-//		HANDLE mProc = GetCurrentProcess();
-
-		// Get the current Affinity
-#if defined (_M_X64)
-//		GetProcessAffinityMask(mProc, (PDWORD_PTR)&mProcMask, (PDWORD_PTR)&mSysMask);
-#else
-//		GetProcessAffinityMask(mProc, &mProcMask, &mSysMask);
-#endif
-//		mThread = GetCurrentThread();
-	}	
+	// A first-call `initialized` guard used to live here; its body was entirely
+	// commented out, so all it did was write a non-atomic flag from every
+	// calling thread (a data race). Removed.
 
 	// 9/26/2010 In China running LuDaShi, QueryPerformanceFrequency has to be called every time because CPU clock speeds can be different
 	MafiaNet::TimeUS curTime;
@@ -177,16 +163,21 @@ MafiaNet::TimeUS GetTimeUS_Windows( void )
 #endif // #if defined(GET_TIME_SPIKE_LIMIT) && GET_TIME_SPIKE_LIMIT>0
 }
 #elif defined(__GNUC__)  || defined(__GCCXML__) || defined(__S3E__)
+static MafiaNet::TimeUS GetInitialTime_Linux( void )
+{
+	timeval tp;
+	gettimeofday( &tp, 0 );
+	return ( tp.tv_sec ) * (MafiaNet::TimeUS) 1000000 + ( tp.tv_usec );
+}
+
 MafiaNet::TimeUS GetTimeUS_Linux( void )
 {
 	timeval tp;
-	if ( initialized == false)
-	{
-		gettimeofday( &tp, 0 );
-		initialized=true;
-		// I do this because otherwise MafiaNet::Time in milliseconds won't work as it will underflow when dividing by 1000 to do the conversion
-		initialTime = ( tp.tv_sec ) * (MafiaNet::TimeUS) 1000000 + ( tp.tv_usec );
-	}
+	// Thread-safe first-use initialization (C++11 magic static). Every RakPeer
+	// startup spawns threads that call this concurrently; a plain lazy-init
+	// bool/global raced, letting a thread read a torn or stale base time.
+	// I subtract an initial time because otherwise MafiaNet::Time in milliseconds won't work as it will underflow when dividing by 1000 to do the conversion
+	static const MafiaNet::TimeUS initialTime = GetInitialTime_Linux();
 
 	// GCC
 	MafiaNet::TimeUS curTime;

Source/src/LocklessTypes.cpp

@@ -3,7 +3,7 @@
  *  All rights reserved.
  *
  *  This source code is licensed under the BSD-style license found in the
- *  RakNet License.txt file in the licenses directory of this source tree. An additional grant 
+ *  RakNet License.txt file in the licenses directory of this source tree. An additional grant
  *  of patent rights can be found in the RakNet Patents.txt file in the same directory.
  *
  *
@@ -17,6 +17,13 @@
 
 using namespace MafiaNet;
 
+// std::atomic replaces the previous per-platform mix of InterlockedIncrement,
+// mutex-guarded arithmetic, and __sync_fetch_and_add. Besides being standard,
+// this fixes two defects of the old code: GetValue() read the counter without
+// any synchronization (a data race with concurrent modifications), and the
+// __sync_fetch_and_add branch returned the value from *before* the change
+// while every other platform returned the value after it.
+
 LocklessUint32_t::LocklessUint32_t()
 {
 	value=0;
@@ -27,31 +34,9 @@ LocklessUint32_t::LocklessUint32_t(uint32_t initial)
 }
 uint32_t LocklessUint32_t::Increment(void)
 {
-#ifdef _WIN32
-	return (uint32_t) InterlockedIncrement(&value);
-#elif defined(ANDROID) || defined(__S3E__) || defined(__APPLE__)
-	uint32_t v;
-	mutex.Lock();
-	++value;
-	v=value;
-	mutex.Unlock();
-	return v;
-#else
-	return __sync_fetch_and_add (&value, (uint32_t) 1);
-#endif
+	return value.fetch_add(1)+1;
 }
 uint32_t LocklessUint32_t::Decrement(void)
 {
-#ifdef _WIN32
-	return (uint32_t) InterlockedDecrement(&value);
-#elif defined(ANDROID) || defined(__S3E__) || defined(__APPLE__)
-	uint32_t v;
-	mutex.Lock();
-	--value;
-	v=value;
-	mutex.Unlock();
-	return v;
-#else
-	return __sync_fetch_and_add (&value, (uint32_t) -1);
-#endif
+	return value.fetch_sub(1)-1;
 }

Source/src/RakNetSocket2.cpp

@@ -212,6 +212,8 @@ unsigned RNS2_Berkley::RecvFromLoopInt(void)
 RNS2_Berkley::RNS2_Berkley()
 {
 	rns2Socket=(RNS2Socket)INVALID_SOCKET;
+	endThreads=false;
+	recvThreadJoinable=false;
 }
 RNS2_Berkley::~RNS2_Berkley()
 {
@@ -231,7 +233,25 @@ int RNS2_Berkley::CreateRecvPollingThread(int threadPriority)
 {
 	endThreads=false;
 
-	int errorCode = MafiaNet::RakThread::Create(RecvFromLoop, this, threadPriority);
+	// Bound the time a blocking recvfrom/recvmmsg can sit in the kernel so the
+	// polling thread re-checks endThreads even if the wake-up datagram sent by
+	// BlockOnStopRecvPollingThread is lost. Both recv loops already treat a
+	// zero/negative return (EAGAIN/WSAETIMEDOUT) as "no data" and loop.
+#if defined(_WIN32)
+	DWORD recvTimeout = 500; // milliseconds
+	setsockopt__(rns2Socket, SOL_SOCKET, SO_RCVTIMEO, (char *) &recvTimeout, sizeof(recvTimeout));
+#else
+	timeval recvTimeout;
+	recvTimeout.tv_sec = 0;
+	recvTimeout.tv_usec = 500000;
+	setsockopt__(rns2Socket, SOL_SOCKET, SO_RCVTIMEO, (char *) &recvTimeout, sizeof(recvTimeout));
+#endif
+
+	// Joinable, not detached: BlockOnStopRecvPollingThread must be able to wait
+	// for the thread to fully exit before the socket and its event handler are
+	// freed (issue #7).
+	int errorCode = MafiaNet::RakThread::CreateJoinable(RecvFromLoop, this, &recvThread, threadPriority);
+	recvThreadJoinable = (errorCode==0);
 	return errorCode;
 }
 void RNS2_Berkley::SignalStopRecvPollingThread(void)
@@ -242,7 +262,11 @@ void RNS2_Berkley::BlockOnStopRecvPollingThread(void)
 {
 	endThreads=true;
 
-	// Get recvfrom to unblock
+	if (recvThreadJoinable==false)
+		return;
+
+	// Get recvfrom to unblock promptly (SO_RCVTIMEO bounds the wait even if
+	// this datagram is lost)
 	RNS2_SendParameters bsp;
 	unsigned long zero=0;
 	bsp.data=(char*) &zero;
@@ -251,13 +275,19 @@ void RNS2_Berkley::BlockOnStopRecvPollingThread(void)
 	bsp.ttl=0;
 	Send(&bsp, _FILE_AND_LINE_);
 
-	MafiaNet::TimeMS timeout = MafiaNet::GetTimeMS()+1000;
-	while ( isRecvFromLoopThreadActive.GetValue()>0 && MafiaNet::GetTimeMS()<timeout )
+	while ( isRecvFromLoopThreadActive.GetValue()>0 )
 	{
 		// Get recvfrom to unblock
 		Send(&bsp, _FILE_AND_LINE_);
 		RakSleep(30);
 	}
+
+	// Never abandon the thread on a deadline: the caller frees this socket and
+	// the RakPeer the thread calls back into right after we return, so leaving
+	// the thread running was a use-after-free (issue #7). Join also gives the
+	// happens-before edge that makes the thread's writes visible.
+	MafiaNet::RakThread::Join(recvThread);
+	recvThreadJoinable=false;
 }
 const RNS2_BerkleyBindParameters *RNS2_Berkley::GetBindings(void) const {return &binding;}
 RNS2Socket RNS2_Berkley::GetSocket(void) const {return rns2Socket;}

Source/src/RakNetSocket2_Berkley.cpp

@@ -452,7 +452,8 @@ void RNS2_Berkley::RecvFromBlockingIPV4And6(RNS2RecvStruct *recvFromStruct)
 		DWORD dwIOError = GetLastError();
 		// 10035 = WSAEWOULDBLOCK (expected for non-blocking sockets)
 		// 10054 = WSAECONNRESET (expected for UDP - prior sendto received ICMP port unreachable)
-		if (dwIOError != 10035 && dwIOError != 10054)
+		// 10060 = WSAETIMEDOUT (expected: SO_RCVTIMEO is set so the recv polling thread can re-check shutdown flags)
+		if (dwIOError != 10035 && dwIOError != 10054 && dwIOError != 10060)
 		{
 			LPVOID messageBuffer;
 			FormatMessage( FORMAT_MESSAGE_ALLOCATE_BUFFER | FORMAT_MESSAGE_FROM_SYSTEM | FORMAT_MESSAGE_IGNORE_INSERTS,