C ++并发队列：> 1个线程较慢

Question

队列与两个不同的锁并发：一个用于 enqueue() 以保护同时入队的多个线程，另一个用于 dequeue() 以达到相同的效果。

Add(enqueue) 只是在队列已满时跳过（返回）插入。 如果队列为空，Remove(dequeue) 会跳过删除。

我使用 doRandom() 生成 0 到 1 之间的随机数。我使用这些数字来决定是否添加/删除。

性能：我尝试使用静态/动态线程分配测试队列。对于两种分配算法，>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

这里没有并行性。请注意，您的所有线程几乎都已序列化，因为您在函数的开头获取锁并在最后释放它。

虽然由于缺乏并行性，工作在线程之间进行了划分，但锁定/解锁的开销占主导地位，并且与单线程相比，总体显示执行时间较长。

所以有并发，但没有没有并行性，我们只是为同步付出了代价，却没有性能优势。