1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
|
#ifndef THREADPOOL_H_INCLUDED
#define THREADPOOL_H_INCLUDED
#include <stdatomic.h>
#include <stddef.h>
#include <ppmlib.h>
/**@brief Funktionszeiger auf eine asynchron auszuführende Funktion.
*
* Der Parameter kann zur Übergabe von Argumenten und den Rückgabewerten
* der Funktion genutzt werden.
*/
typedef void (*ThreadTask_f)(void*);
typedef atomic_char FutureStatus;
enum FurutureStatusEnum { FUT_WAITING = 0, FUT_IN_PROGRESS=1, FUT_DONE = 2};
/**@brief Handle zum zukünftigen Rückgabewert eines asynchronen Funktionsrufes.
*/
typedef struct Future {
ThreadTask_f fn; ///<@brief Zeiger auf die auszuführende Funktion.
/* TODO: benötigte Attribute hinzufügen */
FutureStatus status;
} Future;
/**@brief Initialisiert den globalen Thread-Pool.
*
* Diese Funktion legt \p size viele Worker-Threads an und initialisiert den
* restlichen Zustand des Thread-Pools. Danach werden keine weiteren Threads
* mehr durch den Thread-Pool angelegt.
*
* @param size Anzahl der Worker-Threads
*/
extern int tpInit(size_t size);
/**@brief Beendet alle Worker-Threads im Thread-Pool kontrolliert und gibt
* sämtliche Systemressourcen des Thread-Pools frei.
*
* Bereits gestartete Futures können abgebrochen oder fertiggestellt werden.
*/
extern void tpRelease(void);
/**@brief Startet eine nutzerdefinierte Future.
*
* Die Funktion initialisiert die übergebene Future und fügt sie zur Liste
* der abzuarbeitenden Tasks hinzu. Die Abarbeitungsreihenfolge ist
* unspezifiziert und die Abarbeitung startet sofort (eager).
*
* @param future partiell-initialisiertes Future-Objekt
* @note Der Rufer ist dafür verantwortlich, dass der Speicher nicht vor
* der Auswertung des asynchronen Funktionsrufes freigegeben wird.
*/
extern void tpAsync(Future *future);
/**@brief Erwartet die Abarbeitung einer Future.
*
* Nachdem diese Funktion zurückgekommen ist, ist der asynchrone Funktionsruf
* garantiert abgeschlossen und der Zugriff auf die Future Data-Race-frei.
* Die Funktion gibt auch alle Systemressourcen des Thread-Pools frei.
*
* @param future asynchron gestartetes Future-Objekt
* @note Der Rufer kann danach den Speicher für die Future wieder freigeben.
*/
extern void tpAwait(Future *future);
/**@brief Erzeugt Code für eine spezialisierte Schnittstelle einer Funktion.
*
* Die direkte Nutzung der minimalistischen Schnittstelle des Thread-Pools ist
* unkomfortabel und fehleranfällig, daher ist hier ein Makro definiert,
* mithilfe dessen eine spezifische Schnittstelle für asynchrone Funktionen
* erzeugt werden kann.
*
* Das Makro muss im Filescope expandiert werden und erzeugt folgende Elemente:
* 1. eine spezifische Datenstruktur für Argumente und Rückgabewerte der
* konkreten Funktion mit dem Namen `func_fut`
* 2. eine sogenannte thunk-Funktion mit dem Namen `funcThunk`, welche die
* Funktionsargumente aus dem generischen Argument entpackt, die konkrete
* Funktion ausführt und dann den Rückgabewert zurückschreibt
* 3. eine asynchrone Ruf-Funktion mit dem Namen `funcAsync`, welche ein
* Future-Objekt initialisiert und zurückgibt
* 4. eine synchrone Warte-Funktion mit dem Namen `funcAwait`, welche auf die
* Abarbeitung eines zuvor asynchron gerufenen Future-Objektes wartet und
* das Ergebnis zurückgibt
*
* Zusätzlich dazu wird die konkrete Funktion anfangs deklariert.
* Benutzung ist wie folgt möglich:
* @code
* // deklariert die Schnittstelle einer Funktion ´long fib(long)´
* TASK(long, fib, long);
* @endcode
*
* @param TYPE Rückgabetyp der Funktion
* @param NAME Name der Funktion
* @param ARG Parametertyp der Funktion
*/
#define TASK(TYPE, NAME, ARG) \
TYPE NAME(ARG); \
\
typedef struct { \
Future fut; \
ARG arg; \
TYPE res; \
} NAME ## _fut; \
\
static void NAME ## Thunk(void *args) { \
NAME ## _fut *data = args; \
data->res = NAME(data->arg); \
} \
static inline NAME ## _fut NAME ## Future(ARG arg) { \
return (NAME ## _fut) { \
.fut = { .fn = &NAME ## Thunk, .status=FUT_WAITING}, \
.arg = arg \
}; \
} \
static inline NAME ## _fut* NAME ## Async(NAME ## _fut *future) { \
return tpAsync(&future->fut), future; \
} \
static inline TYPE NAME ## Await(NAME ## _fut *future) { \
return tpAwait(&future->fut), future->res; \
}
#define TASK_QS(TYPE, NAME) \
TYPE NAME(int *arg1, size_t arg2); \
\
typedef struct { \
Future fut; \
int* arg1; \
size_t arg2; \
} NAME ## _fut; \
\
static void NAME ## Thunk(void *args) { \
NAME ## _fut *data = args; \
NAME(data->arg1, data->arg2); \
} \
static inline NAME ## _fut NAME ## Future(int *arg1, size_t arg2) { \
return (NAME ## _fut) { \
.fut = { .fn = &NAME ## Thunk, .status=FUT_WAITING}, \
.arg1 = arg1, \
.arg2 = arg2 \
}; \
} \
static inline NAME ## _fut* NAME ## Async(NAME ## _fut *future) { \
return tpAsync(&future->fut), future; \
} \
static inline TYPE NAME ## Await(NAME ## _fut *future) { \
tpAwait(&future->fut); \
}
#endif
|
