13039341916

Committed 29 Jan 2025 07:56PM UTC coverage: 84.402%. First build

Build # 13039341916

Build Type

Pull #288

github

Committed by

web-flow

Commit Message

Merge 0ba7020e5 into 26de94ded

Pull Request Pull Request #288: coro::task_container gc fix not completing coroutines

Run Details

41 of 59 new or added lines in 6 files covered. (69.49%)

1369 of 1622 relevant lines covered (84.4%)

4302692.69 hits per line

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

89.47

/include/coro/task_container.hpp

#pragma once

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

#include <atomic>
#include <iostream>
#include <list>
#include <memory>
#include <mutex>
#include <thread>
#include <queue>
#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.
     */
    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.
     */
    auto start(coro::task<void>&& user_task) -> bool
    {
        m_size.fetch_add(1, std::memory_order::relaxed);

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

jbaldwin / libcoro / 13039341916

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