diff --git a/.github/workflows/esp32p4-smv-injector.yml b/.github/workflows/esp32p4-smv-injector.yml index 28d37bd06..560f01558 100644 --- a/.github/workflows/esp32p4-smv-injector.yml +++ b/.github/workflows/esp32p4-smv-injector.yml @@ -179,7 +179,11 @@ jobs: assert 'esp_eth_transmit' not in smp_synch assert 'observe_canonical_tx_result' not in smp_synch start_begin = source.index('if (take_start_request())') - start_end = source.index('if (!live_tx_running())', start_begin) + # Match only the publisher's top-level idle branch (8-space + # indent). Guards inside START intentionally test live_tx_running() + # and must remain inside this regression boundary. + start_end = source.index('\n if (!live_tx_running()) {', start_begin) + assert 'live_control_force_stop();' in source[start_begin:start_end] start_block = source[start_begin:start_end] # M3 passes the ETH/PTP epoch context as additional arguments. # Assert that the timer is still armed after the final console log. diff --git a/CMakeLists.txt b/CMakeLists.txt index fcb550a5e..148e0500b 100644 --- a/CMakeLists.txt +++ b/CMakeLists.txt @@ -274,7 +274,8 @@ if(ARIEC61850_BUILD_TESTS) endforeach() add_executable(ariec61850_live_signal_tests tests/test_live_signal.cpp) - target_link_libraries(ariec61850_live_signal_tests PRIVATE ARIEC61850::core) + find_package(Threads REQUIRED) + target_link_libraries(ariec61850_live_signal_tests PRIVATE ARIEC61850::core Threads::Threads) target_compile_features(ariec61850_live_signal_tests PRIVATE cxx_std_20) ariec61850_apply_warnings(ariec61850_live_signal_tests) ariec61850_apply_sanitizers(ariec61850_live_signal_tests) diff --git a/apps/arstack_studio/qml/Main.qml b/apps/arstack_studio/qml/Main.qml index 30a7400ef..7b31cfcb5 100644 --- a/apps/arstack_studio/qml/Main.qml +++ b/apps/arstack_studio/qml/Main.qml @@ -396,7 +396,7 @@ ApplicationWindow { } function validFrequency(value) { - return isFinite(value) && value >= 0 && value <= 1000.0 + return isFinite(value) && (value === 0 || (value >= 1.0 && value <= 1000.0)) } function normalizedAngle(value) { @@ -498,8 +498,15 @@ ApplicationWindow { function applyAllSignals() { if (!workflowBar.session || !workflowBar.session.liveControlReady) return false var ok = workflowBar.session.requestSetFrequency(signalFrequency) - for (var group = 0; group < 2; ++group) - for (var row = 0; row < 4; ++row) ok = sendSignal(group, row) && ok + for (var group = 0; group < 2; ++group) { + for (var row = 0; row < 4; ++row) { + var signal = groupModel(group).get(row) + ok = sendSignal(group, row) && ok + // Enable flags are part of the visible 4I+4V table and must + // not silently retain values from an earlier START. + ok = workflowBar.session.requestSetEnabled(signal.signalId, signal.enabled) && ok + } + } ok = workflowBar.session.requestSetCtSaturation(ctSaturationEnabled, ctDcOffsetPercent, ctHarmonicPercent, ctHarmonicOrder, ctClipPercent) && ok pendingApply = !ok @@ -524,6 +531,17 @@ ApplicationWindow { return draft } + function queueLiveDraft(sendCommand, errorMessage) { + // markScenarioDirty() always sets pendingApply=true. Checking + // !pendingApply here would make every LIVE edit silently do nothing. + if (!workflowBar.injectionRunning || !workflowBar.session || + !workflowBar.session.liveControlReady) return false + var queued = sendCommand() + pendingApply = !queued // QUEUED only, never independent wire proof + if (!queued) showMessage(errorMessage, true) + return queued + } + function setFrequencyValue(value) { if (!validFrequency(value)) return false signalFrequency = value @@ -532,8 +550,8 @@ ApplicationWindow { frequencyField.invalidInput = false refreshPreview() markScenarioDirty() - if (workflowBar.session && workflowBar.session.liveControlReady && !pendingApply) - workflowBar.session.requestSetFrequency(value) + queueLiveDraft(function() { return workflowBar.session.requestSetFrequency(value) }, + "Frequency not queued; press Stop/Start to restore the GUI scenario.") return true } @@ -557,9 +575,10 @@ ApplicationWindow { ctSaturationEnabled = enabled refreshPreview() markScenarioDirty() - if (workflowBar.session && workflowBar.session.liveControlReady && !pendingApply) - workflowBar.session.requestSetCtSaturation(enabled, ctDcOffsetPercent, - ctHarmonicPercent, ctHarmonicOrder, ctClipPercent) + queueLiveDraft(function() { + return workflowBar.session.requestSetCtSaturation(enabled, ctDcOffsetPercent, + ctHarmonicPercent, ctHarmonicOrder, ctClipPercent) + }, "CT shaping not queued; press Stop/Start to restore the GUI scenario.") showMessage(enabled ? "CT saturation stress enabled ยท DC offset + 2nd harmonic + clipping approximation." : "CT saturation stress disabled.", false) @@ -575,11 +594,10 @@ ApplicationWindow { for (var i = 0; i < 3; ++i) model.setProperty(i, "magnitude", value) selectSignal(group, row) markScenarioDirty() - if (workflowBar.session && workflowBar.session.liveControlReady && !pendingApply && !sendLinkedGroup(group)) { - showMessage("ABC update was not accepted; inspect device diagnostics.", true) - return - } - showMessage((pendingApply ? "Local experiment: " : "Device queue: ") + "ABC magnitude queued from " + model.get(row).signalId, false) + var queued = queueLiveDraft(function() { return sendLinkedGroup(group) }, + "ABC magnitude update not queued; retry or Stop/Start.") + if (queued) showMessage("Device queue: ABC magnitude queued from " + model.get(row).signalId, false) + else if (!workflowBar.injectionRunning) showMessage("Local draft: ABC magnitude changed.", false) } function equalGroupAngles(group, row) { @@ -590,11 +608,10 @@ ApplicationWindow { for (var i = 0; i < 3; ++i) model.setProperty(i, "phase", value) selectSignal(group, row) markScenarioDirty() - if (workflowBar.session && workflowBar.session.liveControlReady && !pendingApply && !sendLinkedGroup(group)) { - showMessage("ABC update was not accepted; inspect device diagnostics.", true) - return - } - showMessage((pendingApply ? "Local experiment: " : "Device queue: ") + "ABC angles queued from " + model.get(row).signalId, false) + var queued = queueLiveDraft(function() { return sendLinkedGroup(group) }, + "ABC angles not queued; retry or Stop/Start.") + if (queued) showMessage("Device queue: ABC angles queued from " + model.get(row).signalId, false) + else if (!workflowBar.injectionRunning) showMessage("Local draft: ABC angles changed.", false) } function balanceGroupAngles(group, row) { @@ -608,11 +625,10 @@ ApplicationWindow { model.setProperty(2, "phase", normalizedAngle(base + 120)) selectSignal(group, row) markScenarioDirty() - if (workflowBar.session && workflowBar.session.liveControlReady && !pendingApply && !sendLinkedGroup(group)) { - showMessage("ABC update was not accepted; inspect device diagnostics.", true) - return - } - showMessage((pendingApply ? "Local experiment: " : "Device queue: ") + "Balanced ABC angles queued from " + model.get(row).signalId, false) + var queued = queueLiveDraft(function() { return sendLinkedGroup(group) }, + "Balanced ABC angles not queued; retry or Stop/Start.") + if (queued) showMessage("Device queue: balanced ABC angles queued from " + model.get(row).signalId, false) + else if (!workflowBar.injectionRunning) showMessage("Local draft: balanced ABC angles changed.", false) } function balanced() { @@ -628,7 +644,8 @@ ApplicationWindow { selectSignal(activeGroup, activeRow) refreshPreview() markScenarioDirty() - if (workflowBar.session && workflowBar.session.liveControlReady && !pendingApply) applyAllSignals() + if (workflowBar.injectionRunning && workflowBar.session && workflowBar.session.liveControlReady) + applyAllSignals() } function zeroAll() { @@ -639,8 +656,8 @@ ApplicationWindow { selectSignal(activeGroup, activeRow) refreshPreview() markScenarioDirty() - if (workflowBar.session && workflowBar.session.liveControlReady && !pendingApply) - workflowBar.session.requestZero() + queueLiveDraft(function() { return workflowBar.session.requestZero() }, + "Zero not queued; press Stop/Start to restore the GUI scenario.") } function setActiveQuality(value) { @@ -648,8 +665,9 @@ ApplicationWindow { groupModel(activeGroup).setProperty(activeRow, "quality", unsignedValue) activeQuality = unsignedValue markScenarioDirty() - if (workflowBar.session && workflowBar.session.liveControlReady && !pendingApply) - workflowBar.session.requestSetQuality(activeSignal, unsignedValue) + queueLiveDraft(function() { + return workflowBar.session.requestSetQuality(activeSignal, unsignedValue) + }, "Quality not queued; press Stop/Start to restore the GUI scenario.") } Shortcut { sequence: "Ctrl+B"; enabled: !workflowBar.startQueued; onActivated: root.balanced() } @@ -1043,8 +1061,9 @@ ApplicationWindow { if (value > 0) root.previousAcFrequency = value root.refreshPreview() root.markScenarioDirty() - if (workflowBar.session && workflowBar.session.liveControlReady && !root.pendingApply) - workflowBar.session.requestSetFrequency(value) + root.queueLiveDraft(function() { + return workflowBar.session.requestSetFrequency(value) + }, "Frequency not queued; press Stop/Start to restore the GUI scenario.") } else invalidInput = true } onEditingFinished: { @@ -1052,7 +1071,7 @@ ApplicationWindow { if (!root.validFrequency(value)) { text = root.signalFrequency.toFixed(3) invalidInput = false - root.showMessage("Frequency must be within 0..1000 Hz (0 = DC).", true) + root.showMessage("Frequency must be 0 Hz (DC) or 1..1000 Hz (AC).", true) } else text = value.toFixed(3) } } diff --git a/apps/arstack_studio/qml/SignalMatrix.qml b/apps/arstack_studio/qml/SignalMatrix.qml index ac323221d..91f5a50aa 100644 --- a/apps/arstack_studio/qml/SignalMatrix.qml +++ b/apps/arstack_studio/qml/SignalMatrix.qml @@ -283,8 +283,9 @@ Rectangle { matrix.controller.selectSignal(matrix.groupIndex, signalRow.rowIndex) matrix.controller.refreshPreview() matrix.controller.markScenarioDirty() - if (matrix.session && matrix.session.liveControlReady && !matrix.controller.pendingApply) - matrix.session.requestSetEnabled(signalRow.sid, checked) + matrix.controller.queueLiveDraft(function() { + return matrix.session.requestSetEnabled(signalRow.sid, checked) + }, "Channel enable update not queued; retry or Stop/Start.") } } diff --git a/apps/arstack_studio/src/DeviceController.cpp b/apps/arstack_studio/src/DeviceController.cpp index b0a216e40..b97205b3a 100644 --- a/apps/arstack_studio/src/DeviceController.cpp +++ b/apps/arstack_studio/src/DeviceController.cpp @@ -634,13 +634,17 @@ void DeviceController::handlePortClosed(const QString& portName) { bool DeviceController::sendShow() { return sendCommand(QStringLiteral("SHOW")); } bool DeviceController::start() { return sendCommand(QStringLiteral("START")); } +bool DeviceController::frequencyWithinFirmwareContract(const double hz) noexcept { + return std::isfinite(hz) && (hz == 0.0 || (hz >= 1.0 && hz <= 1000.0)); +} + bool DeviceController::startWithScenario(const QVariantMap& scenario) { if (!deviceVerified_ || running_) return false; bool hzOk = false, currentScaleOk = false, voltageScaleOk = false; const double hz = scenario.value(QStringLiteral("hz")).toDouble(&hzOk); const double currentScale = scenario.value(QStringLiteral("currentScale")).toDouble(¤tScaleOk); const double voltageScale = scenario.value(QStringLiteral("voltageScale")).toDouble(&voltageScaleOk); - if (!hzOk || !std::isfinite(hz) || hz < 0.0 || hz > 1000.0 || + if (!hzOk || !frequencyWithinFirmwareContract(hz) || !currentScaleOk || !voltageScaleOk || !finitePositive(currentScale) || !finitePositive(voltageScale)) { setError(QStringLiteral("Invalid frequency or engineering scale; START refused.")); @@ -710,8 +714,8 @@ bool DeviceController::stop() { return sendCommand(QStringLiteral("STOP")); } bool DeviceController::zero() { return sendCommand(QStringLiteral("ZERO")); } bool DeviceController::setFrequency(const double hz) { - if (!std::isfinite(hz) || hz < 0.0 || hz > 1000.0) { - setError(QStringLiteral("Frequency must be within 0..1000 Hz (0 = DC).")); + if (!frequencyWithinFirmwareContract(hz)) { + setError(QStringLiteral("Frequency must be 0 Hz (DC) or 1..1000 Hz (AC).")); return false; } const auto millihertz = std::llround(hz * 1000.0); diff --git a/apps/arstack_studio/src/DeviceController.hpp b/apps/arstack_studio/src/DeviceController.hpp index 4b0aa7116..cfeb616e8 100644 --- a/apps/arstack_studio/src/DeviceController.hpp +++ b/apps/arstack_studio/src/DeviceController.hpp @@ -187,6 +187,8 @@ class DeviceController : public QObject { return activeGeneration != 0 && eventGeneration == activeGeneration; } + // Exactly match SvLiveSignalBank::validate: 0 Hz is DC, AC is 1..1000 Hz. + [[nodiscard]] static bool frequencyWithinFirmwareContract(double hz) noexcept; [[nodiscard]] static bool parseIdentityLine(const QString& line, DeviceIdentity& identity); [[nodiscard]] static bool identitySupportsCurrentContract( const DeviceIdentity& identity, diff --git a/apps/arstack_studio/src/StudioDeviceController.hpp b/apps/arstack_studio/src/StudioDeviceController.hpp index 374efec41..606db1ab9 100644 --- a/apps/arstack_studio/src/StudioDeviceController.hpp +++ b/apps/arstack_studio/src/StudioDeviceController.hpp @@ -125,8 +125,8 @@ class StudioDeviceController : public DeviceController { } bool setFrequency(const double hz) override { - if (!std::isfinite(hz) || hz < 0.0 || hz > 1000.0) { - emit deviceMessage(QStringLiteral("Frequency must be within 0..1000 Hz.")); + if (!frequencyWithinFirmwareContract(hz)) { + emit deviceMessage(QStringLiteral("Frequency must be 0 Hz (DC) or 1..1000 Hz (AC).")); return false; } if (!deviceVerified()) return false; diff --git a/apps/arstack_studio/src/main.cpp b/apps/arstack_studio/src/main.cpp index 4ed6ed249..775b21c39 100644 --- a/apps/arstack_studio/src/main.cpp +++ b/apps/arstack_studio/src/main.cpp @@ -329,6 +329,14 @@ int checkP0ControllerPolicy(int argc, char* argv[]) { "capabilities=SMV-4I4V,LIVE-SETPOINTS,SESSION-LEASE,PTP-P2,SMPSYNCH-AUTO"), rejectedIdentity); + const bool frequencyMatchesFirmware = + DeviceController::frequencyWithinFirmwareContract(0.0) && + !DeviceController::frequencyWithinFirmwareContract(0.5) && + !DeviceController::frequencyWithinFirmwareContract(-1.0) && + DeviceController::frequencyWithinFirmwareContract(1.0) && + DeviceController::frequencyWithinFirmwareContract(50.0) && + DeviceController::frequencyWithinFirmwareContract(1000.0) && + !DeviceController::frequencyWithinFirmwareContract(1000.001); const bool boundedIdentifyPolicy = DeviceController::identityMaxAttempts() == 5 && DeviceController::identityRetryIntervalMs() == 1100 && @@ -438,7 +446,7 @@ int checkP0ControllerPolicy(int argc, char* argv[]) { if (!currentAccepted || !legacyRejectedAsCurrent || !protocolLegacyParsed || !capabilityFailClosed || !rejectsWrongTarget || !rejectsMissingFirmware || !rejectsMalformedBoot || !boundedIdentifyPolicy || !boundedProfileSyncPolicy || - !workerPolicyAligned || !generationPolicy || !portOwnershipPolicy || + !workerPolicyAligned || !frequencyMatchesFirmware || !generationPolicy || !portOwnershipPolicy || !s6RecoveryPolicy || !singleInstancePolicy || !s7IntentPolicy) { qCritical().noquote() << "S1/S2/S3/S4/S5/S6/S7 control-plane contract: FAIL" @@ -452,6 +460,7 @@ int checkP0ControllerPolicy(int argc, char* argv[]) { << "bounded-identify=" << boundedIdentifyPolicy << "bounded-profile-sync=" << boundedProfileSyncPolicy << "worker-policy-aligned=" << workerPolicyAligned + << "firmware-frequency-contract=" << frequencyMatchesFirmware << "generation-policy=" << generationPolicy << "port-owner-policy=" << portOwnershipPolicy << "s6-recovery-health=" << s6RecoveryPolicy diff --git a/apps/arstack_studio/tests/check_operator_workspace.py b/apps/arstack_studio/tests/check_operator_workspace.py index a605626a8..79615f537 100644 --- a/apps/arstack_studio/tests/check_operator_workspace.py +++ b/apps/arstack_studio/tests/check_operator_workspace.py @@ -42,6 +42,16 @@ def main() -> None: and "check_auto_local_sync_gate_requires_both_clocks_and_fresh_mac" in sv_tests, "retain independently bounded PTP sample-phase and nonzero MAC publication latency") firmware_source = (project_root / "embedded/esp32p4_smv_injector/main/app_main.cpp").read_text(encoding="utf-8") + live_firmware = (project_root / "embedded/esp32p4_smv_injector/main/live_control.cpp").read_text(encoding="utf-8") + bank_header = (project_root / "include/ariec61850/sampled_values/live_signal_state.hpp").read_text(encoding="utf-8") + bank_test = (project_root / "tests/test_live_signal.cpp").read_text(encoding="utf-8") + require("reader_counts_[index].fetch_add(1U)" in bank_header + and "reader_counts_[candidate].load()" in bank_header + and "concurrent_publisher_and_diagnostics_snapshots_are_coherent()" in bank_test + and "g_start_request.store(false, std::memory_order_release);" in live_firmware + and "START rejected: active runtime profile is invalid" in firmware_source + and "live_control_force_stop();" in firmware_source, + "SMV SSOT must protect concurrent publisher/diagnostic snapshots and fail closed on START errors") require("kSvTxEvidenceEpochStride = 2U" in sv_gate and "expected_epoch_stride = kSvTxEvidenceEpochStride" in sv_gate and "kEvidenceStride = kSvTxEvidenceEpochStride" in sv_gate @@ -94,6 +104,23 @@ def main() -> None: and "Firmware recovery" in ribbon_qml, "never infer missing firmware from USB timeout; retry identification and prioritize semantic-verified ports") main_qml = source("Main.qml") + matrix_qml = source("SignalMatrix.qml") + require("function queueLiveDraft(sendCommand, errorMessage)" in main_qml + and "pendingApply = !queued // QUEUED only" in main_qml + and "matrix.controller.queueLiveDraft(function()" in matrix_qml + and "requestSetEnabled(signal.signalId, signal.enabled)" in main_qml + and "requestZero()" in main_qml + and "requestSetQuality(activeSignal, unsignedValue)" in main_qml + and "if (workflowBar.session && workflowBar.session.liveControlReady && !pendingApply)" not in main_qml, + "all RUNNING GUI edits must queue, and Apply must include eight enable bits") + host_controller = (STUDIO / "src/DeviceController.cpp").read_text(encoding="utf-8") + host_header = (STUDIO / "src/DeviceController.hpp").read_text(encoding="utf-8") + studio_controller = (STUDIO / "src/StudioDeviceController.hpp").read_text(encoding="utf-8") + require("frequencyWithinFirmwareContract" in host_header + and "hz == 0.0 || (hz >= 1.0 && hz <= 1000.0)" in host_controller + and "frequencyWithinFirmwareContract(hz)" in studio_controller + and "value === 0 || (value >= 1.0 && value <= 1000.0)" in main_qml, + "frequency validation must never send 0(active_profile.vlan_priority), static_cast(canonical.bytes.size())); + // An operator STOP during template preparation wins over START. + if (!live_tx_running()) { + stop_sample_clock(sample_clock); + continue; + } // Timer commit is the final operation in this branch. // Formatting the profile log after timer enable previously queued // dozens of ISR notifications before the publisher could consume diff --git a/embedded/esp32p4_smv_injector/main/live_control.cpp b/embedded/esp32p4_smv_injector/main/live_control.cpp index df02f1991..03adf780c 100644 --- a/embedded/esp32p4_smv_injector/main/live_control.cpp +++ b/embedded/esp32p4_smv_injector/main/live_control.cpp @@ -445,6 +445,8 @@ void handle_line(char* line) noexcept { return; } g_running.store(false, std::memory_order_release); + // Cancel a queued START before it can create a new sample clock. + g_start_request.store(false, std::memory_order_release); disarm_control_lease(); wake_publisher(); ESP_LOGI(kTag, "STOP accepted: SV transmission suppressed"); diff --git a/include/ariec61850/sampled_values/live_signal_state.hpp b/include/ariec61850/sampled_values/live_signal_state.hpp index 83dce1d7f..aba2d1384 100644 --- a/include/ariec61850/sampled_values/live_signal_state.hpp +++ b/include/ariec61850/sampled_values/live_signal_state.hpp @@ -55,25 +55,29 @@ class SvLiveSignalBank final { banks_[0] = default_state(); banks_[1] = banks_[0]; banks_[2] = banks_[0]; + banks_[3] = banks_[0]; } - // One control-plane writer and any number of serialized realtime snapshots - // are expected. The reader advertises the bank it is copying before the - // copy begins. With three banks, the writer can always select a bank that - // is neither active nor being read, so no non-atomic object is read and - // written concurrently. + // One control-plane writer, multiple *concurrent* readers (the 4 kHz + // publisher and low-rate SHOW/telemetry tasks). A single reader_index_ + // was unsafe: the second reader overwrote the first reader's protection, + // permitting the writer to modify a still-in-use bank. + // + // Count readers PER bank. The reader pins a candidate, then rechecks the + // active index before copying. A writer never modifies the active bank or + // any pinned bank. A spare fourth bank prevents normal two-reader traffic + // from exhausting the writer's update slots. All operations are bounded: + // no mutex, allocation, or I/O in the 4 kHz snapshot path. [[nodiscard]] SvLiveSignalState snapshot() const noexcept { while (true) { - const auto index = active_index_.load(std::memory_order_acquire); - reader_index_.store(index, std::memory_order_release); - - if (active_index_.load(std::memory_order_acquire) != index) { - reader_index_.store(kNoReader, std::memory_order_release); + const auto index = active_index_.load(); + reader_counts_[index].fetch_add(1U); + if (active_index_.load() != index) { + reader_counts_[index].fetch_sub(1U); continue; } - const auto copy = banks_[index]; - reader_index_.store(kNoReader, std::memory_order_release); + reader_counts_[index].fetch_sub(1U); return copy; } } @@ -88,11 +92,9 @@ class SvLiveSignalBank final { } const auto current = active_index_.load(std::memory_order_acquire); - const auto reader = reader_index_.load(std::memory_order_acquire); - std::uint32_t next = kNoReader; for (std::uint32_t candidate = 0U; candidate < banks_.size(); ++candidate) { - if (candidate != current && candidate != reader) { + if (candidate != current && reader_counts_[candidate].load() == 0U) { next = candidate; break; } @@ -103,7 +105,7 @@ class SvLiveSignalBank final { banks_[next] = requested; banks_[next].generation = banks_[current].generation + 1U; - active_index_.store(next, std::memory_order_release); + active_index_.store(next); return true; } @@ -149,8 +151,8 @@ class SvLiveSignalBank final { private: static constexpr std::uint32_t kNoReader = 3U; - std::array banks_{}; - mutable std::atomic reader_index_{kNoReader}; + std::array banks_{}; + mutable std::array, 4> reader_counts_{}; std::atomic active_index_{0U}; }; diff --git a/tests/test_live_signal.cpp b/tests/test_live_signal.cpp index 211cc68ee..ac55750df 100644 --- a/tests/test_live_signal.cpp +++ b/tests/test_live_signal.cpp @@ -3,6 +3,8 @@ #include "ariec61850/sampled_values/live_signal_state.hpp" #include +#include +#include #include #include #include @@ -20,6 +22,59 @@ namespace { } \ } while (false) +// Publisher and SHOW run on independent firmware tasks. Prove every +// concurrently observed generation is a complete published 4I+4V state; +// a shared single-reader index used to make this a data race. +void concurrent_publisher_and_diagnostics_snapshots_are_coherent() { + using namespace ar::iec61850::sampled_values; + SvLiveSignalBank bank; + std::atomic done{false}; + std::atomic inconsistent{false}; + std::atomic read_count{0U}; + std::atomic readers_ready{0U}; + + const auto reader = [&] { + readers_ready.fetch_add(1U); + do { + const auto s = bank.snapshot(); + if (s.generation != 0U) { + const auto number = static_cast(s.generation); + if (s.channels[0].rms_counts != number || + s.channels[1].phase_millidegrees != number || + s.channels[4].rms_counts != number * 2 || + s.channels[7].quality != static_cast(number)) { + inconsistent.store(true); + } + } + read_count.fetch_add(1U); + } while (!done.load()); + }; + std::thread publisher(reader); + std::thread diagnostics(reader); + // Deterministic scheduler handshake: on Windows/MSVC the writer can + // finish 20k updates before a freshly launched reader is scheduled. + // Require both readers to be active, not an arbitrary read-count rate. + while (readers_ready.load() != 2U) std::this_thread::yield(); + + std::uint32_t committed = 0U; + for (std::uint32_t i = 0U; i < 20000U; ++i) { + auto s = bank.snapshot(); + const auto next = static_cast(s.generation + 1U); + s.channels[0].rms_counts = next; + s.channels[1].phase_millidegrees = next; + s.channels[4].rms_counts = next * 2; + s.channels[7].quality = static_cast(next); + if (bank.publish(s)) ++committed; + if ((i % 128U) == 0U) std::this_thread::yield(); + } + done.store(true); + publisher.join(); + diagnostics.join(); + CHECK(committed > 100U); + CHECK(read_count.load() >= 2U); // each initialized reader sampled at least once + CHECK(!inconsistent.load()); +} + void bank_commits_complete_generations() { using namespace ar::iec61850::sampled_values; @@ -221,6 +276,7 @@ void independent_channels_change_without_stream_restart() { int main() { try { bank_commits_complete_generations(); + concurrent_publisher_and_diagnostics_snapshots_are_coherent(); fixed_point_engine_preserves_frequency_and_phase(); dc_engine_outputs_constant_engineering_values(); ct_saturation_shape_is_bounded_and_asymmetric();