Starting threadpool

9 files changed, 527 insertions, 0 deletions

diff --git a/04_exercise/CMakeLists.txt b/04_exercise/CMakeLists.txt
new file mode 100644
index 0000000..0c521a0
--- /dev/null
+++ b/04_exercise/CMakeLists.txt
@@ -0,0 +1,6 @@
+add_executable(quicksort quicksort.c)
+
+target_link_libraries(quicksort PRIVATE threadpool)
+
+add_library(threadpool threadpool.c)
+
diff --git a/04_exercise/Makefile b/04_exercise/Makefile
new file mode 100644
index 0000000..064ec4b
--- /dev/null
+++ b/04_exercise/Makefile
@@ -0,0 +1,34 @@
+#!/usr/bin/make
+.SUFFIXES:
+.PHONY: all run pack clean
+
+SRC = $(wildcard *.c)
+OBJ = $(SRC:%.c=%.o)
+TAR = threadpool
+PCK = lab-4.zip
+
+CFLAGS = -std=gnu11 -c -g -Os -Wall -MMD -MP
+LFLAGS = -pthread
+
+DEP = $(OBJ:%.o=%.d)
+-include $(DEP)
+
+%.o: %.c
+	$(CC) $(CFLAGS) $< -o $@
+
+$(TAR): $(filter-out quicksort.o,$(OBJ))
+	$(CC) $(LFLAGS) -o $@ $^
+
+all: $(TAR)
+
+bench: $(filter-out main.o,$(OBJ))
+	$(CC) $(LFLAGS) -o $@ $^
+
+run: all
+	./$(TAR)
+
+pack: clean
+	zip $(PCK) $(SRC) $(wildcard *.h) $(wildcard *.pdf) $(wildcard *.txt) Makefile
+
+clean:
+	$(RM) $(RMFILES) $(OBJ) $(TAR) bench $(DEP) $(PCK)
diff --git a/04_exercise/array.c b/04_exercise/array.c
new file mode 100644
index 0000000..4f08366
--- /dev/null
+++ b/04_exercise/array.c
@@ -0,0 +1,59 @@
+/***************************************************************************//**
+ * @file array.c
+ * @author Dorian Weber
+ * @brief Implementation des generalisierten Arrays.
+ ******************************************************************************/
+
+#include "array.h"
+#include <stdio.h>
+#include <stdlib.h>
+#include <assert.h>
+
+/* ********************************************************* public functions */
+
+/* (die runden Klammern um einige Funktionsnamen sind notwendig, da Makros
+ * gleichen Namens existieren und der Präprozessor diese expandieren würde) */
+
+void* (arrayInit)(size_t capacity, size_t size) {
+	ArrayHdr *hdr = malloc(sizeof(*hdr) + size*capacity);
+	
+	if (hdr == NULL)
+		return NULL;
+	
+	hdr->len = 0;
+	hdr->cap = capacity;
+	
+	return hdr + 1;
+}
+
+void arrayRelease(void* self) {
+	free(((ArrayHdr*) self) - 1);
+}
+
+void* (arrayPush)(void* self, size_t size) {
+	ArrayHdr *hdr = ((ArrayHdr*) self) - 1;
+	
+	if (hdr->len == hdr->cap) {
+		hdr->cap *= 2;
+		hdr = realloc(hdr, sizeof(*hdr) + size*hdr->cap);
+		
+		if (hdr == NULL) {
+			fputs("program ran out of heap memory\n", stderr);
+			exit(-1);
+		}
+	}
+	
+	++hdr->len;
+	return hdr + 1;
+}
+
+void (arrayPop)(void* self) {
+	ArrayHdr *hdr = ((ArrayHdr*) self) - 1;
+	assert(hdr->len > 0);
+	--hdr->len;
+}
+
+/* Symbol für die Inline-Funktionen erzeugen und aus diesem Modul exportieren */
+extern void arrayClear(void* self);
+extern int arrayIsEmpty(const void* self);
+extern size_t arrayLen(const void* self);
diff --git a/04_exercise/array.h b/04_exercise/array.h
new file mode 100644
index 0000000..abf2dbf
--- /dev/null
+++ b/04_exercise/array.h
@@ -0,0 +1,142 @@
+/***************************************************************************//**
+ * @file array.h
+ * @author Dorian Weber
+ * @brief Generischer Array-Typ.
+ * 
+ * @details
+ * Diese Datei enthält die Schnittstelle eines generischen Array-Typs. Um den
+ * Element-Typ des Arrays festzulegen, deklariert man eine Variable als Zeiger
+ * auf den Element-Typ. Diese kann danach initialisiert und so benutzt werden,
+ * als wäre sie ein Zeiger auf ein Array variabler Länge.
+ * 
+ * Hier ist ein Nutzungsbeispiel:
+ * @code
+ * int *array;
+ * 
+ * arrayInit(array);
+ * arrayPush(array) = 1;
+ * arrayPush(array) = 2;
+ * arrayPush(array) = 3;
+ * 
+ * while (!arrayIsEmpty(array))
+ * 	printf("%i\n", arrayPop(array));
+ * 
+ * arrayRelease(array);
+ * @endcode
+ * 
+ * Viele der genutzten Funktionen sind in Form von Makros implementiert, die die
+ * Variable, die den Zeiger auf das Array hält, aktualisieren, obwohl es so
+ * aussieht, als würde sich dieser Wert niemals ändern. Das macht es riskant
+ * mehr als einen Zeiger auf das Array zu halten, da jede Vergrößerung des
+ * Arrays alle Zeiger auf und in das Array potentiell invalidiert. Obwohl diese
+ * Fehlerquelle subtil und äußerst schwer diagnostizierbar ist - es ist eine
+ * extrem einfache Abstraktion - führt diese Datenstruktur meiner Meinung nach
+ * trotzdem zu einem massiven Produktivitätsgewinn; man sollte sich nur in etwa
+ * im Klaren über die unterliegende Implementation sein, um genau die Untermenge
+ * von Programmiertechniken auszuwählen, die korrekt funktioniert.
+ ******************************************************************************/
+
+#ifndef ARRAY_H_INCLUDED
+#define ARRAY_H_INCLUDED
+
+/* *** includes ************************************************************* */
+
+#include <stddef.h>
+
+/* *** structures *********************************************************** */
+
+/**@brief Arrayheader.
+ * 
+ * Diese Struktur wird jedem Array im Speicher vorangestellt und beinhaltet
+ * Informationen über Kapazität und aktuelle Auslastung des Arrays. Die
+ * Arrayelemente schließen sich dieser Struktur unmittelbar an, so dass der
+ * Nutzer von dieser versteckten Information nichts bemerkt.
+ */
+typedef struct ArrayHdr
+{
+	size_t len; /**<@brief Anzahl der Array-Elemente. */
+	size_t cap; /**<@brief Kapazität des reservierten Speichers. */
+} ArrayHdr;
+
+/* *** interface ************************************************************ */
+
+/**@internal
+ * @brief Initialisiert und gibt einen Zeiger auf den Start des Arrays zurück.
+ * @param capacity  initiale Kapazität
+ * @param size      Größe der Arrayelemente
+ * @return ein Zeiger auf den Start des Arrays, falls erfolgreich,\n
+ *      \c NULL im Falle eines Speicherfehlers
+ */
+extern void* arrayInit(size_t capacity, size_t size);
+
+/**@brief Initialisiert ein neues Array.
+ * @param self  das Array
+ * @return 0, falls keine Fehler bei der Initialisierung aufgetreten sind,\n
+ *        -1 ansonsten
+ */
+#define arrayInit(self) \
+	((self = arrayInit(8, sizeof((self)[0]))) == NULL ? -1 : 0)
+
+/**@brief Gibt das Array und alle assoziierten Strukturen frei.
+ * @param self  das Array
+ */
+extern void arrayRelease(void* self);
+
+/**@internal
+ * @brief Reserviert Platz für einen neuen Wert im Array.
+ * @param self  das Array
+ * @param size  Größe der Arrayelemente
+ * @return der neue Zeiger auf den Start des Arrays
+ */
+extern void* arrayPush(void* self, size_t size);
+
+/**@brief Legt einen Wert auf das Array.
+ * @param self  das Array
+ */
+#define arrayPush(self) \
+	(self = arrayPush(self, sizeof((self)[0])), (self)+arrayLen(self)-1)[0]
+
+/**@brief Entfernt das oberste Element des Arrays.
+ * @param self  das Array
+ */
+extern void arrayPop(void* self);
+
+/**@brief Entfernt und liefert das oberste Element des Arrays.
+ * @param self  das Array
+ * @return das oberste Element von \p self
+ */
+#define arrayPop(self) \
+	(arrayPop(self), (self)+arrayLen(self))[0]
+
+/**@brief Gibt das oberste Element des Arrays zurück.
+ * @param self  das Array
+ * @return das oberste Element von \p self
+ */
+#define arrayTop(self) \
+	(self)[arrayLen(self) - 1]
+
+/**@brief Setzt die Länge des Arrays auf 0 zurück.
+ * @param self  das Array
+ */
+inline void arrayClear(void* self) {
+	((ArrayHdr*) self)[-1].len = 0;
+}
+
+/**@brief Gibt zurück, ob das Array leer ist.
+ * @param self  das Array
+ * @return 0, falls nicht leer\n
+           1, falls leer
+ */
+inline int arrayIsEmpty(const void* self) {
+	return ((ArrayHdr*) self)[-1].len == 0;
+}
+
+/**@brief Gibt die Anzahl der Array-Elemente zurück.
+ * @param self  das Array
+ * @return Anzahl der Elemente des Arrays
+ */
+inline size_t arrayLen(const void* self) {
+	return ((ArrayHdr*) self)[-1].len;
+}
+
+#endif /* ARRAY_H_INCLUDED */
diff --git a/04_exercise/main.c b/04_exercise/main.c
new file mode 100644
index 0000000..cdcaa42
--- /dev/null
+++ b/04_exercise/main.c
@@ -0,0 +1,27 @@
+#include "threadpool.h"
+
+#include <stdio.h>
+#include <stdlib.h>
+
+static TASK(long, fib, long);
+
+long fib(long n) {
+	if (n <= 1)
+		return n;
+	
+	fib_fut *a = fibAsync((fib_fut[]) { fibFuture(n - 1) });
+	fib_fut *b = fibAsync((fib_fut[]) { fibFuture(n - 2) });
+	
+	return fibAwait(a) + fibAwait(b);
+}
+
+int main() {
+	if (tpInit(8) != 0)
+		perror("Thread Pool initialization failed"), exit(-1);
+	atexit(&tpRelease);
+	
+	for (long i = 0; i <= 20; ++i)
+		printf("fib(%2li) = %li\n", i, fib(i));
+	
+	return 0;
+}
diff --git a/04_exercise/quicksort.c b/04_exercise/quicksort.c
new file mode 100644
index 0000000..88a47a2
--- /dev/null
+++ b/04_exercise/quicksort.c
@@ -0,0 +1,119 @@
+#include "threadpool.h"
+
+#include <stdio.h>
+#include <stdlib.h>
+#include <time.h>
+
+static size_t partition(int v[], size_t len);
+
+/* TODO: quicksort-Implementation parallelisieren */
+
+static void quicksort(int list[], size_t len) {
+	if (len <= 1) return;
+	
+	size_t mid = partition(list, len);
+	quicksort(list, mid);
+	quicksort(list + mid + 1, len - mid - 1);
+}
+
+size_t partition(int v[], size_t len) {
+	size_t pivot = len - 1;
+	size_t l = 0, r = len - 2;
+	int t;
+	
+	while (1) {
+		while (l < r && v[l] <  v[pivot]) ++l;
+		while (l < r && v[r] >= v[pivot]) --r;
+		
+		if (l >= r) break;
+		
+		t = v[l];
+		v[l] = v[r];
+		v[r] = t;
+	}
+	
+	if (v[l] >= v[pivot]) {
+		t = v[l];
+		v[l] = v[pivot];
+		v[pivot] = t;
+		return l;
+	}
+	
+	return pivot;
+}
+
+static void list_randomize(int list[], size_t len) {
+	for (int *p = list, *e = p + len; p < e; ++p)
+		*p = ((((double) rand()) / RAND_MAX) - 0.5) * len;
+}
+
+static void list_print(int list[], size_t len) {
+	const char *sep = "";
+	
+	fputs("(", stdout);
+	for (int *p = list, *e = p + len; p < e; ++p, sep = ", ")
+		printf("%s%i", sep, *p);
+	fputs(")\n", stdout);
+}
+
+static int list_is_sorted(int list[], size_t len) {
+	for (int *p = list, *e = p + len - 1; p < e; ++p) {
+		if (p[0] > p[1])
+			return 0;
+	}
+	
+	return 1;
+}
+
+static struct timespec time_diff(struct timespec t1, struct timespec t2) {
+	struct timespec result = {
+		.tv_sec  = t2.tv_sec - t1.tv_sec,
+		.tv_nsec = t2.tv_nsec - t1.tv_nsec
+	};
+	
+	if (t2.tv_nsec - t1.tv_nsec < 0) {
+		result.tv_sec  = t2.tv_sec - t1.tv_sec - 1;
+		result.tv_nsec = 1000000000l + t2.tv_nsec - t1.tv_nsec;
+	}
+	
+	return result;
+}
+
+int main(int argc, const char *argv[]) {
+	if (tpInit(8) != 0)
+		perror("Thread Pool initialization failed"), exit(-1);
+	atexit(&tpRelease);
+	
+	/* read the length of the list from the command line */
+	size_t list_len = 30000000;
+	if (argc >= 2) {
+		long arg = atol(argv[1]);
+		if (arg > 0)
+			list_len = arg;
+	}
+	
+	/* generate a list with random integers */
+	int *list = malloc(sizeof(int) * list_len);
+	list_randomize(list, list_len);
+	printf("Len: %lu\n", list_len);
+	
+	/* print the unsorted list if it's not too long */
+	printf("---- Unsortiert: %i ----\n", list_is_sorted(list, list_len));
+	
+	if (list_len <= 100)
+		list_print(list, list_len);
+	
+	/* sort it and print the sorted list if it's not too long */
+	struct timespec start, end, dt;
+	clock_gettime(CLOCK_PROCESS_CPUTIME_ID, &start);
+	quicksort(list, list_len);
+	clock_gettime(CLOCK_PROCESS_CPUTIME_ID, &end);
+	dt = time_diff(start, end);
+	printf("---- Sortiert:   %i ----\n", list_is_sorted(list, list_len));
+	
+	if (list_len <= 100)
+		list_print(list, list_len);
+	
+	printf("Dauer: %lis %lins\n", dt.tv_sec, dt.tv_nsec);
+	return 0;
+}
diff --git a/04_exercise/threadpool.c b/04_exercise/threadpool.c
new file mode 100644
index 0000000..3172147
--- /dev/null
+++ b/04_exercise/threadpool.c
@@ -0,0 +1,23 @@
+#include "threadpool.h"
+
+#include <stdlib.h>
+#include <unistd.h>
+#include <pthread.h>
+
+typedef struct ThreadPool {
+	/* TODO: benötigte Attribute hinzufügen */
+} ThreadPool;
+
+/* TODO: interne, globale Variablen hinzufügen */
+
+/* TODO: interne Hilfsfunktionen hinzufügen */
+
+int tpInit(size_t size) {
+	return 0;
+}
+
+void tpRelease(void) {}
+
+void tpAsync(Future *future) {}
+
+void tpAwait(Future *future) {}
diff --git a/04_exercise/threadpool.h b/04_exercise/threadpool.h
new file mode 100644
index 0000000..6b758cd
--- /dev/null
+++ b/04_exercise/threadpool.h
@@ -0,0 +1,116 @@
+#ifndef THREADPOOL_H_INCLUDED
+#define THREADPOOL_H_INCLUDED
+
+#include <stddef.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*);
+
+/**@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 */
+} 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 },             \
+			.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;           \
+	}
+
+#endif
diff --git a/CMakeLists.txt b/CMakeLists.txt
index 41de2c8..cc780b0 100644
--- a/CMakeLists.txt
+++ b/CMakeLists.txt
@@ -11,6 +11,7 @@ set(CPACK_PROJECT_VERSION ${PROJECT_VERSION})
 include(CPack)
 
 add_subdirectory(02_exercise)
+add_subdirectory(04_exercise)
 
 set(CLANG_WARNINGS
         -Wall