Skip to content

Fix race in MPSC algo #1812

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

Draft
ccotter wants to merge 7 commits into
base: main
Choose a base branch
from
Draft

Fix race in MPSC algo #1812

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
9b44d94
Fix race in MPSC algo
ccotter Feb 3, 2026
5a875d4
Add Relacy tests to the build
ccotter Feb 4, 2026
c0720dd
Default relacy only fofr GNU
ccotter Feb 4, 2026
0585125
Remove old Relacy Makefile
ccotter Feb 4, 2026
2d595ad
Simplify tests
ccotter Feb 4, 2026
0a9df03
Merge branch 'main' of https://github.com/nvidia/stdexec into HEAD
ccotter Feb 4, 2026
3b6648b
Improve implementation to be near to Dmitry's
maikel Feb 5, 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
8 changes: 8 additions & 0 deletions CMakeLists.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -150,6 +150,14 @@ else()
set(stdexec_compiler_frontend ${CMAKE_CXX_COMPILER_ID})
endif()

# Build relacy tests by default only for GNU compiler and if tests are enabled
if(${STDEXEC_BUILD_TESTS} AND stdexec_compiler_frontend STREQUAL "GNU")
set(STDEXEC_BUILD_RELACY_TESTS_DEFAULT ON)
else()
set(STDEXEC_BUILD_RELACY_TESTS_DEFAULT OFF)
endif()
option(STDEXEC_BUILD_RELACY_TESTS "Build stdexec relacy tests" ${STDEXEC_BUILD_RELACY_TESTS_DEFAULT})

set(stdexec_export_targets)

# Define the main library
Expand Down
5 changes: 4 additions & 1 deletion include/exec/timed_thread_scheduler.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,14 +43,17 @@ namespace exec {
stop
};

// Default ctor for __intrusive_mpsc_queue's internal stub node
constexpr timed_thread_operation_base() = default;

constexpr timed_thread_operation_base(
void (*set_value)(timed_thread_operation_base*) noexcept,
command_type command = command_type::schedule) noexcept
: command_{command}
, set_value_{set_value} {
}

