Jorijn van der Graaf
2c77cdcfab
Frames go on counting while paused, since Pump still runs, but status gave no pause frame and `wait --paused` replied with the current one. A wait that came minutes after `pause --on-check` caught a failure at frame 237 said "paused at frame 5475", which an agent takes for when the check failed. The session records the frame in PauseNow, and status and the paused wait report it as pausedAt. The client prints "PAUSED since frame N", and "paused at frame N (now M)" when the two differ. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1140 lines
52 KiB
C++
1140 lines
52 KiB
C++
// SPDX-License-Identifier: LGPL-3.0-only
|
|
// SPDX-FileCopyrightText: Copyright (C) 2026 Catcrafts®
|
|
|
|
module;
|
|
#include <signal.h>
|
|
#include <sys/prctl.h>
|
|
#include <unistd.h>
|
|
module Crafter.Debug:Session_impl;
|
|
import std;
|
|
import Crafter.Thread;
|
|
import Crafter.Debug.Core;
|
|
import :Registry;
|
|
import :Timeline;
|
|
import :Condition;
|
|
import :Session;
|
|
import :Transport;
|
|
|
|
using namespace Crafter::Debug;
|
|
using Crafter::Debug::Detail::Exchange;
|
|
|
|
// Named, not anonymous: Session::State's members take these types, and a
|
|
// type with internal linkage in the signature of one with module linkage is
|
|
// an exposure clang warns about. Nothing here is exported.
|
|
namespace Crafter::Debug::SessionImpl {
|
|
using Clock = std::chrono::steady_clock;
|
|
|
|
// Request parameters, decoded through Codec so a mistyped request is
|
|
// answered with every mistake in it, as an app command would be.
|
|
struct ListParams {
|
|
std::string glob = "*";
|
|
};
|
|
|
|
struct LogParams {
|
|
std::uint32_t last = 50;
|
|
std::string level = "info";
|
|
std::string category;
|
|
std::string grep;
|
|
std::optional<std::uint64_t> sinceFrame;
|
|
std::uint64_t afterSequence = 0;
|
|
std::vector<std::string> kinds;
|
|
bool follow = false;
|
|
std::uint32_t forMs = 30 * 60 * 1000;
|
|
};
|
|
|
|
struct PauseParams {
|
|
std::optional<std::string> when;
|
|
bool onCheck = false;
|
|
bool trap = false;
|
|
bool clear = false;
|
|
};
|
|
|
|
struct StepParams {
|
|
std::uint32_t count = 1;
|
|
};
|
|
|
|
struct WaitParams {
|
|
std::optional<std::string> condition;
|
|
std::optional<std::uint32_t> frames;
|
|
std::optional<std::string> event;
|
|
bool paused = false;
|
|
};
|
|
|
|
struct QuitParams {
|
|
bool force = false;
|
|
};
|
|
|
|
struct GetParams {
|
|
std::vector<std::string> paths;
|
|
};
|
|
|
|
struct SetParams {
|
|
Required<std::string> path;
|
|
Required<Value> value;
|
|
};
|
|
|
|
struct TraceParams {
|
|
std::vector<std::string> paths;
|
|
std::uint32_t frames = 60;
|
|
bool history = false;
|
|
};
|
|
|
|
struct Job {
|
|
std::string method;
|
|
Value params;
|
|
std::shared_ptr<Exchange> exchange;
|
|
Clock::time_point deadline;
|
|
};
|
|
|
|
struct Pending {
|
|
std::string method;
|
|
std::function<std::optional<Result<Value>>()> poll;
|
|
std::shared_ptr<Exchange> exchange;
|
|
Clock::time_point deadline;
|
|
};
|
|
|
|
enum class WaitKind : std::uint8_t {
|
|
Frames,
|
|
Condition,
|
|
Event,
|
|
Paused,
|
|
Steps,
|
|
};
|
|
|
|
struct Waiter {
|
|
WaitKind kind = WaitKind::Frames;
|
|
std::shared_ptr<Exchange> exchange;
|
|
Clock::time_point deadline;
|
|
Clock::time_point since;
|
|
std::uint64_t targetFrame = 0;
|
|
std::optional<Condition> condition;
|
|
std::string event;
|
|
};
|
|
|
|
struct TraceRun {
|
|
std::vector<std::string> paths;
|
|
std::uint32_t remaining = 0;
|
|
std::vector<std::uint64_t> frames;
|
|
std::vector<std::vector<Value>> series;
|
|
std::shared_ptr<Exchange> exchange;
|
|
Clock::time_point deadline;
|
|
};
|
|
|
|
struct Follower {
|
|
Detail::TimelineQuery query;
|
|
std::shared_ptr<Exchange> exchange;
|
|
Clock::time_point until;
|
|
};
|
|
|
|
// The screenshot source, which an app may set before its Session exists.
|
|
std::mutex captureMutex;
|
|
std::function<Async<std::vector<std::byte>>()> captureSource;
|
|
|
|
// The one Session's state, for the free functions.
|
|
std::atomic<Session::State*> current{ nullptr };
|
|
|
|
template<class P>
|
|
std::optional<P> DecodeParams(const Value& params, Exchange& exchange) {
|
|
P decoded{};
|
|
DecodeContext context;
|
|
Crafter::Debug::Decode(params.IsNull() ? Value::MakeObject() : params, decoded, context);
|
|
if (!context.Ok()) {
|
|
exchange.Reply(MakeErrorReply(Codes::invalidParams, context.Report()));
|
|
return std::nullopt;
|
|
}
|
|
return decoded;
|
|
}
|
|
|
|
void ReplyWith(Exchange& exchange, Result<Value> result) {
|
|
exchange.Reply(result.Ok() ? MakeReply(std::move(result.Get())) : MakeErrorReply(result.GetError().code, result.GetError().message));
|
|
}
|
|
|
|
Value Object(std::initializer_list<std::pair<std::string_view, Value>> members) {
|
|
Value object = Value::MakeObject();
|
|
for (const auto& [key, value] : members) object.Set(key, value);
|
|
return object;
|
|
}
|
|
|
|
std::string ProgramName() {
|
|
std::error_code ec;
|
|
const std::filesystem::path exe = std::filesystem::read_symlink("/proc/self/exe", ec);
|
|
return ec ? std::string("app") : exe.filename().string();
|
|
}
|
|
|
|
std::string NewToken() {
|
|
std::random_device random;
|
|
return std::format("{:08x}{:08x}{:08x}{:08x}", random(), random(), random(), random());
|
|
}
|
|
|
|
// A tracer (a debugger) is attached to this process: TracerPid in
|
|
// /proc/self/status is nonzero.
|
|
bool TracerAttached() {
|
|
std::ifstream status("/proc/self/status");
|
|
std::string line;
|
|
while (std::getline(status, line)) {
|
|
if (line.starts_with("TracerPid:")) return line.find_first_not_of("0\t ", 10) != std::string::npos;
|
|
}
|
|
return false;
|
|
}
|
|
|
|
std::string FileName(std::string_view path) {
|
|
const std::size_t slash = path.rfind('/');
|
|
return std::string(slash == std::string_view::npos ? path : path.substr(slash + 1));
|
|
}
|
|
|
|
std::string Abbreviated(std::string text, std::size_t limit = 200) {
|
|
if (text.size() <= limit) return text;
|
|
text.resize(limit);
|
|
return text + "...";
|
|
}
|
|
|
|
Value EntryValue(const TimelineEntry& entry) {
|
|
Value value = Object({
|
|
{ "sequence", Value(entry.sequence) },
|
|
{ "frame", Value(entry.frame) },
|
|
{ "seconds", Value(entry.seconds) },
|
|
{ "kind", Value(EntryKindName(entry.kind)) },
|
|
{ "category", Value(entry.category) },
|
|
{ "message", Value(entry.message) },
|
|
{ "file", Value(entry.file) },
|
|
{ "line", Value(entry.line) },
|
|
});
|
|
if (!entry.payload.IsNull()) value.Set("payload", entry.payload);
|
|
if (entry.kind == EntryKind::Check) value.Set("count", Value(entry.count));
|
|
return value;
|
|
}
|
|
|
|
std::expected<Value, std::string> ResolvePath(std::string_view path) {
|
|
Detail::Resolved resolved = Detail::Registry::Instance().Resolve(path);
|
|
if (!resolved.Ok()) return std::unexpected(std::move(resolved.error));
|
|
return std::move(resolved.value);
|
|
}
|
|
|
|
// min, max, mean, first, last and how often the value changed, for a
|
|
// traced series; numbers only for the arithmetic ones.
|
|
Value Stats(std::span<const std::uint64_t> frames, std::span<const Value> values) {
|
|
Value stats = Value::MakeObject();
|
|
if (values.empty()) return stats;
|
|
stats.Set("first", values.front());
|
|
stats.Set("last", values.back());
|
|
std::uint64_t changes = 0;
|
|
std::optional<std::uint64_t> firstChange;
|
|
for (std::size_t i = 1; i < values.size(); ++i) {
|
|
if (!(values[i] == values[i - 1])) {
|
|
++changes;
|
|
if (!firstChange) firstChange = frames[i];
|
|
}
|
|
}
|
|
stats.Set("changes", Value(changes));
|
|
if (firstChange) stats.Set("firstChangeFrame", Value(*firstChange));
|
|
if (std::ranges::all_of(values, [](const Value& value) { return value.IsNumber(); })) {
|
|
double sum = 0;
|
|
double low = values.front().AsNumber();
|
|
double high = low;
|
|
for (const Value& value : values) {
|
|
const double number = value.AsNumber();
|
|
sum += number;
|
|
low = std::min(low, number);
|
|
high = std::max(high, number);
|
|
}
|
|
stats.Set("min", Value(low));
|
|
stats.Set("max", Value(high));
|
|
stats.Set("mean", Value(sum / static_cast<double>(values.size())));
|
|
}
|
|
return stats;
|
|
}
|
|
|
|
constexpr std::array<std::string_view, 13> builtinMethods = {
|
|
"debug.hello", "debug.status", "debug.list", "debug.log", "debug.pause", "debug.resume", "debug.step",
|
|
"debug.wait", "debug.quit", "debug.get", "debug.set", "debug.trace", "debug.screenshot",
|
|
};
|
|
}
|
|
|
|
using namespace Crafter::Debug::SessionImpl;
|
|
|
|
namespace Crafter::Debug {
|
|
struct Session::State {
|
|
SessionOptions options;
|
|
std::string token = NewToken();
|
|
std::string name;
|
|
Clock::time_point started = Clock::now();
|
|
std::int64_t pid = ::getpid();
|
|
|
|
// Guards what transport threads, the Pump thread and logging threads share.
|
|
std::mutex mutex;
|
|
std::deque<Job> jobs;
|
|
std::vector<Waiter> waiters;
|
|
std::vector<Follower> followers;
|
|
std::optional<Condition> armedWhen;
|
|
bool armedOnCheck = false;
|
|
bool armedTrap = false;
|
|
std::string pauseReason;
|
|
// The frame the pause began at. Frames go on counting while paused,
|
|
// so the current one says nothing about when it stopped.
|
|
std::uint64_t pausedAt = 0;
|
|
bool shutDown = false;
|
|
|
|
// The Pump thread's own.
|
|
std::vector<Pending> pendings;
|
|
std::vector<TraceRun> traces;
|
|
bool quitHandled = false;
|
|
std::optional<Clock::time_point> lastPump;
|
|
|
|
std::atomic<bool> paused{ false };
|
|
std::atomic<std::uint32_t> steps{ 0 };
|
|
std::atomic<bool> pauseSupported{ false };
|
|
std::atomic<bool> quitRequested{ false };
|
|
std::atomic<std::uint64_t> frame{ 0 };
|
|
std::atomic<std::int64_t> lastPumpNanos{ -1 };
|
|
std::atomic<double> fps{ 0 };
|
|
|
|
std::optional<Group> builtins;
|
|
|
|
// When listening: the transport, and the discovery record that
|
|
// tells clients where it is.
|
|
std::unique_ptr<Detail::Server> server;
|
|
std::optional<Registration> registration;
|
|
|
|
explicit State(SessionOptions given) : options(std::move(given)) {
|
|
name = options.name.empty() ? ProgramName() : options.name;
|
|
}
|
|
|
|
void Start() {
|
|
builtins.emplace("debug");
|
|
builtins->Watch("frame", [this] { return frame.load(); }, { .help = "frames pumped so far" });
|
|
builtins->Watch("paused", [this] { return paused.load(); }, { .help = "whether Advance() holds the app" });
|
|
builtins->Watch("fps", [this] { return fps.load(); }, { .help = "Pumps per second, smoothed" });
|
|
}
|
|
|
|
// Last in the constructor: once the record is written, agents find
|
|
// this instance and send it requests.
|
|
void Listen(const std::shared_ptr<State>& self) {
|
|
// Crafter.Network hands new streams to the pool, and this hands
|
|
// each request on; Start(1) gives the pool no worker to run them.
|
|
if (Crafter::ThreadPool::activeThreads.load() == 0) {
|
|
throw std::logic_error("Crafter.Debug: the Session needs Crafter.Thread's pool with a worker thread. "
|
|
"Call ThreadPool::Start() before creating it (Start(1) starts no worker).");
|
|
}
|
|
const std::weak_ptr<State> weak = self;
|
|
server = std::make_unique<Detail::Server>(options.port, [weak](const Value& request, std::shared_ptr<Exchange> exchange) {
|
|
if (const std::shared_ptr<State> alive = weak.lock()) return alive->Handle(request, exchange);
|
|
exchange->Reply(MakeErrorReply(Codes::shuttingDown, "the app is shutting down"));
|
|
});
|
|
registration.emplace(InstanceRecord{
|
|
.name = name,
|
|
.port = server->Port(),
|
|
.token = token,
|
|
.protocol = protocolVersion,
|
|
.sourceId = std::string(Detail::SourceId()),
|
|
.client = std::string(Detail::ClientPath()),
|
|
});
|
|
}
|
|
|
|
void Agent(std::string_view method, const Value& params) {
|
|
const std::string text = params.IsNull() ? std::string(method) : std::format("{} {}", method, Abbreviated(params.Dump()));
|
|
Detail::Timeline::Instance().Append(Detail::MakeEntry(EntryKind::Agent, "agent", text, std::source_location::current()));
|
|
}
|
|
|
|
void PauseNow(std::string reason, bool trap) {
|
|
{
|
|
std::lock_guard lock(mutex);
|
|
pauseReason = reason;
|
|
pausedAt = frame.load();
|
|
}
|
|
paused.store(true);
|
|
Detail::Timeline::Instance().Append(Detail::MakeEntry(EntryKind::Pause, "debug", std::format("paused: {}", reason), std::source_location::current()));
|
|
if (trap && TracerAttached()) ::raise(SIGTRAP);
|
|
}
|
|
|
|
// ---- requests, on transport threads ----
|
|
|
|
void Handle(const Value& request, const std::shared_ptr<Exchange>& exchange) {
|
|
try {
|
|
Dispatch(request, exchange);
|
|
} catch (const std::exception& error) {
|
|
exchange->Reply(MakeErrorReply(Codes::exception, error.what()));
|
|
} catch (...) {
|
|
exchange->Reply(MakeErrorReply(Codes::exception, "the session threw something that is not a std::exception"));
|
|
}
|
|
}
|
|
|
|
void Dispatch(const Value& request, const std::shared_ptr<Exchange>& exchange) {
|
|
if (!request.IsObject()) {
|
|
return exchange->Reply(MakeErrorReply(Codes::invalidRequest, R"(a request is a JSON object: {"token", "method", "params", "timeoutMs"})"));
|
|
}
|
|
const Value* tokenValue = request.Find("token");
|
|
if (tokenValue == nullptr || !tokenValue->IsString() || tokenValue->AsString() != token) {
|
|
return exchange->Reply(MakeErrorReply(Codes::unauthorized, "wrong or missing token: crafter-debug reads it from the instance's discovery file"));
|
|
}
|
|
const Value* methodValue = request.Find("method");
|
|
if (methodValue == nullptr || !methodValue->IsString()) {
|
|
return exchange->Reply(MakeErrorReply(Codes::invalidRequest, "a request needs a method"));
|
|
}
|
|
const std::string method = methodValue->AsString();
|
|
const Value* paramsValue = request.Find("params");
|
|
const Value params = paramsValue != nullptr ? *paramsValue : Value();
|
|
std::int64_t timeoutMs = 30'000;
|
|
if (const Value* timeout = request.Find("timeoutMs"); timeout != nullptr && timeout->IsInteger() && timeout->AsInteger() > 0) {
|
|
timeoutMs = timeout->AsInteger();
|
|
}
|
|
const Clock::time_point deadline = Clock::now() + std::chrono::milliseconds(timeoutMs);
|
|
{
|
|
std::lock_guard lock(mutex);
|
|
if (shutDown) return exchange->Reply(MakeErrorReply(Codes::shuttingDown, "the app is shutting down"));
|
|
}
|
|
|
|
if (method == "debug.hello") return exchange->Reply(MakeReply(Hello()));
|
|
if (method == "debug.status") return exchange->Reply(MakeReply(Status()));
|
|
if (method == "debug.list") return List(params, *exchange);
|
|
if (method == "debug.log") return Logs(params, exchange, deadline);
|
|
if (method == "debug.pause") return Pause(params, *exchange);
|
|
if (method == "debug.resume") return Resume(*exchange);
|
|
if (method == "debug.step") return Step(params, exchange, deadline);
|
|
if (method == "debug.wait") return Wait(params, exchange, deadline);
|
|
if (method == "debug.quit") return Quit(params, exchange);
|
|
|
|
const bool builtin = std::ranges::find(builtinMethods, method) != builtinMethods.end();
|
|
if (method.starts_with("debug.") && !builtin) {
|
|
const auto suggestion = Suggest(method, builtinMethods);
|
|
return exchange->Reply(MakeErrorReply(Codes::unknownMethod, suggestion ? std::format("no method '{}'; did you mean '{}'?", method, *suggestion)
|
|
: std::format("no method '{}'", method)));
|
|
}
|
|
if (!builtin && Detail::Registry::Instance().FindCommand(method) == nullptr) {
|
|
const auto suggestion = Detail::Registry::Instance().SuggestCommand(method);
|
|
return exchange->Reply(MakeErrorReply(Codes::unknownMethod, suggestion ? std::format("no command '{}'; did you mean '{}'?", method, *suggestion)
|
|
: std::format("no command '{}'; crafter-debug cmds lists them", method)));
|
|
}
|
|
std::lock_guard lock(mutex);
|
|
jobs.push_back({ method, params, exchange, deadline });
|
|
}
|
|
|
|
Value Hello() const {
|
|
return Object({
|
|
{ "protocol", Value(protocolVersion) },
|
|
{ "library", Value(libraryVersion) },
|
|
{ "name", Value(name) },
|
|
{ "pid", Value(pid) },
|
|
});
|
|
}
|
|
|
|
Value Status() {
|
|
Value status = Object({
|
|
{ "name", Value(name) },
|
|
{ "pid", Value(pid) },
|
|
{ "frame", Value(frame.load()) },
|
|
{ "fps", Value(fps.load()) },
|
|
{ "paused", Value(paused.load()) },
|
|
{ "pauseSupported", Value(pauseSupported.load()) },
|
|
{ "quitRequested", Value(quitRequested.load()) },
|
|
});
|
|
const std::int64_t last = lastPumpNanos.load();
|
|
const std::int64_t sinceStart = std::chrono::duration_cast<std::chrono::nanoseconds>(Clock::now() - started).count();
|
|
status.Set("lastPumpAgoMs", last < 0 ? Value() : Value((sinceStart - last) / 1'000'000));
|
|
{
|
|
std::lock_guard lock(mutex);
|
|
status.Set("pauseReason", Value(pauseReason));
|
|
if (paused.load()) status.Set("pausedAt", Value(pausedAt));
|
|
status.Set("queued", Value(jobs.size()));
|
|
status.Set("waiting", Value(waiters.size()));
|
|
status.Set("following", Value(followers.size()));
|
|
Value armed = Value::MakeObject();
|
|
if (armedWhen) armed.Set("when", Value(armedWhen->Text()));
|
|
if (armedOnCheck) armed.Set("onCheck", Value(true));
|
|
status.Set("armed", std::move(armed));
|
|
}
|
|
std::uint64_t failures = 0;
|
|
Value sites = Value::MakeArray();
|
|
for (const Detail::CheckSite& site : Detail::Timeline::Instance().CheckSites()) {
|
|
failures += site.failures;
|
|
sites.Push(Object({ { "file", Value(FileName(site.file)) }, { "line", Value(site.line) }, { "category", Value(site.category) }, { "failures", Value(site.failures) } }));
|
|
}
|
|
status.Set("checkFailures", Value(failures));
|
|
status.Set("checkSites", std::move(sites));
|
|
status.Set("commands", Value(Detail::Registry::Instance().Commands().size()));
|
|
status.Set("watches", Value(Detail::Registry::Instance().Watches().size()));
|
|
return status;
|
|
}
|
|
|
|
void List(const Value& params, Exchange& exchange) {
|
|
const auto decoded = DecodeParams<ListParams>(params, exchange);
|
|
if (!decoded) return;
|
|
Value commands = Value::MakeArray();
|
|
for (const auto& command : Detail::Registry::Instance().Commands(decoded->glob)) {
|
|
commands.Push(Object({ { "name", Value(command->name) }, { "description", Value(command->description) }, { "args", command->args }, { "returns", command->returns } }));
|
|
}
|
|
Value watches = Value::MakeArray();
|
|
for (const auto& watch : Detail::Registry::Instance().Watches(decoded->glob)) {
|
|
watches.Push(Object({ { "path", Value(watch->path) }, { "type", watch->type }, { "help", Value(watch->help) }, { "readOnly", Value(!watch->set) } }));
|
|
}
|
|
Value samples = Value::MakeArray();
|
|
for (const std::string& path : Detail::Registry::Instance().SamplePaths(decoded->glob)) samples.Push(Value(path));
|
|
exchange.Reply(MakeReply(Object({ { "commands", std::move(commands) }, { "watches", std::move(watches) }, { "samples", std::move(samples) } })));
|
|
}
|
|
|
|
void Logs(const Value& params, const std::shared_ptr<Exchange>& exchange, Clock::time_point) {
|
|
const auto decoded = DecodeParams<LogParams>(params, *exchange);
|
|
if (!decoded) return;
|
|
Detail::TimelineQuery query;
|
|
query.afterSequence = decoded->afterSequence;
|
|
query.sinceFrame = decoded->sinceFrame;
|
|
query.category = decoded->category;
|
|
query.grep = decoded->grep;
|
|
if (decoded->level == "info") {
|
|
query.minLevel = 0;
|
|
} else if (decoded->level == "warn") {
|
|
query.minLevel = 1;
|
|
} else if (decoded->level == "error") {
|
|
query.minLevel = 2;
|
|
} else {
|
|
return exchange->Reply(MakeErrorReply(Codes::invalidParams, std::format("level: '{}' is not one of info, warn, error", decoded->level)));
|
|
}
|
|
for (const std::string& kind : decoded->kinds) {
|
|
bool known = false;
|
|
for (std::uint8_t k = 0; k <= static_cast<std::uint8_t>(EntryKind::Stdio); ++k) {
|
|
if (EntryKindName(static_cast<EntryKind>(k)) == kind) {
|
|
query.kinds.push_back(static_cast<EntryKind>(k));
|
|
known = true;
|
|
}
|
|
}
|
|
if (!known) return exchange->Reply(MakeErrorReply(Codes::invalidParams, std::format("kinds: '{}' is not one of log, warn, error, check, event, agent, pause, stdio", kind)));
|
|
}
|
|
// `last` 0 means none, as with tail -n 0: a follower that only
|
|
// wants what comes next. (A TimelineQuery's 0 means all.)
|
|
Detail::TimelineQuery backlog = query;
|
|
backlog.last = decoded->last;
|
|
const auto recent = [&] { return decoded->last == 0 ? std::vector<TimelineEntry>() : Detail::Timeline::Instance().Query(backlog); };
|
|
if (!decoded->follow) {
|
|
Value entries = Value::MakeArray();
|
|
for (const TimelineEntry& entry : recent()) entries.Push(EntryValue(entry));
|
|
return exchange->Reply(MakeReply(Object({ { "entries", std::move(entries) }, { "lastSequence", Value(Detail::Timeline::Instance().LastSequence()) } })));
|
|
}
|
|
// The backlog and the registration happen under one lock, which
|
|
// the listener also takes before streaming, so a follower sees
|
|
// every entry once and in order. Stream never blocks.
|
|
std::lock_guard lock(mutex);
|
|
const std::uint64_t upTo = Detail::Timeline::Instance().LastSequence();
|
|
for (const TimelineEntry& entry : recent()) {
|
|
if (entry.sequence > upTo) continue;
|
|
if (!exchange->Stream(Object({ { "entry", EntryValue(entry) } }))) return;
|
|
}
|
|
Follower follower{ query, exchange, Clock::now() + std::chrono::milliseconds(decoded->forMs) };
|
|
follower.query.afterSequence = std::max(query.afterSequence, upTo);
|
|
followers.push_back(std::move(follower));
|
|
}
|
|
|
|
void Pause(const Value& params, Exchange& exchange) {
|
|
const auto decoded = DecodeParams<PauseParams>(params, exchange);
|
|
if (!decoded) return;
|
|
if (decoded->clear) {
|
|
{
|
|
std::lock_guard lock(mutex);
|
|
armedWhen.reset();
|
|
armedOnCheck = false;
|
|
armedTrap = false;
|
|
}
|
|
Agent("pause --clear", Value());
|
|
return exchange.Reply(MakeReply(Object({ { "armed", Value(false) } })));
|
|
}
|
|
if (!pauseSupported.load()) {
|
|
return exchange.Reply(MakeErrorReply(Codes::pauseUnsupported, "this app never calls Debug::Advance() or Debug::Paused(), so pausing would hold nothing; "
|
|
"gate the simulation on them (see the README)"));
|
|
}
|
|
if (decoded->trap && !decoded->when && !decoded->onCheck) {
|
|
return exchange.Reply(MakeErrorReply(Codes::invalidParams, "trap needs when or onCheck: it stops the app where that happens"));
|
|
}
|
|
if (decoded->when) {
|
|
std::optional<Condition> condition;
|
|
try {
|
|
condition = Condition::Parse(*decoded->when);
|
|
} catch (const ConditionError& error) {
|
|
return exchange.Reply(MakeErrorReply(Codes::conditionError, error.what()));
|
|
}
|
|
{
|
|
std::lock_guard lock(mutex);
|
|
armedWhen = std::move(condition);
|
|
armedTrap = decoded->trap;
|
|
}
|
|
Agent("pause --when", Value(*decoded->when));
|
|
return exchange.Reply(MakeReply(Object({ { "armed", Value("when") }, { "condition", Value(*decoded->when) } })));
|
|
}
|
|
if (decoded->onCheck) {
|
|
{
|
|
std::lock_guard lock(mutex);
|
|
armedOnCheck = true;
|
|
armedTrap = decoded->trap;
|
|
}
|
|
Agent("pause --on-check", Value());
|
|
return exchange.Reply(MakeReply(Object({ { "armed", Value("onCheck") } })));
|
|
}
|
|
Agent("pause", Value());
|
|
PauseNow("an agent asked", false);
|
|
exchange.Reply(MakeReply(Object({ { "paused", Value(true) }, { "frame", Value(frame.load()) } })));
|
|
}
|
|
|
|
void Resume(Exchange& exchange) {
|
|
Agent("resume", Value());
|
|
steps.store(0);
|
|
paused.store(false);
|
|
{
|
|
std::lock_guard lock(mutex);
|
|
pauseReason.clear();
|
|
}
|
|
Detail::Timeline::Instance().Append(Detail::MakeEntry(EntryKind::Pause, "debug", "resumed", std::source_location::current()));
|
|
exchange.Reply(MakeReply(Object({ { "paused", Value(false) }, { "frame", Value(frame.load()) } })));
|
|
}
|
|
|
|
void Step(const Value& params, const std::shared_ptr<Exchange>& exchange, Clock::time_point deadline) {
|
|
const auto decoded = DecodeParams<StepParams>(params, *exchange);
|
|
if (!decoded) return;
|
|
if (!pauseSupported.load()) {
|
|
return exchange->Reply(MakeErrorReply(Codes::pauseUnsupported, "this app never calls Debug::Advance() or Debug::Paused(), so there is nothing to step"));
|
|
}
|
|
if (!paused.load()) return exchange->Reply(MakeErrorReply(Codes::notPaused, "step works while paused: run crafter-debug pause first"));
|
|
if (decoded->count == 0) return exchange->Reply(MakeErrorReply(Codes::invalidParams, "count: step at least 1"));
|
|
Agent("step", Object({ { "count", Value(decoded->count) } }));
|
|
std::lock_guard lock(mutex);
|
|
steps.fetch_add(decoded->count);
|
|
waiters.push_back({ .kind = WaitKind::Steps, .exchange = exchange, .deadline = deadline, .since = Clock::now() });
|
|
}
|
|
|
|
void Wait(const Value& params, const std::shared_ptr<Exchange>& exchange, Clock::time_point deadline) {
|
|
const auto decoded = DecodeParams<WaitParams>(params, *exchange);
|
|
if (!decoded) return;
|
|
const std::int32_t asked = (decoded->condition ? 1 : 0) + (decoded->frames ? 1 : 0) + (decoded->event ? 1 : 0) + (decoded->paused ? 1 : 0);
|
|
if (asked != 1) return exchange->Reply(MakeErrorReply(Codes::invalidParams, "wait for one thing: condition, frames, event or paused"));
|
|
Waiter waiter{ .exchange = exchange, .deadline = deadline, .since = Clock::now() };
|
|
if (decoded->condition) {
|
|
try {
|
|
waiter.condition = Condition::Parse(*decoded->condition);
|
|
} catch (const ConditionError& error) {
|
|
return exchange->Reply(MakeErrorReply(Codes::conditionError, error.what()));
|
|
}
|
|
waiter.kind = WaitKind::Condition;
|
|
} else if (decoded->frames) {
|
|
waiter.kind = WaitKind::Frames;
|
|
waiter.targetFrame = frame.load() + *decoded->frames;
|
|
} else if (decoded->event) {
|
|
waiter.kind = WaitKind::Event;
|
|
waiter.event = *decoded->event;
|
|
} else {
|
|
waiter.kind = WaitKind::Paused;
|
|
if (paused.load()) {
|
|
std::lock_guard lock(mutex);
|
|
return exchange->Reply(MakeReply(Object({ { "frame", Value(frame.load()) }, { "pausedAt", Value(pausedAt) }, { "reason", Value(pauseReason) } })));
|
|
}
|
|
}
|
|
std::lock_guard lock(mutex);
|
|
waiters.push_back(std::move(waiter));
|
|
}
|
|
|
|
void Quit(const Value& params, const std::shared_ptr<Exchange>& exchange) {
|
|
const auto decoded = DecodeParams<QuitParams>(params, *exchange);
|
|
if (!decoded) return;
|
|
Agent(decoded->force ? "quit --force" : "quit", Value());
|
|
if (decoded->force) {
|
|
quitRequested.store(true);
|
|
// From the thread the reply leaves on, so the main loop
|
|
// need not run: --force is for an app whose loop is stuck.
|
|
return exchange->ReplyThen(MakeReply(Object({ { "quitting", Value(true) }, { "force", Value(true) } })), [] { std::_Exit(0); });
|
|
}
|
|
// The reply before the flag: a loop that sees the flag may end
|
|
// the Session at once, and the transport delivers a reply given
|
|
// before its connections close, not one given after.
|
|
exchange->Reply(MakeReply(Object({ { "quitting", Value(true) } })));
|
|
quitRequested.store(true);
|
|
}
|
|
|
|
// ---- the Pump thread ----
|
|
|
|
void Pump() {
|
|
const Clock::time_point now = Clock::now();
|
|
const std::uint64_t current = frame.fetch_add(1) + 1;
|
|
Detail::SetFrame(current);
|
|
if (lastPump) {
|
|
const double seconds = std::chrono::duration<double>(now - *lastPump).count();
|
|
if (seconds > 0) {
|
|
const double previous = fps.load();
|
|
fps.store(previous == 0 ? 1 / seconds : previous * 0.9 + 0.1 / seconds);
|
|
}
|
|
}
|
|
lastPump = now;
|
|
lastPumpNanos.store(std::chrono::duration_cast<std::chrono::nanoseconds>(now - started).count());
|
|
|
|
RunJobs(now);
|
|
PollPendings(now);
|
|
CheckArmed();
|
|
ServeWaiters(now, current);
|
|
RecordTraces(now, current);
|
|
ExpireFollowers(now);
|
|
if (quitRequested.load() && !quitHandled) {
|
|
quitHandled = true;
|
|
if (options.onQuit) options.onQuit();
|
|
}
|
|
}
|
|
|
|
void RunJobs(Clock::time_point now) {
|
|
std::deque<Job> taken;
|
|
{
|
|
std::lock_guard lock(mutex);
|
|
taken.swap(jobs);
|
|
}
|
|
for (Job& job : taken) {
|
|
if (now > job.deadline) {
|
|
job.exchange->Reply(MakeErrorReply(Codes::timeout, "not run: the app did not Pump before the request's timeout"));
|
|
continue;
|
|
}
|
|
try {
|
|
Execute(job);
|
|
} catch (const std::exception& error) {
|
|
job.exchange->Reply(MakeErrorReply(Codes::exception, error.what()));
|
|
} catch (...) {
|
|
job.exchange->Reply(MakeErrorReply(Codes::exception, "something that is not a std::exception was thrown"));
|
|
}
|
|
}
|
|
}
|
|
|
|
void Execute(Job& job) {
|
|
Agent(job.method, job.params);
|
|
if (job.method == "debug.get") return Get(job);
|
|
if (job.method == "debug.set") return Set(job);
|
|
if (job.method == "debug.trace") return StartTrace(job);
|
|
if (job.method == "debug.screenshot") return Screenshot(job);
|
|
const auto command = Detail::Registry::Instance().FindCommand(job.method);
|
|
if (!command) return job.exchange->Reply(MakeErrorReply(Codes::unknownMethod, std::format("'{}' was unregistered before it ran", job.method)));
|
|
Detail::CommandOutcome outcome = command->invoke(job.params);
|
|
if (outcome.done) return ReplyWith(*job.exchange, std::move(*outcome.done));
|
|
pendings.push_back({ job.method, std::move(outcome.pending), job.exchange, job.deadline });
|
|
}
|
|
|
|
void Get(Job& job) {
|
|
const auto decoded = DecodeParams<GetParams>(job.params, *job.exchange);
|
|
if (!decoded) return;
|
|
if (decoded->paths.empty()) return job.exchange->Reply(MakeErrorReply(Codes::invalidParams, "paths: name at least one path, or a glob such as 'fort1.*'"));
|
|
Detail::Registry& registry = Detail::Registry::Instance();
|
|
Value values = Value::MakeObject();
|
|
Value errors = Value::MakeObject();
|
|
const auto read = [&](std::string_view path) {
|
|
Detail::Resolved resolved = registry.Resolve(path);
|
|
if (resolved.Ok()) {
|
|
values.Set(path, std::move(resolved.value));
|
|
} else {
|
|
errors.Set(path, Value(resolved.error));
|
|
}
|
|
};
|
|
for (const std::string& path : decoded->paths) {
|
|
if (path.find_first_of("*?") == std::string::npos) {
|
|
read(path);
|
|
continue;
|
|
}
|
|
const auto watches = registry.Watches(path);
|
|
const auto samples = registry.SamplePaths(path);
|
|
if (watches.empty() && samples.empty()) errors.Set(path, Value("no watch or sample matches"));
|
|
for (const auto& watch : watches) read(watch->path);
|
|
for (const std::string& sample : samples) {
|
|
if (values.Find(sample) == nullptr) read(sample);
|
|
}
|
|
}
|
|
if (values.Size() == 0 && errors.Size() != 0) {
|
|
std::string message;
|
|
for (const auto& [path, error] : errors.Members()) message += message.empty() ? error.AsString() : "; " + error.AsString();
|
|
return job.exchange->Reply(MakeErrorReply(Codes::unknownPath, message));
|
|
}
|
|
Value result = Object({ { "values", std::move(values) } });
|
|
if (errors.Size() != 0) result.Set("errors", std::move(errors));
|
|
job.exchange->Reply(MakeReply(std::move(result)));
|
|
}
|
|
|
|
void Set(Job& job) {
|
|
const auto decoded = DecodeParams<SetParams>(job.params, *job.exchange);
|
|
if (!decoded) return;
|
|
if (const auto error = Detail::Registry::Instance().Set(*decoded->path, *decoded->value)) {
|
|
return job.exchange->Reply(MakeErrorReply(error->code, error->message));
|
|
}
|
|
Detail::Resolved now = Detail::Registry::Instance().Resolve(*decoded->path);
|
|
job.exchange->Reply(MakeReply(Object({ { "path", Value(*decoded->path) }, { "value", now.Ok() ? now.value : *decoded->value } })));
|
|
}
|
|
|
|
void StartTrace(Job& job) {
|
|
const auto decoded = DecodeParams<TraceParams>(job.params, *job.exchange);
|
|
if (!decoded) return;
|
|
if (decoded->paths.empty()) return job.exchange->Reply(MakeErrorReply(Codes::invalidParams, "paths: name at least one path"));
|
|
Detail::Registry& registry = Detail::Registry::Instance();
|
|
if (decoded->history) {
|
|
Value series = Value::MakeObject();
|
|
for (const std::string& path : decoded->paths) {
|
|
const std::vector<Detail::SamplePoint> points = registry.SampleHistory(path);
|
|
if (points.empty()) return job.exchange->Reply(MakeErrorReply(Codes::unknownPath, std::format("no sample history for '{}'; history is kept for Sample() paths", path)));
|
|
std::vector<std::uint64_t> frames;
|
|
std::vector<Value> values;
|
|
for (const Detail::SamplePoint& point : points) {
|
|
frames.push_back(point.frame);
|
|
values.push_back(point.value);
|
|
}
|
|
Value framesValue = Value::MakeArray();
|
|
for (const std::uint64_t f : frames) framesValue.Push(Value(f));
|
|
series.Set(path, Object({ { "frames", std::move(framesValue) }, { "values", Value::MakeArray(values) }, { "stats", Stats(frames, values) } }));
|
|
}
|
|
return job.exchange->Reply(MakeReply(Object({ { "series", std::move(series) } })));
|
|
}
|
|
if (decoded->frames == 0) return job.exchange->Reply(MakeErrorReply(Codes::invalidParams, "frames: trace at least 1 frame"));
|
|
for (const std::string& path : decoded->paths) {
|
|
const Detail::Resolved resolved = registry.Resolve(path);
|
|
if (!resolved.Ok()) return job.exchange->Reply(MakeErrorReply(Codes::unknownPath, resolved.error));
|
|
}
|
|
traces.push_back({ decoded->paths, decoded->frames, {}, std::vector<std::vector<Value>>(decoded->paths.size()), job.exchange, job.deadline });
|
|
}
|
|
|
|
void Screenshot(Job& job) {
|
|
std::function<Async<std::vector<std::byte>>()> source;
|
|
{
|
|
std::lock_guard lock(captureMutex);
|
|
source = captureSource;
|
|
}
|
|
if (!source) {
|
|
return job.exchange->Reply(MakeErrorReply(Codes::noCapture, "this app has no screenshot source: call Debug::SetCapture (the README has the Crafter.Graphics version)"));
|
|
}
|
|
pendings.push_back({
|
|
"debug.screenshot",
|
|
[async = source()]() mutable -> std::optional<Result<Value>> {
|
|
std::optional<Result<std::vector<std::byte>>> polled = async.Poll();
|
|
if (!polled) return std::nullopt;
|
|
if (!polled->Ok()) return Result<Value>(polled->GetError());
|
|
const std::vector<std::byte>& png = polled->Get();
|
|
return Result<Value>(Object({ { "png", Value(Base64Encode(png)) }, { "bytes", Value(png.size()) } }));
|
|
},
|
|
job.exchange,
|
|
job.deadline,
|
|
});
|
|
}
|
|
|
|
void PollPendings(Clock::time_point now) {
|
|
for (auto it = pendings.begin(); it != pendings.end();) {
|
|
try {
|
|
std::optional<Result<Value>> polled = it->poll();
|
|
if (polled) {
|
|
ReplyWith(*it->exchange, std::move(*polled));
|
|
it = pendings.erase(it);
|
|
continue;
|
|
}
|
|
} catch (const std::exception& error) {
|
|
it->exchange->Reply(MakeErrorReply(Codes::exception, std::format("the result threw: {}", error.what())));
|
|
it = pendings.erase(it);
|
|
continue;
|
|
} catch (...) {
|
|
it->exchange->Reply(MakeErrorReply(Codes::exception, "the result threw something that is not a std::exception"));
|
|
it = pendings.erase(it);
|
|
continue;
|
|
}
|
|
if (now > it->deadline) {
|
|
it->exchange->Reply(MakeErrorReply(Codes::timeout, std::format("{} ran, but its result was not ready before the request's timeout", it->method)));
|
|
it = pendings.erase(it);
|
|
continue;
|
|
}
|
|
++it;
|
|
}
|
|
}
|
|
|
|
void CheckArmed() {
|
|
std::optional<Condition> when;
|
|
bool trap = false;
|
|
{
|
|
std::lock_guard lock(mutex);
|
|
if (!armedWhen) return;
|
|
when = armedWhen;
|
|
trap = armedTrap;
|
|
}
|
|
const auto result = when->Evaluate(ResolvePath);
|
|
if (result && !*result) return;
|
|
{
|
|
std::lock_guard lock(mutex);
|
|
armedWhen.reset();
|
|
}
|
|
if (!result) {
|
|
Detail::Timeline::Instance().Append(Detail::MakeEntry(EntryKind::Warn, "debug", std::format("pause --when '{}' disarmed: {}", when->Text(), result.error()), std::source_location::current()));
|
|
return;
|
|
}
|
|
PauseNow(std::format("'{}' became true", when->Text()), trap);
|
|
}
|
|
|
|
std::string TimeoutMessage(const Waiter& waiter) const {
|
|
const auto waited = std::chrono::duration_cast<std::chrono::milliseconds>(Clock::now() - waiter.since).count();
|
|
switch (waiter.kind) {
|
|
case WaitKind::Frames: return std::format("frame {} was not reached within {} ms; the app is at frame {}", waiter.targetFrame, waited, frame.load());
|
|
case WaitKind::Paused: return std::format("the app did not pause within {} ms", waited);
|
|
case WaitKind::Steps: return std::format("the steps were not taken within {} ms: does the app call Debug::Advance() or TakeSteps() every frame?", waited);
|
|
case WaitKind::Event: return std::format("no '{}' event within {} ms", waiter.event, waited);
|
|
case WaitKind::Condition: {
|
|
std::string values;
|
|
for (const std::string& path : waiter.condition->Paths()) {
|
|
const auto value = ResolvePath(path);
|
|
if (!values.empty()) values += ", ";
|
|
values += value ? std::format("{} is {}", path, Abbreviated(value->Dump(), 80)) : std::format("{}: {}", path, value.error());
|
|
}
|
|
return std::format("'{}' was not true within {} ms{}", waiter.condition->Text(), waited, values.empty() ? std::string() : std::format(" ({})", values));
|
|
}
|
|
}
|
|
return "timed out";
|
|
}
|
|
|
|
void ServeWaiters(Clock::time_point now, std::uint64_t current) {
|
|
// Event waiters stay in the list, where the listener finds them.
|
|
std::vector<Waiter> taken;
|
|
{
|
|
std::lock_guard lock(mutex);
|
|
for (auto it = waiters.begin(); it != waiters.end();) {
|
|
if (it->kind == WaitKind::Event) {
|
|
if (now > it->deadline) {
|
|
it->exchange->Reply(MakeErrorReply(Codes::timeout, TimeoutMessage(*it)));
|
|
it = waiters.erase(it);
|
|
} else {
|
|
++it;
|
|
}
|
|
continue;
|
|
}
|
|
taken.push_back(std::move(*it));
|
|
it = waiters.erase(it);
|
|
}
|
|
}
|
|
std::vector<Waiter> remaining;
|
|
for (Waiter& waiter : taken) {
|
|
bool done = false;
|
|
switch (waiter.kind) {
|
|
case WaitKind::Frames:
|
|
if (current >= waiter.targetFrame) {
|
|
waiter.exchange->Reply(MakeReply(Object({ { "frame", Value(current) } })));
|
|
done = true;
|
|
}
|
|
break;
|
|
case WaitKind::Condition: {
|
|
const auto result = waiter.condition->Evaluate(ResolvePath);
|
|
if (!result) {
|
|
waiter.exchange->Reply(MakeErrorReply(Codes::conditionError, result.error()));
|
|
done = true;
|
|
} else if (*result) {
|
|
waiter.exchange->Reply(MakeReply(Object({ { "frame", Value(current) }, { "condition", Value(waiter.condition->Text()) } })));
|
|
done = true;
|
|
}
|
|
break;
|
|
}
|
|
case WaitKind::Paused:
|
|
if (paused.load()) {
|
|
std::string reason;
|
|
std::uint64_t since = 0;
|
|
{
|
|
std::lock_guard lock(mutex);
|
|
reason = pauseReason;
|
|
since = pausedAt;
|
|
}
|
|
waiter.exchange->Reply(MakeReply(Object({ { "frame", Value(current) }, { "pausedAt", Value(since) }, { "reason", Value(reason) } })));
|
|
done = true;
|
|
}
|
|
break;
|
|
case WaitKind::Steps:
|
|
if (steps.load() == 0) {
|
|
waiter.exchange->Reply(MakeReply(Object({ { "frame", Value(current) }, { "paused", Value(paused.load()) } })));
|
|
done = true;
|
|
}
|
|
break;
|
|
case WaitKind::Event: break;
|
|
}
|
|
if (!done && now > waiter.deadline) {
|
|
waiter.exchange->Reply(MakeErrorReply(Codes::timeout, TimeoutMessage(waiter)));
|
|
done = true;
|
|
}
|
|
if (!done) remaining.push_back(std::move(waiter));
|
|
}
|
|
std::lock_guard lock(mutex);
|
|
for (Waiter& waiter : remaining) waiters.push_back(std::move(waiter));
|
|
}
|
|
|
|
void RecordTraces(Clock::time_point now, std::uint64_t current) {
|
|
for (auto it = traces.begin(); it != traces.end();) {
|
|
TraceRun& run = *it;
|
|
run.frames.push_back(current);
|
|
for (std::size_t i = 0; i < run.paths.size(); ++i) {
|
|
const auto value = ResolvePath(run.paths[i]);
|
|
run.series[i].push_back(value ? *value : Value());
|
|
}
|
|
const bool truncated = now > run.deadline;
|
|
if (--run.remaining != 0 && !truncated) {
|
|
++it;
|
|
continue;
|
|
}
|
|
Value frames = Value::MakeArray();
|
|
for (const std::uint64_t f : run.frames) frames.Push(Value(f));
|
|
Value series = Value::MakeObject();
|
|
for (std::size_t i = 0; i < run.paths.size(); ++i) {
|
|
series.Set(run.paths[i], Object({ { "values", Value::MakeArray(run.series[i]) }, { "stats", Stats(run.frames, run.series[i]) } }));
|
|
}
|
|
Value result = Object({ { "frames", std::move(frames) }, { "series", std::move(series) } });
|
|
if (truncated) result.Set("truncated", Value(true));
|
|
run.exchange->Reply(MakeReply(std::move(result)));
|
|
it = traces.erase(it);
|
|
}
|
|
}
|
|
|
|
void ExpireFollowers(Clock::time_point now) {
|
|
std::vector<std::shared_ptr<Exchange>> ended;
|
|
{
|
|
std::lock_guard lock(mutex);
|
|
for (auto it = followers.begin(); it != followers.end();) {
|
|
if (now > it->until) {
|
|
ended.push_back(it->exchange);
|
|
it = followers.erase(it);
|
|
} else {
|
|
++it;
|
|
}
|
|
}
|
|
}
|
|
for (const auto& exchange : ended) exchange->Reply(MakeReply(Object({ { "ended", Value("the follow time elapsed") } })));
|
|
}
|
|
|
|
// ---- every timeline entry, on the thread that recorded it ----
|
|
|
|
void OnEntry(const TimelineEntry& entry) {
|
|
if (entry.kind == EntryKind::Check) {
|
|
bool trigger = false;
|
|
bool trap = false;
|
|
{
|
|
std::lock_guard lock(mutex);
|
|
if (armedOnCheck) {
|
|
armedOnCheck = false;
|
|
trigger = true;
|
|
trap = armedTrap;
|
|
}
|
|
}
|
|
if (trigger) PauseNow(std::format("check failed at {}:{}: {}", FileName(entry.file), entry.line, entry.message), trap);
|
|
}
|
|
if (entry.kind == EntryKind::Event) {
|
|
std::vector<std::shared_ptr<Exchange>> matched;
|
|
{
|
|
std::lock_guard lock(mutex);
|
|
for (auto it = waiters.begin(); it != waiters.end();) {
|
|
if (it->kind == WaitKind::Event && it->event == entry.category) {
|
|
matched.push_back(it->exchange);
|
|
it = waiters.erase(it);
|
|
} else {
|
|
++it;
|
|
}
|
|
}
|
|
}
|
|
for (const auto& exchange : matched) {
|
|
exchange->Reply(MakeReply(Object({ { "frame", Value(entry.frame) }, { "event", Value(entry.category) }, { "payload", entry.payload } })));
|
|
}
|
|
}
|
|
if (entry.sequence == 0) return;
|
|
std::lock_guard lock(mutex);
|
|
std::erase_if(followers, [&](Follower& follower) {
|
|
if (!Detail::Matches(follower.query, entry)) return false;
|
|
return !follower.exchange->Stream(Object({ { "entry", EntryValue(entry) } }));
|
|
});
|
|
}
|
|
|
|
// ---- the end, on the thread destroying the Session ----
|
|
|
|
void Shutdown() {
|
|
Detail::Timeline::Instance().SetListener(nullptr);
|
|
std::deque<Job> queued;
|
|
std::vector<Waiter> waiting;
|
|
std::vector<Follower> following;
|
|
{
|
|
std::lock_guard lock(mutex);
|
|
shutDown = true;
|
|
queued.swap(jobs);
|
|
waiting.swap(waiters);
|
|
following.swap(followers);
|
|
}
|
|
const Value goodbye = MakeErrorReply(Codes::shuttingDown, "the app is shutting down");
|
|
for (const Job& job : queued) job.exchange->Reply(goodbye);
|
|
for (const Waiter& waiter : waiting) waiter.exchange->Reply(goodbye);
|
|
for (const Follower& follower : following) follower.exchange->Reply(goodbye);
|
|
for (const Pending& pending : pendings) pending.exchange->Reply(goodbye);
|
|
for (const TraceRun& run : traces) run.exchange->Reply(goodbye);
|
|
pendings.clear();
|
|
traces.clear();
|
|
builtins.reset();
|
|
}
|
|
};
|
|
}
|
|
|
|
Session::Session(SessionOptions options) : state(std::make_shared<State>(std::move(options))) {
|
|
State* expected = nullptr;
|
|
if (!current.compare_exchange_strong(expected, state.get())) {
|
|
throw std::logic_error("Crafter.Debug: a Session already exists; there is one per process");
|
|
}
|
|
try {
|
|
state->Start();
|
|
std::weak_ptr<State> weak = state;
|
|
Detail::Timeline::Instance().SetListener([weak](const TimelineEntry& entry) {
|
|
if (const std::shared_ptr<State> alive = weak.lock()) alive->OnEntry(entry);
|
|
});
|
|
// A debugger the agent starts is never this process's ancestor,
|
|
// which Yama's ptrace_scope 1 would otherwise require.
|
|
::prctl(PR_SET_PTRACER, PR_SET_PTRACER_ANY, 0, 0, 0);
|
|
if (state->options.listen) state->Listen(state);
|
|
} catch (...) {
|
|
state->registration.reset();
|
|
state->server.reset();
|
|
state->Shutdown();
|
|
current.store(nullptr);
|
|
throw;
|
|
}
|
|
}
|
|
|
|
Session::~Session() {
|
|
// Out of the listing first, so no agent picks an instance on its way
|
|
// out; then every waiting request answered; then the connections
|
|
// closed, which lets those answers leave first.
|
|
state->registration.reset();
|
|
state->Shutdown();
|
|
state->server.reset();
|
|
current.store(nullptr);
|
|
}
|
|
|
|
void Session::Handle(const Value& request, std::shared_ptr<Detail::Exchange> exchange) {
|
|
state->Handle(request, exchange);
|
|
}
|
|
|
|
const std::string& Session::Token() const {
|
|
return state->token;
|
|
}
|
|
|
|
void Crafter::Debug::Pump() {
|
|
if (Session::State* session = current.load()) session->Pump();
|
|
}
|
|
|
|
bool Crafter::Debug::Advance() {
|
|
Session::State* session = current.load();
|
|
if (session == nullptr) return true;
|
|
session->pauseSupported.store(true, std::memory_order_relaxed);
|
|
if (!session->paused.load()) return true;
|
|
std::uint32_t available = session->steps.load();
|
|
while (available > 0) {
|
|
if (session->steps.compare_exchange_weak(available, available - 1)) return true;
|
|
}
|
|
return false;
|
|
}
|
|
|
|
bool Crafter::Debug::Paused() {
|
|
Session::State* session = current.load();
|
|
if (session == nullptr) return false;
|
|
session->pauseSupported.store(true, std::memory_order_relaxed);
|
|
return session->paused.load();
|
|
}
|
|
|
|
std::uint32_t Crafter::Debug::TakeSteps() {
|
|
Session::State* session = current.load();
|
|
if (session == nullptr) return 0;
|
|
session->pauseSupported.store(true, std::memory_order_relaxed);
|
|
return session->steps.exchange(0);
|
|
}
|
|
|
|
bool Crafter::Debug::QuitRequested() {
|
|
Session::State* session = current.load();
|
|
return session != nullptr && session->quitRequested.load();
|
|
}
|
|
|
|
void Crafter::Debug::SetCapture(std::function<Async<std::vector<std::byte>>()> capture) {
|
|
std::lock_guard lock(captureMutex);
|
|
captureSource = std::move(capture);
|
|
}
|