c++ - 线程池 : single thread vs callback tp vs future tp 的计时测试

Question

我在 https://github.com/spakai/threadpool_future 中有 3 个针对此代码 ThreadPool 的单元测试用例

   class ThreadPoolTest : public Test {
    public:
        ThreadPool pool;
        std::condition_variable wasExecuted;
        std::mutex m;
        std::vector<std::shared_ptr<std::thread>> threads; 

        unsigned int count{0};

        void incrementCountAndNotify() {
            std::unique_lock<std::mutex> lock(m);
            ++count;
            std::cout << count << std::endl;
            wasExecuted.notify_all();
        }

        void waitForNotificationOrFailOnTimeout(unsigned expectedCount, int milliseconds=80000) {
            std::unique_lock<std::mutex> lock(m);
            ASSERT_THAT(wasExecuted.wait_for(lock, std::chrono::milliseconds(milliseconds), [&] { return count == expectedCount; }), Eq(true));      

        } 

        bool hasDuplicates(const std::vector<int> & birthdays) {
            std::set<int> uniqueBirthdays(birthdays.begin(), birthdays.end());
            return (uniqueBirthdays.size() != birthdays.size());
        }

        std::vector<int> generateNumbers(const int popSize) {
            std::vector<int> list;
            std::random_device rd;
            std::default_random_engine dre(rd());
            std::uniform_int_distribution<int> di(0,365);
            for(int i{0}; i < popSize ; i++) {
                list.push_back(di(dre));
            } 
            return list;
        }

        void TearDown() override {
            for (auto& t: threads) t->join();
        }
};



TEST_F(ThreadPoolTest,TimingTestWithFuture) {
    pool.start(4);
    std::vector<std::future<unsigned long long>> results;
    auto work = [](int n) {
      unsigned long long factorial = 1;
      for(int i = 1; i <=n; ++i) {
        factorial *= i;
      }

      return factorial;

    };


    TestTimer timer("4-sized-TP with Future",0);
    for (int i = 5; i < 60 ; i++) {
        results.push_back(pool.submit(work,i));
    }


    for(unsigned int i = 0; i< results.size(); i++) {
        results.at(i).get();
    }
}

TEST_F(ThreadPoolTest,TimingTestWithCallback) {
    pool.start(4);
    std::vector<unsigned long long> results;
    TestTimer timer("4-sized-TP-Callback",0);
    for (int n = 5; n < 60 ; n++) {
        auto work = [&]() {
            unsigned long long factorial = 1;
            for(int i = 1; i <=n; ++i) {
              factorial *= i;
            }
            {
                std::lock_guard<std::mutex> guard(m); 
                results.push_back(factorial);
            }
            incrementCountAndNotify();
        };

        pool.add(work);
    }

    waitForNotificationOrFailOnTimeout(55);
}

TEST_F(ThreadPoolTest,TimingTestWithoutTP) {

    std::vector<unsigned long long> results;
    auto work = [](int n) {
      unsigned long long factorial = 1;
      for(int i = 1; i <=n; ++i) {
        factorial *= i;
      }

      return factorial;

    };


    TestTimer timer("In Sequence",0);
    for (int i = 5; i < 60 ; i++) {
        results.push_back(work(i));
    }

     for(unsigned int i = 0; i< results.size(); i++) {
        results.at(i);
    }

}

我在一台 4 CPU 机器上运行。我得到的计时结果显示单线程最快，而返回 future 的线程最慢。

4-sized-TP with Future Time taken = 2.364ms

4-sized-TP-Callback 所用时间 = 1.103ms

按顺序花费的时间 = 0.026 毫秒

我原以为时间顺序会倒过来。我做测试的方式是错误的吗？还是我的代码？

新测试会占用 CPU 资源

TEST_F(ThreadPoolTest,BirthdayParadoxInSequenceTimingTest) {

    std::vector<int> results;

    TestTimer timer("Birthday Paradox :: In Sequence",0);

    std::vector<int> popList = {10,23,30,40,50,60,70,80,90,100,120,150};
    for(auto it=popList.begin(); it!=popList.end(); ++it) {
        int id = *it;
        int dup{0};
        for(int i{0}; i< 100000; i++) {
            auto list = generateNumbers(id);
            if(hasDuplicates(list)) ++dup;
        }

        results.push_back(dup);
    }

        for(unsigned int i = 0; i< results.size(); i++) {
            results.at(i);
        }
}

