Implement allocator

Author: BenBrock

include/binsparse/array.h

@@ -1,6 +1,7 @@
 #pragma once
 
 #include <assert.h>
+#include <binsparse/detail/allocator.h>
 #include <binsparse/detail/shm_tools.h>
 #include <binsparse/types.h>
 #include <stdlib.h>
@@ -10,27 +11,26 @@ typedef struct bsp_array_t {
   void* data;
   size_t size;
   bsp_type_t type;
-  bool shmat_memory;
-  bsp_shm_t shm;
+  bsp_allocator_t allocator;
 } bsp_array_t;
 
 bsp_array_t bsp_construct_default_array_t() {
   bsp_array_t array;
   array.data = NULL;
   array.size = 0;
-  array.shmat_memory = false;
+  array.allocator = bsp_default_allocator;
   return array;
 }
 
 bsp_array_t bsp_construct_array_t(size_t size, bsp_type_t type) {
   size_t byte_size = size * bsp_type_size(type);
 
   bsp_array_t array;
-  array.data = malloc(byte_size);
+  array.allocator = bsp_default_allocator;
+  array.data = array.allocator.malloc(byte_size);
   assert(array.data != NULL);
   array.size = size;
   array.type = type;
-  array.shmat_memory = false;
 
   return array;
 }
