From 1daf12043133d7c900d9d708e83c5927db2e4cab Mon Sep 17 00:00:00 2001 From: Raphael Maenle Date: Wed, 10 Jun 2026 19:33:50 +0200 Subject: [PATCH] =?UTF-8?q?feat(daemon):=20GetVariables=20+=20read=5Fsync?= =?UTF-8?q?=20=E2=80=94=20the=20standard=20mutable-read=20pattern?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit EquipmentRuntime::read_sync establishes THE pattern for reading mutable engine state from gRPC/binding threads (Phase 0 item 6): post the read onto the io thread (the model's single owner), wait on a future with a deadline, nullopt => UNAVAILABLE at the RPC edge. Always truthful, no cache to invalidate; milliseconds are irrelevant at SECS rates. GetVariables: name resolution against the service snapshot (empty query = all; unknown name => INVALID_ARGUMENT naming the offender), values read via read_sync, converted by the new from_item reverse conversion (single-element numeric arrays => scalars, multi-element => List; Boolean/Binary/text per format; C2-as-integer and U8>2^63 wrap documented as TODOs). Tests run the engine in run_async — the daemon's PRODUCTION threading mode, previously untested — and round-trip through both conversions: SetVariables (declared-format write) then GetVariables (read) over a real in-process channel. Daemon suite 41 -> 61 assertions. daemon_interop.py gains a live GetVariables round-trip check vs the running daemon (verified green). Co-Authored-By: Claude Fable 5 --- apps/equipment_service.hpp | 101 ++++++++++++++++++++++++++++++++ docs/DAEMON_ROADMAP.md | 15 +++-- include/secsgem/gem/runtime.hpp | 26 ++++++++ interop/daemon_interop.py | 6 ++ tests/test_daemon_service.cpp | 62 ++++++++++++++++++++ 5 files changed, 204 insertions(+), 6 deletions(-) diff --git a/apps/equipment_service.hpp b/apps/equipment_service.hpp index 995955c..17bf6ec 100644 --- a/apps/equipment_service.hpp +++ b/apps/equipment_service.hpp @@ -15,6 +15,8 @@ #include #include #include +#include +#include #include "secsgem/gem/runtime.hpp" #include "secsgem/secs2/item.hpp" @@ -78,6 +80,68 @@ inline bool value_is_set(const pb::Value& v) { return v.kind_case() != pb::Value::KIND_NOT_SET; } +// SECS-II Item -> proto Value (the read direction). Single-element numeric +// arrays — the overwhelmingly common case — become scalars; multi-element +// arrays become a List so nothing is lost. +// TODO(daemon): C2 (2-byte Unicode) currently surfaces as integer code +// points (it shares storage with U2); convert to text when a tool needs it. +// TODO(daemon): U8 values above 2^63-1 wrap through the sint64 integer +// field; switch Value to a dedicated uint64 arm if that ever bites. +inline pb::Value from_item(const s2::Item& item) { + pb::Value out; + switch (item.format()) { + case s2::Format::List: { + auto* list = out.mutable_list(); + for (const auto& child : item.as_list()) *list->add_items() = from_item(child); + return out; + } + case s2::Format::ASCII: + case s2::Format::JIS8: + out.set_text(item.as_ascii()); + return out; + case s2::Format::Binary: { + const auto& b = item.as_bytes(); + out.set_binary(std::string(b.begin(), b.end())); + return out; + } + case s2::Format::Boolean: { + const auto& b = item.as_bytes(); + if (b.size() == 1) { + out.set_boolean(b[0] != 0); + } else { + auto* list = out.mutable_list(); + for (auto v : b) list->add_items()->set_boolean(v != 0); + } + return out; + } + default: + break; // numeric formats: handled generically below + } + return std::visit( + [&](const auto& vec) -> pb::Value { + using V = std::decay_t; + if constexpr (std::is_same_v || + std::is_same_v) { + return out; // unreachable: handled above + } else { + auto set_one = [](pb::Value& dst, auto v) { + if constexpr (std::is_floating_point_v) + dst.set_real(static_cast(v)); + else + dst.set_integer(static_cast(v)); + }; + if (vec.size() == 1) { + set_one(out, vec[0]); + } else { + auto* list = out.mutable_list(); + for (auto v : vec) set_one(*list->add_items(), v); + } + return out; + } + }, + item.storage()); +} + inline pb::ControlState::State to_proto_state(gem::ControlState s) { switch (s) { case gem::ControlState::EquipmentOffline: return pb::ControlState::EQUIPMENT_OFFLINE; @@ -150,6 +214,43 @@ class EquipmentService final : public pb::Equipment::Service { return grpc::Status::OK; } + grpc::Status GetVariables(grpc::ServerContext*, const pb::VariableQuery* req, + pb::VariableSnapshot* resp) override { + // Resolve names against the snapshot maps (empty query = everything). + std::vector> wanted; + if (req->names().empty()) { + wanted.reserve(vars_.size()); + for (const auto& [name, ref] : vars_) wanted.emplace_back(name, ref.vid); + } else { + for (const auto& name : req->names()) { + auto it = vars_.find(name); + if (it == vars_.end()) + return grpc::Status(grpc::StatusCode::INVALID_ARGUMENT, + "no variable named '" + name + "'"); + wanted.emplace_back(name, it->second.vid); + } + } + // Mutable values live on the io thread — read them there (read_sync is + // the standard pattern; see runtime.hpp). + std::vector vids; + vids.reserve(wanted.size()); + for (const auto& w : wanted) vids.push_back(w.second); + auto values = rt_.read_sync([this, vids]() { + std::vector> out; + out.reserve(vids.size()); + for (auto vid : vids) out.push_back(rt_.model().vid_value(vid)); + return out; + }); + if (!values) + return grpc::Status(grpc::StatusCode::UNAVAILABLE, + "engine io thread did not answer (not running?)"); + for (std::size_t i = 0; i < wanted.size(); ++i) { + if (!(*values)[i]) continue; // vid vanished — defensive, can't happen + (*resp->mutable_values())[wanted[i].first] = from_item(*(*values)[i]); + } + return grpc::Status::OK; + } + grpc::Status GetControlState(grpc::ServerContext*, const pb::Empty*, pb::ControlState* resp) override { // Thread-safe: control_state() reads the runtime's atomic mirror. diff --git a/docs/DAEMON_ROADMAP.md b/docs/DAEMON_ROADMAP.md index 48466a4..4e70e33 100644 --- a/docs/DAEMON_ROADMAP.md +++ b/docs/DAEMON_ROADMAP.md @@ -113,9 +113,9 @@ debts tax every later phase, and the most valuable tests aren't automated. (SVIDs 1/2 `refresh()`, CEIDs 400/401) with YAML role bindings (`control_state_svid:`, `cj_executing_ceid:` …). Gradual; aligns with the capability structure GEM itself defines (S1F19) and enables vendor subsetting. -6. ⬜ **Standardize the mutable-read pattern** for daemon RPCs: post-to-io + - future with deadline (always truthful; latency irrelevant at SECS rates). - First consumer: `GetVariables` (Phase A1) — set the precedent there. +6. ✅ **Standardize the mutable-read pattern** — `EquipmentRuntime::read_sync` + (post-to-io + future with deadline; nullopt => UNAVAILABLE at the RPC edge). + Precedent set by `GetVariables`; every future mutable read copies it. 7. ⬜ Move `apps/equipment_service.hpp` into the library tree (`include/secsgem/daemon/`) once Phase B grows it; add a TSan-built `run_async` + concurrent-RPC daemon test (today's daemon tests only poll()). @@ -126,9 +126,12 @@ debts tax every later phase, and the most valuable tests aren't automated. RPC edge (was silently writing ASCII ""). ### Phase A — finish the universal daemon surface (small, unblock vendors) -1. ⬜ `GetVariables` — needs the reverse `Item → proto Value` conversion - (read via post-to-io + future, or serve from a daemon-side cache of last - set values; decide and document). +1. ✅ `GetVariables` — `from_item` reverse conversion (scalar for 1-element + arrays, List otherwise; C2-as-text and U8>2^63 noted as TODOs) + reads via + `read_sync`. Tested under **run_async (production threading)** — write + through the API, read back through the API — plus empty-query-returns-all, + INVALID_ARGUMENT on unknown names, and a live round-trip check in + `daemon_interop.py`. 2. ⬜ Alarm `name:` config field + `SetAlarm`/`ClearAlarm` RPCs + tests. 3. ⬜ `RequestControlState` (operator online/offline) + control-state atomic mirror (fixes the known race) + `WatchHealth` stream (link state from the diff --git a/include/secsgem/gem/runtime.hpp b/include/secsgem/gem/runtime.hpp index 236ba4f..4ef5695 100644 --- a/include/secsgem/gem/runtime.hpp +++ b/include/secsgem/gem/runtime.hpp @@ -2,12 +2,15 @@ #include #include +#include #include #include +#include #include #include #include #include +#include #include #include @@ -72,6 +75,29 @@ class EquipmentRuntime { void set_alarm(uint32_t alid); void clear_alarm(uint32_t alid); + // ---- reading mutable engine state from outside the io thread ------------- + // THE standard pattern for every read of mutable state (variable values, + // alarm activity, spool depth, ...) from gRPC/binding threads: post the + // read onto the io thread — the model's single owner — and wait with a + // deadline. Always truthful (no cache invalidation), and the milliseconds + // of latency are irrelevant at SECS message rates. Returns nullopt if the + // io thread doesn't service the post in time (not running, or stalled). + // Requires run()/run_async(); in poll() mode there is nobody to serve the + // post, so a same-thread caller should read the model directly instead. + template + auto read_sync(Fn&& fn, + std::chrono::milliseconds timeout = std::chrono::milliseconds(2000)) + -> std::optional&>> { + using R = std::invoke_result_t&>; + auto prom = std::make_shared>(); + auto fut = prom->get_future(); + asio::post(io_, [prom, fn = std::forward(fn)]() mutable { + prom->set_value(fn()); + }); + if (fut.wait_for(timeout) != std::future_status::ready) return std::nullopt; + return fut.get(); + } + // ---- host-command behaviour hook ----------------------------------------- void on_command(std::string rcmd, HostCommandRegistry::Handler h) { model_->commands.set_handler(std::move(rcmd), std::move(h)); diff --git a/interop/daemon_interop.py b/interop/daemon_interop.py index 41180d4..fe86568 100644 --- a/interop/daemon_interop.py +++ b/interop/daemon_interop.py @@ -153,6 +153,12 @@ def run(grpc_addr: str, hsms_host: str, hsms_port: int, session_id: int, check("gRPC SetVariables(ChamberPressure=2.5) -> ACCEPT", ack.code == pb.Ack.ACCEPT, pb.Ack.Code.Name(ack.code)) + # Read back through the API: SetVariables -> engine -> GetVariables. + snap = stub.GetVariables(pb.VariableQuery(names=["ChamberPressure"])) + got = snap.values["ChamberPressure"].real + check("gRPC GetVariables round-trips ChamberPressure=2.5", + abs(got - 2.5) < 0.01, f"got {got}") + ack = stub.FireEvent(pb.Event(name="ProcessStarted")) check("gRPC FireEvent(ProcessStarted) -> ACCEPT", ack.code == pb.Ack.ACCEPT, pb.Ack.Code.Name(ack.code)) diff --git a/tests/test_daemon_service.cpp b/tests/test_daemon_service.cpp index 6c41ccc..2acaa28 100644 --- a/tests/test_daemon_service.cpp +++ b/tests/test_daemon_service.cpp @@ -146,3 +146,65 @@ TEST_CASE("Equipment gRPC service over an in-process channel") { server->Shutdown(); } + +// GetVariables needs a live io thread (read_sync posts onto it), so this case +// runs the engine in run_async() — the daemon's PRODUCTION threading mode — +// with real concurrency between the gRPC handler thread and the io thread. +TEST_CASE("GetVariables round-trip under run_async (production threading mode)") { + gem::EquipmentRuntime rt(test_config()); + dmn::EquipmentService svc(rt); // snapshot BEFORE the io thread starts + rt.run_async(); + + grpc::ServerBuilder builder; + builder.RegisterService(&svc); + std::unique_ptr server(builder.BuildAndStart()); + REQUIRE(server); + auto stub = pb::Equipment::NewStub(server->InProcessChannel(grpc::ChannelArguments{})); + + // Write through the API, then read back through the API: exercises BOTH + // conversions (Value->Item with declared formats, Item->Value) end to end. + { + grpc::ClientContext ctx; + pb::VariableUpdate req; + pb::Ack resp; + (*req.mutable_values())["ChamberPressure"].set_real(2.5); + (*req.mutable_values())["WaferCounter"].set_integer(7); + REQUIRE(stub->SetVariables(&ctx, req, &resp).ok()); + REQUIRE(resp.code() == pb::Ack::ACCEPT); + } + { + grpc::ClientContext ctx; + pb::VariableQuery req; + pb::VariableSnapshot resp; + req.add_names("ChamberPressure"); + req.add_names("WaferCounter"); + auto st = stub->GetVariables(&ctx, req, &resp); + REQUIRE(st.ok()); + REQUIRE(resp.values().count("ChamberPressure") == 1); + REQUIRE(resp.values().count("WaferCounter") == 1); + CHECK(resp.values().at("ChamberPressure").real() == doctest::Approx(2.5)); + CHECK(resp.values().at("WaferCounter").integer() == 7); + } + + SUBCASE("empty query returns every configured variable") { + grpc::ClientContext ctx; + pb::VariableQuery req; + pb::VariableSnapshot resp; + REQUIRE(stub->GetVariables(&ctx, req, &resp).ok()); + const auto expected = rt.model().svids.size() + rt.model().dvids.all().size(); + CHECK(resp.values().size() == expected); + } + + SUBCASE("unknown name is INVALID_ARGUMENT, naming the offender") { + grpc::ClientContext ctx; + pb::VariableQuery req; + pb::VariableSnapshot resp; + req.add_names("definitely_not_a_var"); + auto st = stub->GetVariables(&ctx, req, &resp); + CHECK(st.error_code() == grpc::StatusCode::INVALID_ARGUMENT); + CHECK(st.error_message().find("definitely_not_a_var") != std::string::npos); + } + + server->Shutdown(); + rt.stop(); +}