dietmarkuehl · Nov 1, 2026 · Nov 1, 2026 · Nov 1, 2026 · Nov 1, 2026 · Nov 2, 2026
@@ -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)

@@ -741,13 +741,13 @@ The expression <code>let_stopped(<i>upstream</i>, <i>fun</i>)</code> yields a se
 The expression <code>let_value(<i>upstream</i>, <i>fun</i>)</code> yields a sender <code><i>sndr</i></code> which uses a successful completion (<code>set_value</code>) of <code><i>upstream</i></code> as arguments to invoke <code><i>fun</i></code> which has to return another sender <code><i>inner-sndr</i></code> and the completion of <code><i>inner-sndr</i></code> becomes the completion of <code><i>sndr</i></code>. If this invocation results in an exception <code><i>sndr</i></code> completes with an error completion the result of <code>std::current_exception</code>. If <code><i>upstream</i></code> completes with a cancellation (<code>set_stopped</code>) or with an error (<code>set_error</code>) this completion becomes the completion of <code><i>sndr</i></code>.
 </details>
 <details>
-<summary><code>on(_sched_, _sndr_)</code>, <code>on(_sndr, _sched_, _closure_)</code></summary>
+<summary><code>on(<i>sched</i>, <i>sndr</i>)</code>, <code>on(<i>sndr</i>, <i>sched</i>, <i>closure</i>)</code></summary>
 The <code>on</code> algorithm is a pipeable sender adaptor. Let <code><i>on-sender</i></code> be
 <ul>
-  <li><code>_sndr_</code> when the form <code>on(_sched_, _sndr_)</code> is used;</li>
-  <li><code>_closure_(_sndr_)</code> when the form <code>on(_sndr_, _sched_, _closure_)</code> is used.</li>
+  <li><code><i>sndr</i></code> when the form <code>on(<i>sched</i>, <i>sndr</i>)</code> is used;</li>
+  <li><code><i>closure</i>(<i>sndr</i>)</code> when the form <code>on(<i>sndr</i>, <i>sched</i>, <i>closure</i>)</code> is used.</li>
 </ul>
-The sender <code><i>on-sndr</i></code> is started on <code>_sched_</code>'s execution context.
+The sender <code><i>on-sndr</i></code> is started on <code><i>sched</i></code>'s execution context.
 The <code>on</code> algorithm completes on the original scheduler (obtained using <code>get_start_scheduler</code>) with the result of the <code><i>on-sndr</i></code>.
 </details>
 <details>
@@ -808,6 +808,24 @@ The expression <code>write_env(<i>sender</i>, <i>env</i>)</code> creates a sende
 which passes a receiver to <code><i>sender</i></code> which combines the environment
 <code><i>env</i></code> with the environment from the receiver's environment. The
 queries from <code><i>env</i></code> take precedence over those from the receiver's environment.