STDEXEC::__std::atomic<void*> next_{nullptr};
STDEXEC::__std::atomic<timed_thread_operation_base*> next_{nullptr};
command_type command_;
void (*set_value_)(timed_thread_operation_base*) noexcept;
};
Expand Down
2 changes: 1 addition & 1 deletion include/stdexec/__detail/__atomic.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -61,7 +61,7 @@ namespace STDEXEC::__std {
using std::atomic_thread_fence;
using std::atomic_signal_fence;

# if __cpp_lib_atomic_ref >= 2018'06L && !defined(STDEXEC_RELACY)
# if __cpp_lib_atomic_ref >= 2018'06L
using std::atomic_ref;
# else
inline constexpr int __atomic_flag_map[] = {
Expand Down
93 changes: 56 additions & 37 deletions include/stdexec/__detail/__intrusive_mpsc_queue.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,5 +1,5 @@
/*
* Copyright (c) Dmitiy V'jukov
* Copyright (c) Dmitry V'jukov
* Copyright (c) 2024 Maikel Nadolski
* Copyright (c) 2024 NVIDIA Corporation
*
Expand All @@ -24,59 +24,78 @@

#include "__atomic.hpp"

#include "stdexec/__detail/__config.hpp"

#include "./__spin_loop_pause.hpp"

namespace STDEXEC {
template <auto _Ptr>
class __intrusive_mpsc_queue;

template <class _Node, __std::atomic<void*> _Node::* _Next>
// _Node must be default_initializable only for the queue to construct an
// internal "stub" node - only the _Next data element is accessed internally.
template <class _Node, __std::atomic<_Node*> _Node::* _Next>
requires __std::default_initializable<_Node>
class __intrusive_mpsc_queue<_Next> {
__std::atomic<void*> __back_{&__nil_};
void* __front_{&__nil_};
__std::atomic<_Node*> __nil_ = nullptr;

constexpr void push_back_nil() {
__nil_.store(nullptr, __std::memory_order_relaxed);
auto* __prev = static_cast<_Node*>(__back_.exchange(&__nil_, __std::memory_order_acq_rel));
(__prev->*_Next).store(&__nil_, __std::memory_order_release);
}
__std::atomic<_Node*> __head_{&__stub_};
_Node* __tail_{&__stub_};
_Node __stub_{};

public:

__intrusive_mpsc_queue() {
(__stub_.*_Next).store(nullptr, __std::memory_order_release);
}

constexpr auto push_back(_Node* __new_node) noexcept -> bool {
(__new_node->*_Next).store(nullptr, __std::memory_order_relaxed);
void* __prev_back = __back_.exchange(__new_node, __std::memory_order_acq_rel);
bool __is_nil = __prev_back == static_cast<void*>(&__nil_);
if (__is_nil) {
__nil_.store(__new_node, __std::memory_order_release);
} else {
(static_cast<_Node*>(__prev_back)->*_Next).store(__new_node, __std::memory_order_release);
}
return __is_nil;
_Node* __prev = __head_.exchange(__new_node, __std::memory_order_acq_rel);
(__prev->*_Next).store(__new_node, __std::memory_order_release);
return __prev == &__stub_;
}

constexpr auto pop_front() noexcept -> _Node* {
if (__front_ == static_cast<void*>(&__nil_)) {
_Node* __next = __nil_.load(__std::memory_order_acquire);
if (!__next) {
return nullptr;
_Node* __tail = this->__tail_;
STDEXEC_ASSERT(__tail != nullptr);
_Node* __next = (__tail->*_Next).load(__std::memory_order_acquire);
// If tail is pointing to the stub node we need to advance it once more
if (&__stub_ == __tail) {
if (nullptr == __next) {
return nullptr;
}
__front_ = __next;
this->__tail_ = __next;
__tail = __next;
__next = (__next->*_Next).load(__std::memory_order_acquire);
}
// Normal case: there is a next node and we can just advance the tail
if (nullptr != __next) {
this->__tail_ = __next;
return __tail;
}
auto* __front = static_cast<_Node*>(__front_);
void* __next = (__front->*_Next).load(__std::memory_order_acquire);
if (__next) {
__front_ = __next;
return __front;
// Next is nullptr here means that either:
// 1) There are no more nodes in the queue
// 2) A producer is in the middle of adding a new node
const _Node* __head = this->__head_.load(__std::memory_order_acquire);
// A producer is in the middle of adding a new node
// we cannot return tail as we cannot link the next node yet
if (__tail != __head) {
return nullptr;
}
STDEXEC_ASSERT(!__next);
push_back_nil();
do {
__spin_loop_pause();
__next = (__front->*_Next).load(__std::memory_order_acquire);
} while (!__next);
__front_ = __next;
return __front;
// No more nodes in the queue - we need to insert a stub node
// to be able to link to an eventual empty state (or new nodes)
push_back(&__stub_);
// Now re-attempt to load next
__next = (__tail->*_Next).load(__std::memory_order_acquire);
if (nullptr != __next) {
// Successfully linked either a new node or the stub node
this->__tail_ = __next;
return __tail;
}
// A producer is in the middle of adding a new node since next is still nullptr
// and not our stub node, thus we cannot link the next node yet
return nullptr;
}
};
} // namespace STDEXEC

} // namespace STDEXEC
5 changes: 5 additions & 0 deletions test/CMakeLists.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -62,6 +62,7 @@ set(stdexec_test_sources
stdexec/algos/other/test_execute.cpp
stdexec/detail/test_completion_signatures.cpp
stdexec/detail/test_utility.cpp
stdexec/detail/test_intrusive_mpsc_queue.cpp
stdexec/schedulers/test_task_scheduler.cpp
stdexec/queries/test_env.cpp
stdexec/queries/test_get_forward_progress_guarantee.cpp
Expand Down Expand Up @@ -137,6 +138,10 @@ icm_add_build_failure_test(
FOLDER test
)

if(STDEXEC_BUILD_RELACY_TESTS)
add_subdirectory(rrd)
endif()

# # Adding multiple tests with a glob
# icm_glob_build_failure_tests(
# PATTERN *_fail*.cpp
Expand Down
44 changes: 44 additions & 0 deletions test/rrd/CMakeLists.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,44 @@
include(FetchContent)

FetchContent_Declare(
relacy
GIT_REPOSITORY https://github.com/dvyukov/relacy
GIT_TAG master
)

FetchContent_Populate(relacy)

add_library(relacy INTERFACE)
target_include_directories(relacy INTERFACE
$<BUILD_INTERFACE:${relacy_SOURCE_DIR}/relacy/fakestd>
$<BUILD_INTERFACE:${relacy_SOURCE_DIR}>
$<INSTALL_INTERFACE:include>
)

function(add_relacy_test target_name)
add_executable(${target_name} ${target_name}.cpp)

target_link_libraries(${target_name} PRIVATE relacy)
target_include_directories(${target_name} PRIVATE ../../include)
target_include_directories(${target_name} PRIVATE ${CMAKE_CURRENT_SOURCE_DIR})
set_target_properties(${target_name} PROPERTIES
CXX_STANDARD 20
CXX_STANDARD_REQUIRED ON
CXX_EXTENSIONS OFF)

add_test(NAME relacy-${target_name} COMMAND ${target_name})
endfunction()

set(relacy_tests
async_scope
intrusive_mpsc_queue
split
sync_wait
)

foreach(test ${relacy_tests})
add_relacy_test(${test})
endforeach()

# Target to build all relacy tests
add_custom_target(relacy-tests DEPENDS ${relacy_tests})
43 changes: 0 additions & 43 deletions test/rrd/Makefile
View file
Open in desktop

This file was deleted.

33 changes: 24 additions & 9 deletions test/rrd/README.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -15,13 +15,20 @@ STDEXEC library could needs to use `x.fetch_add(1)` to be compatible with Relacy

## Instructions

Run the following commands from within this directory (`./tests/rrd`).
Configure and build stdexec following the build instructions in the top level
[README.md](../../README.md). There are a couple relacy specific build and ctest
targets, though they are part of the standard build and ctest and will be run
automatically if cmake is configured with `-DSTDEXEC_BUILD_RELACY_TESTS=1`.
`STDEXEC_BUILD_RELACY_TESTS` is set by default for GCC today.

Run the following on a Linux machine with GCC as the toolchain.

```
git clone -b STDEXEC https://github.com/dvyukov/relacy
CXX=g++-11 make -j 4
./build/split
./build/async_scope
mkdir build && cd build
cmake ..
make relacy-tests -j 4
ctest -R relacy # Run all relacy tests
./test/rrd/sync_wait # Run a specific relacy test directly
```

## Recommended use
Expand All @@ -35,8 +42,16 @@ out a more stable build on all environments/compilers, we should revisit this.
## Supported platforms

The STDEXEC Relacy tests have been verified to build and run on
* Linux based GCC+11 with libstdc++ (`x86_64`)
* Mac with Apple Clang 15 with libc++ (`x86_64`)
* Linux based GCC+11-14 with libstdc++ (`x86_64`)
* Mac with Apple Clang 15 and 17 with libc++ (`x86_64`)

## Caveat

Relacy relies on a less than robust approach to implement its runtime: it replaces
std:: names with its own versions, for example, std::atomic and std::mutex, as well
as pthread_* APIs. As libstdc++/libc++ evolve, newer versions may not be compatible with
Relacy. In these cases, changes to Relacy are needed to correctly intercept and replace
std:: names.

G++12 and newer are known to have issues that could be addressed with patches
to Relacy.
When the compilers and standard libraries release new versions, we will need to test the
new versions can compile the stdexec Relacy tests before enabling the new compiler.
25 changes: 17 additions & 8 deletions test/rrd/async_scope.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,17 +1,26 @@
#include "../../relacy/relacy_cli.hpp"
#include "../../relacy/relacy_std.hpp"
/*
* Copyright (c) 2025 NVIDIA Corporation
*
* Licensed under the Apache License Version 2.0 with LLVM Exceptions
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* https://llvm.org/LICENSE.txt
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/

#include <stdexec_relacy.hpp>

#include <exec/async_scope.hpp>
#include <exec/single_thread_context.hpp>
#include <exec/static_thread_pool.hpp>
#include <stdexec/execution.hpp>

#include <stdexcept>

using rl::nvar;
using rl::nvolatile;
using rl::mutex;

namespace ex = STDEXEC;
using exec::async_scope;

Expand Down
Loading