diff --git a/Makefile b/Makefile index 148eb33b..fe1ab31a 100644 --- a/Makefile +++ b/Makefile @@ -37,7 +37,7 @@ PACKAGE=m_lib-$(VERSION) VERSION=0.8.0 # Define the contain of the distribution tarball -HEADER=m-algo.h m-array.h m-atomic.h m-bitset.h m-bptree.h m-buffer.h m-c-mempool.h m-concurrent.h m-core.h m-deque.h m-dict.h m-funcobj.h m-generic.h m-genint.h m-i-list.h m-i-shared.h m-list.h m-mempool.h m-thread.h m-mutex.h m-prioqueue.h m-rbtree.h m-serial-bin.h m-serial-json.h m-shared.h m-snapshot.h m-string.h m-tree.h m-try.h m-tuple.h m-variant.h m-worker.h m-bstring.h m-shared-ptr.h +HEADER=m-algo.h m-array.h m-atomic.h m-bitset.h m-bptree.h m-buffer.h m-c-mempool.h m-concurrent.h m-core.h m-deque.h m-dict.h m-funcobj.h m-generic.h m-genint.h m-i-list.h m-i-shared.h m-list.h m-mempool.h m-thread.h m-mutex.h m-prioqueue.h m-rbtree.h m-serial-bin.h m-serial-json.h m-shared.h m-snapshot.h m-string.h m-tree.h m-try.h m-tuple.h m-variant.h m-worker.h m-bstring.h m-shared-ptr.h m-queue.h DOC1=LICENSE README.md DOC2=doc/API.txt doc/Container.html doc/Container.ods doc/depend.png doc/DEV.md doc/ISSUES.org doc/oplist.odp doc/oplist.png doc/bench-array-log.png doc/bench-array.png doc/bench-list-log.png doc/bench-list.png doc/bench-oset-log.png doc/bench-oset.png doc/bench-umap-log.png doc/bench-umap.png doc/cc.sh EXAMPLE=example/ex11-algo01.c example/ex11-algo02.c example/ex11-json01.c example/ex11-section.c example/ex-algo02.c example/ex-algo03.c example/ex-algo04.c example/ex-array00.c example/ex-array01.c example/ex-array02.c example/ex-array03.c example/ex-array04.c example/ex-array05.c example/ex-bptree01.c example/ex-buffer01.c example/ex-dict01.c example/ex-dict02.c example/ex-dict03.c example/ex-dict04.c example/ex-grep01.c example/ex-list01.c example/ex-mph.c example/ex-multi01.c example/ex11-multi02.c example/ex-multi03.c example/ex-multi04.c example/ex-multi05.c example/ex-rbtree01.c example/ex11-algo02.json example/ex11-json01.json example/Makefile example/ex-defer01.c example/ex-string01.c example/ex-string02.c example/ex-astar.c example/ex-string03.c example/ex11-tstc.c example/ex-no-stdio.c example/ex11-count-lines.c example/ex11-json-net.c example/ex11-rbtree02.c example/ex-curl.c example/ex-dict05.c example/ex-dict06.c example/exm-lib.c example/exm-main.c example/exn-lib.c example/exn-main.c example/exm-header.h example/exn-header.h example/ex-alloc1.c example/ex-alloc2.c example/ex-try01.c example/ex11-generic01.c example/ex-bptree02.c example/ex-i-list.c example/ex11-generic02.c example/ex11-small-name.c example/ex11-variant01.c diff --git a/README.md b/README.md index e3f2ec34..f3ed1cdf 100644 --- a/README.md +++ b/README.md @@ -3878,8 +3878,8 @@ void name_reset(buffer_t buffer) /* Thread safe */ bool name_empty_p(const buffer_t buffer) /* Thread safe */ size_t name_size(const buffer_t buffer) /* Thread safe */ size_t name_capacity(const buffer_t buffer) /* Thread safe */ -bool name_push(buffer_t buffer, const type data) -bool name_pop(type *data, buffer_t buffer) +void name_push(buffer_t buffer, const type data) +void name_pop(type *data, buffer_t buffer) ``` #### Specialized methods diff --git a/doc/API.txt b/doc/API.txt index 248acfef..5cea1ec0 100644 --- a/doc/API.txt +++ b/doc/API.txt @@ -94,4 +94,6 @@ V0.8.0 : * REALLOC / FREE operators are incompatible: This allows more and efficient integration into custom allocators. * M_MEMORY_FULL: replace the failed size allocation size argument into the base type and the number of elements that failed to be allocated. - + * m-buffer: non-thread safe queue & stack are moved into their dedicated header (m-queue) + * m-buffer: Remove buffer blocking policy and thread safe (_push / _pop returns now void) + diff --git a/example/ex11-use-pool.c b/example/ex11-use-pool.c index c56a0dff..785ad189 100644 --- a/example/ex11-use-pool.c +++ b/example/ex11-use-pool.c @@ -150,6 +150,12 @@ M_WORKER_SPAWN_DEF2(worker_str, (str, string_t) ) M_SHARED_PTR_DEF(shared_int, int) M_SHARED_PTR_DEF(shared_str, string_t) +#include "m-queue.h" + +M_QUEUE_DEF(queue_10, int, 10 ) +M_STACK_DEF(stack_20, string_t, 20) + + int main(void) { return 0; diff --git a/m-buffer.h b/m-buffer.h index a537689c..cdcdb779 100644 --- a/m-buffer.h +++ b/m-buffer.h @@ -1,5 +1,5 @@ /* - * M*LIB - Fixed size (Bounded) QUEUE & STACK interface + * M*LIB - Fixed size (Bounded) QUEUE & STACK interface (Thread safe) * * Copyright (c) 2017-2025, Patrick Pelissier * All rights reserved. @@ -31,20 +31,12 @@ /* Define the different kind of policy a lock-based buffer can have: * - the buffer can be either a queue (policy is FIFO) or a stack (policy is FILO), - * - if the push method is by default blocking (waiting for the buffer to has some space) or not, *** deprecated *** - * - if the pop method is by default blocking (waiting for the buffer to has some data) or not, *** deprecated *** - * - if both methods are blocking, *** deprecated *** - * - if it shall be thread safe or not (i.e. remove the mutex lock and atomic costs), * - if the buffer has to be init with empty elements, or if it shall init an element when it is pushed (and moved when popped), * - if the buffer has to overwrite the last element if the buffer is full, * - if the pop of an element is not complete until the call to pop_release (preventing push until this call). */ typedef enum { M_BUFFER_QUEUE = 0, M_BUFFER_STACK = 1, - M_BUFFER_BLOCKING_PUSH = 0, M_BUFFER_UNBLOCKING_PUSH = 2, - M_BUFFER_BLOCKING_POP = 0, M_BUFFER_UNBLOCKING_POP = 4, - M_BUFFER_BLOCKING = 0, M_BUFFER_UNBLOCKING = 6, - M_BUFFER_THREAD_SAFE = 0, M_BUFFER_THREAD_UNSAFE = 8, M_BUFFER_PUSH_INIT_POP_MOVE = 16, M_BUFFER_PUSH_OVERWRITE = 32, M_BUFFER_DEFERRED_POP = 64 @@ -138,80 +130,6 @@ typedef enum { #define M_BUFF3R_POLICY_P(policy, val) \ (((policy) & (val)) != 0) -/* Handle either atomic integer or normal integer in function of the policy - parameter of the buffer BUFFER_THREAD_UNSAFE (BUFFER_THREAD_SAFE is the - default). This enables avoiding to pay the cost of atomic operations if not - applicable. - */ -typedef union m_buff3r_number_s { - unsigned int u; - atomic_uint a; -#ifdef __cplusplus - // Not sure why, but C++ needs an explicit default constructor for this union. - m_buff3r_number_s() : u(0) {}; -#endif -} m_buff3r_number_ct[1]; - -M_INLINE void -m_buff3r_number_init(m_buff3r_number_ct n, unsigned int policy) -{ - if (!M_BUFF3R_POLICY_P(policy, M_BUFFER_THREAD_UNSAFE)) - atomic_init(&n->a, 0U); - else - n->u = 0UL; -} - -M_INLINE unsigned int -m_buff3r_number_load(const m_buff3r_number_ct n, unsigned int policy) -{ - if (!M_BUFF3R_POLICY_P(policy, M_BUFFER_THREAD_UNSAFE)) - // Perform a memory acquire so that further usage of the buffer - // is synchronized. - return atomic_load_explicit(&n->a, memory_order_acquire); - else - return n->u; -} - -M_INLINE void -m_buff3r_number_store(m_buff3r_number_ct n, unsigned int v, unsigned int policy) -{ - if (!M_BUFF3R_POLICY_P(policy, M_BUFFER_THREAD_UNSAFE)) - // This function is used in context where a relaxed access is sufficient. - atomic_store_explicit(&n->a, v, memory_order_relaxed); - else - n->u = v; -} - -M_INLINE void -m_buff3r_number_set(m_buff3r_number_ct n, m_buff3r_number_ct v, unsigned int policy) -{ - if (!M_BUFF3R_POLICY_P(policy, M_BUFFER_THREAD_UNSAFE)) - // This function is used in context where a relaxed access is sufficient. - atomic_store_explicit(&n->a, atomic_load_explicit(&v->a, memory_order_relaxed), memory_order_relaxed); - else - n->u = v->u; -} - -M_INLINE unsigned int -m_buff3r_number_inc(m_buff3r_number_ct n, unsigned int policy) -{ - if (!M_BUFF3R_POLICY_P(policy, M_BUFFER_THREAD_UNSAFE)) - return atomic_fetch_add(&n->a, 1U); - else - return n->u ++; -} - -M_INLINE unsigned int -m_buff3r_number_dec(m_buff3r_number_ct n, unsigned int policy) -{ - if (!M_BUFF3R_POLICY_P(policy, M_BUFFER_THREAD_UNSAFE)) - return atomic_fetch_sub(&n->a, 1U); - else - return n->u --; -} - -/********************************** INTERNAL *********************************/ - /* Test if the size is only run-time or build time */ #define M_BUFF3R_IF_CTE_SIZE(m_size) M_IF(M_BOOL(m_size)) @@ -230,7 +148,7 @@ m_buff3r_number_dec(m_buff3r_number_ct n, unsigned int policy) /* Contract of a buffer within a protected section */ #define M_BUFF3R_PROTECTED_CONTRACT(policy, buffer, size) do { \ - M_ASSERT (m_buff3r_number_load(buffer->number[0], policy) <= M_BUFF3R_SIZE(size)); \ + M_ASSERT (atomic_load_explicit(&buffer->number[0], memory_order_acquire) <= M_BUFF3R_SIZE(size)); \ } while (0) @@ -258,7 +176,7 @@ m_buff3r_number_dec(m_buff3r_number_ct n, unsigned int policy) M_BUFF3R_DEF_TYPE(name, type, m_size, policy, oplist, buffer_t) \ M_CHECK_COMPATIBLE_OPLIST(name, 1, type, oplist) \ M_BUFF3R_DEF_CORE(name, type, m_size, policy, oplist, buffer_t) \ - M_EMPLACE_QUEUE_DEF(name, buffer_t, _emplace, oplist, M_EMPLACE_QUEUE_GENE) + M_EMPLACE_QUEUE_DEF(name, buffer_t, _emplace, oplist, M_BUFF3R_EMPLACE_QUEUE_GENE) /* Define the type of a buffer */ #define M_BUFF3R_DEF_TYPE(name, type, m_size, policy, oplist, buffer_t) \ @@ -267,7 +185,7 @@ m_buff3r_number_dec(m_buff3r_number_ct n, unsigned int policy) by multiple writing threads. No need to align if there is no thread */ \ typedef union M_F(name, _el_s) { \ type x; \ - char align[M_BUFF3R_POLICY_P(policy, M_BUFFER_THREAD_UNSAFE) ? 1 : M_ALIGN_FOR_CACHELINE_EXCLUSION]; \ + char align[M_ALIGN_FOR_CACHELINE_EXCLUSION]; \ } M_F(name, _el_ct); \ \ typedef struct M_F(name, _s) { \ @@ -284,7 +202,7 @@ m_buff3r_number_dec(m_buff3r_number_ct n, unsigned int policy) size_t idx_cons; /* Index of the consumption threads */ \ /* number[0] := Number of elements in the buffer */ \ /* number[1] := [OPTION] Number of elements being deferred in the buffer */ \ - m_buff3r_number_ct number[1 + M_BUFF3R_POLICY_P(policy, M_BUFFER_DEFERRED_POP)]; \ + atomic_uint number[1 + M_BUFF3R_POLICY_P(policy, M_BUFFER_DEFERRED_POP)]; \ /* If fixed size, array of elements, otherwise pointer to element */ \ M_F(name, _el_ct) M_BUFF3R_IF_CTE_SIZE(m_size)(data[m_size], *data); \ } buffer_t[1]; \ @@ -300,23 +218,19 @@ m_buff3r_number_dec(m_buff3r_number_ct n, unsigned int policy) /* Define the core functionalities of a buffer */ #define M_BUFF3R_DEF_CORE(name, type, m_size, policy, oplist, buffer_t) \ \ -M_P(void, name, _init, buffer_t v, size_t size) \ +M_N(void, name, _init, buffer_t v, size_t size) \ { \ M_ASSERT(size <= UINT_MAX); \ - M_ASSERT_POOL(); \ + M_GLOBAL_POOL(); \ M_BUFF3R_IF_CTE_SIZE(m_size)(M_ASSERT(size == m_size), v->capacity = size); \ v->idx_prod = v->idx_cons = v->overwrite = 0; \ - m_buff3r_number_init (v->number[0], policy); \ + atomic_init (&v->number[0], 0U); \ if (M_BUFF3R_POLICY_P(policy, M_BUFFER_DEFERRED_POP)) \ - m_buff3r_number_init (v->number[1], policy); \ - if (!M_BUFF3R_POLICY_P((policy), M_BUFFER_THREAD_UNSAFE)) { \ - m_mutex_init(v->mutexPush); \ - m_mutex_init(v->mutexPop); \ - m_cond_init(v->there_is_data); \ - m_cond_init(v->there_is_room_for_data); \ - } else { \ - M_ASSERT(M_BUFF3R_POLICY_P((policy), M_BUFFER_UNBLOCKING)); \ - } \ + atomic_init (&v->number[1], 0U); \ + m_mutex_init(v->mutexPush); \ + m_mutex_init(v->mutexPop); \ + m_cond_init(v->there_is_data); \ + m_cond_init(v->there_is_room_for_data); \ \ M_BUFF3R_IF_CTE_SIZE(m_size)( /* Statically allocated */ , \ v->data = M_CALL_REALLOC(oplist, M_F(name, _el_ct), NULL, 0, M_BUFF3R_SIZE(m_size)); \ @@ -333,16 +247,16 @@ M_P(void, name, _init, buffer_t v, size_t size) \ } \ \ M_BUFF3R_IF_CTE_SIZE(m_size)( \ - M_P(void, name,_i_init, buffer_t v) \ + M_N(void, name,_i_init, buffer_t v) \ { \ - M_F(name, _init)M_R(v, m_size); \ + M_F(name, _init)(v, m_size); \ } \ , ) \ \ - M_P(void, name, _i_clear_obj, buffer_t v) \ + M_N(void, name, _i_clear_obj, buffer_t v) \ { \ M_BUFF3R_CONTRACT(v,m_size); \ - M_ASSERT_POOL(); \ + M_GLOBAL_POOL(); \ if (!M_BUFF3R_POLICY_P((policy), M_BUFFER_PUSH_INIT_POP_MOVE)) { \ for(size_t i = 0; i < M_BUFF3R_SIZE(m_size); i++) { \ M_CALL_CLEAR(oplist, v->data[i].x); \ @@ -357,64 +271,59 @@ M_P(void, name, _init, buffer_t v, size_t size) \ } \ } \ v->idx_prod = v->idx_cons = 0; \ - m_buff3r_number_store (v->number[0], 0U, policy); \ + atomic_store_explicit (&v->number[0], 0U, memory_order_relaxed); \ if (M_BUFF3R_POLICY_P(policy, M_BUFFER_DEFERRED_POP)) \ - m_buff3r_number_store(v->number[1], 0U, policy); \ + atomic_store_explicit (&v->number[1], 0U, memory_order_relaxed); \ M_BUFF3R_CONTRACT(v,m_size); \ } \ \ - M_P(void, name, _clear, buffer_t v) \ + M_N(void, name, _clear, buffer_t v) \ { \ M_BUFF3R_CONTRACT(v,m_size); \ - M_F(name,_i_clear_obj)M_R(v); \ + M_GLOBAL_POOL(); \ + M_F(name,_i_clear_obj)(v); \ M_BUFF3R_IF_CTE_SIZE(m_size)( , \ M_CALL_FREE(oplist, M_F(name, _el_ct), v->data, M_BUFF3R_SIZE(m_size)); \ v->data = NULL; \ ) \ v->overwrite = 0; \ - if (!M_BUFF3R_POLICY_P((policy), M_BUFFER_THREAD_UNSAFE)) { \ - m_mutex_clear(v->mutexPush); \ - m_mutex_clear(v->mutexPop); \ - m_cond_clear(v->there_is_data); \ - m_cond_clear(v->there_is_room_for_data); \ - } \ + m_mutex_clear(v->mutexPush); \ + m_mutex_clear(v->mutexPop); \ + m_cond_clear(v->there_is_data); \ + m_cond_clear(v->there_is_room_for_data); \ } \ \ - M_P(void, name, _reset, buffer_t v) \ + M_N(void, name, _reset, buffer_t v) \ { \ M_BUFF3R_CONTRACT(v,m_size); \ - if (!M_BUFF3R_POLICY_P((policy), M_BUFFER_THREAD_UNSAFE)) { \ - m_mutex_lock(v->mutexPush); \ - m_mutex_lock(v->mutexPop); \ - } \ + M_GLOBAL_POOL(); \ + m_mutex_lock(v->mutexPush); \ + m_mutex_lock(v->mutexPop); \ M_BUFF3R_PROTECTED_CONTRACT(policy, v, m_size); \ if (M_BUFF3R_POLICY_P((policy), M_BUFFER_PUSH_INIT_POP_MOVE)) \ - M_F(name,_i_clear_obj)M_R(v); \ + M_F(name,_i_clear_obj)(v); \ v->idx_prod = v->idx_cons = 0; \ - m_buff3r_number_store (v->number[0], 0U, policy); \ + atomic_store_explicit (&v->number[0], 0U, memory_order_relaxed); \ if (M_BUFF3R_POLICY_P(policy, M_BUFFER_DEFERRED_POP)) \ - m_buff3r_number_store(v->number[1], 0U, policy); \ - if (!M_BUFF3R_POLICY_P((policy), M_BUFFER_THREAD_UNSAFE)) { \ - m_cond_broadcast(v->there_is_room_for_data); \ - m_mutex_unlock(v->mutexPop); \ - m_mutex_unlock(v->mutexPush); \ - } \ + atomic_store_explicit(&v->number[1], 0U, memory_order_relaxed); \ + m_cond_broadcast(v->there_is_room_for_data); \ + m_mutex_unlock(v->mutexPop); \ + m_mutex_unlock(v->mutexPush); \ M_BUFF3R_CONTRACT(v,m_size); \ } \ \ - M_P(void, name, _init_set, buffer_t dest, const buffer_t src) \ + M_N(void, name, _init_set, buffer_t dest, const buffer_t src) \ { \ - /* un-const 'src', so that we can lock it (semantically it is const) */ \ + M_GLOBAL_POOL(); \ + /* un-const 'src', so that we can lock it (semantically it is const) */ \ M_F(name, _uptr_ct) vu; \ vu.cptr = src; \ M_F(name, _ptr) v = vu.ptr; \ M_ASSERT (dest != v); \ M_BUFF3R_CONTRACT(v,m_size); \ - M_F(name, _init)M_R(dest, M_BUFF3R_SIZE(m_size)); \ - if (!M_BUFF3R_POLICY_P((policy), M_BUFFER_THREAD_UNSAFE)) { \ - m_mutex_lock(v->mutexPush); \ - m_mutex_lock(v->mutexPop); \ - } \ + M_F(name, _init)(dest, M_BUFF3R_SIZE(m_size)); \ + m_mutex_lock(v->mutexPush); \ + m_mutex_lock(v->mutexPop); \ \ M_BUFF3R_PROTECTED_CONTRACT(policy, v, m_size); \ if (!M_BUFF3R_POLICY_P((policy), M_BUFFER_PUSH_INIT_POP_MOVE)) { \ @@ -433,21 +342,20 @@ M_P(void, name, _init, buffer_t v, size_t size) \ \ dest->idx_prod = v->idx_prod; \ dest->idx_cons = v->idx_cons; \ - m_buff3r_number_set (dest->number[0], v->number[0], policy); \ + atomic_store_explicit (&dest->number[0], v->number[0], memory_order_relaxed); \ if (M_BUFF3R_POLICY_P(policy, M_BUFFER_DEFERRED_POP)) \ - m_buff3r_number_set(dest->number[1], v->number[1], policy); \ + atomic_store_explicit(&dest->number[1], v->number[1], memory_order_relaxed); \ \ - if (!M_BUFF3R_POLICY_P((policy), M_BUFFER_THREAD_UNSAFE)) { \ - m_mutex_unlock(v->mutexPop); \ - m_mutex_unlock(v->mutexPush); \ - } \ + m_mutex_unlock(v->mutexPop); \ + m_mutex_unlock(v->mutexPush); \ M_BUFF3R_CONTRACT(v,m_size); \ M_BUFF3R_CONTRACT(dest, m_size); \ } \ \ - M_P(void, name, _set, buffer_t dest, const buffer_t src) \ + M_N(void, name, _set, buffer_t dest, const buffer_t src) \ { \ - /* un-const 'src', so that we can lock it (semantically it is const) */ \ + M_GLOBAL_POOL(); \ + /* un-const 'src', so that we can lock it (semantically it is const) */ \ M_F(name, _uptr_ct) vu; \ vu.cptr = src; \ M_F(name, _ptr) v = vu.ptr; \ @@ -455,24 +363,22 @@ M_P(void, name, _init, buffer_t v, size_t size) \ M_BUFF3R_CONTRACT(v,m_size); \ \ if (dest == v) return; \ - if (!M_BUFF3R_POLICY_P((policy), M_BUFFER_THREAD_UNSAFE)) { \ - /* Case of deadlock: A := B, B:=C, C:=A (all in //) \ - Solution: order the lock by increasing memory */ \ - if (dest < v) { \ - m_mutex_lock(dest->mutexPush); \ - m_mutex_lock(dest->mutexPop); \ - m_mutex_lock(v->mutexPush); \ - m_mutex_lock(v->mutexPop); \ - } else { \ - m_mutex_lock(v->mutexPush); \ - m_mutex_lock(v->mutexPop); \ - m_mutex_lock(dest->mutexPush); \ - m_mutex_lock(dest->mutexPop); \ - } \ + /* Case of deadlock: A := B, B:=C, C:=A (all in //) \ + Solution: order the lock by increasing memory */ \ + if (dest < v) { \ + m_mutex_lock(dest->mutexPush); \ + m_mutex_lock(dest->mutexPop); \ + m_mutex_lock(v->mutexPush); \ + m_mutex_lock(v->mutexPop); \ + } else { \ + m_mutex_lock(v->mutexPush); \ + m_mutex_lock(v->mutexPop); \ + m_mutex_lock(dest->mutexPush); \ + m_mutex_lock(dest->mutexPop); \ } \ \ M_BUFF3R_PROTECTED_CONTRACT(policy, v, m_size); \ - M_F(name,_i_clear_obj)M_R(dest); \ + M_F(name,_i_clear_obj)(dest); \ \ if (!M_BUFF3R_POLICY_P((policy), M_BUFFER_PUSH_INIT_POP_MOVE)) { \ for(size_t i = 0; i < M_BUFF3R_SIZE(m_size); i++) { \ @@ -490,25 +396,22 @@ M_P(void, name, _init, buffer_t v, size_t size) \ \ dest->idx_prod = v->idx_prod; \ dest->idx_cons = v->idx_cons; \ - m_buff3r_number_set (dest->number[0], v->number[0], policy); \ + atomic_store_explicit(&dest->number[0], v->number[0], memory_order_relaxed); \ if (M_BUFF3R_POLICY_P(policy, M_BUFFER_DEFERRED_POP)) \ - m_buff3r_number_set(dest->number[1], v->number[1], policy); \ + atomic_store_explicit(&dest->number[1], v->number[1], memory_order_relaxed); \ \ - if (!M_BUFF3R_POLICY_P((policy), M_BUFFER_THREAD_UNSAFE)) { \ - /* It may be false, but it is not wrong! */ \ - m_cond_broadcast(v->there_is_room_for_data); \ - m_cond_broadcast(v->there_is_data); \ - if (dest < v) { \ - m_mutex_unlock(v->mutexPop); \ - m_mutex_unlock(v->mutexPush); \ - m_mutex_unlock(dest->mutexPop); \ - m_mutex_unlock(dest->mutexPush); \ - } else { \ - m_mutex_unlock(dest->mutexPop); \ - m_mutex_unlock(dest->mutexPush); \ - m_mutex_unlock(v->mutexPop); \ - m_mutex_unlock(v->mutexPush); \ - } \ + m_cond_broadcast(v->there_is_room_for_data); \ + m_cond_broadcast(v->there_is_data); \ + if (dest < v) { \ + m_mutex_unlock(v->mutexPop); \ + m_mutex_unlock(v->mutexPush); \ + m_mutex_unlock(dest->mutexPop); \ + m_mutex_unlock(dest->mutexPush); \ + } else { \ + m_mutex_unlock(dest->mutexPop); \ + m_mutex_unlock(dest->mutexPush); \ + m_mutex_unlock(v->mutexPop); \ + m_mutex_unlock(v->mutexPush); \ } \ M_BUFF3R_CONTRACT(v,m_size); \ M_BUFF3R_CONTRACT(dest, m_size); \ @@ -523,16 +426,16 @@ M_P(void, name, _init, buffer_t v, size_t size) \ deferred pop has reached 0, not the number of items in the \ buffer is 0. */ \ if (M_BUFF3R_POLICY_P(policy, M_BUFFER_DEFERRED_POP)) \ - return m_buff3r_number_load (v->number[1], policy) == 0; \ + return atomic_load_explicit (&v->number[1], memory_order_relaxed) == 0; \ else \ - return m_buff3r_number_load (v->number[0], policy) == 0; \ + return atomic_load_explicit (&v->number[0], memory_order_relaxed) == 0; \ } \ \ M_INLINE bool \ M_F(name, _full_p)(const buffer_t v) \ { \ M_BUFF3R_CONTRACT(v,m_size); \ - return m_buff3r_number_load (v->number[0], policy) \ + return atomic_load_explicit (&v->number[0], memory_order_relaxed) \ == M_BUFF3R_SIZE(m_size); \ } \ \ @@ -540,27 +443,23 @@ M_P(void, name, _init, buffer_t v, size_t size) \ M_F(name, _size)(const buffer_t v) \ { \ M_BUFF3R_CONTRACT(v,m_size); \ - return m_buff3r_number_load (v->number[0], policy); \ + return atomic_load_explicit (&v->number[0], memory_order_relaxed); \ } \ \ - M_P(bool, name, _push_blocking, buffer_t v, type const data, bool blocking) \ + M_N(bool, name, _push_blocking, buffer_t v, type const data, bool blocking) \ { \ M_BUFF3R_CONTRACT(v,m_size); \ - M_ASSERT_POOL(); \ + M_GLOBAL_POOL(); \ /* Producer Mutex lock (mutex lock performs an acquire memory barrier) */ \ - if (!M_BUFF3R_POLICY_P((policy), M_BUFFER_THREAD_UNSAFE)) { \ - m_mutex_lock(v->mutexPush); \ - while (!M_BUFF3R_POLICY_P((policy), M_BUFFER_PUSH_OVERWRITE) \ - && M_F(name, _full_p)(v)) { \ - if (!blocking) { \ - m_mutex_unlock(v->mutexPush); \ - return false; \ - } \ - m_cond_wait(v->there_is_room_for_data, v->mutexPush); \ - } \ - } else if (!M_BUFF3R_POLICY_P((policy), M_BUFFER_PUSH_OVERWRITE) \ - && M_F(name, _full_p)(v)) \ - return false; \ + m_mutex_lock(v->mutexPush); \ + while (!M_BUFF3R_POLICY_P((policy), M_BUFFER_PUSH_OVERWRITE) \ + && M_F(name, _full_p)(v)) { \ + if (!blocking) { \ + m_mutex_unlock(v->mutexPush); \ + return false; \ + } \ + m_cond_wait(v->there_is_room_for_data, v->mutexPush); \ + } \ M_BUFF3R_PROTECTED_CONTRACT(policy, v, m_size); \ \ size_t previousSize, idx = v->idx_prod; \ @@ -598,47 +497,42 @@ M_P(void, name, _init, buffer_t v, size_t size) \ thread which has the push lock. As such, it is an atomic variable \ that performs a release memory barrier. */ \ /* Increment number of elements of the buffer */ \ - previousSize = m_buff3r_number_inc (v->number[0], policy); \ + previousSize = atomic_fetch_add (&v->number[0], 1U); \ if (M_BUFF3R_POLICY_P((policy), M_BUFFER_DEFERRED_POP)) { \ - previousSize = m_buff3r_number_inc (v->number[1], policy); \ + previousSize = atomic_fetch_add (&v->number[1], 1U); \ } \ /* From this point, consumer may read the data in the table */ \ } \ \ /* Producer unlock (mutex unlock performs a release memory barrier) */ \ - if (!M_BUFF3R_POLICY_P((policy), M_BUFFER_THREAD_UNSAFE)) { \ - m_mutex_unlock(v->mutexPush); \ - /* If the number of items in the buffer was 0, some consumer \ - may be waiting. Signal to them the availability of the data \ - We cannot only signal one thread. */ \ - if (previousSize == 0) { \ - m_mutex_lock(v->mutexPop); \ - m_cond_broadcast(v->there_is_data); \ - m_mutex_unlock(v->mutexPop); \ - } \ + m_mutex_unlock(v->mutexPush); \ + /* If the number of items in the buffer was 0, some consumer \ + may be waiting. Signal to them the availability of the data \ + We cannot only signal one thread. */ \ + if (previousSize == 0) { \ + m_mutex_lock(v->mutexPop); \ + m_cond_broadcast(v->there_is_data); \ + m_mutex_unlock(v->mutexPop); \ } \ \ M_BUFF3R_CONTRACT(v,m_size); \ return true; \ } \ \ - M_P(bool, name, _pop_blocking, type *data, buffer_t v, bool blocking) \ + M_N(bool, name, _pop_blocking, type *data, buffer_t v, bool blocking) \ { \ M_BUFF3R_CONTRACT(v,m_size); \ M_ASSERT (data != NULL); \ - M_ASSERT_POOL(); \ + M_GLOBAL_POOL(); \ /* consumer lock (mutex lock performs an acquire memory barrier) */ \ - if (!M_BUFF3R_POLICY_P((policy), M_BUFFER_THREAD_UNSAFE)) { \ - m_mutex_lock(v->mutexPop); \ - while (M_F(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_F(name, _empty_p)(v)) \ - return false; \ + m_mutex_lock(v->mutexPop); \ + while (M_F(name, _empty_p)(v)) { \ + if (!blocking) { \ + m_mutex_unlock(v->mutexPop); \ + return false; \ + } \ + m_cond_wait(v->there_is_data, v->mutexPop); \ + } \ M_BUFF3R_PROTECTED_CONTRACT(policy, v, m_size); \ \ /* POP data from the buffer and update INDEX */ \ @@ -667,24 +561,22 @@ M_P(void, name, _init, buffer_t v, size_t size) \ /* Decrement number of elements in the buffer */ \ size_t previousSize; \ if (!M_BUFF3R_POLICY_P((policy), M_BUFFER_DEFERRED_POP)) { \ - previousSize = m_buff3r_number_dec (v->number[0], policy); \ + previousSize = atomic_fetch_sub (&v->number[0], 1U); \ } else { \ - m_buff3r_number_dec (v->number[1], policy); \ + atomic_fetch_sub (&v->number[1], 1U); \ } \ /* Space may be reused by a producer thread from this point */ \ \ /* consumer unlock (mutex unlock performs a release memory barrier) */ \ - if (!M_BUFF3R_POLICY_P((policy), M_BUFFER_THREAD_UNSAFE)) { \ - m_mutex_unlock(v->mutexPop); \ - /* If the number of items in the buffer was the max, some producer \ - may be waiting. Signal to them the availability of the free room \ - We cannot only signal one thread. */ \ - if ((!M_BUFF3R_POLICY_P((policy), M_BUFFER_DEFERRED_POP)) \ - && previousSize == M_BUFF3R_SIZE(m_size)) { \ - m_mutex_lock(v->mutexPush); \ - m_cond_broadcast(v->there_is_room_for_data); \ - m_mutex_unlock(v->mutexPush); \ - } \ + m_mutex_unlock(v->mutexPop); \ + /* If the number of items in the buffer was the max, some producer \ + may be waiting. Signal to them the availability of the free room \ + We cannot only signal one thread. */ \ + if ((!M_BUFF3R_POLICY_P((policy), M_BUFFER_DEFERRED_POP)) \ + && previousSize == M_BUFF3R_SIZE(m_size)) { \ + m_mutex_lock(v->mutexPush); \ + m_cond_broadcast(v->there_is_room_for_data); \ + m_mutex_unlock(v->mutexPush); \ } \ \ M_BUFF3R_CONTRACT(v,m_size); \ @@ -692,16 +584,18 @@ M_P(void, name, _init, buffer_t v, size_t size) \ } \ \ \ - M_P(bool, name, _push, buffer_t v, type const data) \ + M_N(void, name, _push, buffer_t v, type const data) \ { \ - return M_F(name, _push_blocking)M_R(v, data, \ - !M_BUFF3R_POLICY_P((policy), M_BUFFER_UNBLOCKING_PUSH)); \ + bool b = M_F(name, _push_blocking)(v, data, true); \ + assert(b); \ + (void) b; \ } \ \ - M_P(bool, name, _pop, type *data, buffer_t v) \ + M_N(void, name, _pop, type *data, buffer_t v) \ { \ - return M_F(name, _pop_blocking)M_R(data, v, \ - !M_BUFF3R_POLICY_P((policy), M_BUFFER_UNBLOCKING_POP)); \ + bool b = M_F(name, _pop_blocking)(data, v, true); \ + assert(b); \ + (void) b; \ } \ \ M_INLINE size_t \ @@ -722,7 +616,7 @@ M_P(void, name, _init, buffer_t v, size_t size) \ { \ /* Decrement the effective number of elements in the buffer */ \ if (M_BUFF3R_POLICY_P((policy), M_BUFFER_DEFERRED_POP)) { \ - size_t previousSize = m_buff3r_number_dec (v->number[0], policy); \ + size_t previousSize = atomic_fetch_sub (&v->number[0], 1U); \ if (previousSize == M_BUFF3R_SIZE(m_size)) { \ m_mutex_lock(v->mutexPush); \ m_cond_broadcast(v->there_is_room_for_data); \ @@ -782,7 +676,7 @@ M_P(void, name, _init, buffer_t v, size_t size) \ M_QU3UE_MPMC_DEF_TYPE(name, type, policy, oplist, buffer_t) \ M_CHECK_COMPATIBLE_OPLIST(name, 1, type, oplist) \ M_QU3UE_MPMC_DEF_CORE(name, type, policy, oplist, buffer_t) \ - M_EMPLACE_QUEUE_DEF(name, buffer_t, _emplace, oplist, M_EMPLACE_QUEUE_GENE) + M_EMPLACE_QUEUE_DEF(name, buffer_t, _emplace, oplist, M_BUFF3R_EMPLACE_QUEUE_GENE) /* Define the type of a MPMC queue */ #define M_QU3UE_MPMC_DEF_TYPE(name, type, policy, oplist, buffer_t) \ @@ -817,10 +711,10 @@ M_P(void, name, _init, buffer_t v, size_t size) \ /* Define the core functionalities of a MPMC queue */ #define M_QU3UE_MPMC_DEF_CORE(name, type, policy, oplist, buffer_t) \ - M_P(bool, name, _push, buffer_t table, type const x) \ + M_N(bool, name, _push, buffer_t table, type const x) \ { \ M_QU3UE_MPMC_CONTRACT(table); \ - M_ASSERT_POOL(); \ + M_GLOBAL_POOL(); \ unsigned int idx = atomic_load_explicit(&table->ProdIdx, \ memory_order_relaxed); \ const unsigned int i = idx & (table->size -1); \ @@ -851,11 +745,11 @@ M_P(void, name, _init, buffer_t v, size_t size) \ return true; \ } \ \ - M_P(bool, name, _pop, type *ptr, buffer_t table) \ + M_N(bool, name, _pop, type *ptr, buffer_t table) \ { \ M_QU3UE_MPMC_CONTRACT(table); \ M_ASSERT (ptr != NULL); \ - M_ASSERT_POOL(); \ + M_GLOBAL_POOL(); \ unsigned int iC = atomic_load_explicit(&table->ConsoIdx, \ memory_order_relaxed); \ const unsigned int i = (iC & (table->size -1)); \ @@ -880,12 +774,12 @@ M_P(void, name, _init, buffer_t v, size_t size) \ return true; \ } \ \ - M_P(void, name, _init, buffer_t buffer, size_t size) \ + M_N(void, name, _init, buffer_t buffer, size_t size) \ { \ M_ASSERT (buffer != NULL); \ - M_ASSERT( M_POWEROF2_P(size)); \ - M_ASSERT (0 < size && size <= UINT_MAX); \ - M_ASSERT(((policy) & (M_BUFFER_STACK|M_BUFFER_THREAD_UNSAFE|M_BUFFER_PUSH_OVERWRITE)) == 0); \ + M_ASSERT( M_POWEROF2_P(size) && 0 < size && size <= UINT_MAX); \ + M_ASSERT(((policy) & (M_BUFFER_STACK|M_BUFFER_PUSH_OVERWRITE)) == 0); \ + M_GLOBAL_POOL(); \ atomic_init(&buffer->ProdIdx, (unsigned int) size); \ atomic_init(&buffer->ConsoIdx, (unsigned int) size); \ buffer->size = (unsigned int) size; \ @@ -902,9 +796,10 @@ M_P(void, name, _init, buffer_t v, size_t size) \ M_QU3UE_MPMC_CONTRACT(buffer); \ } \ \ - M_P(void, name, _clear, buffer_t buffer) \ + M_N(void, name, _clear, buffer_t buffer) \ { \ M_QU3UE_MPMC_CONTRACT(buffer); \ + M_GLOBAL_POOL(); \ if (!M_BUFF3R_POLICY_P((policy), M_BUFFER_PUSH_INIT_POP_MOVE)) { \ for(unsigned int j = 0; j < buffer->size; j++) { \ M_CALL_CLEAR(oplist, buffer->Tab[j].x); \ @@ -1014,7 +909,7 @@ M_P(void, name, _init, buffer_t v, size_t size) \ M_QU3UE_SPSC_DEF_TYPE(name, type, policy, oplist, buffer_t) \ M_CHECK_COMPATIBLE_OPLIST(name, 1, type, oplist) \ M_QU3UE_SPSC_DEF_CORE(name, type, policy, oplist, buffer_t) \ - M_EMPLACE_QUEUE_DEF(name, buffer_t, _emplace, oplist, M_EMPLACE_QUEUE_GENE) + M_EMPLACE_QUEUE_DEF(name, buffer_t, _emplace, oplist, M_BUFF3R_EMPLACE_QUEUE_GENE) /* Define the type of a SPSC queue */ #define M_QU3UE_SPSC_DEF_TYPE(name, type, policy, oplist, buffer_t) \ @@ -1031,7 +926,7 @@ M_P(void, name, _init, buffer_t v, size_t size) \ atomic_uint consoIdx; /* Can only increase until overflow */ \ unsigned int size; \ M_F(name, _el_ct) *Tab; \ - M_CACHELINE_ALIGN(align, atomic_uint, size_t, M_F(name, _el_ct) *); \ + M_CACHELINE_ALIGN(align, atomic_uint, unsigned int, M_F(name, _el_ct) *); \ atomic_uint prodIdx; /* Can only increase until overflow */ \ } buffer_t[1]; \ \ @@ -1041,10 +936,10 @@ M_P(void, name, _init, buffer_t v, size_t size) \ /* Define the core functionalities of a SPSC queue */ #define M_QU3UE_SPSC_DEF_CORE(name, type, policy, oplist, buffer_t) \ \ - M_P(bool, name, _push, buffer_t table, type const x) \ + M_N(bool, name, _push, buffer_t table, type const x) \ { \ M_QU3UE_SPSC_CONTRACT(table); \ - M_ASSERT_POOL(); \ + M_GLOBAL_POOL(); \ unsigned int r = atomic_load_explicit(&table->consoIdx, \ memory_order_relaxed); \ unsigned int w = atomic_load_explicit(&table->prodIdx, \ @@ -1062,10 +957,10 @@ M_P(void, name, _init, buffer_t v, size_t size) \ return true; \ } \ \ - M_P(bool, name, _push_move, buffer_t table, type *x) \ + M_N(bool, name, _push_move, buffer_t table, type *x) \ { \ M_QU3UE_SPSC_CONTRACT(table); \ - M_ASSERT_POOL(); \ + M_GLOBAL_POOL(); \ unsigned int r = atomic_load_explicit(&table->consoIdx, \ memory_order_relaxed); \ unsigned int w = atomic_load_explicit(&table->prodIdx, \ @@ -1083,11 +978,11 @@ M_P(void, name, _init, buffer_t v, size_t size) \ return true; \ } \ \ - M_P(bool, name, _pop, type *ptr, buffer_t table) \ + M_N(bool, name, _pop, type *ptr, buffer_t table) \ { \ M_QU3UE_SPSC_CONTRACT(table); \ M_ASSERT (ptr != NULL); \ - M_ASSERT_POOL(); \ + M_GLOBAL_POOL(); \ unsigned int w = atomic_load_explicit(&table->prodIdx, \ memory_order_relaxed); \ unsigned int r = atomic_load_explicit(&table->consoIdx, \ @@ -1105,12 +1000,12 @@ M_P(void, name, _init, buffer_t v, size_t size) \ return true; \ } \ \ - M_P(unsigned, name, _push_bulk, buffer_t table, unsigned n, type const x[]) \ + M_N(unsigned, name, _push_bulk, buffer_t table, unsigned n, type const x[]) \ { \ M_QU3UE_SPSC_CONTRACT(table); \ M_ASSERT (x != NULL); \ M_ASSERT (n <= table->size); \ - M_ASSERT_POOL(); \ + M_GLOBAL_POOL(); \ unsigned int r = atomic_load_explicit(&table->consoIdx, \ memory_order_relaxed); \ unsigned int w = atomic_load_explicit(&table->prodIdx, \ @@ -1131,12 +1026,12 @@ M_P(void, name, _init, buffer_t v, size_t size) \ return max; \ } \ \ - M_P(unsigned, name, _pop_bulk, unsigned n, type ptr[], buffer_t table) \ + M_N(unsigned, name, _pop_bulk, unsigned n, type ptr[], buffer_t table) \ { \ M_QU3UE_SPSC_CONTRACT(table); \ M_ASSERT (ptr != NULL); \ M_ASSERT (n <= table->size); \ - M_ASSERT_POOL(); \ + M_GLOBAL_POOL(); \ unsigned int w = atomic_load_explicit(&table->prodIdx, \ memory_order_relaxed); \ unsigned int r = atomic_load_explicit(&table->consoIdx, \ @@ -1157,10 +1052,10 @@ M_P(void, name, _init, buffer_t v, size_t size) \ return max; \ } \ \ - M_P(void, name, _push_force, buffer_t table, type const x) \ + M_N(void, name, _push_force, buffer_t table, type const x) \ { \ M_QU3UE_SPSC_CONTRACT(table); \ - M_ASSERT_POOL(); \ + M_GLOBAL_POOL(); \ unsigned int r = atomic_load_explicit(&table->consoIdx, \ memory_order_relaxed); \ unsigned int w = atomic_load_explicit(&table->prodIdx, \ @@ -1219,12 +1114,12 @@ M_P(void, name, _init, buffer_t v, size_t size) \ return M_F(name, _size)(v) >= v->size; \ } \ \ - M_P(void, name, _init, buffer_t buffer, size_t size) \ + M_N(void, name, _init, buffer_t buffer, size_t size) \ { \ M_ASSERT (buffer != NULL); \ - M_ASSERT( M_POWEROF2_P(size)); \ - M_ASSERT (0 < size && size <= UINT_MAX); \ - M_ASSERT(((policy) & (M_BUFFER_STACK|M_BUFFER_THREAD_UNSAFE|M_BUFFER_PUSH_OVERWRITE)) == 0); \ + M_ASSERT( M_POWEROF2_P(size) && 0 < size && size <= UINT_MAX); \ + M_ASSERT(((policy) & (M_BUFFER_STACK|M_BUFFER_PUSH_OVERWRITE)) == 0); \ + M_GLOBAL_POOL(); \ atomic_init(&buffer->prodIdx, (unsigned int) size); \ atomic_init(&buffer->consoIdx, (unsigned int) size); \ buffer->size = (unsigned int) size; \ @@ -1240,9 +1135,10 @@ M_P(void, name, _init, buffer_t v, size_t size) \ M_QU3UE_SPSC_CONTRACT(buffer); \ } \ \ - M_P(void, name, _clear, buffer_t buffer) \ + M_N(void, name, _clear, buffer_t buffer) \ { \ M_QU3UE_SPSC_CONTRACT(buffer); \ + M_GLOBAL_POOL(); \ if (!M_BUFF3R_POLICY_P((policy), M_BUFFER_PUSH_INIT_POP_MOVE)) { \ for(unsigned int j = 0; j < buffer->size; j++) { \ M_CALL_CLEAR(oplist, buffer->Tab[j].x); \ @@ -1297,8 +1193,18 @@ M_P(void, name, _init, buffer_t v, size_t size) \ ,GET_SIZE(M_F(name, _size)) \ ) - -/********************************** INTERNAL *********************************/ +/* Same as M_EMPLACE_QUEUE_GENE but use M_GLOBAL_POOL */ +#define M_BUFF3R_EMPLACE_QUEUE_GENE(name, name_t, function_name, oplist, init_func, exp_emplace_type) \ +M_N(void, name, function_name, name_t v M_EMPLACE_LIST_TYPE_VAR(a, exp_emplace_type) ) \ +{ \ + M_GET_TYPE oplist data; \ + M_GLOBAL_POOL(); \ + M_EMPLACE_CALL_FUNC(a, init_func, oplist, data, exp_emplace_type); \ + M_F(name, _push)(v, data); \ + M_CALL_CLEAR(oplist, data); \ +} + + /********************************** INTERNAL *********************************/ #if M_USE_SMALL_NAME #define BUFFER_DEF M_BUFFER_DEF @@ -1312,14 +1218,6 @@ M_P(void, name, _init, buffer_t v, size_t size) \ #define buffer_policy_e m_buffer_policy_e #define BUFFER_QUEUE M_BUFFER_QUEUE #define BUFFER_STACK M_BUFFER_STACK -#define BUFFER_BLOCKING_PUSH M_BUFFER_BLOCKING_PUSH -#define BUFFER_UNBLOCKING_PUSH M_BUFFER_UNBLOCKING_PUSH -#define BUFFER_BLOCKING_POP M_BUFFER_BLOCKING_POP -#define BUFFER_UNBLOCKING_POP M_BUFFER_UNBLOCKING_POP -#define BUFFER_BLOCKING M_BUFFER_BLOCKING -#define BUFFER_UNBLOCKING M_BUFFER_UNBLOCKING -#define BUFFER_THREAD_SAFE M_BUFFER_THREAD_SAFE -#define BUFFER_THREAD_UNSAFE M_BUFFER_THREAD_UNSAFE #define BUFFER_PUSH_INIT_POP_MOVE M_BUFFER_PUSH_INIT_POP_MOVE #define BUFFER_PUSH_OVERWRITE M_BUFFER_PUSH_OVERWRITE #define BUFFER_DEFERRED_POP M_BUFFER_DEFERRED_POP diff --git a/m-core.h b/m-core.h index 95bc138a..51eba8bf 100644 --- a/m-core.h +++ b/m-core.h @@ -461,11 +461,8 @@ M_BEGIN_PROTECTED_CODE # define M_N(rettype, name, suffix, ...) M_INLINE rettype M_F(name, suffix)(__VA_ARGS__) #endif -// TODO: Add for all Assertions: M_ASSERT_POOL() -// to avoid warnings about unused parameter m_pool. -// Pb: Also included for functions without the pool parameter! - -// FIXME: Not sense to do it for a shared pointer: it is shared across all memory contexts! +// TODO: 'pool' seems not the right term. Rename 'pool' into 'localmem' ? 'arena'? memcontext? context? +// TODO: M_USE_GLOBAL_POOL , value for setting the global pool. /************************************************************/ /********************* ERROR handling **********************/ @@ -5162,7 +5159,6 @@ m_core_parse2_enum (const char str[], const char **endptr) { \ M_GET_TYPE oplist data; \ M_EMPLACE_CALL_FUNC(a, init_func, oplist, data, exp_emplace_type); \ - /* FIXME: Why name##_push and not M_CALL_PUSH? */ \ M_F(name, _push) M_R(v, data); \ M_CALL_CLEAR(oplist, data); \ } diff --git a/m-worker.h b/m-worker.h index c70a0d1f..50ca1bc3 100644 --- a/m-worker.h +++ b/m-worker.h @@ -142,8 +142,7 @@ typedef struct m_work3r_thread_s { } m_work3r_thread_ct; /* Definition of the queue that will record the work orders */ -M_BUFFER_DEF(m_work3r_queue, m_work3r_order_ct, 0, - M_BUFFER_QUEUE|M_BUFFER_UNBLOCKING_PUSH|M_BUFFER_BLOCKING_POP|M_BUFFER_THREAD_SAFE|M_BUFFER_DEFERRED_POP, M_WORK3R_OPLIST) +M_BUFFER_DEF(m_work3r_queue, m_work3r_order_ct, 0, M_BUFFER_QUEUE|M_BUFFER_DEFERRED_POP, M_WORK3R_OPLIST) /* Definition the global pool of workers */ typedef struct m_worker_s { @@ -273,7 +272,7 @@ typedef struct m_worker_sync_s { p->callback = callback; \ M_MAP3(M_WORK3R_SPAWN_EXTEND_DEF_EMPLACE_FIELD_INIT_SET, data, __VA_ARGS__) \ const m_work3r_order_ct w = { block, p, M_C3(m_work3r_, name, _callback) M_WORK3R_EXTRA_ORDER }; \ - if (m_work3r_queue_push M_R(block->worker->queue_g, w) == true) { \ + if (m_work3r_queue_push_blocking (block->worker->queue_g, w, false) == true) { \ atomic_fetch_add (&block->num_spawn, 1); \ return; \ } \ @@ -372,7 +371,7 @@ m_work3r_thread(void *arg) } // Waiting for data M_WORK3R_DEBUG ("Waiting for data (queue: %lu / %lu)\n", m_work3r_queue_size(g->queue_g), m_work3r_queue_capacity(g->queue_g)); - m_work3r_queue_pop M_R(&w, g->queue_g); + m_work3r_queue_pop (&w, g->queue_g); // We received a work order // Note: that the work order is still present in the queue // preventing further work order to be pushed in the queue until it finishes doing the work @@ -417,7 +416,7 @@ m_worker_init(m_worker_t g, int numWorker, unsigned int extraQueue, void (*reset if (M_UNLIKELY_NOMEM (g->worker == NULL)) { M_MEMORY_FULL(m_work3r_thread_ct, numWorker_st); } - m_work3r_queue_init M_R(g->queue_g, numWorker_st + extraQueue); + m_work3r_queue_init (g->queue_g, numWorker_st + extraQueue); g->numWorker_g = (unsigned int) numWorker_st; g->resetFunc_g = resetFunc; g->clearFunc_g = clearFunc; @@ -451,7 +450,7 @@ m_worker_clear(m_worker_t g) // Normally all worker threads shall be waiting at this // stage, so all push won't block as the queue is empty. // But for robustness, let's wait. - m_work3r_queue_push_blocking M_R(g->queue_g, w, true); + m_work3r_queue_push_blocking (g->queue_g, w, true); } // Wait for thread termination for(unsigned int i = 0; i < g->numWorker_g; i++) { @@ -461,7 +460,7 @@ m_worker_clear(m_worker_t g) M_MEMORY_FREE(m_work3r_thread_ct, g->worker, g->numWorker_g); m_mutex_clear(g->lock); m_cond_clear(g->a_thread_ends); - m_work3r_queue_clear M_R(g->queue_g); + m_work3r_queue_clear (g->queue_g); } /* Start a new collaboration between workers of pool 'g' @@ -485,7 +484,7 @@ m_worker_spawn(m_worker_sync_t block, void (*func)(void *data), void *data) const m_work3r_order_ct w = { block, data, func M_WORK3R_EXTRA_ORDER }; M_GLOBAL_POOL(); if (M_UNLIKELY (!m_work3r_queue_full_p(block->worker->queue_g)) - && m_work3r_queue_push M_R(block->worker->queue_g, w) == true) { + && m_work3r_queue_push_blocking (block->worker->queue_g, w, false) == true) { M_WORK3R_DEBUG ("Sending data to thread: %p (block: %d / %d)\n", data, block->num_spawn, block->num_terminated_spawn); atomic_fetch_add (&block->num_spawn, 1); return; @@ -504,7 +503,7 @@ m_work3r_spawn_block(m_worker_sync_t block, void (^func)(void *data), void *data const m_work3r_order_ct w = { block, data, NULL, func }; M_GLOBAL_POOL(); if (M_UNLIKELY (!m_work3r_queue_full_p(block->worker->queue_g)) - && m_work3r_queue_push M_R(block->worker->queue_g, w) == true) { + && m_work3r_queue_push_blocking (block->worker->queue_g, w, false) == true) { M_WORK3R_DEBUG ("Sending data to thread as block: %p (block: %d / %d)\n", data, block->num_spawn, block->num_terminated_spawn); atomic_fetch_add (&block->num_spawn, 1); return; @@ -524,7 +523,7 @@ m_work3r_spawn_function(m_worker_sync_t block, std::function f const m_work3r_order_ct w = { block, data, NULL, func }; M_GLOBAL_POOL(); if (M_UNLIKELY (!m_work3r_queue_full_p(block->worker->queue_g)) - && m_work3r_queue_push M_R(block->worker->queue_g, w) == true) { + && m_work3r_queue_push_blocking (block->worker->queue_g, w, false) == true) { M_WORK3R_DEBUG ("Sending data to thread as block: %p (block: %d / %d)\n", data, block->num_spawn, block->num_terminated_spawn); atomic_fetch_add (&block->num_spawn, 1); return; @@ -566,7 +565,7 @@ m_worker_flush(m_worker_t g) { m_work3r_order_ct w; M_GLOBAL_POOL(); - while (m_work3r_queue_pop_blocking M_R(&w, g->queue_g, false) == true) { + while (m_work3r_queue_pop_blocking (&w, g->queue_g, false) == true) { m_work3r_exec(&w); m_work3r_queue_pop_release(g->queue_g); } diff --git a/tests/Makefile b/tests/Makefile index 532fd3bc..c8eff7ec 100644 --- a/tests/Makefile +++ b/tests/Makefile @@ -198,6 +198,7 @@ SYNTHESIS_DATA= M-ALGO test-malgo.c.c test-malgo.synt \ M-GENINT ../m-genint.h test-mgenint.synt \ M-I-LIST test-milist.c.c test-milist.synt \ M-I-SHARED test-mishared.c.c test-mishared.synt \ + M-QUEUE test-mqueue.c.c test-mqueue.synt \ M-LIST test-mlist.c.c test-mlist.synt \ M-MEMPOOL test-mmempool.c.c test-mmempool.synt \ M-PRIOQUEUE test-mprioqueue.c.c test-mprioqueue.synt \ diff --git a/tests/coverage.h b/tests/coverage.h index 076fa28e..a3cb5a30 100644 --- a/tests/coverage.h +++ b/tests/coverage.h @@ -125,6 +125,9 @@ #define M_TR33_CONTRACT(...) (void)0 #undef M_TR33_IT_CONTRACT #define M_TR33_IT_CONTRACT(...) (void)0 +#undef M_QUEU3_CONTRACT +#define M_QUEU3_CONTRACT(...) (void)0 + /* Overwrite M_UNLIKELY_NOMEM so that it can never fails, to avoid spurious coverage */ #undef M_UNLIKELY_NOMEM #define M_UNLIKELY_NOMEM(...) 0 diff --git a/tests/test-mbuffer.c b/tests/test-mbuffer.c index 6fc2032d..435dd771 100644 --- a/tests/test-mbuffer.c +++ b/tests/test-mbuffer.c @@ -26,19 +26,16 @@ START_COVERAGE // Define a fixed queue of unsigned int -BUFFER_DEF(buffer_uint, unsigned int, 10, BUFFER_QUEUE|BUFFER_BLOCKING) +BUFFER_DEF(buffer_uint, unsigned int, 10, BUFFER_QUEUE) QUEUE_MPMC_DEF(queue_uint, unsigned int, BUFFER_QUEUE) QUEUE_SPSC_DEF(squeue_uint, unsigned int, BUFFER_QUEUE) END_COVERAGE // Define a variable stack of float -BUFFER_DEF(buffer_floats, float, 0, BUFFER_STACK|BUFFER_BLOCKING) +BUFFER_DEF(buffer_floats, float, 0, BUFFER_STACK) // Define a fixed stack of char -BUFFER_DEF(buffer_char, char, 10, BUFFER_STACK|BUFFER_UNBLOCKING) - -// Define a fixed queue of long long -BUFFER_DEF(buffer_llong, long long, 16, BUFFER_QUEUE|BUFFER_THREAD_UNSAFE|BUFFER_UNBLOCKING) +BUFFER_DEF(buffer_char, char, 10, BUFFER_STACK) // Define a buffer of complex structure. BUFFER_DEF(buffer_mpz, testobj_t, 32, BUFFER_QUEUE, TESTOBJ_OPLIST) @@ -170,8 +167,7 @@ static void test_global(void) assert (buffer_uint_size(g_buffB) == 5); for(unsigned int i = 0; i < 5;i++) { unsigned int j; - bool b= buffer_uint_pop(&j, g_buffB); - assert(b); + buffer_uint_pop(&j, g_buffB); assert(j == i); } assert (buffer_uint_empty_p(g_buffB)); @@ -182,8 +178,7 @@ static void test_global(void) assert (buffer_uint_size(g_buffB) == 5); for(unsigned int i = 0; i < 5;i++) { unsigned int j; - bool b= buffer_uint_pop(&j, g_buffB); - assert(b); + buffer_uint_pop(&j, g_buffB); assert(j == i); } assert (buffer_uint_empty_p(g_buffB)); @@ -231,56 +226,25 @@ static void test_stack2(void) buffer_char_init(buff, 10); for(int i = 0; i < 10; i++) { c = (char) i; - b = buffer_char_push(buff, c); + b = buffer_char_push_blocking(buff, c, false); assert (b == true); assert (buffer_char_empty_p(buff) == false); } - b = buffer_char_push(buff, c); + b = buffer_char_push_blocking(buff, c, false); assert (b == false); assert (buffer_char_full_p(buff) == true); for(int i = 0; i < 10; i++) { - buffer_char_pop(&c, buff); + b = buffer_char_pop_blocking(&c, buff, false); + assert(b); assert(c == 9-i); assert (buffer_char_full_p(buff) == false); } - b = buffer_char_pop(&c, buff); + b = buffer_char_pop_blocking(&c, buff, false); assert (b == false); assert (buffer_char_empty_p(buff) == true); buffer_char_clear(buff); } -static void test_no_thread(void) -{ - buffer_llong_t buff; - bool b; - long long c; - buffer_llong_init(buff, 16); - for(int i = 0; i < 16; i++) { - c = i; - b = buffer_llong_push(buff, c); - assert (b == true); - assert (buffer_llong_empty_p(buff) == false); - } - b = buffer_llong_push(buff, c); - assert (b == false); - assert (buffer_llong_full_p(buff) == true); - for(int i = 0; i < 16; i++) { - buffer_llong_pop(&c, buff); - assert(c == i); - assert (buffer_llong_full_p(buff) == false); - } - b = buffer_llong_pop(&c, buff); - assert (b == false); - assert (buffer_llong_empty_p(buff) == true); - - b = buffer_llong_push(buff, c); - assert (b == true); - buffer_llong_reset(buff); - assert (buffer_llong_empty_p(buff) == true); - - buffer_llong_clear(buff); -} - static void test_emplace(void) { buffer_mpz_t b; @@ -582,56 +546,41 @@ static void test_spsc(void) static void test_double1(void) { - bool b; M_LET(buffer, BufferDouble1) { - b = BufferDouble1_push(buffer, 0.0); - assert(b); - b= BufferDouble1_push(buffer, 1.0); - assert(b); + BufferDouble1_push(buffer, 0.0); + BufferDouble1_push(buffer, 1.0); double d; - b = BufferDouble1_pop(&d, buffer); - assert(b); + BufferDouble1_pop(&d, buffer); assert(d == 0.0); - b = BufferDouble1_pop(&d, buffer); - assert(b); + BufferDouble1_pop(&d, buffer); assert(d == 1.0); } } static void test_double2(void) { - bool b; BufferDouble2 buffer; BufferDouble2_init(buffer, 2); - b = BufferDouble2_push(buffer, 0.0); - assert(b); - b= BufferDouble2_push(buffer, 1.0); - assert(b); + BufferDouble2_push(buffer, 0.0); + BufferDouble2_push(buffer, 1.0); double d; - b = BufferDouble2_pop(&d, buffer); - assert(b); + BufferDouble2_pop(&d, buffer); assert(d == 0.0); - b = BufferDouble2_pop(&d, buffer); - assert(b); + BufferDouble2_pop(&d, buffer); assert(d == 1.0); BufferDouble2_clear(buffer); } static void test_double3(void) { - bool b; BufferDouble3 buffer; BufferDouble3_init(buffer, 2); - b = BufferDouble3_push(buffer, 0.0); - assert(b); - b= BufferDouble3_push(buffer, 1.0); - assert(b); + BufferDouble3_push(buffer, 0.0); + BufferDouble3_push(buffer, 1.0); double d; - b = BufferDouble3_pop(&d, buffer); - assert(b); + BufferDouble3_pop(&d, buffer); assert(d == 0.0); - b = BufferDouble3_pop(&d, buffer); - assert(b); + BufferDouble3_pop(&d, buffer); assert(d == 1.0); BufferDouble3_clear(buffer); } @@ -668,7 +617,6 @@ int main(void) test_global(); test_stack(); test_stack2(); - test_no_thread(); test_emplace(); test_global_ishared(); test_queue(1000000, 2, 2148371710223136ULL);