io_context.hpp

浏览该文件的文档.

#ifndef NEFORCE_CORE_ASYNC_IO_CONTEXT_HPP__
#define NEFORCE_CORE_ASYNC_IO_CONTEXT_HPP__

 
#include "NeForce/core/async/lock_free_queue.hpp"
#include "NeForce/core/container/flat_unordered_map.hpp"
#include "NeForce/core/container/vector.hpp"
#include "NeForce/core/exception/error_code.hpp"
#include "NeForce/core/functional/function.hpp"
#include "NeForce/core/memory/shared_ptr.hpp"
NEFORCE_BEGIN_NAMESPACE__



NEFORCE_INLINE17 constexpr uint32_t epoll_in = 0x001;
NEFORCE_INLINE17 constexpr uint32_t epoll_out = 0x004;
NEFORCE_INLINE17 constexpr uint32_t epoll_et = 0x80000000;

class NEFORCE_API io_context {
public:
    using handler_type = function<void()>;
    using fd_callback = function<void(int fd, uint32_t events, error_code ec)>;
    using timer_callback = function<void()>;
 
    using native_handle_type =
#ifdef NEFORCE_PLATFORM_WINDOWS
            uintptr_t; 
#else
            int; 
#endif
 
    class executor;
    class work;

    io_context();

    ~io_context();
 
    io_context(const io_context&) = delete;
    io_context& operator=(const io_context&) = delete;

    executor get_executor() noexcept;

    void post(handler_type handler);

    void dispatch(handler_type handler);

    size_t run();

    void run_pool(size_t n);

    size_t run_one(int timeout_ms = -1);

    size_t poll();

    size_t poll_one() { return run_one(0); }

    void stop();

    bool stopped() const noexcept { return stopped_.load(memory_order_acquire); }

    void restart() { stopped_.store(false, memory_order_release); }

    size_t schedule_timer(uint64_t delay_ms, timer_callback handler);

    bool cancel_timer(size_t timer_id);

    void add_fd(native_handle_type fd, uint32_t events, fd_callback cb, bool edge_triggered = true);

    void mod_fd(native_handle_type fd, uint32_t events, bool edge_triggered = true);

    void remove_fd(native_handle_type fd);
 
private:
    friend class file_async;
 
    using file_completion_cb = function<void(error_code, size_t, void* overlapped)>;

    struct timer_entry {
        size_t id;
        uint64_t deadline_ms;
        timer_callback callback;
        bool operator>(const timer_entry& other) const { return deadline_ms > other.deadline_ms; }
    };

    struct fd_info {
        native_handle_type fd;
        uint32_t events;
        fd_callback callback;
    };

    atomic<int> running_{0};
    atomic<bool> stopped_{false};
    atomic<size_t> outstanding_work_{0};
    size_t next_timer_id_{1};

    flat_unordered_map<native_handle_type, fd_info> fd_map_;
    vector<timer_entry> timer_heap_;
    mutable mutex timer_mutex_;

    lock_free_queue<handler_type> external_queue_;
    atomic<size_t> external_queue_count_{0};
 
#ifdef NEFORCE_PLATFORM_WINDOWS
    void* iocp_handle_;
    void* wake_event_;
    flat_unordered_map<native_handle_type, void*> fd_events_;
    mutex fd_mutex_;
    thread monitor_thread_;
    atomic<bool> monitor_running_{false};

    flat_unordered_map<uintptr_t, file_completion_cb> file_completions_;
 
    void monitor_loop();
 
    void register_file_completion(uintptr_t key, file_completion_cb cb);
    void unregister_file_completion(uintptr_t key);
#else
    int epoll_fd_;
    int wake_fd_;
#endif

    vector<thread> pool_threads_;
    mutex pool_mutex_;

    uint64_t next_timer_deadline() const;

    void process_timers(size_t max_count = numeric_traits<size_t>::max());

    void wake();

    size_t drain_handlers(size_t max_count = 256);
};

class io_context::executor {
public:
    executor(const executor& other) noexcept = default;
    executor& operator=(const executor& other) noexcept = default;

    void execute(handler_type handler) const { ctx_->post(move(handler)); }

    NEFORCE_NODISCARD io_context& context() const noexcept { return *ctx_; }

    bool operator==(const executor& other) const noexcept { return ctx_ == other.ctx_; }

    bool operator!=(const executor& other) const noexcept { return ctx_ != other.ctx_; }
 
private:
    friend class io_context;
 
    executor(io_context& ctx) noexcept :
    ctx_(&ctx) {}
 
    io_context* ctx_;
};

class io_context::work {
private:
    io_context* ctx_;
 
public:
    explicit work(io_context& ctx) :
    ctx_(&ctx) {
        ctx_->outstanding_work_.fetch_add(1, memory_order_relaxed);
    }

    ~work() { ctx_->outstanding_work_.fetch_sub(1, memory_order_relaxed); }
 
    work(const work&) = delete;
    work& operator=(const work&) = delete;
};
 // IOContext
 // AsyncComponents
 
NEFORCE_END_NAMESPACE__
#endif // NEFORCE_CORE_ASYNC_IO_CONTEXT_HPP__