C++并发编程----实现线无锁线程安全的数据结构（《C++ Concurrency in Action》 读书笔记）

Posted 2021-07-17 小丑快学习

无锁线程安全的栈

template<typename T>
class lock_free_stack
{
private:
    struct node
    {
        std::shared_ptr<T> data; // 1 指针获取数据 
        node* next;
        node(T const& data_) :
            data(std::make_shared<T>(data_))  
        {}
    };
    std::atomic<node*> head;
public:
	/*
	入栈操作
	*/
    void push(T const& data)
    {
        node* const new_node = new node(data); 
        new_node->next = head.load(); 
        //采用比较-交换技术
        while (!head.compare_exchange_weak(new_node->next, new_node));
    }
	/*
	出栈操作
	*/
	std::shared_ptr<T> pop()
    {
        node* old_head = head.load();
        while (old_head && // 3 在解引用前检查old_head是否为空指针，空代表栈空
            !head.compare_exchange_weak(old_head, old_head->next));
        return old_head ? old_head->data : std::shared_ptr<T>(); //当栈为空时应该返回空指针
    }
};

上述栈的头节点是原子类型的，因此出栈时对于head.compare_exchange_weak(new_node->next, new_node)而言，相当于原子的执行下列操作：

if ( head == new_node->next){
     head = new_node;
     return true;
}
else{
    new_node->next = head;
    return false;
}

也就是说，当没有其他线程修改head的指向时，才能将head指向待插入的节点，倘如此时有其他线程对head进行修改则head == new_node->next将不成立，那么将会重新插入，知道上述的if条件满足条件，节点成功插入，退出循环。

出栈操作则应该检查是否栈为空，如果栈为空，那么将会返回空指针。

节点回收
上述的栈pop操作中，并没有对数据进行删除，因此，这样将会导致内存泄露，但是如果每次在pop操作时就将相应的节点删除，那么有可能导致多个线程删除同一节点，这样必然会导致未定义行为，因此，要将线程安全的亲删除则应该在pop的访问中添加计数器，当其仅有一个线程访问时才能对其数据进行删除，这是我们需要维护一个删除列表，当只有一个线程进行出栈操作时，我们将这个链表说删除。
具体代码的修改如下：

template<typename T>
class lock_free_stack
{
private:
    struct node
    {
        std::shared_ptr<T> data; // 1 指针获取数据 
        node* next;
        node(T const& data_) :
            data(std::make_shared<T>(data_))  
        {}
    };

    std::atomic<node*> head;
    std::atomic<unsigned> threads_in_pop; //记录当前在进行pop（）操作的线程
    std::atomic<node*> to_be_deleted; //指向待删除的链表
    void try_reclaim(node* old_head); //尝试去删除该节点的或者整个待删除链表
    

    /*
     删除一个链表
    */
    static void delete_nodes(node* nodes)
    {
        while (nodes)
        {
            node* next = nodes->next;
            delete nodes;
            nodes = next;
        }
    }
    /*
        尝试去删除一个链表和一个节点
    */
    void try_reclaim(node* old_head)
    {
        if (threads_in_pop == 1) //如果当前的访问pop线程只有一个，则尝试去删除该节点和待删除链表
        {
            node* nodes_to_delete = to_be_deleted.exchange(nullptr);
            if (!--threads_in_pop) 
            {
                delete_nodes(nodes_to_delete); 
            }
            else if (nodes_to_delete) //访问上一条语句时如果有线程访问访问pop，则不能去删除一整条链表 
            {
                chain_pending_nodes(nodes_to_delete); // 将链表挂回待删除链表  
            }
            delete old_head; // 因为pop该节点时只有一个线程，因此删除该节点安全
        }
        else//如果多个线程访问pop，则将该节点挂到待删除链表
        {
            chain_pending_node(old_head); 
            --threads_in_pop;
        }
    }
    void chain_pending_nodes(node* nodes)
    {
        node* last = nodes;
        while (node* const next = last->next) // 让last指针指向链表的末尾
        {
            last = next;
        }
        chain_pending_nodes(nodes, last);
    }
    void chain_pending_nodes(node* first, node* last)
    {
        //采用头插法的将整个链表放回到待删除链表，
        //如果插入时有其他的线程操作过，则应该重新插入
        last->next = to_be_deleted;  
        while (!to_be_deleted.compare_exchange_weak(last->next, first));
    }
    void chain_pending_node(node* n)//插入单个节点
    {
        chain_pending_nodes(n, n); 
    }

public:
    void push(T const& data)
    {
        node* const new_node = new node(data); // 2 
        new_node->next = head.load(); // 3 
        while (!head.compare_exchange_weak(new_node->next, new_node));
        // 4
    }

