一、Michael & Scoot 原版伪代码实现
structure pointer_t {ptr: pointer to node_t, count: unsigned integer}
structure node_t {value: data type, next: pointer_t}
structure queue_t {Head: pointer_t, Tail: pointer_t}
initialize(Q: pointer to queue_t)
node = new_node() // Allocate a free node
node->next.ptr = NULL // Make it the only node in the linked list
Q->Head.ptr = Q->Tail.ptr = node // Both Head and Tail point to it
enqueue(Q: pointer to queue_t, value: data type)
E1: node = new_node() // Allocate a new node from the free list
E2: node->value = value // Copy enqueued value into node
E3: node->next.ptr = NULL // Set next pointer of node to NULL
E4: loop // Keep trying until Enqueue is done
E5: tail = Q->Tail // Read Tail.ptr and Tail.count together
E6: next = tail.ptr->next // Read next ptr and count fields together
E7: if tail == Q->Tail // Are tail and next consistent?
// Was Tail pointing to the last node?
E8: if next.ptr == NULL
// Try to link node at the end of the linked list
E9: if CAS(&tail.ptr->next, next, <node, next.count+1>)
E10: break // Enqueue is done. Exit loop
E11: endif
E12: else // Tail was not pointing to the last node
// Try to swing Tail to the next node
E13: CAS(&Q->Tail, tail, <next.ptr, tail.count+1>)
E14: endif
E15: endif
E16: endloop
// Enqueue is done. Try to swing Tail to the inserted node
E17: CAS(&Q->Tail, tail, <node, tail.count+1>)
dequeue(Q: pointer to queue_t, pvalue: pointer to data type): boolean
D1: loop // Keep trying until Dequeue is done
D2: head = Q->Head // Read Head
D3: tail = Q->Tail // Read Tail
D4: next = head.ptr->next // Read Head.ptr->next
D5: if head == Q->Head // Are head, tail, and next consistent?
D6: if head.ptr == tail.ptr // Is queue empty or Tail falling behind?
D7: if next.ptr == NULL // Is queue empty?
D8: return FALSE // Queue is empty, couldn't dequeue
D9: endif
// Tail is falling behind. Try to advance it
D10: CAS(&Q->Tail, tail, <next.ptr, tail.count+1>)
D11: else // No need to deal with Tail
// Read value before CAS
// Otherwise, another dequeue might free the next node
D12: *pvalue = next.ptr->value
// Try to swing Head to the next node
D13: if CAS(&Q->Head, head, <next.ptr, head.count+1>)
D14: break // Dequeue is done. Exit loop
D15: endif
D16: endif
D17: endif
D18: endloop
D19: free(head.ptr) // It is safe now to free the old node
D20: return TRUE // Queue was not empty, dequeue succeeded
二、C++实现
c++的实现就直接看上述的伪代码跟着实现即可，这里的一些atomic操作也可以看我之前写的博客
template<typename T>
class LockFreeQueue {
private:
// 队列结构
struct Node {
std::shared_ptr<T> data;
std::atomic<Node*> next;
Node() : next(nullptr) {};
};
std::atomic<Node*> head; // 头节点
std::atomic<Node*> tail; // 尾节点
Node* dummy; // 用于回收节点的哑节点
public:
LockFreeQueue() {
dummy = new Node();
head.store(dummy);
tail.store(dummy);
}
~LockFreeQueue() {
T output;
while (dequeue(output)) {}
delete dummy;
}
// 禁止拷贝构造和赋值
LockFreeQueue(const LockFreeQueue&) = delete;
LockFreeQueue& operator=(const LockFreeQueue&) = delete;
void enqueue(const T& value) {
std::shared_ptr<T> new_data(std::make_shared<T>(value)); // 创建数据
Node* new_node = new Node(); // 创建新节点
Node* old_tail;
while (true) {
old_tail = tail.load();
Node* next = old_tail->next.load();
if (old_tail == tail.load()) { // 此时保证tail还没有被其他线程改变
if (next == nullptr) { // 此时保证tail是队列的最后一个节点
if (old_tail->next.compare_exchange_weak(next, new_node)) break; // 插入成功，这是一个原语，可以一次操作
}
else { // 说明tail落后了，推进tail指针
tail.compare_exchange_weak(old_tail, next);
}
}
}
old_tail->data = new_data;
tail.compare_exchange_weak(old_tail, new_node);
}
bool dequeue(T& value) {
Node* old_head;
while (true) {
old_head = head.load(); // 获取当前头部节点
Node* old_tail = tail.load(); // 获取当前尾部节点
Node* next = old_head->next.load();
if (old_head == head.load()) { // 确保head没有被改变
if (old_head == old_tail) { // 队列为空，或者tail落后
if (next == nullptr) { // 队列为空
return false;
}
tail.compare_exchange_weak(old_tail, next);
} else {
// 从队列中移除head，并且读取其值
if (next->data) {
value = *(next->data);
if (head.compare_exchange_weak(old_head, next)) break; // 清除head节点，退出循环
}
}
}
}
delete old_head;
return true;
}
};
三、测试
测试代码如下：
int main() {
LockFreeQueue<int> queue;
std::thread t1([&queue](){
for(int i=0; i<100; i++) {
queue.enqueue(i);
std::this_thread::sleep_for(std::chrono::milliseconds(100));
}
});
std::thread t2([&queue](){
for(int i=100; i<200; i++) {
queue.enqueue(i);
std::this_thread::sleep_for(std::chrono::milliseconds(100));
}
});
std::thread t3([&queue]() {
while (true) {
int i = 0;
if (queue.dequeue(i)) {
std::cout << "t3:" << i << std::endl;
}
}
});
std::thread t4([&queue]() {
while (true) {
int i = 0;
if (queue.dequeue(i)) {
std::cout << "t4:" << i << std::endl;
}
}
});
t1.join();
t2.join();
t3.join();
t4.join();
}
可以看到，多线程下可以顺利的插入与找到数据