TEST_F(ThreadPoolTest,BirthdayParadoxTPWithFutureTimingTest) {
    std::vector<int> popList = {10,23,30,40,50,60,70,80,90,100,120,150};

    pool.start(4);
    std::vector<std::future<int>> results;

    TestTimer timer("4-sized-TP with Future",0);

    for(auto it=popList.begin(); it!=popList.end(); ++it) {
        int id = *it;
        auto work = [&](int pop) {
            int dup{0};
            for(int i{0}; i < 100000 ; i++) {
                auto list = generateNumbers(pop);
                if(hasDuplicates(list)) ++dup; 
            }

            return dup;

        };

        results.push_back(pool.submit(work,id));        
    } 

    for(unsigned int i = 0; i< results.size(); i++) {
        results.at(i).get();
    }
} 



TEST_F(ThreadPoolTest,BirthdayParadoxTPWithCallBackTimingTest) {
    std::vector<int> popList = {10,23,30,40,50,60,70,80,90,100,120,150};

    pool.start(4);
    std::vector<int> results;

    TestTimer timer("4-sized-TP with Callback",0);

    for(auto it=popList.begin(); it!=popList.end(); ++it) {
        int id = *it;
        auto work = [&,id]() {
            int dup{0};
            for(int i{0}; i < 100000 ; i++) {
                auto list = generateNumbers(id);
                if(hasDuplicates(list)) ++dup; 

                {
                    std::lock_guard<std::mutex> guard(m); 
                    results.push_back(dup);

                }
            }

            incrementCountAndNotify();
        };

        pool.add(work);       
    } 

    waitForNotificationOrFailOnTimeout(12);
}

结果还是出乎我的意料

按顺序花费的时间 = 37555.7ms

4-sized-TP with Future Time taken = 62544.8ms

4-sized-TP with Callback Time = 62563.6ms

完整代码和测试在 https://github.com/spakai/threadpool_future 中

Answer 1

你选择的生日悖论问题对 cpu 来说也不是一个具有挑战性的任务。但是要理解您看到的问题，我们首先必须对代码进行一些更改。

我们想测量算法完成所需的时间。内存分配是昂贵的，应该避免在程序中经常重复的部分。创建 vector 或增加它们的大小总是会触发内存分配。创建集合也是如此。为了删除 momory 分配，我将您的代码修改为如下所示:

#include "gmock/gmock.h"
#include <chrono>
#include <condition_variable>
#include <mutex>
#include <random>
#include <set>
#include <vector>

#include "ThreadPool.h"
#include "TestTimer.h"


const unsigned int runs = 100000;

using namespace testing;

class ThreadPoolTest : public Test {
    public:
        ThreadPool pool;
        std::condition_variable wasExecuted;
        std::mutex m;
        std::mutex n;
        std::vector<std::shared_ptr<std::thread>> threads;
        std::vector<int> popList = {10,11,12,23};

        unsigned int count{0};

        void incrementCountAndNotify() {
            {
                std::unique_lock<std::mutex> lock(m);
                ++count;
            }
            wasExecuted.notify_all();
        }

        void waitForNotificationOrFailOnTimeout(unsigned expectedCount, int milliseconds=80000) {
            std::unique_lock<std::mutex> lock(m);
            ASSERT_THAT(wasExecuted.wait_for(lock, std::chrono::milliseconds(milliseconds), [&] { return count == expectedCount; }), Eq(true));

        }

        bool hasDuplicates(const std::vector<int> & birthdays) {
            //This way to check for duplicates is very expensive, since it allocates new memory and copies all values around
            //std::set<int> uniqueBirthdays(birthdays.begin(), birthdays.end());
            //return (uniqueBirthdays.size() != birthdays.size());
            for(unsigned int i = 0; i < birthdays.size(); i++) {
                for(unsigned int j = i+1; j < birthdays.size(); j++) {
                    if(birthdays[i]==birthdays[j]) return true;
                }
            }
            return false;
        }

