thread pool

common/utils/threadPool/Makefile

+all: measurement_display thread-pool-test
+
+measurement_display: measurement_display.c thread-pool.h
+	gcc measurement_display.c -I /data/openairinterface5g.nr/common/utils/  -I. /data/openairinterface5g.nr/common/utils/backtrace.c -lpthread -D TEST_THREAD_POOL -I../LOG -I../../utils/T -o measurement_display
+
+thread-pool-test: thread-pool.c thread-pool.h
+	gcc -g thread-pool.c -I /data/openairinterface5g.nr/common/utils/  -I. /data/openairinterface5g.nr/common/utils/backtrace.c -lpthread -D TEST_THREAD_POOL -I../LOG -I../../utils/T -o thread-pool-test
+

common/utils/threadPool/measurement_display.c

+#define __USE_GNU
+#define _GNU_SOURCE
+#include <stdio.h>
+#include <pthread.h>
+#include <sched.h>
+#include <stdlib.h>
+#include <unistd.h>
+#include <sys/types.h>
+#include <string.h>
+#include <unistd.h>
+#include <sys/syscall.h>
+#include <sys/time.h>
+#include <stdint.h>
+#include <sys/stat.h>
+#include <fcntl.h>
+#include "thread-pool.h"
+
+#define SEP "\t"
+
+uint64_t cpuCyclesMicroSec;
+
+int main(int argc, char *argv[]) {
+  if(argc != 2) {
+    printf("Need one paramter: the trace Linux pipe (fifo)");
+    exit(1);
+  }
+
+  mkfifo(argv[1],0666);
+  int fd=open(argv[1], O_RDONLY);
+
+  if ( fd == -1 ) {
+    perror("open read mode trace file:");
+    exit(1);
+  }
+
+  uint64_t deb=rdtsc();
+  usleep(100000);
+  cpuCyclesMicroSec=(rdtsc()-deb)/100000;
+  printf("Cycles per µs: %lu\n",cpuCyclesMicroSec);
+  printf("Key" SEP "delay to process" SEP "processing time" SEP "delay to be read answer\n");
+  notifiedFIFO_elt_t doneRequest;
+
+  while ( 1 ) {
+    if ( read(fd,&doneRequest, sizeof(doneRequest)) ==  sizeof(doneRequest)) {
+      printf("%lu" SEP "%lu" SEP "%lu" SEP "%lu" "\n",
+             doneRequest.key,
+             (doneRequest.startProcessingTime-doneRequest.creationTime)/cpuCyclesMicroSec,
+             (doneRequest.endProcessingTime-doneRequest.startProcessingTime)/cpuCyclesMicroSec,
+             (doneRequest.returnTime-doneRequest.endProcessingTime)/cpuCyclesMicroSec
+            );
+    } else {
+      printf("no measurements\n");
+      sleep(1);
+    }
+  }
+}

common/utils/threadPool/thread-pool.c

