wheel.hpp

Go to the documentation of this file.

 
#ifndef JOIN_CORE_WHEEL_HPP
#define JOIN_CORE_WHEEL_HPP
 
// libjoin.
#include <join/slab_allocator.hpp>
#include <join/function.hpp>
#include <join/backoff.hpp>
#include <join/memory.hpp>
#include <join/thread.hpp>
#include <join/queue.hpp>
#include <join/error.hpp>
#include <join/clock.hpp>
#include <join/utils.hpp>
 
// C++.
#include <functional>
#include <utility>
#include <chrono>
#include <atomic>
#include <limits>
#include <new>
 
// C.
#include <pthread.h>
#include <cstdint>
 
namespace join
{
    template <class ClockPolicy, class RunPolicy, size_t Capacity, uint64_t TickNs>
    class BasicWheel
    {
        static_assert (Capacity > 0, "capacity must be greater than zero");
        static_assert (TickNs > 0, "tick duration must be greater than zero");
 
        friend RunPolicy;
 
    public:
        template <typename... Args>
        explicit BasicWheel (Args&&... args)
        : _commands (_queueSize)
        , _generationsMem (Capacity * sizeof (std::atomic_uint32_t))
        , _generations (static_cast<std::atomic_uint32_t*> (_generationsMem.get ()))
        , _runner (std::forward<Args> (args)...)
        {
            _allocator.reserveAll ();
 
            for (uint32_t index = 0; index < Capacity; ++index)
            {
                new (&_generations[index]) std::atomic_uint32_t (0);
                new (_allocator.getPtr (index)) Node ();
            }
 
            _allocator.releaseAll ();
 
            _tick = toTick (ClockPolicy::now ());
            _runner.start (*this, resolution ());
        }
 
        BasicWheel (const BasicWheel& other) = delete;
 
        BasicWheel& operator= (const BasicWheel& other) = delete;
 
        BasicWheel (BasicWheel&& other) = delete;
 
        BasicWheel& operator= (BasicWheel&& other) = delete;
 
        ~BasicWheel () noexcept
        {
            _runner.stop ();
            readCommands ();
 
            for (uint32_t index = 0; index < Capacity; ++index)
            {
                static_cast<Node*> (_allocator.getPtr (index))->~Node ();
                _generations[index].~atomic ();
            }
        }
 
        template <class Rep, class Period, typename Func, typename... Args>
        ssize_t setOneShot (std::chrono::duration<Rep, Period> duration, Func&& callback, Args&&... args) noexcept
        {
            return createTimer (ticksFor (duration), 0, std::forward<Func> (callback), std::forward<Args> (args)...);
        }
 
        template <class Rep, class Period, typename Func, typename... Args>
        ssize_t setInterval (std::chrono::duration<Rep, Period> duration, Func&& callback, Args&&... args) noexcept
        {
            const uint64_t ticks = ticksFor (duration);
            return createTimer (ticks, ticks, std::forward<Func> (callback), std::forward<Args> (args)...);
        }
 
        int cancel (ssize_t id, bool sync = false) noexcept
        {
            Node* node = resolve (id);
 
            if (JOIN_UNLIKELY (node == nullptr))
            {
                lastError = make_error_code (Errc::NotFound);
                return -1;
            }
 
            if (JOIN_LIKELY (_runner.isRunnerThread ()))
            {
                disarmTimer (node);
                return 0;
            }
 
            if (JOIN_UNLIKELY (writeCommand ({CommandType::Disarm, placeOf (id), occupantOf (id)}) == -1))
            {
                return -1;  // LCOV_EXCL_LINE
            }
 
            _runner.flush (*this);
 
            if (JOIN_UNLIKELY (sync))
            {
                Backoff backoff;
                while (resolve (id) != nullptr)
                {
                    if (JOIN_UNLIKELY (_runner.flush (*this) == -1))
                    {
                        lastError = make_error_code (Errc::OperationFailed);
                        return -1;
                    }
 
                    backoff ();
                }
            }
 
            return 0;
        }
 
        bool active (ssize_t id) const noexcept
        {
            return resolve (id) != nullptr;
        }
 
