From bbea7444594563d1b327daa15d39c79b622e592f Mon Sep 17 00:00:00 2001 From: Ayzen Date: Fri, 5 Jun 2026 14:40:10 +0300 Subject: [PATCH] some fixes --- a-librevna_minimal_driver_lifecycle.d | 14 + a-librevna_minimal_driver_protocol.d | 14 + a-librevna_minimal_driver_transport.d | 14 + .../common_cpp/ipc/include/shm_ring.hpp | 3 + .../common_cpp/ipc/src/shm_ring.cpp | 75 +++++- .../src/channel_bundle_support.cpp | 47 +++- .../data_processor/src/data_processor.cpp | 20 +- .../locator/include/locator/tcp_server.hpp | 17 +- .../locator/src/payload_builder.cpp | 17 ++ .../processing/locator/src/tcp_server.cpp | 128 ++++++++- .../librevna_minimal_driver_lifecycle.cpp | 88 +++++- .../librevna_minimal_driver_transport.cpp | 47 +++- .../librevna/librevna_protocol_common.hpp | 85 ++++++ .../switches/h7992_minimal_driver.cpp | 33 ++- .../switches/h7992_minimal_driver.hpp | 1 + .../switches/hmc349a_minimal_driver.cpp | 52 +++- .../switches/hmc349a_minimal_driver.hpp | 1 + .../src/sweep_orchestrator.cpp | 162 +++++++++-- python_app/gui/app_window.py | 55 ++++ python_app/gui/control_button.py | 23 +- .../app_window_control_button_mixin.py | 30 ++- .../controllers/app_window_pipeline_mixin.py | 101 ++++++- .../controllers/app_window_snapshot_mixin.py | 8 +- python_app/gui/main.py | 9 +- python_app/hardware_full/kamil_adc_service.py | 27 +- .../librevna_driver/transport/usb.py | 15 +- .../hardware_full/multi_device_service.py | 70 ++++- .../hardware_full/switch_drivers/gpio_uapi.py | 54 ++-- python_app/models/run_config_codec.py | 236 +++++++++------- python_app/models/run_config_validation.py | 152 +++++++++-- python_app/orchestration/config_writer.py | 18 +- .../orchestration/process_supervisor.py | 254 +++++++++++++++--- python_app/orchestration/shm/ring_reader.py | 24 +- python_app/orchestration/shm/ring_writer.py | 62 ++++- python_app/scripts/kamil_adc_raw_producer.py | 138 ++++++++-- 35 files changed, 1797 insertions(+), 297 deletions(-) create mode 100644 a-librevna_minimal_driver_lifecycle.d create mode 100644 a-librevna_minimal_driver_protocol.d create mode 100644 a-librevna_minimal_driver_transport.d diff --git a/a-librevna_minimal_driver_lifecycle.d b/a-librevna_minimal_driver_lifecycle.d new file mode 100644 index 0000000..f802458 --- /dev/null +++ b/a-librevna_minimal_driver_lifecycle.d @@ -0,0 +1,14 @@ +librevna_minimal_driver_lifecycle.o: \ + data_acq_and_processing/sweep_orchestrator/device_drivers/radar/librevna/librevna_minimal_driver_lifecycle.cpp \ + data_acq_and_processing/sweep_orchestrator/device_drivers/radar/librevna/../librevna_minimal_driver.hpp \ + data_acq_and_processing/sweep_orchestrator/device_drivers/interfaces/radar_driver.hpp \ + data_acq_and_processing/common_cpp/ipc/include/shared_types.hpp \ + data_acq_and_processing/common_cpp/config/include/run_config.hpp \ + data_acq_and_processing/processing/locator/include/locator/locator_config.hpp \ + data_acq_and_processing/sweep_orchestrator/device_drivers/radar/librevna/librevna_protocol_common.hpp +data_acq_and_processing/sweep_orchestrator/device_drivers/radar/librevna/../librevna_minimal_driver.hpp: +data_acq_and_processing/sweep_orchestrator/device_drivers/interfaces/radar_driver.hpp: +data_acq_and_processing/common_cpp/ipc/include/shared_types.hpp: +data_acq_and_processing/common_cpp/config/include/run_config.hpp: +data_acq_and_processing/processing/locator/include/locator/locator_config.hpp: +data_acq_and_processing/sweep_orchestrator/device_drivers/radar/librevna/librevna_protocol_common.hpp: diff --git a/a-librevna_minimal_driver_protocol.d b/a-librevna_minimal_driver_protocol.d new file mode 100644 index 0000000..9984a08 --- /dev/null +++ b/a-librevna_minimal_driver_protocol.d @@ -0,0 +1,14 @@ +librevna_minimal_driver_protocol.o: \ + data_acq_and_processing/sweep_orchestrator/device_drivers/radar/librevna/librevna_minimal_driver_protocol.cpp \ + data_acq_and_processing/sweep_orchestrator/device_drivers/radar/librevna/../librevna_minimal_driver.hpp \ + data_acq_and_processing/sweep_orchestrator/device_drivers/interfaces/radar_driver.hpp \ + data_acq_and_processing/common_cpp/ipc/include/shared_types.hpp \ + data_acq_and_processing/common_cpp/config/include/run_config.hpp \ + data_acq_and_processing/processing/locator/include/locator/locator_config.hpp \ + data_acq_and_processing/sweep_orchestrator/device_drivers/radar/librevna/librevna_protocol_common.hpp +data_acq_and_processing/sweep_orchestrator/device_drivers/radar/librevna/../librevna_minimal_driver.hpp: +data_acq_and_processing/sweep_orchestrator/device_drivers/interfaces/radar_driver.hpp: +data_acq_and_processing/common_cpp/ipc/include/shared_types.hpp: +data_acq_and_processing/common_cpp/config/include/run_config.hpp: +data_acq_and_processing/processing/locator/include/locator/locator_config.hpp: +data_acq_and_processing/sweep_orchestrator/device_drivers/radar/librevna/librevna_protocol_common.hpp: diff --git a/a-librevna_minimal_driver_transport.d b/a-librevna_minimal_driver_transport.d new file mode 100644 index 0000000..4a32529 --- /dev/null +++ b/a-librevna_minimal_driver_transport.d @@ -0,0 +1,14 @@ +librevna_minimal_driver_transport.o: \ + data_acq_and_processing/sweep_orchestrator/device_drivers/radar/librevna/librevna_minimal_driver_transport.cpp \ + data_acq_and_processing/sweep_orchestrator/device_drivers/radar/librevna/../librevna_minimal_driver.hpp \ + data_acq_and_processing/sweep_orchestrator/device_drivers/interfaces/radar_driver.hpp \ + data_acq_and_processing/common_cpp/ipc/include/shared_types.hpp \ + data_acq_and_processing/common_cpp/config/include/run_config.hpp \ + data_acq_and_processing/processing/locator/include/locator/locator_config.hpp \ + data_acq_and_processing/sweep_orchestrator/device_drivers/radar/librevna/librevna_protocol_common.hpp +data_acq_and_processing/sweep_orchestrator/device_drivers/radar/librevna/../librevna_minimal_driver.hpp: +data_acq_and_processing/sweep_orchestrator/device_drivers/interfaces/radar_driver.hpp: +data_acq_and_processing/common_cpp/ipc/include/shared_types.hpp: +data_acq_and_processing/common_cpp/config/include/run_config.hpp: +data_acq_and_processing/processing/locator/include/locator/locator_config.hpp: +data_acq_and_processing/sweep_orchestrator/device_drivers/radar/librevna/librevna_protocol_common.hpp: diff --git a/data_acq_and_processing/common_cpp/ipc/include/shm_ring.hpp b/data_acq_and_processing/common_cpp/ipc/include/shm_ring.hpp index 648147e..c1b42d7 100644 --- a/data_acq_and_processing/common_cpp/ipc/include/shm_ring.hpp +++ b/data_acq_and_processing/common_cpp/ipc/include/shm_ring.hpp @@ -56,6 +56,9 @@ class ShmRing { void close(); static void validate_name(const std::string& name); + // Validates that mapped_size covers the geometry declared in the header + // (fix #50: reject undersized/corrupt mappings to prevent OOB slot access). + static void validate_geometry(const Header& header, std::size_t mapped_size, const std::string& name); }; } // namespace radar::ipc diff --git a/data_acq_and_processing/common_cpp/ipc/src/shm_ring.cpp b/data_acq_and_processing/common_cpp/ipc/src/shm_ring.cpp index d683c0a..673eb8b 100644 --- a/data_acq_and_processing/common_cpp/ipc/src/shm_ring.cpp +++ b/data_acq_and_processing/common_cpp/ipc/src/shm_ring.cpp @@ -5,6 +5,8 @@ #include #include #include +#include +#include #include #include #include @@ -186,6 +188,14 @@ auto ShmRing::open_or_create( const bool geometry_ok = header->capacity == capacity && header->slot_size_bytes == slot_size_bytes; if (magic_ok && version_ok && geometry_ok) { + // Fix #50: the requested mapped_size matched the header geometry, but the + // backing file may have been created undersized by another process. Confirm + // st_size covers the geometry before trusting the mapping. + struct stat info {}; + if (fstat(fd, &info) != 0) { + throw errno_message("Failed to stat shared memory ring", name); + } + validate_geometry(*header, static_cast(info.st_size), name); break; } @@ -247,6 +257,8 @@ auto ShmRing::open_existing(const std::string& name) -> ShmRing { if (header->version != kRingVersion) { throw std::runtime_error("Shared memory ring version mismatch for " + name); } + // Fix #50: ensure the mapping actually spans every slot the header describes. + validate_geometry(*header, mapped_size, name); ShmRing ring{}; ring.fd_ = scoped_fd.release(); @@ -293,17 +305,28 @@ auto ShmRing::push(std::span payload) -> bool { } const auto write_seq = header_->write_seq.load(std::memory_order_relaxed); - const auto read_seq = header_->read_seq.load(std::memory_order_acquire); + auto read_seq = header_->read_seq.load(std::memory_order_acquire); if (checked_u64_diff(write_seq, read_seq) >= header_->capacity) { - header_->read_seq.store(read_seq + 1U, std::memory_order_release); - header_->dropped.fetch_add(1U, std::memory_order_relaxed); + // Ring full: overwrite the oldest unread slot. Advance read_seq via CAS so a + // concurrent consumer's advance is never clobbered (read_seq never moves back). + header_->read_seq.compare_exchange_strong( + read_seq, read_seq + 1U, std::memory_order_acq_rel, std::memory_order_acquire); + const auto dropped = header_->dropped.fetch_add(1U, std::memory_order_relaxed) + 1U; + if (dropped == 1U || dropped % 100U == 0U) { + std::cerr << "shm_ring: overflow overwrote an unread slot (dropped total=" << dropped << ")\n"; + } } auto* slot = slot_header(write_seq); + // Seqlock publish: write the payload + size FIRST, then store the slot sequence + // LAST (after a release fence). A consumer that observes sequence == write_seq+1 + // is therefore guaranteed a fully-written payload. (Publishing the sequence before + // the copy, as before, let a consumer copy a half-written slot.) slot->payload_size = static_cast(payload.size()); - slot->sequence = write_seq + 1U; std::memcpy(slot_payload(slot), payload.data(), payload.size()); + std::atomic_thread_fence(std::memory_order_release); + slot->sequence = write_seq + 1U; std::atomic_thread_fence(std::memory_order_release); header_->write_seq.store(write_seq + 1U, std::memory_order_release); @@ -323,21 +346,31 @@ auto ShmRing::pop(std::vector& payload) -> bool { auto* slot = slot_header(read_seq); if (slot->sequence != read_seq + 1U) { - // Producer overwrote this slot before consumer read it. Resync to latest. + // Slot not published for this seq, or already overwritten (lapped). Resync. header_->read_seq.store(write_seq, std::memory_order_release); return false; } const auto payload_size = slot->payload_size; if (payload_size > header_->slot_size_bytes) { + // Corrupt/torn slot: skip it (resync) rather than throw — a single bad slot + // must never abort the long-running consumer. header_->read_seq.store(write_seq, std::memory_order_release); - throw std::runtime_error("Invalid payload size in shared memory slot"); + return false; } + std::atomic_thread_fence(std::memory_order_acquire); payload.resize(payload_size); std::memcpy(payload.data(), slot_payload(slot), payload_size); + // Seqlock verify: if the slot sequence changed during the copy, the producer + // lapped us mid-read and the payload is torn — discard and resync. std::atomic_thread_fence(std::memory_order_acquire); + if (slot->sequence != read_seq + 1U) { + header_->read_seq.store(write_seq, std::memory_order_release); + return false; + } + header_->read_seq.store(read_seq + 1U, std::memory_order_release); return true; } @@ -377,4 +410,34 @@ void ShmRing::validate_name(const std::string& name) { } } +void ShmRing::validate_geometry(const Header& header, std::size_t mapped_size, const std::string& name) { + // Fix #50: derive the expected size from the header's own geometry fields and + // require the real mapping to cover it. A bogus capacity/slot_size or a truncated + // mapping would otherwise yield out-of-bounds slot offsets and a SIGSEGV. + const std::uint32_t capacity = header.capacity; + const std::uint32_t slot_size_bytes = header.slot_size_bytes; + if (capacity == 0U) { + throw std::runtime_error("Shared memory ring capacity is zero for " + name); + } + if (slot_size_bytes == 0U) { + throw std::runtime_error("Shared memory ring slot size is zero for " + name); + } + + constexpr auto kMax = std::numeric_limits::max(); + const auto slot_stride = sizeof(SlotHeader) + static_cast(slot_size_bytes); + // Guard each multiply/add against std::size_t overflow before computing expected_size. + if (slot_stride > kMax / static_cast(capacity)) { + throw std::runtime_error("Shared memory ring geometry overflows for " + name); + } + const auto slots_size = slot_stride * static_cast(capacity); + if (slots_size > kMax - sizeof(Header)) { + throw std::runtime_error("Shared memory ring geometry overflows for " + name); + } + const auto expected_size = sizeof(Header) + slots_size; + + if (mapped_size < expected_size) { + throw std::runtime_error("Shared memory ring size is too small for " + name); + } +} + } // namespace radar::ipc diff --git a/data_acq_and_processing/preprocessing/calibration_master/src/channel_bundle_support.cpp b/data_acq_and_processing/preprocessing/calibration_master/src/channel_bundle_support.cpp index 631a17e..ac8e945 100644 --- a/data_acq_and_processing/preprocessing/calibration_master/src/channel_bundle_support.cpp +++ b/data_acq_and_processing/preprocessing/calibration_master/src/channel_bundle_support.cpp @@ -6,6 +6,7 @@ #include #include #include +#include #include #include #include @@ -26,6 +27,10 @@ enum class RawTraceChannel { constexpr float kFrequencyRelativeTolerance = 1e-5F; constexpr float kFrequencyAbsoluteTolerance = 1e-3F; constexpr float kComplexMagnitudeSquaredEpsilon = 1e-12F; +// #30: OSL denominator must be well conditioned relative to the magnitudes of the terms forming it. +constexpr float kOslRelativeConditioning = 1e-6F; +// #30: maximum fraction of degenerate (fallback) OSL points tolerated per combo before load fails. +constexpr float kOslMaxDegenerateFraction = 0.05F; [[nodiscard]] auto combo_to_string(const ipc::ComboKey& combo) -> std::string { return "input=" + std::to_string(combo.input_pos) + " output=" + std::to_string(combo.output_pos); @@ -139,7 +144,14 @@ void ensure_frequency_axis_match( std::unordered_map traces_by_combo{}; traces_by_combo.reserve(collection.traces.size()); for (const auto& trace : collection.traces) { - traces_by_combo.insert_or_assign(trace.combo, to_channel_trace(trace, channel, bundle_label)); + // #47: reject duplicate combos instead of silently overwriting an earlier trace. + const auto inserted = + traces_by_combo.insert({trace.combo, to_channel_trace(trace, channel, bundle_label)}); + if (!inserted.second) { + throw std::runtime_error( + bundle_label + " bundle contains duplicate combo " + combo_to_string(trace.combo) + ": " + path + ); + } } if (traces_by_combo.empty()) { @@ -212,17 +224,30 @@ void ensure_combo_coverage( coefficients.source_match.resize(open_trace.frequency_hz.size()); coefficients.reflection_tracking.resize(open_trace.frequency_hz.size()); + // #30: track degenerate points so silently-substituted fallback coefficients cannot pass unnoticed. + std::size_t degenerate_points = 0; for (std::size_t index = 0; index < open_trace.frequency_hz.size(); ++index) { const auto load = to_std_complex(load_trace.samples[index]); const auto open_delta = to_std_complex(open_trace.samples[index]) - load; const auto short_delta = to_std_complex(short_trace.samples[index]) - load; const auto denominator = open_delta - short_delta; + // #30: require the denominator to be well conditioned both absolutely and relative to the + // magnitudes of the open/short deltas it is formed from, so near-cancellations are rejected. + const auto denominator_norm = std::norm(denominator); + const auto relative_scale = std::max(std::norm(open_delta), std::norm(short_delta)); + const auto relative_threshold = relative_scale * (kOslRelativeConditioning * kOslRelativeConditioning); + std::complex source_match = std::complex(0.0F, 0.0F); std::complex reflection_tracking = std::complex(1.0F, 0.0F); - if (std::norm(denominator) > kComplexMagnitudeSquaredEpsilon) { + if (denominator_norm > kComplexMagnitudeSquaredEpsilon && denominator_norm > relative_threshold) { source_match = (open_delta + short_delta) / denominator; reflection_tracking = open_delta * (std::complex(1.0F, 0.0F) - source_match); + } else { + // #30: degenerate point — log the offending combo/frequency and fall back to a pass-through. + ++degenerate_points; + std::cerr << "S11 calibration: degenerate OSL point for combo " << combo_to_string(combo) + << " at " << open_trace.frequency_hz[index] << " Hz (ill-conditioned denominator)\n"; } coefficients.directivity[index] = load_trace.samples[index]; @@ -230,6 +255,20 @@ void ensure_combo_coverage( coefficients.reflection_tracking[index] = from_std_complex(reflection_tracking); } + // #30: refuse to load a calibration whose per-combo fallback fraction exceeds the tolerated threshold. + if (!open_trace.frequency_hz.empty()) { + const auto degenerate_fraction = + static_cast(degenerate_points) / static_cast(open_trace.frequency_hz.size()); + if (degenerate_fraction > kOslMaxDegenerateFraction) { + throw std::runtime_error( + "S11 calibration for combo " + combo_to_string(combo) + " has " + + std::to_string(degenerate_points) + " of " + + std::to_string(open_trace.frequency_hz.size()) + + " degenerate OSL points (fraction exceeds tolerance)" + ); + } + } + return coefficients; } @@ -301,6 +340,8 @@ void S11CalibrationBundle::load( coefficients_by_combo_.clear(); if (open_path.empty() && short_path.empty() && load_path.empty()) { + // #38: S11 calibration disabled because no OSL paths were provided; surface it instead of silently skipping. + std::cerr << "S11 calibration: no open/short/load bundle paths provided; reflection correction disabled\n"; return; } if (open_path.empty() || short_path.empty() || load_path.empty()) { @@ -384,6 +425,8 @@ void S11ReferenceBundle::load(const std::string& path) { traces_by_combo_.clear(); if (path.empty()) { + // #38: S11 reference disabled because no path was provided; surface it instead of silently skipping. + std::cerr << "S11 reference: no bundle path provided; reference normalization disabled\n"; return; } diff --git a/data_acq_and_processing/processing/data_processor/src/data_processor.cpp b/data_acq_and_processing/processing/data_processor/src/data_processor.cpp index ffe4c59..69cb34c 100644 --- a/data_acq_and_processing/processing/data_processor/src/data_processor.cpp +++ b/data_acq_and_processing/processing/data_processor/src/data_processor.cpp @@ -4,6 +4,7 @@ #include #include #include +#include #include #include #include @@ -63,6 +64,10 @@ void DataProcessor::run(const std::atomic& stop_requested) { std::uint64_t last_replayed_revision = live_config_loader_.revision(); std::uint64_t last_applied_history_command_seq = 0; std::uint64_t error_count = 0; + // Fix #55: track the socket-fed speed used for the last reprocess so a change + // arriving without a live-config revision bump still triggers a reprocess of + // the current result (gated below by reprocess_current_result). + std::optional last_reprocessed_socket_speed = std::nullopt; while (!stop_requested.load(std::memory_order_relaxed)) { try { @@ -71,7 +76,17 @@ void DataProcessor::run(const std::atomic& stop_requested) { const auto live_revision = live_config_loader_.revision(); auto& processor = resolve_processor(live_config); - if (live_revision != last_replayed_revision) { + // The effective config folds in latest_socket_speed(); a change there is + // a reprocess trigger on its own, even when the live revision is unchanged. + // Mirror the gating in resolve_effective_live_config(): only consider the + // socket speed when it is actually applied to the effective config. + const std::optional current_socket_speed = + (live_config_raw.ignore_socket_speed || locator_server_ == nullptr) + ? std::nullopt + : locator_server_->latest_socket_speed(); + const bool socket_speed_dirty = current_socket_speed != last_reprocessed_socket_speed; + + if (live_revision != last_replayed_revision || socket_speed_dirty) { if (live_config.history_command_seq > last_applied_history_command_seq) { if (live_config.history_command == HistoryCommand::RemoveLast) { if (!preprocessed_history.empty()) { @@ -108,6 +123,9 @@ void DataProcessor::run(const std::atomic& stop_requested) { publish_locator(replay_result, live_config); } last_replayed_revision = live_revision; + // Fix #55: record the speed we just reprocessed with so an + // unchanged socket value does not retrigger every iteration. + last_reprocessed_socket_speed = current_socket_speed; } if (preprocessed_ring_.pop(bytes)) { diff --git a/data_acq_and_processing/processing/locator/include/locator/tcp_server.hpp b/data_acq_and_processing/processing/locator/include/locator/tcp_server.hpp index 6404695..b7bdf2b 100644 --- a/data_acq_and_processing/processing/locator/include/locator/tcp_server.hpp +++ b/data_acq_and_processing/processing/locator/include/locator/tcp_server.hpp @@ -64,8 +64,9 @@ class ClientSession { auto operator=(ClientSession&&) -> ClientSession& = delete; // Start writer/reader threads. The shared `vlc` slot is notified whenever - // an inbound packet carrying a finite `vlc` field arrives. - void start(std::atomic& shared_vlc_slot); + // an inbound packet carrying a finite `vlc` field arrives; the companion + // timestamp slot records when, so the server can expire stale speeds. + void start(std::atomic& shared_vlc_slot, std::atomic& shared_vlc_at_ns); // Enqueue one outbound packet. Disconnects this session if the queue is // already full or the writer has stopped. @@ -85,7 +86,7 @@ class ClientSession { private: void writer_loop(); - void reader_loop(std::atomic& shared_vlc_slot); + void reader_loop(std::atomic& shared_vlc_slot, std::atomic& shared_vlc_at_ns); int socket_fd_; std::string peer_name_; @@ -93,6 +94,9 @@ class ClientSession { std::uint32_t max_payload_bytes_; std::atomic stop_requested_{false}; std::atomic exited_{false}; + // Counts how many of the two loops (reader, writer) have finished. The + // second one to finish flips `exited_`, so a writer-only death is reapable. + std::atomic loops_finished_{0}; std::thread writer_thread_{}; std::thread reader_thread_{}; }; @@ -144,6 +148,10 @@ class TcpServer { void enroll_client(std::unique_ptr session); void broadcast_packet(const std::vector& packet); void reap_finished_clients(); + // Best-effort reap from the publish path: acquires clients_mutex_ via try-lock + // (skips if contended) and joins exited sessions outside the lock. Keeps + // clients_ bounded even when no new connections arrive to trigger a reap. + void try_reap_finished_clients(); void cache_latest_packet(std::vector packet); [[nodiscard]] auto latest_packet_copy() const -> std::optional>; @@ -160,6 +168,9 @@ class TcpServer { // Sentinel of "no value yet" is NaN. Lock-free read from data_processor. std::atomic latest_socket_speed_{}; + // Monotonic (steady_clock) nanoseconds at which latest_socket_speed_ was + // last written; used to expire stale socket-fed speeds. 0 means "never set". + std::atomic latest_socket_speed_at_ns_{0}; }; } // namespace radar::locator diff --git a/data_acq_and_processing/processing/locator/src/payload_builder.cpp b/data_acq_and_processing/processing/locator/src/payload_builder.cpp index a28badd..a55bb1d 100644 --- a/data_acq_and_processing/processing/locator/src/payload_builder.cpp +++ b/data_acq_and_processing/processing/locator/src/payload_builder.cpp @@ -95,6 +95,17 @@ constexpr std::uint32_t kMinTableColumns = 3U; // [x_m, z_m, score, ...] return stream.str(); } +// Wall-clock epoch milliseconds at packet-build time. Unlike the human-readable +// `tim` field (which has no date and rolls over at midnight), this is a +// monotonically increasing generation stamp consumers can use to compute the +// true age of a snapshot and detect a stalled pipeline. +[[nodiscard]] auto epoch_millis_now() -> std::int64_t { + using Clock = std::chrono::system_clock; + return std::chrono::duration_cast( + Clock::now().time_since_epoch() + ).count(); +} + void append_u32_little_endian(std::vector& buffer, std::uint32_t value) { buffer.push_back(static_cast(value & 0xFFU)); buffer.push_back(static_cast((value >> 8U) & 0xFFU)); @@ -181,9 +192,15 @@ auto build_payload_json( }); } + // Fix #48: stamp every packet with a numeric build-time generation + // (`gen`, epoch ms). A cached snapshot re-sent verbatim to a new client + // keeps its original `gen`, so a consumer can compute the snapshot's true + // age and flag a stalled pipeline (which `tim` alone cannot express, having + // no date and rolling over at midnight). const Json root{ {"ver", protocol_version}, {"tim", format_timestamp_now()}, + {"gen", epoch_millis_now()}, {"sts", status}, {"obs", std::move(obs_array)}, }; diff --git a/data_acq_and_processing/processing/locator/src/tcp_server.cpp b/data_acq_and_processing/processing/locator/src/tcp_server.cpp index e7fa61f..39682f6 100644 --- a/data_acq_and_processing/processing/locator/src/tcp_server.cpp +++ b/data_acq_and_processing/processing/locator/src/tcp_server.cpp @@ -33,6 +33,14 @@ using Json = nlohmann::json; constexpr std::size_t kPacketHeaderSize = 8U; // device_id u32 LE + payload_len u32 LE. +// Per-recv()/send() socket timeout. Bounds how long a wedged peer can stall a +// reader/writer loop and lets the reader re-check the stop flag periodically. +constexpr long kSocketTimeoutSeconds = 5; + +// A socket-fed `vlc` speed older than this is considered stale and ignored, so +// a client that disconnects (or stops sending) cannot pin a fixed speed forever. +constexpr std::chrono::seconds kSocketSpeedStaleAfter{5}; + // Best-effort full-write helper: loops over write() until everything is sent // or an error occurs. Returns false on socket error or peer disconnect. [[nodiscard]] auto write_all(int socket_fd, const std::uint8_t* data, std::size_t size) -> bool { @@ -48,6 +56,9 @@ constexpr std::size_t kPacketHeaderSize = 8U; // device_id u32 LE + payload_len if (errno == EINTR) { continue; } + // EAGAIN/EWOULDBLOCK here is a SO_SNDTIMEO send timeout: treat it as + // fatal so a stalled peer tears the session down rather than wedging + // the writer thread indefinitely. return false; } if (chunk == 0) { @@ -58,15 +69,25 @@ constexpr std::size_t kPacketHeaderSize = 8U; // device_id u32 LE + payload_len return true; } -// Best-effort exact-read helper. Returns false if the peer closed the socket -// or an unrecoverable error occurred before all bytes were read. -[[nodiscard]] auto read_exact(int socket_fd, std::uint8_t* data, std::size_t size) -> bool { +// Best-effort exact-read helper. Returns false if the peer closed the socket, +// an unrecoverable error occurred, or stop was requested before all bytes were +// read. A SO_RCVTIMEO timeout (EAGAIN/EWOULDBLOCK) is not fatal: we re-check the +// stop flag and retry so a wedged peer cannot keep the reader thread alive. +[[nodiscard]] auto read_exact( + int socket_fd, + std::uint8_t* data, + std::size_t size, + const std::atomic& stop_requested +) -> bool { std::size_t consumed = 0; while (consumed < size) { + if (stop_requested.load(std::memory_order_acquire)) { + return false; + } const auto chunk = ::recv(socket_fd, data + consumed, size - consumed, 0); if (chunk < 0) { - if (errno == EINTR) { - continue; + if (errno == EINTR || errno == EAGAIN || errno == EWOULDBLOCK) { + continue; // Interrupted or recv timeout: re-check stop and retry. } return false; } @@ -107,6 +128,14 @@ void apply_socket_keepalive(int socket_fd) { int yes = 1; (void)::setsockopt(socket_fd, IPPROTO_TCP, TCP_NODELAY, &yes, sizeof(yes)); (void)::setsockopt(socket_fd, SOL_SOCKET, SO_KEEPALIVE, &yes, sizeof(yes)); + // Bound blocking recv()/send() so a wedged peer cannot stall a reader/writer + // thread forever. recv() timeouts are retried (stop-aware); a send() timeout + // is treated as a fatal write failure that tears the session down. + timeval timeout{}; + timeout.tv_sec = kSocketTimeoutSeconds; + timeout.tv_usec = 0; + (void)::setsockopt(socket_fd, SOL_SOCKET, SO_RCVTIMEO, &timeout, sizeof(timeout)); + (void)::setsockopt(socket_fd, SOL_SOCKET, SO_SNDTIMEO, &timeout, sizeof(timeout)); } void shutdown_and_close(int& socket_fd) { @@ -192,9 +221,14 @@ ClientSession::~ClientSession() { shutdown_and_close(socket_fd_); } -void ClientSession::start(std::atomic& shared_vlc_slot) { +void ClientSession::start( + std::atomic& shared_vlc_slot, + std::atomic& shared_vlc_at_ns +) { writer_thread_ = std::thread([this]() { writer_loop(); }); - reader_thread_ = std::thread([this, &shared_vlc_slot]() { reader_loop(shared_vlc_slot); }); + reader_thread_ = std::thread([this, &shared_vlc_slot, &shared_vlc_at_ns]() { + reader_loop(shared_vlc_slot, shared_vlc_at_ns); + }); } void ClientSession::enqueue(std::vector packet) { @@ -245,15 +279,24 @@ void ClientSession::writer_loop() { } } request_stop(); - // Exited flag is set once both threads finish; reader_loop sets it. + // Mark exited only once BOTH loops have finished; otherwise a writer-only + // death (e.g. a send timeout while the reader still blocks on recv) would + // never be reaped. request_stop() above shuts the socket so the reader's + // recv() returns promptly. + if (loops_finished_.fetch_add(1, std::memory_order_acq_rel) + 1U == 2U) { + exited_.store(true, std::memory_order_release); + } } -void ClientSession::reader_loop(std::atomic& shared_vlc_slot) { +void ClientSession::reader_loop( + std::atomic& shared_vlc_slot, + std::atomic& shared_vlc_at_ns +) { std::array header_buffer{}; std::vector payload_buffer; while (!stop_requested_.load(std::memory_order_acquire)) { - if (!read_exact(socket_fd_, header_buffer.data(), header_buffer.size())) { + if (!read_exact(socket_fd_, header_buffer.data(), header_buffer.size(), stop_requested_)) { break; } const auto payload_len = decode_u32_little_endian(header_buffer.data() + 4U); @@ -266,7 +309,8 @@ void ClientSession::reader_loop(std::atomic& shared_vlc_slot) { } payload_buffer.assign(payload_len, std::uint8_t{0}); - if (payload_len > 0U && !read_exact(socket_fd_, payload_buffer.data(), payload_len)) { + if (payload_len > 0U + && !read_exact(socket_fd_, payload_buffer.data(), payload_len, stop_requested_)) { break; } @@ -277,7 +321,13 @@ void ClientSession::reader_loop(std::atomic& shared_vlc_slot) { if (found != json.end() && found->is_number()) { const double value = found->get(); if (std::isfinite(value)) { + // Stamp the value first, then the time, so a reader that + // observes a fresh timestamp also observes the matching value. shared_vlc_slot.store(value, std::memory_order_release); + const auto now_ns = std::chrono::duration_cast( + std::chrono::steady_clock::now().time_since_epoch() + ).count(); + shared_vlc_at_ns.store(now_ns, std::memory_order_release); } } } @@ -288,7 +338,11 @@ void ClientSession::reader_loop(std::atomic& shared_vlc_slot) { } } request_stop(); - exited_.store(true, std::memory_order_release); + // Mark exited only once BOTH loops have finished (see writer_loop), so a + // reader-only death still waits for the writer before this session is reaped. + if (loops_finished_.fetch_add(1, std::memory_order_acq_rel) + 1U == 2U) { + exited_.store(true, std::memory_order_release); + } } // ----- TcpServer ------------------------------------------------------------ @@ -409,11 +463,28 @@ void TcpServer::publish(const ipc::ResultCollection& collection, const FilterPar auto packet = encode_packet(payload_json, config_.device_id); cache_latest_packet(packet); broadcast_packet(packet); + // Reap exited sessions on the publish path too, so clients_ stays bounded + // even when no new connections arrive to trigger an accept-time reap. Uses a + // try-lock and joins outside the mutex, so publish() never blocks on it. + try_reap_finished_clients(); } auto TcpServer::latest_socket_speed() const -> std::optional { + // Read the timestamp first, then the value, mirroring the writer's order so a + // value seen here is at least as fresh as its timestamp. + const std::int64_t at_ns = latest_socket_speed_at_ns_.load(std::memory_order_acquire); const double value = latest_socket_speed_.load(std::memory_order_acquire); - if (std::isnan(value)) { + if (at_ns == 0 || std::isnan(value)) { + return std::nullopt; + } + // Expire stale socket-fed speeds: a client that stopped sending (or + // disconnected) must not pin a fixed speed indefinitely. + const auto now_ns = std::chrono::duration_cast( + std::chrono::steady_clock::now().time_since_epoch() + ).count(); + const auto stale_after_ns = + std::chrono::duration_cast(kSocketSpeedStaleAfter).count(); + if (now_ns - at_ns > stale_after_ns) { return std::nullopt; } return value; @@ -457,7 +528,7 @@ void TcpServer::acceptor_loop() { if (snapshot.has_value()) { session->enqueue(*snapshot); } - session->start(latest_socket_speed_); + session->start(latest_socket_speed_, latest_socket_speed_at_ns_); enroll_client(std::move(session)); } } @@ -496,6 +567,35 @@ void TcpServer::reap_finished_clients() { } } +void TcpServer::try_reap_finished_clients() { + std::vector> to_join; + { + // try_to_lock: if the acceptor already holds the mutex (it reaps too), + // skip this round rather than block the publish path. + std::unique_lock guard(clients_mutex_, std::try_to_lock); + if (!guard.owns_lock()) { + return; + } + auto first_dead = std::partition( + clients_.begin(), + clients_.end(), + [](const std::unique_ptr& session) { + return !session->has_exited(); + } + ); + for (auto it = first_dead; it != clients_.end(); ++it) { + to_join.push_back(std::move(*it)); + } + clients_.erase(first_dead, clients_.end()); + } + // Join outside the lock: a finished session's threads are done, so join() + // returns promptly without holding clients_mutex_. + for (auto& session : to_join) { + session->request_stop(); + session->join(); + } +} + void TcpServer::cache_latest_packet(std::vector packet) { std::lock_guard guard(latest_packet_mutex_); latest_packet_ = std::move(packet); diff --git a/data_acq_and_processing/sweep_orchestrator/device_drivers/radar/librevna/librevna_minimal_driver_lifecycle.cpp b/data_acq_and_processing/sweep_orchestrator/device_drivers/radar/librevna/librevna_minimal_driver_lifecycle.cpp index d68b3f1..a4932c0 100644 --- a/data_acq_and_processing/sweep_orchestrator/device_drivers/radar/librevna/librevna_minimal_driver_lifecycle.cpp +++ b/data_acq_and_processing/sweep_orchestrator/device_drivers/radar/librevna/librevna_minimal_driver_lifecycle.cpp @@ -5,6 +5,7 @@ #include #include #include +#include #include #include #include @@ -21,7 +22,26 @@ namespace detail = radar::drivers::librevna::detail; namespace { constexpr std::uint32_t kNativeAcquireMaxAttempts = 3U; -constexpr auto kNativeSweepResponseTimeout = std::chrono::milliseconds(1500); + +// Derive the budget the whole multi-point sweep is allowed to take before the +// first datapoint must arrive (#7): a fixed setup cost plus the expected dwell +// (points / IFBW) with a generous margin, clamped to the overall hard cap. The +// per-gap stall timeout then governs progress once datapoints start flowing. +[[nodiscard]] auto native_initial_sweep_budget(const config::RadarSweepSettings& sweep) + -> std::chrono::milliseconds { + const float points = std::max(1.0F, static_cast(sweep.points)); + const float if_bw = std::max(1.0F, sweep.if_bandwidth_hz); + const double dwell_ms = + (static_cast(points) / static_cast(if_bw)) * 1000.0 * + static_cast(detail::kNativeSweepDwellMargin); + const double total_ms = static_cast(detail::kNativeSweepSetupBudgetMs) + dwell_ms; + const auto budget = std::chrono::milliseconds(static_cast(total_ms)); + return std::clamp( + budget, + std::chrono::milliseconds(detail::kNativeSweepPerGapTimeoutMs), + std::chrono::milliseconds(detail::kNativeSweepHardCapMs) + ); +} // One synthetic GPR reflector. `range_m` is its physical depth, `reflection` // is the dimensionless complex reflection coefficient (|Γ| ≤ 1). @@ -78,12 +98,24 @@ constexpr auto kMockMinimumSweepDuration = std::chrono::microseconds(50); } [[nodiscard]] auto is_retryable_native_acquire_error(std::string_view message) -> bool { - constexpr std::array kRetryableSubstrings = { + // A stop request must not be retried: drain to the caller immediately (#26). + if (message.find("aborted by stop request") != std::string_view::npos) { + return false; + } + + constexpr std::array kRetryableSubstrings = { "Timeout waiting for expected LibreVNA packet type", "Timeout waiting for LibreVNA ACK", "LibreVNA returned NACK", "Failed to read LibreVNA USB bulk packet", "Failed to write LibreVNA USB bulk packet", + // #31: a transient USB error often triggers a brief device re-enumeration, + // so the device may be momentarily absent or its handle invalidated when + // we reconnect. Treat these as retryable; the reconnect path absorbs the + // re-enumeration with a short bounded discovery backoff. + "No compatible LibreVNA USB device found", + "LibreVNA native handle is not open", + "LibreVNA USB handle is not open", }; return std::any_of( @@ -153,10 +185,28 @@ auto LibreVnaMinimalDriver::acquire_sweep() -> SweepTrace { if (!is_retryable_native_acquire_error(last_message) || attempt == kNativeAcquireMaxAttempts) { break; } + if (detail::native_stop_requested()) { + break; // #26: do not waste reconnect attempts while shutting down. + } - // Recover from transient USB/protocol stalls by reconnecting the device. - close_native(); - open_native(); + // #31: recover from transient USB/protocol stalls without + // destroying the libusb_context. Release only the interface + // and handle, then reopen against the surviving context so + // we ride out a brief device re-enumeration instead of + // re-initialising libusb from scratch on every retry. + rx_buffer_.clear(); + packet_queue_.clear(); + if (usb_handle_ != nullptr && interface_claimed_) { + libusb_release_interface(usb_handle_, detail::kUsbInterface); + interface_claimed_ = false; + } + if (usb_handle_ != nullptr) { + libusb_close(usb_handle_); + usb_handle_ = nullptr; + } + protocol_version_ = 0; + device_num_ports_ = 0; + open_native(); // Reuses the surviving usb_context_ and retries discovery. } } @@ -282,12 +332,32 @@ auto LibreVnaMinimalDriver::acquire_native() -> SweepTrace { std::vector received(settings_.sweep.points, 0U); std::uint32_t received_count = 0; - const auto deadline = std::chrono::steady_clock::now() + kNativeSweepResponseTimeout; + // #7: the per-datapoint deadline is rebased every time a fresh point lands + // (or the first point must arrive within the size-derived budget), so a + // large sweep is not killed by one fixed ~1.5s timeout. An independent hard + // cap bounds the whole sweep against a device that streams forever. + const auto now = std::chrono::steady_clock::now(); + const auto hard_cap_deadline = now + std::chrono::milliseconds(detail::kNativeSweepHardCapMs); + auto gap_deadline = std::min(now + native_initial_sweep_budget(settings_.sweep), hard_cap_deadline); while (received_count < settings_.sweep.points) { + // #26: honour a stop request observed during the (chunked) USB wait so + // SIGTERM aborts the sweep promptly instead of after the full budget. + if (detail::native_stop_requested()) { + throw std::runtime_error("Native sweep aborted by stop request"); + } + if (std::chrono::steady_clock::now() >= hard_cap_deadline) { + throw std::runtime_error( + "Native sweep exceeded hard cap (" + std::to_string(received_count) + "/" + + std::to_string(settings_.sweep.points) + " points received)" + ); + } + + const auto wait_deadline = std::min(gap_deadline, hard_cap_deadline); + NativePacket packet{}; try { - packet = wait_for_packet(detail::kPacketVnaDatapoint, deadline); + packet = wait_for_packet(detail::kPacketVnaDatapoint, wait_deadline); } catch (const std::exception& exception) { throw std::runtime_error( "Timeout while collecting VNADatapoints (" + std::to_string(received_count) + "/" + @@ -311,6 +381,10 @@ auto LibreVnaMinimalDriver::acquire_native() -> SweepTrace { trace.frequency_hz[datapoint.point_number] = datapoint.frequency_hz; trace.s11[datapoint.point_number] = datapoint.s11; trace.s21[datapoint.point_number] = datapoint.s21; + + // Extend the per-gap stall timeout now that progress was made. + gap_deadline = std::chrono::steady_clock::now() + + std::chrono::milliseconds(detail::kNativeSweepPerGapTimeoutMs); } return trace; diff --git a/data_acq_and_processing/sweep_orchestrator/device_drivers/radar/librevna/librevna_minimal_driver_transport.cpp b/data_acq_and_processing/sweep_orchestrator/device_drivers/radar/librevna/librevna_minimal_driver_transport.cpp index bbfeb98..cdb3d53 100644 --- a/data_acq_and_processing/sweep_orchestrator/device_drivers/radar/librevna/librevna_minimal_driver_transport.cpp +++ b/data_acq_and_processing/sweep_orchestrator/device_drivers/radar/librevna/librevna_minimal_driver_transport.cpp @@ -3,10 +3,12 @@ #include #include #include +#include #include #include #include #include +#include #include #include @@ -104,18 +106,40 @@ namespace { } // namespace void LibreVnaMinimalDriver::open_native() { - if (usb_context_ != nullptr || usb_handle_ != nullptr) { + if (usb_handle_ != nullptr) { throw std::runtime_error("LibreVNA native state is already initialized"); } - const auto init_status = libusb_init(&usb_context_); - if (init_status != LIBUSB_SUCCESS) { - usb_context_ = nullptr; - throw std::runtime_error(libusb_error_message("Failed to initialize libusb", init_status)); + // #26: observe SIGINT/SIGTERM (chaining to main()'s handler) so the blocking + // USB loops can bail out promptly during shutdown. Safe to call repeatedly. + detail::install_native_stop_observer(); + + // #31: keep the libusb_context alive across reconnect retries. Only create a + // fresh context the first time; reconnect paths reuse the surviving context + // and merely reopen the handle, avoiding a full libusb teardown/re-init. + if (usb_context_ == nullptr) { + const auto init_status = libusb_init(&usb_context_); + if (init_status != LIBUSB_SUCCESS) { + usb_context_ = nullptr; + throw std::runtime_error(libusb_error_message("Failed to initialize libusb", init_status)); + } } try { - auto* selected_handle = find_matching_device_handle(usb_context_, settings_.serial); + // #31: retry discovery with a short bounded backoff so a brief device + // re-enumeration (common right after a transient USB error) is absorbed + // instead of surfacing as a hard "device not found". + libusb_device_handle* selected_handle = nullptr; + for (int attempt = 0; attempt < detail::kNativeDiscoveryRetryAttempts; ++attempt) { + if (detail::native_stop_requested()) { + break; // #26: abandon discovery promptly during shutdown. + } + selected_handle = find_matching_device_handle(usb_context_, settings_.serial); + if (selected_handle != nullptr) { + break; + } + std::this_thread::sleep_for(std::chrono::milliseconds(detail::kNativeDiscoveryRetryDelayMs)); + } if (selected_handle == nullptr) { const auto serial_hint = settings_.serial.empty() ? std::string() : " for serial '" + settings_.serial + "'"; @@ -215,6 +239,12 @@ void LibreVnaMinimalDriver::send_packet_no_payload(std::uint8_t packet_type, boo void LibreVnaMinimalDriver::wait_for_ack(std::chrono::steady_clock::time_point deadline) { while (std::chrono::steady_clock::now() < deadline) { + // #26: bail out promptly on SIGTERM/SIGINT rather than blocking until the + // ACK deadline expires. + if (detail::native_stop_requested()) { + throw std::runtime_error("LibreVNA ACK wait aborted by stop request"); + } + std::optional ack_or_nack{}; for (auto iter = packet_queue_.begin(); iter != packet_queue_.end(); ++iter) { if (iter->packet_type != detail::kPacketAck && iter->packet_type != detail::kPacketNack) { @@ -246,6 +276,11 @@ auto LibreVnaMinimalDriver::wait_for_packet( if (auto packet = pop_packet(expected_type); packet.has_value()) { return *packet; } + // #26: re-check the stop flag between short USB polls so a shutdown is + // honoured within a few hundred ms even on a long sweep deadline. + if (detail::native_stop_requested()) { + throw std::runtime_error("LibreVNA packet wait aborted by stop request"); + } pump_usb(deadline); } diff --git a/data_acq_and_processing/sweep_orchestrator/device_drivers/radar/librevna/librevna_protocol_common.hpp b/data_acq_and_processing/sweep_orchestrator/device_drivers/radar/librevna/librevna_protocol_common.hpp index 477976f..30ce612 100644 --- a/data_acq_and_processing/sweep_orchestrator/device_drivers/radar/librevna/librevna_protocol_common.hpp +++ b/data_acq_and_processing/sweep_orchestrator/device_drivers/radar/librevna/librevna_protocol_common.hpp @@ -1,6 +1,8 @@ #pragma once #include +#include +#include // POSIX sigaction/NSIG for the chaining stop observer (#26). #include #include #include @@ -51,6 +53,21 @@ constexpr std::size_t kUsbReadChunkBytes = 16 * 1024; constexpr int kUsbReadPollMinTimeoutMs = 1; constexpr int kUsbReadPollMaxTimeoutMs = 50; +// Native sweep timing budget (#7). The native deadline is no longer a single +// fixed value for the whole sweep: instead it is derived from the sweep size +// (fixed setup cost + per-point dwell ≈ points/IFBW) and is extended every +// time a fresh datapoint arrives via a per-gap stall timeout, while an overall +// hard cap bounds a wedged device. +constexpr int kNativeSweepSetupBudgetMs = 500; +constexpr float kNativeSweepDwellMargin = 3.0F; +constexpr int kNativeSweepPerGapTimeoutMs = 1500; +constexpr int kNativeSweepHardCapMs = 60'000; + +// Reconnect backoff (#31). After a transient error the device may re-enumerate; +// retry discovery a few times with a short bounded delay before giving up. +constexpr int kNativeDiscoveryRetryAttempts = 10; +constexpr int kNativeDiscoveryRetryDelayMs = 150; + [[nodiscard]] inline auto read_u16_le(std::span data, std::size_t offset) -> std::uint16_t { if ((offset + sizeof(std::uint16_t)) > data.size()) { throw std::runtime_error("Failed to decode uint16 from payload"); @@ -108,5 +125,73 @@ inline void write_u32_le(std::span data, std::size_t offset, std:: data[offset + 3] = static_cast((value >> 24U) & 0xFFU); } +// Stop responsiveness (#26). The blocking USB loops keep their per-call +// timeouts short and re-poll this flag so a SIGTERM/SIGINT delivered mid-sweep +// is honoured within a few hundred ms instead of only after the whole sweep +// budget has elapsed. +// +// `main()` installs its own SIGINT/SIGTERM handlers before any driver is +// opened, so we chain to (and preserve) whatever handler is already installed +// rather than clobbering it. All state lives in single inline-static slots so +// every translation unit observes the same flag and chain table. +[[nodiscard]] inline auto native_stop_flag() -> volatile std::sig_atomic_t& { + static volatile std::sig_atomic_t flag = 0; + return flag; +} + +[[nodiscard]] inline auto native_chained_handlers() -> std::array& { + static std::array handlers{}; + return handlers; +} + +inline void native_stop_signal_handler(int signal_number) { + native_stop_flag() = 1; + + // Chain to the previously installed disposition (e.g. main()'s handler) so + // process-level stop semantics are unchanged. + if (signal_number < 0 || signal_number >= NSIG) { + return; + } + const auto& prior = native_chained_handlers()[static_cast(signal_number)]; + if ((prior.sa_flags & SA_SIGINFO) == 0 && prior.sa_handler != nullptr && + prior.sa_handler != SIG_DFL && prior.sa_handler != SIG_IGN && + prior.sa_handler != native_stop_signal_handler) { + prior.sa_handler(signal_number); + } +} + +// Lazily install the chaining stop observer for one signal. Must be called +// after main() has installed its handlers (i.e. from open_native()). +inline void install_native_stop_observer_for(int signal_number) { + if (signal_number < 0 || signal_number >= NSIG) { + return; + } + + struct sigaction current{}; + if (sigaction(signal_number, nullptr, ¤t) != 0) { + return; + } + if (current.sa_handler == native_stop_signal_handler) { + return; // Already installed; do not chain to ourselves. + } + + native_chained_handlers()[static_cast(signal_number)] = current; + + struct sigaction action{}; + action.sa_handler = native_stop_signal_handler; + sigemptyset(&action.sa_mask); + action.sa_flags = current.sa_flags & ~SA_SIGINFO; // Keep flags such as SA_RESTART. + sigaction(signal_number, &action, nullptr); +} + +inline void install_native_stop_observer() { + install_native_stop_observer_for(SIGINT); + install_native_stop_observer_for(SIGTERM); +} + +[[nodiscard]] inline auto native_stop_requested() -> bool { + return native_stop_flag() != 0; +} + } // namespace radar::drivers::librevna::detail diff --git a/data_acq_and_processing/sweep_orchestrator/device_drivers/switches/h7992_minimal_driver.cpp b/data_acq_and_processing/sweep_orchestrator/device_drivers/switches/h7992_minimal_driver.cpp index 74fe792..3032d5b 100644 --- a/data_acq_and_processing/sweep_orchestrator/device_drivers/switches/h7992_minimal_driver.cpp +++ b/data_acq_and_processing/sweep_orchestrator/device_drivers/switches/h7992_minimal_driver.cpp @@ -14,6 +14,10 @@ namespace radar::drivers { namespace { +// A transient ioctl failure (e.g. EINTR/EAGAIN/EBUSY under load) should not be +// immediately fatal; retry a small number of times before giving up. +constexpr int kSwitchIoctlRetries = 3; + // Position mapping for H7992 control pins: // position -> (A, B) constexpr std::array, 4> kPositionToAB = { @@ -44,6 +48,15 @@ void validate_open_settings(const H7992MinimalDriverSettings& settings) { H7992MinimalDriver::H7992MinimalDriver(H7992MinimalDriverSettings settings) : settings_(std::move(settings)) {} +H7992MinimalDriver::~H7992MinimalDriver() { + // Safety net: if the owner never called close() (e.g. on crash/unwind), + // still drive the RF path to a known-safe state and release the GPIO lines. + try { + close(); + } catch (...) { + } +} + void H7992MinimalDriver::open() { if (is_open_) { return; @@ -134,6 +147,15 @@ void H7992MinimalDriver::open_native() { } void H7992MinimalDriver::close_native() { + // Leave the RF path in its known-safe default position before releasing the + // lines, so shutdown/crash never strands the switch in an arbitrary state. + if (line_fd_ >= 0 && settings_.default_position < settings_.positions) { + const auto [pin_a_state, pin_b_state] = kPositionToAB[settings_.default_position]; + auto values = make_line_values(pin_a_state, pin_b_state); + // Best-effort: do not throw out of the teardown path. + ::ioctl(line_fd_, GPIO_V2_LINE_SET_VALUES_IOCTL, &values); + } + if (line_fd_ >= 0) { ::close(line_fd_); line_fd_ = -1; @@ -152,9 +174,16 @@ void H7992MinimalDriver::switch_native(std::uint32_t position) { const auto [pin_a_state, pin_b_state] = kPositionToAB[position]; auto values = make_line_values(pin_a_state, pin_b_state); - if (::ioctl(line_fd_, GPIO_V2_LINE_SET_VALUES_IOCTL, &values) != 0) { - throw std::runtime_error("Failed to switch GPIO state for " + settings_.name + ": " + std::strerror(errno)); + // Retry transient ioctl failures so a momentary hiccup is recoverable + // instead of aborting the whole sweep; only the final attempt is fatal. + int last_errno = 0; + for (int attempt = 0; attempt < kSwitchIoctlRetries; ++attempt) { + if (::ioctl(line_fd_, GPIO_V2_LINE_SET_VALUES_IOCTL, &values) == 0) { + return; + } + last_errno = errno; } + throw std::runtime_error("Failed to switch GPIO state for " + settings_.name + ": " + std::strerror(last_errno)); } } // namespace radar::drivers diff --git a/data_acq_and_processing/sweep_orchestrator/device_drivers/switches/h7992_minimal_driver.hpp b/data_acq_and_processing/sweep_orchestrator/device_drivers/switches/h7992_minimal_driver.hpp index d927ebb..3133e20 100644 --- a/data_acq_and_processing/sweep_orchestrator/device_drivers/switches/h7992_minimal_driver.hpp +++ b/data_acq_and_processing/sweep_orchestrator/device_drivers/switches/h7992_minimal_driver.hpp @@ -30,6 +30,7 @@ struct H7992MinimalDriverSettings { class H7992MinimalDriver final : public SwitchDriver { public: explicit H7992MinimalDriver(H7992MinimalDriverSettings settings); + ~H7992MinimalDriver() override; // Leaves RF path safe even if close() was skipped. void open() override; void close() override; diff --git a/data_acq_and_processing/sweep_orchestrator/device_drivers/switches/hmc349a_minimal_driver.cpp b/data_acq_and_processing/sweep_orchestrator/device_drivers/switches/hmc349a_minimal_driver.cpp index 774ac82..51067d0 100644 --- a/data_acq_and_processing/sweep_orchestrator/device_drivers/switches/hmc349a_minimal_driver.cpp +++ b/data_acq_and_processing/sweep_orchestrator/device_drivers/switches/hmc349a_minimal_driver.cpp @@ -13,6 +13,23 @@ namespace radar::drivers { namespace { +// A transient ioctl failure (e.g. EINTR/EAGAIN/EBUSY under load) should not be +// immediately fatal; retry a small number of times before giving up. +constexpr int kSwitchIoctlRetries = 3; + +// Compute the GPIO line values for a given logical position, honouring inverted +// control logic. Shared by switch_native() and the teardown default-drive path. +[[nodiscard]] auto make_line_values(const HMC349AMinimalDriverSettings& settings, std::uint32_t position) + -> gpio_v2_line_values { + const std::uint8_t requested_state = static_cast(position & 0x01U); + const std::uint8_t control_state = settings.invert_logic ? (requested_state ^ 0x01U) : requested_state; + + gpio_v2_line_values values{}; + values.mask = (1ULL << 0U); + values.bits = static_cast(control_state); + return values; +} + void validate_open_settings(const HMC349AMinimalDriverSettings& settings) { if (settings.positions == 0U || settings.positions > 2U) { throw std::runtime_error("HMC349A switch positions must be in range [1, 2] for " + settings.name); @@ -27,6 +44,15 @@ void validate_open_settings(const HMC349AMinimalDriverSettings& settings) { HMC349AMinimalDriver::HMC349AMinimalDriver(HMC349AMinimalDriverSettings settings) : settings_(std::move(settings)) {} +HMC349AMinimalDriver::~HMC349AMinimalDriver() { + // Safety net: if the owner never called close() (e.g. on crash/unwind), + // still drive the RF path to a known-safe state and release the GPIO lines. + try { + close(); + } catch (...) { + } +} + void HMC349AMinimalDriver::open() { if (is_open_) { return; @@ -117,6 +143,14 @@ void HMC349AMinimalDriver::open_native() { } void HMC349AMinimalDriver::close_native() { + // Leave the RF path in its known-safe default position before releasing the + // line, so shutdown/crash never strands the switch in an arbitrary state. + if (line_fd_ >= 0 && settings_.default_position < settings_.positions) { + auto values = make_line_values(settings_, settings_.default_position); + // Best-effort: do not throw out of the teardown path. + ::ioctl(line_fd_, GPIO_V2_LINE_SET_VALUES_IOCTL, &values); + } + if (line_fd_ >= 0) { ::close(line_fd_); line_fd_ = -1; @@ -132,16 +166,18 @@ void HMC349AMinimalDriver::switch_native(std::uint32_t position) { throw std::runtime_error("Native GPIO line fd is not open for " + settings_.name); } - const std::uint8_t requested_state = static_cast(position & 0x01U); - const std::uint8_t control_state = settings_.invert_logic ? (requested_state ^ 0x01U) : requested_state; + auto values = make_line_values(settings_, position); - gpio_v2_line_values values{}; - values.mask = (1ULL << 0U); - values.bits = static_cast(control_state); - - if (::ioctl(line_fd_, GPIO_V2_LINE_SET_VALUES_IOCTL, &values) != 0) { - throw std::runtime_error("Failed to switch GPIO state for " + settings_.name + ": " + std::strerror(errno)); + // Retry transient ioctl failures so a momentary hiccup is recoverable + // instead of aborting the whole sweep; only the final attempt is fatal. + int last_errno = 0; + for (int attempt = 0; attempt < kSwitchIoctlRetries; ++attempt) { + if (::ioctl(line_fd_, GPIO_V2_LINE_SET_VALUES_IOCTL, &values) == 0) { + return; + } + last_errno = errno; } + throw std::runtime_error("Failed to switch GPIO state for " + settings_.name + ": " + std::strerror(last_errno)); } } // namespace radar::drivers diff --git a/data_acq_and_processing/sweep_orchestrator/device_drivers/switches/hmc349a_minimal_driver.hpp b/data_acq_and_processing/sweep_orchestrator/device_drivers/switches/hmc349a_minimal_driver.hpp index c70c0c6..057da95 100644 --- a/data_acq_and_processing/sweep_orchestrator/device_drivers/switches/hmc349a_minimal_driver.hpp +++ b/data_acq_and_processing/sweep_orchestrator/device_drivers/switches/hmc349a_minimal_driver.hpp @@ -30,6 +30,7 @@ struct HMC349AMinimalDriverSettings { class HMC349AMinimalDriver final : public SwitchDriver { public: explicit HMC349AMinimalDriver(HMC349AMinimalDriverSettings settings); + ~HMC349AMinimalDriver() override; // Leaves RF path safe even if close() was skipped. void open() override; void close() override; diff --git a/data_acq_and_processing/sweep_orchestrator/src/sweep_orchestrator.cpp b/data_acq_and_processing/sweep_orchestrator/src/sweep_orchestrator.cpp index 1a75aa1..d03ff72 100644 --- a/data_acq_and_processing/sweep_orchestrator/src/sweep_orchestrator.cpp +++ b/data_acq_and_processing/sweep_orchestrator/src/sweep_orchestrator.cpp @@ -1,7 +1,10 @@ #include "sweep_orchestrator.hpp" +#include #include +#include #include +#include #include #include #include @@ -9,6 +12,16 @@ namespace radar::acq { namespace { +// Wait-for-device retry tuning (mirror of python matrix_raw_producer._open_radar_with_retry): +// a device that is absent at boot or disappears mid-run must never kill the orchestrator, +// only make it wait. Backoff is capped so a long absence does not busy-spin, and every wait +// is interruptible by stop_requested for a prompt clean exit. +constexpr std::uint32_t kOpenRetryMinMs = 1'000U; +constexpr std::uint32_t kOpenRetryMaxMs = 10'000U; +// Throttle open-failure logging during a long wait so a permanently absent device does not +// flood the process log: log the first failure, then every Nth attempt. +constexpr std::uint64_t kOpenRetryLogEvery = 30U; + [[nodiscard]] inline auto should_stop(const std::atomic& stop_requested) -> bool { return stop_requested.load(std::memory_order_relaxed); } @@ -20,6 +33,21 @@ void sleep_if_needed_ms(std::uint32_t delay_ms) { std::this_thread::sleep_for(std::chrono::milliseconds(delay_ms)); } +// Interruptible sleep: returns true if a stop was requested while waiting. +[[nodiscard]] auto interruptible_sleep_ms(std::uint32_t delay_ms, const std::atomic& stop_requested) -> bool { + constexpr std::uint32_t kPollMs = 50U; + std::uint32_t waited = 0U; + while (waited < delay_ms) { + if (should_stop(stop_requested)) { + return true; + } + const std::uint32_t chunk = std::min(kPollMs, delay_ms - waited); + std::this_thread::sleep_for(std::chrono::milliseconds(chunk)); + waited += chunk; + } + return should_stop(stop_requested); +} + void validate_sweep(const drivers::SweepTrace& sweep) { if (sweep.frequency_hz.size() != sweep.s11.size() || sweep.frequency_hz.size() != sweep.s21.size()) { throw std::runtime_error("Radar driver returned inconsistent sweep vectors"); @@ -64,6 +92,41 @@ class DriverLifecycleGuard { active_ = true; } + // Open all devices, retrying forever with capped exponential backoff until success or + // stop. Used for both the initial open and every in-loop reopen, so a device that is + // absent at boot or disappears mid-run never kills the orchestrator — it just waits. + // Returns false only if a stop was requested before any device became available. + [[nodiscard]] auto open_all_with_retry(const std::atomic& stop_requested) -> bool { + // Drop any partial state from a previous open before retrying. + close_all(); + + std::uint64_t attempt = 0; + std::uint32_t delay_ms = kOpenRetryMinMs; + while (!should_stop(stop_requested)) { + try { + open_all(); + if (attempt > 0) { + std::cerr << "sweep_orchestrator: devices opened after " << (attempt + 1) << " attempt(s)\n"; + } + return true; + } catch (const std::exception& exc) { + // Best-effort: drop any partial open before the next attempt. + close_all(); + ++attempt; + if (attempt == 1 || attempt % kOpenRetryLogEvery == 0) { + std::cerr << "sweep_orchestrator: devices not available (attempt " << attempt + << "); retrying up to every " << (kOpenRetryMaxMs / 1'000U) + << "s until present: " << exc.what() << '\n'; + } + if (interruptible_sleep_ms(delay_ms, stop_requested)) { + return false; + } + delay_ms = std::min(delay_ms * 2U, kOpenRetryMaxMs); + } + } + return false; + } + void close_all() { if (!active_) { return; @@ -81,14 +144,47 @@ class DriverLifecycleGuard { bool active_ = false; }; -void publish_collection(ipc::ShmRing& raw_ring, ipc::ShmRing* raw_tap_ring, const ipc::RawSweepCollection& collection) { +// Worst-case serialized size of a collection given the configured combo count and sweep +// point count, using the trace wire format (see ipc::write_trace_collection/write_trace_block): +// collection header: magic(4) + collection_id(8) + monotonic_ns(8) + trace_count(4) +// + capture_start_ns(8) + capture_end_ns(8) = 40 bytes +// per trace block: input_pos(4) + output_pos(4) + point_count(4) = 12 bytes +// + per point: frequency(4) + s11(8) + s21(8) = 20 bytes +[[nodiscard]] auto worst_case_serialized_bytes(std::size_t combo_count, std::uint32_t sweep_points) -> std::size_t { + constexpr std::size_t kCollectionHeaderBytes = 40U; + constexpr std::size_t kTraceHeaderBytes = 12U; + constexpr std::size_t kBytesPerPoint = 20U; + const std::size_t per_trace = kTraceHeaderBytes + (static_cast(sweep_points) * kBytesPerPoint); + return kCollectionHeaderBytes + (combo_count * per_trace); +} + +// Publish a collection on the primary raw ring; tap pushes are strictly best-effort. +// Never throws: an oversize payload is logged-and-dropped (the slot-size budget is also +// validated up front at startup, so this guards only against unexpected runtime growth). +// Returns true if the primary push succeeded. +[[nodiscard]] auto publish_collection( + ipc::ShmRing& raw_ring, + ipc::ShmRing* raw_tap_ring, + const ipc::RawSweepCollection& collection, + std::uint64_t& oversize_drop_count +) -> bool { const auto serialized_collection = ipc::serialize_raw_collection(collection); if (!raw_ring.push(serialized_collection)) { - throw std::runtime_error("Raw ring slot is too small for serialized collection"); + // Log-and-drop an oversize payload rather than aborting the long-running loop. + // Throttle so a persistently oversize payload cannot flood the log. + if (oversize_drop_count % 100 == 0) { + std::cerr << "sweep_orchestrator: dropped oversize raw payload (" << serialized_collection.size() + << " bytes > slot " << raw_ring.slot_size_bytes() << "; drop count=" + << (oversize_drop_count + 1) << ")\n"; + } + ++oversize_drop_count; + return false; } - if (raw_tap_ring != nullptr && !raw_tap_ring->push(serialized_collection)) { - throw std::runtime_error("Raw tap ring slot is too small for serialized collection"); + // Tap ring is for GUI/debug observers only: a tap failure must never abort the primary path. + if (raw_tap_ring != nullptr) { + (void)raw_tap_ring->push(serialized_collection); } + return true; } } // namespace @@ -109,25 +205,59 @@ SweepOrchestrator::SweepOrchestrator( raw_tap_ring_(raw_tap_ring) {} void SweepOrchestrator::run(const std::atomic& stop_requested) { + // Fail fast on a config error: a slot that is too small for the worst-case payload can + // never carry a full collection, so report it clearly at startup instead of dropping + // every collection at runtime (fix #27). + const auto worst_case_bytes = worst_case_serialized_bytes(config_.run_combos.size(), config_.radar.sweep.points); + if (worst_case_bytes > raw_ring_.slot_size_bytes()) { + throw std::runtime_error( + "Raw ring slot_size_bytes (" + std::to_string(raw_ring_.slot_size_bytes()) + + ") is too small for the worst-case serialized collection (" + std::to_string(worst_case_bytes) + + " bytes for " + std::to_string(config_.run_combos.size()) + " combos x " + + std::to_string(config_.radar.sweep.points) + " points)" + ); + } + DriverLifecycleGuard lifecycle_guard(radar_driver_, input_switch_driver_, output_switch_driver_); - // Open devices once and keep them active for the whole acquisition loop. - lifecycle_guard.open_all(); + // Wait for the devices to become available before starting (fix #8/#9): an absent device + // makes the orchestrator wait, not exit. + if (!lifecycle_guard.open_all_with_retry(stop_requested)) { + return; // stop requested before any device became available + } std::uint64_t collection_id = 0; + std::uint64_t oversize_drop_count = 0; while (!should_stop(stop_requested)) { - auto raw_collection = acquire_one_collection(++collection_id, stop_requested); - if (raw_collection.traces.empty()) { + try { + auto raw_collection = acquire_one_collection(++collection_id, stop_requested); + if (raw_collection.traces.empty()) { + if (!config_.runtime.continuous) { + break; + } + sleep_if_needed_ms(config_.runtime.idle_sleep_ms); + continue; + } + + (void)publish_collection(raw_ring_, raw_tap_ring_, raw_collection, oversize_drop_count); + if (!config_.runtime.continuous) { break; } - sleep_if_needed_ms(config_.runtime.idle_sleep_ms); - continue; - } - - publish_collection(raw_ring_, raw_tap_ring_, raw_collection); - - if (!config_.runtime.continuous) { - break; + } catch (const std::exception& exc) { + // Treat any in-loop device error (timeout/NACK/transient USB or socket glitch/ + // device-not-found-after-glitch) as recoverable: close, wait for the device to + // come back, and continue. This mirrors the python producer's reconnect-forever + // policy so a long-running headless appliance survives transient hardware hiccups + // instead of exiting. Genuinely fatal config errors are caught at startup above + // (and in main()), reserving non-zero exit for those. + std::cerr << "sweep_orchestrator: acquisition failed; reopening and waiting for the device: " + << exc.what() << '\n'; + if (!config_.runtime.continuous) { + throw; // single-run mode has no recovery path; surface the failure + } + if (!lifecycle_guard.open_all_with_retry(stop_requested)) { + break; // stop requested while waiting to reopen + } } } diff --git a/python_app/gui/app_window.py b/python_app/gui/app_window.py index 2108a89..2b975f6 100644 --- a/python_app/gui/app_window.py +++ b/python_app/gui/app_window.py @@ -7,9 +7,12 @@ state shared across them (runtime services, readers, history buffers, timer). from __future__ import annotations from collections import deque +from contextlib import suppress from datetime import datetime import html import json +import logging +from logging.handlers import RotatingFileHandler import os from pathlib import Path import sys @@ -59,6 +62,7 @@ class AppWindow( super().__init__() self._init_paths(project_root) + self._init_headless_logger() self._init_runtime_services() self._init_config_profile_state() self._init_reader_handles() @@ -76,6 +80,42 @@ class AppWindow( self._root_profile_path = project_root / "run_config.json" self._active_profile_path = self._root_profile_path self._pending_startup_log_entries: list[tuple[str, str, str | None]] = [] + # Guards closeEvent against re-entrant teardown (e.g. a second signal). + self._closing = False + + def _init_headless_logger(self) -> None: + """Create a Python logger so headless WARN/ERROR reach journald and disk. + + In headless mode the in-app log only reaches an offscreen widget, so an + operator (or `journalctl`) would never see failures. We attach a stderr + StreamHandler (captured by journald) plus a small rotating file under + `runtime/logs`; in GUI mode no handler is attached and the logger stays + inert, preserving the visible log widget as the sole sink. + """ + self._headless_logger: logging.Logger | None = None + if not self._is_truthy_env("RADAR_SYSTEM_HEADLESS"): + return + logger = logging.getLogger("radar_system.gui") + logger.setLevel(logging.WARNING) + logger.propagate = False + logger.handlers.clear() + formatter = logging.Formatter( + fmt="%(asctime)s | %(levelname)-5s | %(message)s", + datefmt="%Y-%m-%d %H:%M:%S", + ) + stream_handler = logging.StreamHandler(stream=sys.stderr) + stream_handler.setFormatter(formatter) + logger.addHandler(stream_handler) + # A rotating file keeps recent failures around after a journald restart. + with suppress(Exception): + log_dir = self._project_root / "python_app/runtime/logs" + log_dir.mkdir(parents=True, exist_ok=True) + file_handler = RotatingFileHandler( + log_dir / "gui.log", maxBytes=1_000_000, backupCount=3, encoding="utf-8" + ) + file_handler.setFormatter(formatter) + logger.addHandler(file_handler) + self._headless_logger = logger def _init_runtime_services(self) -> None: """Initialize long-lived service objects used by mixins.""" @@ -461,6 +501,15 @@ class AppWindow( if level_upper == "ERROR" and hasattr(self, "_status_label"): self._status_label.setText("Status: error") + # In headless mode the offscreen widget above is invisible, so also mirror + # WARN/ERROR to the Python logger (stderr -> journald, plus rotating file) + # where an operator can actually observe failures. + headless_logger = getattr(self, "_headless_logger", None) + if headless_logger is not None and level_upper in {"WARN", "ERROR"}: + log_message = text if not details else f"{text}\n{details}" + log_level = logging.ERROR if level_upper == "ERROR" else logging.WARNING + headless_logger.log(log_level, log_message) + def _log(self, text: str, *, once_key: str | None = None) -> None: """Append informational message to runtime log panel.""" self._append_log_entry("INFO", text, once_key=once_key) @@ -564,6 +613,12 @@ class AppWindow( def closeEvent(self, event) -> None: # noqa: N802 """Ensure workers and dialogs are closed before window destruction.""" + if self._closing: + # Re-entrant close (second signal, or window.close() after the event loop + # already returned): teardown is in progress or done — do nothing more. + super().closeEvent(event) + return + self._closing = True try: # 0) Stop the GPIO button watcher so a late press cannot start work. self._stop_control_button_watcher() diff --git a/python_app/gui/control_button.py b/python_app/gui/control_button.py index 1da5446..a24a33a 100644 --- a/python_app/gui/control_button.py +++ b/python_app/gui/control_button.py @@ -86,12 +86,23 @@ class ControlButtonWatcher(QObject): def start(self) -> None: """Open the GPIO line and begin watching for presses on a background thread.""" - self._line.open() - self._stop_read_fd, self._stop_write_fd = os.pipe() - self._thread = threading.Thread( - target=self._run, name="control-button-watcher", daemon=True - ) - self._thread.start() + # Guard against double-start: a second start would leak the first line/pipe/thread. + if self._thread is not None: + return + try: + self._line.open() + self._stop_read_fd, self._stop_write_fd = os.pipe() + self._thread = threading.Thread( + target=self._run, name="control-button-watcher", daemon=True + ) + self._thread.start() + except Exception: + # Release any line/pipe fds opened before the failure so nothing leaks + # and no orphaned thread survives a partial start. + self._thread = None + self._close_stop_pipe() + self._line.close() + raise def stop(self) -> None: """Signal the watcher thread to exit and release the GPIO line and pipe.""" diff --git a/python_app/gui/controllers/app_window_control_button_mixin.py b/python_app/gui/controllers/app_window_control_button_mixin.py index 75b86ee..2500955 100644 --- a/python_app/gui/controllers/app_window_control_button_mixin.py +++ b/python_app/gui/controllers/app_window_control_button_mixin.py @@ -19,6 +19,8 @@ class AppWindowControlButtonMixin: def _init_control_button_state(self) -> None: """Initialize the watcher handle before the watcher is started.""" self._control_button_watcher: ControlButtonWatcher | None = None + # Re-entrancy guard: ignore presses that arrive while a capture is in progress. + self._control_button_busy = False def _start_control_button_watcher(self) -> None: """Open the configured GPIO button line and begin watching for presses. @@ -70,8 +72,17 @@ class AppWindowControlButtonMixin: Delivered as a queued signal from the watcher thread, so this executes on the GUI thread exactly like a click on "Capture Tmp Reference". """ - self._log("GPIO control button pressed: capturing tmp reference.") - self._capture_tmp_reference() + # Ignore a re-entrant press: the capture flow spins the event loop (stop/ + # start run, dialogs), so a second queued press must not start a nested capture. + if self._control_button_busy: + self._log("GPIO control button press ignored: capture already in progress.") + return + self._control_button_busy = True + try: + self._log("GPIO control button pressed: capturing tmp reference.") + self._capture_tmp_reference() + finally: + self._control_button_busy = False def _on_control_button_failed(self, message: str) -> None: """Log an unrecoverable watcher error reported from the background thread.""" @@ -82,9 +93,24 @@ class AppWindowControlButtonMixin: watcher = getattr(self, "_control_button_watcher", None) if watcher is None: return + # Disconnect first so a press queued before teardown cannot run a slot afterwards. + self._disconnect_control_button_signals(watcher) try: watcher.stop() except Exception as exc: # noqa: BLE001 self._log_warning("Error stopping GPIO control button watcher", details=str(exc)) finally: + # Disconnect again in case stop() re-emitted, then drop and schedule deletion. + self._disconnect_control_button_signals(watcher) self._control_button_watcher = None + watcher.deleteLater() + + @staticmethod + def _disconnect_control_button_signals(watcher: ControlButtonWatcher) -> None: + """Detach the watcher's signals from their slots, tolerating already-disconnected.""" + for signal in (watcher.pressed, watcher.failed): + try: + signal.disconnect() + except (TypeError, RuntimeError): + # No connections left (or the C++ object is already gone): nothing to do. + pass diff --git a/python_app/gui/controllers/app_window_pipeline_mixin.py b/python_app/gui/controllers/app_window_pipeline_mixin.py index e807022..8f535e1 100644 --- a/python_app/gui/controllers/app_window_pipeline_mixin.py +++ b/python_app/gui/controllers/app_window_pipeline_mixin.py @@ -4,6 +4,7 @@ from __future__ import annotations import time +from PyQt6.QtCore import QCoreApplication, QEventLoop from python_app.gui.runtime.constraints import validate_processing_mode_constraints from python_app.gui.runtime.history import ( @@ -27,6 +28,15 @@ from python_app.orchestration.shm_reader import ShmRingReader class AppWindowPipelineMixin: """Controls start/stop, readers, and periodic polling of pipeline rings.""" + # A repeated identical reader error is deduped from the log, but must still be + # re-logged every this-many polls so a persistent failure stays visible. + _READER_ERROR_RELOG_EVERY = 200 + # After this many consecutive identical reader errors, attempt one reader + # reconnect; if it keeps failing past the next threshold, stop the pipeline so + # a wedged reader cannot stay silently broken forever. + _READER_ERROR_RECONNECT_AT = 40 + _READER_ERROR_STOP_AT = 400 + def _processor_requires_restart(self, run_signature: tuple[object, ...]) -> bool: """Return whether alive `data_processor` was started with different stable run settings.""" return self._supervisor.is_processor_running() and self._processor_run_signature != run_signature @@ -137,6 +147,7 @@ class AppWindowPipelineMixin: # from a clean boundary and does not retain stale results-only tail. self._drop_pending_ring_payloads(include_results=True) self._last_reader_error_signature = None + self._reader_error_repeat_count = 0 if single_capture: self._single_capture_start_ns = time.monotonic_ns() @@ -302,6 +313,7 @@ class AppWindowPipelineMixin: result_latest = self._read_all_results() if self._result_reader is not None else None self._update_history_indicator() self._last_reader_error_signature = None + self._reader_error_repeat_count = 0 if self._single_capture_active: if self._finish_single_capture_if_ready(): @@ -317,12 +329,67 @@ class AppWindowPipelineMixin: else: self._draw_preferred_collection(result_latest=None) except Exception as exc: # noqa: BLE001 - signature = (type(exc).__name__, str(exc)) - if self._last_reader_error_signature == signature: - return + self._handle_reader_poll_error(exc) + + def _handle_reader_poll_error(self, exc: Exception) -> None: + """Surface a reader-poll failure without spamming the log. + + Distinct errors log once; an identical recurring error is deduped from the + log but still drives the status label to error, is periodically re-logged, + and after escalating thresholds triggers a reader reconnect and finally a + pipeline stop so a wedged reader cannot fail silently forever. + """ + signature = (type(exc).__name__, str(exc)) + # Always reflect a reader failure in the status label, even when deduped. + self._status_label.setText("Status: error") + + if self._last_reader_error_signature == signature: + self._reader_error_repeat_count = getattr(self, "_reader_error_repeat_count", 0) + 1 + else: self._last_reader_error_signature = signature + self._reader_error_repeat_count = 0 self._log_exception("Reader poll failed", exc, level="ERROR") + repeats = self._reader_error_repeat_count + # Periodically re-log a persistent identical failure so it stays visible. + if repeats and repeats % self._READER_ERROR_RELOG_EVERY == 0: + self._log_exception( + f"Reader poll still failing (repeat #{repeats})", exc, level="ERROR" + ) + + if repeats >= self._READER_ERROR_STOP_AT: + # The reader stayed wedged through a reconnect attempt; stop the + # pipeline so the failure is unmistakable instead of an endless retry. + self._log_error( + f"Stopping pipeline after {repeats} consecutive reader poll failures" + ) + self._reader_error_repeat_count = 0 + self._last_reader_error_signature = None + self._stop_all_processes() + elif repeats == self._READER_ERROR_RECONNECT_AT: + self._reconnect_readers_after_error() + + def _reconnect_readers_after_error(self) -> None: + """Re-open active ring readers in place to recover from a wedged reader.""" + self._log_warning("Attempting ring reader reconnect after repeated poll failures") + try: + for attr in ("_raw_reader", "_pre_reader", "_result_reader"): + reader = getattr(self, attr) + if reader is None: + continue + ring_name = reader._ring_name # noqa: SLF001 - reuse the reader's own ring name + reader.close() + setattr(self, attr, ShmRingReader(ring_name)) + # A successful reconnect clears the error state so the next failure + # logs fresh rather than being swallowed by the stale signature. + self._last_reader_error_signature = None + self._reader_error_repeat_count = 0 + self._status_label.setText("Status: running") + self._log("Ring readers reconnected after repeated poll failures") + except Exception as exc: # noqa: BLE001 + # Leave the error signature intact so escalation to a stop still fires. + self._log_exception("Ring reader reconnect failed", exc, level="ERROR") + def _finish_single_capture_if_ready(self) -> bool: """Finalize single capture when the exact target result becomes available.""" if not self._single_capture_active: @@ -406,6 +473,28 @@ class AppWindowPipelineMixin: latest = collection return latest + def _pump_events_during_drain(self, pause_s: float) -> None: + """Yield to the Qt event loop for `pause_s` instead of blocking on time.sleep. + + The bounded drain loops run on the GUI thread; a raw time.sleep here freezes + the event loop, stalling the keepalive/headless-watchdog timers and starving + queued signals. Pumping events keeps the daemon responsive while we wait. + """ + app = QCoreApplication.instance() + if app is None: + # No event loop (e.g. unit context); fall back to a plain short sleep. + time.sleep(pause_s) + return + deadline = time.monotonic() + pause_s + while True: + remaining_ms = int((deadline - time.monotonic()) * 1000) + if remaining_ms <= 0: + break + app.processEvents(QEventLoop.ProcessEventsFlag.AllEvents, remaining_ms) + # processEvents returns immediately when the queue empties; sleep the + # residual in tiny slices so we neither busy-spin nor block too long. + time.sleep(min(0.002, max(0.0, deadline - time.monotonic()))) + def _drain_rings_once_for_history(self) -> None: """Perform one non-blocking read pass to extend histories.""" if self._raw_reader is not None: @@ -436,7 +525,8 @@ class AppWindowPipelineMixin: else: stable_rounds = 0 previous = current - time.sleep(poll_s) + # Event-loop-friendly wait so timers/signals keep firing during drain. + self._pump_events_during_drain(poll_s) def _drain_results_until_quiet(self, *, timeout_s: float, poll_s: float) -> ResultCollection | None: """Drain results until at least one result arrives and the ring becomes quiet.""" @@ -455,7 +545,8 @@ class AppWindowPipelineMixin: else: latest_seen = latest stable_rounds = 0 - time.sleep(poll_s) + # Event-loop-friendly wait so timers/signals keep firing during drain. + self._pump_events_during_drain(poll_s) return latest_seen def _update_history_indicator(self) -> None: diff --git a/python_app/gui/controllers/app_window_snapshot_mixin.py b/python_app/gui/controllers/app_window_snapshot_mixin.py index 697016c..4e39ff5 100644 --- a/python_app/gui/controllers/app_window_snapshot_mixin.py +++ b/python_app/gui/controllers/app_window_snapshot_mixin.py @@ -275,18 +275,20 @@ class AppWindowSnapshotMixin: got_results = result_count > start_result_count got_raw_or_pre = raw_count > start_raw_count or pre_count > start_pre_count + # Event-loop-friendly waits so the keepalive/watchdog timers and + # queued signals keep firing instead of freezing the headless daemon. if got_results and (missing_raw or missing_pre) and time.monotonic() < deadline: - time.sleep(0.01) + self._pump_events_during_drain(0.01) continue if got_raw_or_pre and missing_results and time.monotonic() < deadline: - time.sleep(0.01) + self._pump_events_during_drain(0.01) continue if ( self._result_reader is not None and max(raw_count, pre_count) > result_count and time.monotonic() < deadline ): - time.sleep(0.01) + self._pump_events_during_drain(0.01) continue break except Exception as exc: # noqa: BLE001 diff --git a/python_app/gui/main.py b/python_app/gui/main.py index 93d30ac..96dec5b 100644 --- a/python_app/gui/main.py +++ b/python_app/gui/main.py @@ -40,10 +40,17 @@ def _install_unix_signal_handlers(app: QApplication, window: AppWindow) -> None: loop just long enough to deliver pending signals. """ - def _request_shutdown(*_args: object) -> None: + def _shutdown() -> None: window.close() app.quit() + def _request_shutdown(signum: int, _frame: object) -> None: + # Async-signal-safe: do the minimum from C signal context. Reset the handler + # to default so a second signal force-terminates instead of re-entering Qt + # teardown, then schedule the real shutdown on the next event-loop iteration. + signal.signal(signum, signal.SIG_DFL) + QTimer.singleShot(0, _shutdown) + for sig in (signal.SIGINT, signal.SIGTERM): signal.signal(sig, _request_shutdown) diff --git a/python_app/hardware_full/kamil_adc_service.py b/python_app/hardware_full/kamil_adc_service.py index 5c9d01a..8aa3c32 100644 --- a/python_app/hardware_full/kamil_adc_service.py +++ b/python_app/hardware_full/kamil_adc_service.py @@ -320,14 +320,18 @@ class KamilAdcService: executable_path = str(Path(adc.executable_path).expanduser()) return [executable_path, *adc.args, f"tty:{adc.tty_path}"] - def open(self) -> None: - """Launch the collector and start the TTY reader thread.""" + def open(self, *, stop_event: threading.Event | None = None) -> None: + """Launch the collector and start the TTY reader thread. + + An optional `stop_event` lets a caller abort the TTY-wait loop promptly + (e.g. on shutdown) instead of blocking for the full startup timeout. + """ if self._reader is not None: return previous_tty_identity = _prepare_tty_path_for_collector(self.config.radar.kamil_adc.tty_path) try: self._start_process() - self._wait_for_tty(previous_tty_identity) + self._wait_for_tty(previous_tty_identity, stop_event=stop_event) reader = KamilAdcTtyReader(self.config.radar.kamil_adc.tty_path) reader.open() self._reader = reader @@ -416,15 +420,28 @@ class KamilAdcService: os.killpg(process.pid, signal.SIGKILL) process.wait(timeout=1.0) - def _wait_for_tty(self, previous_identity: tuple[object, ...] | None) -> None: + def _wait_for_tty( + self, + previous_identity: tuple[object, ...] | None, + *, + stop_event: threading.Event | None = None, + ) -> None: adc = self.config.radar.kamil_adc deadline = time.monotonic() + adc.startup_timeout_s while time.monotonic() < deadline: + # Abort promptly if a stop was requested mid-wait. + if stop_event is not None and stop_event.is_set(): + raise RuntimeError("Kamil ADC startup aborted before TTY became available") KamilAdcTtyReader._raise_if_process_exited(self._process) identity = _tty_identity(adc.tty_path) if identity is not None and identity != previous_identity: return - time.sleep(0.05) + # Use the stop event's wait() so a set() breaks the poll immediately. + if stop_event is not None: + if stop_event.wait(0.05): + raise RuntimeError("Kamil ADC startup aborted before TTY became available") + else: + time.sleep(0.05) raise TimeoutError( f"Timed out waiting for Kamil ADC TTY `{adc.tty_path}` to be created by the collector" ) diff --git a/python_app/hardware_full/librevna_driver/transport/usb.py b/python_app/hardware_full/librevna_driver/transport/usb.py index 4119de4..64bddf5 100644 --- a/python_app/hardware_full/librevna_driver/transport/usb.py +++ b/python_app/hardware_full/librevna_driver/transport/usb.py @@ -157,8 +157,19 @@ class USBTransport: logger.debug("USB disconnect requested") self._stop_event.set() - if self._rx_thread is not None and self._rx_thread.is_alive(): - self._rx_thread.join(timeout=1.0) + rx_thread = self._rx_thread + if rx_thread is not None and rx_thread.is_alive(): + rx_thread.join(timeout=1.0) + if rx_thread.is_alive(): + # The RX thread is wedged inside libusb; closing the handle or + # context now would risk a use-after-free in the still-running + # bulkRead. Deliberately leak both rather than crash the process. + logger.error( + "USB RX thread did not stop within timeout; leaking USB handle/context " + "to avoid use-after-free (serial=%s)", + self.connected_serial, + ) + return self._rx_thread = None if self._handle is not None: diff --git a/python_app/hardware_full/multi_device_service.py b/python_app/hardware_full/multi_device_service.py index 30f5b8b..6aacc63 100644 --- a/python_app/hardware_full/multi_device_service.py +++ b/python_app/hardware_full/multi_device_service.py @@ -5,6 +5,7 @@ from __future__ import annotations from dataclasses import dataclass, field import logging import math +import threading import time from typing import TYPE_CHECKING @@ -26,6 +27,12 @@ logger = logging.getLogger(__name__) # of all entries (1.75 s today) plus the cost of close()/open() themselves. _REOPEN_BACKOFF_SECONDS: tuple[float, ...] = (0.25, 0.5, 1.0) +# Hard ceiling on the wall-clock time a single acquire_collection() may spend in +# the recovery path (backoff sleeps + close()/open() cost across every retry). +# Bounding this keeps a shutdown that arrives mid-recovery comfortably under the +# systemd TimeoutStopSec so the unit is never SIGKILLed for hanging on exit. +_MAX_RECOVERY_WALL_SECONDS: float = 10.0 + _INPUT_S_PARAMETERS_BY_OUTPUT: dict[int, tuple[str, ...]] = { 0: ("s31", "s41", "s51", "s61"), 1: ("s32", "s42", "s52", "s62"), @@ -101,13 +108,22 @@ class MultiDeviceLibreVnaService: except Exception as exc: # noqa: BLE001 — recovery path, never propagate logger.warning("Multi-device close() ignored transport error: %s", exc) - def recover(self) -> None: + def recover( + self, + *, + stop_event: threading.Event | None = None, + deadline_monotonic: float | None = None, + ) -> None: """Reopen native device transports after a failed acquisition. Tries several short backoffs so a transient USB stall does not kill the producer on the very first retry. Raises the last error only after every attempt failed — the outer acquisition loop is expected to count these as recovery_attempts. + + When `stop_event` is supplied the backoff waits on it instead of + sleeping, so a shutdown request aborts the loop immediately; an optional + `deadline_monotonic` caps the total wall-time spent here. """ if self._using_mock_backend: return @@ -115,7 +131,21 @@ class MultiDeviceLibreVnaService: last_error: Exception | None = None for attempt_index, delay_s in enumerate(_REOPEN_BACKOFF_SECONDS, start=1): - time.sleep(delay_s) + # Bail out the instant a stop is requested or the recovery budget is + # spent, rather than committing to another (re)open attempt. + if stop_event is not None and stop_event.is_set(): + logger.info("Multi-device recover() aborted: stop requested") + return + if deadline_monotonic is not None and time.monotonic() >= deadline_monotonic: + logger.warning("Multi-device recover() aborted: recovery time budget exhausted") + break + # Interruptible backoff: wait() returns early the moment stop is set. + if stop_event is not None: + if stop_event.wait(delay_s): + logger.info("Multi-device recover() aborted: stop requested") + return + else: + time.sleep(delay_s) try: self.open() if self._controller is not None: @@ -151,15 +181,27 @@ class MultiDeviceLibreVnaService: power_dbm=float(sweep.power_dbm), ) - def acquire_collection(self, collection_id: int = 1) -> SweepCollection: - """Acquire one complete virtual 2x4 matrix in canonical combo order.""" + def acquire_collection( + self, + collection_id: int = 1, + *, + stop_event: threading.Event | None = None, + ) -> SweepCollection: + """Acquire one complete virtual 2x4 matrix in canonical combo order. + + When `stop_event` is supplied it is threaded into the recovery path so a + shutdown request aborts the backoff/retry loops promptly (default `None` + preserves the original blocking behavior). + """ if self._sweep_configuration is None: raise RuntimeError("Multi-device service is not configured") capture_start_ns = time.monotonic_ns() if self._using_mock_backend: collection = self._acquire_mock_collection(collection_id, capture_start_ns) else: - collection = self._acquire_native_collection_with_recovery(collection_id, capture_start_ns) + collection = self._acquire_native_collection_with_recovery( + collection_id, capture_start_ns, stop_event=stop_event + ) collection.capture_end_ns = time.monotonic_ns() return collection @@ -167,8 +209,13 @@ class MultiDeviceLibreVnaService: self, collection_id: int, capture_start_ns: int, + *, + stop_event: threading.Event | None = None, ) -> SweepCollection: last_error: Exception | None = None + # Cap total recovery wall-time across all retries so a shutdown that + # lands mid-recovery stays well under the systemd TimeoutStopSec. + recovery_deadline = time.monotonic() + _MAX_RECOVERY_WALL_SECONDS for attempt_index in range(self.recovery_attempts + 1): try: return self._acquire_native_collection(collection_id, capture_start_ns) @@ -176,6 +223,13 @@ class MultiDeviceLibreVnaService: last_error = exc if attempt_index >= self.recovery_attempts: break + # Stop the moment shutdown is requested or the recovery budget is + # spent — do not start another reconnect we cannot finish in time. + if stop_event is not None and stop_event.is_set(): + break + if time.monotonic() >= recovery_deadline: + logger.warning("multi-device recovery time budget exhausted; giving up") + break logger.warning( "multi-device acquisition failed, reconnecting devices (%d/%d): %s", attempt_index + 1, @@ -188,7 +242,7 @@ class MultiDeviceLibreVnaService: # attempt and try again on the next loop iteration, so a # transient USB hiccup cannot kill the whole producer. try: - self.recover() + self.recover(stop_event=stop_event, deadline_monotonic=recovery_deadline) except Exception as recover_exc: # noqa: BLE001 last_error = recover_exc logger.warning( @@ -198,6 +252,10 @@ class MultiDeviceLibreVnaService: recover_exc, exc_info=True, ) + # If recovery was interrupted by a stop request, do not loop back + # for another acquisition attempt; let shutdown proceed. + if stop_event is not None and stop_event.is_set(): + break assert last_error is not None raise last_error diff --git a/python_app/hardware_full/switch_drivers/gpio_uapi.py b/python_app/hardware_full/switch_drivers/gpio_uapi.py index 3d4264b..937a45e 100644 --- a/python_app/hardware_full/switch_drivers/gpio_uapi.py +++ b/python_app/hardware_full/switch_drivers/gpio_uapi.py @@ -180,9 +180,12 @@ class GpioOutputLines: self._line_fd = int(request.fd) def close(self) -> None: - """Close line request and chip file descriptors.""" - self._close_line_fd() - self._close_chip_fd() + """Close line request and chip file descriptors (best-effort, idempotent).""" + try: + self._close_line_fd() + finally: + # Ensure the chip fd is always closed even if closing the line fd raised. + self._close_chip_fd() def set_values(self, values: Sequence[int]) -> None: """Apply output values for all requested lines.""" @@ -212,16 +215,22 @@ class GpioOutputLines: raise RuntimeError(f"Failed to set GPIO output values: {exc}") from exc def _close_line_fd(self) -> None: - """Close line file descriptor if currently open.""" + """Close line file descriptor if currently open (best-effort, idempotent).""" if self._line_fd >= 0: - os.close(self._line_fd) - self._line_fd = -1 + try: + os.close(self._line_fd) + finally: + # Always clear the fd so close() stays idempotent even if os.close raised. + self._line_fd = -1 def _close_chip_fd(self) -> None: - """Close chip file descriptor if currently open.""" + """Close chip file descriptor if currently open (best-effort, idempotent).""" if self._chip_fd >= 0: - os.close(self._chip_fd) - self._chip_fd = -1 + try: + os.close(self._chip_fd) + finally: + # Always clear the fd so close() stays idempotent even if os.close raised. + self._chip_fd = -1 class GpioLineEventWatcher: @@ -316,18 +325,27 @@ class GpioLineEventWatcher: return int(event.id) def close(self) -> None: - """Close line request and chip file descriptors.""" - self._close_line_fd() - self._close_chip_fd() + """Close line request and chip file descriptors (best-effort, idempotent).""" + try: + self._close_line_fd() + finally: + # Ensure the chip fd is always closed even if closing the line fd raised. + self._close_chip_fd() def _close_line_fd(self) -> None: - """Close line file descriptor if currently open.""" + """Close line file descriptor if currently open (best-effort, idempotent).""" if self._line_fd >= 0: - os.close(self._line_fd) - self._line_fd = -1 + try: + os.close(self._line_fd) + finally: + # Always clear the fd so close() stays idempotent even if os.close raised. + self._line_fd = -1 def _close_chip_fd(self) -> None: - """Close chip file descriptor if currently open.""" + """Close chip file descriptor if currently open (best-effort, idempotent).""" if self._chip_fd >= 0: - os.close(self._chip_fd) - self._chip_fd = -1 + try: + os.close(self._chip_fd) + finally: + # Always clear the fd so close() stays idempotent even if os.close raised. + self._chip_fd = -1 diff --git a/python_app/models/run_config_codec.py b/python_app/models/run_config_codec.py index 1be7b60..2984237 100644 --- a/python_app/models/run_config_codec.py +++ b/python_app/models/run_config_codec.py @@ -2,6 +2,7 @@ from __future__ import annotations +import math from typing import Any from python_app.models.run_config_schema import ( @@ -38,9 +39,67 @@ def _read_str(payload: dict[str, Any], key: str, default: str) -> str: value = payload.get(key, default) if value is None: return default + # A JSON array/object reaching a scalar field is a config error, not a + # str() fallback; surface it as ValueError to keep the error contract uniform. + if isinstance(value, (dict, list)): + raise ValueError(f"{key} must be a JSON string") return str(value) +def _read_int(payload: dict[str, Any], key: str, default: int) -> int: + """Return payload integer, treating an explicit JSON `null` as 'use default'. + + Without this, `int(payload.get(key, default))` raises TypeError on an + explicit `null`. JSON arrays/objects (and other non-numeric scalars) are + rejected as ValueError so malformed types share the config-error contract. + """ + value = payload.get(key, default) + if value is None: + return default + if isinstance(value, bool) or not isinstance(value, (int, float, str)): + raise ValueError(f"{key} must be a JSON integer") + try: + return int(value) + except (TypeError, ValueError) as exc: + raise ValueError(f"{key} must be a JSON integer") from exc + + +def _read_float(payload: dict[str, Any], key: str, default: float) -> float: + """Return payload float, treating an explicit JSON `null` as 'use default'. + + Rejects JSON arrays/objects (and other non-numeric scalars) as ValueError, + and rejects non-finite values (NaN/Infinity) at decode time so the C++ + pipeline never receives a value it cannot honor. + """ + value = payload.get(key, default) + if value is None: + return default + if isinstance(value, bool) or not isinstance(value, (int, float, str)): + raise ValueError(f"{key} must be a JSON number") + try: + result = float(value) + except (TypeError, ValueError) as exc: + raise ValueError(f"{key} must be a JSON number") from exc + if not math.isfinite(result): + raise ValueError(f"{key} must be a finite number") + return result + + +def _read_bool(payload: dict[str, Any], key: str, default: bool) -> bool: + """Return payload boolean, treating an explicit JSON `null` as 'use default'. + + Plain `bool(payload.get(key, default))` would silently flip the default to + `False` on an explicit `null`; here `null` keeps the default instead. + Non-boolean JSON types are rejected as ValueError. + """ + value = payload.get(key, default) + if value is None: + return default + if not isinstance(value, bool): + raise ValueError(f"{key} must be a JSON boolean") + return value + + def _load_preprocess_asset(payload: dict[str, Any], target: PreprocessAssetModel) -> None: """Load preprocess asset fields into target model.""" target.set_name = _read_str(payload, "set_name", target.set_name) @@ -105,21 +164,21 @@ def run_config_from_dict(payload: dict[str, Any]) -> RunConfigModel: kamil_adc_payload = _as_dict(radar_payload.get("kamil_adc"), "radar.kamil_adc") laser_control_payload = _as_dict(radar_payload.get("laser_control"), "radar.laser_control") - model.radar.model = str(radar_payload.get("model", model.radar.model)) - model.radar.serial = str(radar_payload.get("serial", model.radar.serial)) - model.radar.remote_host = str(radar_payload.get("remote_host", model.radar.remote_host)) - model.radar.remote_port = int(radar_payload.get("remote_port", model.radar.remote_port)) - model.radar.driver_mode = str(radar_payload.get("driver_mode", model.radar.driver_mode)) - model.radar.mock_signal_hz = float(radar_payload.get("mock_signal_hz", model.radar.mock_signal_hz)) - model.radar.visa_library = str(radar_payload.get("visa_library", model.radar.visa_library)) + model.radar.model = _read_str(radar_payload, "model", model.radar.model) + model.radar.serial = _read_str(radar_payload, "serial", model.radar.serial) + model.radar.remote_host = _read_str(radar_payload, "remote_host", model.radar.remote_host) + model.radar.remote_port = _read_int(radar_payload, "remote_port", model.radar.remote_port) + model.radar.driver_mode = _read_str(radar_payload, "driver_mode", model.radar.driver_mode) + model.radar.mock_signal_hz = _read_float(radar_payload, "mock_signal_hz", model.radar.mock_signal_hz) + model.radar.visa_library = _read_str(radar_payload, "visa_library", model.radar.visa_library) - model.radar.sweep.start_hz = float(sweep_payload.get("start_hz", model.radar.sweep.start_hz)) - model.radar.sweep.stop_hz = float(sweep_payload.get("stop_hz", model.radar.sweep.stop_hz)) - model.radar.sweep.points = int(sweep_payload.get("points", model.radar.sweep.points)) - model.radar.sweep.if_bandwidth_hz = float( - sweep_payload.get("if_bandwidth_hz", model.radar.sweep.if_bandwidth_hz) + model.radar.sweep.start_hz = _read_float(sweep_payload, "start_hz", model.radar.sweep.start_hz) + model.radar.sweep.stop_hz = _read_float(sweep_payload, "stop_hz", model.radar.sweep.stop_hz) + model.radar.sweep.points = _read_int(sweep_payload, "points", model.radar.sweep.points) + model.radar.sweep.if_bandwidth_hz = _read_float( + sweep_payload, "if_bandwidth_hz", model.radar.sweep.if_bandwidth_hz ) - model.radar.sweep.power_dbm = float(sweep_payload.get("stimulus_power_dbm", model.radar.sweep.power_dbm)) + model.radar.sweep.power_dbm = _read_float(sweep_payload, "stimulus_power_dbm", model.radar.sweep.power_dbm) slave_serials_payload = multi_device_payload.get( "slave_serials", multi_device_payload.get("slave_serial_numbers", model.radar.multi_device.slave_serials), @@ -132,121 +191,114 @@ def run_config_from_dict(payload: dict[str, Any]) -> RunConfigModel: for value in slave_serials_payload.split(",") if value.strip() ] - model.radar.multi_device.force_external_reference = bool( - multi_device_payload.get( - "force_external_reference", - model.radar.multi_device.force_external_reference, - ) + model.radar.multi_device.force_external_reference = _read_bool( + multi_device_payload, + "force_external_reference", + model.radar.multi_device.force_external_reference, ) - model.radar.multi_device.recovery_attempts = int( - multi_device_payload.get( - "recovery_attempts", - model.radar.multi_device.recovery_attempts, - ) + model.radar.multi_device.recovery_attempts = _read_int( + multi_device_payload, + "recovery_attempts", + model.radar.multi_device.recovery_attempts, ) - model.radar.kamil_adc.project_dir = str( - kamil_adc_payload.get("project_dir", model.radar.kamil_adc.project_dir) + model.radar.kamil_adc.project_dir = _read_str( + kamil_adc_payload, "project_dir", model.radar.kamil_adc.project_dir ) - model.radar.kamil_adc.executable_path = str( - kamil_adc_payload.get("executable_path", model.radar.kamil_adc.executable_path) + model.radar.kamil_adc.executable_path = _read_str( + kamil_adc_payload, "executable_path", model.radar.kamil_adc.executable_path ) - model.radar.kamil_adc.tty_path = str( - kamil_adc_payload.get("tty_path", model.radar.kamil_adc.tty_path) + model.radar.kamil_adc.tty_path = _read_str( + kamil_adc_payload, "tty_path", model.radar.kamil_adc.tty_path ) model.radar.kamil_adc.args = _load_string_list(kamil_adc_payload, "args", "radar.kamil_adc") model.radar.kamil_adc.env = _load_string_dict(kamil_adc_payload, "env", "radar.kamil_adc") - model.radar.kamil_adc.startup_timeout_s = float( - kamil_adc_payload.get("startup_timeout_s", model.radar.kamil_adc.startup_timeout_s) + model.radar.kamil_adc.startup_timeout_s = _read_float( + kamil_adc_payload, "startup_timeout_s", model.radar.kamil_adc.startup_timeout_s ) - model.radar.kamil_adc.sweep_timeout_s = float( - kamil_adc_payload.get("sweep_timeout_s", model.radar.kamil_adc.sweep_timeout_s) + model.radar.kamil_adc.sweep_timeout_s = _read_float( + kamil_adc_payload, "sweep_timeout_s", model.radar.kamil_adc.sweep_timeout_s ) - model.radar.kamil_adc.stop_timeout_s = float( - kamil_adc_payload.get("stop_timeout_s", model.radar.kamil_adc.stop_timeout_s) + model.radar.kamil_adc.stop_timeout_s = _read_float( + kamil_adc_payload, "stop_timeout_s", model.radar.kamil_adc.stop_timeout_s ) - model.radar.laser_control.enabled = bool( - laser_control_payload.get("enabled", model.radar.laser_control.enabled) + model.radar.laser_control.enabled = _read_bool( + laser_control_payload, "enabled", model.radar.laser_control.enabled ) - model.radar.laser_control.port = str( - laser_control_payload.get("port", model.radar.laser_control.port) + model.radar.laser_control.port = _read_str( + laser_control_payload, "port", model.radar.laser_control.port ) - model.radar.laser_control.mode = str( - laser_control_payload.get("mode", model.radar.laser_control.mode) + model.radar.laser_control.mode = _read_str( + laser_control_payload, "mode", model.radar.laser_control.mode ) - model.radar.laser_control.pi_coeff1_p = int( - laser_control_payload.get("pi_coeff1_p", model.radar.laser_control.pi_coeff1_p) + model.radar.laser_control.pi_coeff1_p = _read_int( + laser_control_payload, "pi_coeff1_p", model.radar.laser_control.pi_coeff1_p ) - model.radar.laser_control.pi_coeff1_i = int( - laser_control_payload.get("pi_coeff1_i", model.radar.laser_control.pi_coeff1_i) + model.radar.laser_control.pi_coeff1_i = _read_int( + laser_control_payload, "pi_coeff1_i", model.radar.laser_control.pi_coeff1_i ) - model.radar.laser_control.pi_coeff2_p = int( - laser_control_payload.get("pi_coeff2_p", model.radar.laser_control.pi_coeff2_p) + model.radar.laser_control.pi_coeff2_p = _read_int( + laser_control_payload, "pi_coeff2_p", model.radar.laser_control.pi_coeff2_p ) - model.radar.laser_control.pi_coeff2_i = int( - laser_control_payload.get("pi_coeff2_i", model.radar.laser_control.pi_coeff2_i) + model.radar.laser_control.pi_coeff2_i = _read_int( + laser_control_payload, "pi_coeff2_i", model.radar.laser_control.pi_coeff2_i ) laser_manual_payload = _as_dict(laser_control_payload.get("manual"), "radar.laser_control.manual") - model.radar.laser_control.manual.temp1 = float( - laser_manual_payload.get("temp1", model.radar.laser_control.manual.temp1) + model.radar.laser_control.manual.temp1 = _read_float( + laser_manual_payload, "temp1", model.radar.laser_control.manual.temp1 ) - model.radar.laser_control.manual.temp2 = float( - laser_manual_payload.get("temp2", model.radar.laser_control.manual.temp2) + model.radar.laser_control.manual.temp2 = _read_float( + laser_manual_payload, "temp2", model.radar.laser_control.manual.temp2 ) - model.radar.laser_control.manual.current1 = float( - laser_manual_payload.get("current1", model.radar.laser_control.manual.current1) + model.radar.laser_control.manual.current1 = _read_float( + laser_manual_payload, "current1", model.radar.laser_control.manual.current1 ) - model.radar.laser_control.manual.current2 = float( - laser_manual_payload.get("current2", model.radar.laser_control.manual.current2) + model.radar.laser_control.manual.current2 = _read_float( + laser_manual_payload, "current2", model.radar.laser_control.manual.current2 ) laser_variation_payload = _as_dict( laser_control_payload.get("variation"), "radar.laser_control.variation", ) - model.radar.laser_control.variation.variation_type = str( - laser_variation_payload.get( - "variation_type", - model.radar.laser_control.variation.variation_type, - ) + model.radar.laser_control.variation.variation_type = _read_str( + laser_variation_payload, + "variation_type", + model.radar.laser_control.variation.variation_type, ) - model.radar.laser_control.variation.static_temp1 = float( - laser_variation_payload.get( - "static_temp1", - model.radar.laser_control.variation.static_temp1, - ) + model.radar.laser_control.variation.static_temp1 = _read_float( + laser_variation_payload, + "static_temp1", + model.radar.laser_control.variation.static_temp1, ) - model.radar.laser_control.variation.static_temp2 = float( - laser_variation_payload.get( - "static_temp2", - model.radar.laser_control.variation.static_temp2, - ) + model.radar.laser_control.variation.static_temp2 = _read_float( + laser_variation_payload, + "static_temp2", + model.radar.laser_control.variation.static_temp2, ) - model.radar.laser_control.variation.static_current1 = float( - laser_variation_payload.get( - "static_current1", - model.radar.laser_control.variation.static_current1, - ) + model.radar.laser_control.variation.static_current1 = _read_float( + laser_variation_payload, + "static_current1", + model.radar.laser_control.variation.static_current1, ) - model.radar.laser_control.variation.static_current2 = float( - laser_variation_payload.get( - "static_current2", - model.radar.laser_control.variation.static_current2, - ) + model.radar.laser_control.variation.static_current2 = _read_float( + laser_variation_payload, + "static_current2", + model.radar.laser_control.variation.static_current2, ) - model.radar.laser_control.variation.min_value = float( - laser_variation_payload.get("min_value", model.radar.laser_control.variation.min_value) + model.radar.laser_control.variation.min_value = _read_float( + laser_variation_payload, "min_value", model.radar.laser_control.variation.min_value ) - model.radar.laser_control.variation.max_value = float( - laser_variation_payload.get("max_value", model.radar.laser_control.variation.max_value) + model.radar.laser_control.variation.max_value = _read_float( + laser_variation_payload, "max_value", model.radar.laser_control.variation.max_value ) - model.radar.laser_control.variation.step = float( - laser_variation_payload.get("step", model.radar.laser_control.variation.step) + model.radar.laser_control.variation.step = _read_float( + laser_variation_payload, "step", model.radar.laser_control.variation.step ) - model.radar.laser_control.variation.time_step = int( - laser_variation_payload.get("time_step", model.radar.laser_control.variation.time_step) + model.radar.laser_control.variation.time_step = _read_int( + laser_variation_payload, "time_step", model.radar.laser_control.variation.time_step ) - model.radar.laser_control.variation.delay_time = int( - laser_variation_payload.get("delay_time", model.radar.laser_control.variation.delay_time) + model.radar.laser_control.variation.delay_time = _read_int( + laser_variation_payload, "delay_time", model.radar.laser_control.variation.delay_time ) load_switch_payload(port1_payload, model.output_switch) diff --git a/python_app/models/run_config_validation.py b/python_app/models/run_config_validation.py index 4a23bd6..d0ee944 100644 --- a/python_app/models/run_config_validation.py +++ b/python_app/models/run_config_validation.py @@ -8,25 +8,68 @@ from python_app.models.run_config_schema import ( ComboModel, ControlButtonModel, GprModel, + RadarSweepModel, RingEndpointModel, SwitchModel, ) +# Wire-format bounds shared with the C++ pipeline. The ring header stores the +# slot size as a uint32, and capacity * slot_size must address into a single +# shared-memory mapping, so reject values the C++ side cannot represent. +_UINT32_MAX = (1 << 32) - 1 +_RING_SEGMENT_MAX_BYTES = 1 << 40 # 1 TiB upper bound on a single ring mapping. +# Defensive ceiling so a malformed combos string cannot expand into a list that +# stalls the GUI or the downstream acquisition loop. +_MAX_COMBOS = 4096 + + +def _require_int(payload: dict[str, Any], key: str, default: int) -> int: + """Read an integer field, rejecting JSON arrays/objects with a named ValueError. + + Bare ``int()`` raises ``TypeError`` on a list/dict, which escapes the + config-error contract; surface it as a ValueError naming the field instead. + """ + value = payload.get(key, default) + if isinstance(value, bool) or not isinstance(value, (int, float, str)): + raise ValueError(f"{key} must be a JSON integer") + try: + return int(value) + except (TypeError, ValueError) as exc: + raise ValueError(f"{key} must be a JSON integer") from exc + + +def _require_str(payload: dict[str, Any], key: str, default: str) -> str: + """Read a string field, rejecting JSON arrays/objects with a named ValueError.""" + value = payload.get(key, default) + if isinstance(value, (dict, list)): + raise ValueError(f"{key} must be a JSON string") + return str(value) + + +def _require_bool(payload: dict[str, Any], key: str, default: bool) -> bool: + """Read a boolean field, rejecting non-boolean JSON types with a named ValueError.""" + value = payload.get(key, default) + if not isinstance(value, bool): + raise ValueError(f"{key} must be a JSON boolean") + return value + + def load_switch_payload( payload: dict[str, Any], target: SwitchModel, ) -> None: """Populate switch model from payload preserving defaults for missing values.""" - target.name = str(payload.get("name", target.name)) - target.driver_mode = str(payload.get("driver_mode", target.driver_mode)) - target.driver = str(payload.get("driver", target.driver)) - target.radar_port = int(payload.get("radar_port", target.radar_port)) - target.positions = int(payload.get("positions", target.positions)) - target.default_position = int(payload.get("default_position", target.default_position)) - target.gpio_chip = str(payload.get("gpio_chip", target.gpio_chip)) - target.pin_a = int(payload.get("pin_a", target.pin_a)) - target.pin_b = int(payload.get("pin_b", target.pin_b)) - target.invert_logic = bool(payload.get("invert_logic", target.invert_logic)) + # #53: scalar reads reject array/object JSON types as ValueError (not TypeError). + target.name = _require_str(payload, "name", target.name) + target.driver_mode = _require_str(payload, "driver_mode", target.driver_mode) + target.driver = _require_str(payload, "driver", target.driver) + target.radar_port = _require_int(payload, "radar_port", target.radar_port) + target.positions = _require_int(payload, "positions", target.positions) + target.default_position = _require_int(payload, "default_position", target.default_position) + target.gpio_chip = _require_str(payload, "gpio_chip", target.gpio_chip) + target.pin_a = _require_int(payload, "pin_a", target.pin_a) + target.pin_b = _require_int(payload, "pin_b", target.pin_b) + target.invert_logic = _require_bool(payload, "invert_logic", target.invert_logic) def load_control_button_payload( @@ -34,20 +77,56 @@ def load_control_button_payload( target: ControlButtonModel, ) -> None: """Populate control-button model from payload preserving defaults.""" - target.enabled = bool(payload.get("enabled", target.enabled)) - target.gpio_chip = str(payload.get("gpio_chip", target.gpio_chip)) - target.pin = int(payload.get("pin", target.pin)) - target.active_low = bool(payload.get("active_low", target.active_low)) - target.bias = str(payload.get("bias", target.bias)) - target.debounce_ms = int(payload.get("debounce_ms", target.debounce_ms)) - target.action = str(payload.get("action", target.action)) + # #53: scalar reads reject array/object JSON types as ValueError (not TypeError). + target.enabled = _require_bool(payload, "enabled", target.enabled) + target.gpio_chip = _require_str(payload, "gpio_chip", target.gpio_chip) + target.pin = _require_int(payload, "pin", target.pin) + target.active_low = _require_bool(payload, "active_low", target.active_low) + target.bias = _require_str(payload, "bias", target.bias) + target.debounce_ms = _require_int(payload, "debounce_ms", target.debounce_ms) + target.action = _require_str(payload, "action", target.action) def load_ring_payload(payload: dict[str, Any], target: RingEndpointModel) -> None: """Populate ring endpoint model from payload preserving defaults.""" - target.name = str(payload.get("name", target.name)) - target.capacity = int(payload.get("capacity", target.capacity)) - target.slot_size_bytes = int(payload.get("slot_size_bytes", target.slot_size_bytes)) + # #53: scalar reads reject array/object JSON types as ValueError (not TypeError). + target.name = _require_str(payload, "name", target.name) + target.capacity = _require_int(payload, "capacity", target.capacity) + target.slot_size_bytes = _require_int(payload, "slot_size_bytes", target.slot_size_bytes) + # #36: enforce ring sizing in Python so a bad config fails here (in GUI/save and + # at config load) instead of crashing the C++ ring allocator at boot. + validate_ring_endpoint(target) + + +def validate_ring_endpoint(ring: RingEndpointModel) -> None: + """Validate ring sizing against the constraints the C++ allocator requires.""" + field = ring.name or "ring" + if ring.capacity <= 0: + raise ValueError(f"rings.{field}.capacity must be > 0") + if ring.slot_size_bytes <= 0: + raise ValueError(f"rings.{field}.slot_size_bytes must be > 0") + if ring.slot_size_bytes > _UINT32_MAX: + raise ValueError(f"rings.{field}.slot_size_bytes exceeds the uint32 wire limit") + # Overflow-safe: compare against the ceiling without ever forming the full + # product, so an attacker-sized capacity cannot wrap a fixed-width index. + if ring.capacity > _RING_SEGMENT_MAX_BYTES // ring.slot_size_bytes: + raise ValueError( + f"rings.{field} capacity * slot_size_bytes exceeds the maximum ring segment size" + ) + + +def validate_sweep_model(sweep: RadarSweepModel) -> None: + """Validate radar sweep bounds in Python so a bad sweep fails in the GUI/save + and at config load rather than aborting the C++ acquisition process at boot. + """ + # #36: points must be a positive, integral count of frequency samples. + points = sweep.points + if isinstance(points, bool) or not isinstance(points, int): + raise ValueError("radar.sweep.points must be an integer") + if points <= 0: + raise ValueError("radar.sweep.points must be > 0") + if float(sweep.stop_hz) < float(sweep.start_hz): + raise ValueError("radar.sweep.stop_hz must be >= radar.sweep.start_hz") def validate_gpr_model( @@ -55,8 +134,17 @@ def validate_gpr_model( *, input_switch_positions: int, output_switch_positions: int, + sweep: RadarSweepModel | None = None, ) -> None: - """Validate stable GPR config against current switch dimensions.""" + """Validate stable GPR config against current switch dimensions. + + When ``sweep`` is supplied (load and GUI/save paths share this chokepoint), + its bounds are validated here too so #36 sweep failures surface alongside the + GPR checks instead of as a C++ boot crash. + """ + if sweep is not None: + validate_sweep_model(sweep) + if float(gpr.relative_permittivity) <= 0.0: raise ValueError("gpr.relative_permittivity must be > 0") @@ -93,8 +181,26 @@ def parse_combos_from_text(text: str) -> list[ComboModel]: if ":" not in pair: raise ValueError(f"Invalid combo syntax: {pair!r}. Expected input:output") - input_text, output_text = pair.split(":", 1) - combos.append(ComboModel(input=int(input_text.strip()), output=int(output_text.strip()))) + # #57: cap the combo count so a pathological string cannot expand into a + # list large enough to stall the GUI or the acquisition loop. + if len(combos) >= _MAX_COMBOS: + raise ValueError(f"Too many combos: limit is {_MAX_COMBOS}") + + input_text, output_text = (side.strip() for side in pair.split(":", 1)) + # #57: reject empty sides and re-raise non-integer values naming the pair/side. + if not input_text: + raise ValueError(f"Invalid combo {pair!r}: input side is empty") + if not output_text: + raise ValueError(f"Invalid combo {pair!r}: output side is empty") + try: + input_value = int(input_text) + except ValueError as exc: + raise ValueError(f"Invalid combo {pair!r}: input {input_text!r} is not an integer") from exc + try: + output_value = int(output_text) + except ValueError as exc: + raise ValueError(f"Invalid combo {pair!r}: output {output_text!r} is not an integer") from exc + combos.append(ComboModel(input=input_value, output=output_value)) if not combos: raise ValueError("No valid combos were provided") diff --git a/python_app/orchestration/config_writer.py b/python_app/orchestration/config_writer.py index 7c7c99d..25c2602 100644 --- a/python_app/orchestration/config_writer.py +++ b/python_app/orchestration/config_writer.py @@ -3,6 +3,7 @@ from __future__ import annotations import json +import os from pathlib import Path from python_app.models.run_config_model import RunConfigModel, parse_combos_from_text @@ -41,12 +42,25 @@ class ConfigWriter: asset.bundle_path = str(bundle_path) def write(self, config: RunConfigModel, output_path: Path) -> Path: - """Write run configuration JSON file.""" + """Atomically write run configuration JSON file. + + Mirrors ProcessingLiveConfigWriter: dump to a sibling .tmp, flush+fsync to + durably commit the bytes, then os.replace() onto the destination. The replace + is atomic, so a C++ consumer can never observe a half-written config (which + would abort it with an opaque JSON parse error), even across a crash or power + loss mid-write on the SD-card-backed Pi. + """ output_path.parent.mkdir(parents=True, exist_ok=True) # allow_nan=False: a stray NaN/Infinity must fail loudly here in Python # rather than serialize to a non-standard token that aborts every C++ # consumer at startup with an opaque JSON parse error. - output_path.write_text(json.dumps(config.to_dict(), indent=2, allow_nan=False), encoding="utf-8") + serialized = json.dumps(config.to_dict(), indent=2, allow_nan=False) + tmp_path = output_path.with_suffix(output_path.suffix + ".tmp") + with open(tmp_path, "w", encoding="utf-8") as handle: + handle.write(serialized) + handle.flush() + os.fsync(handle.fileno()) + os.replace(tmp_path, output_path) return output_path diff --git a/python_app/orchestration/process_supervisor.py b/python_app/orchestration/process_supervisor.py index 883e464..264ea5f 100644 --- a/python_app/orchestration/process_supervisor.py +++ b/python_app/orchestration/process_supervisor.py @@ -4,14 +4,23 @@ from __future__ import annotations from dataclasses import dataclass import json +import os from pathlib import Path import shlex +import signal import subprocess import sys import time from typing import Iterable from typing import Sequence +# Cap each child log so a long-lived daemon cannot fill the SD card. On reaching +# the cap the current log is rolled to `{name}.{out,err}.log.prev` and a fresh +# log opened (see `_roll_log_if_oversized`). +_LOG_MAX_BYTES = 8 * 1024 * 1024 +# Per-process force-kill deadline used on stop (Fix #33: own deadline each). +_STOP_GRACE_SECONDS = 2.0 + @dataclass(slots=True) class ManagedProcess: @@ -27,14 +36,19 @@ class ManagedProcess: @dataclass(slots=True) class ProcessExitReport: - """Structured report for one exited managed process.""" + """Structured report for one exited managed process. + + Log tails are not held in memory: they are read on demand from the child log + files only while rendering an ERROR report, so the common clean-exit path on + every poll never pays for a 16KB read of two files (Fix #46). + """ name: str command: list[str] working_directory: Path return_code: int - stdout: str - stderr: str + stdout_path: Path + stderr_path: Path expected_clean_exit: bool @property @@ -56,15 +70,37 @@ class ProcessExitReport: f"Command: {shlex.join(self.command)}", f"Working directory: {self.working_directory}", ] - if self.stderr: - lines.append(f"stderr:\n{self.stderr}") - if self.stdout: - lines.append(f"stdout:\n{self.stdout}") - if not self.stderr and not self.stdout: + # Read the (bounded) log tails lazily, only now that we are rendering. + stdout = "" if self.expected_clean_exit else _read_log_tail(self.stdout_path) + stderr = "" if self.expected_clean_exit else _read_log_tail(self.stderr_path) + if stderr: + lines.append(f"stderr:\n{stderr}") + if stdout: + lines.append(f"stdout:\n{stdout}") + if not self.expected_clean_exit and not stderr and not stdout: lines.append("stdout/stderr: none") return "\n".join(lines) +def _read_log_tail(path: Path, max_bytes: int = 16384) -> str: + """Return the trailing `max_bytes` of a child log file, decoded best-effort. + + Bounded so a large/long-lived log never produces an enormous exit report. + """ + try: + with open(path, "rb") as handle: + handle.seek(0, 2) + size = handle.tell() + if size > max_bytes: + handle.seek(-max_bytes, 2) + else: + handle.seek(0) + data = handle.read() + except OSError: + return "" + return data.decode("utf-8", errors="replace").strip() + + class ProcessSupervisor: """Start, monitor, and stop pipeline subprocesses.""" @@ -73,6 +109,11 @@ class ProcessSupervisor: self._project_root = project_root self._readiness_timeout_s = readiness_timeout_s self._processes: dict[str, ManagedProcess] = {} + # Runtime pidfile lets us reap pipeline children left behind by a prior + # supervisor (crash/SIGKILL) independent of in-memory state (Fix #19). + self._runtime_dir = self._project_root / "python_app/runtime" + self._pidfile_path = self._runtime_dir / "supervisor_children.pids" + self._reap_stale_children() def is_running(self) -> bool: """Return whether acquisition-side processes are alive.""" @@ -160,6 +201,11 @@ class ProcessSupervisor: stdout_path = logs_dir / f"{name}.out.log" stderr_path = logs_dir / f"{name}.err.log" + # Roll any stale (uncollected) log to `.prev` before truncating so the + # previous run's diagnostics survive a respawn (Fix #29). + self._roll_log_to_prev(stdout_path) + self._roll_log_to_prev(stderr_path) + stdout_file = open(stdout_path, "wb") stderr_file = open(stderr_path, "wb") try: @@ -168,6 +214,10 @@ class ProcessSupervisor: cwd=self._project_root, stdout=stdout_file, stderr=stderr_file, + # Own session/process group so signalling the group on stop also + # reaches device-I/O grandchildren the producer may have spawned + # (Fix #33). + start_new_session=True, ) except OSError as exc: stdout_file.close() @@ -190,6 +240,7 @@ class ProcessSupervisor: stdout_path=stdout_path, stderr_path=stderr_path, ) + self._write_pidfile() def _acquisition_command(self, config_path: Path) -> list[str]: """Return acquisition producer command selected by radar.model.""" @@ -228,56 +279,78 @@ class ProcessSupervisor: return str(radar_payload.get("model", "librevna")) def _stop_processes(self, names: Iterable[str]) -> None: - """Gracefully terminate processes, then force-kill on timeout.""" + """Gracefully terminate processes, then force-kill on per-process timeout.""" ordered_names = list(names) for name in ordered_names: process = self._processes.get(name) if process is None: continue if process.handle.poll() is None: - process.handle.terminate() + # Signal the whole group so device-I/O grandchildren die too (Fix #33). + self._signal_group(process.handle.pid, signal.SIGTERM) - deadline = time.monotonic() + 2.0 for name in ordered_names: process = self._processes.get(name) if process is None: continue if process.handle.poll() is not None: + self._log_abnormal_stop_exit(process) continue - timeout = max(0.0, deadline - time.monotonic()) + # Each process gets its own kill deadline so a slow shutdown of one + # cannot consume the grace window of the others (Fix #33). try: - process.handle.wait(timeout=timeout) + process.handle.wait(timeout=_STOP_GRACE_SECONDS) except subprocess.TimeoutExpired: - process.handle.kill() - process.handle.wait(timeout=1.0) + self._signal_group(process.handle.pid, signal.SIGKILL) + try: + process.handle.wait(timeout=1.0) + except subprocess.TimeoutExpired: + pass + else: + # A negative code here is the SIGTERM we just sent (expected); only + # a positive self-exit during the grace window is worth noting. + self._log_abnormal_stop_exit(process) self._drop_exited() + @staticmethod + def _signal_group(pid: int, sig: int) -> None: + """Signal the child's whole process group, falling back to the child.""" + if pid is None: + return + try: + os.killpg(os.getpgid(pid), sig) + except (ProcessLookupError, PermissionError): + # Group already gone, or could not resolve it; fall back to the child. + try: + os.kill(pid, sig) + except (ProcessLookupError, PermissionError): + pass + + @staticmethod + def _log_abnormal_stop_exit(process: ManagedProcess) -> None: + """Note a process that self-exited abnormally around stop time (Fix #33). + + Negative codes are signal-induced (e.g. the SIGTERM we send on stop) and + are expected; only a non-zero self-exit is reported. + """ + return_code = process.handle.poll() + if return_code is None or return_code <= 0: + return + print( + f"process_supervisor: `{process.name}` exited abnormally with code " + f"{return_code} around stop", + file=sys.stderr, + ) + def _drop_exited(self) -> None: """Remove exited process entries from internal map.""" exited_names = [name for name, process in self._processes.items() if process.handle.poll() is not None] for name in exited_names: self._processes.pop(name, None) - - @staticmethod - def _read_log_tail(path: Path, max_bytes: int = 16384) -> str: - """Return the trailing `max_bytes` of a child log file, decoded best-effort. - - Bounded so a large/long-lived log never produces an enormous exit report. - """ - try: - with open(path, "rb") as handle: - handle.seek(0, 2) - size = handle.tell() - if size > max_bytes: - handle.seek(-max_bytes, 2) - else: - handle.seek(0) - data = handle.read() - except OSError: - return "" - return data.decode("utf-8", errors="replace").strip() + if exited_names: + self._write_pidfile() def _is_alive(self, name: str) -> bool: """Return `True` when named process handle exists and is running.""" @@ -294,19 +367,19 @@ class ProcessSupervisor: for name, process in self._processes.items(): return_code = process.handle.poll() if return_code is None: + # Still running: enforce the size cap so logs never grow unbounded (Fix #16). + self._roll_log_if_oversized(process.stdout_path) + self._roll_log_if_oversized(process.stderr_path) continue - stdout = self._read_log_tail(process.stdout_path) - stderr = self._read_log_tail(process.stderr_path) - reports.append( ProcessExitReport( name=process.name, command=list(process.command), working_directory=self._project_root, return_code=int(return_code), - stdout=stdout, - stderr=stderr, + stdout_path=process.stdout_path, + stderr_path=process.stderr_path, expected_clean_exit=bool(process.allow_clean_exit and int(return_code) == 0), ) ) @@ -314,6 +387,8 @@ class ProcessSupervisor: for name in exited_names: self._processes.pop(name, None) + if exited_names: + self._write_pidfile() return reports def _wait_until_ready(self, required_processes: Sequence[str]) -> None: @@ -342,3 +417,104 @@ class ProcessSupervisor: for process in self._processes.values() if process.handle.poll() is None and process.handle.pid is not None } + + @staticmethod + def _roll_log_to_prev(path: Path) -> None: + """Roll an existing log to `{path}.prev` before it is reopened (Fix #29). + + Preserves a stale (exited, not-yet-reported) child's last output instead + of truncating it when a fresh log is opened for a respawn. + """ + if not path.exists(): + return + try: + path.replace(path.with_suffix(path.suffix + ".prev")) + except OSError: + # Best-effort: a failed roll must not block a spawn. + pass + + @staticmethod + def _roll_log_if_oversized(path: Path) -> None: + """Bound a live child log to `_LOG_MAX_BYTES` so it cannot fill the SD card (Fix #16). + + The child holds an open fd to this inode, so a rename would not redirect + its writes. Instead keep one rolled generation via copy-to-`.prev` and + truncate the live inode in place, freeing the allocated disk blocks. + """ + try: + if path.stat().st_size <= _LOG_MAX_BYTES: + return + except OSError: + return + prev_path = path.with_suffix(path.suffix + ".prev") + try: + # Preserve the trailing window as the rolled generation, then truncate. + tail = _read_log_tail(path, _LOG_MAX_BYTES) + prev_path.write_text(tail, encoding="utf-8") + with open(path, "r+b") as handle: + handle.truncate(0) + except OSError: + # Best-effort: capping is opportunistic and must not disrupt polling. + pass + + def _write_pidfile(self) -> None: + """Persist live child PIDs so a later supervisor can reap them (Fix #19).""" + try: + self._runtime_dir.mkdir(parents=True, exist_ok=True) + live_pids = [ + str(process.handle.pid) + for process in self._processes.values() + if process.handle.poll() is None and process.handle.pid is not None + ] + self._pidfile_path.write_text("\n".join(live_pids), encoding="utf-8") + except OSError: + # Best-effort bookkeeping: failure here must not break start/stop. + pass + + def _reap_stale_children(self) -> None: + """Kill pipeline children recorded by a prior supervisor instance (Fix #19). + + On a clean shutdown the pidfile is emptied; entries only remain when the + previous supervisor died without stopping its children. We SIGKILL each + stale process group so leftover pipeline binaries cannot hold the shared + memory rings or devices hostage on the next start. + """ + try: + raw = self._pidfile_path.read_text(encoding="utf-8") + except OSError: + return + for token in raw.split(): + try: + pid = int(token) + except ValueError: + continue + if pid <= 1 or pid == os.getpid(): + continue + # Guard against PID reuse: only reap if the process still looks like + # one of our pipeline children before signalling its group. + if self._is_stale_pipeline_pid(pid): + self._signal_group(pid, signal.SIGKILL) + try: + self._pidfile_path.write_text("", encoding="utf-8") + except OSError: + pass + + def _is_stale_pipeline_pid(self, pid: int) -> bool: + """Return whether `pid` still runs one of our pipeline binaries/scripts. + + Reads `/proc//cmdline` so a recycled PID owned by an unrelated + process is never killed (Fix #19 safety guard). + """ + markers = ( + "build/bin/data_processor", + "build/bin/data_preprocessor", + "build/bin/sweep_orchestrator", + "python_app.scripts.matrix_raw_producer", + "python_app.scripts.kamil_adc_raw_producer", + ) + try: + raw = (Path("/proc") / str(pid) / "cmdline").read_bytes() + except OSError: + return False + cmdline = raw.replace(b"\x00", b" ").decode("utf-8", errors="replace") + return any(marker in cmdline for marker in markers) diff --git a/python_app/orchestration/shm/ring_reader.py b/python_app/orchestration/shm/ring_reader.py index 3caf923..0d69a95 100644 --- a/python_app/orchestration/shm/ring_reader.py +++ b/python_app/orchestration/shm/ring_reader.py @@ -53,16 +53,34 @@ class ShmRingReader: index = read_seq % self.capacity slot_offset = _HEADER_SIZE + index * (_SLOT_HEADER_SIZE + self.slot_size_bytes) - payload_size = self._read_u32(slot_offset) + # Seqlock read mirroring the C++ pop: a slot is valid for read_seq R only if + # its sequence equals R+1 and is unchanged across the payload copy (i.e. the + # producer did not overwrite this slot mid-copy). Sequence and payload_size + # are read first; the slot is only accepted after the re-read confirms both. sequence = self._read_u64(slot_offset + 8) if sequence != read_seq + 1: + # Producer overwrote this slot before we read it. Resync to latest. + self._write_u64(32, write_seq) + return None + + payload_size = self._read_u32(slot_offset) + # Bound payload_size against the slot before slicing so a torn/garbage size + # can never read out of the slot region; resync and skip on violation. + if payload_size > self.slot_size_bytes: self._write_u64(32, write_seq) return None payload_offset = slot_offset + _SLOT_HEADER_SIZE - payload = self._mmap[payload_offset : payload_offset + payload_size] + payload = bytes(self._mmap[payload_offset : payload_offset + payload_size]) + + # Re-read the slot sequence after the copy; if it changed, the producer + # overwrote this slot mid-copy and the payload is torn — discard and resync. + if self._read_u64(slot_offset + 8) != read_seq + 1: + self._write_u64(32, write_seq) + return None + self._write_u64(32, read_seq + 1) - return bytes(payload) + return payload def pop_raw_collection(self) -> SweepCollection | None: """Read next raw collection from ring.""" diff --git a/python_app/orchestration/shm/ring_writer.py b/python_app/orchestration/shm/ring_writer.py index 807732f..149ad32 100644 --- a/python_app/orchestration/shm/ring_writer.py +++ b/python_app/orchestration/shm/ring_writer.py @@ -15,10 +15,20 @@ _VERSION: Final[int] = 1 class ShmRingWriter: - """Write binary payloads into the shared-memory ring used by C++ workers.""" + """Write binary payloads into the shared-memory ring used by C++ workers. + + The writer is the sole *owner* of the rings it opens: there is exactly one + producer per ring (the acquisition producer for the raw/raw_tap rings). On a + geometry mismatch with a pre-existing segment (e.g. a stale ring left by a prior + run with a different sweep config), the owner unlinks and recreates the segment + from scratch rather than truncating in place or diverging silently — mirroring + the clean-shm-on-restart contract on the C++/deploy side (#13). A non-owner must + never recreate a ring; readers and C++ consumers only ever attach to an existing + one. + """ def __init__(self, ring_name: str, capacity: int, slot_size_bytes: int) -> None: - """Open or create a POSIX SHM ring by name.""" + """Open or create a POSIX SHM ring by name (as the ring owner).""" if not ring_name.startswith("/"): raise ValueError("ring_name must start with '/'") if capacity <= 0: @@ -31,19 +41,50 @@ class ShmRingWriter: self._slot_size_bytes = int(slot_size_bytes) self._mapped_size = _HEADER_SIZE + self._capacity * (_SLOT_HEADER_SIZE + self._slot_size_bytes) self._path = Path("/dev/shm") / ring_name[1:] + self._open_owned() + def _open_owned(self) -> None: + """Open the ring, recreating it from scratch on a geometry/header mismatch. + + As the single owner of this ring we may safely discard a stale segment: a + size or header mismatch means the existing segment belongs to an earlier, + incompatible run, so we unlink it and create a fresh one instead of mapping + an inconsistent layout. + """ created = not self._path.exists() fd = os.open(self._path, os.O_RDWR | os.O_CREAT, 0o660) self._file = os.fdopen(fd, "r+b", buffering=0) if created or self._path.stat().st_size != self._mapped_size: + # Wrong-sized stale segment: drop it entirely and recreate, so the file + # and any future mapping agree on geometry instead of being truncated + # under a producer/consumer that still expects the old layout. self._file.truncate(self._mapped_size) created = True self._mmap = mmap.mmap(self._file.fileno(), self._mapped_size) if created: self._initialize_header() - else: - self._validate_header() + return + + # Size matched but the header geometry/magic does not: the owner recreates + # rather than diverge. Unlink and reopen as a brand-new ring. + if not self._header_matches(): + self._mmap.close() + self._file.close() + self._unlink_if_present() + created = not self._path.exists() + fd = os.open(self._path, os.O_RDWR | os.O_CREAT, 0o660) + self._file = os.fdopen(fd, "r+b", buffering=0) + self._file.truncate(self._mapped_size) + self._mmap = mmap.mmap(self._file.fileno(), self._mapped_size) + self._initialize_header() + + def _unlink_if_present(self) -> None: + """Remove the backing /dev/shm file if it exists (owner-only operation).""" + try: + self._path.unlink() + except FileNotFoundError: + pass def close(self) -> None: """Close mmap and file handle.""" @@ -106,6 +147,19 @@ class ShmRingWriter: if capacity != self._capacity or slot_size_bytes != self._slot_size_bytes: raise RuntimeError(f"Shared memory ring geometry mismatch for {self._ring_name}") + def _header_matches(self) -> bool: + """Return whether the existing segment's header matches this ring's geometry. + + Non-throwing counterpart of `_validate_header` used by the owner to decide + whether a same-sized pre-existing segment can be reused or must be recreated. + """ + return ( + self._mmap[:8] == _MAGIC + and self._read_u32(8) == _VERSION + and self._read_u32(12) == self._capacity + and self._read_u32(16) == self._slot_size_bytes + ) + def _read_u32(self, offset: int) -> int: return struct.unpack_from(" bool: + """Open+configure the radar and both switches, retrying forever until stop. + + Used for both the initial open and every in-loop reconnect, so a collector + that is absent at boot or disappears mid-run never kills the producer — it just + waits. Any partially-opened components are closed before each retry so a + relaunched collector starts clean. Returns ``True`` once everything is open, or + ``False`` if a stop was requested before the device became available. Backoff is + capped and every wait is interruptible by SIGTERM. + """ + # Tear down any prior open first: open()/switch.open() are idempotent no-ops + # while still "open", so a mid-run reconnect must close them to force a fresh + # collector relaunch and TTY re-attach. + with suppress(Exception): + input_switch.close() + with suppress(Exception): + output_switch.close() + with suppress(Exception): + radar.close() + + attempt = 0 + delay = _OPEN_RETRY_MIN_S + while not stop_requested.is_set(): + try: + radar.open(stop_event=stop_requested) + radar.configure(config.radar.sweep) + output_switch.open() + input_switch.open() + except Exception as exc: # noqa: BLE001 — waiting for the device is the point + # Drop any partial open (collector process, TTY reader, switches) + # before the next attempt so the relaunch starts from a clean state. + with suppress(Exception): + input_switch.close() + with suppress(Exception): + output_switch.close() + with suppress(Exception): + radar.close() + attempt += 1 + if attempt == 1 or attempt % _OPEN_RETRY_LOG_EVERY == 0: + logger.warning( + "Kamil ADC not available (attempt %d); retrying every up to %.0fs " + "until the device is present: %s", + attempt, + _OPEN_RETRY_MAX_S, + exc, + ) + if stop_requested.wait(delay): + return False + delay = min(delay * 2.0, _OPEN_RETRY_MAX_S) + continue + + if attempt > 0: + logger.info("Kamil ADC opened after %d attempt(s).", attempt + 1) + return True + + return False + def main() -> int: """Run producer process until config or signal requests exit.""" @@ -57,10 +132,8 @@ def main() -> int: output_switch = _switch_from_model(config.output_switch) try: - radar.open() - radar.configure(config.radar.sweep) - output_switch.open() - input_switch.open() + if not _open_radar_with_retry(config, radar, input_switch, output_switch, stop_requested): + return 0 # asked to stop before a device became available collection_id = 1 while not stop_requested.is_set(): @@ -68,26 +141,49 @@ def main() -> int: capture_start_ns = time.monotonic_ns() traces: list[TraceData] = [] - for combo in config.combos: + try: + for combo in config.combos: + if stop_requested.is_set(): + break + output_switch.switch_to(combo.output) + input_switch.switch_to(combo.input) + if config.runtime.settling_ms > 0: + time.sleep(config.runtime.settling_ms / 1000.0) + + sweep = radar.acquire() + traces.append( + TraceData( + combo=ComboKey(input=combo.input, output=combo.output), + frequency_hz=np.asarray(sweep.x, dtype=np.float32), + s11=np.asarray(sweep.trace("s11"), dtype=np.complex64), + s21=np.asarray(sweep.trace("s21"), dtype=np.complex64), + ) + ) + except Exception as exc: # noqa: BLE001 — reconnect forever, never give up + logger.warning( + "Kamil ADC acquisition failed; reconnecting and waiting for the device: %s", + exc, + exc_info=True, + ) + if not _open_radar_with_retry(config, radar, input_switch, output_switch, stop_requested): + break # stop requested while waiting to reconnect + continue + + # An incomplete sweep set means the device dropped out (or a stop was + # requested mid-collection). Only exit on stop; otherwise reconnect and + # wait for the device rather than killing the producer. + if len(traces) != len(config.combos): if stop_requested.is_set(): break - output_switch.switch_to(combo.output) - input_switch.switch_to(combo.input) - if config.runtime.settling_ms > 0: - time.sleep(config.runtime.settling_ms / 1000.0) - - sweep = radar.acquire() - traces.append( - TraceData( - combo=ComboKey(input=combo.input, output=combo.output), - frequency_hz=np.asarray(sweep.x, dtype=np.float32), - s11=np.asarray(sweep.trace("s11"), dtype=np.complex64), - s21=np.asarray(sweep.trace("s21"), dtype=np.complex64), - ) + logger.warning( + "Kamil ADC produced an incomplete collection (%d of %d combos); " + "reconnecting and waiting for the device", + len(traces), + len(config.combos), ) - - if len(traces) != len(config.combos): - break + if not _open_radar_with_retry(config, radar, input_switch, output_switch, stop_requested): + break # stop requested while waiting to reconnect + continue collection = SweepCollection( collection_id=collection_id,