maki/openairinterface5g
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'origin/remove-thread-pool-abort' into integration_2023_w24

Browse Source
This commit is contained in:
Robert Schmidt
2023-06-16 17:54:11 +02:00
parent d4eeffff81 15b8dbc476
commit 96aacd58aa
48 changed files with 651 additions and 1057 deletions
Download Patch File Download Diff File Expand all files Collapse all files
common/utils/load_module_shlib.c
+40 -45
View File
@@ -63,60 +63,55 @@ void loader_init(void) {
}
/* build the full shared lib name from the module name */
char *loader_format_shlibpath(char *modname, char *version)
static char *loader_format_shlibpath(char *modname, char *version)
{
char *tmpstr;
char *shlibpath =NULL;
char *shlibversion=NULL;
// clang-format off
paramdef_t LoaderParams[] = {
char *tmpstr;
char *shlibpath = NULL;
char *shlibversion = NULL;
// clang-format off
paramdef_t LoaderParams[] = {
{"shlibpath", NULL, 0, .strptr = &shlibpath, .defstrval = NULL, TYPE_STRING, 0, NULL},
{"shlibversion", NULL, 0, .strptr = &shlibversion, .defstrval = "", TYPE_STRING, 0, NULL}
};
// clang-format on
};
// clang-format on
int ret;
int ret;
/* looks for specific path for this module in the config file */
/* specific value for a module path and version is located in a modname subsection of the loader section */
/* shared lib name is formatted as lib<module name><module version>.so */
/* looks for specific path for this module in the config file */
/* specific value for a module path and version is located in a modname subsection of the loader section */
/* shared lib name is formatted as lib<module name><module version>.so */
char cfgprefix[sizeof(LOADER_CONFIG_PREFIX)+strlen(modname)+16];
sprintf(cfgprefix,LOADER_CONFIG_PREFIX ".%s",modname);
ret = config_get( LoaderParams,sizeof(LoaderParams)/sizeof(paramdef_t),cfgprefix);
ret = config_get(LoaderParams, sizeofArray(LoaderParams), cfgprefix);
if (ret <0) {
fprintf(stderr,"[LOADER] %s %d couldn't retrieve config from section %s\n",__FILE__, __LINE__,cfgprefix);
}
/* no specific path, use loader default shared lib path */
if (shlibpath == NULL) {
shlibpath = loader_data.shlibpath ;
}
/* no specific shared lib version */
if (version != NULL) { // version specified as a function parameter
shlibversion=version;
}
if (shlibversion == NULL) { // no specific version specified, neither as a config param or as a function param
shlibversion = "" ;
}
/* alloc memory for full module shared lib file name */
tmpstr = malloc(strlen(shlibpath)+strlen(modname)+strlen(shlibversion)+16);
if (tmpstr == NULL) {
fprintf(stderr,"[LOADER] %s %d malloc error loading module %s, %s\n",__FILE__, __LINE__, modname, strerror(errno));
exit_fun("[LOADER] unrecoverable error");
}
if(shlibpath[0] != 0) {
ret=sprintf(tmpstr,"%s/",shlibpath);
} else {
ret = 0;
}
fprintf(stderr, "[LOADER] %s %d couldn't retrieve config from section %s\n", __FILE__, __LINE__, cfgprefix);
}
/* no specific path, use loader default shared lib path */
if (shlibpath == NULL) {
shlibpath = loader_data.shlibpath;
}
/* no specific shared lib version */
if (version != NULL) { // version specified as a function parameter
shlibversion = version;
}
if (shlibversion == NULL) { // no specific version specified, neither as a config param or as a function param
shlibversion = "";
}
/* alloc memory for full module shared lib file name */
tmpstr = malloc(strlen(shlibpath) + strlen(modname) + strlen(shlibversion) + 16);
if (tmpstr == NULL) {
fprintf(stderr, "[LOADER] %s %d malloc error loading module %s, %s\n", __FILE__, __LINE__, modname, strerror(errno));
exit_fun("[LOADER] unrecoverable error");
}
if (shlibpath[0] != 0) {
ret = sprintf(tmpstr, "%s/", shlibpath);
} else {
ret = 0;
}
sprintf(tmpstr+ret,"lib%s%s.so",modname,shlibversion);
return tmpstr;
sprintf(tmpstr + ret, "lib%s%s.so", modname, shlibversion);
return tmpstr;
}
int load_module_version_shlib(char *modname, char *version, loader_shlibfunc_t *farray, int numf, void *autoinit_arg)
common/utils/threadPool/thread-pool.c
-6
View File
@@ -218,14 +218,8 @@ int main() {
tmp=pullNotifiedFIFO(&myFifo);
printf("pulled: %lu\n", tmp->key);
displayList(&myFifo);
abortNotifiedFIFOJob(&myFifo,1005);
printf("aborted 1005\n");
displayList(&myFifo);
pushNotifiedFIFO(&myFifo,newNotifiedFIFO_elt(sizeof(struct testData), 12345678, NULL, NULL));
displayList(&myFifo);
abortNotifiedFIFOJob(&myFifo,12345678);
printf("aborted 12345678\n");
displayList(&myFifo);
do {
tmp=pollNotifiedFIFO(&myFifo);
common/utils/threadPool/thread-pool.h
-58
View File
@@ -197,31 +197,6 @@ static inline time_stats_t exec_time_stats_NotifiedFIFO(const notifiedFIFO_elt_t
return ts;
}
// 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 int abortNotifiedFIFOJob(notifiedFIFO_t *nf, uint64_t key) {
mutexlock(nf->lockF);
int nbDeleted=0;
notifiedFIFO_elt_t **start=&nf->outF;
while(*start!=NULL) {
if ( (*start)->key == key ) {
notifiedFIFO_elt_t *request=*start;
*start=(*start)->next;
delNotifiedFIFO_elt(request);
nbDeleted++;
} else
start=&(*start)->next;
}
if (nf->outF == NULL)
nf->inF=NULL;
mutexunlock(nf->lockF);
return nbDeleted;
}
// 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) {
@@ -317,39 +292,6 @@ static inline notifiedFIFO_elt_t *tryPullTpool(notifiedFIFO_t *responseFifo, tpo
return msg;
}
static inline int abortTpoolJob(tpool_t *t, uint64_t key) {
int nbRemoved=0;
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);
nbRemoved++;
} else
start=&(*start)->next;
}
if (t->incomingFifo.outF==NULL)
t->incomingFifo.inF=NULL;
struct one_thread *thread = t->allthreads;
while (thread != NULL) {
if (thread->runningOnKey == key) {
thread->dropJob = true;
nbRemoved++;
}
thread = thread->next;
}
mutexunlock(nf->lockF);
return nbRemoved;
}
static inline int abortTpool(tpool_t *t) {
int nbRemoved=0;
/* disables threading: if a message comes in now, we cannot have a race below
common/utils/threadPool/thread-pool.md
+1 -22
View File
@@ -2,8 +2,6 @@
The **thread pool** is a working server, made of a set of worker threads that can be mapped on CPU cores. Each thread pool has an **input queue** ("**FIFO**"), from which its workers pick **jobs** (FIFO element) to execute. When a job is done, the worker sends a message to an output queue, if it has been defined.
A selective abort allows to cancel parallel jobs. This can be useful, e.g., if a client pushed jobs, but from a response of one job, the other linked jobs become useless.
All the thread pool functions are thread safe. The functions executed by worker threads are provided by the thread pool client, so the client has to handle the concurrency/parallel execution of his functions.
## license
@@ -56,7 +54,6 @@ Queues can be used to enqueue messages/jobs, of type `notifiedFIFO_t`.
* `pushNotifiedFIFO()`: Add a job to a queue
* `pullNotifiedFIFO()`: Pull a job from a queue. This call is blocking until a job arrived.
* `pollNotifiedFIFO()`: Pull a job from a queue. This call is not blocking, so it returns always very shortly
* `abortNotifiedFIFOJob()`: Allows to delete all waiting jobs that match a key (see `key` in jobs definition)
* `abortNotifiedFIFO()`: Aborts a FIFO, such that it will always return `NULL`
Note that in 99.9% of cases, `pull()` is better than `poll()`.
@@ -104,12 +101,6 @@ Like `pullNotifiedFIFO()`, but non-blocking: they check if the queue `nf` contai
Note that unlike for `pullNotifiedFIFO()`, returning `NULL` does not inform whether the queue has been aborted; the caller should manually check the `abortFIFO` flag of `nf` in this case.
### `int abortNotifiedFIFOJob(notifiedFIFO_t *nf, uint64_t key)`
Aborts all jobs in FIFO queue `nf` with key `key`. Jobs already under execution will be silently dropped and not put in the FIFO return queue, if any.
Returns the number of aborted jobs.
### `void abortNotifiedFIFO(notifiedFIFO_t *nf)`
Aborts the entire FIFO queue `nf`: all jobs will be dropped, and the FIFO is marked as aborted, such that a call to `pullNotifiedFIFO()` returns `NULL`.
@@ -135,10 +126,6 @@ If they need a return value (e.g., result of a computation), they have to create
## Abort
A abort service `abortTpoolJob()` allows to abort all jobs that match a key (see a job's `key`). When the abort returns, it garanties that no job (matching the key) response will be posted on response queues.
Nevertheless, jobs already performed before the return of `abortTpoolJob()` are pushed in the response Fifo queue.
`abortTpool()` kills all jobs in the Tpool, and terminates the pool.
## API details
@@ -171,7 +158,7 @@ Adds a job for processing in the thread pool.
The job data you can set are, inside `msg`:
* `key`: an arbitrary key to find a job in a response queue, and which can be used to abort jobs using `abortTpoolJob()`.
* `key`: an arbitrary key to find a job in a response queue.
* `reponseFifo`: if non-`NULL`, the message will be sent back on this queue when the job is done. If `NULL`, the thread pool automatically frees the job when it is done.
* `processingFunc`: the function to execute for this job.
@@ -187,14 +174,6 @@ Multiple return queues might be useful. Consider the following example in the eN
The same as `pullTpool()` in a non-blocking fashion (an alternative name would have been `pollTpool()`).
### `int abortTpoolJob(tpool_t *t, uint64_t key)`
Is a facility to cancel work you pushed to a thread pool: every job with a given `key` will be deleted, and results of jobs with such `key` under execution will be dropped.
It returns the number of aborted jobs, including the ones that are currently being executed.
I used it once: when eNB performs turbo decode, I push all segments in the thread pool. But when I get back the decoding results, if one segment can't be decoded, I don't need the results of the other segments of the same UE.
### `int abortTpool(tpool_t *t)`
Aborts the complete Tpool: cancel every work in the input queue, marks to drop existing jobs in processing, and terminates all worker threads. It is afterwards still possible to call functions such as `pushTpool()`, but each calling thread will execute the job itself.