structured_patterns_test.cpp

#include <gtest/gtest.h>

import std;
import mcpplibs.cmp;

namespace {

using namespace mcpplibs::cmp;
using Scheduler = RunLoop::Scheduler;

Task<void> scoped_child(
    Scheduler scheduler,
    std::stop_token token,
    bool cooperative,
    bool fail,
    int& completed,
    int& cancelled) {
    try {
        if (cooperative) {
            co_await scheduler.schedule(token);
        } else {
            co_await scheduler.schedule();
        }
    } catch (const OperationCancelled&) {
        ++cancelled;
    }

    ++completed;

    if (fail) {
        throw std::runtime_error { "child failure" };
    }
}

Task<void> exception_safe_body(
    Scheduler scheduler,
    int failAfter,
    bool cooperative,
    bool failChild,
    int& completed,
    int& cancelled) {
    TaskGroup group {};
    // 在接纳前准备 join 帧，body 的分配异常不会阻止后续收敛
    auto joining = group.join();
    std::exception_ptr bodyError {};
    std::exception_ptr childError {};

    try {
        for (int index {}; index < 3; ++index) {
            group.spawn(scoped_child(
                scheduler, group.get_stop_token(), cooperative,
                failChild && index == 0, completed, cancelled));

            if (index + 1 == failAfter) {
                throw std::runtime_error { "body failure" };
            }
        }
    } catch (...) {
        bodyError = std::current_exception();
        group.request_stop();
    }

    try {
        co_await std::move(joining);
    } catch (...) {
        childError = std::current_exception();
    }

    // body 的错误优先，child 的错误仍在全部结束之后交付
    if (bodyError) {
        std::rethrow_exception(bodyError);
    }

    if (childError) {
        std::rethrow_exception(childError);
    }
}

TEST(CmpStructuredPatternsTest, BodyFailureDrainsAtEveryAdmissionBoundary) {
    for (const bool cooperative : { false, true }) {
        for (int failAfter { 1 }; failAfter <= 3; ++failAfter) {
            RunLoop loop {};
            int completed {};
            int cancelled {};

            try {
                loop.run(exception_safe_body(
                    loop.get_scheduler(), failAfter, cooperative,
                    true, completed, cancelled));
                FAIL() << "body failure must propagate";
            } catch (const std::runtime_error& error) {
                EXPECT_STREQ(error.what(), "body failure");
            }

            EXPECT_EQ(completed, failAfter);
            EXPECT_EQ(cancelled, cooperative ? failAfter : 0);
        }
    }
}

TEST(CmpStructuredPatternsTest, ChildFailureIsDeliveredAfterAllChildrenFinish) {
    RunLoop loop {};
    int completed {};
    int cancelled {};

    try {
        loop.run(exception_safe_body(
            loop.get_scheduler(), 0, false, true, completed, cancelled));
        FAIL() << "child failure must propagate";
    } catch (const std::runtime_error& error) {
        EXPECT_STREQ(error.what(), "child failure");
    }

    EXPECT_EQ(completed, 3);
    EXPECT_EQ(cancelled, 0);
}

Task<void> budgeted_child(
    Scheduler scheduler,
    int& active,
    int& peak,
    int& completed) {
    ++active;
    peak = std::max(peak, active);
    co_await scheduler.schedule_after(std::chrono::milliseconds(1));
    --active;
    ++completed;
}

Task<void> process_in_batches(Scheduler scheduler, int& peak, int& completed) {
    int active {};
    constexpr int BUDGET { 32 };
    constexpr int TOTAL { 1'024 };

    for (int base {}; base < TOTAL; base += BUDGET) {
        TaskGroup group {};
        auto joining = group.join();
        std::exception_ptr bodyError {};

        try {
            for (int index {}; index < std::min(BUDGET, TOTAL - base); ++index) {
                group.spawn(budgeted_child(scheduler, active, peak, completed));
            }
        } catch (...) {
            bodyError = std::current_exception();
            group.request_stop();
        }

        co_await std::move(joining);

        if (bodyError) {
            std::rethrow_exception(bodyError);
        }
    }
}

TEST(CmpStructuredPatternsTest, BatchScopesBoundAdmissionAndReleaseEachBatch) {
    RunLoop loop {};
    int peak {};
    int completed {};
    loop.run(process_in_batches(loop.get_scheduler(), peak, completed));
    EXPECT_EQ(peak, 32);
    EXPECT_EQ(completed, 1'024);
}

Task<void> occupy_blocking_worker(
    ThreadPool::Scheduler blocking,
    Scheduler caller,
    std::binary_semaphore& entered,
    std::binary_semaphore& release) {
    co_await run_blocking(blocking, caller, [&] {
        entered.release();
        release.acquire();
    });
}

Task<void> independent_cpu_progress(
    ThreadPool::Scheduler cpu,
    Scheduler caller,
    std::binary_semaphore& entered,
    std::binary_semaphore& release,
    bool& progressed) {
    co_await cpu.schedule();
    const auto blocked = entered.try_acquire_for(std::chrono::seconds(5));
    co_await caller.schedule();
    progressed = blocked;
    release.release();
}

Task<void> isolated_pools(
    ThreadPool::Scheduler cpu,
    ThreadPool::Scheduler blocking,
    Scheduler caller,
    bool& progressed) {
    std::binary_semaphore entered { 0 };
    std::binary_semaphore release { 0 };
    co_await when_all(
        occupy_blocking_worker(blocking, caller, entered, release),
        independent_cpu_progress(cpu, caller, entered, release, progressed));
}

TEST(CmpStructuredPatternsTest, BlockingPoolSaturationKeepsCpuAndRunLoopProgress) {
    ThreadPool cpu { 1 };
    ThreadPool blocking { 1 };
    RunLoop loop {};
    bool progressed {};
    loop.run(isolated_pools(
        cpu.get_scheduler(), blocking.get_scheduler(), loop.get_scheduler(), progressed));
    EXPECT_TRUE(progressed);
}

}  // namespace