Linux マルチスレッド プログラミングに関する詳細なチュートリアル (スレッドはセマフォを介して通信コードを実装します)

スレッドの分類

スレッドは、スケジューラに従ってユーザーレベルのスレッドとコアレベルのスレッドに分類できます。

(1) ユーザーレベルのスレッド
ユーザーレベルのスレッドは、主にコンテキスト切り替えの問題を解決します。そのスケジューリング アルゴリズムとスケジューリング プロセスはすべてユーザーによって決定され、実行時に特定のカーネル サポートは必要ありません。ここで、オペレーティング システムは多くの場合、スレッドの作成、スケジューリング、キャンセルなどの機能を提供するユーザー空間のスレッド ライブラリを提供しますが、カーネルは依然としてプロセスを管理するだけです。プロセス内のスレッドがブロッキング システム コールを呼び出すと、プロセス内の他のすべてのスレッドを含むプロセスもブロックされます。この種のユーザーレベルのスレッドの主な欠点は、プロセス内で複数のスレッドをスケジュールする際にマルチプロセッサを利用できないことです。

(2) コアレベルのスレッド
この種のスレッドでは、異なるプロセスのスレッドを同じ相対優先度のスケジューリング方法に従ってスケジュールできるため、マルチプロセッサの同時実行の利点を活用できます。
現在、ほとんどのシステムはユーザーレベルのスレッドとコアレベルのスレッドを共存させる方法を採用しています。ユーザーレベルのスレッドは、1 つまたは複数のコアレベルのスレッドに対応できます。これは、「1 対 1」または「多対 1」モデルです。これにより、マルチプロセッサ システムのニーズを満たすだけでなく、スケジューリングのオーバーヘッドも最小限に抑えられます。

Linux スレッドの実装はコアの外部で実行され、コアはプロセスを作成するためのインターフェイス do_fork() を提供します。カーネルは 2 つのシステム コール clone() および fork() を提供し、最終的には異なるパラメーターを使用して do_fork() カーネル API を呼び出します。もちろん、スレッドを実装したい場合は、マルチプロセス (実際には軽量プロセス) 共有データセグメントのコアサポートがなければ不可能です。そのため、do_fork() は CLONE_VM (共有メモリ空間)、CLONE_FS (共有ファイル) を含む多くのパラメータを提供します。システムインフォメーション）、 CLONE_FILES (共有ファイル記述子テーブル)、CLONE_SIGHAND (共有信号ハンドル テーブル)、および CLONE_PID (共有プロセス ID、コア プロセス、つまりプロセス 0 に対してのみ有効)。 fork システムコールを使用する場合、カーネルは共有属性を使用せずに do_fork() を呼び出します。プロセスには独立した実行環境があり、 pthread_create() を使用してスレッドを作成すると、最終的にこれらすべての属性が __clone() を呼び出すように設定され、これらすべてのパラメータがコア内の do_fork() に渡されます。このようにして作成された「プロセス」には共有の実行環境があり、スタックは独立しており、__clone() によって渡されます。

Linux スレッドは、独立したプロセス テーブル エントリを持つコア内の軽量プロセスの形式で存在し、すべての作成、同期、削除、その他の操作はコアの外部の pthread ライブラリで実行されます。 pスレッド ライブラリは、管理スレッド (__pthread_manager()、各プロセスは独立していて一意です) を使用して、スレッドの作成と終了を管理し、スレッド ID をスレッドに割り当て、スレッド関連のシグナル (キャンセルなど) を送信します。 pthread_create()) 呼び出し元は、パイプラインを通じて要求情報を管理スレッドに渡します。

主な機能説明

1.スレッドの作成と終了

pthread_create スレッド作成関数
int pthread_create (pthread_t * thread_id,__const pthread_attr_t * __attr,void *(*__start_routine) (void *),void *__restrict __arg);

スレッド作成関数の最初のパラメータはスレッド識別子へのポインタ、2 番目のパラメータはスレッド属性の設定に使用され、3 番目のパラメータはスレッド実行関数の開始アドレス、最後のパラメータは実行中の関数パラメータ。ここで私たちの関数スレッド パラメータは必要ないため、最後のパラメータは null ポインタに設定されます。また、2 番目のパラメーターを null ポインターに設定します。これにより、デフォルトの属性を持つスレッドが生成されます。スレッドの作成が成功すると、関数は 0 を返します (0 でない場合)。 これは、スレッドの作成が失敗し、一般的なエラー戻りコードが EAGAIN であることを意味します。 そしてアインヴァル。前者は、システムが新しいスレッドの作成を制限していることを意味します。たとえば、スレッドの数が多すぎることを意味します。後者は、2 番目のパラメータで表されるスレッド属性値が不正であることを意味します。スレッドが正常に作成されると、新しく作成されたスレッドはパラメーター 3 とパラメーター 4 で決定された関数を実行し、元のスレッドはコードの次の行の実行を続けます。

