pthread_cond_wait 取りこぼし？

質問

はじめまして。

pthreadのお勉強がてら、パイプライン処理を実装してみようととりあえず実証コードを書いてみましたが、うまく意図した動きをしてくれません。

やりたいことは、処理ステージが２つあって、メインからステージ１をキックし、ステージ１は自分の処理が終わったらステージ２をキックするといった動作です。（メイン、ステージ１、ステージ２を並列に動作させたい）

取りあえず連鎖的に動作するか試したいだけなので、ステージ間のデータの受け渡しとかは、後で考えるとします。

それで、以下のような単純なコードを書きました。
期待する結果は、最後に表示される数値が 10000, 10000, 10000 になることですが、実際は、10000, 4401, 4401 のようにステージ１，２が少なくなります。
一応、それなりに調べて条件変数のセオリーに従い書いたつもりなのですが、どうしてこうなるか、ご教授ください。

test.c （空白を全角にしてあります）
------
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

pthread_mutex_t　　 mutex1,
　　　　　　　　　　mutex2;
pthread_cond_t　　　cond1,
　　　　　　　　　　cond2;
int　　　　　　　　 ready1,
　　　　　　　　　　ready2;
int　　　　　　　　 end1,
　　　　　　　　　　end2;
int　　　　　　　　 count1,
　　　　　　　　　　count2;

void * stage1( void *arg )
{
　　pthread_mutex_lock( &mutex1 );

　　while( 1 ) {

　　　　/* wait for my signal & job */
　　　　while ( ready1 == 0 ) {
　　　　　　pthread_cond_wait( &cond1, &mutex1 );
　　　　}

　　　　if ( end1 == 1 ){　 /* is shutdown thread */
　　　　　　break;
　　　　}

　　　　/* my job. */
　　　　count1++;

　　　　/* job clear */
　　　　ready1 = 0;

　　　　/* forward next stage */
　　　　pthread_mutex_lock( &mutex2 );

　　　　ready2　= 1;

　　　　pthread_cond_signal( &cond2 );
　　　　pthread_mutex_unlock( &mutex2 );
　　}

　　pthread_mutex_unlock( &mutex1 );

　　return NULL;
}

void * stage2( void *arg )
{
　　pthread_mutex_lock( &mutex2 );

　　while( 1 ) {

　　　　while ( ready2 == 0 ) {
　　　　　　pthread_cond_wait( &cond2, &mutex2 );
　　　　}

　　　　if ( end2 == 1 ){
　　　　　　break;
　　　　}

　　　　count2++;

　　　　ready2 = 0;
　　}

　　pthread_mutex_unlock( &mutex2 );

　　return NULL;
}

int main( )
{
　　int　　　　 i;
　　pthread_t　 t1,
　　　　　　　　t2;

　　pthread_mutex_init( &mutex1, 0 );
　　pthread_cond_init ( &cond1,　0 );
　　ready1 = 0;
　　end1　 = 0;
　　count1 = 0;
　　pthread_create( &t1, 0, stage1, NULL );

　　pthread_mutex_init( &mutex2, 0 );
　　pthread_cond_init ( &cond2,　0 );
　　ready2 = 0;
　　end2　 = 0;
　　count2 = 0;
　　pthread_create( &t2, 0, stage2, NULL );

　　for ( i=0; i<10000; i++ ){
　　　　pthread_mutex_lock( &mutex1 );

　　　　ready1　= 1;

　　　　pthread_cond_signal( &cond1 );
　　　　pthread_mutex_unlock( &mutex1 );
　　}

　　pthread_mutex_lock( &mutex1 );
　　ready1　= 1;
　　end1　　= 1;
　　pthread_cond_signal( &cond1 );
　　pthread_mutex_unlock( &mutex1 );
　　pthread_join(t1, 0 );
　　pthread_cond_destroy( &cond1 );
　　pthread_mutex_destroy( &mutex1 );

　　pthread_mutex_lock( &mutex2 );
　　ready2　= 1;
　　end2　　= 1;
　　pthread_cond_signal( &cond2 );
　　pthread_mutex_unlock( &mutex2 );
　　pthread_join(t2, 0 );
　　pthread_cond_destroy( &cond2 );
　　pthread_mutex_destroy( &mutex2 );

　　printf("%d, %d, %d\n", i, count1, count2);

　　return 0;
}
------
gcc -o test -lpthread test.c

以上

回答No.4

hidebun · ベストアンサー

> まだ受信側の前の処理が終わっていないときは、送信側を待たせるというのは難しいのでしょうか。

