シングルスレッドよりマルチスレッドでのベクトル内積のほうが遅くなってしまう理由が知りたい

Question

質問内容

下記コードはなんの変哲もない普通の
整数同士のベクトルの内積を行う関数dotと
複数のcpuに作業を分担させて
計算を高速化させることを目論み実装された
内積関数dot_multi_threadの処理速度を
比較するためのコードです。

コードをコンパイルして実行してみましたが
dot_multi_threadでの計算が
dotよりかなり遅かったです。

こちらのサイトには

使用可能なプロセッサが複数ある場合、スレッドは使用可能なコアで同時に実行されます。

と記載されています。
なので下記コードでもdot_multi_thread関数内の
関数オブジェクトparaがスレッドの数だけ同時に動いて
計算が速くなるような気がしたのですが
結果は予想外なものでした。

なぜdot_multi_thread関数での計算の方が
遅くなってしまったのでしょうか。

どうかご回答宜しくお願いいたします。

コード

#include <iostream>
#include <iomanip>
#include <thread> 
#include <mutex> 
#include <chrono>

//    divisible_byで割り切れる最もclose_to_numに近い整数を算出
inline size_t round_up(const size_t close_to_num, const size_t divisible_by)noexcept
{
    return divisible_by == 0 ? close_to_num : close_to_num + divisible_by - (close_to_num % divisible_by);
}

//  threadを用いない内積
inline long long dot(const size_t n, const long long* const a, const long long* const b)noexcept
{
    long long result = 0;
    for(size_t i = 0; i<n; ++i) result += a[i] * b[i];
    return result;
}

//  threadを用いる内積
long long dot_multi_thread(const size_t n, const long long* const a, const long long* const b, const unsigned thread_num)noexcept
{
    size_t npt = n / thread_num;
    long long result = 0;
    std::mutex mtx;

    //  並列で実行する関数オブジェクト
    auto para = [&](const size_t n, const long long* const a, const long long* const b)
    {
        std::lock_guard<std::mutex> lock(mtx);
        result += dot(n, a, b);
    };
    
    for(size_t i = 0; i < thread_num; ++i)
    {
        std::thread(para, npt, a+npt*i, b+npt*i).join();
    }
    
    return result;
}

int main()
{
    const unsigned thread_num = std::thread::hardware_concurrency();
    //const unsigned thread_num = 2;//    使用するスレッド数

    size_t size = 1234;
    size_t true_size = round_up(size, thread_num);
    
    auto* mem1 = new long long[true_size];
    auto* mem2 = new long long[true_size];

    for(unsigned i = 0; i<true_size; ++i)
    {
        if(i < size){
            mem1[i] = i;
            mem2[i] = i;
        }else
        {
            mem1[i] = 0;
            mem2[i] = 0;
        }
    } 
    
    auto start_1 = std::chrono::system_clock::now();  
    auto res1 = dot(true_size, mem1, mem2);
    auto end_1 = std::chrono::system_clock::now();  
    double time_1 = std::chrono::duration_cast<std::chrono::nanoseconds>(end_1-start_1).count(); 
    
    auto start_2 = std::chrono::system_clock::now(); 
    auto res2 = dot_multi_thread(true_size, mem1, mem2, thread_num);
    auto end_2 = std::chrono::system_clock::now();  
    double time_2 = std::chrono::duration_cast<std::chrono::nanoseconds>(end_2-start_2).count(); 
    
    delete[] mem1;
    delete[] mem2;
    
    std::cout << "--------------------------------------------------------------" << std::endl;
    std::cout << std::fixed << std::setprecision(20);
    std::cout << "単一スレッド   :" << res1 << std::endl;
    std::cout << "マルチスレッド :" << res2 << std::endl;
    
    std::cout << "--------------------------------------------------------------" << std::endl;
    std::cout << std::fixed << std::setprecision(0);
    std::cout << "単一スレッド   :" << time_1 << "ナノ秒" << std::endl;
    std::cout << "マルチスレッド :" << time_2 << "ナノ秒" << std::endl;  
    
    return 0;
}

実行結果

--------------------------------------------------------------
単一スレッド   :625599129
マルチスレッド :625599129
--------------------------------------------------------------
単一スレッド   :9901ナノ秒
マルチスレッド :1247756ナノ秒

