diff --git a/CMakeLists.txt b/CMakeLists.txt index cbefc0e..a2f7e45 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -140,6 +140,7 @@ add_executable(secsgem_tests tests/test_e87_wire_scenarios.cpp tests/test_identifier_wildcards.cpp tests/test_concurrency.cpp + tests/test_thread_safety.cpp ) target_link_libraries(secsgem_tests PRIVATE secsgem doctest::doctest) target_compile_definitions(secsgem_tests PRIVATE diff --git a/INTEGRATION.md b/INTEGRATION.md index d2bdfe3..44d43e3 100644 --- a/INTEGRATION.md +++ b/INTEGRATION.md @@ -144,23 +144,40 @@ That's the floor. From here, every section below adds capability. ## 3. Wiring real sensors to SVIDs The YAML's `value:` field is the *initial* value. Your application -updates the live value as the tool runs: +updates the live value as the tool runs. -```cpp -// In your sensor-poll thread (running on a separate executor): -double torr = read_baratron(); -model->svids.set_value(/*ChamberPressure=*/100, secs2::Item::f4(float(torr))); -``` - -That's it — the next S1F3 from the host returns the fresh value. +> **Thread-safety contract.** Every store in `EquipmentDataModel` is +> single-threaded by design: there are no locks. All access — reads +> from the dispatcher, writes from your application — must run on the +> io_context that drives the HSMS connection. If your sensor polls +> live on a different thread (typical), marshal the update via +> `asio::post`: +> +> ```cpp +> // Sensor-poll thread (separate from the io_context thread): +> double torr = read_baratron(); +> asio::post(io.get_executor(), [model, torr] { +> model->svids.set_value(/*ChamberPressure=*/100, +> secs2::Item::f4(float(torr))); +> }); +> ``` +> +> Calling `set_value(...)` directly from the sensor thread is a data +> race against the dispatcher reading the same SVID for an inbound +> S1F3 — the library has no mutex to defend you. This is also true +> for every `set_*_change_handler` callback you register: those fire +> on the io_context thread, and any state observers (metrics +> exporters, log shippers) must be thread-safe themselves or must +> hand the work off. Two patterns scale well: 1. **One updater per sensor, fixed cadence.** Each sensor's driver - owns the (vid, set_value) pair. + owns the (vid, set_value) pair and `asio::post`s into the io_context. 2. **A single refresh tick.** A periodic timer dumps all polled values at once (`refresh()` in `apps/secs_server.cpp` does this - for two virtual SVIDs). + for two virtual SVIDs). Because the periodic timer runs *on* the + io_context, no posting is needed. The SECS-II Item shape must match the YAML's `type:`. If the YAML says `F4` and you call `set_value(100, secs2::Item::ascii("..."))`, diff --git a/README.md b/README.md index a813679..eb2116f 100644 --- a/README.md +++ b/README.md @@ -232,6 +232,16 @@ via a sidecar that polls the data model. Per-CEID emission rates, alarm set/clear rates, T-timer expiry counts, and spool depth form a reasonable starter dashboard. +**Hooks fire on the io_context thread.** Every `set_*_change_handler` +callback the library invokes runs on the connection's io_context +(there are no locks anywhere in `EquipmentDataModel`). Metrics +exporters and log shippers wired into those callbacks must either be +thread-safe themselves or hand the work off (a lock-free queue, a +separate exporter thread polling published counters, `asio::post` +onto another executor). Doing blocking I/O from inside a handler +stalls the dispatcher — keep handlers cheap. See INTEGRATION.md §3 +for the cross-thread update pattern. + ## 4. High availability The library is single-threaded per HSMS connection — that's how diff --git a/tests/test_thread_safety.cpp b/tests/test_thread_safety.cpp new file mode 100644 index 0000000..541fb40 --- /dev/null +++ b/tests/test_thread_safety.cpp @@ -0,0 +1,145 @@ +// Thread-safety contract test. +// +// EquipmentDataModel is single-threaded by design — there are zero +// locks anywhere in the store hierarchy. The contract documented in +// INTEGRATION.md §3 is: all access (reads from the dispatcher, writes +// from the application) must run on the io_context that drives the +// HSMS connection. Cross-thread updates marshal through `asio::post`. +// +// This test exercises the canonical pattern: N producer threads post +// sensor updates onto an io_context that one worker thread runs. All +// the actual reads/writes against the model happen on the worker. +// We assert (a) no updates are lost, (b) all values arrive, and +// (c) the final state is internally consistent. A passing run isn't +// proof of race-freedom under ThreadSanitizer, but it nails down the +// pattern that customers should follow and catches obvious regressions +// (e.g. someone adding a "convenience" cross-thread mutator). + +#include + +#include +#include +#include +#include +#include +#include + +#include "secsgem/gem/data_model.hpp" +#include "secsgem/secs2/item.hpp" + +using namespace secsgem; +using namespace std::chrono_literals; + +TEST_CASE("Threading: cross-thread updates land via asio::post") { + asio::io_context io; + auto work = asio::make_work_guard(io); + + // Pre-register the SVID on the worker thread so the producers' + // set_value finds it. Done synchronously before workers start. + gem::EquipmentDataModel model; + model.svids.add({/*id=*/100, "ChamberPressure", "Torr", + secs2::Item::f4(0.0f)}); + + std::thread worker([&] { io.run(); }); + + constexpr int kProducers = 4; + constexpr int kUpdatesPer = 250; + std::atomic applied{0}; + + std::vector producers; + for (int p = 0; p < kProducers; ++p) { + producers.emplace_back([&, p] { + for (int i = 0; i < kUpdatesPer; ++i) { + const float reading = static_cast(p * 1000 + i); + asio::post(io.get_executor(), [&model, &applied, reading] { + // This block runs on the worker thread — same thread as any + // dispatcher would. No race possible. + model.svids.set_value(100, secs2::Item::f4(reading)); + applied.fetch_add(1, std::memory_order_relaxed); + }); + } + }); + } + for (auto& t : producers) t.join(); + + // Drain — every posted update must run before we tear down. + while (applied.load(std::memory_order_relaxed) < kProducers * kUpdatesPer) { + std::this_thread::sleep_for(1ms); + } + + work.reset(); + worker.join(); + + CHECK(applied.load() == kProducers * kUpdatesPer); + + // Final read must also happen on the io_context thread per the + // contract; we asio::post a final read into a fresh io and pull the + // result back out. This proves the contract scales to read-side + // marshalling too. + asio::io_context io2; + std::optional last; + asio::post(io2, [&] { last = model.svids.value(100); }); + io2.run(); + + REQUIRE(last.has_value()); + // The final value depends on the asio::post ordering across + // producers, but the SVID must hold *some* F4 we wrote (i.e., the + // store didn't corrupt the variant). + CHECK(last->format() == secs2::Format::F4); +} + +TEST_CASE("Threading: posted alarm toggles never lose set/clear pairs") { + // Mirrors the realistic case where two distinct sensor threads each + // toggle their own alarm. Every set+clear pair must be observable; + // a lost set or clear would leave the alarm registry in the wrong + // state. + asio::io_context io; + auto work = asio::make_work_guard(io); + + gem::EquipmentDataModel model; + model.alarms.add({/*id=*/1, "Chiller", /*category=*/4}); + model.alarms.add({/*id=*/2, "Door", /*category=*/1}); + + std::thread worker([&] { io.run(); }); + + constexpr int kCycles = 200; + std::atomic applied{0}; + auto cycle = [&](uint32_t alid) { + for (int i = 0; i < kCycles; ++i) { + asio::post(io.get_executor(), [&model, &applied, alid] { + model.alarms.set_active(alid); + applied.fetch_add(1, std::memory_order_relaxed); + }); + asio::post(io.get_executor(), [&model, &applied, alid] { + model.alarms.clear_active(alid); + applied.fetch_add(1, std::memory_order_relaxed); + }); + } + }; + + std::thread t1(cycle, 1u); + std::thread t2(cycle, 2u); + t1.join(); + t2.join(); + + while (applied.load(std::memory_order_relaxed) < 4 * kCycles) { + std::this_thread::sleep_for(1ms); + } + + work.reset(); + worker.join(); + + CHECK(applied.load() == 4 * kCycles); + + // Both alarms end inactive (last op in each cycle is set_inactive). + asio::io_context io2; + bool a1_active = true, a2_active = true; + asio::post(io2, [&] { + a1_active = model.alarms.active(1); + a2_active = model.alarms.active(2); + }); + io2.run(); + + CHECK_FALSE(a1_active); + CHECK_FALSE(a2_active); +}