批評我的併發隊列

Question

這是我編寫的併發隊列，我打算在我寫的線程池中使用。我想知道是否有任何性能改進。如果您好奇，atomic_counter被粘貼在下面！

#ifndef NS_CONCURRENT_QUEUE_HPP_INCLUDED 
#define NS_CONCURRENT_QUEUE_HPP_INCLUDED 

#include <ns/atomic_counter.hpp> 
#include <boost/noncopyable.hpp> 
#include <boost/smart_ptr/detail/spinlock.hpp> 
#include <cassert> 
#include <cstddef> 

namespace ns { 
    template<typename T, 
      typename mutex_type = boost::detail::spinlock, 
      typename scoped_lock_type = typename mutex_type::scoped_lock> 
    class concurrent_queue : boost::noncopyable { 
     struct node { 
      node * link; 
      T const value; 
      explicit node(T const & source) : link(0), value(source) { } 
     }; 
     node * m_front; 
     node * m_back; 
     atomic_counter m_counter; 
     mutex_type m_mutex; 
    public: 
     // types 
     typedef T value_type; 

     // construction 
     concurrent_queue() : m_front(0), m_mutex() { } 
     ~concurrent_queue() { clear(); } 

     // capacity 
     std::size_t size() const { return m_counter; } 
     bool empty() const { return (m_counter == 0); } 

     // modifiers 
     void push(T const & source); 
     bool try_pop(T & destination); 
     void clear(); 
    }; 

    template<typename T, typename mutex_type, typename scoped_lock_type> 
    void concurrent_queue<T, mutex_type, scoped_lock_type>::push(T const & source) { 
     node * hold = new node(source); 
     scoped_lock_type lock(m_mutex); 
     if (empty()) 
      m_front = hold; 
     else 
      m_back->link = hold; 
     m_back = hold; 
     ++m_counter; 
    } 

    template<typename T, typename mutex_type, typename scoped_lock_type> 
    bool concurrent_queue<T, mutex_type, scoped_lock_type>::try_pop(T & destination) { 
     node const * hold; 
     { 
      scoped_lock_type lock(m_mutex); 
      if (empty()) 
       return false; 
      hold = m_front; 
      if (m_front == m_back) 
       m_front = m_back = 0; 
      else 
       m_front = m_front->link; 
      --m_counter; 
     } 
     destination = hold->value; 
     delete hold; 
     return true; 
    } 

    template<typename T, typename mutex_type, typename scoped_lock_type> 
    void concurrent_queue<T, mutex_type, scoped_lock_type>::clear() { 
     node * hold; 
     { 
      scoped_lock_type lock(m_mutex); 
      hold = m_front; 
      m_front = 0; 
      m_back = 0; 
      m_counter = 0; 
     } 
     if (hold == 0) 
      return; 
     node * it; 
     while (hold != 0) { 
      it = hold; 
      hold = hold->link; 
      delete it; 
     } 
    } 
} 

#endif 

atomic_counter.hpp

#ifndef NS_ATOMIC_COUNTER_HPP_INCLUDED 
#define NS_ATOMIC_COUNTER_HPP_INCLUDED 

#include <boost/interprocess/detail/atomic.hpp> 
#include <boost/noncopyable.hpp> 

namespace ns { 
    class atomic_counter : boost::noncopyable { 
     volatile boost::uint32_t m_count; 
    public: 
     explicit atomic_counter(boost::uint32_t value = 0) : m_count(value) { } 

     operator boost::uint32_t() const { 
      return boost::interprocess::detail::atomic_read32(const_cast<volatile boost::uint32_t *>(&m_count)); 
     } 

     void operator=(boost::uint32_t value) { 
      boost::interprocess::detail::atomic_write32(&m_count, value); 
     } 

     void operator++() { 
      boost::interprocess::detail::atomic_inc32(&m_count); 
     } 

     void operator--() { 
      boost::interprocess::detail::atomic_dec32(&m_count); 
     } 
    }; 
} 

#endif 

Answer 1

我想你會遇到性能問題，在這種情況下，因爲調用new爲每個新節點的鏈接列表。這不僅僅是因爲調用動態內存分配器很慢。這是因爲調用它經常引入很多併發開銷，因爲免費存儲必須在多線程環境中保持一致。

我會使用一個你調整大小的矢量，因爲它太小而無法保持隊列。我永遠不會調整它更小。

我會安排正面和背面的值，所以向量是一個環形緩衝區。這將需要您在調整大小時移動元素。但這應該是一個相當罕見的事件，並且可以通過在施工時提供建議的矢量大小在一定程度上減輕。

或者，您可以保留鏈接列表結構，但不要銷燬節點。只需將其添加到空閒節點的隊列中即可。不幸的是，免費節點的隊列需要鎖定才能正常管理，而且我不確定你是否真的處於比刪除更新更好的狀態，並且始終處於新狀態。

您還將獲得更好的向量參考位置。但我不太肯定這將如何與高速緩存線交互在CPU之間來回傳遞。

其他一些人建議::std::deque，我不認爲這是一個壞主意，但我懷疑環形緩衝區向量是一個更好的主意。

Answer 2

香草薩特提出了一個無鎖隊列的實現，肯定會超越你的:)

主要的想法是使用一個緩衝圈，隊列的運行過程中完全放棄的內存動態分配。這意味着隊列可能已滿（因此您可能需要等待放入元素），這在您的情況下可能不可接受。

正如Omnifarious指出的那樣，最好不要使用鏈表（用於緩存局部性），除非您爲池分配內存。我會嘗試使用std::deque作爲後端，它對內存更友好，並且保證只要您正常情況下只在隊列中彈出和推送（在前端和後端），就不會有任何重新分配。