trying to understand leak valgrind with opencl

tensor_t/src/tensor_t/cl_tensor_t.c

@@ -50,7 +50,8 @@
     ret = clGetDeviceIDs( *platform_id, CL_DEVICE_TYPE_DEFAULT, 0, NULL, &ret_num_devices);/*platform_id[0] */ \
     checkError(ret,__func__,"Error: Failed to get num devices  ");\
     device_id = malloc(sizeof(cl_device_id)*ret_num_devices);\
-    ret = clGetDeviceIDs( *platform_id, CL_DEVICE_TYPE_DEFAULT, ret_num_devices, device_id, NULL);/*platform_id[0] */ \
+    ret = clGetDeviceIDs( *platform_id,CL_DEVICE_TYPE_GPU /*CL_DEVICE_TYPE_DEFAULT*/, ret_num_devices, device_id, NULL);/*platform_id[0] */ \
+    if(ret == CL_DEVICE_NOT_FOUND) ret = clGetDeviceIDs( *platform_id,CL_DEVICE_TYPE_CPU /*CL_DEVICE_TYPE_DEFAULT*/, ret_num_devices, device_id, NULL);/*platform_id[0] */ \
     checkError(ret,__func__,"Error: Failed to get num devices  ");\
     size_t returned_size = 0;\
 	  size_t max_workgroup_size = 0;\

tensor_t/test_cl/is_good.c

