mirror of https://github.com/trapexit/mergerfs.git
				
				
			
			You can not select more than 25 topics
			Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
		
		
		
		
		
			
		
			
				
					
					
						
							427 lines
						
					
					
						
							12 KiB
						
					
					
				
			
		
		
		
			
			
			
		
		
	
	
							427 lines
						
					
					
						
							12 KiB
						
					
					
				| // Provides an efficient implementation of a semaphore (LightweightSemaphore). | |
| // This is an extension of Jeff Preshing's sempahore implementation (licensed  | |
| // under the terms of its separate zlib license) that has been adapted and | |
| // extended by Cameron Desrochers. | |
|  | |
| #pragma once | |
|  | |
| #include <cstddef> // For std::size_t | |
| #include <atomic> | |
| #include <type_traits> // For std::make_signed<T> | |
|  | |
| #if defined(_WIN32) | |
| // Avoid including windows.h in a header; we only need a handful of | |
| // items, so we'll redeclare them here (this is relatively safe since | |
| // the API generally has to remain stable between Windows versions). | |
| // I know this is an ugly hack but it still beats polluting the global | |
| // namespace with thousands of generic names or adding a .cpp for nothing. | |
| extern "C" { | |
| 	struct _SECURITY_ATTRIBUTES; | |
| 	__declspec(dllimport) void* __stdcall CreateSemaphoreW(_SECURITY_ATTRIBUTES* lpSemaphoreAttributes, long lInitialCount, long lMaximumCount, const wchar_t* lpName); | |
| 	__declspec(dllimport) int __stdcall CloseHandle(void* hObject); | |
| 	__declspec(dllimport) unsigned long __stdcall WaitForSingleObject(void* hHandle, unsigned long dwMilliseconds); | |
| 	__declspec(dllimport) int __stdcall ReleaseSemaphore(void* hSemaphore, long lReleaseCount, long* lpPreviousCount); | |
| } | |
| #elif defined(__MACH__) | |
| #include <mach/mach.h> | |
| #elif defined(__MVS__) | |
| #include <zos-semaphore.h> | |
| #elif defined(__unix__) | |
| #include <semaphore.h> | |
|  | |
| #if defined(__GLIBC_PREREQ) && defined(_GNU_SOURCE) | |
| #if __GLIBC_PREREQ(2,30) | |
| #define MOODYCAMEL_LIGHTWEIGHTSEMAPHORE_MONOTONIC | |
| #endif | |
| #endif | |
| #endif | |
|  | |
| namespace moodycamel | |
| { | |
| namespace details | |
| { | |
| 
 | |
| // Code in the mpmc_sema namespace below is an adaptation of Jeff Preshing's | |
| // portable + lightweight semaphore implementations, originally from | |
| // https://github.com/preshing/cpp11-on-multicore/blob/master/common/sema.h | |
| // LICENSE: | |
| // Copyright (c) 2015 Jeff Preshing | |
| // | |
| // This software is provided 'as-is', without any express or implied | |
| // warranty. In no event will the authors be held liable for any damages | |
| // arising from the use of this software. | |
| // | |
| // Permission is granted to anyone to use this software for any purpose, | |
| // including commercial applications, and to alter it and redistribute it | |
| // freely, subject to the following restrictions: | |
| // | |
| // 1. The origin of this software must not be misrepresented; you must not | |
| //	claim that you wrote the original software. If you use this software | |
| //	in a product, an acknowledgement in the product documentation would be | |
| //	appreciated but is not required. | |
| // 2. Altered source versions must be plainly marked as such, and must not be | |
| //	misrepresented as being the original software. | |
| // 3. This notice may not be removed or altered from any source distribution. | |
| #if defined(_WIN32) | |
| class Semaphore | |
| { | |
| private: | |
| 	void* m_hSema; | |
| 	 | |
| 	Semaphore(const Semaphore& other) MOODYCAMEL_DELETE_FUNCTION; | |
| 	Semaphore& operator=(const Semaphore& other) MOODYCAMEL_DELETE_FUNCTION; | |
| 
 | |
| public: | |
| 	Semaphore(int initialCount = 0) | |
| 	{ | |
| 		assert(initialCount >= 0); | |
| 		const long maxLong = 0x7fffffff; | |
| 		m_hSema = CreateSemaphoreW(nullptr, initialCount, maxLong, nullptr); | |
| 		assert(m_hSema); | |
| 	} | |
| 
 | |
| 	~Semaphore() | |
| 	{ | |
| 		CloseHandle(m_hSema); | |
| 	} | |
| 
 | |
| 	bool wait() | |
| 	{ | |
| 		const unsigned long infinite = 0xffffffff; | |
| 		return WaitForSingleObject(m_hSema, infinite) == 0; | |
| 	} | |
| 	 | |
| 	bool try_wait() | |
| 	{ | |
| 		return WaitForSingleObject(m_hSema, 0) == 0; | |
| 	} | |
| 	 | |
| 	bool timed_wait(std::uint64_t usecs) | |
| 	{ | |
| 		return WaitForSingleObject(m_hSema, (unsigned long)(usecs / 1000)) == 0; | |
| 	} | |
| 
 | |
| 	void signal(int count = 1) | |
| 	{ | |
| 		while (!ReleaseSemaphore(m_hSema, count, nullptr)); | |
| 	} | |
| }; | |
| #elif defined(__MACH__) | |
| //--------------------------------------------------------- | |
| // Semaphore (Apple iOS and OSX) | |
| // Can't use POSIX semaphores due to http://lists.apple.com/archives/darwin-kernel/2009/Apr/msg00010.html | |
| //--------------------------------------------------------- | |
| class Semaphore | |
| { | |
| private: | |
| 	semaphore_t m_sema; | |
| 
 | |
| 	Semaphore(const Semaphore& other) MOODYCAMEL_DELETE_FUNCTION; | |
| 	Semaphore& operator=(const Semaphore& other) MOODYCAMEL_DELETE_FUNCTION; | |
| 
 | |
| public: | |
| 	Semaphore(int initialCount = 0) | |
| 	{ | |
| 		assert(initialCount >= 0); | |
| 		kern_return_t rc = semaphore_create(mach_task_self(), &m_sema, SYNC_POLICY_FIFO, initialCount); | |
| 		assert(rc == KERN_SUCCESS); | |
| 		(void)rc; | |
| 	} | |
| 
 | |
| 	~Semaphore() | |
| 	{ | |
| 		semaphore_destroy(mach_task_self(), m_sema); | |
| 	} | |
| 
 | |
| 	bool wait() | |
| 	{ | |
| 		return semaphore_wait(m_sema) == KERN_SUCCESS; | |
| 	} | |
| 	 | |
| 	bool try_wait() | |
| 	{ | |
| 		return timed_wait(0); | |
| 	} | |
| 	 | |
| 	bool timed_wait(std::uint64_t timeout_usecs) | |
| 	{ | |
| 		mach_timespec_t ts; | |
| 		ts.tv_sec = static_cast<unsigned int>(timeout_usecs / 1000000); | |
| 		ts.tv_nsec = static_cast<int>((timeout_usecs % 1000000) * 1000); | |
| 
 | |
| 		// added in OSX 10.10: https://developer.apple.com/library/prerelease/mac/documentation/General/Reference/APIDiffsMacOSX10_10SeedDiff/modules/Darwin.html | |
| 		kern_return_t rc = semaphore_timedwait(m_sema, ts); | |
| 		return rc == KERN_SUCCESS; | |
| 	} | |
| 
 | |
| 	void signal() | |
| 	{ | |
| 		while (semaphore_signal(m_sema) != KERN_SUCCESS); | |
| 	} | |
| 
 | |
| 	void signal(int count) | |
| 	{ | |
| 		while (count-- > 0) | |
| 		{ | |
| 			while (semaphore_signal(m_sema) != KERN_SUCCESS); | |
| 		} | |
| 	} | |
| }; | |
| #elif defined(__unix__) || defined(__MVS__) | |
| //--------------------------------------------------------- | |
| // Semaphore (POSIX, Linux, zOS) | |
| //--------------------------------------------------------- | |
| class Semaphore | |
| { | |
| private: | |
| 	sem_t m_sema; | |
| 
 | |
| 	Semaphore(const Semaphore& other) MOODYCAMEL_DELETE_FUNCTION; | |
| 	Semaphore& operator=(const Semaphore& other) MOODYCAMEL_DELETE_FUNCTION; | |
| 
 | |
| public: | |
| 	Semaphore(int initialCount = 0) | |
| 	{ | |
| 		assert(initialCount >= 0); | |
| 		int rc = sem_init(&m_sema, 0, static_cast<unsigned int>(initialCount)); | |
| 		assert(rc == 0); | |
| 		(void)rc; | |
| 	} | |
| 
 | |
| 	~Semaphore() | |
| 	{ | |
| 		sem_destroy(&m_sema); | |
| 	} | |
| 
 | |
| 	bool wait() | |
| 	{ | |
| 		// http://stackoverflow.com/questions/2013181/gdb-causes-sem-wait-to-fail-with-eintr-error | |
| 		int rc; | |
| 		do { | |
| 			rc = sem_wait(&m_sema); | |
| 		} while (rc == -1 && errno == EINTR); | |
| 		return rc == 0; | |
| 	} | |
| 
 | |
| 	bool try_wait() | |
| 	{ | |
| 		int rc; | |
| 		do { | |
| 			rc = sem_trywait(&m_sema); | |
| 		} while (rc == -1 && errno == EINTR); | |
| 		return rc == 0; | |
| 	} | |
| 
 | |
| 	bool timed_wait(std::uint64_t usecs) | |
| 	{ | |
| 		struct timespec ts; | |
| 		const int usecs_in_1_sec = 1000000; | |
| 		const int nsecs_in_1_sec = 1000000000; | |
| #ifdef MOODYCAMEL_LIGHTWEIGHTSEMAPHORE_MONOTONIC | |
| 		clock_gettime(CLOCK_MONOTONIC, &ts); | |
| #else | |
| 		clock_gettime(CLOCK_REALTIME, &ts); | |
| #endif | |
| 		ts.tv_sec += (time_t)(usecs / usecs_in_1_sec); | |
| 		ts.tv_nsec += (long)(usecs % usecs_in_1_sec) * 1000; | |
| 		// sem_timedwait bombs if you have more than 1e9 in tv_nsec | |
| 		// so we have to clean things up before passing it in | |
| 		if (ts.tv_nsec >= nsecs_in_1_sec) { | |
| 			ts.tv_nsec -= nsecs_in_1_sec; | |
| 			++ts.tv_sec; | |
| 		} | |
| 
 | |
| 		int rc; | |
| 		do { | |
| #ifdef MOODYCAMEL_LIGHTWEIGHTSEMAPHORE_MONOTONIC | |
| 			rc = sem_clockwait(&m_sema, CLOCK_MONOTONIC, &ts); | |
| #else | |
| 			rc = sem_timedwait(&m_sema, &ts); | |
| #endif | |
| 		} while (rc == -1 && errno == EINTR); | |
| 		return rc == 0; | |
| 	} | |
| 
 | |
| 	void signal() | |
| 	{ | |
| 		while (sem_post(&m_sema) == -1); | |
| 	} | |
| 
 | |
| 	void signal(int count) | |
| 	{ | |
| 		while (count-- > 0) | |
| 		{ | |
| 			while (sem_post(&m_sema) == -1); | |
| 		} | |
| 	} | |
| }; | |
| #else | |
| #error Unsupported platform! (No semaphore wrapper available) | |
| #endif | |
|  | |
| }	// end namespace details | |
|  | |
| 
 | |
| //--------------------------------------------------------- | |
| // LightweightSemaphore | |
| //--------------------------------------------------------- | |
| class LightweightSemaphore | |
| { | |
| public: | |
| 	typedef std::make_signed<std::size_t>::type ssize_t; | |
| 
 | |
| private: | |
| 	std::atomic<ssize_t> m_count; | |
| 	details::Semaphore m_sema; | |
| 	int m_maxSpins; | |
| 
 | |
| 	bool waitWithPartialSpinning(std::int64_t timeout_usecs = -1) | |
| 	{ | |
| 		ssize_t oldCount; | |
| 		int spin = m_maxSpins; | |
| 		while (--spin >= 0) | |
| 		{ | |
| 			oldCount = m_count.load(std::memory_order_relaxed); | |
| 			if ((oldCount > 0) && m_count.compare_exchange_strong(oldCount, oldCount - 1, std::memory_order_acquire, std::memory_order_relaxed)) | |
| 				return true; | |
| 			std::atomic_signal_fence(std::memory_order_acquire);	 // Prevent the compiler from collapsing the loop. | |
| 		} | |
| 		oldCount = m_count.fetch_sub(1, std::memory_order_acquire); | |
| 		if (oldCount > 0) | |
| 			return true; | |
| 		if (timeout_usecs < 0) | |
| 		{ | |
| 			if (m_sema.wait()) | |
| 				return true; | |
| 		} | |
| 		if (timeout_usecs > 0 && m_sema.timed_wait((std::uint64_t)timeout_usecs)) | |
| 			return true; | |
| 		// At this point, we've timed out waiting for the semaphore, but the | |
| 		// count is still decremented indicating we may still be waiting on | |
| 		// it. So we have to re-adjust the count, but only if the semaphore | |
| 		// wasn't signaled enough times for us too since then. If it was, we | |
| 		// need to release the semaphore too. | |
| 		while (true) | |
| 		{ | |
| 			oldCount = m_count.load(std::memory_order_acquire); | |
| 			if (oldCount >= 0 && m_sema.try_wait()) | |
| 				return true; | |
| 			if (oldCount < 0 && m_count.compare_exchange_strong(oldCount, oldCount + 1, std::memory_order_relaxed, std::memory_order_relaxed)) | |
| 				return false; | |
| 		} | |
| 	} | |
| 
 | |
| 	ssize_t waitManyWithPartialSpinning(ssize_t max, std::int64_t timeout_usecs = -1) | |
| 	{ | |
| 		assert(max > 0); | |
| 		ssize_t oldCount; | |
| 		int spin = m_maxSpins; | |
| 		while (--spin >= 0) | |
| 		{ | |
| 			oldCount = m_count.load(std::memory_order_relaxed); | |
| 			if (oldCount > 0) | |
| 			{ | |
| 				ssize_t newCount = oldCount > max ? oldCount - max : 0; | |
| 				if (m_count.compare_exchange_strong(oldCount, newCount, std::memory_order_acquire, std::memory_order_relaxed)) | |
| 					return oldCount - newCount; | |
| 			} | |
| 			std::atomic_signal_fence(std::memory_order_acquire); | |
| 		} | |
| 		oldCount = m_count.fetch_sub(1, std::memory_order_acquire); | |
| 		if (oldCount <= 0) | |
| 		{ | |
| 			if ((timeout_usecs == 0) || (timeout_usecs < 0 && !m_sema.wait()) || (timeout_usecs > 0 && !m_sema.timed_wait((std::uint64_t)timeout_usecs))) | |
| 			{ | |
| 				while (true) | |
| 				{ | |
| 					oldCount = m_count.load(std::memory_order_acquire); | |
| 					if (oldCount >= 0 && m_sema.try_wait()) | |
| 						break; | |
| 					if (oldCount < 0 && m_count.compare_exchange_strong(oldCount, oldCount + 1, std::memory_order_relaxed, std::memory_order_relaxed)) | |
| 						return 0; | |
| 				} | |
| 			} | |
| 		} | |
| 		if (max > 1) | |
| 			return 1 + tryWaitMany(max - 1); | |
| 		return 1; | |
| 	} | |
| 
 | |
| public: | |
| 	LightweightSemaphore(ssize_t initialCount = 0, int maxSpins = 10000) : m_count(initialCount), m_maxSpins(maxSpins) | |
| 	{ | |
| 		assert(initialCount >= 0); | |
| 		assert(maxSpins >= 0); | |
| 	} | |
| 
 | |
| 	bool tryWait() | |
| 	{ | |
| 		ssize_t oldCount = m_count.load(std::memory_order_relaxed); | |
| 		while (oldCount > 0) | |
| 		{ | |
| 			if (m_count.compare_exchange_weak(oldCount, oldCount - 1, std::memory_order_acquire, std::memory_order_relaxed)) | |
| 				return true; | |
| 		} | |
| 		return false; | |
| 	} | |
| 
 | |
| 	bool wait() | |
| 	{ | |
| 		return tryWait() || waitWithPartialSpinning(); | |
| 	} | |
| 
 | |
| 	bool wait(std::int64_t timeout_usecs) | |
| 	{ | |
| 		return tryWait() || waitWithPartialSpinning(timeout_usecs); | |
| 	} | |
| 
 | |
| 	// Acquires between 0 and (greedily) max, inclusive | |
| 	ssize_t tryWaitMany(ssize_t max) | |
| 	{ | |
| 		assert(max >= 0); | |
| 		ssize_t oldCount = m_count.load(std::memory_order_relaxed); | |
| 		while (oldCount > 0) | |
| 		{ | |
| 			ssize_t newCount = oldCount > max ? oldCount - max : 0; | |
| 			if (m_count.compare_exchange_weak(oldCount, newCount, std::memory_order_acquire, std::memory_order_relaxed)) | |
| 				return oldCount - newCount; | |
| 		} | |
| 		return 0; | |
| 	} | |
| 
 | |
| 	// Acquires at least one, and (greedily) at most max | |
| 	ssize_t waitMany(ssize_t max, std::int64_t timeout_usecs) | |
| 	{ | |
| 		assert(max >= 0); | |
| 		ssize_t result = tryWaitMany(max); | |
| 		if (result == 0 && max > 0) | |
| 			result = waitManyWithPartialSpinning(max, timeout_usecs); | |
| 		return result; | |
| 	} | |
| 	 | |
| 	ssize_t waitMany(ssize_t max) | |
| 	{ | |
| 		ssize_t result = waitMany(max, -1); | |
| 		assert(result > 0); | |
| 		return result; | |
| 	} | |
| 
 | |
| 	void signal(ssize_t count = 1) | |
| 	{ | |
| 		assert(count >= 0); | |
| 		ssize_t oldCount = m_count.fetch_add(count, std::memory_order_release); | |
| 		ssize_t toRelease = -oldCount < count ? -oldCount : count; | |
| 		if (toRelease > 0) | |
| 		{ | |
| 			m_sema.signal((int)toRelease); | |
| 		} | |
| 	} | |
| 	 | |
| 	std::size_t availableApprox() const | |
| 	{ | |
| 		ssize_t count = m_count.load(std::memory_order_relaxed); | |
| 		return count > 0 ? static_cast<std::size_t>(count) : 0; | |
| 	} | |
| }; | |
| 
 | |
| }   // end namespace moodycamel
 |