mirrors
/
C-Thread-Pool
mirror of https://github.com/Pithikos/C-Thread-Pool
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Seperate private from public functions

This commit is contained in:
pithikos 2015-01-15 09:12:26 +00:00
parent d301f761c4
commit bf44140c4f
3 changed files with 233 additions and 280 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

62
example.c Normal file
View File

@ -0,0 +1,62 @@
/*
* This is just an example on how to use the thpool library
*
* We create a pool of 4 threads and then add 20 tasks to the pool(10 task1
* functions and 10 task2 functions).
*
* Task1 doesn't take any arguments. Task2 takes an integer. Task2 is used to show
* how to add work to the thread pool with an argument.
*
* As soon as we add the tasks to the pool, the threads will run them. One thread
* may run x tasks in a row so if you see as output the same thread running several
* tasks, it's not an error.
*
* All jobs will not be completed and in fact maybe even none will. You can add a sleep()
* function if you want to complete all tasks in this test file to be able and see clearer
* what is going on.
*
* */
#include <stdio.h>
#include "thpool.h"
/* Some arbitrary task 1 */
void task1(){
printf("# Thread working: %u\n", (int)pthread_self());
printf(" Task 1 running..\n");
}
/* Some arbitrary task 2 */
void task2(int a){
printf("# Thread working: %u\n", (int)pthread_self());
printf(" Task 2 running..\n");
printf("%d\n", a);
}
int main(){
int i;
thpool_t* threadpool; /* make a new thread pool structure */
threadpool=thpool_init(4); /* initialise it to 4 number of threads */
puts("Adding 20 tasks to threadpool");
int a=54;
for (i=0; i<10; i++){
thpool_add_work(threadpool, (void*)task1, NULL);
thpool_add_work(threadpool, (void*)task2, (void*)a);
};
puts("Will kill threadpool");
thpool_destroy(threadpool);
return 0;
}

165
thpool.c
View File