    std::shared_ptr<T> pop()
    {
        ++threads_in_pop; // 2 在做事之前，计数值加1 
        node* old_head = head.load();
        while (old_head &&
            !head.compare_exchange_weak(old_head, old_head->next));
        std::shared_ptr<T> res;
        if (old_head)
        {
            res.swap(old_head->data); // 3 回收删除的节点 
        }
        try_reclaim(old_head); // 4 从节点中直接提取数据，而非拷贝指针 
        return res;
    }
};

测试：

lock_free_stack<int> stack;
void fun1() {
    for (int i = 0; i < 20; i++) {
        stack.push(i);
    }
    for (int i = 0; i < 5; i++) {
        cout << "thread1: " << *stack.pop() << endl;
    }
}

void fun2() {
    for (int i = 0; i < 15; i++) {
        cout << "--thread2: " << *stack.pop() << endl;
    }
}

int main()
{
    thread t1(fun1);
    thread t2(fun2);
    t1.join();
    t2.join();
    return 0;
}

风险指针
维护一个链表，将每个节点和操作他的线程的id绑定，每次检查每个节点指针是否有相应的id和他绑定，如果有指针和其绑定，则不能删除该节点，如果没有id和它绑定，那么可以直接删除该节点。

//风险指针的实现
unsigned const max_hazard_pointers = 100;
struct hazard_pointer
{
    std::atomic<std::thread::id> id;
    std::atomic<void*> pointer;
};
hazard_pointer hazard_pointers[max_hazard_pointers];

/*
    指针拥有者：
    当某个线程创建一个该类的对象，便会自动为该线程寻找一个hazard_pointer类，
   如果不够则抛出异常，当创建该类时会自动将该线程的id设定，并和相应的指针绑定
*/
class hp_owner
{
    hazard_pointer* hp;
public:
    hp_owner(hp_owner const&) = delete;
    hp_owner operator=(hp_owner const&) = delete;
    hp_owner() :
        hp(nullptr)
    {
        for (unsigned i = 0; i < max_hazard_pointers; ++i)
        {
            std::thread::id old_id;
            if (hazard_pointers[i].id.compare_exchange_strong( // 6 尝 
               old_id, std::this_thread::get_id()))
            {
                hp = &hazard_pointers[i];
                break; // 7 
            }
        }
        if (!hp) // 1 
        {
            throw std::runtime_error("No hazard pointers available");
        }
    }
    //返回找到的风险指针的类，并返回该类的指针
    std::atomic<void*>& get_pointer()
    {
        return hp->pointer;
    }
    ~hp_owner() // 2 
    {
        hp->pointer.store(nullptr); // 8 当线程销毁，将指针和数据复原
        hp->id.store(std::thread::id()); // 9 
    }
};
std::atomic<void*>& get_hazard_pointer_for_current_thread() // 
{
    thread_local static hp_owner hazard; // 4 每个线程都有自己的风险指针管理者，知道线程结束
        return hazard.get_pointer(); // 5 
}

//查询当前指针是否在风险指针中
bool outstanding_hazard_pointers_for(void* p)
{
    for (unsigned i = 0; i < max_hazard_pointers; ++i)
    {
        if (hazard_pointers[i].pointer.load() == p)
        {
            return true;
        }
    }
    return false;
}

/*
    删除单个节点
*/
template<typename T>
void do_delete(void* p)
{
    delete static_cast<T*>(p);
}

/*
    删除一个指针指向的节点，仅需要构造一个该类的对象即可
*/
struct data_to_reclaim
{
    void* data;//指向待删除的节点
    std::function<void(void*)> deleter;//删除节点的删除器
    data_to_reclaim* next;//指向下一个节点
    
    template<typename T>
    data_to_reclaim(T* p) : // 1 
        data(p),
        deleter(&do_delete<T>),
        next(0)
    {}

    ~data_to_reclaim()
    {
        deleter(data); // 2 
    }
};

std::atomic<data_to_reclaim*> nodes_to_reclaim;//指向被删除链的头节点

/*
    采用头插法将节点插入到待删除链表中
*/
void add_to_reclaim_list(data_to_reclaim* node) // 
{
    node->next = nodes_to_reclaim.load();
    while (!nodes_to_reclaim.compare_exchange_weak(node -> next, node));
}