+#define _GNU_SOURCE
+#include <sched.h>
+#include <sys/types.h>
+#include <sys/stat.h>
+#include <fcntl.h>
+#include <string.h>
+#include <unistd.h>
+#include <thread-pool.h>
+
+void displayList(notifiedFIFO_t *nf) {
+  int n=0;
+  notifiedFIFO_elt_t *ptr=nf->outF;
+
+  while(ptr) {
+    printf("element: %d, key: %lu\n",++n,ptr->key);
+    ptr=ptr->next;
+  }
+
+  printf("End of list: %d elements\n",n);
+}
+
+static inline  notifiedFIFO_elt_t *pullNotifiedFifoRemember( notifiedFIFO_t *nf, struct one_thread *thr) {
+  mutexlock(nf->lockF);
+
+  while(!nf->outF)
+    condwait(nf->notifF, nf->lockF);
+
+  notifiedFIFO_elt_t *ret=nf->outF;
+  nf->outF=nf->outF->next;
+
+  if (nf->outF==NULL)
+    nf->inF=NULL;
+
+  // For abort feature
+  thr->runningOnKey=ret->key;
+  thr->abortFlag=false;
+  mutexunlock(nf->lockF);
+  return ret;
+}
+
+void *one_thread(void *arg) {
+  struct  one_thread *myThread=(struct  one_thread *) arg;
+  struct  thread_pool *tp=myThread->pool;
+  // configure the thread core assignment
+  // TBD: reserve the core for us exclusively
+  cpu_set_t cpuset;
+  CPU_ZERO(&cpuset);
+  CPU_SET(myThread->coreID, &cpuset);
+  pthread_setaffinity_np(pthread_self(), sizeof(cpu_set_t), &cpuset);
+  //Configure the thread scheduler policy for Linux
+  struct sched_param sparam= {0};
+  sparam.sched_priority = sched_get_priority_max(SCHED_RR);
+  pthread_setschedparam(pthread_self(), SCHED_RR, &sparam);
+  // set the thread name for debugging
+  sprintf(myThread->name,"Tpool_%d",myThread->coreID);
+  pthread_setname_np(pthread_self(), myThread->name );
+
+  // Infinite loop to process requests
+  do {
+    notifiedFIFO_elt_t *elt=pullNotifiedFifoRemember(&tp->incomingFifo, myThread);
+
+    if (tp->measurePerf) elt->startProcessingTime=rdtsc();
+
+    elt->processingFunc(NotifiedFifoData(elt));
+
+    if (tp->measurePerf) elt->endProcessingTime=rdtsc();
+
+    if (elt->reponseFifo) {
+      // Check if the job is still alive, else it has been aborted
+      mutexlock(tp->incomingFifo.lockF);
+
+      if (myThread->abortFlag)
+        delNotifiedFIFO_elt(elt);
+      else
+        pushNotifiedFIFO(elt->reponseFifo, elt);
+
+      mutexunlock(tp->incomingFifo.lockF);
+    }
+  } while (true);
+}
+
+void initTpool(char *params,tpool_t *pool, bool performanceMeas) {
+  memset(pool,0,sizeof(*pool));
+  char *measr=getenv("threadPoolMeasurements");
+  pool->measurePerf=performanceMeas;
+  // force measurement if the output is defined
+  pool->measurePerf=measr!=NULL;
+
+  if (measr) {
+    mkfifo(measr,0666);
+    AssertFatal(-1 != (pool->dummyTraceFd=
+                         open(measr, O_RDONLY| O_NONBLOCK)),"");
+    AssertFatal(-1 != (pool->traceFd=
+                         open(measr, O_WRONLY|O_APPEND|O_NOATIME|O_NONBLOCK)),"");
+  } else
+    pool->traceFd=-1;
+
+  //Configure the thread scheduler policy for Linux
+  struct sched_param sparam= {0};
+  sparam.sched_priority = sched_get_priority_max(SCHED_RR)-1;
+  pthread_setschedparam(pthread_self(), SCHED_RR, &sparam);
+  pool->activated=true;
+  initNotifiedFIFO(&pool->incomingFifo);
+  char *saveptr, * curptr;
+  pool->nbThreads=0;
+  pool->restrictRNTI=false;
+  curptr=strtok_r(params,",",&saveptr);
+
+  while ( curptr!=NULL ) {
+    if (curptr[0] == 'u' || curptr[0] == 'U') {
+      pool->restrictRNTI=true;
+    } else if ( curptr[0]>='0' && curptr[0]<='9' ) {
+      struct one_thread *tmp=pool->allthreads;
+      pool->allthreads=(struct one_thread *)malloc(sizeof(struct one_thread));
+      pool->allthreads->next=tmp;
+      printf("create a thread for core %d\n", atoi(curptr));
+      pool->allthreads->coreID=atoi(curptr);
+      pool->allthreads->id=pool->nbThreads;
+      pool->allthreads->pool=pool;
+      pthread_create(&pool->allthreads->threadID, NULL, one_thread, (void *)pool->allthreads);
+      pool->nbThreads++;
+    } else if (curptr[0] == 'n' || curptr[0] == 'N') {
+      pool->activated=false;
+    } else
+      printf("Error in options for thread pool: %s\n",curptr);
+
+    curptr=strtok_r(NULL,",",&saveptr);
+  }
+
+  if (pool->activated && pool->nbThreads==0) {
+    printf("No servers created in the thread pool, exit\n");
+    exit(1);
+  }
+}
+
+#ifdef TEST_THREAD_POOL
+
+struct testData {
+  int id;
+  char txt[30];
+};
+
+void processing(void *arg) {
+  struct testData *in=(struct testData *)arg;
+  printf("doing: %d, %s, in thr %ld\n",in->id, in->txt,pthread_self() );
+  sprintf(in->txt,"Done by %ld, job %d", pthread_self(), in->id);
+  usleep(rand()%100);
+  printf("done: %d, %s, in thr %ld\n",in->id, in->txt,pthread_self() );
+}
+
+int main() {
+  notifiedFIFO_t myFifo;
+  initNotifiedFIFO(&myFifo);
+  pushNotifiedFIFO(&myFifo,newNotifiedFIFO_elt(sizeof(struct testData), 1234,NULL,NULL));
+
+  for(int i=10; i>1; i--) {
+    pushNotifiedFIFO(&myFifo,newNotifiedFIFO_elt(sizeof(struct testData), 1000+i,NULL,NULL));
+  }
+
+  displayList(&myFifo);
+  notifiedFIFO_elt_t *tmp=pullNotifiedFIFO(&myFifo);
+  printf("pulled: %lu\n", tmp->key);
+  displayList(&myFifo);
+  tmp=pullNotifiedFIFO(&myFifo);
+  printf("pulled: %lu\n", tmp->key);
+  displayList(&myFifo);
+  abortNotifiedFIFO(&myFifo,1005);
+  printf("aborted 1005\n");
+  displayList(&myFifo);
+  pushNotifiedFIFO(&myFifo,newNotifiedFIFO_elt(sizeof(struct testData), 12345678, NULL, NULL));
+  displayList(&myFifo);
+  abortNotifiedFIFO(&myFifo,12345678);
+  printf("aborted 12345678\n");
+  displayList(&myFifo);
+
+  do {
+    tmp=pollNotifiedFIFO(&myFifo);
+
+    if (tmp) {
+      printf("pulled: %lu\n", tmp->key);
+      displayList(&myFifo);
+    } else
+      printf("Empty list \n");
+  } while(tmp);
+
+  tpool_t  pool;
+  char params[]="1,2,3,u";
+  initTpool(params,&pool, true);
+  notifiedFIFO_t worker_back;
+  initNotifiedFIFO(&worker_back);
+
+  for (int i=0; i <1000 ; i++) {
+    notifiedFIFO_elt_t *work=newNotifiedFIFO_elt(sizeof(struct testData), i, &worker_back, processing);
+    struct testData *x=(struct testData *)NotifiedFifoData(work);
+    x->id=i;
+    pushTpool(&pool, work);
+  }
+
+  do {
+    tmp=pullTpool(&worker_back,&pool);
+
+    if (tmp) {
+      struct testData *dd=NotifiedFifoData(tmp);
+      printf("Result: %s\n",dd->txt);
+      delNotifiedFIFO_elt(tmp);
+    } else
+      printf("Empty list \n");
+
+    abortTpool(&pool,510);
+  } while(tmp);
+
+  return 0;
+}
+#endif

