shared_mutex（C++17）
mutexで保護されたデータを多数のスレッドがアクセスするとパフォーマンスが低下します。が、データの変更を伴わないスレッドが多数であった場合はパフォーマンスの低下を抑えることができます。
データの変更を伴わない読み手（reader）スレッドと、データの変更を伴う書き手（writer）スレッドがそれぞれ複数個動いているとしましょう。書き手は他に書き手／読み手がいるときは書くことができません。対して読み手はデータを書き換えることのない読むだけのスレッドですから、他に書き手さえいなければ、いくつの読み手が読んでも構いません。
フツーのmutexは読み手／書き手の区別がないのでデータの変更を行わなくてもロックが取得できるのは1つのスレッドのみです。
C++17で仲間入りとなったstd::shared_mutexは書き手のためのlock()／try_lock()／unlock()に加え、読み手のためのlock_shared()／try_lock_shared()／unlock_shared()が用意されています。
サンプルを書いてみました。writerスレッドはグローバル変数x、yに1～9の乱数、zにxとyの積を書き込みます。readerスレッドはx*yがzに等しいか否かを判定します。1つのwriterスレッドと7つのreaderスレッドを起動し、全スレッドが完了するまでの時間を測ってみました。
#include <thread>
#include <chrono>
#include <random>
#include <iostream>
#include <shared_mutex>
int x;
int y;
int z;
int bad_results;
std::shared_mutex mtx;
void writer() {
using namespace std;
mt19937 gen;
uniform_int_distribution<int> dist(1,9);
for ( int i = 0; i < 10000000; ++i ) {
mtx.lock();
x = dist(gen);
y = dist(gen);
z = x * y;
mtx.unlock();
}
z = -1; // おしまいのシルシ
}
// （write-lockする）フツーのreader
void reader() {
while ( true ) {
mtx.lock();
if ( z < 0 ) { mtx.unlock(); break; }
if ( z != x * y ) ++bad_results;
mtx.unlock();
}
}
// read-lockするshared_reader
void shared_reader() {
while ( true ) {
mtx.lock_shared();
if ( z < 0 ) { mtx.unlock_shared(); break; }
if ( z != x * y ) ++bad_results;
mtx.unlock_shared();
}
}
void run(bool shared) {
using namespace std;
using namespace std::chrono;
x = 0;
y = 0;
z = 0;
bad_results = 0;
const int Nread = 7;
thread threads[Nread+1];
auto start = chrono::high_resolution_clock::now();
threads[0] = thread(writer);
for ( int i = 1; i <= Nread; ++i ) {
threads[i] = thread(shared ? shared_reader : reader);
}
for ( thread& thr : threads ) thr.join();
auto stop = chrono::high_resolution_clock::now();
cout << chrono::duration_cast<chrono::milliseconds>(stop - start).count() << "[ms] "
<< "bad_results = " << bad_results << endl;
}
int main() {
std::cout << "normal: "; run(false);
std::cout << "shared: "; run(true);
}
同時に動ける読み手が増えたことでパフォーマンスが上がっているのが分かります。
lockしない排他：atomic
mutexはたった1つのロックを複数のスレッドが奪い合うことで互いを邪魔しない／させない機構であるがゆえ、前述のconvoyやdeadlockが発生するリスクを伴います。なので望むらくはロックの取得／解放によらずにatomicな処理ができないかってことになるのですが、かなり限定的とは言いながらロックを使わない排他機構が用意されています。それがstd::atomic。std::atomicを使ったサンプルがコチラ。
#include <iostream>
#include <thread>
#include <mutex>
#include <atomic>
void run_mutex(int N) {
using namespace std;
using namespace std::chrono;
mutex mtx;
long count = 0;
auto task = [&](int n, long v) { while( n-- ) { mtx.lock(); count += v; mtx.unlock(); }};
auto start = high_resolution_clock::now();
thread inc(task, N, 1L);
thread dec(task, N,-1L);
inc.join(); dec.join();
auto stop = high_resolution_clock::now();
cout << "mutex: "
<< duration_cast<milliseconds>(stop - start).count() << "[ms]"
" count = " << count << endl;
}
void run_atomic(int N) {
using namespace std;
using namespace std::chrono;
atomic<long> count = 0;
auto task = [&](int n, long v) { while( n-- ) count += v; };
auto start = high_resolution_clock::now();
thread inc(task, N, 1L);
thread dec(task, N,-1L);
inc.join(); dec.join();
auto stop = high_resolution_clock::now();
cout << "atomic: "
<< duration_cast<milliseconds>(stop - start).count() << "[ms]"
" count = " << count << endl;
}
int main() {
const int N = 10000000;
run_mutex(N);
run_atomic(N);
}
run_atomic()内のatomic<long> countが「atomicなlong型変数：count」であり、count += v自体がatomicに処理されるので、その前後にロックの取得／解放がありません。
おまけにmutexより速いという嬉しい結果。
シカケの概略は以下のとおり。まず以下のような関数を考えます。
long compare_and_swap(long* address, long assumed, long value) {
long old = *address;
if ( old == assumed ) *address = value;
return old;
}
「*addressがassumedに等しければ*addressにvalueを書く。変更前の*addressを返す」関数です。この関数と等価な処理を行うCAS（compare and swap）と呼ばれる命令をCPUが持っています。これは機械語の1命令ですから他のスレッドが割り込むことはありません。さらにこのCASを使って、
long atomic_add(long* address, long value) {
lond assumed;
long old = *address;
do {
assumed = old;
old = compare_and_swap(address, assumed, assumed+value);
} while ( old != assumed );
return old;
}
do-whileの中で現在の値：assumedにvalueを加えて*addressにセットしています。もしこのloop内で他のスレッドが割り込んで*addressを書き換えるとassumedとoldは異なる値となるため、loopを再度回って加算をやり直します。ロックを使って他のスレッドを待たせるのではなく、"間違ってたらやり直す"ことでatomic性を保証するってわけ。
atomic<T>のoperator+=()はこんなカラクリで実現されています。ロック不要なのでlock-freeアルゴリズムと呼ばれています。ちなみにlock-freeな加算はWindows-APIならInterlockedAdd()、linuxならgccの組み込み関数__sync_add_and_fetch()で提供されています。
クラステンプレートatomic<T>のテンプレート引数Tに指定できるのはsigned／unsignedなchar～long long（いわゆる整数型）、boolおよびポインタ型で、定義されている演算子は++／--と+=／-=。atomicの適用範囲は限られますが、たった1つの変数に対する加算／減算であればmutexでガードするより簡単で高速なのが売りですね。
スレッドの排他にまつわるトピックは、これであらかた語ったかな。残る同期のおはなしはまた次回。