pthread_join 関数はスレッドの終了を待ちます。
関数のプロトタイプは次のとおりです: int pthread_join (pthread_t __th, void **__thread_return)
最初のパラメータは待機中のスレッドの識別子で、2 番目のパラメータは待機中のスレッドの戻り値を格納するために使用できるユーザー定義のポインタです。この関数はスレッドブロッキング関数であり、この関数を呼び出した関数は、待機中のスレッドが終了するまで待機し、待機中のスレッドのリソースが回復されます。スレッドは 1 つのスレッドによってのみ終了でき、参加可能な状態 (デタッチされていない状態) である必要があります。

pthread_exit 関数
スレッドを終了するには 2 つの方法があります。1 つは、スレッドによって実行されている関数が終了すると、それを呼び出したスレッドも終了する方法です。もう 1 つは、関数 pthread_exit を使用する方法です。 満たすため。その関数のプロトタイプは次のとおりです: void pthread_exit (void *__retval) pthread_join の場合、唯一のパラメーターは関数の戻りコードです。 2 番目のパラメータ thread_return は、 NULL ではない場合、この値は thread_return に渡されます。最後に注意すべきことは、スレッドを複数のスレッドで待機させることはできないということです。そうでない場合は、シグナルを受信した最初のスレッドが正常に戻り、残りのスレッドが pthread_join を呼び出します。 スレッドはエラー コード ESRCH を返します。

pthread_create 関数の 2 番目のパラメーターは、スレッドのプロパティです。この値を NULL に設定します。つまり、スレッドの多くの属性は変更できます。これらのプロパティには主に、バインディング プロパティ、分離プロパティ、スタック アドレス、スタック サイズ、優先順位が含まれます。システムのデフォルト属性は非結合、非分離であり、デフォルトは 1M です。 スタックと親プロセスと同じ優先度レベルを持ちます。まず、バインド属性とデタッチ属性の基本概念について説明します。

バインディング属性: Linux は「1 対 1」スレッド メカニズムを採用しています。つまり、1 つのユーザー スレッドが 1 つのカーネル スレッドに対応します。バインディング属性は、CPU タイム スライスのスケジューリングがカーネル スレッド用であるため、ユーザー スレッドがカーネル スレッドに固定的に割り当てられることを意味します。 (つまり、軽量プロセス)、バインディング属性を持つスレッドは、必要なときにそれに対応するカーネル スレッドが常に存在することを保証できます。対照的に、非拘束属性は、ユーザー スレッドとカーネル スレッド間の関係が常に固定されているわけではなく、システムによって制御および割り当てられることを意味します。

デタッチメント属性: デタッチメント属性は、スレッド自体を終了する方法を決定するために使用されます。非デタッチの場合、スレッドが終了しても、スレッドが占有しているシステム リソースは解放されません。つまり、実際の終了はありません。 pthread_join() 関数が戻った場合にのみ、作成されたスレッドは占有しているシステム リソースを解放できます。切り離された属性の場合、それが占有しているシステム リソースは、スレッドが終了するとすぐに解放されます。
ここで注意すべき点は、スレッドの切り離し属性を設定し、このスレッドが非常に高速に実行される場合、そのスレッドは pthread_create にある可能性が高いということです。 関数は終了後、スレッド番号とシステム リソースが他のスレッドに引き渡される可能性があります。このとき、pthread_create を呼び出すスレッドは間違ったスレッド番号を取得します。

バインディング属性を設定します:

int pthread_attr_init(pthread_attr_t *attr)
int pthread_attr_setscope(pthread_attr_t *attr, int スコープ)
int pthread_attr_getscope(pthread_attr_t *tattr, int *scope)
scope: PTHREAD_SCOPE_SYSTEM: バインディング、このスレッドはシステム内のすべてのスレッドと競合します PTHREAD_SCOPE_PROCESS: アンバウンド、このスレッドはプロセス内の他のスレッドと競合します

デタッチ属性を設定します:

int pthread_attr_setdetachstate(pthread_attr_t *attr, int detachstate)
int pthread_attr_getdetachstate(const pthread_attr_t *tattr,int *detachstate)
detachstate PTHREAD_CREATE_DETACHED: PTHREAD を切り離します _CREATE_JOINABLE: 非分離

