无锁队列示例泄漏内存

Question

我正在尝试从“C++并发操作”(A.Williams)一书中获得无锁队列示例。由于我用sinlge {0}初始化了"null“counter_node_ptr对象，所以没有中止。但是现在我仍然有依赖于未初始化值的内存泄漏和跳转。后者可能是一个勇敢的问题，但前者对我来说是一个问题，我想摆脱它。这是一个完整的例子。欢迎任何帮助：

 // c++ -g lock_free_queue.cpp -latomic -pthread -o lock_free_queue
 // valgrind  ./lock_free_queue

#include <memory>
#include <atomic>
#include <thread>
#include <chrono>


using namespace std::chrono_literals;

// leaking memory
template<typename T>
class lock_free_queue
{
private:
  struct node;

  struct counted_node_ptr
    {
      int external_count;
      node* ptr;

      /*
      counted_node_ptr() noexcept
        : external_count(0)
        , ptr(nullptr) {}
      */
    };
  
    std::atomic<counted_node_ptr> head;
    std::atomic<counted_node_ptr> tail;

  struct node_counter
  {
    unsigned internal_count:30;
    unsigned external_counters:2;
    
    node_counter() noexcept
      : internal_count(0)
      , external_counters(2) {}
  };
  
  struct node
    {    
      std::atomic<T*> data;
      std::atomic<node_counter> count;
      std::atomic<counted_node_ptr> next;

      node() noexcept
      {
        data.store(nullptr);
        node_counter new_count;
        count.store(new_count);
        next.store(counted_node_ptr());
      }
      
      void release_ref()
      {
        node_counter old_counter=
          count.load(std::memory_order_relaxed);
        node_counter new_counter;
        do
          {
            new_counter=old_counter;
            --new_counter.internal_count;
          }
        while(!count.compare_exchange_strong
              (old_counter,new_counter,
               std::memory_order_acquire,std::memory_order_relaxed));
        if(!new_counter.internal_count &&
           !new_counter.external_counters)
          {
            delete this;
          }
      }
  };

  static void increase_external_count(
        std::atomic<counted_node_ptr>& counter,
        counted_node_ptr& old_counter)
    {
        counted_node_ptr new_counter;
        do
        {
            new_counter=old_counter;
            ++new_counter.external_count;
        }
        while(!counter.compare_exchange_strong(
                  old_counter,new_counter,
                  std::memory_order_acquire,std::memory_order_relaxed));
        old_counter.external_count=new_counter.external_count;
    }

  static void free_external_counter(counted_node_ptr &old_node_ptr)
    {
        node* const ptr=old_node_ptr.ptr;
        int const count_increase=old_node_ptr.external_count-2;
        node_counter old_counter=
            ptr->count.load(std::memory_order_relaxed);
        node_counter new_counter;
        do
        {
            new_counter=old_counter;
            --new_counter.external_counters;
            new_counter.internal_count+=count_increase;
        }
        while(!ptr->count.compare_exchange_strong(
                  old_counter,new_counter,
                  std::memory_order_acquire,std::memory_order_relaxed));
        if(!new_counter.internal_count &&
           !new_counter.external_counters)
        {
            delete ptr;
        }
    }

  void set_new_tail(counted_node_ptr &old_tail,
                      counted_node_ptr const &new_tail)
    {
        node* const current_tail_ptr=old_tail.ptr;
        while(!tail.compare_exchange_weak(old_tail,new_tail) &&
              old_tail.ptr==current_tail_ptr);
        if(old_tail.ptr==current_tail_ptr)
            free_external_counter(old_tail);
        else
            current_tail_ptr->release_ref();
    }
  
public:
    lock_free_queue()
    {
      counted_node_ptr new_head;
      new_head.ptr = new node;
      new_head.external_count = 1;
      head.store(new_head);
      tail = head.load();
    }
    lock_free_queue(const lock_free_queue& other)=delete;
    lock_free_queue& operator=(const lock_free_queue& other)=delete;
    ~lock_free_queue()
    {
      while(pop());
      delete head.load().ptr;
    }

  void push(T new_value)
    {
        std::unique_ptr<T> new_data(new T(new_value));
        counted_node_ptr new_next;
        new_next.ptr=new node;
        new_next.external_count=1;
        counted_node_ptr old_tail=tail.load();
        for(;;)
        {
            increase_external_count(tail,old_tail);
            T* old_data=nullptr;
            if(old_tail.ptr->data.compare_exchange_strong(
                   old_data,new_data.get()))
            {
              counted_node_ptr old_next{0};
              if(!old_tail.ptr->next.compare_exchange_strong
                 (old_next,new_next))
                {
                  delete new_next.ptr;
                  new_next=old_next;
                }
              set_new_tail(old_tail, new_next);
              new_data.release();
              break;
            }
            else
            {
              counted_node_ptr old_next{0};
                if(old_tail.ptr->next.compare_exchange_strong(
                       old_next,new_next))
                {
                    old_next=new_next;
                    new_next.ptr=new node;
                }
                set_new_tail(old_tail, old_next);
            }
        }
    }

  std::unique_ptr<T> pop()
    {
        counted_node_ptr old_head=head.load(std::memory_order_relaxed);
        for(;;)
        {
            increase_external_count(head,old_head);
            node* const ptr=old_head.ptr;
            if(ptr==tail.load().ptr)
            {
                return std::unique_ptr<T>();
            }
            counted_node_ptr next=ptr->next.load();
            if(head.compare_exchange_strong(old_head,next))
            {
                T* const res=ptr->data.exchange(nullptr);
                free_external_counter(old_head);
                return std::unique_ptr<T>(res);
            }
            ptr->release_ref();
        }
    }

