Counter in mutex added to spin for a while trying to acquire the lock

common/utils/threadPool/openmp.c

@@ -0,0 +1,92 @@
+#include <assert.h>
+#include <stdint.h>
+#include <time.h>
+#include <stdlib.h>
+#include <stdio.h>
+#include <unistd.h>
+
+#include <omp.h>
+
+  static inline
+int64_t time_now_us(void)
+{
+  struct timespec tms;
+
+  /* The C11 way */
+  /* if (! timespec_get(&tms, TIME_UTC))  */
+
+  /* POSIX.1-2008 way */
+  if (clock_gettime(CLOCK_REALTIME,&tms)) {
+    return -1;
+  }
+  /* seconds, multiplied with 1 million */
+  int64_t micros = tms.tv_sec * 1000000;
+  /* Add full microseconds */
+  int64_t const tv_nsec = tms.tv_nsec;
+  micros += tv_nsec/1000;
+  /* round up if necessary */
+  if (tv_nsec % 1000 >= 500) {
+    ++micros;
+  }
+  return micros;
+}
+
+static inline
+int64_t naive_fibonnacci(int64_t a)
+{
+  if(a < 2)
+    return a;
+  
+  return naive_fibonnacci(a-1) + naive_fibonnacci(a-2);
+}
+
+typedef struct{
+  int64_t a;
+  int64_t time;
+} pair_t;
+
+static __thread
+int acc = 0;
+
+static
+void do_work(void* arg)
+{
+  pair_t* a = (pair_t*)arg;
+  assert(a->a < 10);
+  //naive_fibonnacci(13 + a->a);
+  naive_fibonnacci(19 + a->a);
+
+  if((acc % 4096) == 0)
+    printf("%ld \n", time_now_us() - a->time);
+  acc += 1;
+
+  free(a);
+}
+
+int main()
+{
+
+  omp_set_num_threads(8);
+
+  const int64_t last = time_now_us() ;
+  #pragma omp parallel
+  {
+#pragma omp single
+    {
+  for(int i =0; i < 8*1024*1024; ++i){
+    pair_t* pa = malloc(sizeof(pair_t));
+    pa->a = 0; //i%10;
+    pa->time = last;
+ 
+    #pragma omp task
+    {
+    do_work(pa); 
+    }
+  }
+    }
+  }
+  sleep(15);
+
+  return 0;
+}
+

common/utils/threadPool/task_manager.c