//延迟删除节点
template<typename T>
void reclaim_later(T* data) // 4 
{
    add_to_reclaim_list(new data_to_reclaim(data)); // 5 
}


void delete_nodes_with_no_hazards()
{
    //currnet指向待删除的节点
    data_to_reclaim* current = nodes_to_reclaim.exchange(nullptr);

    while (current)
    {
        data_to_reclaim* const next = current->next;
        if (!outstanding_hazard_pointers_for(current->data)) // 7 
        {
            delete current; //将在析构函数中删除
        }
        else
        {   //如果还有线程id和该指针相对应，则不能删除，需要将链表放回到待删除链表中。
            add_to_reclaim_list(current); 
        }
        current = next;
    }
}

使用引用计数的节点

//引用计数计数节点的方式继续删除节点
template<typename T>
class lock_free_stack
{
private:
    struct node;
    struct counted_node_ptr // 含有外部引用计数的节点
    {
        int external_count;
        node* ptr;
    };
    struct node
    {
        std::shared_ptr<T> data;
        std::atomic<int> internal_count; // 内部引用计数节点
        counted_node_ptr next; 
        node(T const& data_) :data(std::make_shared<T>(data_)),internal_count(0)
        {}
    };
    std::atomic<counted_node_ptr> head;  

    //增加外部节点的计数
    void increase_head_count(counted_node_ptr & old_counter)
    {
        counted_node_ptr new_counter;//类似于copy-on-write的修改技术
        do
        {
            new_counter = old_counter;
            ++new_counter.external_count;
        } while (!head.compare_exchange_strong(old_counter, new_counter));

        old_counter.external_count = new_counter.external_count;
    }
public:
    ~lock_free_stack()
    {
        while (pop());
    }
    void push(T const& data) 
    {
        counted_node_ptr new_node;
        new_node.ptr = new node(data);
        new_node.external_count = 1;//插入一个节点时其外部节点仅有头节点指向，内部节点应该为0
        new_node.ptr->next = head.load();
        while (!head.compare_exchange_weak(new_node.ptr -> next, new_node));
    }

    std::shared_ptr<T> pop()
    {
        counted_node_ptr old_head = head.load();
        for (;;)
        {
            increase_head_count(old_head);//增加外部计数的节点
            node* const ptr = old_head.ptr; // 获取指向数据的指针
            if (!ptr)//指针为空，则数据不存在。返回空节点
            {
                return std::shared_ptr<T>();
            }
            if (head.compare_exchange_strong(old_head, ptr->next)) // 如果没有指针修改果头节点则取数据，否则重新获取头节点取数据。
            {
                std::shared_ptr<T> res;
                res.swap(ptr->data); // 4 
                int const count_increase = old_head.external_count - 2; // 

                if (ptr->internal_count.fetch_add(count_increase) == -count_increase)//如果没有其他节点操作该节点则删除该节点
                {
                    delete ptr;
                }
                return res; // 7 
            }
            else if (ptr->internal_count.fetch_sub(1) == 1)//如果当前仅有一个指针指向该节点则删除节点
            {
                delete ptr; // 8 
            }
        }
    }
};

指定内存顺序的无锁栈

template<typename T>
class lock_free_stack
{
private:
    struct node;
    struct counted_node_ptr // 含有外部引用计数的节点
    {
        int external_count;
        node* ptr;
    };
    struct node
    {
        std::shared_ptr<T> data;
        std::atomic<int> internal_count; // 内部引用计数节点
        counted_node_ptr next;
        node(T const& data_) :data(std::make_shared<T>(data_)), internal_count(0)
        {}
    };
    std::atomic<counted_node_ptr> head;

    //增加外部节点的计数
    void increase_head_count(counted_node_ptr& old_counter);
    
public:
    void push(T const & data);
    shared_ptr<T> pop();
    ~lock_free_stack()
    {
        while (pop());
    }
};


template<typename TC++并发编程----无锁实现线程安全队列（《C++ Concurrency in Action》 读书笔记）

 JUC并发编程 -- 保护共享资源（加锁实现 & 无锁实现）
 java并发编程：管程内存模型无锁并发线程池AQS原理与锁线程安全集合类并发设计模式
 java并发编程：管程内存模型无锁并发线程池AQS原理与锁线程安全集合类并发设计模式
 JAVA - 并发编程 - 线程安全方案
 C++并发编程----利用锁实现线程安全的数据结构（《C++ Concurrency in Action》 读书笔记）