diff --git a/docs/module/manuals/api_description/use_cases.rst b/docs/module/manuals/api_description/use_cases.rst index f7dab446..90a33ec7 100644 --- a/docs/module/manuals/api_description/use_cases.rst +++ b/docs/module/manuals/api_description/use_cases.rst @@ -313,12 +313,13 @@ diagnostics and PTP data sanity checks: -.. warning:: - - Both PTP data callbacks (``TimeSlaveSyncData`` and ``PDelayMeasurementData``) are - **not yet delivered**. Calling ``Subscribe<...>()`` compiles and runs without error, - but the registered callbacks will never be invoked. Delivery will be wired from a - dedicated background thread in a future change. +Delivery is performed by a dedicated worker thread owned by the ``VehicleTime`` backend. +The TimeDaemon publishes into a shared-memory segment without a notification facility, so +the worker polls that segment at a fixed interval (50 ms). The thread is started when the +first callback is registered and runs until the backend is destroyed; while no callback is +registered it sleeps until the next registration. A newly registered callback receives the first +frame polled after its registration; afterwards it is invoked only for frames whose sync or +pDelay content differs from the previously delivered one. .. code-block:: cpp @@ -376,13 +377,6 @@ excluded from the comparison. -.. warning:: - - The ``VehicleTimeStatus`` callback is **not yet delivered**. Calling - ``Subscribe()`` compiles and runs without error, but the registered - callback will never be invoked. Delivery will be wired from a dedicated background - thread in a future change. - .. code-block:: cpp #include "score/time/vehicle_time/src/vehicle_clock.h" diff --git a/docs/module/manuals/examples/vehicle_time.rst b/docs/module/manuals/examples/vehicle_time.rst index 6bebd354..82027188 100644 --- a/docs/module/manuals/examples/vehicle_time.rst +++ b/docs/module/manuals/examples/vehicle_time.rst @@ -114,7 +114,7 @@ Key features: - **Dual time sources**: Both vehicle and local time in single call - **Status monitoring**: Reliability and consistency flags - **Rate tracking**: Clock deviation measurement -- **Callback support**: Status change notifications (future feature) +- **Callback support**: Status change notifications delivered on the backend's worker thread Main Program ~~~~~~~~~~~~ diff --git a/examples/time/vehicle_time/src/vehicle_time_handler.h b/examples/time/vehicle_time/src/vehicle_time_handler.h index c2187f0b..63c692f7 100644 --- a/examples/time/vehicle_time/src/vehicle_time_handler.h +++ b/examples/time/vehicle_time/src/vehicle_time_handler.h @@ -109,9 +109,9 @@ class VehicleTimeHandler /// @brief Registers a callback that is invoked when VehicleTimeStatus flags change. /// - /// @note Delivery is not yet implemented in the backend. The callback can - /// be registered now; it will be invoked once background-thread - /// delivery is wired up in a future change. + /// The callback fires once with the current status right after registration and + /// afterwards whenever the status flags change. It is invoked on the backend's + /// worker thread, so the callback implementation must be thread-safe. void RegisterStatusCallback(score::time::VehicleTime::StatusChangedCallback callback) noexcept { clock_.Subscribe(std::move(callback)); diff --git a/score/time/vehicle_time/src/details/td_impl/BUILD b/score/time/vehicle_time/src/details/td_impl/BUILD index 89ab8698..797b5584 100644 --- a/score/time/vehicle_time/src/details/td_impl/BUILD +++ b/score/time/vehicle_time/src/details/td_impl/BUILD @@ -14,29 +14,78 @@ load("@score_baselibs//:bazel/unit_tests.bzl", "cc_unit_test_suites_for_host_and_qnx") load("@score_baselibs//score/language/safecpp:toolchain_features.bzl", "COMPILER_WARNING_FEATURES") -# Production backend: reads live PTP data from the TimeDaemon via SvtReceiver. +# Production backend: reads live PTP data from the TimeDaemon via SvtReceiver and +# delivers subscription callbacks from a dedicated worker thread. # Link this target (or the forwarding alias //score/time/vehicle_time/src/details:td_impl) # into the production binary to provide CreateBackend(). cc_library( name = "td_impl", srcs = [ + "svt_callback_dispatcher.cpp", "vehicle_clock_backend_impl.cpp", "//score/time/vehicle_time/src/details:logging_contexts", # internal header ], - hdrs = ["vehicle_clock_backend_impl.h"], + hdrs = [ + "svt_callback_dispatcher.h", + "svt_callback_wrapper.h", + "vehicle_clock_backend_impl.h", + ], features = COMPILER_WARNING_FEATURES, visibility = ["//score/time/vehicle_time:__subpackages__"], deps = [ "//score/time/high_res_steady_time/src:high_res_steady_clock", "//score/time/vehicle_time/src:vehicle_clock", "//score/time_daemon/src/ipc:svt_receiver", + "@score_baselibs//score/concurrency:condition_variable", + "@score_baselibs//score/language/futurecpp", "@score_baselibs//score/mw/log:frontend", ], ) +cc_test( + name = "svt_callback_wrapper_test", + srcs = [ + "svt_callback_wrapper.h", + "svt_callback_wrapper_test.cpp", + ], + features = COMPILER_WARNING_FEATURES, + tags = ["unit"], + deps = [ + "@googletest//:gtest", + "@googletest//:gtest_main", + "@score_baselibs//score/language/futurecpp", + ], +) + +cc_test( + name = "svt_callback_dispatcher_test", + srcs = [ + "svt_callback_dispatcher.cpp", + "svt_callback_dispatcher.h", + "svt_callback_dispatcher_test.cpp", + "svt_callback_wrapper.h", + "svt_test_helpers.h", + ], + features = COMPILER_WARNING_FEATURES, + tags = ["unit"], + deps = [ + "//score/time/vehicle_time/src:vehicle_clock", + "//score/time_daemon/src/ipc:svt_receiver_mock", + "@googletest//:gtest", + "@googletest//:gtest_main", + "@score_baselibs//score/concurrency:condition_variable", + "@score_baselibs//score/language/futurecpp", + "@score_baselibs//score/language/safecpp/coverage_termination_handler", + ], +) + cc_test( name = "vehicle_clock_backend_impl_test", srcs = [ + "svt_callback_dispatcher.cpp", + "svt_callback_dispatcher.h", + "svt_callback_wrapper.h", + "svt_test_helpers.h", "vehicle_clock_backend_impl.cpp", "vehicle_clock_backend_impl.h", "vehicle_clock_backend_impl_test.cpp", @@ -53,6 +102,8 @@ cc_test( "//score/time_daemon/src/ipc:svt_receiver_mock", "@googletest//:gtest", "@googletest//:gtest_main", + "@score_baselibs//score/concurrency:condition_variable", + "@score_baselibs//score/language/futurecpp", "@score_baselibs//score/language/safecpp/coverage_termination_handler", "@score_baselibs//score/mw/log:console_only_backend", "@score_baselibs//score/mw/log:frontend", @@ -61,6 +112,10 @@ cc_test( cc_unit_test_suites_for_host_and_qnx( name = "unit_test_suite", - cc_unit_tests = [":vehicle_clock_backend_impl_test"], + cc_unit_tests = [ + ":svt_callback_dispatcher_test", + ":svt_callback_wrapper_test", + ":vehicle_clock_backend_impl_test", + ], visibility = ["//score/time/vehicle_time:__subpackages__"], ) diff --git a/score/time/vehicle_time/src/details/td_impl/svt_callback_dispatcher.cpp b/score/time/vehicle_time/src/details/td_impl/svt_callback_dispatcher.cpp new file mode 100644 index 00000000..3219e007 --- /dev/null +++ b/score/time/vehicle_time/src/details/td_impl/svt_callback_dispatcher.cpp @@ -0,0 +1,269 @@ +/******************************************************************************** + * Copyright (c) 2026 Contributors to the Eclipse Foundation + * + * See the NOTICE file(s) distributed with this work for additional + * information regarding copyright ownership. + * + * This program and the accompanying materials are made available under the + * terms of the Apache License Version 2.0 which is available at + * https://www.apache.org/licenses/LICENSE-2.0 + * + * SPDX-License-Identifier: Apache-2.0 + ********************************************************************************/ +#include "score/time/vehicle_time/src/details/td_impl/svt_callback_dispatcher.h" + +#include + +#include +#include + +namespace score +{ +namespace time +{ +namespace detail +{ + +namespace +{ + +/// @brief Reinterprets an unsigned nanosecond count from the IPC layer as a signed chrono duration. +std::chrono::nanoseconds ToNanoseconds(const std::uint64_t nanoseconds) noexcept +{ + return std::chrono::nanoseconds{static_cast(nanoseconds)}; +} + +PortIdentity ToPortIdentity(const std::uint64_t clock_identity, const std::uint32_t port_number) noexcept +{ + PortIdentity identity{}; + identity.clock_identity = clock_identity; + identity.port_number = static_cast(port_number); + return identity; +} + +/// @brief Converts the IPC sync/follow-up snapshot to the public event type. +TimeSlaveSyncData ConvertSyncData(const score::td::svt::SyncFupSnapshot& sync_data) noexcept +{ + TimeSlaveSyncData converted{}; + converted.precise_origin_timestamp = VehicleTime::Timepoint{ToNanoseconds(sync_data.precise_origin_timestamp)}; + converted.reference_global_timestamp = VehicleTime::Timepoint{ToNanoseconds(sync_data.reference_global_timestamp)}; + converted.reference_local_timestamp = LocalPTPDeviceTimerValue{ToNanoseconds(sync_data.reference_local_timestamp)}; + converted.sync_ingress_timestamp = LocalPTPDeviceTimerValue{ToNanoseconds(sync_data.sync_ingress_timestamp)}; + converted.correction_field = static_cast(sync_data.correction_field); + converted.sequence_id = sync_data.sequence_id; + converted.pdelay = ToNanoseconds(sync_data.pdelay); + converted.source_port_identity = ToPortIdentity(sync_data.clock_identity, sync_data.port_number); + return converted; +} + +/// @brief Converts the IPC pDelay snapshot to the public event type. +PDelayMeasurementData ConvertPDelayData(const score::td::svt::PDelayDataSnapshot& pdelay_data) noexcept +{ + PDelayMeasurementData converted{}; + converted.request_origin_timestamp = LocalPTPDeviceTimerValue{ToNanoseconds(pdelay_data.request_origin_timestamp)}; + converted.request_receipt_timestamp = + MasterPTPDeviceTimerValue{ToNanoseconds(pdelay_data.request_receipt_timestamp)}; + converted.response_origin_timestamp = + MasterPTPDeviceTimerValue{ToNanoseconds(pdelay_data.response_origin_timestamp)}; + converted.response_receipt_timestamp = + LocalPTPDeviceTimerValue{ToNanoseconds(pdelay_data.response_receipt_timestamp)}; + converted.reference_global_timestamp = + VehicleTime::Timepoint{ToNanoseconds(pdelay_data.reference_global_timestamp)}; + converted.reference_local_timestamp = + LocalPTPDeviceTimerValue{ToNanoseconds(pdelay_data.reference_local_timestamp)}; + converted.sequence_id = pdelay_data.sequence_id; + converted.pdelay = ToNanoseconds(pdelay_data.pdelay); + converted.request_port_identity = ToPortIdentity(pdelay_data.req_clock_identity, pdelay_data.req_port_number); + converted.response_port_identity = ToPortIdentity(pdelay_data.resp_clock_identity, pdelay_data.resp_port_number); + return converted; +} + +} // namespace + +ClockStatus ConvertPtpStatus(const score::td::svt::TimeBaseStatus& ptp_status) noexcept +{ + using Flag = VehicleTime::StatusFlag; + if (!ptp_status.is_correct) + { + return ClockStatus{}; + } + ClockStatus status; + if (ptp_status.is_synchronized) + { + status.AddFlag(Flag::kSynchronized); + } + if (ptp_status.is_timeout) + { + status.AddFlag(Flag::kTimeOut); + } + if (ptp_status.is_time_jump_future) + { + status.AddFlag(Flag::kTimeLeapFuture); + } + if (ptp_status.is_time_jump_past) + { + status.AddFlag(Flag::kTimeLeapPast); + } + return status; +} + +SvtCallbackDispatcher::SvtCallbackDispatcher(std::shared_ptr receiver, + const std::chrono::milliseconds poll_interval) noexcept + : svt_receiver_{std::move(receiver)}, + poll_interval_{poll_interval}, + sync_data_slot_{}, + pdelay_slot_{}, + status_slot_{}, + worker_mutex_{}, + enabled_{false}, + callback_registered_{false}, + wakeup_requested_{false}, + worker_wakeup_{}, + worker_{} +{ +} + +SvtCallbackDispatcher::~SvtCallbackDispatcher() noexcept +{ + // The destructor never runs on the worker thread, so joining here is always safe. + if (worker_.joinable()) + { + score::cpp::ignore = worker_.request_stop(); + { + const std::lock_guard guard{worker_mutex_}; + worker_wakeup_.notify_all(); + } + worker_.join(); + } +} + +void SvtCallbackDispatcher::Start() noexcept +{ + const std::lock_guard guard{worker_mutex_}; + enabled_ = true; + StartWorkerIfReadyLocked(); +} + +template +void SvtCallbackDispatcher::SetCallback(Slot& slot, Callback&& callback) noexcept +{ + const bool present = !callback.empty(); + slot.Set(std::move(callback)); + if (present) + { + OnCallbackRegistered(); + } +} + +void SvtCallbackDispatcher::SetTimeSlaveSyncDataReceivedCallback( + VehicleTime::TimeSlaveSyncDataReceivedCallback&& callback) noexcept +{ + SetCallback(sync_data_slot_, std::move(callback)); +} + +void SvtCallbackDispatcher::UnsetTimeSlaveSyncDataReceivedCallback() noexcept +{ + sync_data_slot_.Unset(); +} + +void SvtCallbackDispatcher::SetPDelayMeasurementFinishedCallback( + VehicleTime::PDelayMeasurementFinishedCallback&& callback) noexcept +{ + SetCallback(pdelay_slot_, std::move(callback)); +} + +void SvtCallbackDispatcher::UnsetPDelayMeasurementFinishedCallback() noexcept +{ + pdelay_slot_.Unset(); +} + +void SvtCallbackDispatcher::SetStatusChangedCallback(VehicleTime::StatusChangedCallback&& callback) noexcept +{ + SetCallback(status_slot_, std::move(callback)); +} + +void SvtCallbackDispatcher::UnsetStatusChangedCallback() noexcept +{ + status_slot_.Unset(); +} + +void SvtCallbackDispatcher::OnCallbackRegistered() noexcept +{ + const std::lock_guard guard{worker_mutex_}; + callback_registered_ = true; + wakeup_requested_ = true; + StartWorkerIfReadyLocked(); + // Also delivers the first frame promptly to a fresh registration instead of after a full poll interval. + worker_wakeup_.notify_all(); +} + +void SvtCallbackDispatcher::StartWorkerIfReadyLocked() noexcept +{ + // The worker is never stopped before destruction, so "already running" is simply "already spawned". + // This is also what makes a callback (re-)registering from the worker thread trivially safe. + if (!enabled_ || !callback_registered_ || worker_.joinable()) + { + return; + } + worker_ = score::cpp::jthread{score::cpp::jthread::name_hint{std::string{"vt_cb_dispatch"}}, + [this](const score::cpp::stop_token token) noexcept { + WorkerFunction(token); + }}; +} + +void SvtCallbackDispatcher::WorkerFunction(const score::cpp::stop_token& token) noexcept +{ + while (!token.stop_requested()) + { + // Queried before taking worker_mutex_: the slots must never be locked underneath it, because a + // callback running under its slot mutex takes worker_mutex_ when it (re-)registers. + const bool any_callback_set = IsAnyCallbackSet(); + if (any_callback_set) + { + PollAndDispatch(); + } + + std::unique_lock lock{worker_mutex_}; + const auto wake_up = [this, &token]() noexcept -> bool { + return token.stop_requested() || wakeup_requested_; + }; + if (any_callback_set) + { + score::cpp::ignore = worker_wakeup_.wait_for(lock, token, poll_interval_, wake_up); + } + else + { + // Nobody to deliver to: sleep until the next registration rather than ticking every interval. + score::cpp::ignore = worker_wakeup_.wait(lock, token, wake_up); + } + wakeup_requested_ = false; + } +} + +bool SvtCallbackDispatcher::IsAnyCallbackSet() const noexcept +{ + return sync_data_slot_.IsSet() || pdelay_slot_.IsSet() || status_slot_.IsSet(); +} + +void SvtCallbackDispatcher::PollAndDispatch() noexcept +{ + const auto frame = svt_receiver_->Receive(); + if (!frame.has_value()) + { + return; + } + + const auto& sync_data = frame.value().sync_fup_data; + score::cpp::ignore = sync_data_slot_.InvokeIfChanged(sync_data, ConvertSyncData(sync_data)); + + const auto& pdelay_data = frame.value().pdelay_data; + score::cpp::ignore = pdelay_slot_.InvokeIfChanged(pdelay_data, ConvertPDelayData(pdelay_data)); + + const auto status_flags = ConvertPtpStatus(frame.value().status); + score::cpp::ignore = + status_slot_.InvokeIfChanged(status_flags, VehicleTimeStatus{status_flags, frame.value().rate_deviation}); +} + +} // namespace detail +} // namespace time +} // namespace score diff --git a/score/time/vehicle_time/src/details/td_impl/svt_callback_dispatcher.h b/score/time/vehicle_time/src/details/td_impl/svt_callback_dispatcher.h new file mode 100644 index 00000000..5b21556e --- /dev/null +++ b/score/time/vehicle_time/src/details/td_impl/svt_callback_dispatcher.h @@ -0,0 +1,141 @@ +/******************************************************************************** + * Copyright (c) 2026 Contributors to the Eclipse Foundation + * + * See the NOTICE file(s) distributed with this work for additional + * information regarding copyright ownership. + * + * This program and the accompanying materials are made available under the + * terms of the Apache License Version 2.0 which is available at + * https://www.apache.org/licenses/LICENSE-2.0 + * + * SPDX-License-Identifier: Apache-2.0 + ********************************************************************************/ +#ifndef SCORE_TIME_VEHICLE_TIME_SRC_DETAILS_TD_IMPL_SVT_CALLBACK_DISPATCHER_H +#define SCORE_TIME_VEHICLE_TIME_SRC_DETAILS_TD_IMPL_SVT_CALLBACK_DISPATCHER_H + +// Internal header — include ONLY from translation units under vehicle_time/src/details/td_impl/. +// NOT part of the public API of td_impl. + +#include "score/concurrency/condition_variable.h" +#include "score/time/vehicle_time/src/details/td_impl/svt_callback_wrapper.h" +#include "score/time/vehicle_time/src/vehicle_clock.h" +#include "score/time_daemon/src/ipc/svt/receiver/svt_receiver.h" +#include "score/time_daemon/src/ipc/svt/svt_time_info.h" + +#include +#include + +#include +#include +#include + +namespace score +{ +namespace time +{ +namespace detail +{ + +/// @brief Converts PTP status flags from the TimeDaemon IPC representation to +/// the @c ClockStatus representation. +ClockStatus ConvertPtpStatus(const score::td::svt::TimeBaseStatus& ptp_status) noexcept; + +/// @brief Owns the worker thread that delivers vehicle-time subscription callbacks. +/// +/// The TimeDaemon IPC is a shared-memory segment without a notification facility, so the +/// dispatcher owns a dedicated worker thread that polls the receiver every @p poll_interval. +/// @c Start() enables dispatching once the receiver is initialised. The worker thread is created +/// exactly once, as soon as dispatching is enabled and the first callback has been registered, and +/// runs until the dispatcher is destroyed; while no callback is registered it sleeps until the next +/// registration instead of ticking idly. +/// Every frame read from the receiver is offered part by part to the three @c SvtCallbackWrapper s, which +/// dispatch on the worker thread when the part differs from what they last delivered: +/// - @c TimeSlaveSyncData — the sync/follow-up part; +/// - @c PDelayMeasurementData — the pDelay part; +/// - @c VehicleTimeStatus — keyed on the status flags (rate deviation excluded from the comparison). +/// A newly registered callback (first registration, re-registration, or replacement) always receives +/// the first frame polled after its registration, and afterwards only changes. +/// +/// Set/Unset are safe to call concurrently with an in-flight invocation (see @c SvtCallbackWrapper), +/// including a callback that unsets or replaces itself from the worker thread. +class SvtCallbackDispatcher final +{ + public: + SvtCallbackDispatcher(std::shared_ptr receiver, + std::chrono::milliseconds poll_interval) noexcept; + + ~SvtCallbackDispatcher() noexcept; + SvtCallbackDispatcher(const SvtCallbackDispatcher&) = delete; + SvtCallbackDispatcher& operator=(const SvtCallbackDispatcher&) = delete; + SvtCallbackDispatcher(SvtCallbackDispatcher&&) = delete; + SvtCallbackDispatcher& operator=(SvtCallbackDispatcher&&) = delete; + + /// @brief Enables dispatching once the receiver is initialised. Must be called at most once. + /// + /// The worker thread is spun up as soon as dispatching is enabled and a callback has been + /// registered, in whichever order those two happen. + void Start() noexcept; + + void SetTimeSlaveSyncDataReceivedCallback(VehicleTime::TimeSlaveSyncDataReceivedCallback&& callback) noexcept; + + void UnsetTimeSlaveSyncDataReceivedCallback() noexcept; + + void SetPDelayMeasurementFinishedCallback(VehicleTime::PDelayMeasurementFinishedCallback&& callback) noexcept; + + void UnsetPDelayMeasurementFinishedCallback() noexcept; + + void SetStatusChangedCallback(VehicleTime::StatusChangedCallback&& callback) noexcept; + + void UnsetStatusChangedCallback() noexcept; + + private: + /// @brief Installs @p callback into @p slot. A non-empty callback also starts or wakes the worker; + /// an empty one behaves like the corresponding Unset. + template + void SetCallback(Slot& slot, Callback&& callback) noexcept; + + /// @brief Worker thread body: polls the receiver every @c poll_interval_ while at least one callback is + /// registered, otherwise sleeps until the next registration or stop request. + void WorkerFunction(const score::cpp::stop_token& token) noexcept; + + /// @brief Returns @c true if at least one callback is currently registered. + bool IsAnyCallbackSet() const noexcept; + + /// @brief Reads one frame from the receiver and offers each part to its slot. + void PollAndDispatch() noexcept; + + /// @brief Records that a non-empty callback was registered: starts the worker if dispatching is enabled, + /// or wakes it if it is already running. + void OnCallbackRegistered() noexcept; + + /// @brief Spawns the worker once dispatching is enabled and a callback has been registered; a no-op + /// if it is already running. Must be called with @c worker_mutex_ held. + void StartWorkerIfReadyLocked() noexcept; + + std::shared_ptr svt_receiver_; + const std::chrono::milliseconds poll_interval_; + + SvtCallbackWrapper sync_data_slot_; + SvtCallbackWrapper pdelay_slot_; + SvtCallbackWrapper> status_slot_; + + // Guards the worker state below and is the mutex the worker sleeps on. Never held while a slot + // mutex is taken, so that a callback may (re-)register from the worker thread without risking a + // lock-order inversion. + std::mutex worker_mutex_; + bool enabled_; + // Set once the first non-empty callback is registered and never cleared: the worker outlives + // its subscribers, so a later Unset() must not affect it. + bool callback_registered_; + // Raised by every registration and consumed by the worker: ends an indefinite sleep, and covers a + // registration that happens between the worker's slot check and its wait (no lost wake-up). + bool wakeup_requested_; + score::concurrency::InterruptibleConditionalVariable worker_wakeup_; + score::cpp::jthread worker_; +}; + +} // namespace detail +} // namespace time +} // namespace score + +#endif // SCORE_TIME_VEHICLE_TIME_SRC_DETAILS_TD_IMPL_SVT_CALLBACK_DISPATCHER_H diff --git a/score/time/vehicle_time/src/details/td_impl/svt_callback_dispatcher_test.cpp b/score/time/vehicle_time/src/details/td_impl/svt_callback_dispatcher_test.cpp new file mode 100644 index 00000000..2798d1b9 --- /dev/null +++ b/score/time/vehicle_time/src/details/td_impl/svt_callback_dispatcher_test.cpp @@ -0,0 +1,591 @@ +/******************************************************************************** + * Copyright (c) 2026 Contributors to the Eclipse Foundation + * + * See the NOTICE file(s) distributed with this work for additional + * information regarding copyright ownership. + * + * This program and the accompanying materials are made available under the + * terms of the Apache License Version 2.0 which is available at + * https://www.apache.org/licenses/LICENSE-2.0 + * + * SPDX-License-Identifier: Apache-2.0 + ********************************************************************************/ +#include "score/time/vehicle_time/src/details/td_impl/svt_callback_dispatcher.h" + +#include "score/time/vehicle_time/src/details/td_impl/svt_test_helpers.h" +#include "score/time/vehicle_time/src/vehicle_clock.h" + +#include + +#include +#include +#include +#include +#include +#include + +namespace score +{ +namespace time +{ +namespace +{ + +using namespace std::chrono_literals; +using namespace test_helpers; + +// SvtCallbackWrapper semantics (first delivery after (re-)registration, replacement, Unset) are covered +// in svt_callback_wrapper_test.cpp. The tests here cover what the dispatcher adds on top: the worker +// lifecycle, which part of a frame keys which callback, the field conversions, and Set/Unset from +// other threads while the worker is running. + +class SvtCallbackDispatcherTest : public ::testing::Test +{ + protected: + SvtCallbackDispatcherTest() + : mock_svt_{std::make_shared()}, + frame_source_{}, + dispatcher_{std::make_unique(mock_svt_, kPollInterval)} + { + } + + void ServeFramesFromSource() + { + EXPECT_CALL(*mock_svt_, Receive()).WillRepeatedly([this]() { + return frame_source_.Get(); + }); + } + + /// @brief Enables dispatching and serves @p frame from the mocked receiver. + void StartServing(const std::optional& frame) + { + dispatcher_->Start(); + ServeFramesFromSource(); + frame_source_.Set(frame); + } + + /// @brief Returns the poll counter once it has stopped changing. + /// + /// Unset() does not synchronise with the worker loop: an iteration that already passed its + /// "callback registered?" check may still call Receive() once after Unset() returned. Sampling + /// the counter only after it held still for a few intervals excludes that in-flight poll. + std::size_t PollsOnceIdle() + { + auto polls = frame_source_.Polls(); + for (std::size_t attempt = 0U; attempt < 50U; ++attempt) + { + std::this_thread::sleep_for(5 * kPollInterval); + const auto now = frame_source_.Polls(); + if (now == polls) + { + break; + } + polls = now; + } + return polls; + } + + std::shared_ptr mock_svt_; + FrameSource frame_source_; + std::unique_ptr dispatcher_; +}; + +// --------------------------------------------------------------------------- +// Worker lifecycle +// --------------------------------------------------------------------------- + +TEST_F(SvtCallbackDispatcherTest, WorkerDoesNotPollWhileNoCallbackIsRegistered) +{ + dispatcher_->Start(); + EXPECT_CALL(*mock_svt_, Receive()).Times(0); + + std::this_thread::sleep_for(20 * kPollInterval); +} + +TEST_F(SvtCallbackDispatcherTest, WorkerDoesNotPollBeforeStartAndDeliversOnceStarted) +{ + ServeFramesFromSource(); + frame_source_.Set(MakeFrame(kSynchronizedStatus)); + + Recorder recorder; + dispatcher_->SetStatusChangedCallback(recorder.Callback()); + + std::this_thread::sleep_for(20 * kPollInterval); + EXPECT_EQ(frame_source_.Polls(), 0U); + + // Start() after registration must spin up the worker and deliver to the waiting callback. + dispatcher_->Start(); + ASSERT_TRUE(recorder.WaitForCount(1U)); +} + +TEST_F(SvtCallbackDispatcherTest, WorkerResumesPollingWhenCallbackReRegisteredAfterLastUnset) +{ + StartServing(MakeFrame(kSynchronizedStatus)); + + Recorder recorder; + dispatcher_->SetStatusChangedCallback(recorder.Callback()); + ASSERT_TRUE(recorder.WaitForCount(1U)); + + // Without any callback the worker must stop polling ... + dispatcher_->UnsetStatusChangedCallback(); + const auto polls_after_unset = PollsOnceIdle(); + std::this_thread::sleep_for(20 * kPollInterval); + ASSERT_EQ(frame_source_.Polls(), polls_after_unset); + + // ... and re-registering must make the (still running) worker resume. + dispatcher_->SetStatusChangedCallback(recorder.Callback()); + + ASSERT_TRUE(frame_source_.WaitForAdditionalPolls(1U)); + ASSERT_TRUE(recorder.WaitForCount(2U)); +} + +TEST_F(SvtCallbackDispatcherTest, SettingEmptyCallbackDoesNotInvokeAnything) +{ + StartServing(MakeFrame(kSynchronizedStatus)); + + dispatcher_->SetStatusChangedCallback(VehicleTime::StatusChangedCallback{}); + dispatcher_->SetTimeSlaveSyncDataReceivedCallback(VehicleTime::TimeSlaveSyncDataReceivedCallback{}); + dispatcher_->SetPDelayMeasurementFinishedCallback(VehicleTime::PDelayMeasurementFinishedCallback{}); + + // No callback is registered, so the worker must not poll at all. + std::this_thread::sleep_for(20 * kPollInterval); + EXPECT_EQ(frame_source_.Polls(), 0U); +} + +TEST_F(SvtCallbackDispatcherTest, NoCallbackIsDeliveredWhileReceiveReturnsNullopt) +{ + StartServing(std::nullopt); + + Recorder status_recorder; + Recorder> sync_recorder; + Recorder> pdelay_recorder; + dispatcher_->SetStatusChangedCallback(status_recorder.Callback()); + dispatcher_->SetTimeSlaveSyncDataReceivedCallback(sync_recorder.Callback()); + dispatcher_->SetPDelayMeasurementFinishedCallback(pdelay_recorder.Callback()); + + ASSERT_TRUE(frame_source_.WaitForAdditionalPolls(5U)); + EXPECT_EQ(status_recorder.Count(), 0U); + EXPECT_EQ(sync_recorder.Count(), 0U); + EXPECT_EQ(pdelay_recorder.Count(), 0U); +} + +TEST_F(SvtCallbackDispatcherTest, DestructorJoinsWorkerWhileCallbacksAreRegistered) +{ + StartServing(MakeFrame(kSynchronizedStatus)); + + Recorder recorder; + dispatcher_->SetStatusChangedCallback(recorder.Callback()); + ASSERT_TRUE(recorder.WaitForCount(1U)); + + dispatcher_.reset(); + + const auto polls_after_destruction = frame_source_.Polls(); + std::this_thread::sleep_for(20 * kPollInterval); + EXPECT_EQ(frame_source_.Polls(), polls_after_destruction); +} + +TEST_F(SvtCallbackDispatcherTest, DestructorJoinsIdleWorkerAfterLastCallbackUnset) +{ + StartServing(MakeFrame(kSynchronizedStatus)); + + Recorder recorder; + dispatcher_->SetStatusChangedCallback(recorder.Callback()); + ASSERT_TRUE(recorder.WaitForCount(1U)); + dispatcher_->UnsetStatusChangedCallback(); + + // The worker keeps running idle after the last unset; destruction must still stop and join it + // without hanging, and nothing may poll afterwards. + dispatcher_.reset(); + + const auto polls_after_destruction = frame_source_.Polls(); + std::this_thread::sleep_for(20 * kPollInterval); + EXPECT_EQ(frame_source_.Polls(), polls_after_destruction); +} + +// --------------------------------------------------------------------------- +// VehicleTimeStatus +// --------------------------------------------------------------------------- + +TEST_F(SvtCallbackDispatcherTest, StatusCallbackFiresOnFirstFrameAfterRegistration) +{ + StartServing(MakeFrame(kSynchronizedStatus, 2.5)); + + Recorder recorder; + dispatcher_->SetStatusChangedCallback(recorder.Callback()); + + ASSERT_TRUE(recorder.WaitForCount(1U)); + EXPECT_TRUE(recorder.Last().IsFlagActive(VehicleTime::StatusFlag::kSynchronized)); + EXPECT_FALSE(recorder.Last().IsFlagActive(VehicleTime::StatusFlag::kTimeOut)); + EXPECT_DOUBLE_EQ(recorder.Last().RateDeviation(), 2.5); +} + +TEST_F(SvtCallbackDispatcherTest, StatusCallbackFiresForInvalidFrameWithEmptyFlags) +{ + StartServing(MakeFrame(SvtStatus{true, false, false, false, false})); + + Recorder recorder; + dispatcher_->SetStatusChangedCallback(recorder.Callback()); + + ASSERT_TRUE(recorder.WaitForCount(1U)); + EXPECT_FALSE(recorder.Last().IsConsistent()); +} + +TEST_F(SvtCallbackDispatcherTest, StatusCallbackDoesNotRepeatWhileFlagsAreUnchanged) +{ + StartServing(MakeFrame(kSynchronizedStatus)); + + Recorder recorder; + dispatcher_->SetStatusChangedCallback(recorder.Callback()); + + ASSERT_TRUE(recorder.WaitForCount(1U)); + ASSERT_TRUE(frame_source_.WaitForAdditionalPolls(5U)); + EXPECT_EQ(recorder.Count(), 1U); +} + +TEST_F(SvtCallbackDispatcherTest, StatusCallbackFiresWhenFlagsChange) +{ + StartServing(MakeFrame(kSynchronizedStatus)); + + Recorder recorder; + dispatcher_->SetStatusChangedCallback(recorder.Callback()); + ASSERT_TRUE(recorder.WaitForCount(1U)); + + frame_source_.Set(MakeFrame(kTimeoutStatus)); + + ASSERT_TRUE(recorder.WaitForCount(2U)); + EXPECT_TRUE(recorder.Last().IsFlagActive(VehicleTime::StatusFlag::kTimeOut)); + EXPECT_TRUE(recorder.Last().IsFlagActive(VehicleTime::StatusFlag::kSynchronized)); +} + +TEST_F(SvtCallbackDispatcherTest, StatusCallbackIgnoresRateDeviationChanges) +{ + StartServing(MakeFrame(kSynchronizedStatus, 1.0)); + + Recorder recorder; + dispatcher_->SetStatusChangedCallback(recorder.Callback()); + ASSERT_TRUE(recorder.WaitForCount(1U)); + + frame_source_.Set(MakeFrame(kSynchronizedStatus, 9.0)); + + ASSERT_TRUE(frame_source_.WaitForAdditionalPolls(5U)); + EXPECT_EQ(recorder.Count(), 1U); + EXPECT_DOUBLE_EQ(recorder.Last().RateDeviation(), 1.0); +} + +TEST_F(SvtCallbackDispatcherTest, StatusCallbackReRegisteredReceivesUnchangedStatusAgain) +{ + StartServing(MakeFrame(kSynchronizedStatus)); + + Recorder first_recorder; + dispatcher_->SetStatusChangedCallback(first_recorder.Callback()); + ASSERT_TRUE(first_recorder.WaitForCount(1U)); + + dispatcher_->UnsetStatusChangedCallback(); + + Recorder second_recorder; + dispatcher_->SetStatusChangedCallback(second_recorder.Callback()); + + ASSERT_TRUE(second_recorder.WaitForCount(1U)); + EXPECT_TRUE(second_recorder.Last().IsFlagActive(VehicleTime::StatusFlag::kSynchronized)); + EXPECT_EQ(first_recorder.Count(), 1U); + + // Afterwards only changes are delivered. + ASSERT_TRUE(frame_source_.WaitForAdditionalPolls(5U)); + EXPECT_EQ(second_recorder.Count(), 1U); +} + +TEST_F(SvtCallbackDispatcherTest, StatusCallbackReplacedWithoutUnsetReceivesUnchangedStatusAgain) +{ + StartServing(MakeFrame(kSynchronizedStatus)); + + Recorder first_recorder; + dispatcher_->SetStatusChangedCallback(first_recorder.Callback()); + ASSERT_TRUE(first_recorder.WaitForCount(1U)); + + Recorder second_recorder; + dispatcher_->SetStatusChangedCallback(second_recorder.Callback()); + + ASSERT_TRUE(second_recorder.WaitForCount(1U)); + EXPECT_TRUE(second_recorder.Last().IsFlagActive(VehicleTime::StatusFlag::kSynchronized)); + EXPECT_EQ(first_recorder.Count(), 1U); +} + +TEST_F(SvtCallbackDispatcherTest, StatusCallbackRegisteredWhileWorkerIsActiveReceivesCurrentStatusFirst) +{ + StartServing(MakeFrame(kTimeoutStatus)); + + // Worker already running and has seen the timeout status through another callback. + dispatcher_->SetPDelayMeasurementFinishedCallback([](const PDelayMeasurementData&) {}); + ASSERT_TRUE(frame_source_.WaitForAdditionalPolls(5U)); + + Recorder recorder; + dispatcher_->SetStatusChangedCallback(recorder.Callback()); + + ASSERT_TRUE(recorder.WaitForCount(1U)); + EXPECT_TRUE(recorder.Last().IsFlagActive(VehicleTime::StatusFlag::kTimeOut)); + + frame_source_.Set(MakeFrame(kSynchronizedStatus)); + ASSERT_TRUE(recorder.WaitForCount(2U)); + EXPECT_FALSE(recorder.Last().IsFlagActive(VehicleTime::StatusFlag::kTimeOut)); +} + +TEST_F(SvtCallbackDispatcherTest, UnsetStatusCallbackStopsDelivery) +{ + StartServing(MakeFrame(kSynchronizedStatus)); + + Recorder recorder; + dispatcher_->SetStatusChangedCallback(recorder.Callback()); + ASSERT_TRUE(recorder.WaitForCount(1U)); + + dispatcher_->UnsetStatusChangedCallback(); + frame_source_.Set(MakeFrame(kTimeoutStatus)); + + // Keep the worker polling through another registered callback and verify status stays silent. + dispatcher_->SetPDelayMeasurementFinishedCallback([](const PDelayMeasurementData&) {}); + ASSERT_TRUE(frame_source_.WaitForAdditionalPolls(5U)); + EXPECT_EQ(recorder.Count(), 1U); +} + +// --------------------------------------------------------------------------- +// TimeSlaveSyncData +// --------------------------------------------------------------------------- + +TEST_F(SvtCallbackDispatcherTest, SyncDataCallbackFiresOnEachNewFrame) +{ + auto frame = MakeFrame(kSynchronizedStatus); + frame.sync_fup_data.sequence_id = 1U; + StartServing(frame); + + Recorder> recorder; + dispatcher_->SetTimeSlaveSyncDataReceivedCallback(recorder.Callback()); + ASSERT_TRUE(recorder.WaitForCount(1U)); + EXPECT_EQ(recorder.Last().sequence_id, 1U); + + frame.sync_fup_data.sequence_id = 2U; + frame_source_.Set(frame); + ASSERT_TRUE(recorder.WaitForCount(2U)); + EXPECT_EQ(recorder.Last().sequence_id, 2U); + + // Unchanged frame: no further delivery. + ASSERT_TRUE(frame_source_.WaitForAdditionalPolls(5U)); + EXPECT_EQ(recorder.Count(), 2U); + + frame.sync_fup_data.sequence_id = 3U; + frame_source_.Set(frame); + ASSERT_TRUE(recorder.WaitForCount(3U)); + EXPECT_EQ(recorder.Last().sequence_id, 3U); +} + +TEST_F(SvtCallbackDispatcherTest, SyncDataCallbackReceivesConvertedFields) +{ + StartServing(MakeFrame(kSynchronizedStatus)); + + Recorder> recorder; + dispatcher_->SetTimeSlaveSyncDataReceivedCallback(recorder.Callback()); + ASSERT_TRUE(recorder.WaitForCount(1U)); + + auto frame = MakeFrame(kSynchronizedStatus); + frame.sync_fup_data = SvtSyncData{101ULL, 202ULL, 303ULL, 404ULL, 0x10000ULL, 55U, 606ULL, 7U, 0xABCDULL}; + frame_source_.Set(frame); + + ASSERT_TRUE(recorder.WaitForCount(2U)); + const auto data = recorder.Last(); + EXPECT_EQ(data.precise_origin_timestamp.time_since_epoch(), 101ns); + EXPECT_EQ(data.reference_global_timestamp.time_since_epoch(), 202ns); + EXPECT_EQ(data.reference_local_timestamp.time_since_epoch(), 303ns); + EXPECT_EQ(data.sync_ingress_timestamp.time_since_epoch(), 404ns); + EXPECT_EQ(data.correction_field, 0x10000LL); + EXPECT_EQ(data.sequence_id, 55U); + EXPECT_EQ(data.pdelay, 606ns); + EXPECT_EQ(data.source_port_identity.port_number, 7U); + EXPECT_EQ(data.source_port_identity.clock_identity, 0xABCDULL); +} + +// --------------------------------------------------------------------------- +// PDelayMeasurementData +// --------------------------------------------------------------------------- + +TEST_F(SvtCallbackDispatcherTest, PDelayCallbackFiresOnEachNewFrame) +{ + auto frame = MakeFrame(kSynchronizedStatus); + frame.pdelay_data.sequence_id = 1U; + StartServing(frame); + + Recorder> recorder; + dispatcher_->SetPDelayMeasurementFinishedCallback(recorder.Callback()); + ASSERT_TRUE(recorder.WaitForCount(1U)); + EXPECT_EQ(recorder.Last().sequence_id, 1U); + + frame.pdelay_data.sequence_id = 2U; + frame_source_.Set(frame); + ASSERT_TRUE(recorder.WaitForCount(2U)); + EXPECT_EQ(recorder.Last().sequence_id, 2U); + + // Unchanged frame: no further delivery. + ASSERT_TRUE(frame_source_.WaitForAdditionalPolls(5U)); + EXPECT_EQ(recorder.Count(), 2U); + + frame.pdelay_data.sequence_id = 3U; + frame_source_.Set(frame); + ASSERT_TRUE(recorder.WaitForCount(3U)); + EXPECT_EQ(recorder.Last().sequence_id, 3U); +} + +TEST_F(SvtCallbackDispatcherTest, PDelayCallbackReceivesConvertedFields) +{ + StartServing(MakeFrame(kSynchronizedStatus)); + + Recorder> recorder; + dispatcher_->SetPDelayMeasurementFinishedCallback(recorder.Callback()); + ASSERT_TRUE(recorder.WaitForCount(1U)); + + auto frame = MakeFrame(kSynchronizedStatus); + frame.pdelay_data = + SvtPDelayData{11ULL, 22ULL, 33ULL, 44ULL, 55ULL, 66ULL, 77U, 88ULL, 3U, 0x1111ULL, 4U, 0x2222ULL}; + frame_source_.Set(frame); + + ASSERT_TRUE(recorder.WaitForCount(2U)); + const auto data = recorder.Last(); + EXPECT_EQ(data.request_origin_timestamp.time_since_epoch(), 11ns); + EXPECT_EQ(data.request_receipt_timestamp.time_since_epoch(), 22ns); + EXPECT_EQ(data.response_origin_timestamp.time_since_epoch(), 33ns); + EXPECT_EQ(data.response_receipt_timestamp.time_since_epoch(), 44ns); + EXPECT_EQ(data.reference_global_timestamp.time_since_epoch(), 55ns); + EXPECT_EQ(data.reference_local_timestamp.time_since_epoch(), 66ns); + EXPECT_EQ(data.sequence_id, 77U); + EXPECT_EQ(data.pdelay, 88ns); + EXPECT_EQ(data.request_port_identity.port_number, 3U); + EXPECT_EQ(data.request_port_identity.clock_identity, 0x1111ULL); + EXPECT_EQ(data.response_port_identity.port_number, 4U); + EXPECT_EQ(data.response_port_identity.clock_identity, 0x2222ULL); +} + +// --------------------------------------------------------------------------- +// Multiple callbacks and thread safety of Set / Unset +// --------------------------------------------------------------------------- + +TEST_F(SvtCallbackDispatcherTest, AllThreeCallbacksAreDeliveredFromTheSameFrame) +{ + auto frame = MakeFrame(kSynchronizedStatus); + StartServing(frame); + + Recorder status_recorder; + Recorder> sync_recorder; + Recorder> pdelay_recorder; + dispatcher_->SetTimeSlaveSyncDataReceivedCallback(sync_recorder.Callback()); + dispatcher_->SetPDelayMeasurementFinishedCallback(pdelay_recorder.Callback()); + dispatcher_->SetStatusChangedCallback(status_recorder.Callback()); + ASSERT_TRUE(status_recorder.WaitForCount(1U)); + ASSERT_TRUE(sync_recorder.WaitForCount(1U)); + ASSERT_TRUE(pdelay_recorder.WaitForCount(1U)); + + frame.status = kTimeoutStatus; + frame.sync_fup_data.sequence_id = 5U; + frame.pdelay_data.sequence_id = 6U; + frame_source_.Set(frame); + + ASSERT_TRUE(status_recorder.WaitForCount(2U)); + ASSERT_TRUE(sync_recorder.WaitForCount(2U)); + ASSERT_TRUE(pdelay_recorder.WaitForCount(2U)); + EXPECT_TRUE(status_recorder.Last().IsFlagActive(VehicleTime::StatusFlag::kTimeOut)); + EXPECT_EQ(sync_recorder.Last().sequence_id, 5U); + EXPECT_EQ(pdelay_recorder.Last().sequence_id, 6U); +} + +TEST_F(SvtCallbackDispatcherTest, ChangeInOneFramePartFiresOnlyThatCallback) +{ + auto frame = MakeFrame(kSynchronizedStatus); + StartServing(frame); + + Recorder status_recorder; + Recorder> sync_recorder; + Recorder> pdelay_recorder; + dispatcher_->SetTimeSlaveSyncDataReceivedCallback(sync_recorder.Callback()); + dispatcher_->SetPDelayMeasurementFinishedCallback(pdelay_recorder.Callback()); + dispatcher_->SetStatusChangedCallback(status_recorder.Callback()); + ASSERT_TRUE(status_recorder.WaitForCount(1U)); + ASSERT_TRUE(sync_recorder.WaitForCount(1U)); + ASSERT_TRUE(pdelay_recorder.WaitForCount(1U)); + + // Only the sync part changes: sync fires, the other two stay silent. + frame.sync_fup_data.sequence_id = 5U; + frame_source_.Set(frame); + ASSERT_TRUE(sync_recorder.WaitForCount(2U)); + ASSERT_TRUE(frame_source_.WaitForAdditionalPolls(5U)); + EXPECT_EQ(status_recorder.Count(), 1U); + EXPECT_EQ(pdelay_recorder.Count(), 1U); + + // Only the pDelay part changes: pDelay fires, the other two stay silent. + frame.pdelay_data.sequence_id = 6U; + frame_source_.Set(frame); + ASSERT_TRUE(pdelay_recorder.WaitForCount(2U)); + ASSERT_TRUE(frame_source_.WaitForAdditionalPolls(5U)); + EXPECT_EQ(status_recorder.Count(), 1U); + EXPECT_EQ(sync_recorder.Count(), 2U); +} + +TEST_F(SvtCallbackDispatcherTest, UnsetFromWithinCallbackDoesNotDeadlock) +{ + StartServing(MakeFrame(kSynchronizedStatus)); + + Recorder recorder; + dispatcher_->SetStatusChangedCallback([this, &recorder](const VehicleTimeStatus& status) { + recorder.Record(status); + dispatcher_->UnsetStatusChangedCallback(); + }); + ASSERT_TRUE(recorder.WaitForCount(1U)); + + // Keep the worker polling through another registered callback; the status callback must stay silent. + dispatcher_->SetPDelayMeasurementFinishedCallback([](const PDelayMeasurementData&) {}); + frame_source_.Set(MakeFrame(kTimeoutStatus)); + + ASSERT_TRUE(frame_source_.WaitForAdditionalPolls(5U)); + EXPECT_EQ(recorder.Count(), 1U); +} + +TEST_F(SvtCallbackDispatcherTest, SubscribingToAnotherEventFromWithinCallbackDoesNotDeadlock) +{ + auto frame = MakeFrame(kSynchronizedStatus); + StartServing(frame); + + Recorder status_recorder; + Recorder> sync_recorder; + dispatcher_->SetStatusChangedCallback([this, &status_recorder, &sync_recorder](const VehicleTimeStatus& status) { + status_recorder.Record(status); + dispatcher_->SetTimeSlaveSyncDataReceivedCallback(sync_recorder.Callback()); + }); + ASSERT_TRUE(status_recorder.WaitForCount(1U)); + ASSERT_TRUE(sync_recorder.WaitForCount(1U)); + + frame.sync_fup_data.sequence_id = 42U; + frame_source_.Set(frame); + + ASSERT_TRUE(sync_recorder.WaitForCount(2U)); + EXPECT_EQ(sync_recorder.Last().sequence_id, 42U); +} + +TEST_F(SvtCallbackDispatcherTest, UnsetBlocksUntilInFlightCallbackReturns) +{ + StartServing(MakeFrame(kSynchronizedStatus)); + + std::promise callback_entered; + std::promise release_callback; + auto release_future = release_callback.get_future().share(); + dispatcher_->SetStatusChangedCallback([&callback_entered, release_future](const VehicleTimeStatus&) { + callback_entered.set_value(); + release_future.wait(); + }); + ASSERT_EQ(callback_entered.get_future().wait_for(kWaitTimeout), std::future_status::ready); + + auto unset_done = std::async(std::launch::async, [this]() { + dispatcher_->UnsetStatusChangedCallback(); + }); + EXPECT_EQ(unset_done.wait_for(50 * kPollInterval), std::future_status::timeout); + + release_callback.set_value(); + EXPECT_EQ(unset_done.wait_for(kWaitTimeout), std::future_status::ready); +} + +} // namespace +} // namespace time +} // namespace score diff --git a/score/time/vehicle_time/src/details/td_impl/svt_callback_wrapper.h b/score/time/vehicle_time/src/details/td_impl/svt_callback_wrapper.h new file mode 100644 index 00000000..58cc69cb --- /dev/null +++ b/score/time/vehicle_time/src/details/td_impl/svt_callback_wrapper.h @@ -0,0 +1,118 @@ +/******************************************************************************** + * Copyright (c) 2026 Contributors to the Eclipse Foundation + * + * See the NOTICE file(s) distributed with this work for additional + * information regarding copyright ownership. + * + * This program and the accompanying materials are made available under the + * terms of the Apache License Version 2.0 which is available at + * https://www.apache.org/licenses/LICENSE-2.0 + * + * SPDX-License-Identifier: Apache-2.0 + ********************************************************************************/ +#ifndef SCORE_TIME_VEHICLE_TIME_SRC_DETAILS_TD_IMPL_SVT_CALLBACK_WRAPPER_H +#define SCORE_TIME_VEHICLE_TIME_SRC_DETAILS_TD_IMPL_SVT_CALLBACK_WRAPPER_H + +// Internal header — include ONLY from translation units under vehicle_time/src/details/td_impl/. +// NOT part of the public API of td_impl. + +#include +#include +#include +#include + +namespace score +{ +namespace time +{ +namespace detail +{ + +/// @brief Thread-safe holder for a single move-only callback that is invoked from a dedicated worker thread +/// whenever the observed value changes. +/// +/// The slot remembers the @p Data of the last value delivered to the current callback, so +/// @c InvokeIfChanged() delivers only on change. @c Set() forgets that key together with the old +/// callback: the first @c InvokeIfChanged() after any (re-)registration therefore always delivers. +/// +/// Guarantees: +/// - @c Set() / @c Unset() may be called from any thread at any time. +/// - @c InvokeIfChanged() must be called from a single worker thread only. The callback runs while +/// the slot's recursive mutex is held, so: +/// - @c Set() / @c Unset() from another thread block until the in-flight invocation has returned. +/// Once they return, the previously stored callback is neither running nor will it ever be +/// invoked again — the caller may safely destroy whatever the callback referenced. +/// - @c Set() / @c Unset() called re-entrantly from inside the callback take effect immediately; +/// the running invocation completes normally on a shared handle that outlives the slot contents. +/// +/// @tparam Callback A callable wrapper offering @c empty() and @c operator() (e.g. @c score::cpp::callback). +/// @tparam Data Equality-comparable, copyable type identifying the value last delivered. +template +class SvtCallbackWrapper final +{ + public: + SvtCallbackWrapper() noexcept = default; + ~SvtCallbackWrapper() noexcept = default; + SvtCallbackWrapper(const SvtCallbackWrapper&) = delete; + SvtCallbackWrapper& operator=(const SvtCallbackWrapper&) = delete; + SvtCallbackWrapper(SvtCallbackWrapper&&) = delete; + SvtCallbackWrapper& operator=(SvtCallbackWrapper&&) = delete; + + /// @brief Installs @p callback, replacing any previous one. An empty callback behaves like @c Unset(). + /// + /// Forgets the last delivered key, so the next @c InvokeIfChanged() delivers unconditionally. + void Set(Callback&& callback) noexcept + { + const std::lock_guard lock{mutex_}; + callback_ = callback.empty() ? nullptr : std::make_shared(std::move(callback)); + last_data_.reset(); + } + + /// @brief Removes the stored callback. + void Unset() noexcept + { + const std::lock_guard lock{mutex_}; + callback_.reset(); + last_data_.reset(); + } + + /// @brief Returns @c true if a callback is currently installed. + bool IsSet() const noexcept + { + const std::lock_guard lock{mutex_}; + return callback_ != nullptr; + } + + /// @brief Invokes the stored callback with @p argument unless @p data equals the data of the + /// previous delivery to the same callback. + /// + /// Must be called from the worker thread only. + /// + /// @return @c true if the callback was invoked, @c false if none is installed or @p data is unchanged. + template + bool InvokeIfChanged(const Data& data, const Argument& argument) noexcept + { + const std::lock_guard lock{mutex_}; + if ((callback_ == nullptr) || (last_data_.has_value() && (last_data_.value() == data))) + { + return false; + } + last_data_ = data; + + // Local copy keeps the callback alive should it Unset() or replace itself while running. + const std::shared_ptr callback = callback_; + (*callback)(argument); + return true; + } + + private: + mutable std::recursive_mutex mutex_; + std::shared_ptr callback_{}; + std::optional last_data_{}; +}; + +} // namespace detail +} // namespace time +} // namespace score + +#endif // SCORE_TIME_VEHICLE_TIME_SRC_DETAILS_TD_IMPL_SVT_CALLBACK_WRAPPER_H diff --git a/score/time/vehicle_time/src/details/td_impl/svt_callback_wrapper_test.cpp b/score/time/vehicle_time/src/details/td_impl/svt_callback_wrapper_test.cpp new file mode 100644 index 00000000..d9744c46 --- /dev/null +++ b/score/time/vehicle_time/src/details/td_impl/svt_callback_wrapper_test.cpp @@ -0,0 +1,211 @@ +/******************************************************************************** + * Copyright (c) 2026 Contributors to the Eclipse Foundation + * + * See the NOTICE file(s) distributed with this work for additional + * information regarding copyright ownership. + * + * This program and the accompanying materials are made available under the + * terms of the Apache License Version 2.0 which is available at + * https://www.apache.org/licenses/LICENSE-2.0 + * + * SPDX-License-Identifier: Apache-2.0 + ********************************************************************************/ +#include "score/time/vehicle_time/src/details/td_impl/svt_callback_wrapper.h" + +#include + +#include + +#include +#include +#include +#include +#include + +namespace score +{ +namespace time +{ +namespace detail +{ +namespace +{ + +using TestCallback = score::cpp::callback; +using Slot = SvtCallbackWrapper; + +TEST(SvtCallbackWrapperTest, InvokeIfChangedReturnsFalseWhenNoCallbackIsSet) +{ + Slot slot; + EXPECT_FALSE(slot.IsSet()); + EXPECT_FALSE(slot.InvokeIfChanged(1, 1)); +} + +TEST(SvtCallbackWrapperTest, InvokeIfChangedCallsStoredCallbackWithArgumentOnEveryNewKey) +{ + Slot slot; + std::vector received; + slot.Set([&received](const int& value) { + received.push_back(value); + }); + + EXPECT_TRUE(slot.IsSet()); + EXPECT_TRUE(slot.InvokeIfChanged(7, 70)); + EXPECT_TRUE(slot.InvokeIfChanged(8, 80)); + EXPECT_EQ(received, (std::vector{70, 80})); +} + +TEST(SvtCallbackWrapperTest, InvokeIfChangedSkipsRepeatedKey) +{ + Slot slot; + int invocations{0}; + slot.Set([&invocations](const int&) { + ++invocations; + }); + + EXPECT_TRUE(slot.InvokeIfChanged(7, 7)); + EXPECT_FALSE(slot.InvokeIfChanged(7, 7)); + EXPECT_FALSE(slot.InvokeIfChanged(7, 8)); // argument differs but key does not + EXPECT_TRUE(slot.InvokeIfChanged(9, 9)); + EXPECT_EQ(invocations, 2); +} + +TEST(SvtCallbackWrapperTest, SetForgetsLastKeySoNewCallbackIsInvokedWithUnchangedKey) +{ + Slot slot; + slot.Set([](const int&) {}); + EXPECT_TRUE(slot.InvokeIfChanged(7, 7)); + EXPECT_FALSE(slot.InvokeIfChanged(7, 7)); + + int replacement_invocations{0}; + slot.Set([&replacement_invocations](const int&) { + ++replacement_invocations; + }); + EXPECT_TRUE(slot.InvokeIfChanged(7, 7)); + EXPECT_EQ(replacement_invocations, 1); +} + +TEST(SvtCallbackWrapperTest, UnsetRemovesCallbackAndForgetsLastKey) +{ + Slot slot; + int invocations{0}; + slot.Set([&invocations](const int&) { + ++invocations; + }); + EXPECT_TRUE(slot.InvokeIfChanged(7, 7)); + slot.Unset(); + + EXPECT_FALSE(slot.IsSet()); + EXPECT_FALSE(slot.InvokeIfChanged(7, 7)); + + slot.Set([&invocations](const int&) { + ++invocations; + }); + EXPECT_TRUE(slot.InvokeIfChanged(7, 7)); + EXPECT_EQ(invocations, 2); +} + +TEST(SvtCallbackWrapperTest, SettingEmptyCallbackBehavesLikeUnset) +{ + Slot slot; + slot.Set([](const int&) {}); + slot.Set(TestCallback{}); + + EXPECT_FALSE(slot.IsSet()); + EXPECT_FALSE(slot.InvokeIfChanged(0, 0)); +} + +TEST(SvtCallbackWrapperTest, UnsetFromWithinCallbackDoesNotDeadlockAndTakesEffectAfterwards) +{ + Slot slot; + int invocations{0}; + slot.Set([&slot, &invocations](const int&) { + ++invocations; + slot.Unset(); + }); + + EXPECT_TRUE(slot.InvokeIfChanged(0, 0)); + EXPECT_FALSE(slot.IsSet()); + EXPECT_FALSE(slot.InvokeIfChanged(0, 0)); + EXPECT_EQ(invocations, 1); +} + +TEST(SvtCallbackWrapperTest, SetFromWithinCallbackReplacesCallbackForNextInvocation) +{ + Slot slot; + std::vector trace; + slot.Set([&slot, &trace](const int&) { + trace.push_back(1); + slot.Set([&trace](const int&) { + trace.push_back(2); + }); + }); + + EXPECT_TRUE(slot.InvokeIfChanged(0, 0)); + EXPECT_TRUE(slot.InvokeIfChanged(0, 0)); + EXPECT_EQ(trace, (std::vector{1, 2})); +} + +TEST(SvtCallbackWrapperTest, UnsetFromAnotherThreadBlocksUntilInFlightInvocationReturns) +{ + Slot slot; + std::promise callback_entered; + std::promise release_callback; + auto release_future = release_callback.get_future().share(); + slot.Set([&callback_entered, release_future](const int&) { + callback_entered.set_value(); + release_future.wait(); + }); + + std::thread invoker{[&slot]() { + std::ignore = slot.InvokeIfChanged(0, 0); + }}; + callback_entered.get_future().wait(); + + auto unset_done = std::async(std::launch::async, [&slot]() { + slot.Unset(); + }); + EXPECT_EQ(unset_done.wait_for(std::chrono::milliseconds{50}), std::future_status::timeout); + + release_callback.set_value(); + EXPECT_EQ(unset_done.wait_for(std::chrono::seconds{5}), std::future_status::ready); + invoker.join(); + EXPECT_FALSE(slot.IsSet()); +} + +TEST(SvtCallbackWrapperTest, SetFromAnotherThreadBlocksUntilInFlightInvocationReturns) +{ + Slot slot; + std::promise callback_entered; + std::promise release_callback; + auto release_future = release_callback.get_future().share(); + slot.Set([&callback_entered, release_future](const int&) { + callback_entered.set_value(); + release_future.wait(); + }); + + std::thread invoker{[&slot]() { + std::ignore = slot.InvokeIfChanged(0, 0); + }}; + callback_entered.get_future().wait(); + + int replacement_invocations{0}; + auto set_done = std::async(std::launch::async, [&slot, &replacement_invocations]() { + slot.Set([&replacement_invocations](const int&) { + ++replacement_invocations; + }); + }); + EXPECT_EQ(set_done.wait_for(std::chrono::milliseconds{50}), std::future_status::timeout); + + release_callback.set_value(); + EXPECT_EQ(set_done.wait_for(std::chrono::seconds{5}), std::future_status::ready); + invoker.join(); + + EXPECT_TRUE(slot.InvokeIfChanged(0, 0)); + EXPECT_EQ(replacement_invocations, 1); +} + +} // namespace +} // namespace detail +} // namespace time +} // namespace score diff --git a/score/time/vehicle_time/src/details/td_impl/svt_test_helpers.h b/score/time/vehicle_time/src/details/td_impl/svt_test_helpers.h new file mode 100644 index 00000000..d1c7290f --- /dev/null +++ b/score/time/vehicle_time/src/details/td_impl/svt_test_helpers.h @@ -0,0 +1,145 @@ +/******************************************************************************** + * Copyright (c) 2026 Contributors to the Eclipse Foundation + * + * See the NOTICE file(s) distributed with this work for additional + * information regarding copyright ownership. + * + * This program and the accompanying materials are made available under the + * terms of the Apache License Version 2.0 which is available at + * https://www.apache.org/licenses/LICENSE-2.0 + * + * SPDX-License-Identifier: Apache-2.0 + ********************************************************************************/ +#ifndef SCORE_TIME_VEHICLE_TIME_SRC_DETAILS_TD_IMPL_SVT_TEST_HELPERS_H +#define SCORE_TIME_VEHICLE_TIME_SRC_DETAILS_TD_IMPL_SVT_TEST_HELPERS_H + +// Test-only helpers shared between the SvtCallbackDispatcher and VehicleClockBackendImpl unit tests. + +#include "score/time_daemon/src/ipc/receiver_mock.h" +#include "score/time_daemon/src/ipc/svt/svt_time_info.h" + +#include +#include +#include +#include +#include + +namespace score +{ +namespace time +{ +namespace test_helpers +{ + +using SvtMock = score::td::ReceiverMock; +using SvtSnapshot = score::td::svt::TimeBaseSnapshot; +using SvtStatus = score::td::svt::TimeBaseStatus; +using SvtSyncData = score::td::svt::SyncFupSnapshot; +using SvtPDelayData = score::td::svt::PDelayDataSnapshot; + +constexpr SvtStatus kSynchronizedStatus{true, false, false, false, true}; +constexpr SvtStatus kTimeoutStatus{true, true, false, false, true}; +constexpr std::chrono::milliseconds kPollInterval{1}; +constexpr std::chrono::seconds kWaitTimeout{5}; + +inline SvtSnapshot MakeFrame(const SvtStatus status, const double rate_deviation = 0.0) noexcept +{ + return SvtSnapshot{1000ULL, 0ULL, rate_deviation, status, {}, {}}; +} + +/// @brief Thread-safe frame supplier for the mocked receiver; counts how often the worker polled. +class FrameSource +{ + public: + void Set(const std::optional& frame) noexcept + { + const std::lock_guard lock{mutex_}; + frame_ = frame; + } + + std::optional Get() noexcept + { + // Notify under the lock so the object can be destroyed as soon as a waiter resumes. + const std::lock_guard lock{mutex_}; + ++polls_; + polled_.notify_all(); + return frame_; + } + + std::size_t Polls() const noexcept + { + const std::lock_guard lock{mutex_}; + return polls_; + } + + /// @brief Blocks until the worker polled at least @p additional more times than right now. + bool WaitForAdditionalPolls(const std::size_t additional) noexcept + { + std::unique_lock lock{mutex_}; + const std::size_t target = polls_ + additional; + return polled_.wait_for(lock, kWaitTimeout, [this, target]() noexcept { + return polls_ >= target; + }); + } + + private: + mutable std::mutex mutex_; + std::condition_variable polled_; + std::optional frame_{}; + std::size_t polls_{0U}; +}; + +/// @brief Records callback invocations so the test thread can wait for them. +template +class Recorder +{ + public: + void Record(const Event& event) noexcept + { + // Notify under the lock so the recorder can be destroyed as soon as a waiter resumes. + const std::lock_guard lock{mutex_}; + last_ = event; + ++count_; + recorded_.notify_all(); + } + + /// @brief Returns a callable that records into this recorder, ready to pass to a Set*Callback(). + auto Callback() noexcept + { + return [this](const Event& event) noexcept { + Record(event); + }; + } + + bool WaitForCount(const std::size_t count) noexcept + { + std::unique_lock lock{mutex_}; + return recorded_.wait_for(lock, kWaitTimeout, [this, count]() noexcept { + return count_ >= count; + }); + } + + std::size_t Count() const noexcept + { + const std::lock_guard lock{mutex_}; + return count_; + } + + Event Last() const noexcept + { + const std::lock_guard lock{mutex_}; + return last_; + } + + private: + mutable std::mutex mutex_; + std::condition_variable recorded_; + Event last_{}; + std::size_t count_{0U}; +}; + +} // namespace test_helpers +} // namespace time +} // namespace score + +#endif // SCORE_TIME_VEHICLE_TIME_SRC_DETAILS_TD_IMPL_SVT_TEST_HELPERS_H diff --git a/score/time/vehicle_time/src/details/td_impl/vehicle_clock_backend_impl.cpp b/score/time/vehicle_time/src/details/td_impl/vehicle_clock_backend_impl.cpp index 42731db4..cac97ae1 100644 --- a/score/time/vehicle_time/src/details/td_impl/vehicle_clock_backend_impl.cpp +++ b/score/time/vehicle_time/src/details/td_impl/vehicle_clock_backend_impl.cpp @@ -20,6 +20,7 @@ #include #include +#include namespace score { @@ -29,8 +30,13 @@ namespace detail { VehicleClockBackendImpl::VehicleClockBackendImpl(std::shared_ptr receiver, - HighResSteadyClock local_clock) noexcept - : is_ready_{false}, init_mutex_{}, svt_receiver_{std::move(receiver)}, local_clock_{std::move(local_clock)} + HighResSteadyClock local_clock, + const std::chrono::milliseconds poll_interval) noexcept + : is_ready_{false}, + init_mutex_{}, + svt_receiver_{std::move(receiver)}, + local_clock_{std::move(local_clock)}, + dispatcher_{svt_receiver_, poll_interval} { } @@ -87,11 +93,14 @@ bool VehicleClockBackendImpl::Init() noexcept { score::mw::log::LogError(kVehicleTimeLogContext) << "VehicleClockBackendImpl: failed to open TimeDaemon shared memory segment."; + return false; } - is_ready_.store(ok, std::memory_order_release); + // The worker only ever reads from the receiver, so it must not run before the receiver is initialised. + dispatcher_.Start(); + is_ready_.store(true, std::memory_order_release); - return ok; + return true; } bool VehicleClockBackendImpl::IsAvailable() const noexcept @@ -120,63 +129,35 @@ bool VehicleClockBackendImpl::WaitUntilAvailable(const score::cpp::stop_token& t } void VehicleClockBackendImpl::SetTimeSlaveSyncDataReceivedCallback( - VehicleTime::TimeSlaveSyncDataReceivedCallback&& /*callback*/) noexcept + VehicleTime::TimeSlaveSyncDataReceivedCallback&& callback) noexcept { - // TODO(https://github.com/eclipse-score/inc_time/issues/59): implement callback delivery. + dispatcher_.SetTimeSlaveSyncDataReceivedCallback(std::move(callback)); } void VehicleClockBackendImpl::UnsetTimeSlaveSyncDataReceivedCallback() noexcept { - // TODO(https://github.com/eclipse-score/inc_time/issues/59): implement callback delivery. + dispatcher_.UnsetTimeSlaveSyncDataReceivedCallback(); } void VehicleClockBackendImpl::SetPDelayMeasurementFinishedCallback( - VehicleTime::PDelayMeasurementFinishedCallback&& /*callback*/) noexcept + VehicleTime::PDelayMeasurementFinishedCallback&& callback) noexcept { - // TODO(https://github.com/eclipse-score/inc_time/issues/59): implement callback delivery. + dispatcher_.SetPDelayMeasurementFinishedCallback(std::move(callback)); } void VehicleClockBackendImpl::UnsetPDelayMeasurementFinishedCallback() noexcept { - // TODO(https://github.com/eclipse-score/inc_time/issues/59): implement callback delivery. + dispatcher_.UnsetPDelayMeasurementFinishedCallback(); } -void VehicleClockBackendImpl::SetStatusChangedCallback(VehicleTime::StatusChangedCallback&& /*callback*/) noexcept +void VehicleClockBackendImpl::SetStatusChangedCallback(VehicleTime::StatusChangedCallback&& callback) noexcept { - // TODO(https://github.com/eclipse-score/inc_time/issues/59): implement callback delivery. + dispatcher_.SetStatusChangedCallback(std::move(callback)); } void VehicleClockBackendImpl::UnsetStatusChangedCallback() noexcept { - // TODO(https://github.com/eclipse-score/inc_time/issues/59): implement callback delivery. -} - -ClockStatus VehicleClockBackendImpl::ConvertPtpStatus( - const score::td::svt::TimeBaseStatus& ptp_status) noexcept -{ - using Flag = VehicleTime::StatusFlag; - if (!ptp_status.is_correct) - { - return ClockStatus{}; - } - ClockStatus status; - if (ptp_status.is_synchronized) - { - status.AddFlag(Flag::kSynchronized); - } - if (ptp_status.is_timeout) - { - status.AddFlag(Flag::kTimeOut); - } - if (ptp_status.is_time_jump_future) - { - status.AddFlag(Flag::kTimeLeapFuture); - } - if (ptp_status.is_time_jump_past) - { - status.AddFlag(Flag::kTimeLeapPast); - } - return status; + dispatcher_.UnsetStatusChangedCallback(); } } // namespace detail diff --git a/score/time/vehicle_time/src/details/td_impl/vehicle_clock_backend_impl.h b/score/time/vehicle_time/src/details/td_impl/vehicle_clock_backend_impl.h index 15dd77ae..0f1ddd07 100644 --- a/score/time/vehicle_time/src/details/td_impl/vehicle_clock_backend_impl.h +++ b/score/time/vehicle_time/src/details/td_impl/vehicle_clock_backend_impl.h @@ -17,6 +17,7 @@ // NOT part of the public API of td_impl. #include "score/time/high_res_steady_time/src/high_res_steady_clock.h" +#include "score/time/vehicle_time/src/details/td_impl/svt_callback_dispatcher.h" #include "score/time/vehicle_time/src/vehicle_clock.h" #include "score/time/vehicle_time/src/vehicle_clock_backend.h" #include "score/time_daemon/src/ipc/svt/receiver/svt_receiver.h" @@ -47,8 +48,11 @@ namespace detail /// where the local reference clock is supplied via @c HighResSteadyClock::GetInstance() /// (captured once at construction to avoid per-call mutex overhead). /// -/// Callback registration is not yet supported; all Set/Unset methods are no-ops -/// until the TimeDaemon IPC layer provides a subscription facility. +/// @par Callback delivery +/// Subscription callbacks are delivered by an owned @c SvtCallbackDispatcher, which runs a dedicated +/// worker thread polling the receiver. Dispatching is enabled by the first successful @c Init(); the +/// worker thread starts once dispatching is enabled and the first callback is registered, and runs +/// until this backend is destroyed. /// /// @note Placed in @c score::time::detail (rather than an anonymous namespace) so /// that vehicle_clock_backend_impl_test.cpp can construct it directly with injected mocks. @@ -56,11 +60,17 @@ namespace detail class VehicleClockBackendImpl final : public VehicleClockBackend { public: + /// @brief Default interval at which the worker thread polls the receiver for new frames. + static constexpr std::chrono::milliseconds kDefaultPollInterval{50}; + /// @brief Constructs backend with injected TimeDaemon receiver and local steady clock. /// - /// @param receiver SvtReceiver instance used to read TimeDaemon SVT samples. - /// @param local_clock Local steady clock used for PTP-to-now extrapolation. - VehicleClockBackendImpl(std::shared_ptr receiver, HighResSteadyClock local_clock) noexcept; + /// @param receiver SvtReceiver instance used to read TimeDaemon SVT samples. + /// @param local_clock Local steady clock used for PTP-to-now extrapolation. + /// @param poll_interval Interval at which the worker thread polls the receiver for new frames. + VehicleClockBackendImpl(std::shared_ptr receiver, + HighResSteadyClock local_clock, + std::chrono::milliseconds poll_interval = kDefaultPollInterval) noexcept; ~VehicleClockBackendImpl() noexcept override = default; VehicleClockBackendImpl(const VehicleClockBackendImpl&) = delete; @@ -89,36 +99,33 @@ class VehicleClockBackendImpl final : public VehicleClockBackend bool WaitUntilAvailable(const score::cpp::stop_token& token, std::chrono::steady_clock::time_point until) const noexcept override; - /// @brief No-op in production backend until callback transport support is available. + /// @brief Installs the time-sync data callback, delivered from the worker thread. void SetTimeSlaveSyncDataReceivedCallback( VehicleTime::TimeSlaveSyncDataReceivedCallback&& callback) noexcept override; - /// @brief No-op in production backend until callback transport support is available. + /// @brief Removes the time-sync data callback, waiting for an in-flight invocation. void UnsetTimeSlaveSyncDataReceivedCallback() noexcept override; - /// @brief No-op in production backend until callback transport support is available. + /// @brief Installs the pDelay measurement callback, delivered from the worker thread. void SetPDelayMeasurementFinishedCallback( VehicleTime::PDelayMeasurementFinishedCallback&& callback) noexcept override; - /// @brief No-op in production backend until callback transport support is available. + /// @brief Removes the pDelay measurement callback, waiting for an in-flight invocation. void UnsetPDelayMeasurementFinishedCallback() noexcept override; - /// @brief No-op in production backend until callback transport support is available. + /// @brief Installs the status-changed callback, delivered from the worker thread. void SetStatusChangedCallback(VehicleTime::StatusChangedCallback&& callback) noexcept override; - /// @brief No-op in production backend until callback transport support is available. + /// @brief Removes the status-changed callback, waiting for an in-flight invocation. void UnsetStatusChangedCallback() noexcept override; private: - /// @brief Converts PTP status flags from the TimeDaemon IPC representation to - /// the @c ClockStatus representation. - static ClockStatus ConvertPtpStatus( - const score::td::svt::TimeBaseStatus& ptp_status) noexcept; - std::atomic_bool is_ready_; std::mutex init_mutex_; std::shared_ptr svt_receiver_; HighResSteadyClock local_clock_; + + SvtCallbackDispatcher dispatcher_; }; } // namespace detail diff --git a/score/time/vehicle_time/src/details/td_impl/vehicle_clock_backend_impl_test.cpp b/score/time/vehicle_time/src/details/td_impl/vehicle_clock_backend_impl_test.cpp index ea350047..45ec858c 100644 --- a/score/time/vehicle_time/src/details/td_impl/vehicle_clock_backend_impl_test.cpp +++ b/score/time/vehicle_time/src/details/td_impl/vehicle_clock_backend_impl_test.cpp @@ -16,13 +16,14 @@ #include "score/time/clock/src/no_status.h" #include "score/time/clock/src/scoped_clock_override.h" #include "score/time/high_res_steady_time/src/high_res_steady_clock_backend_mock.h" -#include "score/time_daemon/src/ipc/receiver_mock.h" -#include "score/time_daemon/src/ipc/svt/svt_time_info.h" +#include "score/time/vehicle_time/src/details/td_impl/svt_test_helpers.h" #include #include +#include #include +#include namespace score { @@ -32,12 +33,9 @@ namespace { using namespace std::chrono_literals; +using namespace test_helpers; using ::testing::Return; -using SvtMock = score::td::ReceiverMock; -using SvtSnapshot = score::td::svt::TimeBaseSnapshot; -using SvtStatus = score::td::svt::TimeBaseStatus; - class VehicleClockBackendImplTest : public ::testing::Test { protected: @@ -45,13 +43,30 @@ class VehicleClockBackendImplTest : public ::testing::Test : mock_hirs_{std::make_shared()}, hirs_guard_{mock_hirs_}, mock_svt_{std::make_shared()}, - impl_{std::make_unique(mock_svt_, HighResSteadyClock::GetInstance())} + frame_source_{}, + impl_{std::make_unique(mock_svt_, + HighResSteadyClock::GetInstance(), + kPollInterval)} + { + } + + void InitBackend() { + EXPECT_CALL(*mock_svt_, Init()).WillOnce(Return(true)); + EXPECT_TRUE(impl_->Init()); + } + + void ServeFramesFromSource() + { + EXPECT_CALL(*mock_svt_, Receive()).WillRepeatedly([this]() { + return frame_source_.Get(); + }); } std::shared_ptr mock_hirs_; test_utils::ScopedClockOverride hirs_guard_; std::shared_ptr mock_svt_; + FrameSource frame_source_; std::unique_ptr impl_; }; @@ -260,14 +275,52 @@ TEST_F(VehicleClockBackendImplTest, WaitUntilAvailableReturnsFalseWhenDeadlinePa EXPECT_FALSE(impl_->WaitUntilAvailable(score::cpp::stop_source{}.get_token(), past_deadline)); } -TEST_F(VehicleClockBackendImplTest, CallbackMethodsAreNoOps) +// --------------------------------------------------------------------------- +// Init gating of callback delivery (the worker itself is owned by SvtCallbackDispatcher +// and covered in svt_callback_dispatcher_test.cpp) +// --------------------------------------------------------------------------- + +TEST_F(VehicleClockBackendImplTest, CallbacksAreNotDeliveredBeforeInit) +{ + EXPECT_CALL(*mock_svt_, Receive()).Times(0); + + Recorder recorder; + impl_->SetStatusChangedCallback([&recorder](const VehicleTimeStatus& status) { + recorder.Record(status); + }); + + std::this_thread::sleep_for(20 * kPollInterval); + EXPECT_EQ(recorder.Count(), 0U); +} + +TEST_F(VehicleClockBackendImplTest, CallbacksAreNotDeliveredWhenInitFails) { - impl_->SetTimeSlaveSyncDataReceivedCallback([](const TimeSlaveSyncData&) {}); - impl_->UnsetTimeSlaveSyncDataReceivedCallback(); - impl_->SetPDelayMeasurementFinishedCallback([](const PDelayMeasurementData&) {}); - impl_->UnsetPDelayMeasurementFinishedCallback(); - impl_->SetStatusChangedCallback([](const VehicleTimeStatus&) {}); - impl_->UnsetStatusChangedCallback(); + EXPECT_CALL(*mock_svt_, Init()).WillOnce(Return(false)); + EXPECT_CALL(*mock_svt_, Receive()).Times(0); + EXPECT_FALSE(impl_->Init()); + + Recorder recorder; + impl_->SetStatusChangedCallback([&recorder](const VehicleTimeStatus& status) { + recorder.Record(status); + }); + + std::this_thread::sleep_for(20 * kPollInterval); + EXPECT_EQ(recorder.Count(), 0U); +} + +TEST_F(VehicleClockBackendImplTest, CallbackRegisteredBeforeInitIsDeliveredOnceInitSucceeds) +{ + Recorder recorder; + impl_->SetStatusChangedCallback([&recorder](const VehicleTimeStatus& status) { + recorder.Record(status); + }); + + frame_source_.Set(MakeFrame(kSynchronizedStatus)); + ServeFramesFromSource(); + InitBackend(); + + ASSERT_TRUE(recorder.WaitForCount(1U)); + EXPECT_TRUE(recorder.Last().IsFlagActive(VehicleTime::StatusFlag::kSynchronized)); } } // namespace diff --git a/score/time/vehicle_time/src/vehicle_clock_backend.h b/score/time/vehicle_time/src/vehicle_clock_backend.h index a5a1fd74..47c15936 100644 --- a/score/time/vehicle_time/src/vehicle_clock_backend.h +++ b/score/time/vehicle_time/src/vehicle_clock_backend.h @@ -63,36 +63,53 @@ class VehicleClockBackend virtual bool WaitUntilAvailable(const score::cpp::stop_token& token, std::chrono::steady_clock::time_point until) const noexcept = 0; - /// @brief Installs callback invoked when new time-sync data arrives. + /// @brief Installs the callback invoked when new time-sync data arrives. /// - /// Trigger conditions depend on backend implementation. + /// Fires for the first Sync/Follow-Up frame received from the TimeDaemon after registration + /// and afterwards for every frame whose content differs from the previously delivered one. + /// Invoked on the backend's dedicated worker thread. + /// + /// Replacing an installed callback is safe while an invocation is in flight: the call + /// returns only once the previous callback is no longer running (unless made from + /// within that callback itself). virtual void SetTimeSlaveSyncDataReceivedCallback( VehicleTime::TimeSlaveSyncDataReceivedCallback&& callback) noexcept = 0; /// @brief Removes the time-sync data callback. + /// + /// Returns only once an in-flight invocation has completed (unless called from within + /// the callback itself), so callers may release captured resources afterwards. virtual void UnsetTimeSlaveSyncDataReceivedCallback() noexcept = 0; - /// @brief Installs callback invoked after finished pDelay measurement. + /// @brief Installs the callback invoked after a finished pDelay measurement. /// - /// Trigger conditions depend on backend implementation. + /// Fires for the first pDelay measurement result received from the TimeDaemon after + /// registration and afterwards for every result that differs from the previously delivered + /// one. Invoked on the backend's dedicated worker thread. Same replacement guarantees as + /// @c SetTimeSlaveSyncDataReceivedCallback(). virtual void SetPDelayMeasurementFinishedCallback( VehicleTime::PDelayMeasurementFinishedCallback&& callback) noexcept = 0; /// @brief Removes the pDelay measurement callback. + /// + /// Same completion guarantee as @c UnsetTimeSlaveSyncDataReceivedCallback(). virtual void UnsetPDelayMeasurementFinishedCallback() noexcept = 0; /// @brief Installs the callback invoked when VehicleTimeStatus flags change. /// /// The callback fires: - /// - unconditionally on the first \c Now() call after registration; and - /// - on every subsequent \c Now() call where the status flags differ from - /// the last fired value (rate deviation is ignored for comparison). + /// - unconditionally on the first status frame received after registration; and + /// - afterwards only when the status flags differ from the last delivered value + /// (rate deviation is ignored for comparison). /// - /// The callback is invoked on the thread that calls \c Now() — typically - /// a background polling thread. The implementation must be thread-safe. + /// The callback is invoked on the backend's dedicated worker thread — the callback + /// implementation must be thread-safe. Same replacement guarantees as + /// @c SetTimeSlaveSyncDataReceivedCallback(). virtual void SetStatusChangedCallback(VehicleTime::StatusChangedCallback&& callback) noexcept = 0; /// @brief Removes the status-changed callback. + /// + /// Same completion guarantee as @c UnsetTimeSlaveSyncDataReceivedCallback(). virtual void UnsetStatusChangedCallback() noexcept = 0; };