c++ - 使用 shared_ptr 时出现 SEGFAULT

Question

我正在尝试使用 shared_ptr 在 C++ 中实现Lazy Concurrent List-based Set。我的推理是 unreachable nodes 将被最后一个 shared_ptr 自动释放。根据我的理解，shared_ptr 的引用计数 的递增和递减操作是原子的。这意味着只有引用该节点的last shared_ptr 应该为该节点调用delete/free。我为多线程 运行程序，但我的程序崩溃并出现错误double free called 或只是Segmentation Fault(SIGSEGV)。我不明白这怎么可能。下面给出了我的实现代码，方法名称表示它们的预期操作。

#include<thread>
#include<iostream>
#include<mutex>
#include<climits>

using namespace std;

class Thread
{
   public:
      std::thread t;  
};
int n=50,ki=100,kd=100,kc=100;`/*no of threads, no of inserts,deletes & searches*/`


class Node
{
public:
      int key;
      shared_ptr<Node> next;
      bool marked;
      std::mutex nodeLock;

      Node() {
         key=0;
         next = nullptr;
         marked = false;
      }

      Node(int k) {
         key = k;
         next = nullptr;
         marked = false;
      }

      void lock() {
         nodeLock.lock();
      }

      void unlock() {
         nodeLock.unlock();
      }

      ~Node()
      {
      }
};

class List {
   shared_ptr<Node> head;
   shared_ptr<Node> tail;

public:

   bool validate(shared_ptr<Node> pred, shared_ptr<Node> curr) {
      return !(pred->marked) && !(curr->marked) && ((pred->next) == curr);
   }

   List() {
      head=make_shared<Node>(INT_MIN);
      tail=make_shared<Node>(INT_MAX);
      head->next=tail;
   }

   bool add(int key)
   {
      while(true)
      {
         /*shared_ptr<Node> pred = head;
         shared_ptr<Node> curr = pred->next;*/
        auto pred = head;
        auto curr = pred->next;

         while (key>(curr->key))
         {
            pred = curr;
            curr = curr->next;
         }

         pred->lock();
         curr->lock();

         if (validate(pred,curr))
         {
            if (curr->key == key)
            {
               curr->unlock();
               pred->unlock();
               return false;
            }
            else
            {
                shared_ptr<Node> newNode(new Node(key));
               //auto newNode = make_shared<Node>(key);
                //shared_ptr<Node> newNode = make_shared<Node>(key);
                newNode->next = curr;
                pred->next = newNode;
                curr->unlock();
                pred->unlock();
                return true;
            }
         }
         curr->unlock();
         pred->unlock();
      }
   }

   bool remove(int key)
   {
      while(true)
      {
         /*shared_ptr<Node> pred = head;
         shared_ptr<Node> curr = pred->next;*/

        auto pred = head;
        auto curr = pred->next;

         while (key>(curr->key))
         {
            pred = curr;
            curr = curr->next;
         }

         pred->lock();
         curr->lock();

         if (validate(pred,curr))
         {
            if (curr->key != key)
            {
               curr->unlock();
               pred->unlock();
               return false;
            }
            else
            {
               curr->marked = true;
               pred->next = curr->next;
               curr->unlock();
               pred->unlock();
               return true;
            }
         }
         curr->unlock();
         pred->unlock();
      }
   }

   bool contains(int key) {
      //shared_ptr<Node> curr = head->next;
    auto curr = head->next;

      while (key>(curr->key)) {
         curr = curr->next;
      }
      return curr->key == key && !curr->marked;
   }
}list;

void test(int curr)
{
   bool test;
    int time;

    int val, choice;
    int total,k=0;
    total=ki+kd+kc;

    int i=0,d=0,c=0;

    while(k<total)
    {
        choice = (rand()%3)+1;

        if(choice==1)
        {
            if(i<ki)
            {
                val = (rand()%99)+1;
                test = list.add(val);
                i++;
                k++;
            }
        }
        else if(choice==2)
        {
            if(d<kd)
            {
                val = (rand()%99)+1;
                test = list.remove(val);
                d++;
                k++;
            }
        }
        else if(choice==3)
        {
            if(c<kc)
            {
                val = (rand()%99)+1;
                test = list.contains(val);
                c++;
                k++;
            }
        }
    }
}

int main()
{
   int i;

   vector<Thread>thr(n);

   for(i=0;i<n;i++)
   {
      thr[i].t = thread(test,i+1);
   }
   for(i=0;i<n;i++)
   {
      thr[i].t.join();
   }
   return 0;
}

