feat(daemon): GetVariables + read_sync — the standard mutable-read pattern
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 <noreply@anthropic.com>
This commit is contained in:
@@ -15,6 +15,8 @@
|
|||||||
#include <cstdint>
|
#include <cstdint>
|
||||||
#include <map>
|
#include <map>
|
||||||
#include <string>
|
#include <string>
|
||||||
|
#include <utility>
|
||||||
|
#include <vector>
|
||||||
|
|
||||||
#include "secsgem/gem/runtime.hpp"
|
#include "secsgem/gem/runtime.hpp"
|
||||||
#include "secsgem/secs2/item.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;
|
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<decltype(vec)>;
|
||||||
|
if constexpr (std::is_same_v<V, s2::Item::List> ||
|
||||||
|
std::is_same_v<V, std::string>) {
|
||||||
|
return out; // unreachable: handled above
|
||||||
|
} else {
|
||||||
|
auto set_one = [](pb::Value& dst, auto v) {
|
||||||
|
if constexpr (std::is_floating_point_v<decltype(v)>)
|
||||||
|
dst.set_real(static_cast<double>(v));
|
||||||
|
else
|
||||||
|
dst.set_integer(static_cast<int64_t>(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) {
|
inline pb::ControlState::State to_proto_state(gem::ControlState s) {
|
||||||
switch (s) {
|
switch (s) {
|
||||||
case gem::ControlState::EquipmentOffline: return pb::ControlState::EQUIPMENT_OFFLINE;
|
case gem::ControlState::EquipmentOffline: return pb::ControlState::EQUIPMENT_OFFLINE;
|
||||||
@@ -150,6 +214,43 @@ class EquipmentService final : public pb::Equipment::Service {
|
|||||||
return grpc::Status::OK;
|
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<std::pair<std::string, uint32_t>> 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<uint32_t> vids;
|
||||||
|
vids.reserve(wanted.size());
|
||||||
|
for (const auto& w : wanted) vids.push_back(w.second);
|
||||||
|
auto values = rt_.read_sync([this, vids]() {
|
||||||
|
std::vector<std::optional<s2::Item>> 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*,
|
grpc::Status GetControlState(grpc::ServerContext*, const pb::Empty*,
|
||||||
pb::ControlState* resp) override {
|
pb::ControlState* resp) override {
|
||||||
// Thread-safe: control_state() reads the runtime's atomic mirror.
|
// Thread-safe: control_state() reads the runtime's atomic mirror.
|
||||||
|
|||||||
@@ -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
|
(SVIDs 1/2 `refresh()`, CEIDs 400/401) with YAML role bindings
|
||||||
(`control_state_svid:`, `cj_executing_ceid:` …). Gradual; aligns with the
|
(`control_state_svid:`, `cj_executing_ceid:` …). Gradual; aligns with the
|
||||||
capability structure GEM itself defines (S1F19) and enables vendor subsetting.
|
capability structure GEM itself defines (S1F19) and enables vendor subsetting.
|
||||||
6. ⬜ **Standardize the mutable-read pattern** for daemon RPCs: post-to-io +
|
6. ✅ **Standardize the mutable-read pattern** — `EquipmentRuntime::read_sync`
|
||||||
future with deadline (always truthful; latency irrelevant at SECS rates).
|
(post-to-io + future with deadline; nullopt => UNAVAILABLE at the RPC edge).
|
||||||
First consumer: `GetVariables` (Phase A1) — set the precedent there.
|
Precedent set by `GetVariables`; every future mutable read copies it.
|
||||||
7. ⬜ Move `apps/equipment_service.hpp` into the library tree
|
7. ⬜ Move `apps/equipment_service.hpp` into the library tree
|
||||||
(`include/secsgem/daemon/`) once Phase B grows it; add a TSan-built
|
(`include/secsgem/daemon/`) once Phase B grows it; add a TSan-built
|
||||||
`run_async` + concurrent-RPC daemon test (today's daemon tests only poll()).
|
`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 "").
|
RPC edge (was silently writing ASCII "").
|
||||||
|
|
||||||
### Phase A — finish the universal daemon surface (small, unblock vendors)
|
### Phase A — finish the universal daemon surface (small, unblock vendors)
|
||||||
1. ⬜ `GetVariables` — needs the reverse `Item → proto Value` conversion
|
1. ✅ `GetVariables` — `from_item` reverse conversion (scalar for 1-element
|
||||||
(read via post-to-io + future, or serve from a daemon-side cache of last
|
arrays, List otherwise; C2-as-text and U8>2^63 noted as TODOs) + reads via
|
||||||
set values; decide and document).
|
`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.
|
2. ⬜ Alarm `name:` config field + `SetAlarm`/`ClearAlarm` RPCs + tests.
|
||||||
3. ⬜ `RequestControlState` (operator online/offline) + control-state atomic
|
3. ⬜ `RequestControlState` (operator online/offline) + control-state atomic
|
||||||
mirror (fixes the known race) + `WatchHealth` stream (link state from the
|
mirror (fixes the known race) + `WatchHealth` stream (link state from the
|
||||||
|
|||||||
@@ -2,12 +2,15 @@
|
|||||||
|
|
||||||
#include <asio.hpp>
|
#include <asio.hpp>
|
||||||
#include <atomic>
|
#include <atomic>
|
||||||
|
#include <chrono>
|
||||||
#include <cstdint>
|
#include <cstdint>
|
||||||
#include <functional>
|
#include <functional>
|
||||||
|
#include <future>
|
||||||
#include <memory>
|
#include <memory>
|
||||||
#include <optional>
|
#include <optional>
|
||||||
#include <string>
|
#include <string>
|
||||||
#include <thread>
|
#include <thread>
|
||||||
|
#include <type_traits>
|
||||||
#include <utility>
|
#include <utility>
|
||||||
#include <vector>
|
#include <vector>
|
||||||
|
|
||||||
@@ -72,6 +75,29 @@ class EquipmentRuntime {
|
|||||||
void set_alarm(uint32_t alid);
|
void set_alarm(uint32_t alid);
|
||||||
void clear_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 <typename Fn>
|
||||||
|
auto read_sync(Fn&& fn,
|
||||||
|
std::chrono::milliseconds timeout = std::chrono::milliseconds(2000))
|
||||||
|
-> std::optional<std::invoke_result_t<std::decay_t<Fn>&>> {
|
||||||
|
using R = std::invoke_result_t<std::decay_t<Fn>&>;
|
||||||
|
auto prom = std::make_shared<std::promise<R>>();
|
||||||
|
auto fut = prom->get_future();
|
||||||
|
asio::post(io_, [prom, fn = std::forward<Fn>(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 -----------------------------------------
|
// ---- host-command behaviour hook -----------------------------------------
|
||||||
void on_command(std::string rcmd, HostCommandRegistry::Handler h) {
|
void on_command(std::string rcmd, HostCommandRegistry::Handler h) {
|
||||||
model_->commands.set_handler(std::move(rcmd), std::move(h));
|
model_->commands.set_handler(std::move(rcmd), std::move(h));
|
||||||
|
|||||||
@@ -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",
|
check("gRPC SetVariables(ChamberPressure=2.5) -> ACCEPT",
|
||||||
ack.code == pb.Ack.ACCEPT, pb.Ack.Code.Name(ack.code))
|
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"))
|
ack = stub.FireEvent(pb.Event(name="ProcessStarted"))
|
||||||
check("gRPC FireEvent(ProcessStarted) -> ACCEPT",
|
check("gRPC FireEvent(ProcessStarted) -> ACCEPT",
|
||||||
ack.code == pb.Ack.ACCEPT, pb.Ack.Code.Name(ack.code))
|
ack.code == pb.Ack.ACCEPT, pb.Ack.Code.Name(ack.code))
|
||||||
|
|||||||
@@ -146,3 +146,65 @@ TEST_CASE("Equipment gRPC service over an in-process channel") {
|
|||||||
|
|
||||||
server->Shutdown();
|
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<grpc::Server> 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();
|
||||||
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user