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

标签 c++ multithreading c++11 c++14 threadpool

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

   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 大小的 TP， future 时间 = 2.364ms

4-sized-TP-Callback 耗时 = 1.103ms

序列中花费的时间 = 0.026ms

我原以为时间顺序会相反。 难道是我做测试的方式不对？还是我的代码？

新的测试会占用大量 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 大小的 TP， future 时间 = 62544.8ms

4-sized-TP，回调时间= 62563.6ms

完整的代码和测试位于 https://github.com/spakai/threadpool_future

最佳答案

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

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

#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

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

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

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;
    }
}

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

关于c++ - 线程池的计时测试 : single thread vs callback tp vs future tp，我们在Stack Overflow上找到一个类似的问题： https://stackoverflow.com/questions/43860261/

上一篇：c++ - 如何在 C++11 中创建/传递完成处理程序回调作为函数参数？

下一篇：c++ - 洪水填充算法 - 房间面积

相关文章：

linux - 使用在双引号中包含空格的参数调用 std::system

c++ - 如何使可变参数模板类方法将函数指针作为从函数模板派生的类型的参数？

c++ - 继承层次结构中的特定类作为 boost::signals2 回调中的类型

c# - 在STA线程上显示忙碌指示器

java - 如何让Web Service使用多线程？

java - 下载数据时 UI-Thread 似乎滞后

c++ - 括号序列的类型

c++ - 一个奇怪的字符出现在我的字符数组的第一个

c++ - “Default member initializer needed within definition of enclosing class outside of member functions” - 我的代码格式不正确吗？

c++ - Qt 创建者 : Widget is not added to layout

©2024 IT工具网  联系我们