        std::chrono::nanoseconds remaining (ssize_t id) const noexcept
        {
            const Node* node = resolve (id);
 
            if (JOIN_UNLIKELY (node == nullptr))
            {
                return std::chrono::nanoseconds::zero ();
            }
 
            const uint64_t tick = toTick (ClockPolicy::now ());
            const uint64_t deadline = node->deadline.load (std::memory_order_acquire);
 
            if (JOIN_UNLIKELY (resolve (id) == nullptr))
            {
                return std::chrono::nanoseconds::zero ();  // LCOV_EXCL_LINE
            }
 
            return std::chrono::nanoseconds ((deadline > tick + 1) ? ((deadline - tick - 1) * TickNs) : 0);
        }
 
        std::chrono::nanoseconds interval (ssize_t id) const noexcept
        {
            const Node* node = resolve (id);
 
            if (JOIN_UNLIKELY (node == nullptr))
            {
                return std::chrono::nanoseconds::zero ();
            }
 
            const uint64_t ticks = node->interval.load (std::memory_order_acquire);
 
            if (JOIN_UNLIKELY (resolve (id) == nullptr))
            {
                return std::chrono::nanoseconds::zero ();  // LCOV_EXCL_LINE
            }
 
            return std::chrono::nanoseconds (ticks * TickNs);
        }
 
        bool oneShot (ssize_t id) const noexcept
        {
            const Node* node = resolve (id);
 
            if (JOIN_UNLIKELY (node == nullptr))
            {
                return false;
            }
 
            const uint64_t ticks = node->interval.load (std::memory_order_acquire);
 
            return (resolve (id) != nullptr) && (ticks == 0);
        }
 
#ifdef JOIN_HAS_NUMA
        int mbind (int numa) const noexcept
        {
            if ((_allocator.mbind (numa) == -1) || (_commands.mbind (numa) == -1))
            {
                return -1;
            }
 
            return _generationsMem.mbind (numa);
        }
#endif
 
        int mlock () const noexcept
        {
            if ((_allocator.mlock () == -1) || (_commands.mlock () == -1))
            {
                return -1;  // LCOV_EXCL_LINE
            }
 
            return _generationsMem.mlock ();
        }
 
        const RunPolicy& runner () const noexcept
        {
            return _runner;
        }
 
        static constexpr std::chrono::nanoseconds resolution () noexcept
        {
            return std::chrono::nanoseconds (TickNs);
        }
 
        static constexpr size_t capacity () noexcept
        {
            return Capacity;
        }
 
    private:
        enum class CommandType : uint8_t
        {
            Arm,    
            Disarm, 
        };
 
        enum class State : uint8_t
        {
            Idle,      
            Linked,    
            Firing,    
            Cancelled, 
        };
 
        struct Node
        {
            Node* prev = nullptr;
 
            Node* next = nullptr;
 
            std::atomic_uint64_t deadline{0};
 
            std::atomic_uint64_t interval{0};
 
            uint16_t slot = 0;
 
            uint8_t level = 0;
 
            State state = State::Idle;
 
            Function<void (), 48> callback;
        };
 
        static_assert (sizeof (Node) == 128, "node must fill exactly two cache lines");
 
        struct Command
        {
            CommandType type;  
            uint32_t place;    
            uint32_t occupant; 
        };
 
        uint64_t advance () noexcept
        {
            readCommands ();
 
            if (_armed == 0)
            {
                return 0;
            }
 
            const uint64_t target = toTick (ClockPolicy::now ());
            const uint64_t start = _tick;
 
            if (JOIN_UNLIKELY (target < start))
            {
                return 0;  // LCOV_EXCL_LINE
            }
 
            if (target >= _due)
            {
                uint64_t tick = start;
                uint64_t next;
 
                while ((next = nextTick (tick)) <= target)
                {
                    tick = next;
                    _tick = tick;
 
                    if (JOIN_UNLIKELY ((tick & _slotMask) == 0))
                    {
                        cascade (tick);
                    }
 
                    fire (tick & _slotMask);
                }
 
                _due = next;
            }
 
            _tick = target;
 
            return target - start;
        }
 
        std::chrono::nanoseconds next () const noexcept
        {
            if (_armed == 0)
            {
                return std::chrono::nanoseconds::max ();
            }
 
            const uint64_t now = static_cast<uint64_t> (ClockPolicy::now ().time_since_epoch ().count ());
            const uint64_t tick = now / TickNs;
            const uint64_t ticks = (_due > tick) ? (_due - tick) : 0;
            const uint64_t limit = static_cast<uint64_t> (std::numeric_limits<int64_t>::max () / 2) / TickNs;
            const uint64_t elapsed = (ticks != 0) ? (now % TickNs) : 0;
 
            return std::chrono::nanoseconds (((ticks < limit) ? ticks : limit) * TickNs - elapsed);
        }
 
