1 / 6
EPICS Collaboration Meeting · C++ Edition
Reactive Programming
for EPICS Controls
C++17 — PVXS PVA + RxCpp, zero-overhead reactive pipelines
RxEpics/cpp · observable<T> PV streams · first EPICS conformance test in the suite
Igor Khokhriakov  ·  Principal Software Engineer
github.com/scientific-software-hub/rx-controls-suite
2 / 6
Pattern 1 Correlated Multi-PV Reads Beyond sequential pvxs::client::Context::get()
The scenario
"Every 500 ms, read BEAM:CURRENT and ORBIT:XPOS — only process the pair if both arrive in the same polling window. Discard if either times out."
With today's tools…
  • Two sequential ctx.get(pv).exec()->wait(5.0) — a timing gap
  • std::thread pair + std::future to parallelise — manual pairing and error handling
  • No built-in "atomic multi-PV read" in PVXS — every project rolls its own
EPICS C++ applications reimplementing this same parallel-read pattern from scratch is the rule, not the exception. Each variant handles edge cases slightly differently.
With Rx — rxcpp::observable<>::zip()
C++17 · observable<>::zip()
rxcpp::observable<>::interval(500ms)
  .flat_map([&ctx](long) {
    // Both reads fire on background threads in parallel
    return rxcpp::observable<>::zip(
      [](double c, double x) {
        return process(c, x); },
      rxepics::read_pv<double>(
          "BEAM:CURRENT", ctx),
      rxepics::read_pv<double>(
          "ORBIT:XPOS",   ctx)
    );
  })
  .subscribe([](auto result) {
    publish(result); });
✓  Zero std::promise. Zero shared state.
✓  If either PV times out the pair is silently dropped.
✓  PVXS Value extracted via .as<T>() — template selects type.
3 / 6
Pattern 2 Real-Time Monitor Stream Processing Beyond raw PVXS monitor callbacks
The scenario
"Monitor TEST:CALC (10 Hz IOC push), compute a 5-sample sliding average, write back only if it deviates more than 10% from the last written value."
With today's tools…
  • ctx.monitor(pv).event(callback).exec() — own ring buffer + write-guard variable
  • std::mutex on callback thread — manual serialisation every time
  • Cancellation: destroy the Monitor handle — easy to leak
EPICS C++ monitor code is callback spaghetti — same ring buffer and mutex pattern repeated in every project, each slightly different.
With Rx — monitor_pv + buffer + distinct_until_changed
C++17 · buffer(5,1) + distinct_until_changed
rxepics::monitor_pv<double>("TEST:CALC")

  // sliding window — no deque, no index arithmetic
  .buffer(5, 1)
  .map([](const auto& w) {
    return std::accumulate(
        w.begin(), w.end(), 0.0) / w.size();
  })

  // skip write if within 10% of last value
  .distinct_until_changed(
      [](double p, double c) {
        return std::abs(c-p)/p < 0.1; })

  .flat_map([&ctx](double avg) {
    return rxepics::write_pv<double>(
        "TEST:DOUBLE", avg, ctx); });
✓  No ring buffer. No mutex. No write-guard variable.
✓  Dispose: PVXS Monitor handle destroyed — subscription cancelled.
4 / 6
Pattern 3 Alarm Fan-In Beyond per-PV caMonitor registrations
The scenario
"Watch three PVs. When any one exceeds a threshold, notify the operator immediately. The others must continue monitoring unaffected."
With today's tools…
  • One ctx.monitor(pv).event(cb).exec() per PV — three registrations
  • Shared alarm queue + std::mutex for fan-in — or a dedicated thread
  • No first-class PVXS primitive for merging subscriptions
Each additional PV is another registration, another callback, another piece of shared state to protect.
With Rx — rxcpp::observable<>::merge()
C++17 · observable<>::merge()
rxcpp::observable<>::merge(
  rxepics::monitor_pv<double>("PV:ONE"),
  rxepics::monitor_pv<double>("PV:TWO"),
  rxepics::monitor_pv<double>("PV:THREE")
)
.filter([](double v) {
  return std::abs(v) > THRESHOLD; })
.subscribe([](double v) {
  notify_operator(v); });
✓  Any one out-of-range PV fires — others continue.
✓  Add a fourth PV: one line.
✓  PVXS callbacks serialised by monitor's internal mutex — no data races.
5 / 6
Library RxEpics/cpp — Data Flow reference implementation

EPICS IOC · PVXS Context

Single-shot reads via ctx.get(pv).exec()->wait(); writes via ctx.put(pv).set("value",v).exec(); push via ctx.monitor(pv).event(cb).exec().
No commands — write to a command PV instead.

Reactive Observable (single-shot)

read_pv<T> · write_pv<T>
Dispatched on std::thread; emits one value then completes. 5 s timeout via wait(5.0).

Reactive Observable (push)

monitor_pv<T> — PVXS monitor callback delivers values to rxcpp::subscriber<T> via shared_ptr<mutex>. Dispose destroys the Monitor handle → subscription cancelled.

RxCpp Operators

flat_map · zip · merge · buffer · filter · scan · sample_with_time · distinct_until_changed

Application / Downstream

Write-back, alarm, HMI — pure functions, no shared state.
First EPICS subproject in the suite with a ReactiveX conformance test.

Showstopper pipeline
Fluent EpicsClient chain
Read → calibrate → write — all in one chain; no intermediate variables.
EpicsClient — fluent builder
rxepics::EpicsClient()
  .read("TEST:CALC")
  .map([](std::any v) -> std::any {
    return std::any_cast<double>(v) * 1.0328;
  })
  .write("TEST:DOUBLE")
  .subscribe(
    [](std::any v) {
      std::cout << std::any_cast<double>(v);
    });
Header-only · CMake + FetchContent · C++17 · PVXS
Verify: EPICS_PVA_ADDR_LIST=localhost ./build/tests/verify_contract
6 / 6
Thank you
Same ReactiveX idioms.
Every system.
poll  ·  zip  ·  sliding average  ·  backpressure  ·  fluent pipelines
AGPL-3.0 · C++17 · CMake + FetchContent · RxCpp · PVXS
github.com/scientific-software-hub/rx-controls-suite
Python edition RxEpics/python · RxPY v4 · caproto Tango C++ edition RxTango/cpp · RxCpp · cppTango Tango Java edition RxTango/java · RxJava3 · jbang