開発環境の備考

種類	名前	バージョン	備考
os	Linux Mint	20	-
cpu	Intel© Core™ i7-7700 CPU @ 3.60GHz × 4	-	コア数4 スレッド数8
コンパイラ	clang++	10	コンパイルオプションに-pthreadを記載

Answer 1

これでかなり速くなってんじゃねぇかと。

#include <iostream>
#include <iomanip>
#include <thread> 
#include <mutex> 
#include <chrono>
#include <vector>
#include <atomic>

//    divisible_byで割り切れる最もclose_to_numに近い整数を算出
inline size_t round_up(const size_t close_to_num, const size_t divisible_by)noexcept
{
    return divisible_by == 0 ? close_to_num : close_to_num + divisible_by - (close_to_num % divisible_by);
}

//  threadを用いない内積
inline long long dot(const size_t n, const long long* const a, const long long* const b)noexcept
{
    long long result = 0;
    for(size_t i = 0; i<n; ++i) result += a[i] * b[i];
    return result;
}

//  threadを用いる内積
long long dot_multi_thread(const size_t n, const long long* const a, const long long* const b, const unsigned thread_num)noexcept
{
    size_t npt = n / thread_num;
    std::atomic<long long> result = 0;

    //  並列で実行する関数オブジェクト
    auto para = [&](const size_t n, const long long* const a, const long long* const b)
    {
        result += dot(n, a, b);
    };

    std::vector<std::thread> threads;

    for(size_t i = 0; i < thread_num; ++i)
    {
        threads.emplace_back(para, npt, a+npt*i, b+npt*i);
    }

    for ( auto& thr : threads ) {
      thr.join();
    }

    return result;
}

int main()
{
    const unsigned thread_num = std::thread::hardware_concurrency();
    //const unsigned thread_num = 2;//    使用するスレッド数

    size_t size = 1234;
    size_t true_size = round_up(size, thread_num);

    auto* mem1 = new long long[true_size];
    auto* mem2 = new long long[true_size];

    for(unsigned i = 0; i<true_size; ++i)
    {
        if(i < size){
            mem1[i] = i;
            mem2[i] = i;
        }else
        {
            mem1[i] = 0;
            mem2[i] = 0;
        }
    } 

    auto start_1 = std::chrono::system_clock::now();  
    auto res1 = dot(true_size, mem1, mem2);
    auto end_1 = std::chrono::system_clock::now();  
    double time_1 = std::chrono::duration_cast<std::chrono::nanoseconds>(end_1-start_1).count(); 

    auto start_2 = std::chrono::system_clock::now(); 
    auto res2 = dot_multi_thread(true_size, mem1, mem2, thread_num);
    auto end_2 = std::chrono::system_clock::now();  
    double time_2 = std::chrono::duration_cast<std::chrono::nanoseconds>(end_2-start_2).count(); 

    delete[] mem1;
    delete[] mem2;

    std::cout << "--------------------------------------------------------------" << std::endl;
    std::cout << std::fixed << std::setprecision(20);
    std::cout << "単一スレッド   :" << res1 << std::endl;
    std::cout << "マルチスレッド :" << res2 << std::endl;

    std::cout << "--------------------------------------------------------------" << std::endl;
    std::cout << std::fixed << std::setprecision(0);
    std::cout << "単一スレッド   :" << time_1 << "ナノ秒" << std::endl;
    std::cout << "マルチスレッド :" << time_2 << "ナノ秒" << std::endl;  

    return 0;
}

size_t size = 1234;

こいつがもっと大きくないとマルチスレッドのウマミが出ないっしょね。
※ 僕とこ(8-core)では 千倍(1234000)にしてようやくシングルスレッド≒マルチスレッド になりました

Comments

[退会済みユーザー]

epistemeさん
ご回答誠に有難うございます。

確かにベクトルサイズを1234000とした私のコードと貴方のコードを
両方動かしてみましたが結果に差が出ました。

私のコード:
単一スレッド :2467573ナノ秒
マルチスレッド :3096476ナノ秒

貴方のコード:
単一スレッド :2243516ナノ秒
マルチスレッド :1185997ナノ秒

