図? シングルスレッドのプロセスとマルチスレッドのプロセス
並列分散ソフトウェア
電子・情報工学系
新城 靖
<yas@is.tsukuba.ac.jp>
このページは、次の URL にあります。
http://www.hlla.is.tsukuba.ac.jp/~yas/sie/pdsoft-2003/2004-01-08
あるいは、次のページから手繰っていくこともできます。
http://www.hlla.is.tsukuba.ac.jp/~yas/sie/
http://www.is.tsukuba.ac.jp/~yas/index-j.html
http://www.hlla.is.tsukuba.ac.jp/~yas/index-j.html
軽量プロセスというと、内部にループを含むような語感がある。
マルチスレッドプログラミングは、非常に難しい。
どこでスレッドを使うべきか。
Pthread を利用したプログラムを書く時には、次のようなヘッダ・ファイルを 読み込む。
#include <pthread.h>
Solaris (Unix International系)でのコンパイルとリンク。
-D_REENTRANTと-lpthread を付ける。
セマフォを使う時、Solaris 6 (SunOS 5.6, sakura) では、リンク時に---------------------------------------------------------------------- % cc -D_REENTRANT -o create create.c -lpthread% ./create
----------------------------------------------------------------------
-lposix4
オプションを付ける。Solaris 7-9 (SunOS 5.7, 5.8, 5.9) では、リンク時に -lrt オプションを付ける。
SGI (O2, Origin) でのコンパイルとリンク。-lpthread を付ける。
---------------------------------------------------------------------- % cc -o create create.c -lpthread
図? fork-joinモデルの実現
後で join する必要がない時には、pthread_detach() を使って切り離す。 (joinしなくてもゾンビが残らない。)
----------------------------------------------------------------------
1:
2: /*
3: create-join-2.c -- スレッドを２つ作るプログラム
4: */
5:
6: #include <pthread.h>
7:
8: void func1( int x );
9: void func2( int x );
10:
11: main()
12: {
13: pthread_t t1 ;
14: pthread_t t2 ;
15: pthread_create( &t1, NULL, (void *)func1, (void *)10 );
16: pthread_create( &t2, NULL, (void *)func2, (void *)20 );
17: printf("main()\n");
18: pthread_join( t1, NULL );
19: pthread_join( t2, NULL );
20: }
21:
22: void func1( int x )
23: {
24: int i ;
25: for( i = 0 ; i<3 ; i++ )
26: {
27: printf("func1( %d ): %d \n",x, i );
28: }
29: }
30:
31: void func2( int x )
32: {
33: int i ;
34: for( i = 0 ; i<3 ; i++ )
35: {
36: printf("func2( %d ): %d \n",x, i );
37: }
38: }
----------------------------------------------------------------------
実行例。
この例では、次の３つのスレッドが作られる。---------------------------------------------------------------------- % wget http://www.hlla.is.tsukuba.ac.jp/~yas/sie/pdsoft-2003/2004-01-08/ex/create-join-2.c
func2 から作られたスレッド t2
func1 から作られたスレッド t1
pthread_create() により作られる。
どういう順序で実行されるかは、決まっていない。 決まっていない。スレッドは、もともと順番を決めないような処理、 非同期的(asynchronous) な処理を表現するためのもの。どうしても他のスレッドと同期を行なう必要が 出てきた時には、mutex や条件変数といった同期機能を使う。
pthread_create()で指定された関数からリターンすると、そ
のスレッドが終了する。pthread_exit() を呼び出してもよい。
ただし、
初期スレッド
が終了すると、プロセス全体が終了する。
exit() システムコールを呼び出しても終了する。
プロセスの場合、ファイル、端末、ウインドウ、主記憶、CPU時間。
主記憶やCPU時間といった資源は、横取りしても平気なので、あまり問題になりない。
スレッドの場合は、プログラミング言語の変数。
変更されない変数の値を読むだけなら、特に問題は起きない。それ意外の時、 特に、複数のスレッドで値を読み書きする時に問題が起きる。
----------------------------------------------------------------------
1:
2: /*
3: * mutex-nolock.c -- 共有資源をロックなしでアクセスするプログラム
4: */
5:
6: #include <pthread.h>
7:
8: void thread_A(), thread_B();
9: int shared_resource ;
10:
11: main() {
12: pthread_t t1 ;
13: pthread_t t2 ;
14: shared_resource = 0 ;
15: pthread_setconcurrency( 2 );
16: pthread_create( &t1, NULL, (void *)thread_A, 0 );
17: pthread_create( &t2, NULL, (void *)thread_B, 0 );
18: pthread_join( t1, NULL );
19: pthread_join( t2, NULL );
20: printf("main(): shared_resource == %d\n", shared_resource );
21: }
22:
23: void thread_A()
24: {
25: int i, x ;
26: for( i = 0 ; i<1000000 ; i++ )
27: {
28: x = shared_resource ;
29: x = x + 1 ;
30: shared_resource = x ;
31: }
32: }
33:
34: void thread_B() {
35: int i, x ;
36: for( i = 0 ; i<1000000 ; i++ ) {
37: x = shared_resource ;
38: x = x + 1 ;
39: shared_resource = x ;
40: }
41: }
----------------------------------------------------------------------
pthread_setconcurrency() は、利用したい CPU 数をシステムに要求するもの。 Solaris 6 (SunOS 5.6, sakura) では、thr_setconcurrency() を使う。
---------------------------------------------------------------------- % wget http://www.hlla.is.tsukuba.ac.jp/~yas/sie/pdsoft-2003/2004-01-08/ex/mutex-nolock.c
thread_A(), thread_B()と２つのスレッドが作
られている。整数型の変数 shared_resource は、その２つの
スレッドの両方からアクセスされる。２つのスレッドで合計2000000 増える予
定だが、実際に実行すると、そうはならない。実行する度に答えが違う。
このプログラムが共有メモリ型マルチプロセッサで動いているとして、動きを 考えてえる。
---------------------------------------------------------------------- thread_A() thread_B() : : : : 28: x = shared_resource ; 37: x = shared_resource ; 29: x = x + 1 ; 38: x = x + 1 ; 30: shared_resource = x ; 39: shared_resource = x ; : : : : ----------------------------------------------------------------------
shared_resource++と書いてもだめ。複数の機械語命令に分割
されるので。
相互排除(mutual exclusion)： ある資源をアクセスできるスレッドの数を 多くても１つにする。
プログラムの字面上、相互排除が必要な部分を 際どい部分(critical section) ( クリティカルセクション ) という。
Pthread では、相互排除を行なうために、 mutex という仕組みがある。次のようにして、相互排除を行ないたい部分を lock と unlock で囲む。
pthread_mutex_t mutex1 ;
:
:
pthread_mutex_lock( &murex1 );
＜相互排除したい部分（際どい部分）＞
pthread_mutex_unlock( &mutex1 );
----------------------------------------------------------------------
1: /*
2: * mutex-lock.c -- 共有資源をロックしながらアクセスするプログラム
3: */
4:
5: #include <pthread.h>
6:
7: void thread_A(), thread_B();
8: int shared_resource ;
9: pthread_mutex_t mutex1 ;
10:
11: main() {
12: pthread_t t1 ;
13: pthread_t t2 ;
14: shared_resource = 0 ;
15: pthread_mutex_init( &mutex1, NULL );
16: pthread_setconcurrency( 2 );
17: pthread_create( &t1, NULL, (void *)thread_A, 0 );
18: pthread_create( &t2, NULL, (void *)thread_B, 0 );
19: pthread_join( t1, NULL );
20: pthread_join( t2, NULL );
21: printf("main(): shared_resource == %d\n", shared_resource );
22: }
23:
24: void thread_A()
25: {
26: int i, x ;
27: for( i = 0 ; i<1000000 ; i++ )
28: {
29: pthread_mutex_lock( &mutex1 );
30: x = shared_resource ;
31: x = x + 1 ;
32: shared_resource = x ;
33: pthread_mutex_unlock( &mutex1 );
34: }
35: }
36:
37: void thread_B() {
38: int i, x ;
39: for( i = 0 ; i<1000000 ; i++ ) {
40: pthread_mutex_lock( &mutex1 );
41: x = shared_resource ;
42: x = x + 1 ;
43: shared_resource = x ;
44: pthread_mutex_unlock( &mutex1 );
45: }
46: }
----------------------------------------------------------------------
---------------------------------------------------------------------- % wget http://www.hlla.is.tsukuba.ac.jp/~yas/sie/pdsoft-2003/2004-01-08/ex/mutex-lock.c
---------------------------------------------------------------------- thread_A() thread_B() : : 29: pthread_mutex_lock( &mutex1 ); : 40: pthread_mutex_lock( &mutex1 ); 30: x = shared_resource ; ＜他のスレッドが実行中なので 31: x = x + 1 ; この状態でしばらく待たされる＞ 32: shared_resource = x ; : 33: pthread_mutex_unlock( &mutex1 ); : : ＜実行再開＞ : 41: x = shared_resource ; 42: x = x + 1 ; 43: shared_resource = x ; 44: pthread_mutex_unlock( &mutex1 ); ----------------------------------------------------------------------後から来た
thread_B() は、他のスレッドが実行中は、待たさ
れる。
２つのスレッドの間には、バッファを置く。---------------------------------------------------------------------- thread_A | thread_B ----------------------------------------------------------------------
バッファが空の時、thread_B() は、thread_A() が何かデー タをバッファに入れるのを待つ。バッファがいっぱいの時、thread_A() は、thread_B() がバッファから何かデータを取り出すのを待つ。
条件変数の操作：
----------------------------------------------------------------------
1:
2: /*
3: * condv-buffer.c -- 条件変数を使った環状バッファ
4: */
5:
6: #include <pthread.h>
7:
8: void thread_A(), thread_B();
9:
10: #define BUFFER_SIZE 4 /* バッファの大きさ */
11: struct circular_buffer
12: {
13: int rp ; /* 読み出す位置 */
14: int wp ; /* 書き込む位置 */
15: int used ; /* バッファ内の要素数 */
16: int data[BUFFER_SIZE]; /* データを保存する場所 */
17: pthread_mutex_t mutex ; /* この構造体の相互排除のための mutex */
18: pthread_cond_t not_full ; /* バッファが一杯ではない状態を待つための条件変数 */
19: pthread_cond_t not_empty ; /* バッファが空ではない状態を待つための条件変数 */
20: };
21:
22: void put( struct circular_buffer *b,int x )
23: {
24: pthread_mutex_lock( &b->mutex );
25: loop: if( b->used == BUFFER_SIZE )
26: {
27: pthread_cond_wait( &b->not_full,&b->mutex );
28: goto loop;
29: }
30: b->data[ b->wp++ ] = x ;
31: if( b->wp >= BUFFER_SIZE )
32: b->wp = 0 ;
33: b->used ++ ;
34: pthread_cond_signal( &b->not_empty );
35: pthread_mutex_unlock( &b->mutex );
36: }
37:
38: int get( struct circular_buffer *b )
39: {
40: int x ;
41: pthread_mutex_lock( &b->mutex );
42: loop: if( b->used == 0 )
43: {
44: pthread_cond_wait( &b->not_empty,&b->mutex );
45: goto loop;
46: }
47: x = b->data[ b->rp++ ] ;
48: if( b->rp >= BUFFER_SIZE )
49: b->rp = 0 ;
50: b->used -- ;
51: pthread_cond_signal( &b->not_full );
52: pthread_mutex_unlock( &b->mutex );
53: return( x );
54: }
55:
56: main()
57: {
58: pthread_t t1 ;
59: pthread_t t2 ;
60: struct circular_buffer *b ;
61: b = (struct circular_buffer *)malloc(sizeof(struct circular_buffer));
62: if( b == NULL )
63: {
64: perror("no memory for struct buffer\n");
65: exit( -1 );
66: }
67: b->rp = 0 ;
68: b->wp = 0 ;
69: b->used = 0 ;
70: pthread_mutex_init( &b->mutex, NULL );
71: pthread_cond_init( &b->not_full,NULL );
72: pthread_cond_init( &b->not_empty,NULL );
73: pthread_setconcurrency( 2 );
74: pthread_create( &t1, NULL, (void *)thread_A, (void *)b );
75: pthread_create( &t2, NULL, (void *)thread_B, (void *)b );
76: pthread_join( t1, NULL );
77: pthread_join( t2, NULL );
78: }
79:
80: void thread_A( struct circular_buffer *b ) /* producer */
81: {
82: int i,x ;
83: for( i = 0 ; i<10 ; i++ )
84: {
85: x = i ;
86: printf("thread_A(): put( %d )\n",x );
87: put( b,x );
88: }
89: }
90:
91: void thread_B( struct circular_buffer *b ) /* consumer */
92: {
93: int i, x ;
94: for( i = 0 ; i<10 ; i++ )
95: {
96: x = get( b );
97: printf("thread_B(): get() %d.\n", x );
98: }
99: }
----------------------------------------------------------------------
put() は、バッファにデータを追加する時に使う手続き。
基本的には、入口で pthread_mutex_lock() し、
出口で pthread_mutex_unlock() する。
バッファが一杯の時には、条件変数b->not_full で、
一杯でないという条件になるまで待つ。
待っている間は、mutex のロックは解除される。
pthread_cond_wait() からリターンして来る時には、もう一度
ロックされた状態に戻るが、待っている間に、他の変数
(rp,wp,data)が書き換えられている可能性があるので、もう一
度最初から調べる。
get() は、バッファからデータを取り出す時に使う手
続き。put()とほぼ対称形。バッファが空の時に、wait
し、バッファがもはや一杯ではないことをsignal する。
thread_A() は、１０回バッファにデータを書き込むスレッド。
thread_B() は逆に、１０回バッファからデータを読み出すス
レッド。
複数のスレッドが同時に動いている。バッファにためられるのは、最大4なの に、put() が 5 回連続して成功しているように見える。printf() の順番と put(), get() の順番は違うことがある。---------------------------------------------------------------------- % wget http://www.hlla.is.tsukuba.ac.jp/~yas/sie/pdsoft-2003/2004-01-08/ex/condv-buffer.c
バッファに要素を「１つずつ」追加しているので、
pthread_cond_signal() でもよい。
pthread_cond_broadcast() に変えても動くようにプログラムを
作る。
pthread_cond_wait() で待っている間に条件が変わっているかもしれないので、 最初から調べ直す。signal で１人だけしか起き上がらないと仮定してはいけ ない。
「１つずつ」ではなく、複数個同時に読み書きする時には、
pthread_cond_broadcast() でないとだめ。
迷った時には、pthread_cond_broadcast()。
＜−＞再帰呼出し
スレッド間でポインタを渡す時には、スレッドの寿命にも注意。
シングルスレッドのプログラムでは、static変数は、プログラムのモジュール性 を高めるために有効に使われてきた。
マルチスレッドと相性が非常に悪い。static変数もextern変数と同様に複数の スレッドで共有される。変更する場合には、mutex でロックが必要になる。
gethostbyname() は、
static変数に値をセットして、リターン・バリューとして返してる。
----------------------------------------------------------------------
struct hostent *gethostbyname( char *name ){
static struct hostent ret ;
.....
return( &ret );
}
----------------------------------------------------------------------
複数のスレッドが同時にこの関数を呼び出した場合、同じstatic変数が使われ
る。
externやstaticを使わず、auto変数やmalloc()だけを使っているような手続き は、スレッド・セーフ。
別のスレッド・セーフでない手続きを呼んでいれば、それはスレッド・セーフ ではない。
Sun のマニュアルより:
----------------------------------------------------------------------
struct hostent *gethostbyname(const char *name);
struct hostent *gethostbyname_r(const char *name,
struct hostent *result, char *buffer, int buflen,
int *h_errnop);
----------------------------------------------------------------------
O2には、gethostbyname_r() はない。
次のような関数が利用可能である。
#include <semaphore.h>
int sem_init(sem_t *sem, int pshared, unsigned int value)
int sem_wait(sem_t * sem)
int sem_trywait(sem_t * sem)
sem_wait()。0でも止まらずエラーを返す。
int sem_post(sem_t * sem)
int sem_getvalue(sem_t * sem, int * sval)
int sem_destroy(sem_t * sem)
注意：名前付きのセマフォもある。sem_open() で作成／初期化し、 sem_unlink() で削除する。
注意：Solaris には、POSIX のセマフォとは別に、カーネル内でのデバイス・ ドライバ作成のためのセマフォが用意されている。
Java には、最初から言語のレベルでスレッドの機能が入っている。
---------------------------------------------------------------------- Java Pthread ---------------------------------------------------------------------- new Thread(); start(); pthread_create() join() pthread_join() synchronized pthread_mutex_lock()とpthread_mutex_unlock()の組 wait() pthread_cond_wait() wait(long timeout) pthread_cond_timedwait() notify() pthread_cond_signal() notifyAll() pthread_cond_broadcast() ----------------------------------------------------------------------
Java の synchronized は、再帰可能。PTHREAD_MUTEX_RECURSIVE 相当。 １つのスレッドが２度同じオブジェクトをロックしてもよい。
Pthreads のプログラムで、１つの mutex と１つの条件変数を使ったものなら、 Java で簡単に書き直せる。
循環バッファのプログラムは、１つの mutex で２つの条件変数を使っているので、単純には Java で書き直せない。 生産者側と消費者側が同時に待つことはないという性質を利用する。
Java で書かれたスレッドのプログラムは、汚いものがけっこうある。スレッ ド「間」の同期で、対称系になるべき所を、片方のスレッドのメソッドにして、 非対称になっていることがある。Java でプログラムを書く時にも、active object (thread) と passive object (threadなし)をきちんと分けた方がよい。
long と double 以外の型は、原始的。 volatile long, volatle double は原始的。
一種の弱い一貫性(weak consistency)。release consistency に近い。
----------------------------------------------------------------------
1:
2: /*
3: * CircularBuffer.java -- Java による環状バッファ
4: */
5:
6: class CircularBuffer
7: {
8: static final int BUFFER_SIZE = 4 ;
9: int rp ; // 読み出す位置
10: int wp ; // 書き込む位置
11: int data[]; // データを保存する場所
12: int used ; // バッファ内の要素数
13: CircularBuffer()
14: {
15: data = new int[BUFFER_SIZE];
16: rp = 0 ;
17: wp = 0 ;
18: }
19:
20: public synchronized void put( int x ) throws InterruptedException
21: {
22: while( used == data.length )
23: wait();
24: data[ wp++ ] = x ;
25: if( wp == data.length )
26: wp = 0 ;
27: if( used++ == 0 )
28: notifyAll();
29: }
30:
31: public synchronized int get() throws InterruptedException
32: {
33: int x ;
34: while( used == 0 )
35: wait();
36: x = data[ rp++ ] ;
37: if( rp >= data.length )
38: rp = 0 ;
39: if( used-- == data.length )
40: notifyAll();
41: return( x );
42: }
43:
44: static void thread_A( CircularBuffer b ) // Producer
45: {
46: int i,x ;
47: try {
48: for( i = 0 ; i<10 ; i++ )
49: {
50: x = i ;
51: System.out.println("thread_A(): put( "+x+" )");
52: b.put( x );
53: }
54: }
55: catch( InterruptedException e )
56: {
57: System.err.println("thread_A(): Interrupted");
58: }
59: }
60:
61: static void thread_B( CircularBuffer b ) // Consumer
62: {
63: int i, x ;
64: try {
65: for( i = 0 ; i<10 ; i++ )
66: {
67: x = b.get();
68: System.out.println("thread_B(): get() "+x+".");
69: }
70: }
71: catch( InterruptedException e )
72: {
73: System.err.println("thread_B(): Interrupted");
74: }
75: }
76:
77: static void main(String argv[])
78: {
79: final CircularBuffer b = new CircularBuffer();
80: Thread t1 = new Thread( new Runnable()
81: { public void run() {thread_A(b);} } );
82: t1.start();
83: Thread t2 = new Thread( new Runnable()
84: { public void run() {thread_B(b);} } );
85: t2.start();
86: try {
87: t1.join();
88: t2.join();
89: }
90: catch( InterruptedException e )
91: {
92: System.err.println("main(): Interrupted");
93: }
94: }
95: }
----------------------------------------------------------------------
一度に wait する必要があるのは、put() 側か get() のどちらか一方だけである。 よって、両方とも同じキューにつないでも、動作する。
レポートは、次のような形式の電子メールで送ること。
---------------------------------------------------------------------- To: yas@is.tsukuba.ac.jp Subject: [pdsoft/report-thread-1] ＜内容に関したサブジェクト＞ 学籍番号 200301234 (各自の学籍番号で置き換える) 名前 漢字の名前 ＜本文＞ ---------------------------------------------------------------------- ＜内容＞
注意：sakura の新城のログイン名は、yshinjo である。 レポートは、yas@is.tsukuba.ac.jp に送ること。 Subject: には、pdsoft という文字とレポートの番号(report-thread-1)を必 ず含めなさい。
電子メールの本文の先頭に学籍番号と名前（漢字の名前がある人は、漢字で） を書きなさい。
内容は、文法的に正しい、日本語、または英語で書くこと。文章には、述語を 付ける。体言止めは、使ってはいけない。単にプログラムを含めるのではなく、 「以下に○○のプログラムを示す」と書く。実行結果を必ず付ける。
送られたレポートには、最初の締切り以降にまとめて確認応答(acknowledge) を返す。メッセージの紛失、クラッシュ等に対応するために、時間切れによる 再送などのリカバリは適宜やること。
送ったレポートは、講義が終るまで保存しなさい。
１度に１方向の車しか通さない狭い橋がある。橋の上に同時に３台の車が通る と橋が落ちる。この橋が落ちないように、車の交通整理を実現するプログラム を書きなさい。車は、スレッドで実現されるものとする。そして、次のような 手続きを呼び出す。
vehicle(int direction)
{
arrive_bridge( direction );
cross_bridge( direction );
exit_bridge( direction );
}
このコードで direction は、0 または 1 であり、橋のどの方向に車
が渡ろうとしているかを示している。
手続き arrive_bridge() と exit_bridge() を、mutex と条件
変数を使って書きなさい。arrive_bridge() は、安全にその方向で車
が通れるまでリターンしてはならない。衝突したり、重量オーバーで橋が落ち
たりすることがないようにしなければならない。デバッグのためのメッセージ
を適宜画面に出力する。exit_bridge() は、橋を渡り終えことを告げ
るために呼ばれる。この時、可能ならば、待っている車を渡らせ始める
(arrive_bridge() からリターンさせる)。
ここでは、公平性は実現しなくてもよい。また、飢餓状態にならないことを保 証しなくてもよい。
完成したプログラムにおいて、新しい車が来た時、既に別の車が待っていたと する。この場合、新しい車が先に橋を渡るか、それとも古い車が先に橋を渡る か、それとも、予想できないか（非決定的か）。その理由を簡単に説明しなさ い。
水素原子や酸素原子ではなく、水素分子 (H2)や酸素分子 (O2)を使ってもよい。
pthread_create( ..., NULL, (void *)H_func, ... );
pthread_create( ..., NULL, (void *)H_func, ... );
pthread_create( ..., NULL, (void *)H_func, ... );
pthread_create( ..., NULL, (void *)H_func, ... );
pthread_create( ..., NULL, (void *)O_func, ... );
pthread_create( ..., NULL, (void *)O_func, ... );
pthread_create( ..., NULL, (void *)O_func, ... );
H_func( ... )
{
printf(...);
H_Ready(...);
}
O_func( ... )
{
printf(...);
O_Ready(...);
}
H2O_func( ... )
{
printf(...);
}
H_Ready(...)
{
....
}
O_Ready(...)
{
....
}
make_water()
{
pthread_create( ..., NULL, (void *)H2O_func, ... );
}
セマフォを使った循環バッファのプログラムの作成は、次の教科書の練習問 題にもなっている。
清水謙多郎: "オペレーティングシステム",岩波書店 (1992). ISBN: 4000078526.