Skip to content

Task sender #75

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
dietmarkuehl merged 15 commits into from
Mar 23, 2026
Merged

Task sender #75

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
15 commits
Select commit Hold shift + click to select a range
ae8cbd0
an experiment propagating an allocator via a sender
dietmarkuehl Feb 18, 2026
92c8ab7
demo of a sender for delayed coroutine frame allocation
dietmarkuehl Feb 22, 2026
89e367f
fixed some issues to match the most recent execution main
dietmarkuehl Mar 8, 2026
a3c57d5
added a demo of defering coroutine frame creation
dietmarkuehl Mar 15, 2026
1a5b363
adding an allocation example
dietmarkuehl Mar 23, 2026
e57ca49
Merge branch 'main' into task-sender
dietmarkuehl Mar 23, 2026
6a8bb7f
worked around using a possibly older version of execution
dietmarkuehl Mar 23, 2026
4a21699
Merge branch 'main' into task-sender
dietmarkuehl Mar 23, 2026
b1c78a1
fix problem if allocator types don't match
dietmarkuehl Mar 23, 2026
c86bfde
clang format
dietmarkuehl Mar 23, 2026
9c651f9
fix formatting issue
dietmarkuehl Mar 23, 2026
7f90c9c
fixed missing operator delete
dietmarkuehl Mar 23, 2026
e71e234
fix memory leak in example
dietmarkuehl Mar 23, 2026
3ab8250
remove task-sender from compiled examples for now
dietmarkuehl Mar 23, 2026
107abb3
restore task-sender for non-MSVC
dietmarkuehl Mar 23, 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
2 changes: 2 additions & 0 deletions examples/CMakeLists.txt
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -31,6 +31,8 @@ set(ALL_EXAMPLES
if(NOT MSVC)
list(
APPEND ALL_EXAMPLES
task-sender
alloc-1
bulk
c++now-allocator
c++now-cancel
Expand Down
124 changes: 124 additions & 0 deletions examples/alloc-1.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,124 @@
// examples/alloc-1.cpp -*-C++-*-
// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception

#include <beman/execution/task.hpp>
#include <beman/execution/execution.hpp>
#include <functional>
#include <iostream>
#include <memory>
#include <new>
#include <cstdlib>
#include <utility>

Comment on lines +4 to +12
Copy link

Copilot AI Mar 23, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This example relies on std::pmr types (std::pmr::memory_resource, std::pmr::polymorphic_allocator) but does not include <memory_resource>. Add <memory_resource> (and any other direct headers) to ensure portable compilation.

Copilot uses AI. Check for mistakes.
namespace ex = beman::execution;

// ----------------------------------------------------------------------------
// --- defer_frame turns yields a sender calling a coroutine upon start() ---

template <typename Mem, typename Self = void>
struct defer_frame {
Mem mem;
Self self;
template <typename... Arg>
auto operator()(Arg&&... arg) const {
return ex::let_value(ex::read_env(ex::get_allocator),
[mem = this->mem, self = this->self, ... a = std::forward<Arg>(arg)](auto alloc) {
return std::invoke(mem, self, std::allocator_arg, alloc, std::move(a)...);
});
}
template <typename Alloc, typename... Arg>
auto operator()(::std::allocator_arg_t, Alloc alloc, Arg&&... arg) const {
return ex::let_value(ex::just(alloc),
[&mem = this->mem, self = this->self, ... a = std::forward<Arg>(arg)](auto alloc) {
return std::invoke(mem, self, std::allocator_arg, alloc, std::move(a)...);
});
}
auto operator()(::std::allocator_arg_t) const = delete;
};

template <typename Task>
struct defer_frame<Task, void> {
Task task;
template <typename... Arg>
auto operator()(Arg&&... arg) const {
return ex::let_value(ex::read_env(ex::get_allocator),
[task = this->task, ... a = std::forward<Arg>(arg)](auto alloc) {
return std::invoke(task, std::allocator_arg, alloc, std::move(a)...);
});
}
template <typename Alloc, typename... Arg>
auto operator()(::std::allocator_arg_t, Alloc alloc, Arg&&... arg) const {
return ex::let_value(ex::just(alloc), [&task = this->task, ... a = std::forward<Arg>(arg)](auto alloc) {
return std::invoke(task, std::allocator_arg, alloc, std::move(a)...);
});
}
auto operator()(::std::allocator_arg_t) const = delete;
};

// ----------------------------------------------------------------------------

void* operator new(std::size_t n) {
auto p = std::malloc(n);
std::cout << " global new(" << n << ")->" << p << "\n";
return p;
}
void operator delete(void* ptr) noexcept {
std::cout << " global operator delete(" << ptr << ")\n";
std::free(ptr);
}
void operator delete(void* ptr, std::size_t size) noexcept {
std::cout << " global operator delete(" << ptr << ", " << size << ")\n";
std::free(ptr);
}

struct resource : std::pmr::memory_resource {
void* do_allocate(std::size_t n, std::size_t) override {
auto p{std::malloc(n)};
std::cout << " resource::allocate(" << n << ")->" << p << "\n";
return p;
}
void do_deallocate(void* p, std::size_t n, std::size_t) override {
std::cout << " resource::deallocate(" << p << ", " << n << ")\n";
std::free(p);
}
bool do_is_equal(const std::pmr::memory_resource& other) const noexcept override { return this == &other; }
};

// ----------------------------------------------------------------------------

using allocator_type = std::pmr::polymorphic_allocator<std::byte>;
struct alloc_env {
using allocator_type = ::allocator_type;
};
template <typename T = void>
using a_task = ex::task<T, alloc_env>;

a_task<int> hidden_async_fun(std::allocator_arg_t, ::allocator_type, int value) { co_return value; }
auto async_fun(int value) { return defer_frame(&hidden_async_fun)(value); }

int main() {
std::cout << std::unitbuf;
resource res{};
allocator_type alloc(&res);

std::cout << "not setting up an allocator:\n";
ex::sync_wait([]() -> a_task<> {
auto result{co_await async_fun(17)};
std::cout << " result=" << result << "\n";
}());

std::cout << "setting up an allocator:\n";
ex::sync_wait(ex::write_env(
[]() -> a_task<> {
auto result{co_await async_fun(17)};
std::cout << " result=" << result << "\n";
}(),
ex::env{ex::prop{ex::get_allocator, alloc}}));

std::cout << "setting up an allocator and using defer_frame:\n";
ex::sync_wait(ex::write_env(defer_frame([](auto, auto) -> a_task<> {
auto result{co_await async_fun(17)};
std::cout << " result=" << result << "\n";
})(),
ex::env{ex::prop{ex::get_allocator, alloc}}));
}
28 changes: 28 additions & 0 deletions examples/odd-completions.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,28 @@
// examples/hello.cpp -*-C++-*-
Copy link

Copilot AI Mar 23, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This file header comment says examples/hello.cpp, but the filename is odd-completions.cpp. Update the header comment so it matches the actual file name (consistent with other examples like examples/odd-return.cpp).

Suggested change
// examples/hello.cpp -*-C++-*-
// examples/odd-completions.cpp -*-C++-*-

Copilot uses AI. Check for mistakes.
// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception

#include <beman/execution/task.hpp>
#include <beman/execution/execution.hpp>
#include <iostream>

namespace ex = beman::execution;

int main() {
return std::get<0>(ex::sync_wait([]() -> ex::task<int> {
std::cout << "Hello, world!\n";
co_return co_await ex::just(0);
}())
.value_or(std::tuple(-1)));
ex::sync_wait([](int value) -> ex::task<int> {
switch (value) {
default:
Comment on lines +10 to +18
Copy link

Copilot AI Mar 23, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There is an early return at the start of main, so the subsequent ex::sync_wait(...) block is unreachable and will never run. Either remove the early return or restructure main so both examples execute as intended.

Copilot uses AI. Check for mistakes.
co_return value;
case -1:
co_yield ex::with_error(std::make_exception_ptr(value));
case 2:
throw value;
case 0:
co_await ex::just_stopped();
}
}(0));
}
160 changes: 160 additions & 0 deletions examples/task-sender.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,160 @@
// examples/task_sender.hpp -*-C++-*-
// SPDX-License-Identifier: Apache-2.0 WITH LLVM-exception

#include <beman/execution/task.hpp>
#include <beman/execution/execution.hpp>
#include <functional>
#include <iostream>
#include <memory>
Copy link

Copilot AI Mar 23, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This file uses std::pmr::polymorphic_allocator and std::pmr::memory_resource but doesn’t include <memory_resource>, which is the standard header that declares these types. Add the proper header(s) to avoid relying on transitive includes.

Suggested change
#include <memory>
#include <memory>
#include <memory_resource>

Copilot uses AI. Check for mistakes.
#include <new>
#include <cstdlib>
#include <utility>

namespace ex = beman::execution;

void* operator new(std::size_t n) {
auto p = std::malloc(n);
std::cout << "global new(" << n << ")->" << p << "\n";
return p;
}
void operator delete(void* ptr) noexcept {
std::cout << " global operator delete(" << ptr << ")\n";
std::free(ptr);
}
void operator delete(void* ptr, std::size_t size) noexcept {
std::cout << " global operator delete(" << ptr << ", " << size << ")\n";
std::free(ptr);
}

template <typename Mem, typename Self = void>
struct defer_frame {
Mem mem;
Self self;
template <typename... Arg>
auto operator()(Arg&&... arg) const {
return ex::let_value(ex::read_env(ex::get_allocator),
[mem = this->mem, self = this->self, ... a = std::forward<Arg>(arg)](auto alloc) {
return std::invoke(mem, self, std::allocator_arg, alloc, std::move(a)...);
});
}
template <typename Alloc, typename... Arg>
auto operator()(::std::allocator_arg_t, Alloc alloc, Arg&&... arg) const {
return ex::let_value(ex::just(alloc),
[&mem = this->mem, self = this->self, ... a = std::forward<Arg>(arg)](auto alloc) {
return std::invoke(mem, self, std::allocator_arg, alloc, std::move(a)...);
});
}
auto operator()(::std::allocator_arg_t) const = delete;
};

template <typename Task>
struct defer_frame<Task, void> {
Task task;
template <typename... Arg>
auto operator()(Arg&&... arg) const {
return ex::let_value(ex::read_env(ex::get_allocator),
[&task = this->task, ... a = std::forward<Arg>(arg)](auto alloc) {
return std::invoke(task, std::allocator_arg, alloc, std::move(a)...);
});
}
template <typename Alloc, typename... Arg>
auto operator()(::std::allocator_arg_t, Alloc alloc, Arg&&... arg) const {
return ex::let_value(ex::just(alloc), [&task = this->task, ... a = std::forward<Arg>(arg)](auto alloc) {
return std::invoke(task, std::allocator_arg, alloc, std::move(a)...);
});
}
auto operator()(::std::allocator_arg_t) const = delete;
};

struct env {
using allocator_type = std::pmr::polymorphic_allocator<std::byte>;
};

auto lambda{[](int i, auto&&...) -> ex::task<void, env> {
auto alloc = co_await ex::read_env(ex::get_allocator);
alloc.deallocate(alloc.allocate(1), 1);
std::cout << "lambda(" << i << ")\n";
co_return;
}};

class example {
ex::task<void, env> member_(std::allocator_arg_t, std::pmr::polymorphic_allocator<std::byte>, int);
ex::task<void, env> const_member_(std::allocator_arg_t, std::pmr::polymorphic_allocator<std::byte>, int) const;

public:
auto member(int i) { return defer_frame(&example::member_, this)(i); }
auto const_member(int i) { return defer_frame(&example::const_member_, this)(i); }
};

inline ex::task<void, env> example::member_(std::allocator_arg_t, std::pmr::polymorphic_allocator<std::byte>, int i) {
std::cout << "example::member(" << i << ")\n";
co_return;
}
inline ex::task<void, env>
example::const_member_(std::allocator_arg_t, std::pmr::polymorphic_allocator<std::byte>, int i) const {
std::cout << "example::const member(" << i << ")\n";
co_return;
}

struct resource : std::pmr::memory_resource {
void* do_allocate(std::size_t n, std::size_t) override {
auto p{std::malloc(n)};
std::cout << " resource::allocate(" << n << ")->" << p << "\n";
return p;
}
void do_deallocate(void* p, std::size_t n, std::size_t) override {
std::cout << " resource::deallocate(" << p << ", " << n << ")\n";
std::free(p);
}
bool do_is_equal(const std::pmr::memory_resource& other) const noexcept override { return this == &other; }
};

int main() {
resource res{};
std::pmr::polymorphic_allocator<std::byte> alloc(&res);
std::cout << "direct allocator use:\n";
alloc.deallocate(alloc.allocate(1), 1);
std::cout << "write_env/just/then use:\n";
ex::sync_wait(ex::write_env(ex::just(alloc) | ex::then([](auto a) { a.deallocate(a.allocate(1), 1); }),
ex::env{ex::prop{ex::get_allocator, alloc}}));
std::cout << "write_env/read_env/then use:\n";
ex::sync_wait(
ex::write_env(ex::read_env(ex::get_allocator) | ex::then([](auto a) { a.deallocate(a.allocate(1), 1); }),
ex::env{ex::prop{ex::get_allocator, alloc}}));
std::cout << "write_env/let_value/then use:\n";
ex::sync_wait(ex::write_env(ex::just() | ex::let_value([] {
return ex::read_env(ex::get_allocator) |
ex::then([](auto a) { a.deallocate(a.allocate(1), 1); });
}),
ex::env{ex::prop{ex::get_allocator, alloc}}));
std::cout << "write_env/task<>:\n";
ex::sync_wait(ex::write_env(
[]() -> ex::task<> {
auto a = co_await ex::read_env(ex::get_allocator);
a.deallocate(a.allocate(1), 1);
}(),
ex::env{ex::prop{ex::get_allocator, alloc}}));
std::cout << "write_env/task<void, env>:\n";
ex::sync_wait(ex::write_env(
[](auto&&...) -> ex::task<void, env> {
auto a = co_await ex::read_env(ex::get_allocator);
a.deallocate(a.allocate(1), 1);
}(std::allocator_arg, alloc),
ex::env{ex::prop{ex::get_allocator, alloc}}));
std::cout << "write_env/defer_frame<task<void, env>>:\n";
static constexpr defer_frame t0([](auto, auto, int i) -> ex::task<void, env> {
std::cout << " i=" << i << "\n";
auto a = co_await ex::read_env(ex::get_allocator);
a.deallocate(a.allocate(1), 1);
});
ex::sync_wait(ex::write_env(t0(17), ex::env{ex::prop{ex::get_allocator, alloc}}));
std::cout << "write_env/temporary defer_frame<task<void, env>>:\n";
ex::sync_wait(ex::write_env(defer_frame([](auto, auto, auto i) -> ex::task<void, env> {
std::cout << " i=" << i << "\n";
co_await std::suspend_never{};
auto a = co_await ex::read_env(ex::get_allocator);
a.deallocate(a.allocate(1), 1);
})(42),
ex::env{ex::prop{ex::get_allocator, alloc}}));
std::cout << "done\n";
}
4 changes: 4 additions & 0 deletions include/beman/task/detail/inline_scheduler.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,10 @@ struct inline_scheduler {
struct sender {
using sender_concept = ::beman::execution::sender_t;
using completion_signatures = ::beman::execution::completion_signatures<::beman::execution::set_value_t()>;
template <typename...>
static consteval auto get_completion_signatures() noexcept -> completion_signatures {
return {};
}

env get_env() const noexcept { return {}; }
template <::beman::execution::receiver Receiver>
Expand Down
2 changes: 1 addition & 1 deletion include/beman/task/detail/promise_type.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -67,7 +67,7 @@ class promise_type
auto get_return_object() noexcept { return Coroutine(::beman::task::detail::handle<promise_type>(this)); }

template <::beman::execution::sender Sender>
auto await_transform(Sender&& sender) noexcept {
auto await_transform(Sender&& sender) {
if constexpr (requires {
::std::forward<Sender>(sender).as_awaitable(*this);
// typename ::std::remove_cvref_t<Sender>::task_concept;
Expand Down
7 changes: 5 additions & 2 deletions include/beman/task/detail/state.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -48,8 +48,11 @@ struct state : ::beman::task::detail::state_base<T, C>, ::beman::task::detail::s
return std::noop_coroutine();
}
auto do_get_allocator() -> allocator_type override {
if constexpr (requires { ::beman::execution::get_allocator(::beman::execution::get_env(this->receiver)); })
return ::beman::execution::get_allocator(::beman::execution::get_env(this->receiver));
if constexpr (requires {
allocator_type(
::beman::execution::get_allocator(::beman::execution::get_env(this->receiver)));
})
return allocator_type(::beman::execution::get_allocator(::beman::execution::get_env(this->receiver)));
else
return allocator_type{};
}
Expand Down
14 changes: 9 additions & 5 deletions include/beman/task/detail/task.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -46,15 +46,19 @@ class task {
using stop_token_type = decltype(std::declval<stop_source_type>().get_token());

public:
using task_concept = int;
using sender_concept = ::beman::execution::sender_t;
using xcompletion_signatures = ::beman::execution::detail::meta::combine<
using task_concept = int;
using sender_concept = ::beman::execution::sender_t;
using completion_signatures = ::beman::execution::detail::meta::combine<
::beman::execution::completion_signatures<beman::task::detail::completion_t<Value>,
::beman::execution::set_stopped_t()>,
::beman::task::detail::error_types_of_t<Env> >;
template <typename Ev>
auto get_completion_signatures(const Ev&) const& noexcept {
return xcompletion_signatures{};
auto get_completion_signatures(const Ev&) const& noexcept -> completion_signatures {
return {};
}
template <typename...>
static consteval auto get_completion_signatures() noexcept -> completion_signatures {
return {};
}

using promise_type = ::beman::task::detail::promise_type<task, Value, Env>;
Expand Down
4 changes: 4 additions & 0 deletions include/beman/task/detail/task_scheduler.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -126,6 +126,10 @@ class task_scheduler {
public:
using sender_concept = ::beman::execution::sender_t;
using completion_signatures = ::beman::execution::completion_signatures<::beman::execution::set_value_t()>;
template <typename...>
static consteval auto get_completion_signatures() noexcept -> completion_signatures {
return {};
}
Comment on lines 126 to +132
Copy link

Copilot AI Mar 23, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The class documentation lists ex::set_stopped() as a possible completion, but the sender’s declared completion_signatures only contains set_value_t(). Either update the documentation or include set_stopped_t() in the declared completion signatures so they stay consistent.

Copilot uses AI. Check for mistakes.

template <::beman::execution::scheduler S>
explicit sender(S&& s) : inner_sender(static_cast<concrete<S>*>(nullptr), std::forward<S>(s)) {}
Expand Down
Loading