Crafter.Thread/tests/EnqueueBatch/main.cpp

//SPDX-License-Identifier: LGPL-3.0-only
//SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts®

// ThreadPool::Enqueue(vector) runs every task exactly once, whatever the
// task count. When the count didn't divide by the number of worker threads,
// the last worker got only the remainder and tasksPerThread tasks were
// dropped. The counts here cover fewer tasks than workers, an exact multiple,
// one more and one less than a multiple, and a large uneven batch, then a
// batch of single Enqueue calls, which rely on the per-thread task counts the
// batch path also kept wrong.

#include <cstdlib>
import Crafter.Thread;
import std;
using namespace Crafter;

namespace {

int failures = 0;

void Check(bool ok, std::string_view what) {
    std::println("{}  {}", ok ? "PASS" : "FAIL", what);
    if (!ok) ++failures;
}

// Waits until `done` reaches `expected` or a few seconds pass.
bool WaitFor(const std::atomic<std::uint32_t>& done, std::uint32_t expected) {
    const auto deadline = std::chrono::steady_clock::now() + std::chrono::seconds(5);
    while (done.load() < expected) {
        if (std::chrono::steady_clock::now() > deadline) return false;
        std::this_thread::sleep_for(std::chrono::milliseconds(1));
    }
    return true;
}

void RunBatch(std::uint32_t count) {
    std::atomic<std::uint32_t> done = 0;
    std::vector<std::atomic<std::uint32_t>> runs(count);
    std::vector<std::function<void(void)>> tasks;
    tasks.reserve(count);
    for (std::uint32_t i = 0; i < count; ++i) {
        tasks.emplace_back([&runs, &done, i] {
            runs[i]++;
            done++;
        });
    }
    ThreadPool::Enqueue(std::move(tasks));
    const bool finished = WaitFor(done, count);
    // A little longer, in case something runs twice.
    std::this_thread::sleep_for(std::chrono::milliseconds(20));
    const bool once = std::ranges::all_of(runs, [](const std::atomic<std::uint32_t>& r) { return r.load() == 1; });
    Check(finished && once, std::format("a batch of {} tasks runs every task exactly once ({} of {} done)", count, done.load(), count));
}

} // namespace

int main() {
    ThreadPool::Start(4);
    const std::uint32_t workers = ThreadPool::activeThreads;
    std::println("      {} worker threads", workers);

    for (std::uint32_t count : {1u, workers - 1, workers, workers + 1, 3 * workers + 2, 1000u}) {
        RunBatch(count);
    }

    // Single Enqueue calls pick the least loaded worker by its task count;
    // a count left wrong by a batch would starve or flood one of them.
    std::atomic<std::uint32_t> done = 0;
    for (std::uint32_t i = 0; i < 200; ++i) {
        ThreadPool::Enqueue([&done] { done++; });
    }
    Check(WaitFor(done, 200), std::format("200 single tasks after the batches all run ({} done)", done.load()));

    ThreadPool::Stop();
    std::println("{}", failures == 0 ? "all checks passed" : "check(s) failed");
    return failures == 0 ? EXIT_SUCCESS : EXIT_FAILURE;
}