そこで疑問が出てきたのですが
両者のコードはスレッド間で共有する変数をガードするための方法が違うように見えますが
std::atmicとstd::mutexは何が違うのでしょうか。

そこがわかればなぜ高速化ができたのか理解に繋がりそうです。

お返事どうかお待ちしております。

[episteme]

> そこがわかればなぜ高速化ができたのか理解に繋がりそうです。

違います。
あなたのコードは
- スレッドを起こし
- そのスレッドが停止するのを待つ
を繰り返しています。
よーするにマルチスレッドになっていません。同時に複数のスレッドを起こしていませんから。

[退会済みユーザー]

epistemeさん
お返事誠にありがとうございます。

私の質問すべき内容がおかしかったです
失礼いたしました。

恐らくbjnesさんの仰っているマルチスレッドになっていないという事ですね。
bjnesさんへの返信に記載したコードをご覧になってください。

std::threadではオブジェクトが右辺値か左辺値かで挙動が
変わってしまうように見えます。

それがなぜかわかりません。
それがなぜか教えていただけましたら幸いです。

あとstd::atmicとstd::mutexについても宜しくお願いいたします。

どうか御返事お待ちしております。

[episteme]

> std::threadではオブジェクトが右辺値か左辺値かで挙動が
> 変わってしまうように見えます。

どう違うんですか?

[episteme]

> あとstd::atmicとstd::mutexについても宜しくお願いいたします。

atomicによる排他は整数/ポインタの加減乗除程度に限定されるが、mutexより軽い。

[退会済みユーザー]

epistemeさん
お返事誠にありがとうございます。

auto t1 = std::thread(f);
auto t2 = std::thread(f);

t1.join();
t2.join();
このように書けば並列に実行されている(標準出力される数列はおかしい部分がある)

しかし
std::thread(f).join();
std::thread(f).join();
このように書けば並列に実行されていないように見える(標準出力される数列は規則正しく並んでいる)

ということです。

どうか御返事お待ちしております。

[episteme]

join() はスレッドが停止するまで「待つ(次の行に制御が移らない)」からです。

std::thread(f).join(); // [1] スレッドを起こし、それが停止するまで[2]が行われない
std::thread(f).join(); // [2]

[退会済みユーザー]

epistemeさん
お返事誠にありがとうございます。

> 待つ(次の行に制御が移らない)
私はjoin();が実行されてから初めてfが実行され始めると思っていましたがもしかして
auto t1 = std::thread(f);の部分で
すでにfは実行され初めており
join()に移ってからfが終了するまで待つということでしょうか。

もしそうならなぜスレッドとなっていなかったのかイメージできます。

どうか御返事お待ちしております。

[episteme]

> auto t1 = std::thread(f);の部分で
> すでにfは実行され初めており
> join()に移ってからfが終了するまで待つということでしょうか。

YES.

[退会済みユーザー]

epistemeさん

疑問が晴れてよかったです。
std::atmicとstd::mutexの違いの説明や
解釈の確認をしていただきありがとうございました。

Answer 2

スレッド分割に係るマネージメントのコストというのはかなり高いです
これはシングルスレッドから見れば全く無駄な実行コストとなりますんで、たかだか数マイクロsec程度のコードをマルチスレッドにしたところで、そーゆー結果になるってことなんでしょう。

Comments

[退会済みユーザー]

y_waiwaiさん

epistemeさんのご回答にもある通りベクトルサイズが小さすぎて
高速化できた時間よりオーバーヘッドのほうが上回った結果
こうなってしまったようですね。

ご回答誠に有難うございました。

Answer 3

ななめ読みで失礼します。

並列のオーバーヘッドが問題になってそうだと感じました。

  auto para = [&](const size_t n, const long long* const a, const long long* const b)
    {
        std::lock_guard<std::mutex> lock(mtx);
        result += dot(n, a, b);
    };

std::lockguardで同期しているので
result += dot(n, a, b)が実行されるスレッド数が制限されています。

この間のスレッド待ち処理がオーバーヘッドの原因です。

並列処理させる場合はなるべく個々のスレッドが待たないような工夫が必要です。
処理の最初でmutex lockさせるようなプログラムはあまり効率が出ないと思います。

あと、join関数は処理が終わるまで待ちます。
そのため、

   for(size_t i = 0; i < thread_num; ++i)
    {
        std::thread(para, npt, a+npt*i, b+npt*i).join();
    }

