trapexit/mergerfs
1
0
Fork 0
mirror of https://github.com/trapexit/mergerfs.git synced 2026-10-12 02:23:45 +00:00
Code Issues Packages Projects Releases Wiki Activity

vendor moodycamel queue and upgrade to 1.0.5

Browse Source
This commit is contained in:
Antonio SJ Musumeci
2026-04-30 08:43:42 -04:00
committed by trapexit
parent 943f91ff98
commit f71e27644e
3 changed files with 77 additions and 33 deletions
Download Patch File Download Diff File Expand all files Collapse all files
vendored/libfuse/include/moodycamel/blockingconcurrentqueue.h → vendored/moodycamel/blockingconcurrentqueue.h
+29 -19
View File
@@ -56,18 +56,18 @@ public:
// includes making the memory effects of construction visible, possibly with a
// memory barrier).
explicit BlockingConcurrentQueue(size_t capacity = 6 * BLOCK_SIZE)
: inner(capacity), sema(create<LightweightSemaphore, ssize_t, int>(0, (int)Traits::MAX_SEMA_SPINS), &BlockingConcurrentQueue::template destroy<LightweightSemaphore>)
: inner(capacity), sema(create<LightweightSemaphore, ssize_t, int>(0, static_cast<int>(Traits::MAX_SEMA_SPINS)), &BlockingConcurrentQueue::template destroy<LightweightSemaphore>)
{
assert(reinterpret_cast<ConcurrentQueue*>((BlockingConcurrentQueue*)1) == &((BlockingConcurrentQueue*)1)->inner && "BlockingConcurrentQueue must have ConcurrentQueue as its first member");
assert(reinterpret_cast<ConcurrentQueue*>(reinterpret_cast<BlockingConcurrentQueue*>(1)) == &(reinterpret_cast<BlockingConcurrentQueue*>(1))->inner && "BlockingConcurrentQueue must have ConcurrentQueue as its first member");
if (!sema) {
MOODYCAMEL_THROW(std::bad_alloc());
}
}
BlockingConcurrentQueue(size_t minCapacity, size_t maxExplicitProducers, size_t maxImplicitProducers)
: inner(minCapacity, maxExplicitProducers, maxImplicitProducers), sema(create<LightweightSemaphore, ssize_t, int>(0, (int)Traits::MAX_SEMA_SPINS), &BlockingConcurrentQueue::template destroy<LightweightSemaphore>)
: inner(minCapacity, maxExplicitProducers, maxImplicitProducers), sema(create<LightweightSemaphore, ssize_t, int>(0, static_cast<int>(Traits::MAX_SEMA_SPINS)), &BlockingConcurrentQueue::template destroy<LightweightSemaphore>)
{
assert(reinterpret_cast<ConcurrentQueue*>((BlockingConcurrentQueue*)1) == &((BlockingConcurrentQueue*)1)->inner && "BlockingConcurrentQueue must have ConcurrentQueue as its first member");
assert(reinterpret_cast<ConcurrentQueue*>(reinterpret_cast<BlockingConcurrentQueue*>(1)) == &(reinterpret_cast<BlockingConcurrentQueue*>(1))->inner && "BlockingConcurrentQueue must have ConcurrentQueue as its first member");
if (!sema) {
MOODYCAMEL_THROW(std::bad_alloc());
}
@@ -179,12 +179,13 @@ public:
inline bool enqueue_bulk(It itemFirst, size_t count)
{
if ((details::likely)(inner.enqueue_bulk(std::forward<It>(itemFirst), count))) {
sema->signal((LightweightSemaphore::ssize_t)(ssize_t)count);
assert(static_cast<ssize_t>(count) >= 0);
sema->signal(static_cast<LightweightSemaphore::ssize_t>(static_cast<ssize_t>(count)));
return true;
}
return false;
}
// Enqueues several items using an explicit producer token.
// Allocates memory if required. Only fails if memory allocation fails
// (or Traits::MAX_SUBQUEUE_SIZE has been defined and would be surpassed).
@@ -195,7 +196,8 @@ public:
inline bool enqueue_bulk(producer_token_t const& token, It itemFirst, size_t count)
{
if ((details::likely)(inner.enqueue_bulk(token, std::forward<It>(itemFirst), count))) {
sema->signal((LightweightSemaphore::ssize_t)(ssize_t)count);
assert(static_cast<ssize_t>(count) >= 0);
sema->signal(static_cast<LightweightSemaphore::ssize_t>(static_cast<ssize_t>(count)));
return true;
}
return false;
@@ -264,12 +266,13 @@ public:
inline bool try_enqueue_bulk(It itemFirst, size_t count)
{
if (inner.try_enqueue_bulk(std::forward<It>(itemFirst), count)) {
sema->signal((LightweightSemaphore::ssize_t)(ssize_t)count);
assert(static_cast<ssize_t>(count) >= 0);
sema->signal(static_cast<LightweightSemaphore::ssize_t>(static_cast<ssize_t>(count)));
return true;
}
return false;
}
// Enqueues several items using an explicit producer token.
// Does not allocate memory. Fails if not enough room to enqueue.
// Note: Use std::make_move_iterator if the elements should be moved
@@ -279,7 +282,8 @@ public:
inline bool try_enqueue_bulk(producer_token_t const& token, It itemFirst, size_t count)
{
if (inner.try_enqueue_bulk(token, std::forward<It>(itemFirst), count)) {
sema->signal((LightweightSemaphore::ssize_t)(ssize_t)count);
assert(static_cast<ssize_t>(count) >= 0);
sema->signal(static_cast<LightweightSemaphore::ssize_t>(static_cast<ssize_t>(count)));
return true;
}
return false;
@@ -327,13 +331,14 @@ public:
inline size_t try_dequeue_bulk(It itemFirst, size_t max)
{
size_t count = 0;
max = (size_t)sema->tryWaitMany((LightweightSemaphore::ssize_t)(ssize_t)max);
assert(static_cast<ssize_t>(max) >= 0);
max = static_cast<size_t>(sema->tryWaitMany(static_cast<LightweightSemaphore::ssize_t>(static_cast<ssize_t>(max))));
while (count != max) {
count += inner.template try_dequeue_bulk<It&>(itemFirst, max - count);
}
return count;
}
// Attempts to dequeue several elements from the queue using an explicit consumer token.
// Returns the number of items actually dequeued.
// Returns 0 if all producer streams appeared empty at the time they
@@ -343,7 +348,8 @@ public:
inline size_t try_dequeue_bulk(consumer_token_t& token, It itemFirst, size_t max)
{
size_t count = 0;
max = (size_t)sema->tryWaitMany((LightweightSemaphore::ssize_t)(ssize_t)max);
assert(static_cast<ssize_t>(max) >= 0);
max = static_cast<size_t>(sema->tryWaitMany(static_cast<LightweightSemaphore::ssize_t>(static_cast<ssize_t>(max))));
while (count != max) {
count += inner.template try_dequeue_bulk<It&>(token, itemFirst, max - count);
}
@@ -447,13 +453,14 @@ public:
inline size_t wait_dequeue_bulk(It itemFirst, size_t max)
{
size_t count = 0;
max = (size_t)sema->waitMany((LightweightSemaphore::ssize_t)(ssize_t)max);
assert(static_cast<ssize_t>(max) >= 0);
max = static_cast<size_t>(sema->waitMany(static_cast<LightweightSemaphore::ssize_t>(static_cast<ssize_t>(max))));
while (count != max) {
count += inner.template try_dequeue_bulk<It&>(itemFirst, max - count);
}
return count;
}
// Attempts to dequeue several elements from the queue.
// Returns the number of items actually dequeued, which can
// be 0 if the timeout expires while waiting for elements,
@@ -465,7 +472,8 @@ public:
inline size_t wait_dequeue_bulk_timed(It itemFirst, size_t max, std::int64_t timeout_usecs)
{
size_t count = 0;
max = (size_t)sema->waitMany((LightweightSemaphore::ssize_t)(ssize_t)max, timeout_usecs);
assert(static_cast<ssize_t>(max) >= 0);
max = static_cast<size_t>(sema->waitMany(static_cast<LightweightSemaphore::ssize_t>(static_cast<ssize_t>(max)), timeout_usecs));
while (count != max) {
count += inner.template try_dequeue_bulk<It&>(itemFirst, max - count);
}
@@ -492,7 +500,8 @@ public:
inline size_t wait_dequeue_bulk(consumer_token_t& token, It itemFirst, size_t max)
{
size_t count = 0;
max = (size_t)sema->waitMany((LightweightSemaphore::ssize_t)(ssize_t)max);
assert(static_cast<ssize_t>(max) >= 0);
max = static_cast<size_t>(sema->waitMany(static_cast<LightweightSemaphore::ssize_t>(static_cast<ssize_t>(max))));
while (count != max) {
count += inner.template try_dequeue_bulk<It&>(token, itemFirst, max - count);
}
@@ -510,7 +519,8 @@ public:
inline size_t wait_dequeue_bulk_timed(consumer_token_t& token, It itemFirst, size_t max, std::int64_t timeout_usecs)
{
size_t count = 0;
max = (size_t)sema->waitMany((LightweightSemaphore::ssize_t)(ssize_t)max, timeout_usecs);
assert(static_cast<ssize_t>(max) >= 0);
max = static_cast<size_t>(sema->waitMany(static_cast<LightweightSemaphore::ssize_t>(static_cast<ssize_t>(max)), timeout_usecs));
while (count != max) {
count += inner.template try_dequeue_bulk<It&>(token, itemFirst, max - count);
}
@@ -537,7 +547,7 @@ public:
// Thread-safe.
inline size_t size_approx() const
{
return (size_t)sema->availableApprox();
return static_cast<size_t>(sema->availableApprox());
}
vendored/libfuse/include/moodycamel/concurrentqueue.h → vendored/moodycamel/concurrentqueue.h
+30 -8
View File
@@ -217,7 +217,7 @@ namespace moodycamel { namespace details {
// VS2013 doesn't support `thread_local`, and MinGW-w64 w/ POSIX threading has a crippling bug: http://sourceforge.net/p/mingw-w64/bugs/445
// g++ <=4.7 doesn't support thread_local either.
// Finally, iOS/ARM doesn't have support for it either, and g++/ARM allows it to compile but it's unconfirmed to actually work
#if (!defined(_MSC_VER) || _MSC_VER >= 1900) && (!defined(__MINGW32__) && !defined(__MINGW64__) || !defined(__WINPTHREADS_VERSION)) && (!defined(__GNUC__) || __GNUC__ > 4 || (__GNUC__ == 4 && __GNUC_MINOR__ >= 8)) && (!defined(__APPLE__) || !TARGET_OS_IPHONE) && !defined(__arm__) && !defined(_M_ARM) && !defined(__aarch64__) && !defined(__MVS__)
#if (!defined(_MSC_VER) || _MSC_VER >= 1900) && (!defined(__MINGW32__) && !defined(__MINGW64__) || !defined(__WINPTHREADS_VERSION)) && (!defined(__GNUC__) || __GNUC__ > 4 || (__GNUC__ == 4 && __GNUC_MINOR__ >= 8)) && (!defined(__APPLE__) || !TARGET_OS_IPHONE) && !defined(__arm__) && !defined(_M_ARM) && !defined(__aarch64__) && !defined(__MVS__) && !defined(MOODYCAMEL_NO_THREAD_LOCAL)
// Assume `thread_local` is fully supported in all other C++11 compilers/platforms
#define MOODYCAMEL_CPP11_THREAD_LOCAL_SUPPORTED // tentatively enabled for now; years ago several users report having problems with it on
#endif
@@ -355,6 +355,12 @@ struct ConcurrentQueueDefaultTraits
// How many full blocks can be expected for a single implicit producer? This should
// reflect that number's maximum for optimal performance. Must be a power of 2.
// Note: This impacts the maximum number of elements that can be enqueued by a
// single implicit producer when using try_enqueue/try_enqeue_bulk exclusively (which
// cannot allocate), since it limits the number of blocks that the producer can hold to
// store elements. When pre-allocating blocks for use with try-enqueueing, configure
// this initial size to the desired maximum number of blocks per implicit producer.
// Alternately, use the regular enqueue methods, which can grow the index as needed.
static const size_t IMPLICIT_INITIAL_INDEX_SIZE = 32;
// The initial size of the hash table mapping thread IDs to implicit producers.
@@ -469,8 +475,15 @@ namespace details
static inline size_t hash_thread_id(thread_id_t id)
{
static_assert(sizeof(thread_id_t) <= 8, "Expected a platform where thread IDs are at most 64-bit values");
#if defined(__GNUC__) && !defined(__clang__)
#pragma GCC diagnostic push
#pragma GCC diagnostic ignored "-Wuseless-cast"
#endif
return static_cast<size_t>(hash_32_or_64<sizeof(thread_id_converter<thread_id_t>::thread_id_hash_t)>::hash(
thread_id_converter<thread_id_t>::prehash(id)));
#if defined(__GNUC__) && !defined(__clang__)
#pragma GCC diagnostic pop
#endif
}
template<typename T>
@@ -509,9 +522,9 @@ namespace details
template<typename T>
static inline void swap_relaxed(std::atomic<T>& left, std::atomic<T>& right)
{
T temp = std::move(left.load(std::memory_order_relaxed));
left.store(std::move(right.load(std::memory_order_relaxed)), std::memory_order_relaxed);
right.store(std::move(temp), std::memory_order_relaxed);
T temp = left.load(std::memory_order_relaxed);
left.store(right.load(std::memory_order_relaxed), std::memory_order_relaxed);
right.store(temp, std::memory_order_relaxed);
}
template<typename T>
@@ -1056,6 +1069,9 @@ public:
// Does not allocate memory. Fails if not enough room to enqueue (or implicit
// production is disabled because Traits::INITIAL_IMPLICIT_PRODUCER_HASH_SIZE
// is 0).
// Note: If using only try_enqueue/try_enqueue_bulk with pre-allocated blocks, configure
// Traits::IMPLICIT_INITIAL_INDEX_SIZE appropriately to ensure the index has sufficient
// capacity for the number of blocks each producer may need.
// Thread-safe.
inline bool try_enqueue(T const& item)
{
@@ -1067,6 +1083,9 @@ public:
// Does not allocate memory (except for one-time implicit producer).
// Fails if not enough room to enqueue (or implicit production is
// disabled because Traits::INITIAL_IMPLICIT_PRODUCER_HASH_SIZE is 0).
// Note: If using only try_enqueue/try_enqueue_bulk with pre-allocated blocks, configure
// Traits::IMPLICIT_INITIAL_INDEX_SIZE appropriately to ensure the index has sufficient
// capacity for the number of blocks each producer may need.
// Thread-safe.
inline bool try_enqueue(T&& item)
{
@@ -1094,6 +1113,9 @@ public:
// Does not allocate memory (except for one-time implicit producer).
// Fails if not enough room to enqueue (or implicit production is
// disabled because Traits::INITIAL_IMPLICIT_PRODUCER_HASH_SIZE is 0).
// Note: If using only try_enqueue/try_enqueue_bulk with pre-allocated blocks, configure
// Traits::IMPLICIT_INITIAL_INDEX_SIZE appropriately to ensure the index has sufficient
// capacity for the number of blocks each producer may need.
// Note: Use std::make_move_iterator if the elements should be moved
// instead of copied.
// Thread-safe.
@@ -1479,7 +1501,7 @@ private:
while (head != nullptr) {
auto prevHead = head;
auto refs = head->freeListRefs.load(std::memory_order_relaxed);
if ((refs & REFS_MASK) == 0 || !head->freeListRefs.compare_exchange_strong(refs, refs + 1, std::memory_order_acquire, std::memory_order_relaxed)) {
if ((refs & REFS_MASK) == 0 || !head->freeListRefs.compare_exchange_strong(refs, refs + 1, std::memory_order_acquire)) {
head = freeListHead.load(std::memory_order_acquire);
continue;
}
@@ -1529,7 +1551,7 @@ private:
node->freeListRefs.store(1, std::memory_order_release);
if (!freeListHead.compare_exchange_strong(head, node, std::memory_order_release, std::memory_order_relaxed)) {
// Hmm, the add failed, but we can only try again when the refcount goes back to zero
if (node->freeListRefs.fetch_add(SHOULD_BE_ON_FREELIST - 1, std::memory_order_release) == 1) {
if (node->freeListRefs.fetch_add(SHOULD_BE_ON_FREELIST - 1, std::memory_order_acq_rel) == 1) {
continue;
}
}
@@ -1604,7 +1626,7 @@ private:
}
else {
// Increment counter
auto prevVal = elementsCompletelyDequeued.fetch_add(1, std::memory_order_release);
auto prevVal = elementsCompletelyDequeued.fetch_add(1, std::memory_order_acq_rel);
assert(prevVal < BLOCK_SIZE);
return prevVal == BLOCK_SIZE - 1;
}
@@ -1627,7 +1649,7 @@ private:
}
else {
// Increment counter
auto prevVal = elementsCompletelyDequeued.fetch_add(count, std::memory_order_release);
auto prevVal = elementsCompletelyDequeued.fetch_add(count, std::memory_order_acq_rel);
assert(prevVal + count <= BLOCK_SIZE);
return prevVal + count == BLOCK_SIZE;
}
vendored/libfuse/include/moodycamel/lightweightsemaphore.h → vendored/moodycamel/lightweightsemaphore.h
+18 -6
View File
@@ -5,8 +5,12 @@
#pragma once
#include <cstddef> // For std::size_t
#include <atomic>
#include <cassert> // assert
#include <cerrno> // For EINTR
#include <cstddef> // For std::size_t
#include <cstdint> // For std::uint64_t
#include <ctime> // For clock_gettime
#include <type_traits> // For std::make_signed<T>
#if defined(_WIN32)
@@ -36,6 +40,14 @@ extern "C" {
#endif
#endif
#ifndef MOODYCAMEL_DELETE_FUNCTION
#if __cplusplus >= 201103L || _MSC_VER >= 1900
#define MOODYCAMEL_DELETE_FUNCTION = delete
#else
#define MOODYCAMEL_DELETE_FUNCTION
#endif
#endif
namespace moodycamel
{
namespace details
@@ -222,8 +234,8 @@ public:
#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;
ts.tv_sec += static_cast<time_t>(usecs / usecs_in_1_sec);
ts.tv_nsec += static_cast<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) {
@@ -294,7 +306,7 @@ private:
if (m_sema.wait())
return true;
}
if (timeout_usecs > 0 && m_sema.timed_wait((std::uint64_t)timeout_usecs))
if (timeout_usecs > 0 && m_sema.timed_wait(static_cast<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
@@ -330,7 +342,7 @@ private:
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)))
if ((timeout_usecs == 0) || (timeout_usecs < 0 && !m_sema.wait()) || (timeout_usecs > 0 && !m_sema.timed_wait(static_cast<std::uint64_t>(timeout_usecs))))
{
while (true)
{
@@ -413,7 +425,7 @@ public:
ssize_t toRelease = -oldCount < count ? -oldCount : count;
if (toRelease > 0)
{
m_sema.signal((int)toRelease);
m_sema.signal(static_cast<int>(toRelease));
}
}