15811413062

Committed 22 Jun 2025 10:26PM UTC coverage: 88.141%. First build

Build # 15811413062

Build Type

Pull #354

github

Committed by

web-flow

Commit Message

Merge e0efc064b into e87eb5a59

Pull Request Pull Request #354: task_container: add back task_container

Run Details

18 of 20 new or added lines in 1 file covered. (90.0%)

1650 of 1872 relevant lines covered (88.14%)

13348820.88 hits per line

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

90.0

/include/coro/task_container.hpp

#pragma once

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

#include <atomic>
#include <iostream>
#include <list>
#include <memory>
#include <mutex>
#include <queue>
#include <thread>
#include <vector>

namespace coro
{
class io_scheduler;

template<concepts::executor executor_type>
class task_container
{
public:
    /**
     * @param e Tasks started in the container are scheduled onto this executor.  For tasks created
     *           from a coro::io_scheduler, this would usually be that coro::io_scheduler instance.
     * @param opts Task container options.
     */
    explicit task_container(std::shared_ptr<executor_type> e) : m_executor(std::move(e))
    {
        if (m_executor == nullptr)
        {
            throw std::runtime_error{"task_container cannot have a nullptr executor"};
        }
    }
    task_container(const task_container&)                    = delete;
    task_container(task_container&&)                         = delete;
    auto operator=(const task_container&) -> task_container& = delete;
    auto operator=(task_container&&) -> task_container&      = delete;
    ~task_container()
    {
        // This will hang the current thread.. but if tasks are not complete thats also pretty bad.
        while (!empty())
        {
            // Sleep a bit so the cpu doesn't totally churn.
            std::this_thread::sleep_for(std::chrono::milliseconds{10});
        }
    }

    /**
     * Stores a user task and starts its execution on the container's thread pool.
     * @param user_task The scheduled user's task to store in this task container and start its execution.
     * @return True if the task was succesfully started into the task container. This can fail if the task
     *         is already completed or does not contain a valid coroutine anymore.
     */
    auto start(coro::task<void>&& user_task) -> bool
    {
        m_size.fetch_add(1, std::memory_order::relaxed);

        auto task = make_self_deleting_task(std::move(user_task));
        // Hook the promise to decrement the size upon its self deletion of the coroutine frame.
        task.promise().executor_size(m_size);
        return m_executor->resume(task.handle());
    }

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

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

    /**
     * Will continue to garbage collect and yield until all tasks are complete.  This method can be
     * co_await'ed to make it easier to wait for the task container to have all its tasks complete.
     *
     * This does not shut down the task container, but can be used when shutting down, or if your
     * logic requires all the tasks contained within to complete, it is similar to coro::latch.
     */
    auto yield_until_empty() -> coro::task<void>
    {
        while (!empty())
        {
            co_await m_executor->yield();
        }
    }

private:
    auto make_self_deleting_task(task<void> user_task) -> detail::task_self_deleting
    {
        co_await user_task;
        co_return;
    }

    /// The number of alive tasks.
    std::atomic<std::size_t> m_size{};
    /// The executor to schedule tasks that have just started.
    std::shared_ptr<executor_type> m_executor{nullptr};
};

} // namespace coro

1	#pragma once
2
3	#include "coro/attribute.hpp"
4	#include "coro/concepts/executor.hpp"
5	#include "coro/detail/task_self_deleting.hpp"
6	#include "coro/task.hpp"
7
8	#include <atomic>
9	#include <iostream>
10	#include <list>
11	#include <memory>
12	#include <mutex>
13	#include <queue>
14	#include <thread>
15	#include <vector>
16
17	namespace coro
18	{
19	class io_scheduler;
20
21	template<concepts::executor executor_type>
22	class task_container
23	{
24	public:
25	/**
26	* @param e Tasks started in the container are scheduled onto this executor. For tasks created
27	* from a coro::io_scheduler, this would usually be that coro::io_scheduler instance.
28	* @param opts Task container options.
29	*/
30	explicit task_container(std::shared_ptr<executor_type> e) : m_executor(std::move(e))	2✔
31	{
32	if (m_executor == nullptr)	2✔
33	{
NEW 34	throw std::runtime_error{"task_container cannot have a nullptr executor"};	×
35	}
36	}	2✔
37	task_container(const task_container&) = delete;
38	task_container(task_container&&) = delete;
39	auto operator=(const task_container&) -> task_container& = delete;
40	auto operator=(task_container&&) -> task_container& = delete;
41	~task_container()	2✔
42	{
43	// This will hang the current thread.. but if tasks are not complete thats also pretty bad.
44	while (!empty())	2✔
45	{
46	// Sleep a bit so the cpu doesn't totally churn.
NEW 47	std::this_thread::sleep_for(std::chrono::milliseconds{10});	×
48	}
49	}	2✔
50
51	/**
52	* Stores a user task and starts its execution on the container's thread pool.
53	* @param user_task The scheduled user's task to store in this task container and start its execution.
54	* @return True if the task was succesfully started into the task container. This can fail if the task
55	* is already completed or does not contain a valid coroutine anymore.
56	*/
57	auto start(coro::task<void>&& user_task) -> bool	1,001✔
58	{
59	m_size.fetch_add(1, std::memory_order::relaxed);	1,001✔
60
61	auto task = make_self_deleting_task(std::move(user_task));	1,001✔
62	// Hook the promise to decrement the size upon its self deletion of the coroutine frame.
63	task.promise().executor_size(m_size);	1,001✔
64	return m_executor->resume(task.handle());	2,002✔
65	}	1,001✔
66
67	/**
68	* @return The number of active tasks in the container.
69	*/
70	auto size() const -> std::size_t { return m_size.load(std::memory_order::acquire); }	18✔
71
72	/**
73	* @return True if there are no active tasks in the container.
74	*/
75	auto empty() const -> bool { return size() == 0; }	8✔
76
77	/**
78	* Will continue to garbage collect and yield until all tasks are complete. This method can be
79	* co_await'ed to make it easier to wait for the task container to have all its tasks complete.
80	*
81	* This does not shut down the task container, but can be used when shutting down, or if your
82	* logic requires all the tasks contained within to complete, it is similar to coro::latch.
83	*/
84	auto yield_until_empty() -> coro::task<void>	2✔
85	{
86	while (!empty())
87	{
88	co_await m_executor->yield();
89	}
90	}	4✔
91
92	private:
93	auto make_self_deleting_task(task<void> user_task) -> detail::task_self_deleting	1,001✔
94	{
95	co_await user_task;
96	co_return;
97	}	2,002✔
98
99	/// The number of alive tasks.
100	std::atomic<std::size_t> m_size{};
101	/// The executor to schedule tasks that have just started.
102	std::shared_ptr<executor_type> m_executor{nullptr};
103	};
104
105	} // namespace coro

jbaldwin / libcoro / 15811413062

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