尝试实现无锁队列时发生堆栈溢出
Stack overflow when trying to implement lock-free queue
我根据 Maged M. Michael 和 Michael L. Scott 工作中指定的算法实现了一个无锁队列 简单、快速、实用的非阻塞和阻塞
Concurrent Queue Algorithms(算法请跳转到第4页)
我在 shared_ptr
上使用了原子操作,例如 std::atomic_load_explicit
等
仅在一个线程中使用队列时,一切正常,但在不同线程中使用时,出现堆栈溢出异常。
很遗憾,我无法追查问题的根源。似乎当一个 shared_ptr
超出范围时,它会减少下一个 ConcurrentQueueNode
的引用数量并导致无限递归,但我不明白为什么..
代码:
队列节点:
template<class T>
struct ConcurrentQueueNode {
T m_Data;
std::shared_ptr<ConcurrentQueueNode> m_Next;
template<class ... Args>
ConcurrentQueueNode(Args&& ... args) :
m_Data(std::forward<Args>(args)...) {}
std::shared_ptr<ConcurrentQueueNode>& getNext() {
return m_Next;
}
T getValue() {
return std::move(m_Data);
}
};
并发队列(注意:胆小者勿入):
template<class T>
class ConcurrentQueue {
std::shared_ptr<ConcurrentQueueNode<T>> m_Head, m_Tail;
public:
ConcurrentQueue(){
m_Head = m_Tail = std::make_shared<ConcurrentQueueNode<T>>();
}
template<class ... Args>
void push(Args&& ... args) {
auto node = std::make_shared<ConcurrentQueueNode<T>>(std::forward<Args>(args)...);
std::shared_ptr<ConcurrentQueueNode<T>> tail;
for (;;) {
tail = std::atomic_load_explicit(&m_Tail, std::memory_order_acquire);
std::shared_ptr<ConcurrentQueueNode<T>> next =
std::atomic_load_explicit(&tail->getNext(),std::memory_order_acquire);
if (tail == std::atomic_load_explicit(&m_Tail, std::memory_order_acquire)) {
if (next.get() == nullptr) {
auto currentNext = std::atomic_load_explicit(&m_Tail, std::memory_order_acquire)->getNext();
auto res = std::atomic_compare_exchange_weak(&tail->getNext(), &next, node);
if (res) {
break;
}
}
else {
std::atomic_compare_exchange_weak(&m_Tail, &tail, next);
}
}
}
std::atomic_compare_exchange_strong(&m_Tail, &tail, node);
}
bool tryPop(T& dest) {
std::shared_ptr<ConcurrentQueueNode<T>> head;
for (;;) {
head = std::atomic_load_explicit(&m_Head, std::memory_order_acquire);
auto tail = std::atomic_load_explicit(&m_Tail,std::memory_order_acquire);
auto next = std::atomic_load_explicit(&head->getNext(), std::memory_order_acquire);
if (head == std::atomic_load_explicit(&m_Head, std::memory_order_acquire)) {
if (head.get() == tail.get()) {
if (next.get() == nullptr) {
return false;
}
std::atomic_compare_exchange_weak(&m_Tail, &tail, next);
}
else {
dest = next->getValue();
auto res = std::atomic_compare_exchange_weak(&m_Head, &head, next);
if (res) {
break;
}
}
}
}
return true;
}
};
重现问题的示例用法:
int main(){
ConcurrentQueue<int> queue;
std::thread threads[4];
for (auto& thread : threads) {
thread = std::thread([&queue] {
for (auto i = 0; i < 100'000; i++) {
queue.push(i);
int y;
queue.tryPop(y);
}
});
}
for (auto& thread : threads) {
thread.join();
}
return 0;
}
问题是竞争条件会导致队列中的每个节点都等待一次全部释放 - 这是递归的并且会破坏您的堆栈。
如果您将测试更改为仅使用一个线程但不弹出,您每次都会遇到相同的堆栈溢出错误。
for (auto i = 1; i < 100000; i++) {
queue.push(i);
//int y;
//queue.tryPop(y);
}
您需要非递归删除节点链:
__forceinline ~ConcurrentQueueNode() {
if (!m_Next || m_Next.use_count() > 1)
return;
KillChainOfDeath();
}
void KillChainOfDeath() {
auto pThis = this;
std::shared_ptr<ConcurrentQueueNode> Next, Prev;
while (1) {
if (pThis->m_Next.use_count() > 1)
break;
Next.swap(pThis->m_Next); // unwire node
Prev = NULL; // free previous node that we unwired in previous loop
if (!(pThis = Next.get())) // move to next node
break;
Prev.swap(Next); // else Next.swap will free before unwire.
}
}
我以前没用过shared_ptr,不知道有没有更快的方法。另外,由于我以前从未使用过shared_ptr,我不知道你的算法是否会遇到ABA问题。除非在 shared_ptr 实现中有一些特殊的东西来防止 ABA 我担心以前释放的节点可以被重用,欺骗 CAS。当我 运行 你的代码时,我似乎从来没有遇到过这个问题。
我根据 Maged M. Michael 和 Michael L. Scott 工作中指定的算法实现了一个无锁队列 简单、快速、实用的非阻塞和阻塞
Concurrent Queue Algorithms(算法请跳转到第4页)
我在 shared_ptr
上使用了原子操作,例如 std::atomic_load_explicit
等
仅在一个线程中使用队列时,一切正常,但在不同线程中使用时,出现堆栈溢出异常。
很遗憾,我无法追查问题的根源。似乎当一个 shared_ptr
超出范围时,它会减少下一个 ConcurrentQueueNode
的引用数量并导致无限递归,但我不明白为什么..
代码:
队列节点:
template<class T>
struct ConcurrentQueueNode {
T m_Data;
std::shared_ptr<ConcurrentQueueNode> m_Next;
template<class ... Args>
ConcurrentQueueNode(Args&& ... args) :
m_Data(std::forward<Args>(args)...) {}
std::shared_ptr<ConcurrentQueueNode>& getNext() {
return m_Next;
}
T getValue() {
return std::move(m_Data);
}
};
并发队列(注意:胆小者勿入):
template<class T>
class ConcurrentQueue {
std::shared_ptr<ConcurrentQueueNode<T>> m_Head, m_Tail;
public:
ConcurrentQueue(){
m_Head = m_Tail = std::make_shared<ConcurrentQueueNode<T>>();
}
template<class ... Args>
void push(Args&& ... args) {
auto node = std::make_shared<ConcurrentQueueNode<T>>(std::forward<Args>(args)...);
std::shared_ptr<ConcurrentQueueNode<T>> tail;
for (;;) {
tail = std::atomic_load_explicit(&m_Tail, std::memory_order_acquire);
std::shared_ptr<ConcurrentQueueNode<T>> next =
std::atomic_load_explicit(&tail->getNext(),std::memory_order_acquire);
if (tail == std::atomic_load_explicit(&m_Tail, std::memory_order_acquire)) {
if (next.get() == nullptr) {
auto currentNext = std::atomic_load_explicit(&m_Tail, std::memory_order_acquire)->getNext();
auto res = std::atomic_compare_exchange_weak(&tail->getNext(), &next, node);
if (res) {
break;
}
}
else {
std::atomic_compare_exchange_weak(&m_Tail, &tail, next);
}
}
}
std::atomic_compare_exchange_strong(&m_Tail, &tail, node);
}
bool tryPop(T& dest) {
std::shared_ptr<ConcurrentQueueNode<T>> head;
for (;;) {
head = std::atomic_load_explicit(&m_Head, std::memory_order_acquire);
auto tail = std::atomic_load_explicit(&m_Tail,std::memory_order_acquire);
auto next = std::atomic_load_explicit(&head->getNext(), std::memory_order_acquire);
if (head == std::atomic_load_explicit(&m_Head, std::memory_order_acquire)) {
if (head.get() == tail.get()) {
if (next.get() == nullptr) {
return false;
}
std::atomic_compare_exchange_weak(&m_Tail, &tail, next);
}
else {
dest = next->getValue();
auto res = std::atomic_compare_exchange_weak(&m_Head, &head, next);
if (res) {
break;
}
}
}
}
return true;
}
};
重现问题的示例用法:
int main(){
ConcurrentQueue<int> queue;
std::thread threads[4];
for (auto& thread : threads) {
thread = std::thread([&queue] {
for (auto i = 0; i < 100'000; i++) {
queue.push(i);
int y;
queue.tryPop(y);
}
});
}
for (auto& thread : threads) {
thread.join();
}
return 0;
}
问题是竞争条件会导致队列中的每个节点都等待一次全部释放 - 这是递归的并且会破坏您的堆栈。
如果您将测试更改为仅使用一个线程但不弹出,您每次都会遇到相同的堆栈溢出错误。
for (auto i = 1; i < 100000; i++) {
queue.push(i);
//int y;
//queue.tryPop(y);
}
您需要非递归删除节点链:
__forceinline ~ConcurrentQueueNode() {
if (!m_Next || m_Next.use_count() > 1)
return;
KillChainOfDeath();
}
void KillChainOfDeath() {
auto pThis = this;
std::shared_ptr<ConcurrentQueueNode> Next, Prev;
while (1) {
if (pThis->m_Next.use_count() > 1)
break;
Next.swap(pThis->m_Next); // unwire node
Prev = NULL; // free previous node that we unwired in previous loop
if (!(pThis = Next.get())) // move to next node
break;
Prev.swap(Next); // else Next.swap will free before unwire.
}
}
我以前没用过shared_ptr,不知道有没有更快的方法。另外,由于我以前从未使用过shared_ptr,我不知道你的算法是否会遇到ABA问题。除非在 shared_ptr 实现中有一些特殊的东西来防止 ABA 我担心以前释放的节点可以被重用,欺骗 CAS。当我 运行 你的代码时,我似乎从来没有遇到过这个问题。