ntp_time_service.hpp

Go to the documentation of this file.

// SPDX-License-Identifier: MIT
#pragma once
#ifndef _TIME_SHIELD_NTP_TIME_SERVICE_HPP_INCLUDED
#define _TIME_SHIELD_NTP_TIME_SERVICE_HPP_INCLUDED
 
#include "config.hpp"
 
#if TIME_SHIELD_ENABLE_NTP_CLIENT
 
#include "ntp_client_pool.hpp"
#include "ntp_client_pool_runner.hpp"
#include "time_utils.hpp"
 
#include <atomic>
#include <chrono>
#include <condition_variable>
#include <cstdlib>
#include <cstdint>
#include <memory>
#include <mutex>
#include <thread>
#include <utility>
#include <vector>
 
namespace time_shield {
 
    template <class RunnerT>
    class NtpTimeServiceT;
 
    namespace detail {
        template <class RunnerT>
        struct NtpTimeServiceSingleton;
 
        template <class RunnerT>
        struct NtpTimeServiceTestAccess;
 
#ifdef TIME_SHIELD_TEST_FAKE_NTP
        class FakeNtpRunner {
        public:
            FakeNtpRunner() = default;
            explicit FakeNtpRunner(NtpClientPool) {}
 
            FakeNtpRunner(const FakeNtpRunner&) = delete;
            FakeNtpRunner& operator=(const FakeNtpRunner&) = delete;

            ~FakeNtpRunner() {
                stop();
            }

            bool start(std::chrono::milliseconds interval = std::chrono::seconds(30),
                       bool measure_immediately = true) {
                if (m_is_running.load()) {
                    return false;
                }
                if (interval.count() <= 0) {
                    interval = std::chrono::milliseconds(1);
                }
                m_interval = interval;
                m_is_stop_requested.store(false);
                m_is_force_requested.store(false);
                m_is_running.store(true);
                m_measure_immediately = measure_immediately;
                try {
                    m_thread = std::thread(&FakeNtpRunner::run_loop, this);
                } catch (...) {
                    m_is_running.store(false);
                    return false;
                }
                return true;
            }

            bool start(int interval_ms, bool measure_immediately = true) {
                return start(std::chrono::milliseconds(interval_ms), measure_immediately);
            }

            void stop() {
                m_is_stop_requested.store(true);
                m_cv.notify_all();
                if (m_thread.joinable()) {
                    m_thread.join();
                }
                m_is_running.store(false);
            }

            bool running() const noexcept { return m_is_running.load(); }

            bool force_measure() {
                if (!m_is_running.load()) {
                    return false;
                }
                m_is_force_requested.store(true);
                m_cv.notify_one();
                return true;
            }

            bool measure_now() {
                return do_measure();
            }

            int64_t offset_us() const noexcept { return m_offset_us.load(); }
            int64_t utc_time_us() const noexcept { return now_realtime_us() + m_offset_us.load(); }
            int64_t utc_time_ms() const noexcept { return utc_time_us() / 1000; }
            int64_t utc_time_sec() const noexcept { return utc_time_us() / 1000000; }

            bool last_measure_ok() const noexcept { return m_last_measure_ok.load(); }
            uint64_t measure_count() const noexcept { return m_measure_count.load(); }
            uint64_t fail_count() const noexcept { return m_fail_count.load(); }
            int64_t last_update_realtime_us() const noexcept { return m_last_update_realtime_us.load(); }
            int64_t last_success_realtime_us() const noexcept { return m_last_success_realtime_us.load(); }

            std::vector<NtpSample> last_samples() const { return {}; }
 
        private:
            void run_loop() {
                const auto sleep_interval = m_interval;
                bool is_first = m_measure_immediately;
                while (!m_is_stop_requested.load()) {
                    if (is_first) {
                        do_measure();
                        is_first = false;
                    } else {
                        std::unique_lock<std::mutex> lk(m_cv_mtx);
                        m_cv.wait_for(lk, sleep_interval, [this]() {
                            return m_is_stop_requested.load() || m_is_force_requested.load();
                        });
                        if (m_is_stop_requested.load()) {
                            break;
                        }
                        m_is_force_requested.store(false);
                        do_measure();
                    }
                }
                m_is_running.store(false);
            }

            bool do_measure() {
                const uint64_t count = m_measure_count.fetch_add(1) + 1;
                m_offset_us.store(static_cast<int64_t>(count * 1000));
                m_last_measure_ok.store(true);
                const int64_t now = now_realtime_us();
                m_last_update_realtime_us.store(now);
                m_last_success_realtime_us.store(now);
                return true;
            }
 