@ -30,6 +30,77 @@ static volatile int threads_on_hold;
/* ========================== STRUCTURES ============================ */
/* Binary semaphore */
typedef struct bsem_t {
pthread_mutex_t mutex;
pthread_cond_t cond;
int v;
} bsem_t;
/* Job */
typedef struct job_t{
void* (*function)(void* arg); /* function pointer */
void* arg; /* function's argument */
struct job_t* prev; /* pointer to previous job */
} job_t;
/* Job queue */
typedef struct jobqueue_t{
pthread_mutex_t rwmutex; /* used for queue r/w access */
job_t *front; /* pointer to front of queue */
job_t *rear; /* pointer to rear of queue */
bsem_t *has_jobs; /* flag as binary semaphore */
int len; /* number of jobs in queue */
} jobqueue_t;
/* Thread */
typedef struct thread_t{
int id; /* friendly id */
pthread_t pthread; /* pointer to actual thread */
struct thpool_t* thpool; /* access to thpool */
} thread_t;
/* Threadpool */
typedef struct thpool_t{
thread_t** threads; /* pointer to threads */
int threads_alive; /* threads currently alive */
pthread_mutex_t thcount_lock; /* used for thread count etc */
jobqueue_t* jobqueue; /* pointer to the job queue */
} thpool_t;
/* ========================== PROTOTYPES ============================ */
static void thread_init(thpool_t* thpool, thread_t** thread, int id);
static void* thread_do(thread_t* thread);
static void thread_hold();
static void thread_destroy(thread_t* thread);
static int jobqueue_init(thpool_t* thpool);
static void jobqueue_clear(thpool_t* thpool);
static void jobqueue_push(thpool_t* thpool, job_t* newjob);
static job_t* jobqueue_pull(thpool_t* thpool);
static void jobqueue_destroy(thpool_t* thpool);
static void bsem_init(bsem_t *bsem, int value);
static void bsem_reset(bsem_t *bsem);
static void bsem_post(bsem_t *bsem);
static void bsem_post_all(bsem_t *bsem);
static void bsem_wait(bsem_t *bsem);
/* ========================== THREADPOOL ============================ */
/* Initialise thread pool */
@ -169,6 +240,19 @@ void thpool_resume(thpool_t* thpool) {
/* ============================ THREAD ============================== */
/**
* @brief Initialize a thread in the thread pool
*
* Will initialize a new thread for the given threadpool and give the
* the thread an ID
*
* Notice also that the thread's id is not populated automatically.
*
* @param thread address to the pointer of the thread to be created
* @param id id to be given to the thread
*
*/
static void thread_init (thpool_t *thpool, thread_t **thread, int id){
*thread = (thread_t*)malloc(sizeof(thread_t));
@ -185,6 +269,11 @@ static void thread_init (thpool_t *thpool, thread_t **thread, int id){
}
/**
* @brief Sets the calling thread on hold until threads_on_hold is set to 1
* @param thread
*/
static void thread_hold () {
threads_on_hold = 1;
while (threads_on_hold){
@ -193,6 +282,15 @@ static void thread_hold () {
}
/**
* @brief What each thread is doing
*
* In principle this is an endless loop. The only time this loop gets interuppted is once
* thpool_destroy() is invoked or the program exits.
*
* @param thread thread that will run this function
* @return nothing
*/
static void* thread_do(thread_t* thread){
/* Assure all threads have been created before starting serving */
@ -240,6 +338,10 @@ static void* thread_do(thread_t* thread){
}
/**
* @brief Frees a thread
* @param thread
*/
static void thread_destroy (thread_t* thread){
//puts("Destroying thread");
free(thread);
@ -250,7 +352,11 @@ static void thread_destroy (thread_t* thread){
/* ============================ JOB QUEUE =========================== */
/* Initialise queue */
/**
* @brief Initialize queue
* @return 0 on success,
* -1 on memory allocation error
*/
static int jobqueue_init(thpool_t* thpool){
thpool->jobqueue=(jobqueue_t*)malloc(sizeof(jobqueue_t));
if (thpool->jobqueue==NULL){
@ -263,7 +369,9 @@ static int jobqueue_init(thpool_t* thpool){
}
/* Clear queue */
/**
* @brief Clears the queue
*/
static void jobqueue_clear(thpool_t* thpool){
while(thpool->jobqueue->len){
@ -278,7 +386,17 @@ static void jobqueue_clear(thpool_t* thpool){
}
/* Add job to queue */
/**
* @brief Add job to queue
*
* A new job will be added to the queue. The new job MUST be allocated
* before passed to this function or else other functions like jobqueue_empty()
* will be broken.
*
* CALLER MUST HOLD A MUTEX
*
* @param pointer to the new (allocated) job
*/
static void jobqueue_push(thpool_t* thpool, job_t* newjob){ /* remember that job prev and next point to NULL */
newjob->prev = NULL;
@ -300,7 +418,17 @@ static void jobqueue_push(thpool_t* thpool, job_t* newjob){ /* remember that job
}
/* Get first element from queue */
/**
* @brief Get first job from queue(removes it from queue)
*
* This does not free allocated memory so be sure to have peeked() \n
* before invoking this as else there will result lost memory pointers.
*
* CALLER MUST HOLD A MUTEX
*
* @return point to job on success,
* NULL if there is no job in queue
*/
static job_t* jobqueue_pull(thpool_t* thpool){
job_t* job;
@ -331,7 +459,14 @@ static job_t* jobqueue_pull(thpool_t* thpool){
}
/* Remove and deallocate all jobs in queue */
/**
* @brief Remove and deallocate all jobs in queue
*
* This function will deallocate all jobs in the queue and set the
* jobqueue to its initialization values, thus tail and head pointing
* to NULL and amount of jobs equal to 0.
*
* */
static void jobqueue_destroy(thpool_t* thpool){
jobqueue_clear(thpool);
free(thpool->jobqueue->has_jobs);
@ -344,7 +479,9 @@ static void jobqueue_destroy(thpool_t* thpool){
/* ======================== SYNCHRONISATION ========================= */
/* Binary semaphore init */
/**
* @brief Inits semaphore to given value (1 or 0)
* */
static void bsem_init(bsem_t *bsem, int value) {
if (value < 0 || value > 1) {
printf("ERROR: bsem_init(): Binary semaphore can take only values 1 or 0");
@ -354,13 +491,17 @@ static void bsem_init(bsem_t *bsem, int value) {
}
/* Binary semaphore reset */
/**
* @brief Resets semaphore to 0
* */
static void bsem_reset(bsem_t *bsem) {
bsem_init(bsem, 0);
}
/* Binary semaphore post */
/**
* @brief Sets semaphore to one and notifies at least one thread
* */
static void bsem_post(bsem_t *bsem) {
pthread_mutex_lock(&bsem->mutex);
bsem->v = 1;
@ -369,7 +510,9 @@ static void bsem_post(bsem_t *bsem) {
}
/* Binary semaphore post */
/**
* @brief Sets semaphore to one and notifies all threads
* */
static void bsem_post_all(bsem_t *bsem) {
pthread_mutex_lock(&bsem->mutex);
bsem->v = 1;
@ -378,7 +521,9 @@ static void bsem_post_all(bsem_t *bsem) {
}
/* Binary semaphore wait */
/**
* @brief Waits on semaphore until semaphore has value 0
* */
static void bsem_wait(bsem_t *bsem) {
pthread_mutex_lock(&bsem->mutex);
while (bsem->v != 1) {

286
thpool.h
View File

@ -5,57 +5,6 @@
*
**********************************/
/* Description: Library providing a threading pool where you can add work on the fly. The number
* of threads in the pool is adjustable when creating the pool. In most cases
* this should equal the number of threads supported by your cpu.
*
* For an example on how to use the threadpool, check the main.c file or just read
* the documentation found in the README.md file.
*
* In this header file a detailed overview of the functions and the threadpool's logical
* scheme is presented in case you wish to tweak or alter something.
*
*
*
* _______________________________________________________
* / \
* | JOB QUEUE | job1 | job2 | job3 | job4 | .. |
* | |
* | threadpool | thread1 | thread2 | .. |
* \_______________________________________________________/
*
*
* Description: Jobs are added to the job queue. Once a thread in the pool
* is idle, it is assigned with the first job from the queue(and
* erased from the queue). It's each thread's job to read from
* the queue serially(using lock) and executing each job
* until the queue is empty.
*
*
* Scheme:
*
* thpool______ jobqueue____ ______
* | | | | .----------->|_job0_| Newly added job
* | | | rear ----------' |_job1_|
* | jobqueue----------------->| | |_job2_|
* | | | front ----------. |__..__|
* |___________| |___________| '----------->|_jobn_| Job for thread to take
*
*
* job0________
* | |
* | function---->
* | |
* | arg------->
* | | job1________
* | next-------------->| |
* |___________| | |..
*
*
*
*/
#ifndef _THPOOL_
#define _THPOOL_
@ -64,74 +13,25 @@
#include <semaphore.h>
/* ========================== STRUCTURES ============================ */
/* Binary semaphore */
typedef struct bsem_t {
pthread_mutex_t mutex;
pthread_cond_t cond;
int v;
} bsem_t;
/* Job */
typedef struct job_t{
void* (*function)(void* arg); /* function pointer */
void* arg; /* function's argument */
struct job_t* prev; /* pointer to previous job */
} job_t;
/* Job queue */
typedef struct jobqueue_t{
pthread_mutex_t rwmutex; /* used for queue r/w access */
job_t *front; /* pointer to front of queue */
job_t *rear; /* pointer to rear of queue */
bsem_t *has_jobs; /* flag as binary semaphore */
int len; /* number of jobs in queue */
} jobqueue_t;
/* Thread */
typedef struct thread_t{
int id; /* friendly id */
pthread_t pthread; /* pointer to actual thread */
struct thpool_t* thpool; /* access to thpool */
} thread_t;
/* Threadpool */
typedef struct thpool_t{
thread_t** threads; /* pointer to threads */
int threads_alive; /* threads currently alive */
pthread_mutex_t thcount_lock; /* used for thread count etc */
jobqueue_t* jobqueue; /* pointer to the job queue */
} thpool_t;
/* thpool is a pointer to a thpool_t data structure */
typedef struct thpool_t* thpool;
/* =========================== FUNCTIONS ============================ */
/* ---------------------- Threadpool specific ----------------------- */
/**
* @brief Initialize threadpool
*
* Initializes a threadpool. This function will not return untill all
* threads have initialized succesfully.
* threads have initialized successfully.
*
* @param number of threads to be used
* @return threadpool struct on success,
* NULL on error
* @param threadsN number of threads to be created in the threadpool
* @return thpool pointer to created threadpool on success,
* pointer to NULL on error
*/
thpool_t* thpool_init(int threadsN);
extern thpool* thpool_init(int threadsN);
/**
@ -143,12 +43,12 @@ thpool_t* thpool_init(int threadsN);
*
* NOTICE: You have to cast both the function and argument to not get warnings.
*
* @param threadpool to where the work will be added to
* @param function to add as work
* @param thpool threadpool to which the work will be added
* @param function function to add as work
* @param argument to the above function
* @return int
*/
int thpool_add_work(thpool_t* thpool, void *(*function_p)(void*), void* arg_p);
extern int thpool_add_work(thpool, void *(*function)(void*), void* arg_p);
/**
@ -159,7 +59,7 @@ int thpool_add_work(thpool_t* thpool, void *(*function_p)(void*), void* arg_p);
* @param threadpool to wait for
* @return nothing
*/
void thpool_wait(thpool_t* thpool);
extern void thpool_wait(thpool);
/**
@ -174,7 +74,7 @@ void thpool_wait(thpool_t* thpool);
* @param threadpool where the threads should be paused
* @return nothing
*/
void thpool_pause(thpool_t* thpool);
extern void thpool_pause(thpool);
/**
@ -183,175 +83,21 @@ void thpool_pause(thpool_t* thpool);
* @param threadpool where the threads should be unpaused
* @return nothing
*/
void thpool_resume(thpool_t* thpool);
extern void thpool_resume(thpool);
/**
* @brief Destroy the threadpool
*
* This will 'kill' the threadpool and free up memory. If threads are active when this
* is called, they will finish what they are doing and then they will get destroyied.
* This will wait for the currently active threads to finish and then 'kill'
* the whole threadpool to free up memory.
*
* @param threadpool a pointer to the threadpool structure you want to destroy
* @return nothing
*/
void thpool_destroy(thpool_t* thpool);
extern void thpool_destroy(thpool);
/* ----------------------- Thread specific --------------------------- */
/**
* @brief Initialize a thread in the thread pool
*
* Will initialize a new thread for the given threadpool and give the
* the thread an ID
*
* Notice also that the thread's id is not populated automatically.
*
* @param threadpool threadpool to create thread
* @param thread address to the pointer of the thread to be created
* @param id id to be given to thread
*
*/
static void thread_init(thpool_t* thpool, thread_t** thread, int id);
/**
* @brief What each thread is doing
*
* In principle this is an endless loop. The only time this loop gets interuppted is once
* thpool_destroy() is invoked.
*
* @param threadpool to use
* @return nothing
*/
static void* thread_do(thread_t* thread);
/**
* @brief Sets the calling thread on hold until threads_on_hold is set to 1
* @param nothing
* @return nothing
*/
static void thread_hold();
/**
* @brief Frees a thread
* @param thread
* @return nothing
*/
static void thread_destroy(thread_t* thread);
/* ----------------------- Queue specific --------------------------- */
/**
* @brief Initialize queue
* @param pointer to threadpool
* @return 0 on success,
* -1 on memory allocation error
*/
static int jobqueue_init(thpool_t* thpool);
/**
* @brief Clears the queue
* @param pointer to threadpool
* @return nothing
*/
static void jobqueue_clear(thpool_t* thpool);
/**
* @brief Add job to queue
*
* A new job will be added to the queue. The new job MUST be allocated
* before passed to this function or else other functions like jobqueue_empty()
* will be broken.
*
* MUST HOLD MUTEX WHEN CALLING
*
* @param pointer to threadpool
* @param pointer to the new job(MUST BE ALLOCATED)
* @return nothing
*/
static void jobqueue_push(thpool_t* thpool, job_t* newjob);
/**
* @brief Get first job from queue(removes it from queue)
*
* This does not free allocated memory so be sure to have peeked() \n
* before invoking this as else there will result lost memory pointers.
*
* MUST HOLD MUTEX WHEN CALLING
*
* @param pointer to threadpool
* @return point to job on success,
* NULL if there is no job in queue
*/
static job_t* jobqueue_pull(thpool_t* thpool);
/**
* @brief Remove and deallocate all jobs in queue
*
* This function will deallocate all jobs in the queue and set the
* jobqueue to its initialization values, thus tail and head pointing
* to NULL and amount of jobs equal to 0.
*
* @param pointer to threadpool structure
* */
static void jobqueue_destroy(thpool_t* thpool);
/* ----------------------- Synchronisation -------------------------- */
/**
* @brief Inits semaphore to given value
* @param binary semaphore
* @param value 1 or 0
* */
static void bsem_init(bsem_t *bsem, int value);
/**
* @brief Resets semaphore to 0
* @param binary semaphore
* */
static void bsem_reset(bsem_t *bsem);
/**
* @brief Sets semaphore to one and notifies at least one thread
* @param binary semaphore
* */
static void bsem_post(bsem_t *bsem);
/**
* @brief Sets semaphore to one and notifies all threads
* @param binary semaphore
* */
static void bsem_post_all(bsem_t *bsem);
/**
* @brief Waits on semaphore until semaphore has value 0
* @param binary semaphore
* */
static void bsem_wait(bsem_t *bsem);
#endif