common/utils/threadPool/thread-pool.h

+#ifndef THREAD_POOL_H
+#define THREAD_POOL_H
+#include <stdbool.h>
+#include <stdint.h>
+#include <pthread.h>
+#include <sys/syscall.h>
+#include <assertions.h>
+#include <log.h>
+
+#ifdef DEBUG
+  #define THREADINIT   PTHREAD_ERRORCHECK_MUTEX_INITIALIZER_NP
+#else
+  #define THREADINIT   PTHREAD_MUTEX_INITIALIZER
+#endif
+#define mutexinit(mutex)   AssertFatal(pthread_mutex_init(&mutex,NULL)==0,"");
+#define condinit(signal)   AssertFatal(pthread_cond_init(&signal,NULL)==0,"");
+#define mutexlock(mutex)   AssertFatal(pthread_mutex_lock(&mutex)==0,"");
+#define mutextrylock(mutex)   pthread_mutex_trylock(&mutex)
+#define mutexunlock(mutex) AssertFatal(pthread_mutex_unlock(&mutex)==0,"");
+#define condwait(condition, mutex) AssertFatal(pthread_cond_wait(&condition, &mutex)==0,"");
+#define condbroadcast(signal) AssertFatal(pthread_cond_broadcast(&signal)==0,"");
+#define condsignal(signal)    AssertFatal(pthread_cond_broadcast(&signal)==0,"");
+
+typedef struct notifiedFIFO_elt_s {
+  struct notifiedFIFO_elt_s *next;
+  uint64_t key; //To filter out elements
+  struct notifiedFIFO_s *reponseFifo;
+  void (*processingFunc)(void *);
+  bool malloced;
+  uint64_t creationTime;
+  uint64_t startProcessingTime;
+  uint64_t endProcessingTime;
+  uint64_t returnTime;
+  void *msgData;
+}  notifiedFIFO_elt_t;
+
+typedef struct notifiedFIFO_s {
+  notifiedFIFO_elt_t *outF;
+  notifiedFIFO_elt_t *inF;
+  pthread_mutex_t lockF;
+  pthread_cond_t  notifF;
+} notifiedFIFO_t;
+
+// You can use this allocator or use any piece of memory
+static inline notifiedFIFO_elt_t *newNotifiedFIFO_elt(int size,
+    uint64_t key,
+    notifiedFIFO_t *reponseFifo,
+    void (*processingFunc)(void *)) {
+  notifiedFIFO_elt_t *ret;
+  size_t sz=sizeof(notifiedFIFO_elt_t)+size;
+  AssertFatal( NULL != (ret=(notifiedFIFO_elt_t *) malloc((sz/32+1)*32)), "");
+  ret->next=NULL;
+  ret->key=key;
+  ret->reponseFifo=reponseFifo;
+  ret->processingFunc=processingFunc;
+  // We set user data piece aligend 32 bytes to be able to process it with SIMD
+  ret->msgData=(void *)ret+(sizeof(notifiedFIFO_elt_t)/32+1)*32;
+  ret->malloced=true;
+  return ret;
+}
+
+static inline void *NotifiedFifoData(notifiedFIFO_elt_t *elt) {
+  return elt->msgData;
+}
+
+static inline void delNotifiedFIFO_elt(notifiedFIFO_elt_t *elt) {
+  bool tmp=elt->malloced;
+
+  if (elt->malloced) {
+    elt->malloced=false;
+    free(elt);
+  } else
+    printf("delNotifiedFIFO on something not allocated by newNotifiedFIFO\n");
+
+  //LOG_W(UTIL,"delNotifiedFIFO on something not allocated by newNotifiedFIFO\n");
+}
+
+static inline void initNotifiedFIFO(notifiedFIFO_t *nf) {
+  mutexinit(nf->lockF);
+  condinit (nf->notifF);
+  nf->inF=NULL;
+  nf->outF=NULL;
+  // No delete function: the creator has only to free the memory
+}
+
+static inline void pushNotifiedFIFO(notifiedFIFO_t *nf, notifiedFIFO_elt_t *msg) {
+  mutexlock(nf->lockF);
+  msg->next=NULL;
+
+  if (nf->outF == NULL)
+    nf->outF = msg;
+
+  if (nf->inF)
+    nf->inF->next = msg;
+
+  nf->inF = msg;
+  condbroadcast(nf->notifF);
+  mutexunlock(nf->lockF);
+}
+
+static inline  notifiedFIFO_elt_t *pullNotifiedFIFO(notifiedFIFO_t *nf) {
+  mutexlock(nf->lockF);
+
+  while(!nf->outF)
+    condwait(nf->notifF, nf->lockF);
+
+  notifiedFIFO_elt_t *ret=nf->outF;
+  nf->outF=nf->outF->next;
+
+  if (nf->outF==NULL)
+    nf->inF=NULL;
+
+  mutexunlock(nf->lockF);
+  return ret;
+}
+
+static inline  notifiedFIFO_elt_t *pollNotifiedFIFO(notifiedFIFO_t *nf) {
+  int tmp=mutextrylock(nf->lockF);
+  if (tmp != 0 )
+    return NULL;
+  
+  notifiedFIFO_elt_t *ret=nf->outF;
+
+  if (ret!=NULL)
+    nf->outF=nf->outF->next;
+
+  if (nf->outF==NULL)
+    nf->inF=NULL;
+
+  mutexunlock(nf->lockF);
+  return ret;
+}
+
+// This function aborts all messages matching the key
+// If the queue is used in thread pools, it doesn't cancels already running processing
+// because the message has already been picked
+static inline void abortNotifiedFIFO(notifiedFIFO_t *nf, uint64_t key) {
+  mutexlock(nf->lockF);
+  notifiedFIFO_elt_t **start=&nf->outF;
+
+  while(*start!=NULL) {
+    if ( (*start)->key == key ) {
+      notifiedFIFO_elt_t *request=*start;
+      *start=(*start)->next;
+      delNotifiedFIFO_elt(request);
+    }
+
+    if (*start != NULL)
+      start=&(*start)->next;
+  }
+
+  mutexunlock(nf->lockF);
+}
+
+struct one_thread {
+  pthread_t  threadID;
+  int id;
+  int coreID;
+  char name[256];
+  uint64_t runningOnKey;
+  bool abortFlag;
+  struct thread_pool *pool;
+  struct one_thread *next;
+};
+
+typedef struct thread_pool {
+  int activated;
+  bool measurePerf;
+  int traceFd;
+  int dummyTraceFd;
+  uint64_t cpuCyclesMicroSec;
+  uint64_t startProcessingUE;
+  int nbThreads;
+  bool restrictRNTI;
+  notifiedFIFO_t incomingFifo;
+  struct one_thread *allthreads;
+} tpool_t;
+
+static inline void pushTpool(tpool_t *t, notifiedFIFO_elt_t *msg) {
+  if (t->measurePerf) msg->creationTime=rdtsc();
+
+  pushNotifiedFIFO(&t->incomingFifo, msg);
+}
+
+static inline notifiedFIFO_elt_t *pullTpool(notifiedFIFO_t *responseFifo, tpool_t *t) {
+  notifiedFIFO_elt_t *msg= pullNotifiedFIFO(responseFifo);
+
+  if (t->measurePerf)
+    msg->returnTime=rdtsc();
+
+  if (t->traceFd)
+    (void)write(t->traceFd, msg, sizeof(*msg));
+
+  return msg;
+}
+
+static inline notifiedFIFO_elt_t *tryPullTpool(notifiedFIFO_t *responseFifo, tpool_t *t) {
+  notifiedFIFO_elt_t *msg= pullNotifiedFIFO(responseFifo);
+  if (msg == NULL)
+    return NULL;
+  
+  if (t->measurePerf)
+    msg->returnTime=rdtsc();
+
+  if (t->traceFd)
+    (void)write(t->traceFd, msg, sizeof(*msg));
+
+  return msg;
+}
+
+static inline void abortTpool(tpool_t *t, uint64_t key) {
+  notifiedFIFO_t *nf=&t->incomingFifo;
+  mutexlock(nf->lockF);
+  notifiedFIFO_elt_t **start=&nf->outF;
+
+  while(*start!=NULL) {
+    if ( (*start)->key == key ) {
+      notifiedFIFO_elt_t *request=*start;
+      *start=(*start)->next;
+      delNotifiedFIFO_elt(request);
+    }
+
+    if (*start != NULL)
+      start=&(*start)->next;
+  }
+
+  struct one_thread *ptr=t->allthreads;
+
+  while(ptr!=NULL) {
+    if (ptr->runningOnKey==key)
+      ptr->abortFlag=true;
+
+    ptr=ptr->next;
+  }
+
+  mutexunlock(nf->lockF);
+}
+
+#endif