        template <typename Func, typename... Args>
        ssize_t createTimer (uint64_t ticks, uint64_t interval, Func&& callback, Args&&... args) noexcept
        {
            void* ptr = _allocator.tryAllocate (sizeof (Node));
 
            if (JOIN_UNLIKELY (ptr == nullptr))
            {
                lastError = make_error_code (Errc::OutOfMemory);
                return -1;
            }
 
            Node* node = static_cast<Node*> (ptr);
            node->callback = std::bind (std::forward<Func> (callback), std::forward<Args> (args)...);
            node->interval.store (interval, std::memory_order_release);
            node->state = State::Idle;
            const uint64_t now = toTick (ClockPolicy::now ());
            node->deadline.store (now + ticks + 1, std::memory_order_release);
 
            const uint32_t place = _allocator.getIndex (node);
            const uint32_t occupant = _generations[place].load (std::memory_order_relaxed) & _occupantMask;
 
            if (JOIN_LIKELY (_runner.isRunnerThread ()))
            {
                if (_armed++ == 0)
                {
                    _tick = now;
                    _due = std::numeric_limits<uint64_t>::max ();
                }
 
                armTimer (node);
                return makeId (occupant, place);
            }
 
            if (JOIN_UNLIKELY (writeCommand ({CommandType::Arm, place, occupant}) == -1))
            {
                // LCOV_EXCL_START
                releaseTimer (node);
                return -1;
                // LCOV_EXCL_STOP
            }
 
            _runner.flush (*this);
 
            return makeId (occupant, place);
        }
 
        void armTimer (Node* node, bool sameTick = false) noexcept
        {
            const uint64_t tick = _tick;
            const uint64_t deadline = node->deadline.load (std::memory_order_relaxed);
            const uint64_t horizon = tick + ((uint64_t (1) << (_levels * _slotBits)) - 1);
            const uint64_t due = (deadline > tick) ? deadline : (sameTick ? tick : (tick + 1));
            const uint64_t expires = (due < horizon) ? due : horizon;
            const uint64_t delta = (expires > tick) ? (expires - tick) : 1;
 
            size_t level = static_cast<size_t> (63 - __builtin_clzll (delta)) / _slotBits;
            level = (level < _levels) ? level : (_levels - 1);
 
            const size_t slot = (expires >> (level * _slotBits)) & _slotMask;
 
            node->level = static_cast<uint8_t> (level);
            node->slot = static_cast<uint16_t> (slot);
            node->state = State::Linked;
            node->prev = nullptr;
            node->next = _wheel[level][slot];
 
            if (node->next != nullptr)
            {
                node->next->prev = node;
            }
 
            _wheel[level][slot] = node;
            _occupied[level][slot >> 6] |= uint64_t (1) << (slot & 63);
 
            const uint64_t event = (expires >> (level * _slotBits)) << (level * _slotBits);
 
            if (event < _due)
            {
                _due = event;
            }
        }
 
        void disarmTimer (Node* node) noexcept
        {
            if (JOIN_UNLIKELY (node->state == State::Firing))
            {
                node->state = State::Cancelled;
                return;
            }
 
            if (JOIN_UNLIKELY (node->state == State::Cancelled))
            {
                return;
            }
 
            if (JOIN_LIKELY (node->state == State::Linked))
            {
                unlink (node);
                --_armed;
            }
 
            releaseTimer (node);
        }
 
        void releaseTimer (Node* node) noexcept
        {
            const uint32_t place = _allocator.getIndex (node);
 
            node->callback.reset ();
            _generations[place].fetch_add (1, std::memory_order_release);
            _allocator.deallocate (node);
        }
 
        uint64_t nextTick (uint64_t tick) const noexcept
        {
            uint64_t next = std::numeric_limits<uint64_t>::max ();
 
            for (size_t level = 0; level < _levels; ++level)
            {
                const size_t shift = level * _slotBits;
                const uint64_t base = (tick >> shift) + 1;
                const size_t start = base & _slotMask;
                const size_t word = start >> 6;
                uint64_t bits = _occupied[level][word] & (~uint64_t (0) << (start & 63));
 
                for (size_t i = 0; i <= _slots / 64; ++i)
                {
                    if (bits != 0)
                    {
                        const size_t slot = ((word + i) % (_slots / 64)) * 64 + __builtin_ctzll (bits);
                        const uint64_t candidate = (base + ((slot - start) & _slotMask)) << shift;
                        next = (candidate < next) ? candidate : next;
                        break;
                    }
 
                    bits = _occupied[level][(word + i + 1) % (_slots / 64)];
                }
 
                if (next < (((tick >> (shift + _slotBits)) + 1) << (shift + _slotBits)))
                {
                    break;
                }
            }
 
            return next;
        }
 
