diff --git a/.github/workflows/ci.cpu.yml b/.github/workflows/ci.cpu.yml index cf1fe0232..0051c53eb 100644 --- a/.github/workflows/ci.cpu.yml +++ b/.github/workflows/ci.cpu.yml @@ -43,6 +43,7 @@ jobs: - { name: "CPU (gcc 12, Release, TSAN)", build: "Release", tag: gcc12-cuda12.9, cxxstd: "20", cxxflags: "-fsanitize=thread" } - { name: "CPU (gcc 13, Debug)", build: "Debug", tag: gcc13-cuda12.9, cxxstd: "20", cxxflags: "", } - { name: "CPU (gcc 14, Debug)", build: "Debug", tag: gcc14-cuda12.9, cxxstd: "20", cxxflags: "", } + - { name: "CPU (gcc 14, Debug, 32-bit)", build: "Debug", tag: gcc14-cuda12.9, cxxstd: "20", cxxflags: "-m32 -D_GLIBCXX_USE_TBB_PAR_BACKEND=0", arch: 32 } - { name: "CPU (gcc 14, Debug, ASAN)", build: "Debug", tag: gcc14-cuda12.9, cxxstd: "20", cxxflags: "-fsanitize=address" } - { name: "CPU (gcc 14, Debug, TSAN)", build: "Debug", tag: gcc14-cuda12.9, cxxstd: "20", cxxflags: "-fsanitize=thread" } - { name: "CPU (gcc 14, Release, ASAN)", build: "Release", tag: gcc14-cuda12.9, cxxstd: "20", cxxflags: "-fsanitize=address" } @@ -186,6 +187,14 @@ jobs: aws-region: us-east-2 role-duration-seconds: 28800 # 8 hours role-to-assume: arn:aws:iam::279114543810:role/gha-oidc-NVIDIA + # gcc-multilib provides /usr/include/asm for -m32; g++-14-multilib only + # recommends it. + - if: ${{ matrix.arch == 32 }} + name: Install the 32-bit toolchain + shell: bash + run: | + apt-get update + DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends g++-14-multilib gcc-multilib - name: Build and test CPU schedulers env: ASAN_OPTIONS: alloc_dealloc_mismatch=0 @@ -229,7 +238,7 @@ jobs: cmake -S . -B build -GNinja \ -DCMAKE_BUILD_TYPE=${{ matrix.build }} \ -DCMAKE_CXX_FLAGS="${{ matrix.cxxflags }}" \ - -DSTDEXEC_ENABLE_TBB:BOOL=${{ !contains(matrix.cxxflags, '-fsanitize') && !contains(matrix.name, 'modules') }} \ + -DSTDEXEC_ENABLE_TBB:BOOL=${{ !contains(matrix.cxxflags, '-fsanitize') && !contains(matrix.name, 'modules') && matrix.arch != 32 }} \ -DSTDEXEC_ENABLE_TASKFLOW:BOOL=${{ matrix.name == 'CPU (clang 22, Debug)' }} \ -DSTDEXEC_ENABLE_ASIO:BOOL=TRUE \ -DSTDEXEC_ASIO_IMPLEMENTATION:STRING=boost \ diff --git a/include/exec/detail/bwos_lifo_queue.hpp b/include/exec/detail/bwos_lifo_queue.hpp index e6b40b1d3..59c20b615 100644 --- a/include/exec/detail/bwos_lifo_queue.hpp +++ b/include/exec/detail/bwos_lifo_queue.hpp @@ -41,19 +41,22 @@ import stdexec; * * BLOCK COUNTER ENCODING: * ---------------------- - * Block counters (last_block_, start_block_, and internal counters) are 64-bit values with: - * - Bits [63:32]: Round number (32-bit unsigned, wraps on overflow) - * - Bits [31:0]: Block index within the circular block array (masked by mask_) + * The queue-level block counters (last_block_, start_block_) count blocks + * linearly: counter = round * num_blocks() + index. Because num_blocks() is a + * power of two, the index is (counter & mask_) and the round is + * (counter >> round_shift_). Moving to the next or previous block is counter + 1 + * or counter - 1; the round changes exactly when the index wraps. * - * The round number increments each time the index wraps around from mask_ to 0. - * This encoding allows: + * The counters are std::uint64_t regardless of the width of std::size_t, so the + * round always has at least 32 bits and wraps in step with the 32-bit block-level + * rounds below. This encoding allows: * - Direct comparison of counters to determine relative order * - ABA problem prevention through round tracking - * - Efficient atomic operations on a single 64-bit value + * - Atomic operations on a single 64-bit value * * TWO-LEVEL ROUND SYSTEM: * ---------------------- - * 1. Queue-level rounds: Tracked in last_block_/start_block_ upper 32 bits + * 1. Queue-level rounds: counter >> round_shift_ of last_block_/start_block_ * - Increments when the owner wraps around the circular block array * - Passed to blocks during reclaim() to initialize block-level rounds * @@ -176,24 +179,28 @@ namespace experimental::execution::bwos std::vector ring_buffer_; }; - auto advance_get_index(std::size_t& owner, std::size_t owner_index) noexcept -> bool; - auto advance_steal_index(std::size_t& thief) noexcept -> bool; - auto advance_put_index(std::size_t& owner) noexcept -> bool; + // Linear block counter, see BLOCK COUNTER ENCODING. Deliberately not + // std::size_t: the round must keep 32 bits on 32-bit targets. + using block_counter = std::uint64_t; - auto increase_block_counter(std::size_t counter) const noexcept -> std::size_t; - auto decrease_block_counter(std::size_t counter) const noexcept -> std::size_t; + auto advance_get_index(block_counter& owner, std::size_t owner_index) noexcept -> bool; + auto advance_steal_index(block_counter& thief) noexcept -> bool; + auto advance_put_index(block_counter& owner) noexcept -> bool; - // Block counter (round in upper 32 bits, index in lower bits) for the most + auto index_of(block_counter counter) const noexcept -> std::size_t; + auto round_of(block_counter counter) const noexcept -> std::uint32_t; + + // Linear block counter (see BLOCK COUNTER ENCODING) for the most // recent block owned by push/pop operations. Modified only by the owner // thread. - alignas(hardware_destructive_interference_size) STDEXEC::__std::atomic last_block_{ - 0}; + alignas( + hardware_destructive_interference_size) STDEXEC::__std::atomic last_block_{0}; // Block counter for the oldest block available for stealing. Modified only // by the owner thread when it advances to a new block and needs to reclaim // the block at start_block_ position. alignas( - hardware_destructive_interference_size) STDEXEC::__std::atomic start_block_{0}; + hardware_destructive_interference_size) STDEXEC::__std::atomic start_block_{0}; // Circular array of blocks. Size is always a power of 2 for efficient // masking. @@ -201,6 +208,9 @@ namespace experimental::execution::bwos // Bitmask (blocks_.size() - 1) for extracting block index from counters. std::size_t mask_{}; + + // log2(blocks_.size()) for extracting the round from counters. + int round_shift_{}; }; ////////////////////////////////////////////////////////////////////////////// @@ -214,6 +224,7 @@ namespace experimental::execution::bwos block_type(block_size, allocator), allocator_of_t(allocator)) , mask_(blocks_.size() - 1) + , round_shift_(std::countr_zero(blocks_.size())) { blocks_[0].reclaim(0); } @@ -221,11 +232,11 @@ namespace experimental::execution::bwos template auto lifo_queue::pop_back() noexcept -> Tp { - std::size_t owner = last_block_.load(STDEXEC::__std::memory_order_relaxed); - std::size_t owner_index{}; + block_counter owner = last_block_.load(STDEXEC::__std::memory_order_relaxed); + std::size_t owner_index{}; do { - owner_index = owner & mask_; + owner_index = index_of(owner); block_type& current_block = blocks_[owner_index]; auto [ec, value] = current_block.get(); if (ec == lifo_queue_error_code::success) @@ -245,11 +256,11 @@ namespace experimental::execution::bwos template auto lifo_queue::steal_front() noexcept -> Tp { - std::size_t thief = start_block_.load(STDEXEC::__std::memory_order_relaxed); + block_counter thief = start_block_.load(STDEXEC::__std::memory_order_relaxed); do { - auto const thief_round = static_cast(thief >> 32); - std::size_t const thief_index = thief & mask_; + auto const thief_round = round_of(thief); + std::size_t const thief_index = index_of(thief); block_type& block = blocks_[thief_index]; fetch_result result = block.steal(thief_round); while (result.status != lifo_queue_error_code::done) @@ -273,10 +284,10 @@ namespace experimental::execution::bwos template auto lifo_queue::push_back(Tp value) noexcept -> bool { - std::size_t owner = last_block_.load(STDEXEC::__std::memory_order_relaxed); + block_counter owner = last_block_.load(STDEXEC::__std::memory_order_relaxed); do { - std::size_t owner_index = owner & mask_; + std::size_t owner_index = index_of(owner); block_type& current_block = blocks_[owner_index]; auto ec = current_block.put(value); if (ec == lifo_queue_error_code::success) @@ -293,10 +304,10 @@ namespace experimental::execution::bwos template auto lifo_queue::push_back(Iterator first, Sentinel last) noexcept -> Iterator { - std::size_t owner = last_block_.load(STDEXEC::__std::memory_order_relaxed); + block_counter owner = last_block_.load(STDEXEC::__std::memory_order_relaxed); do { - std::size_t owner_index = owner & mask_; + std::size_t owner_index = index_of(owner); block_type& current_block = blocks_[owner_index]; first = current_block.bulk_put(first, last); } @@ -317,45 +328,32 @@ namespace experimental::execution::bwos } template - auto lifo_queue::increase_block_counter(std::size_t counter) const noexcept - -> std::size_t + auto lifo_queue::index_of(block_counter counter) const noexcept -> std::size_t { - std::uint32_t round = static_cast(counter >> 32); - std::size_t index = counter & mask_; - std::size_t next_index = (index + 1) & mask_; - std::uint32_t next_round = round + static_cast(next_index == 0); - return (static_cast(next_round) << 32) | next_index; + return static_cast(counter) & mask_; } template - auto lifo_queue::decrease_block_counter(std::size_t counter) const noexcept - -> std::size_t + auto lifo_queue::round_of(block_counter counter) const noexcept -> std::uint32_t { - std::uint32_t round = static_cast(counter >> 32); - std::size_t index = counter & mask_; - std::size_t prev_index = static_cast(static_cast(index) - 1) - & mask_; - std::uint32_t prev_round = round - static_cast(index == 0); - return (static_cast(prev_round) << 32) | prev_index; + return static_cast(counter >> round_shift_); } template - auto - lifo_queue::advance_get_index(std::size_t& owner, std::size_t owner_index) noexcept - -> bool + auto lifo_queue::advance_get_index(block_counter& owner, + std::size_t owner_index) noexcept -> bool { - std::size_t start = start_block_.load(STDEXEC::__std::memory_order_relaxed); + block_counter start = start_block_.load(STDEXEC::__std::memory_order_relaxed); if (start == owner) { // Cannot move backward past start_block_ - queue is empty return false; } - // Move to predecessor block, properly decrementing round if wrapping - // backward - std::size_t predecessor = decrease_block_counter(owner); - std::size_t predecessor_index = predecessor & mask_; - block_type& previous_block = blocks_[predecessor_index]; - block_type& current_block = blocks_[owner_index]; + // Move to predecessor block; the round drops when the index wraps backward + block_counter predecessor = owner - 1; + std::size_t predecessor_index = index_of(predecessor); + block_type& previous_block = blocks_[predecessor_index]; + block_type& current_block = blocks_[owner_index]; current_block.reduce_round(); previous_block.takeover(); last_block_.store(predecessor, STDEXEC::__std::memory_order_relaxed); @@ -364,43 +362,43 @@ namespace experimental::execution::bwos } template - auto lifo_queue::advance_put_index(std::size_t& owner) noexcept -> bool + auto lifo_queue::advance_put_index(block_counter& owner) noexcept -> bool { - std::size_t next_index = (owner + 1ul) & mask_; - std::size_t owner_index = owner & mask_; + block_counter next = owner + 1; + std::size_t next_index = index_of(next); + std::size_t owner_index = index_of(owner); if (next_index == owner_index) { // Wrap-around would hit the same block - queue is full return false; } - // Calculate next round, incrementing when wrapping to index 0 - std::uint32_t next_round = static_cast(owner >> 32) - + static_cast(next_index == 0); - block_type& next_block = blocks_[next_index]; + // The round increments exactly when next_index wraps to 0 + std::uint32_t next_round = round_of(next); + block_type& next_block = blocks_[next_index]; if (!next_block.is_writable(next_round)) { // Block is not yet safe to reuse (thieves still accessing previous generation) return false; } - std::size_t first = start_block_.load(STDEXEC::__std::memory_order_relaxed); - std::size_t first_index = first & mask_; + block_counter first = start_block_.load(STDEXEC::__std::memory_order_relaxed); + std::size_t first_index = index_of(first); if (next_index == first_index) { // About to reuse the block at start_block_, so advance start_block_ forward - start_block_.store(increase_block_counter(first), STDEXEC::__std::memory_order_relaxed); + start_block_.store(first + 1, STDEXEC::__std::memory_order_relaxed); } block_type& current_block = blocks_[owner_index]; current_block.grant(); - owner = (static_cast(next_round) << 32) | next_index; + owner = next; next_block.reclaim(next_round); last_block_.store(owner, STDEXEC::__std::memory_order_relaxed); return true; } template - auto lifo_queue::advance_steal_index(std::size_t& thief) noexcept -> bool + auto lifo_queue::advance_steal_index(block_counter& thief) noexcept -> bool { - thief = increase_block_counter(thief); + ++thief; return thief < last_block_.load(STDEXEC::__std::memory_order_relaxed); } diff --git a/include/exec/linux/io_uring_context.hpp b/include/exec/linux/io_uring_context.hpp index 66557157c..11ec08691 100644 --- a/include/exec/linux/io_uring_context.hpp +++ b/include/exec/linux/io_uring_context.hpp @@ -201,7 +201,7 @@ namespace experimental::execution { ::io_uring_cqe __cqe{}; __cqe.res = -ECANCELED; - __cqe.user_data = bit_cast<__u64>(__op); + __cqe.user_data = static_cast<__u64>(reinterpret_cast(__op)); __op->__vtable_->__complete_(__op, __cqe); } @@ -268,7 +268,7 @@ namespace experimental::execution } else { - __sqe.user_data = bit_cast<__u64>(__op); + __sqe.user_data = static_cast<__u64>(reinterpret_cast(__op)); __array_[__index] = __index; ++__result.__n_submitted; ++__tail; @@ -321,7 +321,7 @@ namespace experimental::execution { __u32 const __index = __head & __mask_; ::io_uring_cqe const & __cqe = __entries_[__index]; - auto* __op = bit_cast<__task*>(__cqe.user_data); + auto* __op = reinterpret_cast<__task*>(static_cast(__cqe.user_data)); __op->__vtable_->__complete_(__op, __cqe); ++__head; ++__count; @@ -332,7 +332,7 @@ namespace experimental::execution { __task* __op = __ready.pop_front(); ::io_uring_cqe __dummy_cqe{}; - __dummy_cqe.user_data = bit_cast<__u64>(__op); + __dummy_cqe.user_data = static_cast<__u64>(reinterpret_cast(__op)); __op->__vtable_->__complete_(__op, __dummy_cqe); } return __count; @@ -362,7 +362,7 @@ namespace experimental::execution __wakeup_operation& __self = *static_cast<__wakeup_operation*>(__pointer); __entry = ::io_uring_sqe{}; __entry.fd = __self.__eventfd_; - __entry.addr = bit_cast<__u64>(&__self.__buffer_); + __entry.addr = static_cast<__u64>(reinterpret_cast(&__self.__buffer_)); # ifdef STDEXEC_HAS_IORING_OP_READ __entry.opcode = IORING_OP_READ; __entry.len = sizeof(__self.__buffer_); @@ -918,7 +918,7 @@ namespace experimental::execution { std::memset(&__sqe, 0, sizeof(__sqe)); __sqe.opcode = IORING_OP_ASYNC_CANCEL; - __sqe.addr = bit_cast<__u64>(__op_->__parent_); + __sqe.addr = static_cast<__u64>(reinterpret_cast(__op_->__parent_)); } # else __op_->submit_stop(__sqe); @@ -1193,14 +1193,14 @@ namespace experimental::execution # ifdef STDEXEC_HAS_IO_URING_ASYNC_CANCELLATION ::io_uring_sqe __sqe_{}; __sqe_.opcode = IORING_OP_TIMEOUT; - __sqe_.addr = bit_cast<__u64>(&__duration_); + __sqe_.addr = static_cast<__u64>(reinterpret_cast(&__duration_)); __sqe_.len = 1; __sqe = __sqe_; # else ::io_uring_sqe __sqe_{}; __sqe_.opcode = IORING_OP_READV; __sqe_.fd = __timerfd_; - __sqe_.addr = bit_cast<__u64>(&__iov_); + __sqe_.addr = static_cast<__u64>(reinterpret_cast(&__iov_)); __sqe_.len = 1; __sqe = __sqe_; # endif diff --git a/test/exec/test_bwos_lifo_queue.cpp b/test/exec/test_bwos_lifo_queue.cpp index 830a3e144..bdb37b9f9 100644 --- a/test/exec/test_bwos_lifo_queue.cpp +++ b/test/exec/test_bwos_lifo_queue.cpp @@ -500,4 +500,51 @@ TEST_CASE("exec::bwos::lifo_queue - takeover grant synchronization", "[bwos]") owner.join(); thief.join(); -} \ No newline at end of file +} + +TEST_CASE("exec::bwos::lifo_queue - keeps all items across wraps of the block ring", "[bwos]") +{ + // 8 blocks of 4. Rounds drained by steal_front() leave the owner's position + // where it is, so it keeps moving forward and laps the ring a few hundred + // times. Every third round is drained by pop_back() instead, which walks the + // owner back across the wrap point as well. A round pushes at most 12 items + // on top of at most one block (4) left over by the previous round, so the + // queue never holds more than 16 of its 32 slots and no push may be refused. + // Catches block counters that lose the round, e.g. because they are only + // 32 bits wide. + exec::bwos::lifo_queue queue(8, 4); + std::vector items(12); + long pushed = 0; + long taken = 0; + for (int round = 0; round < 2000; ++round) + { + int const n = 1 + round % 12; + for (int i = 0; i < n; ++i) + { + REQUIRE(queue.push_back(&items[i])); + ++pushed; + } + if (round % 3 == 2) + { + while (queue.pop_back() != nullptr) + { + ++taken; + } + } + else + { + // Thieves cannot reach the owner's current block, so this may leave a + // few items behind for a later pop_back() round. + while (queue.steal_front() != nullptr) + { + ++taken; + } + } + } + while (queue.pop_back() != nullptr) + { + ++taken; + } + CHECK(queue.steal_front() == nullptr); + CHECK(pushed == taken); +}