Nach diesem kurzen Abstecher nun wieder zur\u00fcck zum eigentlichen Thema dieses Beitrages, denn wie auch im verlinkten Beitrag, sowie den beiden darin enthaltenen Videos erkl\u00e4rt, m\u00f6chte man in aller Regel den Setup-Aufwand f\u00fcr viele kleine Aufgaben m\u00f6glichst gering halten. Zu diesem Zweck bietet Windows sogenannte Threadpools<\/a>. Etwas \u00e4hnliches bietet Linux nicht von Haus aus; doch das l\u00e4sst sich recht einfach \u00e4ndern.<\/p>\n F\u00fcr ein eigenes Projekt unter Linux war ich n\u00e4mlich selber auf der Suche und wurde nach etwas Suchen auch bereits f\u00fcndig. Die gefundene Implementation<\/a> ist zwar funktional und minimalistisch gehalten, hat jedoch ein paar kleinere Ecken und Kanten, weshalb ich sie f\u00fcr meine Zwecke noch einmal \u00fcberarbeitet habe.<\/p>\n Da der Source somit eh einmal komplett bearbeitet wurde, nutze ich die Chance (insbesondere auf Grund des vielfachen Wunsches eines einzelnen Herren), um die Funktionsweise einmal direkt am Source des Thread Pools zu erkl\u00e4ren. Eine Vorwarnung m\u00f6chte ich aber bereits jetzt geben: Das wird etwas technisch \ud83d\ude09<\/p>\n Okay, Ihr wolltet es so! Als ersten Schritt auf dem Weg zu unserem Thread Pool schauen wir uns die ben\u00f6tigten Informationen an. Hierzu werfen wir zuerst einen Blick in den Header des Threadpools:<\/p>\n Die Struktur ist an sich nicht weiter spannend. Neben der Gr\u00f6\u00dfe und der Anzahl laufender Threads im Pool enth\u00e4lt diese Struktur noch den Status (ob der Pool aktiv ist, oder beendet werden soll), sowie die Liste aller Threads, die zu diesem Pool geh\u00f6ren. Als letztes ist hier noch ein Feld f\u00fcr den Zeitpunkt, wann dieser Pool erzeugt wurde. An sich also recht unspektakul\u00e4r.<\/p>\n Die vier gerade nicht erw\u00e4hnten Felder machen nun die gesamte Magie des Thread Pools aus: W\u00e4hrend in queue die gesamten Arbeitsauftr\u00e4ge stehen, sorgen mutex, job_posted und job_taken f\u00fcr die Synchronisation der Arbeit. Aber zur Synchronisation nachher gleich mehr, denn zuerst werfen wir einmal einen kurzen Blick auf die Auftragswarteschlange.<\/p>\n Okay, dieser Teil besteht aus 2 Teilen: Zum Einen aus einer doppelt verketteten Liste und zum anderen aus den Auftr\u00e4gen, die in dieser verwaltet werden. Die Art der Nutzung der Warteschlange sorgt daf\u00fcr, dass alle Arbeitsauftr\u00e4ge in der Liste zyklisch beschrieben werden und somit eine Art Ringpuffer entsteht. Die Kapazit\u00e4t wird hierbei von zwei Feldern geregelt: capacity_cur (aktuelle Anzahl genutzter Eintr\u00e4ge) und capacity_max (Maximal verf\u00fcgbare Queue-L\u00e4nge). Diese Werte werden beim Erzeugen des Threadpools gesetzt, bzw. dynamisch in bestimmten Grenzen angepasst, wenn die Warteschlange vergr\u00f6\u00dfert werden muss. Auch hierzu gleich mehr.<\/p>\n Denn viel spannender als eine stumpfe doppelt verkettete Liste ist deren Inhalt. In unserem Fall handelt es sich dabei um die auszuf\u00fchrenden Auftr\u00e4ge. Diese sind aus 2 Callbacks und den dazugeh\u00f6rigen Argumenten aufgebaut. Ein Callback ist dabei eine einfache Funktion, der ein Void-Pointer \u00fcbergeben werden kann: Quasi also beliebige Daten \ud83d\ude09<\/p>\n Wer sich nun wundert, warum es neben einer Job-Funktion auch immer eine Cleanup-Funktion gibt, dem sei gesagt, dass Speicher gerne aufger\u00e4umt werden m\u00f6chte. Gleiches gilt auch f\u00fcr andere Ressourcen, die man sich vom Betriebssystem holt. Und selbst wenn man das f\u00fcr seinen Arbeitsauftrag an sich tut, kann es passieren, dass dieser auf Grund des Multithreading abgebrochen wird, bevor er die angeforderten Ressourcen wieder korrekt freigeben konnte. Und genau an dieser Stelle springt einem die Cleanup-Funktion dann zur Seite, da diese automatisch ausgef\u00fchrt wird, wenn der Arbeitsauftrag abgebrochen wurde.<\/p>\n Nach dem wir nun den prinzipiellen Aufbau des Threadpools im Speicher gekl\u00e4rt haben, k\u00f6nnen wir uns dem wichtigen Teil zuwenden: Der eigentlichen Implementierung. Bei dieser beginnen wir der Einfachheit halber mit dem Erstellen des Pools, da dieses noch vergleichsweise Straight Forward ist.<\/p>\n Ich denke, weite Teile dieser Routine d\u00fcrften selbsterkl\u00e4rend sein. Einzig die letzte Schleife, in der unsere Arbeitsthreads erzeugt werden bedarf wahrscheinlich etwas Erl\u00e4uterung:<\/p>\n Mit pthread_create wird hierbei ein neuer pthread als Kind des aktuellen Threads erzeugt. Dieser w\u00fcrde normalerweise erst dann finalisiert werden, wenn unser Thread auch beendet wird. Um den Arbeitsthread unabh\u00e4ngig von unserem aktuellen Thread (ja, das Hauptprogramm ist auch nur ein Thread), auszuf\u00fchren, m\u00fcssen wir diesen abkoppeln. Dies geschieht in der letzten Zeile der Schleife.<\/p>\n Im Kontext der Create-Methode werden nun noch zwei weitere Methoden referenziert. Die erste ist _threadpool_makeQueue, der die gew\u00fcnschte Gr\u00f6\u00dfe des Thread Pools \u00fcbergeben wird und die entsprechend die oben erw\u00e4hnte, doppelt verkettete Liste initialisiert. Bliebe noch die zweite Funktion zu erw\u00e4hnen: _threadpool_dowork.<\/p>\n Wie der Name dieser Funktion bereits vermuten l\u00e4sst, erledigt sie die ganze Arbeit unseres Thread Pools. Werfen wir also einfach einen Blick drauf:<\/p>\n Auch hier ist der Ablauf an sich relativ einfach, aber wie versprochen, wollte ich ja noch mal auf das Thema Synchronisation zu sprechen kommen. Und genau diese Funktion bietet einen wunderbaren Ansatz daf\u00fcr, weil sie alles enth\u00e4lt, was es an Synchronisationsaufgaben zu erledigen gibt.<\/p>\n Das Grundproblem bei nebenl\u00e4ufigen Anwendungen ist n\u00e4mlich, wie man sich auf einen bestimmten Stand einigt, den alle Beteiligten koh\u00e4rent f\u00fcr ihre Berechnungen benutzen. Nehmen wir hierzu einmal an, auf dem Tisch vor uns steht eine bereits geschnittene Torte mit X Tortenst\u00fccken drauf. Ferner k\u00f6nnen wir f\u00fcr jedes der St\u00fccken nur entweder feststellen, ob es noch da ist, oder uns ein St\u00fcck Torte wegnehmen. Wenn wir nur einen Berechtigten haben, der die St\u00fccken der Torte verteilt, ist die Aufgabe relativ klar und die Situation unproblematisch: Derjenige guckt nach, ob ein St\u00fcck noch da ist und nimmt sich dieses St\u00fcck im n\u00e4chsten Zeitschritt einfach. Nun ist so eine Torte unter Party-G\u00e4sten sehr begehrt und das DRM beim Zugriff auf die einzelnen St\u00fccken nat\u00fcrlich nicht existent. Auch erfolgt der Zugriff auf die Torte durch jeden Gast \u00fcblicherweise gleichzeitig, denn alle wollen die leckere Torte gern essen.<\/p>\n Nehmen wir an, die Tortenst\u00fccke sind unterscheidbar (die Menge der Tortenst\u00fccke brauch jedoch nicht abz\u00e4hlbar sein ;-)), so kann es passieren, dass zwei G\u00e4ste das selbe Tortenst\u00fcck essen m\u00f6chten. Da der Zugriff der G\u00e4ste aber nicht synchronisiert erfolgt, k\u00f6nnen folgende 3 Situationen auftreten:<\/p>\n In allen 3 F\u00e4llen d\u00fcrfte der Spa\u00df der G\u00e4ste durch die unzureichende Synchronisation des Zugriffs auf die Torte getr\u00fcbt werden. Was k\u00f6nnen wir also tun? Die erste Variante w\u00e4re, nur genau einen die Tortenst\u00fccke austeilen zu lassen. Da wir dann aber nicht mehr sehr effizient die Torte verteilen k\u00f6nnten (wir nehmen einmal an, von unseren N G\u00e4sten k\u00f6nnen fast alle gleichzeitig mindestens ein Tortenst\u00fcck erreichen), weil sonst jeder sich anstellen m\u00fcsste, verwerfen wir diesen Vorschlag: Wir wollen das Verteilen der Torte ja schlie\u00dflich parallelisieren, so dass jeder schnellstm\u00f6glich sein St\u00fcck bekommt.<\/p>\n Ein Ansatz, der hier auffallen d\u00fcrfte, ist das Problem, dass wir es irgendwie schaffen m\u00fcssen, dass die G\u00e4ste koordinieren k\u00f6nnen, welche St\u00fccken bereits „vergeben“ sind, da wir dann Zugriffe auf das selbe St\u00fcck Kuchen unterbinden k\u00f6nnen. F\u00fchren wir dazu eine Operation „Nachschauen und Markieren“ ein, die auf dem St\u00fcck Torte ein F\u00e4hnchen setzt, sobald das St\u00fcck Torte von jemandem als seines markiert wird. Zus\u00e4tzlich sind die F\u00e4hnchen so beschaffen, dass immer nur eines gesetzt werden kann, aber jeder wei\u00df, wenn er sein eigenes F\u00e4hnchen korrekt gesetzt hat: Wenn Zwei ein F\u00e4hnchen versuchen zu setzen, hat einer von beiden (z.B. vom Universum ausgew\u00fcrfelt ;-)) Gl\u00fcck, w\u00e4hrend der andere eindeutig wei\u00df, dass dieses St\u00fcck nicht verf\u00fcgbar ist (unabh\u00e4ngig, davon, ob er wei\u00df, ob ein St\u00fcck Kuchen da ist oder nicht).<\/p>\n F\u00fcr die drei F\u00e4lle oben hei\u00dft das Folgendes:<\/p>\n Und schon sind alle unsere G\u00e4ste zufrieden, weil sie immer garantiert bekommen, dass sie ein St\u00fcck Torte auch bekommen, wenn sie einmal festgestellt haben, dass eines da ist, bzw. sie sich ein anderes suchen m\u00fcssen, wenn keines mehr am gew\u00fcnschten Ort liegt.<\/p>\n Was bedeutet das nun f\u00fcr unser Problem im Thread Pool? Nun, wie auch unsere G\u00e4ste mit den Tortenst\u00fccken haben unsere Threads im Thread Pool das Problem, dass auf Speicherbereiche zugegriffen werden muss. Zus\u00e4tzlich zu Speicher vorhanden oder nicht, m\u00fcssen unsere Threads jedoch auch mit Ver\u00e4nderungen zurecht kommen, um einen koh\u00e4renten Stand der Abarbeitung zu gew\u00e4hrleisten, d.h. so dass niemand zwischen dem Lesen einer Speicherstelle und deren Ver\u00e4nderung eine weitere Ver\u00e4nderung vornehmen kann.<\/p>\n Bei Threads wird diese Synchronisation der Zugriffe typischer Weise durch mehrere Techniken realisiert:<\/p>\n M\u00f6chte man, dass die Threads untereinander kooperieren, ist das Signalisieren von Status\u00e4nderungen oftmals von Bedeutung. Hierf\u00fcr kann man Events (Ereignisse, auf deren Eintreten passiv gewartet werden kann) bzw. Condition Variablen (Variablen, die \u00fcber einen Mutex synchronisiert werden) verwenden.<\/p>\n Nach dem wir nun die Grundlagen f\u00fcr die Erkl\u00e4rung der verwendeten Funktionalit\u00e4ten gelegt haben, gehen wir einfach einmal der Reihe nach durch:<\/p>\n Kaum in der Routine und wir haben es bereits mit Synchronisation zu tun \ud83d\ude1b Was diese Zeile macht ist an sich etwas ganz triviales: Wir z\u00e4hlen, wieviele Threads grad aktiv sind. Und da wir grade zum Leben erweckt wurden, teilen wir dies der Pool-Verwaltung mit. Nun machen das au\u00dfer uns noch X andere Threads, was zum oben erw\u00e4hnten Problem mit dem gleichzeitigen Zugriff auf den Speicher f\u00fchrt. Damit dies nicht passiert nutzen wir einen sogenannten atomaren Inkrement. Dies ist eine Spezielle Art eine Variable zu inkrementieren, bei der uns vom Compiler garantiert wird, dass der erzeugte Code auf der Zielplattform ohne Unterbrechung, also in einem einzigen Schritt, ausgef\u00fchrt wird. Auf diese Weise kann also immer nur ein Thread gleichzeitig sein Auferstehen mitteilen, w\u00e4hrend seine Konkurrenten warten m\u00fcssen.<\/p>\n In diesen 4 Variablen merken wir uns sp\u00e4ter, was wir derzeit zu haben. Die Job-Funktion ist dabei die regul\u00e4r auszuf\u00fchrende Funktion, w\u00e4hrend die Cleanup-Funktion nur ausgef\u00fchrt wird, wenn wir w\u00e4hrend der Ausf\u00fchrung unseres Arbeitsauftrags unterbrochen werden.<\/p>\n Und die erste Magie in unserer Routine. Diese ist n\u00f6tig, um dem Betriebssystem mitzuteilen, dass es unseren Thread jederzeit unterbrechen bzw. beenden darf. Zus\u00e4tzlich teilen wir dem Betriebssystem mit, dass wir im Falle einer Unterbrechung bitte ein Unlock des Pool-Mutex ausf\u00fchren m\u00f6chten. Diesen holen wir uns n\u00e4mlich im n\u00e4chsten Schritt und m\u00fcssen daher daf\u00fcr sorgen, dass wir im Falle einer Unterbrechung diesen freigeben, damit wir in keine Deadlocks oder Starving-Situationen laufen k\u00f6nnen.<\/p>\n Dadurch, dass wir uns den Mutex geholt haben, stellen wir sicher, dass sich die Threads im Pool in eine gewisse Ordnung beim Warten auf Arbeit begeben und keine zwei Threads gleichzeitig versuchen aus der Arbeitswarteschlange zu lesen.<\/p>\n Die Aufgabe unseres Workers ist, sich bis an sein Lebensende Aufgaben zu holen. Dazu fragen wir ab, ob in der Warteschlange Arbeit vorhanden ist. Da die Warteschlange diese Information nicht atomar liefern kann, darf immer nur ein Thread gleichzeitig nachschauen. Da wir zu diesem Zeitpunkt jedoch den Pool-Mutex (siehe vorigen Abschnitt) haben, ist dies gegeben. Da wir aber f\u00fcr den Fall, dass wir keine Arbeit finden anderen eine Chance geben m\u00fcssen, uns Arbeit zu hinterlegen, geben wir mit Hilfe einer Condition-Variable den Pool-Mutex frei. Dies friert unseren eigenen Thread und gibt gleichzeitig den Mutex f\u00fcr andere frei. Sobald jemand neue Arbeit signalisiert, wird unser Thread wieder aus seinem Schlaf erweckt und pr\u00fcft erneut die Verf\u00fcgbarkeit von Arbeit. Beim Aufwecken wird uns zudem wieder der Besitz des Mutex zugestanden, so dass wir wieder garantiert der einzige sind, der auf den Pool und seinen Inhalt zugreift.<\/p>\n An dieser Stelle \u00fcberpr\u00fcfen wir, bevor wir versuchen, wirklich etwas zu tun, ob unsere anstehende Aufgabe ist, uns selber zu beenden. Wenn dies der Fall ist, dann verlassen wir einfach unsere Arbeitsschleife und starten die Aufr\u00e4umarbeiten. Dazu aber gleich mehr. Denn aller Voraussicht wird es was zu tun geben.<\/p>\n Und da wir gewissenhafte Arbeiter sind, fragen wir einfach nach, was unser Auftrag ist und teilen unserer Verwaltung mit, dass wir uns einen Arbeitsauftrag genommen haben. Das Bescheidgeben erfolgt hierbei mit dem Gegenst\u00fcck zur weiter oben bereits gesehenen Funktion pthread_cond_wait. Die Gegenstelle werden wir an anderer Stelle noch genauer erl\u00e4utern.<\/p>\n Da wir nun unsere Arbeit haben, k\u00f6nnen wir auch andere auf die Verwaltungsstruktur zugreifen lassen, da wir bis zur Beendigung unserer Arbeit keinen weiteren Zugriff auf die Verwaltungsstruktur ben\u00f6tigen.<\/p>\n\r\n\/\/ Type to tell if threads in the pool should keep running or should terminate\r\ntypedef enum threadpool_state {\r\n ALL_RUN,\r\n ALL_EXIT\r\n} threadpool_state_t;\r\n\r\ntypedef struct threadpool {\r\n size_t size; \/\/ Number of entries in array.\r\n size_t live; \/\/ Number of live threads in pool (when\r\n \/\/ pool is being destroyed, live<=size)\r\n threadpool_state_t state; \/\/ Threads check this before getting job.\r\n pthread_t * array; \/\/ The threads themselves.\r\n\r\n threadpool_queue_head_t * queue; \/\/ queue of work orders\r\n\r\n pthread_mutex_t mutex; \/\/ protects all vars declared below.\r\n pthread_cond_t job_posted; \/\/ dispatcher: \"Hey guys, there's a job!\"\r\n pthread_cond_t job_taken; \/\/ a worker: \"Got it!\"\r\n\r\n struct timeval created; \/\/ When the threadpool was created.\r\n} threadpool_t;\r\n<\/pre>\n\r\n\/\/ \"dispatch_func_t\" declares a typed function pointer.\r\n\/\/ A variable of type \"dispatch_func_t\" points to a \r\n\/\/ function with the following signature:\r\n\/\/ void dispatch_function(void *arg);\r\ntypedef void (* dispatch_func_t)(void *);\r\n\r\ntypedef struct threadpool_queue_node {\r\n struct threadpool_queue_node * next;\r\n struct threadpool_queue_node * prev;\r\n\r\n dispatch_func_t job_func;\r\n void * job_arg;\r\n dispatch_func_t cleanup_func;\r\n void * cleanup_arg;\r\n} threadpool_queue_node_t;\r\n\r\ntypedef struct threadpool_queue_head {\r\n threadpool_queue_node_t * head;\r\n threadpool_queue_node_t * tail;\r\n\r\n threadpool_queue_node_t * freeHead;\r\n threadpool_queue_node_t * freeTail;\r\n\r\n size_t capacity_cur;\r\n size_t capacity_max;\r\n} threadpool_queue_head_t;\r\n<\/pre>\n\r\n\/*\r\n * Create a thread pool.\r\n *\/\r\nthreadpool_t *threadpool_create(int poolsize) {\r\n threadpool_t *pool; \/\/ pool we create and hand back\r\n int i; \/\/ work var\r\n\r\n \/\/ sanity check the argument\r\n if ((poolsize <= 0) || (poolsize > THREADPOOL_MAX_SIZE)) {\r\n return NULL;\r\n }\r\n\r\n \/\/ create the threadpool_t struct\r\n pool = (threadpool_t *)malloc(sizeof(threadpool_t));\r\n if (pool == NULL) {\r\n fprintf(stderr, \"\\n\\nOut of memory creating a new threadpool!\\n\");\r\n return NULL;\r\n }\r\n\r\n \/\/ initialize everything but the array and live thread count\r\n gettimeofday(&pool->created, NULL);\r\n pool->size = poolsize;\r\n pool->state = ALL_RUN;\r\n pthread_mutex_init(&(pool->mutex), NULL);\r\n pthread_cond_init(&(pool->job_posted), NULL);\r\n pthread_cond_init(&(pool->job_taken), NULL);\r\n\r\n \/\/ create the array of threads within the pool\r\n pool->array = (pthread_t *) malloc(pool->size * sizeof(pthread_t));\r\n if (!pool->array) {\r\n fprintf(stderr, \"\\n\\nOut of memory allocating thread array!\\n\");\r\n free(pool);\r\n pool = NULL;\r\n return NULL;\r\n }\r\n\r\n pool->queue = _threadpool_makeQueue(poolsize);\r\n if (!pool->queue) {\r\n fprintf(stderr, \"\\n\\nOut of memory allocating task queue!\\n\");\r\n free(pool->array);\r\n free(pool);\r\n pool = NULL;\r\n return NULL;\r\n }\r\n\r\n \/\/ bring each thread to life (update counters in loop so threads can\r\n \/\/ access pool->live to find out their ID#\r\n for (i = 0; i < pool->size; ++i) {\r\n if (0 != pthread_create(pool->array + i, NULL, (void *(*)(void *))_threadpool_dowork, (void *) pool)) {\r\n perror(\"\\n\\nThread creation failed:\");\r\n \/\/ TODO: Do proper finalization!\r\n exit(EXIT_FAILURE);\r\n return NULL;\r\n }\r\n\r\n \/\/ automatic cleanup when thread exits.\r\n pthread_detach(pool->array[i]);\r\n }\r\n\r\n return pool;\r\n}\r\n<\/pre>\n\r\n \/\/ bring each thread to life (update counters in loop so threads can\r\n \/\/ access pool->live to find out their ID#\r\n for (i = 0; i < pool->size; ++i) {\r\n if (0 != pthread_create(pool->array + i, NULL, (void *(*)(void *))_threadpool_dowork, (void *) pool)) {\r\n perror(\"\\n\\nThread creation failed:\");\r\n \/\/ TODO: Do proper finalization!\r\n exit(EXIT_FAILURE);\r\n return NULL;\r\n }\r\n\r\n \/\/ automatic cleanup when thread exits.\r\n pthread_detach(pool->array[i]);\r\n }\r\n<\/pre>\n\r\n\/*\r\n * Define the life of a working thread.\r\n *\/\r\nvoid *_threadpool_dowork(threadpool_t *pool) {\r\n \/\/ Remember my creation sequence number\r\n\/\/ int myid =\r\n __sync_add_and_fetch(&pool->live, 1);\r\n\r\n \/\/ When we get a posted job, we copy it into these local vars.\r\n dispatch_func_t job_func;\r\n void * job_arg;\r\n\r\n dispatch_func_t cleanup_func;\r\n void * cleanup_arg;\r\n\r\n pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS, NULL);\r\n pthread_cleanup_push((dispatch_func_t)pthread_mutex_unlock, (void *) &pool->mutex);\r\n\r\n \/\/ Grab mutex so we can begin waiting for a job\r\n if (0 != pthread_mutex_lock(&pool->mutex)) {\r\n perror(\"\\nMutex lock failed!:\");\r\n exit(EXIT_FAILURE);\r\n }\r\n\r\n \/\/ Main loop: wait for job posting, do job(s) ... forever\r\n while(1) {\r\n\r\n while(!_threadpool_jobAvailable(pool->queue)) {\r\n pthread_cond_wait(&pool->job_posted, &pool->mutex);\r\n }\r\n\r\n \/\/ We've just woken up and we have the mutex. Check pool's state\r\n if (ALL_EXIT == pool->state) {\r\n break;\r\n }\r\n\r\n \/\/ while we find work to do\r\n _threadpool_getWorkOrder(pool->queue, &job_func, &job_arg, &cleanup_func, &cleanup_arg);\r\n pthread_cond_signal(&pool->job_taken);\r\n\r\n \/\/ Yield mutex so other jobs can be posted\r\n if (0 != pthread_mutex_unlock(&pool->mutex)) {\r\n perror(\"\\n\\nMutex unlock failed!:\");\r\n exit(EXIT_FAILURE);\r\n }\r\n\r\n \/\/ Run the job we've taken\r\n if(cleanup_func) {\r\n pthread_cleanup_push(cleanup_func, cleanup_arg);\r\n }\r\n\r\n \/\/GCC fails if I DON'T write a while loop here. Let's make it happy!\r\n while(0);\r\n\r\n job_func(job_arg);\r\n\r\n if(cleanup_func) {\r\n pthread_cleanup_pop(1);\r\n }\r\n\r\n \/\/ Grab mutex so we can grab posted job, or (if no job is posted)\r\n \/\/ begin waiting for next posting.\r\n if (0 != pthread_mutex_lock(&pool->mutex)) {\r\n perror(\"\\n\\nMutex lock failed!:\");\r\n exit(EXIT_FAILURE);\r\n }\r\n }\r\n\r\n \/\/ If we get here, we broke from loop because state is ALL_EXIT.\r\n __sync_sub_and_fetch(&pool->live, 1);\r\n\r\n \/\/ We're not really taking a job ... but this signals the destroyer\r\n \/\/ that one thread has exited, so it can keep on destroying.\r\n pthread_cond_signal(&pool->job_taken);\r\n\r\n if (0 != pthread_mutex_unlock(&pool->mutex)) {\r\n perror(\"\\n\\nMutex unlock failed!:\");\r\n exit(EXIT_FAILURE);\r\n }\r\n\r\n pthread_cleanup_pop(1);\r\n return NULL;\r\n}\r\n<\/pre>\n\n
\n
\n
\r\n\/*\r\n * Define the life of a working thread.\r\n *\/\r\nvoid *_threadpool_dowork(threadpool_t *pool) {\r\n \/\/ Remember my creation sequence number\r\n\/\/ int myid =\r\n __sync_add_and_fetch(&pool->live, 1);\r\n<\/pre>\n\r\n \/\/ When we get a posted job, we copy it into these local vars.\r\n dispatch_func_t job_func;\r\n void * job_arg;\r\n\r\n dispatch_func_t cleanup_func;\r\n void * cleanup_arg;\r\n<\/pre>\n
\r\n pthread_setcanceltype(PTHREAD_CANCEL_ASYNCHRONOUS, NULL);\r\n pthread_cleanup_push((dispatch_func_t)pthread_mutex_unlock, (void *) &pool->mutex);\r\n<\/pre>\n
\r\n \/\/ Grab mutex so we can begin waiting for a job\r\n if (0 != pthread_mutex_lock(&pool->mutex)) {\r\n perror(\"\\nMutex lock failed!:\");\r\n exit(EXIT_FAILURE);\r\n }\r\n<\/pre>\n\r\n \/\/ Main loop: wait for job posting, do job(s) ... forever\r\n while(1) {\r\n\r\n while(!_threadpool_jobAvailable(pool->queue)) {\r\n pthread_cond_wait(&pool->job_posted, &pool->mutex);\r\n }\r\n<\/pre>\n\r\n \/\/ We've just woken up and we have the mutex. Check pool's state\r\n if (ALL_EXIT == pool->state) {\r\n break;\r\n }\r\n<\/pre>\n\r\n \/\/ while we find work to do\r\n _threadpool_getWorkOrder(pool->queue, &job_func, &job_arg, &cleanup_func, &cleanup_arg);\r\n pthread_cond_signal(&pool->job_taken);\r\n<\/pre>\n
\r\n \/\/ Yield mutex so other jobs can be posted\r\n if (0 != pthread_mutex_unlock(&pool->mutex)) {\r\n perror(\"\\n\\nMutex unlock failed!:\");\r\n exit(EXIT_FAILURE);\r\n }\r\n<\/pre>\n\r\n \/\/ Run the job we've taken\r\n if(cleanup_func) {\r\n pthread_cleanup_push(cleanup_func, cleanup_arg);\r\n }\r\n\r\n job_func(job_arg);\r\n\r\n if(cleanup_func) {\r\n pthread_cleanup_pop(1);\r\n }\r\n<\/pre>\n
![\"Flattr](\"http:\/\/blog.benny-baumann.de\/wp-content\/plugins\/flattr\/img\/flattr-badge-large.png\"){kind=icon}
<\/a><\/p>","protected":false},"excerpt":{"rendered":"