lock_free_queue.hpp

浏览该文件的文档.

#ifndef MSTL_CORE_ASYNC_LOCK_FREE_QUEUE_HPP__
#define MSTL_CORE_ASYNC_LOCK_FREE_QUEUE_HPP__

 
#include "MSTL/core/memory/unique_ptr.hpp"
#include "MSTL/core/async/atomic.hpp"
MSTL_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(_MSTL 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) {
        _MSTL 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, _MSTL memory_order_relaxed);
    }

    unique_ptr<T> pop() {
        counted_node_ptr old_head = head.load(_MSTL 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 _MSTL 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, _MSTL 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(_MSTL 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 _MSTL 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, _MSTL memory_order_relaxed);
                return _MSTL unique_ptr<T>(res);
            }
 
            ptr->release_ref();
        }
 
        return _MSTL make_unique<T>();
    }

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

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

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