使用shared_ptr时出现SEGFAULT错误

Question

我正在尝试使用shared_ptr在C++中实现延迟并发基于列表的集合。我的理解是，只有最后一个引用节点的shared_ptr才会自动释放不可达节点。根据我的理解，对shared_ptr的引用计数进行的增减操作是原子的。这意味着仅具有对该节点的最后一个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。

- Dee Jay

---

关于使用shared_ptr，很好。我敬佩实验精神。 - user4581301

@user4581301，我应该开一个新问题，包括main的完整代码吗？还是只在评论中发布它？ - Dee Jay

@DeeJay 请编辑您当前的帖子，加入main函数。无需发起另一个问题。同时，请勿在评论中发布代码。 - PaulMcKenzie

@user4581301，不是的，但这个列表的主要思想就是如此。一旦你锁定了pred＆curr，就会调用“validate”函数来检测任何同步冲突。 validate检查pred＆curr是否未被标记，并且pred仍指向curr。即使删除了pred或curr，它们也会首先被标记（remove方法）。因此，“Lazy Synchronization”是该列表的名称。 - Dee Jay

@user4581301 添加了必要的代码编辑。 - Dee Jay

@PaulMcKenzie 添加了必要的代码编辑。 - Dee Jay

Answer 1

问题在于你没有使用原子存储和加载操作来处理你的shared_ptr。

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

因此，每个线程拥有自己的shared_ptr对象引用同一共享对象是安全的，但是当至少有一个线程正在使用非const成员函数（这就是你重新分配next指针时所做的事情）时，多个线程访问相同的shared_ptr是不安全的。

下面展示了问题的图示例：

让我们看一下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已经在第二个线程开始删除之前初始化，它仍然指向现在已删除的节点。当尝试比较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进行的，那么当一个线程仅修改m_ptr而不是引用计数时，另一个线程开始读取或修改相同的shared_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。

- Corristo

---

好的...我的想法是，当一个线程发现一个对象（使用shared_ptr）时，引用计数会被原子地增加。我假设上述两个步骤，即读取下一个指针和引用计数增加，将是一个原子步骤。因此，如果另一个线程首先获得引用（读取下一个指针），则没有线程能够释放节点。 - Dee Jay

@DeeJay 问题在于创建一个shared_ptr的副本总是需要至少执行两个操作：增加控制块中的引用计数，然后将指针复制到控制块中。假设引用计数首先被修改。那么你可能会遇到以下情况：一个线程即将重新分配next指针。它已经减少了旧对象的引用计数并注意到引用计数为零。在实际删除对象之前，该线程被调度出去。 - Corristo

那么，使用“原子操作”如何解决这个问题呢？看来我需要更深入地了解“原子操作”和“shared_ptr”。 - Dee Jay

@DeeJay std::atomic_load和std::atomic_store会锁定互斥量，以确保在进行atomic_load操作时，不能同时进行对同一个shared_ptr的atomic_store（或另一个std::atomic_load）操作。 - Corristo

@DeeJay，我已经扩展了我的答案，以展示两个线程的示例运行中出现的问题。 - Corristo