+<details>
+<summary>Example</summary>
+
+This example shows using `write_env` providing an environment with an allocator:
+
+```c++
+std::pmr::monotonic_buffer_resource        mbr;
+std::pmr::polymorphic_allocator<std::byte> 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 `|`.
+</details>
 </details>
 
 
@@ -890,7 +908,7 @@ an object <code><i>loop</i></code> of type <code>run_loop</code> are:
 <ul>
 <li><code><i>loop</i>.get_scheduler()</code> to get a scheduler scheduling work on <code><i>loop</i></code>.</li>
 <li><code><i>loop</i>.finish()</code> to request <code><i>loop</i></code> to exit processing work items. Note that <code>run_loop</code> doesn't maintain a stop source, i.e., when this operation is invoked the work doesn't get cancelled.</li>
-<li><code><i>loop</i>.run()</code> to have the current thread executed work items scheduled on <code><i>loop</i></code>. The thread will continue processing work items as long as <code><i>loop</i>.finish()</code> wasn't called or there is work scheduled on <code><i>loop</i></code>.
+<li><code><i>loop</i>.run()</code> to have the current thread execute work items scheduled on <code><i>loop</i></code>. The thread will continue processing work items as long as <code><i>loop</i>.finish()</code> wasn't called or there is work scheduled on <code><i>loop</i></code>.
 </ul>
 </details>
 
@@ -1167,3 +1185,19 @@ This concept determines if an object of type <code>Rcvr</code> supports the comp
 <summary><code>concept <i>valid-specialization</i>&lt;T, Args...&gt;</code></summary>
 This concept determines the class template <code>T</code> can be specialized with arguments <code>Args...</code>, i.e., if <code>T&lt;Args...&gt;</code> is valid.
 </details>
+
+
+<details>
+<summary><code>concept <i>simple-allocator</i>&lt;Alloc&gt;</code></summary>
+
+The concept <code><i>simple-allocator</i>&lt;Alloc&gt;</code> determines if
+<code>Alloc</code> is a basic allocator. If `alloc` is an object of type
+`Alloc`, the following need to be `true`:
+
+<ul>
+    <li>`alloc.allocate(n)` has type `Alloc::value_type*`.</li>
+    <li>`alloc.deallocate(alloc.allocate(n), n)` needs to be valid.</i>
+    <li>The type `Alloc` needs to be copy-constructible.</i>
+    <li>Two objects of type `Alloc` need to be equality comparable.</i>
+</ul>
+</details>
@@ -32,7 +32,7 @@ template <typename Allocator, typename Alloc, typename... A>
 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 <typename Allocator, typename A0, typename... A>

@@ -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<Sched>, Alloc>>(
-              alloc, ::std::forward<Sched>(sched), ::std::move(alloc))) {}
+              alloc, ::std::forward<Sched>(sched), alloc)) {}
 
     template <::beman::execution::detail::simple_allocator Alloc>
     explicit task_scheduler(task_scheduler sched, const Alloc&) noexcept : task_scheduler(::std::move(sched)) {}

@@ -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 <beman/execution/detail/prop.hpp>
 #include <beman/execution/detail/read_env.hpp>
 #include <beman/execution/detail/receiver.hpp>
+#include <beman/execution/detail/run_loop.hpp>
 #include <beman/execution/detail/schedule_from.hpp>
 #include <beman/execution/detail/schedule.hpp>
 #include <beman/execution/detail/scheduler.hpp>
+#include <beman/execution/detail/scheduler_tag.hpp>
 #include <beman/execution/detail/scope_token.hpp>
 #include <beman/execution/detail/sender_adaptor_closure.hpp>
 #include <beman/execution/detail/sender_in.hpp>

@@ -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

@@ -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 <execution>
+namespace ex = std::execution;
+#elif 1
+#include <beman/execution.hpp>
+namespace ex = beman::execution;
+#else
+#include <stdexec/execution.hpp>
+namespace ex = stdexec;
+#endif
+
+#ifdef BEMAN_HAS_IMPORT_STD
+import std;
+#else
+#include <array>
+#include <cassert>
+#include <cstddef>
+#include <cstdlib>
+#include <iostream>
+#include <memory>
+#include <memory_resource>
+#include <type_traits>
+#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<bool> 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 <typename T>
+struct allocator {
+    using value_type = T;
+    std::pmr::memory_resource* resource{};
+    allocator(std::pmr::memory_resource* r) : resource(r) {}
+    template <typename U>
+    allocator(const allocator<U>& other) : resource(other.resource) {}
+    T*   allocate(std::size_t n) { return static_cast<T*>(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<std::byte>;
+};
+} // namespace
+
+// ----------------------------------------------------------------------------
+
+int main(int ac, char*[]) {
+    allocator<std::byte> 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();
+}
@@ -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 <execution>
+namespace ex = std::execution;
+#elif 1
+#include <beman/execution.hpp>
+namespace ex = beman::execution;
+#else
+#include <stdexec/execution.hpp>
+namespace ex = stdexec;
+#endif
+
+#ifdef BEMAN_HAS_IMPORT_STD
+import std;
+#else
+#include <array>
+#include <cassert>
+#include <cstddef>
+#include <cstdlib>
+#include <memory>
+#include <memory_resource>
+#include <type_traits>
+#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<char, big_size> data{};
+
+    template <ex::receiver R>
+    struct state {
+        using operation_state_concept = ex::operation_state_tag;
+        std::remove_reference_t<R> receiver;
+        std::array<char, big_size> data{};
+        state(R&& r) : receiver(std::forward<R>(r)) {}
+        void start() noexcept { ex::set_value(std::move(receiver)); }
+    };
+
+    struct sender {
+        using sender_concept = ex::sender_tag;
+        std::array<char, big_size> data{};
+
+        struct env {
+            auto query(const ex::get_completion_scheduler_t<ex::set_value_t>&) const noexcept {
+                return big_scheduler{};
+            }
+        };
+        auto get_env() const noexcept { return env{}; }
+
+        template <typename... Env>
+        static consteval auto get_completion_signatures() {
+            return ex::completion_signatures<ex::set_value_t()>{};
+        }
+
+        template <ex::receiver R>
+        state<R> connect(R&& r) && {
+            return state<R>{std::forward<R>(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<big_scheduler::sender>);
+static_assert(ex::scheduler<big_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 <typename T>
+struct allocator {
+    using value_type = T;
+    std::pmr::memory_resource* resource{};
+    allocator(std::pmr::memory_resource* r) : resource(r) {}
+    template <typename U>
+    allocator(const allocator<U>& other) : resource(other.resource) {}
+    T*   allocate(std::size_t n) { return static_cast<T*>(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<std::byte>;
+};
+} // namespace
+
+// ----------------------------------------------------------------------------
+
+int main(int ac, char*[]) {
+    allocator<std::byte> 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<void, allocator_env> {
+        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}})));
+}