Skip to content

ci: run the bwos lifo_queue tests on i386 under UBSan #2328

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ericniebler merged 5 commits into from
Oct 14, 2026
+128 −74
Merged

ci: run the bwos lifo_queue tests on i386 under UBSan #2328

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
bbbec2b
ci: add a 32-bit gcc 14 row
pysco68 Oct 12, 2026
4beca5c
io_uring: convert pointers through std::uintptr_t, not bit_cast
pysco68 Oct 12, 2026
1a2ff83
style: clang-format the bwos and io_uring changes
pysco68 Oct 12, 2026
250f294
bwos: count the queue-level blocks linearly in 64 bits
pysco68 Oct 12, 2026
4115d32
Merge branch 'main' into bwos-32bit-block-counter
pysco68 Oct 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
11 changes: 10 additions & 1 deletion .github/workflows/ci.cpu.yml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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" }
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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 \
Expand Down
126 changes: 62 additions & 64 deletions include/exec/detail/bwos_lifo_queue.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
*
Expand Down Expand Up @@ -176,31 +179,38 @@ namespace experimental::execution::bwos
std::vector<Tp, Allocator> 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<std::size_t> last_block_{
0};
alignas(
hardware_destructive_interference_size) STDEXEC::__std::atomic<block_counter> 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<std::size_t> start_block_{0};
hardware_destructive_interference_size) STDEXEC::__std::atomic<block_counter> start_block_{0};

// Circular array of blocks. Size is always a power of 2 for efficient
// masking.
std::vector<block_type, allocator_of_t<block_type>> blocks_{};

// 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_{};
};

//////////////////////////////////////////////////////////////////////////////
Expand All @@ -214,18 +224,19 @@ namespace experimental::execution::bwos
block_type(block_size, allocator),
allocator_of_t<block_type>(allocator))
, mask_(blocks_.size() - 1)
, round_shift_(std::countr_zero(blocks_.size()))
{
blocks_[0].reclaim(0);
}

template <class Tp, class Allocator>
auto lifo_queue<Tp, Allocator>::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)
Expand All @@ -245,11 +256,11 @@ namespace experimental::execution::bwos
template <class Tp, class Allocator>
auto lifo_queue<Tp, Allocator>::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<std::uint32_t>(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)
Expand All @@ -273,10 +284,10 @@ namespace experimental::execution::bwos
template <class Tp, class Allocator>
auto lifo_queue<Tp, Allocator>::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)
Expand All @@ -293,10 +304,10 @@ namespace experimental::execution::bwos
template <class Iterator, class Sentinel>
auto lifo_queue<Tp, Allocator>::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);
}
Expand All @@ -317,45 +328,32 @@ namespace experimental::execution::bwos
}

template <class Tp, class Allocator>
auto lifo_queue<Tp, Allocator>::increase_block_counter(std::size_t counter) const noexcept
-> std::size_t
auto lifo_queue<Tp, Allocator>::index_of(block_counter counter) const noexcept -> std::size_t
{
std::uint32_t round = static_cast<std::uint32_t>(counter >> 32);
std::size_t index = counter & mask_;
std::size_t next_index = (index + 1) & mask_;
std::uint32_t next_round = round + static_cast<std::uint32_t>(next_index == 0);
return (static_cast<std::size_t>(next_round) << 32) | next_index;
return static_cast<std::size_t>(counter) & mask_;
}

template <class Tp, class Allocator>
auto lifo_queue<Tp, Allocator>::decrease_block_counter(std::size_t counter) const noexcept
-> std::size_t
auto lifo_queue<Tp, Allocator>::round_of(block_counter counter) const noexcept -> std::uint32_t
{
std::uint32_t round = static_cast<std::uint32_t>(counter >> 32);
std::size_t index = counter & mask_;
std::size_t prev_index = static_cast<std::size_t>(static_cast<std::ptrdiff_t>(index) - 1)
& mask_;
std::uint32_t prev_round = round - static_cast<std::uint32_t>(index == 0);
return (static_cast<std::size_t>(prev_round) << 32) | prev_index;
return static_cast<std::uint32_t>(counter >> round_shift_);
}

template <class Tp, class Allocator>
auto
lifo_queue<Tp, Allocator>::advance_get_index(std::size_t& owner, std::size_t owner_index) noexcept
-> bool
auto lifo_queue<Tp, Allocator>::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);
Expand All @@ -364,43 +362,43 @@ namespace experimental::execution::bwos
}

template <class Tp, class Allocator>
auto lifo_queue<Tp, Allocator>::advance_put_index(std::size_t& owner) noexcept -> bool
auto lifo_queue<Tp, Allocator>::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<std::uint32_t>(owner >> 32)
+ static_cast<std::uint32_t>(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<std::size_t>(next_round) << 32) | next_index;
owner = next;
next_block.reclaim(next_round);
last_block_.store(owner, STDEXEC::__std::memory_order_relaxed);
return true;
}

template <class Tp, class Allocator>
auto lifo_queue<Tp, Allocator>::advance_steal_index(std::size_t& thief) noexcept -> bool
auto lifo_queue<Tp, Allocator>::advance_steal_index(block_counter& thief) noexcept -> bool
{
thief = increase_block_counter(thief);
++thief;
return thief < last_block_.load(STDEXEC::__std::memory_order_relaxed);
}

Expand Down
16 changes: 8 additions & 8 deletions include/exec/linux/io_uring_context.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<std::uintptr_t>(__op));
__op->__vtable_->__complete_(__op, __cqe);
}

Expand Down Expand Up @@ -268,7 +268,7 @@ namespace experimental::execution
}
else
{
__sqe.user_data = bit_cast<__u64>(__op);
__sqe.user_data = static_cast<__u64>(reinterpret_cast<std::uintptr_t>(__op));
__array_[__index] = __index;
++__result.__n_submitted;
++__tail;
Expand Down Expand Up @@ -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<std::uintptr_t>(__cqe.user_data));
__op->__vtable_->__complete_(__op, __cqe);
++__head;
++__count;
Expand All @@ -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<std::uintptr_t>(__op));
__op->__vtable_->__complete_(__op, __dummy_cqe);
}
return __count;
Expand Down Expand Up @@ -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<std::uintptr_t>(&__self.__buffer_));
# ifdef STDEXEC_HAS_IORING_OP_READ
__entry.opcode = IORING_OP_READ;
__entry.len = sizeof(__self.__buffer_);
Expand Down Expand Up @@ -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<std::uintptr_t>(__op_->__parent_));
}
# else
__op_->submit_stop(__sqe);
Expand Down Expand Up @@ -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<std::uintptr_t>(&__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<std::uintptr_t>(&__iov_));
__sqe_.len = 1;
__sqe = __sqe_;
# endif
Expand Down
Loading
Loading