Frame Allocators

std::pmr::monotonic_buffer_resource resource;
std::pmr::polymorphic_allocator<std::byte> alloc(&resource);

run_async(executor, alloc)(my_task());

std::pmr::monotonic_buffer_resource resource;
run_async(executor, &resource)(my_task());

// A memory resource that pools freed blocks by size. When a coroutine
// frame is freed, its block is kept; the next frame of the same size
// reuses it instead of allocating again -- the strategy that makes
// recycling_memory_resource fast. The two counters let the test observe
// upstream allocations versus reuse.
struct pooling_resource : std::pmr::memory_resource
{
    std::size_t upstream = 0;   // blocks taken from the heap
    std::size_t reused = 0;     // blocks served from the freelist

    void*
    do_allocate(std::size_t bytes, std::size_t) override
    {
        auto& blocks = pool_[bytes];
        if(! blocks.empty())
        {
            ++reused;
            void* p = blocks.back();
            blocks.pop_back();
            return p;
        }
        ++upstream;
        return ::operator new(bytes);
    }

    void
    do_deallocate(void* p, std::size_t bytes, std::size_t) override
    {
        pool_[bytes].push_back(p);   // keep the block for the next frame
    }

    bool
    do_is_equal(memory_resource const& other) const noexcept override
    {
        return this == &other;
    }

    ~pooling_resource() override
    {
        for(auto& [bytes, blocks] : pool_)
            for(void* p : blocks)
                ::operator delete(p);
    }

    std::unordered_map<std::size_t, std::vector<void*>> pool_;
};

// Run the same task repeatedly through one pooling resource. The
// first run has an empty pool, so its frames come from upstream.
// Once the task completes, its frames go back into the pool, so
// every later run reuses a freed block of the right size.
pooling_resource pooling;

auto run_once = [&]
{
    thread_pool pool(1);
    run_async(pool.get_executor(), &pooling)(my_task());
    pool.join();   // task done: its frames are back in the pool
};

run_once();                                     // cold: fills pool
std::size_t const upstream_when_warm = pooling.upstream;

for(int i = 0; i < 7; ++i)
    run_once();                                 // warm: reuses pool

// HALO can apply: task is awaited immediately
int result = co_await compute();

// HALO cannot apply: task escapes to storage
auto t = compute();
tasks.push_back(std::move(t));

void process_batch(std::vector<item> const& items)
{
    std::array<std::byte, 64 * 1024> buffer;
    std::pmr::monotonic_buffer_resource resource(
        buffer.data(), buffer.size());

    for (auto const& item : items)
    {
        run_async(executor, &resource)(process(item));
    }
    // All frames deallocated when resource goes out of scope
}

// BAD: buf lives in frame across all subsequent co_awaits
task<> process(stream& s)
{
    char buf[4096];
    auto [ec, n] = co_await s.read_some(buf);
    co_await do_work(buf, n);
    co_await s.write_some(reply);   // buf wastes 4K in frame
}

// GOOD: braces end buf's lifetime before next suspend
task<> process(stream& s)
{
    std::size_t n;
    {
        char buf[4096];
        auto [ec, n_] = co_await s.read_some(buf);
        n = n_;
        co_await do_work(buf, n);
    }
    co_await s.write_some(reply);  // 4K saved
}

// BAD: both arrays in frame simultaneously (8K)
task<> pipeline(stream& in, stream& out)
{
    char read_buf[4096];
    auto [ec1, n] = co_await in.read_some(read_buf);

    char write_buf[4096];
    prepare(write_buf, read_buf, n);
    co_await out.write_some(write_buf);
}

// GOOD: non-overlapping scopes allow frame reuse (4K)
task<> pipeline(stream& in, stream& out)
{
    std::size_t n;
    {
        char read_buf[4096];
        auto [ec, n_] = co_await in.read_some(read_buf);
        n = n_;
    }
    {
        char write_buf[4096];
        prepare(write_buf, n);
        co_await out.write_some(write_buf);
    }
}