26024: fix shutdown crash if run from Apache Arrow

src/Amalgam/Concurrency.h

@@ -44,6 +44,31 @@ namespace Concurrency
 	//standard write lock on a read-write shared mutex
 	typedef std::unique_lock<ReadWriteMutex> WriteLock;
 
+	//backing flag for spinlock
+	//needed only when a lock could be required to release a thread local
+	typedef std::atomic_flag SpinMutex;
+
+	//spinlock for access to SpinMutex
+	//guaranteed to be lock-free, but may require arbitrary CPU cycles to
+	//claim the lock.  RAII semantics.
+	class SpinLock {
+	public:
+		inline SpinLock(SpinMutex &flag) noexcept : flag(flag)
+		{
+			while(flag.test_and_set(std::memory_order_acquire))
+				std::this_thread::yield();
+		}
+
+		SpinLock(const SpinLock &other) = delete;
+
+		inline ~SpinLock() noexcept {
+			flag.clear(std::memory_order_release);
+		}
+
+	private:
+		SpinMutex &flag;
+	};
+
 	//Object to perform scope-based unlocking of a vector of locks of LockType for an existing buffer
 	template<typename LockBufferType>
 	class MultipleLockBufferObject

src/Amalgam/evaluablenode/EvaluableNodeManagement.cpp

@@ -24,11 +24,7 @@ EvaluableNodeManager::~EvaluableNodeManager()
 	Concurrency::WriteLock lock(managerAttributesMutex);
 
 	//clear from any threads
-	LocalAllocationBuffer::IterateFunctionOverRegisteredLabs(
-		[this](LocalAllocationBuffer *lab)
-	{
-		lab->Clear(this);
-	});
+	LocalAllocationBuffer::ClearAllRegisteredLabs(this);
 #else
 	localAllocationBuffer.Clear(this);
 #endif
@@ -73,13 +69,9 @@ void EvaluableNodeManager::CollectGarbage()
 
 	//clear regardless of what's in the buffer
 	localAllocationBuffer.Clear();
-	//clear all threads' local allocation buffers that are using this enm
 #ifdef MULTITHREAD_SUPPORT
-	LocalAllocationBuffer::IterateFunctionOverRegisteredLabs(
-		[this](LocalAllocationBuffer *lab)
-		{
-			lab->Clear(this);
-		});
+	//clear all threads' local allocation buffers that are using this enm
+	LocalAllocationBuffer::ClearAllRegisteredLabs(this);
 #endif
 
 	MarkAllReferencedNodesInUse(firstUnusedNodeIndex);
