Replace TAB by SPACE

This commit is contained in:
Patrick Pelissier committed 2020-07-25 19:17:25 +02:00
1 parent efef35c1e1
commit 5f9feb833c
10 files changed
+1067 -1061

No files matched your search

+251 -248
View File
@@ -63,14 +63,14 @@ typedef enum {
USAGE: BUFFER_OPLIST(name[, oplist of the type]) */
#define BUFFER_OPLIST(...) \
BUFFERI_OPLIST_P1(M_IF_NARGS_EQ1(__VA_ARGS__) \
((__VA_ARGS__, M_DEFAULT_OPLIST), \
((__VA_ARGS__, M_DEFAULT_OPLIST), \
(__VA_ARGS__ )))
/* Define a nearly lock-free queue for Many Producer Many Consummer
Much faster than queue of BUFFER_DEF in heavy communication scenario
Size of created queue can only a power of 2.
*/
#define QUEUE_MPMC_DEF(name, type, ...) \
#define QUEUE_MPMC_DEF(name, type, ...) \
M_BEGIN_PROTECTED_CODE \
QUEUEI_MPMC_DEF_P1(M_IF_NARGS_EQ1(__VA_ARGS__) \
((name, type, __VA_ARGS__, M_GLOBAL_OPLIST_OR_DEF(type)(), M_C(name,_t)), \
@@ -81,7 +81,7 @@ typedef enum {
Much faster than queue of BUFFER_DEF in heavy communication scenario
Size of created queue can only a power of 2.
*/
#define QUEUE_SPSC_DEF(name, type, ...) \
#define QUEUE_SPSC_DEF(name, type, ...) \
M_BEGIN_PROTECTED_CODE \
QUEUEI_SPSC_DEF_P1(M_IF_NARGS_EQ1(__VA_ARGS__) \
((name, type, __VA_ARGS__, M_GLOBAL_OPLIST_OR_DEF(type)(), M_C(name,_t)), \
@@ -99,13 +99,13 @@ typedef enum {
#define BUFFERI_POLICY_P(policy, val) \
(((policy) & (val)) != 0)
#define BUFFERI_CONTRACT(buffer, size) do { \
assert (buffer != NULL); \
assert (buffer->data != NULL); \
#define BUFFERI_CONTRACT(buffer, size) do { \
assert (buffer != NULL); \
assert (buffer->data != NULL); \
}while (0)
#define BUFFERI_PROTECTED_CONTRACT(buffer, size) do { \
assert (atomic_load(&buffer->number[0]) <= BUFFERI_SIZE(size)); \
assert (atomic_load(&buffer->number[0]) <= BUFFERI_SIZE(size)); \
} while (0)
/* Deferred evaluation for the definition,
@@ -122,7 +122,7 @@ typedef enum {
#define BUFFERI_DEF_P3(name, type, m_size, policy, oplist, buffer_t) \
\
typedef struct M_C(name, _s) { \
typedef struct M_C(name, _s) { \
m_mutex_t mutexPush; /* MUTEX used for pushing elements */ \
size_t idx_prod; /* Index of the production threads */ \
size_t overwrite; /* Number of overwritten values */ \
@@ -158,9 +158,11 @@ M_C(name, _init)(buffer_t v, size_t size) \
m_mutex_init(v->mutexPop); \
m_cond_init(v->there_is_data); \
m_cond_init(v->there_is_room_for_data); \
} else \
} else { \
assert(BUFFERI_POLICY_P((policy), BUFFER_UNBLOCKING)); \
BUFFERI_IF_CTE_SIZE(m_size)( , \
} \
\
BUFFERI_IF_CTE_SIZE(m_size)( /* Statically allocated */ , \
v->data = M_CALL_REALLOC(oplist, type, NULL, BUFFERI_SIZE(m_size)); \
if (v->data == NULL) { \
M_MEMORY_FULL (BUFFERI_SIZE(m_size)*sizeof(type)); \
@@ -168,12 +170,12 @@ M_C(name, _init)(buffer_t v, size_t size) \
} \
) \
if (!BUFFERI_POLICY_P((policy), BUFFER_PUSH_INIT_POP_MOVE)) { \
for(size_t i = 0; i < size; i++) { \
for(size_t i = 0; i < size; i++) { \
M_CALL_INIT(oplist, v->data[i]); \
} \
} \
} \
BUFFERI_CONTRACT(v,m_size); \
} \
BUFFERI_CONTRACT(v,m_size); \
} \
\
BUFFERI_IF_CTE_SIZE(m_size)( \
static inline void \
@@ -184,13 +186,13 @@ M_C(name, _init)(buffer_t v, size_t size) \
, ) \
\
static inline void \
M_C(name, _int_clear_obj)(buffer_t v) \
M_C(name, _int_clear_obj)(buffer_t v) \
{ \
BUFFERI_CONTRACT(v,m_size); \
BUFFERI_CONTRACT(v,m_size); \
if (!BUFFERI_POLICY_P((policy), BUFFER_PUSH_INIT_POP_MOVE)) { \
for(size_t i = 0; i < BUFFERI_SIZE(m_size); i++) { \
for(size_t i = 0; i < BUFFERI_SIZE(m_size); i++) { \
M_CALL_CLEAR(oplist, v->data[i]); \
} \
} \
} else { \
size_t i = BUFFERI_POLICY_P((policy), BUFFER_STACK) ? 0 : v->idx_cons; \
while (i != v->idx_prod) { \
@@ -201,17 +203,17 @@ M_C(name, _init)(buffer_t v, size_t size) \
} \
} \
v->idx_prod = v->idx_cons = 0; \
atomic_store_explicit (&v->number[0], 0UL, memory_order_relaxed); \
atomic_store_explicit (&v->number[0], 0UL, memory_order_relaxed); \
if (BUFFERI_POLICY_P(policy, BUFFER_DEFERRED_POP)) \
atomic_store_explicit(&v->number[1], 0UL, memory_order_relaxed); \
BUFFERI_CONTRACT(v,m_size); \
atomic_store_explicit(&v->number[1], 0UL, memory_order_relaxed); \
BUFFERI_CONTRACT(v,m_size); \
} \
\
\
static inline void \
M_C(name, _clear)(buffer_t v) \
M_C(name, _clear)(buffer_t v) \
{ \
BUFFERI_CONTRACT(v,m_size); \
M_C(name, _int_clear_obj)(v); \
BUFFERI_CONTRACT(v,m_size); \
M_C(name, _int_clear_obj)(v); \
BUFFERI_IF_CTE_SIZE(m_size)( , \
M_CALL_FREE(oplist, v->data); \
v->data = NULL; \
@@ -224,30 +226,30 @@ M_C(name, _init)(buffer_t v, size_t size) \
m_cond_clear(v->there_is_room_for_data); \
} \
} \
\
\
static inline void \
M_C(name, _clean)(buffer_t v) \
M_C(name, _clean)(buffer_t v) \
{ \
BUFFERI_CONTRACT(v,m_size); \
BUFFERI_CONTRACT(v,m_size); \
if (!BUFFERI_POLICY_P((policy), BUFFER_THREAD_UNSAFE)) { \
m_mutex_lock(v->mutexPush); \
m_mutex_lock(v->mutexPop); \
} \
BUFFERI_PROTECTED_CONTRACT(v, m_size); \
if (BUFFERI_POLICY_P((policy), BUFFER_PUSH_INIT_POP_MOVE)) \
M_C(name, _int_clear_obj)(v); \
BUFFERI_PROTECTED_CONTRACT(v, m_size); \
if (BUFFERI_POLICY_P((policy), BUFFER_PUSH_INIT_POP_MOVE)) \
M_C(name, _int_clear_obj)(v); \
v->idx_prod = v->idx_cons = 0; \
atomic_store_explicit (&v->number[0], 0UL, memory_order_relaxed); \
atomic_store_explicit (&v->number[0], 0UL, memory_order_relaxed); \
if (BUFFERI_POLICY_P(policy, BUFFER_DEFERRED_POP)) \
atomic_store_explicit(&v->number[1], 0UL, memory_order_relaxed); \
atomic_store_explicit(&v->number[1], 0UL, memory_order_relaxed); \
if (!BUFFERI_POLICY_P((policy), BUFFER_THREAD_UNSAFE)) { \
m_cond_broadcast(v->there_is_room_for_data); \
m_mutex_unlock(v->mutexPop); \
m_mutex_unlock(v->mutexPush); \
} \
BUFFERI_CONTRACT(v,m_size); \
BUFFERI_CONTRACT(v,m_size); \
} \
\
\
static inline void \
M_C(name, _init_set)(buffer_t dest, const buffer_t src) \
{ \
@@ -263,11 +265,11 @@ M_C(name, _init)(buffer_t v, size_t size) \
m_mutex_lock(v->mutexPop); \
} \
\
BUFFERI_PROTECTED_CONTRACT(v, m_size); \
BUFFERI_PROTECTED_CONTRACT(v, m_size); \
if (!BUFFERI_POLICY_P((policy), BUFFER_PUSH_INIT_POP_MOVE)) { \
for(size_t i = 0; i < BUFFERI_SIZE(m_size); i++) { \
for(size_t i = 0; i < BUFFERI_SIZE(m_size); i++) { \
M_CALL_INIT_SET(oplist, dest->data[i], v->data[i]); \
} \
} \
} else { \
size_t i = BUFFERI_POLICY_P((policy), BUFFER_STACK) ? 0 : v->idx_cons; \
while (i != v->idx_prod) { \
@@ -291,7 +293,7 @@ M_C(name, _init)(buffer_t v, size_t size) \
BUFFERI_CONTRACT(v,m_size); \
BUFFERI_CONTRACT(dest, m_size); \
} \
\
\
static inline void \
M_C(name, _set)(buffer_t dest, const buffer_t src) \
{ \
@@ -319,13 +321,13 @@ M_C(name, _init)(buffer_t v, size_t size) \
} \
} \
\
BUFFERI_PROTECTED_CONTRACT(v, m_size); \
M_C(name, _int_clear_obj)(dest); \
BUFFERI_PROTECTED_CONTRACT(v, m_size); \
M_C(name, _int_clear_obj)(dest); \
\
if (!BUFFERI_POLICY_P((policy), BUFFER_PUSH_INIT_POP_MOVE)) { \
for(size_t i = 0; i < BUFFERI_SIZE(m_size); i++) { \
for(size_t i = 0; i < BUFFERI_SIZE(m_size); i++) { \
M_CALL_INIT_SET(oplist, dest->data[i], v->data[i]); \
} \
} \
} else { \
size_t i = BUFFERI_POLICY_P((policy), BUFFER_STACK) ? 0 : v->idx_cons; \
while (i != v->idx_prod) { \
@@ -361,11 +363,11 @@ M_C(name, _init)(buffer_t v, size_t size) \
BUFFERI_CONTRACT(v,m_size); \
BUFFERI_CONTRACT(dest, m_size); \
} \
\
\
static inline bool \
M_C(name, _empty_p)(buffer_t v) \
M_C(name, _empty_p)(buffer_t v) \
{ \
BUFFERI_CONTRACT(v,m_size); \
BUFFERI_CONTRACT(v,m_size); \
/* If the buffer has been configured with deferred pop \
we considered the queue as empty when the number of \
deferred pop has reached 0, not the number of items in the \
@@ -375,32 +377,32 @@ M_C(name, _init)(buffer_t v, size_t size) \
else \
return atomic_load_explicit (&v->number[0], memory_order_relaxed) == 0; \
} \
\
\
static inline bool \
M_C(name, _full_p)(buffer_t v) \
{ \
BUFFERI_CONTRACT(v,m_size); \
return atomic_load_explicit (&v->number[0], memory_order_relaxed) \
== BUFFERI_SIZE(m_size); \
BUFFERI_CONTRACT(v,m_size); \
return atomic_load_explicit (&v->number[0], memory_order_relaxed) \
== BUFFERI_SIZE(m_size); \
} \
\
static inline size_t \
\
static inline size_t \
M_C(name, _size)(buffer_t v) \
{ \
BUFFERI_CONTRACT(v,m_size); \
return atomic_load_explicit (&v->number[0], memory_order_relaxed); \
BUFFERI_CONTRACT(v,m_size); \
return atomic_load_explicit (&v->number[0], memory_order_relaxed); \
} \
\
\
static inline bool \
M_C(name, _push_blocking)(buffer_t v, type const data, bool blocking) \
{ \
BUFFERI_CONTRACT(v,m_size); \
\
/* BUFFER lock */ \
BUFFERI_CONTRACT(v,m_size); \
\
/* BUFFER lock */ \
if (!BUFFERI_POLICY_P((policy), BUFFER_THREAD_UNSAFE)) { \
m_mutex_lock(v->mutexPush); \
while (!BUFFERI_POLICY_P((policy), BUFFER_PUSH_OVERWRITE) \
&& M_C(name, _full_p)(v)) { \
&& M_C(name, _full_p)(v)) { \
if (!blocking) { \
m_mutex_unlock(v->mutexPush); \
return false; \
@@ -408,15 +410,15 @@ M_C(name, _init)(buffer_t v, size_t size) \
m_cond_wait(v->there_is_room_for_data, v->mutexPush); \
} \
} else if (!BUFFERI_POLICY_P((policy), BUFFER_PUSH_OVERWRITE) \
&& M_C(name, _full_p)(v)) \
&& M_C(name, _full_p)(v)) \
return false; \
BUFFERI_PROTECTED_CONTRACT(v, m_size); \
\
BUFFERI_PROTECTED_CONTRACT(v, m_size); \
\
size_t previousSize, idx = v->idx_prod; \
/* INDEX computation if we have to overwrite the last element */ \
if (M_UNLIKELY (BUFFERI_POLICY_P((policy), BUFFER_PUSH_OVERWRITE) \
&& M_C(name, _full_p)(v))) { \
v->overwrite++; \
/* INDEX computation if we have to overwrite the last element */ \
if (M_UNLIKELY (BUFFERI_POLICY_P((policy), BUFFER_PUSH_OVERWRITE) \
&& M_C(name, _full_p)(v))) { \
v->overwrite++; \
/* Let's overwrite the last element */ \
/* Compute the index of the last push element */ \
idx--; \
@@ -424,7 +426,7 @@ M_C(name, _init)(buffer_t v, size_t size) \
idx = idx >= BUFFERI_SIZE(m_size) ? BUFFERI_SIZE(m_size)-1 : idx; \
} \
/* Update data in the buffer */ \
M_CALL_SET(oplist, v->data[idx], data); \
M_CALL_SET(oplist, v->data[idx], data); \
previousSize = BUFFERI_SIZE(m_size); \
\
} else { \
@@ -449,13 +451,13 @@ M_C(name, _init)(buffer_t v, size_t size) \
thread which has the push lock. As such, it has to be handled \
like an atomic variable. */ \
/* Increment number of elements of the buffer */ \
previousSize = atomic_fetch_add (&v->number[0], 1UL); \
previousSize = atomic_fetch_add (&v->number[0], 1UL); \
if (BUFFERI_POLICY_P((policy), BUFFER_DEFERRED_POP)) { \
previousSize = atomic_fetch_add (&v->number[1], 1UL); \
previousSize = atomic_fetch_add (&v->number[1], 1UL); \
} \
} \
\
/* BUFFER unlock */ \
/* BUFFER unlock */ \
if (!BUFFERI_POLICY_P((policy), BUFFER_THREAD_UNSAFE)) { \
m_mutex_unlock(v->mutexPush); \
/* If the number of items in the buffer was 0, some consummer \
@@ -467,35 +469,35 @@ M_C(name, _init)(buffer_t v, size_t size) \
m_mutex_unlock(v->mutexPop); \
} \
} \
\
BUFFERI_CONTRACT(v,m_size); \
\
BUFFERI_CONTRACT(v,m_size); \
return true; \
} \
\
static inline bool \
M_C(name, _pop_blocking)(type *data, buffer_t v, bool blocking) \
{ \
BUFFERI_CONTRACT(v,m_size); \
assert (data != NULL); \
\
/* BUFFER lock */ \
BUFFERI_CONTRACT(v,m_size); \
assert (data != NULL); \
\
/* BUFFER lock */ \
if (!BUFFERI_POLICY_P((policy), BUFFER_THREAD_UNSAFE)) { \
m_mutex_lock(v->mutexPop); \
while (M_C(name, _empty_p)(v)) { \
while (M_C(name, _empty_p)(v)) { \
if (!blocking) { \
m_mutex_unlock(v->mutexPop); \
return false; \
} \
m_cond_wait(v->there_is_data, v->mutexPop); \
} \
} else if (M_C(name, _empty_p)(v)) \
} else if (M_C(name, _empty_p)(v)) \
return false; \
BUFFERI_PROTECTED_CONTRACT(v, m_size); \
\
BUFFERI_PROTECTED_CONTRACT(v, m_size); \
\
/* POP data from the buffer and update INDEX */ \
if (!BUFFERI_POLICY_P((policy), BUFFER_STACK)) { \
/* FIFO queue */ \
if (!BUFFERI_POLICY_P((policy), BUFFER_PUSH_INIT_POP_MOVE)) { \
if (!BUFFERI_POLICY_P((policy), BUFFER_PUSH_INIT_POP_MOVE)) { \
M_CALL_SET(oplist, *data, v->data[v->idx_cons]); \
} else { \
M_DO_INIT_MOVE (oplist, *data, v->data[v->idx_cons]); \
@@ -518,12 +520,12 @@ M_C(name, _init)(buffer_t v, size_t size) \
/* Decrement number of elements in the buffer */ \
size_t previousSize; \
if (!BUFFERI_POLICY_P((policy), BUFFER_DEFERRED_POP)) { \
previousSize = atomic_fetch_sub (&v->number[0], 1UL); \
previousSize = atomic_fetch_sub (&v->number[0], 1UL); \
} else { \
atomic_fetch_sub (&v->number[1], 1UL); \
atomic_fetch_sub (&v->number[1], 1UL); \
} \
\
/* BUFFER unlock */ \
/* BUFFER unlock */ \
if (!BUFFERI_POLICY_P((policy), BUFFER_THREAD_UNSAFE)) { \
m_mutex_unlock(v->mutexPop); \
/* If the number of items in the buffer was the max, some producer \
@@ -536,31 +538,31 @@ M_C(name, _init)(buffer_t v, size_t size) \
m_mutex_unlock(v->mutexPush); \
} \
} \
\
BUFFERI_CONTRACT(v,m_size); \
\
BUFFERI_CONTRACT(v,m_size); \
return true; \
} \
\
\
\
\
static inline bool \
M_C(name, _push)(buffer_t v, type const data) \
M_C(name, _push)(buffer_t v, type const data) \
{ \
return M_C(name, _push_blocking)(v, data, \
!BUFFERI_POLICY_P((policy), BUFFER_UNBLOCKING_PUSH)); \
} \
\
static inline bool \
M_C(name, _pop)(type *data, buffer_t v) \
M_C(name, _pop)(type *data, buffer_t v) \
{ \
return M_C(name, _pop_blocking)(data, v, \
!BUFFERI_POLICY_P((policy), BUFFER_UNBLOCKING_POP)); \
} \
\
static inline size_t \
M_C(name, _overwrite)(const buffer_t v) \
static inline size_t \
M_C(name, _overwrite)(const buffer_t v) \
{ \
return v->overwrite; \
} \
return v->overwrite; \
} \
\
static inline size_t \
M_C(name, _capacity)(const buffer_t v) \
@@ -574,7 +576,7 @@ M_C(name, _init)(buffer_t v, size_t size) \
{ \
/* Decrement the effective number of elements in the buffer */ \
if (BUFFERI_POLICY_P((policy), BUFFER_DEFERRED_POP)) { \
size_t previousSize = atomic_fetch_sub (&v->number[0], 1UL); \
size_t previousSize = atomic_fetch_sub (&v->number[0], 1UL); \
if (previousSize == BUFFERI_SIZE(m_size)) { \
m_mutex_lock(v->mutexPush); \
m_cond_broadcast(v->there_is_room_for_data); \
@@ -600,11 +602,11 @@ M_C(name, _init)(buffer_t v, size_t size) \
#define QUEUEI_MPMC_DEF_P1(arg) QUEUEI_MPMC_DEF_P2 arg
/* Validate the value oplist before going further */
#define QUEUEI_MPMC_DEF_P2(name, type, policy, oplist, buffer_t) \
#define QUEUEI_MPMC_DEF_P2(name, type, policy, oplist, buffer_t) \
M_IF_OPLIST(oplist)(QUEUEI_MPMC_DEF_P3, QUEUEI_MPMC_DEF_FAILURE)(name, type, policy, oplist, buffer_t)
/* Stop processing with a compilation failure */
#define QUEUEI_MPMC_DEF_FAILURE(name, type, policy, oplist, buffer_t) \
#define QUEUEI_MPMC_DEF_FAILURE(name, type, policy, oplist, buffer_t) \
M_STATIC_FAILURE(M_LIB_NOT_AN_OPLIST, "(QUEUE_MPMC_DEF): the given argument is not a valid oplist: " M_AS_STR(oplist))
#ifdef NDEBUG
@@ -621,54 +623,55 @@ M_C(name, _init)(buffer_t v, size_t size) \
} while (0)
#endif
#define QUEUEI_MPMC_DEF_P3(name, type, policy, oplist, buffer_t) \
\
/* The sequence number of an element will be equal to either \
- 2* the index of the production which creates it, \
- 1 + 2* the index of the consumption which consummes it \
\
/* The sequence number of an element will be equal to either \
- 2* the index of the production which creates it, \
- 1 + 2* the index of the consumption which consummes it \
In case of wrapping, as there is no order comparison but only \
equal comparison, there is no special issue. \
Each element is put in a separate cache line to avoid false \
sharing. \
*/ \
typedef struct M_C(name, _el_s) { \
atomic_uint seq; /* Can only increase until wrapping */ \
type x; \
*/ \
typedef struct M_C(name, _el_s) { \
atomic_uint seq; /* Can only increase until wrapping */ \
type x; \
M_CACHELINE_ALIGN(align, atomic_uint, type); \
} M_C(name, _el_t); \
\
} M_C(name, _el_t); \
\
/* If there is only one producer and one consummer, then they won't \
typically use the same cache line, increasing performance. */ \
typedef struct M_C(name, _s) { \
typedef struct M_C(name, _s) { \
atomic_uint ProdIdx; /* Can only increase until wrapping */ \
M_CACHELINE_ALIGN(align1, atomic_uint); \
atomic_uint ConsoIdx; /* Can only increase until wrapping */ \
M_CACHELINE_ALIGN(align2, atomic_uint); \
M_C(name, _el_t) *Tab; \
unsigned int size; \
} buffer_t[1]; \
\
static inline bool \
unsigned int size; \
} buffer_t[1]; \
\
static inline bool \
M_C(name, _push)(buffer_t table, type const x) \
{ \
{ \
QUEUEI_MPMC_CONTRACT(table); \
unsigned int idx = atomic_load_explicit(&table->ProdIdx, \
memory_order_relaxed); \
const unsigned int i = idx & (table->size -1); \
const unsigned int i = idx & (table->size -1); \
const unsigned int seq = atomic_load_explicit(&table->Tab[i].seq, \
memory_order_acquire); \
if (M_UNLIKELY (2*(idx - table->size) + 1 != seq)) { \
if (M_UNLIKELY (2*(idx - table->size) + 1 != seq)) { \
/* Buffer full (or unlikely preemption). Can not push */ \
return false; \
} \
return false; \
} \
if (M_UNLIKELY (!atomic_compare_exchange_strong_explicit(&table->ProdIdx, \
&idx, idx+1, memory_order_relaxed, memory_order_relaxed))) { \
/* Thread has been preempted by another one. */ \
return false; \
} \
&idx, idx+1, memory_order_relaxed, memory_order_relaxed))) { \
/* Thread has been preempted by another one. */ \
return false; \
} \
/* If it is interrupted here, it may block pop method (not push) */ \
if (!BUFFERI_POLICY_P((policy), BUFFER_PUSH_INIT_POP_MOVE)) { \
M_CALL_SET(oplist, table->Tab[i].x, x); \
M_CALL_SET(oplist, table->Tab[i].x, x); \
} else { \
M_CALL_INIT_SET(oplist, table->Tab[i].x, x); \
} \
@@ -676,27 +679,27 @@ M_C(name, _init)(buffer_t v, size_t size) \
atomic_store_explicit(&table->Tab[i].seq, 2*idx, memory_order_release); \
QUEUEI_MPMC_CONTRACT(table); \
return true; \
} \
\
static inline bool \
M_C(name, _pop)(type *ptr, buffer_t table) \
{ \
} \
\
static inline bool \
M_C(name, _pop)(type *ptr, buffer_t table) \
{ \
QUEUEI_MPMC_CONTRACT(table); \
assert (ptr != NULL); \
unsigned int iC = atomic_load_explicit(&table->ConsoIdx, \
memory_order_relaxed); \
const unsigned int i = (iC & (table->size -1)); \
memory_order_relaxed); \
const unsigned int i = (iC & (table->size -1)); \
const unsigned int seq = atomic_load_explicit(&table->Tab[i].seq, \
memory_order_acquire); \
if (seq != 2 * iC) { \
if (seq != 2 * iC) { \
/* Nothing in buffer to consumme (or unlikely preemption) */ \
return false; \
} \
return false; \
} \
if (M_UNLIKELY (!atomic_compare_exchange_strong_explicit(&table->ConsoIdx, \
&iC, iC+1, memory_order_relaxed, memory_order_relaxed))) { \
/* Thread has been preempted by another one */ \
return false; \
} \
&iC, iC+1, memory_order_relaxed, memory_order_relaxed))) { \
/* Thread has been preempted by another one */ \
return false; \
} \
if (!BUFFERI_POLICY_P((policy), BUFFER_PUSH_INIT_POP_MOVE)) { \
M_CALL_SET(oplist, *ptr, table->Tab[i].x); \
} else { \
@@ -705,40 +708,40 @@ M_C(name, _init)(buffer_t v, size_t size) \
atomic_store_explicit(&table->Tab[i].seq, 2*iC + 1, memory_order_release); \
QUEUEI_MPMC_CONTRACT(table); \
return true; \
} \
\
static inline void \
M_C(name, _init)(buffer_t buffer, size_t size) \
{ \
assert (buffer != NULL); \
assert( M_POWEROF2_P(size)); \
} \
\
static inline void \
M_C(name, _init)(buffer_t buffer, size_t size) \
{ \
assert (buffer != NULL); \
assert( M_POWEROF2_P(size)); \
assert (0 < size && size <= UINT_MAX); \
assert(((policy) & (BUFFER_STACK|BUFFER_THREAD_UNSAFE|BUFFER_PUSH_OVERWRITE)) == 0); \
atomic_init(&buffer->ProdIdx, (unsigned int) size); \
atomic_init(&buffer->ConsoIdx, (unsigned int) size); \
buffer->size = (unsigned int) size; \
buffer->Tab = M_CALL_REALLOC(oplist, M_C(name, _el_t), NULL, size); \
if (buffer->Tab == NULL) { \
M_MEMORY_FULL (size*sizeof(M_C(name, _el_t) )); \
return; \
} \
if (buffer->Tab == NULL) { \
M_MEMORY_FULL (size*sizeof(M_C(name, _el_t) )); \
return; \
} \
for(unsigned int j = 0; j < (unsigned int) size; j++) { \
atomic_init(&buffer->Tab[j].seq, 2*j+1U); \
if (!BUFFERI_POLICY_P((policy), BUFFER_PUSH_INIT_POP_MOVE)) { \
M_CALL_INIT(oplist, buffer->Tab[j].x); \
M_CALL_INIT(oplist, buffer->Tab[j].x); \
} \
} \
} \
QUEUEI_MPMC_CONTRACT(buffer); \
} \
\
static inline void \
M_C(name, _clear)(buffer_t buffer) \
{ \
} \
\
static inline void \
M_C(name, _clear)(buffer_t buffer) \
{ \
QUEUEI_MPMC_CONTRACT(buffer); \
if (!BUFFERI_POLICY_P((policy), BUFFER_PUSH_INIT_POP_MOVE)) { \
for(unsigned int j = 0; j < buffer->size; j++) { \
M_CALL_CLEAR(oplist, buffer->Tab[j].x); \
} \
} \
} else { \
unsigned int iP = atomic_load_explicit(&buffer->ProdIdx, memory_order_relaxed); \
unsigned int i = iP & (buffer->size -1); \
@@ -751,19 +754,19 @@ M_C(name, _init)(buffer_t v, size_t size) \
j = 0; \
} \
} \
M_CALL_FREE(oplist, buffer->Tab); \
M_CALL_FREE(oplist, buffer->Tab); \
buffer->Tab = NULL; /* safer */ \
} \
\
static inline size_t \
} \
\
static inline size_t \
M_C(name, _size)(buffer_t table) \
{ \
{ \
QUEUEI_MPMC_CONTRACT(table); \
const unsigned int iC = atomic_load_explicit(&table->ConsoIdx, memory_order_relaxed); \
const unsigned int iP = atomic_load_explicit(&table->ProdIdx, memory_order_acquire); \
/* We return an approximation as we can't read both iC & iP atomically \
As we read producer index after consummer index, \
and they are atomic variables without reordering \
and they are atomic variables without reordering \
producer index is always greater or equal than consumer index \
(or on overflow occurs, in which case as we compute with modulo \
arithmetic, the right result is computed). \
@@ -771,27 +774,27 @@ M_C(name, _init)(buffer_t v, size_t size) \
if the function is interrupted a long time between reading iC & \
iP. the function is not protected against it. \
*/ \
return iP-iC; \
} \
return iP-iC; \
} \
\
static inline size_t \
M_C(name, _capacity)(buffer_t v) \
M_C(name, _capacity)(buffer_t v) \
{ \
QUEUEI_MPMC_CONTRACT(v); \
return v->size; \
} \
\
static inline bool \
M_C(name, _empty_p)(buffer_t v) \
{ \
return M_C(name, _size) (v) == 0; \
} \
\
static inline bool \
M_C(name, _full_p)(buffer_t v) \
{ \
static inline bool \
M_C(name, _empty_p)(buffer_t v) \
{ \
return M_C(name, _size) (v) == 0; \
} \
\
static inline bool \
M_C(name, _full_p)(buffer_t v) \
{ \
return M_C(name, _size)(v) >= v->size; \
} \
} \
@@ -810,11 +813,11 @@ M_C(name, _init)(buffer_t v, size_t size) \
#define QUEUEI_SPSC_DEF_P1(arg) QUEUEI_SPSC_DEF_P2 arg
/* Validate the value oplist before going further */
#define QUEUEI_SPSC_DEF_P2(name, type, policy, oplist, buffer_t) \
#define QUEUEI_SPSC_DEF_P2(name, type, policy, oplist, buffer_t) \
M_IF_OPLIST(oplist)(QUEUEI_SPSC_DEF_P3, QUEUEI_SPSC_DEF_FAILURE)(name, type, policy, oplist, buffer_t)
/* Stop processing with a compilation failure */
#define QUEUEI_SPSC_DEF_FAILURE(name, type, policy, oplist, buffer_t) \
#define QUEUEI_SPSC_DEF_FAILURE(name, type, policy, oplist, buffer_t) \
M_STATIC_FAILURE(M_LIB_NOT_AN_OPLIST, "(QUEUE_SPSC_DEF): the given argument is not a valid oplist: " M_AS_STR(oplist))
#ifdef NDEBUG
@@ -833,10 +836,10 @@ M_C(name, _init)(buffer_t v, size_t size) \
#define QUEUEI_SPSC_DEF_P3(name, type, policy, oplist, buffer_t) \
\
typedef struct M_C(name, _el_s) { \
type x; \
} M_C(name, _el_t); \
\
typedef struct M_C(name, _el_s) { \
type x; \
} M_C(name, _el_t); \
\
typedef struct M_C(name, _s) { \
atomic_uint consoIdx; /* Can only increase until overflow */ \
unsigned int size; \
@@ -845,9 +848,9 @@ M_C(name, _init)(buffer_t v, size_t size) \
atomic_uint prodIdx; /* Can only increase until overflow */ \
} buffer_t[1]; \
\
static inline bool \
static inline bool \
M_C(name, _push)(buffer_t table, type const x) \
{ \
{ \
QUEUEI_SPSC_CONTRACT(table); \
unsigned int r = atomic_load_explicit(&table->consoIdx, \
memory_order_relaxed); \
@@ -857,18 +860,18 @@ M_C(name, _init)(buffer_t v, size_t size) \
return false; \
unsigned int i = w & (table->size -1); \
if (!BUFFERI_POLICY_P((policy), BUFFER_PUSH_INIT_POP_MOVE)) { \
M_CALL_SET(oplist, table->Tab[i].x, x); \
M_CALL_SET(oplist, table->Tab[i].x, x); \
} else { \
M_CALL_INIT_SET(oplist, table->Tab[i].x, x); \
} \
atomic_store_explicit(&table->prodIdx, w+1, memory_order_release); \
atomic_store_explicit(&table->prodIdx, w+1, memory_order_release); \
QUEUEI_SPSC_CONTRACT(table); \
return true; \
} \
\
static inline bool \
static inline bool \
M_C(name, _push_move)(buffer_t table, type x) \
{ \
{ \
QUEUEI_SPSC_CONTRACT(table); \
unsigned int r = atomic_load_explicit(&table->consoIdx, \
memory_order_relaxed); \
@@ -878,24 +881,24 @@ M_C(name, _init)(buffer_t v, size_t size) \
return false; \
unsigned int i = w & (table->size -1); \
if (!BUFFERI_POLICY_P((policy), BUFFER_PUSH_INIT_POP_MOVE)) { \
M_DO_MOVE(oplist, table->Tab[i].x, x); \
M_DO_MOVE(oplist, table->Tab[i].x, x); \
} else { \
M_DO_INIT_MOVE(oplist, table->Tab[i].x, x); \
} \
atomic_store_explicit(&table->prodIdx, w+1, memory_order_release); \
atomic_store_explicit(&table->prodIdx, w+1, memory_order_release); \
QUEUEI_SPSC_CONTRACT(table); \
return true; \
} \
\
static inline bool \
static inline bool \
M_C(name, _pop)(type *ptr, buffer_t table) \
{ \
{ \
QUEUEI_SPSC_CONTRACT(table); \
assert (ptr != NULL); \
unsigned int w = atomic_load_explicit(&table->prodIdx, \
memory_order_relaxed); \
memory_order_relaxed); \
unsigned int r = atomic_load_explicit(&table->consoIdx, \
memory_order_acquire); \
memory_order_acquire); \
if (w-r == 0) \
return false; \
unsigned int i = r & (table->size -1); \
@@ -904,14 +907,14 @@ M_C(name, _init)(buffer_t v, size_t size) \
} else { \
M_DO_INIT_MOVE (oplist, *ptr, table->Tab[i].x); \
} \
atomic_store_explicit(&table->consoIdx, r+1, memory_order_release); \
atomic_store_explicit(&table->consoIdx, r+1, memory_order_release); \
QUEUEI_SPSC_CONTRACT(table); \
return true; \
} \
\
static inline unsigned \
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); \
@@ -935,16 +938,16 @@ M_C(name, _init)(buffer_t v, size_t size) \
return max; \
} \
\
static inline unsigned \
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); \
memory_order_relaxed); \
unsigned int r = atomic_load_explicit(&table->consoIdx, \
memory_order_acquire); \
memory_order_acquire); \
if (w-r == 0) \
return 0; \
unsigned int max = M_MIN(w-r, n); \
@@ -961,9 +964,9 @@ M_C(name, _init)(buffer_t v, size_t size) \
return max; \
} \
\
static inline void \
static inline void \
M_C(name, _push_force)(buffer_t table, type const x) \
{ \
{ \
QUEUEI_SPSC_CONTRACT(table); \
unsigned int r = atomic_load_explicit(&table->consoIdx, \
memory_order_relaxed); \
@@ -976,11 +979,11 @@ M_C(name, _init)(buffer_t v, size_t size) \
} \
unsigned int i = w & (table->size -1); \
if (!BUFFERI_POLICY_P((policy), BUFFER_PUSH_INIT_POP_MOVE)) { \
M_CALL_SET(oplist, table->Tab[i].x, x); \
M_CALL_SET(oplist, table->Tab[i].x, x); \
} else { \
M_CALL_INIT_SET(oplist, table->Tab[i].x, x); \
} \
atomic_store_explicit(&table->prodIdx, w+1, memory_order_release); \
atomic_store_explicit(&table->prodIdx, w+1, memory_order_release); \
QUEUEI_SPSC_CONTRACT(table); \
} \
\
@@ -989,12 +992,12 @@ M_C(name, _init)(buffer_t v, size_t size) \
{ \
QUEUEI_SPSC_CONTRACT(table); \
unsigned int r = atomic_load_explicit(&table->consoIdx, \
memory_order_relaxed); \
memory_order_relaxed); \
unsigned int w = atomic_load_explicit(&table->prodIdx, \
memory_order_acquire); \
memory_order_acquire); \
/* We return an approximation as we can't read both r & w atomically \
As we read producer index after consummer index, \
and they are atomic variables without reordering \
and they are atomic variables without reordering \
producer index is always greater or equal than consumer index \
(or on overflow occurs, in which case as we compute with modulo \
arithmetic, the right result is computed). \
@@ -1006,54 +1009,54 @@ M_C(name, _init)(buffer_t v, size_t size) \
} \
\
static inline size_t \
M_C(name, _capacity)(buffer_t v) \
M_C(name, _capacity)(buffer_t v) \
{ \
return v->size; \
} \
\
static inline bool \
M_C(name, _empty_p)(buffer_t v) \
{ \
return M_C(name, _size) (v) == 0; \
} \
\
static inline bool \
M_C(name, _full_p)(buffer_t v) \
{ \
return M_C(name, _size)(v) >= v->size; \
} \
static inline bool \
M_C(name, _empty_p)(buffer_t v) \
{ \
return M_C(name, _size) (v) == 0; \
} \
\
static inline void \
M_C(name, _init)(buffer_t buffer, size_t size) \
{ \
assert (buffer != NULL); \
assert( M_POWEROF2_P(size)); \
static inline bool \
M_C(name, _full_p)(buffer_t v) \
{ \
return M_C(name, _size)(v) >= v->size; \
} \
\
static inline void \
M_C(name, _init)(buffer_t buffer, size_t size) \
{ \
assert (buffer != NULL); \
assert( M_POWEROF2_P(size)); \
assert (0 < size && size <= UINT_MAX); \
assert(((policy) & (BUFFER_STACK|BUFFER_THREAD_UNSAFE|BUFFER_PUSH_OVERWRITE)) == 0); \
atomic_init(&buffer->prodIdx, (unsigned int) size); \
atomic_init(&buffer->consoIdx, (unsigned int) size); \
buffer->size = (unsigned int) size; \
buffer->Tab = M_CALL_REALLOC(oplist, M_C(name, _el_t), NULL, size); \
if (buffer->Tab == NULL) { \
M_MEMORY_FULL (size*sizeof(M_C(name, _el_t) )); \
return; \
} \
if (buffer->Tab == NULL) { \
M_MEMORY_FULL (size*sizeof(M_C(name, _el_t) )); \
return; \
} \
if (!BUFFERI_POLICY_P((policy), BUFFER_PUSH_INIT_POP_MOVE)) { \
for(unsigned int j = 0; j < (unsigned int) size; j++) { \
M_CALL_INIT(oplist, buffer->Tab[j].x); \
M_CALL_INIT(oplist, buffer->Tab[j].x); \
} \
} \
} \
QUEUEI_SPSC_CONTRACT(buffer); \
} \
\
static inline void \
M_C(name, _clear)(buffer_t buffer) \
{ \
} \
\
static inline void \
M_C(name, _clear)(buffer_t buffer) \
{ \
QUEUEI_SPSC_CONTRACT(buffer); \
if (!BUFFERI_POLICY_P((policy), BUFFER_PUSH_INIT_POP_MOVE)) { \
for(unsigned int j = 0; j < buffer->size; j++) { \
M_CALL_CLEAR(oplist, buffer->Tab[j].x); \
} \
} \
} else { \
unsigned int iP = atomic_load_explicit(&buffer->prodIdx, memory_order_relaxed); \
unsigned int i = iP & (buffer->size -1); \
@@ -1066,33 +1069,33 @@ M_C(name, _init)(buffer_t v, size_t size) \
j = 0; \
} \
} \
M_CALL_FREE(oplist, buffer->Tab); \
M_CALL_FREE(oplist, buffer->Tab); \
buffer->Tab = NULL; /* safer */ \
} \
\
} \
\
/* Deferred evaluation for the definition,
so that all arguments are evaluated before further expansion */
#define BUFFERI_OPLIST_P1(arg) BUFFERI_OPLIST_P2 arg
/* Validation of the given oplist */
#define BUFFERI_OPLIST_P2(name, oplist) \
#define BUFFERI_OPLIST_P2(name, oplist) \
M_IF_OPLIST(oplist)(BUFFERI_OPLIST_P3, BUFFERI_OPLIST_FAILURE)(name, oplist)
/* Prepare a clean compilation failure */
#define BUFFERI_OPLIST_FAILURE(name, oplist) \
#define BUFFERI_OPLIST_FAILURE(name, oplist) \
((M_LIB_ERROR(ARGUMENT_OF_BUFFER_OPLIST_IS_NOT_AN_OPLIST, name, oplist)))
/* OPLIST defininition of a buffer */
#define BUFFERI_OPLIST_P3(name, oplist) \
#define BUFFERI_OPLIST_P3(name, oplist) \
(INIT(M_C(name, _int_init)) \
,INIT_SET(M_C(name, _init_set)) \
,SET(M_C(name, _set)) \
,CLEAR(M_C(name, _clear)) \
,TYPE(M_C(name,_t)) \
,SUBTYPE(M_C(name, _type_t)) \
,CLEAN(M_C(name,_clean)) \
,PUSH(M_C(name,_push)) \
,INIT_SET(M_C(name, _init_set)) \
,SET(M_C(name, _set)) \
,CLEAR(M_C(name, _clear)) \
,TYPE(M_C(name,_t)) \
,SUBTYPE(M_C(name, _type_t)) \
,CLEAN(M_C(name,_clean)) \
,PUSH(M_C(name,_push)) \
,POP(M_C(name,_pop)) \
,OPLIST(oplist) \
,TEST_EMPTY(M_C(name, _empty_p)), \