diff --git a/Makefile b/Makefile index ed03c97c..771a028c 100644 --- a/Makefile +++ b/Makefile @@ -116,7 +116,7 @@ SIMPLE_BUILD = build/simple-$(shell uname -s) simple: simple-test simple-config: - cmake -G Ninja -S . -B $(SIMPLE_BUILD) -DBEMAN_USE_MODULES=OFF -DCXXFLAGS= + cmake -G Ninja -S . -B $(SIMPLE_BUILD) -DBEMAN_USE_MODULES=OFF -DCMAKE_BUILD_TYPE=Debug simple-build: simple-config CXXFLAGS= cmake --build $(SIMPLE_BUILD) diff --git a/docs/overview.md b/docs/overview.md index 9e6b0287..f529bf4e 100644 --- a/docs/overview.md +++ b/docs/overview.md @@ -741,13 +741,13 @@ The expression let_stopped(upstream, fun) yields a se The expression let_value(upstream, fun) yields a sender sndr which uses a successful completion (set_value) of upstream as arguments to invoke fun which has to return another sender inner-sndr and the completion of inner-sndr becomes the completion of sndr. If this invocation results in an exception sndr completes with an error completion the result of std::current_exception. If upstream completes with a cancellation (set_stopped) or with an error (set_error) this completion becomes the completion of sndr.
-on(_sched_, _sndr_), on(_sndr, _sched_, _closure_) +on(sched, sndr), on(sndr, sched, closure) The on algorithm is a pipeable sender adaptor. Let on-sender be -The sender on-sndr is started on _sched_'s execution context. +The sender on-sndr is started on sched's execution context. The on algorithm completes on the original scheduler (obtained using get_start_scheduler) with the result of the on-sndr.
@@ -808,6 +808,24 @@ The expression write_env(sender, env) creates a sende which passes a receiver to sender which combines the environment env with the environment from the receiver's environment. The queries from env take precedence over those from the receiver's environment. +
+Example + +This example shows using `write_env` providing an environment with an allocator: + +```c++ +std::pmr::monotonic_buffer_resource mbr; +std::pmr::polymorphic_allocator mbr_alloc(&mbr); +ex::sync_wait( + ex::read_env(ex::get_allocator) + | ex::then([&](auto alloc){ assert(mbr_alloc == alloc); }) + | ex::write_env(ex::env{ex::prop{ex::get_allocator, mbr_alloc}}) +); +``` + +Note that `write_env` affects the sender passed as first argument or, when using +`|`, the sender before the `|`. +
@@ -890,7 +908,7 @@ an object loop of type run_loop are: @@ -1167,3 +1185,19 @@ This concept determines if an object of type Rcvr supports the comp concept valid-specialization<T, Args...> This concept determines the class template T can be specialized with arguments Args..., i.e., if T<Args...> is valid. + + +
+concept simple-allocator<Alloc> + +The concept simple-allocator<Alloc> determines if +Alloc is a basic allocator. If `alloc` is an object of type +`Alloc`, the following need to be `true`: + + +
diff --git a/include/beman/execution/detail/find_allocator.hpp b/include/beman/execution/detail/find_allocator.hpp index 9f94ee31..51b61e18 100644 --- a/include/beman/execution/detail/find_allocator.hpp +++ b/include/beman/execution/detail/find_allocator.hpp @@ -32,7 +32,7 @@ template Allocator find_allocator(const ::std::allocator_arg_t&, const Alloc& alloc, const A&...) noexcept { static_assert( requires(const Alloc& a) { Allocator(a); }, - "The allocator needs to be constructible from the argument following std::allocator"); + "The allocator needs to be constructible from the argument following std::allocator_arg"); return Allocator(alloc); } template diff --git a/include/beman/execution/detail/task_scheduler.hpp b/include/beman/execution/detail/task_scheduler.hpp index 504933c8..a5833590 100644 --- a/include/beman/execution/detail/task_scheduler.hpp +++ b/include/beman/execution/detail/task_scheduler.hpp @@ -344,7 +344,7 @@ class task_scheduler { explicit task_scheduler(Sched&& sched, Alloc alloc = {}) : backend_(::std::allocate_shared< ::beman::execution::detail::task_scheduler_backend_for<::std::remove_cvref_t, Alloc>>( - alloc, ::std::forward(sched), ::std::move(alloc))) {} + alloc, ::std::forward(sched), alloc)) {} template <::beman::execution::detail::simple_allocator Alloc> explicit task_scheduler(task_scheduler sched, const Alloc&) noexcept : task_scheduler(::std::move(sched)) {} diff --git a/include/beman/execution/execution.hpp b/include/beman/execution/execution.hpp index a56fd3cb..3bc4c091 100644 --- a/include/beman/execution/execution.hpp +++ b/include/beman/execution/execution.hpp @@ -51,9 +51,11 @@ import beman.execution.detail.parallel_scheduler_replacement; import beman.execution.detail.prop; import beman.execution.detail.read_env; import beman.execution.detail.receiver; +import beman.execution.detail.run_loop; import beman.execution.detail.schedule_from; import beman.execution.detail.schedule; import beman.execution.detail.scheduler; +import beman.execution.detail.scheduler_tag; import beman.execution.detail.scope_token; import beman.execution.detail.sender_adaptor_closure; import beman.execution.detail.sender_in; @@ -127,9 +129,11 @@ import beman.execution.detail.write_env; #include #include #include +#include #include #include #include +#include #include #include #include diff --git a/tests/beman/execution/CMakeLists.txt b/tests/beman/execution/CMakeLists.txt index 07e05189..0065e0ab 100644 --- a/tests/beman/execution/CMakeLists.txt +++ b/tests/beman/execution/CMakeLists.txt @@ -10,6 +10,8 @@ list(APPEND unsupported_execution_tests exec-split.test) list( APPEND execution_tests allocator-requirements-general.test + dis-task_scheduler-allocation.test + dis-counting_scope-allocation.test exec-affine.test exec-associate.test exec-awaitable.test diff --git a/tests/beman/execution/dis-counting_scope-allocation.test.cpp b/tests/beman/execution/dis-counting_scope-allocation.test.cpp new file mode 100644 index 00000000..329365cf --- /dev/null +++ b/tests/beman/execution/dis-counting_scope-allocation.test.cpp @@ -0,0 +1,116 @@ +// tests/beman/execution/dis-task_scheduler-allocation.test.cpp -*-C++-*- +// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + +#if 0 +#include +namespace ex = std::execution; +#elif 1 +#include +namespace ex = beman::execution; +#else +#include +namespace ex = stdexec; +#endif + +#ifdef BEMAN_HAS_IMPORT_STD +import std; +#else +#include +#include +#include +#include +#include +#include +#include +#include +#endif + +// ---------------------------------------------------------------------------- + +#if defined(__SANITIZE_THREAD__) +#define BEMAN_TSAN_ENABLED 1 +#elif defined(__has_feature) +#if __has_feature(thread_sanitizer) +#define BEMAN_TSAN_ENABLED 1 +#endif +#endif + +std::atomic new_disabled{false}; + +void disable_new() { new_disabled = true; } + +#ifndef BEMAN_TSAN_ENABLED +void* operator new(std::size_t size) { + if (new_disabled) + std::abort(); + return std::malloc(size); +} +void* operator new[](std::size_t size) { + if (new_disabled) + std::abort(); + return std::malloc(size); +} +void operator delete(void* ptr) noexcept { std::free(ptr); } +void operator delete(void* ptr, std::size_t) noexcept { std::free(ptr); } +void operator delete[](void* ptr) noexcept { std::free(ptr); } +void operator delete[](void* ptr, std::size_t) noexcept { std::free(ptr); } +#endif + +// ---------------------------------------------------------------------------- + +namespace { +inline std::pmr::memory_resource* malloc_resource() { + struct resource_t : std::pmr::memory_resource { + void* do_allocate(std::size_t bytes, std::size_t) override { return std::malloc(bytes); } + void do_deallocate(void* p, std::size_t, std::size_t) override { std::free(p); } + bool do_is_equal(const std::pmr::memory_resource& other) const noexcept override { return this == &other; } + }; + static constinit resource_t res{}; + return &res; +} + +template +struct allocator { + using value_type = T; + std::pmr::memory_resource* resource{}; + allocator(std::pmr::memory_resource* r) : resource(r) {} + template + allocator(const allocator& other) : resource(other.resource) {} + T* allocate(std::size_t n) { return static_cast(resource->allocate(n * sizeof(T))); } + void deallocate(T* p, std::size_t) noexcept { resource->deallocate(p, 0, alignof(T)); } + bool operator==(const allocator&) const noexcept = default; + operator std::pmr::memory_resource*() const noexcept { return resource; } +}; + +struct allocator_env { + using allocator_type = std::pmr::polymorphic_allocator; +}; +} // namespace + +// ---------------------------------------------------------------------------- + +int main(int ac, char*[]) { + allocator alloc(ac == 1 ? malloc_resource() : std::pmr::new_delete_resource()); + ex::env env{ex::prop{ex::get_allocator, alloc}}; + + ex::run_loop loop{}; + std::thread thread([&] { loop.run(); }); + disable_new(); + + ex::counting_scope scope{}; + + ex::spawn( + ex::starts_on(loop.get_scheduler(), ex::just() | ex::then([]() noexcept { std::cout << "hello, world\n"; })), + scope.get_token(), + env); + + ex::spawn(ex::starts_on(loop.get_scheduler(), ex::just() | ex::then([]() noexcept { + std::cout << "goodbye, world\n"; + }) | ex::then([&]() noexcept { loop.finish(); })), + scope.get_token(), + env); + + ex::sync_wait(scope.join()); + + thread.join(); +} diff --git a/tests/beman/execution/dis-task_scheduler-allocation.test.cpp b/tests/beman/execution/dis-task_scheduler-allocation.test.cpp new file mode 100644 index 00000000..4e88ab55 --- /dev/null +++ b/tests/beman/execution/dis-task_scheduler-allocation.test.cpp @@ -0,0 +1,140 @@ +// tests/beman/execution/dis-task_scheduler-allocation.test.cpp -*-C++-*- +// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception + +#if 0 +#include +namespace ex = std::execution; +#elif 1 +#include +namespace ex = beman::execution; +#else +#include +namespace ex = stdexec; +#endif + +#ifdef BEMAN_HAS_IMPORT_STD +import std; +#else +#include +#include +#include +#include +#include +#include +#include +#endif + +// ---------------------------------------------------------------------------- + +#if defined(__SANITIZE_THREAD__) +#define BEMAN_TSAN_ENABLED 1 +#elif defined(__has_feature) +#if __has_feature(thread_sanitizer) +#define BEMAN_TSAN_ENABLED 1 +#endif +#endif + +#ifndef BEMAN_TSAN_ENABLED +void* operator new(std::size_t size) { + std::abort(); + return std::malloc(size); +} +void* operator new[](std::size_t size) { + std::abort(); + return std::malloc(size); +} +void operator delete(void* ptr) noexcept { std::free(ptr); } +void operator delete(void* ptr, std::size_t) noexcept { std::free(ptr); } +void operator delete[](void* ptr) noexcept { std::free(ptr); } +void operator delete[](void* ptr, std::size_t) noexcept { std::free(ptr); } +#endif + +// ---------------------------------------------------------------------------- + +namespace { +constexpr std::size_t big_size{40}; +struct big_scheduler { + using scheduler_concept = ex::scheduler_tag; + std::array data{}; + + template + struct state { + using operation_state_concept = ex::operation_state_tag; + std::remove_reference_t receiver; + std::array data{}; + state(R&& r) : receiver(std::forward(r)) {} + void start() noexcept { ex::set_value(std::move(receiver)); } + }; + + struct sender { + using sender_concept = ex::sender_tag; + std::array data{}; + + struct env { + auto query(const ex::get_completion_scheduler_t&) const noexcept { + return big_scheduler{}; + } + }; + auto get_env() const noexcept { return env{}; } + + template + static consteval auto get_completion_signatures() { + return ex::completion_signatures{}; + } + + template + state connect(R&& r) && { + return state{std::forward(r)}; + } + }; + + auto schedule() const noexcept { return sender{}; } + + auto query(ex::get_forward_progress_guarantee_t) const noexcept { + return ex::forward_progress_guarantee::parallel; + } + + bool operator==(const big_scheduler&) const noexcept = default; +}; + +static_assert(big_size <= sizeof(big_scheduler)); +static_assert(ex::sender); +static_assert(ex::scheduler); + +inline std::pmr::memory_resource* malloc_resource() { + struct resource_t : std::pmr::memory_resource { + void* do_allocate(std::size_t bytes, std::size_t) override { return std::malloc(bytes); } + void do_deallocate(void* p, std::size_t, std::size_t) override { std::free(p); } + bool do_is_equal(const std::pmr::memory_resource& other) const noexcept override { return this == &other; } + }; + static constinit resource_t res{}; + return &res; +} + +template +struct allocator { + using value_type = T; + std::pmr::memory_resource* resource{}; + allocator(std::pmr::memory_resource* r) : resource(r) {} + template + allocator(const allocator& other) : resource(other.resource) {} + T* allocate(std::size_t n) { return static_cast(resource->allocate(n * sizeof(T))); } + void deallocate(T* p, std::size_t) noexcept { resource->deallocate(p, 0, alignof(T)); } + bool operator==(const allocator&) const noexcept = default; + operator std::pmr::memory_resource*() const noexcept { return resource; } +}; + +struct allocator_env { + using allocator_type = std::pmr::polymorphic_allocator; +}; +} // namespace + +// ---------------------------------------------------------------------------- + +int main(int ac, char*[]) { + allocator alloc(ac == 1 ? malloc_resource() : std::pmr::new_delete_resource()); + ex::sync_wait(ex::starts_on(big_scheduler{}, [](std::allocator_arg_t, auto) -> ex::task { + auto s = co_await ex::read_env(ex::get_start_scheduler); + co_await ex::starts_on(s, ex::just()); + }(std::allocator_arg, alloc) | ex::write_env(ex::env{ex::prop{ex::get_allocator, alloc}}))); +}