        void cascade (uint64_t tick) noexcept
        {
            for (size_t level = 1; level < _levels; ++level)
            {
                const size_t slot = (tick >> (level * _slotBits)) & _slotMask;
                Node* node;
 
                while ((node = _wheel[level][slot]) != nullptr)
                {
                    unlink (node);
                    armTimer (node, true);
                }
 
                if (slot != 0)
                {
                    break;
                }
            }
        }
 
        void fire (size_t slot) noexcept
        {
            Node* node = _wheel[0][slot];
 
            while (node != nullptr)
            {
                unlink (node);
 
                node->state = State::Firing;
                node->callback ();
 
                const bool cancelled = (node->state == State::Cancelled);
                node->state = State::Idle;
 
                if (JOIN_LIKELY (!cancelled && (node->interval.load (std::memory_order_relaxed) != 0)))
                {
                    node->deadline.fetch_add (node->interval.load (std::memory_order_relaxed),
                                              std::memory_order_release);
                    armTimer (node);
                }
                else
                {
                    releaseTimer (node);
                    --_armed;
                }
 
                node = _wheel[0][slot];
            }
        }
 
        void unlink (Node* node) noexcept
        {
            if (node->prev != nullptr)
            {
                node->prev->next = node->next;
            }
            else
            {
                _wheel[node->level][node->slot] = node->next;
 
                if (node->next == nullptr)
                {
                    _occupied[node->level][node->slot >> 6] &= ~(uint64_t (1) << (node->slot & 63));
                }
            }
 
            if (node->next != nullptr)
            {
                node->next->prev = node->prev;
            }
 
            node->prev = nullptr;
            node->next = nullptr;
            node->state = State::Idle;
        }
 
        int writeCommand (const Command& cmd) noexcept
        {
            if (JOIN_UNLIKELY (_commands.tryPush (cmd) == -1))
            {
                // LCOV_EXCL_START
                lastError = make_error_code (Errc::TemporaryError);
                return -1;
                // LCOV_EXCL_STOP
            }
 
            return 0;
        }
 
        void readCommands () noexcept
        {
            Command cmd;
 
            while (_commands.tryPop (cmd) == 0)
            {
                processCommand (cmd);
            }
        }
 
        void processCommand (const Command& cmd) noexcept
        {
            if (JOIN_LIKELY ((_generations[cmd.place].load (std::memory_order_acquire) & _occupantMask) ==
                             cmd.occupant))
            {
                Node* node = static_cast<Node*> (_allocator.getPtr (cmd.place));
 
                if (cmd.type == CommandType::Arm)
                {
                    if (_armed++ == 0)
                    {
                        _tick = toTick (ClockPolicy::now ());
                        _due = std::numeric_limits<uint64_t>::max ();
                    }
 
                    armTimer (node);
                }
                else
                {
                    disarmTimer (node);
                }
            }
        }
 
        Node* resolve (ssize_t id) const noexcept
        {
            if (JOIN_UNLIKELY (id <= 0))
            {
                return nullptr;
            }
 
            const uint32_t place = placeOf (id);
 
            if (JOIN_UNLIKELY (place >= Capacity))
            {
                return nullptr;
            }
 
            if (JOIN_UNLIKELY ((_generations[place].load (std::memory_order_acquire) & _occupantMask) !=
                               occupantOf (id)))
            {
                return nullptr;
            }
 
            return static_cast<Node*> (_allocator.getPtr (place));
        }
 
        static constexpr ssize_t makeId (uint32_t occupant, uint32_t place) noexcept
        {
            return (static_cast<ssize_t> (occupant & _occupantMask) << 32) | (static_cast<ssize_t> (place) + 1);
        }
 
        static constexpr uint32_t placeOf (ssize_t id) noexcept
        {
            return static_cast<uint32_t> ((id & 0xFFFFFFFF) - 1);
        }
 
        static constexpr uint32_t occupantOf (ssize_t id) noexcept
        {
            return static_cast<uint32_t> (id >> 32) & _occupantMask;
        }
 