スケジュール ポリシーの設定:

int pthread_attr_setschedpolicy(pthread_attr_t * tattr, intpolicy)
int pthread_attr_getschedpolicy(pthread_attr_t *tattr, int *policy)
policy SCHED_FIFO: 先入れ先出し SCHED_RR: ループ SCHED_OTHER: 実装定義メソッド

優先順位の設定:

int pthread_attr_setschedparam (pthread_attr_t *attr, struct sched_pa​​ram *param)
int pthread_attr_getschedparam (pthread_attr_t *attr, struct sched_pa​​ram *param)

3.スレッドアクセス制御

1) ミューテックスロック(mutex)
は、ロック機構によりスレッド間の同期を実現します。一度に 1 つのスレッドだけが、コードの 1 つの重要なセクションを実行できます。

1 int pthread_mutex_init(pthread_mutex_t *mutex,const pthread_mutex_attr_t *mutexattr);
2 int pthread_mutex_lock(pthread_mutex_t *mutex);
3 int pthread_mutex_unlock(pthread_mutex_t *mutex);
4 int pthread_mutex_destroy(pthread_mutex_t *mutex);

(1) まずロック init() を初期化するか、pthread_mutex_t を静的に割り当てます mutex=PTHREAD_MUTEX_INITIALIER
(2) ロック、ロック、trylock、ロックを待機しているロックブロック、trylock はすぐに EBUSY を返します
(3) ロック解除、ロック解除はロック状態でなければならず、ロックスレッドによってロック解除される必要があります
(4) lock、destroy (このときロックを解除する必要があります。ロックを解除しないと EBUSY が返されます)

mutex は 2 つのタイプに分けられます: 再帰的と非再帰的。これは POSIX 名です。もう 1 つはリエントラントです。 非リエントラント。スレッド間同期ツールとして使用する場合、これら 2 つのミューテックスに違いはありません。唯一の違いは、同じスレッドが繰り返しミューテックスを再帰的に実行できることです。 ミューテックスはロックされていますが、非再帰ミューテックスは繰り返しロックできません。 ロック。
パフォーマンスのためではなく、設計意図を反映するために、非再帰ミューテックスが推奨されます。非再帰的と再帰的 使用するカウンターが 1 つ少なく、前者の方が少し速いだけなので、2 つの間のパフォーマンスの差は実際には大きくありません。同じスレッド内で非再帰ミューテックスを複数回繰り返す ロックするとすぐにデッドロックが発生します。これは、コードのロック要件を検討し、問題を早期に (コーディング段階で) 検出するのに役立つと思います。間違いなく再帰ミューテックスです スレッドのロック自体を心配する必要がないため、Java と Windows がデフォルトで再帰ミューテックスを提供しているのはこのためだと思います。 (ジャワ この言語に付属する固有のロックは再入可能であり、その同時ロックは このライブラリは ReentrantLock を提供しており、Windows の CRITICAL_SECTION もリエントラントです。それらのどれも軽量の非再帰を提供していないようです ミューテックス。 )

2) 条件変数(cond)
スレッド間で共有されるグローバル変数を同期に使用する仕組み。

1 int pthread_cond_init(pthread_cond_t *cond,pthread_condattr_t *cond_attr);
2 int pthread_cond_wait(pthread_cond_t *cond,pthread_mutex_t *mutex);
3 int pthread_cond_timedwait(pthread_cond_t *cond,pthread_mutex_t *mutex,const timespec *abstime);
4 int pthread_cond_destroy(pthread_cond_t *cond); int pthread_cond_signal(pthread_cond_t *cond);
6 int pthread_cond_broadcast(pthread_cond_t *cond); //すべてのスレッドのブロックを解除します

(1) init() または pthread_cond_t cond=PTHREAD_COND_INITIALIER; 属性を NULL に設定します
(2) 条件が成立するまで待ちます。 pthread_cond_wait、pthread_cond_timedwait。
wait() はロックを解放し、条件変数が true になるのを待ちます。
timedwait() は待機時間を設定します (ロックにより、スレッドの待機が 1 つだけになることが保証されます)。
(3) 条件変数をアクティブにする: pthread_cond_signal、pthread_cond_broadcast (待機中のすべてのスレッドをアクティブにする)
(4) 条件変数をクリアする: destroy; 待機中のスレッドはありません。そうでない場合は、EBUSY が返されます

