使用 shared_ptr 时的 SEGFAULT

Question

我正在尝试使用shared_ptr 在 C++ 中实现 基于惰性并发列表的集合。我的理由是unreachable nodes 将被最后一个shared_ptr 自动释放。据我了解，shared_ptr's reference count 上的递增和递减操作是原子的。这意味着只有参考节点的 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)

您能否指出我在上面的代码中做错了什么？以及如何修复该错误？
编辑：添加了一个非常粗略的test function，它随机调用三个list methods。此外，线程数和每个操作的数量是全局声明的。粗略的编程，但它重新创建了 SEGFAULT。

Answer 1

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

确实，shared_ptr 的控制块中的引用计数（参与特定共享对象所有权的每个 shared_ptr 都有一个指向的指针）是原子的，但是，@ 的数据成员987654324@ 本身不是。

因此，让多个线程各自拥有自己的shared_ptr 到一个共享对象是安全的，但是让多个线程访问相同的shared_ptr 并不安全，只要其中至少一个线程使用非-const 成员函数，这是您在重新分配 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_ptrs 的读取和写入都通过std::atomic_load/std::atomic_store resp。在另一个线程开始读取或修改相同的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。