amir/filehasher
1
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Updating the LF MPMC queue and replacing DirQueue with it

Browse Source 
Making the MPMC queue support when producers are consumers at the same
time by adding a variable work, mpmc_push_work() that increments work
and mpmc_task_done() that decrements work, and if work = 0 calls
mpmc_producers_finished() that pushes poinsons to wake up sleeping
threads and make them return NULL

Replacing DirQueue, a queue growable with realloc with the MPMC queue
This commit is contained in:
amir
2026-03-11 17:48:36 +01:00
parent 0e3ec5b09c
commit 06ea07d9f6
4 changed files with 39 additions and 145 deletions
Download Patch File Download Diff File Expand all files Collapse all files

46
lf_mpmc.h
View File

@@ -279,52 +279,6 @@ static void *mpmc_pop(MPMCQueue *q) {
return data;
}
/* ----------------------------------------------------------- */
/* TRY POP (non blocking) */
/* ----------------------------------------------------------- */
// static b32 mpmc_try_pop(MPMCQueue *q, void **out) {
//
// if (!plat_sem_trywait(&q->items_sem))
// return false;
//
// MPMCSlot *slot;
// size_t pos;
//
// int spins = 0;
//
// for (;;) {
//
// pos = atomic_load_explicit(&q->head, memory_order_relaxed);
// slot = &q->slots[pos & q->mask];
//
// size_t seq = atomic_load_explicit(&slot->seq, memory_order_acquire);
// intptr_t diff = (intptr_t)seq - (intptr_t)(pos + 1);
//
// if (likely(diff == 0)) {
//
// if (atomic_compare_exchange_weak_explicit(&q->head, &pos, pos + 1,
// memory_order_relaxed,
// memory_order_relaxed))
// break;
//
// } else {
//
// if (++spins > 10) {
// SwitchToThread();
// spins = 0;
// } else {
// cpu_pause();
// }
// }
// }
//
// *out = slot->data;
//
// atomic_store_explicit(&slot->seq, pos + q->capacity, memory_order_release);
//
// return true;
// }
/* ----------------------------------------------------------- */
/* PUSH POISON */
/* ----------------------------------------------------------- */