timer.hpp

浏览该文件的文档.

#ifndef NEFORCE_CORE_ASYNC_TIMER_HPP__
#define NEFORCE_CORE_ASYNC_TIMER_HPP__

 
#include "NeForce/core/async/atomic.hpp"
#include "NeForce/core/async/condition_variable.hpp"
#include "NeForce/core/async/packaged_task.hpp"
#include "NeForce/core/async/thread.hpp"
#include "NeForce/core/container/map.hpp"
#include "NeForce/core/container/set.hpp"
#include "NeForce/core/functional/function.hpp"
#include "NeForce/core/memory/shared_ptr.hpp"
NEFORCE_BEGIN_NAMESPACE__



template <typename Clock>
class timer_scheduler {
public:
    using clock_type = Clock;                           
    using time_point = typename clock_type::time_point; 
    using duration = typename clock_type::duration;     
    using token = size_t;                               
    using handler_type = function<void()>;              
 
private:
    struct node {
        time_point expire;    
        token id;             
        handler_type handler; 
 
        node(time_point exp, const token tid, handler_type&& h) :
        expire(exp),
        id(tid),
        handler(move(h)) {}
 
        node(const node&) = default;
        node& operator=(const node&) = default;
        node(node&&) = default;
        node& operator=(node&&) = default;
 
        ~node() = default;

        bool operator<(const node& other) const {
            if (expire < other.expire) {
                return true;
            }
            if (expire > other.expire) {
                return false;
            }
            return id < other.id;
        }
    };
 
    set<node> nodes_;                                   
    map<token, typename set<node>::iterator> node_map_; 
    map<token, shared_ptr<atomic<bool>>> cancel_flags_; 
    map<token, shared_ptr<promise<void>>> promises_;    
 
    thread thread_;               
    mutable mutex mutex_;         
    condition_variable cv_;       
    token next_id_{1};            
    atomic<bool> stopped_{false}; 
 
    friend class thread_pool;
 
private:
    void run() {
        while (!stopped_.load()) {
            unique_lock<mutex> lock(mutex_);
 
            if (nodes_.empty()) {
                cv_.wait_for(lock, 500_ms, [this] { return stopped_.load() || !nodes_.empty(); });
                if (stopped_.load()) {
                    break;
                }
            }
 
            time_point now = clock_type::now();
            while (!nodes_.empty() && nodes_.begin()->expire <= now) {
                auto it = nodes_.begin();
                node current_node = *it;
                nodes_.erase(it);
                node_map_.erase(current_node.id);
 
                auto flag_it = cancel_flags_.find(current_node.id);
                const bool cancelled = (flag_it != cancel_flags_.end() && flag_it->second->load(memory_order_acquire));
 
                lock.unlock_quiet();
                if (!stopped_.load() && !cancelled) {
                    current_node.handler();
                }
                lock.lock_quiet();
 
                cancel_flags_.erase(current_node.id);
                promises_.erase(current_node.id);
                now = clock_type::now();
            }
 
            if (!nodes_.empty()) {
                auto expire_time = nodes_.begin()->expire;
                cv_.wait_until(lock, expire_time, [this] {
                    return stopped_.load(memory_order_acquire) || nodes_.empty() ||
                           nodes_.begin()->expire <= clock_type::now();
                });
            }
        }
    }
 
public:
    timer_scheduler() { thread_ = thread(&timer_scheduler::run, this); }

    ~timer_scheduler() { stop(); }

    void stop() {
        stopped_.store(true);
        cv_.notify_one();
        if (thread_.joinable()) {
            thread_.join();
        }
    }
 
    timer_scheduler(const timer_scheduler&) = delete;
    timer_scheduler& operator=(const timer_scheduler&) = delete;
    timer_scheduler(timer_scheduler&&) = delete;
    timer_scheduler& operator=(timer_scheduler&&) = delete;

    token add_task(time_point expire, handler_type&& handler) {
        unique_lock<mutex> lock(mutex_);
        token id = next_id_++;
 
        auto flag = make_shared<atomic<bool>>(false);
        auto promise = make_shared<_NEFORCE promise<void>>();
        handler_type wrapped = [flag, h = move(handler), promise]() {
            if (!flag->load(memory_order_acquire)) {
                h();
            }
            promise->set_value();
        };
 
        const bool is_earliest = nodes_.empty() || expire < nodes_.begin()->expire;
 
        node new_node(expire, id, _NEFORCE move(wrapped));
        auto result = nodes_.insert(new_node);
        node_map_[id] = result.first;
        cancel_flags_[id] = move(flag);
        promises_[id] = promise;
 
        if (is_earliest) {
            cv_.notify_one();
        }
        lock.unlock_quiet();
        return id;
    }

