tpool.c

Go to the documentation of this file.

/* ********************************
 * Author:       Johan Hanssen Seferidis
 * License:      MIT
 * Description:  Library providing a threading pool where you can add
 *               work. For usage, check the tpool.h file or README.md
 *
 *//*
 *
 ********************************/

#define _POSIX_C_SOURCE 200809L
#include <unistd.h>
#include <signal.h>
#include <stdio.h>
#include <stdlib.h>
#if defined(__APPLE__) && defined(__MACH__)
#define _DARWIN_C_SOURCE 1
#endif
#include <pthread.h>
#include <time.h>
#if defined(__linux__)
#include <sys/prctl.h>
#endif

#include "libsheepyObject.h" // for finalizeLibsheepyRecycleContainers
#include "tpool.h"

#ifdef tpool_DEBUG
#define tpool_DEBUG 1
#else
#define tpool_DEBUG 0
#endif

#if !defined(DISABLE_PRINT) || defined(tpool_DEBUG)
#define err(str) logE(str)
#else
#define err(str)
#endif

static volatile int threads_keepalive;
static volatile int threads_on_hold;


// locks, max 512 threads
pthread_mutex_t tpoolLocks[tpoolLockCount]; // only one thread can have the lock

threadpool tpool;

/* ========================== STRUCTURES ============================ */


/* Binary semaphore */
typedef struct bsem {
  pthread_mutex_t mutex;
  pthread_cond_t   cond;
  int v;
} bsem;


/* Job */
typedef struct job{
  struct job*  prev;                   /* pointer to previous job   */
  void   (*function)(void* arg);       /* function pointer          */
  void*  arg;                          /* function's argument       */
} job;


/* Job queue */
typedef struct jobqueue{
  pthread_mutex_t rwmutex;             /* used for queue r/w access */
  job  *front;                         /* pointer to front of queue */
  job  *rear;                          /* pointer to rear  of queue */
  bsem *has_jobs;                      /* flag as binary semaphore  */
  int   len;                           /* number of jobs in queue   */
} jobqueue;


/* Thread */
typedef struct thread{
  int       id;                        /* friendly id               */
  pthread_t pthread;                   /* pointer to actual thread  */
  threadpool tpool_p;            /* access to tpool          */
} thread;


/* Threadpool */
typedef struct tpool_{
  thread**   threads;                  /* pointer to threads        */
  volatile int num_threads_alive;      /* threads currently alive   */
  volatile int num_threads_working;    /* threads currently working */
  pthread_mutex_t  thcount_lock;       /* used for thread count etc */
  pthread_cond_t  threads_all_idle;    /* signal to tpool_wait     */
  jobqueue  jobqueue;                  /* job queue                 */
} tpool_;





/* ========================== PROTOTYPES ============================ */


static int  thread_init(threadpool tpool_p, struct thread** thread_p, int id);
static void* thread_do(struct thread* thread_p);
static void  thread_hold(int sig_id);
static void  thread_destroy(struct thread* thread_p);

static int   jobqueue_init(jobqueue* jobqueue_p);
static void  jobqueue_clear(jobqueue* jobqueue_p);
static void  jobqueue_push(jobqueue* jobqueue_p, struct job* newjob_p);
static struct job* jobqueue_pull(jobqueue* jobqueue_p);
static void  jobqueue_destroy(jobqueue* jobqueue_p);

static void  bsem_init(struct bsem *bsem_p, int value);
static void  bsem_reset(struct bsem *bsem_p);
static void  bsem_post(struct bsem *bsem_p);
static void  bsem_post_all(struct bsem *bsem_p);
static void  bsem_wait(struct bsem *bsem_p);





/* ========================== THREADPOOL ============================ */

// use lock in the tpoolLocks array
void tpoolLock(int position) {
  pthread_mutex_lock(&tpoolLocks[position]);
}

// unlock a lock in the tpoolLocks array
void tpoolUlock(int position) {
  pthread_mutex_unlock(&tpoolLocks[position]);
}

