19916794195

Committed 04 Dec 2025 03:36AM UTC coverage: 86.289%. First build

Build # 19916794195

Build Type

Pull #425

github

Committed by

web-flow

Commit Message

Merge 55932264e into f671edb4c

Pull Request Pull Request #425: task_group::start(task) reintroduce

Run Details

20 of 22 new or added lines in 1 file covered. (90.91%)

1674 of 1940 relevant lines covered (86.29%)

5139599.76 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

87.1

/include/coro/task_group.hpp

#pragma once

#include "coro/concepts/executor.hpp"
#include "coro/concepts/range_of.hpp"
#include "coro/detail/task_self_deleting.hpp"
#include "coro/event.hpp"
#include "coro/task.hpp"

#include <atomic>
#include <memory>
#include <thread>

namespace coro
{
class io_scheduler;

template<concepts::executor executor_type>
class task_group
{
public:
    /**
     * Creates a task group, use the start method to schedule tasks into this group.
     * @param executor Tasks started in the group are scheduled onto this executor.
     */
    explicit task_group(std::unique_ptr<executor_type>& executor) : m_executor(executor.get())
    {
        if (executor == nullptr)
        {
            throw std::runtime_error{"task_group cannot have a nullptr executor"};
        }
    }

    /**
     * Creates a task group with a single task to start oon the executor.
     * @param executor Tasks started in the group are scheduled onto this executor.
     * @param task The first task to start in the group.
     */
    explicit task_group(std::unique_ptr<executor_type>& executor, coro::task<void>&& task) : task_group(executor)
    {
        (void)start(std::forward<coro::task<void>>(task));
    }

    /**
     * Creates a task group and starts the given range of tasks on the executor.
     * @tparam range_type The range type.
     * @param executor Tasks started in the group are scheduled onto this executor.
     * @param tasks The group of tasks to track.
     */
    template<coro::concepts::range_of<coro::task<void>> range_type>
    explicit task_group(std::unique_ptr<executor_type>& executor, range_type tasks) : task_group(executor)
    {
        for (auto& task : tasks)
        {
            (void)start(std::move(task));
        }
    }
    task_group(const task_group&)                    = delete;
    task_group(task_group&&)                         = delete;
    auto operator=(const task_group&) -> task_group& = delete;
    auto operator=(task_group&&) -> task_group&      = delete;

    /**
     * When the task group destructs it waits for all of its tasks to complete. It is advisable to
     * use `co_await task_group` to efficiently wait for all tasks to complete as
     * destructors cannot be co_await'ed this will hang the calling thread until they are done.
     */
    ~task_group()
    {
        while (!empty())
        {
            // Sleep a bit so the cpu doesn't totally churn.
            std::this_thread::sleep_for(std::chrono::milliseconds{10});
        }
    }

    /**
     * Starts a task into this group.
     * @param task The task to include into this group.
     * @return True if the task was started, this will only return false if the executor
     *         has been shutdown.
     */
    [[nodiscard]] auto start(coro::task<void>&& task) -> bool
    {
        // Make sure the event isn't triggered, or can trigger again.
        m_on_empty_event.reset();
        // We're about to start another task.
        m_size.fetch_add(1, std::memory_order::release);
        auto wrapper_task = detail::make_task_self_deleting(std::move(task));
        wrapper_task.promise().user_final_suspend([this]() -> void { count_down(); });

        // Kick it.
        if (!m_executor->resume(wrapper_task.handle()))
        {
            count_down();
            return false;
        }
        return true;
    }

    /**
     * @return The number of active tasks in the group.
     */
    [[nodiscard]] auto size() const -> std::size_t { return m_size.load(std::memory_order::acquire); }

    /**
     * @return True if there are no active tasks in the group.
     */
    [[nodiscard]] auto empty() const -> bool { return size() == 0; }

    /**
     * Wait until the task group's tasks are completed (zero tasks remain).
     * NOTE: that this can happen multiple times if you are using start(task) as its possible
     * for the task count to drop to zero and then increase again. It is advisable to not
     * await this method on the task group until you know all tasks in the group have started.
     */
    auto operator co_await() const noexcept -> event::awaiter { return m_on_empty_event.operator co_await(); }

private:
    executor_type* m_executor{nullptr};
    /// The number of alive tasks.
    std::atomic<uint64_t> m_size{};
    /// Event to trigger if m_size goes to zero.
    coro::event m_on_empty_event{};

    auto count_down() -> void
    {
        if (m_size.fetch_sub(1, std::memory_order::acq_rel) == 1)
        {
            m_on_empty_event.set();
        }
    }
};

} // namespace coro

