diff --git a/m-buffer.h b/m-buffer.h index 670899fd..802cf23a 100644 --- a/m-buffer.h +++ b/m-buffer.h @@ -81,10 +81,10 @@ typedef enum { assert (buffer->data != NULL); \ }while (0) -#define BUFFERI_PROTECTED_CONTRACT(buffer, size) do { \ - assert (buffer->number[0] <= BUFFERI_SIZE(size)); \ - assert (buffer->idx_prod <= BUFFERI_SIZE(size)); \ - assert (buffer->idx_cons <= BUFFERI_SIZE(size)); \ +#define BUFFERI_PROTECTED_CONTRACT(buffer, size) do { \ + assert (atomic_load(&buffer->number[0]) <= BUFFERI_SIZE(size)); \ + assert (buffer->idx_prod <= BUFFERI_SIZE(size)); \ + assert (buffer->idx_cons <= BUFFERI_SIZE(size)); \ } while (0) #define BUFFERI_DEF(arg) BUFFERI_DEF2 arg @@ -97,7 +97,7 @@ typedef enum { m_cond_t there_is_room_for_data; \ BUFFERI_IF_CTE_SIZE(m_size)( ,size_t size;) \ size_t idx_prod, idx_cons, overwrite; \ - size_t number[1 + BUFFERI_POLICY_P(policy, BUFFER_DEFERRED_POP)]; \ + atomic_ulong number[1 + BUFFERI_POLICY_P(policy, BUFFER_DEFERRED_POP)]; \ type *data; \ } buffer_t[1]; \ \ @@ -106,9 +106,10 @@ M_C(name, _init)(buffer_t v, size_t size) \ { \ assert( size > 0); \ BUFFERI_IF_CTE_SIZE(m_size)(assert(size == m_size), v->size = size); \ - v->idx_prod = v->idx_cons = v->number[0] = v->overwrite = 0; \ + v->idx_prod = v->idx_cons = v->overwrite = 0; \ + atomic_init (&v->number[0], 0); \ if (BUFFERI_POLICY_P(policy, BUFFER_DEFERRED_POP)) \ - v->number[1] = 0; \ + atomic_init (&v->number[1], 0); \ if (!BUFFERI_POLICY_P((policy), BUFFER_THREAD_UNSAFE)) { \ m_mutex_init(v->mutex); \ m_cond_init(v->there_is_data); \ @@ -145,9 +146,10 @@ M_C(name, _init)(buffer_t v, size_t size) \ i = 0; \ } \ } \ - v->idx_prod = v->idx_cons = v->number[0] = 0; \ + v->idx_prod = v->idx_cons = 0; \ + atomic_store (&v->number[0], 0); \ if (BUFFERI_POLICY_P(policy, BUFFER_DEFERRED_POP)) \ - v->number[1] = 0; \ + atomic_store(&v->number[1], 0); \ BUFFERI_CONTRACT(v,m_size); \ } \ \ @@ -173,12 +175,12 @@ M_C(name, _init)(buffer_t v, size_t size) \ if (!BUFFERI_POLICY_P((policy), BUFFER_THREAD_UNSAFE)) \ m_mutex_lock(v->mutex); \ BUFFERI_PROTECTED_CONTRACT(v, m_size); \ - if (BUFFERI_POLICY_P((policy), BUFFER_PUSH_INIT_POP_MOVE)) \ + if (BUFFERI_POLICY_P((policy), BUFFER_PUSH_INIT_POP_MOVE)) \ M_C(name, _int_clear_obj)(v); \ - else \ - v->idx_prod = v->idx_cons = v->number[0] = 0; \ + v->idx_prod = v->idx_cons = 0; \ + atomic_store (&v->number[0], 0); \ if (BUFFERI_POLICY_P(policy, BUFFER_DEFERRED_POP)) \ - v->number[1] = 0; \ + atomic_store(&v->number[1], 0); \ if (!BUFFERI_POLICY_P((policy), BUFFER_THREAD_UNSAFE)) { \ m_cond_signal(v->there_is_room_for_data); \ m_mutex_unlock(v->mutex); \ @@ -191,23 +193,23 @@ M_C(name, _init)(buffer_t v, size_t size) \ { \ BUFFERI_CONTRACT(v,m_size); \ if (BUFFERI_POLICY_P(policy, BUFFER_DEFERRED_POP)) \ - return v->number[1] == 0; \ + return atomic_load (&v->number[1]) == 0; \ else \ - return v->number[0] == 0; \ + return atomic_load (&v->number[0]) == 0; \ } \ \ static inline bool \ M_C(name, _full_p)(const buffer_t v) \ { \ BUFFERI_CONTRACT(v,m_size); \ - return v->number[0] == BUFFERI_SIZE(m_size); \ + return atomic_load (&v->number[0]) == BUFFERI_SIZE(m_size); \ } \ \ static inline size_t \ M_C(name, _size)(const buffer_t v) \ { \ BUFFERI_CONTRACT(v,m_size); \ - return v->number[0]; \ + return atomic_load (&v->number[0]); \ } \ \ static inline bool \ @@ -233,7 +235,7 @@ M_C(name, _init)(buffer_t v, size_t size) \ \ /* INDEX computation if we have to overwrite the last element */ \ if (M_UNLIKELY (BUFFERI_POLICY_P((policy), BUFFER_PUSH_OVERWRITE) \ - && v->number[0] == BUFFERI_SIZE(m_size) )) { \ + && atomic_load (&v->number[0]) == BUFFERI_SIZE(m_size) )) { \ v->overwrite++; \ /* Let's clear the last element to overwrite it */ \ if (!BUFFERI_POLICY_P((policy), BUFFER_STACK)) { \ @@ -241,9 +243,9 @@ M_C(name, _init)(buffer_t v, size_t size) \ } else { \ (v->idx_prod) --; \ } \ - v->number[0] --; \ + atomic_store (&v->number[0], atomic_load(&v->number[0]) - 1); \ if (BUFFERI_POLICY_P((policy), BUFFER_DEFERRED_POP)) \ - v->number[1] --; \ + atomic_store (&v->number[1], atomic_load(&v->number[1]) - 1); \ if (BUFFERI_POLICY_P((policy), BUFFER_PUSH_INIT_POP_MOVE)) { \ M_GET_CLEAR oplist(v->data[v->idx_prod]); \ } \ @@ -260,9 +262,9 @@ M_C(name, _init)(buffer_t v, size_t size) \ } else { \ (v->idx_prod) ++; \ } \ - v->number[0] ++; \ + atomic_store (&v->number[0], atomic_load(&v->number[0]) + 1); \ if (BUFFERI_POLICY_P((policy), BUFFER_DEFERRED_POP)) \ - v->number[1] ++; \ + atomic_store (&v->number[1], atomic_load(&v->number[1]) + 1); \ \ /* BUFFER unlock */ \ if (!BUFFERI_POLICY_P((policy), BUFFER_THREAD_UNSAFE)) { \ @@ -311,9 +313,9 @@ M_C(name, _init)(buffer_t v, size_t size) \ } \ } \ if (!BUFFERI_POLICY_P((policy), BUFFER_DEFERRED_POP)) \ - v->number[0] --; \ + atomic_store (&v->number[0], atomic_load(&v->number[0]) - 1); \ else \ - v->number[1] --; \ + atomic_store (&v->number[1], atomic_load(&v->number[1]) - 1); \ /* BUFFER unlock */ \ if (!BUFFERI_POLICY_P((policy), BUFFER_THREAD_UNSAFE)) { \ m_cond_signal(v->there_is_room_for_data); \ @@ -341,7 +343,7 @@ M_C(name, _init)(buffer_t v, size_t size) \ M_C(name, _pop_release)(buffer_t v) \ { \ if (BUFFERI_POLICY_P((policy), BUFFER_DEFERRED_POP)) \ - v->number[0] --; \ + atomic_store (&v->number[0], atomic_load(&v->number[0]) - 1); \ } \