@@ -33,7 +33,8 @@ TEST(rank){
   tensor_TYPE_FLOAT *tf = CREATE_TENSOR_TYPE_FLOAT(D);
   EXPECT_EQ(tf->dim->rank, 180);
 
-  free_dimension(D);
+//  free_dimension(D);
+  free_tensor_TYPE_FLOAT(tf);
 }
 
 void print_tensor_float(tensor_TYPE_FLOAT *M, char *msg){
@@ -204,7 +205,7 @@ TEST(tensorContractnProd_TYPE_FLOAT2 ){
   extern long int  PRECISION_TYPE_DOUBLE;
 
 TEST(cl_tensorContractnProd_TYPE_FLOAT2 ){
-  PRECISION_TYPE_DOUBLE=1000;
+  PRECISION_TYPE_FLOAT=100;
 
   dimension *d0=create_dim(3);
   dimension *d1=create_dim(3);
@@ -254,6 +255,7 @@ TEST(cl_tensorContractnProd_TYPE_FLOAT2 ){
 }
 
 TEST(cl_tensorContractnProd_TYPE_DOUBLE2 ){
+  PRECISION_TYPE_DOUBLE=1000;
   dimension *d0=create_dim(3);
   dimension *d1=create_dim(3);
 

y_socket_t/src/y_socket_t/y_socket_t.c

@@ -329,7 +329,8 @@ void* y_socket_handler_(void *arg){
       pthread_mutex_unlock(sock->mut_go_on);
 //      kill_all_workers(argw);
 //      printf("debug: kill_all\n");
-    } 
+    }
+
   }
   return NULL;
 }
@@ -436,6 +437,22 @@ void *y_socket_poll_fds(void *arg){
         char * temp_all_buf = NULL;
         copy_list_y_ptr_STRING_to_one_string(&temp_all_buf , m_str);
         push_back_list_TYPE_PTR(list_arg, temp_all_buf);
+  
+    if(strncmp(temp_all_buf,"update standby",14)==0){
+      //pthread_mutex_lock(sock->mut_go_on);
+      //sock->go_on = 0;
+      //pthread_mutex_unlock(sock->mut_go_on);
+      standby_all_workers(workers->begin_list->value->arg);
+//      printf("debug: kill_all\n");
+    }
+    else if(strncmp(temp_all_buf,"update wakeup",13)==0){
+      //pthread_mutex_lock(sock->mut_go_on);
+      //sock->go_on = 0;
+      //pthread_mutex_unlock(sock->mut_go_on);
+      wakeup_all_workers(workers->begin_list->value->arg);
+//      printf("debug: kill_all\n");
+    }
+    else{
         struct arg_handler_ *ptr_argHandl = malloc(sizeof(struct arg_handler_));
           ptr_argHandl->buf = temp_all_buf;
           ptr_argHandl->fds=fds;
@@ -450,6 +467,7 @@ void *y_socket_poll_fds(void *arg){
           .status=TASK_PENDING,
         };
         push_tasQ(argx->tasQ, task_handl);
+      }
 			///
 				//y_socket_handler_(temp_all_buf, fds, argSock);
 

y_worker_t/include/y_worker_t/y_worker_t.h

@@ -10,17 +10,19 @@
 
 #define KILL_WORKER 0
 #define GO_ON_WORKER 1
+#define STANDBY_WORKER 2
 
 #define WORKER_ON 1
 #define WORKER_OFF 0
+#define WORKER_STANDBY 2
 
 
 struct argWorker;
 
 typedef struct y_worker_t{
   int exec;
-  pthread_mutex_t *mut_exec;
+//  pthread_mutex_t *mut_exec;
-  pthread_cond_t *cond_exec;
+//  pthread_cond_t *cond_exec;
   int status;
   pthread_mutex_t *mut_worker;
   pthread_cond_t *cond_worker;
@@ -64,6 +66,9 @@ void* execute_work(void* arg);
 
 void kill_all_workers(struct argWorker *argw);
 
+void wakeup_all_workers ( struct argWorker *argw);
+void standby_all_workers ( struct argWorker *argw);
+
 //void kill_all_workers(struct main_list_ptr_y_WORKER_T * workers, struct argExecTasQ *argx);
 //void wait_workers(struct main_list_ptr_y_WORKER_T *workers);
 //void free_workers(struct main_list_ptr_y_WORKER_T *workers);

y_worker_t/src/y_worker_t/y_worker_t.c

@@ -18,10 +18,6 @@ ptr_y_WORKER_T create_ptr_y_WORKER_T(struct main_list_ptr_y_WORKER_T * workers,
                                      struct argExecTasQ *argx, int exec, int id ){
 	ptr_y_WORKER_T pworker = malloc(sizeof(y_WORKER_T));
 	pworker->exec=exec;
-	pworker->mut_exec = malloc(sizeof(pthread_mutex_t));
-	pthread_mutex_init(pworker->mut_exec, NULL);
-	pworker->cond_exec = malloc(sizeof(pthread_cond_t));
-	pthread_cond_init(pworker->cond_exec, NULL);
 	pworker->status=WORKER_OFF;
 	pworker->mut_worker = malloc(sizeof(pthread_mutex_t));
 	pthread_mutex_init(pworker->mut_worker, NULL);
@@ -43,18 +39,14 @@ ptr_y_WORKER_T create_ptr_y_WORKER_T(struct main_list_ptr_y_WORKER_T * workers,
   push_back_list_ptr_y_WORKER_T(workers, pworker);
   pthread_mutex_unlock(mut_workers);
   
-  //pthread_create(pworker->thread,NULL,execute_work,(void*)(pworker->arg));
+  pthread_create(pworker->thread,NULL,execute_work,(void*)(pworker->arg));
-  pthread_create(pworker->thread,NULL,execute_task,(void*)(pworker->arg->argx));
+  //pthread_create(pworker->thread,NULL,execute_task,(void*)(pworker->arg->argx));
   
 
 	return pworker;
 }
 void __free_ptr_y_WORKER_T(void* p_worker){
-	ptr_y_WORKER_T pworker = (struct y_worker_t*)p_worker;
+  ptr_y_WORKER_T pworker = (ptr_y_WORKER_T)p_worker;
-	pthread_mutex_destroy(pworker->mut_exec);
-	free(pworker->mut_exec);
-	pthread_cond_destroy(pworker->cond_exec);
-	free(pworker->cond_exec);
 	pthread_mutex_destroy(pworker->mut_worker);
 	free(pworker->mut_worker);
 	pthread_cond_destroy(pworker->cond_worker);
@@ -93,22 +85,34 @@ void* execute_work(void* arg){
   pworker->id_thread=id_thread;
   pthread_mutex_unlock(pworker->mut_worker);
 	pthread_cond_signal(pworker->cond_worker);
-    //printf("debug: ############################ execute_task call : thread_id:%ld, self=%ld \n",pworker->id,id_thread);
+   
-//  do{
+  pthread_mutex_lock(pworker->mut_worker);
+  while(pworker->exec == STANDBY_WORKER){
+    pworker->status=WORKER_STANDBY;
+    pthread_cond_wait(pworker->cond_worker, pworker->mut_worker);
+  }
+  exec = pworker->exec;
+  pworker->status = exec;
+  pthread_mutex_unlock(pworker->mut_worker);
+
+   //printf("debug: ############################ execute_task call : thread_id:%ld, self=%ld \n",pworker->id,id_thread);
+  while(exec){
     //printf("debug: execute_task call : thread_id:%ld, self=%ld \n",pworker->id,id_thread);
     execute_task((void*)argx);
     //printf("debug: <<<<>>>> execute_task end, worker exec=%d id:%ld self:%ld \n",exec,pworker->id, pworker->id_thread);
-/*
+    pthread_mutex_lock(pworker->mut_worker);
-  pthread_mutex_lock(pworker->mut_exec);
+    while(pworker->exec == STANDBY_WORKER){
-  while(pworker->exec == 1){
+      pworker->status=WORKER_STANDBY;
-    pthread_cond_wait(pworker->cond_exec, pworker->mut_exec);
+      pthread_cond_wait(pworker->cond_worker, pworker->mut_worker);
-  }
+    }
-  exec = pworker->exec;
+    exec = pworker->exec;
-  pthread_mutex_unlock(pworker->mut_exec);
+    pworker->status = exec;
-*/
+    pthread_mutex_unlock(pworker->mut_worker);
-    printf("debug: execute_task end, worker exec=%d id:%ld self:%ld \n",exec,pworker->id, pworker->id_thread);
-//  }while(exec);
 
+
+  };
+
+    printf("debug: execute_task end, worker exec=%d id:%ld self:%ld \n",exec,pworker->id, pworker->id_thread);
   
   pthread_mutex_lock(pworker->mut_worker);
   pworker->status=WORKER_OFF;
@@ -139,10 +143,6 @@ int check_worker_status(struct y_worker_t * pworker ){
 
 GEN_FUNC_PTR_LIST_FREE(ptr_y_WORKER_T){
   ptr_y_WORKER_T pworker = (struct y_worker_t*)arg;
-	pthread_mutex_destroy(pworker->mut_exec);
-	free(pworker->mut_exec);
-	pthread_cond_destroy(pworker->cond_exec);
-	free(pworker->cond_exec);
   pthread_mutex_destroy(pworker->mut_worker);
   free(pworker->mut_worker);
   pthread_cond_destroy(pworker->cond_worker);
@@ -150,7 +150,58 @@ GEN_FUNC_PTR_LIST_FREE(ptr_y_WORKER_T){
   free(pworker->thread);
   free(pworker->arg);
   free(pworker);
-} 
+}
+void wakeup_all_workers ( struct argWorker *argw){
+  	struct main_list_ptr_y_WORKER_T * workers  = argw->workers;
+	  struct argExecTasQ *argx = argw->argx;
+
+    pthread_mutex_lock(argx->tasQ->mut_tasQ);
+    (argx->go_on)=1;
+    pthread_mutex_unlock(argx->tasQ->mut_tasQ);
+
+//  for(move_current_to_begin_list_ptr_y_WORKER_T(workers); workers->current_list;increment_list_ptr_y_WORKER_T(workers)){}
+  for(struct list_ptr_y_WORKER_T * loc_current_list = workers->begin_list; loc_current_list; loc_current_list = loc_current_list->next ){
+      pthread_mutex_lock(loc_current_list->value->mut_worker);
+      loc_current_list->value->exec=GO_ON_WORKER;
+      pthread_mutex_unlock(loc_current_list->value->mut_worker);
+      pthread_cond_signal(loc_current_list->value->cond_worker);
+  }
+  printf("debug: wakeup_all_workers !\n"); 
+
+}
+void standby_all_workers ( struct argWorker *argw){
+  	struct main_list_ptr_y_WORKER_T * workers  = argw->workers;
+	  struct argExecTasQ *argx = argw->argx;
+
+//  for(move_current_to_begin_list_ptr_y_WORKER_T(workers); workers->current_list;increment_list_ptr_y_WORKER_T(workers)){}
+  for(struct list_ptr_y_WORKER_T * loc_current_list = workers->begin_list; loc_current_list; loc_current_list = loc_current_list->next ){
+      pthread_mutex_lock(loc_current_list->value->mut_worker);
+      loc_current_list->value->exec=STANDBY_WORKER;
+      pthread_mutex_unlock(loc_current_list->value->mut_worker);
+      pthread_cond_signal(loc_current_list->value->cond_worker);
+  }
+
+  pthread_mutex_lock(argx->tasQ->mut_tasQ);
+  (argx->go_on)=0;
+  pthread_mutex_unlock(argx->tasQ->mut_tasQ);
+
+//  for(move_current_to_begin_list_ptr_y_WORKER_T(workers); workers->current_list;increment_list_ptr_y_WORKER_T(workers)){}
+  for(struct list_ptr_y_WORKER_T * loc_current_list = workers->begin_list; loc_current_list; loc_current_list = loc_current_list->next ){
+///    if(check_worker_status(workers->current_list->value) == WORKER_ON)
+    {
+      struct y_task_t task = {
+        .func=NULL,
+        .arg=NULL,
+        .status=TASK_DONE,
+      };
+      push_tasQ(argx->tasQ, task);
+    }
+  }
+  
+  printf("debug: standby_all_workers !\n"); 
+
+}
+
 
 void kill_all_workers ( struct argWorker *argw){
   	struct main_list_ptr_y_WORKER_T * workers  = argw->workers;
@@ -158,10 +209,10 @@ void kill_all_workers ( struct argWorker *argw){
 
 //  for(move_current_to_begin_list_ptr_y_WORKER_T(workers); workers->current_list;increment_list_ptr_y_WORKER_T(workers)){}
   for(struct list_ptr_y_WORKER_T * loc_current_list = workers->begin_list; loc_current_list; loc_current_list = loc_current_list->next ){
-      pthread_mutex_lock(loc_current_list->value->mut_exec);
+      pthread_mutex_lock(loc_current_list->value->mut_worker);
       loc_current_list->value->exec=KILL_WORKER;
-      pthread_mutex_unlock(loc_current_list->value->mut_exec);
+      pthread_mutex_unlock(loc_current_list->value->mut_worker);
-      pthread_cond_signal(loc_current_list->value->cond_exec);
+      pthread_cond_signal(loc_current_list->value->cond_worker);
   }
 
   pthread_mutex_lock(argx->tasQ->mut_tasQ);
@@ -197,6 +248,7 @@ void kill_all_workers ( struct argWorker *argw){
   
   free_argExecTasQ(argx);
   
+  printf("debug: kill_all_workers !\n"); 
 
 
 }

y_worker_t/test/is_good.c

@@ -244,12 +244,20 @@ int nb_task=9;
 
  //getchar();
 
+#if 0
  struct pollfd fd[1];
  fd[0].fd=0;
  fd[0].events = POLLIN | POLLRDNORM | POLLRDBAND | POLLPRI;
  poll(fd, 1, -1);
+#endif 
+  usleep(2000000);
+
+	standby_all_workers(workers->begin_list->value->arg);
+  usleep(2000000);
+
+  wakeup_all_workers(workers->begin_list->value->arg);
+  usleep(2000000);
 
-//  usleep(200000000);
  
 	kill_all_workers(workers->begin_list->value->arg);
 

ytest_t/test/src/Makefile

@@ -1,11 +1,12 @@
 
 NAME_TEST=is_good
 CC=gcc
-LDFLAGS=-lpthread
 ROOT_DIR=$(shell pwd)
-INCLUDE_DIR=$(ROOT_DIR)
+YTESTDIR=$(ROOT_DIR)/../..
+INCLUDE_DIR=$(YTESTDIR)/include_ytest/include/
 PERMDIR=$(ROOT_DIR)/../../../ypermutation_t
 CFLAGS=-I$(INCLUDE_DIR) -I$(PERMDIR)/src/
+LDFLAGS=-L$(YTESTDIR)  -lpthread -lytest
 SRC_DIR=$(ROOT_DIR)
 SRC=$(wildcard src/*/*.c) 
 OBJ=$(SRC:.c=.o)