fanasina/y_project
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

worker: trying to generalize thread pool workers

Browse Source
This commit is contained in:
fanasina
2025-04-30 23:08:05 +02:00
parent 840276d8b9
commit 923a2f0d0f
7 changed files with 505 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
y_worker_t/Makefile
+92
View File
@@ -0,0 +1,92 @@
# lib: -lyworker
PROJECT_LIB=libyworker.so
CC=gcc
INCLUDE_DIRS=$(PWD)/include
SOCDIR=$(PWD)
#$(wildcard $(PWD)/**/include)
YLISTDIR=$(PWD)/../list_t
INCLUDE=-I$(INCLUDE_DIRS) -I$(YLISTDIR)/src
CFLAGS=-g -lpthread -Wall -Werror -fpic $(INCLUDE) #"-D DEBUG=1"
#LDFLAGS=
YWORKSRC=$(PWD)/src/y_worker_t/y_worker_t.c
YWORKSRC_O=$(YWORKSRC:.c=.o)
YTASKSRC=$(PWD)/src/y_worker_t/y_task_t.c
YTASKSRC_O=$(YTASKSRC:.c=.o)
YLISTSRC=$(YLISTDIR)/src/list_t/list_t.c
YLISTSRC_O=$(YLISTSRC:.c=.o)
TOPTARGETS := all clean
DEPS=$(YLISTDIR)
OBJ=$(YWORKSRC_O) $(YLISTSRC_O)
$(TOPTARGETS): $(DEPS)
all: $(PROJECT_LIB)
$(PROJECT_LIB): $(OBJ)
echo "objects:" $(OBJ)
#$(CC) -shared -o $@ $^ $(INCLUDE) $(LDFLAGS)
#$(CC) -shared -o $@ $^ $(LDFLAGS)
$(CC) -shared -o $@ $^ $(CFLAGS)
$(YWORKSRC_O): $(YWORKSRC) $(YLISTSRC_O) $(YTASKSRC_O)
$(CC) -o $@ -c $< $(CFLAGS)
$(YTASKSRC_O): $(YTASKSRC) $(YLISTSRC_O)
$(CC) -o $@ -c $< $(CFLAGS)
$(DEPS):
$(MAKE) -C $@ $(MAKECMDGOALS)
# .PHONY: $(TOPTARGETS) $(SUBDIRS)
.PHONY: clean
clean:
rm -f $(OBJ)
mrproper: clean
rm -f $(PROJECT_LIB)
install:
cp libyworker.so /usr/lib/
@if [ -d /usr/local/include ] ; then\
echo "copy include to /usr/local/include/" ;\
cp -r include/* /usr/local/include/;\
else\
echo "copy include to /usr/include/" ;\
cp -r include/* /usr/include/;\
fi
uninstall:
rm /usr/lib/libyworker.so
@if [ -d /usr/local/include ] ; then\
echo "remove from /usr/local/include/" ;\
rm -r /usr/local/include/y_worker_t ;\
else\
echo "remove from /usr/include/" ;\
rm -r /usr/include/y_worker_t ;\
fi
#SRC_test=test/is_good.c
#compile: $(SRC_test) $(PROJECT_LIB)
# $(CC) -o launch_is_good_m $< -L. test/src/permutation_t/permutation_t.o test/src/set_theoric_t/set_theoric_t.o -lytest -I./test/src -I./include_ytest
y_worker_t/include/y_worker_t/y_task_t.h
+46
View File
@@ -0,0 +1,46 @@
#ifndef Y_TASK_T_H__C
#define Y_TASK_T_H__C
#include <pthread.h>
#include "list_t/list_t.h"
#define TASK_PENDING 2
#define TASK_DONE 0
#define TASK_NOT_DONE 1
typedef struct y_task_t{
void* (*func)(void*);
void* arg;
void* ret;
int status;
} y_TASK_T;
GENERATE_LIST_ALL(y_TASK_T)
struct y_tasQ{ // taskQueue
struct main_list_y_TASK_T* list_tasQ;
pthread_mutex_t *mut_tasQ;
pthread_cond_t *cond_tasQ;
};
struct y_tasQ * create_y_tasQ();
void free_y_tasQ(struct y_tasQ * tasQ);
void push_tasQ(struct y_tasQ *tasQ, struct y_task_t task);
struct list_y_TASK_T* pull_tasQ(struct y_tasQ *tasQ);
struct argExecTasQ{
int * go_on;
struct y_tasQ *tasQ;
struct y_tasQ *historytasQ;
};
struct argExecTasQ * create_argExecTasQ();
void free_argExecTasQ(struct argExecTasQ * arg);
void * execute_task(void *arg);
#endif /* Y_TASK_T_H__C */
y_worker_t/include/y_worker_t/y_worker_t.h
+33
View File
@@ -0,0 +1,33 @@
#ifndef Y_WORKER_T_H__C
#define Y_WORKER_T_H__C
#include <stdlib.h>
#include <pthread.h>
#include "list_t/list_t.h"
#include "y_worker_t/y_task_t.h"
#define KILL_WORKER 0
#define GO_ON_WORKER 1
typedef struct y_worker_t{
int exec;
// pthread_mutex_t *mut_worker;
// pthread_cond_t *cond_worker;
size_t id;
pthread_t thread;
} y_WORKER_T;
GENERATE_LIST_ALL(y_WORKER_T)
struct argWorker {
struct argExecTasQ *argx;
struct list_y_WORKER_T *worker;
};
void* execute_work(void* arg);
#endif /* Y_WORKER_T_H__C */
y_worker_t/src/y_worker_t/y_task_t.c
+91
View File
@@ -0,0 +1,91 @@
#include "y_worker_t/y_task_t.h"
GEN_LIST_ALL(y_TASK_T)
struct y_tasQ * create_y_tasQ(){
struct y_tasQ *tmp_ytsQ = malloc(sizeof(struct y_tasQ));
tmp_ytsQ->list_tasQ = create_var_list_y_TASK_T();
tmp_ytsQ->mut_tasQ = malloc(sizeof(pthread_mutex_t));
tmp_ytsQ->cond_tasQ = malloc(sizeof(pthread_cond_t));
pthread_mutex_init(tmp_ytsQ->mut_tasQ, NULL);
pthread_cond_init(tmp_ytsQ->cond_tasQ, NULL);
return tmp_ytsQ;
}
void free_y_tasQ(struct y_tasQ * tasQ){
pthread_mutex_destroy(tasQ->mut_tasQ);
pthread_cond_destroy(tasQ->cond_tasQ);
free(tasQ->mut_tasQ);
free(tasQ->cond_tasQ);
free_all_var_list_y_TASK_T(tasQ->list_tasQ);
free(tasQ);
}
void push_tasQ(struct y_tasQ *tasQ, struct y_task_t task){
printf("debug: push_tasQ debut\n");
pthread_mutex_lock(tasQ->mut_tasQ);
push_back_list_y_TASK_T(tasQ->list_tasQ, task);
pthread_mutex_unlock(tasQ->mut_tasQ);
pthread_cond_signal(tasQ->cond_tasQ);
printf("debug: push_tasQ fin\n");
}
struct list_y_TASK_T* pull_tasQ(struct y_tasQ *tasQ){
printf("debug: pull_tasQ debut\n");
struct list_y_TASK_T *valueRet = NULL;
pthread_mutex_lock(tasQ->mut_tasQ);
while(tasQ->list_tasQ->end_list == NULL){
pthread_cond_wait(tasQ->cond_tasQ, tasQ->mut_tasQ);
}
printf("debug: call pull_end_from_list_y_TASK_T debut\n");
valueRet = pull_end_from_list_y_TASK_T(tasQ->list_tasQ);
printf("debug: call pull_end_from_list_y_TASK_T fin, is tasQ NULL? : %d\nis tasQ->list_tasQ NULL?:%d\n", tasQ==NULL, tasQ->list_tasQ == NULL);
pthread_mutex_unlock(tasQ->mut_tasQ);
printf("debug: pull_tasQ fin : is valueRet NULL ? = %d\n", valueRet == NULL);
return valueRet;
}
struct argExecTasQ * create_argExecTasQ(){
struct argExecTasQ * retasQ = malloc(sizeof(struct argExecTasQ));
retasQ->go_on = malloc(sizeof(int));
*(retasQ->go_on)=1;
retasQ->tasQ = create_y_tasQ();
retasQ->historytasQ = create_y_tasQ();
return retasQ;
}
void free_argExecTasQ(struct argExecTasQ * arg){
free(arg->go_on);
free_y_tasQ(arg->tasQ);
free_y_tasQ(arg->historytasQ);
free(arg);
}
void * execute_task(void *arg){
struct argExecTasQ *argx = (struct argExecTasQ *)arg;
struct y_tasQ *tasQ = argx->tasQ;
struct y_tasQ *historytasQ = argx->historytasQ;
struct list_y_TASK_T *l_task=NULL;
while(*(argx->go_on)){
l_task = pull_tasQ(tasQ);
printf("debug: is l_task NULL? = %d\n", l_task==NULL);
if(l_task){
if((l_task->value.status != TASK_DONE) && (l_task->value.func!=NULL))
l_task->value.ret = l_task->value.func(l_task->value.arg);
l_task->value.status = TASK_DONE;
}
pthread_mutex_lock(historytasQ->mut_tasQ);
if(l_task) append_list_y_TASK_T(historytasQ->list_tasQ, l_task);
pthread_mutex_unlock(historytasQ->mut_tasQ);
}
}
y_worker_t/src/y_worker_t/y_worker_t.c
+18
View File
@@ -0,0 +1,18 @@
#include "y_worker_t/y_worker_t.h"
GEN_LIST_ALL(y_WORKER_T)
void* execute_work(void* arg){
struct argWorker *argw = (struct argWorker*)arg;
struct argExecTasQ *argx=argw->argx;
struct list_y_WORKER_T * worker = argw->worker;
size_t id_thread=pthread_self();
do{
printf("debug: execute_task call : thread_id:%ld, self=%ld \n",worker->value.id,id_thread);
execute_task((void*)argx);
printf("debug: execute_task end, worker exec=%d \n",worker->value.exec);
}while(worker->value.exec);
//free_y_worker_t(worker);
}
y_worker_t/test/Makefile
+68
View File
@@ -0,0 +1,68 @@
NAME_TEST=is_good
CC=gcc
YTESTDIR=$(PWD)/../../ytest_t
YLISTDIR=$(PWD)/../../list_t
ROOT_DIR=$(PWD)/..
INCLUDE_DIR=$(ROOT_DIR)/include
CFLAGS=-I$(INCLUDE_DIR) -I$(YTESTDIR)/include_ytest/include -I$(YLISTDIR)/src
LDFLAGS=-L$(YTESTDIR) -lytest -lpthread -lm -lOpenCL
#SRC_DIR=$(ROOT_DIR)/src
#SRC=$(wildcard */*/*.c)
#SRC=$(wildcard **/**/*.c)
#OBJ=$(SRC:.c=.o)
#HEADS=$(OBJS:.o=.h)
#TEST_DIR=$(PWD)
EXECSRC=$(NAME_TEST).c
EXEC=launch_$(NAME_TEST)_m
YWORKSRC=$(ROOT_DIR)/src/y_worker_t/y_worker_t.c
YWORKSRC_O=$(YWORKSRC:.c=.o)
YTASKSRC=$(ROOT_DIR)/src/y_worker_t/y_task_t.c
YTASKSRC_O=$(YTASKSRC:.c=.o)
YLISTSRC=$(YLISTDIR)/src/list_t/list_t.c
YLISTSRC_O=$(YLISTSRC:.c=.o)
TOPTARGETS := all clean
DEPS=$(YTESTDIR) $(YLISTDIR) $(ROOT_DIR)
OBJ=$(YWORKSRC_O) $(YTASKSRC_O) $(YLISTSRC_O)
LIB_YTEST=$(YTESTDIR)/libytest.so
LIB_YWORK=$(PWD)/../libyworker.so
$(TOPTARGETS): $(DEPS)
$(DEPS):
$(MAKE) -C $@ $(MAKECMDGOALS)
all: $(EXEC) $(LIB_YWORK) $(LIB_YTEST)
$(EXEC): $(EXECSRC) $(OBJ)
$(CC) -o $@ $^ $(CFLAGS) $(LDFLAGS)
.PHONY: clean mrproper
clean:
rm -f $(OBJ)
mrproper: clean
rm -f $(EXEC)
run: $(EXEC)
$(EXEC) -h
y_worker_t/test/is_good.c
+157
View File
@@ -0,0 +1,157 @@
#include <stdio.h>
#include <stdlib.h>
#include <stdbool.h>
#include <time.h>
// for sleep !
#ifdef __linux__
#include <unistd.h>
#elif _WIN32
#include <windows.h>
#endif
#include "ftest/ftest.h"
#include "ftest/ftest_array.h"
#include "fmock/fmock.h"
//#include "permutation_t/permutation_t.h"
#include "y_worker_t/y_worker_t.h"
#include "y_worker_t/y_task_t.h"
//#include <netinet/in.h>
//#include <sys/worker.h>
//#include <arpa/inet.h>
#define VALGRIND_ 1
TEST(first){
struct y_tasQ tasQ0 ;
struct y_tasQ *tasQ = create_y_tasQ();
free_y_tasQ(tasQ);
}
TEST(worker){
struct main_list_y_WORKER_T * workers = create_var_list_y_WORKER_T();
struct y_worker_t w1;
w1.exec=GO_ON_WORKER;
push_back_list_y_WORKER_T(workers, w1);
struct y_worker_t w2;
w2.exec=KILL_WORKER;//only one task
push_back_list_y_WORKER_T(workers, w2);
free_all_var_list_y_WORKER_T(workers);
}
void* funcPrintSelf(void* arg){
LOG("begin fun call\n");
struct argExecTasQ *argx=(struct argExecTasQ*)arg;
int randomm=rand()%1000;
size_t thread_id=pthread_self();
LOG("my status=%d, idfunc=%d; threadid=%ld\n",*(argx->go_on), randomm, thread_id);
usleep(200000);
LOG("end func%d; in thread=%ld\n",randomm,thread_id);
}
TEST(threads){
srand(time(NULL));
struct main_list_y_WORKER_T * workers = create_var_list_y_WORKER_T();
struct argExecTasQ *argx = create_argExecTasQ();
for(int i=0; i<4; ++i){
struct y_worker_t w={.exec=GO_ON_WORKER, .id=i};
struct list_y_WORKER_T * liwo=malloc(sizeof(struct list_y_WORKER_T));
liwo->value=w;
//liwo->preview=NULL;
liwo->next=NULL;
append_list_y_WORKER_T(workers, liwo);
//push_back_list_y_WORKER_T(workers, w);
struct argWorker argw;
argw.argx = argx;
argw.worker=workers->end_list;
LOG("%d workers %ld created\n",4,argw.worker->value.id);
pthread_create(&(w.thread),NULL,execute_work,(void*)&argw);
//usleep(5000);
}
usleep(500000);
for(int i=0; i<10;++i)
{
struct y_task_t task = {
.func=funcPrintSelf,
.arg=argx,
.status=TASK_PENDING,
};
push_tasQ(argx->tasQ, task);
usleep(10000);
}
LOG("%d tasks created\n",40);
for(int i=4; i<8; ++i){
struct y_worker_t w={.exec=GO_ON_WORKER, .id=i};
struct list_y_WORKER_T * liwo=malloc(sizeof(struct list_y_WORKER_T));
liwo->value=w;
//liwo->preview=NULL;
liwo->next=NULL;
append_list_y_WORKER_T(workers, liwo);
//push_back_list_y_WORKER_T(workers, w);
struct argWorker argw;
argw.argx = argx;
argw.worker=workers->end_list;
pthread_create(&(w.thread),NULL,execute_work,(void*)&argw);
usleep(5000);
}
LOG("another %d workers created\n",4);
// usleep(400000);
for(move_current_to_begin_list_y_WORKER_T(workers); workers->current_list;increment_list_y_WORKER_T(workers)){
workers->current_list->value.exec=KILL_WORKER;
}
*(argx->go_on)=0;
for(move_current_to_begin_list_y_WORKER_T(workers); workers->current_list;increment_list_y_WORKER_T(workers)){
struct y_task_t task = {
.func=NULL,
.arg=argx,
.status=TASK_DONE,
};
push_tasQ(argx->tasQ, task);
}
for(move_current_to_begin_list_y_WORKER_T(workers); workers->current_list;increment_list_y_WORKER_T(workers)){
pthread_join(workers->current_list->value.thread ,NULL);
LOG("debug: JOIN donei, thread id:%ld\n",workers->current_list->value.id);
}
// usleep(5000000);
free_argExecTasQ(argx);
free_all_var_list_y_WORKER_T(workers);
}
int main(int argc, char **argv){
run_all_tests_args(argc, argv);
return 0;
}