はpara処理を一個ずつスレッドを立てて、そのスレッドが終わるまで待って、次のスレッドを立てて、また終わるまで待って、を繰り返す。
になります。あれ、これ、、並列処理されてなくないですか。

並列処理させる部分と同期待ちさせる部分をそれぞれ別に書くことが必要ですね。

Comments

[退会済みユーザー]

bjnesさん
ご回答誠に有難うございます。

#include <iostream>
#include <thread>
#include <vector>

int main()
{
auto f = []()
{
for(unsigned i=0; i<100; i++)std::cout << i <<std::endl;
};

#define PARA// コード切り替え用のスイッチ

#if defined(PARA)// 並列に実行できている(数字の並びがおかしくなる)

auto t1 = std::thread(f);
auto t2 = std::thread(f);

t1.join();
t2.join();

#else// 並列に実行できていない(数字はきれいに並んでいる)

std::thread(f).join();
std::thread(f).join();

#endif

return 0;
}

この上記コード動かしてみて
貴方の仰ると通り並列に動いていなかった事がわかりました。

しかし疑問がございます。
std::threadではオブジェクトが右辺値か左辺値かで挙動が
変わってしまうようですがなぜでしょうか。

私にはいまいちjoinの挙動がイメージできません。

どうか御返事お待ちしております。

[bjnes]

そのまま、プログラムを、素直に読んでください。
.join();が発行された時点でそのスレッドクラスのコンストラクタに登録された処理が終わるまで、
プログラムが一時停止します。

なので、
std::thread(f).join();
std::thread(f).join();
は
ｆ実行→join()待ち→f実行→join()待ちとなります。

一方で、
auto t1 = std::thread(f);
auto t2 = std::thread(f);
は、fを実行してthread情報をt1に入れ、fを実行してthread情報をt２に入れる。
となります。

[bjnes]

後、マルチスレッドプログラムにおいて、計算途中の結果を一つの場所に集積するのはあまり良くないです。

例えば１６スレッドあったとして、f()を並列にする場合、
int result[16];
としてそれぞれのスレッド結果を保存する配列をつくってから、そこに結果を代入。

合計するときは、
まず８スレッドで
result[threadnum] = result[threadnum] +result[threadnum+8]
を実行して、
次に４スレッドで
result[threadnum] = result[threadnum] +result[threadnum+4]
次に２スレッドで
result[threadnum] = result[threadnum] +result[threadnum+2]
次に１スレッドで
result[0] = result[0] +result[１]

とすると、16回も同期待ちせずに、少ない同期待ちで高速に足し算できますよ？

[退会済みユーザー]

bjnesさん
ご回答誠に有難うございます。

epistemeさんに同じことを質問してしまった後なのですが失礼します。
join()はコンストラクタで登録された関数の実行を始め、終了するまで待つ。
ためのメソッドかと考えていましたが

std::threadクラスのコンストラクタに関数が渡された時点で処理は実行され始めており、
その処理が終了するまで待つ役割を果たすメソッドがjoinだとしたら
なぜスレッドとならなかったのかイメージがつくのですが
そういった解釈で正しいのでしょうか。

> 16回も同期待ちせずに、少ない同期待ちで高速に足し算できますよ？
確かに言われてみればそちらのほうが同期待ちの時間は減らせそうです。
教えてくださりありがとうございます。

どうか御返事お待ちしております。

[退会済みユーザー]

bjnesさん

上記の疑問点について
epistemeさんよりjoinの挙動のイメージの解釈が
正しいとのご返答を頂いたの疑問は晴れました。

あと同期待ちを減らす方法についての部分が
思ってもいないことだったの今後役に立ちそうです。

ご回答誠にありがとうございました。

[bjnes]

＞そういった解釈で正しいのでしょうか。
そうですね。正しいです。

あと小規模なベクトル演算ならば、CPU拡張命令を使い倒す方向に動いたほうが良きかと。
性能求めるなら、OpenBLASとかその手のライブラリを習得したほうが良きかと。
マルチスレッドプログラミングになれたいならこのまま、頑張ってください

[退会済みユーザー]

OpenBLASですか...
初耳なので難しそうですが調べてみます。