#include // Because I wanted the C++11 memory order semantics, of which gb.h does not offer (because it was a C89 library) template struct MPMCQueueNode { T data; std::atomic idx; }; typedef char CacheLinePad[64]; // Multiple Producer Multiple Consumer Queue template struct MPMCQueue { CacheLinePad pad0; isize mask; Array> buffer; gbMutex mutex; std::atomic count; CacheLinePad pad1; std::atomic head_idx; CacheLinePad pad2; std::atomic tail_idx; CacheLinePad pad3; }; template void mpmc_init(MPMCQueue *q, gbAllocator a, isize size) { size = next_pow2_isize(size); GB_ASSERT(gb_is_power_of_two(size)); gb_mutex_init(&q->mutex); q->mask = size-1; array_init(&q->buffer, a, size); for (isize i = 0; i < size; i++) { q->buffer[i].idx.store(i, std::memory_order_relaxed); } } template void mpmc_destroy(MPMCQueue *q) { gb_mutex_destroy(&q->mutex); gb_free(q->buffer.allocator, q->buffer.data); } template bool mpmc_enqueue(MPMCQueue *q, T const &data) { isize head_idx = q->head_idx.load(std::memory_order_relaxed); for (;;) { auto node = &q->buffer.data[head_idx & q->mask]; isize node_idx = node->idx.load(std::memory_order_acquire); isize diff = node_idx - head_idx; if (diff == 0) { isize next_head_idx = head_idx+1; if (q->head_idx.compare_exchange_weak(head_idx, next_head_idx)) { node->data = data; node->idx.store(next_head_idx, std::memory_order_release); q->count.fetch_add(1, std::memory_order_release); return true; } } else if (diff < 0) { gb_mutex_lock(&q->mutex); isize old_size = q->buffer.count; isize new_size = old_size*2; array_resize(&q->buffer, new_size); if (q->buffer.data == nullptr) { GB_PANIC("Unable to resize enqueue: %td -> %td", old_size, new_size); gb_mutex_unlock(&q->mutex); return false; } for (isize i = old_size; i < new_size; i++) { q->buffer.data[i].idx.store(i, std::memory_order_relaxed); } q->mask = new_size-1; gb_mutex_unlock(&q->mutex); } else { head_idx = q->head_idx.load(std::memory_order_relaxed); } } } template bool mpmc_dequeue(MPMCQueue *q, T *data_) { isize tail_idx = q->tail_idx.load(std::memory_order_relaxed); for (;;) { auto node = &q->buffer.data[tail_idx & q->mask]; isize node_idx = node->idx.load(std::memory_order_acquire); isize diff = node_idx - (tail_idx+1); if (diff == 0) { isize next_tail_idx = tail_idx+1; if (q->tail_idx.compare_exchange_weak(tail_idx, next_tail_idx)) { if (data_) *data_ = node->data; node->idx.store(tail_idx + q->mask + 1, std::memory_order_release); q->count.fetch_sub(1, std::memory_order_release); return true; } } else if (diff < 0) { return false; } else { tail_idx = q->tail_idx.load(std::memory_order_relaxed); } } }