代码之家  ›  专栏  ›  技术社区  ›  tcb

并发队列在展开时卡住

  •  -1
  • tcb  · 技术社区  · 7 年前

    我编写了以下代码来实现并发队列。

    template <typename T>
    class ConcurrentQueue
    {
        // Internal storage for a queue element
        struct Element
        {
            T m_elem;
            std::mutex m_mtx;
            std::condition_variable m_cv;
            bool m_hasElement = false;
        };
    
    public:
        // The number of enqueued elements cannot go beyond p_capacity.
        ConcurrentQueue(size_t p_capacity) :
            m_elements(p_capacity),
            m_approxCount(0),
            m_actualCount(0),
            m_front(0),
            m_back(0)
        {}
    
        // Enqueues an element to the queue. Returns true on success and false
        // if the enqueue failed due to the capacity being reached.
        bool Enqueue(T p_element)
        {
            if (++m_approxCount > m_elements.size())
            {
                --m_approxCount;
                return false;
            }
    
            ++m_actualCount;
    
            size_t slot = m_back.fetch_add(1) % m_elements.size();
            auto& element = m_elements[slot];
    
            std::lock_guard<std::mutex> lk(element.m_mtx);
            element.m_elem = std::move(p_element);
            element.m_hasElement = true;
            element.m_cv.notify_one();
    
            return true;
        }
    
        // Dequeues an element from the queue. Returns true on success and false
        // if the dequeue failed due to the queue being empty.
        bool Dequeue(T& p_element)
        {
            size_t count = m_actualCount.load();
            if (count == 0)
            {
                return false;
            }
    
            while (!m_actualCount.compare_exchange_strong(count, count - 1))
            {
                if (count == 0)
                {
                    return false;
                }
            }
    
            size_t slot = m_front.fetch_add(1) % m_elements.size();
            auto& element = m_elements[slot];
    
            std::unique_lock<std::mutex> lk(element.m_mtx);
            element.m_cv.wait(lk, [&element] { return element.m_hasElement; });
            p_element = std::move(element.m_elem);
            element.m_hasElement = false;
            --m_approxCount;
    
            return true;
        }
    
    private:
        // Fixed size vector that stores the elements
        std::vector<Element> m_elements;
    
        // Approx count of number of elements in the queue.
        std::atomic<size_t> m_approxCount;
    
        // Actual count of the number of elements in the queue
        std::atomic<size_t> m_actualCount;
    
        // Index to the front of the queue
        std::atomic<size_t> m_front;
    
        // Index to the back of the queue
        std::atomic<size_t> m_back;
    };
    

    以及下面的测试来验证它的功能

    int main()
    {
        int numElements = 1000;
        int numThreads = 10;
        ConcurrentQueue<int> q(numElements * numThreads / 2);
    
        std::vector<std::thread> enqueueThreads;
        for (int i = 0; i < numThreads; ++i)
        {
            enqueueThreads.emplace_back([&q, i, numElements]
            {
                for (int j = 0; j < numElements; ++j)
                {
                    while (!q.Enqueue(i * numElements + j));
                }
            });
        }
    
        std::atomic<int> toDequeue = numElements * numThreads;
        std::vector<std::thread> dequeueThreads;
        for (int i = 0; i < numThreads; ++i)
        {
            dequeueThreads.emplace_back([&q, &toDequeue]
            {
                while (toDequeue > 0)
                {
                    int element;
                    if (q.Dequeue(element))
                    {
                        --toDequeue;
                    }
                }
            });
        }
    
        for (auto& t : enqueueThreads)
        {
            t.join();
        }
    
        for (auto& t : dequeueThreads)
        {
            t.join();
        }
    }
    

    在调试版本(VS2017)中,测试运行良好,但在零售版本中 main 函数不返回 Dequeue 线程未完成)指示 ConcurrentQueue 实施。里面的虫子是什么 Enqueue 出列 方法?

    1 回复  |  直到 7 年前
        1
  •  0
  •   tcb    7 年前

    这个 Enqueue 如果出列程序没有释放插槽,则方法需要等待插槽释放。 以下代码修复了该问题。

    template <typename T>
    bool ConcurrentQueue<T>::Enqueue(T p_element)
    {
        if (++m_approxCount > m_elements.size())
        {
            --m_approxCount;
            return false;
        }
    
        size_t slot = m_back.fetch_add(1) % m_elements.size();
        auto& element = m_elements[slot];
    
        {
            std::unique_lock<std::mutex> lk(element.m_mtx);
            element.m_cv.wait(lk, [&element] { return !element.m_hasElement; });
            element.m_elem = std::move(p_element);
            element.m_hasElement = true;
            element.m_cv.notify_all();
        }
    
        ++m_actualCount;
        return true;
    }
    
    template <typename T>
    bool ConcurrentQueue<T>::Dequeue(T& p_element)
    {
        size_t count = UINT64_MAX;
        while (!m_actualCount.compare_exchange_strong(count, count - 1))
        {
            if (count == 0)
            {
                return false;
            }
        }
    
        size_t slot = m_front.fetch_add(1) % m_elements.size();
        auto& element = m_elements[slot];
    
        {
            std::unique_lock<std::mutex> lk(element.m_mtx);
            element.m_cv.wait(lk, [&element] { return element.m_hasElement; });
            p_element = std::move(element.m_elem);
            element.m_hasElement = false;
            element.m_cv.notify_all();
        }
    
        --m_approxCount;
        return true;
    }