Skip to content

Commit 16e8875

Browse files
m-filam-fila
authored
add schedule_on algorithm for alien coroutine (#57)
* add schedule on algorithm compatible with alien coroutine * update examples readme
1 parent 18093b9 commit 16e8875

4 files changed

Lines changed: 261 additions & 1 deletion

File tree

‎examples/CMakeLists.txt‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,8 @@ add_executable(alien_sync_wait alien_sync_wait.cpp)
3434
target_link_libraries(alien_sync_wait PRIVATE CoroutineTests)
3535
add_executable(alien_counting_scope alien_counting_scope.cpp)
3636
target_link_libraries(alien_counting_scope PRIVATE CoroutineTests)
37-
37+
add_executable(alien_schedule_on alien_schedule_on.cpp)
38+
target_link_libraries(alien_schedule_on PRIVATE CoroutineTests)
3839

3940
if(BUILD_STDEXEC)
4041
add_executable(exec_task_stdexec exec_task.cpp)

‎examples/README.md‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -144,6 +144,12 @@ Link: [alien_counting_scope.cpp](alien_counting_scope.cpp)
144144

145145
This example demonstrates dynamic work submitting with `counting_scope` compatible with coroutine semantic as in ["Alien" example](#alien). `spawn` schedules execution of a coroutine that returns `void`. `join()` blocks the current thread until all submitted coroutines are finished (and rethrows the first exception captured from submitted work, if any).
146146

147+
## Alien schedule_on
148+
149+
Link: [alien_schedule_on.cpp](alien_schedule_on.cpp)
150+
151+
This example demonstrates changing scheduler used by part of a chain of coroutines following the semantics from ["Alien" example](#alien). The `schedule_on` algorithm can be used to adapt a coroutine to use a different scheduler than its parent. Starting the coroutine suspends its parent and reschedule work on the new scheduler, finishing the coroutine reschedules continuation of its parent using parent's scheduler.
152+
147153
## Capy task
148154

149155
Link: [capy_task.cpp](capy_task.cpp)

‎examples/alien_schedule_on.cpp‎

Lines changed: 74 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,74 @@
1+
#include <coroutine>
2+
#include <functional>
3+
#include <string_view>
4+
5+
#include "CoroutineTests/alien/schedule_on.hpp"
6+
#include "CoroutineTests/alien/sync_wait.hpp"
7+
#include "CoroutineTests/alien/tool.hpp"
8+
#include "CoroutineTests/threadpool.hpp"
9+
#include "alien_timer.hpp" // AsyncTimer
10+
#include "logging_utils.hpp" // log, format_name
11+
12+
using namespace CoroutineTests::alien;
13+
14+
tool::Task<void> inner_most(std::string_view parent) {
15+
const auto self = format_name(parent, "inner_most");
16+
log(self) << "Starting inner_most" << std::endl;
17+
log(self) << "Calling async API in inner_most" << std::endl;
18+
auto status = co_await AsyncTimer{std::chrono::milliseconds(75),
19+
StatusCode::SUCCESS, self};
20+
log(self) << "Result from async API in inner_most: " << status << std::endl;
21+
log(self) << "Finishing inner_most" << std::endl;
22+
co_return;
23+
}
24+
25+
tool::Task<tool::StatusCode> inner(std::string_view parent) {
26+
const auto self = format_name(parent, "inner");
27+
log(self) << "Starting inner" << std::endl;
28+
co_await inner_most(self);
29+
log(self) << "Finishing inner" << std::endl;
30+
co_return tool::StatusCode::SUCCESS;
31+
}
32+
33+
tool::Task<void> outer(
34+
std::function<void(std::coroutine_handle<>)> inner_scheduler,
35+
std::string_view parent) {
36+
const auto self = format_name(parent, "outer");
37+
log(self) << "Starting outer" << std::endl;
38+
auto status = co_await schedule_on(inner_scheduler, inner(self));
39+
log(self) << "Received status " << status << " from inner" << std::endl;
40+
log(self) << "Finishing outer" << std::endl;
41+
co_return;
42+
}
43+
44+
tool::Task<tool::StatusCode> outer_most(
45+
std::function<void(std::coroutine_handle<>)> inner_scheduler,
46+
std::string_view parent) {
47+
const auto self = format_name(parent, "outer_most");
48+
log(self) << "Starting outer_most" << std::endl;
49+
co_await outer(inner_scheduler, self);
50+
log(self) << "Finishing outer_most" << std::endl;
51+
co_return tool::StatusCode::SUCCESS;
52+
}
53+
54+
int main() {
55+
log() << "main Starting" << std::endl;
56+
CoroutineTests::Threadpool threadpool(1);
57+
auto scheduler = [&threadpool](std::coroutine_handle<> handle) {
58+
log() << "scheduler called, enqueuing work" << std::endl;
59+
threadpool.enqueue_task(handle);
60+
};
61+
62+
CoroutineTests::Threadpool inner_threadpool(1);
63+
auto inner_scheduler = [&inner_threadpool](std::coroutine_handle<> handle) {
64+
log() << "inner_scheduler called, enqueuing work" << std::endl;
65+
inner_threadpool.enqueue_task(handle);
66+
};
67+
68+
log() << "main Launching outer_most and waiting for completion..."
69+
<< std::endl;
70+
auto status = CoroutineTests::alien::sync_wait(
71+
scheduler, outer_most(inner_scheduler, "main"));
72+
log() << "main Final status of outer_most " << status << "" << std::endl;
73+
return 0;
74+
}

‎include/CoroutineTests/alien/schedule_on.hpp‎

Lines changed: 179 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,179 @@
1+
#pragma once
2+
3+
#include <concepts>
4+
#include <coroutine>
5+
#include <exception>
6+
#include <functional>
7+
#include <optional>
8+
9+
namespace CoroutineTests::alien {
10+
11+
namespace detail::schedule_on {
12+
namespace concepts {
13+
template <typename T>
14+
concept HasScheduler = requires(T t) {
15+
{
16+
t.get_scheduler()
17+
} -> std::convertible_to<std::function<void(std::coroutine_handle<>)>>;
18+
};
19+
} // namespace concepts
20+
21+
// Helper coroutine type allowing for having different scheduler than parent
22+
// coroutine.
23+
template <typename ResultType>
24+
class [[nodiscard]] Task {
25+
26+
static_assert(std::movable<ResultType> || std::same_as<ResultType, void>,
27+
"Task<ResultType> requires ResultType to be movable or void");
28+
29+
public:
30+
using result_type = ResultType;
31+
32+
struct promise_type; // typedef required by coroutines
33+
using handle_type =
34+
std::coroutine_handle<promise_type>; // not required but useful
35+
36+
// Constructor from coroutine handle
37+
Task(handle_type coroutine_handle) : m_coroutine(coroutine_handle) {}
38+
~Task() {
39+
if (m_coroutine) {
40+
m_coroutine.destroy();
41+
}
42+
}
43+
Task() = default;
44+
Task(const Task&) = delete;
45+
Task& operator=(const Task&) = delete;
46+
Task(Task&& other) noexcept : m_coroutine{other.m_coroutine} {
47+
other.m_coroutine = {};
48+
}
49+
Task& operator=(Task&& other) noexcept {
50+
if (this != &other) {
51+
if (m_coroutine) {
52+
m_coroutine.destroy();
53+
}
54+
m_coroutine = other.m_coroutine;
55+
other.m_coroutine = {};
56+
}
57+
return *this;
58+
}
59+
60+
// Awaitable interface: always suspend to allow async execution
61+
bool await_ready() const noexcept { return false; }
62+
// Awaitable interface: setup parent relationship, suspend parent and
63+
// schedule this coroutine on its scheduler
64+
template <detail::schedule_on::concepts::HasScheduler T>
65+
inline void await_suspend(std::coroutine_handle<T> handle) noexcept;
66+
// Awaitable interface: return result or rethrow exception on resume
67+
result_type await_resume() const;
68+
69+
private:
70+
handle_type m_coroutine = nullptr;
71+
};
72+
73+
// Helper for handling co_return in promise_type, default implementation for
74+
// non-void ResultType
75+
template <typename ResultType>
76+
struct ReturnHelper {
77+
// Storage for the co_return result value
78+
std::optional<ResultType> m_value;
79+
80+
// Required by coroutines, mutually exclusive with return_void
81+
// Store the co_return value
82+
template <typename T>
83+
requires std::constructible_from<ResultType, T&&>
84+
void return_value(T&& value) {
85+
m_value.emplace(std::forward<T>(value));
86+
}
87+
// Overload to resolve ambiguity
88+
void return_value(ResultType value) { m_value.emplace(std::move(value)); }
89+
};
90+
91+
// Specialization for void return type
92+
template <>
93+
struct ReturnHelper<void> {
94+
// Required by coroutines, mutually exclusive with return_value
95+
// Handle co_return without value
96+
void return_void() {}
97+
};
98+
99+
template <typename ResultType>
100+
struct Task<ResultType>::promise_type
101+
: public ReturnHelper<typename Task<ResultType>::result_type> {
102+
// Storage for exceptions thrown in the coroutine body
103+
std::exception_ptr m_exception;
104+
// Handle to the parent coroutine that co_awaited this task
105+
std::coroutine_handle<> m_parent;
106+
// Handle to scheduler to resume this coroutine and propagate to children
107+
std::function<void(std::coroutine_handle<>)> m_scheduler;
108+
// Handle to the scheduler to resume parent coroutine
109+
std::function<void(std::coroutine_handle<>)> m_parent_scheduler;
110+
111+
// Non-default constructor to pass the scheduler. The constructor will be
112+
// used if coroutine function has the same signature. The unused parameters
113+
// are here only to match the signature.
114+
template <typename Coro>
115+
promise_type(std::function<void(std::coroutine_handle<>)> scheduler,
116+
Coro&&) noexcept
117+
: m_scheduler(scheduler) {}
118+
119+
// Accessor for scheduler used by child coroutines
120+
const auto& get_scheduler() const { return m_scheduler; }
121+
// Schedule resumption of this task
122+
void reschedule() { m_scheduler(handle_type::from_promise(*this)); }
123+
124+
// Required by coroutines: create the object
125+
Task get_return_object() { return {handle_type::from_promise(*this)}; }
126+
// Required by coroutines: suspend immediately on start (lazy execution)
127+
std::suspend_always initial_suspend() const { return {}; }
128+
// Required by coroutines: handle completion and resume parent
129+
auto final_suspend() const noexcept {
130+
struct final_awaiter {
131+
// Don't skip final suspension
132+
bool await_ready() const noexcept { return false; }
133+
// Resume parent coroutine on its own scheduler
134+
void await_suspend(handle_type handle) noexcept {
135+
auto parent = handle.promise().m_parent;
136+
auto parent_scheduler = handle.promise().m_parent_scheduler;
137+
if (parent && parent_scheduler) {
138+
parent_scheduler(parent);
139+
}
140+
}
141+
// No action needed on resume
142+
void await_resume() const noexcept {}
143+
};
144+
return final_awaiter{};
145+
}
146+
// Required by coroutines: capture exceptions for later rethrowing
147+
void unhandled_exception() { m_exception = std::current_exception(); }
148+
};
149+
150+
template <typename ResultType>
151+
template <detail::schedule_on::concepts::HasScheduler T>
152+
inline void Task<ResultType>::await_suspend(
153+
std::coroutine_handle<T> handle) noexcept {
154+
m_coroutine.promise().m_parent = handle;
155+
m_coroutine.promise().m_parent_scheduler = handle.promise().get_scheduler();
156+
m_coroutine.promise().reschedule();
157+
}
158+
159+
template <typename ResultType>
160+
inline typename Task<ResultType>::result_type Task<ResultType>::await_resume()
161+
const {
162+
if (m_coroutine.promise().m_exception) {
163+
std::rethrow_exception(m_coroutine.promise().m_exception);
164+
}
165+
if constexpr (std::same_as<result_type, void>) {
166+
return;
167+
} else {
168+
return std::move(m_coroutine.promise().m_value).value();
169+
}
170+
}
171+
} // namespace detail::schedule_on
172+
173+
template <typename Coro>
174+
auto schedule_on(std::function<void(std::coroutine_handle<>)>, Coro coro)
175+
-> detail::schedule_on::Task<typename Coro::result_type> {
176+
co_return co_await coro;
177+
}
178+
179+
} // namespace CoroutineTests::alien

0 commit comments

Comments
 (0)