\n
[ トップページ ]
[ ___ _CommandPrompt ]
[ ___ _JScript ]
[ ___ _MySQL ]
[ ___ _Cygwin ]
[ ___ _Java ]
[ ___ _Emacs ]
[ ___ _Make ]
[ ___ _Perl ]
[ ___ _Python ]
[ ___ _OpenGL ]
[ ___ _C# ]
[ ___ _StyleSheet ]
[ ___ _C++ ]
[ ___ _Winsock ]
[ ___ _Thread ]
[ ___ _VisualStudio ]
[ ___ _C ]
[ ___ _Win32API ]
[ ___ _Lua ]
[ ___ _PhotoShop ]
ヘッダ検索
■ リファレンス(Reference)
■ CreateThread
SYNTAX
HANDLE CreateThread(
LPSECURITY_ATTRIBUTES Lpthreadattributes, // Thread の取得属性 ( NULL で Default )
DWORD dwStackSize, // Stack の予約サイズ (0 で プライマリスレッドと同じサイズ )
LPTHREAD_START_ROUTINE lpStartAddress, // ThreadFunc の関数
LPVOID lpParameter, // ThreadFunc にわたすパラメータ
DWORD dwCreationFlags, // flag
// CREATE_SUSPENDED : ResumeThread が呼ばれるまで待機
// 0 : 即時に実行
LPDWORD lpThreadId // 識別 id がセットされる ( 不要ならば NULL にする )
)
DESC
Thread を生成して Thread の識別子である HANDLE をかえす
HANDLE trd[2];
struct ThreadArg {
int i;
};
ThreadArg arg[2];
// Thread を 2個つくる
for( int i=0; i< 2; i++ ){
trd[i] = CreateThread( 0, 0,
(LPTHREAD_START_ROUTINE)funcTrd,
&(arg[i]), 0, 0 );
}
■ WaitForSingleObject
SYNTAX
DWORD WaitForSingleObject(HANDLE h, DWORD msec );
DESC
ハンドルで指定したオブジェクトを待機する。
ハンドルに Thread を指定した場合はスレッドが終了するというイベントをまつ。
RET
WAIT_OBJECT_0 : シグナル状態になった
WAIT_TIMEOUT : タイムアウトによって戻ってきた
WAIT_ABANDONED: 指定したオブジェクトは既に別スレッドが破棄している
WAIT_FAILED : 関数が失敗
TIP
// Thread 完了まち
if( WaitForSingleObject( hEvent, INFINITE ) == WAIT_OBJECT_0 )
{
// 完了
}
WaitForSingleObject( 2, trd, 1, INFINITE );
■ GetCurrentThread
SYNTAX
HANDLE GetCurrentThread();
DESC
現在のスレッドの疑似ハンドルが返る
現在の Thread のパラメータを変更する際に便利.
■ GetCurrentThreadId
SYNTAX
DWORD GetCurrentThreadId();
DESC
呼び出し側スレッドのスレッド識別子を返す
■ Sleep
SYNTAX
void Sleep( DWORD msec );
DESC
指定時間 Thread を休止させる
TIP
0 を指定すると 同じ Priority の 他の Thread に制御をゆずる ( CPU 時間の割り当てがされる )
なければ, そのまま処理が継続される
■ ResumeThread
SYNTAX
DWORD ResumeThread(
HANDLE hThread // スレッドのハンドル
);
DESC
スレッドを再実行させる。
スレッドには SuspendCount（中断）カウントがあり 1 減らす
SuspendCount が 0 になると、スレッドの実行が再開される
■ SuspendThread
SYNTAX
SuspendThread( HANDLE )
DESC
指定したスレッドを停止させる。
停止したスレッドはCPU時間が割り当てられなくなる。
WARNING
自分自身で自発的に止める場合は Sleep() を利用する。
■ ExitThread
SYNTAX
ExitThread( DWORD exitcode )
DESC
スレッドを終了させる。
終了したいスレッド自身からコールする。
外部から殺すには TerminateThread() を使う
指定した終了コードは外部から GetExitCodeThread() で取得できる。
■ TerminateThread
SYNTAX
TerminateThread( HANDLE, DWORD exitcode )
DESC
指定したスレッドを終了させる。
■ GetExitCodeThread
SYNTAX
BOOL GetExitCodeThread(
HANDLE hThread, // スレッドのハンドル
LPDWORD lpExitCode // 終了ステータス
);
RETVAL
0 : FAIL
!0 : OK
STILL_ACTIVE : 実行中
DESC
別スレッドから指定したスレッドが終了したか調べる
thread::getExitCode() で利用されている
// 終了待ちをする
while ( 1 ) {
GetExitCodeThread( trd, &ret );
if ( ret == STILL_ACTIVE ){
printf("still working\n" )
Sleep(0);
}
else {
break;
}
}
■ 用語 ■ thread
■ Thread.とは
プログラムの実行される単位のこと。
プログラムを一本の流れとしてとらえると、この流れから分岐するひとつの糸のこと。
CPU に割り振られることで実行される
WARNING
切り替わるタイミングは CPU がアセンブラ単位で切り替える。
C++ で１行でかいてあるコードも実際はその中の任意の場所でかわるということ。
( ソースコード 1 行単位ではない )
同期とは終わるまでまつことで、非同期とは終わるまでまたない方法のこと。
通常の関数よびだしは同期する( おわるまで待つ )。
■ スレッドのつかいどころ
WARNING
スレッドを増やしても、意味のない用途がある。
スレッドを使うときは、ハードの資源以上は超えられない。
たとえば、 ハードディスクが１つしかなければ
ファイルアクセスをするスレッドを２つ以上、作成しても実際は１つしか実行されない。
ファイルのロードなどは 処理時間の大半が待ち時間 ( latency )なので
その空き時間を他のことに回すことができる。
■ スレッドの危険性をしる
SAMPLE
グロバール変数を同時に操作する
どのスレッドを実行するかは、OS が気まぐれで変更するため
共有オブジェクトへの操作は危険になる。
int gCnt;
// 共有オブジェクト gCnt をインクリメントする仕事
void jobAdd( void *ptr )
{
DWORD id = GetCurrentThreadId();
printf( "start thread id = %d\n" , id );
for( int i=0; i< 1000*1000; i++ ){
gCnt ++; // 複数のスレッドから共有物を変更する。ここが危険
}
printf( "finish thread\n" );
}
const int N = 3;
HANDLE trd[N];
int i;
for( i=0; i< N; i++ ){
trd[i] = CreateThread( 0, 0,
(LPTHREAD_START_ROUTINE)jobAdd,
NULL, 0, 0 );
}
// 終了をまつ
for( i=0; i< N; i++ ){
WaitForSingleObject( trd[i], 10*1000 );
}
// この結果が毎回ことなる。
printf( "result gCnt %d\n" , gCnt );
gCnt ++;
1. メモリ上にある gCnt を レジスタへ移動 ( MOV )
2. CPUで +1 ( ADD )する
3. その結果を メモリ gCnt へ戻す ( MOV )
gCnt ++;
1. A さん : メモリ上にある gCnt を レジスタへ移動 ( MOV )
2. B さん : メモリ上にある gCnt を レジスタへ移動 ( MOV )
3. A さん : CPUで +1 ( ADD )する
4. A さん : その結果を メモリ gCnt へ戻す ( MOV )
5. B さん : CPUで +1 ( ADD )する
6. B さん : その結果を メモリ gCnt へ戻す ( MOV )
■ 共有物にはロック(鍵)をかける
SAMPLE
鍵をかけてからグロバール変数を同時に操作する
そこで共有物を操作する間は、他のスレッドがさわらないようにロックをする。
これで自分だけが操作できる状態になる。
それには Mutex をつかう。
Mutex は複数のスレッドの交通整理をする。
Mutex は bool 変数のように ON/OFF できる。
他のスレッドが、鍵のかかった Mutex をさわろうとすると眠ってしまう。
Win32 Thread では EnterCriticalSection を利用する
// 鍵自体は共有物
CRITICAL_SECTION cs;
void jobAdd( void *ptr )
{
DWORD id = GetCurrentThreadId();
printf( "start thread id = %d\n" , id );
// 鍵をかけてから
EnterCriticalSection( &cs );
// 自分だけ共有物を操作する
for( int i=0; i< 1000*1000; i++ ){
gCnt ++;
}
// 仕事が終わったら、すばやく鍵をはずすこと
LeaveCriticalSecttion( &cs );
printf( "finish thread\n" );
}
{
// 鍵の初期化と終了処理をついか
InitializeCriticalSection( &cs );
CreateThread( ... )
DeleteCriticalSection( &cs );
}
REFERENCE Mutexクラスをつくる
複数のスレッドからよんでも安全な関数をスレッドセーフであるという。
スレッドセーフとある関数は、複数のスレッドから読んでもいい。
例えば、
あるクラスのインスタンスがあって、そのメンバ変数を変更するメソッドを
複数のスレッドから呼ぶ場合はスレッドセーフではない。
vector< int > a;
// これを複数のスレッドからよびだしてはいけない。
a.push_back( 10 );
■ 状態
スレッドにはいくつかの状態がある ( Program を利用する際は意識する必要なし )
待ち
何かの理由で動作できない状態
例えばセマフォの資源が取得できるようになるのを待っている, といった状態がこれに当たる
待ち状態のスレッドに CPU が割り当てられることはない
起動直後のスレッドは待ち状態か実行可能状態のどちらかになる
実行可能
CPU が割り当てられれば動作することができる状態
例えばセマフォの資源は取得できたが, 空いている CPU がない, といった状態
実行可能状態になったスレッドはスケジューラのキューに登録される
スケジューラは CPU に空きができるとキューからスレッドをとりだし CPU を割り当てる
CPU を割り当てられたスレッドは実行状態になる
実行
CPU が割り当てられ, スレッドが動作している状態
この状態で資源の残っていないセマフォの取得を行なったりすると待ち状態になる
POINT
sleep は明示的に待ち状態に移行する操作です
タイムアウト付きの sleep は一定時間たつと待ち状態から実行可能状態に移行する
yield は実行可能状態に移行する操作
自スレッド以外に実行可能状態の スレッドがいれば, そのスレッドに CPU が渡される
いなかった場合は再び自スレッドが実行状態になる
終了
スレッドが終了した状態
スレッドは終了してもすぐに全てのリソースを 解放するわけではない
終了状態のスレッドに CPU が割り当てられることはなく、別の状態に移行することもない
■ 排他処理
ある処理しているスレッドを同時に 1 つだけに限定させるような処理のこと
排他処理をしている区間をクリティカルセクション( 危険領域 ) という
排他処理はクリティカルセクションに入る, クリティカルセクションから出る,
と いった操作で実装できる
POINT
排他処理のオブジェクトは １つのデータを複数のスレッドで操作する場合に
破壊されないように保護するために使う。
// var に対する処理を 1 つのスレッドだけに限定させるため に排他処理をする
void Add(int n)
{
EnterCriticalSection();
var += n;
LeaveCriticalSecttion();
}
■ Mutex
ミューテックスは[ 排他処理 ]をするために使用されるオブジェクト
ミューテックスには取得されている, いないの 2つの状態があり
一度にミュー テックスを取得できるのは 1 つのスレッドだけ
取得できなかったスレッドは待ち状態になる
ミューテックスでクリティカルセクションをつくる場合は
クリティカルセク ションに入る前にミューテックスを取得し, クリティカルセクションから抜ける時にミューテックスを返却する
ミューテックスをもっていたスレッドがミューテックスを返却すると
ミューテックスを待っていたスレッドは実行可能状態になる
複数のスレッドが待ち状態になっていた場合
どのスレッドが実行可能状態に なるのかはスケジューラに依存する
Windows ではクリティカルセクションの作成に限定したオブジェクトがある
このオブジェクトはミューテックスよりも効率よく動作するので
排他処理を行なう場合, Windows ではミューテックスよりもクリティカルセクションを使う
POINT
ミューテックスは取得しているスレッドを覚えていて
取得していないミューテックスを返却することはできない
■ セマフォ(Semaphore)
セマフォもミューテックスと同じように取得されている, いないの状態をもつが
セマフォでは状態を資源数で管理する
資源数の最大値と現在値を持ち
セマフォを取得すると資源数の現在値が減り, 返却すると増える
セマフォは取得をしたスレッドを覚えていないので
セマフォを取得した状態でさらに取得することができ
セマフォを取得していないスレッドが返却もできる
1 つしか資源のないセマフォはミューテックスと同じようにクリティカルセクションを作るのにも使える
しかしミューテックスの方が効率的
POINT
セマフォは決められた数だけのスレッドがはいる時に使う。
ジョブキューでいうと、マスタが Increment すると、ワーカーは目を覚まして Decrement をする。
■ 排他処理
スレッドはあらゆるタイミングで別スレッドに切り替わる可能性がある
この特徴はスレッド間で共有の領域を操作している時に問題になる
1 をタス Code
static int counter;
void Increment()
{
counter++;
}
Increment: Register を利用する必要があるんだ. ^ ^/
0: load r0, counter # counter から r0 に値をロードする.
1: add r0, r0, 1 # r0+1 を r0 に設定する.
2: store counter, r0 # r0 を couner にストアする.
3: ret # 関数から帰る.
// ミューテックスをつかって排他処理をしてみる
// 共有 Object
static int cnt;
void inc()
{
mutex.lock(); // mutex.lock() はそれ以降のコードを実行するスレッドを 1 つに限定する効果がある
cnt++;
mutex.unlock(); // この効果は mutex.unlock() まで続く
}
■ 同期処理
スレッド間で[ 足並みを揃える ]ための処理
別のスレッドでしている処理の完了をまったり
別のスレッドからのリクエストを待つ, といった時に使う
/// 別のスレッドからのリクエストを待って, 何らかの処理を始め るために同期処理をする
void Main()
{
WaitEvent();
:
}
■ 同期オブジェクト
排他処理, 同期処理を実装する際には同期オブジェクトを使う
同期オブジェクトはある条件を満たすまでスレッドを待機させることができるようなオブジェクト
ミューテックスとセマフォはほとんどの環境にある
もしくは比較的簡単に実装することが可能
一方, 条件変数はプラット フォームのサポートがないと実装するのが難しいオブジェクト
■ Event(イベント)
Windows でよく使われる同期オブジェクト
イベントにはシグナルと非シグナルの 2 つの状態があり
プログラムからは イベントのシグナル状態を問い合わせたり待つことができる
状態が 2 つしかない点はミューテックスに似ているが
イベントはどのスレッドからでもシグナル, 非シグナル状態にできる
イベントにはシグナル状態の問い合わせ成功と同時に非シグナル状態になるものと
問い合わせ成功後もシグナル状態のままになっているものがある
POINT
イベントは mutex とちがってクエリをすることができる。
■ 同期処理
複数のスレッドはそれぞれ勝手なタイミングで動く
Thread A の結果を受けて Thread B が処理をする必要がある場合
Thread B は Thread A の動作が完了待ちをする必要がある
このように複数のスレッドが[ 足並みを揃える ]ような処理を同期処理という
POINT
同期処理はセマフォ, イベント, 条件変数などの同期オブジェクトを使って実装する
// Thread A の動作が完了したら Thread A は FinishThreadA() を呼ぶ
void FinishThreadA()
{
event.signal();
}
void WaitThreadA()
{
event.waitSignal();
}
// ThreadA が終わるのをまつ
void WaitThreadA()
{
// 実行可能なまま処理をつづけるのが問題. ( CPU を無駄にもちつづける. )
// 一方同期オブジェ クトを使って待ち状態になれば, CPU は一旦スケジューラに返却され
// 他の実行可能状態のスレッドに割り当てることができる
while (event != 1) {
}
}
また, WaitThreadA() のコードを最適化すると,
コンパイラは無限ループするようなコードを出力する可能性がある
while() 文の中に event 変数を 変更している箇所がないため,
コンパイラが while の条件式は変化しない, と勘違いする可能性があるため
それを回避するために volatile を使うことがある
これによりコンパイラは while の条件式が変化するものである, と判断することができるため,
無限ループになる可能性はなくなる
しかし, まだ安全ではない
volatile はコンパイラ向けの指示であって CPU 向けの指示ではない, というのが理由
同期オブジェクトを使わないで同期処理をするのは得策ではない
問題がなければ, 素直に同期オブジェクトをつかって同期処理を実装すること
■ アトミック操作
別のスレッドに割り込まれずに処理できるような[ 操作 ]をアトミックであるという
マルチスレッドプログラミングではアトミックであるかどうかが非常に重要
マルチスレッドで動作するプログラムであっても, 全ての処理をアトミックにする必要はない
必要があるのは, 全体のほんの一部 だけ
POINT
[ スレッド間で共有する領域を使っている処理はアトミックにする ] 必要がある
それ以外はアトミックである必要はない
static int counter;
// 共有領域なので, Atomic に操作する必要あり.
void Increment()
{
counter++;
}
void Decrement()
{
counter--;
}
// Threed Local なので
int Add(int counter, int val)
{
return counter+val;
}
int Sub(int counter, int val)
{
return counter-val;
}
POINT
2 つ以上の命令になってしまう処理は全て割り込まれる可能性がある
なにがアトミックなのか
どこまでをアトミックにするかを判断すること
Node *top;
void Push(Node *node)
{
node->prev = 0;
node->next = top;
if (top != 0) {
top->prev = node;
}
top = node;
■ リーダライタロック
マルチスレッドプログラミングでは, スレッド間で共有される領域を操作する
ときは排他処理をする必要がある
排他処理はミューテックスを用いて実装することが多いが
ミューテックスでは効率的ではないケースがある
共有領域に対する操作は読み込み, 書き込みに分類されるが
ほとんどの操作が読込みで, 書込みはほとんどしないことがある
POINT
読込み処理と読込み処理同士であれば排他処理は不要
読み込み処理同士は並列動作可能
ミューテックスでつくられたクリティカルセクションでは読込みしか行なわない操作であったとしても
同時に実行できるのは必ず 1 つのスレッド だけになる
この制限を解消するのがリーダライタロック
リーダライタロックには読込むためのロックと書き込むためのロックがある
ミューテックスと異なるのは読込みロック時でも読込みロックが成功する点だけ
このルールにより, 読み込み動作が並列で動作できるようになる
■ ロックフリー
マルチスレッドプログラミングではアトミックであることが非常に重要
アトミックな操作を実装するのには同期オブジェクトを使った排他処理がよくされるが
いくつかの問題がある
排他処理はコスト 0 の操作ではなく、それなりにコストがかかる処理
同期オブジェクトを使ってアトミックな処理を実装することは簡単で判りやすいが
パフォーマンス面で問題になることがある
ロックフリーは[ アトミックな操作を同期オブジェクトなどのロックを伴わずに実装する方法 ]です
ロックフリー なアルゴリズムでは同期オブジェクトのかわりに CPU で用意されている特別な命令を使う
たとえば compare-and-swap という操作がある compare-and-swap は以 下のような操作
int CAS(int *var, int cmp, int swap)
{
int prev = *var;
if (prev == cmp) {
*var = swap;
}
return prev;
}
多くの CPU には compare-and-swap をアトミックに.する命令が用意されている
■ チュートリアル(Tutorial)
■ スレッドをつくる
スレッドをつくるには CreateThread() を実行する。
スレッドで実行する関数は次のシグネーチャをもつ
#include< stdio.h>
#include< windows.h>
void job( void *ptr )
{
while( true ) {
Sleep( 1000 );
printf( "thread working\n" );
}
}
int main()
{
HANDLE trd[2];
// Thread を 2個つくる
for( int i=0; i< 2; i++ ){
trd[i] = CreateThread( 0, 0,
(LPTHREAD_START_ROUTINE)job,
NULL, 0, 0 );
}
return 0;
}
■ スレッドの終了をまつ
スレッドのハンドルを保持しておき、 WaitForSingleObject() でまつ。
int main()
{
HANDLE trd[2];
// Thread を 2個つくる
for( int i=0; i< 2; i++ ){
trd[i] = CreateThread( 0, 0,
(LPTHREAD_START_ROUTINE)job,
NULL, 0, 0 );
}
for( i=0; i< 2; i++ ){
WaitForSingleObject( trd[i], INFINITE );
}
// 最大でまつ時間も指定できる。
WaitForSingleObject( trd[i], 3000 );
return 0;
}
■ スレッドを停止させる
SAMPLE
スレッドを停止させる
スレッドの動きを制御する
スレッドは指定した関数が最後まで到達すると終了するため
制御用の変数で制御する。
メインスレッド側でユーザーの入力をまち、スレッドの停止をすることができる。
class Job {
public:
int flag;
Job() : flag(0){}
};
void jobtest( void *ptr )
{
Job *j = ( Job *)ptr;
while( true ) {
Sleep( 1000 );
printf( "thread working\n" );
// 終了条件
if ( j->flag ) {
printf( "stop\n" );
break;
}
}
}
int main()
{
HANDLE trd[2];
Job j;
int i;
for( i=0; i< 2; i++ ){
trd[i] = CreateThread( 0, 0,
(LPTHREAD_START_ROUTINE)jobtest,
&j, 0, 0 );
}
// メッセージループ
{
while ( true ) {
printf( "スレッドを止めるためには 1 を入力をしてください\n" );
char buf[256];
gets( buf );
puts( buf );
// スレッドを止めるため変数を変更する
if ( strcmp(buf, "1" ) == 0 ) {
j.flag = 1;
break;
}
}
}
return 0;
}
■ スレッドの処理の完了をチェックする
制御用の変数をそのままチェック用の変数に使う。
■ スレッド間の動作を連携させる
SAMPLE
グローバル変数をつかってスレッド間で通信する
スレッド側で値を更新して、更新後にメインスレッド側でその値を表示する。
スレッド間で通信するには、一番単純にはグローバル変数で値を共有すればいい。
// スレッド間で共有する変数
Mutex gReadMtx;
Mutex gWriteMtx;
volatile bool gRead;
volatile bool gWrite;
int gX;
class TestJob : public Job
{
public:
void main( void *ptr ) {
while ( true ) {
// メインスレッドが読むまで待つ
while ( !gRead ) {
;
}
gX += 2;
gReadMtx.lock();
gRead = false;
gReadMtx.unlock();
// 書いたことを true にしてメインスレッドに教える
Sleep(1000);
printf ( "thread write gX\n" );
gWriteMtx.lock();
gWrite = true;
gWriteMtx.unlock();
}
printf ( "thread work finish\n" );
}
};
int main()
{
Worker w;
TestJob job;
w.addJob( &job );
// 最初に書き込みをはじめる
gRead = true;
gWrite = false;
w.start();
while ( true ){
// スレッドが書くまでまつ
while ( !gWrite ) {
;
}
// 元にもどす
gWrite = false;
printf( "main tread Read %d\n" , gX );
// 読んだ
gRead = true;;
}
w.join();
return 0;
}
■ volatile.をつかう
WARNING
上のコードでは コンパイラが最適化をかけることで無限ループになる可能性がある。
while ( gWrite ) {
// ループ内で gWrite を変更する処理がないため、コンパイラが gWrite の評価をなくしてしまう。
// コンパイラからすると別スレッドが操作するとは、わからない。
;
}
■ スレッドの動作を連携させる
メインスレッドで入力をうけて, 変更があった場合にスレッド側で表示する。
SAMPLE
スレッドの処理の終了まちをチェックする
int main()
{
HANDLE trd[2];
Job j;
int i;
for( i=0; i< 2; i++ ){
trd[i] = CreateThread( 0, 0,
(LPTHREAD_START_ROUTINE)jobtest,
&j, 0, 0 );
}
// メッセージループ
while ( true ) {
printf( "スレッドがデータをロード中 %d\n" , j.flag );
Sleep( 1000 );
if ( j.flag == 10 ) {
printf( "スレッドの処理が終わりました\n" );
break;
}
}
return 0;
}
■ スレッドをクラスにしてみる
SAMPLE
スレッドクラス
スレッドを実行する処理が定型句なのでクラス化してみる。
スレッドに渡すスタティック関数をラップしてみる。
class Worker
{
public:
// 停止状態でスレッドを生成
Worker() {
DWORD id;
trd = CreateThread( 0, 0, (LPTHREAD_START_ROUTINE)( Worker::workMain ), this,
CREATE_SUSPENDED, &id );
}
// スレッドを実行
void start() {
ResumeThread( trd );
}
// スレッドの終了をまつ
void Worker::join()
{
WaitForSingleObject( trd, INFINITE );
}
// 仕事を追加する
void addJob() {
}
static void workMain( void *ptr ) {
Worker *p = ( Worker *)ptr;
p->job->main( p->job );
}
private:
HANDLE trd;
};
class Job
{
public:
/// Worker が実行する仕事
virtual void main( void *ptr ) = 0;
virtual ~Job(){}
};
class TestJob : public Job
{
public:
int flag;
TestJob() : flag(1){}
void main( void *ptr ) {
while ( flag ) {
Sleep(1000);
printf ( "thread working ...\n" );
}
}
};
int main()
{
Worker w;
TestJob job;
w.addJob( &job );
w.start();
char buf[256];
gets(buf);
job.flag = 0;
w.join();
return 0;
}
■ スレッドプール(ThreadPool)
SAMPLE
スレッドプール
スレッドプールGUI
POINT
スレッド(ワーカー)をマスタと通信する箇所をひとつのモジュールにまとめておく。
ここにはワーカーに依頼する仕事をリスト(キュー)につめておく。
ひとつのキューを複数のスレッドで操作をすることは危険な処理。
たとえば ワーカーがキューから仕事をとる最中に、マスタが仕事をキューに入れようとすると
キューを破壊する可能性がある。
そこで共有物であるキューを Mutex で保護をしてあげる。
[ マスタ ( メインスレッド) ]
|
| (仕事をつめる)
|
[ 仕事依頼リスト ]
[ 仕事１ ][ 仕事２ ][ 仕事３ ][ 仕事４ ]
|
| (仕事をする)
|
[ ワーカー ][ ワーカー ][ ワーカー ]
template< class T >
class JobQ
{
volatile unsigned int cnt;
public:
// キューにつめる
void enQ( const T &ref ){
// 空き数カウンタが０より大きくなるのをまつ。
// いっぱいならば Worker が仕事を取るまで待つ
WaitForSingleObject( smp_empty, INFINITE );
EnterCriticalSection( &(s) );
cnt ++;
data[wp] = ref;
wp ++;
if ( wp == NRQ ) wp = 0;
LeaveCriticalSection( &(s) );
// 残り数を１つ増やして待機中のスレッドを動かす。
// [○][○][][][]
ReleaseSemaphore( smp_count, 1, 0 );
}
// キューからとりだす
void deQ( T &ref ){
// 仕事がはいるのをまつ。
WaitForSingleObject( smp_count, INFINITE );
EnterCriticalSection( &(s) );
ref = data[rp];
cnt --;
rp ++;
if ( rp == NRQ ) rp = 0;
LeaveCriticalSection( &(s) );
// 空き数カウンタを増やす。
// [][][][○][○]
ReleaseSemaphore( smp_empty, 1, 0 );
}
// コンストラクタ
JobQ() : cnt(0){
rp = 0;
wp = 0;
// 現在のつめた個数
// 最初は 空なので "0"
// Worker は カウンタ があれば Queue から取り出す
smp_count = CreateSemaphore( 0, 0, NRQ, 0 );
// こちらは空きカウンタ なので いっぱいにする -> ( == 満タン )
// Master は 空きカウンタ があれば Queue に 仕事( Request ) をつめる
smp_empty = CreateSemaphore( 0, NRQ, NRQ, 0 );
InitializeCriticalSection( &(s) );
}
unsigned int getCount() {
EnterCriticalSection( &(s) );
unsigned int ret = cnt;
LeaveCriticalSection( &(s) );
return ret;
}
private:
T data[NRQ];
int rp, wp; // 読み込み位置, 書き込み位置
CRITICAL_SECTION s; // Q read/write の排他制御.
HANDLE smp_count; // Productor Semaphore ( １個追加したら ++ )
HANDLE smp_empty; // Consumer Semaphore ( １個とったら -- )
};
int main()
{
JobQ< string > q;
// ワーカースレッドを作成して
Worker *w = new Worker();
Worker *w1 = new Worker();
// 仕事をあたえる
w->addJob( new JobTest( &q ) );
w1->addJob( new JobTest( &q ) );
w->work();
w1->work();
// メインスレッドはリクエストだけを受け付けてキューにつめることを繰り返す。
while (1) {
string s;
cin >> s;
cout < < "マスタ: リクエスト( " < < s < < " )いただきました" < < endl;
// キューにつめる。
q.enQ( s );
}
return 0;
}
class JobTest : public Job
{
public:
JobTest( JobQ< string > *_q ) : q( _q ){}
void main( void *ptr ) {
while ( true ) {
string s;
cout < < Worker::currentWorker()->getName();
cout < < ": リクエストまってます" < < endl;
q->deQ( s );
cout < < Worker::currentWorker()->getName();
cout < < "仕事をもらったので実行します " < < s < < endl;
}
}
private:
JobQ< string > *q;
};
■ Mutexクラスをつくる
SAMPLE
スレッドの動作を排他的にする
// Class にしてみる
class Mutex {
public:
void lock() {
EnterCriticalSection( &cs );
}
void unlock() {
LeaveCriticalSection( &cs );
}
Mutex() {
InitializeCriticalSection( &cs );
}
~Mutex() {
DeleteCriticalSection( &cs );
}
private:
CRITICAL_SECTION cs;
};
// 使う
static Mutex m;
// 排他処理をする効果を見るため, Sleep() する
void threadFunc( void *) {
m.lock();
m.unlock();
}
■ InitializeCriticalSection
SYNTAX
InitializeCriticalSection( LPCRITICAL_SECTION )
DESC
CriticalSection を初期化する
■ EnterCriticalSection
SYNTAX
EnterCriticalSection( LPCRITICAL_SECTION )
DESC
排他制御変数 を使って排他制御を開始する
■ LeaveCriticalSection
SYNTAX
LeaveCriticalSection( LPCRITICAL_SECTION )
DESC
排他制御を終了する
■ DeleteCriticalSection
SYNTAX
void DeleteCriticalSection( LPCRITICAL_SECTION )
DESC
CriticalSection を破棄する
■ 同期処理
class Event {
public:
// ON にする
void signal()
// Thread を停止させる
void block() {
WaitForSingleObject( evt, INFINITE );
}
// 作成時は信号を ON にしておく
Event() {
evt = CreateEvent( 0, false, true, NULL );
}
~Event() {
CloseHandle( evt );
}
private:
HANDLE evt;
}
■ CreateEvent
SYNTAX
bool CreateEvent (
lpEventAttributes : DWORD
bManualReset : DWORD // False: 自動 Reset
bInitialState : DWORD // 初期状態. True: Signal False: NoSignal
lpName : DWORD
);
RET
hdl :
0 : Fail
TIP
Event の状態を監視したい( 信号まちをしたい ) Thread は
WaitForSingleObject を呼ぶことで待機状態にする
■ SetEvent
SYNTAX
BOOL SetEvent(
HANDLE hEvent // EventObject Handle
);
RET
!0 : Success
0 : Fail
DESC
Thread 同士と同期をとるために利用する信号
Thread 間で処理の Timing の通信をする際に利用する
EventObject はふたつの状態をもつ.
ON : Signal 状態.
OFF: NoSignal 状態.
POINT
NoSignal 状態ならば, イベントオブジェクトを待ち受けるスレッドが休止状態である。
スレッドが待ち状態であるとき、ほとんどCPUに負荷はかからない。
class UpdateTask : public Task {
public:
void main() {
// 描画処理はまかせて更新処理をくりかえす.
// 問題は, vector< Object > をつついてしまうこと.
// Buffer ( 要は push , pop する側に分ける必要がある )
while ( 1 ) {
App::update();
// 終了したら信号を青にする
evt.signal( 1 );
}
}
};
void main() {
// 更新処理.
UpdateTask task;
Thread A;
A.addTask( &task );
// MessageLoop
while ( 1 ) {
App::display();
// update まち
evt.block();
}
}
■ ResetEvent
SYNTAX
BOOL ResetEvent(
HANDLE hEvent // EventObject Handle
);
DESC
Manual の場合は, Reset されるまでは, Signal 状態を維持する.
自動の場合
Thread 解放時に System が NoSignal にする.
Thread がひとつも待機していないときは, Signal 状態を維持する.
Ret
!0 : Ok
0 : Fail
SyncObject を有効にする. ( 信号を青にすると考えておく. )
Event MainLoopSignalt;
待つ側:
// 1 になるのをまつ.
待たせる側:
// 信号を 1 にする. ( 同期をとる timing で 青にかえてあげる. )
MainLoopSignal.signal(1);
待つ側
// 1 になるのをまつ.
::WaitForSingleObject( hMainLoopSync, INFINITE );
待たせる側
// 信号を 1 にする. ( 同期をとる timing で 青にかえてあげる. )
::SetEvent( hMainLoopSync );
// Mutex でも実装できる
待つ側:
mutex.lock();
待たせる側:
mutex.unlock();
■ CreateSemaphore
SYNTAX
HANDLE CreateSemaphore(
LPSECURITY_ATTRIBUTES lpSemaphoreAttributes, // セキュリティ記述子
LONG lInitialCount, // 初期カウント
LONG lMaximumCount, // 最大カウント数
LPCTSTR lpName // オブジェクトの名前
);
DESC
セマフォオブジェクトを作成または開く。
// 資源数５、初期カウント０のセマフォをつくる
smp = CreateSemaphore( 0, 0, 5, 0 );
// セマフォを取得するまでスレッドはまつ。
// シグナル状態になったセマフォを取得するとカウント数はひとつ減る。
WaitForSingleObject( smp, INFINITE );
// 待機中のスレッドは動かすには別のスレッドから資源数を増やす。
ReleaseSemaphore( smp, 1, 0 );
■ ReleaseSemaphore
SYNTAX
BOOL ReleaseSemaphore(
HANDLE hSemaphore, // セマフォのハンドル
LONG lReleaseCount, // 増やすカウント数
LPLONG lpPreviousCount // 変更前のカウント( 知る必要がないなら NULL )
);
DESC
指定されたセマフォオブジェクトのカウントを、指定した数だけ増やす。
■ 並列化
POINT
何を並列にすべきか考える
■ Data.の並列化
SAMPLE
画像変換の並列処理
DESC
処理する対象データを 小分けして処理を並列にする
画像のピクセル単位の処理が分りやすい例。
各ピクセルデータ間には依存関係がないため並列して処理ができる。
// 画像をモノクロに変換する
void funcTrd( void *ptr )
{
Data *w = (Data *)ptr;
unsigned int i;
unsigned char y;
/*
WARNING
1. unsigned でないと問題あり.
char *p = 0000 = 255 = 0x00010000
*/
unsigned char *p = (unsigned char *)(w->data);
//char *p = (char *)(w->data);
for( i=0; i< w->nr; i++ ){
y = 0.299f * (int)(p[2])
+ 0.587f * (int)(p[1])
+ 0.114f * (int)(p[0]);
p[0] = p[1] = p[2] = y;
p += sizeof(char) * 4;
}
}
int main( void )
{
int i, nr;
HANDLE trd[NRTRD];
Data w[NRTRD];
image_w img;
// 画像の読み込み
loadBmp( "test.bmp" , &img );
nr = ( (img.w * img.h ) + (NRTRD - 1) ) / NRTRD;
for( i=0; i< NRTRD; i++ ){
w[i].id = i;
w[i].nr = nr;
w[i].data = ( (char*)(img.data)
+ i*nr*(sizeof(char)*4)
);
// 残り Pixel の調整
if ( i == NRTRD - 1) {
w[i].nr = (img.w * img.h) - nr*(NRTRD-1);
}
// Thread を作成して小分けしたデータを渡す
trd[i] = CreateThread( 0, 0,
(LPTHREAD_START_ROUTINE)funcTrd,
&(w[i]), 0, 0 );
}
WaitForMultipleObjects( NRTRD, trd, 1, INFINITE );
/* 画像の出力 */
writeBmp("test_mono.bmp" , &img );
return 0;
}
金利比較.COM
_
_
_
