From b0cb370fd0a6f0dc21dd7557144c7f295c49aa8d Mon Sep 17 00:00:00 2001 From: Will Binns Date: Thu, 22 Oct 2026 21:58:51 +0000 Subject: [PATCH] OPS-SAT run 3, version 2: each replica in a process of its own (v0.7.0) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit MruQuorum, the manager process, starts three MruReplica processes, one ReplicaHost each, linked through F´'s GenericHub over SOCK_SEQPACKET Unix socket pairs (no network, no framing), and pins replica 0 to core 0 and replicas 1 and 2 to core 1. The kill is real: RunRecorder calls the deployment at the start of each cycle and the manager sends SIGKILL to replica 0's process at cycle 240. SeqPacketDriver turns off reopening and ends its reconnect thread once open. The payload is its own module. Requirements MRU-DEP-005 to 007, MRU-RR-006; CI runs version 2; the release adds its ARM build. Full length in ESA's image: 148 s, the same counts and events as version 1. --- .github/workflows/ci.yml | 2 + .github/workflows/release.yml | 9 + .gitignore | 3 + CHANGELOG.md | 13 + CITATION.cff | 2 +- Cargo.lock | 8 +- Cargo.toml | 2 +- README.md | 11 +- docs/fprime-design-fpp.md | 42 +- docs/fprime-design.md | 4 +- docs/fprime-requirements.md | 4 + docs/opssat.md | 27 +- fprime/CMakeLists.txt | 4 + fprime/Mru/CMakeLists.txt | 1 + fprime/Mru/Components/CMakeLists.txt | 1 + .../Mru/Components/ReplicaHost/CMakeLists.txt | 2 + .../Components/ReplicaHost/ReplicaHost.cpp | 39 +- .../Components/ReplicaHost/ReplicaHost.hpp | 5 +- .../Mru/Components/RunRecorder/CMakeLists.txt | 4 +- .../Components/RunRecorder/RunRecorder.cpp | 11 +- .../Components/RunRecorder/RunRecorder.hpp | 10 +- .../test/ut/RunRecorderTestMain.cpp | 7 + .../RunRecorder/test/ut/RunRecorderTester.cpp | 30 +- .../RunRecorder/test/ut/RunRecorderTester.hpp | 3 + .../Components/SeqPacketDriver/CMakeLists.txt | 12 + .../SeqPacketDriver/SeqPacketDriver.cpp | 72 +++ .../SeqPacketDriver/SeqPacketDriver.fpp | 19 + .../SeqPacketDriver/SeqPacketDriver.hpp | 41 ++ .../SeqPacketDriver/SeqPacketSocket.cpp | 53 ++ .../SeqPacketDriver/SeqPacketSocket.hpp | 39 ++ fprime/Mru/Payload/CMakeLists.txt | 9 + fprime/Mru/Payload/Payload.cpp | 55 ++ fprime/Mru/Payload/Payload.hpp | 34 ++ fprime/MruQuorum/CMakeLists.txt | 18 + fprime/MruQuorum/Main.cpp | 504 ++++++++++++++++++ fprime/MruQuorum/Top/CMakeLists.txt | 16 + fprime/MruQuorum/Top/MruQuorumTopology.cpp | 115 ++++ fprime/MruQuorum/Top/MruQuorumTopology.hpp | 39 ++ .../MruQuorum/Top/MruQuorumTopologyDefs.hpp | 58 ++ fprime/MruQuorum/Top/instances.fpp | 69 +++ fprime/MruQuorum/Top/system.fpp | 3 + fprime/MruQuorum/Top/topology.fpp | 132 +++++ fprime/MruReplica/CMakeLists.txt | 15 + fprime/MruReplica/Main.cpp | 106 ++++ fprime/MruReplica/Top/CMakeLists.txt | 12 + fprime/MruReplica/Top/MruReplicaTopology.cpp | 65 +++ fprime/MruReplica/Top/MruReplicaTopology.hpp | 23 + .../MruReplica/Top/MruReplicaTopologyDefs.hpp | 31 ++ fprime/MruReplica/Top/instances.fpp | 18 + fprime/MruReplica/Top/system.fpp | 3 + fprime/MruReplica/Top/topology.fpp | 37 ++ opssat/fprime2/build.sh | 29 + opssat/fprime2/package.sh | 149 ++++++ opssat/fprime2/restart.seq | 14 + opssat/fprime2/run | 172 ++++++ opssat/fprime2/shrink.seq | 15 + opssat/fprime2/tmr.seq | 12 + scripts/fprime-deployment.sh | 47 +- 58 files changed, 2208 insertions(+), 72 deletions(-) create mode 100644 fprime/Mru/Components/SeqPacketDriver/CMakeLists.txt create mode 100644 fprime/Mru/Components/SeqPacketDriver/SeqPacketDriver.cpp create mode 100644 fprime/Mru/Components/SeqPacketDriver/SeqPacketDriver.fpp create mode 100644 fprime/Mru/Components/SeqPacketDriver/SeqPacketDriver.hpp create mode 100644 fprime/Mru/Components/SeqPacketDriver/SeqPacketSocket.cpp create mode 100644 fprime/Mru/Components/SeqPacketDriver/SeqPacketSocket.hpp create mode 100644 fprime/Mru/Payload/CMakeLists.txt create mode 100644 fprime/Mru/Payload/Payload.cpp create mode 100644 fprime/Mru/Payload/Payload.hpp create mode 100644 fprime/MruQuorum/CMakeLists.txt create mode 100644 fprime/MruQuorum/Main.cpp create mode 100644 fprime/MruQuorum/Top/CMakeLists.txt create mode 100644 fprime/MruQuorum/Top/MruQuorumTopology.cpp create mode 100644 fprime/MruQuorum/Top/MruQuorumTopology.hpp create mode 100644 fprime/MruQuorum/Top/MruQuorumTopologyDefs.hpp create mode 100644 fprime/MruQuorum/Top/instances.fpp create mode 100644 fprime/MruQuorum/Top/system.fpp create mode 100644 fprime/MruQuorum/Top/topology.fpp create mode 100644 fprime/MruReplica/CMakeLists.txt create mode 100644 fprime/MruReplica/Main.cpp create mode 100644 fprime/MruReplica/Top/CMakeLists.txt create mode 100644 fprime/MruReplica/Top/MruReplicaTopology.cpp create mode 100644 fprime/MruReplica/Top/MruReplicaTopology.hpp create mode 100644 fprime/MruReplica/Top/MruReplicaTopologyDefs.hpp create mode 100644 fprime/MruReplica/Top/instances.fpp create mode 100644 fprime/MruReplica/Top/system.fpp create mode 100644 fprime/MruReplica/Top/topology.fpp create mode 100755 opssat/fprime2/build.sh create mode 100755 opssat/fprime2/package.sh create mode 100644 opssat/fprime2/restart.seq create mode 100755 opssat/fprime2/run create mode 100644 opssat/fprime2/shrink.seq create mode 100644 opssat/fprime2/tmr.seq diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index f364c7c..0df7369 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -72,6 +72,8 @@ jobs: run: fprime-util generate && fprime-util build -j4 - name: Deployment end to end (OPS-SAT run 3 package, counts against the demo) run: ./scripts/fprime-deployment.sh + - name: Deployment end to end, version 2 (replicas as processes, a real kill) + run: VERSION=2 ./scripts/fprime-deployment.sh - name: Unit tests and parity with the demo (MRU-SYS-001) working-directory: fprime run: fprime-util generate --ut && fprime-util check --all -j4 diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml index 95008fe..71072a0 100644 --- a/.github/workflows/release.yml +++ b/.github/workflows/release.yml @@ -71,6 +71,15 @@ jobs: cp opssat/fprime/dist/* "pkg/$name/" cp opssat/fprime/run opssat/fprime/shrink.seq opssat/fprime/tmr.seq opssat/fprime/restart.seq LICENSE "pkg/$name/" tar -C pkg -czf "dist/$name.tar.gz" "$name" + - name: Build version 2 (replicas as processes) and compile its sequences + run: ./opssat/fprime2/build.sh + - name: Package version 2 + run: | + name=mru-fprime-processes-${{ github.ref_name }}-armv7-linux-static + mkdir -p "pkg/$name" + cp opssat/fprime2/dist/* "pkg/$name/" + cp opssat/fprime2/run opssat/fprime2/shrink.seq opssat/fprime2/tmr.seq opssat/fprime2/restart.seq LICENSE "pkg/$name/" + tar -C pkg -czf "dist/$name.tar.gz" "$name" - uses: actions/upload-artifact@v4 with: name: fprime-armv7-linux-static diff --git a/.gitignore b/.gitignore index e9ac54f..e2f0eaf 100644 --- a/.gitignore +++ b/.gitignore @@ -14,3 +14,6 @@ opssat/fprime/dist-native/ opssat/fprime/toGround/ # F´ install folder written by fprime-util fprime/build-artifacts/ +opssat/fprime2/dist/ +opssat/fprime2/dist-native/ +opssat/fprime2/toGround/ diff --git a/CHANGELOG.md b/CHANGELOG.md index 924c9f5..645fe9e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -3,6 +3,19 @@ All notable changes to Mru Flight. Versions follow [Semantic Versioning](https://semver.org). Release packages are on the [releases page](https://github.com/mruspace/flight/releases). +## 0.7.0 (2026-10-22) + +### Added + +- Run 3, version 2: each replica in a process of its own. `MruQuorum`, the manager process (RedundancyManager, HealthRegistry, FaultInjector, RunRecorder and F´'s CdhCore services), starts three `MruReplica` processes, one ReplicaHost each, linked through F´'s GenericHub. Requirements MRU-DEP-005 to 007. +- `SeqPacketDriver`: a byte-stream driver over one end of a `SOCK_SEQPACKET` Unix socket pair, which keeps message boundaries, so the hub needs no framing and the link no network. After it opens, it turns off reopening and ends its reconnect thread, so a send to a dead replica fails at once. +- Each replica process pins itself to a core (`-C`, default 0,1,1) and reports it. The kill is real: `RunRecorder` calls the deployment at the start of each cycle (MRU-RR-006, one unit test, 45 in all) and the manager sends SIGKILL to replica 0's process at cycle 240; the summary reports how each replica process ended and the experiment's CPU share. +- `opssat/fprime2/`: the sequences, `run`, `build.sh` and `package.sh` for version 2 (`exp4051-mru-flight-fprime-processes-VERSION.tar.gz`); the release adds `mru-fprime-processes-VERSION-armv7-linux-static.tar.gz`. `VERSION=2 scripts/fprime-deployment.sh` runs it in CI. Full length in ESA's image: 148 s, the same counts and events as version 1, replica 0 killed by SIGKILL at cycle 240, three processes pinned in every segment, 19.9 MB of address space, the experiment's CPU 2.9% to 3.75% under emulation. + +### Changed + +- The payload (the demo's hash over a 4 KiB block) is the module `Mru/Payload`, shared by ReplicaHost and RunRecorder; ReplicaHost keeps `compute` and `expectedKnownAnswer`. + ## 0.6.0 (2026-10-22) ### Fixed diff --git a/CITATION.cff b/CITATION.cff index 1c14100..4cb23c3 100644 --- a/CITATION.cff +++ b/CITATION.cff @@ -8,7 +8,7 @@ authors: license: Apache-2.0 repository-code: "https://github.com/mruspace/flight" url: "https://mru.space" -version: 0.6.0 +version: 0.7.0 date-released: 2026-10-22 keywords: - fault tolerance diff --git a/Cargo.lock b/Cargo.lock index 2f564cc..31fa14c 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4,14 +4,14 @@ version = 4 [[package]] name = "bench" -version = "0.6.0" +version = "0.7.0" dependencies = [ "quorum", ] [[package]] name = "demo" -version = "0.6.0" +version = "0.7.0" dependencies = [ "libc", "quorum", @@ -25,11 +25,11 @@ checksum = "ce5d3ddc6d3fa000eb1536d85e147bfe31aacaba692ed6a876f95cb7c855be78" [[package]] name = "quorum" -version = "0.6.0" +version = "0.7.0" [[package]] name = "quorum-ffi" -version = "0.6.0" +version = "0.7.0" dependencies = [ "quorum", ] diff --git a/Cargo.toml b/Cargo.toml index ba8b09e..115a1a9 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -5,7 +5,7 @@ exclude = ["fprime"] resolver = "2" [workspace.package] -version = "0.6.0" +version = "0.7.0" edition = "2021" license = "Apache-2.0" repository = "https://github.com/mruspace/flight" diff --git a/README.md b/README.md index 28ee7ae..0b3712c 100644 --- a/README.md +++ b/README.md @@ -63,7 +63,7 @@ F´ components have not run on flight hardware. |---|---| | [`quorum`](quorum/src/lib.rs) | The decision core: vote, compare, self-check, and the health record that decides when a replica must be diagnosed. `no_std`, no allocation, no dependencies. Unit-tested, checked with [Kani](https://github.com/model-checking/kani), and built in CI for a bare-metal ARM Cortex-M target. | | [`quorum-ffi`](quorum-ffi/src/lib.rs) | The C interface to the core ([`quorum.h`](quorum-ffi/include/quorum.h)), as a `no_std` static library. Kani proves that its decide, mode, strike and clear functions give the core's answer. Used by the F´ components, and usable from any C or C++ framework. | -| [`fprime`](fprime) | F´ v4.4.1 components: `RedundancyManager`, `HealthRegistry`, `ReplicaHost`, and the test-only `FaultInjector` and `RunRecorder`, with spares, standby and state that survives resets. 44 unit tests, and a system test against the demo on the bench schedules: wrong results match in all 60 cases, and correct results match exactly in 36 ([requirements](docs/fprime-requirements.md), [design and results](docs/fprime-design-fpp.md)). `MruOpsSat`, the first F´ deployment: the components with F´'s command, event, telemetry, health and sequencing services, driven by a Linux timer, built as a static 32-bit ARM binary for OPS-SAT PRETTY's processor ([run 3 package](docs/opssat.md)). | +| [`fprime`](fprime) | F´ v4.4.1 components: `RedundancyManager`, `HealthRegistry`, `ReplicaHost`, and the test-only `FaultInjector` and `RunRecorder`, with spares, standby and state that survives resets. 45 unit tests, and a system test against the demo on the bench schedules: wrong results match in all 60 cases, and correct results match exactly in 36 ([requirements](docs/fprime-requirements.md), [design and results](docs/fprime-design-fpp.md)). `MruOpsSat`, the first F´ deployment: the components with F´'s command, event, telemetry, health and sequencing services, driven by a Linux timer, built as a static 32-bit ARM binary for OPS-SAT PRETTY's processor ([run 3 package](docs/opssat.md)). | | [`bench`](bench/src/main.rs) | Times the decision core on the host: nanoseconds per call of `decide()` for each policy and mode, and of `Health::strike()`. No dependencies. | | [`demo`](demo/src/main.rs) | The `quorum` program: three replica processes, a voter built on the core, fault injection, enforced resource limits, and a report of what was delivered and what it cost. Runs on Linux and macOS; release builds are fully static binaries for 64-bit and 32-bit ARM Linux and x86_64 Linux. Its only dependency is `libc`, for the system calls behind the limits. | @@ -393,9 +393,12 @@ Changes are in [CHANGELOG.md](CHANGELOG.md). 1. **F´ components**: `RedundancyManager`, `HealthRegistry`, `ReplicaHost`, and the test-only `FaultInjector` and `RunRecorder` are built ([design](docs/fprime-design.md)). The first deployment, `MruOpsSat`, runs - them with F´'s own services. It is built for OPS-SAT PRETTY's processor and - ran full length in ESA's experiment environment; a run on the SEPP is - planned, not yet proposed. Next: `PowerAccountant` and `ScrubScheduler`. + them with F´'s own services. A second, `MruQuorum` with three `MruReplica` + processes, puts each replica in a process of its own on the SEPP's two + cores, linked through F´'s GenericHub, with a real kill. Both are built for + OPS-SAT PRETTY's processor and ran full length in ESA's experiment + environment; a run on the SEPP is planned, not yet proposed. Next: + `PowerAccountant` and `ScrubScheduler`. 2. **Measurements on representative processors**: static ARM Linux binaries are built and tested in CI under emulation. The 32-bit ARM binary ran on the SEPP, the flight processor of OPS-SAT PRETTY, on the FlatSat and in orbit diff --git a/docs/fprime-design-fpp.md b/docs/fprime-design-fpp.md index 5aae598..0f683a1 100644 --- a/docs/fprime-design-fpp.md +++ b/docs/fprime-design-fpp.md @@ -172,7 +172,7 @@ topology that forgets `loadParameters()` does not test every cycle. |---|---| | Kani, `quorum` and `quorum-ffi` | 9 of 9 proofs verified | | C test of `quorum.h` | passes | -| F´ unit tests | 44 (HealthRegistry 5, ReplicaHost 5, FaultInjector 8, RedundancyManager 23, RunRecorder 3), with address and undefined-behavior sanitizers | +| F´ unit tests | 45 (HealthRegistry 5, ReplicaHost 5, FaultInjector 8, RedundancyManager 23, RunRecorder 4), with address and undefined-behavior sanitizers | | MRU-SYS-001, 60 cases (5 schedules, 2 policies, 3 upset rates, 2 seeds, 4,000 cycles) | 36 identical to the demo. In the other 24, a kill that leaves one replica costs the shrinking quorum 2 correct results and adds 3 caught disagreements, as predicted. Wrong results are identical in all 60. | The expected counts come from the demo itself (`scripts/parity-cases.sh`), @@ -218,6 +218,46 @@ RedundancyManager ─resultOut─▶ RunRecorder ─TlmGet─▶ telemetry datab has (`--allow-multiple-definition`), and names a personality routine that is never called with `panic=abort` (`MruOpsSat/RustSupport.cpp`). +### Version 2: `MruQuorum` and `MruReplica` + +Added on 22 October 2026, option 2 of the [design note](fprime-design.md): +each replica in a process of its own, for the most representative setup the +OPS-SAT rules allow. + +``` +MruQuorum (manager process) MruReplica (x3) + manager -> injector -> hub_i -> adapter_i -> link_i ~~ link -> adapter -> hub -> host + manager <- injector <- hub_i <- adapter_i <- link_i ~~ link <- adapter <- hub <- host +``` + +- **The link**: `SeqPacketDriver`, a byte-stream driver with the shape of F´'s + `TcpClient`, over one end of a `SOCK_SEQPACKET` socket pair. Such a socket + keeps message boundaries, so each hub message arrives whole and the hub + needs no framing. The manager makes the pairs and starts the replica + processes before F´ starts (`posix_spawn`), handing each its end; there is + no address, no listening and no network. After the link opens, the driver + turns off F´'s automatic reopen and ends the reconnect thread: a send to a + replica that is gone fails at once instead of waiting a second on the + manager's thread (found in the first test, where the manager's queue + overflowed after the kill). +- **Cores**: each replica process pins itself (`sched_setaffinity`, allowed + for an unprivileged user) before its threads start; replica 0 on core 0 and + replicas 1 and 2 on core 1 by default, so the pair that compares after the + stuck replica is retired runs on different cores. +- **Real faults**: `RunRecorder` calls the deployment at the start of each + cycle, before the manager (MRU-RR-006); the manager process sends SIGKILL + (or SIGSTOP) to the replica process on the scheduled cycle. The stuck value + and the bit flip stay in `FaultInjector`. The summary reports how each + replica process ended (`0=signal 9 1=exit 0 2=exit 0`) and the CPU share of + all four processes. +- **The payload** moved to `Mru/Payload`, shared by `ReplicaHost` and + `RunRecorder`, so the manager process links the task without the replica + component. +- **A gap in F´'s autocoder** (v4.4.1, direct port calls): the hub's + serial-to-typed calls are written in namespace `Svc` and name the port + serializer without its module; the two topology headers make the Mru + serializers visible there. + ## Change to MRU-SYS-001 The first MRU-SYS-001 asked for the same counts as the demo on the bench diff --git a/docs/fprime-design.md b/docs/fprime-design.md index 8b3b279..7fe2ced 100644 --- a/docs/fprime-design.md +++ b/docs/fprime-design.md @@ -60,7 +60,9 @@ involved. between replicas. 2. **Three processes.** One deployment per replica, linked through F´'s generic hub over local sockets. Matches the demo, and an OPS-SAT experiment can run - this way on one Linux processor. + this way on one Linux processor. Built on 22 October 2026: `MruQuorum` and + `MruReplica`, over Unix socket pairs, the replicas pinned to the SEPP's two + cores ([design](fprime-design-fpp.md#version-2-mruquorum-and-mrureplica)). 3. **Separate processors.** The same deployments on separate boards of the hardware-in-the-loop bench, then on flight hardware with independent processors. diff --git a/docs/fprime-requirements.md b/docs/fprime-requirements.md index 1a4a038..27e5ee0 100644 --- a/docs/fprime-requirements.md +++ b/docs/fprime-requirements.md @@ -106,6 +106,7 @@ Added on 22 October 2026 for the OPS-SAT deployment (run 3). It stands in for th | MRU-RR-003 | The component shall give the run's counts with the manager's last policy, mode, live replicas, caught count and slot states, read from the telemetry database. | Unit test | | MRU-RR-004 | The component shall record the end of the command sequence and whether every command in it succeeded. | Unit test | | MRU-RR-005 | The component shall not be part of a flight build. | Inspection of the build | +| MRU-RR-006 | The component shall call a function the deployment gives it with the number of each cycle, at the start of the cycle, before the manager's work for it. | Unit test | ## Deployment `MruOpsSat` @@ -117,6 +118,9 @@ Added on 22 October 2026: the first F´ deployment of the components, for OPS-SA | MRU-DEP-002 | The policy and the faults shall be commanded by an F´ command sequence that ends before cycle 1. | Integration test | | MRU-DEP-003 | The deployment shall stop at a time cap or on SIGTERM and write a summary of the run. | Integration test | | MRU-DEP-004 | A static 32-bit ARM Linux build shall run in ESA's OPS-SAT experiment environment as user `exp`, within 128 MB of address space and the registered 5% CPU share. | Full-length run in ESA's environment image | +| MRU-DEP-005 | Version 2 (`MruQuorum` and `MruReplica`, added 22 October 2026) shall run each replica as a process of its own, linked to the manager process through F´'s GenericHub over a Unix socket pair that keeps message boundaries (no network, no framing). | Integration test (`VERSION=2 scripts/fprime-deployment.sh`) | +| MRU-DEP-006 | Each replica process shall pin itself to a configured core and report the core it runs on. | Integration test; full-length run in ESA's environment image | +| MRU-DEP-007 | Version 2 shall apply a kill or a hang as a real signal to a replica process (SIGKILL, SIGSTOP) at the start of a configured cycle, and the summary shall report how each replica process ended. | Integration test | ## System requirement diff --git a/docs/opssat.md b/docs/opssat.md index 6855c7b..ceabb91 100644 --- a/docs/opssat.md +++ b/docs/opssat.md @@ -18,7 +18,7 @@ satellite, what it touches, its limits, how to start and stop it, what it sends down, and what it can and cannot show. It describes the `quorum` program that ran in runs 1 and 2. The package for a planned third run is the F´ deployment of the same decision core, with F´'s command, event and -telemetry interfaces ([run 3](#run-3-v060-the-f-deployment-planned)). +telemetry interfaces ([run 3](#run-3-v070-the-f-deployment-planned)). ## What runs @@ -143,7 +143,7 @@ target, checks it against `SHA256SUMS`, adds `run`, the schedules, a the required ownership and no hidden entries. `QUORUM_BIN=path ./package.sh` packages a binary you built yourself. -### Run 3 (v0.6.0, the F´ deployment, planned) +### Run 3 (v0.7.0, the F´ deployment, planned) Run 1 flew the demo, `quorum`, and run 2 uses it too. Run 3, planned and not yet proposed to ESA, is the package for the F´ components ([design](fprime-design-fpp.md)) in a real F´ deployment, @@ -224,6 +224,29 @@ and events as v0.5.1 (342 / 3 / 0 over 448 cycles, 239 / 109 / 0, the restore at cycle 303), no cycle withheld, the timed run at 1.25%, 1.47% and 1.30%. The package is built by `package.sh` from the v0.6.0 release asset. +**Version 2, the package planned for run 3: each replica in a process of +its own** (v0.7.0, [`opssat/fprime2`](../opssat/fprime2)). Version 1 runs +the replicas as threads of one process and simulates the kill. Version 2 is +the most representative setup the OPS-SAT rules allow: `MruQuorum`, the +manager process, starts three `MruReplica` processes, one `ReplicaHost` +each, linked through F´'s GenericHub over Unix socket pairs (no network), +and pins replica 0 to core 0 and replicas 1 and 2 to core 1 of the SEPP. The +stuck value and the bit flip come from `FaultInjector` as before; the kill is +real: the manager sends SIGKILL to replica 0's process at the start of cycle +240. The segments, caps and sequences are the same, except that the kill is +no longer in the sequence. Eight files: `run`, `MruQuorum`, `MruReplica`, +the three sequences, `README.txt` and the license. The package built locally +by `package.sh` and run full length in ESA's image on 22 October 2026: 148 +seconds, every exit as planned, the same counts and events as version 1 +(342 / 3 / 0 over 448 cycles with replica 0 lost at 242; 239 then caught, +0 wrong; the restore at cycle 303), replica 0 ended by signal 9 and the +others cleanly, the three replica processes pinned in every segment, 19.9 MB +of address space for the manager. The experiment's CPU, all four processes, +was 2.9% to 3.75% of wall time under emulation (the manager 1.5% to 1.8%). +Most of the replicas' share is their start-up under the emulator, which +translates each binary; on the SEPP that cost is far lower, and the FlatSat +test measures it. Version 1 stays the fallback. + The CPU share comes from the cycle rate, not a throttle: the deployment is idle between timer ticks. The optimized build matters: an unoptimized F´ build spent 84% of its time in the payload hash and ran at 8% at a 20 ms diff --git a/fprime/CMakeLists.txt b/fprime/CMakeLists.txt index 5af79a7..d53bba5 100644 --- a/fprime/CMakeLists.txt +++ b/fprime/CMakeLists.txt @@ -46,3 +46,7 @@ add_fprime_subdirectory("${CMAKE_CURRENT_LIST_DIR}/Mru") # The OPS-SAT deployment (exp4051, run 3) add_fprime_subdirectory("${CMAKE_CURRENT_LIST_DIR}/MruOpsSat") + +# OPS-SAT run 3, version 2: the replicas as processes of their own +add_fprime_subdirectory("${CMAKE_CURRENT_LIST_DIR}/MruReplica") +add_fprime_subdirectory("${CMAKE_CURRENT_LIST_DIR}/MruQuorum") diff --git a/fprime/Mru/CMakeLists.txt b/fprime/Mru/CMakeLists.txt index d34d7a0..1e866e7 100644 --- a/fprime/Mru/CMakeLists.txt +++ b/fprime/Mru/CMakeLists.txt @@ -1,5 +1,6 @@ # Mru project-wide types, ports and components. add_fprime_subdirectory("${CMAKE_CURRENT_LIST_DIR}/Types") add_fprime_subdirectory("${CMAKE_CURRENT_LIST_DIR}/Ports") +add_fprime_subdirectory("${CMAKE_CURRENT_LIST_DIR}/Payload") add_fprime_subdirectory("${CMAKE_CURRENT_LIST_DIR}/Components") add_fprime_subdirectory("${CMAKE_CURRENT_LIST_DIR}/Test/Parity") diff --git a/fprime/Mru/Components/CMakeLists.txt b/fprime/Mru/Components/CMakeLists.txt index c65bd38..41233ef 100644 --- a/fprime/Mru/Components/CMakeLists.txt +++ b/fprime/Mru/Components/CMakeLists.txt @@ -3,3 +3,4 @@ add_fprime_subdirectory("${CMAKE_CURRENT_LIST_DIR}/HealthRegistry") add_fprime_subdirectory("${CMAKE_CURRENT_LIST_DIR}/ReplicaHost") add_fprime_subdirectory("${CMAKE_CURRENT_LIST_DIR}/FaultInjector") add_fprime_subdirectory("${CMAKE_CURRENT_LIST_DIR}/RunRecorder") +add_fprime_subdirectory("${CMAKE_CURRENT_LIST_DIR}/SeqPacketDriver") diff --git a/fprime/Mru/Components/ReplicaHost/CMakeLists.txt b/fprime/Mru/Components/ReplicaHost/CMakeLists.txt index d231b33..76f89e2 100644 --- a/fprime/Mru/Components/ReplicaHost/CMakeLists.txt +++ b/fprime/Mru/Components/ReplicaHost/CMakeLists.txt @@ -5,6 +5,8 @@ register_fprime_module( "${CMAKE_CURRENT_LIST_DIR}/ReplicaHost.cpp" HEADERS "${CMAKE_CURRENT_LIST_DIR}/ReplicaHost.hpp" + DEPENDS + Mru_Payload ) register_fprime_ut( diff --git a/fprime/Mru/Components/ReplicaHost/ReplicaHost.cpp b/fprime/Mru/Components/ReplicaHost/ReplicaHost.cpp index f19df0a..c98a6bb 100644 --- a/fprime/Mru/Components/ReplicaHost/ReplicaHost.cpp +++ b/fprime/Mru/Components/ReplicaHost/ReplicaHost.cpp @@ -11,51 +11,16 @@ namespace Mru { constexpr FwSizeType ReplicaHost::BLOCK_WORDS; constexpr U64 ReplicaHost::KNOWN_ANSWER_INPUT; -namespace { - -constexpr U64 FILL_MULTIPLIER = 0x9E3779B97F4A7C15ULL; // golden-ratio constant, as in the demo -constexpr U64 FNV_OFFSET = 0xcbf29ce484222325ULL; // FNV-1a 64-bit offset basis -constexpr U64 FNV_PRIME = 0x00000100000001b3ULL; // FNV-1a 64-bit prime -constexpr U32 BITS_PER_WORD = 64; -constexpr U32 BYTES_PER_WORD = 8; -constexpr U32 BITS_PER_BYTE = 8; -constexpr U64 BYTE_MASK = 0xff; - -//! xorshift64 (13, 7, 17), as in the demo -U64 xorshift(U64& s) { - s ^= s << 13; - s ^= s >> 7; - s ^= s << 17; - return s; -} - -} // namespace - ReplicaHost ::ReplicaHost(const char* const compName) : ReplicaHostComponentBase(compName), m_block(), m_enabled(true) {} ReplicaHost ::~ReplicaHost() {} U64 ReplicaHost ::compute(U64 (&block)[BLOCK_WORDS], U64 input, bool flip, U16 word, U8 bit) { - U64 s = (input * FILL_MULTIPLIER) + 1; - for (FwSizeType i = 0; i < BLOCK_WORDS; i++) { - block[i] = xorshift(s); - } - if (flip) { - block[word % BLOCK_WORDS] ^= (static_cast(1) << (bit % BITS_PER_WORD)); - } - U64 h = FNV_OFFSET; - for (FwSizeType i = 0; i < BLOCK_WORDS; i++) { - for (U32 b = 0; b < BYTES_PER_WORD; b++) { - h ^= (block[i] >> (BITS_PER_BYTE * b)) & BYTE_MASK; - h *= FNV_PRIME; - } - } - return h; + return Payload::compute(block, input, flip, word, bit); } U64 ReplicaHost ::expectedKnownAnswer() { - U64 block[BLOCK_WORDS] = {}; - return compute(block, KNOWN_ANSWER_INPUT, false, 0, 0); + return Payload::expectedKnownAnswer(); } void ReplicaHost ::requestIn_handler(FwIndexType portNum, diff --git a/fprime/Mru/Components/ReplicaHost/ReplicaHost.hpp b/fprime/Mru/Components/ReplicaHost/ReplicaHost.hpp index ed60823..f688154 100644 --- a/fprime/Mru/Components/ReplicaHost/ReplicaHost.hpp +++ b/fprime/Mru/Components/ReplicaHost/ReplicaHost.hpp @@ -7,16 +7,17 @@ #define Mru_ReplicaHost_HPP #include "Mru/Components/ReplicaHost/ReplicaHostComponentAc.hpp" +#include "Mru/Payload/Payload.hpp" namespace Mru { class ReplicaHost final : public ReplicaHostComponentBase { public: //! Words in the working block: 4 KiB, as in the quorum demo - static constexpr FwSizeType BLOCK_WORDS = 512; + static constexpr FwSizeType BLOCK_WORDS = Payload::BLOCK_WORDS; //! Input of the known-answer test - static constexpr U64 KNOWN_ANSWER_INPUT = 0; + static constexpr U64 KNOWN_ANSWER_INPUT = Payload::KNOWN_ANSWER_INPUT; //! Construct ReplicaHost object explicit ReplicaHost(const char* const compName); diff --git a/fprime/Mru/Components/RunRecorder/CMakeLists.txt b/fprime/Mru/Components/RunRecorder/CMakeLists.txt index 8c51410..8fc5a08 100644 --- a/fprime/Mru/Components/RunRecorder/CMakeLists.txt +++ b/fprime/Mru/Components/RunRecorder/CMakeLists.txt @@ -7,7 +7,7 @@ register_fprime_module( "${CMAKE_CURRENT_LIST_DIR}/RunRecorder.hpp" DEPENDS Mru_Components_RedundancyManager - Mru_Components_ReplicaHost + Mru_Payload Svc_Seq ) @@ -20,7 +20,7 @@ register_fprime_ut( DEPENDS STest Mru_Components_RedundancyManager - Mru_Components_ReplicaHost + Mru_Payload Svc_Seq UT_AUTO_HELPERS ) diff --git a/fprime/Mru/Components/RunRecorder/RunRecorder.cpp b/fprime/Mru/Components/RunRecorder/RunRecorder.cpp index 72f5b9f..1b92e14 100644 --- a/fprime/Mru/Components/RunRecorder/RunRecorder.cpp +++ b/fprime/Mru/Components/RunRecorder/RunRecorder.cpp @@ -26,7 +26,7 @@ class ManagerChannels : public RedundancyManagerComponentBase { } // namespace RunRecorder ::RunRecorder(const char* const compName) - : RunRecorderComponentBase(compName), m_managerIdBase(0), m_totals(), m_block() {} + : RunRecorderComponentBase(compName), m_managerIdBase(0), m_cycleHook(nullptr), m_totals(), m_block() {} RunRecorder ::~RunRecorder() {} @@ -37,7 +37,7 @@ void RunRecorder ::configure(FwIdType managerIdBase) { void RunRecorder ::resultIn_handler(FwIndexType portNum, U64 cycle, U64 value) { (void)portNum; // the ground truth: the same task, computed here without any fault - if (value == ReplicaHost::compute(this->m_block, cycle, false, 0, 0)) { + if (value == Payload::compute(this->m_block, cycle, false, 0, 0)) { this->m_totals.correct++; this->m_totals.lastCorrect = cycle; this->tlmWrite_Correct(this->m_totals.correct); @@ -48,10 +48,17 @@ void RunRecorder ::resultIn_handler(FwIndexType portNum, U64 cycle, U64 value) { } } +void RunRecorder ::setCycleHook(void (*hook)(U64 cycle)) { + this->m_cycleHook = hook; +} + void RunRecorder ::schedIn_handler(FwIndexType portNum, U32 context) { (void)portNum; (void)context; this->m_totals.cycles++; + if (this->m_cycleHook != nullptr) { + this->m_cycleHook(this->m_totals.cycles); + } } void RunRecorder ::seqDoneIn_handler(FwIndexType portNum, diff --git a/fprime/Mru/Components/RunRecorder/RunRecorder.hpp b/fprime/Mru/Components/RunRecorder/RunRecorder.hpp index 91ea855..0c285f7 100644 --- a/fprime/Mru/Components/RunRecorder/RunRecorder.hpp +++ b/fprime/Mru/Components/RunRecorder/RunRecorder.hpp @@ -6,7 +6,7 @@ #ifndef Mru_RunRecorder_HPP #define Mru_RunRecorder_HPP -#include "Mru/Components/ReplicaHost/ReplicaHost.hpp" +#include "Mru/Payload/Payload.hpp" #include "Mru/Components/RunRecorder/RunRecorderComponentAc.hpp" #include "Mru/Types/ModeEnumAc.hpp" #include "Mru/Types/PolicyEnumAc.hpp" @@ -50,6 +50,11 @@ class RunRecorder final : public RunRecorderComponentBase { //! Ask the command sequencer to run a sequence file void startSequence(const char* path); + //! A function called with the cycle number at the start of every cycle, + //! before the manager's: a deployment acts on the cycle there, e.g. a + //! real fault in a replica process. nullptr for none. + void setCycleHook(void (*hook)(U64 cycle)); + private: void resultIn_handler(FwIndexType portNum, U64 cycle, U64 value) override; void schedIn_handler(FwIndexType portNum, U32 context) override; @@ -62,8 +67,9 @@ class RunRecorder final : public RunRecorderComponentBase { bool readChannel(FwChanIdType localId, Fw::TlmBuffer& val); FwIdType m_managerIdBase; + void (*m_cycleHook)(U64 cycle); Totals m_totals; - U64 m_block[ReplicaHost::BLOCK_WORDS]; + U64 m_block[Payload::BLOCK_WORDS]; }; } // namespace Mru diff --git a/fprime/Mru/Components/RunRecorder/test/ut/RunRecorderTestMain.cpp b/fprime/Mru/Components/RunRecorder/test/ut/RunRecorderTestMain.cpp index c056f86..c0ae617 100644 --- a/fprime/Mru/Components/RunRecorder/test/ut/RunRecorderTestMain.cpp +++ b/fprime/Mru/Components/RunRecorder/test/ut/RunRecorderTestMain.cpp @@ -29,6 +29,13 @@ TEST(Nominal, CountsCyclesAndSequence) { tester.countsCyclesAndSequence(); } +TEST(Nominal, CallsTheCycleHook) { + COMMENT("The cycle hook is called with each cycle's number"); + REQUIREMENT("MRU-RR-006"); + Mru::RunRecorderTester tester; + tester.callsTheCycleHook(); +} + int main(int argc, char** argv) { ::testing::InitGoogleTest(&argc, argv); return RUN_ALL_TESTS(); diff --git a/fprime/Mru/Components/RunRecorder/test/ut/RunRecorderTester.cpp b/fprime/Mru/Components/RunRecorder/test/ut/RunRecorderTester.cpp index a426a19..cb29e60 100644 --- a/fprime/Mru/Components/RunRecorder/test/ut/RunRecorderTester.cpp +++ b/fprime/Mru/Components/RunRecorder/test/ut/RunRecorderTester.cpp @@ -20,6 +20,17 @@ class ManagerChannels : public RedundancyManagerComponentBase { using RedundancyManagerComponentBase::CHANNELID_SLOTS; }; +//! The cycles the hook was called with +U64 g_hookCycles[8]; +U32 g_hookCalls = 0; + +void hook(U64 cycle) { + if (g_hookCalls < 8) { + g_hookCycles[g_hookCalls] = cycle; + } + g_hookCalls++; +} + } // namespace RunRecorderTester ::RunRecorderTester() @@ -34,8 +45,8 @@ RunRecorderTester ::~RunRecorderTester() { } U64 RunRecorderTester ::correctFor(U64 cycle) { - U64 block[ReplicaHost::BLOCK_WORDS]; - return ReplicaHost::compute(block, cycle, false, 0, 0); + U64 block[Payload::BLOCK_WORDS]; + return Payload::compute(block, cycle, false, 0, 0); } Fw::TlmValid RunRecorderTester ::from_tlmGet_handler(FwIndexType portNum, @@ -109,4 +120,19 @@ void RunRecorderTester ::countsCyclesAndSequence() { ASSERT_TRUE(this->component.totals().seqOk); } +void RunRecorderTester ::callsTheCycleHook() { + g_hookCalls = 0; + this->invoke_to_schedIn(0, 0); // no hook yet + ASSERT_EQ(g_hookCalls, 0u); + this->component.setCycleHook(hook); + this->invoke_to_schedIn(0, 0); + this->invoke_to_schedIn(0, 0); + ASSERT_EQ(g_hookCalls, 2u); + ASSERT_EQ(g_hookCycles[0], 2u); + ASSERT_EQ(g_hookCycles[1], 3u); + this->component.setCycleHook(nullptr); + this->invoke_to_schedIn(0, 0); + ASSERT_EQ(g_hookCalls, 2u); +} + } // namespace Mru diff --git a/fprime/Mru/Components/RunRecorder/test/ut/RunRecorderTester.hpp b/fprime/Mru/Components/RunRecorder/test/ut/RunRecorderTester.hpp index e3d0591..ee3149a 100644 --- a/fprime/Mru/Components/RunRecorder/test/ut/RunRecorderTester.hpp +++ b/fprime/Mru/Components/RunRecorder/test/ut/RunRecorderTester.hpp @@ -31,6 +31,9 @@ class RunRecorderTester final : public RunRecorderGTestBase { //! Cycles are counted, the sequence's end is recorded (MRU-RR-003, 004) void countsCyclesAndSequence(); + //! The cycle hook is called with each cycle, before anything else (MRU-RR-006) + void callsTheCycleHook(); + private: void connectPorts(); void initComponents(); diff --git a/fprime/Mru/Components/SeqPacketDriver/CMakeLists.txt b/fprime/Mru/Components/SeqPacketDriver/CMakeLists.txt new file mode 100644 index 0000000..23faced --- /dev/null +++ b/fprime/Mru/Components/SeqPacketDriver/CMakeLists.txt @@ -0,0 +1,12 @@ +register_fprime_module( + AUTOCODER_INPUTS + "${CMAKE_CURRENT_LIST_DIR}/SeqPacketDriver.fpp" + SOURCES + "${CMAKE_CURRENT_LIST_DIR}/SeqPacketDriver.cpp" + "${CMAKE_CURRENT_LIST_DIR}/SeqPacketSocket.cpp" + HEADERS + "${CMAKE_CURRENT_LIST_DIR}/SeqPacketDriver.hpp" + "${CMAKE_CURRENT_LIST_DIR}/SeqPacketSocket.hpp" + DEPENDS + Drv_Ip +) diff --git a/fprime/Mru/Components/SeqPacketDriver/SeqPacketDriver.cpp b/fprime/Mru/Components/SeqPacketDriver/SeqPacketDriver.cpp new file mode 100644 index 0000000..9e90366 --- /dev/null +++ b/fprime/Mru/Components/SeqPacketDriver/SeqPacketDriver.cpp @@ -0,0 +1,72 @@ +// ====================================================================== +// \title SeqPacketDriver.cpp +// \brief A byte-stream driver over an inherited SOCK_SEQPACKET socket. +// The shape of F´'s TcpClient; only the socket differs. +// ====================================================================== + +#include "Mru/Components/SeqPacketDriver/SeqPacketDriver.hpp" + +#include "Fw/Types/Assert.hpp" + +namespace Mru { + +SeqPacketDriver ::SeqPacketDriver(const char* const compName) + : SeqPacketDriverComponentBase(compName), Drv::SocketComponentHelper(), m_socket(), m_allocationSize(1024) {} + +SeqPacketDriver ::~SeqPacketDriver() {} + +Drv::SocketIpStatus SeqPacketDriver ::configure(int fd, FwSizeType bufferSize) { + this->m_allocationSize = bufferSize; + return this->m_socket.configureDescriptor(fd, SOCKET_SEND_TIMEOUT_SECONDS, SOCKET_SEND_TIMEOUT_MICROSECONDS); +} + +Drv::IpSocket& SeqPacketDriver ::getSocketHandler() { + return this->m_socket; +} + +Fw::Buffer SeqPacketDriver ::getBuffer() { + return this->allocate_out(0, this->m_allocationSize); +} + +void SeqPacketDriver ::sendBuffer(Fw::Buffer buffer, Drv::SocketIpStatus status) { + FW_ASSERT((status != Drv::SOCK_SUCCESS) || (buffer.getData() != nullptr)); + Drv::ByteStreamStatus recvStatus = Drv::ByteStreamStatus::OTHER_ERROR; + if (status == Drv::SOCK_SUCCESS) { + recvStatus = Drv::ByteStreamStatus::OP_OK; + } else if (status == Drv::SOCK_NO_DATA_AVAILABLE) { + recvStatus = Drv::ByteStreamStatus::RECV_NO_DATA; + } + this->recv_out(0, buffer, recvStatus); +} + +void SeqPacketDriver ::connected() { + // a socket pair cannot be reopened: once the peer is gone, a send must + // fail at once, not wait for a reconnect on the caller's thread (the + // manager's, for a replica that was killed) + this->setAutomaticOpen(false); + // and there is nothing for the reconnect thread to do: let it end, + // rather than wake every 50 ms for the rest of the run + this->stopReconnect(); + if (this->isConnected_ready_OutputPort(0)) { + this->ready_out(0); + } +} + +Drv::ByteStreamStatus SeqPacketDriver ::send_handler(const FwIndexType portNum, Fw::Buffer& fwBuffer) { + (void)portNum; + const Drv::SocketIpStatus status = this->send(fwBuffer.getData(), fwBuffer.getSize()); + if (status == Drv::SOCK_SUCCESS) { + return Drv::ByteStreamStatus::OP_OK; + } + if (status == Drv::SOCK_INTERRUPTED_TRY_AGAIN) { + return Drv::ByteStreamStatus::SEND_RETRY; + } + return Drv::ByteStreamStatus::OTHER_ERROR; +} + +void SeqPacketDriver ::recvReturnIn_handler(FwIndexType portNum, Fw::Buffer& fwBuffer) { + (void)portNum; + this->deallocate_out(0, fwBuffer); +} + +} // namespace Mru diff --git a/fprime/Mru/Components/SeqPacketDriver/SeqPacketDriver.fpp b/fprime/Mru/Components/SeqPacketDriver/SeqPacketDriver.fpp new file mode 100644 index 0000000..2206300 --- /dev/null +++ b/fprime/Mru/Components/SeqPacketDriver/SeqPacketDriver.fpp @@ -0,0 +1,19 @@ +module Mru { + + @ Bytes over one end of a Unix socket pair that keeps message boundaries + @ (SOCK_SEQPACKET): one send is one receive. Links a replica process to the + @ manager process through F´'s GenericHub, with no framing and no network. + @ The manager makes the pair and passes one end to each replica process. + passive component SeqPacketDriver { + + import Drv.ByteStreamDriver + + @ Allocation for received data + output port allocate: Fw.BufferGet + + @ Deallocation of allocated buffers + output port deallocate: Fw.BufferSend + + } + +} diff --git a/fprime/Mru/Components/SeqPacketDriver/SeqPacketDriver.hpp b/fprime/Mru/Components/SeqPacketDriver/SeqPacketDriver.hpp new file mode 100644 index 0000000..c5e027c --- /dev/null +++ b/fprime/Mru/Components/SeqPacketDriver/SeqPacketDriver.hpp @@ -0,0 +1,41 @@ +// ====================================================================== +// \title SeqPacketDriver.hpp +// \brief A byte-stream driver over an inherited SOCK_SEQPACKET socket +// ====================================================================== + +#ifndef Mru_SeqPacketDriver_HPP +#define Mru_SeqPacketDriver_HPP + +#include "Drv/Ip/SocketComponentHelper.hpp" +#include "Mru/Components/SeqPacketDriver/SeqPacketDriverComponentAc.hpp" +#include "Mru/Components/SeqPacketDriver/SeqPacketSocket.hpp" + +namespace Mru { + +class SeqPacketDriver final : public SeqPacketDriverComponentBase, public Drv::SocketComponentHelper { + public: + explicit SeqPacketDriver(const char* const compName); + ~SeqPacketDriver() override; + + SeqPacketDriver(const SeqPacketDriver&) = delete; + SeqPacketDriver& operator=(const SeqPacketDriver&) = delete; + + //! Use this end of a socket pair. `bufferSize` must hold the largest message. + Drv::SocketIpStatus configure(int fd, FwSizeType bufferSize = 1024); + + private: + Drv::IpSocket& getSocketHandler() override; + Fw::Buffer getBuffer() override; + void sendBuffer(Fw::Buffer buffer, Drv::SocketIpStatus status) override; + void connected() override; + + Drv::ByteStreamStatus send_handler(const FwIndexType portNum, Fw::Buffer& fwBuffer) override; + void recvReturnIn_handler(FwIndexType portNum, Fw::Buffer& fwBuffer) override; + + SeqPacketSocket m_socket; + FwSizeType m_allocationSize; +}; + +} // namespace Mru + +#endif diff --git a/fprime/Mru/Components/SeqPacketDriver/SeqPacketSocket.cpp b/fprime/Mru/Components/SeqPacketDriver/SeqPacketSocket.cpp new file mode 100644 index 0000000..0d4aea8 --- /dev/null +++ b/fprime/Mru/Components/SeqPacketDriver/SeqPacketSocket.cpp @@ -0,0 +1,53 @@ +// ====================================================================== +// \title SeqPacketSocket.cpp +// \brief F´'s socket helper over an inherited SOCK_SEQPACKET descriptor +// ====================================================================== + +#include "Mru/Components/SeqPacketDriver/SeqPacketSocket.hpp" + +#include + +namespace Mru { + +SeqPacketSocket::SeqPacketSocket() : Drv::IpSocket(), m_fd(-1), m_used(false) {} + +Drv::SocketIpStatus SeqPacketSocket::configureDescriptor(int fd, U32 sendTimeoutSeconds, U32 sendTimeoutMicroseconds) { + if (fd < 0 || sendTimeoutMicroseconds >= 1000000) { + return Drv::SOCK_INVALID_CALL; + } + this->m_fd = fd; + this->m_used = false; + this->m_timeoutSeconds = sendTimeoutSeconds; + this->m_timeoutMicroseconds = sendTimeoutMicroseconds; + return Drv::SOCK_SUCCESS; +} + +Drv::SocketIpStatus SeqPacketSocket::openProtocol(Drv::SocketDescriptor& socketDescriptor) { + // the descriptor is opened once; after the peer is gone there is nothing to reopen + if (this->m_fd < 0 || this->m_used) { + return Drv::SOCK_FAILED_TO_CONNECT; + } + if (this->setupTimeouts(this->m_fd) != Drv::SOCK_SUCCESS) { + return Drv::SOCK_FAILED_TO_SET_SOCKET_OPTIONS; + } + this->m_used = true; + socketDescriptor.fd = this->m_fd; + return Drv::SOCK_SUCCESS; +} + +FwSignedSizeType SeqPacketSocket::sendProtocol(const Drv::SocketDescriptor& socketDescriptor, + const U8* const data, + const FwSizeType size) { + // a send to a peer that is gone returns an error, never SIGPIPE + return static_cast( + ::send(socketDescriptor.fd, data, static_cast(size), SEND_NO_SIGNAL_FLAGS)); +} + +FwSignedSizeType SeqPacketSocket::recvProtocol(const Drv::SocketDescriptor& socketDescriptor, + U8* const data, + const FwSizeType size) { + // one call returns one whole message; 0 when the peer is gone + return static_cast(::recv(socketDescriptor.fd, data, static_cast(size), 0)); +} + +} // namespace Mru diff --git a/fprime/Mru/Components/SeqPacketDriver/SeqPacketSocket.hpp b/fprime/Mru/Components/SeqPacketDriver/SeqPacketSocket.hpp new file mode 100644 index 0000000..99912ec --- /dev/null +++ b/fprime/Mru/Components/SeqPacketDriver/SeqPacketSocket.hpp @@ -0,0 +1,39 @@ +// ====================================================================== +// \title SeqPacketSocket.hpp +// \brief F´'s socket helper over an inherited SOCK_SEQPACKET descriptor +// ====================================================================== + +#ifndef Mru_SeqPacketSocket_HPP +#define Mru_SeqPacketSocket_HPP + +#include "Drv/Ip/IpSocket.hpp" + +namespace Mru { + +//! One end of a socket pair made before the process started. F´'s socket +//! helper calls open once; there is no address, so it cannot reconnect: when +//! the peer is gone, the link is gone. +class SeqPacketSocket final : public Drv::IpSocket { + public: + SeqPacketSocket(); + + //! Use this descriptor, with these send timeouts + Drv::SocketIpStatus configureDescriptor(int fd, U32 sendTimeoutSeconds, U32 sendTimeoutMicroseconds); + + protected: + Drv::SocketIpStatus openProtocol(Drv::SocketDescriptor& socketDescriptor) override; + FwSignedSizeType sendProtocol(const Drv::SocketDescriptor& socketDescriptor, + const U8* const data, + const FwSizeType size) override; + FwSignedSizeType recvProtocol(const Drv::SocketDescriptor& socketDescriptor, + U8* const data, + const FwSizeType size) override; + + private: + int m_fd; + bool m_used; +}; + +} // namespace Mru + +#endif diff --git a/fprime/Mru/Payload/CMakeLists.txt b/fprime/Mru/Payload/CMakeLists.txt new file mode 100644 index 0000000..710f6a3 --- /dev/null +++ b/fprime/Mru/Payload/CMakeLists.txt @@ -0,0 +1,9 @@ +# The replicas' task, shared by ReplicaHost and RunRecorder +register_fprime_module( + SOURCES + "${CMAKE_CURRENT_LIST_DIR}/Payload.cpp" + HEADERS + "${CMAKE_CURRENT_LIST_DIR}/Payload.hpp" + DEPENDS + Fw_Types +) diff --git a/fprime/Mru/Payload/Payload.cpp b/fprime/Mru/Payload/Payload.cpp new file mode 100644 index 0000000..54d43b7 --- /dev/null +++ b/fprime/Mru/Payload/Payload.cpp @@ -0,0 +1,55 @@ +// ====================================================================== +// \title Payload.cpp +// \brief The replicas' task: the demo's hash over a 4 KiB block (D6) +// ====================================================================== + +#include "Mru/Payload/Payload.hpp" + +namespace Mru { +namespace Payload { + +namespace { + +constexpr U64 FILL_MULTIPLIER = 0x9E3779B97F4A7C15ULL; // golden-ratio constant, as in the demo +constexpr U64 FNV_OFFSET = 0xcbf29ce484222325ULL; // FNV-1a 64-bit offset basis +constexpr U64 FNV_PRIME = 0x00000100000001b3ULL; // FNV-1a 64-bit prime +constexpr U32 BITS_PER_WORD = 64; +constexpr U32 BYTES_PER_WORD = 8; +constexpr U32 BITS_PER_BYTE = 8; +constexpr U64 BYTE_MASK = 0xff; + +//! xorshift64 (13, 7, 17), as in the demo +U64 xorshift(U64& s) { + s ^= s << 13; + s ^= s >> 7; + s ^= s << 17; + return s; +} + +} // namespace + +U64 compute(U64 (&block)[BLOCK_WORDS], U64 input, bool flip, U16 word, U8 bit) { + U64 s = (input * FILL_MULTIPLIER) + 1; + for (FwSizeType i = 0; i < BLOCK_WORDS; i++) { + block[i] = xorshift(s); + } + if (flip) { + block[word % BLOCK_WORDS] ^= (static_cast(1) << (bit % BITS_PER_WORD)); + } + U64 h = FNV_OFFSET; + for (FwSizeType i = 0; i < BLOCK_WORDS; i++) { + for (U32 b = 0; b < BYTES_PER_WORD; b++) { + h ^= (block[i] >> (BITS_PER_BYTE * b)) & BYTE_MASK; + h *= FNV_PRIME; + } + } + return h; +} + +U64 expectedKnownAnswer() { + U64 block[BLOCK_WORDS] = {}; + return compute(block, KNOWN_ANSWER_INPUT, false, 0, 0); +} + +} // namespace Payload +} // namespace Mru diff --git a/fprime/Mru/Payload/Payload.hpp b/fprime/Mru/Payload/Payload.hpp new file mode 100644 index 0000000..cf1d845 --- /dev/null +++ b/fprime/Mru/Payload/Payload.hpp @@ -0,0 +1,34 @@ +// ====================================================================== +// \title Payload.hpp +// \brief The replicas' task: the demo's hash over a 4 KiB block (D6) +// +// A module of its own so that a deployment can check results without the +// ReplicaHost component: the manager process of OPS-SAT run 3 has the +// replicas in other processes, and RunRecorder needs only the task. +// ====================================================================== + +#ifndef Mru_Payload_HPP +#define Mru_Payload_HPP + +#include + +namespace Mru { +namespace Payload { + +//! Words in the working block: 4 KiB, as in the quorum demo +constexpr FwSizeType BLOCK_WORDS = 512; + +//! Input of the known-answer test +constexpr U64 KNOWN_ANSWER_INPUT = 0; + +//! Fill the working block for `input`, apply an optional bit flip in it (as +//! radiation would), and hash it. The same task as the demo. +U64 compute(U64 (&block)[BLOCK_WORDS], U64 input, bool flip, U16 word, U8 bit); + +//! The correct result of the known-answer test (D7) +U64 expectedKnownAnswer(); + +} // namespace Payload +} // namespace Mru + +#endif diff --git a/fprime/MruQuorum/CMakeLists.txt b/fprime/MruQuorum/CMakeLists.txt new file mode 100644 index 0000000..7f150c0 --- /dev/null +++ b/fprime/MruQuorum/CMakeLists.txt @@ -0,0 +1,18 @@ +##### +# MruQuorum: the manager process of Mru Flight for OPS-SAT PRETTY (exp4051, +# run 3, version 2). RedundancyManager, HealthRegistry, FaultInjector and +# RunRecorder with F´'s own command, event, telemetry, health and sequencing +# services, driven by a Linux timer. Each replica is a process of its own +# (MruReplica), linked through F´'s GenericHub over a Unix socket pair. No +# communications stack and no network. Events are printed as text lines. +##### + +add_fprime_subdirectory("${CMAKE_CURRENT_LIST_DIR}/Top/") + +register_fprime_deployment( + SOURCES + "${CMAKE_CURRENT_LIST_DIR}/Main.cpp" + "${CMAKE_CURRENT_LIST_DIR}/../MruOpsSat/RustSupport.cpp" + DEPENDS + ${FPRIME_CURRENT_MODULE}_Top +) diff --git a/fprime/MruQuorum/Main.cpp b/fprime/MruQuorum/Main.cpp new file mode 100644 index 0000000..21e9b3f --- /dev/null +++ b/fprime/MruQuorum/Main.cpp @@ -0,0 +1,504 @@ +// ====================================================================== +// \title Main.cpp +// \brief The MruQuorum deployment: the manager process of one run, with +// each replica in a process of its own (MruReplica), stopped by a +// time cap or a signal, with a summary +// +// Usage: MruQuorum [-s sequence.bin] [-t seconds] [-o summary.txt] +// [-f state-file] [-p cycle-ms] [-m memory-MB] +// [-R replica-binary] [-C core,core,core] +// [-k kind:replica@cycle,...] +// +// Before F´ starts, the manager makes one SOCK_SEQPACKET socket pair per +// replica and starts the three replica processes, each pinned to its core +// (-C, default 0,1,1: replicas 0 and 1 on different cores). -k schedules +// real faults in the replica processes, sent at the start of a cycle before +// its requests: kill (SIGKILL, the process ends) and stop (SIGSTOP, it +// hangs). At the end the summary says how each replica process ended. +// +// The run: set up the topology, run the command sequence (the policy and the +// faults, as an operator would command them) and wait for it to finish, then +// drive the rate groups from the Linux timer until the time cap or SIGTERM. +// Signals are blocked in every thread and taken by one thread with sigwait, +// so stopping the timer never runs inside a signal handler. +// ====================================================================== +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include + +#include "Fw/Logger/Logger.hpp" +#include "MruQuorum/Top/MruQuorumTopology.hpp" +#include "Os/Os.hpp" +#include "Os/Task.hpp" + +extern char** environ; + +namespace { + +constexpr int REPLICAS = 3; + +//! The replica processes, started before F´ +pid_t g_pids[REPLICAS] = {-1, -1, -1}; + +//! A real fault in a replica process at the start of a cycle +struct ProcessFault { + U64 cycle; + int replica; + int signal; + bool sent; +}; +constexpr int MAX_FAULTS = 6; +ProcessFault g_faults[MAX_FAULTS]; +int g_nFaults = 0; + +//! Called by RunRecorder at the start of every cycle, before the manager +void onCycle(U64 cycle) { + for (int i = 0; i < g_nFaults; i++) { + ProcessFault& f = g_faults[i]; + if (!f.sent && f.cycle == cycle && g_pids[f.replica] > 0) { + f.sent = true; + const int r = kill(g_pids[f.replica], f.signal); + std::printf("FAULT: replica %d (pid %d) sent %s at the start of cycle %llu: %s\n", f.replica, + static_cast(g_pids[f.replica]), f.signal == SIGKILL ? "SIGKILL" : "SIGSTOP", + static_cast(cycle), r == 0 ? "ok" : "failed"); + std::fflush(stdout); + } + } +} + +//! Parse "kill:0@240,stop:2@60"; false on a malformed entry +bool parseFaults(const char* text) { + const char* p = text; + while (*p != '\0') { + char kind[8] = {}; + int replica = -1; + unsigned long long cycle = 0; + int used = 0; + if (std::sscanf(p, "%7[a-z]:%d@%llu%n", kind, &replica, &cycle, &used) != 3 || replica < 0 || + replica >= REPLICAS || cycle == 0 || g_nFaults >= MAX_FAULTS) { + return false; + } + int sig = 0; + if (std::strcmp(kind, "kill") == 0) { + sig = SIGKILL; + } else if (std::strcmp(kind, "stop") == 0) { + sig = SIGSTOP; + } else { + return false; + } + g_faults[g_nFaults++] = ProcessFault{cycle, replica, sig, false}; + p += used; + if (*p == ',') { + p++; + } + } + return true; +} + +//! Start replica `i` with `fd` as descriptor 100 + i, pinned to `core` +pid_t startReplica(const char* binary, int i, int fd, int core, unsigned memoryMb) { + const int target = 100 + i; + char fdArg[16]; + char replicaArg[16]; + char coreArg[16]; + char memArg[16]; + (void)std::snprintf(fdArg, sizeof(fdArg), "%d", target); + (void)std::snprintf(replicaArg, sizeof(replicaArg), "%d", i); + (void)std::snprintf(coreArg, sizeof(coreArg), "%d", core); + (void)std::snprintf(memArg, sizeof(memArg), "%u", memoryMb); + char* const args[] = {const_cast(binary), const_cast("-f"), fdArg, const_cast("-r"), + replicaArg, const_cast("-c"), coreArg, const_cast("-m"), memArg, nullptr}; + posix_spawn_file_actions_t actions; + posix_spawn_file_actions_init(&actions); + // the descriptor survives the exec under its new number; every other + // socket was made close-on-exec + posix_spawn_file_actions_adddup2(&actions, fd, target); + pid_t pid = -1; + const int r = posix_spawn(&pid, binary, &actions, nullptr, args, environ); + posix_spawn_file_actions_destroy(&actions); + return (r == 0) ? pid : -1; +} + +//! Wait for the replica processes to end (they end when the links close), +//! then describe how each one ended +void reapReplicas(char* out, FwSizeType size) { + int status[REPLICAS] = {0, 0, 0}; + bool done[REPLICAS] = {false, false, false}; + for (int waited = 0; waited < 3000; waited += 20) { + bool all = true; + for (int i = 0; i < REPLICAS; i++) { + if (!done[i] && g_pids[i] > 0) { + const pid_t r = waitpid(g_pids[i], &status[i], WNOHANG); + done[i] = (r == g_pids[i]); + } + all = all && (done[i] || g_pids[i] <= 0); + } + if (all) { + break; + } + // a stopped replica cannot see its link close: let it go + for (int i = 0; i < REPLICAS; i++) { + if (!done[i] && g_pids[i] > 0 && waited == 1000) { + (void)kill(g_pids[i], SIGKILL); + } + } + (void)usleep(20000); + } + FwSizeType used = 0; + for (int i = 0; i < REPLICAS && used < size; i++) { + char what[32] = "not started"; + if (g_pids[i] > 0 && !done[i]) { + (void)std::snprintf(what, sizeof(what), "still running"); + } else if (g_pids[i] > 0 && WIFEXITED(status[i])) { + (void)std::snprintf(what, sizeof(what), "exit %d", WEXITSTATUS(status[i])); + } else if (g_pids[i] > 0 && WIFSIGNALED(status[i])) { + (void)std::snprintf(what, sizeof(what), "signal %d", WTERMSIG(status[i])); + } + const int n = std::snprintf(out + used, size - used, "%s%d=%s", i == 0 ? "" : " ", i, what); + used += (n > 0) ? static_cast(n) : 0; + } +} + +//! Why the run stopped, set by the signal thread +const char* volatile g_stopped = "time-limit"; + +sigset_t g_signals; + +//! Takes SIGTERM, SIGINT, SIGHUP and the time cap's SIGALRM, stops the timer +void* signalThread(void*) { + int sig = 0; + while (sigwait(&g_signals, &sig) == 0) { + g_stopped = (sig == SIGALRM) ? "time-limit" : "signal"; + MruQuorum::stopRateGroups(); + } + return nullptr; +} + +double seconds(const struct timespec& t) { + return static_cast(t.tv_sec) + static_cast(t.tv_nsec) / 1e9; +} + +double seconds(const struct timeval& t) { + return static_cast(t.tv_sec) + static_cast(t.tv_usec) / 1e6; +} + +//! CPU time of the whole process (every thread) and the wall clock at one +//! moment. The summary splits the run at two of them, so a share above the +//! expected one can be placed: start-up, the timed run, or the end. +struct Sample { + double user; //!< CPU time in user space + double system; //!< CPU time in the kernel, on behalf of the process + double wall; //!< Monotonic clock +}; + +Sample sample() { + struct rusage u; + (void)getrusage(RUSAGE_SELF, &u); + struct timespec now; + (void)clock_gettime(CLOCK_MONOTONIC, &now); + return Sample{seconds(u.ru_utime), seconds(u.ru_stime), seconds(now)}; +} + +const char* modeName(const Mru::Mode& m) { + switch (m.e) { + case Mru::Mode::VOTE: + return "vote"; + case Mru::Mode::COMPARE: + return "compare"; + case Mru::Mode::SELF_CHECK: + return "self-check"; + case Mru::Mode::HALTED: + return "halted"; + default: + return "unknown"; + } +} + +const char* slotName(const Mru::SlotState& s) { + switch (s.e) { + case Mru::SlotState::ACTIVE: + return "active"; + case Mru::SlotState::STANDBY: + return "standby"; + case Mru::SlotState::RETIRED: + return "retired"; + default: + return "unknown"; + } +} + +//! The process's address space in KiB, from /proc/self/maps: what RLIMIT_AS +//! counts. (VmSize in /proc/self/status is the emulator's under qemu-user; +//! the maps file is the program's own view on hardware and in emulation.) +unsigned long addressSpaceKiB() { + FILE* maps = fopen("/proc/self/maps", "r"); + if (maps == nullptr) { + return 0; + } + unsigned long total = 0; + unsigned long lo = 0; + unsigned long hi = 0; + char line[512]; + while (fgets(line, sizeof(line), maps) != nullptr) { + if (sscanf(line, "%lx-%lx", &lo, &hi) == 2 && hi > lo) { + total += (hi - lo) / 1024; + } + } + fclose(maps); + return total; +} + +void usage(const char* app) { + Fw::Logger::log( + "Usage: %s [-s sequence.bin] [-t seconds] [-o summary.txt] [-f state-file] [-p cycle-ms] [-m memory-MB]\n" + " [-R replica-binary] [-C core,core,core] [-k kind:replica@cycle,...]\n", + app); +} + +} // namespace + +int main(int argc, char* argv[]) { + struct timespec start; + (void)clock_gettime(CLOCK_MONOTONIC, &start); + + const char* sequence = nullptr; + const char* summary = nullptr; + const char* statePath = ""; + unsigned maxSeconds = 0; + unsigned cycleMs = 100; + const char* replicaBinary = nullptr; + int cores[REPLICAS] = {0, 1, 1}; + unsigned memoryMb = 128; + int option = 0; + while ((option = getopt(argc, argv, "hs:t:o:f:p:m:R:C:k:")) != -1) { + switch (option) { + case 's': + sequence = optarg; + break; + case 't': + maxSeconds = static_cast(atoi(optarg)); + break; + case 'o': + summary = optarg; + break; + case 'f': + statePath = optarg; + break; + case 'p': + cycleMs = static_cast(atoi(optarg)); + break; + case 'm': + memoryMb = static_cast(atoi(optarg)); + break; + case 'R': + replicaBinary = optarg; + break; + case 'C': + if (std::sscanf(optarg, "%d,%d,%d", &cores[0], &cores[1], &cores[2]) != 3) { + usage(argv[0]); + return 1; + } + break; + case 'k': + if (!parseFaults(optarg)) { + usage(argv[0]); + return 1; + } + break; + case 'h': + default: + usage(argv[0]); + return (option == 'h') ? 0 : 1; + } + } + if (cycleMs == 0 || cycleMs >= 1000) { + usage(argv[0]); + return 1; + } + + // the same limit as the demo: address space per process + if (memoryMb > 0) { + struct rlimit limit; + limit.rlim_cur = static_cast(memoryMb) * 1024 * 1024; + limit.rlim_max = limit.rlim_cur; + if (setrlimit(RLIMIT_AS, &limit) != 0) { + Fw::Logger::log("Could not set the memory limit\n"); + return 1; + } + } + + // the replica processes first, before any thread or blocked signal: a + // child starts with this process's signal mask + char defaultReplica[512]; + if (replicaBinary == nullptr) { + const char* slash = std::strrchr(argv[0], '/'); + const int dirLen = (slash == nullptr) ? 0 : static_cast(slash - argv[0]); + (void)std::snprintf(defaultReplica, sizeof(defaultReplica), "%.*s%sMruReplica", dirLen, argv[0], + (slash == nullptr) ? "./" : "/"); + replicaBinary = defaultReplica; + } + int ends[REPLICAS][2]; + for (int i = 0; i < REPLICAS; i++) { + if (socketpair(AF_UNIX, SOCK_SEQPACKET | SOCK_CLOEXEC, 0, ends[i]) != 0) { + Fw::Logger::log("Could not make the socket pair for replica %d\n", i); + return 1; + } + } + for (int i = 0; i < REPLICAS; i++) { + g_pids[i] = startReplica(replicaBinary, i, ends[i][1], cores[i], memoryMb); + if (g_pids[i] <= 0) { + Fw::Logger::log("Could not start replica %d (%s)\n", i, replicaBinary); + return 1; + } + (void)::close(ends[i][1]); // the replica's end lives in the replica + } + + // block the signals here, before any task starts, so every thread + // inherits the mask; the signal thread takes them + sigemptyset(&g_signals); + sigaddset(&g_signals, SIGTERM); + sigaddset(&g_signals, SIGINT); + sigaddset(&g_signals, SIGHUP); + sigaddset(&g_signals, SIGALRM); + (void)pthread_sigmask(SIG_BLOCK, &g_signals, nullptr); + + Os::init(); + MruQuorum::TopologyState state; + state.statePath = statePath; + state.cycleMs = cycleMs; + for (int i = 0; i < REPLICAS; i++) { + state.fds[i] = ends[i][0]; + } + MruQuorum::setupTopology(state); + MruQuorum::setCycleHook(onCycle); + + pthread_t sigThread; + if (pthread_create(&sigThread, nullptr, signalThread, nullptr) != 0) { + Fw::Logger::log("Could not start the signal thread\n"); + return 1; + } + + // every replica linked before anything is asked of it + const char* seqStatus = "none"; + bool linked = false; + for (U32 waited = 0; waited < 5000 && !linked; waited += 10) { + linked = MruQuorum::linksOpen(); + if (!linked) { + Os::Task::delay(Fw::TimeInterval(0, 10000)); + } + } + if (!linked) { + seqStatus = "no-link"; + } + + // the sequence first: the policy and the faults are set before cycle 1 + if (linked && sequence != nullptr) { + MruQuorum::runSequence(sequence); + seqStatus = "timeout"; + for (U32 waited = 0; waited < 10000; waited += 10) { + const Mru::RunRecorder::Totals t = MruQuorum::totals(); + if (t.seqDone) { + seqStatus = t.seqOk ? "ok" : "failed"; + break; + } + Os::Task::delay(Fw::TimeInterval(0, 10000)); + } + } + + if (maxSeconds > 0) { + // the cap counts from the start of the process, start-up included + struct timespec now; + (void)clock_gettime(CLOCK_MONOTONIC, &now); + const double left = static_cast(maxSeconds) - (seconds(now) - seconds(start)); + struct itimerval cap; + std::memset(&cap, 0, sizeof(cap)); + const double l = (left > 0.001) ? left : 0.001; + cap.it_value.tv_sec = static_cast(l); + cap.it_value.tv_usec = static_cast((l - static_cast(cap.it_value.tv_sec)) * 1e6); + (void)setitimer(ITIMER_REAL, &cap, nullptr); + } + // start-up ends here: topology, tasks and the sequence are done + const Sample started = sample(); + if (std::strcmp(seqStatus, "none") == 0 || std::strcmp(seqStatus, "ok") == 0) { + MruQuorum::startRateGroups(state); + } else { + g_stopped = "sequence"; + } + const Sample stopped = sample(); + + // let the last cycle's replies and telemetry reach their queues' end + Os::Task::delay(Fw::TimeInterval(0, 200000)); + const Mru::RunRecorder::Totals t = MruQuorum::totals(); + // every thread is still running here: the largest the address space gets + const unsigned long vmKiB = addressSpaceKiB(); + MruQuorum::teardownTopology(state); + // the links are closed: each replica sees its socket close and ends + for (int i = 0; i < REPLICAS; i++) { + (void)::close(ends[i][0]); + } + char replicas[160] = {}; + reapReplicas(replicas, sizeof(replicas)); + struct rusage children; + (void)getrusage(RUSAGE_CHILDREN, &children); + const double replicaCpu = seconds(children.ru_utime) + seconds(children.ru_stime); + + const Sample ended = sample(); + struct rusage usage; + (void)getrusage(RUSAGE_SELF, &usage); + const double cpu = ended.user + ended.system; + const double wall = ended.wall - seconds(start); + // the three phases: start-up (process start to the first timer cycle), + // the timed run, and the end (draining the queues, teardown) + const double runUser = stopped.user - started.user; + const double runSystem = stopped.system - started.system; + const double runWall = stopped.wall - started.wall; + const double runShare = (runWall > 0.0) ? 100.0 * (runUser + runSystem) / runWall : 0.0; + const double endCpu = (ended.user + ended.system) - (stopped.user + stopped.system); + + char text[2048]; + const int n = snprintf( + text, sizeof(text), + "policy=%s cycles=%llu cycle_ms=%u sequence=%s\n" + "useful=%llu detected=%llu wrong=%llu late=%llu last_useful=%llu live_at_end=%u mode=%s " + "slots=%s,%s,%s stopped=%s\n" + "footprint: max_rss=%ldKiB address_space=%luKiB cpu=%.3fs wall=%.3fs\n" + "cpu: startup_user=%.3fs startup_system=%.3fs startup_wall=%.3fs run_user=%.3fs run_system=%.3fs " + "run_wall=%.3fs run_share=%.2f%% end_cpu=%.3fs\n" + "replicas: %s cores=%d,%d,%d cpu=%.3fs experiment_share=%.2f%%\n" + "limits: memory=%uMB/process max_run=%us\n", + t.policy.e == Mru::Policy::TMR ? "tmr" : "shrink", static_cast(t.cycles), cycleMs, + seqStatus, static_cast(t.correct), static_cast(t.caught), + static_cast(t.wrong), static_cast(t.late), + static_cast(t.lastCorrect), static_cast(t.live), modeName(t.mode), + slotName(t.slots[0]), slotName(t.slots[1]), slotName(t.slots[2]), g_stopped, usage.ru_maxrss, vmKiB, cpu, wall, + started.user, started.system, started.wall - seconds(start), runUser, runSystem, runWall, runShare, endCpu, + replicas, cores[0], cores[1], cores[2], replicaCpu, (wall > 0.0) ? 100.0 * (cpu + replicaCpu) / wall : 0.0, + memoryMb, maxSeconds); + if (n > 0) { + fputs(text, stdout); + if (summary != nullptr) { + FILE* f = fopen(summary, "w"); + if (f != nullptr) { + fputs(text, f); + fclose(f); + } + } + } + fflush(stdout); + // the signal thread is still in sigwait; leave without joining it + return (std::strcmp(seqStatus, "failed") == 0 || std::strcmp(seqStatus, "timeout") == 0 || + std::strcmp(seqStatus, "no-link") == 0) + ? 1 + : 0; +} diff --git a/fprime/MruQuorum/Top/CMakeLists.txt b/fprime/MruQuorum/Top/CMakeLists.txt new file mode 100644 index 0000000..b0aa1d5 --- /dev/null +++ b/fprime/MruQuorum/Top/CMakeLists.txt @@ -0,0 +1,16 @@ +register_fprime_module( + AUTOCODER_INPUTS + "${CMAKE_CURRENT_LIST_DIR}/instances.fpp" + "${CMAKE_CURRENT_LIST_DIR}/system.fpp" + "${CMAKE_CURRENT_LIST_DIR}/topology.fpp" + SOURCES + "${CMAKE_CURRENT_LIST_DIR}/MruQuorumTopology.cpp" + DEPENDS + Fw_Logger + Mru_Components_RedundancyManager + Mru_Components_HealthRegistry + Mru_Payload + Mru_Components_SeqPacketDriver + Mru_Components_FaultInjector + Mru_Components_RunRecorder +) diff --git a/fprime/MruQuorum/Top/MruQuorumTopology.cpp b/fprime/MruQuorum/Top/MruQuorumTopology.cpp new file mode 100644 index 0000000..2dc46e1 --- /dev/null +++ b/fprime/MruQuorum/Top/MruQuorumTopology.cpp @@ -0,0 +1,115 @@ +// ====================================================================== +// \title MruQuorumTopology.cpp +// \brief Set up, run and tear down the MruQuorum topology +// ====================================================================== +#include "MruQuorum/Top/MruQuorumTopologyAc.hpp" +#include "Fw/Types/MallocAllocator.hpp" +#include "Fw/Types/String.hpp" +#include "Mru/Payload/Payload.hpp" + +#include + +namespace MruQuorum { + +namespace { + +Fw::MallocAllocator mallocator; +Svc::ActiveRateGroup::ContextArray rateGroupFastContext(0); +Svc::ActiveRateGroup::ContextArray rateGroupSlowContext(0); + +//! Room for the sequence file: a few commands +constexpr FwSizeType SEQUENCE_BUFFER = 5 * 1024; + +//! Hub messages are small: a request or a reply is under 100 bytes +constexpr FwSizeType MESSAGE_BYTES = 1024; +constexpr U16 MESSAGE_BUFFERS = 192; +constexpr U16 BUFFER_MANAGER_ID = 100; + +//! Stack of each link thread +constexpr Os::Task::ParamType LINK_STACK = 128 * 1024; + +void configureTopology(const TopologyState& state) { + // the fast group runs every cycle, the slow one once a second + Svc::RateGroupDriver::DividerSet dividers; + const FwSizeType perSecond = (state.cycleMs > 0 && state.cycleMs < 1000) ? 1000 / state.cycleMs : 1; + dividers.dividers[Ports_RateGroups::rateGroupFast] = Svc::RateGroupDriver::Divider(1, 0); + dividers.dividers[Ports_RateGroups::rateGroupSlow] = Svc::RateGroupDriver::Divider(perSecond, 0); + rateGroupDriver.configure(dividers); + rateGroupFast.configure(rateGroupFastContext); + rateGroupSlow.configure(rateGroupSlowContext); + cmdSeq.allocateBuffer(0, mallocator, SEQUENCE_BUFFER); + // parameters are not loaded (no parameter database): the manager uses + // the model's defaults, a known-answer test every 64 cycles and a state + // save every 1,000 + manager.configure(Mru::Payload::expectedKnownAnswer(), state.statePath); + recorder.configure(manager.getIdBase()); + Svc::BufferManager::BufferBins bins; + std::memset(&bins, 0, sizeof(bins)); + bins.bins[0].bufferSize = MESSAGE_BYTES; + bins.bins[0].numBuffers = MESSAGE_BUFFERS; + buffers.setup(BUFFER_MANAGER_ID, 0, mallocator, bins); + (void)link0.configure(state.fds[0], MESSAGE_BYTES); + (void)link1.configure(state.fds[1], MESSAGE_BYTES); + (void)link2.configure(state.fds[2], MESSAGE_BYTES); +} + +} // namespace + +void setupTopology(const TopologyState& state) { + initComponents(state); + setBaseIds(); + connectComponents(); + regCommands(); + configComponents(state); + configureTopology(state); + startTasks(state); + // the links' read and reconnect threads, with small stacks (the default + // is the C library's 8 MB per thread) + link0.start(Fw::String("link0"), Os::Task::TASK_PRIORITY_DEFAULT, LINK_STACK, Os::Task::TASK_DEFAULT, + Os::Task::TASK_PRIORITY_DEFAULT, LINK_STACK); + link1.start(Fw::String("link1"), Os::Task::TASK_PRIORITY_DEFAULT, LINK_STACK, Os::Task::TASK_DEFAULT, + Os::Task::TASK_PRIORITY_DEFAULT, LINK_STACK); + link2.start(Fw::String("link2"), Os::Task::TASK_PRIORITY_DEFAULT, LINK_STACK, Os::Task::TASK_DEFAULT, + Os::Task::TASK_PRIORITY_DEFAULT, LINK_STACK); +} + +bool linksOpen() { + return link0.isOpened() && link1.isOpened() && link2.isOpened(); +} + +void setCycleHook(void (*hook)(U64 cycle)) { + recorder.setCycleHook(hook); +} + +void runSequence(const char* path) { + recorder.startSequence(path); +} + +Mru::RunRecorder::Totals totals() { + return recorder.totals(); +} + +void startRateGroups(const TopologyState& state) { + timer.startTimer(Fw::TimeInterval(state.cycleMs / 1000, (state.cycleMs % 1000) * 1000)); +} + +void stopRateGroups() { + timer.quit(); +} + +void teardownTopology(const TopologyState& state) { + link0.stop(); + link1.stop(); + link2.stop(); + (void)link0.join(); + (void)link1.join(); + (void)link2.join(); + stopTasks(state); + freeThreads(state); + cmdSeq.deallocateBuffer(mallocator); + buffers.cleanup(); + tearDownComponents(state); + deinitComponents(state); +} + +} // namespace MruQuorum diff --git a/fprime/MruQuorum/Top/MruQuorumTopology.hpp b/fprime/MruQuorum/Top/MruQuorumTopology.hpp new file mode 100644 index 0000000..3a84987 --- /dev/null +++ b/fprime/MruQuorum/Top/MruQuorumTopology.hpp @@ -0,0 +1,39 @@ +// ====================================================================== +// \title MruQuorumTopology.hpp +// \brief Set up, run and tear down the MruQuorum topology +// ====================================================================== +#ifndef MRUQUORUM_MRUQUORUMTOPOLOGY_HPP +#define MRUQUORUM_MRUQUORUMTOPOLOGY_HPP + +#include "Mru/Components/RunRecorder/RunRecorder.hpp" +#include "MruQuorum/Top/MruQuorumTopologyDefs.hpp" + +namespace MruQuorum { + +//! Initialize, connect, configure and start the components +void setupTopology(const TopologyState& state); + +//! Ask the command sequencer to run a sequence file +void runSequence(const char* path); + +//! True when the links to all three replica processes are open +bool linksOpen(); + +//! Called with the cycle number at the start of every cycle, before the manager +void setCycleHook(void (*hook)(U64 cycle)); + +//! The counts and the manager's last telemetry +Mru::RunRecorder::Totals totals(); + +//! Drive the rate groups from the Linux timer; blocks until stopRateGroups() +void startRateGroups(const TopologyState& state); + +//! Stop the timer; startRateGroups() returns after the current cycle +void stopRateGroups(); + +//! Stop and join the tasks, free their resources +void teardownTopology(const TopologyState& state); + +} // namespace MruQuorum + +#endif diff --git a/fprime/MruQuorum/Top/MruQuorumTopologyDefs.hpp b/fprime/MruQuorum/Top/MruQuorumTopologyDefs.hpp new file mode 100644 index 0000000..7268740 --- /dev/null +++ b/fprime/MruQuorum/Top/MruQuorumTopologyDefs.hpp @@ -0,0 +1,58 @@ +// ====================================================================== +// \title MruQuorumTopologyDefs.hpp +// \brief Definitions the MruQuorum topology autocoder requires +// ====================================================================== +#ifndef MRUQUORUM_MRUQUORUMTOPOLOGYDEFS_HPP +#define MRUQUORUM_MRUQUORUMTOPOLOGYDEFS_HPP + +#include "MruQuorum/Top/FppConstantsAc.hpp" +#include "MruQuorum/Top/Ports_RateGroupsEnumAc.hpp" +#include "Svc/Subtopologies/CdhCore/PingEntries.hpp" +#include "Svc/Subtopologies/CdhCore/SubtopologyTopologyDefs.hpp" + +//! Missed health pings before a warning and before a fatal event, per active +//! component of the main topology. The replicas are processes of their own +//! and are not pinged: a replica that stops is found by its silence. +#include "Mru/Ports/ReplicaEnablePortAc.hpp" +#include "Mru/Ports/ReplicaReplyPortAc.hpp" +#include "Mru/Ports/ReplicaRequestPortAc.hpp" + +// F´'s topology autocoder (v4.4.1, direct port calls) writes the hub's +// serial-to-typed calls inside namespace Svc, and names the port serializer +// without its module. These make the Mru serializers visible there. +namespace Svc { +using Mru::ReplicaEnablePortSerializer; +using Mru::ReplicaReplyPortSerializer; +using Mru::ReplicaRequestPortSerializer; +} // namespace Svc + +namespace PingEntries { +namespace MruQuorum_rateGroupFast { +enum { WARN = 3, FATAL = 5 }; +} +namespace MruQuorum_rateGroupSlow { +enum { WARN = 3, FATAL = 5 }; +} +namespace MruQuorum_cmdSeq { +enum { WARN = 3, FATAL = 5 }; +} +namespace MruQuorum_manager { +enum { WARN = 3, FATAL = 5 }; +} +} // namespace PingEntries + +namespace MruQuorum { + +//! What Main.cpp passes to the topology +struct TopologyState { + const char* statePath; //!< The manager's state file, "" for none + U32 cycleMs; //!< The cycle period in milliseconds + int fds[3]; //!< This process's end of each replica's socket pair + CdhCore::SubtopologyState cdhCore; +}; + +namespace PingEntries = ::PingEntries; + +} // namespace MruQuorum + +#endif diff --git a/fprime/MruQuorum/Top/instances.fpp b/fprime/MruQuorum/Top/instances.fpp new file mode 100644 index 0000000..848115d --- /dev/null +++ b/fprime/MruQuorum/Top/instances.fpp @@ -0,0 +1,69 @@ +module MruQuorum { + + # Base IDs: 0x1000CCxx in the main topology; CdhCore uses 0x01000000. + + module Default { + constant QUEUE_SIZE = 10 + constant STACK_SIZE = 128 * 1024 + } + + # ---------------------------------------------------------------------- + # Active component instances + # ---------------------------------------------------------------------- + + @ Every cycle: the manager, then the recorder + instance rateGroupFast: Svc.ActiveRateGroup base id 0x10001000 \ + queue size Default.QUEUE_SIZE \ + stack size Default.STACK_SIZE \ + priority 43 + + @ Once a second: telemetry, commands, health pings, events, sequencer + instance rateGroupSlow: Svc.ActiveRateGroup base id 0x10002000 \ + queue size Default.QUEUE_SIZE \ + stack size Default.STACK_SIZE \ + priority 42 + + instance cmdSeq: Svc.CmdSequencer base id 0x10003000 \ + queue size Default.QUEUE_SIZE \ + stack size Default.STACK_SIZE \ + priority 40 + + instance manager: Mru.RedundancyManager base id 0x10004000 \ + queue size 32 \ + stack size Default.STACK_SIZE \ + priority 44 + + # ---------------------------------------------------------------------- + # Passive component instances + # ---------------------------------------------------------------------- + + instance chronoTime: Svc.ChronoTime base id 0x10010000 + + instance rateGroupDriver: Svc.RateGroupDriver base id 0x10011000 + + instance timer: Svc.LinuxTimer base id 0x10012000 + + instance registry: Mru.HealthRegistry base id 0x10013000 + + @ Test builds only: the faults, by command + instance injector: Mru.FaultInjector base id 0x10014000 + + @ Experiment builds only: checks every delivered result, and acts on the + @ cycle for the real faults in replica processes + instance recorder: Mru.RunRecorder base id 0x10015000 + + @ One hub, link and adapter per replica process + instance hub0: Svc.GenericHub base id 0x10016000 + instance hub1: Svc.GenericHub base id 0x10017000 + instance hub2: Svc.GenericHub base id 0x10018000 + instance link0: Mru.SeqPacketDriver base id 0x10019000 + instance link1: Mru.SeqPacketDriver base id 0x1001A000 + instance link2: Mru.SeqPacketDriver base id 0x1001B000 + instance adapter0: Drv.ByteStreamBufferAdapter base id 0x1001C000 + instance adapter1: Drv.ByteStreamBufferAdapter base id 0x1001D000 + instance adapter2: Drv.ByteStreamBufferAdapter base id 0x1001E000 + + @ Hub message buffers, for all three links + instance buffers: Svc.BufferManager base id 0x1001F000 + +} diff --git a/fprime/MruQuorum/Top/system.fpp b/fprime/MruQuorum/Top/system.fpp new file mode 100644 index 0000000..725cf11 --- /dev/null +++ b/fprime/MruQuorum/Top/system.fpp @@ -0,0 +1,3 @@ +module MruQuorum { + system MruQuorumSystem: MruQuorum +} diff --git a/fprime/MruQuorum/Top/topology.fpp b/fprime/MruQuorum/Top/topology.fpp new file mode 100644 index 0000000..2bbc268 --- /dev/null +++ b/fprime/MruQuorum/Top/topology.fpp @@ -0,0 +1,132 @@ +module MruQuorum { + + enum Ports_RateGroups { + rateGroupFast + rateGroupSlow + } + + deployment topology MruQuorum { + + import CdhCore.Subtopology + + instance chronoTime + instance rateGroupFast + instance rateGroupSlow + instance rateGroupDriver + instance timer + instance cmdSeq + instance manager + instance registry + instance injector + instance recorder + instance hub0 + instance hub1 + instance hub2 + instance link0 + instance link1 + instance link2 + instance adapter0 + instance adapter1 + instance adapter2 + instance buffers + + command connections instance CdhCore.cmdDisp + event connections instance CdhCore.events + telemetry connections instance CdhCore.tlmSend + text event connections instance CdhCore.textLogger + health connections instance CdhCore.$health + time connections instance chronoTime + + connections RateGroups { + timer.CycleOut -> rateGroupDriver.CycleIn + + rateGroupDriver.CycleOut[Ports_RateGroups.rateGroupFast] -> rateGroupFast.CycleIn + # the recorder first: a real fault for a cycle lands before its requests + rateGroupFast.RateGroupMemberOut[0] -> recorder.schedIn + rateGroupFast.RateGroupMemberOut[1] -> manager.schedIn + + rateGroupDriver.CycleOut[Ports_RateGroups.rateGroupSlow] -> rateGroupSlow.CycleIn + rateGroupSlow.RateGroupMemberOut[0] -> CdhCore.tlmSend.Run + rateGroupSlow.RateGroupMemberOut[1] -> CdhCore.cmdDisp.run + rateGroupSlow.RateGroupMemberOut[2] -> CdhCore.$health.Run + rateGroupSlow.RateGroupMemberOut[3] -> CdhCore.events.run + rateGroupSlow.RateGroupMemberOut[4] -> cmdSeq.schedIn + } + + connections Sequencer { + cmdSeq.comCmdOut -> CdhCore.cmdDisp.seqCmdBuff + CdhCore.cmdDisp.seqCmdStatus -> cmdSeq.cmdResponseIn + cmdSeq.seqDone -> recorder.seqDoneIn + recorder.seqRunOut -> cmdSeq.seqRunIn + } + + @ The test topology of the design note, with each replica in a process + @ of its own behind a hub: serial port 0 carries requests, 1 enables, + @ and the reply comes back on 0 (MruReplica's hub has the same numbers) + connections Quorum { + manager.requestOut[0] -> injector.requestIn[0] + manager.requestOut[1] -> injector.requestIn[1] + manager.requestOut[2] -> injector.requestIn[2] + injector.requestOut[0] -> hub0.serialIn[0] + injector.requestOut[1] -> hub1.serialIn[0] + injector.requestOut[2] -> hub2.serialIn[0] + manager.enableOut[0] -> hub0.serialIn[1] + manager.enableOut[1] -> hub1.serialIn[1] + manager.enableOut[2] -> hub2.serialIn[1] + hub0.serialOut[0] -> injector.replyIn[0] + hub1.serialOut[0] -> injector.replyIn[1] + hub2.serialOut[0] -> injector.replyIn[2] + injector.replyOut[0] -> manager.replyIn[0] + injector.replyOut[1] -> manager.replyIn[1] + injector.replyOut[2] -> manager.replyIn[2] + + manager.healthStrike -> registry.strikeIn + manager.healthClear -> registry.clearIn + + manager.resultOut -> recorder.resultIn + recorder.tlmGet -> CdhCore.tlmSend.TlmGet + } + + @ Hub, adapter and driver per replica: F´'s hub pattern + connections Links { + hub0.allocate -> buffers.bufferGetCallee + hub0.deallocate -> buffers.bufferSendIn + hub0.toBufferDriver -> adapter0.bufferIn + adapter0.bufferInReturn -> hub0.toBufferDriverReturn + adapter0.bufferOut -> hub0.fromBufferDriver + hub0.fromBufferDriverReturn -> adapter0.bufferOutReturn + adapter0.toByteStreamDriver -> link0.$send + link0.ready -> adapter0.byteStreamDriverReady + link0.$recv -> adapter0.fromByteStreamDriver + adapter0.fromByteStreamDriverReturn -> link0.recvReturnIn + link0.allocate -> buffers.bufferGetCallee + link0.deallocate -> buffers.bufferSendIn + hub1.allocate -> buffers.bufferGetCallee + hub1.deallocate -> buffers.bufferSendIn + hub1.toBufferDriver -> adapter1.bufferIn + adapter1.bufferInReturn -> hub1.toBufferDriverReturn + adapter1.bufferOut -> hub1.fromBufferDriver + hub1.fromBufferDriverReturn -> adapter1.bufferOutReturn + adapter1.toByteStreamDriver -> link1.$send + link1.ready -> adapter1.byteStreamDriverReady + link1.$recv -> adapter1.fromByteStreamDriver + adapter1.fromByteStreamDriverReturn -> link1.recvReturnIn + link1.allocate -> buffers.bufferGetCallee + link1.deallocate -> buffers.bufferSendIn + hub2.allocate -> buffers.bufferGetCallee + hub2.deallocate -> buffers.bufferSendIn + hub2.toBufferDriver -> adapter2.bufferIn + adapter2.bufferInReturn -> hub2.toBufferDriverReturn + adapter2.bufferOut -> hub2.fromBufferDriver + hub2.fromBufferDriverReturn -> adapter2.bufferOutReturn + adapter2.toByteStreamDriver -> link2.$send + link2.ready -> adapter2.byteStreamDriverReady + link2.$recv -> adapter2.fromByteStreamDriver + adapter2.fromByteStreamDriverReturn -> link2.recvReturnIn + link2.allocate -> buffers.bufferGetCallee + link2.deallocate -> buffers.bufferSendIn + } + + } + +} diff --git a/fprime/MruReplica/CMakeLists.txt b/fprime/MruReplica/CMakeLists.txt new file mode 100644 index 0000000..c9e91e8 --- /dev/null +++ b/fprime/MruReplica/CMakeLists.txt @@ -0,0 +1,15 @@ +##### +# MruReplica: one replica of Mru Flight as its own F´ deployment, for OPS-SAT +# run 3. A ReplicaHost behind F´'s GenericHub, linked to the manager process +# (MruQuorum) over one end of a Unix socket pair that the manager made. +##### + +add_fprime_subdirectory("${CMAKE_CURRENT_LIST_DIR}/Top/") + +register_fprime_deployment( + SOURCES + "${CMAKE_CURRENT_LIST_DIR}/Main.cpp" + "${CMAKE_CURRENT_LIST_DIR}/../MruOpsSat/RustSupport.cpp" + DEPENDS + ${FPRIME_CURRENT_MODULE}_Top +) diff --git a/fprime/MruReplica/Main.cpp b/fprime/MruReplica/Main.cpp new file mode 100644 index 0000000..8cf2c40 --- /dev/null +++ b/fprime/MruReplica/Main.cpp @@ -0,0 +1,106 @@ +// ====================================================================== +// \title Main.cpp +// \brief One replica of Mru Flight as its own F´ process +// +// Usage: MruReplica -f fd [-r replica] [-c core] [-m memory-MB] +// +// The manager process (MruQuorum) starts each replica with one end of a +// SOCK_SEQPACKET socket pair as descriptor `fd`. The replica pins itself to +// `core` (a processor of its own where there is one), answers the manager's +// requests through F´'s GenericHub, and exits when the link closes: when the +// manager stops, or is killed. A replica killed by a fault simply ends. +// ====================================================================== +#include +#include +#include +#include +#include + +#include +#include + +#include "MruReplica/Top/MruReplicaTopology.hpp" +#include "Os/Os.hpp" +#include "Os/Task.hpp" + +namespace { + +volatile sig_atomic_t g_stop = 0; + +void onSignal(int) { + g_stop = 1; +} + +} // namespace + +int main(int argc, char* argv[]) { + int fd = -1; + int replica = -1; + int core = -1; + unsigned memoryMb = 128; + int option = 0; + while ((option = getopt(argc, argv, "f:r:c:m:")) != -1) { + switch (option) { + case 'f': + fd = atoi(optarg); + break; + case 'r': + replica = atoi(optarg); + break; + case 'c': + core = atoi(optarg); + break; + case 'm': + memoryMb = static_cast(atoi(optarg)); + break; + default: + std::fprintf(stderr, "Usage: %s -f fd [-r replica] [-c core] [-m memory-MB]\n", argv[0]); + return 1; + } + } + if (fd < 0) { + std::fprintf(stderr, "MruReplica: no socket descriptor (-f)\n"); + return 1; + } + if (memoryMb > 0) { + struct rlimit limit; + limit.rlim_cur = static_cast(memoryMb) * 1024 * 1024; + limit.rlim_max = limit.rlim_cur; + (void)setrlimit(RLIMIT_AS, &limit); + } + // pin before any thread starts: every thread of the process inherits it + const char* pinned = "not pinned"; + if (core >= 0) { + cpu_set_t set; + CPU_ZERO(&set); + CPU_SET(core, &set); + pinned = (sched_setaffinity(0, sizeof(set), &set) == 0) ? "pinned" : "pin refused"; + } + std::printf("replica %d: pid %d, %s, on core %d\n", replica, static_cast(getpid()), pinned, sched_getcpu()); + std::fflush(stdout); + + struct sigaction action; + action.sa_handler = onSignal; + sigemptyset(&action.sa_mask); + action.sa_flags = 0; + (void)sigaction(SIGTERM, &action, nullptr); + (void)sigaction(SIGINT, &action, nullptr); + + Os::init(); + MruReplica::TopologyState state; + state.fd = fd; + MruReplica::setupTopology(state); + + // run until the link has opened and closed again, or a stop signal + bool opened = false; + for (U32 waited = 0; g_stop == 0; waited += 50) { + const bool open = MruReplica::linkOpen(); + opened = opened || open; + if ((opened && !open) || (!opened && waited > 10000)) { + break; + } + Os::Task::delay(Fw::TimeInterval(0, 50000)); + } + MruReplica::teardownTopology(state); + return 0; +} diff --git a/fprime/MruReplica/Top/CMakeLists.txt b/fprime/MruReplica/Top/CMakeLists.txt new file mode 100644 index 0000000..4e7fc67 --- /dev/null +++ b/fprime/MruReplica/Top/CMakeLists.txt @@ -0,0 +1,12 @@ +register_fprime_module( + AUTOCODER_INPUTS + "${CMAKE_CURRENT_LIST_DIR}/instances.fpp" + "${CMAKE_CURRENT_LIST_DIR}/system.fpp" + "${CMAKE_CURRENT_LIST_DIR}/topology.fpp" + SOURCES + "${CMAKE_CURRENT_LIST_DIR}/MruReplicaTopology.cpp" + DEPENDS + Fw_Logger + Mru_Components_ReplicaHost + Mru_Components_SeqPacketDriver +) diff --git a/fprime/MruReplica/Top/MruReplicaTopology.cpp b/fprime/MruReplica/Top/MruReplicaTopology.cpp new file mode 100644 index 0000000..4a8fc02 --- /dev/null +++ b/fprime/MruReplica/Top/MruReplicaTopology.cpp @@ -0,0 +1,65 @@ +// ====================================================================== +// \title MruReplicaTopology.cpp +// \brief Set up, run and tear down the MruReplica topology +// ====================================================================== +#include "MruReplica/Top/MruReplicaTopologyAc.hpp" + +#include + +#include "Fw/Types/MallocAllocator.hpp" +#include "Fw/Types/String.hpp" + +namespace MruReplica { + +namespace { + +Fw::MallocAllocator mallocator; + +//! Hub messages are small: a request or a reply is under 100 bytes +constexpr FwSizeType MESSAGE_BYTES = 1024; +constexpr U16 MESSAGE_BUFFERS = 64; +constexpr U16 BUFFER_MANAGER_ID = 200; + +//! Stack of each link thread +constexpr Os::Task::ParamType LINK_STACK = 128 * 1024; + +void configureTopology(const TopologyState& state) { + Svc::BufferManager::BufferBins bins; + std::memset(&bins, 0, sizeof(bins)); + bins.bins[0].bufferSize = MESSAGE_BYTES; + bins.bins[0].numBuffers = MESSAGE_BUFFERS; + buffers.setup(BUFFER_MANAGER_ID, 0, mallocator, bins); + (void)link.configure(state.fd, MESSAGE_BYTES); +} + +} // namespace + +void setupTopology(const TopologyState& state) { + initComponents(state); + setBaseIds(); + connectComponents(); + regCommands(); + configComponents(state); + configureTopology(state); + startTasks(state); + // the link's read and reconnect threads, with small stacks (the default + // is the C library's 8 MB per thread) + link.start(Fw::String("link"), Os::Task::TASK_PRIORITY_DEFAULT, LINK_STACK, Os::Task::TASK_DEFAULT, + Os::Task::TASK_PRIORITY_DEFAULT, LINK_STACK); +} + +bool linkOpen() { + return link.isOpened(); +} + +void teardownTopology(const TopologyState& state) { + link.stop(); + (void)link.join(); + stopTasks(state); + freeThreads(state); + buffers.cleanup(); + tearDownComponents(state); + deinitComponents(state); +} + +} // namespace MruReplica diff --git a/fprime/MruReplica/Top/MruReplicaTopology.hpp b/fprime/MruReplica/Top/MruReplicaTopology.hpp new file mode 100644 index 0000000..3b5923c --- /dev/null +++ b/fprime/MruReplica/Top/MruReplicaTopology.hpp @@ -0,0 +1,23 @@ +// ====================================================================== +// \title MruReplicaTopology.hpp +// \brief Set up, run and tear down the MruReplica topology +// ====================================================================== +#ifndef MRUREPLICA_MRUREPLICATOPOLOGY_HPP +#define MRUREPLICA_MRUREPLICATOPOLOGY_HPP + +#include "MruReplica/Top/MruReplicaTopologyDefs.hpp" + +namespace MruReplica { + +//! Initialize, connect, configure and start the components and the link +void setupTopology(const TopologyState& state); + +//! True once the link has opened and while it stays open +bool linkOpen(); + +//! Stop and join the tasks, free their resources +void teardownTopology(const TopologyState& state); + +} // namespace MruReplica + +#endif diff --git a/fprime/MruReplica/Top/MruReplicaTopologyDefs.hpp b/fprime/MruReplica/Top/MruReplicaTopologyDefs.hpp new file mode 100644 index 0000000..b1616c5 --- /dev/null +++ b/fprime/MruReplica/Top/MruReplicaTopologyDefs.hpp @@ -0,0 +1,31 @@ +// ====================================================================== +// \title MruReplicaTopologyDefs.hpp +// \brief Definitions the MruReplica topology autocoder requires +// ====================================================================== +#ifndef MRUREPLICA_MRUREPLICATOPOLOGYDEFS_HPP +#define MRUREPLICA_MRUREPLICATOPOLOGYDEFS_HPP + + +#include "Mru/Ports/ReplicaEnablePortAc.hpp" +#include "Mru/Ports/ReplicaReplyPortAc.hpp" +#include "Mru/Ports/ReplicaRequestPortAc.hpp" + +// F´'s topology autocoder (v4.4.1, direct port calls) writes the hub's +// serial-to-typed calls inside namespace Svc, and names the port serializer +// without its module. These make the Mru serializers visible there. +namespace Svc { +using Mru::ReplicaEnablePortSerializer; +using Mru::ReplicaReplyPortSerializer; +using Mru::ReplicaRequestPortSerializer; +} // namespace Svc + +namespace MruReplica { + +//! What Main.cpp passes to the topology +struct TopologyState { + int fd; //!< This process's end of the socket pair +}; + +} // namespace MruReplica + +#endif diff --git a/fprime/MruReplica/Top/instances.fpp b/fprime/MruReplica/Top/instances.fpp new file mode 100644 index 0000000..26ac7a3 --- /dev/null +++ b/fprime/MruReplica/Top/instances.fpp @@ -0,0 +1,18 @@ +module MruReplica { + + instance host: Mru.ReplicaHost base id 0x20001000 \ + queue size 16 \ + stack size 128 * 1024 \ + priority 45 + + @ The far end of the manager's hub for this replica + instance hub: Svc.GenericHub base id 0x20002000 + + @ This end of the socket pair + instance link: Mru.SeqPacketDriver base id 0x20003000 + + instance adapter: Drv.ByteStreamBufferAdapter base id 0x20004000 + + instance buffers: Svc.BufferManager base id 0x20005000 + +} diff --git a/fprime/MruReplica/Top/system.fpp b/fprime/MruReplica/Top/system.fpp new file mode 100644 index 0000000..4d3403d --- /dev/null +++ b/fprime/MruReplica/Top/system.fpp @@ -0,0 +1,3 @@ +module MruReplica { + system MruReplicaSystem: MruReplica +} diff --git a/fprime/MruReplica/Top/topology.fpp b/fprime/MruReplica/Top/topology.fpp new file mode 100644 index 0000000..37dbc2d --- /dev/null +++ b/fprime/MruReplica/Top/topology.fpp @@ -0,0 +1,37 @@ +module MruReplica { + + deployment topology MruReplica { + + instance host + instance hub + instance link + instance adapter + instance buffers + + @ The manager's ports, through the hub (the serial port numbers match + @ the manager's hub for this replica) + connections Replica { + hub.serialOut[0] -> host.requestIn + hub.serialOut[1] -> host.enableIn + host.replyOut -> hub.serialIn[0] + } + + @ Hub, adapter and driver: F´'s hub pattern over a byte-stream driver + connections Link { + hub.allocate -> buffers.bufferGetCallee + hub.deallocate -> buffers.bufferSendIn + hub.toBufferDriver -> adapter.bufferIn + adapter.bufferInReturn -> hub.toBufferDriverReturn + adapter.bufferOut -> hub.fromBufferDriver + hub.fromBufferDriverReturn -> adapter.bufferOutReturn + adapter.toByteStreamDriver -> link.$send + link.ready -> adapter.byteStreamDriverReady + link.$recv -> adapter.fromByteStreamDriver + adapter.fromByteStreamDriverReturn -> link.recvReturnIn + link.allocate -> buffers.bufferGetCallee + link.deallocate -> buffers.bufferSendIn + } + + } + +} diff --git a/opssat/fprime2/build.sh b/opssat/fprime2/build.sh new file mode 100755 index 0000000..691faaf --- /dev/null +++ b/opssat/fprime2/build.sh @@ -0,0 +1,29 @@ +#!/bin/sh +# build.sh: build run 3, version 2, for the SEPP and compile its command +# sequences, in the F´ build container. Output in opssat/fprime2/dist/: +# MruQuorum and MruReplica (static 32-bit ARM, optimized) and shrink.bin, tmr.bin, +# restart.bin, compiled against that build's dictionary, plus the dictionary. +# +# ./opssat/fprime2/build.sh the ARM build for the package +# NATIVE=1 ./opssat/fprime2/build.sh a native Linux build, for tests +set -eu +cd "$(dirname "$0")/../.." +if [ -n "${NATIVE:-}" ]; then + platform= cache=/build/native-rel bin=Linux out=opssat/fprime2/dist-native +else + platform=armv7-linux-static cache=/build/arm-rel bin=armv7-linux-static out=opssat/fprime2/dist +fi +./scripts/fprime-docker.sh sh -c " +set -e +cd fprime +[ -f $cache/CMakeCache.txt ] || fprime-util generate $platform --build-cache $cache -DCMAKE_BUILD_TYPE=Release >/tmp/generate.log 2>&1 || { tail -30 /tmp/generate.log; exit 1; } +fprime-util build $platform --build-cache $cache -j8 >/tmp/build.log 2>&1 || { grep -E 'error|FAILED' /tmp/build.log | head -30; exit 1; } +cd /src +rm -rf $out && mkdir -p $out +cp $cache/bin/$bin/MruQuorum $cache/bin/$bin/MruReplica $out/ +cp $cache/MruQuorum/Top/MruQuorumTopologyDictionary.json $out/ +for s in shrink tmr restart; do + fprime-seqgen --dictionary $out/MruQuorumTopologyDictionary.json opssat/fprime2/\$s.seq $out/\$s.bin >/dev/null +done +" +ls -l "$out" diff --git a/opssat/fprime2/package.sh b/opssat/fprime2/package.sh new file mode 100755 index 0000000..f8d7794 --- /dev/null +++ b/opssat/fprime2/package.sh @@ -0,0 +1,149 @@ +#!/bin/sh +# package.sh: build the run 3 experiment package for OPS-SAT PRETTY [exp4051], +# version 2: the F´ deployment with each replica in a process of its own. +# +# Output: exp4051-mru-flight-fprime-processes-VERSION.tar.gz in the current folder, with +# all paths relative and all files owned by user and group "exp", as the +# OPS-SAT experiment environment README requires. The runner unpacks it into +# $HOME of user exp and executes ./run. +# +# Contents: run (entry point), MruQuorum (the manager process) and MruReplica +# (one replica process), static 32-bit ARM hard-float binaries from the +# GitHub release, checksum verified; shrink.bin, tmr.bin and restart.bin (the +# F´ command sequences, compiled against the manager's dictionary, from the +# same release asset); README.txt (for the operators) and LICENSE. +# +# ./package.sh uses the release in VERSION (default v0.7.0) +# DIST=opssat/fprime2/dist ./package.sh uses a local build (build.sh) +set -eu + +# macOS tar stores extended attributes as extra "._NAME" entries; both +# settings below stop them (ESA found them in the run 1 package). +export COPYFILE_DISABLE=1 + +here=$(cd "$(dirname "$0")" && pwd) +VERSION=${VERSION:-v0.7.0} +PLATFORM=armv7-linux-static +REPO=mruspace/flight +work=$(mktemp -d) +trap 'rm -rf "$work"' EXIT +files="MruQuorum MruReplica shrink.bin tmr.bin restart.bin" + +if [ -n "${DIST:-}" ]; then + for f in $files; do cp "$DIST/$f" "$work/$f"; done + source="local build in $DIST" +else + name="mru-fprime-processes-$VERSION-$PLATFORM" + asset="$name.tar.gz" + base="https://github.com/$REPO/releases/download/$VERSION" + curl -fsSL -o "$work/$asset" "$base/$asset" + curl -fsSL -o "$work/SHA256SUMS" "$base/SHA256SUMS" + (cd "$work" && grep " $asset\$" SHA256SUMS | (sha256sum -c - 2>/dev/null || shasum -a 256 -c -)) + tar -xzf "$work/$asset" -C "$work" + for f in $files; do mv "$work/$name/$f" "$work/$f"; done + source="release $VERSION, $asset" +fi + +stage="$work/stage" +mkdir -p "$stage" +for f in $files; do cp "$work/$f" "$stage/$f"; done +cp "$here/run" "$stage/run" +cp "$here/../../LICENSE" "$stage/LICENSE" +chmod 755 "$stage/run" "$stage/MruQuorum" "$stage/MruReplica" +chmod 644 "$stage/shrink.bin" "$stage/tmr.bin" "$stage/restart.bin" "$stage/LICENSE" +sha() { (sha256sum "$1" 2>/dev/null || shasum -a 256 "$1") | cut -d' ' -f1; } +sum=$(sha "$stage/MruQuorum") +sumReplica=$(sha "$stage/MruReplica") +cat >"$stage/README.txt" </dev/null | grep -q bsdtar; then + # shellcheck disable=SC2086 + tar -C "$stage" --no-mac-metadata --uname exp --gname exp --uid 1024 --gid 1024 -czf "$out" $entries +else + # shellcheck disable=SC2086 + tar -C "$stage" --owner=exp:1024 --group=exp:1024 -czf "$out" $entries +fi +echo "wrote $out" + +# List and check the archive with Python's tarfile, which hides nothing: +# exactly the eight files, no "._" entries, every entry owned by exp. +python3 -I - "$out" <<'PY' +import sys, tarfile +want = ["run", "MruQuorum", "MruReplica", "shrink.bin", "tmr.bin", "restart.bin", "README.txt", "LICENSE"] +with tarfile.open(sys.argv[1]) as t: + members = t.getmembers() +for m in members: + print(f"{m.mode:o} {m.uname}/{m.gname} {m.size:>8} {m.name}") +names = [m.name for m in members] +bad = [n for n in names if n.startswith("._") or "/._" in n] +if bad: + sys.exit(f"error: hidden metadata entries in the package: {bad}") +if sorted(names) != sorted(want): + sys.exit(f"error: package entries {names} differ from {want}") +if any(m.uname != "exp" or m.gname != "exp" for m in members): + sys.exit("error: an entry is not owned by exp:exp") +print(f"checked: {len(members)} entries, no hidden files, all owned by exp") +PY diff --git a/opssat/fprime2/restart.seq b/opssat/fprime2/restart.seq new file mode 100644 index 0000000..bf683d6 --- /dev/null +++ b/opssat/fprime2/restart.seq @@ -0,0 +1,14 @@ +; Run 3 on OPS-SAT PRETTY (exp4051), version 2, segment 3: the shrinking +; quorum with the manager's state file, each replica in a process of its own. +; cycle 200 replica 1 returns the same wrong value: outvoted on cycles +; 200, 201 and 202, fails its known-answer test and is retired +; The state file is saved on every change and, set here by command, every +; 100 cycles. The entry script then kills the manager process without +; warning (SIGKILL); its replica processes see their links close and end. +; The manager is started again with no sequence and new replica processes: +; it restores the policy, the slots (replica 1 retired) and the cycle +; counter from the file. +R00:00:00 MruQuorum.manager.SET_POLICY SHRINK +R00:00:00 MruQuorum.manager.STATESAVEPERIOD_PRM_SET 100 +R00:00:00 MruQuorum.injector.SET_UPSETS 0.0 0.05 1 +R00:00:00 MruQuorum.injector.INJECT STUCK 1 200 diff --git a/opssat/fprime2/run b/opssat/fprime2/run new file mode 100755 index 0000000..5ac4aba --- /dev/null +++ b/opssat/fprime2/run @@ -0,0 +1,172 @@ +#!/bin/sh +# run: entry point for the Mru Flight experiment on OPS-SAT PRETTY [exp4051], +# run 3, version 2: the F´ deployment with each replica in a process of its +# own, spread over the processor's two cores. +# +# The OPS-SAT runner executes ./run as user exp in $HOME, which holds the +# files of this package, and collects everything written under $HOME when the +# run ends. This script runs in the foreground, three segments one after the +# other, writes all output to toGround/, and exits. It stops by itself inside +# the five-minute window: the caps below add up to 145 seconds. +# +# MruQuorum is the manager process: Mru's RedundancyManager, HealthRegistry +# and FaultInjector with F´'s command dispatcher, event manager, telemetry +# database, health pings and command sequencer, driven by a Linux timer every +# CYCLE_MS. It starts three MruReplica processes, one ReplicaHost each, +# linked through F´'s GenericHub over Unix socket pairs (no network), pinned +# to the cores in CORES. The stuck value and the bit flip are commanded by +# F´ command sequences (shrink.bin, tmr.bin, restart.bin) and applied by +# FaultInjector; the kill is real: the manager sends SIGKILL to replica 0's +# process at the start of cycle KILL_CYCLE. +# +# 1 shrink the shrinking quorum on the faults of run 1, the kill real, +# cap SHRINK_SECONDS +# 2 tmr the same faults under fixed TMR, cap TMR_SECONDS +# 3 restart the shrinking quorum with the manager's state file. After +# RESTART_AFTER seconds the manager process is killed without +# warning (SIGKILL, our own process, in user space; nothing on +# the platform is touched); its replica processes see their +# links close and end. The state file is copied aside, and the +# manager is started again with no sequence and new replica +# processes. It restores the policy, the slots and the cycle +# counter, and runs for RESTART_SECONDS more. +# +# Settings (environment variables, defaults in brackets): +# SEGMENTS segments to run, in order [shrink tmr restart] +# SHRINK_SECONDS cap of segment 1 [45] +# TMR_SECONDS cap of segment 2 [35] +# RESTART_AFTER seconds before the kill [35] +# RESTART_SECONDS cap of the restarted process [30] +# CYCLE_MS cycle period in milliseconds [100] +# MEMORY_MB address-space limit, each [128] +# CORES replicas' cores [0,1,1] +# KILL_CYCLE cycle of the real kill [240] +# DEPLOYMENT the manager binary [./MruQuorum] +# REPLICA the replica binary [./MruReplica] +# OUT_DIR output folder [./toGround] +set -u + +here=$(cd "$(dirname "$0")" && pwd) +cd "$here" || exit 2 +SEGMENTS=${SEGMENTS:-"shrink tmr restart"} +SHRINK_SECONDS=${SHRINK_SECONDS:-45} +TMR_SECONDS=${TMR_SECONDS:-35} +RESTART_AFTER=${RESTART_AFTER:-35} +RESTART_SECONDS=${RESTART_SECONDS:-30} +CYCLE_MS=${CYCLE_MS:-100} +MEMORY_MB=${MEMORY_MB:-128} +CORES=${CORES:-0,1,1} +KILL_CYCLE=${KILL_CYCLE:-240} +DEPLOYMENT=${DEPLOYMENT:-$here/MruQuorum} +REPLICA=${REPLICA:-$here/MruReplica} +OUT_DIR=${OUT_DIR:-$here/toGround} + +mkdir -p "$OUT_DIR" || exit 2 +info=$OUT_DIR/run-info.txt +now() { date -u '+%Y-%m-%dT%H:%M:%SZ'; } +{ + echo "experiment=exp4051 package=mru-flight-fprime-processes run=3" + echo "start_utc=$(now)" + echo "uname=$(uname -a 2>/dev/null)" + echo "user=$(id 2>/dev/null)" + echo "nproc=$(nproc 2>/dev/null) loadavg=$(cat /proc/loadavg 2>/dev/null)" + echo "--- /proc/cpuinfo ---"; cat /proc/cpuinfo 2>/dev/null + echo "--- /proc/meminfo (head) ---"; head -5 /proc/meminfo 2>/dev/null + echo "---" + echo "deployment_sha256=$(sha256sum "$DEPLOYMENT" 2>/dev/null | cut -d' ' -f1)" + echo "replica_sha256=$(sha256sum "$REPLICA" 2>/dev/null | cut -d' ' -f1)" + for s in shrink tmr restart; do + echo "sequence_${s}_sha256=$(sha256sum "$here/$s.bin" 2>/dev/null | cut -d' ' -f1)" + done + echo "settings: segments=$SEGMENTS shrink_seconds=$SHRINK_SECONDS tmr_seconds=$TMR_SECONDS restart_after=$RESTART_AFTER restart_seconds=$RESTART_SECONDS cycle_ms=$CYCLE_MS memory_mb=$MEMORY_MB cores=$CORES kill_cycle=$KILL_CYCLE" +} >"$info" + +child= +sleeper= +stop() { + # Forward a stop request to the running deployment; it stops its timer, + # writes its summary and exits. + [ -n "$sleeper" ] && kill "$sleeper" 2>/dev/null + if [ -n "$child" ]; then + kill -TERM "$child" 2>/dev/null + wait "$child" + fi + echo "stopped_by_signal_utc=$(now)" >>"$info" + exit 0 +} +trap stop TERM INT HUP + +# deployment NAME SECONDS EVENTS_SUFFIX [more options]: start the deployment +# in the background; its pid is in $child. +deployment() { + d_name=$1 d_secs=$2 d_out=$3 + shift 3 + echo "$d_name: start_utc=$(now) max_seconds=$d_secs" >>"$info" + "$DEPLOYMENT" -t "$d_secs" -p "$CYCLE_MS" -m "$MEMORY_MB" -R "$REPLICA" -C "$CORES" \ + -o "$OUT_DIR/summary-$d_name.txt" "$@" >"$OUT_DIR/events-$d_out.txt" 2>&1 & + child=$! +} + +status=0 +for segment in $SEGMENTS; do + case $segment in + shrink | tmr) + deployment "$segment" "$([ "$segment" = shrink ] && echo "$SHRINK_SECONDS" || echo "$TMR_SECONDS")" "$segment" \ + -s "$here/$segment.bin" -k "kill:0@$KILL_CYCLE" + wait "$child" + rc=$? + child= + echo "$segment: exit=$rc end_utc=$(now)" >>"$info" + [ "$rc" -eq 0 ] || status=1 + ;; + restart) + state=$OUT_DIR/state-restart.dat + # A state file left in $HOME by an earlier run of this package would + # be restored by the first process. The segment starts from cycle 1. + if [ -e "$state" ] || [ -e "$OUT_DIR/state-restart-at-kill.dat" ]; then + rm -f "$state" "$state.tmp" "$OUT_DIR/state-restart-at-kill.dat" + echo "restart: removed a state file left by an earlier run" >>"$info" + fi + # The first process gets a cap of its own, in case the kill below + # never happens: it then stops at the cap, keeps its state file, and + # the second process still restores from it. + deployment restart-1 $((RESTART_AFTER + 10)) restart-1 -s "$here/restart.bin" -f "$state" + sleep "$RESTART_AFTER" & + sleeper=$! + wait "$sleeper" + sleeper= + if kill -0 "$child" 2>/dev/null; then + kill -KILL "$child" + echo "restart: deployment killed (SIGKILL) at $(now), as planned" >>"$info" + fi + wait "$child" + rc=$? + child= + echo "restart-1: exit=$rc end_utc=$(now)" >>"$info" + # 137 is the planned kill; 0 means the cap ended it first (still restorable) + [ "$rc" -eq 137 ] || [ "$rc" -eq 0 ] || status=1 + # the manager is gone; its replica processes end as their links close + sleep 1 + if [ -f "$state" ]; then + cp "$state" "$OUT_DIR/state-restart-at-kill.dat" + echo "restart: state file at the kill: $(wc -c <"$state" | tr -d ' ') bytes" >>"$info" + else + echo "restart: no state file found at the kill" >>"$info" + status=1 + fi + sleep 1 + deployment restart "$RESTART_SECONDS" restart-2 -f "$state" + wait "$child" + rc=$? + child= + echo "restart-2: exit=$rc end_utc=$(now)" >>"$info" + [ "$rc" -eq 0 ] || status=1 + ;; + *) + echo "unknown segment $segment" >>"$info" + status=1 + ;; + esac +done +echo "end_utc=$(now) status=$status" >>"$info" +exit "$status" diff --git a/opssat/fprime2/shrink.seq b/opssat/fprime2/shrink.seq new file mode 100644 index 0000000..f94f344 --- /dev/null +++ b/opssat/fprime2/shrink.seq @@ -0,0 +1,15 @@ +; Run 3 on OPS-SAT PRETTY (exp4051), version 2, segment 1: the shrinking +; quorum with each replica in a process of its own, replicas 0 and 1 on +; different cores. The faults of run 1, as in version 1, except the kill: +; cycle 60 one bit flip in replica 2's computation (FaultInjector) +; cycle 120 replica 1 returns the same wrong value (FaultInjector): +; outvoted three times, fails its known-answer test, retired +; cycle 240 replica 0's process is killed with SIGKILL at the start of +; the cycle (by the manager process, `-k kill:0@240` in run): +; a real crash; lost after three silent cycles; replica 2 +; goes on alone in self-check +; SET_UPSETS sets no random upsets and seed 1, the demo's seed. +R00:00:00 MruQuorum.manager.SET_POLICY SHRINK +R00:00:00 MruQuorum.injector.SET_UPSETS 0.0 0.05 1 +R00:00:00 MruQuorum.injector.INJECT CORRUPT 2 60 +R00:00:00 MruQuorum.injector.INJECT STUCK 1 120 diff --git a/opssat/fprime2/tmr.seq b/opssat/fprime2/tmr.seq new file mode 100644 index 0000000..fecf7e5 --- /dev/null +++ b/opssat/fprime2/tmr.seq @@ -0,0 +1,12 @@ +; Run 3 on OPS-SAT PRETTY (exp4051), version 2, segment 2: fixed TMR with +; each replica in a process of its own, on the same faults: +; cycle 60 one bit flip in replica 2's computation (FaultInjector) +; cycle 120 replica 1 returns the same wrong value (FaultInjector): +; outvoted on every cycle; fixed TMR does not diagnose it +; cycle 240 replica 0's process is killed with SIGKILL (by the manager +; process): lost; fixed TMR compares replicas 1 and 2, which +; disagree on every cycle, and delivers nothing more +R00:00:00 MruQuorum.manager.SET_POLICY TMR +R00:00:00 MruQuorum.injector.SET_UPSETS 0.0 0.05 1 +R00:00:00 MruQuorum.injector.INJECT CORRUPT 2 60 +R00:00:00 MruQuorum.injector.INJECT STUCK 1 120 diff --git a/scripts/fprime-deployment.sh b/scripts/fprime-deployment.sh index 4b4b9e3..0d76f19 100755 --- a/scripts/fprime-deployment.sh +++ b/scripts/fprime-deployment.sh @@ -1,26 +1,38 @@ #!/bin/sh # fprime-deployment.sh: the OPS-SAT run 3 package, end to end with short caps, -# against a native build of the MruOpsSat deployment. Runs opssat/fprime/run -# with its three sequences, then checks each segment against the quorum demo +# against a native build. Version 1 (VERSION=1, the default) runs +# opssat/fprime/run with MruOpsSat, the replicas as threads; version 2 +# (VERSION=2) runs opssat/fprime2/run with MruQuorum and MruReplica, each +# replica a process of its own and the kill a real SIGKILL. Then it checks +# each segment against the quorum demo # on the same faults (MRU-SYS-001: the same wrong count, and a kill may cost # up to three correct results; each cycle the manager withheld because it # closed late, MRU-RM-023, may cost one more), the events on their cycles, the restore from # the state file, and a second run in the same folder. Exit status is # non-zero if any check fails. # -# MRU_DEPLOYMENT the MruOpsSat binary [fprime/build-artifacts/Linux/MruOpsSat/bin/MruOpsSat] -# MRU_DICTIONARY its dictionary [fprime/build-artifacts/Linux/MruOpsSat/dict/MruOpsSatTopologyDictionary.json] +# MRU_DEPLOYMENT the (manager) binary [fprime/build-artifacts/Linux/NAME/bin/NAME] +# MRU_REPLICA version 2's replica [fprime/build-artifacts/Linux/MruReplica/bin/MruReplica] +# MRU_DICTIONARY its dictionary [fprime/build-artifacts/Linux/NAME/dict/NAMETopologyDictionary.json] # QUORUM_BIN the demo binary [target/release/quorum, built if missing] set -eu cd "$(dirname "$0")/.." -art=fprime/build-artifacts/Linux/MruOpsSat -deployment=${MRU_DEPLOYMENT:-$art/bin/MruOpsSat} -dictionary=${MRU_DICTIONARY:-$art/dict/MruOpsSatTopologyDictionary.json} +version=${VERSION:-1} +if [ "$version" = 2 ]; then + name=MruQuorum package=opssat/fprime2 +else + name=MruOpsSat package=opssat/fprime +fi +art=fprime/build-artifacts/Linux/$name +deployment=${MRU_DEPLOYMENT:-$art/bin/$name} +replica=${MRU_REPLICA:-fprime/build-artifacts/Linux/MruReplica/bin/MruReplica} +dictionary=${MRU_DICTIONARY:-$art/dict/${name}TopologyDictionary.json} if [ -z "${QUORUM_BIN:-}" ]; then cargo build --release --quiet -p demo fi Q=${QUORUM_BIN:-target/release/quorum} case "$deployment" in /*) ;; *) deployment="$PWD/$deployment" ;; esac +case "$replica" in /*) ;; *) replica="$PWD/$replica" ;; esac fails=0 pass() { printf 'ok %s\n' "$1"; } @@ -39,14 +51,14 @@ at() { dir=$(mktemp -d) trap 'rm -rf "$dir"' EXIT -cp opssat/fprime/run "$dir/run" +cp "$package/run" "$dir/run" for s in shrink tmr restart; do - fprime-seqgen --dictionary "$dictionary" "opssat/fprime/$s.seq" "$dir/$s.bin" >/dev/null + fprime-seqgen --dictionary "$dictionary" "$package/$s.seq" "$dir/$s.bin" >/dev/null done # 20 ms cycles: segment 1 and 2 reach cycle 400 in 8 s, past the kill at 240; # segment 3 is killed near cycle 400, after the periodic saves at 300 and 400 -env DEPLOYMENT="$deployment" CYCLE_MS=20 SHRINK_SECONDS=8 TMR_SECONDS=8 RESTART_AFTER=8 RESTART_SECONDS=3 \ +env DEPLOYMENT="$deployment" REPLICA="$replica" CYCLE_MS=20 SHRINK_SECONDS=8 TMR_SECONDS=8 RESTART_AFTER=8 RESTART_SECONDS=3 \ "$dir/run" >"$dir/run.out" 2>&1 || fail "run: exit status $? ($(tail -n 5 "$dir/toGround/run-info.txt" | tr '\n' ' '))" out=$dir/toGround @@ -97,8 +109,21 @@ else fail "restart: restored at '$at' ($r / $(tail -n 6 "$out/run-info.txt" | tr '\n' ' ') / events: $(grep -hE 'Retired|Withheld|StateRestored' "$out/events-restart-1.txt" "$out/events-restart-2.txt" | sed 's/.*) //' | tr '\n' ' '))" fi +# version 2: each replica a process on its core, replica 0 killed for real +if [ "$version" = 2 ]; then + r1=$(sed -n 's/^replicas: \(.*\) cores=.*/\1/p' "$out/summary-shrink.txt") + if [ "$r1" = "0=signal 9 1=exit 0 2=exit 0" ] && + grep -q 'FAULT: replica 0 (pid [0-9]*) sent SIGKILL at the start of cycle 240: ok' "$out/events-shrink.txt" && + [ "$(grep -c 'replica [0-2]: pid [0-9]*, pinned, on core' "$out/events-shrink.txt")" = 3 ] && + grep -q '^replicas: 0=exit 0 1=exit 0 2=exit 0' "$out/summary-restart.txt"; then + pass "processes: three replica processes pinned, replica 0 killed by SIGKILL at cycle 240, the others ended cleanly" + else + fail "processes: '$r1' / $(grep -h 'FAULT\|pinned' "$out/events-shrink.txt" | tr '\n' ' ')" + fi +fi + # the same package again in the same folder: the old state file is removed -env DEPLOYMENT="$deployment" CYCLE_MS=20 SEGMENTS=restart RESTART_AFTER=6 RESTART_SECONDS=2 \ +env DEPLOYMENT="$deployment" REPLICA="$replica" CYCLE_MS=20 SEGMENTS=restart RESTART_AFTER=6 RESTART_SECONDS=2 \ "$dir/run" >"$dir/run2.out" 2>&1 || true if grep -q 'removed a state file left by an earlier run' "$out/run-info.txt" && grep -q 'No valid state file' "$out/events-restart-1.txt" &&