为何无界并发队列实现中需保留这两个if语句?
我在学习一款无界并发队列(UnboundedQueue)的实现时,对代码中两处高亮的if语句必要性存疑。我自己推导觉得这两个if语句没必要,但不敢否定教材内容,希望得到解释。
以下是队列的实现代码:
template<typename T> struct Node { Node(T val, Node* next = nullptr) : value(val), next(next) {} T value; std::atomic<Node*> next; }; template<typename T> struct UnboundedQueue { std::atomic<Node*> head; std::atomic<Node*> tail; void enq(T val); std::optional<T> deq(void); }; template<typename T> void UnboundedQueue::enq(T val) { Node* node = new Node(val); while(true) { Node* last = tail.load(); Node* next = last->next.load(); // ********* 此处if语句 if(last == tail.load()) { if(next == nullptr) { if(last->next.compare_exchange_weak(next, node)) { tail.compare_exchange_weak(last, node); return; } } else { tail.compare_exchange_weak(last, next); } } } } template<typename T> std::optional<T> UnboundedQueue::deq(void) { while(true) { Node* first = head.load(); Node* last = tail.load(); Node* next = first->next.load(); // ********* 此处if语句 if(first == head.load()) { if(last == first) { if(first->next == nullptr) { return std::nullopt; } else { tail.compare_exchange_weak(last, next); } } else { if(head.compare_exchange_weak(first, next)) { T res = first->value; // 此处存在未定义行为,但本次问题不涉及 delete first; return res; } } } } }
推导前提:两个不变量及证明
我先提出两个不变量并给出非正式证明:
- 不变量1:head节点永远不会超过tail节点
非正式证明:若要出现head超过tail的情况,需满足head == tail且head == first(以执行CAS操作)。但由于我们在加载first之后才加载last,这意味着last == tail,因此会进入第一个if语句,不会让head超过tail。
- 不变量2:若某个节点曾是tail节点且其next值为nullptr,则它当前仍是tail节点
非正式证明:我们从未将节点的next值设为nullptr。因此节点要失去tail身份,必须先在其后添加另一个节点,将其next值设为非nullptr。
移除if语句后的场景分析
我假设移除这两个if语句,分析不同场景下队列状态不会出现错误:
假设last != tail.load()
场景1:next == nullptr,根据第二个不变量,该假设不成立。
场景2:next != nullptr。此时执行tail.compare_exchange_weak(last, next),由于tail != last,该操作不会产生任何效果。
假设first != head.load()
场景1:last == first
场景1.1:first->next == nullptr。
此时last == first,说明first曾等于tail。根据第二个不变量,first == tail,再结合第一个不变量(head不会超过tail),可得first == head,因此该假设不成立。
场景1.2:first->next != nullptr场景1.2.1:first != tail:执行
tail.compare_exchange_weak(last, next)不会产生任何效果。
场景1.2.2:first == tail:结合第一个不变量(head不会超过tail),可得first == head,假设不成立。
场景2:last != first:
执行head.compare_exchange_weak(first, next),由于head != first(假设条件),该操作不会产生任何效果。
注:我在并发编程方面经验有限,推导可能存在错误。
内容的提问来源于stack exchange,提问作者Thornsider3