        private:
            std::chrono::milliseconds m_interval{std::chrono::seconds(30)};
            bool m_measure_immediately{true};
 
            std::thread m_thread;
            std::condition_variable m_cv;
            std::mutex m_cv_mtx;
 
            std::atomic<bool> m_is_running{false};
            std::atomic<bool> m_is_stop_requested{false};
            std::atomic<bool> m_is_force_requested{false};
 
            std::atomic<bool> m_last_measure_ok{false};
            std::atomic<uint64_t> m_measure_count{0};
            std::atomic<uint64_t> m_fail_count{0};
            std::atomic<int64_t> m_last_update_realtime_us{0};
            std::atomic<int64_t> m_last_success_realtime_us{0};
            std::atomic<int64_t> m_offset_us{0};
        };
#endif // _TIME_SHIELD_TEST_FAKE_NTP
    } // namespace detail

    template <class RunnerT>
    class NtpTimeServiceT {
        friend struct detail::NtpTimeServiceSingleton<RunnerT>;
        friend struct detail::NtpTimeServiceTestAccess<RunnerT>;
 
    public:
        static NtpTimeServiceT& instance() noexcept {
            return detail::NtpTimeServiceSingleton<RunnerT>::instance();
        }
 
        NtpTimeServiceT(const NtpTimeServiceT&) = delete;
        NtpTimeServiceT& operator=(const NtpTimeServiceT&) = delete;

        bool init() {
            return init(m_interval, m_measure_immediately);
        }

        bool init(std::chrono::milliseconds interval, bool measure_immediately = true) {
            if (is_process_shutting_down()) {
                return false;
            }
 
            std::unique_ptr<RunnerT> local_runner;
            std::chrono::milliseconds start_interval = interval;
            bool use_measure_immediately = measure_immediately;
            {
                std::unique_lock<std::mutex> lk(m_mtx);
                while (is_transitioning_locked()) {
                    m_cv.wait(lk);
                }
                if (is_process_shutting_down_locked()) {
                    return false;
                }
                if (m_state == State::running && is_running_locked()) {
                    return true;
                }
                if (start_interval.count() <= 0) {
                    start_interval = std::chrono::milliseconds(1);
                }
                m_interval = start_interval;
                m_measure_immediately = use_measure_immediately;
                m_state = State::starting;
 
                local_runner = build_runner_locked();
                if (!local_runner) {
                    m_state = State::stopped;
                    lk.unlock();
                    m_cv.notify_all();
                    return false;
                }
            }
 
            bool has_started = false;
            bool is_ok = false;
            try {
                has_started = local_runner->start(start_interval, use_measure_immediately);
                if (has_started) {
                    is_ok = local_runner->measure_now();
                }
            } catch (...) {
                is_ok = false;
            }
 
            if (!has_started || !is_ok || is_process_shutting_down()) {
                try {
                    if (has_started) {
                        local_runner->stop();
                    }
                } catch (...) {
                    // no-throw
                }
                local_runner.reset();
            }
 
            {
                std::lock_guard<std::mutex> lk(m_mtx);
                if (local_runner) {
                    m_last_offset_us.store(local_runner->offset_us(), std::memory_order_relaxed);
                    m_runner = std::move(local_runner);
                    m_state = State::running;
                } else {
                    m_runner.reset();
                    m_state = State::stopped;
                }
            }
            m_cv.notify_all();
            return is_ok;
        }

        bool init(int interval_ms, bool measure_immediately = true) {
            return init(std::chrono::milliseconds(interval_ms), measure_immediately);
        }

        void shutdown() {
            std::unique_ptr<RunnerT> local_runner;
            {
                std::unique_lock<std::mutex> lk(m_mtx);
                while (is_transitioning_locked()) {
                    m_cv.wait(lk);
                }
                if (m_state == State::stopped || !m_runner) {
                    return;
                }
                m_state = State::stopping;
                local_runner = std::move(m_runner);
            }
            try {
                local_runner->stop();
            } catch (...) {
                // no-throw
            }
            {
                std::lock_guard<std::mutex> lk(m_mtx);
                m_state = State::stopped;
            }
            m_cv.notify_all();
        }

        bool running() const noexcept {
            std::lock_guard<std::mutex> lk(m_mtx);
            return m_state == State::running && is_running_locked();
        }

        void ensure_started() noexcept {
            if (is_process_shutting_down()) {
                return;
            }
            if (running()) {
                return;
            }
            (void)init();
        }

        int64_t offset_us() noexcept {
            if (is_process_shutting_down()) {
                return m_last_offset_us.load(std::memory_order_relaxed);
            }
            ensure_started();
            std::lock_guard<std::mutex> lk(m_mtx);
            if (!m_runner) return 0;
            const int64_t offset = m_runner->offset_us();
            m_last_offset_us.store(offset, std::memory_order_relaxed);
            return offset;
        }

        int64_t utc_time_us() noexcept {
            if (is_process_shutting_down()) {
                return now_realtime_us() + m_last_offset_us.load(std::memory_order_relaxed);
            }
            ensure_started();
            std::lock_guard<std::mutex> lk(m_mtx);
            if (!m_runner) return now_realtime_us();
            m_last_offset_us.store(m_runner->offset_us(), std::memory_order_relaxed);
            return m_runner->utc_time_us();
        }

        int64_t utc_time_ms() noexcept {
            return utc_time_us() / 1000;
        }

        int64_t utc_time_sec() noexcept {
            return utc_time_us() / 1000000;
        }

        bool last_measure_ok() const noexcept {
            std::lock_guard<std::mutex> lk(m_mtx);
            if (!m_runner) return false;
            return m_runner->last_measure_ok();
        }

        uint64_t measure_count() const noexcept {
            std::lock_guard<std::mutex> lk(m_mtx);
            if (!m_runner) return 0;
            return m_runner->measure_count();
        }

        uint64_t fail_count() const noexcept {
            std::lock_guard<std::mutex> lk(m_mtx);
            if (!m_runner) return 0;
            return m_runner->fail_count();
        }

        int64_t last_update_realtime_us() const noexcept {
            std::lock_guard<std::mutex> lk(m_mtx);
            if (!m_runner) return 0;
            return m_runner->last_update_realtime_us();
        }

        int64_t last_success_realtime_us() const noexcept {
            std::lock_guard<std::mutex> lk(m_mtx);
            if (!m_runner) return 0;
            return m_runner->last_success_realtime_us();
        }

        bool stale(std::chrono::milliseconds max_age) const noexcept {
            const int64_t last = last_update_realtime_us();
            if (last == 0) {
                return true;
            }
            const int64_t age = now_realtime_us() - last;
            return age > static_cast<int64_t>(max_age.count()) * 1000;
        }

        bool stale(int max_age_ms) const noexcept {
            return stale(std::chrono::milliseconds(max_age_ms));
        }

        bool set_servers(std::vector<NtpServerConfig> servers) {
            std::lock_guard<std::mutex> lk(m_mtx);
            if (!is_reconfigurable_locked()) {
                return false;
            }
            m_has_custom_servers = true;
            m_servers = std::move(servers);
            return true;
        }

        bool set_default_servers() {
            std::lock_guard<std::mutex> lk(m_mtx);
            if (!is_reconfigurable_locked()) {
                return false;
            }
            m_has_custom_servers = true;
            m_servers = NtpClientPool::build_default_servers();
            return true;
        }

        bool clear_servers() {
            std::lock_guard<std::mutex> lk(m_mtx);
            if (!is_reconfigurable_locked()) {
                return false;
            }
            m_has_custom_servers = false;
            m_servers.clear();
            return true;
        }

        bool set_pool_config(NtpPoolConfig cfg) {
            std::lock_guard<std::mutex> lk(m_mtx);
            if (!is_reconfigurable_locked()) {
                return false;
            }
            m_has_custom_pool_cfg = true;
            m_pool_cfg = std::move(cfg);
            return true;
        }

        NtpPoolConfig pool_config() const {
            std::lock_guard<std::mutex> lk(m_mtx);
            if (m_has_custom_pool_cfg) {
                return m_pool_cfg;
            }
            return NtpPoolConfig{};
        }

        std::vector<NtpSample> last_samples() const {
            std::lock_guard<std::mutex> lk(m_mtx);
            if (!m_runner) return {};
            return m_runner->last_samples();
        }

        NtpTimeServiceT() = default;
        ~NtpTimeServiceT() = default;

        bool apply_config_now() {
            if (is_process_shutting_down()) {
                return false;
            }
 
            std::unique_ptr<RunnerT> new_runner;
            std::unique_ptr<RunnerT> old_runner;
            std::chrono::milliseconds interval;
            bool measure_immediately = true;
            bool was_running = false;
            {
                std::unique_lock<std::mutex> lk(m_mtx);
                while (is_transitioning_locked()) {
                    m_cv.wait(lk);
                }
                if (is_process_shutting_down_locked()) {
                    return false;
                }
                was_running = m_state == State::running && is_running_locked();
                m_state = State::starting;
                new_runner = build_runner_locked();
                if (!new_runner) {
                    m_state = was_running ? State::running : State::stopped;
                    lk.unlock();
                    m_cv.notify_all();
                    return false;
                }
                interval = m_interval;
                measure_immediately = m_measure_immediately;
                old_runner = std::move(m_runner);
            }
 
            if (old_runner) {
                try {
                    old_runner->stop();
                } catch (...) {
                }
            }
 
            bool has_started = false;
            bool is_ok = false;
            try {
                has_started = new_runner->start(interval, measure_immediately);
                if (has_started) {
                    is_ok = new_runner->measure_now();
                }
            } catch (...) {
                is_ok = false;
            }
 
            if (!has_started || !is_ok || is_process_shutting_down()) {
                try {
                    if (has_started) {
                        new_runner->stop();
                    }
                } catch (...) {
                    // no-throw
                }
                new_runner.reset();
            }
 
            {
                std::lock_guard<std::mutex> lk(m_mtx);
                if (new_runner) {
                    m_last_offset_us.store(new_runner->offset_us(), std::memory_order_relaxed);
                    m_runner = std::move(new_runner);
                    m_state = State::running;
                } else {
                    m_runner.reset();
                    m_state = State::stopped;
                }
            }
            m_cv.notify_all();
            return is_ok;
        }
 
    private:
        enum class ProcessState : uint8_t {
            alive,
            shutting_down
        };

        enum class State : uint8_t {
            stopped,
            starting,
            running,
            stopping
        };

        void register_process_shutdown_handler() noexcept {
            if (std::atexit(&detail::NtpTimeServiceSingleton<RunnerT>::handle_process_exit) == 0) {
                m_atexit_registration_count.fetch_add(1, std::memory_order_relaxed);
            }
        }

        void begin_process_shutdown() noexcept {
            m_process_state.store(ProcessState::shutting_down, std::memory_order_release);
 
            std::unique_ptr<RunnerT> local_runner;
            {
                std::unique_lock<std::mutex> lk(m_mtx);
                while (is_transitioning_locked()) {
                    m_cv.wait(lk);
                }
                if (m_runner) {
                    m_last_offset_us.store(m_runner->offset_us(), std::memory_order_relaxed);
                    m_state = State::stopping;
                    local_runner = std::move(m_runner);
                } else {
                    m_state = State::stopped;
                }
            }
 
            if (local_runner) {
                try {
                    local_runner->stop();
                } catch (...) {
                    // no-throw
                }
            }
 
            {
                std::lock_guard<std::mutex> lk(m_mtx);
                m_state = State::stopped;
            }
            m_cv.notify_all();
        }

        bool is_process_shutting_down() const noexcept {
            return m_process_state.load(std::memory_order_acquire) == ProcessState::shutting_down;
        }

        bool is_process_shutting_down_locked() const noexcept {
            return is_process_shutting_down();
        }

        uint32_t atexit_registration_count() const noexcept {
            return m_atexit_registration_count.load(std::memory_order_relaxed);
        }

        bool is_running_locked() const noexcept {
            return m_runner && m_runner->running();
        }

        bool is_transitioning_locked() const noexcept {
            return m_state == State::starting || m_state == State::stopping;
        }

        bool is_reconfigurable_locked() const noexcept {
            return !is_process_shutting_down_locked() && m_state == State::stopped && !m_runner;
        }

        std::unique_ptr<RunnerT> build_runner_locked() {
            std::vector<NtpServerConfig> servers;
            if (m_has_custom_servers) {
                servers = m_servers;
            } else {
                servers = NtpClientPool::build_default_servers();
            }
 
            NtpPoolConfig cfg = m_has_custom_pool_cfg ? m_pool_cfg : NtpPoolConfig{};
            NtpClientPool pool(cfg);
            pool.set_servers(std::move(servers));
 
            std::unique_ptr<RunnerT> runner;
            try {
                runner.reset(new RunnerT(std::move(pool)));
            } catch (...) {
                return nullptr;
            }
            return runner;
        }
 
    private:
        mutable std::mutex m_mtx;
        std::condition_variable m_cv;
        State m_state{State::stopped};
        std::atomic<ProcessState> m_process_state{ProcessState::alive};
        std::atomic<int64_t> m_last_offset_us{0};
        std::atomic<uint32_t> m_atexit_registration_count{0};
        std::chrono::milliseconds m_interval{std::chrono::seconds(30)};
        bool m_measure_immediately{true};
 
        bool m_has_custom_servers{false};
        std::vector<NtpServerConfig> m_servers;
 
        bool m_has_custom_pool_cfg{false};
        NtpPoolConfig m_pool_cfg{};
 
        std::unique_ptr<RunnerT> m_runner;
    };
 
    namespace detail {
        template <class RunnerT>
        struct NtpTimeServiceSingleton final {
            static NtpTimeServiceT<RunnerT>& instance() noexcept {
                static NtpTimeServiceT<RunnerT>* p_instance = []() noexcept {
                    NtpTimeServiceT<RunnerT>* p_service = new NtpTimeServiceT<RunnerT>{};
                    p_service->register_process_shutdown_handler();
                    return p_service;
                }();
                return *p_instance;
            }
 
            static void handle_process_exit() noexcept {
                instance().begin_process_shutdown();
            }
        };
 
        template <class RunnerT>
        struct NtpTimeServiceTestAccess final {
            static void begin_process_shutdown() noexcept {
                NtpTimeServiceSingleton<RunnerT>::instance().begin_process_shutdown();
            }
 
            static bool is_process_shutting_down() noexcept {
                return NtpTimeServiceSingleton<RunnerT>::instance().is_process_shutting_down();
            }
 
            static uint32_t atexit_registration_count() noexcept {
                return NtpTimeServiceSingleton<RunnerT>::instance().atexit_registration_count();
            }
        };
    } // namespace detail
 
#if defined(TIME_SHIELD_TEST_FAKE_NTP)
    using NtpTimeService = NtpTimeServiceT<detail::FakeNtpRunner>;
#else
    using NtpTimeService = NtpTimeServiceT<NtpClientPoolRunner>;
#endif
 
namespace ntp {

    inline bool init(std::chrono::milliseconds interval = std::chrono::seconds(30),
                     bool measure_immediately = true) {
        return NtpTimeService::instance().init(interval, measure_immediately);
    }

    inline bool init(int interval_ms,
                     bool measure_immediately = true) {
        return NtpTimeService::instance().init(std::chrono::milliseconds(interval_ms), measure_immediately);
    }
    
    inline void shutdown() {
        NtpTimeService::instance().shutdown();
    }
    
    inline int64_t offset_us() noexcept {
        return NtpTimeService::instance().offset_us();
    }
    
    inline int64_t utc_time_us() noexcept {
        return NtpTimeService::instance().utc_time_us();
    }
    
    inline int64_t utc_time_ms() noexcept {
        return NtpTimeService::instance().utc_time_ms();
    }
    
    inline int64_t utc_time_sec() noexcept {
        return NtpTimeService::instance().utc_time_sec();
    }
    
    inline bool last_measure_ok() noexcept {
        return NtpTimeService::instance().last_measure_ok();
    }
    
    inline uint64_t measure_count() noexcept {
        return NtpTimeService::instance().measure_count();
    }
    
    inline uint64_t fail_count() noexcept {
        return NtpTimeService::instance().fail_count();
    }
    
    inline int64_t last_update_realtime_us() noexcept {
        return NtpTimeService::instance().last_update_realtime_us();
    }
    
    inline int64_t last_success_realtime_us() noexcept {
        return NtpTimeService::instance().last_success_realtime_us();
    }
    
    inline bool stale(std::chrono::milliseconds max_age) noexcept {
        return NtpTimeService::instance().stale(max_age);
    }

    inline bool stale(int max_age_ms) noexcept {
        return NtpTimeService::instance().stale(max_age_ms);
    }
 
} // namespace ntp
 
} // namespace time_shield
 
#else // TIME_SHIELD_ENABLE_NTP_CLIENT
 
namespace time_shield {
    class NtpTimeService {
    public:
        static NtpTimeService& instance() {
            static_assert(sizeof(void*) == 0, "NtpTimeService is disabled by configuration.");
            return *reinterpret_cast<NtpTimeService*>(0);
        }
    };
} // namespace time_shield
 
#endif // _TIME_SHIELD_ENABLE_NTP_CLIENT
 
#endif // _TIME_SHIELD_NTP_TIME_SERVICE_HPP_INCLUDED