我无法弄清楚上面的代码有什么问题。错误每次都不同，其中一些只是 SEGFAULTS 或

pure virtual method called
terminate called without an active exception
Aborted (core dumped)

你能指出我在上面的代码中做错了什么吗？以及如何修复该错误？
编辑: 添加了一个非常粗糙的测试函数，它随机调用三个列表方法。此外，全局声明线程数和每个操作数。粗略的编程，但它重新创建了 SEGFAULT。

Answer 1

问题是您没有为您的 shared_ptr 使用原子存储和加载操作。

确实，shared_ptr 的控制 block 中的引用计数(参与特定共享对象所有权的每个 shared_ptr 都有一个指针)是然而，shared_ptr 的数据成员本身不是原子的。

因此让多个线程每个都有自己的共享对象的shared_ptr是安全的，但是让多个线程尽快访问同一个shared_ptr是不安全的因为至少其中一个正在使用非常量成员函数，这就是您在重新分配 next 指针时所做的事情。

说明问题

让我们看一下 libstdc++ 的 shared_ptr 实现的(简化和美化的)复制构造函数:

shared_ptr(const shared_ptr& rhs)
 : m_ptr(rhs.m_ptr),
   m_refcount(rhs.m_refcount) 
{ }

这里 m_ptr 只是指向共享对象的原始指针，而 m_refcount 是一个进行引用计数并处理对象 最终删除的类m_ptr 指向。

这只是一个可能出错的例子:假设当前一个线程正在试图弄清楚列表中是否包含一个特定的键。它从 List::contains 中的复制初始化 auto curr = head->next 开始。在成功初始化 curr.m_ptr 之后，操作系统调度程序决定此线程必须暂停并安排到另一个线程中。

另一个线程正在删除 head 的后继者。因为 head->next 的引用计数仍然是 1(毕竟 head->next 的引用计数还没有被线程 1 修改)，当第二个线程完成删除节点时，它正在被删除。

然后一段时间后第一个线程继续。它完成了curr的初始化，但是由于m_ptr在线程2开始删除之前就已经初始化了，所以它仍然指向现在被删除的节点。当尝试比较 key > curr->key 时，线程 1 将访问无效内存。

使用 std::atomic_load 和 std::atomic_store 来防止这个问题

std::atomic_load 和 std::atomic_store 通过在调用复制构造函数/复制赋值运算符之前锁定互斥锁来防止问题发生通过指针传入的shared_ptr。如果对多个线程共享的 shared_ptr 的所有读取和写入都是通过 std::atomic_load/std::atomic_store 响应的。在另一个线程开始读取或修改相同的 shared_ptr 时，永远不会发生一个线程只修改了 m_ptr 但没有修改引用计数的情况。

有了必要的原子加载和存储，List 成员函数应该如下所示:

bool validate(Node const& pred, Node const& curr) {
   return !(pred.marked) && !(curr.marked) && 
          (std::atomic_load(&pred.next).get() == &curr);
}

bool add(int key) {
    while (true) {
        auto pred = std::atomic_load(&head);
        auto curr = std::atomic_load(&pred->next);

        while (key > (curr->key)) {
            pred = std::move(curr);
            curr = std::atomic_load(&pred->next);
        }

        std::scoped_lock lock{pred->nodeLock, curr->nodeLock};
        if (validate(*pred, *curr)) {
            if (curr->key == key) {
                return false;
            } else {
                auto new_node = std::make_shared<Node>(key);

                new_node->next = std::move(curr);
                std::atomic_store(&pred->next, std::move(new_node));
                return true;
            }
        }
    }
}

bool remove(int key) {
    while (true) {
        auto pred = std::atomic_load(&head);
        auto curr = std::atomic_load(&pred->next);

        while (key > (curr->key)) {
            pred = std::move(curr);
            curr = std::atomic_load(&pred->next);
        }

        std::scoped_lock lock{pred->nodeLock, curr->nodeLock};
        if (validate(*pred, *curr)) {
            if (curr->key != key) {
                return false;
            } else {
                curr->marked = true;
                std::atomic_store(&pred->next, std::atomic_load(&curr->next));
                return true;
            }
        }
    }
}

bool contains(int key) {
    auto curr = std::atomic_load(&head->next);

    while (key > (curr->key)) {
        curr = std::atomic_load(&curr->next);
    }
    return curr->key == key && !curr->marked;
}

此外，您还应该将 Node::marked 设为 std::atomic_bool。