10.7 Threads synchronisieren
 
In vielen Fällen, eigentlich bei Threads fast immer, werden ja mehrere parallel laufende Prozesse benötigt, die gemeinsame Daten verwenden und/oder austauschen. Einfachstes Beispiel, ein Thread schreibt gerade etwas in eine Datei, während ein anderer Thread daraus etwas liest. Selbes Problem haben Sie auch beim Zugriff auf globale Variablen. Wenn mehrere Threads darauf zugreifen müssen und Sie hierbei keine Vorkehrungen getroffen haben, ist nicht vorherzusagen, welcher Thread die Variable gerade bearbeitet. Sind hierbei z. B. mathematische Arbeiten auf mehreren Threads aufgeteilt, kann man mit fast 100 %iger Sicherheit sagen, dass das Ergebnis nicht richtig sein wird.
Hierfür sei folgendes einfaches Beispiel gegeben. Zwei Threads greifen auf eine globale Variable zu – hier auf einen geöffneten FILE-Zeiger. Ein Thread wird erzeugt, um etwas in diese Datei zu schreiben, und ein weiterer Thread soll diese wieder auslesen. Ein simples Beispiel, wie es scheint, nur dass es hierbei schon zu Synchronisationsproblemen (Race Conditions) kommt. Aber testen Sie selbst.
/* thread5.c */
#include <stdio.h>
#include <stdlib.h>
#include <unistd. h.>
#include <pthread. h.>
#define MAX_THREADS 2
#define BUF 255
#define COUNTER 10000000
/* Globale Variable */
static FILE *fz;
static void open_file(const char *file) {
fz = fopen( file, "w+" );
if( fz == NULL ) {
printf("Konnte Datei %s nicht öffnen\n", file);
exit(EXIT_FAILURE);
}
}
static void thread_schreiben(void *name) {
char string[BUF];
printf("Bitte Eingabe machen: ");
fgets(string, BUF, stdin);
fputs(string, fz);
fflush(fz);
/* Thread-Ende */
pthread_exit((void *)pthread_self());
}
static void thread_lesen(void *name) {
char string[BUF];
rewind(fz);
fgets(string, BUF, fz);
printf("Ausgabe Thread %ld: ", pthread_self());
fputs(string, stdout);
fflush(stdout);
/* Thread-Ende */
pthread_exit((void *)pthread_self());
}
int main (void) {
static pthread_t th1, th2;
static int ret1, ret2;
printf("->Haupt-Thread (ID:%ld) gestartet ...\n",
pthread_self());
open_file("testfile");
/* Threads erzeugen */
if (pthread_create( &th1, NULL,
&thread_schreiben, NULL)!=0) {
fprintf (stderr, "Konnte keinen Thread erzeugen\n");
exit (EXIT_FAILURE);
}
/* Threads erzeugen */
if (pthread_create(&th2,NULL,&thread_lesen,NULL) != 0) {
fprintf (stderr, "Konnte keinen Thread erzeugen\n");
exit (EXIT_FAILURE);
}
pthread_join(th1, &ret1);
pthread_join(th2, &ret2);
printf("<-Thread %ld fertig\n", th1);
printf("<-Thread %ld fertig\n", th1);
printf("<-Haupt-Thread (ID:%ld) fertig ...\n",
pthread_self());
return EXIT_SUCCESS;
}
Das Programm bei der Ausführung:
$ gcc -o thread5 thread5.c -lpthread
$ ./thread5
->Haupt-Thread (ID:-1209412512) gestartet...
Bitte Eingabe machen: Ausgabe Thread -1217807440: Hallo, das ist ein Test
<-Thread -1209414736 fertig
<-Thread -1209414736 fertig
<-Haupt-Thread (ID:-1209412512) fertig ...
$ cat testfile
Hallo, das ist ein Test
Bei der Eingabe können Sie schon erkennen, dass der Thread »thread_lesen« schon mit seiner Ausgabe begonnen hat und sich schon wieder beendet hat, bevor Sie etwas von der Tastatur eingeben konnten. Richtig ausgeführt sollte hier folgende Ausgabe bei der Programmausführung entstehen:
$ ./thread5
->Haupt-Thread (ID:-1209412512) gestartet ...
Bitte Eingabe machen: Hallo Welt
Ausgabe Thread -1217807440: Hallo Welt
<-Thread -1209414736 fertig
<-Thread -1209414736 fertig
<-Haupt-Thread (ID:-1209412512) fertig ...
Zugegeben, als echter C-Guru würde Ihnen jetzt hier schon etwas einfallen, z. B. eine »pollende« Schleife mit einem sleep() um den Lese-Thread herumzubauen, die immer wieder abfragt, ob fgets() etwas eingelesen hat. Na ja, das wäre wohl nicht im Sinne des Erfinders, und wenn Sie wirklich die Threads für Echtzeitanwendungen verwenden wollen, ist das wohl auch das Ende Ihrer Programmiererkarriere, wenn die Weichenschaltung einer U-Bahn »in einer pollenden Schleife« warten muss, bevor diese gestellt werden kann! Und außerdem gibt es für solche Fälle einige Synchronisationsmöglichkeiten, die Ihnen die Thread-Bibliothek anbietet.
10.7.1 Mutexe
Wenn Sie mehrere Threads starten und diese quasi parallel ablaufen, können Sie nicht erkennen, wie weit welcher Thread gerade mit der Verarbeitung von Daten ist. Wenn mehrere Threads beispielsweise an ein und derselben Aufgabe abhängig voneinander arbeiten, wird eine Synchronisation erforderlich. Genauso ist dies erforderlich, wenn Threads globale Variablen oder die Hardware wie z. B. die Tastatur (stdin) verwenden, da sonst ein Thread diese Variable einfach überschreiben würde, bevor sie noch verwendet wird.
Um Threads zu synchronisieren, haben Sie zwei Möglichkeiten, zum einen mit so genannten Locks, die Sie in diesem Kapitel mit den Mutexen durchgehen werden, und zum anderen mit einem Monitor. Mit dem Monitor werden so genannte Condition-Variablen verwendet.
|
Hinweis Der Begriff »Mutex« steht für »Mutual Exclusion Device« (= gegenseitiger Ausschluss). Ein Mutex ist somit ohne Besitzer oder gehört genau einem Thread.
|
Die Funktionsweise von Mutexen ähnelt den Semaphoren bei den Prozessen. Genauer noch: Ein Mutex ist nichts weiter als ein Semaphor, was wiederum nur eine atomare Operation auf eine Variable ist. Trotzdem lassen sich diese aber wesentlich einfacher erstellen. Das Prinzip ist simpel. Ein Thread arbeitet mit einer globalen oder statischen Variablen, die für alle anderen Threads von einem Mutex blockiert (gesperrt) wird. Benötigt der Thread diese Variable nicht mehr, gibt er diese frei.
Anhand dieser Erklärung dürfte auch klar sein, dass man selbst dafür verantwortlich ist, keinen Deadlock zu erzeugen. In folgenden Fällen könnten auch bei Threads Deadlocks auftreten:
|
Threads können Ressourcen anfordern, obwohl sie bereits Ressourcen besitzen. |
|
|
Ein Thread gibt seine Ressource nicht mehr frei. |
|
|
Eine Resource ist frei oder im Besitz eines »exklusiven« Threads. |
Im Falle eines Deadlocks kann keiner der beteiligten Threads seine Arbeit mehr fortsetzen, und somit ist meist keine normale Beendigung mehr möglich. Datenverlust kann die Folge sein.
Statische Mutexe
Um eine Mutex-Variable als statisch zu definieren, müssen Sie diese mit der Konstante PTHREAD_MUTEX_INITIALIZER initialisieren. Folgende Funktionen stehen Ihnen zur Verfügung, um Mutexe zu sperren und wieder freizugeben:
#include <pthread. h.>
int pthread_mutex_lock(pthread_mutex_t *mutex);
int pthread_mutex_trylock(pthread_mutex_t *mutex);
int pthread_mutex_unlock(pthread_mutex_t *mutex);
Mit pthread_mutex_lock() sperren Sie einen Mutex. Wenn hierbei z. B. ein Thread versucht, mit demselben Mutex ebenfalls eine Sperre einzurichten, so wird dieser so lange blockiert, bis der Mutex von einem anderen Thread wieder mittels pthread_mutex_unlock() freigegeben wird.
Die Funktion pthread_mutex_trylock() ist ähnlich wie pthread_mutex_lock(), nur dass diese Funktion den aufrufenden Thread nicht blockiert, wenn ein Mutex durch einen anderen Thread blockiert wird. pthread_mutex_trylock() kehrt stattdessen mit dem Fehlercode (errno) EBUSY zurück und macht mit der Ausführung des aufrufenden Threads weiter.
Das folgende Beispiel ist dasselbe, wie Sie es schon vom Listing »thread5.c« her kennen, nur dass jetzt das Synchronisationsproblem mithilfe eines Mutex behoben wird. Zuerst wird global der Mutex mit der Konstante PTHREAD_MUTEX_INITIALIZER statisch initialisiert, und anschließend werden im Beispiel die Sperren dort gesetzt und wieder freigegeben, wo dies sinnvoll ist.
/* thread6.c */
#include <stdio.h>
#include <stdlib.h>
#include <unistd. h.>
#include <pthread. h.>
#define MAX_THREADS 2
#define BUF 255
#define COUNTER 10000000
static FILE *fz;
/* Statische Mutex-Variable */
pthread_mutex_t fz_mutex=PTHREAD_MUTEX_INITIALIZER;
static void open_file(const char *file) {
fz = fopen( file, "w+" );
if( fz == NULL ) {
printf("Konnte Datei %s nicht öffnen\n", file);
exit(EXIT_FAILURE);
}
}
static void thread_schreiben(void *name) {
char string[BUF];
printf("Bitte Eingabe machen: ");
fgets(string, BUF, stdin);
fputs(string, fz);
fflush(fz);
/* Mutex wieder freigeben */
pthread_mutex_unlock( &fz_mutex );
/* Thread-Ende */
pthread_exit((void *)pthread_self());
}
static void thread_lesen(void *name) {
char string[BUF];
/* Mutex sperren */
pthread_mutex_lock( &fz_mutex );
rewind(fz);
fgets(string, BUF, fz);
printf("Ausgabe Thread %ld: ", pthread_self());
fputs(string, stdout);
fflush(stdout);
/* Mutex wieder freigeben */
pthread_mutex_unlock( &fz_mutex );
/* Thread-Ende */
pthread_exit((void *)pthread_self());
}
int main (void) {
static pthread_t th1, th2;
static int ret1, ret2;
printf("->Haupt-Thread (ID:%ld) gestartet ...\n",
pthread_self());
open_file("testfile");
/* Mutex sperren */
pthread_mutex_lock( &fz_mutex );
/* Threads erzeugen */
if( pthread_create( &th1, NULL, &thread_schreiben,
NULL)!=0) {
fprintf (stderr, "Konnte keinen Thread erzeugen\n");
exit (EXIT_FAILURE);
}
/* Threads erzeugen */
if(pthread_create(&th2,NULL, &thread_lesen, NULL) != 0) {
fprintf (stderr, "Konnte keinen Thread erzeugen\n");
exit (EXIT_FAILURE);
}
pthread_join(th1, &ret1);
pthread_join(th2, &ret2);
printf("<-Thread %ld fertig\n", th1);
printf("<-Thread %ld fertig\n", th1);
printf("<-Haupt-Thread (ID:%ld) fertig ...\n",
pthread_self());
return EXIT_SUCCESS;
}
Das Programm bei der Ausführung:
$ gcc -o thread6 thread6.c -lpthread
$ ./thread6
->Haupt-Thread (ID:-1209412512) gestartet ...
Bitte Eingabe machen: Hallo Welt mit Mutexe
Ausgabe Thread -1217807440: Hallo Welt mit Mutexe
<-Thread -1209414736 fertig
<-Thread -1209414736 fertig
<-Haupt-Thread (ID:-1209412512) fertig ...
Natürlich können Sie den Lese-Thread mit pthread_mutex_trylock() als eine nicht blockierende Mutex-Anforderung ausführen. Hierzu müssten Sie nur die Funktion »thread_lesen« ein wenig umändern. Hier ein solcher möglicher Ansatz:
static void thread_lesen(void *name) {
char string[BUF];
/* Versuche Mutex zu sperren */
while( (pthread_mutex_trylock( &fz_mutex )) == EBUSY) {
sleep(10);
printf("Lese-Thread wartet auf Arbeit ...\n");
printf("Bitte Eingabe machen: ");
fflush(stdout);
}
rewind(fz);
fgets(string, BUF, fz);
printf("Ausgabe Thread %ld: ", pthread_self());
fputs(string, stdout);
fflush(stdout);
/* Mutex wieder freigeben */
pthread_mutex_unlock( &fz_mutex );
/* Thread-Ende */
pthread_exit((void *)pthread_self());
}
Hierbei wird versucht, alle zehn Sekunden den Mutex zu sperren. Solange EBUSY zurückgegeben wird, ist der Mutex noch von einem anderen Thread gesperrt. Während dieser Zeit könnte der wartende Thread ja andere Arbeiten ausführen (es gibt immer was zu tun). Das Programm bei der Ausführung mit pthread_mutex_trylock():
$ gcc -o thread7 thread7.c -lpthread
$ ./thread7
->Haupt-Thread (ID:-1209412512) gestartet ...
Bitte Eingabe machen: Lese-Thread wartet auf Arbeit ...
Bitte Eingabe machen: Lese-Thread wartet auf Arbeit ...
Bitte Eingabe machen: Hallo Mutex, Du bist frei
Ausgabe Thread -1217807440: Hallo Mutex, Du bist frei
<-Thread -1209414736 fertig
<-Thread -1209414736 fertig
<-Haupt-Thread (ID:-1209412512) fertig ...
Dynamische Mutexe
Wenn Sie Mutexe in einer Struktur verwenden wollen, was durchaus eine gängige Praxis ist, können Sie dynamische Mutexe verwenden. Dies sind dann Mutexe, für die zur Laufzeit mit z. B. malloc() Speicher angefordert wird. Für dynamische Mutexe stehen folgende Funktionen zur Verfügung:
#include <pthread. h.>
int pthread_mutex_init(
pthread_mutex_t *mutex,
const pthread_mutex_attr_t *mutexattr );
int pthread_mutex_destroy(pthread_mutex_t *mutex);
Mit pthread_mutex_init() initialisieren Sie das Mutex mutex. Mit dem Parameter mutexattr können Sie Attribute für das Mutex verwenden. Wird hierbei NULL angegeben, werden die Standardattribute verwendet. Auf die Attribute von Mutexen wird in Kürze eingegangen. Freigeben können Sie einen solchen dynamisch angelegten Mutex wieder mit pthread_mutex_destroy(). Hierzu nochmals dasselbe Beispiel wie eben mit »thread6.c«, nur mit dynamisch angelegtem Mutex.
/* thread8.c */
#include <stdio.h>
#include <stdlib.h>
#include <unistd. h.>
#include <errno.h>
#include <pthread. h.>
#define BUF 255
struct data {
FILE *fz;
char filename[BUF];
pthread_mutex_t mutex;
};
static void thread_schreiben(void *arg) {
char string[BUF];
struct data *d=(struct data *)arg;
printf("Bitte Eingabe machen: ");
fgets(string, BUF, stdin);
fputs(string, d->fz);
fflush(d->fz);
/* Mutex wieder freigeben */
pthread_mutex_unlock( &d->mutex );
/* Thread-Ende */
pthread_exit((void *)pthread_self());
}
static void thread_lesen(void *arg) {
char string[BUF];
struct data *d=(struct data *)arg;
/* Mutex sperren */
while( (pthread_mutex_trylock( &d->mutex )) == EBUSY) {
sleep(10);
printf("Lese-Thread wartet auf Arbeit ...\n");
printf("Bitte Eingabe machen: ");
fflush(stdout);
}
rewind(d->fz);
fgets(string, BUF, d->fz);
printf("Ausgabe Thread %ld: ", pthread_self());
fputs(string, stdout);
fflush(stdout);
/* Mutex wieder freigeben */
pthread_mutex_unlock( &d->mutex );
/* Thread-Ende */
pthread_exit((void *)pthread_self());
}
int main (void) {
static pthread_t th1, th2;
static int ret1, ret2;
struct data *d;
/* Speicher für die Struktur reservieren */
d = malloc(sizeof(struct data));
if(d == NULL) {
printf("Konnte keinen Speicher reservieren ...!\n");
exit(EXIT_FAILURE);
}
printf("->Haupt-Thread (ID:%ld) gestartet ...\n",
pthread_self());
strncpy(d->filename, "testfile", BUF);
d->fz = fopen( d->filename, "w+" );
if( d->fz == NULL ) {
printf("Konnte Datei %s nicht öffnen\n", d->filename);
exit(EXIT_FAILURE);
}
/* Mutex initialisieren */
pthread_mutex_init( &d->mutex, NULL );
/* Mutex sperren */
pthread_mutex_lock( &d->mutex );
/* Threads erzeugen */
if(pthread_create (&th1,NULL,&thread_schreiben,d) != 0) {
fprintf (stderr, "Konnte keinen Thread erzeugen\n");
exit (EXIT_FAILURE);
}
/* Threads erzeugen */
if (pthread_create (&th2,NULL, &thread_lesen, d) != 0) {
fprintf (stderr, "Konnte keinen Thread erzeugen\n");
exit (EXIT_FAILURE);
}
pthread_join(th1, &ret1);
pthread_join(th2, &ret2);
/* Dynamisch angelegten Mutex löschen */
pthread_mutex_destroy( &d->mutex );
printf("<-Thread %ld fertig\n", th1);
printf("<-Thread %ld fertig\n", th1);
printf("<-Haupt-Thread (ID:%ld) fertig ...\n",
pthread_self());
return EXIT_SUCCESS;
}
Das Programm bei der Ausführung kann ich mir hier ersparen, da es exakt dem Beispiel »thread6.c« entspricht, nur dass hierbei eben ein dynamischer Mutex statt eines statischen verwendet wurde.
Mutex-Attribute
Mit den folgenden Funktionen können Sie Mutex-Attribute verändern oder abfragen:
#include <pthread. h.>
int pthread_mutexattr_init( pthread_mutexattr_t *attr );
int pthread_mutexattr_destroy( pthread_mutexattr_t *attr );
int pthread_mutexattr_settype( pthread_mutexattr_t *attr,
int kind );
int pthread_mutexattr_gettype(
const pthread_mutexattr_t *attr, int *kind );
Mit dem Mutex-Attribut legen Sie fest, was passiert, wenn ein Thread versuchen sollte, einen Mutex nochmals zu sperren, obwohl dieser bereits mit pthread_mutex_lock() gesperrt wurde. Mit der Funktion pthread_mutexattr_init() initialisieren Sie zunächst das Mutex-Attributobjekt attr. Zunächst wird hierbei die Standardeinstellung (PTHREAD_MUTEX_FAST_NP) verwendet. Ändern können Sie dieses Attribut mit pthread_mutexattr_settype(). Damit setzen Sie die Attribute des Mutex-Attributobjekts auf kind. Folgende Konstanten können Sie hierbei für kind verwenden:
|
|
PTHREAD_MUTEX_FAST_NP (Standardeinstellung) – pthread_mutex_lock() blockiert den aufrufenden Thread für immer. Also ein Deadlock. |
|
|
PTHREAD_MUTEX_RECURSIVE_NP – pthread_mutex_lock() blockiert nicht und kehrt sofort erfolgreich zurück. Wird ein Thread mit diesem Mutex gesperrt, so wird ein Zähler für jede Sperrung um den Wert 1 erhöht. Damit die Sperrung eines rekursiven Mutex aufgehoben wird, muss dieser ebenso oft freigegeben werden, wie er gesperrt wurde. |
|
|
PTHREAD_MUTEX_ERRORCHECK_NP – pthread_mutex_lock() kehrt sofort wieder mit dem Fehlercode EDEADLK zurück, also ähnlich wie mit pthread_mutex_trylock(), nur dass hier eben EBUSY zurückgegeben wird. |
|
Hinweis Da die Variablen hierbei mit dem Suffix _NP (non-portable) verbunden sind, sind diese nicht mit dem POSIX-Standard vereinbar und somit nicht geeignet für portable Programme.
|
10.7.2 Condition-Variablen (Bedingungsvariablen)
Bedingungsvariablen werden dazu verwendet, auf das Eintreffen einer bestimmten Bedingung zu warten bzw. die Erfüllung oder den Eintritt einer Bedingung zu zeigen. Bedingungsvariablen werden außerdem mit den Mutexen verknüpft. Dabei wird beim Warten auf eine Bedingung eine Sperre zu einem dazu verknüpften Mutex freigegeben (natürlich musste zuvor eine Sperre auf den Mutex erfolgt sein).
Andersherum sollte vor einem Eintreffen auf einer Bedingung eine Sperre auf den verknüpften Mutex erfolgen, so dass nach dem Warten auf diesen Mutex auch die Sperre auf den Mutex wieder vorhanden ist. Erfolgte keine Sperre vor dem Signal, wartet ein Thread wieder, bis eine Sperre auf den Mutex möglich ist.
Statische Bedingungsvariablen
Für die Bedingungsvariablen wird der Datentyp pthread_cond_t verwendet. Damit eine solche Bedingungsvariable überhaupt als statisch definiert ist, muss diese mit der Konstante PTHREAD_COND_INITIALIZER initialisiert werden. Hier die Funktionen, womit Sie mit Condition-Variablen operieren können:
#include <pthread. h.>
int pthread_cond_signal( pthread_cond_t *cond );
int pthread_cond_broadcast( pthread_cond_t *cond );
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);
Bevor Sie zunächst die Funktion pthread_cond_wait() verwenden, müssen Sie beim aufrufenden Thread das Mutex mutex sperren. Mit einem anschließenden pthread_cond_wait() wird der Mutex dann freigegeben, und der Thread wird mit der Bedingungsvariablen cond, bis zum Eintreffen einer bestimmten Bedingung, blockiert. Bei einem erfolgreichen Aufruf von pthread_cond_wait() wird auch für den Mutex automatisch die Sperre wieder eingerichtet – oder einfach, es herrscht wieder der Zustand wie vor dem pthread_cond_wait-Aufruf.
Threads, die auf die Bedingungsvariable cond warten, können Sie mit phtread_cond_signal() wieder aufwecken und weiter ausführen. Bei mehreren Threads, die auf die Bedingungsvariable cond warten, bekommt der Thread mit der höchsten Priorität den Zuschlag.
Wollen Sie hingegen alle Threads aufwecken, die auf die Bedingungsvariable cond warten, können Sie die Funktion pthread_cond_broadcast() verwenden.
Natürlich gibt es auch noch eine Funktion, womit Sie, im Gegensatz zu phtread_cond_wait(), nur eine gewisse Zeit auf die Bedingungsvariable cond warten, bevor sie zum aufrufenden Thread zurückkehrt und wieder automatisch die Sperre von Mutex einrichtet – pthread_cond_timewait(). Als Zeit können Sie hierbei abstime verwenden, womit Sie eine absolute Zeit in Sekunden und Nanosekunden angeben, die seit dem 1.1.1970 vergangen sind.
struct timespec {
time_t tv_sec; // Sekunden
long tv_nsec; // Nanosekunden
};
Hierzu ein recht einfaches Beispiel, das rein die Funktionalität von Bedingungsvariablen und vor allem deren Verwendung demonstriert.
/* thread9.c */
#include <stdio.h>
#include <pthread. h.>
#include <unistd. h.>
#include <stdlib.h>
#define THREAD_MAX 3
pthread_mutex_t mutex=PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
static void *threads (void *arg) {
printf("\t->Thread %ld wartet auf Bedingung\n",
pthread_self());
pthread_mutex_lock(&mutex);
pthread_cond_wait(&cond, &mutex);
printf("\t->Thread %ld hat Bedingung erhalten\n",
pthread_self());
printf("\t->Thread %ld: Sende wieder die "
"Bedingungsvariable\n", pthread_self());
pthread_cond_signal(&cond);
pthread_mutex_unlock(&mutex);
return NULL;
}
int main (void) {
int i;
pthread_t th[THREAD_MAX];
printf("->Main-Thread %ld gestartet\n", pthread_self());
for(i=0; i<THREAD_MAX; i++)
if (pthread_create (&th[i],NULL, &threads, NULL)!=0) {
printf ("Konnte keinen Thread erzeugen\n");
exit (EXIT_FAILURE);
}
printf("->Main-Thread: habe soeben %d Threads erzeugt\n",
THREAD_MAX);
/* Kurz ruhig legen, damit der Main-Thread als Erstes die
* Bedingungsvariable sendet */
sleep(1);
printf("->Main-Thread: Sende die Bedingungsvariable\n");
pthread_cond_signal(&cond);
for(i=0; i<THREAD_MAX; i++)
pthread_join (th[i], NULL);
printf("->Main-Thread %ld beendet\n", pthread_self());
pthread_exit(NULL);
}
Das Programm bei der Ausführung:
$ gcc -o thread9 thread9.c -lpthread
$ ./thread9
->Main-Thread -1209416608 gestartet
->Main-Thread: habe soeben 3 Threads erzeugt
->Thread -1209418832 wartet auf Bedingung
->Thread -1217811536 wartet auf Bedingung
->Thread -1226204240 wartet auf Bedingung
->Main-Thread: Sende die Bedingungsvariable
->Thread -1209418832 hat Bedingung erhalten
->Thread -1209418832: Sende wieder die Bedingungsvariable
->Thread -1217811536 hat Bedingung erhalten
->Thread -1217811536: Sende wieder die Bedingungsvariable
->Thread -1226204240 hat Bedingung erhalten
->Thread -1226204240: Sende wieder die Bedingungsvariable
->Main-Thread -1209416608 beendet
Sie sehen hierbei, dass, sobald der Haupt-Thread eine Bedingungsvariable »sendet«, eine Kettenreaktion der weiteren Threads entsteht. Hier werden die Threads, entsprechend wie sie in der Queue angelegt wurden, abgearbeitet.
Dazu ein simples Beispiel. In diesem Beispiel wartet der Thread Nummer 2 auf die Condition-Variable von Thread 1. Thread 1 weist einem globalen Zahlenarray werte zehn Werte zu, die Thread 2 anschließend berechnet. Dies ist natürlich auch wieder ein primitives Beispiel und soll nur die Funktion von Condition-Variablen demonstrieren.
/* thread10.c */
#include <stdio.h>
#include <pthread. h.>
#include <unistd. h.>
#include <stdlib.h>
static int werte[10];
pthread_mutex_t mutex=PTHREAD_MUTEX_INITIALIZER;
pthread_cond_t cond = PTHREAD_COND_INITIALIZER;
static void thread1 (void *arg) {
int ret, i;
printf ("\t->Thread %ld gestartet ...\n",
pthread_self ());
sleep (1);
ret = pthread_mutex_lock (&mutex);
if (ret != 0) {
printf ("Fehler bei lock in Thread:%ld\n",
pthread_self());
exit (EXIT_FAILURE);
}
/* Kritischer Codeabschnitt */
for (i = 0; i < 10; i++)
werte[i] = i;
/* Kritischer Codeausschnitt Ende */
printf ("\t->Thread %ld sendet Bedingungsvariable\n",
pthread_self());
pthread_cond_signal (&cond);
ret = pthread_mutex_unlock (&mutex);
if (ret != 0) {
printf ("Fehler bei unlock in Thread: %ld\n",
pthread_self ());
exit (EXIT_FAILURE);
}
printf ("\t->Thread %ld ist fertig\n",pthread_self());
pthread_exit ((void *) 0);
}
static void thread2 (void *arg) {
int i;
int summe = 0;
printf ("\t->Thread %ld wartet auf Bedingungsvariable\n",
pthread_self ());
pthread_cond_wait (&cond, &mutex);
printf ("\t->Thread %ld gestartet ...\n",
pthread_self ());
for (i = 0; i < 10; i++)
summe += werte[i];
printf ("\t->Thread %ld fertig\n",pthread_self());
printf ("Summe aller Zahlen beträgt: %d\n", summe);
pthread_exit ((void *) 0);
}
int main (void) {
pthread_t th[2];
printf("->Main-Thread %ld gestartet\n", pthread_self());
pthread_create (&th[0], NULL, thread1, NULL);
pthread_create (&th[1], NULL, thread2, NULL);
pthread_join (th[0], NULL);
pthread_join (th[1], NULL);
printf("->Main-Thread %ld beendet\n", pthread_self());
return EXIT_SUCCESS;
}
Das Programm bei der Ausführung:
$ gcc -o thread10 thread10.c -lpthread
$ ./thread10
->Main-Thread -1209416608 gestartet
->Thread -1209418832 gestartet ...
->Thread -1217811536 wartet auf Bedingungsvariable
->Thread -1209418832 sendet Bedingungsvariable
->Thread -1209418832 ist fertig
->Thread -1217811536 gestartet ...
->Thread -1217811536 fertig
Summe aller Zahlen beträgt: 45
->Main-Thread -1209416608 beendet
|
Hinweis In diesem und auch in vielen anderen Beispielen wurde das eine oder andere Mal auf eine Fehlerüberprüfung verzichtet, was man in der Praxis natürlich tunlichst vermeiden sollte. Allerdings würde ein »perfekt« geschriebenes Programm zu viele Buchseiten in Anspruch nehmen.
|
Dynamische Bedingungsvariablen
Natürlich steht Ihnen hierzu auch die Möglichkeit zur Verfügung, Bedingungsvariablen dynamisch anzulegen, wie dies häufig mit Datenstrukturen der Fall ist. Hierzu stehen Ihnen die folgenden Funktionen zur Verfügung:
#include <pthread. h.>
int pthread_cond_init( pthread_cond_t *cond,
pthread_condattr_t *cond_attr );
int pthread_cond_destroy( pthread_cond_t *cond );
Mit pthread_cond_init() initialisieren Sie die Bedingungsvariable cond mit den über attr festgelegten Attributen (hierauf wird im nächsten Abschnitt eingegangen). Verwenden Sie für attr NULL, werden die standardmäßig voreingestellten Bedingungsvariablen verwendet. Freigeben können Sie die dynamisch angelegte Bedingungsvariable cond wieder mit der Funktion pthread_cond_destroy().
Hierzu dasselbe Beispiel wie schon im Beispiel »thread10.c« zuvor, nur eben als dynamische Variante.
/* thread11.c */
#include <stdio.h>
#include <pthread. h.>
#include <unistd. h.>
#include <stdlib.h>
struct data {
int werte[10];
pthread_mutex_t mutex;
pthread_cond_t cond;
};
static void thread1 (void *arg) {
struct data *d=(struct data *)arg;
int ret, i;
printf ("\t->Thread %ld gestartet ...\n",
pthread_self ());
sleep (1);
ret = pthread_mutex_lock (&d->mutex);
if (ret != 0) {
printf ("Fehler bei lock in Thread:%ld\n",
pthread_self());
exit (EXIT_FAILURE);
}
/* Kritischer Codeabschnitt */
for (i = 0; i < 10; i++)
d->werte[i] = i;
/* Kritischer Codeausschnitt Ende */
printf ("\t->Thread %ld sendet Bedingungsvariable\n",
pthread_self());
pthread_cond_signal (&d->cond);
ret = pthread_mutex_unlock (&d->mutex);
if (ret != 0) {
printf ("Fehler bei unlock in Thread: %ld\n",
pthread_self ());
exit (EXIT_FAILURE);
}
printf ("\t->Thread %ld ist fertig\n", pthread_self());
pthread_exit ((void *) 0);
}
static void thread2 (void *arg) {
struct data *d=(struct data *)arg;
int i;
int summe = 0;
printf ("\t->Thread %ld wartet auf Bedingungsvariable\n",
pthread_self ());
pthread_cond_wait (&d->cond, &d->mutex);
printf ("\t->Thread %ld gestartet ...\n",
pthread_self ());
for (i = 0; i < 10; i++)
summe += d->werte[i];
printf ("\t->Thread %ld fertig\n",pthread_self());
printf ("Summe aller Zahlen beträgt: %d\n", summe);
pthread_exit ((void *) 0);
}
int main (void) {
pthread_t th[2];
struct data *d;
/* Speicher für die Struktur reservieren */
d = malloc(sizeof(struct data));
if(d == NULL) {
printf("Konnte keinen Speicher reservieren ...!\n");
exit(EXIT_FAILURE);
}
/* Bedingungsvariablen initialisieren */
pthread_cond_init(&d->cond, NULL);
printf("->Main-Thread %ld gestartet\n", pthread_self());
pthread_create (&th[0], NULL, thread1, d);
pthread_create (&th[1], NULL, thread2, d);
pthread_join (th[0], NULL);
pthread_join (th[1], NULL);
/* Bedingungsvariable freigeben */
pthread_cond_destroy(&d->cond);
printf("->Main-Thread %ld beendet\n", pthread_self());
return EXIT_SUCCESS;
}
Hierzu noch ein typisches Anwendungsbeispiel. Wir simulieren ein Programm, das Daten empfängt, und erzeugen dabei zwei Threads. Jeder dieser beiden Threads wird mit pthread_cond_wait() in einen Wartezustand geschickt und wartet auf das Signal pthread_cond_signal() vom Haupt-Thread. Ein einfaches Server-Client-Prinzip also. Der Haupt-Thread simuliert dann, er würde zwei Datenpakete an einen Client-Thread verschicken. Der Client-Thread simuliert anschließend, er würde die Datenpakete bearbeiten. Im Beispiel wurden statische Bedingungsvariablen verwendet. Die Ausgabe und der Ablauf des Programms sollten den Sachverhalt außerdem von selbst erklären:
/* thread12.c */
#define _MULTI_THREADED
#include <pthread. h.>
#include <stdio.h>
#include <stdlib.h>
#include <unistd. h.>
#define NUMTHREADS 2
static void checkResults (const char *string, int val) {
if (val) {
printf ("Fehler mit %d bei %s", val, string);
exit (EXIT_FAILURE);
}
}
static pthread_mutex_t dataMutex =
PTHREAD_MUTEX_INITIALIZER;
static pthread_cond_t DatenVorhandenCondition =
PTHREAD_COND_INITIALIZER;
static int DatenVorhanden = 0;
static int geteilteDaten = 0;
static void *theThread (void *parm) {
int rc;
// Datenpaket in zwei Verarbeitungsschritten
int retries = 2;
printf ("\t->Client %ld: gestartet\n", pthread_self ());
rc = pthread_mutex_lock (&dataMutex);
checkResults ("pthread_mutex_lock()\n", rc);
while (retries--) {
while (!DatenVorhanden) {
printf ("\t->Client %ld: Warte auf Daten ...\n",
pthread_self ());
rc = pthread_cond_wait ( &DatenVorhandenCondition,
&dataMutex);
if (rc) {
printf ("Client %ld: pthread_cond_wait()"
" Fehler rc=%d\n", rc, pthread_self ());
pthread_mutex_unlock (&dataMutex);
exit (EXIT_FAILURE);
}
}
printf("\t->Client %ld: Daten wurden gemeldet --->\n"
"\t----> Bearbeite die Daten, solange sie "
"geschützt sind (lock)\n", pthread_self ());
if (geteilteDaten == 0) {
DatenVorhanden = 0;
}
}//Ende while(retries--)
printf ("Client %ld: Alles erledigt\n",
pthread_self ());
rc = pthread_mutex_unlock (&dataMutex);
checkResults ("pthread_mutex_unlock()\n", rc);
return NULL;
}
int main (int argc, char **argv) {
pthread_t thread[NUMTHREADS];
int rc = 0;
// Gesamtanzahl der Datenpakete
int anzahlDaten = 4;
int i;
printf ("->Main-Thread %ld gestartet ...\n");
for (i = 0; i < NUMTHREADS; ++i) {
rc=pthread_create (&thread[i], NULL, theThread, NULL);
checkResults ("pthread_create()\n", rc);
}
/* Server-Schleife */
while (anzahlDaten--) {
sleep (3); // Eine Bremse zum "Mitverfolgen"
printf ("->Server: Daten gefunden\n");
/* Schütze geteilte (shared) Daten und Flags */
rc = pthread_mutex_lock (&dataMutex);
checkResults ("pthread_mutex_lock()\n", rc);
printf ("->Server: Sperre die Daten und gib eine "
"Meldung an Consumer\n");
++geteilteDaten; /* Füge "shared" Daten hinzu */
DatenVorhanden = 1; /* ein vorhandenes Datenpaket */
/* Client wieder aufwecken */
rc = pthread_cond_signal (&DatenVorhandenCondition);
if (rc) {
pthread_mutex_unlock (&dataMutex);
printf ("Server: Fehler beim Aufwecken von "
"Client, rc=%d\n", rc);
exit (EXIT_FAILURE);
}
printf("->Server: Gibt die gesperrten Daten"
" wieder frei\n");
rc = pthread_mutex_unlock (&dataMutex);
checkResults ("pthread_mutex_lock()\n", rc);
}//Ende while(anzahlDaten--)
for (i = 0; i < NUMTHREADS; ++i) {
rc = pthread_join (thread[i], NULL);
checkResults ("pthread_join()\n", rc);
}
printf ("->Main-Thread ist fertig\n");
return EXIT_SUCCESS;
}
Das Programm bei der Ausführung:
$ gcc -o thread12 thread12.c -lpthread
$ ./thread12
->Main-Thread -1073743916 gestartet...
->Client -1209418832: gestartet
->Client -1209418832: Warte auf Daten ...
->Client -1217811536: gestartet
->Client -1217811536: Warte auf Daten ...
->Server: Daten gefunden
->Server: Sperre die Daten und gib eine Meldung an Consumer
->Server: Gibt die gesperrten Daten wieder frei
->Client -1209418832: Daten wurden gemeldet --->
----> Bearbeite die Daten, solange sie geschützt sind (lock)
->Client -1209418832: Daten wurden gemeldet --->
----> Bearbeite die Daten, solange sie geschützt sind (lock)
Client -1209418832: Alles erledigt
->Server: Daten gefunden
->Server: Sperre die Daten und gib eine Meldung an Consumer
->Server: Gibt die gesperrten Daten wieder frei
->Client -1217811536: Daten wurden gemeldet --->
----> Bearbeite die Daten, solange sie geschützt sind (lock)
->Client -1217811536: Daten wurden gemeldet --->
----> Bearbeite die Daten, solange sie geschützt sind (lock)
Client -1217811536: Alles erledigt
->Server: Daten gefunden
->Server: Sperre die Daten und gib eine Meldung an Consumer
->Server: Gibt die gesperrten Daten wieder frei
->Server: Daten gefunden
->Server: Sperre die Daten und gib eine Meldung an Consumer
->Server: Gibt die gesperrten Daten wieder frei
->Main-Thread ist fertig
Condition-Variablen-Attribute
Für die Attribute von Bedingungsvariablen stehen Ihnen folgende Funktionen zur Verfügung:
#include <pthread. h.>
int pthread_condattr_init( pthread_condattr_t *attr );
int pthread_condattr_destroy( pthread_condattr_t *attr );
Allerdings machen diese Funktionen noch keinen Sinn, da Linux-Threads noch keine Attribute für Bedingungsvariablen anbieten. Diese Funktionen wurden dennoch implementiert, um den POSIX-Standard zu erfüllen.
10.7.3 Semaphore
Threads können auch mit Semaphoren synchronisiert werden. Wie Sie bereits aus dem Kapitel zur Interprozesskommunikation erfahren haben, sind Semaphore nichts anderes als nicht negative Zählvariablen, die man beim Eintritt in einen kritischen Bereich dekrementiert und beim Verlassen wieder inkrementiert. Hierzu stehen Ihnen folgende Funktionen zur Verfügung:
#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 );
int sem_post( sem_t * sem );
int sem_getvalue( sem_t * sem, int * sval );
int sem_destroy( sem_t * sem );
Alle Funktionen geben bei Erfolg 0 oder bei einem Fehler -1 zurück. Mit der Funktion sem_init() initialisieren Sie das Semaphor sem mit dem Anfangswert value. Geben Sie für den zweiten Parameter pshared einen Wert ungleich 0 an, kann das Semaphor gemeinsam von mehreren Prozessen und deren Threads verwendet werden, oder aber, wenn gleich 0 verwendet wird, das Semaphor kann nur »lokal« für die Threads des aktuellen Prozesses verwendet werden.
Die Funktion sem_wait() wird zum Suspendieren eines aufrufenden Threads verwendet. sem_wait() wartet so lange, bis der Zähler sem einen Wert ungleich 0 besitzt. Sobald der Wert von sem ungleich 0 ist, also z. B. um 1 inkrementiert wurde, kann der suspendierende Thread mit seiner Ausführung fortfahren. Des Weiteren dekrementiert sem_wait, wenn diese Funktion »aufgeweckt« wurde, den Zähler des Semaphors wieder um 1. Im Gegensatz zu sem_wait() blockiert sem_trywait() nicht, wenn sem gleich 0 ist, und kehrt sofort mit dem Rückgabewert -1 zurück.
Den Zähler des Semaphors sem können Sie mit der Funktion sem_post() um 1 erhöhen. Wollen Sie also einen anderen Thread, der mit sem_wait() suspendiert wurde, aufwecken, müssen Sie nur sem_post() aus einem anderen Thread aufrufen. sem_post() ist eine nicht blockierende Funktion.
Wollen Sie überprüfen, welchen Wert das Semaphor gerade hat, können Sie die Funktion sem_getvalue() verwenden. Mit der Funktion sem_destroy() löschen Sie das Semaphor sem wieder.
Das folgende Beispiel entspricht dem Listing »thread10.c«, nur dass hier anstatt Bedingungsvariablen und Mutexen eben ein Semaphor verwendet wird. Mithilfe der Semaphore lässt sich eine Synchronisation (meiner Meinung nach) erheblich einfacher realisieren.
/* thread13.c */
#include <stdio.h>
#include <pthread. h.>
#include <unistd. h.>
#include <stdlib.h>
#include <semaphore.h>
static int werte[10];
sem_t sem;
static void thread1 (void *arg) {
int ret, i, val;
printf ("\t->Thread %ld gestartet ...\n",
pthread_self ());
/* Kritischer Codeabschnitt */
for (i = 0; i < 10; i++)
werte[i] = i;
/* Kritischer Codeausschnitt Ende */
/* Semaphor um 1 inkrementieren */
sem_post(&sem);
/* Aktuellen Wert ermitteln */
sem_getvalue(&sem, &val);
printf("\t->Semaphor inkrementiert (Wert: %d)\n", val);
printf ("\t->Thread %ld ist fertig\n\n",pthread_self());
pthread_exit ((void *) 0);
}
static void thread2 (void *arg) {
int i;
int summe = 0;
/* Semaphor suspendiert, bis der Wert ungleich 0 ist */
sem_wait(&sem);
printf ("\t->Thread %ld gestartet ...\n",
pthread_self ());
for (i = 0; i < 10; i++)
summe += werte[i];
printf ("\t->Summe aller Zahlen beträgt: %d\n", summe);
printf ("\t->Thread %ld fertig\n\n",pthread_self());
pthread_exit ((void *) 0);
}
int main (void) {
pthread_t th[2];
int val;
printf("->Main-Thread %ld gestartet\n", pthread_self());
/* Semaphor initialisieren */
sem_init(&sem, 0, 0);
/* Aktuellen Wert abfragen */
sem_getvalue(&sem, &val);
printf("->Semaphor initialisiert (Wert: %d)\n\n", val);
/* Mit Absicht anders herum */
pthread_create (&th[1], NULL, thread2, NULL);
pthread_create (&th[0], NULL, thread1, NULL);
pthread_join (th[0], NULL);
pthread_join (th[1], NULL);
/* Aktuellen Wert abfragen */
sem_getvalue(&sem, &val);
printf("->Semaphor (Wert: %d)\n", val);
/* Semphor löschen */
sem_destroy(&sem);
printf("->Semaphor gelöscht\n");
printf("->Main-Thread %ld beendet\n", pthread_self());
return EXIT_SUCCESS;
}
Das Programm bei der Ausführung:
$ gcc -o thread13 thread13.c -lpthread
$ ./thread13
->Main-Thread -1209416608 gestartet
->Semaphor initialisiert (Wert: 0)
->Thread -1217811536 gestartet ...
->Thread -1209418832 gestartet ...
->Summe aller Zahlen beträgt: 45
->Thread -1209418832 fertig
->Semaphor inkrementiert (Wert: 0)
->Thread -1217811536 ist fertig
->Semaphor (Wert: 0)
->Semaphor gelöscht
->Main-Thread -1209416608 beendet
10.7.4 Weitere Synchronisationstechniken im Überblick
Neben den hier vorgestellten Synchronisationsmechanismen bietet Ihnen die phtread-Bibliothek noch drei weitere an, worauf hier allerdings nur kurz eingegangen werden soll.
|
|
RW-Locks – Mit RW-Locks (Read-Write-Locks) können Sie es einrichten, dass mehrere Threads aus einem (shared) Datenbereich lesen, aber nur ein Thread zum selben Zeitpunkt darin etwas schreiben darf (one-writer, many-reader). Alle Funktionen dazu beginnen mit dem Präfix pthread_rwlock_. |
|
|
Barrier – Als Barrier bezeichnet man einen Punkt, der als (unüberwindbare) Barriere verwendet wird, die erst überwunden werden kann, wenn eine bestimmte Anzahl von Threads an diese Barriere kommt, eben das Prinzip der hohen Mauer bei den Pfadpfindern, die man nur im Team (mit einer gewissen Anzahl von Personen) überwinden kann. Solange eine gewisse Anzahl von Threads nicht vorhanden ist, müssen eben alle Threads vor der Barriere warten. Soll z. B. ein bestimmter Thread erst ausgeführt werden, wenn viele andere Threads parallel mehrere Teilaufgaben erledigt haben, sind Barriers eine prima Synchronisationsmöglichkeit. Alle Funktionen zu den Barriers beginnen mit dem Präfix pthread_barrier_. |
|
|
Spinlocks – Spinlocks sind nur für Multiprozessorsystemen interessant. Das Prinzip ist dasselbe wie bei den Mutexen, nur dass - anders als bei den Mutexen - ein Thread, der auf einen Spinlock wartet, nicht die CPU freigibt, sondern eine so genannte »busy Loop« (Schleife) ausführt, bis der Spinlock frei ist. Dadurch bleibt ein Kontexwechsel (Contex Switch) erspart. Contex Switch: Beim Kontexwechsel wird der Thread blockiert, und alle Informationen, die für das Weiterlaufen benötigt werden, müssen gespeichert werden. Bei vielen Kontexwechseln ist dies ein Menge ersparter Zeit, die man mit Spinlocks gewinnen kann. Alle Funktionen zu den Spinlocks beginnen mit dem Präfix pthread_spin_. |
|
