feat(transport): serve the session over QUIC on loopback
A Session with listen = true (the default) now listens on 127.0.0.1 with the ALPN crafter-debug/1 and writes its discovery record (port, token, protocol, source id, client path), which goes first when it ends. One request per stream the client opens: the request and a FIN in, zero or more stream lines and then the reply and a FIN out, one JSON object per line. It holds no thread while nothing arrives. Crafter.Network delivers a new stream on the pool, its bytes on msquic's worker through OnRecieve, and replies leave through SendAsync, which never waits. The handler, which takes the session's mutex, runs on a pool task rather than inside the stream's callback, so no msquic worker ever waits on the session, and the session's threads only send, which Network allows under any lock. Stream wrappers are let go outside every lock their callbacks take. The rules head the implementation. Limits: a request over 1 MiB is answered with an error, not buffered, and a log follower more than 8 MiB behind is reset. quit --force exits 200 ms after its reply was taken, from a thread of its own, since "sent" means msquic has the bytes, not that they left. A client killed without closing is let go at the 10 s idle timeout. The Session refuses a pool without a worker: Network hands it new streams there, and with Start(1) nothing would run them. The partition exports nothing, so an app's view of the module is unchanged and it does not see Crafter.Network's types. Crafter.Network comes from forgejo, unpinned, with the args unchanged; --local takes ../Crafter.Network, and $CRAFTER_DEBUG_NETWORK any project.cpp, for the branch the callback API is on until it lands. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
parent
d75f935bf2
commit
47f2325664
5 changed files with 416 additions and 3 deletions
|
|
@ -7,11 +7,13 @@ module;
|
|||
#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;
|
||||
|
|
@ -285,6 +287,11 @@ namespace Crafter::Debug {
|
|||
|
||||
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;
|
||||
}
|
||||
|
|
@ -296,6 +303,30 @@ namespace Crafter::Debug {
|
|||
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()));
|
||||
|
|
@ -1033,14 +1064,23 @@ Session::Session(SessionOptions options) : state(std::make_shared<State>(std::mo
|
|||
// 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);
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Reference in a new issue