Merge pull request #583 from jpcima/atomic_queue

Update atomic_queue for a bugfix
This commit is contained in:
JP Cimalando 2020-12-14 13:28:58 +01:00 committed by GitHub
commit e78edbb7ce
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
3 changed files with 19 additions and 19 deletions

View file

@ -55,15 +55,13 @@ struct GetIndexShuffleBits<false, array_size, elements_per_cache_line> {
// the element within the cache line) with the next N bits (which are the index of the cache line) // the element within the cache line) with the next N bits (which are the index of the cache line)
// of the element index. // of the element index.
template<int BITS> template<int BITS>
constexpr unsigned remap_index_with_mix(unsigned index, unsigned mix) constexpr unsigned remap_index_with_mix(unsigned index, unsigned mix) {
{
return index ^ mix ^ (mix << BITS); return index ^ mix ^ (mix << BITS);
} }
template<int BITS> template<int BITS>
constexpr unsigned remap_index(unsigned index) noexcept { constexpr unsigned remap_index(unsigned index) noexcept {
return remap_index_with_mix<BITS>( return remap_index_with_mix<BITS>(index, (index ^ (index >> BITS)) & ((1u << BITS) - 1));
index, (index ^ (index >> BITS)) & ((1u << BITS) - 1));
} }
template<> template<>
@ -133,7 +131,7 @@ protected:
// The special member functions are not thread-safe. // The special member functions are not thread-safe.
AtomicQueueCommon() = default; AtomicQueueCommon() noexcept = default;
AtomicQueueCommon(AtomicQueueCommon const& b) noexcept AtomicQueueCommon(AtomicQueueCommon const& b) noexcept
: head_(b.head_.load(X)) : head_(b.head_.load(X))
@ -213,7 +211,7 @@ protected:
else { else {
for(;;) { for(;;) {
unsigned char expected = STORED; unsigned char expected = STORED;
if(ATOMIC_QUEUE_LIKELY(state.compare_exchange_strong(expected, LOADING, X, X))) { if(ATOMIC_QUEUE_LIKELY(state.compare_exchange_strong(expected, LOADING, A, X))) {
T element{std::move(q_element)}; T element{std::move(q_element)};
state.store(EMPTY, R); state.store(EMPTY, R);
return element; return element;
@ -238,7 +236,7 @@ protected:
else { else {
for(;;) { for(;;) {
unsigned char expected = EMPTY; unsigned char expected = EMPTY;
if(ATOMIC_QUEUE_LIKELY(state.compare_exchange_strong(expected, STORING, X, X))) { if(ATOMIC_QUEUE_LIKELY(state.compare_exchange_strong(expected, STORING, A, X))) {
q_element = std::forward<U>(element); q_element = std::forward<U>(element);
state.store(STORED, R); state.store(STORED, R);
return; return;
@ -318,11 +316,16 @@ public:
} }
bool was_empty() const noexcept { bool was_empty() const noexcept {
return static_cast<int>(head_.load(X) - tail_.load(X)) <= 0; return !was_size();
} }
bool was_full() const noexcept { bool was_full() const noexcept {
return static_cast<int>(head_.load(X) - tail_.load(X)) >= static_cast<int>(static_cast<Derived const&>(*this).size_); return was_size() >= static_cast<int>(static_cast<Derived const&>(*this).size_);
}
unsigned was_size() const noexcept {
// tail_ can be greater than head_ because of consumers doing pop, rather that try_pop, when the queue is empty.
return std::max(static_cast<int>(head_.load(X) - tail_.load(X)), 0);
} }
unsigned capacity() const noexcept { unsigned capacity() const noexcept {
@ -400,7 +403,7 @@ class AtomicQueue2 : public AtomicQueueCommon<AtomicQueue2<T, SIZE, MINIMIZE_CON
public: public:
using value_type = T; using value_type = T;
AtomicQueue2() = default; AtomicQueue2() noexcept = default;
AtomicQueue2(AtomicQueue2 const&) = delete; AtomicQueue2(AtomicQueue2 const&) = delete;
AtomicQueue2& operator=(AtomicQueue2 const&) = delete; AtomicQueue2& operator=(AtomicQueue2 const&) = delete;
}; };

View file

@ -16,7 +16,6 @@ static inline void spin_loop_pause() noexcept {
} }
} // namespace atomic_queue } // namespace atomic_queue
#elif defined(__arm__) || defined(__aarch64__) #elif defined(__arm__) || defined(__aarch64__)
// TODO: These need to be verified as I do not have access to ARM platform.
namespace atomic_queue { namespace atomic_queue {
constexpr int CACHE_LINE_SIZE = 64; constexpr int CACHE_LINE_SIZE = 64;
static inline void spin_loop_pause() noexcept { static inline void spin_loop_pause() noexcept {
@ -48,9 +47,11 @@ namespace atomic_queue {
#if defined(__GNUC__) || defined(__clang__) #if defined(__GNUC__) || defined(__clang__)
#define ATOMIC_QUEUE_LIKELY(expr) __builtin_expect(static_cast<bool>(expr), 1) #define ATOMIC_QUEUE_LIKELY(expr) __builtin_expect(static_cast<bool>(expr), 1)
#define ATOMIC_QUEUE_UNLIKELY(expr) __builtin_expect(static_cast<bool>(expr), 0) #define ATOMIC_QUEUE_UNLIKELY(expr) __builtin_expect(static_cast<bool>(expr), 0)
#define ATOMIC_QUEUE_NOINLINE __attribute__((noinline))
#else #else
#define ATOMIC_QUEUE_LIKELY(expr) expr #define ATOMIC_QUEUE_LIKELY(expr) (expr)
#define ATOMIC_QUEUE_UNLIKELY(expr) expr #define ATOMIC_QUEUE_UNLIKELY(expr) (expr)
#define ATOMIC_QUEUE_NOINLINE
#endif #endif
//////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////// ////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////

View file

@ -342,13 +342,9 @@ private:
struct QueuedFileData struct QueuedFileData
{ {
using TimePoint = std::chrono::time_point<std::chrono::high_resolution_clock>; using TimePoint = std::chrono::time_point<std::chrono::high_resolution_clock>;
QueuedFileData() = default; QueuedFileData() noexcept {}
QueuedFileData(std::weak_ptr<FileId> id, FileData* data, TimePoint queuedTime) QueuedFileData(std::weak_ptr<FileId> id, FileData* data, TimePoint queuedTime) noexcept
: id(id), data(data), queuedTime(queuedTime) {} : id(id), data(data), queuedTime(queuedTime) {}
QueuedFileData(const QueuedFileData&) = default;
QueuedFileData& operator=(const QueuedFileData&) = default;
QueuedFileData(QueuedFileData&&) = default;
QueuedFileData& operator=(QueuedFileData&&) = default;
std::weak_ptr<FileId> id; std::weak_ptr<FileId> id;
FileData* data { nullptr }; FileData* data { nullptr };
TimePoint queuedTime {}; TimePoint queuedTime {};