スレッドが起こされるタイミングはOSに任されていますから、
送信と受信が完全に1対1で同期しなければならないのであれば、そもそも
スレッドを分割する設計が合わないのではないかという話にもなりそうです。

やるとすれば、受信側は受信処理を完了したら完了フラグを立てて、次の受信を待つ、
送信側は、完了フラグが立っていなければ、sleepするぐらいでしょうか。
ただ、送信の都度sleep処理が走るので、遅くなりますが。

お礼

再度のご回答ありがとうございます。

下記のようにすると同期しますが、仰る通り遅くなりますね。
実際にはステージ間にバッファ（キュー）を入れてクッションを作るようにしないと駄目なことが想像できました。

------
　　for ( i=0; i<10000; i++ ){
　　　　while ( ready1 == 1 ){
　　　　　　sched_yield();
　　　　}
　　　　pthread_mutex_lock( &mutex1 );

　　　　ready1　= 1;

　　　　pthread_cond_signal( &cond1 );
　　　　pthread_mutex_unlock( &mutex1 );
　　}
------

回答No.3

tatsu99

＞まだ受信側の前の処理が終わっていないときは、送信側を待たせるというのは難しいのでしょうか。

簡単な解としては、スリープして、他のスレッドを動かすことになるかと思います。

main　関数にsleepを入れれば、とりあえずは、全て同じ10000が表示されます。

　　for ( i=0; i<10000; i++ ){
　　　　pthread_mutex_lock( &mutex1 );

　　　　ready1　= 1;

　　　　pthread_cond_signal( &cond1 );
　　　　pthread_mutex_unlock( &mutex1 );
　　　　usleep(1);・・・・・・ここでスリープ
　　}
なお、実行時は
#include <unistd.h>を追加してください。

お礼

再度のご回答ありがとうございます。

usleep() は、入れてみたりしましたが、やはりバラツキます。
（環境によって結果が違いました）

回答No.2

hidebun

参考URLに、
> pthread_cond_wait は ( pthread_mutex_unlock による) mutex の
> アンロックと条件変数 cond の送信に対する待機を一息で行う

つまり、stage1の関数内部全体がロックされているわけではないですね。
pthread_cond_wait()を呼び出した時点でロックが解除されます。
これが想定外の動作でしたら、まともに動かないでしょう。

ともかく、ログを埋めてみてください。
メインのfor文もロックを返却してから、次にロックを取得するまでの間隔が短すぎて、
stage1がロックを取得できない状況にもなっているのではないかと想像します。

参考URL：

http://www.linux.or.jp/JM/html/glibc-linuxthreads/man3/pthread_cond_init.3.html

お礼

ご回答ありがとうございます。

確かに、pthread_cond_wait()から抜けないことがありますね。

参考URLに

> mutex のアンロックと条件変数 cond の送信に対する待機を一息で行う。

とあり、

> 呼び出し側のスレッドに戻る前に pthread_cond_wait は mutex を ( pthread_mutex_lock によって)再び獲得する。

とありますが、これはつまり、
呼び出し時のアンロックと待機は、一息で行われるが、
復帰時の受信とロックは、一息ではないということですね。

考えてみれば、送信側でまだロックしていれば、ロックを獲得できないので受信から間が空いてしまい、そして送信側がアンロックしたからと言って、必ずしも自分がロックを獲得できるわけではないということですよね...なるほど。

まだ受信側の前の処理が終わっていないときは、送信側を待たせるというのは難しいのでしょうか。

回答No.1

hidebun

for ( i=0; i<10000; i++ ){
　　　　pthread_mutex_lock( &mutex1 );

　　　　ready1　= 1;

　　　　pthread_cond_signal( &cond1 );
　　　　pthread_mutex_unlock( &mutex1 );
　　}

ここですけれども、メインスレッドがひたすら一人で動作し続けて、
他のスレッドに処理が渡っていないのでは？
メイン, stage1, stage2の各スレッドが協調動作しているかどうか、
それぞれにprintfでログを仕込んで出力してみましょう。
上記のfor文の最後に、usleep(10);なんて加えてみると、動作が変わるかもしれません。

お礼

回答を付けて頂きありがとうございます。

私は何か根本的な勘違いをしているのでしょうか...
メインでpthread_cond_signal()すると、ステージ１ではpthread_cond_wait()で シグナル捕捉と同時にmutexを確保し関数を抜け、自分の処理（ここではカウントアップ）を実行し、次にpthread_cond_wait()が呼び出されるまでは、 mutexは解放されないと理解しているのですが違うのでしょうか？
そうなっていれば、ステージ１の処理中は、メインでは、pthread_mutex_lock()で、ロックするので一人で動作し続けるようなことは無いように思うのですが....