common/utils/threadPool/thread-pool.md

+# Thread pool
+
+The thread pool is a working server, made of a set of worker threads that can be mapped on CPU cores.
+
+Each worker loop on pick from the same input queue jobs to do.
+When a job is done, the worker sends a return if a return is defined.
+
+A selective abort allows to cancel parallel jobs (usage: a client pushed jobs, but from a response of one job, the other linked jobs becomes useless).
+
+All the thread pool functions are thread safe, nevertheless the working functions are implemented by the thread pool client, so the client has to tackle the parallel execution of his functions called "processingFunc" hereafter.
+
+# jobs
+
+A job is a message (notifiedFIFO_elt_t):
+next: internal FIFO chain, do not set it
+key:  a long int that the client can use to identify a message or a group of messages
+responseFifo: if the client defines a response FIFO, the message will be posted back after processing
+processingFunc: any funtion (type void processingFunc(void *)) that the worker will launch
+msgData: the data passed to processingFunc. It can be added automatically, or you can set it to a buffer you are managing
+malloced: a boolean that enable internal free in these cases: no return Fifo or Abort feature
+
+The job messages can be created with newNotifiedFIFO_elt() and delNotifiedFIFO_elt() or managed by the client.
+
+# Queues of jobs
+
+Queues are type of: notifiedFIFO_t that must be initialized by init_notifiedFIFO()
+No delete function is required, the creator has only to free the data of type notifiedFIFO_t
+
+push_notifiedFIFO() add a job in the queue
+pull_notifiedFIFO() is blocking, poll_notifiedFIFO() is non blocking
+
+abort_notifiedFIFO() allows the customer to delete all waiting jobs that match with the key (see key in jobs definition)
+
+# Thread pools
+
+## initialization
+The clients can create one or more thread pools with init_tpool()
+the params string structure: describes a list of cores, separated by "," that run a worker thread
+
+The threads are all Linux real time scheduler, their name is set automatically is "Tpool_<core id>"
+
+## adding jobs
+The client create their jobs messages as a notifiedFIFO_elt_t, then they push it with pushTpool() (that internally calls push_notifiedFIFO())
+
+If they need a return, they have to create response queues with init_notifiedFIFO() and set this FIFO pointer in the notifiedFIFO_elt_t before pushing the job.
+
+## abort
+
+A abort service abortTpool() allows to abort all jobs that match a key (see jobs "key"). When the abort returns, it garanties no job (matching the key) response will be posted on response queues.
+
+Nevertheless, jobs already performed before the return of abortTpool() are pushed in the response Fifo queue.
+
+## Performance measurements
+
+A performance measurement is integrated: the pool will automacillay fill timestamps:
+
+* creationTime: time the request is push to the pool;
+* startProcessingTime: time a worker start to run on the job
+* endProcessingTime: time the worker finished the job
+* returnTime: time the client reads the result
+
+if you set the environement variable: thread-pool-measurements to a valid file name
+These measurements will be wrote to this Linux pipe.
+
+A tool to read the linux fifo and display it in ascii will be provided (TBD)