/* Initialise thread pool */
threadpool tpool_init(int num_threads){

  threads_on_hold   = 0;
  threads_keepalive = 1;

  if (num_threads < 0){
    num_threads = 0;
  }

  /* Make new thread pool */
  threadpool tpool_p;
  tpool_p = (threadpool)malloc(sizeof(struct tpool_));
  if (tpool_p == NULL){
    err("tpool_init(): Could not allocate memory for thread pool\n");
    return NULL;
  }
  tpool_p->num_threads_alive   = 0;
  tpool_p->num_threads_working = 0;

  /* Initialise the job queue */
  if (jobqueue_init(&tpool_p->jobqueue) == -1){
    err("tpool_init(): Could not allocate memory for job queue\n");
    free(tpool_p);
    return NULL;
  }

  /* Make threads in pool */
  tpool_p->threads = (struct thread**)malloc((size_t)num_threads * sizeof(struct thread *));
  if (tpool_p->threads == NULL){
    err("tpool_init(): Could not allocate memory for threads\n");
    jobqueue_destroy(&tpool_p->jobqueue);
    free(tpool_p);
    return NULL;
  }

  pthread_mutex_init(&(tpool_p->thcount_lock), NULL);
  pthread_cond_init(&tpool_p->threads_all_idle, NULL);

  /* Thread init */
  int n;
  for (n=0; n<num_threads; n++){
    thread_init(tpool_p, &tpool_p->threads[n], n);
#if tpool_DEBUG
      printf("tpool_DEBUG: Created thread %d in pool \n", n);
#endif
  }

  /* Wait for threads to initialize */
  while (tpool_p->num_threads_alive != num_threads) {}

  return tpool_p;
}


/* Add work to the thread pool */
int tpool_add_work(threadpool tpool_p, void (*function_p)(void*), void* arg_p){
  job* newjob;

  if (!tpool_p){
    err("tpool_add_work(): tpool_p is NULL\n");
    return -1;
  }
  if (!function_p){
    err("tpool_add_work(): function_p is NULL\n");
    return -1;
  }
  newjob=(struct job*)malloc(sizeof(struct job));
  if (newjob==NULL){
    err("tpool_add_work(): Could not allocate memory for new job\n");
    return -1;
  }

  /* add function and argument */
  newjob->function=function_p;
  newjob->arg=arg_p;

  /* add job to queue */
  jobqueue_push(&tpool_p->jobqueue, newjob);

  return 0;
}


/* Wait until all jobs have finished */
void tpool_wait(threadpool tpool_p){
  if (!tpool_p) return;
  pthread_mutex_lock(&tpool_p->thcount_lock);
  while (tpool_p->jobqueue.len || tpool_p->num_threads_working) {
    pthread_cond_wait(&tpool_p->threads_all_idle, &tpool_p->thcount_lock);
  }
  pthread_mutex_unlock(&tpool_p->thcount_lock);
}


/* Destroy the threadpool */
void tpool_destroy(threadpool tpool_p){
  /* No need to destory if it's NULL */
  if (!tpool_p) return;

  volatile int threads_total = tpool_p->num_threads_alive;

  /* End each thread 's infinite loop */
  threads_keepalive = 0;

  /* Give one second to kill idle threads */
  double TIMEOUT = 1.0;
  time_t start, end;
  double tpassed = 0.0;
  time (&start);
  while (tpassed < TIMEOUT && tpool_p->num_threads_alive){
    bsem_post_all(tpool_p->jobqueue.has_jobs);
    time (&end);
    tpassed = difftime(end,start);
  }

  /* Poll remaining threads */
  while (tpool_p->num_threads_alive){
    bsem_post_all(tpool_p->jobqueue.has_jobs);
    sleep(1);
  }

  /* Job queue cleanup */
  jobqueue_destroy(&tpool_p->jobqueue);
  int n;
  /* wait until all threads are finished because when recycleContainers is enabled
   * all threads must finish before running dArrayFree in finalizeLibsheepyAtExit
   * in the main process */
  for (n=0; n < threads_total; n++){
    /* ignore error code from pthread_join because the threads might already be finished at this time */
    int result_code;
    result_code = pthread_join(tpool_p->threads[n]->pthread, NULL);
    if (result_code == EINVAL) {
      logE("not a joinable thread or another thread is already waiting to join with this thread.");
    }
  }
  /* Deallocs */
  for (n=0; n < threads_total; n++){
    thread_destroy(tpool_p->threads[n]);
  }
  free(tpool_p->threads);
  free(tpool_p);
}


/* Pause all threads in threadpool */
void tpool_pause(threadpool tpool_p) {
  if (!tpool_p) return;
  int n;
  for (n=0; n < tpool_p->num_threads_alive; n++){
    pthread_kill(tpool_p->threads[n]->pthread, SIGUSR2);
  }
}


/* Resume all threads in threadpool */
void tpool_resume(threadpool tpool_p) {
  if (!tpool_p) return;

    // resuming a single threadpool hasn't been
    // implemented yet, meanwhile this supresses
    // the warnings
    (void)tpool_p;

  threads_on_hold = 0;
}


int tpool_num_threads_working(threadpool tpool_p){
  if (!tpool_p) return -1;
  return tpool_p->num_threads_working;
}





/* ============================ THREAD ============================== */


/* Initialize a thread in the thread pool
 *
 * @param thread        address to the pointer of the thread to be created
 * @param id            id to be given to the thread
 * @return 0 on success, -1 otherwise.
 */
static int thread_init (threadpool tpool_p, struct thread** thread_p, int id){

  *thread_p = (struct thread*)malloc(sizeof(struct thread));
  if (thread_p == NULL){
    err("thread_init(): Could not allocate memory for thread\n");
    return -1;
  }

  (*thread_p)->tpool_p = tpool_p;
  (*thread_p)->id       = id;

  pthread_create(&(*thread_p)->pthread, NULL, (void *)thread_do, (*thread_p));
  // keep thread attached to be able to join them in tpool_destroy
  // all threads must finish before running dArrayFree in finalizeLibsheepyAtExit
  /* pthread_detach((*thread_p)->pthread); */
  return 0;
}


/* Sets the calling thread on hold */
static void thread_hold(int sig_id) {
    (void)sig_id;
  threads_on_hold = 1;
  while (threads_on_hold){
    sleep(1);
  }
}


/* What each thread is doing
*
* In principle this is an endless loop. The only time this loop gets interuppted is once
* tpool_destroy() is invoked or the program exits.
*
* @param  thread        thread that will run this function
* @return nothing
*/
static void* thread_do(struct thread* thread_p){

  /* Set thread name for profiling and debuging */
  char thread_name[128] = {0};
  sprintf(thread_name, "thread-pool-%d", thread_p->id);

#if defined(__linux__)
  /* Use prctl instead to prevent using _GNU_SOURCE flag and implicit declaration */
  prctl(PR_SET_NAME, thread_name);
#elif defined(__APPLE__) && defined(__MACH__)
  pthread_setname_np(thread_name);
#else
  #warning "thread_do(): pthread_setname_np is not supported on this system"
#endif

  /* Assure all threads have been created before starting serving */
  threadpool tpool_p = thread_p->tpool_p;

  /* Register signal handler */
  struct sigaction act;
  sigemptyset(&act.sa_mask);
  act.sa_flags = 0;
  act.sa_handler = thread_hold;
  if (sigaction(SIGUSR2, &act, NULL) == -1) {
    err("thread_do(): cannot handle SIGUSR2");
  }

  #if recycleContainers
  /* the containers are recycled in the thread local storage, the unused containers are freed at thread exit */
  pthread_key_t key;
  if (pthread_key_create(&key, finalizeLibsheepyRecycleContainers)) {
    err("pthread_key_create, finalizeLibsheepyRecycleContainers not registered, there will be memory leaks when the thread exits");
  }
  else {
    // set value to call the destructor at exit
    pthread_setspecific(key, (const void *) 1);
  }
  #endif // recycleContainers

  /* Mark thread as alive (initialized) */
  pthread_mutex_lock(&tpool_p->thcount_lock);
  tpool_p->num_threads_alive += 1;
  pthread_mutex_unlock(&tpool_p->thcount_lock);

  while(threads_keepalive){

    bsem_wait(tpool_p->jobqueue.has_jobs);

    if (threads_keepalive){

      pthread_mutex_lock(&tpool_p->thcount_lock);
      tpool_p->num_threads_working++;
      pthread_mutex_unlock(&tpool_p->thcount_lock);

      /* Read job from queue and execute it */
      void (*func_buff)(void*);
      void*  arg_buff;
      job* job_p = jobqueue_pull(&tpool_p->jobqueue);
      if (job_p) {
        func_buff = job_p->function;
        arg_buff  = job_p->arg;
        func_buff(arg_buff);
        free(job_p);
      }

      pthread_mutex_lock(&tpool_p->thcount_lock);
      tpool_p->num_threads_working--;
      if (!tpool_p->num_threads_working) {
        pthread_cond_signal(&tpool_p->threads_all_idle);
      }
      pthread_mutex_unlock(&tpool_p->thcount_lock);

    }
  }
  pthread_mutex_lock(&tpool_p->thcount_lock);
  tpool_p->num_threads_alive --;
  pthread_mutex_unlock(&tpool_p->thcount_lock);

  return NULL;
}