  std::shared_ptr<T> wait_and_pop()
  {
    std::unique_ptr<T> value_ptr = pop();
    while (!value_ptr)
      {
        std::this_thread::yield();
        value_ptr = pop();
      }
    return value_ptr;
  }

};

#define MAIN
#ifdef MAIN

#include <iostream>
int main()
{
  lock_free_queue<int> hallo;
  int a = 0;
  int b = 0;
  int c = 0;
  int d = 0;
  int e = 0;
  int f = 0;


  //#define WAIT_AND_POP
#ifdef WAIT_AND_POP
  std::thread task2 =  std::thread( [&] () {
                                      //std::this_thread::sleep_for(2s);
                                      a = *(hallo.wait_and_pop());
                                      b = *(hallo.wait_and_pop());
                                    });
  std::thread task3 =  std::thread( [&] () {
                                      //std::this_thread::sleep_for(2s);g
                                      c = *(hallo.wait_and_pop());
                                      d = *(hallo.wait_and_pop());
                                    });
 
#else // #findef WAIT_AND_POP
  std::thread task2 =  std::thread( [&] () {
                                      //std::this_thread::sleep_for(2s);
                                      std::unique_ptr<int> ap = hallo.pop();
                                      a = (ap ? *(ap.get()) : 0);
                                      std::unique_ptr<int> bp = hallo.pop();
                                      b = (bp ? *(bp.get()) : 0);
                                      std::unique_ptr<int> ep = hallo.pop();
                                      e = (ep ? *(ep.get()) : 0);
                                    });
  std::thread task3 =  std::thread( [&] () {
                                      //std::this_thread::sleep_for(2s);
                                      std::unique_ptr<int> cp = hallo.pop();
                                      c = (cp ? *(cp.get()) : 0);
                                      std::unique_ptr<int> dp = hallo.pop();
                                      d = (dp ? *(dp.get()) : 0);
                                      std::unique_ptr<int> fp = hallo.pop();
                                      f = (fp ? *(fp.get()) : 0);
                                    });
#endif

  std::this_thread::sleep_for(1s);
  
  std::thread task0 =  std::thread( [&] () {
                                      hallo.push(10);
                                      hallo.push(2); 
                                    });
  
  std::thread task1 =  std::thread( [&] () {
                                      hallo.push(5);
                                      hallo.push(6); 
                                    });
  std::thread task4 =  std::thread( [&] () {
                                      hallo.push(5);
                                      hallo.push(6); 
                                    });
  task1.join();
  task2.join();
  task3.join();
  task0.join();
  task4.join();

  std::cout << "a = " << a << " b = " << b << " c = " << c << " d = " << d << std::endl;
  if (hallo.pop())
    {
      std::cout << "more data than expected" << std::endl;
    }
  else
    {
      std::cout << "ok" << std::endl;
    }
}

#endif

我会犯这样的错误：

==48636== HEAP SUMMARY:
==48636==     in use at exit: 32 bytes in 1 blocks
==48636==   total heap usage: 31 allocs, 30 frees, 75,688 bytes allocated
==48636== 
==48636== Thread 1:
==48636== 32 bytes in 1 blocks are definitely lost in loss record 1 of 1
==48636==    at 0x483BE63: operator new(unsigned long) (in /usr/lib/x86_64-linux-gnu/valgrind/vgpreload_memcheck-amd64-linux.so)
==48636==    by 0x10C75B: lock_free_queue<int>::lock_free_queue() (lock_free_queue.cpp:136)
==48636==    by 0x10A6BF: main (lock_free_queue.cpp:229)
==48636== 
==48636== LEAK SUMMARY:
==48636==    definitely lost: 32 bytes in 1 blocks
==48636==    indirectly lost: 0 bytes in 0 blocks
==48636==      possibly lost: 0 bytes in 0 blocks
==48636==    still reachable: 0 bytes in 0 blocks
==48636==         suppressed: 0 bytes in 0 blocks

当然，我已经检查了队列的构造函数和析构函数.我错过了什么？

Answer 1

我遇到了同样的问题张贴在其他地方，这是唯一的其他提到，我可以在网上找到。

尽管std::atomic在普通平台上并不是免费的，但代码仍然应该能够正常工作。

结果发现，插片有个ABA问题-

T* old_data=nullptr;

if(old_tail.ptr->data.compare_exchange_strong(
       old_data, new_data.get()))
{

在两种情况下，当初始创建空节点时，以及(作者忽略了)当节点一直在队列中传递并已被消耗时，可以将old_t.ptr->数据设置为nullptr：

T* const res = ptr->data.exchange(nullptr);

在从读取old_tail到尝试这个compare_exchange之间不太可能循环整个队列的情况下，我们将将数据分配到一个已经消耗的节点中，当old_data被释放时，该节点将被解除分配。

您可以通过在使用后不将指针设置为null来解决问题--它无论如何也不会在任何地方使用--或者使用不同的哨位值而不是nullptr来使调试更加容易。

T* const res = ptr->data.load();
// or T* const res = ptr->data.exchange((T*)1);

下面是一个显示它按预期工作的https://godbolt.org/z/4Y5WsboE4链接