Update documentation on QUEUE_MPMC

This commit is contained in:
Patrick Pelissier committed 2020-06-06 11:09:55 +02:00
1 parent b722227b39
commit f4b9185ceb
2 files changed
+13 -10

No files matched your search

+5 -3
View File
@@ -3364,7 +3364,8 @@ Otherwise it does nothing.
Define the MPMC queue 'name##\_t' and its associated methods as "static inline" functions. Define the MPMC queue 'name##\_t' and its associated methods as "static inline" functions.
A MPMC queue is a fixed circular queue implementing a queue (or stack) interface. A MPMC queue is a fixed circular queue implementing a queue (or stack) interface.
It can be used to transfer message from Multiple Producer threads to Multiple Consumer threads. It can be used to transfer message from Multiple Producer threads to Multiple Consumer threads.
This is done internally using lock-free objects. This queue is not stricly lock free but [has](https://stackoverflow.com/questions/45907210/lock-free-progress-guarantees)
a lot of the properties of such algorithms.
The size is specified only at run-time and shall be a power of 2. The size is specified only at run-time and shall be a power of 2.
@@ -3376,8 +3377,9 @@ An additional policy can be applied to the buffer by performing a logical or of
* BUFFER\_PUSH\_INIT\_POP\_MOVE : change the behavior of PUSH to push a new initialized object, and POP as moving this new object into the new emplacement (this is mostly used for performance reasons or to handle properly a shared_ptr semantic). In practice, it works as if POP performs the initialization of the object. * BUFFER\_PUSH\_INIT\_POP\_MOVE : change the behavior of PUSH to push a new initialized object, and POP as moving this new object into the new emplacement (this is mostly used for performance reasons or to handle properly a shared_ptr semantic). In practice, it works as if POP performs the initialization of the object.
This container is designed to be used for easy synchronization inter-threads This container is designed to be used for easy synchronization inter-threads
in a context of very fast communication. in a context of very fast communication (the variable shall be a global shared one).
(the variable shall be a global shared one). There should not have more threads using this queue than they are available hardware cores
due to the only partial protection on Context-switch Immunity of this structure.
It shall be done once per type and per compilation unit. It shall be done once per type and per compilation unit.
+8 -7
View File
@@ -64,7 +64,7 @@ typedef enum {
((__VA_ARGS__, M_DEFAULT_OPLIST), \ ((__VA_ARGS__, M_DEFAULT_OPLIST), \
(__VA_ARGS__ ))) (__VA_ARGS__ )))
/* Define a lock-free queue for Many Producer Many Consummer /* Define a nearly lock-free queue for Many Producer Many Consummer
Much faster than queue of BUFFER_DEF in heavy communication scenario Much faster than queue of BUFFER_DEF in heavy communication scenario
Size of created queue can only a power of 2. Size of created queue can only a power of 2.
*/ */
@@ -581,8 +581,7 @@ M_C(name, _init)(buffer_t v, size_t size) \
/* Definition of a a QUEUE for Many Produccer / Many Consummer /* Definition of a a QUEUE for Many Produccer / Many Consummer
for high bandwidth scenario: for high bandwidth scenario:
* lock-free, * nearly lock-free,
* wait-free but only because it accepts to fail spuriously,
* quite fast * quite fast
* no blocking calls. * no blocking calls.
* only queue (no stack) * only queue (no stack)
@@ -621,13 +620,13 @@ M_C(name, _init)(buffer_t v, size_t size) \
/* The sequence number of an element will be equal to either \ /* The sequence number of an element will be equal to either \
- 2* the index of the production which creates it, \ - 2* the index of the production which creates it, \
- 1 + 2* the index of the consumption which consummes it \ - 1 + 2* the index of the consumption which consummes it \
In case of overflow, as there is no order comparison but only \ In case of wrapping, as there is no order comparison but only \
equal comparison, there is no special issue. \ equal comparison, there is no special issue. \
Each element is put in a separate cache line to avoid false \ Each element is put in a separate cache line to avoid false \
sharing. \ sharing. \
*/ \ */ \
typedef struct M_C(name, _el_s) { \ typedef struct M_C(name, _el_s) { \
atomic_uint seq; /* Can only increase until overflow */ \ atomic_uint seq; /* Can only increase until wrapping */ \
type x; \ type x; \
M_CACHELINE_ALIGN(align, atomic_uint, type); \ M_CACHELINE_ALIGN(align, atomic_uint, type); \
} M_C(name, _el_t); \ } M_C(name, _el_t); \
@@ -635,9 +634,9 @@ M_C(name, _init)(buffer_t v, size_t size) \
/* If there is only one producer and one consummer, then they won't \ /* If there is only one producer and one consummer, then they won't \
typically use the same cache line, increasing performance. */ \ 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 overflow */ \ atomic_uint ProdIdx; /* Can only increase until wrapping */ \
M_CACHELINE_ALIGN(align1, atomic_uint); \ M_CACHELINE_ALIGN(align1, atomic_uint); \
atomic_uint ConsoIdx; /* Can only increase until overflow */ \ atomic_uint ConsoIdx; /* Can only increase until wrapping */ \
M_CACHELINE_ALIGN(align2, atomic_uint); \ M_CACHELINE_ALIGN(align2, atomic_uint); \
M_C(name, _el_t) *Tab; \ M_C(name, _el_t) *Tab; \
unsigned int size; \ unsigned int size; \
@@ -661,11 +660,13 @@ M_C(name, _init)(buffer_t v, size_t size) \
/* Thread has been preempted by another one. */ \ /* Thread has been preempted by another one. */ \
return false; \ return false; \
} \ } \
/* If it is interrupted here, it may block pop method (not push) */ \
if (!BUFFERI_POLICY_P((policy), BUFFER_PUSH_INIT_POP_MOVE)) { \ 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 { \ } else { \
M_CALL_INIT_SET(oplist, table->Tab[i].x, x); \ M_CALL_INIT_SET(oplist, table->Tab[i].x, x); \
} \ } \
/* Finish transaction */ \
atomic_store_explicit(&table->Tab[i].seq, 2*idx, memory_order_release); \ atomic_store_explicit(&table->Tab[i].seq, 2*idx, memory_order_release); \
QUEUEI_MPMC_CONTRACT(table); \ QUEUEI_MPMC_CONTRACT(table); \
return true; \ return true; \