// Git repo at https://gitlab.com/AaronErhardt/threadboost // Actual code starts at line 200 /* threadboost.h * * Copyright (C) 2019-2020 Aaron Erhardt * * This program is free software: you can redistribute it and/or modify * it under the terms of the GNU General Public License as published by * the Free Software Foundation, either version 3 of the License, or * (at your option) any later version. * * This program is distributed in the hope that it will be useful, * but WITHOUT ANY WARRANTY; without even the implied warranty of * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the * GNU General Public License for more details. * * You should have received a copy of the GNU General Public License * along with this program. If not, see . */ #ifndef __THBOOST #define __THBOOST #ifdef __cplusplus extern "C" { #endif /* ============================== ABOUT THREADBOOST ============================ * * Threadboost is a highly optimized thread pool that aims to run as fast as possible. * In certain situations in can run over 200 times faster than regular thread pools * yet its design might have some drawbacks for certain applications. */ /* ================================== THE API ================================== */ /* >>> struct thboost_t * * This struct stores all relevant information about the thread pool. * It is not intended to be accessed manually. */ typedef struct pub_thboost_t thboost_t; /* >>> thboost_t* thboost_init(unsigned int num_of_threads) * * DESCRIPTION: returns a new thread pool with the specified number of worker threads. * * RETURN: a new threadpool. * * ARGUMENT: the number of worker threads, * recommended maximum is one less than the number of CPU cores. */ thboost_t* thboost_init(unsigned int num_of_threads); /* >>> void thboost_destroy(thboost_t* th_pool); * * DESCRIPTION: destroys a given thread pool and frees all resources. * * ARGUMENT: pointer to the thread pool that should be destroyed. */ void thboost_destroy(thboost_t* th_pool); /* >>> void thboost_work(thboost_t* th_pool); * * DESCRIPTION: lets all worker threads do their assigned work, * returns when all worker threads are finished. * * ARGUMENT: pointer to the thread pool that should do the work. */ void thboost_work(thboost_t* th_pool); /* >>> void thboost_add_work(thboost_t* th_pool, unsigned int thread_num, void* func, void* args); * * DESCRIPTION: Assigns work (a function + argument) to a specific worker thread. * * ARGUMENTS: - th_pool: pointer to the thread pool of the targeted worker thread * - thread_num: number of the worker thread (0 is the first, 1 the second etc.) * - func: function pointer assigned to the worker thread * - args: argument, the assigned function should take (pass multiple arguments as a struct) */ void thboost_add_work(thboost_t* th_pool, unsigned int thread_num, void* func, void* args); /* >>> void thboost_add_work_all(thboost_t* th_pool, void* func, void* args); * * DESCRIPTION: Assigns the same work (a function + argument) to all worker threads. * * ARGUMENTS: - th_pool: pointer to the thread pool of the targeted worker threads * - func: function pointer assigned to all worker threads * - args: argument, the assigned function should take (pass multiple arguments as a struct) */ void thboost_add_work_all(thboost_t* th_pool, void* func, void* args); /* >>> void thboost_add_func(thboost_t* th_pool, unsigned int thread_num, void* func); * * DESCRIPTION: Assigns a function to a specific worker thread. * * ARGUMENTS: - th_pool: pointer to the thread pool of the targeted worker thread * - thread_num: number of the worker thread (0 is the first, 1 the second etc.) * - func: function pointer assigned to the worker thread */ void thboost_add_func(thboost_t* th_pool, unsigned int thread_num, void* func); /* >>> void thboost_add_func_all(thboost_t* th_pool, void* func); * * DESCRIPTION: Assigns the same function to all worker threads. * * ARGUMENTS: - th_pool: pointer to the thread pool of the targeted worker threads * - func: function pointer assigned to all worker threads */ void thboost_add_func_all(thboost_t* th_pool, void* func); /* >>> void thboost_add_args(thboost_t* th_pool, unsigned int thread_num, void* args); * * DESCRIPTION: Assigns a argument to a specific worker thread. * * ARGUMENTS: - th_pool: pointer to the thread pool of the targeted worker thread * - thread_num: number of the worker thread (0 is the first, 1 the second etc.) * - args: argument, the assigned function should take (pass multiple arguments as a struct) */ void thboost_add_args(thboost_t* th_pool, unsigned int thread_num, void* args); /* >>> void thboost_add_args_all(thboost_t* th_pool, void* args); * * DESCRIPTION: Assigns the same argument to all worker threads. * * ARGUMENTS: - th_pool: pointer to the thread pool of the targeted worker threads * - args: argument, the assigned function should take (pass multiple arguments as a struct) */ void thboost_add_args_all(thboost_t* th_pool, void* args); /* ============================== THE INTERNAL API ============================= * * The internal API provides the same functions as the default API (see above) except that the * functions have names that start with "thboost_internal..." instead of "thboost..." * and always work with the internal thread pool. Therefore you also never need to pass * a pointer to a thread pool to functions of the internal API. * Since the internal API behaves similar to the default API only a few points documented * in this section. For reference have look at the functions of the default API. */ /* >>> void thboost_internal_init(unsigned int num_of_threads); * * NOTE: creates the internal thread pool similar to thboost_init. * This function can be called multiple times but will create the internal * thread pool only if it doesn't exist already. * There can only be ONE INTERNAL THREAD POOL. */ void thboost_internal_init(unsigned int num_of_threads); void thboost_internal_work(); void thboost_internal_add_work(unsigned int thread_num, void* func, void* args); void thboost_internal_add_work_all(void* func, void* args); void thboost_internal_add_func(unsigned int thread_num, void* func); void thboost_internal_add_func_all(void* func); void thboost_internal_add_args(unsigned int thread_num, void* args); void thboost_internal_add_args_all(void* args); void thboost_internal_destroy(); #ifdef __cplusplus } #endif #endif // END OF thboost.h // ============== ACTUAL CODE ============== #include #define NUM_OF_THREADS 5 void print_num(void * args) { int* number = (int*)args; printf("NUMBER: %i\n", *number); } int main(int argc, char *argv[]) { /* creating new thread pool */ thboost_t* thpool = thboost_init(NUM_OF_THREADS); int test[] = {0, 1, 2, 3, 4}; /* assigning the function print_num and a number to each worker thread */ for (long t = 0; t < NUM_OF_THREADS; ++t) { thboost_add_work(thpool, t, print_num, (void*)(&test[t])); } /* running all worker threads at once */ thboost_work(thpool); for (long t = 0; t < NUM_OF_THREADS; ++t) { test[t] = t + 5; } printf("\n"); /* running all worker threads at once */ thboost_work(thpool); /* destroying the thread pool */ thboost_destroy(thpool); return 0; } /* threadboost.c * * Copyright (c) 2019-2020 Aaron Erhardt * * This program is free software: you can redistribute it and/or modify * it under the terms of the GNU General Public License as published by * the Free Software Foundation, either version 3 of the License, or * (at your option) any later version. * * This program is distributed in the hope that it will be useful, * but WITHOUT ANY WARRANTY; without even the implied warranty of * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the * GNU General Public License for more details. * * You should have received a copy of the GNU General Public License * along with this program. If not, see . */ #include #include #include //#include "thboost.h" /* ============================= DEBUG FLAGS ============================ */ /* * THBOOST_DEBUG: * Enables printing of error messages (to stderr). * * THBOOST_INFO_MAIN * Enables printing of useful information about the status of the main thread. * * THBOOST_INFO_THREAD * Enables printing of useful information about the status of the worker threads. * * THBOOST_INFO_ALL * Enables both THBOOST_INFO_MAIN and THBOOST_INFO_THREAD. * * THBOOST_TIME_INFO * Enables printing of time information of the worker threads. * This includes the times spent for executing the assigned worker function and * a whole worker loop. * * THBOOST_CPU_STATS * Enables collecting and printing of information about the CPU time and the real * time a worker thread uses. If the CPU time is significantly lower than the real * time you should probably reduce the amount of worker threads and background * processes on your system. * */ /* UNCOMMNENT THE FLAGS YOU NEED HERE */ //#define THBOOST_DEBUG //#define THBOOST_INFO_MAIN //#define THBOOST_INFO_THREAD //#define THBOOST_INFO_ALL //#define THBOOST_TIME_INFO //#define THBOOST_CPU_STATS /* =========================== OPTIONAL DELAY =========================== */ /* Uncomment (and edit) the following section to add a delay to the spinning loops * of the thread pool in order to avoid high CPU usage caused by the spinning of the threads. * But keep in mind that this will HARM THE PERFORMANCE. */ /* #include static struct timespec THBOOST_DELAY = {.tv_sec = 0, .tv_nsec = 1}; #define THBOOST_EVENT_DELAY() nanosleep(&THBOOST_DELAY, NULL) */ /* ============================ PREPROCESSOR ============================ */ #if defined(THBOOST_DEBUG) || defined(THBOOST_INFO) || \ defined(THBOOST_TIME_INFO) || defined(THBOOST_CPU_STATS) #include #endif #if defined(THBOOST_DEBUG) #define err(str) \ fprintf(stderr, "line %d: %s() -> %s\n", __LINE__, __func__, str) #else #define err(str) #endif #if defined(THBOOST_INFO_MAIN) || defined(THBOOST_INFO_ALL) #define INFO_MAIN(str) printf("MAIN: " str "\n") #else #define INFO_MAIN(str) #endif #if defined(THBOOST_INFO_THREAD) || defined(THBOOST_INFO_ALL) #define INFO_THREAD(str) printf("THREAD: " str "\n") #else #define INFO_THREAD(str) #endif #if defined(THBOOST_TIME_INFO) #define INFO_TIME(str, time) printf("TIME: " str " %ld\n", time) #else #define INFO_TIME(str) #endif /* ========================== MACROS ============================ */ /* easily allocate memory and avoid errors with automatic type casting */ #define easy_malloc(data_type, num_of_elems) \ (data_type *) malloc(sizeof(data_type) * num_of_elems) /* ========================== STRUCTURES ============================ */ /* >>> worker_job_t * * Stores two pointers to the function and the argument a worker thread needs * to run its assgigned job. */ typedef struct worker_job_t { void (*volatile func)(void *); /* Function a worker thread will execute */ void volatile *args; /* Arguments used for the function */ } worker_job_t; /* >>> worker_args_t * * Stores the pointers needed to initialize the worker threads, * used in the function worker_loop. */ typedef struct worker_args_t { worker_job_t *job; uint_fast8_t volatile *start_condition; uint_fast8_t volatile *stop_condition; uint_fast8_t volatile *exit_condition; } worker_args_t; /* >>> pub_thboost_t aka. thboost_t * * Stores all information related to the threadpool and required by the API. */ typedef struct pub_thboost_t { pthread_t *threads; pthread_attr_t th_attr; unsigned int num_of_threads; uint_fast8_t volatile start_condition; uint_fast8_t volatile exit_condition; uint_fast8_t volatile *stop_conditions; uint_fast8_t volatile *last_stop_condition; worker_job_t *main_job; worker_job_t *jobs; worker_args_t *worker_args; } pub_thboost_t; /* ========================== STATIC FUNCTIONS ============================ */ static void worker_loop(void *worker_args); static void empty_func(void *empty_args) {} /* ========================== THREADBOOST ============================ */ pub_thboost_t *thboost_init(unsigned int num_of_threads) { /* "num_of_workers" is one less than the actual num_of_threads * because the main thread also acts like a worker thread.*/ unsigned int num_of_workers = num_of_threads - 1; INFO_MAIN("Allocating memory for the thboost struct"); pub_thboost_t *this = easy_malloc(pub_thboost_t, 1); if (this == NULL) { err("Could not allocate memory for thboost structure"); return NULL; } this->num_of_threads = num_of_threads; this->exit_condition = 0; this->start_condition = 0; /* Set the pthread attribute to make all new threads detached in order * to save up system resources */ pthread_attr_init(&this->th_attr); pthread_attr_setdetachstate(&this->th_attr, PTHREAD_CREATE_DETACHED); INFO_MAIN("Allocating memory for the worker threads"); this->threads = easy_malloc(pthread_t, num_of_workers); if (this->threads == NULL) { err("Could not allocate memory for the worker threads"); free(this); return NULL; } INFO_MAIN("Allocating memory for the stop conditions of the worker threads"); this->stop_conditions = easy_malloc(uint_fast8_t, num_of_workers); if (this->stop_conditions == NULL) { err("Could not allocate memory for stop conditions of the worker threads"); free(this->threads); free(this); return NULL; } INFO_MAIN("Allocating memory for the jobs of the worker threads"); this->jobs = easy_malloc(worker_job_t, num_of_threads); if (this->jobs == NULL) { err("Could not allocate memory for the jobs of the worker threads"); free(this->threads); free((void *)this->stop_conditions); free(this); return NULL; } INFO_MAIN("Allocating memory for the worker arguments"); this->worker_args = easy_malloc(worker_args_t, num_of_workers); if (this->worker_args == NULL) { err("Could not allocate memory for worker arguments"); free(this->threads); free((void *)this->stop_conditions); free(this->jobs); free(this); return NULL; } // first thread is MAIN (thread No 0), so no worker thread needs to be spawned this->jobs[0].args = NULL; this->jobs[0].func = empty_func; this->main_job = &this->jobs[0]; INFO_MAIN("Initializing worker threads"); int rc; for (uint_fast8_t t = 0; t < num_of_workers; ++t) { /* Initialize worker threads to execute empty_func with argument "NULL" */ this->jobs[t + 1].args = NULL; this->jobs[t + 1].func = empty_func; /* Initialize worker args with all info the worker thread needs */ this->worker_args[t].job = &this->jobs[t + 1]; this->worker_args[t].stop_condition = &this->stop_conditions[t]; this->worker_args[t].start_condition = &this->start_condition; this->worker_args[t].exit_condition = &this->exit_condition; /* create pthreads */ rc = pthread_create(&this->threads[t], &this->th_attr, (void *)worker_loop, (void *)&this->worker_args[t]); if (rc) { err("Couldn't create thread!"); exit(-1); } } /* calculating the pointer of the last stop condition to optimize a loop in thboost_work */ this->last_stop_condition = this->stop_conditions + num_of_workers; INFO_MAIN("thboost_init finished successfully"); return this; } void thboost_work(pub_thboost_t *this) { #if defined(THBOOST_DEBUG) int count_waiting_iterations = 0; #endif INFO_MAIN("Waking up worker threads"); this->start_condition++; INFO_MAIN("Doing work"); this->main_job->func((void *)this->main_job->args); INFO_MAIN("Finished work"); register uint_fast8_t volatile *stop_condition = this->stop_conditions; register uint_fast8_t volatile start_condition = this->start_condition; while (stop_condition < this->last_stop_condition) { INFO_MAIN("Waiting for worker threads to finish"); while (*stop_condition != start_condition) { #if defined(THBOOST_DEBUG) count_waiting_iterations++; #endif #ifdef THBOOST_EVENT_DELAY THBOOST_EVENT_DELAY(); #endif } stop_condition++; } #if defined(THBOOST_DEBUG) if (count_waiting_iterations > 10000) { err("WARNING: Slow worker threads: Maybe too many threads active!"); } #endif INFO_MAIN("thboost_work finished successfully"); } void thboost_destroy(pub_thboost_t *this) { INFO_MAIN("Destroying threads"); this->exit_condition = 1; this->start_condition++; for (uint_fast8_t volatile *stop_condition = this->stop_conditions; stop_condition < this->last_stop_condition;) { INFO_MAIN("Waiting for worker threads to exit"); if (*stop_condition == this->start_condition) { stop_condition++; } else { #ifdef THBOOST_EVENT_DELAY THBOOST_EVENT_DELAY(); #endif } } INFO_MAIN("Freeing memory"); free(this->threads); free((void *)this->stop_conditions); free(this->jobs); free(this->worker_args); pthread_attr_destroy(&this->th_attr); free(this); INFO_MAIN("thboost_destroy finished successfully"); } void thboost_add_work(thboost_t *this, unsigned int thread_num, void *func, void *args) { this->jobs[thread_num].func = func; this->jobs[thread_num].args = args; INFO_MAIN("thboost_add_work finished successfully"); } void thboost_add_work_all(thboost_t *this, void *func, void *args) { for (register uint_fast8_t t = 0; t < this->num_of_threads; ++t) { this->jobs[t].func = func; this->jobs[t].args = args; } INFO_MAIN("thboost_add_work_all finished successfully"); } void thboost_add_func(thboost_t *this, unsigned int thread_num, void *func) { this->jobs[thread_num].func = func; INFO_MAIN("thboost_add_func finished successfully"); } void thboost_add_func_all(thboost_t *this, void *func) { for (register uint_fast8_t t = 0; t < this->num_of_threads; ++t) { this->jobs[t].func = func; } INFO_MAIN("thboost_add_func_all finished successfully"); } void thboost_add_args(thboost_t *this, unsigned int thread_num, void *args) { this->jobs[thread_num].args = args; INFO_MAIN("thboost_add_args finished successfully"); } void thboost_add_args_all(thboost_t *this, void *args) { for (register uint_fast8_t t = 0; t < this->num_of_threads; ++t) { this->jobs[t].args = args; } INFO_MAIN("thboost_add_args_all finished successfully"); } /* ========================== HELPER FUNCTIONS ============================ */ static void worker_loop(void *worker_args) { #if defined(THBOOST_TIME_INFO) clock_t last_time = 0; #endif #if defined(THBOOST_CPU_STATS) struct timespec real_time; struct timespec cpu_time; clock_gettime(CLOCK_MONOTONIC, &real_time); clock_gettime(CLOCK_THREAD_CPUTIME_ID, &cpu_time); #endif INFO_THREAD("Worker loop initiated!"); /* Type casting the data type of the argument */ worker_args_t *args = (worker_args_t *)worker_args; /* Storing important pointers in registers to speed up execution */ register void (*volatile * func_ptr)(void *) = &args->job->func; register void *volatile *args_ptr = (void **)&args->job->args; register uint_fast8_t volatile *start_condition_ptr = args->start_condition; register uint_fast8_t volatile *stop_condition_ptr = args->stop_condition; register uint_fast8_t volatile *exit_condition_ptr = args->exit_condition; *stop_condition_ptr = 0; for (;;) { if (*stop_condition_ptr != *start_condition_ptr) { if (*exit_condition_ptr) { INFO_THREAD("EXIT!"); *stop_condition_ptr = *start_condition_ptr; #if defined(THBOOST_CPU_STATS) struct timespec real_time_now; struct timespec cpu_time_now; clock_gettime(CLOCK_MONOTONIC, &real_time_now); clock_gettime(CLOCK_THREAD_CPUTIME_ID, &cpu_time_now); real_time_now.tv_sec -= real_time.tv_sec; real_time_now.tv_nsec -= real_time.tv_nsec; if (real_time_now.tv_nsec < 0) { real_time_now.tv_nsec += 1000000000; // + 1s real_time_now.tv_sec--; // -1s } cpu_time_now.tv_sec -= cpu_time.tv_sec; cpu_time_now.tv_nsec -= cpu_time.tv_nsec; if (cpu_time_now.tv_nsec < 0) { cpu_time_now.tv_nsec += 1000000000; // + 1s cpu_time_now.tv_sec--; // -1s } printf("REAL TIME: %lds, %ld\n", real_time_now.tv_sec, real_time_now.tv_nsec); printf("CPU TIME: %lds, %ld\n", cpu_time_now.tv_sec, cpu_time_now.tv_nsec); #endif pthread_exit(NULL); } INFO_THREAD("Doing work!"); #if defined(THBOOST_TIME_INFO) clock_t current_time = clock(); INFO_TIME("whole worker iteration took", current_time - last_time); last_time = current_time; #endif /* Execute the assigned function */ (*func_ptr)((void *)*args_ptr); #if defined(THBOOST_TIME_INFO) current_time = clock(); INFO_TIME("worker job took", current_time - last_time); #endif /* Set the stop condition so the main thread nows work is finished */ *stop_condition_ptr = *start_condition_ptr; } else { #ifdef THBOOST_EVENT_DELAY THBOOST_EVENT_DELAY(); #endif INFO_THREAD("Sleeping!"); } } } /* ====================== INTERNAL THREADBOOST API ====================== */ static thboost_t *internal_thboost = NULL; void thboost_internal_init(unsigned int num_of_threads) { if (internal_thboost == NULL) { internal_thboost = thboost_init(num_of_threads); } } void thboost_internal_work() { thboost_work(internal_thboost); } void thboost_internal_add_work(unsigned int thread_num, void *func, void *args) { thboost_add_work(internal_thboost, thread_num, func, args); } void thboost_internal_add_work_all(void *func, void *args) { thboost_add_work_all(internal_thboost, func, args); } void thboost_internal_add_args(unsigned int thread_num, void *args) { thboost_add_args(internal_thboost, thread_num, args); } void thboost_internal_add_args_all(void *args) { thboost_add_args_all(internal_thboost, args); } void thboost_internal_add_func(unsigned int thread_num, void *func) { thboost_add_func(internal_thboost, thread_num, func); } void thboost_internal_add_func_all(void *func) { thboost_add_func_all(internal_thboost, func); } void thboost_internal_destroy() { thboost_destroy(internal_thboost); internal_thboost = NULL; }

The worlds fastest thread pool [C]