c++ - 使用异步 C++ 的 vector 并行乘法

Question

我正在尝试并行化一段代码，该代码将两个复数 float vector 相乘并对结果求和。为此，我尝试将 std::async 与 future 一起使用。我的想法是将 vector 分成 8 个部分，然后对这 8 个部分中的每一个部分并行执行乘法，然后再将它们相加以获得最终结果。为此，我创建了 8 个 future，每个都包含一个 lambda，用于将两个 vector 相乘并对结果求和。每个 future 都会传递指向 vector 不同位置的指针，该位置表示该特定 future 应作用的 vector 部分。

然而，它似乎并没有给我预期的速度提升，它可能将这部分代码的速度提高了 20-30%，但仅此而已，此外，负载似乎并没有分散到我的核心(4 或 8 个超线程)，但似乎都在一个 100% 的核心上。

我已经包含了下面的代码。任何建议将不胜感激。

size_t size = Input1.size()/8;

std::vector<std::future<complex<float> > > futures;
futures.reserve(8);

for(int i = 0; i<8; ++i)
{
    futures.push_back(std::async( [](complex<float>* pos, complex<float>*pos2, size_t siz)
    {
        complex<float> resum(0,0);
        for(int i = 0; i < siz; ++i)
            resum += pos[i]*pos2[i];
        return resum;
    }, &Input1[i*size], &Input2[i*size], size));
}


complex<float> ResSum(0,0);
for(int i = 0; i < futures.size(); ++i)
    ResSum += futures.at(i).get();

Answer 1

这取决于您向其输入多少数据。

在下面的示例中，4096 个条目将通过一个简单的循环更快。但是对于 1000*4096 个条目，并行版本更快。

因此，您的 20-30% 改进结果可能恰好落在相关硬件的范围内。

这是我使用的测试程序。

第一次运行是简单循环，第二次来自问题，第三次使用 std::launch::async。

Plain       From        With
loop        question    launch::async
First       Second      Third
166         1067        607     
166         614         434     
166         523         509     
265993      94633       66231       
182981      60594       69537       
237767      65731       57256   

这是 live result .

#include <vector>
#include <thread>
#include <future>
#include <complex>
#include <string>
#include <iostream>
#include <chrono>
#include <random>
#include <ratio>

float get_random()
{
    static std::default_random_engine e;
    static std::uniform_real_distribution<> dis(0,1); // rage 0 - 1
    return static_cast<float>(dis(e));
}

void do_tests(float val1, float val2, float val3, float val4, int multiplier)
{
    {
        std::vector<std::complex<float>> Input1(4096*multiplier,std::complex<float>{val1,val2});
        std::vector<std::complex<float>> Input2(4096*multiplier,std::complex<float>{val3,val4});
        std::complex<float> ResSum(0,0);
        auto start{std::chrono::high_resolution_clock::now()};

        size_t size = Input1.size();
        for (int i=0; i<size; ++i) {
            ResSum += Input1[i]*Input2[i];
        }

        auto end{std::chrono::high_resolution_clock::now()};
        auto time_used{end-start};
        std::cout << std::chrono::duration_cast<std::chrono::microseconds>(time_used).count() << "\t\t";
    }

    {
        std::vector<std::complex<float>> Input1(4096*multiplier,std::complex<float>{val1,val2});
        std::vector<std::complex<float>> Input2(4096*multiplier,std::complex<float>{val3,val4});
        std::complex<float> ResSum(0,0);
        auto start{std::chrono::high_resolution_clock::now()};

        size_t size = Input1.size()/8;
        std::vector<std::future<std::complex<float>>> futures;
        futures.reserve(8);

        for (int i = 0; i<8; ++i) {
            futures.push_back(
                std::async(
                    [](std::complex<float>* pos,std::complex<float>*pos2,size_t siz) {
                std::complex<float> resum(0,0);
                for (int i = 0; i < siz; ++i)
                    resum += pos[i]*pos2[i];
                return resum;
            }
                    ,&Input1[i*size],&Input2[i*size],size
                )
                );
        }

        for (int i = 0; i < futures.size(); ++i)
            ResSum += futures.at(i).get();

        auto end{std::chrono::high_resolution_clock::now()};
        auto time_used{end-start};
        std::cout << std::chrono::duration_cast<std::chrono::microseconds>(time_used).count() << "\t\t";
    }


    {
        std::vector<std::complex<float>> Input1(4096*multiplier,std::complex<float>{val1,val2});
        std::vector<std::complex<float>> Input2(4096*multiplier,std::complex<float>{val3,val4});
        std::complex<float> ResSum(0,0);
        auto start{std::chrono::high_resolution_clock::now()};

        size_t size = Input1.size()/8;
        std::vector<std::future<std::complex<float>>> futures;
        futures.reserve(8);

        for (int i = 0; i<8; ++i) {
            futures.push_back(
                std::async(std::launch::async,
                    [](std::complex<float>* pos,std::complex<float>*pos2,size_t siz) {
                std::complex<float> resum(0,0);
                for (int i = 0; i < siz; ++i)
                    resum += pos[i]*pos2[i];
                return resum;
            }
                    ,&Input1[i*size],&Input2[i*size],size
                )
                );
        }

        for (int i = 0; i < futures.size(); ++i)
            ResSum += futures.at(i).get();

        auto end{std::chrono::high_resolution_clock::now()};
        auto time_used{end-start};
        std::cout << std::chrono::duration_cast<std::chrono::microseconds>(time_used).count() << "\t\t";
    }

    std::cout << '\n';

}

int main()
{
    float val1{get_random()};
    float val2{get_random()};
    float val3{get_random()};
    float val4{get_random()};

    std::cout << "First\t\tSecond\t\tThird\n";

    do_tests(val1, val2, val3, val4, 1);
    do_tests(val1, val2, val3, val4, 1);
    do_tests(val1, val2, val3, val4, 1);
    do_tests(val1, val2, val3, val4, 1000);
    do_tests(val1, val2, val3, val4, 1000);
    do_tests(val1, val2, val3, val4, 1000);


}