implement thread pools

5 files changed, 170 insertions, 0 deletions

diff --git a/src/threading/CMakeLists.txt b/src/threading/CMakeLists.txt
new file mode 100644
index 0000000..b2bd203
--- /dev/null
+++ b/src/threading/CMakeLists.txt
@@ -0,0 +1,17 @@
+set(THREADS_PREFER_PTHREAD_FLAG ON)
+find_package(Threads REQUIRED)
+
+list(APPEND
+	THREADING_SOURCES
+    ${CMAKE_CURRENT_SOURCE_DIR}/cpu.c
+    ${CMAKE_CURRENT_SOURCE_DIR}/cpu.h
+    ${CMAKE_CURRENT_SOURCE_DIR}/pool.c
+    ${CMAKE_CURRENT_SOURCE_DIR}/pool.h
+)
+
+
+add_library(threading STATIC ${THREADING_SOURCES})
+target_compile_options(threading PUBLIC ${CFLAGS})
+target_link_libraries(threading PRIVATE Threads::Threads)
+
+target_include_directories(threading PUBLIC ${CMAKE_CURRENT_SOURCE_DIR})
\ No newline at end of file
diff --git a/src/threading/cpu.c b/src/threading/cpu.c
new file mode 100644
index 0000000..45671df
--- /dev/null
+++ b/src/threading/cpu.c
@@ -0,0 +1,6 @@
+#include <sys/sysinfo.h>
+
+int get_core_count()
+{
+	return get_nprocs();
+}
diff --git a/src/threading/cpu.h b/src/threading/cpu.h
new file mode 100644
index 0000000..98fc1d7
--- /dev/null
+++ b/src/threading/cpu.h
@@ -0,0 +1,14 @@
+#ifndef CPU_H
+#define CPU_H
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+int get_core_count();
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif
\ No newline at end of file
diff --git a/src/threading/pool.c b/src/threading/pool.c
new file mode 100644
index 0000000..266254e
--- /dev/null
+++ b/src/threading/pool.c
@@ -0,0 +1,95 @@
+#include <stdlib.h>
+#include <stddef.h>
+#include <stdio.h>
+#include <pthread.h>
+
+#include "cpu.h"
+#include "pool.h"
+
+struct worker_t {
+    struct pool_t* pool;
+    int id;
+};
+
+struct pool_t* pool_init()
+{
+    struct pool_t* pool = malloc(sizeof(struct pool_t));
+
+    pool->workers = get_core_count();
+    pool->tasks = NULL;
+    pool->task_next = NULL;
+    pool->pool_size = 0;
+    pool->condition = NULL;
+
+    return pool;
+}
+
+void pool_free(struct pool_t* pool)
+{
+    if (pool->tasks)
+        free(pool->tasks);
+    free(pool);
+}
+
+void pool_submit(struct pool_t* pool, worker_func func, void* arg)
+{
+    pool->pool_size++;
+    pool->tasks = realloc(pool->tasks, sizeof(struct pool_task_t) * (pool->pool_size));
+
+    struct pool_task_t* task = &pool->tasks[pool->pool_size-1];
+    task->func = func;
+    task->arg = arg;
+    task->done = 0;
+}
+
+static void* task_executor(void* pinfo)
+{
+    struct worker_t* worker = (struct worker_t*)pinfo;
+    struct pool_t* pool = worker->pool;
+
+    if (pool->task_next)
+    {
+        struct pool_task_t* pool_end = pool->tasks + pool->pool_size;
+        struct pool_task_t* task = pool->task_next++;
+        while (pool_end > task)
+        {
+            if (!task->done)
+            {
+                task->func(task->arg);
+                task->done = 1;
+            }
+
+            task = pool->task_next++;
+        }
+    }
+
+    return NULL;
+}
+
+void pool_complete(struct pool_t* pool)
+{
+    pthread_t* threads = malloc(sizeof(pthread_t) * pool->workers);
+    struct worker_t* workers = malloc(sizeof(struct worker_t) * pool->workers);
+
+    if (pool->pool_size)
+        pool->task_next = &pool->tasks[0];
+
+    for (int i = 0; i < pool->workers && (pool->condition == NULL || *pool->condition); ++i)
+    {
+        struct worker_t* worker = &workers[i];
+        pthread_t* thread = &threads[i];
+
+        worker->pool = pool;
+        worker->id = i;
+
+        pthread_create(thread, NULL, task_executor, worker);
+    }
+
+    for (int i = 0; i < pool->workers; ++i)
+    {
+        pthread_join(threads[i], NULL);
+    }
+
+    free(threads);
+    free(workers);
+}
\ No newline at end of file
diff --git a/src/threading/pool.h b/src/threading/pool.h
new file mode 100644
index 0000000..73655d0
--- /dev/null
+++ b/src/threading/pool.h
@@ -0,0 +1,38 @@
+#ifndef POOL_H
+#define POOL_H
+
+#ifdef __cplusplus
+extern "C" {
+#endif
+
+#include <stddef.h>
+
+typedef void *(*worker_func)(void *);
+
+struct pool_task_t {
+    worker_func func;
+    void* arg;
+    int done;
+};
+
+struct pool_t {
+    int workers;
+
+    struct pool_task_t* tasks;
+    struct pool_task_t* task_next;
+    size_t pool_size;
+
+    int* condition;
+};
+
+struct pool_t* pool_init();
+void pool_free(struct pool_t*);
+
+void pool_submit(struct pool_t*, worker_func, void*);
+void pool_complete(struct pool_t*);
+
+#ifdef __cplusplus
+}
+#endif
+
+#endif
\ No newline at end of file