From c15db051999e51f5c8ecc45fb084e6fb76c9b5c2 Mon Sep 17 00:00:00 2001 From: gingerBill Date: Thu, 12 Jan 2023 12:41:53 +0000 Subject: [PATCH] Implement `MPSCQueue` --- src/checker.cpp | 8 ++--- src/checker.hpp | 3 +- src/queue.cpp | 82 +++++++++++++++++++++++++++++++++++++++++++++++++ 3 files changed, 87 insertions(+), 6 deletions(-) diff --git a/src/checker.cpp b/src/checker.cpp index 053bb0e17..5ffdfab55 100644 --- a/src/checker.cpp +++ b/src/checker.cpp @@ -1161,7 +1161,7 @@ gb_internal void init_checker_info(CheckerInfo *i) { TIME_SECTION("checker info: mpmc queues"); mpmc_init(&i->entity_queue, a, 1<<20); - mpmc_init(&i->definition_queue, a, 1<<20); + mpsc_init(&i->definition_queue, a); //, 1<<20); mpmc_init(&i->required_global_variable_queue, a, 1<<10); mpmc_init(&i->required_foreign_imports_through_force_queue, a, 1<<10); mpmc_init(&i->intrinsics_entry_point_usage, a, 1<<10); // just waste some memory here, even if it probably never used @@ -1182,7 +1182,7 @@ gb_internal void destroy_checker_info(CheckerInfo *i) { array_free(&i->required_foreign_imports_through_force); mpmc_destroy(&i->entity_queue); - mpmc_destroy(&i->definition_queue); + mpsc_destroy(&i->definition_queue); mpmc_destroy(&i->required_global_variable_queue); mpmc_destroy(&i->required_foreign_imports_through_force_queue); @@ -1493,7 +1493,7 @@ gb_internal void add_entity_definition(CheckerInfo *i, Ast *identifier, Entity * GB_ASSERT(entity != nullptr); identifier->Ident.entity = entity; entity->identifier = identifier; - mpmc_enqueue(&i->definition_queue, entity); + mpsc_enqueue(&i->definition_queue, entity); } gb_internal bool redeclaration_error(String name, Entity *prev, Entity *found) { @@ -5583,7 +5583,7 @@ gb_internal void check_add_entities_from_queues(Checker *c) { gb_internal void check_add_definitions_from_queues(Checker *c) { isize cap = c->info.definitions.count + c->info.definition_queue.count.load(std::memory_order_relaxed); array_reserve(&c->info.definitions, cap); - for (Entity *e; mpmc_dequeue(&c->info.definition_queue, &e); /**/) { + for (Entity *e; mpsc_dequeue(&c->info.definition_queue, &e); /**/) { array_add(&c->info.definitions, e); } } diff --git a/src/checker.hpp b/src/checker.hpp index 806eb2e51..356dd1fc8 100644 --- a/src/checker.hpp +++ b/src/checker.hpp @@ -303,7 +303,6 @@ struct UntypedExprInfo { }; typedef PtrMap UntypedExprInfoMap; -typedef MPMCQueue ProcBodyQueue; enum ObjcMsgKind : u32 { ObjcMsg_normal, @@ -380,7 +379,7 @@ struct CheckerInfo { // NOTE(bill): These are actually MPSC queues // TODO(bill): Convert them to be MPSC queues - MPMCQueue definition_queue; + MPSCQueue definition_queue; MPMCQueue entity_queue; MPMCQueue required_global_variable_queue; MPMCQueue required_foreign_imports_through_force_queue; diff --git a/src/queue.cpp b/src/queue.cpp index 8f279bb21..845f87310 100644 --- a/src/queue.cpp +++ b/src/queue.cpp @@ -1,3 +1,85 @@ +template +struct MPSCNode { + std::atomic *> next; + T value; +}; + +// +// Multiple Producer Single Consumer Lockless Queue +// URL: https://www.1024cores.net +// +template +struct MPSCQueue { + std::atomic *> head; + std::atomic *> tail; + std::atomic count; + MPSCNode sentinel; + gbAllocator allocator; +}; + +template gb_internal void mpsc_init (MPSCQueue *q, gbAllocator const &allocator); +template gb_internal void mpsc_destroy(MPSCQueue *q); +template gb_internal isize mpsc_enqueue(MPSCQueue *q, T const &value); +template gb_internal bool mpsc_dequeue(MPSCQueue *q, T *value_); +template gb_internal MPSCNode *mpsc_tail (MPSCQueue *q); + +template +gb_internal void mpsc_init(MPSCQueue *q, gbAllocator const &allocator) { + q->allocator = allocator; + q->count.store(0, std::memory_order_relaxed); + q->head.store(&q->sentinel, std::memory_order_relaxed); + q->tail.store(&q->sentinel, std::memory_order_relaxed); + q->sentinel.next.store(nullptr, std::memory_order_relaxed); +} + +template +gb_internal void mpsc_destroy(MPSCQueue *q) { + while (mpsc_dequeue(q, (T *)nullptr)) {} + // DO NOTHING for the time being +} + + +template +gb_internal MPSCNode *mpsc_alloc_node(MPSCQueue *q, T const &value) { + auto node = gb_alloc_item(q->allocator, MPSCNode); + node->value = value; + return node; +} + +template +gb_internal isize mpsc_enqueue(MPSCQueue *q, MPSCNode *node) { + node->next.store(nullptr, std::memory_order_relaxed); + auto prev = q->head.exchange(node, std::memory_order_acq_rel); + prev->next.store(node, std::memory_order_release); + isize count = 1 + q->count.fetch_add(1, std::memory_order_acq_rel); + return count; +} + +template +gb_internal isize mpsc_enqueue(MPSCQueue *q, T const &value) { + auto node = mpsc_alloc_node(q, value); + return mpsc_enqueue(q, node); +} + + +template +gb_internal bool mpsc_dequeue(MPSCQueue *q, T *value_) { + auto tail = q->tail.load(std::memory_order_relaxed); + auto next = tail->next.load(std::memory_order_relaxed); + if (next) { + q->tail.store(next, std::memory_order_relaxed); + // `tail` is now "dead" and needs to be "freed" + if (*value_) *value_ = next->value; + q->count.fetch_sub(1, std::memory_order_acq_rel); + return true; + } + return false; +} + +//////////////////////////// + + + #define MPMC_CACHE_LINE_SIZE 64 typedef std::atomic MPMCQueueAtomicIdx;