@@ -49,6 +49,7 @@ bsp_array_t bsp_complex_array_to_fp(bsp_array_t other) {
   bsp_array_t array;
   array.data = other.data;
   array.size = other.size * 2;
+  array.allocator = other.allocator;
 
   if (other.type == BSP_COMPLEX_FLOAT32) {
     array.type = BSP_FLOAT32;
@@ -65,6 +66,7 @@ bsp_array_t bsp_fp_array_to_complex(bsp_array_t other) {
   bsp_array_t array;
   array.data = other.data;
   array.size = other.size / 2;
+  array.allocator = other.allocator;
 
   if (other.type == BSP_FLOAT32) {
     array.type = BSP_COMPLEX_FLOAT32;
@@ -76,11 +78,7 @@ bsp_array_t bsp_fp_array_to_complex(bsp_array_t other) {
 }
 
 void bsp_destroy_array_t(bsp_array_t array) {
-  if (array.shmat_memory == false) {
-    free(array.data);
-  } else {
-    bsp_shm_detach(array.data);
-  }
+  array.allocator.free(array.data);
 }
 
 bool bsp_array_equal(bsp_array_t x, bsp_array_t y) {

include/binsparse/detail/allocator.h

@@ -0,0 +1,11 @@
+#pragma once
+
+#include <stdlib.h>
+
+typedef struct bsp_allocator_t {
+  void* (*malloc)(size_t);
+  void (*free)(void*);
+} bsp_allocator_t;
+
+const static bsp_allocator_t bsp_default_allocator = {.malloc = malloc,
+                                                      .free = free};

include/binsparse/detail/shm_tools.h

@@ -1,5 +1,6 @@
 #pragma once
 
+#include <binsparse/detail/allocator.h>
 #include <stddef.h>
 #include <stdio.h>
 #include <sys/ipc.h>
@@ -42,3 +43,20 @@ void* bsp_shm_attach(bsp_shm_t shm) {
 void bsp_shm_detach(void* data) {
   shmdt(data);
 }
+
+void* bsp_shm_malloc(size_t size) {
+  bsp_shm_t shm_id = bsp_shm_new(size);
+
+  void* ptr = bsp_shm_attach(shm_id);
+
+  bsp_shm_delete(shm_id);
+
+  return ptr;
+}
+
+void bsp_shm_free(void* ptr) {
+  bsp_shm_detach(ptr);
+}
+
+const static bsp_allocator_t bsp_shm_allocator = {.malloc = bsp_shm_malloc,
+                                                  .free = bsp_shm_free};

include/binsparse/hdf5_wrapper.h

@@ -9,6 +9,10 @@
 
 #include <binsparse/detail/shm_tools.h>
 
+#if __STDC_VERSION__ >= 201112L
+#include <stdatomic.h>
+#endif
+
 // Write an array to a dataset / file
 // Returns 0 on success, nonzero on error.
 int bsp_write_array(hid_t f, const char* label, bsp_array_t array,
@@ -65,6 +69,7 @@ int bsp_write_array(hid_t f, const char* label, bsp_array_t array,
   return 0;
 }
 
+#if __STDC_VERSION__ >= 201112L
 bsp_array_t bsp_read_array_parallel(hid_t f, const char* label,
                                     int num_threads) {
   hid_t dset = H5Dopen2(f, label, H5P_DEFAULT);
@@ -91,12 +96,19 @@ bsp_array_t bsp_read_array_parallel(hid_t f, const char* label,
 
   bsp_type_t type = bsp_get_bsp_type(hdf5_type);
 
+  // Array will be written into a POSIX shared memory.
   bsp_shm_t array_shm = bsp_shm_new(dims[0] * bsp_type_size(type));
   bsp_array_t array;
   array.type = type;
   array.size = dims[0];
-  array.shm = array_shm;
-  array.shmat_memory = true;
+  array.allocator = bsp_shm_allocator;
+
+  bsp_shm_t active_children_shm = bsp_shm_new(sizeof(_Atomic int));
+
+  _Atomic int* active_children = bsp_shm_attach(active_children_shm);
+  bsp_shm_delete(active_children_shm);
+
+  *active_children = num_threads - 1;
 
   pid_t* pids = (pid_t*) malloc(sizeof(pid_t) * num_threads);
 
@@ -115,7 +127,7 @@ bsp_array_t bsp_read_array_parallel(hid_t f, const char* label,
 
   array.data = bsp_shm_attach(array_shm);
   if (thread_num == 0) {
-    bsp_shm_delete(array.shm);
+    bsp_shm_delete(array_shm);
   }
 
   hsize_t chunk_size = (array.size + num_threads - 1) / num_threads;
@@ -131,22 +143,29 @@ bsp_array_t bsp_read_array_parallel(hid_t f, const char* label,
     hid_t memspace_id = H5Screate_simple(1, &count, NULL);
 
     H5Dread(dset, bsp_get_hdf5_native_type(type), memspace_id, fspace,
-            H5P_DEFAULT, array.data + start * bsp_type_size(type));
+            H5P_DEFAULT, ((char*) array.data) + start * bsp_type_size(type));
     H5Sclose(memspace_id);
   }
 
   H5Dclose(dset);
   H5Sclose(fspace);
 
   if (thread_num > 0) {
+    atomic_fetch_add_explicit(active_children, -1, memory_order_relaxed);
+    bsp_shm_detach(active_children);
     bsp_shm_detach(array.data);
     exit(0);
   }
 
   free(pids);
 
+  while (atomic_load(active_children) > 0) {
+  }
+  bsp_shm_detach(active_children);
+
   return array;
 }
+#endif
 
 bsp_array_t bsp_read_array(hid_t f, const char* label) {
   hid_t dset = H5Dopen2(f, label, H5P_DEFAULT);

include/binsparse/read_matrix.h

@@ -4,6 +4,7 @@
 #include <binsparse/matrix.h>
 #include <cJSON/cJSON.h>
 
+#if __STDC_VERSION__ >= 201112L
 bsp_matrix_t bsp_read_matrix_from_group_parallel(hid_t f, int num_threads) {
   bsp_matrix_t matrix = bsp_construct_default_matrix_t();
 
@@ -107,6 +108,7 @@ bsp_matrix_t bsp_read_matrix_from_group_parallel(hid_t f, int num_threads) {
 
   return matrix;
 }
+#endif
 
 bsp_matrix_t bsp_read_matrix_from_group(hid_t f) {
   bsp_matrix_t matrix = bsp_construct_default_matrix_t();
@@ -219,6 +221,7 @@ size_t bsp_final_dot(const char* str) {
   return dot_idx;
 }
 
+#if __STDC_VERSION__ >= 201112L
 bsp_matrix_t bsp_read_matrix_parallel(const char* file_name, const char* group,
                                       int num_threads) {
   if (group == NULL) {
@@ -243,6 +246,7 @@ bsp_matrix_t bsp_read_matrix_parallel(const char* file_name, const char* group,
     return matrix;
   }
 }
+#endif
 
 bsp_matrix_t bsp_read_matrix(const char* file_name, const char* group) {
   if (group == NULL) {