admin/mlib
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Fix data race

Browse Source
This commit is contained in:
Patrick Pelissier committed 2017-12-25 23:35:04 +01:00
1 parent b4cf464e98
commit 31820e7a5d
1 file changed
+27 -25
Download Patch File Download Diff File Expand all files Collapse all files
m-buffer.h
+27 -25
View File
@@ -81,10 +81,10 @@ typedef enum {
assert (buffer->data != NULL); \ assert (buffer->data != NULL); \
}while (0) }while (0)
#define BUFFERI_PROTECTED_CONTRACT(buffer, size) do { \ #define BUFFERI_PROTECTED_CONTRACT(buffer, size) do { \
assert (buffer->number[0] <= BUFFERI_SIZE(size)); \ assert (atomic_load(&buffer->number[0]) <= BUFFERI_SIZE(size)); \
assert (buffer->idx_prod <= BUFFERI_SIZE(size)); \ assert (buffer->idx_prod <= BUFFERI_SIZE(size)); \
assert (buffer->idx_cons <= BUFFERI_SIZE(size)); \ assert (buffer->idx_cons <= BUFFERI_SIZE(size)); \
} while (0) } while (0)
#define BUFFERI_DEF(arg) BUFFERI_DEF2 arg #define BUFFERI_DEF(arg) BUFFERI_DEF2 arg
@@ -97,7 +97,7 @@ typedef enum {
m_cond_t there_is_room_for_data; \ m_cond_t there_is_room_for_data; \
BUFFERI_IF_CTE_SIZE(m_size)( ,size_t size;) \ BUFFERI_IF_CTE_SIZE(m_size)( ,size_t size;) \
size_t idx_prod, idx_cons, overwrite; \ 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; \ type *data; \
} buffer_t[1]; \ } buffer_t[1]; \
\ \
@@ -106,9 +106,10 @@ M_C(name, _init)(buffer_t v, size_t size) \
{ \ { \
assert( size > 0); \ assert( size > 0); \
BUFFERI_IF_CTE_SIZE(m_size)(assert(size == m_size), v->size = size); \ 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)) \ 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)) { \ if (!BUFFERI_POLICY_P((policy), BUFFER_THREAD_UNSAFE)) { \
m_mutex_init(v->mutex); \ m_mutex_init(v->mutex); \
m_cond_init(v->there_is_data); \ m_cond_init(v->there_is_data); \
@@ -145,9 +146,10 @@ M_C(name, _init)(buffer_t v, size_t size) \
i = 0; \ 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)) \ if (BUFFERI_POLICY_P(policy, BUFFER_DEFERRED_POP)) \
v->number[1] = 0; \ atomic_store(&v->number[1], 0); \
BUFFERI_CONTRACT(v,m_size); \ 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)) \ if (!BUFFERI_POLICY_P((policy), BUFFER_THREAD_UNSAFE)) \
m_mutex_lock(v->mutex); \ m_mutex_lock(v->mutex); \
BUFFERI_PROTECTED_CONTRACT(v, m_size); \ 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); \ M_C(name, _int_clear_obj)(v); \
else \ v->idx_prod = v->idx_cons = 0; \
v->idx_prod = v->idx_cons = v->number[0] = 0; \ atomic_store (&v->number[0], 0); \
if (BUFFERI_POLICY_P(policy, BUFFER_DEFERRED_POP)) \ 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)) { \ if (!BUFFERI_POLICY_P((policy), BUFFER_THREAD_UNSAFE)) { \
m_cond_signal(v->there_is_room_for_data); \ m_cond_signal(v->there_is_room_for_data); \
m_mutex_unlock(v->mutex); \ m_mutex_unlock(v->mutex); \
@@ -191,23 +193,23 @@ M_C(name, _init)(buffer_t v, size_t size) \
{ \ { \
BUFFERI_CONTRACT(v,m_size); \ BUFFERI_CONTRACT(v,m_size); \
if (BUFFERI_POLICY_P(policy, BUFFER_DEFERRED_POP)) \ if (BUFFERI_POLICY_P(policy, BUFFER_DEFERRED_POP)) \
return v->number[1] == 0; \ return atomic_load (&v->number[1]) == 0; \
else \ else \
return v->number[0] == 0; \ return atomic_load (&v->number[0]) == 0; \
} \ } \
\ \
static inline bool \ static inline bool \
M_C(name, _full_p)(const buffer_t v) \ M_C(name, _full_p)(const buffer_t v) \
{ \ { \
BUFFERI_CONTRACT(v,m_size); \ 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 \ static inline size_t \
M_C(name, _size)(const buffer_t v) \ M_C(name, _size)(const buffer_t v) \
{ \ { \
BUFFERI_CONTRACT(v,m_size); \ BUFFERI_CONTRACT(v,m_size); \
return v->number[0]; \ return atomic_load (&v->number[0]); \
} \ } \
\ \
static inline bool \ 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 */ \ /* INDEX computation if we have to overwrite the last element */ \
if (M_UNLIKELY (BUFFERI_POLICY_P((policy), BUFFER_PUSH_OVERWRITE) \ 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++; \ v->overwrite++; \
/* Let's clear the last element to overwrite it */ \ /* Let's clear the last element to overwrite it */ \
if (!BUFFERI_POLICY_P((policy), BUFFER_STACK)) { \ if (!BUFFERI_POLICY_P((policy), BUFFER_STACK)) { \
@@ -241,9 +243,9 @@ M_C(name, _init)(buffer_t v, size_t size) \
} else { \ } else { \
(v->idx_prod) --; \ (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)) \ 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)) { \ if (BUFFERI_POLICY_P((policy), BUFFER_PUSH_INIT_POP_MOVE)) { \
M_GET_CLEAR oplist(v->data[v->idx_prod]); \ M_GET_CLEAR oplist(v->data[v->idx_prod]); \
} \ } \
@@ -260,9 +262,9 @@ M_C(name, _init)(buffer_t v, size_t size) \
} else { \ } else { \
(v->idx_prod) ++; \ (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)) \ if (BUFFERI_POLICY_P((policy), BUFFER_DEFERRED_POP)) \
v->number[1] ++; \ atomic_store (&v->number[1], atomic_load(&v->number[1]) + 1); \
\ \
/* BUFFER unlock */ \ /* BUFFER unlock */ \
if (!BUFFERI_POLICY_P((policy), BUFFER_THREAD_UNSAFE)) { \ 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)) \ if (!BUFFERI_POLICY_P((policy), BUFFER_DEFERRED_POP)) \
v->number[0] --; \ atomic_store (&v->number[0], atomic_load(&v->number[0]) - 1); \
else \ else \
v->number[1] --; \ atomic_store (&v->number[1], atomic_load(&v->number[1]) - 1); \
/* BUFFER unlock */ \ /* BUFFER unlock */ \
if (!BUFFERI_POLICY_P((policy), BUFFER_THREAD_UNSAFE)) { \ if (!BUFFERI_POLICY_P((policy), BUFFER_THREAD_UNSAFE)) { \
m_cond_signal(v->there_is_room_for_data); \ 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) \ M_C(name, _pop_release)(buffer_t v) \
{ \ { \
if (BUFFERI_POLICY_P((policy), BUFFER_DEFERRED_POP)) \ if (BUFFERI_POLICY_P((policy), BUFFER_DEFERRED_POP)) \
v->number[0] --; \ atomic_store (&v->number[0], atomic_load(&v->number[0]) - 1); \
} \ } \