Add when_all() and when_all_ready(). · olegabr/cppcoro@67fea7f · GitHub
Skip to content

Commit 67fea7f

Browse files
lewissbakerlewissbaker
committed
Add when_all() and when_all_ready().
Add 'get_starter()' methods to all task types. Add detail::unwrap_reference_t helper. Add detail::when_all_awaitable helper.
1 parent 671ac0b commit 67fea7f

12 files changed

Lines changed: 1545 additions & 0 deletions

‎include/cppcoro/detail/unwrap_reference.hpp‎

Lines changed: 31 additions & 0 deletions

‎include/cppcoro/detail/when_all_awaitable.hpp‎

Lines changed: 59 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,59 @@
1+
///////////////////////////////////////////////////////////////////////////////
2+
// Copyright (c) Lewis Baker
3+
// Licenced under MIT license. See LICENSE.txt for details.
4+
///////////////////////////////////////////////////////////////////////////////
5+
#ifndef CPPCORO_DETAIL_WHEN_ALL_AWAITABLE_HPP_INCLUDED
6+
#define CPPCORO_DETAIL_WHEN_ALL_AWAITABLE_HPP_INCLUDED
7+
8+
#include <cppcoro/detail/continuation.hpp>
9+
10+
#include <atomic>
11+
12+
namespace cppcoro
13+
{
14+
namespace detail
15+
{
16+
class when_all_awaitable
17+
{
18+
public:
19+
20+
when_all_awaitable(std::size_t count) noexcept
21+
: m_refCount(count + 1)
22+
{}
23+
24+
detail::continuation get_continuation() noexcept
25+
{
26+
return detail::continuation{ &when_all_awaitable::resumer_callback, this };
27+
}
28+
29+
bool await_ready() noexcept
30+
{
31+
return m_refCount.load(std::memory_order_acquire) == 1;
32+
}
33+
34+
bool await_suspend(std::experimental::coroutine_handle<> awaiter) noexcept
35+
{
36+
m_awaiter = awaiter;
37+
return m_refCount.fetch_sub(1, std::memory_order_acq_rel) > 1;
38+
}
39+
40+
void await_resume() noexcept {}
41+
42+
private:
43+
44+
static void resumer_callback(void* state) noexcept
45+
{
46+
auto* that = static_cast<when_all_awaitable*>(state);
47+
if (that->m_refCount.fetch_sub(1, std::memory_order_acq_rel) == 1)
48+
{
49+
that->m_awaiter.resume();
50+
}
51+
}
52+
53+
std::atomic<std::size_t> m_refCount;
54+
std::experimental::coroutine_handle<> m_awaiter;
55+
};
56+
}
57+
}
58+
59+
#endif

‎include/cppcoro/lazy_task.hpp‎

Lines changed: 32 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -356,6 +356,38 @@ namespace cppcoro
356356
return awaitable{ m_coroutine };
357357
}
358358

359+
// Internal helper method for when_all() implementation.
360+
auto get_starter() const noexcept
361+
{
362+
class starter
363+
{
364+
public:
365+
366+
starter(std::experimental::coroutine_handle<promise_type> coroutine) noexcept
367+
: m_coroutine(coroutine)
368+
{}
369+
370+
void start(detail::continuation c) noexcept
371+
{
372+
if (m_coroutine && !m_coroutine.promise().is_ready())
373+
{
374+
m_coroutine.promise().set_continuation(c);
375+
m_coroutine.resume();
376+
}
377+
else
378+
{
379+
c.resume();
380+
}
381+
}
382+
383+
private:
384+
385+
std::experimental::coroutine_handle<promise_type> m_coroutine;
386+
};
387+
388+
return starter{ m_coroutine };
389+
}
390+
359391
private:
360392

361393
std::experimental::coroutine_handle<promise_type> m_coroutine;

‎include/cppcoro/shared_lazy_task.hpp‎

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -446,6 +446,37 @@ namespace cppcoro
446446
return awaitable{ m_coroutine };
447447
}
448448

449+
auto get_starter() const noexcept
450+
{
451+
struct starter
452+
{
453+
public:
454+
455+
explicit starter(std::experimental::coroutine_handle<promise_type> coroutine)
456+
: m_coroutine(coroutine)
457+
{}
458+
459+
void start(detail::continuation c) noexcept
460+
{
461+
m_waiter.m_continuation = c;
462+
463+
if (!m_coroutine ||
464+
m_coroutine.promise().is_ready() ||
465+
!m_coroutine.promise().try_await(&m_waiter, m_coroutine))
466+
{
467+
// Task completed synchronously, resume immediately.
468+
c.resume();
469+
}
470+
}
471+
472+
private:
473+
std::experimental::coroutine_handle<promise_type> m_coroutine;
474+
detail::shared_lazy_task_waiter m_waiter;
475+
};
476+
477+
return starter{ m_coroutine };
478+
}
479+
449480
private:
450481

451482
template<typename U>

‎include/cppcoro/shared_task.hpp‎

Lines changed: 28 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -409,6 +409,34 @@ if (m_coroutine)
409409
return awaitable{ m_coroutine };
410410
}
411411

412+
auto get_starter() const noexcept
413+
{
414+
struct starter
415+
{
416+
public:
417+
418+
explicit starter(std::experimental::coroutine_handle<promise_type> coroutine)
419+
: m_coroutine(coroutine)
420+
{}
421+
422+
void start(detail::continuation c) noexcept
423+
{
424+
m_waiter.m_continuation = c;
425+
if (!m_coroutine || m_coroutine.promise().is_ready() || !m_coroutine.promise().try_await(&m_waiter))
426+
{
427+
// Task completed synchronously, resume immediately.
428+
c.resume();
429+
}
430+
}
431+
432+
private:
433+
std::experimental::coroutine_handle<promise_type> m_coroutine;
434+
detail::shared_task_waiter m_waiter;
435+
};
436+
437+
return starter{ m_coroutine };
438+
}
439+
412440
private:
413441

414442
template<typename U>

‎include/cppcoro/task.hpp‎

Lines changed: 28 additions & 0 deletions

0 commit comments

Comments
 (0)