/* * Licensed to the OpenAirInterface (OAI) Software Alliance under one or more * contributor license agreements. See the NOTICE file distributed with * this work for additional information regarding copyright ownership. * The OpenAirInterface Software Alliance licenses this file to You under * the OAI Public License, Version 1.1 (the "License"); you may not use this file * except in compliance with the License. * You may obtain a copy of the License at * * http://www.openairinterface.org/?page_id=698 * * Author and copyright: Laurent Thomas, open-cells.com * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * See the License for the specific language governing permissions and * limitations under the License. *------------------------------------------------------------------------------- * For more information about the OpenAirInterface (OAI) Software Alliance: * contact@openairinterface.org */ #ifndef THREAD_POOL_H #define THREAD_POOL_H #include <stdbool.h> #include <stdint.h> #include <malloc.h> #include <stdalign.h> #include <pthread.h> #include <unistd.h> #include <string.h> #include <sys/syscall.h> #include "assertions.h" #include "common/utils/time_meas.h" #include "common/utils/system.h" #ifdef DEBUG #define THREADINIT PTHREAD_ERRORCHECK_MUTEX_INITIALIZER_NP #else #define THREADINIT PTHREAD_MUTEX_INITIALIZER #endif #define mutexinit(mutex) {int ret=pthread_mutex_init(&mutex,NULL); \ AssertFatal(ret==0,"ret=%d\n",ret);} #define condinit(signal) {int ret=pthread_cond_init(&signal,NULL); \ AssertFatal(ret==0,"ret=%d\n",ret);} #define mutexlock(mutex) {int ret=pthread_mutex_lock(&mutex); \ AssertFatal(ret==0,"ret=%d\n",ret);} #define mutextrylock(mutex) pthread_mutex_trylock(&mutex) #define mutexunlock(mutex) {int ret=pthread_mutex_unlock(&mutex); \ AssertFatal(ret==0,"ret=%d\n",ret);} #define condwait(condition, mutex) {int ret=pthread_cond_wait(&condition, &mutex); \ AssertFatal(ret==0,"ret=%d\n",ret);} #define condbroadcast(signal) {int ret=pthread_cond_broadcast(&signal); \ AssertFatal(ret==0,"ret=%d\n",ret);} #define condsignal(signal) {int ret=pthread_cond_signal(&signal); \ AssertFatal(ret==0,"ret=%d\n",ret);} #define tpool_nbthreads(tpool) (tpool.nbThreads) 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; oai_cputime_t creationTime; oai_cputime_t startProcessingTime; oai_cputime_t endProcessingTime; oai_cputime_t returnTime; // use alignas(32) to align msgData to 32b // user data behind it will be aligned to 32b as well // important! this needs to be the last member in the struct alignas(32) 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; bool abortFIFO; // if set, the FIFO always returns NULL -> abort condition } 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 = (notifiedFIFO_elt_t *)memalign(32, sizeof(notifiedFIFO_elt_t) + size); AssertFatal(NULL != ret, "out of memory\n"); 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 // msgData is aligned to 32bytes, so everything after will be as well ret->msgData = ((uint8_t *)ret) + sizeof(notifiedFIFO_elt_t); 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) { if (elt->malloced) { elt->malloced = false; free(elt); } /* it is allowed to call delNotifiedFIFO_elt when the memory is managed by * the caller */ } static inline void initNotifiedFIFO_nothreadSafe(notifiedFIFO_t *nf) { nf->inF=NULL; nf->outF=NULL; nf->abortFIFO = false; } static inline void initNotifiedFIFO(notifiedFIFO_t *nf) { mutexinit(nf->lockF); condinit (nf->notifF); initNotifiedFIFO_nothreadSafe(nf); // No delete function: the creator has only to free the memory } static inline void pushNotifiedFIFO_nothreadSafe(notifiedFIFO_t *nf, notifiedFIFO_elt_t *msg) { msg->next=NULL; if (nf->outF == NULL) nf->outF = msg; if (nf->inF != NULL) nf->inF->next = msg; nf->inF = msg; } static inline void pushNotifiedFIFO(notifiedFIFO_t *nf, notifiedFIFO_elt_t *msg) { mutexlock(nf->lockF); if (!nf->abortFIFO) { pushNotifiedFIFO_nothreadSafe(nf,msg); condsignal(nf->notifF); } mutexunlock(nf->lockF); } static inline notifiedFIFO_elt_t *pullNotifiedFIFO_nothreadSafe(notifiedFIFO_t *nf) { if (nf->outF == NULL) return NULL; if (nf->abortFIFO) return NULL; notifiedFIFO_elt_t *ret=nf->outF; AssertFatal(nf->outF != nf->outF->next,"Circular list in thread pool: push several times the same buffer is forbidden\n"); nf->outF=nf->outF->next; if (nf->outF==NULL) nf->inF=NULL; return ret; } static inline notifiedFIFO_elt_t *pullNotifiedFIFO(notifiedFIFO_t *nf) { mutexlock(nf->lockF); notifiedFIFO_elt_t *ret = NULL; while((ret=pullNotifiedFIFO_nothreadSafe(nf)) == NULL && !nf->abortFIFO) condwait(nf->notifF, nf->lockF); mutexunlock(nf->lockF); return ret; } static inline notifiedFIFO_elt_t *pollNotifiedFIFO(notifiedFIFO_t *nf) { int tmp=mutextrylock(nf->lockF); if (tmp != 0 ) return NULL; if (nf->abortFIFO) { mutexunlock(nf->lockF); return NULL; } notifiedFIFO_elt_t *ret=pullNotifiedFIFO_nothreadSafe(nf); mutexunlock(nf->lockF); return ret; } static inline time_stats_t exec_time_stats_NotifiedFIFO(const notifiedFIFO_elt_t* elt) { time_stats_t ts = {0}; if (elt->startProcessingTime == 0 && elt->endProcessingTime == 0) return ts; /* no measurements done */ ts.in = elt->startProcessingTime; ts.diff = elt->endProcessingTime - ts.in; ts.p_time = ts.diff; ts.diff_square = ts.diff * ts.diff; ts.max = ts.diff; ts.trials = 1; return ts; } // This functions aborts all messages in the queue, and marks the queue as // "aborted", such that every call to it will return NULL static inline void abortNotifiedFIFO(notifiedFIFO_t *nf) { mutexlock(nf->lockF); nf->abortFIFO = true; notifiedFIFO_elt_t **elt = &nf->outF; while(*elt != NULL) { notifiedFIFO_elt_t *p = *elt; *elt = (*elt)->next; delNotifiedFIFO_elt(p); } if (nf->outF == NULL) nf->inF = NULL; condbroadcast(nf->notifF); mutexunlock(nf->lockF); } struct one_thread { pthread_t threadID; int id; int coreID; char name[256]; uint64_t runningOnKey; bool dropJob; bool terminate; struct thread_pool *pool; struct one_thread *next; }; typedef struct thread_pool { bool activated; bool measurePerf; int traceFd; int dummyKeepReadingTraceFd; uint64_t cpuCyclesMicroSec; int nbThreads; 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_oai(); if ( t->activated) pushNotifiedFIFO(&t->incomingFifo, msg); else { if (t->measurePerf) msg->startProcessingTime=rdtsc_oai(); msg->processingFunc(NotifiedFifoData(msg)); if (t->measurePerf) msg->endProcessingTime=rdtsc_oai(); if (msg->reponseFifo) pushNotifiedFIFO(msg->reponseFifo, msg); else delNotifiedFIFO_elt(msg); } } static inline notifiedFIFO_elt_t *pullTpool(notifiedFIFO_t *responseFifo, tpool_t *t) { notifiedFIFO_elt_t *msg= pullNotifiedFIFO(responseFifo); if (msg == NULL) return NULL; AssertFatal(t->traceFd != 0, "Thread pool used while not initialized"); if (t->measurePerf) msg->returnTime=rdtsc_oai(); if (t->traceFd > 0) { ssize_t b = write(t->traceFd, msg, sizeof(*msg)); AssertFatal(b == sizeof(*msg), "error in write(): %d, %s\n", errno, strerror(errno)); } return msg; } static inline notifiedFIFO_elt_t *tryPullTpool(notifiedFIFO_t *responseFifo, tpool_t *t) { notifiedFIFO_elt_t *msg= pollNotifiedFIFO(responseFifo); AssertFatal(t->traceFd != 0, "Thread pool used while not initialized"); if (msg == NULL) return NULL; if (t->measurePerf) msg->returnTime=rdtsc_oai(); if (t->traceFd > 0) { ssize_t b = write(t->traceFd, msg, sizeof(*msg)); AssertFatal(b == sizeof(*msg), "error in write(): %d, %s\n", errno, strerror(errno)); } return msg; } static inline int abortTpool(tpool_t *t) { int nbRemoved=0; /* disables threading: if a message comes in now, we cannot have a race below * as each thread will simply execute the message itself */ t->activated = false; notifiedFIFO_t *nf=&t->incomingFifo; mutexlock(nf->lockF); nf->abortFIFO = true; notifiedFIFO_elt_t **start=&nf->outF; /* mark threads to abort them */ struct one_thread *thread = t->allthreads; while (thread != NULL) { thread->dropJob = true; thread->terminate = true; nbRemoved++; thread = thread->next; } /* clear FIFOs */ while(*start!=NULL) { notifiedFIFO_elt_t **request=start; *start=(*start)->next; delNotifiedFIFO_elt(*request); *request = NULL; nbRemoved++; } if (t->incomingFifo.outF==NULL) t->incomingFifo.inF=NULL; condbroadcast(t->incomingFifo.notifF); mutexunlock(nf->lockF); /* join threads that are still runing */ thread = t->allthreads; while (thread != NULL) { pthread_cancel(thread->threadID); struct one_thread *next = thread->next; free(thread); thread = next; } return nbRemoved; } void initNamedTpool(char *params,tpool_t *pool, bool performanceMeas, char *name); void initFloatingCoresTpool(int nbThreads,tpool_t *pool, bool performanceMeas, char *name); #define initTpool(PARAMPTR,TPOOLPTR, MEASURFLAG) initNamedTpool(PARAMPTR,TPOOLPTR, MEASURFLAG, NULL) #endif