Commit 0ceab97f authored by Makarand Kulkarni's avatar Makarand Kulkarni

Initial change set to abstract notifiedFIFO

parent fe613541
...@@ -34,6 +34,9 @@ ...@@ -34,6 +34,9 @@
#include <threadPool/thread-pool.h> #include <threadPool/thread-pool.h>
void displayList(notifiedFIFO_t *nf) { void displayList(notifiedFIFO_t *nf) {
((notifiedFIFO_api_t *)nf->opaqueptr)->display(nf);
/*
int n=0; int n=0;
notifiedFIFO_elt_t *ptr=nf->outF; notifiedFIFO_elt_t *ptr=nf->outF;
...@@ -43,12 +46,23 @@ void displayList(notifiedFIFO_t *nf) { ...@@ -43,12 +46,23 @@ void displayList(notifiedFIFO_t *nf) {
} }
printf("End of list: %d elements\n",n); printf("End of list: %d elements\n",n);
*/
}
void initNotifiedFIFO_typeFIFO(notifiedFIFO_t *nf) {
nf->opaqueptr = (notifiedFIFO_typeFIFO_t *)malloc(sizeof(struct notifiedFIFO_typeFIFO_s));
AssertFatal(nf->opaqueptr != NULL, "createNotifiedFIFO_typeFIFO's opaqueptr is NULL");
((notifiedFIFO_api_t *)nf->opaqueptr)->init = _initNotifiedFIFO_typeFIFO;
_initNotifiedFIFO(nf);/* Now call the old init function */
} }
static inline notifiedFIFO_elt_t *pullNotifiedFifoRemember( notifiedFIFO_t *nf, struct one_thread *thr) { static inline notifiedFIFO_elt_t *pullNotifiedFifoRemember( notifiedFIFO_t *nf, struct one_thread *thr) {
notifiedFIFO_api_t* notifiedFIFO_api = (notifiedFIFO_api_t *)nf->opaqueptr;
mutexlock(nf->lockF); mutexlock(nf->lockF);
while(!nf->outF && !thr->terminate) while(notifiedFIFO_api->empty(nf) && !thr->terminate)
condwait(nf->notifF, nf->lockF); condwait(nf->notifF, nf->lockF);
if (thr->terminate) { if (thr->terminate) {
...@@ -56,11 +70,7 @@ static inline notifiedFIFO_elt_t *pullNotifiedFifoRemember( notifiedFIFO_t *nf, ...@@ -56,11 +70,7 @@ static inline notifiedFIFO_elt_t *pullNotifiedFifoRemember( notifiedFIFO_t *nf,
return NULL; return NULL;
} }
notifiedFIFO_elt_t *ret=nf->outF; notifiedFIFO_elt_t *ret=notifiedFIFO_api->pull(nf);
nf->outF=nf->outF->next;
if (nf->outF==NULL)
nf->inF=NULL;
// For abort feature // For abort feature
thr->runningOnKey=ret->key; thr->runningOnKey=ret->key;
......
...@@ -54,6 +54,9 @@ ...@@ -54,6 +54,9 @@
#define condsignal(signal) {int ret=pthread_cond_signal(&signal); \ #define condsignal(signal) {int ret=pthread_cond_signal(&signal); \
AssertFatal(ret==0,"ret=%d\n",ret);} AssertFatal(ret==0,"ret=%d\n",ret);}
#define tpool_nbthreads(tpool) (tpool.nbThreads) #define tpool_nbthreads(tpool) (tpool.nbThreads)
#define initNotifiedFIFO(nf) initNotifiedFIFO_typeFIFO(nf)
typedef struct notifiedFIFO_elt_s { typedef struct notifiedFIFO_elt_s {
struct notifiedFIFO_elt_s *next; struct notifiedFIFO_elt_s *next;
uint64_t key; //To filter out elements uint64_t key; //To filter out elements
...@@ -68,13 +71,22 @@ typedef struct notifiedFIFO_elt_s { ...@@ -68,13 +71,22 @@ typedef struct notifiedFIFO_elt_s {
} notifiedFIFO_elt_t; } notifiedFIFO_elt_t;
typedef struct notifiedFIFO_s { typedef struct notifiedFIFO_s {
notifiedFIFO_elt_t *outF;
notifiedFIFO_elt_t *inF;
pthread_mutex_t lockF; pthread_mutex_t lockF;
pthread_cond_t notifF; pthread_cond_t notifF;
bool abortFIFO; // if set, the FIFO always returns NULL -> abort condition bool abortFIFO; // if set, the FIFO always returns NULL -> abort condition
void *opaqueptr;
} notifiedFIFO_t; } notifiedFIFO_t;
typedef struct notifiedFIFO_api_s {
void (*init)(notifiedFIFO_t *nf);
void (*push)(notifiedFIFO_t *nf, notifiedFIFO_elt_t *msg);
notifiedFIFO_elt_t* (*pull)(notifiedFIFO_t *nf);
int (*abort)(notifiedFIFO_t *nf);
bool (*empty)(notifiedFIFO_t *nf);
void (*display)(notifiedFIFO_t *nf);
} notifiedFIFO_api_t;
// You can use this allocator or use any piece of memory // You can use this allocator or use any piece of memory
static inline notifiedFIFO_elt_t *newNotifiedFIFO_elt(int size, static inline notifiedFIFO_elt_t *newNotifiedFIFO_elt(int size,
uint64_t key, uint64_t key,
...@@ -105,62 +117,125 @@ static inline void delNotifiedFIFO_elt(notifiedFIFO_elt_t *elt) { ...@@ -105,62 +117,125 @@ static inline void delNotifiedFIFO_elt(notifiedFIFO_elt_t *elt) {
* the caller */ * the caller */
} }
static inline void initNotifiedFIFO_nothreadSafe(notifiedFIFO_t *nf) { void initNotifiedFIFO_typeFIFO(notifiedFIFO_t *nf);
nf->inF=NULL;
nf->outF=NULL; typedef struct notifiedFIFO_typeFIFO_s {
nf->abortFIFO = false; notifiedFIFO_api_t api; // Public methods which work on opaqueptr impl
} notifiedFIFO_elt_t *outF;
static inline void initNotifiedFIFO(notifiedFIFO_t *nf) { notifiedFIFO_elt_t *inF;
mutexinit(nf->lockF);
condinit (nf->notifF); } notifiedFIFO_typeFIFO_t;
initNotifiedFIFO_nothreadSafe(nf);
// No delete function: the creator has only to free the memory static inline void displayNotifiedFIFO_typeFIFO(notifiedFIFO_t *nf) {
notifiedFIFO_typeFIFO_t *nf_typeFIFO = ((notifiedFIFO_typeFIFO_t *)nf->opaqueptr);
int n=0;
notifiedFIFO_elt_t *ptr=nf_typeFIFO->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 void pushNotifiedFIFO_nothreadSafe(notifiedFIFO_t *nf, notifiedFIFO_elt_t *msg) { static inline void pushNotifiedFIFO_typeFIFO(notifiedFIFO_t *nf, notifiedFIFO_elt_t *msg) {
notifiedFIFO_typeFIFO_t *nf_typeFIFO = ((notifiedFIFO_typeFIFO_t *)nf->opaqueptr);
msg->next=NULL; msg->next=NULL;
if (nf->outF == NULL) if (nf_typeFIFO->outF == NULL)
nf->outF = msg; nf_typeFIFO->outF = msg;
if (nf->inF != NULL) if (nf_typeFIFO->inF != NULL)
nf->inF->next = msg; nf_typeFIFO->inF->next = msg;
nf->inF = msg; nf_typeFIFO->inF = msg;
} }
static inline void pushNotifiedFIFO(notifiedFIFO_t *nf, notifiedFIFO_elt_t *msg) { static inline notifiedFIFO_elt_t *pullNotifiedFIFO_typeFIFO(notifiedFIFO_t *nf) {
mutexlock(nf->lockF); notifiedFIFO_typeFIFO_t *nf_typeFIFO = ((notifiedFIFO_typeFIFO_t *)nf->opaqueptr);
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_typeFIFO->outF == NULL)
if (nf->outF == NULL)
return NULL; return NULL;
if (nf->abortFIFO) if (nf->abortFIFO)
return NULL; return NULL;
notifiedFIFO_elt_t *ret=nf->outF; notifiedFIFO_elt_t *ret=nf_typeFIFO->outF;
AssertFatal(nf->outF != nf->outF->next,"Circular list in thread pool: push several times the same buffer is forbidden\n"); AssertFatal(nf_typeFIFO->outF != nf_typeFIFO->outF->next,"Circular list in thread pool: push several times the same buffer is forbidden\n");
nf->outF=nf->outF->next; nf_typeFIFO->outF=nf_typeFIFO->outF->next;
if (nf->outF==NULL) if (nf_typeFIFO->outF==NULL)
nf->inF=NULL; nf_typeFIFO->inF=NULL;
return ret; return ret;
} }
static inline int abortNotifiedFIFO_typeFIFO(notifiedFIFO_t *nf) {
notifiedFIFO_typeFIFO_t *nf_typeFIFO = ((notifiedFIFO_typeFIFO_t *)nf->opaqueptr);
int nbRemoved=0;
notifiedFIFO_elt_t **elt = &nf_typeFIFO->outF;
while(*elt != NULL) {
notifiedFIFO_elt_t *p = *elt;
*elt = (*elt)->next;
delNotifiedFIFO_elt(p);
nbRemoved++;
}
if (nf_typeFIFO->outF == NULL)
nf_typeFIFO->inF = NULL;
return nbRemoved;
}
static inline bool is_emptyNotifiedFIFO_typeFIFO(notifiedFIFO_t *nf) {
notifiedFIFO_typeFIFO_t *nf_typeFIFO = ((notifiedFIFO_typeFIFO_t *)nf->opaqueptr);
return (nf_typeFIFO->outF == NULL);
}
static inline void _initNotifiedFIFO_typeFIFO(notifiedFIFO_t *nf) {
notifiedFIFO_typeFIFO_t *nf_typeFIFO = ((notifiedFIFO_typeFIFO_t *)nf->opaqueptr);
nf_typeFIFO->inF=NULL;
nf_typeFIFO->outF=NULL;
nf_typeFIFO->api.push = pushNotifiedFIFO_typeFIFO;
nf_typeFIFO->api.pull = pullNotifiedFIFO_typeFIFO;
nf_typeFIFO->api.abort = abortNotifiedFIFO_typeFIFO;
nf_typeFIFO->api.display = displayNotifiedFIFO_typeFIFO;
nf_typeFIFO->api.empty = is_emptyNotifiedFIFO_typeFIFO;
}
static inline void _initNotifiedFIFO(notifiedFIFO_t *nf) {
mutexinit(nf->lockF);
condinit (nf->notifF);
nf->abortFIFO = false;
// The caller sets the init function in opaqueptr to avoid proto
// changes
((notifiedFIFO_api_t *)nf->opaqueptr)->init(nf);
// 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);
if (!nf->abortFIFO) {
((notifiedFIFO_api_t *)nf->opaqueptr)->push(nf,msg);
condsignal(nf->notifF);
}
mutexunlock(nf->lockF);
}
static inline notifiedFIFO_elt_t *pullNotifiedFIFO(notifiedFIFO_t *nf) { static inline notifiedFIFO_elt_t *pullNotifiedFIFO(notifiedFIFO_t *nf) {
mutexlock(nf->lockF); mutexlock(nf->lockF);
notifiedFIFO_elt_t *ret = NULL; notifiedFIFO_elt_t *ret = NULL;
while((ret=pullNotifiedFIFO_nothreadSafe(nf)) == NULL && !nf->abortFIFO) while((ret= ((notifiedFIFO_api_t *)nf->opaqueptr)->pull(nf)) == NULL && !nf->abortFIFO)
condwait(nf->notifF, nf->lockF); condwait(nf->notifF, nf->lockF);
mutexunlock(nf->lockF); mutexunlock(nf->lockF);
...@@ -178,7 +253,7 @@ static inline notifiedFIFO_elt_t *pollNotifiedFIFO(notifiedFIFO_t *nf) { ...@@ -178,7 +253,7 @@ static inline notifiedFIFO_elt_t *pollNotifiedFIFO(notifiedFIFO_t *nf) {
return NULL; return NULL;
} }
notifiedFIFO_elt_t *ret=pullNotifiedFIFO_nothreadSafe(nf); notifiedFIFO_elt_t *ret=((notifiedFIFO_api_t *)nf->opaqueptr)->pull(nf);
mutexunlock(nf->lockF); mutexunlock(nf->lockF);
return ret; return ret;
} }
...@@ -197,22 +272,19 @@ static inline time_stats_t exec_time_stats_NotifiedFIFO(const notifiedFIFO_elt_t ...@@ -197,22 +272,19 @@ static inline time_stats_t exec_time_stats_NotifiedFIFO(const notifiedFIFO_elt_t
return ts; return ts;
} }
// This functions aborts all messages in the queue, and marks the queue as // This functions aborts all messages in the queue, and marks the queue as
// "aborted", such that every call to it will return NULL // "aborted", such that every call to it will return NULL
static inline void abortNotifiedFIFO(notifiedFIFO_t *nf) { static inline int abortNotifiedFIFO(notifiedFIFO_t *nf) {
mutexlock(nf->lockF); mutexlock(nf->lockF);
int nbRemoved = 0;
nf->abortFIFO = true; 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) nbRemoved = ((notifiedFIFO_api_t *)nf->opaqueptr)->abort(nf);
nf->inF = NULL;
condbroadcast(nf->notifF); condbroadcast(nf->notifF);
mutexunlock(nf->lockF); mutexunlock(nf->lockF);
return nbRemoved;
} }
struct one_thread { struct one_thread {
...@@ -300,7 +372,7 @@ static inline int abortTpool(tpool_t *t) { ...@@ -300,7 +372,7 @@ static inline int abortTpool(tpool_t *t) {
notifiedFIFO_t *nf=&t->incomingFifo; notifiedFIFO_t *nf=&t->incomingFifo;
mutexlock(nf->lockF); mutexlock(nf->lockF);
nf->abortFIFO = true; nf->abortFIFO = true;
notifiedFIFO_elt_t **start=&nf->outF; /* notifiedFIFO_elt_t **start=&nf->outF; */
/* mark threads to abort them */ /* mark threads to abort them */
struct one_thread *thread = t->allthreads; struct one_thread *thread = t->allthreads;
...@@ -311,7 +383,7 @@ static inline int abortTpool(tpool_t *t) { ...@@ -311,7 +383,7 @@ static inline int abortTpool(tpool_t *t) {
thread = thread->next; thread = thread->next;
} }
/* clear FIFOs */ /* clear FIFOs
while(*start!=NULL) { while(*start!=NULL) {
notifiedFIFO_elt_t **request=start; notifiedFIFO_elt_t **request=start;
*start=(*start)->next; *start=(*start)->next;
...@@ -326,6 +398,8 @@ static inline int abortTpool(tpool_t *t) { ...@@ -326,6 +398,8 @@ static inline int abortTpool(tpool_t *t) {
condbroadcast(t->incomingFifo.notifF); condbroadcast(t->incomingFifo.notifF);
mutexunlock(nf->lockF); mutexunlock(nf->lockF);
*/
nbRemoved = abortNotifiedFIFO(nf);
/* join threads that are still runing */ /* join threads that are still runing */
thread = t->allthreads; thread = t->allthreads;
while (thread != NULL) { while (thread != NULL) {
......
...@@ -769,7 +769,7 @@ void *UE_thread(void *arg) { ...@@ -769,7 +769,7 @@ void *UE_thread(void *arg) {
initNotifiedFIFO(&txFifo); initNotifiedFIFO(&txFifo);
notifiedFIFO_t freeBlocks; notifiedFIFO_t freeBlocks;
initNotifiedFIFO_nothreadSafe(&freeBlocks); initNotifiedFIFO(&freeBlocks);
int timing_advance = UE->timing_advance; int timing_advance = UE->timing_advance;
NR_UE_MAC_INST_t *mac = get_mac_inst(0); NR_UE_MAC_INST_t *mac = get_mac_inst(0);
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment