tpool.c

/* 
 * 
 * Copyright (c) Year(s), 2013, Luis M. S. Russo and Alexandre 
 * P. Francisco / KDBIO / INESC-ID, <qhv@kdbio.inesc-id.pt> 
 *  
 * Any published media that is related with to use of the distributed 
 * software, or derived software, must contain a reference to "Extending 
 * quick hypervolume. Luís M. S. Russo, Alexandre P. Francisco: 
 * J. Heuristics 22(3): 245-271 (2016)".
 *  
 * Permission to use, copy, modify, and/or distribute this software for 
 * any purpose with or without fee is hereby granted, provided that the 
 * above copyright notice and this permission notice appear in all 
 * copies. 
 *  
 * THE SOFTWARE IS PROVIDED "AS IS" AND THE AUTHOR DISCLAIMS ALL 
 * WARRANTIES WITH REGARD TO THIS SOFTWARE INCLUDING ALL IMPLIED 
 * WARRANTIES OF MERCHANTABILITY AND FITNESS. IN NO EVENT SHALL THE 
 * AUTHOR BE LIABLE FOR ANY SPECIAL, DIRECT, INDIRECT, OR CONSEQUENTIAL 
 * DAMAGES OR ANY DAMAGES WHATSOEVER RESULTING FROM LOSS OF USE, DATA OR 
 * PROFITS, WHETHER IN AN ACTION OF CONTRACT, NEGLIGENCE OR OTHER 
 * TORTIOUS ACTION, ARISING OUT OF OR IN CONNECTION WITH THE USE OR 
 * PERFORMANCE OF THIS SOFTWARE. 
 * 
 */

#include <stdlib.h>
#include <pthread.h>
#include <stdio.h>
#include <assert.h>
#include <stdio.h>
#include <stdlib.h>
#include <sys/types.h>  /* basic system data types */
#include <limits.h>     /* PIPE_BUF */
#include <string.h>

#include "tpool.h"

/* This is the maximal number of cores in PC */
#ifndef CORES
#define CORES 64
#endif /* CORES */

/* Number of dimensions import from existing code */

/* Maximum number of simultaneous jobs */
#define SIMJs (CORES+1)
/* Maximum number of simultaneous tasks */
/* The last SIMJs, guarantees a task per job. */
#define SIMTs (4*SIMJs)

struct task {
  toolBox* b;                   
  int jid;                      
  taskData d;
};

typedef struct task* task;

enum Qstate  {active, stopped};

struct fns{
  void  (*run)(toolBox* b, taskData* d, const jobData* jd);
  void  (*openBox)(toolBox* b);
  void  (*closeBox)(toolBox* b);
  int   (*priority)(taskData* d);     
  void  (*jobCopy)(jobData* des, jobData* src);
  void  (*taskCopy)(taskData* dst, taskData* src);
  void  (*merge)(jobData* dst, taskData* src);
  void  (*publish)(int jid, jobData* jd);
};

struct jobQueue{
  pthread_mutex_t* mutex;       
  pthread_cond_t cond;          
  int head;                     
  int tail;                     
  int C[SIMJs];                 
  jobData R[SIMJs];             
};

struct taskQueue {
  pthread_mutex_t mutex;        
  pthread_cond_t cond;          
  pthread_cond_t mem;           
  int waiters;                  
  int minGrade;                 
  enum Qstate s;                
  int memDrop;                  
  int memGet;                   
  int head;                     
  int tail;                     
  int heavyTail;                
  task A[SIMTs];                
};

static struct fns F;
static struct taskQueue Q = {
  PTHREAD_MUTEX_INITIALIZER,
  PTHREAD_COND_INITIALIZER,
  PTHREAD_COND_INITIALIZER
};
static struct jobQueue Jq = {
  &(Q.mutex),
  PTHREAD_COND_INITIALIZER
};

/* An array of tool boxes, one for each worker. */
static toolBox toodles[CORES];

/* static void printState(void); */


static int activeQ(void);

/* #define  pthread_mutex_lock(A)  pthread_mutex_lock(A); printf("mutex Locked\n"); printState(); */
/* #define  pthread_mutex_unlock(A)  pthread_mutex_unlock(A); printf("mutex UnLocked\n") */

static void halt(void)
{
  int actv;

  pthread_mutex_lock(&Q.mutex);
  Q.s = stopped;
  actv = !activeQ();
  pthread_mutex_unlock(&Q.mutex);

  if(actv)
    pthread_cond_broadcast(&Q.cond); /* wake sleeping threads. */
}


static int resources(void)
{
  return ((Q.memGet - Q.heavyTail)*4)/SIMTs;
}

static int tasksN(void)
{
  return Q.head - Q.tail;
}

static int jobsN(void)
{
  return Jq.head - Jq.tail;
}

static int activeQ(void)
{
  return (active == Q.s) || (Jq.tail < Jq.head);
}

/* static void printState(void) */
/* { */
/*   printf("Q info: %d jobs, %d tasks, %d waiters\n", jobsN(), tasksN(), Q.waiters ); */
/* } */

static task pop(void)
{
  task t;

  assert(tasksN() > 0);

  t = Q.A[Q.tail % SIMTs];
  Q.A[Q.tail % SIMTs] = NULL;
  Q.tail++;

  return t;
}

static int submit(int jid, taskData* d)
{
  int signal;

  assert(Jq.tail <= jid && jid <= Jq.head);

  (*F.merge)(&(Jq.R[jid % SIMJs]), d);
  /* Jq.R[jid % SIMJs]+=r; */
  Jq.C[jid % SIMJs]--;          
  signal = (jobsN() == SIMJs && 0 == Jq.C[Jq.tail % SIMJs]);

  while(jobsN() > 0 && 0 == Jq.C[Jq.tail % SIMJs])
    {                           
      (*F.publish)(Jq.tail, &(Jq.R[Jq.tail % SIMJs]));
      /* Do not clean up, jobs are initialized */
      Jq.tail++;
    }

  return signal;
}

static int recycle(task t)
{
  int signal;

  signal = (resources() == 4);  
  Q.A[Q.memDrop % SIMTs] = t;
  Q.memDrop++;
  Q.heavyTail++;                 
  assert(Q.heavyTail <= Q.tail);

  return signal;
}

static task queryTs(int p)
{
  task t;
  
  t = NULL;

  /* printf("Admission criteria: Q.minGrade %d < p %d\n", Q.minGrade, p); */

  if(Q.minGrade < p)
    {    /* Hurray, you passed task admission */
      Q.minGrade += resources();   /* Complain about lack of resources */
      if(resources() < 4)         
        { /* Huff, you still got the resource. */
          t = Q.A[Q.memGet % SIMTs];
          Q.memGet++;

          if(Q.memGet > Q.memDrop) /* But need to alloc it. */
            {
              Q.memDrop++;
              t = (task) calloc(1, sizeof(struct task));
            }

          assert(Q.memGet <= Q.memDrop);
        }
    }
  else /* We regret to inform you ... bla bla ... no task admission. */
    Q.minGrade += resources() - 3;   /* Complain about waisted resources */

  return t;
}

static int post(int pri, int jid, taskData* d)
{
  int r;
  task t;
  int signal;
  int p;                        
  /* printf("Trying to post task\n"); */
  /* printState(); */

  signal = 0;

  p = (*F.priority)(d);
  if(0 == pri)
    p = INT_MAX;
  t = queryTs(p);

  if(0 == pri)
    {
      while(t == NULL)
        { /* LOOP until there is room to post task */
          pthread_cond_wait(&Q.mem, &Q.mutex);
          t = queryTs(p);
        }
    }

  r = (t != NULL);
  if(r)
    {                   
      /* printf("Success post\n"); */
      t->jid = jid;
      Jq.C[t->jid % SIMJs]++;         
      pthread_mutex_unlock(&Q.mutex); 
      (*F.taskCopy)(&(t->d), d);
      t->b = NULL;              
      pthread_mutex_lock(&Q.mutex);
      signal = (activeQ() && Q.waiters > 0); 
      Q.A[Q.head % SIMTs] = t;      
      Q.head++;      

      assert(Q.head <= Q.memGet);
      assert(Q.memGet <= Q.memDrop);
    }
  pthread_mutex_unlock(&Q.mutex); /* Also releases by default */

  if(signal)
    pthread_cond_signal(&Q.cond); /* wake sleeping threads. */

  return r;
}

void postJob(jobData* jd, taskData* d)
{
  int jid;

  pthread_mutex_lock(&Q.mutex);

  while(jobsN() == SIMJs) /* Job queue is full */
    pthread_cond_wait(&Jq.cond, Jq.mutex);

  jid = Jq.head;             
  Jq.head++;                 /* Clean space is available, just update counter */

  (*F.jobCopy)(&(Jq.R[jid % SIMJs]), jd);

  post(0, jid, d);            
}

void prun(taskData* d, int p)
{
  task t;

  t = (task) container_of(d, struct task, d); /* The d's mean diferent things. */

  /* Assuming taskData actually is in a task */
  if(p)
    { /* Try to Parallelize the computation */
      pthread_mutex_lock(&Q.mutex);
      if(!post(1, t->jid, d)) /* Not crucial, use task priority */
        (*(F.run))(t->b, d, &(Jq.R[t->jid % SIMJs]));  
    }
  else
    (*(F.run))(t->b, d, &(Jq.R[t->jid % SIMJs]));  
}

void* worker(void *threadid)
{
  task t;
  int tid;
  int signalMem;
  int signalJob;
  int signalDismiss;
  toolBox* b;

  tid = *((int*)threadid);

  b = &(toodles[tid-1]);
  (*F.openBox)(b);             
  do
    {
      signalJob = 0;
      signalMem = 0;
      signalDismiss = 0;

      pthread_mutex_lock(&Q.mutex);
      while(activeQ() && 0 == tasksN())      
        {
          Q.waiters++;
          pthread_cond_wait(&Q.cond, &Q.mutex);
          Q.waiters--;
          /* printf("Got a Job\n"); */
          /* printState(); */
        }

      if(activeQ())
        { /* Otherwise stop worker */
          t = pop();                
          pthread_mutex_unlock(&Q.mutex);
          t->b = b;
          (*F.run)(t->b, &(t->d), &(Jq.R[t->jid % SIMJs]));  
          pthread_mutex_lock(&Q.mutex);
          signalJob = submit(t->jid, &(t->d));  
          signalMem = recycle(t);               
          signalDismiss = !activeQ();
        }
      pthread_mutex_unlock(&Q.mutex);

      /* Signal events */
      if(signalJob)
        pthread_cond_signal(&Jq.cond); /* wake sleeping threads. */        
      if(signalMem)
        pthread_cond_signal(&Q.mem);   /* wake sleeping threads. */        
      if(signalDismiss)
        pthread_cond_broadcast(&Q.cond); /* wake sleeping threads. */
    }
  while(activeQ());

  (*F.closeBox)(b);             
  pthread_exit(NULL);
}

void launch(
            int w,
            void* (*console)(void *threadid),
            void  (*run)(toolBox* b, taskData* d, const jobData* jd), 
            void  (*openBox)(toolBox* b),
            void  (*closeBox)(toolBox* b),
            int   (*priority)(taskData* d),
            void  (*jobCopy)(jobData* des, jobData* src),
            void  (*taskCopy)(taskData* dst, taskData* src),
            void  (*merge)(jobData* dst, taskData* src),
            void  (*publish)(int jid, jobData* jd)
            )
{
  int i;
  pthread_t wths[CORES];        
  pthread_t cthread;            
  int tids[CORES];

  F.run = run;
  F.openBox = openBox;
  F.closeBox = closeBox;
  F.priority = priority;
  F.jobCopy = jobCopy;
  F.taskCopy = taskCopy;
  F.merge = merge;
  F.publish = publish;

  if(CORES < w)
    w = CORES;
  if(1 > w)
    w = 1;

  /* start all the worker threads */
  printf("Using %d threads\n", w);
  /* toodles = (toolBox*) calloc(w, sizeof(toolBox)); */
  memset(toodles, 0x0, w*sizeof(toolBox));

  i = 0;
  while(i < w)
    {
      tids[i] = i+1;
      pthread_create(&wths[i], NULL, worker, &tids[i]);
      i++;
    }

  /* Create console only after you have the workers */

  pthread_create(&cthread, NULL, console, NULL);
  pthread_join(cthread, NULL);  
  halt();

  /* Afther console finishes wait for workers. Forced to finish by halt. */

  i = 0;
  while(i < w)
    {
      pthread_join(wths[i], NULL);
      i++;
    }
  
  i = 0;
  while(i < SIMTs)
    {
      if(NULL != Q.A[i])
        free(Q.A[i]);
      i++;
    }

  /* free(toodles); */
}