C并发队列：>1个线程时速度较慢

Question

队列并发两个不同的锁：一个用于 enqueue（） 以保护同时排队的多个线程

如果队列已满，Add（enqueue）将跳过（返回）插入。如果队列为空，则删除（出列）将跳过删除。

我使用doRandon（）生成了一堆0到1之间的随机数。我使用这些数字来决定是否添加/删除。

性能：我已尝试使用静态/动态线程分配测试队列。的执行时间

   //g++ -std=c++0x -pthread -o block blocking.cpp;./block

    #include <iostream>
    #include <thread>
    #include <mutex>
    #include <atomic>
    #include <random>
    #include <stdio.h>  
    #include <time.h>
    #include <stdlib.h>
    #include <ctime>
    #include <condition_variable>
    using namespace std;

    #define NUMBER_OF_THREADS 1
    #define NUMBER_OF_OPERATIONS 10000000
    #define QUEUE_CAPACITY 1000000

    std::vector<double> getRandom();


    template <class T> class BoundedQueue {

        private:
            T * array;  
            int head, tail, capacity;
            std::mutex enqLock, deqLock;
            std::atomic<long> sharedCounter;
            std::condition_variable notEmptyCondition, notFullCondition;    

        public:
            void staticAllocation(double randomNumbers[], int threadID);
            void dynamicAllocation(double randomNumbers[]);
            void add (T x);
            BoundedQueue ();
            void remove ();
    };

    template <class T> BoundedQueue<T>::BoundedQueue () {
        capacity = QUEUE_CAPACITY;
        array = new T[capacity];
        head = 0;
        tail = 0;
        sharedCounter = 0;
    }

    template <class T> void BoundedQueue<T>::add (T x) {

            enqLock.lock();

            if (tail - head == capacity) {
                enqLock.unlock();
                return;
            }
            array[tail % capacity] = x;
            tail++;

            enqLock.unlock();
    }   

    template <class T> void BoundedQueue<T>::remove() {

            deqLock.lock();
            if (tail - head == 0) {
                deqLock.unlock();
                return;
            }
            T result = array [head % capacity];
            head++;

            deqLock.unlock();
    }   

    template <class T> void BoundedQueue<T>::dynamicAllocation(double randomNumbers[]) {

        long i = 0;
        while (i < QUEUE_CAPACITY) {
            i = sharedCounter.fetch_add(1, std::memory_order_relaxed);
            if(randomNumbers[i] <= 0.5) add(0);
            else remove();
        }   
    }

    template <class T> void BoundedQueue<T>::staticAllocation (double randomNumbers[], int threadID) {

        int split = NUMBER_OF_OPERATIONS / NUMBER_OF_THREADS;   
        for (int i = threadID * split; i < (threadID * split) + split; i++) {
            if(randomNumbers[i] <= 0.5) add(0);
            else remove();
        }
    }

    std::vector<double> getRandom() {

        std::vector<double> numbers;
        std::random_device rd;
        std::mt19937 gen(rd());
        std::uniform_real_distribution<> dis(0,1);

        for(int i = 0; i < NUMBER_OF_OPERATIONS; i++) numbers.push_back(dis(gen));

        return numbers;
    }   

    int main () {

        BoundedQueue<int> bQ;
        std::vector<double> temp = getRandom();
        double* randomNumbers = &temp[0];

        std::thread myThreads[NUMBER_OF_THREADS];

        clock_t begin = clock();

        for(int i = 0; i < NUMBER_OF_THREADS; i++) {
                    myThreads[i] = std::thread ( [&] {bQ.dynamicAllocation(randomNumbers); });
        }

        for(int i = 0; i < NUMBER_OF_THREADS; i++) {
            if(myThreads[i].joinable()) myThreads[i].join();
        }   

        clock_t end = clock();

        cout << double(end-begin) * 1000 / CLOCKS_PER_SEC;
        return 0;
    }

Answer 1

这里没有相似之处。请注意，所有的线程几乎都被序列化了，因为您在函数开始时获取了一个锁，并在函数结束时释放了它。

尽管由于缺乏并行性，工作在线程之间进行分配，但锁定/解锁开销占主导地位，总体而言，与单线程相比，执行时间较长。

因此，存在并发性，但没有并行性，我们只是付出了同步的代价，而没有性能优势。