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

Remove trailing spaces in code

Browse Source
This commit is contained in:
nil0x42 2015-04-11 00:51:59 +02:00
parent 470dcc7521
commit 29c8e4e401
1 changed files with 37 additions and 37 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

74
thpool.c
View File

@ -1,11 +1,11 @@
/* ******************************** /* ********************************
* Author: Johan Hanssen Seferidis * Author: Johan Hanssen Seferidis
* License: MIT * License: MIT
* Description: Library providing a threading pool where you can add * Description: Library providing a threading pool where you can add
* 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 *//*
* *
********************************/ ********************************/
@ -15,7 +15,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>
#include "thpool.h" #include "thpool.h"
@ -72,7 +72,7 @@ typedef struct thpool_{
volatile int num_threads_alive; /* threads currently alive */ volatile int num_threads_alive; /* threads currently alive */
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 */
jobqueue* jobqueue_p; /* pointer to the job queue */ jobqueue* jobqueue_p; /* pointer to the job queue */
} thpool_; } thpool_;
@ -139,7 +139,7 @@ struct thpool_* thpool_init(int num_threads){
fprintf(stderr, "thpool_init(): Could not allocate memory for threads\n"); fprintf(stderr, "thpool_init(): Could not allocate memory for threads\n");
exit(1); exit(1);
} }
/* Thread init */ /* Thread init */
int n; int n;
for (n=0; n<num_threads; n++){ for (n=0; n<num_threads; n++){
@ -148,7 +148,7 @@ struct thpool_* thpool_init(int num_threads){
printf("THPOOL_DEBUG: Created thread %d in pool \n", n); printf("THPOOL_DEBUG: Created thread %d in pool \n", n);
#endif #endif
} }
/* 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) {}
@ -187,8 +187,8 @@ void thpool_wait(thpool_* thpool_p){
time_t start, end; time_t start, end;
double tpassed = 0.0; double tpassed = 0.0;
time (&start); time (&start);
while (tpassed < timeout && while (tpassed < timeout &&
(thpool_p->jobqueue_p->len || thpool_p->num_threads_working)) (thpool_p->jobqueue_p->len || thpool_p->num_threads_working))
{ {
time (&end); time (&end);
tpassed = difftime(end,start); tpassed = difftime(end,start);
@ -199,11 +199,11 @@ void thpool_wait(thpool_* thpool_p){
long new_nano; long new_nano;
double multiplier = 1.01; double multiplier = 1.01;
int max_secs = 20; int max_secs = 20;
struct timespec polling_interval; struct timespec polling_interval;
polling_interval.tv_sec = 0; polling_interval.tv_sec = 0;
polling_interval.tv_nsec = init_nano; polling_interval.tv_nsec = init_nano;
while (thpool_p->jobqueue_p->len || thpool_p->num_threads_working) while (thpool_p->jobqueue_p->len || thpool_p->num_threads_working)
{ {
nanosleep(&polling_interval, NULL); nanosleep(&polling_interval, NULL);
@ -216,7 +216,7 @@ void thpool_wait(thpool_* thpool_p){
} }
else break; else break;
} }
/* Fall back to max polling */ /* Fall back to max polling */
while (thpool_p->jobqueue_p->len || thpool_p->num_threads_working){ while (thpool_p->jobqueue_p->len || thpool_p->num_threads_working){
sleep(max_secs); sleep(max_secs);
@ -226,12 +226,12 @@ void thpool_wait(thpool_* thpool_p){
/* Destroy the threadpool */ /* Destroy the threadpool */
void thpool_destroy(thpool_* thpool_p){ void thpool_destroy(thpool_* thpool_p){
volatile int threads_total = thpool_p->num_threads_alive; volatile int threads_total = thpool_p->num_threads_alive;
/* 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;
@ -242,7 +242,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);
@ -252,7 +252,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++){
@ -285,13 +285,13 @@ void thpool_resume(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
* *
*/ */
static void thread_init (thpool_* thpool_p, struct thread** thread_p, int id){ static void 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, "thpool_init(): Could not allocate memory for thread\n"); fprintf(stderr, "thpool_init(): Could not allocate memory for thread\n");
@ -303,7 +303,7 @@ static void thread_init (thpool_* thpool_p, struct thread** thread_p, int id){
pthread_create(&(*thread_p)->pthread, NULL, (void *)thread_do, (*thread_p)); pthread_create(&(*thread_p)->pthread, NULL, (void *)thread_do, (*thread_p));
pthread_detach((*thread_p)->pthread); pthread_detach((*thread_p)->pthread);
} }
@ -317,10 +317,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
*/ */
@ -328,14 +328,14 @@ 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;
act.sa_handler = thread_hold; act.sa_handler = thread_hold;
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;
@ -346,11 +346,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;
@ -364,7 +364,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--;
pthread_mutex_unlock(&thpool_p->thcount_lock); pthread_mutex_unlock(&thpool_p->thcount_lock);
@ -393,19 +393,19 @@ 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));
pthread_mutex_init(&(thpool_p->jobqueue_p->rwmutex), NULL); pthread_mutex_init(&(thpool_p->jobqueue_p->rwmutex), NULL);
if (thpool_p->jobqueue_p == NULL){ if (thpool_p->jobqueue_p == NULL){
return -1; return -1;
} }
thpool_p->jobqueue_p->has_jobs = (struct bsem*)malloc(sizeof(struct bsem)); thpool_p->jobqueue_p->has_jobs = (struct bsem*)malloc(sizeof(struct bsem));
if (thpool_p->jobqueue_p->has_jobs == NULL){ if (thpool_p->jobqueue_p->has_jobs == NULL){
return -1; return -1;
} }
bsem_init(thpool_p->jobqueue_p->has_jobs, 0); bsem_init(thpool_p->jobqueue_p->has_jobs, 0);
jobqueue_clear(thpool_p); jobqueue_clear(thpool_p);
return 0; return 0;
} }
@ -444,16 +444,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){
@ -462,21 +462,21 @@ 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 */
return NULL; return NULL;
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;
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--;
/* Make sure has_jobs has right value */ /* Make sure has_jobs has right value */
if (thpool_p->jobqueue_p->len > 0) { if (thpool_p->jobqueue_p->len > 0) {
bsem_post(thpool_p->jobqueue_p->has_jobs); bsem_post(thpool_p->jobqueue_p->has_jobs);