C++のマルチスレッドプログラミングとは、1つのプログラム内で複数の実行の流れを動かし、複数の処理を並行して進めるプログラミング手法です。
C++では、C++11から標準ライブラリに本格的な並行処理機能が導入されました。
代表的な機能には、次のようなものがあります。
std::thread：スレッドの作成と管理std::mutex：共有データの排他制御std::condition_variable：スレッド間の待機と通知std::atomic：アトミックなデータ操作std::future、std::async：非同期処理と結果の取得std::jthread：停止要求と終了待ちを管理しやすいスレッドstd::semaphore、std::latch、std::barrier：複数スレッドの進行制御
マルチスレッドを利用すると、重い計算を複数のCPUコアへ分散したり、時間のかかる処理をバックグラウンドで実行したりできます。
一方で、複数のスレッドが同じデータへアクセスする場合は、データ競合やデッドロックなどの問題が発生する可能性があります。
そのため、C++のマルチスレッドプログラミングでは、単にスレッドを増やすだけではなく、共有データ、同期、スレッドの寿命を正しく設計することが重要です。
並行処理と並列処理の違い
マルチスレッドを理解するうえでは、「並行処理」と「並列処理」の違いを押さえておく必要があります。
並行処理とは
並行処理とは、複数の処理が一定期間の中で少しずつ進行している状態です。
CPUコアが1つしかない場合でも、OSが短い間隔で実行するスレッドを切り替えることで、複数の処理が同時に進んでいるように見せられます。
たとえば、次の2つの処理があるとします。
- ファイルを読み込む処理
- ユーザー入力を受け付ける処理
CPUがそれぞれを短時間ずつ切り替えて実行すれば、どちらも停止せずに進行しているように見えます。
並列処理とは
並列処理とは、複数の処理が実際に同じ時刻に実行されている状態です。
たとえば、4コアのCPUで4つのスレッドが別々のコアに割り当てられれば、複数の処理を物理的に同時実行できる可能性があります。
ただし、マルチスレッドを使ったからといって、必ず並列実行されるわけではありません。
実際の実行方法は、次のような要因によって変わります。
- CPUのコア数
- 論理プロセッサ数
- OSのスケジューリング
- 同時に動作しているほかのプログラム
- スレッドの待機状態
- 処理内容
- 実行環境
マルチスレッドは、複数の処理を並行して進めるための仕組みであり、実際に並列実行されるかどうかは実行環境に依存します。
マルチスレッドを利用する主な目的
C++でマルチスレッドを利用する目的は、大きく分けて処理性能の向上と応答性の向上です。
処理性能を向上させる
大きな計算処理を複数の小さな処理に分割し、複数のスレッドで処理することで、実行時間を短縮できる場合があります。
たとえば、次のような処理です。
- 画像処理
- 動画エンコード
- 3Dレンダリング
- 科学技術計算
- 大量データの集計
- ファイルの圧縮や展開
- 機械学習の前処理
- シミュレーション
ただし、スレッド数を増やせば必ず高速になるわけではありません。
マルチスレッドには、次のような追加コストがあります。
- スレッドの作成と破棄
- スレッドの切り替え
- mutexの取得と解放
- CPUキャッシュの競合
- スレッド間のデータ受け渡し
- タスク分割と結果統合
処理が小さすぎる場合は、マルチスレッド化による管理コストのほうが大きくなり、かえって遅くなることもあります。
アプリケーションの応答性を向上させる
時間のかかる処理を別スレッドへ移すことで、メインスレッドが停止するのを防げます。
たとえば、GUIアプリケーションでは、メインスレッドが画面描画やユーザー操作を担当し、別スレッドが重い処理を担当します。
メインスレッド
├─ 画面描画
├─ マウスやキーボード入力
└─ ユーザー操作への応答
ワーカースレッド
├─ ファイルの読み込み
├─ ネットワーク通信
└─ 重い計算処理
このように役割を分けることで、ファイル読み込みや計算中に画面が固まるのを防げます。
std::threadによる基本的なスレッド作成
std::threadは、C++でスレッドを直接作成するための基本クラスです。
std::threadオブジェクトを構築すると、指定した関数やラムダ式の実行が新しいスレッド上で開始されます。
基本的なスレッド作成例
#include <iostream>
#include <thread>
void worker()
{
std::cout << "ワーカースレッドで実行中\n";
}
int main()
{
std::thread thread(worker);
std::cout << "メインスレッドで実行中\n";
thread.join();
return 0;
}
このプログラムでは、メインスレッドとは別に、worker()を実行する新しいスレッドが作成されます。
ただし、出力順序は保証されません。
メインスレッドで実行中
ワーカースレッドで実行中
となる場合もあれば、次のようになる場合もあります。
ワーカースレッドで実行中
メインスレッドで実行中
どちらのスレッドが先に動くかは、OSのスケジューラが決定します。
joinの役割
join()は、対象スレッドが終了するまで、呼び出し元のスレッドを待機させます。
thread.join();
先ほどの例では、メインスレッドがワーカースレッドの終了を待っています。
メインスレッド
↓
ワーカースレッドを作成
↓
join()で終了待ち
↓
ワーカースレッドが終了
↓
メインスレッドが再開
std::threadオブジェクトが実行中のスレッドを保持したまま破棄されると、std::terminate()が呼び出されます。
そのため、通常はstd::threadが破棄される前に、join()またはdetach()を呼ぶ必要があります。
joinableによる状態確認
joinable()を使うと、std::threadオブジェクトが実行スレッドを保持しているか確認できます。
#include <thread>
void worker()
{
}
int main()
{
std::thread thread(worker);
if (thread.joinable()) {
thread.join();
}
return 0;
}
ただし、通常の明確な処理では、必ずしも毎回joinable()を確認する必要はありません。
次のように、スレッドを作成した直後に1回だけjoin()することが明らかな場合は、そのまま呼び出せます。
std::thread thread(worker);
thread.join();
joinable()は、次のような複雑なケースで役立ちます。
- 条件によってスレッドを作成する場合
- スレッドオブジェクトがmoveされた可能性がある場合
- すでに
join()またはdetach()された可能性がある場合 - 汎用的な終了処理を作成する場合
detachの役割と注意点
detach()を使うと、スレッドをstd::threadオブジェクトから切り離して独立実行させられます。
#include <chrono>
#include <iostream>
#include <thread>
void worker()
{
std::this_thread::sleep_for(
std::chrono::seconds(1)
);
std::cout << "処理完了\n";
}
int main()
{
std::thread thread(worker);
thread.detach();
std::this_thread::sleep_for(
std::chrono::seconds(2)
);
}
ただし、detach()を呼び出すと、元のstd::threadオブジェクトからスレッドの寿命を管理できなくなります。
切り離した後は、次の操作が困難になります。
- スレッドの終了を待つ
- 処理結果を受け取る
- 例外を受け取る
- 停止を要求する
- スレッドが使用するデータの寿命を管理する
特に、切り離したスレッドがローカル変数を参照するコードは危険です。
void start()
{
int value = 10;
std::thread([&value] {
// start()終了後はvalueが破棄されている可能性がある
std::cout << value << '\n';
}).detach();
}
start()が終了するとvalueは破棄されます。
その後、切り離したスレッドがvalueへアクセスすると、ダングリング参照による未定義動作が発生します。
特別な理由がない限り、detach()よりもjoin()やstd::jthreadを利用する設計が安全です。
スレッドに引数を渡す方法
std::threadには、実行する関数だけでなく、その関数へ渡す引数も指定できます。
値として引数を渡す
#include <iostream>
#include <string>
#include <thread>
void printMessage(std::string message, int count)
{
for (int i = 0; i < count; ++i) {
std::cout << message << '\n';
}
}
int main()
{
std::thread thread(
printMessage,
"Hello",
3
);
thread.join();
}
std::threadへ渡した引数は、基本的に内部へコピーまたはムーブされます。
参照として引数を渡す
変数を参照として渡す場合は、std::ref()を使用します。
#include <functional>
#include <iostream>
#include <thread>
void increment(int& value)
{
++value;
}
int main()
{
int number = 10;
std::thread thread(
increment,
std::ref(number)
);
thread.join();
std::cout << number << '\n';
}
実行結果は次のようになります。
11
ただし、std::ref()は参照先の寿命を延長しません。
参照先の変数がスレッドより先に破棄されると、未定義動作になります。
参照をスレッドへ渡す場合は、参照先がスレッドの終了まで確実に生存することを保証しなければなりません。
ラムダ式を使う
実際のC++コードでは、ラムダ式でスレッド処理を記述することもよくあります。
#include <iostream>
#include <thread>
int main()
{
int value = 100;
std::thread thread([value] {
std::cout << value << '\n';
});
thread.join();
}
この例では、valueを値としてコピーしています。
参照キャプチャを使えば、元の変数を書き換えられます。
#include <iostream>
#include <thread>
int main()
{
int value = 100;
std::thread thread([&value] {
value += 50;
});
thread.join();
std::cout << value << '\n';
}
この例では、join()が完了してからメインスレッドがvalueを読み取っているため、適切に同期されています。
ただし、複数のスレッドが同じ変数へ同時にアクセスする場合は、mutexやatomicなどによる同期が必要です。
データ競合とは
マルチスレッドプログラミングで特に注意しなければならない問題が、データ競合です。
データ競合は、複数のスレッドが同じメモリ位置へ競合するアクセスを行い、適切な同期がない場合に発生します。
一般的には、次の条件が重なるとデータ競合になります。
- 複数のスレッドが同じメモリ位置へアクセスする
- 少なくとも一方が書き込みを行う
- 少なくとも一方が非アトミック操作である
- スレッド間に適切な同期関係がない
通常の非アトミック変数でデータ競合が発生すると、C++では未定義動作になります。
データ競合が発生する例
#include <iostream>
#include <thread>
int counter = 0;
void increment()
{
for (int i = 0; i < 100000; ++i) {
++counter;
}
}
int main()
{
std::thread thread1(increment);
std::thread thread2(increment);
thread1.join();
thread2.join();
std::cout << counter << '\n';
}
期待する結果は次の値です。
200000
しかし、実際には200000より小さい値になったり、別の異常動作が発生したりする可能性があります。
++counterは、C++のメモリモデル上、複数スレッドに対してアトミックな操作ではありません。
概念的には、次の処理を行っています。
1. counterの値を読み込む
2. 値に1を加える
3. counterへ書き戻す
2つのスレッドが同じ値を同時に読み込むと、片方の更新が失われる可能性があります。
counter = 10
スレッドA：10を読み込む
スレッドB：10を読み込む
スレッドA：11を書き込む
スレッドB：11を書き込む
本来は12になるべきところが、11になります。
ただし、実際に何個のCPU命令へ変換されるかは、CPUやコンパイラ、最適化によって異なります。
重要なのは、通常のintに対するインクリメントには、スレッド間の安全性が保証されていないという点です。
std::mutexによる排他制御
std::mutexは、共有データへ同時にアクセスできるスレッドを1つに制限するための同期機構です。
mutexを取得したスレッドだけが、保護された共有データを操作できます。
mutexを使った修正例
#include <iostream>
#include <mutex>
#include <thread>
int counter = 0;
std::mutex counterMutex;
void increment()
{
for (int i = 0; i < 100000; ++i) {
counterMutex.lock();
++counter;
counterMutex.unlock();
}
}
int main()
{
std::thread thread1(increment);
std::thread thread2(increment);
thread1.join();
thread2.join();
std::cout << counter << '\n';
}
counterMutexを取得したスレッドだけがcounterを変更できます。
別のスレッドが同じmutexを取得しようとすると、先に取得したスレッドが解放するまで待機します。
ただし、lock()とunlock()を直接呼び出す方法は、例外や途中のreturnによってunlock()が実行されない危険があります。
そのため、通常はRAII形式のロック管理を使います。
lock_guardを使う
単純な排他制御では、std::lock_guardを利用できます。
void increment()
{
for (int i = 0; i < 100000; ++i) {
std::lock_guard<std::mutex> lock(
counterMutex
);
++counter;
}
}
std::lock_guardは、生成時にmutexをロックし、スコープを抜ける際に自動的にアンロックします。
lock_guardを生成
↓
mutexをロック
↓
共有データを操作
↓
スコープを抜ける
↓
自動的にアンロック
例外が発生してもデストラクタによってロックが解放されるため、手動のunlock()忘れを防げます。
このように、オブジェクトの寿命を利用してリソースを管理する考え方をRAIIと呼びます。
scoped_lockを使う
複数のmutexを一度に取得する場合は、std::scoped_lockが便利です。
#include <mutex>
std::mutex mutexA;
std::mutex mutexB;
void process()
{
std::scoped_lock lock(
mutexA,
mutexB
);
// mutexAとmutexBの両方で保護される処理
}
std::scoped_lockは、渡された複数のmutexを単一の操作として取得する際に、デッドロックを回避するように動作します。
ただし、std::scoped_lockを使えば、プログラム内のすべてのデッドロックを防げるわけではありません。
別の場所でmutexを個別に異なる順序で取得している場合などは、依然としてデッドロックする可能性があります。
std::unique_lockによる柔軟なロック管理
std::unique_lockは、std::lock_guardよりも柔軟なロック管理を行えるクラスです。
#include <mutex>
std::mutex mutex;
void process()
{
std::unique_lock<std::mutex> lock(mutex);
// mutexで保護された処理
lock.unlock();
// mutexを必要としない処理
lock.lock();
// 再びmutexで保護された処理
}
std::unique_lockには、次の特徴があります。
- 任意のタイミングでロックとアンロックができる
- ロック取得を遅らせられる
- ロックの所有権をmoveできる
std::condition_variableと組み合わせられる
単純な排他制御ではstd::lock_guardやstd::scoped_lockを優先し、柔軟な制御が必要な場合にstd::unique_lockを使うのが一般的です。
デッドロックとは
デッドロックとは、複数のスレッドが互いにロックの解放を待ち続け、処理が永久に進まなくなる状態です。
デッドロックが発生する例
#include <mutex>
#include <thread>
std::mutex mutexA;
std::mutex mutexB;
void task1()
{
std::lock_guard<std::mutex> lockA(mutexA);
std::lock_guard<std::mutex> lockB(mutexB);
}
void task2()
{
std::lock_guard<std::mutex> lockB(mutexB);
std::lock_guard<std::mutex> lockA(mutexA);
}
int main()
{
std::thread thread1(task1);
std::thread thread2(task2);
thread1.join();
thread2.join();
}
次の状態になると、処理が進まなくなります。
スレッド1：mutexAを取得
スレッド2：mutexBを取得
スレッド1：mutexBの解放待ち
スレッド2：mutexAの解放待ち
両方のスレッドが相手のmutexを待つため、どちらも処理を再開できません。
デッドロックを防ぐ方法
デッドロック対策として、次の方法があります。
- mutexの取得順序を統一する
- 複数mutexには
std::scoped_lockを使う - ロック範囲を短くする
- ロック中に時間のかかる処理をしない
- ロック中に外部コールバックを呼び出さない
- mutexの所有関係を明確にする
たとえば、重い計算をロックの外で行い、共有データへ反映する部分だけをロックします。
Result result = performHeavyCalculation();
{
std::lock_guard<std::mutex> lock(mutex);
sharedResult = result;
}
これにより、mutexを占有する時間を短くできます。
std::atomicによるアトミック操作
単純なカウンターやフラグを複数スレッドから操作する場合は、std::atomicを利用できます。
atomicカウンターの例
#include <atomic>
#include <iostream>
#include <thread>
std::atomic<int> counter{0};
void increment()
{
for (int i = 0; i < 100000; ++i) {
++counter;
}
}
int main()
{
std::thread thread1(increment);
std::thread thread2(increment);
thread1.join();
thread2.join();
std::cout << counter.load() << '\n';
}
std::atomic<int>に対するインクリメントは、アトミックな読み出し・変更・書き込み操作として実行されます。
atomicが適している処理
std::atomicは、次のような単純な共有状態に適しています。
- カウンター
- 停止フラグ
- 状態フラグ
- 進捗値
- 単純な参照数
- ポインターの交換
std::atomic<bool> stopRequested{false};
複数のatomic操作は一体ではない
atomic変数に対する個々の操作はアトミックですが、複数の操作を組み合わせた処理全体がアトミックになるわけではありません。
たとえば、次のコードには問題があります。
std::atomic<int> balance{100};
void withdraw(int amount)
{
if (balance.load() >= amount) {
balance.fetch_sub(amount);
}
}
load()とfetch_sub()は、それぞれ個別にはアトミックです。
しかし、次の処理全体はアトミックではありません。
残高を確認する
↓
十分なら残高を減らす
2つのスレッドが同時に残高を確認すると、どちらも引き出し可能と判断し、残高が不足する可能性があります。
複数の状態や処理の整合性を維持する必要がある場合は、mutexを使うか、compare_exchangeを用いた適切な再試行処理が必要です。
mutexを使うべきケース
複数の変数を一貫した状態で更新する場合は、mutexのほうが適しています。
struct Account {
int balance;
int transactionCount;
};
balanceとtransactionCountを同時に更新し、両方の整合性を保ちたい場合は、構造体全体をmutexで保護します。
#include <mutex>
std::mutex accountMutex;
Account account;
void deposit(int amount)
{
std::lock_guard<std::mutex> lock(
accountMutex
);
account.balance += amount;
++account.transactionCount;
}
メモリオーダーとは
std::atomicには、メモリ操作の順序を指定するstd::memory_orderがあります。
代表的なものは次のとおりです。
std::memory_order_relaxedstd::memory_order_acquirestd::memory_order_releasestd::memory_order_acq_relstd::memory_order_seq_cst
指定を省略した場合、多くのatomic操作では最も強いstd::memory_order_seq_cstが使用されます。
counter.fetch_add(
1,
std::memory_order_relaxed
);
memory_order_relaxedを使っても、カウンター更新自体の不可分性は維持されます。
弱くなるのは、そのatomic操作と周辺の通常のメモリアクセスとの順序保証です。
単純な統計カウンターのように、ほかのデータとの順序関係が不要な場合は、memory_order_relaxedを利用できることがあります。
ただし、メモリオーダーは非常に難しい分野です。
誤った指定は、特定のCPUや最適化条件でのみ発生する不具合につながります。
明確な理由と検証がない限り、まずはデフォルトのメモリオーダーを使用するのが安全です。
条件変数による待機と通知
std::condition_variableは、ある条件が成立するまでスレッドを効率的に待機させるための仕組みです。
単なるループで条件を監視するビジーウェイトよりも、CPU資源を無駄に消費せずに待機できます。
ビジーウェイトの問題
次のような待機方法は適切ではありません。
while (!ready) {
// readyになるまで繰り返す
}
この方法では、待機中もCPUを使い続けます。
また、readyが通常の変数で、適切な同期がない場合はデータ競合も発生します。
条件変数を使った例
#include <condition_variable>
#include <iostream>
#include <mutex>
#include <thread>
std::mutex mutex;
std::condition_variable condition;
bool ready = false;
void worker()
{
std::unique_lock<std::mutex> lock(mutex);
condition.wait(lock, [] {
return ready;
});
std::cout << "処理を開始します\n";
}
int main()
{
std::thread thread(worker);
{
std::lock_guard<std::mutex> lock(mutex);
ready = true;
}
condition.notify_one();
thread.join();
}
ワーカースレッドは、readyがtrueになるまで待機します。
待機中はmutexを一時的に解放し、通知を受けるとmutexを再取得して条件を確認します。
ワーカースレッド
↓
mutexを取得
↓
readyを確認
↓
falseなので待機
↓
mutexを一時的に解放
↓
通知を受ける
↓
mutexを再取得
↓
readyを再確認
↓
処理を再開
必ず条件を再確認する
条件変数では、通知されていないにもかかわらず待機が解除されるspurious wakeupが起きる可能性があります。
そのため、次のように述語付きのwait()を使います。
condition.wait(lock, [] {
return ready;
});
これは概念的には、次の処理と同じです。
while (!ready) {
condition.wait(lock);
}
また、述語に使用する共有状態は、原則として同じmutexで保護します。
書き込み側だけmutexを使わずにreadyを変更すると、データ競合や通知の取りこぼしにつながる可能性があります。
notify_oneとnotify_all
notify_one()は、待機中のスレッドのうち1つを起こします。
condition.notify_one();
notify_all()は、待機中のすべてのスレッドを起こします。
condition.notify_all();
1つのスレッドだけ処理すればよい場合は、notify_one()が適しています。
notify_all()で多数のスレッドを起こすと、同じmutexの取得を巡って競合が発生する可能性があります。
producer-consumerパターン
条件変数の代表的な利用例が、producer-consumer、つまり生産者・消費者パターンです。
- 生産者スレッド：タスクをキューへ追加する
- 消費者スレッド：キューからタスクを取り出す
#include <condition_variable>
#include <iostream>
#include <mutex>
#include <queue>
#include <thread>
std::queue<int> tasks;
std::mutex tasksMutex;
std::condition_variable tasksCondition;
bool finished = false;
void producer()
{
for (int i = 1; i <= 5; ++i) {
{
std::lock_guard<std::mutex> lock(
tasksMutex
);
tasks.push(i);
}
tasksCondition.notify_one();
}
{
std::lock_guard<std::mutex> lock(
tasksMutex
);
finished = true;
}
tasksCondition.notify_all();
}
void consumer()
{
while (true) {
int task = 0;
{
std::unique_lock<std::mutex> lock(
tasksMutex
);
tasksCondition.wait(lock, [] {
return !tasks.empty() || finished;
});
if (tasks.empty() && finished) {
break;
}
task = tasks.front();
tasks.pop();
}
std::cout
<< "処理中: "
<< task
<< '\n';
}
}
int main()
{
std::thread producerThread(producer);
std::thread consumerThread(consumer);
producerThread.join();
consumerThread.join();
}
重要なのは、タスクをキューから取り出したあと、mutexを解放してから実際の処理を行っている点です。
{
std::unique_lock<std::mutex> lock(
tasksMutex
);
task = tasks.front();
tasks.pop();
}
// 重い処理はロックの外で行う
process(task);
これにより、タスク処理中でも生産者やほかの消費者がキューへアクセスできます。
ただし、複数の生産者が存在する場合は、単一のfinishedフラグだけでは不十分な場合があります。
その場合は、次のような終了管理が必要です。
- 残っている生産者数をカウントする
- キューに
close()の概念を持たせる - 専用の終了メッセージを投入する
- 全体のライフサイクルを管理する
std::asyncとstd::future
スレッドそのものを直接管理する必要がなく、「非同期に関数を実行して、あとで結果を受け取りたい」という場合は、std::asyncとstd::futureを利用できます。
非同期処理の結果を受け取る
#include <future>
#include <iostream>
int calculate()
{
return 100 + 200;
}
int main()
{
std::future<int> future =
std::async(
std::launch::async,
calculate
);
std::cout << "別の処理を実行\n";
int result = future.get();
std::cout << result << '\n';
}
future.get()は、非同期処理が終了していなければ完了まで待機し、完了後に結果を返します。
引数を渡す
#include <future>
#include <iostream>
int add(int a, int b)
{
return a + b;
}
int main()
{
auto future =
std::async(
std::launch::async,
add,
10,
20
);
std::cout << future.get() << '\n';
}
例外を受け取る
非同期関数内で発生した例外は、future.get()を呼び出した場所で再送出されます。
#include <future>
#include <iostream>
#include <stdexcept>
int calculate()
{
throw std::runtime_error(
"計算に失敗しました"
);
}
int main()
{
auto future =
std::async(
std::launch::async,
calculate
);
try {
future.get();
} catch (const std::exception& exception) {
std::cerr
<< exception.what()
<< '\n';
}
}
一方、std::threadのトップレベル関数から例外が外へ漏れると、std::terminate()が呼ばれます。
そのため、std::thread内では例外を捕捉するか、std::promiseなどを使って呼び出し元へ伝える必要があります。
起動方針を省略する場合の注意
次のように起動方針を省略した場合、必ずしも別スレッドで実行されるとは限りません。
auto future = std::async(calculate);
実装は、次のいずれかを選択できます。
- 非同期に実行する
get()やwait()が呼ばれた時点で遅延実行する
確実に非同期実行を要求したい場合は、std::launch::asyncを明示します。
auto future =
std::async(
std::launch::async,
calculate
);
future破棄時の待機に注意する
std::launch::asyncで作成された非同期処理では、対応するstd::futureの破棄時に、処理完了を待つ場合があります。
そのため、次のコードは完全なfire-and-forgetにはなりません。
std::async(
std::launch::async,
[] {
heavyTask();
}
);
一時的なfutureが式の終端で破棄され、その場で非同期処理の完了を待つ可能性があります。
非同期処理を利用する場合は、futureの寿命を明示的に管理する必要があります。
std::jthreadによるスレッド管理
C++20以降では、可能であればstd::threadだけでなくstd::jthreadも検討できます。
std::jthreadには、次の特徴があります。
- 破棄時に停止要求を送る
- 破棄時に自動的に終了を待つ
std::stop_tokenを関数へ渡せるjoin()忘れによるstd::terminate()を防ぎやすい
基本的な使用例
#include <iostream>
#include <thread>
void worker()
{
std::cout << "実行中\n";
}
int main()
{
std::jthread thread(worker);
// スコープ終了時に自動的に終了を待つ
}
std::jthreadのデストラクタは、スレッドがjoinableである場合、概念的に次の処理を行います。
request_stop();
join();
単純にjoin()するだけではなく、まず停止要求を送り、その後でスレッドの終了を待ちます。
stop_tokenを使った協調的停止
#include <chrono>
#include <iostream>
#include <stop_token>
#include <thread>
void worker(std::stop_token stopToken)
{
while (!stopToken.stop_requested()) {
std::cout << "処理中\n";
std::this_thread::sleep_for(
std::chrono::milliseconds(200)
);
}
std::cout << "停止します\n";
}
int main()
{
std::jthread thread(worker);
std::this_thread::sleep_for(
std::chrono::seconds(1)
);
thread.request_stop();
}
request_stop()は、スレッドを強制終了する機能ではありません。
ワーカースレッド側がstop_requested()を確認し、自ら処理を終了します。
この方式を協調的キャンセルと呼びます。
jthreadでも停止できない場合がある
std::jthreadを使っても、ワーカースレッドが停止要求を確認しなければ終了しません。
std::jthread thread([] {
while (true) {
// 停止要求を確認しない永久ループ
}
});
この場合、std::jthreadのデストラクタは停止要求を送りますが、スレッドは終了しません。
そのため、デストラクタ内のjoin()が永遠に戻らない可能性があります。
std::jthreadはスレッドを強制終了する仕組みではなく、停止要求と終了待ちを管理しやすくする仕組みです。
複数スレッドへ計算を分割する例
配列の合計値を2つのスレッドで計算する例です。
#include <cstddef>
#include <iostream>
#include <numeric>
#include <thread>
#include <vector>
long long partialSum(
const std::vector<int>& values,
std::size_t begin,
std::size_t end
)
{
return std::accumulate(
values.begin()
+ static_cast<std::ptrdiff_t>(begin),
values.begin()
+ static_cast<std::ptrdiff_t>(end),
0LL
);
}
int main()
{
std::vector<int> values(
1'000'000,
1
);
const std::size_t middle =
values.size() / 2;
long long result1 = 0;
long long result2 = 0;
std::thread thread1([&] {
result1 = partialSum(
values,
0,
middle
);
});
std::thread thread2([&] {
result2 = partialSum(
values,
middle,
values.size()
);
});
thread1.join();
thread2.join();
std::cout
<< result1 + result2
<< '\n';
}
この例では、各スレッドが異なる変数へ書き込んでいます。
スレッド1 → result1
スレッド2 → result2
メインスレッドは、両方のjoin()が完了した後に結果を読み取ります。
そのため、result1とresult2にmutexやatomicを使う必要はありません。
ただし、join()より前にメインスレッドが結果を読み取ると、データ競合が発生する可能性があります。
std::thread thread([&] {
result1 = 100;
});
std::cout << result1 << '\n'; // 問題あり
thread.join();
マルチスレッド処理では、共有データを細かくロックするよりも、各スレッドに独立したデータを持たせ、最後に結果を統合する設計が有効です。
スレッド数の決め方
利用可能な並行実行数の目安は、std::thread::hardware_concurrency()で取得できます。
#include <iostream>
#include <thread>
int main()
{
const unsigned int count =
std::thread::hardware_concurrency();
std::cout
<< "並行実行数の目安: "
<< count
<< '\n';
}
ただし、この値はあくまでヒントです。
必ずしも次の値を意味するわけではありません。
- 物理CPUコア数
- 最適なスレッド数
- プロセスが実際に利用できるCPU数
- コンテナや仮想環境で割り当てられたCPU数
実装が値を取得できない場合は、0を返すこともあります。
CPU負荷が高い処理
計算中心の処理では、論理CPU数付近を出発点として検証する方法があります。
ワーカースレッド数
≒
論理CPU数
ただし、メインスレッドやほかの処理もCPUを使用するため、常に同数が最適とは限りません。
I/O待ちが多い処理
ネットワーク通信やファイルアクセスなど、待機時間が長い処理では、CPU数より多いスレッドが有効な場合があります。
ただし、スレッドを増やしすぎると、次のコストが増加します。
- スレッドごとのスタックメモリ
- コンテキストスイッチ
- スケジューリング負荷
- mutex競合
- CPUキャッシュミス
- メモリ帯域の競合
実際のスレッド数は、処理内容と実行環境に合わせてベンチマークで決めることが重要です。
スレッドプールとは
処理のたびにstd::threadを作成すると、スレッドの作成と破棄にコストがかかります。
そこで実務では、一定数のワーカースレッドを事前に作成し、タスクをキューへ投入するスレッドプールがよく利用されます。
タスクを投入
↓
タスクキュー
↓
┌──────────┐
│ Worker 1 │
│ Worker 2 │
│ Worker 3 │
│ Worker 4 │
└──────────┘
スレッドプールには、次の利点があります。
- スレッドを毎回作成しなくてよい
- 同時実行数を制限できる
- タスク管理を一元化できる
- 大量のスレッド生成を防げる
安全性を考慮した簡易スレッドプール
次のコードは、学習用に簡略化したスレッドプールです。
#include <condition_variable>
#include <cstddef>
#include <functional>
#include <mutex>
#include <queue>
#include <stdexcept>
#include <stop_token>
#include <thread>
#include <utility>
#include <vector>
class ThreadPool {
public:
explicit ThreadPool(std::size_t threadCount)
{
if (threadCount == 0) {
throw std::invalid_argument(
"threadCount must be greater than zero"
);
}
workers_.reserve(threadCount);
for (std::size_t i = 0;
i < threadCount;
++i) {
workers_.emplace_back(
[this](std::stop_token stopToken) {
workerLoop(stopToken);
}
);
}
}
~ThreadPool()
{
{
std::lock_guard<std::mutex> lock(
mutex_
);
stopping_ = true;
}
for (auto& worker : workers_) {
worker.request_stop();
}
condition_.notify_all();
// workers_はデストラクタ本体終了後に破棄される。
// 各jthreadのデストラクタがjoin()を行う。
}
ThreadPool(const ThreadPool&) = delete;
ThreadPool& operator=(
const ThreadPool&
) = delete;
void submit(std::function<void()> task)
{
if (!task) {
throw std::invalid_argument(
"task must not be empty"
);
}
{
std::lock_guard<std::mutex> lock(
mutex_
);
if (stopping_) {
throw std::runtime_error(
"cannot submit to a stopped ThreadPool"
);
}
tasks_.push(std::move(task));
}
condition_.notify_one();
}
private:
void workerLoop(std::stop_token stopToken)
{
while (true) {
std::function<void()> task;
{
std::unique_lock<std::mutex> lock(
mutex_
);
condition_.wait(
lock,
[this, &stopToken] {
return stopping_
|| stopToken.stop_requested()
|| !tasks_.empty();
}
);
if (
(
stopping_
|| stopToken.stop_requested()
)
&& tasks_.empty()
) {
return;
}
task = std::move(
tasks_.front()
);
tasks_.pop();
}
try {
task();
} catch (...) {
// 本番コードでは、ログ記録や
// futureへの例外格納などを行う。
}
}
}
std::queue<std::function<void()>> tasks_;
std::mutex mutex_;
std::condition_variable condition_;
bool stopping_ = false;
// 最後に宣言することで、
// 破棄時には最初にworkers_が破棄される。
std::vector<std::jthread> workers_;
};
使用例は次のとおりです。
#include <iostream>
int main()
{
ThreadPool pool(4);
for (int i = 0; i < 10; ++i) {
pool.submit([i] {
std::cout
<< "Task: "
<< i
<< '\n';
});
}
}
メンバーの宣言順に注意する
C++のクラスメンバーは、宣言された順番とは逆の順番で破棄されます。
そのため、ワーカースレッドが利用するキュー、mutex、条件変数は、ワーカースレッドよりも長く生存しなければなりません。
次の順番で宣言すると、workers_が最初に破棄されます。
std::queue<std::function<void()>> tasks_;
std::mutex mutex_;
std::condition_variable condition_;
bool stopping_ = false;
std::vector<std::jthread> workers_;
破棄順は次のとおりです。
1. workers_
2. stopping_
3. condition_
4. mutex_
5. tasks_
これにより、ワーカースレッドの終了待ちが完了するまで、mutex、条件変数、タスクキューが生存します。
スレッドプールでは終了方針を決める
スレッドプールを実務で使用する場合は、停止時の挙動を明確にしなければなりません。
主な検討事項は次のとおりです。
- 残ったタスクを最後まで実行するか
- 未実行タスクを破棄するか
- 実行中のタスクをキャンセルできるか
- タスクの戻り値をどう返すか
- タスク内の例外をどう扱うか
- キューの最大サイズを設けるか
- タスクの優先順位を設けるか
- 停止中に
submit()された場合どうするか
学習用の簡易実装を、そのまま本番環境で利用するのは避けるべきです。
std::shared_mutexによる読み書き制御
読み取りが多く、書き込みが少ない共有データでは、std::shared_mutexを利用できる場合があります。
- 読み取り：複数スレッドが同時に取得できる
- 書き込み：1つのスレッドだけが排他的に取得できる
#include <mutex>
#include <shared_mutex>
#include <string>
#include <unordered_map>
class Cache {
public:
std::string get(
const std::string& key
) const
{
std::shared_lock lock(mutex_);
const auto iterator =
values_.find(key);
if (iterator == values_.end()) {
return {};
}
return iterator->second;
}
void set(
const std::string& key,
const std::string& value
)
{
std::unique_lock lock(mutex_);
values_[key] = value;
}
private:
mutable std::shared_mutex mutex_;
std::unordered_map<
std::string,
std::string
> values_;
};
get()では共有ロックを取得するため、複数の読み取り処理を同時に実行できます。
set()では排他ロックを取得するため、書き込み中はほかの読み取りや書き込みが待機します。
ただし、std::shared_mutexが常にstd::mutexより高速とは限りません。
読み書きロックには追加の管理コストがあります。
また、実装や利用状況によっては、読み取りが連続することで書き込み側が長時間待たされる可能性もあります。
利用する場合は、実際の処理で性能を計測する必要があります。
マルチスレッドで起こりやすい問題
マルチスレッドプログラミングでは、データ競合以外にもさまざまな問題が発生します。
データ競合
複数のスレッドが、同期なしで同じデータへ競合するアクセスを行う問題です。
主な対策は次のとおりです。
std::mutexを使うstd::atomicを使う- スレッドごとに独立したデータを持つ
- 不変オブジェクトを利用する
- 共有データへのアクセスをカプセル化する
デッドロック
複数のスレッドが互いのロック解放を待ち続ける問題です。
主な対策は次のとおりです。
- mutexの取得順序を統一する
std::scoped_lockを利用する- ロック範囲を短くする
- ロック中に外部関数を呼び出さない
- ロック中にコールバックを実行しない
ライブロック
スレッド自体は動作しているものの、互いに譲り合うなどして処理が進まない状態です。
たとえば、2人が狭い通路ですれ違う際、同時に同じ方向へ避け続け、いつまでも通れない状態に似ています。
スターベーション
特定のスレッドだけが、長時間リソースを取得できない状態です。
あるスレッドがmutexを頻繁に取得したり、スケジューリング上の不均衡があったりすると発生する可能性があります。
false sharing
異なるスレッドが別々の変数を操作していても、それらの変数が同じCPUキャッシュラインに配置されていると、性能が低下することがあります。
#include <atomic>
struct Counters {
std::atomic<long long> counter1{0};
std::atomic<long long> counter2{0};
};
counter1とcounter2を別々のCPUコアから頻繁に更新すると、キャッシュラインの所有権移動が繰り返される可能性があります。
C++17以降では、std::hardware_destructive_interference_sizeを利用できる実装もあります。
#include <atomic>
#include <new>
struct alignas(
std::hardware_destructive_interference_size
) AlignedCounter {
std::atomic<long long> value{0};
};
ただし、これを使えば必ずfalse sharingを完全に解消できるわけではありません。
過剰なパディングによって、メモリ使用量やキャッシュ効率が悪化する可能性もあります。
性能対策は、実際の計測結果に基づいて行うことが重要です。
スレッドセーフとは
スレッドセーフとは、複数のスレッドから同時に利用しても、仕様どおり正しく動作する性質を指します。
ただし、「スレッドセーフ」という言葉だけでは、どの操作が安全なのか分からない場合があります。
異なるオブジェクトなら同時利用できる
Object object1;
Object object2;
object1とobject2を別のスレッドから操作できるケースです。
同じクラスでも、同じインスタンスを同時に操作できるとは限りません。
同じオブジェクトでも読み取りなら安全
同じオブジェクトを複数スレッドから読み取るだけなら安全でも、書き込みが混ざると危険な場合があります。
同じオブジェクトへ同時に書き込める
内部で適切な同期処理が行われており、複数スレッドから同時に更新しても安全なケースです。
ライブラリやクラスの説明で「スレッドセーフ」と書かれている場合は、次の点を確認する必要があります。
- 同じインスタンスへ同時アクセスできるか
- 読み取りだけが安全なのか
- 書き込みも安全なのか
- 呼び出し側で同期が必要なのか
- コールバック中の再入が許可されるか
安全なマルチスレッド設計の基本
マルチスレッドの安全性を高めるには、同期機能だけでなく、設計そのものを単純にすることが重要です。
共有データをできるだけ減らす
マルチスレッドの問題の多くは、共有された可変データから発生します。
可能であれば、各スレッドに独立したデータを持たせ、最後に結果だけを統合します。
共有データを全スレッドで更新
↓
ロックが必要
↓
競合やデッドロックの可能性
データを分割して各スレッドで処理
↓
最後に結果を統合
↓
同期箇所が少ない
共有状態を減らすことは、安全性だけでなく性能の向上にもつながります。
mutexが保護するデータを明確にする
どのmutexが、どのデータを保護しているのかを明確にします。
#include <mutex>
class Counter {
public:
void increment()
{
std::lock_guard<std::mutex> lock(
mutex_
);
++value_;
}
int get() const
{
std::lock_guard<std::mutex> lock(
mutex_
);
return value_;
}
private:
mutable std::mutex mutex_;
int value_ = 0;
};
value_へ外部から直接アクセスできないようにカプセル化することで、ロック漏れを防ぎやすくなります。
ロック中の処理を短くする
次のような処理は、可能な限りロックの外で行います。
- 重い計算
- ファイル入出力
- ネットワーク通信
- 長時間の待機
- ログ出力
- ユーザー定義コールバック
ロック時間が長いほど、ほかのスレッドが待たされ、性能低下やデッドロックの可能性が高まります。
ロック中に未知のコードを呼び出さない
次のように、mutexを保持したまま外部のコールバックを呼び出すコードは危険です。
std::lock_guard<std::mutex> lock(mutex);
callback();
callback()の内部で同じmutexを取得すると、デッドロックする可能性があります。
必要なデータやコールバックをロック中にコピーし、ロックを解放してから呼び出します。
Callback callbackCopy;
{
std::lock_guard<std::mutex> lock(mutex);
callbackCopy = callback;
}
callbackCopy();
スレッドの寿命をRAIIで管理する
スレッドの所有者と終了責任を明確にします。
std::jthreadや、スレッドを所有する専用クラスを使うことで、終了待ちの漏れを防ぎやすくなります。
生のstd::threadオブジェクトを広範囲に受け渡すと、誰がjoin()するのか分かりにくくなります。
コンパイル方法
GCCやClangでは、スレッド機能を使う際に-pthreadを指定するのが一般的です。
GCCでコンパイルする
g++ \
-std=c++20 \
-O2 \
-Wall \
-Wextra \
-pthread \
main.cpp \
-o main
Clangでコンパイルする
clang++ \
-std=c++20 \
-O2 \
-Wall \
-Wextra \
-pthread \
main.cpp \
-o main
Microsoft Visual C++でコンパイルする
Visual Studioの開発者コマンドプロンプトでは、次のようにコンパイルできます。
cl /std:c++20 /EHsc /W4 main.cpp
利用できるC++20以降の機能は、コンパイラと標準ライブラリのバージョンによって異なります。
std::jthread、停止トークン、セマフォなどを利用する場合は、使用環境の対応状況を確認する必要があります。
マルチスレッドプログラムのデバッグ
マルチスレッドの不具合は、通常のプログラムよりも再現が難しい傾向があります。
スレッドの実行順序が毎回変わるため、次のような現象が起こります。
- デバッグビルドでは発生しない
- 最適化すると発生する
- 高負荷時だけ発生する
- 特定のCPUでだけ発生する
- 数時間に1回しか発生しない
- ログを追加すると発生しなくなる
ThreadSanitizerを使う
GCCやClangでは、ThreadSanitizerを利用してデータ競合を検出できる場合があります。
g++ \
-std=c++20 \
-g \
-O1 \
-fsanitize=thread \
-fno-omit-frame-pointer \
-pthread \
main.cpp \
-o main
作成したプログラムを実行します。
./main
ThreadSanitizerは、データ競合の発見に非常に有効です。
ただし、すべての並行処理バグを検出できるわけではありません。
主な制限は次のとおりです。
- 実行されなかった処理経路のバグは検出できない
- すべてのデッドロックを検出するわけではない
- ライブロックやスターベーションを完全には検出できない
- 実行速度やメモリ使用量が大幅に増える
- 利用できるプラットフォームに制限がある
- 低レベルな同期処理では誤検出や見逃しの可能性がある
ThreadSanitizerで警告が出なかったからといって、プログラムが完全にスレッドセーフであるとは限りません。
設計レビュー、ストレステスト、停止処理のテスト、タイムアウトテストなども必要です。
C++マルチスレッド機能の使い分け
C++の代表的な並行処理機能は、次のように使い分けられます。
| 目的 | 主な機能 |
|---|---|
| スレッドを直接作成する | std::thread |
| 停止要求と終了待ちを管理する | std::jthread |
| 共有データを排他的に保護する | std::mutex |
| 単純なロックをRAIIで管理する | std::lock_guard |
| 複数mutexをまとめて取得する | std::scoped_lock |
| 柔軟にロックを管理する | std::unique_lock |
| 単純な値をアトミックに操作する | std::atomic |
| 条件成立まで待機する | std::condition_variable |
| 非同期処理の結果を受け取る | std::future、std::async |
| 読み取りを同時実行する | std::shared_mutex |
C++マルチスレッドプログラミングのまとめ
C++のマルチスレッドプログラミングでは、スレッドを作成すること自体よりも、共有データとスレッドの寿命を正しく管理することが重要です。
特に、次の点を意識する必要があります。
共有変数には必ず同期方法を検討する
同じ変数を複数スレッドから読み書きする場合は、mutexやatomicなどによる同期が必要です。
通常の変数へ同期なしでアクセスすると、データ競合による未定義動作が発生する可能性があります。
RAII形式のロックを優先する
lock()とunlock()を直接呼び出すよりも、std::lock_guard、std::scoped_lock、std::unique_lockを利用するほうが安全です。
例外や途中のreturnがあっても、自動的にロックが解放されます。
デッドロックを防ぐ
複数のmutexを利用する場合は、取得順序を統一し、ロック範囲を短くします。
ロック中に外部コールバックや時間のかかる処理を実行しないことも重要です。
atomicを過信しない
std::atomicは個々の操作をアトミックにしますが、複数の操作からなる処理全体の整合性を自動的に保証するものではありません。
複数の変数や条件を一貫して更新する場合は、mutexの利用を検討します。
スレッドの寿命を明確にする
スレッドを作成したら、誰が終了を待つのかを明確にします。
std::jthreadは、停止要求と終了待ちを管理しやすくしますが、ワーカースレッド側が停止要求へ協力する必要があります。
共有状態を減らす
最も安全で保守しやすい方法は、共有状態を減らすことです。
各スレッドに独立したデータを持たせ、最後に結果を統合する設計は、ロックの数を減らし、データ競合やデッドロックを防ぎやすくします。
C++のマルチスレッドプログラミングでは、高度な同期技術を多用することよりも、共有データを減らし、単純で明確な構造にすることが、正確性、保守性、性能のすべてにおいて重要です。
以上、C++によるマルチスレッドプログラミングについてでした。
最後までお読みいただき、ありがとうございました。
