Patrick Pelissier
b50813713a
it is used to control the inline semantics. By combining it with optimize option "-fno-inline" for the internal M*LIB function (using _Pragma optimize) we can control easily if we want to use an inline version or call of an external function. Define new user API: * M_USE_EXTERN_DECL : define it before including any MLIB header to requests to call the M*LIB functions and don't inline then. * M_USE_EXTERN_DEF : define it once in a translation unit to define the M*LIB functions. This translation unit shall contain all templated macro too. If you use one option, you need to use the other one too for another translation unit.
815 lines
45 KiB
C
815 lines
45 KiB
C
/*
|
|
* M*LIB - SNAPSHOT Module
|
|
*
|
|
* Copyright (c) 2017-2023, Patrick Pelissier
|
|
* All rights reserved.
|
|
* Redistribution and use in source and binary forms, with or without
|
|
* modification, are permitted provided that the following conditions are met:
|
|
* + Redistributions of source code must retain the above copyright
|
|
* notice, this list of conditions and the following disclaimer.
|
|
* + Redistributions in binary form must reproduce the above copyright
|
|
* notice, this list of conditions and the following disclaimer in the
|
|
* documentation and/or other materials provided with the distribution.
|
|
*
|
|
* THIS SOFTWARE IS PROVIDED BY THE REGENTS AND CONTRIBUTORS ``AS IS'' AND ANY
|
|
* EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
|
|
* WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
|
|
* DISCLAIMED. IN NO EVENT SHALL THE REGENTS AND CONTRIBUTORS BE LIABLE FOR ANY
|
|
* DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
|
|
* (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
|
|
* LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
|
|
* ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
|
|
* (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
|
|
* SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
|
|
*/
|
|
#ifndef MSTARLIB_SNAPSHOT_H
|
|
#define MSTARLIB_SNAPSHOT_H
|
|
|
|
#include "m-atomic.h"
|
|
#include "m-core.h"
|
|
#include "m-genint.h"
|
|
|
|
M_BEGIN_PROTECTED_CODE
|
|
|
|
/* Define a Single Producer Single Consummer snapshot and its functions
|
|
USAGE: SNAPSHOT_SPSC_DEF(name, type[, oplist]) */
|
|
#define M_SNAPSHOT_SPSC_DEF(name, ...) \
|
|
M_SNAPSHOT_SPSC_DEF_AS(name, M_F(name,_t), __VA_ARGS__)
|
|
|
|
|
|
/* Define a Single Producer Single Consummer snapshot and its functions
|
|
as the given name name_t
|
|
USAGE: SNAPSHOT_SPSC_DEF_AS(name, name_t, type[, oplist]) */
|
|
#define M_SNAPSHOT_SPSC_DEF_AS(name, name_t, ...) \
|
|
M_BEGIN_PROTECTED_CODE \
|
|
M_SNAPSH0T_SPSC_DEF_P1(M_IF_NARGS_EQ1(__VA_ARGS__) \
|
|
((name, __VA_ARGS__, M_GLOBAL_OPLIST_OR_DEF(__VA_ARGS__)(), name_t ), \
|
|
(name, __VA_ARGS__ , name_t ))) \
|
|
M_END_PROTECTED_CODE
|
|
|
|
|
|
/* Define a Single Producer Multiple Consummer snapshot and its functions
|
|
USAGE: SNAPSHOT_SPMC_DEF(name, type[, oplist]) */
|
|
#define M_SNAPSHOT_SPMC_DEF(name, ...) \
|
|
M_SNAPSHOT_SPMC_DEF_AS(name, M_F(name,_t), __VA_ARGS__)
|
|
|
|
|
|
/* Define a Single Producer Multiple Consummer snapshot and its functions
|
|
as the given name name_t
|
|
USAGE: SNAPSHOT_SPMC_DEF_AS(name, type[, oplist]) */
|
|
#define M_SNAPSHOT_SPMC_DEF_AS(name, name_t, ...) \
|
|
M_BEGIN_PROTECTED_CODE \
|
|
M_SNAPSH0T_SPMC_DEF_P1(M_IF_NARGS_EQ1(__VA_ARGS__) \
|
|
((name, __VA_ARGS__, M_GLOBAL_OPLIST_OR_DEF(__VA_ARGS__)(), name_t ), \
|
|
(name, __VA_ARGS__ , name_t ))) \
|
|
M_END_PROTECTED_CODE
|
|
|
|
|
|
/* Define a Multiple Producer Multiple Consummer snapshot and its functions
|
|
USAGE: SNAPSHOT_MPMC_DEF(name, type[, oplist]) */
|
|
#define M_SNAPSHOT_MPMC_DEF(name, ...) \
|
|
M_SNAPSHOT_MPMC_DEF_AS(name, M_F(name,_t), __VA_ARGS__)
|
|
|
|
|
|
/* Define a Multiple Producer Multiple Consummer snapshot and its functions
|
|
as the given name name_t
|
|
USAGE: SNAPSHOT_MPMC_DEF_AS(name, name_t, type[, oplist]) */
|
|
#define M_SNAPSHOT_MPMC_DEF_AS(name, name_t, ...) \
|
|
M_BEGIN_PROTECTED_CODE \
|
|
M_SNAPSH0T_MPMC_DEF_P1(M_IF_NARGS_EQ1(__VA_ARGS__) \
|
|
((name, __VA_ARGS__, M_GLOBAL_OPLIST_OR_DEF(__VA_ARGS__)(), name_t ), \
|
|
(name, __VA_ARGS__ , name_t ))) \
|
|
M_END_PROTECTED_CODE
|
|
|
|
|
|
/* Define the oplist of a snapshot (SPSC, SPMC or MPMC).
|
|
USAGE: SNAPSHOT_OPLIST(name[, oplist]) */
|
|
#define M_SNAPSHOT_OPLIST(...) \
|
|
M_SNAPSH0T_OPLIST_P1(M_IF_NARGS_EQ1(__VA_ARGS__) \
|
|
((__VA_ARGS__, M_GLOBAL_OPLIST_OR_DEF(__VA_ARGS__)() ), \
|
|
(__VA_ARGS__ )))
|
|
|
|
|
|
/*****************************************************************************/
|
|
/********************************** INTERNAL *********************************/
|
|
/*****************************************************************************/
|
|
|
|
// deferred evaluation of the input
|
|
#define M_SNAPSH0T_OPLIST_P1(arg) M_SNAPSH0T_OPLIST_P2 arg
|
|
|
|
/* Validation of the given oplist */
|
|
#define M_SNAPSH0T_OPLIST_P2(name, oplist) \
|
|
M_IF_OPLIST(oplist)(M_SNAPSH0T_OPLIST_P3, M_SNAPSH0T_OPLIST_FAILURE)(name, oplist)
|
|
|
|
/* Prepare a clean compilation failure */
|
|
#define M_SNAPSH0T_OPLIST_FAILURE(name, oplist) \
|
|
((M_LIB_ERROR(ARGUMENT_OF_SNAPSHOT_OPLIST_IS_NOT_AN_OPLIST, name, oplist)))
|
|
|
|
/* Define the oplist of a snapshot */
|
|
#define M_SNAPSH0T_OPLIST_P3(name, oplist) \
|
|
(INIT(M_F(name, _init)) \
|
|
,INIT_SET(M_F(name, _init_set)) \
|
|
,SET(M_F(name, _set)) \
|
|
,CLEAR(M_F(name, _clear)) \
|
|
,NAME(name) \
|
|
,TYPE(M_F(name, _ct)) \
|
|
,SUBTYPE(M_F(name, _subtype_ct)) \
|
|
,OPLIST(oplist) \
|
|
,M_IF_METHOD(INIT_MOVE, oplist)(INIT_MOVE(M_F(name, _init_move)),) \
|
|
,M_IF_METHOD(MOVE, oplist)(MOVE(M_F(name, _move)),) \
|
|
)
|
|
|
|
|
|
/********************************** INTERNAL *********************************/
|
|
|
|
/* Flag defining the atomic state of a snapshot:
|
|
* - r: Index of the read buffer Range [0..2]
|
|
* - w: Index of the write buffer Range [0..2]
|
|
* - f: Next index of the write buffer when a shot is taken Range [0..2]
|
|
* - b: Boolean indicating that the read buffer shall be updated
|
|
* all fields packed in an unsigned char type.
|
|
*/
|
|
#define M_SNAPSH0T_SPSC_FLAG(r, w, f, b) \
|
|
((unsigned char)( ( (r) << 4) | ((w) << 2) | ((f)) | ((b) << 6)))
|
|
#define M_SNAPSH0T_SPSC_R(flags) \
|
|
(((unsigned int) (flags) >> 4) & 0x03u)
|
|
#define M_SNAPSH0T_SPSC_W(flags) \
|
|
(((unsigned int) (flags) >> 2) & 0x03u)
|
|
#define M_SNAPSH0T_SPSC_F(flags) \
|
|
(((unsigned int) (flags) >> 0) & 0x03u)
|
|
#define M_SNAPSH0T_SPSC_B(flags) \
|
|
(((unsigned int) (flags) >> 6) & 0x01u)
|
|
|
|
/* NOTE: Due to atomic_load only accepting non-const pointer,
|
|
we can't have any const in the interface. */
|
|
#define M_SNAPSH0T_SPSC_FLAGS_CONTRACT(flags) \
|
|
M_ASSERT(M_SNAPSH0T_SPSC_R(flags) != M_SNAPSH0T_SPSC_W(flags) \
|
|
&& M_SNAPSH0T_SPSC_R(flags) != M_SNAPSH0T_SPSC_F(flags) \
|
|
&& M_SNAPSH0T_SPSC_W(flags) != M_SNAPSH0T_SPSC_F(flags))
|
|
|
|
#define M_SNAPSH0T_SPSC_CONTRACT(snap) do { \
|
|
M_ASSERT((snap) != NULL); \
|
|
unsigned char f = atomic_load (&(snap)->flags); \
|
|
M_SNAPSH0T_SPSC_FLAGS_CONTRACT(f); \
|
|
} while (0)
|
|
|
|
// A snapshot is basically an atomic triple buffer (Lock Free)
|
|
// between a single producer thread and a single consummer thread.
|
|
#define M_SNAPSH0T_SPSC_MAX_BUFFER 3
|
|
|
|
// Defered evaluation of the arguments.
|
|
#define M_SNAPSH0T_SPSC_DEF_P1(arg) M_ID( M_SNAPSH0T_SPSC_DEF_P2 arg )
|
|
|
|
/* Validate the oplist before going further */
|
|
#define M_SNAPSH0T_SPSC_DEF_P2(name, type, oplist, snapshot_t) \
|
|
M_IF_OPLIST(oplist)(M_SNAPSH0T_SPSC_DEF_P3, M_SNAPSH0T_SPSC_DEF_FAILURE)(name, type, oplist, snapshot_t)
|
|
|
|
/* Stop processing with a compilation failure */
|
|
#define M_SNAPSH0T_SPSC_DEF_FAILURE(name, type, oplist, snapshot_t) \
|
|
M_STATIC_FAILURE(M_LIB_NOT_AN_OPLIST, "(SNAPSHOT_SPSC_DEF): the given argument is not a valid oplist: " #oplist)
|
|
|
|
/* Expand the type and the functions of a SPSC snapshot */
|
|
#define M_SNAPSH0T_SPSC_DEF_P3(name, type, oplist, snapshot_t) \
|
|
M_SNAPSH0T_SPSC_DEF_TYPE(name, type, oplist, snapshot_t) \
|
|
M_CHECK_COMPATIBLE_OPLIST(name, 1, type, oplist) \
|
|
M_SNAPSH0T_SPSC_DEF_CORE(name, type, oplist, snapshot_t) \
|
|
|
|
/* Define the type */
|
|
#define M_SNAPSH0T_SPSC_DEF_TYPE(name, type, oplist, snapshot_t) \
|
|
\
|
|
/* Create an aligned type to avoid false sharing between threads */ \
|
|
typedef struct M_F(name, _aligned_type_s) { \
|
|
type x; \
|
|
M_CACHELINE_ALIGN(align, type); \
|
|
} M_F(name, _aligned_type_ct); \
|
|
\
|
|
typedef struct M_F(name, _s) { \
|
|
M_F(name, _aligned_type_ct) data[M_SNAPSH0T_SPSC_MAX_BUFFER]; \
|
|
atomic_uchar flags; \
|
|
} snapshot_t[1]; \
|
|
typedef struct M_F(name, _s) *M_F(name, _ptr); \
|
|
typedef const struct M_F(name, _s) *M_F(name, _srcptr); \
|
|
\
|
|
/* Define internal types for oplist */ \
|
|
typedef snapshot_t M_F(name, _ct); \
|
|
typedef type M_F(name, _subtype_ct); \
|
|
|
|
/* Define the core functions */
|
|
#define M_SNAPSH0T_SPSC_DEF_CORE(name, type, oplist, snapshot_t) \
|
|
\
|
|
M_INLINE void \
|
|
M_F(name, _init)(snapshot_t snap) \
|
|
{ \
|
|
M_ASSERT(snap != NULL); \
|
|
for(int i = 0; i < M_SNAPSH0T_SPSC_MAX_BUFFER; i++) { \
|
|
M_CALL_INIT(oplist, snap->data[i].x); \
|
|
} \
|
|
atomic_init (&snap->flags, M_SNAPSH0T_SPSC_FLAG(0, 1, 2, 0)); \
|
|
M_SNAPSH0T_SPSC_CONTRACT(snap); \
|
|
} \
|
|
\
|
|
M_INLINE void \
|
|
M_F(name, _clear)(snapshot_t snap) \
|
|
{ \
|
|
M_SNAPSH0T_SPSC_CONTRACT(snap); \
|
|
for(int i = 0; i < M_SNAPSH0T_SPSC_MAX_BUFFER; i++) { \
|
|
M_CALL_CLEAR(oplist, snap->data[i].x); \
|
|
} \
|
|
} \
|
|
\
|
|
/* const is missing for org due to use of atomic_load of org */ \
|
|
M_INLINE void \
|
|
M_F(name, _init_set)(snapshot_t snap, snapshot_t org) \
|
|
{ \
|
|
M_SNAPSH0T_SPSC_CONTRACT(org); \
|
|
M_ASSERT(snap != NULL && snap != org); \
|
|
for(int i = 0; i < M_SNAPSH0T_SPSC_MAX_BUFFER; i++) { \
|
|
M_CALL_INIT_SET(oplist, snap->data[i].x, org->data[i].x); \
|
|
} \
|
|
atomic_init (&snap->flags, atomic_load(&org->flags)); \
|
|
M_SNAPSH0T_SPSC_CONTRACT(snap); \
|
|
} \
|
|
\
|
|
/* const is missing for org due to use of atomic_load of org */ \
|
|
M_INLINE void \
|
|
M_F(name, _set)(snapshot_t snap, snapshot_t org) \
|
|
{ \
|
|
M_SNAPSH0T_SPSC_CONTRACT(snap); \
|
|
M_SNAPSH0T_SPSC_CONTRACT(org); \
|
|
for(int i = 0; i < M_SNAPSH0T_SPSC_MAX_BUFFER; i++) { \
|
|
M_CALL_SET(oplist, snap->data[i].x, org->data[i].x); \
|
|
} \
|
|
atomic_init (&snap->flags, atomic_load(&org->flags)); \
|
|
M_SNAPSH0T_SPSC_CONTRACT(snap); \
|
|
} \
|
|
\
|
|
M_IF_METHOD(INIT_MOVE, oplist)( \
|
|
M_INLINE void \
|
|
M_F(name, _init_move)(snapshot_t snap, snapshot_t org) \
|
|
{ \
|
|
M_SNAPSH0T_SPSC_CONTRACT(org); \
|
|
M_ASSERT(snap != NULL && snap != org); \
|
|
for(int i = 0; i < M_SNAPSH0T_SPSC_MAX_BUFFER; i++) { \
|
|
M_CALL_INIT_MOVE(oplist, snap->data[i].x, org->data[i].x); \
|
|
} \
|
|
atomic_store (&snap->flags, atomic_load(&org->flags)); \
|
|
atomic_store (&org->flags, M_SNAPSH0T_SPSC_FLAG(0,0,0,0) ); \
|
|
M_SNAPSH0T_SPSC_CONTRACT(snap); \
|
|
} \
|
|
,) /* IF_METHOD (INIT_MOVE) */ \
|
|
\
|
|
M_IF_METHOD(MOVE, oplist)( \
|
|
M_INLINE void \
|
|
M_F(name, _move)(snapshot_t snap, \
|
|
snapshot_t org) \
|
|
{ \
|
|
M_SNAPSH0T_SPSC_CONTRACT(snap); \
|
|
M_SNAPSH0T_SPSC_CONTRACT(org); \
|
|
M_ASSERT(snap != org); \
|
|
for(int i = 0; i < M_SNAPSH0T_SPSC_MAX_BUFFER; i++) { \
|
|
M_CALL_MOVE(oplist, snap->data[i].x, org->data[i].x); \
|
|
} \
|
|
atomic_store (&snap->flags, atomic_load(&org->flags)); \
|
|
atomic_store (&org->flags, M_SNAPSH0T_SPSC_FLAG(0,0,0,0) ); \
|
|
M_SNAPSH0T_SPSC_CONTRACT(snap); \
|
|
} \
|
|
,) /* IF_METHOD (MOVE) */ \
|
|
\
|
|
M_INLINE type * \
|
|
M_F(name, _write)(snapshot_t snap) \
|
|
{ \
|
|
M_SNAPSH0T_SPSC_CONTRACT(snap); \
|
|
unsigned char nextFlags, origFlags = atomic_load (&snap->flags); \
|
|
/* Atomic CAS operation */ \
|
|
do { \
|
|
/* Swap F and W buffer, setting exchange flag */ \
|
|
nextFlags = M_SNAPSH0T_SPSC_FLAG(M_SNAPSH0T_SPSC_R(origFlags), \
|
|
M_SNAPSH0T_SPSC_F(origFlags), \
|
|
M_SNAPSH0T_SPSC_W(origFlags), 1); \
|
|
/* exponential backoff is not needed as there can't be more \
|
|
than 2 threads which try to update the data. */ \
|
|
} while (!atomic_compare_exchange_weak (&snap->flags, &origFlags, \
|
|
nextFlags)); \
|
|
/* Return new write buffer for new updating */ \
|
|
return &snap->data[M_SNAPSH0T_SPSC_W(nextFlags)].x; \
|
|
} \
|
|
\
|
|
M_INLINE type const * \
|
|
M_F(name, _read)(snapshot_t snap) \
|
|
{ \
|
|
M_SNAPSH0T_SPSC_CONTRACT(snap); \
|
|
unsigned char nextFlags, origFlags = atomic_load (&snap->flags); \
|
|
/* Atomic CAS operation */ \
|
|
do { \
|
|
/* If no exchange registered, do nothing and keep the same */ \
|
|
if (!M_SNAPSH0T_SPSC_B(origFlags)) { \
|
|
nextFlags = origFlags; \
|
|
break; \
|
|
} \
|
|
/* Swap R and F buffer, clearing exchange flag */ \
|
|
nextFlags = M_SNAPSH0T_SPSC_FLAG(M_SNAPSH0T_SPSC_F(origFlags), \
|
|
M_SNAPSH0T_SPSC_W(origFlags), \
|
|
M_SNAPSH0T_SPSC_R(origFlags), 0); \
|
|
/* exponential backoff is not needed as there can't be more \
|
|
than 2 threads which try to update the data. */ \
|
|
} while (!atomic_compare_exchange_weak (&snap->flags, &origFlags, \
|
|
nextFlags)); \
|
|
/* Return current read buffer */ \
|
|
return M_CONST_CAST(type, &snap->data[M_SNAPSH0T_SPSC_R(nextFlags)].x); \
|
|
} \
|
|
\
|
|
/* Non const due to use of atomic_load */ \
|
|
M_INLINE bool \
|
|
M_F(name, _updated_p)(snapshot_t snap) \
|
|
{ \
|
|
M_SNAPSH0T_SPSC_CONTRACT(snap); \
|
|
unsigned char flags = atomic_load (&snap->flags); \
|
|
return M_SNAPSH0T_SPSC_B(flags); \
|
|
} \
|
|
\
|
|
/* Non const due to use of atomic_load */ \
|
|
M_INLINE type * \
|
|
M_F(name, _get_write_buffer)(snapshot_t snap) \
|
|
{ \
|
|
M_SNAPSH0T_SPSC_CONTRACT(snap); \
|
|
unsigned char flags = atomic_load(&snap->flags); \
|
|
return &snap->data[M_SNAPSH0T_SPSC_W(flags)].x; \
|
|
} \
|
|
\
|
|
/* Non const due to use of atomic_load */ \
|
|
M_INLINE type const * \
|
|
M_F(name, _get_read_buffer)(snapshot_t snap) \
|
|
{ \
|
|
M_SNAPSH0T_SPSC_CONTRACT(snap); \
|
|
unsigned char flags = atomic_load(&snap->flags); \
|
|
return M_CONST_CAST(type, &snap->data[M_SNAPSH0T_SPSC_R(flags)].x); \
|
|
} \
|
|
|
|
|
|
/********************************** INTERNAL *********************************/
|
|
|
|
#define M_SNAPSH0T_SPMC_INT_FLAG(w, n) ( ((w) << 1) | (n) )
|
|
#define M_SNAPSH0T_SPMC_INT_FLAG_W(f) ((f) >> 1)
|
|
#define M_SNAPSH0T_SPMC_INT_FLAG_N(f) ((f) & 1)
|
|
|
|
// 2 more buffer than the number of readers are needed
|
|
#define M_SNAPSH0T_SPMC_EXTRA_BUFFER 2
|
|
|
|
#define M_SNAPSH0T_SPMC_MAX_READER (M_GENINT_MAX_ALLOC-M_SNAPSH0T_SPMC_EXTRA_BUFFER)
|
|
|
|
/* Internal structure to handle SPMC snapshot but return an unique index in the buffer array.
|
|
- lastNext: last published written index + next flag (format M_SNAPSH0T_SPMC_INT_FLAG)
|
|
- currentWrite: the index being currently written.
|
|
- n_reader : number of readers
|
|
- cptTab: ref counter array to keep track of how many readers use the corresponding buffer.
|
|
- freeList: a pool of free integers.
|
|
*/
|
|
typedef struct m_snapsh0t_mrsw_s {
|
|
atomic_uint lastNext;
|
|
unsigned int currentWrite;
|
|
size_t n_reader;
|
|
atomic_uint *cptTab;
|
|
m_genint_t freeList;
|
|
} m_snapsh0t_mrsw_ct[1];
|
|
|
|
// can't check currentWrite due to potential data race on it
|
|
#define M_SNAPSH0T_SPMC_INT_CONTRACT(s) do { \
|
|
M_ASSERT (s != NULL); \
|
|
M_ASSERT (s->n_reader > 0 && s->n_reader <= M_SNAPSH0T_SPMC_MAX_READER); \
|
|
M_ASSERT ((size_t)M_SNAPSH0T_SPMC_INT_FLAG_W(atomic_load(&s->lastNext)) \
|
|
<= s->n_reader + M_SNAPSH0T_SPMC_EXTRA_BUFFER); \
|
|
M_ASSERT (s->cptTab != NULL); \
|
|
} while (0)
|
|
|
|
/* Initialize m_snapsh0t_mrsw_ct for n readers (constructor) */
|
|
M_INLINE void
|
|
m_snapsh0t_mrsw_init(m_snapsh0t_mrsw_ct s, size_t n)
|
|
{
|
|
M_ASSERT (s != NULL);
|
|
M_ASSERT (n >= 1 && n <= M_SNAPSH0T_SPMC_MAX_READER);
|
|
s->n_reader = n;
|
|
n += M_SNAPSH0T_SPMC_EXTRA_BUFFER;
|
|
|
|
// Initialize the counters to zero (no reader use it)
|
|
atomic_uint *ptr = M_MEMORY_REALLOC (atomic_uint, NULL, n);
|
|
if (M_UNLIKELY_NOMEM (ptr == NULL)) {
|
|
M_MEMORY_FULL(sizeof (atomic_uint) * n);
|
|
return;
|
|
}
|
|
s->cptTab = ptr;
|
|
for(size_t i = 0; i < n; i++)
|
|
atomic_init(&s->cptTab[i], 0U);
|
|
m_genint_init (s->freeList, (unsigned int) n);
|
|
|
|
// Get a free buffer and set it as available for readers
|
|
unsigned int w = m_genint_pop(s->freeList);
|
|
M_ASSERT (w != M_GENINT_ERROR);
|
|
atomic_store(&s->cptTab[w], 1U);
|
|
atomic_init(&s->lastNext, M_SNAPSH0T_SPMC_INT_FLAG(w, true));
|
|
|
|
// Get working buffer
|
|
s->currentWrite = m_genint_pop(s->freeList);
|
|
M_ASSERT (s->currentWrite != M_GENINT_ERROR);
|
|
atomic_store(&s->cptTab[s->currentWrite], 1U);
|
|
M_SNAPSH0T_SPMC_INT_CONTRACT(s);
|
|
}
|
|
|
|
/* Clear m_snapsh0t_mrsw_ct (destructor) */
|
|
M_INLINE void
|
|
m_snapsh0t_mrsw_clear(m_snapsh0t_mrsw_ct s)
|
|
{
|
|
M_SNAPSH0T_SPMC_INT_CONTRACT(s);
|
|
M_MEMORY_FREE (s->cptTab);
|
|
m_genint_clear(s->freeList);
|
|
s->cptTab = NULL;
|
|
s->n_reader = 0;
|
|
}
|
|
|
|
/* Return the current index that is written in the buffer */
|
|
M_INLINE unsigned int
|
|
m_snapsh0t_mrsw_get_write_idx(m_snapsh0t_mrsw_ct s)
|
|
{
|
|
M_SNAPSH0T_SPMC_INT_CONTRACT(s);
|
|
return s->currentWrite;
|
|
}
|
|
|
|
/* Return the number of readers */
|
|
M_INLINE unsigned int
|
|
m_snapsh0t_mrsw_size(m_snapsh0t_mrsw_ct s)
|
|
{
|
|
M_SNAPSH0T_SPMC_INT_CONTRACT(s);
|
|
return (unsigned int) s->n_reader;
|
|
}
|
|
|
|
/* Give the current index that is written to the readers,
|
|
and return new available index for the writer thread */
|
|
M_INLINE unsigned int
|
|
m_snapsh0t_mrsw_write_idx(m_snapsh0t_mrsw_ct s, unsigned int idx)
|
|
{
|
|
M_SNAPSH0T_SPMC_INT_CONTRACT(s);
|
|
|
|
// Provide the finalized written buffer to the readers.
|
|
unsigned int newNext, previous = atomic_load(&s->lastNext);
|
|
do {
|
|
newNext = M_SNAPSH0T_SPMC_INT_FLAG(idx, true);
|
|
} while (!atomic_compare_exchange_weak(&s->lastNext, &previous, newNext));
|
|
|
|
if (M_SNAPSH0T_SPMC_INT_FLAG_N(previous)) {
|
|
// Reuse previous buffer as it was not used by any reader
|
|
idx = M_SNAPSH0T_SPMC_INT_FLAG_W(previous);
|
|
// Some other read threads may already have try to reserve this index
|
|
// So atomic_load(&s->cptTab[idx]) can be greater than 1.
|
|
// However they will fail to ack it in lastNext,
|
|
// so they will remove their reservation later
|
|
} else {
|
|
// Remove the writer thread counter from the count of the previous buffer
|
|
idx = M_SNAPSH0T_SPMC_INT_FLAG_W(previous);
|
|
unsigned int c = atomic_fetch_sub(&s->cptTab[idx], 1U);
|
|
M_ASSERT (c != 0 && c <= s->n_reader + 1);
|
|
// Get a new buffer.
|
|
if (c != 1) {
|
|
// If someone else keeps a ref on the buffer, we can't reuse it
|
|
// get another free one.
|
|
idx = m_genint_pop(s->freeList);
|
|
M_ASSERT(idx != M_GENINT_ERROR);
|
|
} else {
|
|
// No other thread keep track of this buffer.
|
|
// Reuse it.
|
|
}
|
|
M_ASSERT (idx < s->n_reader + M_SNAPSH0T_SPMC_EXTRA_BUFFER);
|
|
M_ASSERT (atomic_load(&s->cptTab[idx]) == 0);
|
|
atomic_store(&s->cptTab[idx], 1U);
|
|
}
|
|
M_SNAPSH0T_SPMC_INT_CONTRACT(s);
|
|
return idx;
|
|
}
|
|
|
|
/* Perform a swap of the current write buffer and return a new one */
|
|
M_INLINE unsigned int
|
|
m_snapsh0t_mrsw_write(m_snapsh0t_mrsw_ct s)
|
|
{
|
|
s->currentWrite = m_snapsh0t_mrsw_write_idx(s, s->currentWrite);
|
|
M_SNAPSH0T_SPMC_INT_CONTRACT(s);
|
|
return s->currentWrite;
|
|
}
|
|
|
|
/* Start writing to the write buffer and return its index */
|
|
M_INLINE unsigned int
|
|
m_snapsh0t_mrsw_write_start(m_snapsh0t_mrsw_ct s)
|
|
{
|
|
M_SNAPSH0T_SPMC_INT_CONTRACT(s);
|
|
// Get a new buffer.
|
|
unsigned int idx = m_genint_pop(s->freeList);
|
|
M_ASSERT (idx != M_GENINT_ERROR);
|
|
M_ASSERT (idx < s->n_reader + M_SNAPSH0T_SPMC_EXTRA_BUFFER);
|
|
M_ASSERT (atomic_load(&s->cptTab[idx]) == 0);
|
|
atomic_store(&s->cptTab[idx], 1U);
|
|
M_SNAPSH0T_SPMC_INT_CONTRACT(s);
|
|
return idx;
|
|
}
|
|
|
|
/* End writing to the given write buffer */
|
|
M_INLINE void
|
|
m_snapsh0t_mrsw_write_end(m_snapsh0t_mrsw_ct s, unsigned int idx)
|
|
{
|
|
M_SNAPSH0T_SPMC_INT_CONTRACT(s);
|
|
|
|
// Provide this write bufer to the readers
|
|
unsigned int newNext, previous = atomic_load(&s->lastNext);
|
|
do {
|
|
newNext = M_SNAPSH0T_SPMC_INT_FLAG(idx, true);
|
|
} while (!atomic_compare_exchange_weak(&s->lastNext, &previous, newNext));
|
|
|
|
// Free the previous write buffer
|
|
idx = M_SNAPSH0T_SPMC_INT_FLAG_W(previous);
|
|
unsigned int c = atomic_fetch_sub(&s->cptTab[idx], 1U);
|
|
M_ASSERT (c != 0 && c <= s->n_reader + 1);
|
|
if (c == 1) {
|
|
m_genint_push(s->freeList, idx);
|
|
}
|
|
M_SNAPSH0T_SPMC_INT_CONTRACT(s);
|
|
}
|
|
|
|
/* Start reading the latest written buffer and return the index to it */
|
|
M_INLINE unsigned int
|
|
m_snapsh0t_mrsw_read_start(m_snapsh0t_mrsw_ct s)
|
|
{
|
|
M_SNAPSH0T_SPMC_INT_CONTRACT(s);
|
|
unsigned int idx, previous;
|
|
reload:
|
|
// Load the last published index + Next flag
|
|
previous = atomic_load(&s->lastNext);
|
|
while (true) {
|
|
// Get the last published index
|
|
idx = M_SNAPSH0T_SPMC_INT_FLAG_W(previous);
|
|
// Load the number of threads using this index
|
|
unsigned int c = atomic_load(&s->cptTab[idx]);
|
|
M_ASSERT (c <= s->n_reader + 1);
|
|
// Reserve the index if it still being reserved by someone else
|
|
if (M_UNLIKELY (c == 0
|
|
|| !atomic_compare_exchange_strong(&s->cptTab[idx], &c, c+1)))
|
|
goto reload;
|
|
// Try to ack it
|
|
unsigned int newNext = M_SNAPSH0T_SPMC_INT_FLAG(idx, false);
|
|
reforce:
|
|
if (M_LIKELY (atomic_compare_exchange_strong(&s->lastNext, &previous, newNext)))
|
|
break;
|
|
// We have been preempted by another thread
|
|
if (idx == M_SNAPSH0T_SPMC_INT_FLAG_W(previous)) {
|
|
// This is still ok if the index has not changed
|
|
// We can get previous to true again if the writer has recycled the index,
|
|
// while we reserved it, and the reader get prempted until its CAS.
|
|
if (M_UNLIKELY (M_SNAPSH0T_SPMC_INT_FLAG_N(previous) == true)) goto reforce;
|
|
break;
|
|
}
|
|
// Free the reserved index as we failed it to ack it
|
|
c = atomic_fetch_sub(&s->cptTab[idx], 1U);
|
|
M_ASSERT (c != 0 && c <= s->n_reader + 1);
|
|
if (c == 1) {
|
|
m_genint_push(s->freeList, idx);
|
|
}
|
|
}
|
|
M_SNAPSH0T_SPMC_INT_CONTRACT(s);
|
|
return idx;
|
|
}
|
|
|
|
/* End the reading the given buffer */
|
|
M_INLINE void
|
|
m_snapsh0t_mrsw_read_end(m_snapsh0t_mrsw_ct s, unsigned int idx)
|
|
{
|
|
M_SNAPSH0T_SPMC_INT_CONTRACT(s);
|
|
M_ASSERT (idx < s->n_reader + M_SNAPSH0T_SPMC_EXTRA_BUFFER);
|
|
// Decrement reference counter of the buffer
|
|
unsigned int c = atomic_fetch_sub(&s->cptTab[idx], 1U);
|
|
M_ASSERT (c != 0 && c <= s->n_reader + 1);
|
|
if (c == 1) {
|
|
// Buffer no longer used by any reader thread.
|
|
// Push back index in free list
|
|
m_genint_push(s->freeList, idx);
|
|
}
|
|
M_SNAPSH0T_SPMC_INT_CONTRACT(s);
|
|
}
|
|
|
|
|
|
/********************************** INTERNAL *********************************/
|
|
|
|
/* Contract of a SPMC snapshot.
|
|
Nothing notable as it can be accessed concurrently */
|
|
#define M_SNAPSH0T_SPMC_CONTRACT(snap) do { \
|
|
M_ASSERT (snap != NULL); \
|
|
M_ASSERT (snap->data != NULL); \
|
|
} while (0)
|
|
|
|
|
|
// Defered evaluation
|
|
#define M_SNAPSH0T_SPMC_DEF_P1(arg) M_ID( M_SNAPSH0T_SPMC_DEF_P2 arg )
|
|
|
|
/* Validate the oplist before going further */
|
|
#define M_SNAPSH0T_SPMC_DEF_P2(name, type, oplist, snapshot_t) \
|
|
M_IF_OPLIST(oplist)(M_SNAPSH0T_SPMC_DEF_P3, M_SNAPSH0T_SPMC_DEF_FAILURE)(name, type, oplist, snapshot_t)
|
|
|
|
/* Stop processing with a compilation failure */
|
|
#define M_SNAPSH0T_SPMC_DEF_FAILURE(name, type, oplist, snapshot_t) \
|
|
M_STATIC_FAILURE(M_LIB_NOT_AN_OPLIST, "(SNAPSHOT_SPMC_DEF): the given argument is not a valid oplist: " #oplist)
|
|
|
|
/* Expand the type and the functions of a SPMC snapshot */
|
|
#define M_SNAPSH0T_SPMC_DEF_P3(name, type, oplist, snapshot_t) \
|
|
M_SNAPSH0T_SPMC_DEF_TYPE(name, type, oplist, snapshot_t) \
|
|
M_CHECK_COMPATIBLE_OPLIST(name, 1, type, oplist) \
|
|
M_SNAPSH0T_SPMC_DEF_CORE(name, type, oplist, snapshot_t) \
|
|
|
|
/* Define the type */
|
|
#define M_SNAPSH0T_SPMC_DEF_TYPE(name, type, oplist, snapshot_t) \
|
|
\
|
|
/* Create an aligned type to avoid false sharing between threads */ \
|
|
typedef struct M_F(name, _aligned_type_s) { \
|
|
type x; \
|
|
M_CACHELINE_ALIGN(align, type); \
|
|
} M_F(name, _aligned_type_ct); \
|
|
\
|
|
typedef struct M_F(name, _s) { \
|
|
M_F(name, _aligned_type_ct) *data; \
|
|
m_snapsh0t_mrsw_ct core; \
|
|
} snapshot_t[1]; \
|
|
\
|
|
/* Define internal types for oplist */ \
|
|
typedef snapshot_t M_F(name, _ct); \
|
|
typedef type M_F(name, _subtype_ct); \
|
|
|
|
/* Define the core functions */
|
|
#define M_SNAPSH0T_SPMC_DEF_CORE(name, type, oplist, snapshot_t) \
|
|
\
|
|
M_INLINE void \
|
|
M_F(name, _init)(snapshot_t snap, size_t nReader) \
|
|
{ \
|
|
M_ASSERT (snap != NULL); \
|
|
M_ASSERT (nReader > 0 && nReader <= M_SNAPSH0T_SPMC_MAX_READER); \
|
|
snap->data = M_CALL_REALLOC(oplist, M_F(name, _aligned_type_ct), \
|
|
NULL, nReader+M_SNAPSH0T_SPMC_EXTRA_BUFFER); \
|
|
if (M_UNLIKELY_NOMEM (snap->data == NULL)) { \
|
|
M_MEMORY_FULL(sizeof(M_F(name, _aligned_type_ct)) * \
|
|
(nReader+M_SNAPSH0T_SPMC_EXTRA_BUFFER)); \
|
|
return; \
|
|
} \
|
|
for(size_t i = 0; i < nReader + M_SNAPSH0T_SPMC_EXTRA_BUFFER; i++) { \
|
|
M_CALL_INIT(oplist, snap->data[i].x); \
|
|
} \
|
|
m_snapsh0t_mrsw_init(snap->core, nReader); \
|
|
M_SNAPSH0T_SPMC_CONTRACT(snap); \
|
|
} \
|
|
\
|
|
M_INLINE void \
|
|
M_F(name, _clear)(snapshot_t snap) \
|
|
{ \
|
|
M_SNAPSH0T_SPMC_CONTRACT(snap); \
|
|
size_t nReader = m_snapsh0t_mrsw_size(snap->core); \
|
|
for(size_t i = 0; i < nReader + M_SNAPSH0T_SPMC_EXTRA_BUFFER; i++) { \
|
|
M_CALL_CLEAR(oplist, snap->data[i].x); \
|
|
} \
|
|
M_CALL_FREE(oplist, snap->data); \
|
|
m_snapsh0t_mrsw_clear(snap->core); \
|
|
} \
|
|
\
|
|
M_INLINE type * \
|
|
M_F(name, _write)(snapshot_t snap) \
|
|
{ \
|
|
M_SNAPSH0T_SPMC_CONTRACT(snap); \
|
|
const unsigned int idx = m_snapsh0t_mrsw_write(snap->core); \
|
|
return &snap->data[idx].x; \
|
|
} \
|
|
\
|
|
M_INLINE type const * \
|
|
M_F(name, _read_start)(snapshot_t snap) \
|
|
{ \
|
|
M_SNAPSH0T_SPMC_CONTRACT(snap); \
|
|
const unsigned int idx = m_snapsh0t_mrsw_read_start(snap->core); \
|
|
return M_CONST_CAST(type, &snap->data[idx].x); \
|
|
} \
|
|
\
|
|
M_INLINE void \
|
|
M_F(name, _read_end)(snapshot_t snap, type const *old) \
|
|
{ \
|
|
M_SNAPSH0T_SPMC_CONTRACT(snap); \
|
|
M_ASSERT (old != NULL); \
|
|
const M_F(name, _aligned_type_ct) *oldx; \
|
|
oldx = M_CTYPE_FROM_FIELD(M_F(name, _aligned_type_ct), old, type, x); \
|
|
M_ASSERT (oldx >= snap->data); \
|
|
M_ASSERT (oldx < snap->data + snap->core->n_reader + M_SNAPSH0T_SPMC_EXTRA_BUFFER); \
|
|
M_ASSERT(snap->core->n_reader +M_SNAPSH0T_SPMC_EXTRA_BUFFER < UINT_MAX); \
|
|
const unsigned int idx = (unsigned int) (oldx - snap->data); \
|
|
m_snapsh0t_mrsw_read_end(snap->core, idx); \
|
|
} \
|
|
\
|
|
M_INLINE type * \
|
|
M_F(name, _get_write_buffer)(snapshot_t snap) \
|
|
{ \
|
|
M_SNAPSH0T_SPMC_CONTRACT(snap); \
|
|
const unsigned int idx = m_snapsh0t_mrsw_get_write_idx(snap->core); \
|
|
return &snap->data[idx].x; \
|
|
} \
|
|
\
|
|
|
|
|
|
/********************************** INTERNAL *********************************/
|
|
|
|
// MPMC is built upon SPMC
|
|
|
|
// Defered evaluation
|
|
#define M_SNAPSH0T_MPMC_DEF_P1(arg) M_ID( M_SNAPSH0T_MPMC_DEF_P2 arg )
|
|
|
|
/* Validate the oplist before going further */
|
|
#define M_SNAPSH0T_MPMC_DEF_P2(name, type, oplist, snapshot_t) \
|
|
M_IF_OPLIST(oplist)(M_SNAPSH0T_MPMC_DEF_P3, M_SNAPSH0T_MPMC_DEF_FAILURE)(name, type, oplist, snapshot_t)
|
|
|
|
/* Stop processing with a compilation failure */
|
|
#define M_SNAPSH0T_MPMC_DEF_FAILURE(name, type, oplist, snapshot_t) \
|
|
M_STATIC_FAILURE(M_LIB_NOT_AN_OPLIST, "(SNAPSHOT_MPMC_DEF): the given argument is not a valid oplist: " #oplist)
|
|
|
|
/* Expand the type and the functions of a MPMC snapshot */
|
|
#define M_SNAPSH0T_MPMC_DEF_P3(name, type, oplist, snapshot_t) \
|
|
M_SNAPSH0T_SPMC_DEF_P1((M_F(name, _mrsw), type, oplist, M_F(name, _mrsw_pct))) \
|
|
M_SNAPSH0T_MPMC_DEF_TYPE(name, type, oplist, snapshot_t) \
|
|
M_CHECK_COMPATIBLE_OPLIST(name, 1, type, oplist) \
|
|
M_SNAPSH0T_MPMC_DEF_CORE(name, type, oplist, snapshot_t) \
|
|
|
|
/* Define the types */
|
|
#define M_SNAPSH0T_MPMC_DEF_TYPE(name, type, oplist, snapshot_t) \
|
|
\
|
|
typedef struct M_F(name, _s) { \
|
|
M_F(name, _mrsw_pct) core; \
|
|
} snapshot_t[1]; \
|
|
\
|
|
/* Define internal types for oplist */ \
|
|
typedef snapshot_t M_F(name, _ct); \
|
|
typedef type M_F(name, _subtype_ct); \
|
|
|
|
/* Define the core functions */
|
|
#define M_SNAPSH0T_MPMC_DEF_CORE(name, type, oplist, snapshot_t) \
|
|
\
|
|
M_INLINE void \
|
|
M_F(name, _init)(snapshot_t snap, size_t nReader, size_t nWriter) \
|
|
{ \
|
|
M_F(name, _mrsw_init)(snap->core, nReader + nWriter -1 ); \
|
|
unsigned int idx = snap->core->core->currentWrite; \
|
|
snap->core->core->currentWrite = M_GENINT_ERROR; \
|
|
m_snapsh0t_mrsw_write_end(snap->core->core, idx); \
|
|
} \
|
|
\
|
|
M_INLINE void \
|
|
M_F(name, _clear)(snapshot_t snap) \
|
|
{ \
|
|
M_F(name, _mrsw_clear)(snap->core); \
|
|
} \
|
|
\
|
|
M_INLINE type * \
|
|
M_F(name, _write_start)(snapshot_t snap) \
|
|
{ \
|
|
M_SNAPSH0T_SPMC_CONTRACT(snap->core); \
|
|
const unsigned int idx = m_snapsh0t_mrsw_write_start(snap->core->core); \
|
|
return &snap->core->data[idx].x; \
|
|
} \
|
|
\
|
|
M_INLINE void \
|
|
M_F(name, _write_end)(snapshot_t snap, type *old) \
|
|
{ \
|
|
M_SNAPSH0T_SPMC_CONTRACT(snap->core); \
|
|
const M_F(name, _mrsw_aligned_type_ct) *oldx; \
|
|
oldx = M_CTYPE_FROM_FIELD(M_F(name, _mrsw_aligned_type_ct), old, type, x); \
|
|
M_ASSERT (oldx >= snap->core->data); \
|
|
M_ASSERT (oldx < snap->core->data + snap->core->core->n_reader + M_SNAPSH0T_SPMC_EXTRA_BUFFER); \
|
|
M_ASSERT(snap->core->core->n_reader + M_SNAPSH0T_SPMC_EXTRA_BUFFER < UINT_MAX); \
|
|
const unsigned int idx = (unsigned int) (oldx - snap->core->data); \
|
|
m_snapsh0t_mrsw_write_end(snap->core->core, idx); \
|
|
} \
|
|
\
|
|
M_INLINE type const * \
|
|
M_F(name, _read_start)(snapshot_t snap) \
|
|
{ \
|
|
return M_F(name, _mrsw_read_start)(snap->core); \
|
|
} \
|
|
\
|
|
M_INLINE void \
|
|
M_F(name, _read_end)(snapshot_t snap, type const *old) \
|
|
{ \
|
|
M_F(name, _mrsw_read_end)(snap->core, old); \
|
|
} \
|
|
\
|
|
|
|
//FIXME: Evaluate the needs for the methods _set_, _init_set.
|
|
|
|
M_END_PROTECTED_CODE
|
|
|
|
/********************************** INTERNAL *********************************/
|
|
|
|
#if M_USE_SMALL_NAME
|
|
#define SNAPSHOT_SPSC_DEF M_SNAPSHOT_SPSC_DEF
|
|
#define SNAPSHOT_SPSC_DEF_AS M_SNAPSHOT_SPSC_DEF_AS
|
|
#define SNAPSHOT_SPMC_DEF M_SNAPSHOT_SPMC_DEF
|
|
#define SNAPSHOT_SPMC_DEF_AS M_SNAPSHOT_SPMC_DEF_AS
|
|
#define SNAPSHOT_MPMC_DEF M_SNAPSHOT_MPMC_DEF
|
|
#define SNAPSHOT_MPMC_DEF_AS M_SNAPSHOT_MPMC_DEF_AS
|
|
#define SNAPSHOT_OPLIST M_SNAPSHOT_OPLIST
|
|
#endif
|
|
|
|
#endif
|