timer.hpp

Go to the documentation of this file.

 
#ifndef JOIN_CORE_TIMER_HPP
#define JOIN_CORE_TIMER_HPP
 
// libjoin.
#include <join/proactor.hpp>
#include <join/function.hpp>
#include <join/backoff.hpp>
#include <join/clock.hpp>
 
// C++.
#include <functional>
#include <chrono>
#include <atomic>
#include <memory>
#include <new>
 
// C.
#include <sys/timerfd.h>
#include <unistd.h>
#include <cstdlib>
 
namespace join
{
    template <class ClockPolicy, class ProactorType>
    class BasicTimer : protected CompletionHandler
    {
    public:
        enum class State
        {
            Idle,     
            Armed,    
            Invoking, 
            Closed,   
        };
 
        explicit BasicTimer (ProactorType& proactor = ProactorThread::proactor ())
        : _handle (timerfd_create (ClockPolicy::type (), TFD_NONBLOCK | TFD_CLOEXEC))
        , _proactor (proactor)
        {
            if (_handle == -1)
            {
                throw std::system_error (errno, std::system_category (), "timerfd_create failed");  // LCOV_EXCL_LINE
            }
 
            _ops->op = IoOperation::makeRead (_handle, &_ops->expirations,
                                              static_cast<uint32_t> (sizeof (_ops->expirations)), this);
 
            if (_proactor.submit (_ops->op, true, true) == -1)
            {
                // LCOV_EXCL_START
                close (_handle);
                throw std::system_error (lastError, "timerfd submit failed");
                // LCOV_EXCL_STOP
            }
        }
 
        BasicTimer (const BasicTimer& other) = delete;
 
        BasicTimer& operator= (const BasicTimer& other) = delete;
 
        BasicTimer (BasicTimer&& other) = delete;
 
        BasicTimer& operator= (BasicTimer&& other) = delete;
 
        ~BasicTimer () noexcept
        {
            cancel ();
 
            Backoff backoff;
            while (_state.load (std::memory_order_acquire) != State::Closed)
            {
                _proactor.cancel (_ops->op, true, true);
                backoff ();
            }
 
            close (_handle);
        }
 
        template <class Rep, class Period, typename Func, typename... Args>
        void setOneShot (std::chrono::duration<Rep, Period> duration, Func&& callback, Args&&... args)
        {
            cancel ();
 
            auto ns = std::chrono::duration_cast<std::chrono::nanoseconds> (duration);
            _callback = std::bind (std::forward<Func> (callback), std::forward<Args> (args)...);
            _oneShot.store (true, std::memory_order_relaxed);
            _ns.store (std::chrono::nanoseconds::zero (), std::memory_order_relaxed);
            _state.store (State::Armed, std::memory_order_release);
 
            auto ts = toTimerSpec (ns);
            timerfd_settime (_handle, 0, &ts, nullptr);
        }
 
        template <class Clock, class Duration, typename Func, typename... Args>
        void setOneShot (std::chrono::time_point<Clock, Duration> timePoint, Func&& callback, Args&&... args)
        {
            static_assert (
                (std::is_same_v<ClockPolicy, RealTime> && std::is_same_v<Clock, std::chrono::system_clock>) ||
                    (std::is_same_v<ClockPolicy, Monotonic> && std::is_same_v<Clock, std::chrono::steady_clock>),
                "Clock type mismatch timer policy");
 
            cancel ();
 
            auto elapsed = timePoint.time_since_epoch ();
            auto ns = std::chrono::duration_cast<std::chrono::nanoseconds> (elapsed);
            _callback = std::bind (std::forward<Func> (callback), std::forward<Args> (args)...);
            _oneShot.store (true, std::memory_order_relaxed);
            _ns.store (std::chrono::nanoseconds::zero (), std::memory_order_relaxed);
            _state.store (State::Armed, std::memory_order_release);
 
            auto ts = toTimerSpec (ns);
            timerfd_settime (_handle, TFD_TIMER_ABSTIME, &ts, nullptr);
        }
 
        template <class Rep, class Period, typename Func, typename... Args>
        void setInterval (std::chrono::duration<Rep, Period> duration, Func&& callback, Args&&... args)
        {
            cancel ();
 
            auto ns = std::chrono::duration_cast<std::chrono::nanoseconds> (duration);
            _callback = std::bind (std::forward<Func> (callback), std::forward<Args> (args)...);
            _oneShot.store (false, std::memory_order_relaxed);
            _ns.store (ns, std::memory_order_relaxed);
            _state.store (State::Armed, std::memory_order_release);
 
            auto ts = toTimerSpec (ns, true);
            timerfd_settime (_handle, 0, &ts, nullptr);
        }
 
        void cancel () noexcept
        {
            _oneShot.store (true, std::memory_order_relaxed);
            _ns.store (std::chrono::nanoseconds::zero (), std::memory_order_relaxed);
 
            struct itimerspec ts = {};
            timerfd_settime (_handle, 0, &ts, nullptr);
 
            if (JOIN_UNLIKELY (_proactor.isProactorThread ()))
            {
                _state.store (State::Idle, std::memory_order_release);
                return;
            }
 
            Backoff backoff;
            State expected = State::Armed;
 
            while (!_state.compare_exchange_strong (expected, State::Idle, std::memory_order_acq_rel,
                                                    std::memory_order_acquire))
            {
                if (expected != State::Invoking)
                {
                    return;
                }
 
                backoff ();
                expected = State::Armed;
            }
 
            _callback = nullptr;
        }
 
        bool active () const noexcept
        {
            struct itimerspec ts = {};
            timerfd_gettime (_handle, &ts);
            const bool hasValue = (ts.it_value.tv_sec != 0 || ts.it_value.tv_nsec != 0);
            const bool hasInterval = (ts.it_interval.tv_sec != 0 || ts.it_interval.tv_nsec != 0);
            return hasValue || hasInterval;
        }
 
        std::chrono::nanoseconds remaining () const noexcept
        {
            struct itimerspec ts = {};
            timerfd_gettime (_handle, &ts);
            return std::chrono::seconds (ts.it_value.tv_sec) + std::chrono::nanoseconds (ts.it_value.tv_nsec);
        }
 
        std::chrono::nanoseconds interval () const noexcept
        {
            return _ns.load (std::memory_order_relaxed);
        }
 
        bool oneShot () const noexcept
        {
            return _oneShot.load (std::memory_order_relaxed);
        }
 
        static constexpr int type () noexcept
        {
            return ClockPolicy::type ();
        }
 
    private:
        void onComplete ([[maybe_unused]] IoOperation& op, int result) override
        {
            uint64_t expirations = _ops->expirations;
 
            if (JOIN_UNLIKELY (result != static_cast<int> (sizeof (_ops->expirations)) && result != -EAGAIN &&
                               result != -EINTR))
            {
                // LCOV_EXCL_START
                _state.store (State::Closed, std::memory_order_release);
                return;
                // LCOV_EXCL_STOP
            }
 
            if (JOIN_UNLIKELY (_proactor.submit (_ops->op) == -1))
            {
                // LCOV_EXCL_START
                _state.store (State::Closed, std::memory_order_release);
                return;
                // LCOV_EXCL_STOP
            }
 
            if (JOIN_UNLIKELY (result != static_cast<int> (sizeof (_ops->expirations))))
            {
                return;  // LCOV_EXCL_LINE
            }
 
            State expected = State::Armed;
 
            if (JOIN_LIKELY (_state.compare_exchange_strong (expected, State::Invoking, std::memory_order_acq_rel,
                                                             std::memory_order_acquire)))
            {
                for (uint64_t i = 0; i < expirations; ++i)
                {
                    _callback ();
                }
 
                expected = State::Invoking;
                _state.compare_exchange_strong (expected, State::Armed, std::memory_order_acq_rel,
                                                std::memory_order_acquire);
            }
        }
 
        void onCancel ([[maybe_unused]] IoOperation& op, [[maybe_unused]] int result) override
        {
            _state.store (State::Closed, std::memory_order_release);
        }
 
        static constexpr itimerspec toTimerSpec (std::chrono::nanoseconds ns, bool periodic = false) noexcept
        {
            itimerspec ts{};
            if (periodic)
            {
                ts.it_interval.tv_sec = ns.count () / _nsPerSec;
                ts.it_interval.tv_nsec = ns.count () % _nsPerSec;
            }
            if (ns.count () < 1)
            {
                ns = std::chrono::nanoseconds (1);
            }
            ts.it_value.tv_sec = ns.count () / _nsPerSec;
            ts.it_value.tv_nsec = ns.count () % _nsPerSec;
            return ts;
        }
 
        struct Ops
        {
            static void* operator new (size_t size)
            {
                void* mem = ::aligned_alloc (alignof (Ops), size);
 
                if (mem == nullptr)
                {
                    throw std::bad_alloc ();  // LCOV_EXCL_LINE
                }
 
                return mem;
            }
 
            static void operator delete (void* mem) noexcept
            {
                ::free (mem);
            }
 
            IoOperation op = {};
 
            uint64_t expirations = 0;
        };
 
        static constexpr uint64_t _nsPerSec = 1000000000ULL;
 
        const std::unique_ptr<Ops> _ops{new Ops ()};
 
        std::atomic<State> _state{State::Idle};
 
        Function<void ()> _callback;
 
        std::atomic<std::chrono::nanoseconds> _ns{std::chrono::nanoseconds::zero ()};
 
        std::atomic_bool _oneShot{true};
 
        int _handle = -1;
 
        ProactorType& _proactor;
    };
}
 
#endif