Fix use after free in coro::task_container

[ripose-jp]

In coro::task_container, this is potential for a use after free bug to occur between the final co_return of make_cleanup_task and another thread executing gc_internal. This can happen because there is a delay between the std::scoped_lock being destroyed upon the co_return of make_cleanup_task and its final suspend. This leads to a race condition where another thread can acquire the lock and initiate a garbage collection, where that cleanup task is destroyed before it's final suspend. This leads to a use after free.

This commit fixes the issue by reworking how tasks get garbage collected. Instead of unconditionally destroying tasks scheduled for deletion, the gc_internal calls is_ready on the coro::task first. This function will only be true in the cases where the task has entered final suspend, has been default constructed, or has been destroyed. This means a task can get scheduled for deletion, but not be deleted when gc_internal is called. This allows time for the task to enter final suspend, which avoids the use after free bug.

To make this implementation easier, several aspects of the implementation were changed. m_task_indexes was removed in favor of a queue which keeps track of free indices into m_tasks. This makes things easier since it removes the hard constraint of keeping used indices behind m_free_pos and free indices at or after. m_tasks_to_delete was then changed to a list in order to allow for constant time deletion of elements in place. This is helpful because it is possible for tasks to be queue for deletion, but not yet safe to delete. This means unconditionally clearing the m_tasks_to_delete data structure is no longer viable.

---

Here's some example code that can cause the use-after-free bug. Since it's a race condition, it's non-deterministic so it may take some time before the bug occurs.

#include <coro/coro.hpp>

#include <cstdint>
#include <iostream>
#include <numeric>
#include <vector>

static constexpr size_t DUMMY_TASK_COUNT{10};

static coro::task<void> do_dummy_work()
{
    constexpr size_t SIZE_DOUBLE_WORDS{100 * 1024 * 1024 / sizeof(uint64_t)};

    [[maybe_unused]] std::vector<uint64_t> buffer;
    buffer.resize(SIZE_DOUBLE_WORDS);
    std::iota(std::begin(buffer), std::end(buffer), 0);
    co_return;
}

static coro::task<void> run_async(
    std::shared_ptr<coro::thread_pool> &tp,
    coro::task_container<coro::thread_pool> &tc)
{
    co_await tp->schedule();

    while (true)
    {
        for (size_t i{0}; i < DUMMY_TASK_COUNT; ++i)
        {
            tc.start(do_dummy_work());
        }
        co_await tc.garbage_collect_and_yield_until_empty();
        std::cout << "Dummy work iteration complete\n";
    }
}

int main(void)
{
    std::shared_ptr<coro::thread_pool> tp = std::make_shared<coro::thread_pool>(
        coro::thread_pool::options{ .thread_count = DUMMY_TASK_COUNT, }
    );
    coro::task_container tc{tp};
    coro::sync_wait(run_async(tp, tc));
    return 0;
}

[jbaldwin]

I'll take a deep dive into this as soon as I can, the description of the use after free makes sense, and thank you for pushing a detailed PR with a solution to the problem.