/* Frees a thread  */
static void thread_destroy (thread* thread_p){
  free(thread_p);
}





/* ============================ JOB QUEUE =========================== */


/* Initialize queue */
static int jobqueue_init(jobqueue* jobqueue_p){
  jobqueue_p->len = 0;
  jobqueue_p->front = NULL;
  jobqueue_p->rear  = NULL;

  jobqueue_p->has_jobs = (struct bsem*)malloc(sizeof(struct bsem));
  if (jobqueue_p->has_jobs == NULL){
    return -1;
  }

  pthread_mutex_init(&(jobqueue_p->rwmutex), NULL);
  bsem_init(jobqueue_p->has_jobs, 0);

  return 0;
}


/* Clear the queue */
static void jobqueue_clear(jobqueue* jobqueue_p){

  while(jobqueue_p->len){
    free(jobqueue_pull(jobqueue_p));
  }

  jobqueue_p->front = NULL;
  jobqueue_p->rear  = NULL;
  bsem_reset(jobqueue_p->has_jobs);
  jobqueue_p->len = 0;

}


/* Add (allocated) job to queue
 */
static void jobqueue_push(jobqueue* jobqueue_p, struct job* newjob){

  pthread_mutex_lock(&jobqueue_p->rwmutex);
  newjob->prev = NULL;

  switch(jobqueue_p->len){

    case 0:  /* if no jobs in queue */
          jobqueue_p->front = newjob;
          jobqueue_p->rear  = newjob;
          break;

    default: /* if jobs in queue */
          jobqueue_p->rear->prev = newjob;
          jobqueue_p->rear = newjob;

  }
  jobqueue_p->len++;

  bsem_post(jobqueue_p->has_jobs);
  pthread_mutex_unlock(&jobqueue_p->rwmutex);
}


/* Get first job from queue(removes it from queue)
 *
 * Notice: Caller MUST hold a mutex
 */
static struct job* jobqueue_pull(jobqueue* jobqueue_p){

  pthread_mutex_lock(&jobqueue_p->rwmutex);
  job* job_p = jobqueue_p->front;

  switch(jobqueue_p->len){

    case 0:  /* if no jobs in queue */
            break;

    case 1:  /* if one job in queue */
          jobqueue_p->front = NULL;
          jobqueue_p->rear  = NULL;
          jobqueue_p->len = 0;
          break;

    default: /* if >1 jobs in queue */
          jobqueue_p->front = job_p->prev;
          jobqueue_p->len--;
          /* more than one job in queue -> post it */
          bsem_post(jobqueue_p->has_jobs);

  }

  pthread_mutex_unlock(&jobqueue_p->rwmutex);
  return job_p;
}


/* Free all queue resources back to the system */
static void jobqueue_destroy(jobqueue* jobqueue_p){
  jobqueue_clear(jobqueue_p);
  free(jobqueue_p->has_jobs);
}





/* ======================== SYNCHRONISATION ========================= */


/* Init semaphore to 1 or 0 */
static void bsem_init(bsem *bsem_p, int value) {
  if (value < 0 || value > 1) {
    err("bsem_init(): Binary semaphore can take only values 1 or 0");
    exit(1);
  }
  pthread_mutex_init(&(bsem_p->mutex), NULL);
  pthread_cond_init(&(bsem_p->cond), NULL);
  bsem_p->v = value;
}


/* Reset semaphore to 0 */
static void bsem_reset(bsem *bsem_p) {
  bsem_init(bsem_p, 0);
}


/* Post to at least one thread */
static void bsem_post(bsem *bsem_p) {
  pthread_mutex_lock(&bsem_p->mutex);
  bsem_p->v = 1;
  pthread_cond_signal(&bsem_p->cond);
  pthread_mutex_unlock(&bsem_p->mutex);
}


/* Post to all threads */
static void bsem_post_all(bsem *bsem_p) {
  pthread_mutex_lock(&bsem_p->mutex);
  bsem_p->v = 1;
  pthread_cond_broadcast(&bsem_p->cond);
  pthread_mutex_unlock(&bsem_p->mutex);
}


/* Wait on semaphore until semaphore has value 0 */
static void bsem_wait(bsem* bsem_p) {
  pthread_mutex_lock(&bsem_p->mutex);
  while (bsem_p->v != 1) {
    pthread_cond_wait(&bsem_p->cond, &bsem_p->mutex);
  }
  bsem_p->v = 0;
  pthread_mutex_unlock(&bsem_p->mutex);
}