// Copyright (c) 2026 Progmasoft.
// SPDX-License-Identifier: MPL-2.0 WITH AdditionRef-Progmasoft-Exception-2.0
#include <algorithm>
#include <chrono>
#include <condition_variable>
#include <cstdio>
#include <deque>
#include <filesystem>
#include <map>
#include <memory>
#include <mutex>
#include <optional>
#include <set>
#include <thread>
#include "Execution.hpp"
#include "Files.hpp"
#include "Process.hpp"
#include "Protocol.hpp"
#include "ResultsJson.hpp"
#include "Scheduler.hpp"
#include "SchedulerModel.hpp"
namespace Progmasoft::Catch3::Runner
{
namespace
{
using Scheduling::AppendBounded;
using Scheduling::Bisection;
using Scheduling::Clock;
using Scheduling::Event;
using Scheduling::Job;
using Scheduling::kMaximumProbes;
using Scheduling::kMaximumProcessOutput;
using Scheduling::kMaximumRecordedAssertions;
using Scheduling::kStartupFailuresToGiveUp;
using Scheduling::kStartupTimeout;
using Scheduling::Microseconds;
using Scheduling::Passing;
using Scheduling::Running;
using Scheduling::Satisfies;
using Scheduling::TestState;
using Scheduling::Worker;
using std::chrono::milliseconds;
class Scheduler final
{
public:
Scheduler(const Options &options,
const Plan &plan,
const Catalog &catalog,
std::string program,
std::string programName,
const SchedulerEvents &events)
: options_{ options }
, plan_{ plan }
, catalog_{ catalog }
, program_{ std::move(program) }
, programName_{ std::move(programName) }
, events_{ events }
, runStart_{ Clock::now() }
{
std::error_code error;
temporaryRoot_ = std::filesystem::temp_directory_path(error)
/ ("catch3-" + UniqueName());
const unsigned hardware = HardwareThreads();
maximumWorkers_ = options.jobs == 0 ? hardware : options.jobs;
maximumWorkers_ = std::max<std::size_t>(
1,
std::min(maximumWorkers_, plan.tests.size()));
retryBudget_ = std::max<std::size_t>(5, plan.tests.size() / 10);
states_.resize(plan.tests.size());
for (std::size_t position = 0; position < plan.tests.size();
++position)
{
const PlannedTest &planned = plan.tests[position];
const TestEntry &entry = *planned.entry;
TestResult &result = states_[position].result;
result.id = entry.key;
result.name = entry.name;
result.className = entry.className;
result.file = entry.file;
result.line = entry.line;
result.tags = entry.tags;
result.selectionReason = planned.selectionReason;
result.quarantined = planned.quarantined;
result.predictedDurationMicroseconds
= planned.predictedMicroseconds;
result.predictedFailureProbability
= planned.failureProbability;
if (planned.quarantined)
{
result.analysis.push_back("in quarantine: "
+ planned.quarantineReason);
}
states_[position].repeatsLeft
= options.repeat > 0 ? options.repeat - 1 : 0;
}
}
ScheduleOutcome
Run()
{
for (std::size_t position = 0; position < plan_.tests.size();
++position)
{
Job job;
job.test = position;
const PlannedTest &planned = plan_.tests[position];
if (planned.replaySequence)
{
job.before = *planned.replaySequence;
job.fresh = true;
job.seed = planned.replaySeed;
}
Enqueue(std::move(job));
}
if (options_.inProcess)
{
RunInProcess();
}
else
{
RunInWorkers();
}
ScheduleOutcome outcome;
for (auto &state : states_)
{
if (!state.finished)
{
state.result.outcome = Outcome::kNotRun;
if (infrastructureFailed_)
{
state.result.category
= FailureCategory::kInfrastructure;
}
}
outcome.tests.push_back(std::move(state.result));
}
outcome.workers = workersStarted_;
outcome.notes = std::move(notes_);
outcome.infrastructureFailed = infrastructureFailed_;
std::error_code error;
// Only an empty root goes; kept directories hold failures.
std::filesystem::remove(temporaryRoot_, error);
return outcome;
}
private:
// Queue and dispatch.
void
Enqueue(Job job)
{
++states_[job.test].pendingJobs;
states_[job.test].queued = true;
if (job.kind == Job::Kind::kProbe || !job.before.empty()
|| states_[job.test].result.attempts.size() > 0)
{
// Retries and probes go ahead of first attempts: they
// finish a verdict that is already overdue.
queue_.push_front(std::move(job));
}
else
{
queue_.push_back(std::move(job));
}
}
/// Whether the dependencies of a test have their verdicts:
/// returns false while one is still open.
bool
DependenciesSettled(std::size_t position) const
{
for (const std::size_t dependency :
plan_.tests[position].dependencies)
{
if (!states_[dependency].finished)
{
return false;
}
}
return true;
}
bool
ResourcesFree(std::size_t position) const
{
for (const auto &resource :
plan_.tests[position].entry->resources)
{
if (heldResources_.count(resource) > 0)
{
return false;
}
}
return true;
}
std::size_t
RunningCount() const
{
std::size_t count = 0;
for (const auto &worker : workers_)
{
count += worker->job.has_value() ? 1 : 0;
}
return count;
}
/// The next job that may start now, removed from the queue.
std::optional<Job>
TakeRunnable()
{
for (auto job = queue_.begin(); job != queue_.end(); ++job)
{
const std::size_t position = job->test;
if (job->kind == Job::Kind::kAttempt
&& !DependenciesSettled(position))
{
continue;
}
const bool serial = plan_.tests[position].entry->serial;
if (serial && RunningCount() > 0)
{
// Nothing else starts until the serial test has run.
return std::nullopt;
}
if (serialRunning_)
{
return std::nullopt;
}
if (!ResourcesFree(position))
{
continue;
}
Job taken = std::move(*job);
queue_.erase(job);
return taken;
|