        static uint64_t toTick (typename ClockPolicy::TimePoint timePoint) noexcept
        {
            return static_cast<uint64_t> (timePoint.time_since_epoch ().count ()) / TickNs;
        }
 
        template <class Rep, class Period>
        static uint64_t ticksFor (std::chrono::duration<Rep, Period> duration) noexcept
        {
            auto ns = std::chrono::duration_cast<std::chrono::nanoseconds> (duration);
 
            if (ns.count () < 1)
            {
                return 1;
            }
 
            return (static_cast<uint64_t> (ns.count ()) + TickNs - 1) / TickNs;
        }
 
        static constexpr size_t _slots = 256;
 
        static constexpr size_t _slotBits = 8;
 
        static constexpr size_t _slotMask = _slots - 1;
 
        static constexpr size_t _levels = 6;
 
        static constexpr uint32_t _occupantMask = 0x7FFFFFFF;
 
        static constexpr size_t _queueSize = 2 * Capacity;
 
        LocalMem::SlabAllocator<Capacity, sizeof (Node)> _allocator;
 
        LocalMem::Mpsc::Queue<Command> _commands;
 
        LocalMem _generationsMem;
 
        std::atomic_uint32_t* const _generations;
 
        size_t _armed = 0;
 
        uint64_t _occupied[_levels][_slots / 64] = {};
 
        uint64_t _due = std::numeric_limits<uint64_t>::max ();
 
        alignas (64) Node* _wheel[_levels][_slots] = {};
 
        alignas (64) uint64_t _tick = 0;
 
        ClockPolicy _clock;
 
        RunPolicy _runner;
    };
 
    template <class ClockPolicy>
    class SpinPolicy
    {
    public:
        explicit SpinPolicy (int core = -1, int prio = 0) noexcept
        : _core (core)
        , _priority (prio)
        {
        }
 
        SpinPolicy (const SpinPolicy& other) = delete;
 
        SpinPolicy& operator= (const SpinPolicy& other) = delete;
 
        SpinPolicy (SpinPolicy&& other) = delete;
 
        SpinPolicy& operator= (SpinPolicy&& other) = delete;
 
        ~SpinPolicy () noexcept
        {
            stop ();
        }
 
        template <class Wheel>
        int start (Wheel& wheel, std::chrono::nanoseconds)
        {
            _running.store (true, std::memory_order_release);
            _dispatcher = Thread (_core, _priority, &SpinPolicy::loop<Wheel>, this, &wheel);
 
            Backoff backoff;
            while (_threadId.load (std::memory_order_acquire) == Thread::invalidId)
            {
                backoff ();
            }
 
            return 0;
        }
 
        void stop () noexcept
        {
            _running.store (false, std::memory_order_release);
 
            if (JOIN_UNLIKELY (isRunnerThread ()))
            {
                return;  // LCOV_EXCL_LINE
            }
 
            Backoff backoff;
            while (_threadId.load (std::memory_order_acquire) != Thread::invalidId)
            {
                backoff ();
            }
        }
 
        bool isRunnerThread () const noexcept
        {
            return _threadId.load (std::memory_order_acquire) == Thread::currentId ();
        }
 
        template <class Wheel>
        int flush (Wheel&) noexcept
        {
            return 0;
        }
 
        int affinity () const noexcept
        {
            return _dispatcher.affinity ();
        }
 
        int priority () const noexcept
        {
            return _dispatcher.priority ();
        }
 
        pthread_t handle () const noexcept
        {
            return _dispatcher.handle ();
        }
 
    private:
        template <class Wheel>
        void loop (Wheel* wheel) noexcept
        {
            _threadId.store (Thread::currentId (), std::memory_order_release);
 
            while (JOIN_LIKELY (_running.load (std::memory_order_relaxed)))
            {
                wheel->advance ();
 
                if (wheel->_armed == 0)
                {
#if defined(__x86_64__) || defined(__i386__)
                    _mm_pause ();
#elif defined(__aarch64__) || defined(__arm__)
                    __asm__ __volatile__ ("isb" ::: "memory");
#endif
                }
            }
 
            _threadId.store (Thread::invalidId, std::memory_order_release);
        }
 
        alignas (64) std::atomic<pid_t> _threadId{Thread::invalidId};
 
        alignas (64) std::atomic_bool _running{false};
 
        int _core = -1;
 
        int _priority = 0;
 
        Thread _dispatcher;
    };
}
 
#endif