    bool cancel(token id) {
        unique_lock<mutex> lock(mutex_);
 
        const auto flag_it = cancel_flags_.find(id);
        if (flag_it == cancel_flags_.end()) {
            return false;
        }
        flag_it->second->store(true, memory_order_release);
 
        auto node_it = node_map_.find(id);
        if (node_it != node_map_.end()) {
            const bool is_earliest = (node_it->second == nodes_.begin());
            nodes_.erase(node_it->second);
            node_map_.erase(node_it);
 
            if (is_earliest) {
                cv_.notify_one();
            }
 
            cancel_flags_.erase(flag_it);
            promises_.erase(id);
        } else {
            const auto prom_it = promises_.find(id);
            shared_ptr<promise<void>> prom;
            if (prom_it != promises_.end()) {
                prom = prom_it->second;
                promises_.erase(prom_it);
            }
            lock.unlock_quiet();
 
            if (prom) {
                prom->get_future().wait();
            }
        }
 
        return true;
    }

    void cancel_all() {
        unique_lock<mutex> lock(mutex_);
        for (const auto& cancel_flag: cancel_flags_) {
            cancel_flag.second->store(true, memory_order_release);
        }
        nodes_.clear();
        node_map_.clear();
        cancel_flags_.clear();
        lock.unlock_quiet();
        cv_.notify_one();
    }

    NEFORCE_NODISCARD size_t size() const {
        lock<mutex> lock(mutex_);
        return nodes_.size();
    }

    NEFORCE_NODISCARD bool is_pending(token id) const {
        lock<mutex> lock(mutex_);
        return node_map_.find(id) != node_map_.end();
    }
};

template <typename Clock>
class basic_timer {
public:
    using clock_type = Clock;                                           
    using time_point = typename clock_type::time_point;                 
    using duration = typename clock_type::duration;                     
    using token = typename timer_scheduler<Clock>::token;               
    using handler_type = typename timer_scheduler<Clock>::handler_type; 
 
private:
    shared_ptr<timer_scheduler<Clock>> scheduler_; 
    token task_id_{0};                             
    time_point expire_{clock_type::now()};         
 
public:
    basic_timer() :
    scheduler_(make_shared<timer_scheduler<Clock>>()) {}

    ~basic_timer() {
        if (scheduler_) {
            cancel();
        }
    }
 
    basic_timer(const basic_timer&) = delete;
    basic_timer& operator=(const basic_timer&) = delete;

    basic_timer(basic_timer&& other) noexcept :
    scheduler_(_NEFORCE move(other.scheduler_)),
    task_id_(other.task_id_),
    expire_(other.expire_) {
        other.task_id_ = 0;
    }

    basic_timer& operator=(basic_timer&& other) noexcept {
        if (_NEFORCE addressof(other) == this) {
            return *this;
        }
 
        cancel();
        scheduler_ = _NEFORCE move(other.scheduler_);
        task_id_ = other.task_id_;
        expire_ = other.expire_;
        other.task_id_ = 0;
 
        return *this;
    }

    void expires_at(const time_point& expiry_time) {
        cancel();
        expire_ = expiry_time;
    }

    void expires_after(const duration& expiry_duration) {
        cancel();
        expire_ = clock_type::now() + expiry_duration;
    }

    void expires_from_now(const int64_t ms) { expires_after(milliseconds(ms)); }

    NEFORCE_NODISCARD time_point expiry() const { return expire_; }

    NEFORCE_NODISCARD bool is_active() const { return task_id_ != 0 && scheduler_ && scheduler_->is_pending(task_id_); }

    template <typename WaitHandler>
    void async_wait(WaitHandler&& handler) {
        cancel();
        task_id_ = scheduler_->add_task(expire_, handler_type(_NEFORCE forward<WaitHandler>(handler)));
    }

    void cancel() {
        if (scheduler_ && task_id_ != 0) {
            scheduler_->cancel(task_id_);
            task_id_ = 0;
        }
    }
};

using steady_timer = basic_timer<steady_clock>;

using system_timer = basic_timer<system_clock>;
 // AsyncTimer
 // AsyncComponents
 
NEFORCE_END_NAMESPACE__
#endif // NEFORCE_CORE_ASYNC_TIMER_HPP__