13041658666

Committed 29 Jan 2025 10:28PM UTC coverage: 84.577%. First build

Build # 13041658666

Build Type

Pull #288

github

Committed by

web-flow

Commit Message

Merge f299a9da1 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%)

1371 of 1621 relevant lines covered (84.58%)

4304824.16 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_deleting.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.
     * @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().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_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 <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	* @return True if the task was succesfully started into the task container. This can fail if the task
58	* is already completed or does not contain a valid coroutine anymore.
59	*/
60	auto start(coro::task<void>&& user_task) -> bool	400,504✔
61	{
62	m_size.fetch_add(1, std::memory_order::relaxed);	400,504✔
63
64	auto task = make_self_deleting_task(std::move(user_task));	400,504✔
65	// Hook the promise to decrement the size upon its self deletion of the coroutine frame.
66	task.promise().task_container_size(m_size);	400,504✔
67	return m_executor->resume(task.handle());	801,008✔
68	}	400,504✔
69
70	/**
71	* @return The number of active tasks in the container.
72	*/
73	auto size() const -> std::size_t { return m_size.load(std::memory_order::acquire); }	4✔
74
75	/**
76	* @return True if there are no active tasks in the container.
77	*/
78	auto empty() const -> bool { return size() == 0; }	2✔
79
80	/**
81	* Will continue to garbage collect and yield until all tasks are complete. This method can be
82	* co_await'ed to make it easier to wait for the task container to have all its tasks complete.
83	*
84	* This does not shut down the task container, but can be used when shutting down, or if your
85	* logic requires all the tasks contained within to complete, it is similar to coro::latch.
86	*/
87	auto yield_until_empty() -> coro::task<void>
88	{
89	while (!empty())
90	{
91	co_await m_executor->yield();
92	}
93	}
94
95	private:
96	auto make_self_deleting_task(task<void> user_task) -> detail::task_self_deleting	400,504✔
97	{
98	co_await user_task;
99	co_return;
100	}	801,008✔
101
102	/// The number of alive tasks.
103	std::atomic<std::size_t> m_size{};
104	/// The executor to schedule tasks that have just started.
105	std::shared_ptr<executor_type> m_executor{nullptr};
106	};
107
108	} // namespace coro

jbaldwin / libcoro / 13041658666

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