@@ -0,0 +1,561 @@
+#include "task_manager.h"
+
+#include <assert.h> 
+#include <stdlib.h>
+#include <stdbool.h>
+#include <string.h>
+#include <errno.h>
+
+
+//////////////////////////////////////////////
+//////////////////////////////////////////////
+////////////////  Seq Ring  //////////
+//////////////////////////////////////////////
+//////////////////////////////////////////////
+
+typedef struct seq_ring_buf_s
+{
+  const size_t elt_size;
+  uint8_t* array;
+
+  size_t cap;
+  uint32_t head;
+  uint32_t tail;
+} seq_ring_t;
+
+// For working correctly, maintain the default elements to a multiple of 2
+#define DEFAULT_ELM 32768 
+
+inline static
+uint32_t mask(uint32_t cap, uint32_t val)
+{
+  return val & (cap-1);
+}
+
+inline static
+size_t seq_ring_size(seq_ring_t* r)
+{
+  return r->head - r->tail; 
+};
+
+static inline 
+bool full(seq_ring_t* r)
+{
+  return seq_ring_size(r) == r->cap -1;
+}
+
+static
+void enlarge_buffer(seq_ring_t* r)
+{
+  assert(r != NULL);
+  assert(full(r));
+
+  const uint32_t factor = 2;
+  uint8_t* tmp_buffer = calloc(r->cap * factor, r->elt_size);
+  assert(tmp_buffer != NULL);
+
+  const uint32_t head_pos = mask(r->cap, r->head);
+  const uint32_t tail_pos = mask(r->cap, r->tail);
+
+  if(head_pos > tail_pos){
+    memcpy(tmp_buffer, r->array + tail_pos*r->elt_size , (head_pos-tail_pos)*r->elt_size );
+  } else {
+    memcpy(tmp_buffer, r->array + tail_pos*r->elt_size , (r->cap-tail_pos)*r->elt_size);
+    memcpy(tmp_buffer + (r->cap-tail_pos)*r->elt_size, r->array, head_pos*r->elt_size);
+  }
+  r->cap *= factor;
+  free(r->array);
+  r->array = tmp_buffer;
+  r->tail = 0;
+  r->head = r->cap/2 - 1;
+}
+
+static
+void seq_ring_init(seq_ring_t* r, size_t elt_size)
+{
+  assert(r != NULL);
+  uint8_t* tmp_buffer = calloc(DEFAULT_ELM, elt_size); 
+  assert(tmp_buffer != NULL);
+  seq_ring_t tmp = {.elt_size = elt_size, .array = tmp_buffer, .head = 0, .tail = 0, .cap = DEFAULT_ELM};
+  memcpy(r, &tmp, sizeof(seq_ring_t));
+
+}
+
+static
+void* seq_ring_front(seq_ring_t* r)
+{
+  return r->array + mask(r->cap, r->tail)*r->elt_size;
+};
+
+static
+void* seq_ring_next(seq_ring_t* r, void* it)
+{
+  assert(r != NULL);
+  assert(it != NULL);
+  const uint32_t pos = ((uint8_t*)it - r->array)/r->elt_size;
+
+  if(pos + 1 == r->cap) 
+    return r->array;
+  return it + r->elt_size;
+}
+
+static inline
+void* seq_ring_end(seq_ring_t* r)
+{
+  return r->array + mask(r->cap, r->head)*r->elt_size;
+}
+
+static inline
+void* seq_ring_at(seq_ring_t* r, uint32_t pos)
+{
+  return r->array + mask(r->cap, r->tail + pos)*r->elt_size;
+}
+
+static inline
+int32_t seq_ring_dist(seq_ring_t* r, void* first, void* last)
+{
+  // Precondition: random access iterators belonging to a sequence.
+  assert(r != NULL);
+  assert(first != NULL);
+  assert(last != NULL);
+
+  const uint32_t last_pos = ((uint8_t*)last - r->array) / r->elt_size; 
+  const uint32_t first_pos = ((uint8_t*)first - r->array) / r->elt_size; 
+  if(last_pos > first_pos)
+    return last_pos - first_pos; 
+
+  return (r->cap - first_pos) + last_pos;
+};
+
+static inline
+bool seq_ring_equal(void* it_1,void* it_2)
+{
+  return (uint8_t*)it_1 == (uint8_t*)it_2;  
+}
+
+typedef void (*seq_free_func)(void*);
+
+static
+void seq_ring_free(seq_ring_t* cb, seq_free_func fp)
+{
+  assert(cb != NULL);
+  if(fp != NULL){
+    void* start_it = seq_ring_front(cb);
+    void* end_it = seq_ring_end(cb);
+    while(start_it != end_it){
+      fp(start_it);
+      start_it = seq_ring_next(cb, start_it);
+    }
+  }
+
+  free(cb->array);
+}
+
+static
+void seq_ring_push_back(seq_ring_t* r, uint8_t* data, size_t len)
+{
+  assert(r != NULL);
+  assert(len == r->elt_size);
+
+  if(full(r))
+    enlarge_buffer(r);
+  
+  const uint32_t pos = mask(r->cap, r->head);
+  memcpy(r->array + r->elt_size*pos, data, r->elt_size);
+  r->head += 1;
+};
+
+static
+void seq_ring_erase(seq_ring_t* r, void* it_start, void* it_end)
+{
+  assert(it_start == r->array + mask(r->cap,r->tail)*r->elt_size && "Only deletions from tail implemented");
+  if(it_start == it_end){
+    return; // closed semi-open range [ ) 
+  }
+  const uint32_t start_pos = ((uint8_t*)it_start - r->array)/r->elt_size; 
+  const uint32_t end_pos = ((uint8_t*)it_end - r->array)/r->elt_size; 
+  assert(end_pos <= r->cap);
+  assert(end_pos != start_pos);
+
+  if(end_pos > start_pos){
+    memset(it_start, 0, (end_pos - start_pos)*r->elt_size);
+    r->tail += end_pos - start_pos;
+  } else {
+    assert(end_pos < start_pos); 
+    memset(it_start, 0, (r->cap - start_pos)*r->elt_size);
+    memset(r->array, 0, end_pos*r->elt_size);
+    r->tail +=  r->cap - start_pos + end_pos ;
+  }
+  assert(mask(r->cap, r->tail) == end_pos);
+}
+
+static
+void seq_ring_move(seq_ring_t* r, void* it, void* end, void* out)
+{
+  assert(it == r->array + mask(r->cap,r->tail)*r->elt_size && "Only deletions from tail implemented");
+  if(it == end){
+    return; // closed semi-open range [ ) 
+  }
+  const uint32_t start_pos = ((uint8_t*)it - r->array)/r->elt_size; 
+  const uint32_t end_pos = ((uint8_t*)end - r->array)/r->elt_size; 
+  assert(end_pos <= r->cap);
+  assert(end_pos != start_pos);
+
+  if(end_pos > start_pos){
+    memcpy(out, it, (end_pos - start_pos)*r->elt_size);
+    memset(it, 0, (end_pos - start_pos)*r->elt_size);
+    r->tail += end_pos - start_pos;
+  } else {
+    assert(end_pos < start_pos); 
+    memcpy(out, it, (r->cap - start_pos)*r->elt_size); 
+    memset(it, 0, (r->cap - start_pos)*r->elt_size);
+
+    memcpy(out, r->array, end_pos*r->elt_size); // (r->cap - start_pos)*r->elt_size); 
+    memset(r->array, 0, end_pos*r->elt_size);
+    r->tail +=  r->cap - start_pos + end_pos ;
+  }
+  assert(mask(r->cap, r->tail) == end_pos);
+}
+
+
+static
+void seq_ring_swap(seq_ring_t* cb, void* it_1, void* it_2)
+{
+  assert(0!=0 && "Not implemented");
+  assert(cb != NULL);
+  assert(it_1 != NULL);
+  assert(it_2 != NULL);
+};
+
+
+//////////////////////////////////////////////
+//////////////////////////////////////////////
+//////////////// End Seq Ring  //////////
+//////////////////////////////////////////////
+//////////////////////////////////////////////
+
+
+
+
+//////////////////////////////////////////////
+//////////////////////////////////////////////
+//////////////// Notification Queue //////////
+//////////////////////////////////////////////
+//////////////////////////////////////////////
+
+typedef struct {
+  pthread_mutex_t mtx;
+  pthread_cond_t cv;
+  seq_ring_t r;
+  int done;
+} not_q_t;
+
+typedef struct{
+  task_t t[4096];
+  int len;
+  bool success;
+} ret_try_t;
+
+static
+void init_not_q(not_q_t* q)
+{
+  assert(q != NULL);
+
+  q->done = 0;
+  seq_ring_init(&q->r, sizeof(task_t));
+
+  pthread_mutexattr_t attr = {0};
+#ifdef _DEBUG
+  int const rc_mtx = pthread_mutexattr_settype(&attr, PTHREAD_MUTEX_ERRORCHECK);
+  assert(rc_mtx == 0);
+#endif
+  int rc = pthread_mutex_init(&q->mtx, &attr);
+  assert(rc == 0 && "Error while creating the mtx");
+
+  pthread_condattr_t* c_attr = NULL; 
+  rc = pthread_cond_init(&q->cv, c_attr);
+  assert(rc == 0);
+}
+
+static
+void free_not_q(not_q_t* q, void (*clean)(void*) )
+{
+  assert(q != NULL);
+  assert(q->done == 1);
+
+  seq_ring_free(&q->r, clean);
+
+  int rc = pthread_mutex_destroy(&q->mtx);
+  assert(rc == 0);
+
+  rc = pthread_cond_destroy(&q->cv);
+  assert(rc == 0);
+}
+
+static
+bool try_push_not_q(not_q_t* q, task_t t)
+{
+  assert(q != NULL);
+  assert(q->done == 0 || q->done ==1);
+  assert(t.func != NULL);
+  assert(t.args != NULL);
+
+  if(pthread_mutex_trylock(&q->mtx ) != 0)
+    return false;
+
+  seq_ring_push_back(&q->r, (uint8_t*)&t, sizeof(task_t));
+
+  int rc = pthread_mutex_unlock(&q->mtx);
+  assert(rc == 0);
+
+  pthread_cond_signal(&q->cv);
+
+  return true;
+}
+
+static
+void push_not_q(not_q_t* q, task_t t)
+{
+  assert(q != NULL);
+  assert(q->done == 0 || q->done ==1);
+  assert(t.func != NULL);
+  //assert(t.args != NULL);
+
+  int rc = pthread_mutex_lock(&q->mtx);
+  assert(rc == 0);
+
+  seq_ring_push_back(&q->r, (void*)&t, sizeof(task_t));
+
+  pthread_mutex_unlock(&q->mtx);
+
+  pthread_cond_signal(&q->cv);
+}
+
+static
+ret_try_t try_pop_not_q(not_q_t* q)
+{
+  assert(q != NULL);
+
+  ret_try_t  ret = {.success = false}; 
+
+  int rc = pthread_mutex_trylock(&q->mtx);
+  assert(rc == 0 || rc == EBUSY);
+
+  if(rc == EBUSY)
+    return ret;
+
+  assert(q->done == 0 || q->done ==1);
+
+  size_t sz = seq_ring_size(&q->r); 
+  if(sz == 0){
+    rc = pthread_mutex_unlock(&q->mtx);
+    assert(rc == 0);
+
+    return ret;
+  }
+
+  ret.len = sz < 4096 ? sz : 4096;
+  for(int i = 0; i < ret.len; ++i){
+  void* it = seq_ring_at(&q->r, i);
+    memcpy(&ret.t[i], it, sizeof(task_t)); 
+  }
+
+//   void* it = seq_ring_front(&q->r);
+//    assert(it != seq_ring_end(&q->r));
+//    memcpy(&ret.t, it, sizeof(task_t)); 
+
+  seq_ring_erase(&q->r, seq_ring_front(&q->r), seq_ring_at(&q->r, ret.len) );
+
+
+//  seq_ring_move(&q->r, seq_ring_front(&q->r), seq_ring_at(&q->r, ret.len), ret.t);
+
+  assert(sz == ret.len + seq_ring_size(&q->r)); 
+
+  rc = pthread_mutex_unlock(&q->mtx);
+  assert(rc == 0);
+  ret.success = true; 
+
+  return ret;
+}
+
+static
+bool pop_not_q(not_q_t* q, ret_try_t* out)
+{
+  assert(q != NULL);
+  assert(out != NULL);
+
+  pthread_mutex_lock(&q->mtx);
+  assert(q->done == 0 || q->done ==1);
+
+  while(seq_ring_size(&q->r) == 0 && q->done == 0)
+    pthread_cond_wait(&q->cv , &q->mtx);
+
+  if(q->done == 1){
+    int rc = pthread_mutex_unlock(&q->mtx);
+    assert(rc == 0);
+    return false;
+  }
+
+  const size_t sz =  seq_ring_size(&q->r);
+
+  out->len = sz < 4096 ? sz : 4096; 
+  for(int i =0; i < out->len; ++i){
+     void* it = seq_ring_at(&q->r, i);
+    assert(it != seq_ring_end(&q->r));
+    memcpy(&out->t[i], it, sizeof(task_t)); 
+  }
+
+//   void* it = seq_ring_at(&q->r, 0);
+//    assert(it != seq_ring_end(&q->r));
+//    memcpy(&out->t, it, sizeof(task_t)); 
+
+  seq_ring_erase(&q->r, seq_ring_front(&q->r), seq_ring_at(&q->r, out->len));
+
+//  seq_ring_move(&q->r, seq_ring_front(&q->r), seq_ring_at(&q->r, out->len), out->t);
+
+
+  int rc = pthread_mutex_unlock(&q->mtx);
+  assert(rc == 0);
+
+  return true;
+}
+
+static
+void done_not_q(not_q_t* q)
+{
+  assert(q != NULL);
+
+  int rc = pthread_mutex_lock(&q->mtx);
+  assert(rc == 0);
+  
+  q->done = 1;
+
+  rc = pthread_mutex_unlock(&q->mtx);
+  assert(rc == 0);
+
+  rc = pthread_cond_signal(&q->cv);
+  assert(rc == 0);
+}
+
+//////////////////////////////////////////////
+//////////////////////////////////////////////
+//////////////// End Notification Queue //////////
+//////////////////////////////////////////////
+//////////////////////////////////////////////
+
+
+
+//////////////////////////////////////////////
+//////////////////////////////////////////////
+/////////// Task Manager /////////////////////
+//////////////////////////////////////////////
+//////////////////////////////////////////////
+
+
+typedef struct{
+  task_manager_t* man;
+  int idx;
+} task_thread_args_t;
+
+static
+void* worker_thread(void* arg)
+{
+  assert(arg != NULL);
+
+  task_thread_args_t* args = (task_thread_args_t*)arg; 
+  int const idx = args->idx;
+  task_manager_t* man = args->man;
+
+  uint32_t const len = man->len_thr;
+  int const num_it = 3*(man->len_thr + idx); 
+
+  not_q_t* q_arr = (not_q_t*)man->q_arr;
+  ret_try_t ret; 
+  for(;;){
+    for(int i = idx; i < num_it; ++i){
+      ret = try_pop_not_q(&q_arr[i%len]);
+      if(ret.success == true){
+        break;
+      } 
+    }
+
+    if(ret.success == false && pop_not_q(&q_arr[idx], &ret) == false)
+      break;
+    
+    for(int i =0; i < ret.len; ++i )
+      ret.t[i].func(ret.t[i].args); 
+  }
+
+  free(args);
+  return NULL;
+}
+
+void init_task_manager(task_manager_t* man, uint32_t num_threads)
+{
+  assert(man != NULL);
+  assert(num_threads > 0 && num_threads < 33 && "Do you have zero or more than 32 processors??");
+
+  man->q_arr = calloc(num_threads, sizeof(not_q_t));
+  assert(man->q_arr != NULL && "Memory exhausted");
+   
+  not_q_t* q_arr = (not_q_t*)man->q_arr;
+
+  for(uint32_t i = 0; i < num_threads; ++i){
+    init_not_q(&q_arr[i]);   
+  }
+
+  man->t_arr = calloc(num_threads, sizeof(pthread_t));
+  assert(man->t_arr != NULL && "Memory exhausted" );
+  man->len_thr = num_threads;
+
+  for(uint32_t i = 0; i < num_threads; ++i){
+    task_thread_args_t* args = malloc(sizeof(task_thread_args_t) ); 
+    args->idx = i;
+    args->man = man;
+
+    int rc = pthread_create(&man->t_arr[i], NULL, worker_thread, args);
+    assert(rc == 0);
+  }
+
+  man->index = 0;
+}
+
+void free_task_manager(task_manager_t* man, void (*clean)(void*))
+{
+  not_q_t* q_arr = (not_q_t*)man->q_arr;
+  for(uint32_t i = 0; i < man->len_thr; ++i){
+    done_not_q(&q_arr[i]);
+  }
+
+  for(uint32_t i = 0; i < man->len_thr; ++i){
+    pthread_join(man->t_arr[i], NULL); 
+  }
+
+  for(uint32_t i = 0; i < man->len_thr; ++i){
+    free_not_q(&q_arr[i], clean); 
+  }
+
+  free(man->q_arr);
+
+  free(man->t_arr);
+}
+
+void async_task_manager(task_manager_t* man, task_t t)
+{
+  assert(man != NULL);
+  assert(t.func != NULL);
+  //assert(t.args != NULL);
+
+  uint64_t const index = man->index++;
+//  atomic_fetch_add_explicit(&man->index, 1, memory_order_relaxed);
+
+  not_q_t* q_arr = (not_q_t*)man->q_arr;
+  for(uint32_t i = 0; i < man->len_thr; ++i){
+    if(try_push_not_q(&q_arr[(i+index) % man->len_thr], t)){
+      return;
+    }
+  }
+
+  push_not_q(&q_arr[index%man->len_thr], t);
+}
+

