From 0f3bf6ab8265a444271950a527f30a33afb3e221 Mon Sep 17 00:00:00 2001 From: Patrick Pelissier Date: Tue, 15 Jan 2019 23:27:02 +0100 Subject: [PATCH] Add _bulk version for SPSC queue. To test/bench --- m-buffer.h | 56 ++++++++++++++++++++++++++++++++++++++++++++++++++++-- 1 file changed, 54 insertions(+), 2 deletions(-) diff --git a/m-buffer.h b/m-buffer.h index b822f3fe..d7eb2f65 100644 --- a/m-buffer.h +++ b/m-buffer.h @@ -641,7 +641,7 @@ M_C(name, _init)(buffer_t v, size_t size) \ } buffer_t[1]; \ \ static inline bool \ - M_C(name, _push)(buffer_t table, type x) \ + M_C(name, _push)(buffer_t table, type const x) \ { \ QUEUEI_MPMC_CONTRACT(table); \ unsigned int idx = atomic_load_explicit(&table->ProdIdx, \ @@ -825,7 +825,7 @@ M_C(name, _init)(buffer_t v, size_t size) \ } buffer_t[1]; \ \ static inline bool \ - M_C(name, _push)(buffer_t table, type x) \ + M_C(name, _push)(buffer_t table, type const x) \ { \ QUEUEI_SPSC_CONTRACT(table); \ unsigned int r = atomic_load_explicit(&table->consoIdx, \ @@ -867,6 +867,58 @@ M_C(name, _init)(buffer_t v, size_t size) \ return true; \ } \ \ + static inline unsigned \ + M_C(name, _push_bulk)(buffer_t table, unsigned n, type const x[]) \ + { \ + QUEUEI_SPSC_CONTRACT(table); \ + assert (x != NULL); \ + assert (n <= table->size); \ + unsigned int r = atomic_load_explicit(&table->consoIdx, \ + memory_order_relaxed); \ + unsigned int w = atomic_load_explicit(&table->prodIdx, \ + memory_order_acquire); \ + unsigned int max = M_MIN(n, table->size - (w-r) ); \ + if (max == 0) \ + return 0; \ + for(unsigned int k = 0; k < max; k++) { \ + unsigned int i = (w+k) & (table->size -1); \ + if (!BUFFERI_POLICY_P((policy), BUFFER_PUSH_INIT_POP_MOVE)) { \ + M_CALL_SET(oplist, table->Tab[i].x, x[k]); \ + } else { \ + M_CALL_INIT_SET(oplist, table->Tab[i].x, x[k]); \ + } \ + } \ + atomic_store_explicit(&table->prodIdx, w+max, memory_order_release); \ + QUEUEI_SPSC_CONTRACT(table); \ + return max; \ + } \ + \ + static inline unsigned \ + M_C(name, _pop_bulk)(unsigned int n, type ptr[], buffer_t table) \ + { \ + QUEUEI_SPSC_CONTRACT(table); \ + assert (ptr != NULL); \ + assert (n <= table->size); \ + unsigned int w = atomic_load_explicit(&table->prodIdx, \ + memory_order_relaxed); \ + unsigned int r = atomic_load_explicit(&table->consoIdx, \ + memory_order_acquire); \ + if (w-r == 0) \ + return 0; \ + unsigned int max = M_MIN(w-r, n); \ + for(unsigned int k = 0; k < max; k++) { \ + unsigned int i = (r+k) & (table->size -1); \ + if (!BUFFERI_POLICY_P((policy), BUFFER_PUSH_INIT_POP_MOVE)) { \ + M_CALL_SET(oplist, ptr[k], table->Tab[i].x); \ + } else { \ + M_DO_INIT_MOVE (oplist, ptr[k], table->Tab[i].x); \ + } \ + } \ + atomic_store_explicit(&table->consoIdx, r+max, memory_order_release); \ + QUEUEI_SPSC_CONTRACT(table); \ + return max; \ + } \ + \ static inline size_t \ M_C(name, _size)(buffer_t table) \ { \