int pthread_cond_wait(pthread_cond_t *cond, pthread_mutex_t *mutex);
int pthread_cond_timedwait(pthread_cond_t *cond, pthread_mutex_t *mutex, const struct timespec *abstime);

#include <stdio.h>
#include <stdlib.h>
#include <time.h>
#include "pthread.h"
#define BUFFER_SIZE 16
struct prodcons  
{  
int buffer[BUFFER_SIZE];  
pthread_mutex_t lock;  //mutex ensuring exclusive access to buffer
int readpos,writepos;  //position for reading and writing
pthread_cond_t notempty;  //signal when buffer is not empty
pthread_cond_t notfull;  //signal when buffer is not full
};  
//initialize a buffer
void init(struct prodcons* b)  
{  
pthread_mutex_init(&b->lock,NULL);  
pthread_cond_init(&b->notempty,NULL);  
pthread_cond_init(&b->notfull,NULL);  
b->readpos = 0;  
b->writepos = 0;  
}  
//store an integer in the buffer
void put(struct prodcons* b, int data)  
{  
pthread_mutex_lock(&b->lock);  
//wait until buffer is not full
while((b->writepos+1)%BUFFER_SIZE == b->readpos)  
{  
printf("wait for not full\n");  
pthread_cond_wait(&b->notfull,&b->lock);  
}
b->buffer[b->writepos] = data;  
b->writepos++;
b->writepos %= BUFFER_SIZE;
pthread_cond_signal(&b->notempty); //signal buffer is not empty
pthread_mutex_unlock(&b->lock);  
}
//read and remove an integer from the buffer
int get(struct prodcons* b)  
{  
int data;  
pthread_mutex_lock(&b->lock);  
//wait until buffer is not empty
while(b->writepos == b->readpos)  
{  
printf("wait for not empty\n");  
pthread_cond_wait(&b->notempty,&b->lock);  
}
data=b->buffer[b->readpos];  
b->readpos++;
b->readpos %= BUFFER_SIZE;
pthread_cond_signal(&b->notfull);  //signal buffer is not full
pthread_mutex_unlock(&b->lock);  
return data;
}
#define OVER -1
struct prodcons buffer;  
void * producer(void * data)  
{  
int n;  
for(n=0; n<50; ++n)  
{
printf("put-->%d\n",n);  
put(&buffer,n);  
}  
put(&buffer,OVER);  
printf("producer stopped\n");  
return NULL;  
}  
void * consumer(void * data)  
{  
int n;  
while(1)  
{  
int d = get(&buffer);  
if(d == OVER) break;  
printf("get-->%d\n",d);  
}
printf("consumer stopped\n");  
return NULL;  
}  
int main()  
{  
pthread_t tha,thb;  
void * retval;  
init(&buffer);  
pthread_creare(&tha,NULL,producer,0);  
pthread_creare(&thb,NULL,consumer,0);  
pthread_join(tha,&retval);  
pthread_join(thb,&retval);  
return 0;  
}

プロセスと同様、スレッドもセマフォを介して通信できますが、セマフォは軽量です。

#include <semaphore.h>
int sem_init(sem_t *sem , int pshared, unsigned int value);

int sem_wait(sem_t *sem); //给信号量减1，对一个值为0的信号量调用sem_wait，这个函数将会等待直到有其它线程使它不再是0为止。
int sem_post(sem_t *sem); //给信号量的值加1
int sem_destroy(sem_t *sem);

#define BSIZE 64
typedef struct 
{
char buf[BSIZE];
sem_t occupied;
sem_t empty;
int nextin;
int nextout;
sem_t pmut;
sem_t cmut;
}buffer_t;
buffer_t buffer;
void init(buffer_t * b)
{
sem_init(&b->occupied, 0, 0);
sem_init(&b->empty,0, BSIZE);
sem_init(&b->pmut, 0, 1);
sem_init(&b->cmut, 0, 1);
b->nextin = b->nextout = 0;
}
void producer(buffer_t *b, char item) 
{
sem_wait(&b->empty);
sem_wait(&b->pmut);
b->buf[b->nextin] = item;
b->nextin++;
b->nextin %= BSIZE;
sem_post(&b->pmut);
sem_post(&b->occupied);
}
char consumer(buffer_t *b) 
{
char item;
sem_wait(&b->occupied);
sem_wait(&b->cmut);
item = b->buf[b->nextout];
b->nextout++;
b->nextout %= BSIZE;
sem_post(&b->cmut);
sem_post(&b->empty);
return item;
}