lock_free_queue.hpp

浏览该文件的文档.

#ifndef NEFORCE_CORE_ASYNC_LOCK_FREE_QUEUE_HPP__
#define NEFORCE_CORE_ASYNC_LOCK_FREE_QUEUE_HPP__

 
#include "NeForce/core/async/atomic.hpp"
#include "NeForce/core/memory/unique_ptr.hpp"
NEFORCE_BEGIN_NAMESPACE__



template <typename T>
class lock_free_queue {
private:
    struct node_counter {
        uint32_t internal_count : 30;
        uint32_t external_counters : 2;
    };
 
    struct node;

    struct counted_node_ptr {
        int external_count = 0; 
        node* ptr = nullptr;    

        counted_node_ptr() noexcept {}
    };

    struct node {
        atomic<T*> data;               
        atomic<node_counter> count;    
        atomic<counted_node_ptr> next; 

        explicit node(int external_count = 2) {
            node_counter new_count;
            new_count.internal_count = 0;
            new_count.external_counters = external_count;
            count.store(new_count);
 
            counted_node_ptr node_ptr;
            node_ptr.ptr = nullptr;
            node_ptr.external_count = 0;
 
            next.store(node_ptr);
        }

        void release_ref() {
            node_counter old_counter = count.load(memory_order_relaxed);
            node_counter new_counter;
            do {
                new_counter = old_counter;
                --new_counter.internal_count;
            } while (!count.compare_exchange_strong(old_counter, new_counter, memory_order_acquire,
                                                    memory_order_relaxed));
            if (!new_counter.internal_count && !new_counter.external_counters) {
                delete this;
                destruct_count.fetch_add(1);
            }
        }
    };
 
private:
    static atomic<int> destruct_count;
    static atomic<int> construct_count;
 
    atomic<counted_node_ptr> head; 
    atomic<counted_node_ptr> tail; 
    atomic<size_t> push_count_{0}; 
    atomic<size_t> pop_count_{0};  
 
private:
    void set_new_tail(counted_node_ptr& old_tail, counted_node_ptr const& new_tail) {
        node* const current_tail_ptr = old_tail.ptr;
        while (!tail.compare_exchange_weak(old_tail, new_tail) && old_tail.ptr == current_tail_ptr) {
            this_thread::relax();
        }
        if (old_tail.ptr == current_tail_ptr) {
            lock_free_queue::free_external_counter(old_tail);
        } else {
            current_tail_ptr->release_ref();
        }
    }

    static void free_external_counter(counted_node_ptr& old_node_ptr) {
        node* const ptr = old_node_ptr.ptr;
        int const count_increase = old_node_ptr.external_count - 2;
        node_counter old_counter = ptr->count.load(_NEFORCE memory_order_relaxed);
        node_counter new_counter;
        do {
            new_counter = old_counter;
            --new_counter.external_counters;
            new_counter.internal_count += count_increase;
        } while (!ptr->count.compare_exchange_strong(old_counter, new_counter, memory_order_acquire,
                                                     memory_order_relaxed));
        if (!new_counter.internal_count && !new_counter.external_counters) {
            destruct_count.fetch_add(1);
            delete ptr;
        }
    }

    static void increase_external_count(atomic<counted_node_ptr>& counter, counted_node_ptr& old_counter) {
        counted_node_ptr new_counter;
        do {
            new_counter = old_counter;
            ++new_counter.external_count;
        } while (
                !counter.compare_exchange_strong(old_counter, new_counter, memory_order_acquire, memory_order_relaxed));
        old_counter.external_count = new_counter.external_count;
    }
 
public:
    lock_free_queue() {
        counted_node_ptr new_next;
        new_next.ptr = new node();
        new_next.external_count = 1;
        tail.store(new_next);
        head.store(new_next);
    }

    ~lock_free_queue() {
        while (pop()) {
            this_thread::relax();
        }
        auto head_counted_node = head.load();
        delete head_counted_node.ptr;
    }

    void push(T new_value) {
        unique_ptr<T> new_data(new T(new_value));
        counted_node_ptr new_next;
        new_next.ptr = new node;
        new_next.external_count = 1;
        counted_node_ptr old_tail = tail.load();
        for (;;) {
            lock_free_queue::increase_external_count(tail, old_tail);
            T* old_data = nullptr;
            if (old_tail.ptr->data.compare_exchange_strong(old_data, new_data.get())) {
                counted_node_ptr old_next;
                counted_node_ptr now_next = old_tail.ptr->next.load();
                if (!old_tail.ptr->next.compare_exchange_strong(old_next, new_next)) {
                    delete new_next.ptr;
                    new_next = old_next;
                }
                this->set_new_tail(old_tail, new_next);
                new_data.release();
                break;
            }
            counted_node_ptr old_next;
            if (old_tail.ptr->next.compare_exchange_strong(old_next, new_next)) {
                old_next = new_next;
                new_next.ptr = new node;
            }
            this->set_new_tail(old_tail, old_next);
        }
        ++construct_count;
        push_count_.fetch_add(1, memory_order_relaxed);
    }

    unique_ptr<T> pop() {
        counted_node_ptr old_head = head.load(memory_order_relaxed);
        for (;;) {
            lock_free_queue::increase_external_count(head, old_head);
            node* const ptr = old_head.ptr;
            if (ptr == tail.load().ptr) {
                ptr->release_ref();
                return _NEFORCE make_unique<T>();
            }
            counted_node_ptr next = ptr->next.load();
            if (head.compare_exchange_strong(old_head, next)) {
                T* res = ptr->data.exchange(nullptr);
                lock_free_queue::free_external_counter(old_head);
                pop_count_.fetch_add(1, memory_order_relaxed);
                return unique_ptr<T>(res);
            }
            ptr->release_ref();
        }
    }

    unique_ptr<T> try_pop() {
        counted_node_ptr old_head{};
 
        for (int retry = 0; retry < 3; ++retry) {
            old_head = head.load(memory_order_relaxed);
            lock_free_queue::increase_external_count(head, old_head);
            node* ptr = old_head.ptr;
 
            if (ptr == tail.load().ptr) {
                ptr->release_ref();
                return make_unique<T>();
            }
 
            counted_node_ptr next = ptr->next.load();
            if (head.compare_exchange_strong(old_head, next)) {
                T* res = ptr->data.exchange(nullptr);
                lock_free_queue::free_external_counter(old_head);
                pop_count_.fetch_add(1, _NEFORCE memory_order_relaxed);
                return unique_ptr<T>(res);
            }
 
            ptr->release_ref();
        }
 
        return make_unique<T>();
    }

    bool empty() const noexcept {
        const counted_node_ptr head_ptr = head.load(memory_order_acquire);
        const counted_node_ptr tail_ptr = tail.load(memory_order_acquire);
        return head_ptr.ptr == tail_ptr.ptr;
    }

    size_t size() const noexcept {
        const size_t push_cnt = push_count_.load(memory_order_relaxed);
        const size_t pop_cnt = pop_count_.load(memory_order_relaxed);
        return push_cnt - pop_cnt;
    }

    void clear() {
        while (unique_ptr<T> ptr = this->pop()) {
            this_thread::relax();
        }
    }
};
 
template <typename T>
atomic<int> lock_free_queue<T>::destruct_count{0};
 
template <typename T>
atomic<int> lock_free_queue<T>::construct_count{0};
 // LockFreeQueue
 // AsyncComponents
 
NEFORCE_END_NAMESPACE__
#endif // NEFORCE_CORE_ASYNC_LOCK_FREE_QUEUE_HPP__