        //I added the parameter list, to avoid the allocation of new memory
        //The list will also have the needed size, so that we dont need to it here
        std::vector<int> generateNumbers(std::vector<int>& list) {
            //It is not exactly specified how the random_device works, it may read from /dev/random, which can not be done in parallel
            //To make the measurements compareable over multiple machiens i removed this code
            //std::random_device rd;
            std::default_random_engine dre(0);
            std::uniform_int_distribution<int> di(0,365);
            int counter = 0;
            for(int& i : list) {
                i = di(dre);
            }
            return list;
        }

        void TearDown() override {
            for (auto& t: threads) t->join();
        }
};


TEST_F(ThreadPoolTest,BirthdayParadoxInSequenceTimingTest) {

    std::vector<int> results;

    TestTimer timer("Birthday Paradox :: In Sequence",0);

    for(auto it=popList.begin(); it!=popList.end(); ++it) {
        std::cout << "TID " << std::this_thread::get_id() << std::endl;

        int id = *it;
        int dup{0};
        std::vector<int> list(id); //Allocate memory in the right size only once for all 100000 runs
        for(int i{0}; i < runs ; i++) {
                generateNumbers(list);
            if(hasDuplicates(list)) ++dup;
        }

        results.push_back(dup); //This push_back is ok, since it is only called 4 times in total
    }

        for(unsigned int i = 0; i< results.size(); i++) {
            results.at(i);
        }
}

TEST_F(ThreadPoolTest,BirthdayParadoxTPWithFutureTimingTest) {
    pool.start(4);
    std::vector<std::future<int>> results;

    TestTimer timer("4-sized-TP with Future",0);

    for(auto it=popList.begin(); it!=popList.end(); ++it) {
        int id = *it;
        auto work = [&](int pop) {
            std::cout << "TID " << std::this_thread::get_id() << std::endl;

            int dup{0};
            std::vector<int> list(pop); //Same as above
            for(int i{0}; i < runs ; i++) {
                generateNumbers(list);
                if(hasDuplicates(list)) ++dup;
            }

            return dup;

        };

        results.push_back(pool.submit(work,id));
    }

    for(unsigned int i = 0; i< results.size(); i++) {
        results.at(i).get();
    }
}


TEST_F(ThreadPoolTest,BirthdayParadoxTPWithCallBackTimingTest) {
    pool.start(4);
    std::vector<int> results;

    TestTimer timer("4-sized-TP with Callback",0);

    for(auto it=popList.begin(); it!=popList.end(); ++it) {
        int id = *it;
        auto work = [&,id]() {
            std::cout << "TID " << std::this_thread::get_id() << std::endl;

            int dup{0};
            std::vector<int> list(id); //Same here too
            for(int i{0}; i < runs ; i++) {
                generateNumbers(list);
                if(hasDuplicates(list)) ++dup;

                        {
                        std::lock_guard<std::mutex> guard(n);
                        results.push_back(dup);
                    }
            }

            incrementCountAndNotify();
        };

        pool.add(work);
    }
    waitForNotificationOrFailOnTimeout(4);
}

现在，我们的内存管理正确了，我们可以开始推理运行时了。我使用 2 个内核和超线程运行代码，因此如果我们使用多线程，我们期望加速为 2 或更高。让我们看看结果:

Birthday Paradox :: In Sequence Time taken = 680.96ms
4-sized-TP with Future Time taken = 1838.28ms
4-sized-TP with Callback Time taken = 1861.07ms

如果我将线程池中的线程数量限制为一个，那么所有版本的运行时间几乎相同。

我们看到这种不直观行为的原因是，问题是内存限制的。速度下降的原因在于检查重复项。

for(unsigned int i = 0; i < birthdays.size(); i++) {
    for(unsigned int j = i+1; j < birthdays.size(); j++) {
        if(birthdays[i]==birthdays[j]) return true;
    }
}

生日的访问在内存中很好地对齐。如果有多个线程在运行，则算法不会提高速度，因为所有线程都只是在等待值。更糟糕的是，不同的线程正在从不同的位置读取，因此，它们可能会丢弃可能被其他线程使用的缓存行。这就是性能下降的原因。