From 4cd7bf43d5c7e3dcaca926bc7273f98146866033 Mon Sep 17 00:00:00 2001 From: tzcnt Date: Fri, 3 Jul 2026 14:51:06 -0700 Subject: [PATCH 1/4] fix: timestamp counter on armhf --- include/tmc/detail/compat.hpp | 24 ++++++++++++++++++++++++ 1 file changed, 24 insertions(+) diff --git a/include/tmc/detail/compat.hpp b/include/tmc/detail/compat.hpp index 4bc10d8a..654a05db 100644 --- a/include/tmc/detail/compat.hpp +++ b/include/tmc/detail/compat.hpp @@ -90,6 +90,8 @@ static inline void TMC_CPU_PAUSE() noexcept { // Clang defines __yield intrinsic, but GCC doesn't, so we use asm asm volatile("yield"); } +#if defined(__aarch64__) || defined(_M_ARM64) +// AArch64: the generic timer is read through dedicated system registers. // Read the ARM "Virtual Counter" register. // This ticks at a frequency independent of the processor frequency. // https://developer.arm.com/documentation/ddi0406/cb/System-Level-Architecture/The-Generic-Timer/About-the-Generic-Timer/The-virtual-counter?lang=en @@ -104,6 +106,28 @@ static inline size_t TMC_ARM_CPU_FREQ() noexcept { asm volatile("mrs %0, cntfrq_el0; isb; " : "=r"(freq)::"memory"); return freq; } +#else +// AArch32 (e.g. armhf): the AArch64 `cntvct_el0` / `cntfrq_el0` system +// registers don't exist here. The ARMv7-A Generic Timer exposes the same +// counters through CP15 coprocessor accesses instead. +// https://developer.arm.com/documentation/ddi0406/cb/System-Level-Architecture/The-Generic-Timer/Register-descriptions?lang=en +// Read the 64-bit "Virtual Counter" (CNTVCT) via MRRC into a register pair. +// size_t is 32-bit here, so we return only the low word - matching the 32-bit +// x86 path, which likewise truncates __rdtsc(). The counter ticks at a +// frequency independent of the processor frequency. +static inline size_t TMC_CPU_TIMESTAMP() noexcept { + uint32_t lo, hi; + asm volatile("mrrc p15, 1, %0, %1, c14" : "=r"(lo), "=r"(hi)::"memory"); + (void)hi; + return lo; +} +// Read the "Virtual Counter" frequency (CNTFRQ, a 32-bit register) via MRC. +static inline size_t TMC_ARM_CPU_FREQ() noexcept { + uint32_t freq; + asm volatile("mrc p15, 0, %0, c14, c0, 0" : "=r"(freq)::"memory"); + return freq; +} +#endif static inline const size_t TMC_CPU_FREQ = TMC_ARM_CPU_FREQ(); #elif defined(__loongarch__) && defined(__LP64__) // Use some barrier instructions to generate a delay, as there's From 46ee6dc6761076f3e3f7c2ca73fd3bc3928c9fac Mon Sep 17 00:00:00 2001 From: tzcnt Date: Fri, 3 Jul 2026 15:21:27 -0700 Subject: [PATCH 2/4] add fallbacks for ARMv7-A without timer counter, and unknown architectures --- include/tmc/detail/compat.hpp | 82 ++++++++++++++++++++++++++++++----- 1 file changed, 72 insertions(+), 10 deletions(-) diff --git a/include/tmc/detail/compat.hpp b/include/tmc/detail/compat.hpp index 654a05db..5abd611c 100644 --- a/include/tmc/detail/compat.hpp +++ b/include/tmc/detail/compat.hpp @@ -42,7 +42,7 @@ #ifdef __has_cpp_attribute -#if __has_cpp_attribute(clang::coro_await_elidable_argument) && \ +#if __has_cpp_attribute(clang::coro_await_elidable) && \ __has_cpp_attribute(clang::coro_await_elidable_argument) #define TMC_CORO_AWAIT_ELIDABLE [[clang::coro_await_elidable]] #define TMC_CORO_AWAIT_ELIDABLE_ARGUMENT [[clang::coro_await_elidable_argument]] @@ -62,8 +62,8 @@ #define TMC_SIZED_DEALLOCATION 0 #endif -#if defined(__x86_64__) || defined(_M_AMD64) || defined(i386) || \ - defined(__i386__) || defined(__i386) || defined(_M_IX86) +#if defined(__x86_64__) || defined(_M_AMD64) || defined(i386) || defined(__i386__) || \ + defined(__i386) || defined(_M_IX86) #ifdef _MSC_VER #include #else @@ -83,14 +83,45 @@ static inline size_t TMC_CPU_TIMESTAMP() noexcept { // clustering threshold in tmc::channel) this seems like reasonable behavior // anyway. static inline const size_t TMC_CPU_FREQ = 3500000000; -#elif defined(__arm__) || defined(_M_ARM) || defined(_M_ARM64) || \ - defined(__aarch64__) || defined(__ARM_ACLE) +#elif defined(__arm__) || defined(_M_ARM) || defined(_M_ARM64) || defined(__aarch64__) +#if defined(__aarch64__) || defined(_M_ARM64) || \ + (defined(__ARM_ARCH_PROFILE) && __ARM_ARCH_PROFILE == 'A') || \ + (defined(__ARM_ARCH_PROFILE) && __ARM_ARCH_PROFILE == 'R') #define TMC_CPU_ARM +#endif +#ifdef _MSC_VER +#include +#endif static inline void TMC_CPU_PAUSE() noexcept { - // Clang defines __yield intrinsic, but GCC doesn't, so we use asm + // Clang defines __yield intrinsic, but GCC doesn't, so we use asm where we + // know the ARM yield instruction is available. +#if defined(__aarch64__) || defined(_M_ARM64) || (defined(__ARM_ARCH) && __ARM_ARCH >= 7) +#if defined(_MSC_VER) + __yield(); +#else asm volatile("yield"); +#endif +#endif +} +#if defined(_MSC_VER) && defined(_M_ARM64) +#define TMC_ARM64_SYSREG(op0, op1, crn, crm, op2) \ + (((op0 & 1) << 14) | ((op1 & 7) << 11) | ((crn & 15) << 7) | \ + ((crm & 15) << 3) | ((op2 & 7) << 0)) +static inline size_t TMC_CPU_TIMESTAMP() noexcept { + return static_cast(_ReadStatusReg(TMC_ARM64_SYSREG(3, 3, 14, 0, 2))); +} +static inline size_t TMC_ARM_CPU_FREQ() noexcept { + return static_cast(_ReadStatusReg(TMC_ARM64_SYSREG(3, 3, 14, 0, 0))); +} +#undef TMC_ARM64_SYSREG +#elif defined(_MSC_VER) && defined(_M_ARM) +static inline size_t TMC_CPU_TIMESTAMP() noexcept { + return static_cast(_MoveFromCoprocessor64(15, 1, 14)); +} +static inline size_t TMC_ARM_CPU_FREQ() noexcept { + return static_cast(_MoveFromCoprocessor(15, 0, 14, 0, 0)); } -#if defined(__aarch64__) || defined(_M_ARM64) +#elif defined(__aarch64__) || defined(_M_ARM64) // AArch64: the generic timer is read through dedicated system registers. // Read the ARM "Virtual Counter" register. // This ticks at a frequency independent of the processor frequency. @@ -106,16 +137,36 @@ static inline size_t TMC_ARM_CPU_FREQ() noexcept { asm volatile("mrs %0, cntfrq_el0; isb; " : "=r"(freq)::"memory"); return freq; } -#else +#elif defined(__ARM_ARCH) && __ARM_ARCH >= 7 && defined(__ARM_ARCH_PROFILE) && \ + (__ARM_ARCH_PROFILE == 'A' || __ARM_ARCH_PROFILE == 'R') +#if defined(__linux__) +#include +#include +#endif // AArch32 (e.g. armhf): the AArch64 `cntvct_el0` / `cntfrq_el0` system // registers don't exist here. The ARMv7-A Generic Timer exposes the same // counters through CP15 coprocessor accesses instead. // https://developer.arm.com/documentation/ddi0406/cb/System-Level-Architecture/The-Generic-Timer/Register-descriptions?lang=en +// The Generic Timer is optional on ARMv7-A, and user-mode access is controlled +// by the OS. On Linux, HWCAP_EVTSTRM indicates that the kernel configured the +// architected timer event stream; if that is absent, fall back to inert timing +// rather than trapping on an undefined/privileged CP15 access. +static inline bool TMC_ARM_HAS_GENERIC_TIMER_IMPL() noexcept { +#if defined(__linux__) && defined(HWCAP_EVTSTRM) + return (getauxval(AT_HWCAP) & HWCAP_EVTSTRM) != 0; +#else + return false; +#endif +} +static inline const bool TMC_ARM_HAS_GENERIC_TIMER = TMC_ARM_HAS_GENERIC_TIMER_IMPL(); // Read the 64-bit "Virtual Counter" (CNTVCT) via MRRC into a register pair. // size_t is 32-bit here, so we return only the low word - matching the 32-bit // x86 path, which likewise truncates __rdtsc(). The counter ticks at a // frequency independent of the processor frequency. static inline size_t TMC_CPU_TIMESTAMP() noexcept { + if (!TMC_ARM_HAS_GENERIC_TIMER) { + return 0; + } uint32_t lo, hi; asm volatile("mrrc p15, 1, %0, %1, c14" : "=r"(lo), "=r"(hi)::"memory"); (void)hi; @@ -123,10 +174,16 @@ static inline size_t TMC_CPU_TIMESTAMP() noexcept { } // Read the "Virtual Counter" frequency (CNTFRQ, a 32-bit register) via MRC. static inline size_t TMC_ARM_CPU_FREQ() noexcept { + if (!TMC_ARM_HAS_GENERIC_TIMER) { + return 1000000000; + } uint32_t freq; asm volatile("mrc p15, 0, %0, c14, c0, 0" : "=r"(freq)::"memory"); return freq; } +#else +static inline size_t TMC_CPU_TIMESTAMP() noexcept { return 0; } +static inline size_t TMC_ARM_CPU_FREQ() noexcept { return 1000000000; } #endif static inline const size_t TMC_CPU_FREQ = TMC_ARM_CPU_FREQ(); #elif defined(__loongarch__) && defined(__LP64__) @@ -145,10 +202,10 @@ static inline size_t TMC_CPU_TIMESTAMP() noexcept { // Calculate the LoongArch stable counter frequency static inline size_t TMC_LOONGARCH_CPU_FREQ() noexcept { size_t cc_freq; - asm ("cpucfg %0, %1" : "=r"(cc_freq) : "r"(0x4)); + asm("cpucfg %0, %1" : "=r"(cc_freq) : "r"(0x4)); size_t cc_mul_div; - asm ("cpucfg %0, %1" : "=r"(cc_mul_div) : "r"(0x5)); + asm("cpucfg %0, %1" : "=r"(cc_mul_div) : "r"(0x5)); size_t cc_mul = cc_mul_div & 0xffff; size_t cc_div = (cc_mul_div >> 16) & 0xffff; @@ -204,6 +261,11 @@ static inline size_t TMC_CPU_TIMESTAMP() noexcept { return count; } static inline const size_t TMC_CPU_FREQ = TMC_RISCV_READ_TIMEBASE_FREQ(); +#else +// Fallback for unknown architectures +static inline void TMC_CPU_PAUSE() noexcept {} +static inline size_t TMC_CPU_TIMESTAMP() noexcept { return 0; } +static inline const size_t TMC_CPU_FREQ = 1000000000; #endif // clang-format tries to collapse the pragmas into one line... From 706613eabc9cf03046560b9812479b8220c931b2 Mon Sep 17 00:00:00 2001 From: tzcnt Date: Fri, 3 Jul 2026 16:08:56 -0700 Subject: [PATCH 3/4] move architecture-specific timer behavior into its own header --- include/tmc/channel.hpp | 286 +++++++++------------------- include/tmc/detail/compat.hpp | 156 --------------- include/tmc/detail/timer_compat.hpp | 188 ++++++++++++++++++ 3 files changed, 283 insertions(+), 347 deletions(-) create mode 100644 include/tmc/detail/timer_compat.hpp diff --git a/include/tmc/channel.hpp b/include/tmc/channel.hpp index eab70ed3..e1614402 100644 --- a/include/tmc/channel.hpp +++ b/include/tmc/channel.hpp @@ -28,6 +28,7 @@ #include "tmc/detail/compat.hpp" #include "tmc/detail/concepts_awaitable.hpp" #include "tmc/detail/qu_storage.hpp" +#include "tmc/detail/timer_compat.hpp" #include "tmc/detail/tiny_lock.hpp" #include "tmc/ex_any.hpp" #include "tmc/task.hpp" @@ -71,8 +72,7 @@ struct chan_default_config { /// but access to a single token from multiple threads is not. /// To access the channel from multiple threads or tasks concurrently, /// make a copy of the token for each (by using the copy constructor). -template -class chan_tok; +template class chan_tok; /// Creates a new channel and returns an access token to it. template @@ -115,11 +115,9 @@ namespace detail { // remain unconditionally seq_cst - it is followed by a plain seq_cst load // (not an RMW), which provides no StoreLoad ordering even on x86. #if defined(TMC_CPU_X86) || defined(TMC_CPU_ARM) -inline constexpr std::memory_order hazptr_protect_order = - std::memory_order_relaxed; +inline constexpr std::memory_order hazptr_protect_order = std::memory_order_relaxed; #else -inline constexpr std::memory_order hazptr_protect_order = - std::memory_order_seq_cst; +inline constexpr std::memory_order hazptr_protect_order = std::memory_order_seq_cst; #endif // Hazard pointer type used internally by channel to track in-use blocks. @@ -141,8 +139,8 @@ class alignas(TMC_CACHE_LINE_SIZE) hazard_ptr { template friend class tmc::channel; template friend class tmc::chan_zc_scope; - static inline constexpr size_t InactiveHazptrOffset = - TMC_ONE_BIT << (TMC_PLATFORM_BITS - 2); + static inline constexpr size_t InactiveHazptrOffset = TMC_ONE_BIT + << (TMC_PLATFORM_BITS - 2); inline void release_blocks() noexcept { // These elements may be read (by try_reclaim_block()) after @@ -249,8 +247,7 @@ template class chan_zc_scope { chan_zc_scope(const chan_zc_scope&) = delete; chan_zc_scope& operator=(const chan_zc_scope&) = delete; chan_zc_scope(chan_zc_scope&& Other) noexcept - : haz_ptr{Other.haz_ptr}, data{Other.data}, - release_idx{Other.release_idx} { + : haz_ptr{Other.haz_ptr}, data{Other.data}, release_idx{Other.release_idx} { Other.data = nullptr; } @@ -284,8 +281,7 @@ template class channel { static inline constexpr size_t BlockSize = Config::BlockSize; static inline constexpr size_t BlockSizeMask = BlockSize - 1; static_assert( - BlockSize && ((BlockSize & (BlockSize - 1)) == 0), - "BlockSize must be a power of 2" + BlockSize && ((BlockSize & (BlockSize - 1)) == 0), "BlockSize must be a power of 2" ); // Ensure that the subtraction of unsigned offsets always results in a value @@ -303,8 +299,8 @@ template class channel { // elements and then cannot free any blocks until that thread wakes. This is // extremely unlikely, and not an error - it will just prevent block // reclamation. On 64 bit in practice this will never happen. - static inline constexpr size_t InactiveHazptrOffset = - TMC_ONE_BIT << (TMC_PLATFORM_BITS - 2); + static inline constexpr size_t InactiveHazptrOffset = TMC_ONE_BIT + << (TMC_PLATFORM_BITS - 2); // Defaults to 2M items per second; this is 1 item every 500ns. static inline constexpr size_t DefaultHeavyLoadThreshold = 2000000; @@ -316,8 +312,7 @@ template class channel { static inline constexpr size_t ClusterPeriod = 10000; friend chan_tok; - template - friend chan_tok make_channel() noexcept; + template friend chan_tok make_channel() noexcept; public: class aw_pull; @@ -345,16 +340,14 @@ template class channel { static constexpr size_t UNPADLEN = sizeof(size_t) + sizeof(void*) + sizeof(tmc::detail::qu_storage); - static constexpr size_t WANTLEN = (UNPADLEN + TMC_CACHE_LINE_SIZE - 1) & - static_cast( - 0 - TMC_CACHE_LINE_SIZE - ); // round up to TMC_CACHE_LINE_SIZE - static constexpr size_t PADLEN = - UNPADLEN < WANTLEN ? (WANTLEN - UNPADLEN) : 999; + static constexpr size_t WANTLEN = + (UNPADLEN + TMC_CACHE_LINE_SIZE - 1) & + static_cast(0 - TMC_CACHE_LINE_SIZE); // round up to TMC_CACHE_LINE_SIZE + static constexpr size_t PADLEN = UNPADLEN < WANTLEN ? (WANTLEN - UNPADLEN) : 999; struct empty {}; - using Padding = std::conditional_t< - Config::PackingLevel == 0 && PADLEN != 999, char[PADLEN], empty>; + using Padding = + std::conditional_t; TMC_NO_UNIQUE_ADDRESS Padding pad; // If this returns false, data is ready and consumer should not wait. @@ -371,8 +364,7 @@ template class channel { consumer_base* set_data_ready_or_get_waiting_consumer() noexcept { uintptr_t expected = 0; if (flags.compare_exchange_strong( - expected, DATA_BIT, std::memory_order_acq_rel, - std::memory_order_acquire + expected, DATA_BIT, std::memory_order_acq_rel, std::memory_order_acquire )) { return nullptr; } else { @@ -384,9 +376,7 @@ template class channel { // because it saw the closed flag and returned. // This does not need to be called during normal operation - only after the // closed flag is observed. - void set_not_waiting() noexcept { - flags.store(BOTH_BITS, std::memory_order_release); - } + void set_not_waiting() noexcept { flags.store(BOTH_BITS, std::memory_order_release); } // Used by close() to find consumers that were already waiting on a slot // that will never receive data. @@ -449,9 +439,7 @@ template class channel { // This does not need to be called during normal operation - only after the // closed flag is observed. void set_not_waiting() noexcept { - flags.store( - reinterpret_cast(BOTH_BITS), std::memory_order_release - ); + flags.store(reinterpret_cast(BOTH_BITS), std::memory_order_release); } // Used by close() to find consumers that were already waiting on a slot @@ -547,8 +535,7 @@ template class channel { std::atomic tail_block; struct empty {}; - using EmbeddedBlock = - std::conditional_t; + using EmbeddedBlock = std::conditional_t; TMC_NO_UNIQUE_ADDRESS EmbeddedBlock embedded_block; channel() noexcept { @@ -621,9 +608,7 @@ template class channel { } for (size_t i = 0; i < ClusterOn.size(); ++i) { - ClusterOn[i].id->requested_thread_index.store( - closest, std::memory_order_relaxed - ); + ClusterOn[i].id->requested_thread_index.store(closest, std::memory_order_relaxed); } } @@ -723,8 +708,7 @@ template class channel { // (see hazptr_protect_order), this guarantees that if a token observed the // pre-CAS block pointer, the revalidation here will observe that token's // active_offset protection. - static inline void - keep_min(size_t& Dst, std::atomic const& Src) noexcept { + static inline void keep_min(size_t& Dst, std::atomic const& Src) noexcept { size_t val = Src.load(std::memory_order_seq_cst); if (circular_less_than(val, Dst)) { Dst = val; @@ -754,9 +738,7 @@ template class channel { return; } if (circular_less_than( - static_cast(block)->offset.load( - std::memory_order_relaxed - ), + static_cast(block)->offset.load(std::memory_order_relaxed), NewHead->offset.load(std::memory_order_relaxed) )) { if (!DstBlock.compare_exchange_strong( @@ -769,9 +751,8 @@ template class channel { // If this hazptr updated its own block, but the updated block is // still earlier than the new head, then we cannot free that block. keep_min( - MinProtected, static_cast(block)->offset.load( - std::memory_order_relaxed - ) + MinProtected, + static_cast(block)->offset.load(std::memory_order_relaxed) ); } // Reload hazptr after trying to modify block to ensure that if it was @@ -784,9 +765,8 @@ template class channel { // the first block that is protected by a hazard pointer. This block is // returned to become the NewHead. If OldHead is protected, then it will be // returned unchanged, and no blocks can be reclaimed. - data_block* try_advance_head( - hazard_ptr* Haz, data_block* OldHead, size_t ProtectIdx - ) noexcept { + data_block* + try_advance_head(hazard_ptr* Haz, data_block* OldHead, size_t ProtectIdx) noexcept { // In the current implementation, this is called only from consumers. // Therefore, this token's hazptr will be active, and protecting read_block. // However, if producers are lagging behind, and no producer is currently @@ -808,9 +788,9 @@ template class channel { // Find the block associated with this offset. data_block* newHead = OldHead; - while (circular_less_than( - newHead->offset.load(std::memory_order_relaxed), ProtectIdx - )) { + while ( + circular_less_than(newHead->offset.load(std::memory_order_relaxed), ProtectIdx) + ) { newHead = newHead->next.load(std::memory_order_acquire); } @@ -829,13 +809,11 @@ template class channel { // ProtectIdx may have been reduced by the double-check in // try_advance_block. If so, reduce newHead as well. - if (circular_less_than( - ProtectIdx, newHead->offset.load(std::memory_order_relaxed) - )) { + if (circular_less_than(ProtectIdx, newHead->offset.load(std::memory_order_relaxed))) { newHead = OldHead; - while (circular_less_than( - newHead->offset.load(std::memory_order_relaxed), ProtectIdx - )) { + while ( + circular_less_than(newHead->offset.load(std::memory_order_relaxed), ProtectIdx) + ) { newHead = newHead->next; } } @@ -890,14 +868,11 @@ template class channel { } // Actually unlink the blocks from the head of the queue. // They stay linked to each other. - unlinked[unlinkedCount - 1]->next.store( - nullptr, std::memory_order_release - ); + unlinked[unlinkedCount - 1]->next.store(nullptr, std::memory_order_release); while (true) { // Update their offsets to the end of the queue. - size_t boff = - tailBlock->offset.load(std::memory_order_relaxed) + BlockSize; + size_t boff = tailBlock->offset.load(std::memory_order_relaxed) + BlockSize; for (size_t i = 0; i < unlinkedCount; ++i) { unlinked[i]->offset.store(boff, std::memory_order_relaxed); boff += BlockSize; @@ -905,8 +880,7 @@ template class channel { // Re-link the tail of the queue to the head of the unlinked blocks. if (tailBlock->next.compare_exchange_strong( - next, unlinked[0], std::memory_order_acq_rel, - std::memory_order_acquire + next, unlinked[0], std::memory_order_acq_rel, std::memory_order_acquire )) { break; } @@ -977,8 +951,7 @@ template class channel { if (next == nullptr) { data_block* newBlock = new data_block(offset + BlockSize); if (Block->next.compare_exchange_strong( - next, newBlock, std::memory_order_acq_rel, - std::memory_order_acquire + next, newBlock, std::memory_order_acq_rel, std::memory_order_acquire )) { next = newBlock; } else { @@ -1014,8 +987,7 @@ template class channel { // load below sees the value close() stored before its release of `closed`. // - Non-blocking try_pull may pass acquire: it never blocks, so eventual // visibility is sufficient. - template - bool is_closed_past(size_t Idx) const noexcept { + template bool is_closed_past(size_t Idx) const noexcept { auto closedState = closed.load(Order); if (0 == closedState) [[likely]] { return false; @@ -1025,9 +997,7 @@ template class channel { TMC_CPU_PAUSE(); closedState = closed.load(std::memory_order_acquire); } - return circular_less_than( - write_closed_at.load(std::memory_order_relaxed), 1 + Idx - ); + return circular_less_than(write_closed_at.load(std::memory_order_relaxed), 1 + Idx); } // Idx will be initialized by this function @@ -1038,12 +1008,10 @@ template class channel { // seq_cst is needed here to create a StoreLoad barrier between setting // hazptr and loading the block Idx = write_offset.fetch_add(1, std::memory_order_seq_cst); - data_block* block = static_cast( - Haz->write_block.load(std::memory_order_seq_cst) - ); + data_block* block = + static_cast(Haz->write_block.load(std::memory_order_seq_cst)); - [[maybe_unused]] size_t boff = - block->offset.load(std::memory_order_relaxed); + [[maybe_unused]] size_t boff = block->offset.load(std::memory_order_relaxed); assert(circular_less_than(actOff, 1 + Idx)); assert(circular_less_than(boff, 1 + Idx)); @@ -1056,9 +1024,7 @@ template class channel { if (is_closed_past(Idx)) [[unlikely]] { // Nothing will be written; release the hazard pointer so that this // token doesn't block reclamation while it is idle. - Haz->active_offset.store( - Idx + InactiveHazptrOffset, std::memory_order_release - ); + Haz->active_offset.store(Idx + InactiveHazptrOffset, std::memory_order_release); return nullptr; } block = find_block(block, Idx); @@ -1080,12 +1046,10 @@ template class channel { // hazptr and loading the block StartIdx = write_offset.fetch_add(Count, std::memory_order_seq_cst); EndIdx = StartIdx + Count; - data_block* block = static_cast( - Haz->write_block.load(std::memory_order_seq_cst) - ); + data_block* block = + static_cast(Haz->write_block.load(std::memory_order_seq_cst)); - [[maybe_unused]] size_t boff = - block->offset.load(std::memory_order_relaxed); + [[maybe_unused]] size_t boff = block->offset.load(std::memory_order_relaxed); assert(circular_less_than(actOff, 1 + StartIdx)); assert(circular_less_than(boff, 1 + StartIdx)); @@ -1098,9 +1062,7 @@ template class channel { if (is_closed_past(StartIdx)) [[unlikely]] { // Nothing will be written; release the hazard pointer so that this // token doesn't block reclamation while it is idle. - Haz->active_offset.store( - EndIdx + InactiveHazptrOffset, std::memory_order_release - ); + Haz->active_offset.store(EndIdx + InactiveHazptrOffset, std::memory_order_release); return nullptr; } @@ -1118,8 +1080,7 @@ template class channel { // Update last known block. Haz->write_block.store(protectBlock, std::memory_order_release); Haz->next_protect_write.store( - protectBlock->offset.load(std::memory_order_relaxed), - std::memory_order_relaxed + protectBlock->offset.load(std::memory_order_relaxed), std::memory_order_relaxed ); return startBlock; } @@ -1135,8 +1096,7 @@ template class channel { data_block* block = static_cast(Haz->read_block.load(std::memory_order_seq_cst)); - [[maybe_unused]] size_t boff = - block->offset.load(std::memory_order_relaxed); + [[maybe_unused]] size_t boff = block->offset.load(std::memory_order_relaxed); assert(circular_less_than(actOff, 1 + Idx)); assert(circular_less_than(boff, 1 + Idx)); @@ -1152,9 +1112,7 @@ template class channel { element* elem = &block->values[Idx & BlockSizeMask]; elem->set_not_waiting(); // Also release the hazard pointer now (nothing else to read) - Haz->active_offset.store( - Idx + InactiveHazptrOffset, std::memory_order_release - ); + Haz->active_offset.store(Idx + InactiveHazptrOffset, std::memory_order_release); return nullptr; } block = find_block(block, Idx); @@ -1207,9 +1165,7 @@ template class channel { haz->write_count.store(0, std::memory_order_relaxed); haz->next_protect_write.store(HeadOff, std::memory_order_relaxed); haz->next_protect_read.store(HeadOff, std::memory_order_relaxed); - haz->active_offset.store( - HeadOff + InactiveHazptrOffset, std::memory_order_relaxed - ); + haz->active_offset.store(HeadOff + InactiveHazptrOffset, std::memory_order_relaxed); haz->read_block.store(head, std::memory_order_relaxed); haz->write_block.store(head, std::memory_order_relaxed); @@ -1263,9 +1219,7 @@ template class channel { write_element(elem, std::forward(ConstructArgs)...); // Then release the hazard pointer - Haz->active_offset.store( - idx + InactiveHazptrOffset, std::memory_order_release - ); + Haz->active_offset.store(idx + InactiveHazptrOffset, std::memory_order_release); return true; } @@ -1297,9 +1251,7 @@ template class channel { } // Then release the hazard pointer - Haz->active_offset.store( - endIdx + InactiveHazptrOffset, std::memory_order_release - ); + Haz->active_offset.store(endIdx + InactiveHazptrOffset, std::memory_order_release); return true; } @@ -1332,10 +1284,9 @@ template class channel { } release_idx = idx + InactiveHazptrOffset; - if (parent.haz_ptr->write_count + parent.haz_ptr->read_count >= - ClusterPeriod) [[unlikely]] { - if (parent.haz_ptr->write_count + parent.haz_ptr->read_count == - ClusterPeriod) { + if (parent.haz_ptr->write_count + parent.haz_ptr->read_count >= ClusterPeriod) + [[unlikely]] { + if (parent.haz_ptr->write_count + parent.haz_ptr->read_count == ClusterPeriod) { size_t elapsed = parent.haz_ptr->elapsed(); size_t readerCount = 0; parent.haz_ptr->for_each_owned_hazptr( @@ -1359,14 +1310,10 @@ template class channel { // Try to get rti. Suspend if we can get it. // If we don't get it on this call, don't suspend and try // again to get it on the next call. - int rti = parent.haz_ptr->requested_thread_index.load( - std::memory_order_relaxed - ); + int rti = parent.haz_ptr->requested_thread_index.load(std::memory_order_relaxed); if (rti == -1) { if (parent.chan.try_cluster(parent.haz_ptr)) { - rti = parent.haz_ptr->requested_thread_index.load( - std::memory_order_relaxed - ); + rti = parent.haz_ptr->requested_thread_index.load(std::memory_order_relaxed); } } if (rti != -1) { @@ -1394,13 +1341,10 @@ template class channel { } bool await_suspend(std::coroutine_handle<> Outer) noexcept { - int rti = - parent.haz_ptr->requested_thread_index.load(std::memory_order_relaxed); + int rti = parent.haz_ptr->requested_thread_index.load(std::memory_order_relaxed); if (rti != -1) { thread_hint = static_cast(rti); - parent.haz_ptr->requested_thread_index.store( - -1, std::memory_order_relaxed - ); + parent.haz_ptr->requested_thread_index.store(-1, std::memory_order_relaxed); } base.continuation = Outer; @@ -1427,23 +1371,19 @@ template class channel { friend channel; - aw_pull_base(channel& Chan, hazard_ptr* Haz) noexcept - : chan(Chan), haz_ptr{Haz} {} + aw_pull_base(channel& Chan, hazard_ptr* Haz) noexcept : chan(Chan), haz_ptr{Haz} {} }; class aw_pull final : public aw_pull_base { friend chan_tok; - aw_pull(channel& Chan, hazard_ptr* Haz) noexcept - : aw_pull_base(Chan, Haz) {} + aw_pull(channel& Chan, hazard_ptr* Haz) noexcept : aw_pull_base(Chan, Haz) {} struct aw_pull_impl final : public aw_pull_base_impl { aw_pull_impl(aw_pull_base& Parent) noexcept : aw_pull_base_impl(Parent) {} TMC_AWAIT_RESUME std::optional await_resume() noexcept { if (aw_pull_base_impl::base.ok) { - std::optional result( - std::move(aw_pull_base_impl::elem->data.value) - ); + std::optional result(std::move(aw_pull_base_impl::elem->data.value)); aw_pull_base_impl::elem->data.destroy(); aw_pull_base_impl::parent.haz_ptr->active_offset.store( aw_pull_base_impl::release_idx, std::memory_order_release @@ -1462,13 +1402,11 @@ template class channel { class aw_pull_zc final : public aw_pull_base { friend chan_tok; - aw_pull_zc(channel& Chan, hazard_ptr* Haz) noexcept - : aw_pull_base(Chan, Haz) {} + aw_pull_zc(channel& Chan, hazard_ptr* Haz) noexcept : aw_pull_base(Chan, Haz) {} // Same as aw_pull but returns a scoped object instead struct aw_pull_zc_impl final : public aw_pull_base_impl { - aw_pull_zc_impl(aw_pull_base& Parent) noexcept - : aw_pull_base_impl(Parent) {} + aw_pull_zc_impl(aw_pull_base& Parent) noexcept : aw_pull_base_impl(Parent) {} TMC_AWAIT_RESUME std::optional> await_resume() noexcept { if (aw_pull_base_impl::base.ok) { @@ -1483,9 +1421,7 @@ template class channel { }; public: - aw_pull_zc_impl operator co_await() && noexcept { - return aw_pull_zc_impl(*this); - } + aw_pull_zc_impl operator co_await() && noexcept { return aw_pull_zc_impl(*this); } }; std::variant try_pull(hazard_ptr* Haz) { @@ -1504,30 +1440,24 @@ template class channel { if (circular_less_than(woff, Idx + 1)) { // If closed, continue draining until the channel is empty. if (is_closed_past(Idx)) [[unlikely]] { - Haz->active_offset.store( - Idx + InactiveHazptrOffset, std::memory_order_release - ); + Haz->active_offset.store(Idx + InactiveHazptrOffset, std::memory_order_release); return std::variant( std::in_place_index ); } // Release the hazard pointer so that this token doesn't block // reclamation while it is idle. - Haz->active_offset.store( - Idx + InactiveHazptrOffset, std::memory_order_release - ); + Haz->active_offset.store(Idx + InactiveHazptrOffset, std::memory_order_release); return std::variant( std::in_place_index ); } // Queue appears non-empty. See if data is ready for consumption at our // speculative Idx. - data_block* block = static_cast( - Haz->read_block.load(std::memory_order_seq_cst) - ); + data_block* block = + static_cast(Haz->read_block.load(std::memory_order_seq_cst)); - [[maybe_unused]] size_t boff = - block->offset.load(std::memory_order_relaxed); + [[maybe_unused]] size_t boff = block->offset.load(std::memory_order_relaxed); assert(circular_less_than(actOff, 1 + Idx)); assert(circular_less_than(boff, 1 + Idx)); @@ -1557,9 +1487,7 @@ template class channel { std::in_place_index, std::move(elem->data.value) ); elem->data.destroy(); - Haz->active_offset.store( - Idx + InactiveHazptrOffset, std::memory_order_release - ); + Haz->active_offset.store(Idx + InactiveHazptrOffset, std::memory_order_release); return result; } } else { @@ -1581,18 +1509,14 @@ template class channel { // the consumer retries. Otherwise no data can ever arrive at or // after Idx, and everything before Idx has been consumed. if (is_closed_past(Idx)) [[unlikely]] { - Haz->active_offset.store( - Idx + InactiveHazptrOffset, std::memory_order_release - ); + Haz->active_offset.store(Idx + InactiveHazptrOffset, std::memory_order_release); return std::variant( std::in_place_index ); } // Release the hazard pointer so that this token doesn't block // reclamation while it is idle. - Haz->active_offset.store( - Idx + InactiveHazptrOffset, std::memory_order_release - ); + Haz->active_offset.store(Idx + InactiveHazptrOffset, std::memory_order_release); return std::variant( std::in_place_index ); @@ -1616,10 +1540,9 @@ template class channel { aw_push_impl(aw_push& Parent) noexcept : parent{Parent} {} bool await_ready() noexcept { - if (parent.haz_ptr->write_count + parent.haz_ptr->read_count >= - ClusterPeriod) [[unlikely]] { - if (parent.haz_ptr->write_count + parent.haz_ptr->read_count == - ClusterPeriod) { + if (parent.haz_ptr->write_count + parent.haz_ptr->read_count >= ClusterPeriod) + [[unlikely]] { + if (parent.haz_ptr->write_count + parent.haz_ptr->read_count == ClusterPeriod) { size_t elapsed = parent.haz_ptr->elapsed(); size_t writerCount = 0; parent.haz_ptr->for_each_owned_hazptr( @@ -1644,14 +1567,12 @@ template class channel { // Try to get rti. Suspend if we can get it. // If we don't get it on this call, don't suspend and try // again to get it on the next call. - int rti = parent.haz_ptr->requested_thread_index.load( - std::memory_order_relaxed - ); + int rti = + parent.haz_ptr->requested_thread_index.load(std::memory_order_relaxed); if (rti == -1) { if (parent.chan.try_cluster(parent.haz_ptr)) { - rti = parent.haz_ptr->requested_thread_index.load( - std::memory_order_relaxed - ); + rti = + parent.haz_ptr->requested_thread_index.load(std::memory_order_relaxed); } } if (rti != -1) { @@ -1665,11 +1586,8 @@ template class channel { } void await_suspend(std::coroutine_handle<> Outer) noexcept { - size_t target = - static_cast(parent.haz_ptr->requested_thread_index); - parent.haz_ptr->requested_thread_index.store( - -1, std::memory_order_relaxed - ); + size_t target = static_cast(parent.haz_ptr->requested_thread_index); + parent.haz_ptr->requested_thread_index.store(-1, std::memory_order_relaxed); tmc::detail::post_checked( tmc::detail::this_thread::executor(), std::move(Outer), tmc::detail::this_thread::this_task().prio, target @@ -1693,8 +1611,7 @@ template class channel { size_t idx = StartIdx; block = find_block(block, idx); while (circular_less_than(idx, EndIdx)) { - auto cons = - block->values[idx & BlockSizeMask].spin_wait_for_waiting_consumer(); + auto cons = block->values[idx & BlockSizeMask].spin_wait_for_waiting_consumer(); if (cons != nullptr) { cons->ok = false; tmc::detail::post_checked( @@ -1717,9 +1634,7 @@ template class channel { size_t woff = write_offset.load(std::memory_order_seq_cst); // Setting this to a distant-but-greater value before setting closed // prevents consumers from exiting too early. - write_closed_at.store( - woff + InactiveHazptrOffset, std::memory_order_seq_cst - ); + write_closed_at.store(woff + InactiveHazptrOffset, std::memory_order_seq_cst); closed.store(WRITE_CLOSING_BIT, std::memory_order_seq_cst); @@ -1729,9 +1644,7 @@ template class channel { // the sentinel and will still be produced. woff = write_offset.fetch_add(1, std::memory_order_seq_cst); write_closed_at.store(woff, std::memory_order_seq_cst); - closed.store( - WRITE_CLOSING_BIT | WRITE_CLOSED_BIT, std::memory_order_seq_cst - ); + closed.store(WRITE_CLOSING_BIT | WRITE_CLOSED_BIT, std::memory_order_seq_cst); // Readers that already claimed slots >= write_closed_at will never // receive data. Wake them now. @@ -2008,8 +1921,7 @@ template class chan_tok { /// `chan_tok`, and before this `chan_tok` goes out of scope! The safest way /// to accomplish this is to tie its scope to the loop: /// `while (auto data = co_await chan.pull_zc()) { process(data->get()); }` - [[nodiscard("You must co_await pull_zc().")]] chan_t::aw_pull_zc - pull_zc() noexcept { + [[nodiscard("You must co_await pull_zc().")]] chan_t::aw_pull_zc pull_zc() noexcept { ASSERT_NO_CONCURRENT_ACCESS(); hazard_ptr* haz = get_hazard_ptr(); return typename chan_t::aw_pull_zc(*chan, haz); @@ -2050,9 +1962,7 @@ template class chan_tok { // Implementing handling for throwing construction is not possible with the // current design. This assert will also fire if no matching constructor can // be found for the iterator's dereferenced value. - static_assert( - std::is_nothrow_constructible_v - ); + static_assert(std::is_nothrow_constructible_v); hazard_ptr* haz = get_hazard_ptr(); return chan->post_bulk(haz, static_cast(Begin), Count); } @@ -2076,9 +1986,7 @@ template class chan_tok { // Implementing handling for throwing construction is not possible with the // current design. This assert will also fire if no matching constructor can // be found for the iterator's dereferenced value. - static_assert( - std::is_nothrow_constructible_v - ); + static_assert(std::is_nothrow_constructible_v); hazard_ptr* haz = get_hazard_ptr(); return chan->post_bulk( haz, static_cast(Begin), static_cast(End - Begin) @@ -2105,10 +2013,8 @@ template class chan_tok { // Implementing handling for throwing construction is not possible with the // current design. This assert will also fire if no matching constructor can // be found for the iterator's dereferenced value. - static_assert( - std::is_nothrow_constructible_v< - T, decltype(std::move(*static_cast(Range).begin()))> - ); + static_assert(std::is_nothrow_constructible_v< + T, decltype(std::move(*static_cast(Range).begin()))>); hazard_ptr* haz = get_hazard_ptr(); auto begin = static_cast(Range).begin(); auto end = static_cast(Range).end(); @@ -2184,8 +2090,7 @@ template class chan_tok { /// value of 2,000,000 represents an item being pushed every 500ns. This /// behavior can be disabled entirely by setting this to 0. chan_tok& set_heavy_load_threshold(size_t Threshold) noexcept { - size_t cycles = - Threshold == 0 ? 0 : TMC_CPU_FREQ * chan_t::ClusterPeriod / Threshold; + size_t cycles = Threshold == 0 ? 0 : TMC_CPU_FREQ * chan_t::ClusterPeriod / Threshold; chan->MinClusterCycles.store(cycles, std::memory_order_relaxed); return *this; } @@ -2195,8 +2100,7 @@ template class chan_tok { /// /// If the other token is from a different channel, this token will now point /// to that channel. - chan_tok(const chan_tok& Other) noexcept - : chan(Other.chan), haz_ptr{nullptr} {} + chan_tok(const chan_tok& Other) noexcept : chan(Other.chan), haz_ptr{nullptr} {} /// Copy Assignment: If the other token is from a different channel, this /// token will now point to that channel. diff --git a/include/tmc/detail/compat.hpp b/include/tmc/detail/compat.hpp index 5abd611c..c633ffb5 100644 --- a/include/tmc/detail/compat.hpp +++ b/include/tmc/detail/compat.hpp @@ -71,18 +71,6 @@ #endif #define TMC_CPU_X86 #define TMC_CPU_PAUSE _mm_pause -static inline size_t TMC_CPU_TIMESTAMP() noexcept { - return static_cast(__rdtsc()); -} -// Assume a 3.5GHz CPU if we can't get the value (on x86). -// Yes, this is hacky. Getting the real RDTSC freq requires -// waiting for another time source (system timer) and then dividing by -// that duration. This takes real time and would have to be done on -// startup. Using a 3.5GHz default means that slower processors will appear to -// be running faster, and vice versa. For the current usage of this (the -// clustering threshold in tmc::channel) this seems like reasonable behavior -// anyway. -static inline const size_t TMC_CPU_FREQ = 3500000000; #elif defined(__arm__) || defined(_M_ARM) || defined(_M_ARM64) || defined(__aarch64__) #if defined(__aarch64__) || defined(_M_ARM64) || \ (defined(__ARM_ARCH_PROFILE) && __ARM_ARCH_PROFILE == 'A') || \ @@ -103,89 +91,6 @@ static inline void TMC_CPU_PAUSE() noexcept { #endif #endif } -#if defined(_MSC_VER) && defined(_M_ARM64) -#define TMC_ARM64_SYSREG(op0, op1, crn, crm, op2) \ - (((op0 & 1) << 14) | ((op1 & 7) << 11) | ((crn & 15) << 7) | \ - ((crm & 15) << 3) | ((op2 & 7) << 0)) -static inline size_t TMC_CPU_TIMESTAMP() noexcept { - return static_cast(_ReadStatusReg(TMC_ARM64_SYSREG(3, 3, 14, 0, 2))); -} -static inline size_t TMC_ARM_CPU_FREQ() noexcept { - return static_cast(_ReadStatusReg(TMC_ARM64_SYSREG(3, 3, 14, 0, 0))); -} -#undef TMC_ARM64_SYSREG -#elif defined(_MSC_VER) && defined(_M_ARM) -static inline size_t TMC_CPU_TIMESTAMP() noexcept { - return static_cast(_MoveFromCoprocessor64(15, 1, 14)); -} -static inline size_t TMC_ARM_CPU_FREQ() noexcept { - return static_cast(_MoveFromCoprocessor(15, 0, 14, 0, 0)); -} -#elif defined(__aarch64__) || defined(_M_ARM64) -// AArch64: the generic timer is read through dedicated system registers. -// Read the ARM "Virtual Counter" register. -// This ticks at a frequency independent of the processor frequency. -// https://developer.arm.com/documentation/ddi0406/cb/System-Level-Architecture/The-Generic-Timer/About-the-Generic-Timer/The-virtual-counter?lang=en -static inline size_t TMC_CPU_TIMESTAMP() noexcept { - size_t count; - asm volatile("mrs %0, cntvct_el0; " : "=r"(count)::"memory"); - return count; -} -// Read the ARM "Virtual Counter" frequency. -static inline size_t TMC_ARM_CPU_FREQ() noexcept { - size_t freq; - asm volatile("mrs %0, cntfrq_el0; isb; " : "=r"(freq)::"memory"); - return freq; -} -#elif defined(__ARM_ARCH) && __ARM_ARCH >= 7 && defined(__ARM_ARCH_PROFILE) && \ - (__ARM_ARCH_PROFILE == 'A' || __ARM_ARCH_PROFILE == 'R') -#if defined(__linux__) -#include -#include -#endif -// AArch32 (e.g. armhf): the AArch64 `cntvct_el0` / `cntfrq_el0` system -// registers don't exist here. The ARMv7-A Generic Timer exposes the same -// counters through CP15 coprocessor accesses instead. -// https://developer.arm.com/documentation/ddi0406/cb/System-Level-Architecture/The-Generic-Timer/Register-descriptions?lang=en -// The Generic Timer is optional on ARMv7-A, and user-mode access is controlled -// by the OS. On Linux, HWCAP_EVTSTRM indicates that the kernel configured the -// architected timer event stream; if that is absent, fall back to inert timing -// rather than trapping on an undefined/privileged CP15 access. -static inline bool TMC_ARM_HAS_GENERIC_TIMER_IMPL() noexcept { -#if defined(__linux__) && defined(HWCAP_EVTSTRM) - return (getauxval(AT_HWCAP) & HWCAP_EVTSTRM) != 0; -#else - return false; -#endif -} -static inline const bool TMC_ARM_HAS_GENERIC_TIMER = TMC_ARM_HAS_GENERIC_TIMER_IMPL(); -// Read the 64-bit "Virtual Counter" (CNTVCT) via MRRC into a register pair. -// size_t is 32-bit here, so we return only the low word - matching the 32-bit -// x86 path, which likewise truncates __rdtsc(). The counter ticks at a -// frequency independent of the processor frequency. -static inline size_t TMC_CPU_TIMESTAMP() noexcept { - if (!TMC_ARM_HAS_GENERIC_TIMER) { - return 0; - } - uint32_t lo, hi; - asm volatile("mrrc p15, 1, %0, %1, c14" : "=r"(lo), "=r"(hi)::"memory"); - (void)hi; - return lo; -} -// Read the "Virtual Counter" frequency (CNTFRQ, a 32-bit register) via MRC. -static inline size_t TMC_ARM_CPU_FREQ() noexcept { - if (!TMC_ARM_HAS_GENERIC_TIMER) { - return 1000000000; - } - uint32_t freq; - asm volatile("mrc p15, 0, %0, c14, c0, 0" : "=r"(freq)::"memory"); - return freq; -} -#else -static inline size_t TMC_CPU_TIMESTAMP() noexcept { return 0; } -static inline size_t TMC_ARM_CPU_FREQ() noexcept { return 1000000000; } -#endif -static inline const size_t TMC_CPU_FREQ = TMC_ARM_CPU_FREQ(); #elif defined(__loongarch__) && defined(__LP64__) // Use some barrier instructions to generate a delay, as there's // no instruction dedicated for the delay in spinlock :(. @@ -193,31 +98,7 @@ static inline void TMC_CPU_PAUSE() noexcept { for (int i = 0; i < 32; i++) asm volatile("ibar 0"); } -// Read the LoongArch stable counter -static inline size_t TMC_CPU_TIMESTAMP() noexcept { - size_t count; - asm volatile("rdtime.d %0, $zero" : "=r"(count)); - return count; -} -// Calculate the LoongArch stable counter frequency -static inline size_t TMC_LOONGARCH_CPU_FREQ() noexcept { - size_t cc_freq; - asm("cpucfg %0, %1" : "=r"(cc_freq) : "r"(0x4)); - - size_t cc_mul_div; - asm("cpucfg %0, %1" : "=r"(cc_mul_div) : "r"(0x5)); - - size_t cc_mul = cc_mul_div & 0xffff; - size_t cc_div = (cc_mul_div >> 16) & 0xffff; - - return cc_freq * cc_mul / cc_div; -} - -static inline const size_t TMC_CPU_FREQ = TMC_LOONGARCH_CPU_FREQ(); #elif defined(__riscv) -#if defined(__linux__) -#include -#endif #define TMC_CPU_RISCV static inline void TMC_CPU_PAUSE() noexcept { #if defined(__riscv_zihintpause) @@ -226,46 +107,9 @@ static inline void TMC_CPU_PAUSE() noexcept { asm volatile("nop" ::: "memory"); #endif } -static inline size_t TMC_RISCV_READ_TIMEBASE_FREQ() noexcept { -#if defined(__linux__) - const char* paths[] = { - "/proc/device-tree/cpus/timebase-frequency", - "/sys/firmware/devicetree/base/cpus/timebase-frequency", - }; - for (const char* path : paths) { - auto* file = std::fopen(path, "rb"); - if (file == nullptr) { - continue; - } - unsigned char bytes[8] = {}; - const auto size = std::fread(bytes, 1, sizeof(bytes), file); - std::fclose(file); - if (size == 4 || size == 8) { - size_t result = 0; - for (size_t i = 0; i != size; ++i) { - result = (result << 8) | bytes[i]; - } - if (result != 0) { - return result; - } - } - } -#endif - // This is only used for heuristics if the real RISC-V timebase is unavailable. - // Use the 2GHz clock frequency of SG2042-class server chips as a default. - return 2000000000; -} -static inline size_t TMC_CPU_TIMESTAMP() noexcept { - size_t count; - asm volatile("rdtime %0" : "=r"(count)); - return count; -} -static inline const size_t TMC_CPU_FREQ = TMC_RISCV_READ_TIMEBASE_FREQ(); #else // Fallback for unknown architectures static inline void TMC_CPU_PAUSE() noexcept {} -static inline size_t TMC_CPU_TIMESTAMP() noexcept { return 0; } -static inline const size_t TMC_CPU_FREQ = 1000000000; #endif // clang-format tries to collapse the pragmas into one line... diff --git a/include/tmc/detail/timer_compat.hpp b/include/tmc/detail/timer_compat.hpp new file mode 100644 index 00000000..a929785b --- /dev/null +++ b/include/tmc/detail/timer_compat.hpp @@ -0,0 +1,188 @@ +// Copyright (c) 2023-2026 Logan McDougall +// +// Distributed under the Boost Software License, Version 1.0. (See accompanying +// file LICENSE or copy at http://www.boost.org/LICENSE_1_0.txt) + +#pragma once + +// CPU timestamp-counter facilities (TMC_CPU_TIMESTAMP / TMC_CPU_FREQ) used by +// tmc::channel for load-based clustering. Split out of compat.hpp because +// channel.hpp is the only consumer. The architecture #if/#elif chain below +// mirrors the one in compat.hpp; when adding support for a new architecture, +// update both. + +#include +#include // IWYU pragma: keep + +#if defined(__x86_64__) || defined(_M_AMD64) || defined(i386) || defined(__i386__) || \ + defined(__i386) || defined(_M_IX86) +#ifdef _MSC_VER +#include +#else +#include +#endif +static inline size_t TMC_CPU_TIMESTAMP() noexcept { + return static_cast(__rdtsc()); +} +// Assume a 3.5GHz CPU if we can't get the value (on x86). +// Yes, this is hacky. Getting the real RDTSC freq requires +// waiting for another time source (system timer) and then dividing by +// that duration. This takes real time and would have to be done on +// startup. Using a 3.5GHz default means that slower processors will appear to +// be running faster, and vice versa. For the current usage of this (the +// clustering threshold in tmc::channel) this seems like reasonable behavior +// anyway. +static inline const size_t TMC_CPU_FREQ = 3500000000; +#elif defined(__arm__) || defined(_M_ARM) || defined(_M_ARM64) || defined(__aarch64__) +#ifdef _MSC_VER +#include +#endif +#if defined(_MSC_VER) && defined(_M_ARM64) +#define TMC_ARM64_SYSREG(op0, op1, crn, crm, op2) \ + (((op0 & 1) << 14) | ((op1 & 7) << 11) | ((crn & 15) << 7) | ((crm & 15) << 3) | \ + ((op2 & 7) << 0)) +static inline size_t TMC_CPU_TIMESTAMP() noexcept { + return static_cast(_ReadStatusReg(TMC_ARM64_SYSREG(3, 3, 14, 0, 2))); +} +static inline size_t TMC_ARM_CPU_FREQ() noexcept { + return static_cast(_ReadStatusReg(TMC_ARM64_SYSREG(3, 3, 14, 0, 0))); +} +#undef TMC_ARM64_SYSREG +#elif defined(_MSC_VER) && defined(_M_ARM) +static inline size_t TMC_CPU_TIMESTAMP() noexcept { + return static_cast(_MoveFromCoprocessor64(15, 1, 14)); +} +static inline size_t TMC_ARM_CPU_FREQ() noexcept { + return static_cast(_MoveFromCoprocessor(15, 0, 14, 0, 0)); +} +#elif defined(__aarch64__) || defined(_M_ARM64) +// AArch64: the generic timer is read through dedicated system registers. +// Read the ARM "Virtual Counter" register. +// This ticks at a frequency independent of the processor frequency. +// https://developer.arm.com/documentation/ddi0406/cb/System-Level-Architecture/The-Generic-Timer/About-the-Generic-Timer/The-virtual-counter?lang=en +static inline size_t TMC_CPU_TIMESTAMP() noexcept { + size_t count; + asm volatile("mrs %0, cntvct_el0; " : "=r"(count)::"memory"); + return count; +} +// Read the ARM "Virtual Counter" frequency. +static inline size_t TMC_ARM_CPU_FREQ() noexcept { + size_t freq; + asm volatile("mrs %0, cntfrq_el0; isb; " : "=r"(freq)::"memory"); + return freq; +} +#elif defined(__ARM_ARCH) && __ARM_ARCH >= 7 && defined(__ARM_ARCH_PROFILE) && \ + (__ARM_ARCH_PROFILE == 'A' || __ARM_ARCH_PROFILE == 'R') +#if defined(__linux__) +#include +#include +#endif +// AArch32 (e.g. armhf): the AArch64 `cntvct_el0` / `cntfrq_el0` system +// registers don't exist here. The ARMv7-A Generic Timer exposes the same +// counters through CP15 coprocessor accesses instead. +// https://developer.arm.com/documentation/ddi0406/cb/System-Level-Architecture/The-Generic-Timer/Register-descriptions?lang=en +// The Generic Timer is optional on ARMv7-A, and user-mode access is controlled +// by the OS. On Linux, HWCAP_EVTSTRM indicates that the kernel configured the +// architected timer event stream; if that is absent, fall back to inert timing +// rather than trapping on an undefined/privileged CP15 access. +static inline bool TMC_ARM_HAS_GENERIC_TIMER_IMPL() noexcept { +#if defined(__linux__) && defined(HWCAP_EVTSTRM) + return (getauxval(AT_HWCAP) & HWCAP_EVTSTRM) != 0; +#else + return false; +#endif +} +static inline const bool TMC_ARM_HAS_GENERIC_TIMER = TMC_ARM_HAS_GENERIC_TIMER_IMPL(); +// Read the 64-bit "Virtual Counter" (CNTVCT) via MRRC into a register pair. +// size_t is 32-bit here, so we return only the low word - matching the 32-bit +// x86 path, which likewise truncates __rdtsc(). The counter ticks at a +// frequency independent of the processor frequency. +static inline size_t TMC_CPU_TIMESTAMP() noexcept { + if (!TMC_ARM_HAS_GENERIC_TIMER) { + return 0; + } + uint32_t lo, hi; + asm volatile("mrrc p15, 1, %0, %1, c14" : "=r"(lo), "=r"(hi)::"memory"); + (void)hi; + return lo; +} +// Read the "Virtual Counter" frequency (CNTFRQ, a 32-bit register) via MRC. +static inline size_t TMC_ARM_CPU_FREQ() noexcept { + if (!TMC_ARM_HAS_GENERIC_TIMER) { + return 1000000000; + } + uint32_t freq; + asm volatile("mrc p15, 0, %0, c14, c0, 0" : "=r"(freq)::"memory"); + return freq; +} +#else +static inline size_t TMC_CPU_TIMESTAMP() noexcept { return 0; } +static inline size_t TMC_ARM_CPU_FREQ() noexcept { return 1000000000; } +#endif +static inline const size_t TMC_CPU_FREQ = TMC_ARM_CPU_FREQ(); +#elif defined(__loongarch__) && defined(__LP64__) +// Read the LoongArch stable counter +static inline size_t TMC_CPU_TIMESTAMP() noexcept { + size_t count; + asm volatile("rdtime.d %0, $zero" : "=r"(count)); + return count; +} +// Calculate the LoongArch stable counter frequency +static inline size_t TMC_LOONGARCH_CPU_FREQ() noexcept { + size_t cc_freq; + asm("cpucfg %0, %1" : "=r"(cc_freq) : "r"(0x4)); + + size_t cc_mul_div; + asm("cpucfg %0, %1" : "=r"(cc_mul_div) : "r"(0x5)); + + size_t cc_mul = cc_mul_div & 0xffff; + size_t cc_div = (cc_mul_div >> 16) & 0xffff; + + return cc_freq * cc_mul / cc_div; +} + +static inline const size_t TMC_CPU_FREQ = TMC_LOONGARCH_CPU_FREQ(); +#elif defined(__riscv) +#if defined(__linux__) +#include +#endif +static inline size_t TMC_RISCV_READ_TIMEBASE_FREQ() noexcept { +#if defined(__linux__) + const char* paths[] = { + "/proc/device-tree/cpus/timebase-frequency", + "/sys/firmware/devicetree/base/cpus/timebase-frequency", + }; + for (const char* path : paths) { + auto* file = std::fopen(path, "rb"); + if (file == nullptr) { + continue; + } + unsigned char bytes[8] = {}; + const auto size = std::fread(bytes, 1, sizeof(bytes), file); + std::fclose(file); + if (size == 4 || size == 8) { + size_t result = 0; + for (size_t i = 0; i != size; ++i) { + result = (result << 8) | bytes[i]; + } + if (result != 0) { + return result; + } + } + } +#endif + // This is only used for heuristics if the real RISC-V timebase is unavailable. + // Use the 2GHz clock frequency of SG2042-class server chips as a default. + return 2000000000; +} +static inline size_t TMC_CPU_TIMESTAMP() noexcept { + size_t count; + asm volatile("rdtime %0" : "=r"(count)); + return count; +} +static inline const size_t TMC_CPU_FREQ = TMC_RISCV_READ_TIMEBASE_FREQ(); +#else +// Fallback for unknown architectures +static inline size_t TMC_CPU_TIMESTAMP() noexcept { return 0; } +static inline const size_t TMC_CPU_FREQ = 1000000000; +#endif From 61b62b23da3c19f6c571725b0bc4ad0c5e68d979 Mon Sep 17 00:00:00 2001 From: tzcnt Date: Fri, 3 Jul 2026 16:12:28 -0700 Subject: [PATCH 4/4] un-format channel --- include/tmc/channel.hpp | 285 +++++++++++++++++++++++++++------------- 1 file changed, 191 insertions(+), 94 deletions(-) diff --git a/include/tmc/channel.hpp b/include/tmc/channel.hpp index e1614402..a2ac2fd1 100644 --- a/include/tmc/channel.hpp +++ b/include/tmc/channel.hpp @@ -72,7 +72,8 @@ struct chan_default_config { /// but access to a single token from multiple threads is not. /// To access the channel from multiple threads or tasks concurrently, /// make a copy of the token for each (by using the copy constructor). -template class chan_tok; +template +class chan_tok; /// Creates a new channel and returns an access token to it. template @@ -115,9 +116,11 @@ namespace detail { // remain unconditionally seq_cst - it is followed by a plain seq_cst load // (not an RMW), which provides no StoreLoad ordering even on x86. #if defined(TMC_CPU_X86) || defined(TMC_CPU_ARM) -inline constexpr std::memory_order hazptr_protect_order = std::memory_order_relaxed; +inline constexpr std::memory_order hazptr_protect_order = + std::memory_order_relaxed; #else -inline constexpr std::memory_order hazptr_protect_order = std::memory_order_seq_cst; +inline constexpr std::memory_order hazptr_protect_order = + std::memory_order_seq_cst; #endif // Hazard pointer type used internally by channel to track in-use blocks. @@ -139,8 +142,8 @@ class alignas(TMC_CACHE_LINE_SIZE) hazard_ptr { template friend class tmc::channel; template friend class tmc::chan_zc_scope; - static inline constexpr size_t InactiveHazptrOffset = TMC_ONE_BIT - << (TMC_PLATFORM_BITS - 2); + static inline constexpr size_t InactiveHazptrOffset = + TMC_ONE_BIT << (TMC_PLATFORM_BITS - 2); inline void release_blocks() noexcept { // These elements may be read (by try_reclaim_block()) after @@ -247,7 +250,8 @@ template class chan_zc_scope { chan_zc_scope(const chan_zc_scope&) = delete; chan_zc_scope& operator=(const chan_zc_scope&) = delete; chan_zc_scope(chan_zc_scope&& Other) noexcept - : haz_ptr{Other.haz_ptr}, data{Other.data}, release_idx{Other.release_idx} { + : haz_ptr{Other.haz_ptr}, data{Other.data}, + release_idx{Other.release_idx} { Other.data = nullptr; } @@ -281,7 +285,8 @@ template class channel { static inline constexpr size_t BlockSize = Config::BlockSize; static inline constexpr size_t BlockSizeMask = BlockSize - 1; static_assert( - BlockSize && ((BlockSize & (BlockSize - 1)) == 0), "BlockSize must be a power of 2" + BlockSize && ((BlockSize & (BlockSize - 1)) == 0), + "BlockSize must be a power of 2" ); // Ensure that the subtraction of unsigned offsets always results in a value @@ -299,8 +304,8 @@ template class channel { // elements and then cannot free any blocks until that thread wakes. This is // extremely unlikely, and not an error - it will just prevent block // reclamation. On 64 bit in practice this will never happen. - static inline constexpr size_t InactiveHazptrOffset = TMC_ONE_BIT - << (TMC_PLATFORM_BITS - 2); + static inline constexpr size_t InactiveHazptrOffset = + TMC_ONE_BIT << (TMC_PLATFORM_BITS - 2); // Defaults to 2M items per second; this is 1 item every 500ns. static inline constexpr size_t DefaultHeavyLoadThreshold = 2000000; @@ -312,7 +317,8 @@ template class channel { static inline constexpr size_t ClusterPeriod = 10000; friend chan_tok; - template friend chan_tok make_channel() noexcept; + template + friend chan_tok make_channel() noexcept; public: class aw_pull; @@ -340,14 +346,16 @@ template class channel { static constexpr size_t UNPADLEN = sizeof(size_t) + sizeof(void*) + sizeof(tmc::detail::qu_storage); - static constexpr size_t WANTLEN = - (UNPADLEN + TMC_CACHE_LINE_SIZE - 1) & - static_cast(0 - TMC_CACHE_LINE_SIZE); // round up to TMC_CACHE_LINE_SIZE - static constexpr size_t PADLEN = UNPADLEN < WANTLEN ? (WANTLEN - UNPADLEN) : 999; + static constexpr size_t WANTLEN = (UNPADLEN + TMC_CACHE_LINE_SIZE - 1) & + static_cast( + 0 - TMC_CACHE_LINE_SIZE + ); // round up to TMC_CACHE_LINE_SIZE + static constexpr size_t PADLEN = + UNPADLEN < WANTLEN ? (WANTLEN - UNPADLEN) : 999; struct empty {}; - using Padding = - std::conditional_t; + using Padding = std::conditional_t< + Config::PackingLevel == 0 && PADLEN != 999, char[PADLEN], empty>; TMC_NO_UNIQUE_ADDRESS Padding pad; // If this returns false, data is ready and consumer should not wait. @@ -364,7 +372,8 @@ template class channel { consumer_base* set_data_ready_or_get_waiting_consumer() noexcept { uintptr_t expected = 0; if (flags.compare_exchange_strong( - expected, DATA_BIT, std::memory_order_acq_rel, std::memory_order_acquire + expected, DATA_BIT, std::memory_order_acq_rel, + std::memory_order_acquire )) { return nullptr; } else { @@ -376,7 +385,9 @@ template class channel { // because it saw the closed flag and returned. // This does not need to be called during normal operation - only after the // closed flag is observed. - void set_not_waiting() noexcept { flags.store(BOTH_BITS, std::memory_order_release); } + void set_not_waiting() noexcept { + flags.store(BOTH_BITS, std::memory_order_release); + } // Used by close() to find consumers that were already waiting on a slot // that will never receive data. @@ -439,7 +450,9 @@ template class channel { // This does not need to be called during normal operation - only after the // closed flag is observed. void set_not_waiting() noexcept { - flags.store(reinterpret_cast(BOTH_BITS), std::memory_order_release); + flags.store( + reinterpret_cast(BOTH_BITS), std::memory_order_release + ); } // Used by close() to find consumers that were already waiting on a slot @@ -535,7 +548,8 @@ template class channel { std::atomic tail_block; struct empty {}; - using EmbeddedBlock = std::conditional_t; + using EmbeddedBlock = + std::conditional_t; TMC_NO_UNIQUE_ADDRESS EmbeddedBlock embedded_block; channel() noexcept { @@ -608,7 +622,9 @@ template class channel { } for (size_t i = 0; i < ClusterOn.size(); ++i) { - ClusterOn[i].id->requested_thread_index.store(closest, std::memory_order_relaxed); + ClusterOn[i].id->requested_thread_index.store( + closest, std::memory_order_relaxed + ); } } @@ -708,7 +724,8 @@ template class channel { // (see hazptr_protect_order), this guarantees that if a token observed the // pre-CAS block pointer, the revalidation here will observe that token's // active_offset protection. - static inline void keep_min(size_t& Dst, std::atomic const& Src) noexcept { + static inline void + keep_min(size_t& Dst, std::atomic const& Src) noexcept { size_t val = Src.load(std::memory_order_seq_cst); if (circular_less_than(val, Dst)) { Dst = val; @@ -738,7 +755,9 @@ template class channel { return; } if (circular_less_than( - static_cast(block)->offset.load(std::memory_order_relaxed), + static_cast(block)->offset.load( + std::memory_order_relaxed + ), NewHead->offset.load(std::memory_order_relaxed) )) { if (!DstBlock.compare_exchange_strong( @@ -751,8 +770,9 @@ template class channel { // If this hazptr updated its own block, but the updated block is // still earlier than the new head, then we cannot free that block. keep_min( - MinProtected, - static_cast(block)->offset.load(std::memory_order_relaxed) + MinProtected, static_cast(block)->offset.load( + std::memory_order_relaxed + ) ); } // Reload hazptr after trying to modify block to ensure that if it was @@ -765,8 +785,9 @@ template class channel { // the first block that is protected by a hazard pointer. This block is // returned to become the NewHead. If OldHead is protected, then it will be // returned unchanged, and no blocks can be reclaimed. - data_block* - try_advance_head(hazard_ptr* Haz, data_block* OldHead, size_t ProtectIdx) noexcept { + data_block* try_advance_head( + hazard_ptr* Haz, data_block* OldHead, size_t ProtectIdx + ) noexcept { // In the current implementation, this is called only from consumers. // Therefore, this token's hazptr will be active, and protecting read_block. // However, if producers are lagging behind, and no producer is currently @@ -788,9 +809,9 @@ template class channel { // Find the block associated with this offset. data_block* newHead = OldHead; - while ( - circular_less_than(newHead->offset.load(std::memory_order_relaxed), ProtectIdx) - ) { + while (circular_less_than( + newHead->offset.load(std::memory_order_relaxed), ProtectIdx + )) { newHead = newHead->next.load(std::memory_order_acquire); } @@ -809,11 +830,13 @@ template class channel { // ProtectIdx may have been reduced by the double-check in // try_advance_block. If so, reduce newHead as well. - if (circular_less_than(ProtectIdx, newHead->offset.load(std::memory_order_relaxed))) { + if (circular_less_than( + ProtectIdx, newHead->offset.load(std::memory_order_relaxed) + )) { newHead = OldHead; - while ( - circular_less_than(newHead->offset.load(std::memory_order_relaxed), ProtectIdx) - ) { + while (circular_less_than( + newHead->offset.load(std::memory_order_relaxed), ProtectIdx + )) { newHead = newHead->next; } } @@ -868,11 +891,14 @@ template class channel { } // Actually unlink the blocks from the head of the queue. // They stay linked to each other. - unlinked[unlinkedCount - 1]->next.store(nullptr, std::memory_order_release); + unlinked[unlinkedCount - 1]->next.store( + nullptr, std::memory_order_release + ); while (true) { // Update their offsets to the end of the queue. - size_t boff = tailBlock->offset.load(std::memory_order_relaxed) + BlockSize; + size_t boff = + tailBlock->offset.load(std::memory_order_relaxed) + BlockSize; for (size_t i = 0; i < unlinkedCount; ++i) { unlinked[i]->offset.store(boff, std::memory_order_relaxed); boff += BlockSize; @@ -880,7 +906,8 @@ template class channel { // Re-link the tail of the queue to the head of the unlinked blocks. if (tailBlock->next.compare_exchange_strong( - next, unlinked[0], std::memory_order_acq_rel, std::memory_order_acquire + next, unlinked[0], std::memory_order_acq_rel, + std::memory_order_acquire )) { break; } @@ -951,7 +978,8 @@ template class channel { if (next == nullptr) { data_block* newBlock = new data_block(offset + BlockSize); if (Block->next.compare_exchange_strong( - next, newBlock, std::memory_order_acq_rel, std::memory_order_acquire + next, newBlock, std::memory_order_acq_rel, + std::memory_order_acquire )) { next = newBlock; } else { @@ -987,7 +1015,8 @@ template class channel { // load below sees the value close() stored before its release of `closed`. // - Non-blocking try_pull may pass acquire: it never blocks, so eventual // visibility is sufficient. - template bool is_closed_past(size_t Idx) const noexcept { + template + bool is_closed_past(size_t Idx) const noexcept { auto closedState = closed.load(Order); if (0 == closedState) [[likely]] { return false; @@ -997,7 +1026,9 @@ template class channel { TMC_CPU_PAUSE(); closedState = closed.load(std::memory_order_acquire); } - return circular_less_than(write_closed_at.load(std::memory_order_relaxed), 1 + Idx); + return circular_less_than( + write_closed_at.load(std::memory_order_relaxed), 1 + Idx + ); } // Idx will be initialized by this function @@ -1008,10 +1039,12 @@ template class channel { // seq_cst is needed here to create a StoreLoad barrier between setting // hazptr and loading the block Idx = write_offset.fetch_add(1, std::memory_order_seq_cst); - data_block* block = - static_cast(Haz->write_block.load(std::memory_order_seq_cst)); + data_block* block = static_cast( + Haz->write_block.load(std::memory_order_seq_cst) + ); - [[maybe_unused]] size_t boff = block->offset.load(std::memory_order_relaxed); + [[maybe_unused]] size_t boff = + block->offset.load(std::memory_order_relaxed); assert(circular_less_than(actOff, 1 + Idx)); assert(circular_less_than(boff, 1 + Idx)); @@ -1024,7 +1057,9 @@ template class channel { if (is_closed_past(Idx)) [[unlikely]] { // Nothing will be written; release the hazard pointer so that this // token doesn't block reclamation while it is idle. - Haz->active_offset.store(Idx + InactiveHazptrOffset, std::memory_order_release); + Haz->active_offset.store( + Idx + InactiveHazptrOffset, std::memory_order_release + ); return nullptr; } block = find_block(block, Idx); @@ -1046,10 +1081,12 @@ template class channel { // hazptr and loading the block StartIdx = write_offset.fetch_add(Count, std::memory_order_seq_cst); EndIdx = StartIdx + Count; - data_block* block = - static_cast(Haz->write_block.load(std::memory_order_seq_cst)); + data_block* block = static_cast( + Haz->write_block.load(std::memory_order_seq_cst) + ); - [[maybe_unused]] size_t boff = block->offset.load(std::memory_order_relaxed); + [[maybe_unused]] size_t boff = + block->offset.load(std::memory_order_relaxed); assert(circular_less_than(actOff, 1 + StartIdx)); assert(circular_less_than(boff, 1 + StartIdx)); @@ -1062,7 +1099,9 @@ template class channel { if (is_closed_past(StartIdx)) [[unlikely]] { // Nothing will be written; release the hazard pointer so that this // token doesn't block reclamation while it is idle. - Haz->active_offset.store(EndIdx + InactiveHazptrOffset, std::memory_order_release); + Haz->active_offset.store( + EndIdx + InactiveHazptrOffset, std::memory_order_release + ); return nullptr; } @@ -1080,7 +1119,8 @@ template class channel { // Update last known block. Haz->write_block.store(protectBlock, std::memory_order_release); Haz->next_protect_write.store( - protectBlock->offset.load(std::memory_order_relaxed), std::memory_order_relaxed + protectBlock->offset.load(std::memory_order_relaxed), + std::memory_order_relaxed ); return startBlock; } @@ -1096,7 +1136,8 @@ template class channel { data_block* block = static_cast(Haz->read_block.load(std::memory_order_seq_cst)); - [[maybe_unused]] size_t boff = block->offset.load(std::memory_order_relaxed); + [[maybe_unused]] size_t boff = + block->offset.load(std::memory_order_relaxed); assert(circular_less_than(actOff, 1 + Idx)); assert(circular_less_than(boff, 1 + Idx)); @@ -1112,7 +1153,9 @@ template class channel { element* elem = &block->values[Idx & BlockSizeMask]; elem->set_not_waiting(); // Also release the hazard pointer now (nothing else to read) - Haz->active_offset.store(Idx + InactiveHazptrOffset, std::memory_order_release); + Haz->active_offset.store( + Idx + InactiveHazptrOffset, std::memory_order_release + ); return nullptr; } block = find_block(block, Idx); @@ -1165,7 +1208,9 @@ template class channel { haz->write_count.store(0, std::memory_order_relaxed); haz->next_protect_write.store(HeadOff, std::memory_order_relaxed); haz->next_protect_read.store(HeadOff, std::memory_order_relaxed); - haz->active_offset.store(HeadOff + InactiveHazptrOffset, std::memory_order_relaxed); + haz->active_offset.store( + HeadOff + InactiveHazptrOffset, std::memory_order_relaxed + ); haz->read_block.store(head, std::memory_order_relaxed); haz->write_block.store(head, std::memory_order_relaxed); @@ -1219,7 +1264,9 @@ template class channel { write_element(elem, std::forward(ConstructArgs)...); // Then release the hazard pointer - Haz->active_offset.store(idx + InactiveHazptrOffset, std::memory_order_release); + Haz->active_offset.store( + idx + InactiveHazptrOffset, std::memory_order_release + ); return true; } @@ -1251,7 +1298,9 @@ template class channel { } // Then release the hazard pointer - Haz->active_offset.store(endIdx + InactiveHazptrOffset, std::memory_order_release); + Haz->active_offset.store( + endIdx + InactiveHazptrOffset, std::memory_order_release + ); return true; } @@ -1284,9 +1333,10 @@ template class channel { } release_idx = idx + InactiveHazptrOffset; - if (parent.haz_ptr->write_count + parent.haz_ptr->read_count >= ClusterPeriod) - [[unlikely]] { - if (parent.haz_ptr->write_count + parent.haz_ptr->read_count == ClusterPeriod) { + if (parent.haz_ptr->write_count + parent.haz_ptr->read_count >= + ClusterPeriod) [[unlikely]] { + if (parent.haz_ptr->write_count + parent.haz_ptr->read_count == + ClusterPeriod) { size_t elapsed = parent.haz_ptr->elapsed(); size_t readerCount = 0; parent.haz_ptr->for_each_owned_hazptr( @@ -1310,10 +1360,14 @@ template class channel { // Try to get rti. Suspend if we can get it. // If we don't get it on this call, don't suspend and try // again to get it on the next call. - int rti = parent.haz_ptr->requested_thread_index.load(std::memory_order_relaxed); + int rti = parent.haz_ptr->requested_thread_index.load( + std::memory_order_relaxed + ); if (rti == -1) { if (parent.chan.try_cluster(parent.haz_ptr)) { - rti = parent.haz_ptr->requested_thread_index.load(std::memory_order_relaxed); + rti = parent.haz_ptr->requested_thread_index.load( + std::memory_order_relaxed + ); } } if (rti != -1) { @@ -1341,10 +1395,13 @@ template class channel { } bool await_suspend(std::coroutine_handle<> Outer) noexcept { - int rti = parent.haz_ptr->requested_thread_index.load(std::memory_order_relaxed); + int rti = + parent.haz_ptr->requested_thread_index.load(std::memory_order_relaxed); if (rti != -1) { thread_hint = static_cast(rti); - parent.haz_ptr->requested_thread_index.store(-1, std::memory_order_relaxed); + parent.haz_ptr->requested_thread_index.store( + -1, std::memory_order_relaxed + ); } base.continuation = Outer; @@ -1371,19 +1428,23 @@ template class channel { friend channel; - aw_pull_base(channel& Chan, hazard_ptr* Haz) noexcept : chan(Chan), haz_ptr{Haz} {} + aw_pull_base(channel& Chan, hazard_ptr* Haz) noexcept + : chan(Chan), haz_ptr{Haz} {} }; class aw_pull final : public aw_pull_base { friend chan_tok; - aw_pull(channel& Chan, hazard_ptr* Haz) noexcept : aw_pull_base(Chan, Haz) {} + aw_pull(channel& Chan, hazard_ptr* Haz) noexcept + : aw_pull_base(Chan, Haz) {} struct aw_pull_impl final : public aw_pull_base_impl { aw_pull_impl(aw_pull_base& Parent) noexcept : aw_pull_base_impl(Parent) {} TMC_AWAIT_RESUME std::optional await_resume() noexcept { if (aw_pull_base_impl::base.ok) { - std::optional result(std::move(aw_pull_base_impl::elem->data.value)); + std::optional result( + std::move(aw_pull_base_impl::elem->data.value) + ); aw_pull_base_impl::elem->data.destroy(); aw_pull_base_impl::parent.haz_ptr->active_offset.store( aw_pull_base_impl::release_idx, std::memory_order_release @@ -1402,11 +1463,13 @@ template class channel { class aw_pull_zc final : public aw_pull_base { friend chan_tok; - aw_pull_zc(channel& Chan, hazard_ptr* Haz) noexcept : aw_pull_base(Chan, Haz) {} + aw_pull_zc(channel& Chan, hazard_ptr* Haz) noexcept + : aw_pull_base(Chan, Haz) {} // Same as aw_pull but returns a scoped object instead struct aw_pull_zc_impl final : public aw_pull_base_impl { - aw_pull_zc_impl(aw_pull_base& Parent) noexcept : aw_pull_base_impl(Parent) {} + aw_pull_zc_impl(aw_pull_base& Parent) noexcept + : aw_pull_base_impl(Parent) {} TMC_AWAIT_RESUME std::optional> await_resume() noexcept { if (aw_pull_base_impl::base.ok) { @@ -1421,7 +1484,9 @@ template class channel { }; public: - aw_pull_zc_impl operator co_await() && noexcept { return aw_pull_zc_impl(*this); } + aw_pull_zc_impl operator co_await() && noexcept { + return aw_pull_zc_impl(*this); + } }; std::variant try_pull(hazard_ptr* Haz) { @@ -1440,24 +1505,30 @@ template class channel { if (circular_less_than(woff, Idx + 1)) { // If closed, continue draining until the channel is empty. if (is_closed_past(Idx)) [[unlikely]] { - Haz->active_offset.store(Idx + InactiveHazptrOffset, std::memory_order_release); + Haz->active_offset.store( + Idx + InactiveHazptrOffset, std::memory_order_release + ); return std::variant( std::in_place_index ); } // Release the hazard pointer so that this token doesn't block // reclamation while it is idle. - Haz->active_offset.store(Idx + InactiveHazptrOffset, std::memory_order_release); + Haz->active_offset.store( + Idx + InactiveHazptrOffset, std::memory_order_release + ); return std::variant( std::in_place_index ); } // Queue appears non-empty. See if data is ready for consumption at our // speculative Idx. - data_block* block = - static_cast(Haz->read_block.load(std::memory_order_seq_cst)); + data_block* block = static_cast( + Haz->read_block.load(std::memory_order_seq_cst) + ); - [[maybe_unused]] size_t boff = block->offset.load(std::memory_order_relaxed); + [[maybe_unused]] size_t boff = + block->offset.load(std::memory_order_relaxed); assert(circular_less_than(actOff, 1 + Idx)); assert(circular_less_than(boff, 1 + Idx)); @@ -1487,7 +1558,9 @@ template class channel { std::in_place_index, std::move(elem->data.value) ); elem->data.destroy(); - Haz->active_offset.store(Idx + InactiveHazptrOffset, std::memory_order_release); + Haz->active_offset.store( + Idx + InactiveHazptrOffset, std::memory_order_release + ); return result; } } else { @@ -1509,14 +1582,18 @@ template class channel { // the consumer retries. Otherwise no data can ever arrive at or // after Idx, and everything before Idx has been consumed. if (is_closed_past(Idx)) [[unlikely]] { - Haz->active_offset.store(Idx + InactiveHazptrOffset, std::memory_order_release); + Haz->active_offset.store( + Idx + InactiveHazptrOffset, std::memory_order_release + ); return std::variant( std::in_place_index ); } // Release the hazard pointer so that this token doesn't block // reclamation while it is idle. - Haz->active_offset.store(Idx + InactiveHazptrOffset, std::memory_order_release); + Haz->active_offset.store( + Idx + InactiveHazptrOffset, std::memory_order_release + ); return std::variant( std::in_place_index ); @@ -1540,9 +1617,10 @@ template class channel { aw_push_impl(aw_push& Parent) noexcept : parent{Parent} {} bool await_ready() noexcept { - if (parent.haz_ptr->write_count + parent.haz_ptr->read_count >= ClusterPeriod) - [[unlikely]] { - if (parent.haz_ptr->write_count + parent.haz_ptr->read_count == ClusterPeriod) { + if (parent.haz_ptr->write_count + parent.haz_ptr->read_count >= + ClusterPeriod) [[unlikely]] { + if (parent.haz_ptr->write_count + parent.haz_ptr->read_count == + ClusterPeriod) { size_t elapsed = parent.haz_ptr->elapsed(); size_t writerCount = 0; parent.haz_ptr->for_each_owned_hazptr( @@ -1567,12 +1645,14 @@ template class channel { // Try to get rti. Suspend if we can get it. // If we don't get it on this call, don't suspend and try // again to get it on the next call. - int rti = - parent.haz_ptr->requested_thread_index.load(std::memory_order_relaxed); + int rti = parent.haz_ptr->requested_thread_index.load( + std::memory_order_relaxed + ); if (rti == -1) { if (parent.chan.try_cluster(parent.haz_ptr)) { - rti = - parent.haz_ptr->requested_thread_index.load(std::memory_order_relaxed); + rti = parent.haz_ptr->requested_thread_index.load( + std::memory_order_relaxed + ); } } if (rti != -1) { @@ -1586,8 +1666,11 @@ template class channel { } void await_suspend(std::coroutine_handle<> Outer) noexcept { - size_t target = static_cast(parent.haz_ptr->requested_thread_index); - parent.haz_ptr->requested_thread_index.store(-1, std::memory_order_relaxed); + size_t target = + static_cast(parent.haz_ptr->requested_thread_index); + parent.haz_ptr->requested_thread_index.store( + -1, std::memory_order_relaxed + ); tmc::detail::post_checked( tmc::detail::this_thread::executor(), std::move(Outer), tmc::detail::this_thread::this_task().prio, target @@ -1611,7 +1694,8 @@ template class channel { size_t idx = StartIdx; block = find_block(block, idx); while (circular_less_than(idx, EndIdx)) { - auto cons = block->values[idx & BlockSizeMask].spin_wait_for_waiting_consumer(); + auto cons = + block->values[idx & BlockSizeMask].spin_wait_for_waiting_consumer(); if (cons != nullptr) { cons->ok = false; tmc::detail::post_checked( @@ -1634,7 +1718,9 @@ template class channel { size_t woff = write_offset.load(std::memory_order_seq_cst); // Setting this to a distant-but-greater value before setting closed // prevents consumers from exiting too early. - write_closed_at.store(woff + InactiveHazptrOffset, std::memory_order_seq_cst); + write_closed_at.store( + woff + InactiveHazptrOffset, std::memory_order_seq_cst + ); closed.store(WRITE_CLOSING_BIT, std::memory_order_seq_cst); @@ -1644,7 +1730,9 @@ template class channel { // the sentinel and will still be produced. woff = write_offset.fetch_add(1, std::memory_order_seq_cst); write_closed_at.store(woff, std::memory_order_seq_cst); - closed.store(WRITE_CLOSING_BIT | WRITE_CLOSED_BIT, std::memory_order_seq_cst); + closed.store( + WRITE_CLOSING_BIT | WRITE_CLOSED_BIT, std::memory_order_seq_cst + ); // Readers that already claimed slots >= write_closed_at will never // receive data. Wake them now. @@ -1921,7 +2009,8 @@ template class chan_tok { /// `chan_tok`, and before this `chan_tok` goes out of scope! The safest way /// to accomplish this is to tie its scope to the loop: /// `while (auto data = co_await chan.pull_zc()) { process(data->get()); }` - [[nodiscard("You must co_await pull_zc().")]] chan_t::aw_pull_zc pull_zc() noexcept { + [[nodiscard("You must co_await pull_zc().")]] chan_t::aw_pull_zc + pull_zc() noexcept { ASSERT_NO_CONCURRENT_ACCESS(); hazard_ptr* haz = get_hazard_ptr(); return typename chan_t::aw_pull_zc(*chan, haz); @@ -1962,7 +2051,9 @@ template class chan_tok { // Implementing handling for throwing construction is not possible with the // current design. This assert will also fire if no matching constructor can // be found for the iterator's dereferenced value. - static_assert(std::is_nothrow_constructible_v); + static_assert( + std::is_nothrow_constructible_v + ); hazard_ptr* haz = get_hazard_ptr(); return chan->post_bulk(haz, static_cast(Begin), Count); } @@ -1986,7 +2077,9 @@ template class chan_tok { // Implementing handling for throwing construction is not possible with the // current design. This assert will also fire if no matching constructor can // be found for the iterator's dereferenced value. - static_assert(std::is_nothrow_constructible_v); + static_assert( + std::is_nothrow_constructible_v + ); hazard_ptr* haz = get_hazard_ptr(); return chan->post_bulk( haz, static_cast(Begin), static_cast(End - Begin) @@ -2013,8 +2106,10 @@ template class chan_tok { // Implementing handling for throwing construction is not possible with the // current design. This assert will also fire if no matching constructor can // be found for the iterator's dereferenced value. - static_assert(std::is_nothrow_constructible_v< - T, decltype(std::move(*static_cast(Range).begin()))>); + static_assert( + std::is_nothrow_constructible_v< + T, decltype(std::move(*static_cast(Range).begin()))> + ); hazard_ptr* haz = get_hazard_ptr(); auto begin = static_cast(Range).begin(); auto end = static_cast(Range).end(); @@ -2090,7 +2185,8 @@ template class chan_tok { /// value of 2,000,000 represents an item being pushed every 500ns. This /// behavior can be disabled entirely by setting this to 0. chan_tok& set_heavy_load_threshold(size_t Threshold) noexcept { - size_t cycles = Threshold == 0 ? 0 : TMC_CPU_FREQ * chan_t::ClusterPeriod / Threshold; + size_t cycles = + Threshold == 0 ? 0 : TMC_CPU_FREQ * chan_t::ClusterPeriod / Threshold; chan->MinClusterCycles.store(cycles, std::memory_order_relaxed); return *this; } @@ -2100,7 +2196,8 @@ template class chan_tok { /// /// If the other token is from a different channel, this token will now point /// to that channel. - chan_tok(const chan_tok& Other) noexcept : chan(Other.chan), haz_ptr{nullptr} {} + chan_tok(const chan_tok& Other) noexcept + : chan(Other.chan), haz_ptr{nullptr} {} /// Copy Assignment: If the other token is from a different channel, this /// token will now point to that channel.