@@ -123,11 +115,7 @@ void EvaluableNodeManager::CollectGarbageWithConcurrentAccess(Concurrency::ReadL
 		Concurrency::WriteLock write_lock(activeInterpreters->memoryModificationMutex);
 
 		//clear all threads' local allocation buffers that are using this enm
-		LocalAllocationBuffer::IterateFunctionOverRegisteredLabs(
-			[this](LocalAllocationBuffer *lab)
-		{
-			lab->Clear(this);
-		});
+		LocalAllocationBuffer::ClearAllRegisteredLabs(this);
 
 		size_t cur_first_unused_node_index = firstUnusedNodeIndex;
 		//clear firstUnusedNodeIndex to signal to other threads that they won't need to do garbage collection
@@ -141,7 +129,7 @@ void EvaluableNodeManager::CollectGarbageWithConcurrentAccess(Concurrency::ReadL
 		MarkAllReferencedNodesInUse(cur_first_unused_node_index);
 		FreeAllNodesExceptReferencedNodes(cur_first_unused_node_index);
 
-		//wake up remaining threads 
+		//wake up remaining threads
 		{
 			//lock the notification mutex to prevent other threads from waking up and seeing
 			//an outdated state of garbageCollectionThreadSelectionFlag

src/Amalgam/evaluablenode/EvaluableNodeManagement.h

@@ -85,15 +85,15 @@ class EvaluableNodeManager
 			: lastEvaluableNodeManager(nullptr)
 		{
 		#ifdef MULTITHREAD_SUPPORT
-			Concurrency::Lock lock(registryMutex);
+			Concurrency::SpinLock lock(registryMutex);
 			registry.push_back(this);
 		#endif
 		}
 
 		~LocalAllocationBuffer()
 		{
 		#ifdef MULTITHREAD_SUPPORT
-			Concurrency::Lock lock(registryMutex);
+			Concurrency::SpinLock lock(registryMutex);
 
 			auto it = std::find(registry.begin(), registry.end(), this);
 			if(it != registry.end())
@@ -114,6 +114,19 @@ class EvaluableNodeManager
 			lastEvaluableNodeManager = nullptr;
 		}
 
+	#ifdef MULTITHREAD_SUPPORT
+		//removes all EvaluableNodes from all local allocation buffers
+		//it will clear each buffer if only_clear_if_current_enm is nullptr
+		//or if that buffer's lastEvaluableNodeManager is the same as
+		//only_clear_if_current_enm
+		static inline void ClearAllRegisteredLabs(EvaluableNodeManager *only_clear_if_current_enm = nullptr)
+		{
+			Concurrency::SpinLock lock(registryMutex);
+			for(auto lab : LocalAllocationBuffer::registry)
+				lab->Clear(only_clear_if_current_enm);
+		}
+	#endif
+
 		//gets a pointer to the next available node from the local allocation buffer
 		//nullptr if it cannot
 		inline EvaluableNode *AllocNodeIfAvailable(EvaluableNodeManager *cur_enm)
@@ -146,18 +159,7 @@ class EvaluableNodeManager
 			buffer.push_back(en);
 		}
 
-	#ifdef MULTITHREAD_SUPPORT
-		//calls func on all registered local allocation buffers for each thread
-		template<typename Func>
-		static inline void IterateFunctionOverRegisteredLabs(Func func)
-		{
-			Concurrency::Lock lock(registryMutex);
-			for(auto lab : LocalAllocationBuffer::registry)
-				func(lab);
-		}
-	#endif
-
-		// Keeps track of the the last EvaluableNodeManager that accessed 
+		// Keeps track of the the last EvaluableNodeManager that accessed
 		// the local allocation buffer for a each thread.
 		// A given local allocation buffer should only have nodes associated with one manager.
 		// If a different manager accesses the buffer, it is cleared to maintain this invariant.
@@ -171,7 +173,13 @@ class EvaluableNodeManager
 	#ifdef MULTITHREAD_SUPPORT
 		//registry that keeps track of all local allocation buffers
 		static inline std::vector<LocalAllocationBuffer *> registry;
-		static inline Concurrency::SingleMutex registryMutex;
+		//spinlock to manage access to registry
+		//if this is integrated into other libraries, it is possible
+		//individual LABs in thread-local variables will be destroyed after
+		//this object's destructor fires; use a spinlock since regular
+		//std::mutex can be invalidated at destruction but std::atomic_flag
+		//if just an int
+		static inline Concurrency::SpinMutex registryMutex;
 	#endif
 	};
 
@@ -422,7 +430,7 @@ class EvaluableNodeManager
 
 		ShrinkMemoryToCurrentUtilizationWithLock();
 	}
-	
+
 	//frees an EvaluableNode (must be owned by this EvaluableNodeManager)
 	// if place_nodes_in_lab is true, then it will update the local allocation buffer and place nodes in it
 	inline void FreeNode(EvaluableNode *en, bool place_nodes_in_lab = true)
@@ -915,7 +923,7 @@ class EvaluableNodeManager
 	thread_local inline static LocalAllocationBuffer localAllocationBuffer;
 #else
 	inline static LocalAllocationBuffer localAllocationBuffer;
-#endif		
+#endif
 
 	//debug diagnostic variables for localAllocationBuffer
 #ifdef DEBUG_REPORT_LAB_USAGE