mirrors/C-Thread-Pool
1
0
Fork 0
mirror of https://github.com/Pithikos/C-Thread-Pool synced 2024-11-21 21:21:23 +03:00
Code Issues Packages Projects Releases Wiki Activity

Fix whitespace

Browse Source
This commit is contained in:
Pithikos 2017-01-17 19:59:29 +00:00
parent 48ae25d35a
commit 2ba0bd59fa
4 changed files with 53 additions and 54 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
README.md
View File

@ -37,7 +37,6 @@ you can use `thpool_wait(thpool);`. If you want to destroy the pool you can use
`thpool_destroy(thpool);`. `thpool_destroy(thpool);`.
## API ## API
For a deeper look into the documentation check in the [thpool.h](https://github.com/Pithikos/C-Thread-Pool/blob/master/thpool.h) file. Below is a fast practical overview. For a deeper look into the documentation check in the [thpool.h](https://github.com/Pithikos/C-Thread-Pool/blob/master/thpool.h) file. Below is a fast practical overview.

0
tests/funcs.sh Normal file → Executable file
View File

52
thpool.c
View File

@ -5,7 +5,7 @@
* work. For usage, check the thpool.h file or README.md * work. For usage, check the thpool.h file or README.md
* *
*//** @file thpool.h *//* *//** @file thpool.h *//*
* *
********************************/ ********************************/
#include <unistd.h> #include <unistd.h>
@ -14,7 +14,7 @@
#include <stdlib.h> #include <stdlib.h>
#include <pthread.h> #include <pthread.h>
#include <errno.h> #include <errno.h>
#include <time.h> #include <time.h>
#if defined(__linux__) #if defined(__linux__)
#include <sys/prctl.h> #include <sys/prctl.h>
#endif #endif
@ -75,7 +75,7 @@ typedef struct thpool_{
volatile int num_threads_working; /* threads currently working */ volatile int num_threads_working; /* threads currently working */
pthread_mutex_t thcount_lock; /* used for thread count etc */ pthread_mutex_t thcount_lock; /* used for thread count etc */
pthread_cond_t threads_all_idle; /* signal to thpool_wait */ pthread_cond_t threads_all_idle; /* signal to thpool_wait */
jobqueue* jobqueue_p; /* pointer to the job queue */ jobqueue* jobqueue_p; /* pointer to the job queue */
} thpool_; } thpool_;
@ -148,7 +148,7 @@ struct thpool_* thpool_init(int num_threads){
pthread_mutex_init(&(thpool_p->thcount_lock), NULL); pthread_mutex_init(&(thpool_p->thcount_lock), NULL);
pthread_cond_init(&thpool_p->threads_all_idle, NULL); pthread_cond_init(&thpool_p->threads_all_idle, NULL);
/* Thread init */ /* Thread init */
int n; int n;
for (n=0; n<num_threads; n++){ for (n=0; n<num_threads; n++){
@ -156,7 +156,7 @@ struct thpool_* thpool_init(int num_threads){
if (THPOOL_DEBUG) if (THPOOL_DEBUG)
printf("THPOOL_DEBUG: Created thread %d in pool \n", n); printf("THPOOL_DEBUG: Created thread %d in pool \n", n);
} }
/* Wait for threads to initialize */ /* Wait for threads to initialize */
while (thpool_p->num_threads_alive != num_threads) {} while (thpool_p->num_threads_alive != num_threads) {}
@ -206,7 +206,7 @@ void thpool_destroy(thpool_* thpool_p){
/* End each thread 's infinite loop */ /* End each thread 's infinite loop */
threads_keepalive = 0; threads_keepalive = 0;
/* Give one second to kill idle threads */ /* Give one second to kill idle threads */
double TIMEOUT = 1.0; double TIMEOUT = 1.0;
time_t start, end; time_t start, end;
@ -217,7 +217,7 @@ void thpool_destroy(thpool_* thpool_p){
time (&end); time (&end);
tpassed = difftime(end,start); tpassed = difftime(end,start);
} }
/* Poll remaining threads */ /* Poll remaining threads */
while (thpool_p->num_threads_alive){ while (thpool_p->num_threads_alive){
bsem_post_all(thpool_p->jobqueue_p->has_jobs); bsem_post_all(thpool_p->jobqueue_p->has_jobs);
@ -227,7 +227,7 @@ void thpool_destroy(thpool_* thpool_p){
/* Job queue cleanup */ /* Job queue cleanup */
jobqueue_destroy(thpool_p); jobqueue_destroy(thpool_p);
free(thpool_p->jobqueue_p); free(thpool_p->jobqueue_p);
/* Deallocs */ /* Deallocs */
int n; int n;
for (n=0; n < threads_total; n++){ for (n=0; n < threads_total; n++){
@ -265,13 +265,13 @@ int thpool_num_threads_working(thpool_* thpool_p){
/* Initialize a thread in the thread pool /* Initialize a thread in the thread pool
* *
* @param thread address to the pointer of the thread to be created * @param thread address to the pointer of the thread to be created
* @param id id to be given to the thread * @param id id to be given to the thread
* @return 0 on success, -1 otherwise. * @return 0 on success, -1 otherwise.
*/ */
static int thread_init (thpool_* thpool_p, struct thread** thread_p, int id){ static int thread_init (thpool_* thpool_p, struct thread** thread_p, int id){
*thread_p = (struct thread*)malloc(sizeof(struct thread)); *thread_p = (struct thread*)malloc(sizeof(struct thread));
if (thread_p == NULL){ if (thread_p == NULL){
fprintf(stderr, "thread_init(): Could not allocate memory for thread\n"); fprintf(stderr, "thread_init(): Could not allocate memory for thread\n");
@ -297,10 +297,10 @@ static void thread_hold () {
/* What each thread is doing /* What each thread is doing
* *
* In principle this is an endless loop. The only time this loop gets interuppted is once * In principle this is an endless loop. The only time this loop gets interuppted is once
* thpool_destroy() is invoked or the program exits. * thpool_destroy() is invoked or the program exits.
* *
* @param thread thread that will run this function * @param thread thread that will run this function
* @return nothing * @return nothing
*/ */
@ -321,7 +321,7 @@ static void* thread_do(struct thread* thread_p){
/* Assure all threads have been created before starting serving */ /* Assure all threads have been created before starting serving */
thpool_* thpool_p = thread_p->thpool_p; thpool_* thpool_p = thread_p->thpool_p;
/* Register signal handler */ /* Register signal handler */
struct sigaction act; struct sigaction act;
sigemptyset(&act.sa_mask); sigemptyset(&act.sa_mask);
@ -330,7 +330,7 @@ static void* thread_do(struct thread* thread_p){
if (sigaction(SIGUSR1, &act, NULL) == -1) { if (sigaction(SIGUSR1, &act, NULL) == -1) {
fprintf(stderr, "thread_do(): cannot handle SIGUSR1"); fprintf(stderr, "thread_do(): cannot handle SIGUSR1");
} }
/* Mark thread as alive (initialized) */ /* Mark thread as alive (initialized) */
pthread_mutex_lock(&thpool_p->thcount_lock); pthread_mutex_lock(&thpool_p->thcount_lock);
thpool_p->num_threads_alive += 1; thpool_p->num_threads_alive += 1;
@ -341,11 +341,11 @@ static void* thread_do(struct thread* thread_p){
bsem_wait(thpool_p->jobqueue_p->has_jobs); bsem_wait(thpool_p->jobqueue_p->has_jobs);
if (threads_keepalive){ if (threads_keepalive){
pthread_mutex_lock(&thpool_p->thcount_lock); pthread_mutex_lock(&thpool_p->thcount_lock);
thpool_p->num_threads_working++; thpool_p->num_threads_working++;
pthread_mutex_unlock(&thpool_p->thcount_lock); pthread_mutex_unlock(&thpool_p->thcount_lock);
/* Read job from queue and execute it */ /* Read job from queue and execute it */
void (*func_buff)(void* arg); void (*func_buff)(void* arg);
void* arg_buff; void* arg_buff;
@ -359,7 +359,7 @@ static void* thread_do(struct thread* thread_p){
func_buff(arg_buff); func_buff(arg_buff);
free(job_p); free(job_p);
} }
pthread_mutex_lock(&thpool_p->thcount_lock); pthread_mutex_lock(&thpool_p->thcount_lock);
thpool_p->num_threads_working--; thpool_p->num_threads_working--;
if (!thpool_p->num_threads_working) { if (!thpool_p->num_threads_working) {
@ -391,7 +391,7 @@ static void thread_destroy (thread* thread_p){
/* Initialize queue */ /* Initialize queue */
static int jobqueue_init(thpool_* thpool_p){ static int jobqueue_init(thpool_* thpool_p){
thpool_p->jobqueue_p = (struct jobqueue*)malloc(sizeof(struct jobqueue)); thpool_p->jobqueue_p = (struct jobqueue*)malloc(sizeof(struct jobqueue));
if (thpool_p->jobqueue_p == NULL){ if (thpool_p->jobqueue_p == NULL){
return -1; return -1;
@ -445,16 +445,16 @@ static void jobqueue_push(thpool_* thpool_p, struct job* newjob){
default: /* if jobs in queue */ default: /* if jobs in queue */
thpool_p->jobqueue_p->rear->prev = newjob; thpool_p->jobqueue_p->rear->prev = newjob;
thpool_p->jobqueue_p->rear = newjob; thpool_p->jobqueue_p->rear = newjob;
} }
thpool_p->jobqueue_p->len++; thpool_p->jobqueue_p->len++;
bsem_post(thpool_p->jobqueue_p->has_jobs); bsem_post(thpool_p->jobqueue_p->has_jobs);
} }
/* Get first job from queue(removes it from queue) /* Get first job from queue(removes it from queue)
* *
* Notice: Caller MUST hold a mutex * Notice: Caller MUST hold a mutex
*/ */
static struct job* jobqueue_pull(thpool_* thpool_p){ static struct job* jobqueue_pull(thpool_* thpool_p){
@ -463,24 +463,24 @@ static struct job* jobqueue_pull(thpool_* thpool_p){
job_p = thpool_p->jobqueue_p->front; job_p = thpool_p->jobqueue_p->front;
switch(thpool_p->jobqueue_p->len){ switch(thpool_p->jobqueue_p->len){
case 0: /* if no jobs in queue */ case 0: /* if no jobs in queue */
break; break;
case 1: /* if one job in queue */ case 1: /* if one job in queue */
thpool_p->jobqueue_p->front = NULL; thpool_p->jobqueue_p->front = NULL;
thpool_p->jobqueue_p->rear = NULL; thpool_p->jobqueue_p->rear = NULL;
thpool_p->jobqueue_p->len = 0; thpool_p->jobqueue_p->len = 0;
break; break;
default: /* if >1 jobs in queue */ default: /* if >1 jobs in queue */
thpool_p->jobqueue_p->front = job_p->prev; thpool_p->jobqueue_p->front = job_p->prev;
thpool_p->jobqueue_p->len--; thpool_p->jobqueue_p->len--;
/* more than one job in queue -> post it */ /* more than one job in queue -> post it */
bsem_post(thpool_p->jobqueue_p->has_jobs); bsem_post(thpool_p->jobqueue_p->has_jobs);
} }
return job_p; return job_p;
} }

54
thpool.h
View File

@ -1,7 +1,7 @@
/********************************** /**********************************
* @author Johan Hanssen Seferidis * @author Johan Hanssen Seferidis
* License: MIT * License: MIT
* *
**********************************/ **********************************/
#ifndef _THPOOL_ #ifndef _THPOOL_
@ -19,17 +19,17 @@ typedef struct thpool_* threadpool;
/** /**
* @brief Initialize threadpool * @brief Initialize threadpool
* *
* Initializes a threadpool. This function will not return untill all * Initializes a threadpool. This function will not return untill all
* threads have initialized successfully. * threads have initialized successfully.
* *
* @example * @example
* *
* .. * ..
* threadpool thpool; //First we declare a threadpool * threadpool thpool; //First we declare a threadpool
* thpool = thpool_init(4); //then we initialize it to 4 threads * thpool = thpool_init(4); //then we initialize it to 4 threads
* .. * ..
* *
* @param num_threads number of threads to be created in the threadpool * @param num_threads number of threads to be created in the threadpool
* @return threadpool created threadpool on success, * @return threadpool created threadpool on success,
* NULL on error * NULL on error
@ -39,26 +39,26 @@ threadpool thpool_init(int num_threads);
/** /**
* @brief Add work to the job queue * @brief Add work to the job queue
* *
* Takes an action and its argument and adds it to the threadpool's job queue. * Takes an action and its argument and adds it to the threadpool's job queue.
* If you want to add to work a function with more than one arguments then * If you want to add to work a function with more than one arguments then
* a way to implement this is by passing a pointer to a structure. * a way to implement this is by passing a pointer to a structure.
* *
* NOTICE: You have to cast both the function and argument to not get warnings. * NOTICE: You have to cast both the function and argument to not get warnings.
* *
* @example * @example
* *
* void print_num(int num){ * void print_num(int num){
* printf("%d\n", num); * printf("%d\n", num);
* } * }
* *
* int main() { * int main() {
* .. * ..
* int a = 10; * int a = 10;
* thpool_add_work(thpool, (void*)print_num, (void*)a); * thpool_add_work(thpool, (void*)print_num, (void*)a);
* .. * ..
* } * }
* *
* @param threadpool threadpool to which the work will be added * @param threadpool threadpool to which the work will be added
* @param function_p pointer to function to add as work * @param function_p pointer to function to add as work
* @param arg_p pointer to an argument * @param arg_p pointer to an argument
@ -69,19 +69,19 @@ int thpool_add_work(threadpool, void (*function_p)(void*), void* arg_p);
/** /**
* @brief Wait for all queued jobs to finish * @brief Wait for all queued jobs to finish
* *
* Will wait for all jobs - both queued and currently running to finish. * Will wait for all jobs - both queued and currently running to finish.
* Once the queue is empty and all work has completed, the calling thread * Once the queue is empty and all work has completed, the calling thread
* (probably the main program) will continue. * (probably the main program) will continue.
* *
* Smart polling is used in wait. The polling is initially 0 - meaning that * Smart polling is used in wait. The polling is initially 0 - meaning that
* there is virtually no polling at all. If after 1 seconds the threads * there is virtually no polling at all. If after 1 seconds the threads
* haven't finished, the polling interval starts growing exponentially * haven't finished, the polling interval starts growing exponentially
* untill it reaches max_secs seconds. Then it jumps down to a maximum polling * untill it reaches max_secs seconds. Then it jumps down to a maximum polling
* interval assuming that heavy processing is being used in the threadpool. * interval assuming that heavy processing is being used in the threadpool.
* *
* @example * @example
* *
* .. * ..
* threadpool thpool = thpool_init(4); * threadpool thpool = thpool_init(4);
* .. * ..
@ -90,7 +90,7 @@ int thpool_add_work(threadpool, void (*function_p)(void*), void* arg_p);
* thpool_wait(thpool); * thpool_wait(thpool);
* puts("All added work has finished"); * puts("All added work has finished");
* .. * ..
* *
* @param threadpool the threadpool to wait for * @param threadpool the threadpool to wait for
* @return nothing * @return nothing
*/ */
@ -99,22 +99,22 @@ void thpool_wait(threadpool);
/** /**
* @brief Pauses all threads immediately * @brief Pauses all threads immediately
* *
* The threads will be paused no matter if they are idle or working. * The threads will be paused no matter if they are idle or working.
* The threads return to their previous states once thpool_resume * The threads return to their previous states once thpool_resume
* is called. * is called.
* *
* While the thread is being paused, new work can be added. * While the thread is being paused, new work can be added.
* *
* @example * @example
* *
* threadpool thpool = thpool_init(4); * threadpool thpool = thpool_init(4);
* thpool_pause(thpool); * thpool_pause(thpool);
* .. * ..
* // Add a bunch of work * // Add a bunch of work
* .. * ..
* thpool_resume(thpool); // Let the threads start their magic * thpool_resume(thpool); // Let the threads start their magic
* *
* @param threadpool the threadpool where the threads should be paused * @param threadpool the threadpool where the threads should be paused
* @return nothing * @return nothing
*/ */
@ -123,14 +123,14 @@ void thpool_pause(threadpool);
/** /**
* @brief Unpauses all threads if they are paused * @brief Unpauses all threads if they are paused
* *
* @example * @example
* .. * ..
* thpool_pause(thpool); * thpool_pause(thpool);
* sleep(10); // Delay execution 10 seconds * sleep(10); // Delay execution 10 seconds
* thpool_resume(thpool); * thpool_resume(thpool);
* .. * ..
* *
* @param threadpool the threadpool where the threads should be unpaused * @param threadpool the threadpool where the threads should be unpaused
* @return nothing * @return nothing
*/ */
@ -139,10 +139,10 @@ void thpool_resume(threadpool);
/** /**
* @brief Destroy the threadpool * @brief Destroy the threadpool
* *
* This will wait for the currently active threads to finish and then 'kill' * This will wait for the currently active threads to finish and then 'kill'
* the whole threadpool to free up memory. * the whole threadpool to free up memory.
* *
* @example * @example
* int main() { * int main() {
* threadpool thpool1 = thpool_init(2); * threadpool thpool1 = thpool_init(2);
@ -152,7 +152,7 @@ void thpool_resume(threadpool);
* .. * ..
* return 0; * return 0;
* } * }
* *
* @param threadpool the threadpool to destroy * @param threadpool the threadpool to destroy
* @return nothing * @return nothing
*/ */