common/utils/threadPool/task_manager.h

@@ -0,0 +1,27 @@
+#ifndef TASK_MANAGER_WORKING_STEALING_H
+#define TASK_MANAGER_WORKING_STEALING_H 
+
+#include <pthread.h>
+#include <stdatomic.h>
+#include <stdint.h>
+
+typedef struct{
+  void* args;
+  void (*func)(void* args);
+} task_t;
+
+typedef struct{
+  pthread_t* t_arr;
+  size_t len_thr;
+  atomic_uint_fast64_t index;
+  void* q_arr;
+} task_manager_t;
+
+void init_task_manager(task_manager_t* man, uint32_t num_threads);
+
+void free_task_manager(task_manager_t* man, void (*clean)(void* args) );
+
+void async_task_manager(task_manager_t* man, task_t t);
+
+#endif
+

common/utils/threadPool/thread-pool.c

@@ -23,6 +23,7 @@
 
 
 #define _GNU_SOURCE
+#include <assert.h>
 #include <sched.h>
 #include <sys/types.h>
 #include <sys/stat.h>
@@ -194,99 +195,82 @@ void processing(void *arg) {
   //printf("done: %d, %s, in thr %ld\n",in->id, in->txt,pthread_self() );
 }
 
-int main() {
-  notifiedFIFO_t myFifo;
-  initNotifiedFIFO(&myFifo);
-  pushNotifiedFIFO(&myFifo,newNotifiedFIFO_elt(sizeof(struct testData), 1234,NULL,NULL));
+static inline
+int64_t time_now_us(void)
+{
+  struct timespec tms;
 
-  for(int i=10; i>1; i--) {
-    pushNotifiedFIFO(&myFifo,newNotifiedFIFO_elt(sizeof(struct testData), 1000+i,NULL,NULL));
+  /* The C11 way */
+  /* if (! timespec_get(&tms, TIME_UTC))  */
+
+  /* POSIX.1-2008 way */
+  if (clock_gettime(CLOCK_REALTIME,&tms)) {
+    return -1;
   }
+  /* seconds, multiplied with 1 million */
+  int64_t micros = tms.tv_sec * 1000000;
+  /* Add full microseconds */
+  micros += tms.tv_nsec/1000;
+  /* round up if necessary */
+  if (tms.tv_nsec % 1000 >= 500) {
+    ++micros;
+  }
+  return micros;
+}
 
-  displayList(&myFifo);
-  notifiedFIFO_elt_t *tmp=pullNotifiedFIFO(&myFifo);
-  printf("pulled: %lu\n", tmp->key);
-  displayList(&myFifo);
-  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);
+static
+int64_t naive_fibonnacci(int64_t a)
+{
+  if(a < 2)
+    return 1;
+  
+  return naive_fibonnacci(a-1) + naive_fibonnacci(a-2);
+}
 
-  do {
-    tmp=pollNotifiedFIFO(&myFifo);
+typedef struct{
+  int64_t a;
+  int64_t time;
+}pair_t;
 
-    if (tmp) {
-      printf("pulled: %lu\n", tmp->key);
-      displayList(&myFifo);
-    } else
-      printf("Empty list \n");
-  } while(tmp);
 
+
+static
+__thread int acc = 0;
+
+static
+void do_work(void* arg)
+{
+  pair_t* a = (pair_t*)arg;
+  assert(a->a < 10);
+  naive_fibonnacci(19 + a->a);
+
+  if((acc % 4096) == 0)
+    printf("%ld \n", time_now_us() -a->time   );  
+  acc += 1;
+
+//  free(a);
+}
+
+int main(void)
+{
   tpool_t  pool;
-  char params[]="1,2,3,4,5";
-  initTpool(params,&pool, true);
+  char params[]="0,1,2,3,4,5,6,7";
+  initTpool(params,&pool, false);
   notifiedFIFO_t worker_back;
   initNotifiedFIFO(&worker_back);
 
   sleep(1);
-  int cumulProcessTime=0, cumulTime=0;
-  struct timespec st,end;
-  clock_gettime(CLOCK_MONOTONIC, &st);
-  int nb_jobs=4;
-  for (int i=0; i <1000 ; i++) {
-    int parall=nb_jobs;
-    for (int j=0; j <parall ; j++) {
-      notifiedFIFO_elt_t *work=newNotifiedFIFO_elt(sizeof(struct testData), i, &worker_back, processing);
-      struct testData *x=(struct testData *)NotifiedFifoData(work);
-      x->id=i;
+
+  int64_t now = time_now_us();
+  for (int i=0; i < 8*1024*1024; i++) {
+      notifiedFIFO_elt_t *work=newNotifiedFIFO_elt(sizeof(pair_t), i, &worker_back, do_work);
+      pair_t* x =( pair_t *)NotifiedFifoData(work);
+      x->a=0; //i%10;
+      x->time = now; 
       pushTpool(&pool, work);
     }
-    int sleepmax=0;
-    while (parall) {
-      tmp=pullTpool(&worker_back,&pool);
-      if (tmp) {
-	parall--;
-	struct testData *dd=NotifiedFifoData(tmp);
-	if (dd->sleepTime > sleepmax)
-	  sleepmax=dd->sleepTime;
-	delNotifiedFIFO_elt(tmp);
-      }
-    }
-    cumulProcessTime+=sleepmax;
-  }
-  clock_gettime(CLOCK_MONOTONIC, &end);
-  long long dur=(end.tv_sec-st.tv_sec)*1000*1000+(end.tv_nsec-st.tv_nsec)/1000;
-  printf("In µs, Total time per group of %d job:%lld, work time per job %d, overhead per job %lld\n",
-	 nb_jobs, dur/1000, cumulProcessTime/1000, (dur-cumulProcessTime)/(1000*nb_jobs));
-
-	/*	
-  for (int i=0; i <1000 ; i++) {
-    notifiedFIFO_elt_t *work=newNotifiedFIFO_elt(sizeof(struct testData), i, &worker_back, processing);
-    struct testData *x=(struct testData *)NotifiedFifoData(work);
-    x->id=i;
-    pushTpool(&pool, work);
-  }
-
-  do {
-    tmp=pullTpool(&worker_back,&pool);
-
-    if (tmp) {
-      struct testData *dd=NotifiedFifoData(tmp);
-      printf("Result: %s\n",dd->txt);
-      delNotifiedFIFO_elt(tmp);
-    } else
-      printf("Empty list \n");
-
-    abortTpoolJob(&pool,510);
-  } while(tmp);
-	*/
-  return 0;
+  
+  sleep(20);
 }
+
 #endif

common/utils/threadPool/thread-pool.h

@@ -33,20 +33,84 @@
 #include "common/utils/time_meas.h"
 #include "common/utils/system.h"
 
+#include <stdatomic.h>
+
+typedef struct{
+  _Atomic bool lock;  
+} spinlock_t;
+
+static
+void lock_spinlock(spinlock_t* s)
+{
+    for (;;) {
+      // Optimistically assume the lock is free on the first try
+      if(!atomic_exchange_explicit(&s->lock, true, memory_order_acquire) ){
+        return;
+      }
+
+      // Wait for lock to be released without generating cache misses
+      while (atomic_load_explicit(&s->lock, memory_order_relaxed)){
+        // Issue X86 PAUSE or ARM YIELD instruction to reduce contention between
+        // hyper-threads
+        __builtin_ia32_pause();
+      }
+    }
+
+}
+
+static
+bool try_lock_spinlock(spinlock_t* s)
+{
+    // First do a relaxed load to check if lock is free in order to prevent
+    // unnecessary cache misses if someone does while(!try_lock())
+    return !atomic_load_explicit(&s->lock, memory_order_relaxed) &&
+           !atomic_exchange_explicit(&s->lock, true, memory_order_acquire);
+}
+
+
+
+static
+void unlock_spinlock(spinlock_t* s)
+{
+  atomic_store_explicit(&s->lock, false, memory_order_release);
+}
+
 #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 mutexlock(mutex)   {  int mtx_cnt = 2048;\
+                              int ret_mtx_cnt = 1;\
+                              while(ret_mtx_cnt != 0 &&  mtx_cnt != 0){\
+                               ret_mtx_cnt = pthread_mutex_trylock(&mutex);\
+                               mtx_cnt -=1; \
+                              }\
+                              if(ret_mtx_cnt != 0){\
+                              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 mutexinit(mutex)   { (&mutex)->lock = false }
+//#define mutexlock(mutex)   {   lock_spinlock(&mutex); }
+//#define mutextrylock(mutex)   { try_lock_spinlock(&mutex); } 
+//#define mutexunlock(mutex) { unlock_spinlock(&mutex); }
+
+
+
+
+#define condinit(signal)   {int ret=pthread_cond_init(&signal,NULL); \
+                            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); \
@@ -54,6 +118,19 @@
 #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
@@ -71,6 +148,8 @@ typedef struct notifiedFIFO_s {
   notifiedFIFO_elt_t *outF;
   notifiedFIFO_elt_t *inF;
   pthread_mutex_t lockF;
+// spinlock_t lockF;
+
   pthread_cond_t  notifF;
   bool abortFIFO; // if set, the FIFO always returns NULL -> abort condition
 } notifiedFIFO_t;

common/utils/threadPool/tm_main.c

@@ -0,0 +1,117 @@
+/*
+MIT License
+
+Copyright (c) 2022 Mikel Irazabal
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.
+*/
+
+#include "task_manager.h"
+
+#include <assert.h>
+#include <stdlib.h>
+#include <stdio.h>
+#include <time.h>
+#include <pthread.h>
+#include <sys/types.h>
+#include <sys/syscall.h>
+#include <unistd.h>
+
+static inline
+int64_t time_now_us(void)
+{
+  struct timespec tms;
+
+  /* The C11 way */
+  /* if (! timespec_get(&tms, TIME_UTC))  */
+
+  /* POSIX.1-2008 way */
+  if (clock_gettime(CLOCK_REALTIME,&tms)) {
+    return -1;
+  }
+  /* seconds, multiplied with 1 million */
+  int64_t micros = tms.tv_sec * 1000000;
+  /* Add full microseconds */
+  int64_t const tv_nsec = tms.tv_nsec;
+  micros += tv_nsec/1000;
+  /* round up if necessary */
+  if (tv_nsec % 1000 >= 500) {
+    ++micros;
+  }
+  return micros;
+}
+
+static inline
+int64_t naive_fibonnacci(int64_t a)
+{
+  if(a < 2)
+    return a;
+  
+  return naive_fibonnacci(a-1) + naive_fibonnacci(a-2);
+}
+
+typedef struct{
+  int64_t a;
+  int64_t time;
+} pair_t;
+
+static __thread
+int acc = 0;
+
+static
+void do_work(void* arg)
+{
+  pair_t* a = (pair_t*)arg;
+  assert(a->a < 10);
+  //naive_fibonnacci(13 + a->a);
+  naive_fibonnacci(19 + a->a);
+
+  if((acc % 4096) == 0)
+    printf("%ld \n", time_now_us() - a->time);
+  acc += 1;
+
+  free(a);
+}
+
+void clean_func(void* arg)
+{
+  task_t* t = (task_t*)arg;
+  free((pair_t*)t->args);
+}
+
+int main()
+{
+  task_manager_t man;
+  init_task_manager(&man, 8);
+
+  const int64_t last = time_now_us() ;
+  for(int i = 0; i < 8*1024*1024; ++i){
+    pair_t* pa = malloc(sizeof(pair_t));
+    pa->a = 0; //i%10;
+    pa->time = last;
+    task_t t = {.args = pa, t.func = do_work};
+    async_task_manager(&man, t);
+  }
+
+  sleep(20); 
+  free_task_manager(&man, clean_func);
+
+  return EXIT_SUCCESS;
+}
+