plat_sem_destroy(&q->items_sem);

base.h

@@ -97,6 +97,8 @@ typedef double f64;
 
 #if defined(_WIN32) || defined(_WIN64)
 
+// Memory allocation
+
 static u32 plat_get_pagesize(void) {
   SYSTEM_INFO sysinfo = {0};
   GetSystemInfo(&sysinfo);
@@ -121,8 +123,35 @@ static b32 plat_mem_release(void *ptr, u64 size) {
   return VirtualFree(ptr, size, MEM_RELEASE);
 }
 
+// Semaphores
+typedef struct plat_sem {
+  HANDLE handle;
+} plat_sem;
+
+static b32 plat_sem_init(plat_sem *s, u32 initial) {
+  s->handle = CreateSemaphore(NULL, initial, LONG_MAX, NULL);
+  return s->handle != NULL;
+}
+
+static void plat_sem_wait(plat_sem *s) {
+  WaitForSingleObject(s->handle, INFINITE);
+}
+
+static void plat_sem_post(plat_sem *s, u32 count) {
+  ReleaseSemaphore(s->handle, count, NULL);
+}
+
+static void plat_sem_destroy(plat_sem *s) {
+  if (s->handle) {
+    CloseHandle(s->handle);
+    s->handle = NULL;
+  }
+}
+
 #elif defined(__linux__)
 
+// Memory allocation
+
 #ifndef _DEFAULT_SOURCE
 #define _DEFAULT_SOURCE
 #endif
@@ -158,4 +187,28 @@ static b32 plat_mem_release(void *ptr, u64 size) {
   return ret == 0;
 }
 
+// Semaphores
+#include <semaphore.h>
+
+typedef struct plat_sem {
+  sem_t sem;
+} plat_sem;
+
+static b32 plat_sem_init(plat_sem *s, u32 initial) {
+  return sem_init(&s->sem, 0, initial) == 0;
+}
+
+static void plat_sem_wait(plat_sem *s) {
+  while (sem_wait(&s->sem) == -1 && errno == EINTR) {
+  }
+}
+
+static void plat_sem_post(plat_sem *s, u32 count) {
+  for (u32 i = 0; i < count; i++) {
+    sem_post(&s->sem);
+  }
+}
+
+static void plat_sem_destroy(plat_sem *s) { sem_destroy(&s->sem); }
+
 #endif

binaries/changelog.txt

@@ -29,3 +29,5 @@ Small improvements of the LF MPMC queue
 Making the LF MPMC queue generic and in a seperate header file
 
 4.0: Implementing a semaphore in the LF MPMC queue to wake up consumers when there is items in the queue instead of spinning (busy waiting) or sleeping, this makes the queue spin only when the slots are transitionning (multiple consumers claiming the same slot)
+
+Making the MPMC queue platform agnostic

lf_mpmc.h

@@ -23,6 +23,9 @@ static void cpu_pause(void) {
   _mm_pause();
 #endif
 }
+
+typedef struct plat_sem plat_sem;
+
 typedef struct CACHE_ALIGN {
   atomic_size_t seq;
   void *data;
@@ -40,7 +43,7 @@ typedef struct {
   size_t commit_step;
   atomic_flag commit_lock;
 
-  HANDLE items_sem;
+  plat_sem items_sem;
 
   MPMCSlot *slots;
 } MPMCQueue;
@@ -87,7 +90,7 @@ static void mpmc_init(MPMCQueue *q, size_t max_capacity) {
   atomic_init(&q->head, 0);
   atomic_init(&q->tail, 0);
 
-  q->items_sem = CreateSemaphore(NULL, 0, LONG_MAX, NULL);
+  plat_sem_init(&q->items_sem, 0);
 }
 
 /* ----------------------------------------------------------- */
@@ -176,7 +179,7 @@ static void mpmc_push(MPMCQueue *q, void *item) {
 
   atomic_store_explicit(&slot->seq, pos + 1, memory_order_release);
 
-  ReleaseSemaphore(q->items_sem, 1, NULL);
+  plat_sem_post(&q->items_sem, 1);
 }
 
 /* ----------------------------------------------------------- */
@@ -184,7 +187,7 @@ static void mpmc_push(MPMCQueue *q, void *item) {
 /* ----------------------------------------------------------- */
 static void *mpmc_pop(MPMCQueue *q) {
 
-  WaitForSingleObject(q->items_sem, INFINITE);
+  plat_sem_wait(&q->items_sem);
 
   MPMCSlot *slot;
   size_t pos;
@@ -249,8 +252,7 @@ static void mpmc_finish(MPMCQueue *q) {
     q->slots = NULL;
   }
 
-  if (q->items_sem)
-    CloseHandle(q->items_sem);
+  plat_sem_destroy(&q->items_sem);
 
   q->capacity = 0;
   q->mask = 0;