1	#pragma once
2
3	#include "coro/concepts/executor.hpp"
4	#include "coro/concepts/range_of.hpp"
5	#include "coro/detail/task_self_deleting.hpp"
6	#include "coro/event.hpp"
7	#include "coro/task.hpp"
8
9	#include <atomic>
10	#include <memory>
11	#include <thread>
12
13	namespace coro
14	{
15	class io_scheduler;
16
17	template<concepts::executor executor_type>
18	class task_group
19	{
20	public:
21	/**
22	* Creates a task group, use the start method to schedule tasks into this group.
23	* @param executor Tasks started in the group are scheduled onto this executor.
24	*/
25	explicit task_group(std::unique_ptr<executor_type>& executor) : m_executor(executor.get())	4✔
26	{
27	if (executor == nullptr)	4✔
28	{
29	throw std::runtime_error{"task_group cannot have a nullptr executor"};	×
30	}
31	}	4✔
32
33	/**
34	* Creates a task group with a single task to start oon the executor.
35	* @param executor Tasks started in the group are scheduled onto this executor.
36	* @param task The first task to start in the group.
37	*/
38	explicit task_group(std::unique_ptr<executor_type>& executor, coro::task<void>&& task) : task_group(executor)	2✔
39	{
40	(void)start(std::forward<coro::task<void>>(task));	2✔
41	}	2✔
42
43	/**
44	* Creates a task group and starts the given range of tasks on the executor.
45	* @tparam range_type The range type.
46	* @param executor Tasks started in the group are scheduled onto this executor.
47	* @param tasks The group of tasks to track.
48	*/
49	template<coro::concepts::range_of<coro::task<void>> range_type>
50	explicit task_group(std::unique_ptr<executor_type>& executor, range_type tasks) : task_group(executor)	1✔
51	{
52	for (auto& task : tasks)	1,001✔
53	{
54	(void)start(std::move(task));	1,000✔
55	}
56	}	1✔
57	task_group(const task_group&) = delete;
58	task_group(task_group&&) = delete;
59	auto operator=(const task_group&) -> task_group& = delete;
60	auto operator=(task_group&&) -> task_group& = delete;
61
62	/**
63	* When the task group destructs it waits for all of its tasks to complete. It is advisable to
64	* use `co_await task_group` to efficiently wait for all tasks to complete as
65	* destructors cannot be co_await'ed this will hang the calling thread until they are done.
66	*/
67	~task_group()	4✔
68	{
69	while (!empty())	4✔
70	{
71	// Sleep a bit so the cpu doesn't totally churn.
72	std::this_thread::sleep_for(std::chrono::milliseconds{10});	×
73	}
74	}	4✔
75
76	/**
77	* Starts a task into this group.
78	* @param task The task to include into this group.
79	* @return True if the task was started, this will only return false if the executor
80	* has been shutdown.
81	*/
82	[[nodiscard]] auto start(coro::task<void>&& task) -> bool	2,003✔
83	{
84	// Make sure the event isn't triggered, or can trigger again.
85	m_on_empty_event.reset();	2,003✔
86	// We're about to start another task.
87	m_size.fetch_add(1, std::memory_order::release);	2,003✔
88	auto wrapper_task = detail::make_task_self_deleting(std::move(task));	2,003✔
89	wrapper_task.promise().user_final_suspend([this]() -> void { count_down(); });	4,006✔
90
91	// Kick it.
92	if (!m_executor->resume(wrapper_task.handle()))	2,003✔
93	{
NEW 94	count_down();	×
NEW 95	return false;	×
96	}
97	return true;	2,003✔
98	}
99
100	/**
101	* @return The number of active tasks in the group.
102	*/
103	[[nodiscard]] auto size() const -> std::size_t { return m_size.load(std::memory_order::acquire); }	24✔
104
105	/**
106	* @return True if there are no active tasks in the group.
107	*/
108	[[nodiscard]] auto empty() const -> bool { return size() == 0; }	9✔
109
110	/**
111	* Wait until the task group's tasks are completed (zero tasks remain).
112	* NOTE: that this can happen multiple times if you are using start(task) as its possible
113	* for the task count to drop to zero and then increase again. It is advisable to not
114	* await this method on the task group until you know all tasks in the group have started.
115	*/
116	auto operator co_await() const noexcept -> event::awaiter { return m_on_empty_event.operator co_await(); }	5✔
117
118	private:
119	executor_type* m_executor{nullptr};
120	/// The number of alive tasks.
121	std::atomic<uint64_t> m_size{};
122	/// Event to trigger if m_size goes to zero.
123	coro::event m_on_empty_event{};
124
125	auto count_down() -> void	2,003✔
126	{
127	if (m_size.fetch_sub(1, std::memory_order::acq_rel) == 1)	4,006✔
128	{
129	m_on_empty_event.set();	19✔
130	}
131	}	2,003✔
132	};
133
134	} // namespace coro

jbaldwin / libcoro / 19916794195

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous