From 10d918bbc9dd9c133b8b42f6b35626fa30a9a953 Mon Sep 17 00:00:00 2001 From: Patrick Pelissier Date: Wed, 26 Mar 2025 23:57:41 +0100 Subject: [PATCH] Port to new memory model (add global context 0) --- example/ex11-use-pool.c | 5 +++++ m-core.h | 17 +++++++++++++++++ m-worker.h | 27 ++++++++++++++++++--------- 3 files changed, 40 insertions(+), 9 deletions(-) diff --git a/example/ex11-use-pool.c b/example/ex11-use-pool.c index 382cb816..4b2b34b6 100644 --- a/example/ex11-use-pool.c +++ b/example/ex11-use-pool.c @@ -140,6 +140,11 @@ M_TUPLE_DEF2(tuple2, (num, int), (name, string_t, M_STRING_POOL_OPLIST) ) M_VARIANT_DEF2(variant1, (num, int) ) M_VARIANT_DEF2(variant2, (num, int), (name, string_t, M_STRING_POOL_OPLIST) ) +#include "m-worker.h" + +M_WORKER_SPAWN_DEF2(worker_int, (in, int) ) +M_WORKER_SPAWN_DEF2(worker_str, (str, string_t, M_STRING_POOL_OPLIST) ) + int main(void) { return 0; diff --git a/m-core.h b/m-core.h index 0d894e68..5fdecb70 100644 --- a/m-core.h +++ b/m-core.h @@ -413,6 +413,21 @@ M_BEGIN_PROTECTED_CODE // FIXME: Rename to M_USE_MEMORY_CONTEXT? +/* In order to support context for memory functions, the prototypes and usages of some functions are modified + * so that they accept a first argument 'm_pool' which is the memory context parameter. + * For some functions, there is no meaning of a local memory context parameter: for all data shared between threads, + * only a global memory context has any meaning. For such cases, the memory context parameter is not requested to + * the used but set to 0. + * several macros are defined **globally**. Theses macros expand to nothing if memory context is not requested. + * * M_P_EXPAND: expand to the first parameter 'm_pool' of type M_USE_POOL for a function that may use memory context. + * * M_P_EXPAND_void: same but assume that there is no argument + * * M_R_EXPAND: expand to the first parameter 'm_pool' + * * M_R_EXPAND_void: same but assume that there is no argument + * * M_R: add the parameter m_pool to the function call if needed, + * * M_P: define the prototype of a function that may have the parameter m_pool. + * * M_ASSERT_POOL: avoid warning of unused m_pool argument + * * M_GLOBAL_POOL: define m_pool to zero, meaning it is a global memory context. + */ #ifndef M_USE_POOL // Memory pool parameter not used for memory function. Stub it with 0 as it won't be used (global definition). # define M_P_EXPAND @@ -421,6 +436,7 @@ M_BEGIN_PROTECTED_CODE # define M_R_EXPAND_void # define M_R # define M_ASSERT_POOL() +# define M_GLOBAL_POOL() #else // Memory pool parameter is used for function using memory. (local definition to the function). // Expand to a prototype to a function that may not trigger exception @@ -430,6 +446,7 @@ M_BEGIN_PROTECTED_CODE # define M_R_EXPAND_void m_pool # define M_R(...) (m_pool, __VA_ARGS__) # define M_ASSERT_POOL() (void) m_pool +# define M_GLOBAL_POOL() M_USE_POOL m_pool = { 0 } #endif // M_P: Expand to a prototype of a function that may trigger exception / accept memory context diff --git a/m-worker.h b/m-worker.h index e8d58d70..c70a0d1f 100644 --- a/m-worker.h +++ b/m-worker.h @@ -235,6 +235,7 @@ typedef struct m_worker_sync_s { M_INLINE void \ M_C3(m_work3r_, name, _clear)(struct M_C3(m_worker_, name, _s) *p) \ { \ + M_GLOBAL_POOL(); \ M_MAP3(M_WORK3R_SPAWN_EXTEND_DEF_CALLBACK_CLEAR, data, __VA_ARGS__) \ /* TODO: Overload */ \ M_MEMORY_DEL(p); \ @@ -263,6 +264,7 @@ typedef struct m_worker_sync_s { M_MAP3_C(M_WORK3R_SPAWN_EXTEND_DEF_EMPLACE_FIELD, data, __VA_ARGS__) \ ) \ { \ + M_GLOBAL_POOL(); \ if (!m_work3r_queue_full_p(block->worker->queue_g)) { \ struct M_C3(m_worker_, name, _s) *p = M_MEMORY_ALLOC ( struct M_C3(m_worker_, name, _s)); \ if (M_UNLIKELY_NOMEM(p == NULL)) { \ @@ -271,7 +273,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 (block->worker->queue_g, w) == true) { \ + if (m_work3r_queue_push M_R(block->worker->queue_g, w) == true) { \ atomic_fetch_add (&block->num_spawn, 1); \ return; \ } \ @@ -361,6 +363,7 @@ m_work3r_thread(void *arg) { // Get back the given argument struct m_worker_s *g = M_ASSIGN_CAST(struct m_worker_s *, arg); + M_GLOBAL_POOL(); while (true) { m_work3r_order_ct w; // If needed, reset the global state of the worker @@ -369,7 +372,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(&w, g->queue_g); + m_work3r_queue_pop M_R(&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 @@ -409,11 +412,12 @@ m_worker_init(m_worker_t g, int numWorker, unsigned int extraQueue, void (*reset // numWorker can still be 0 if it is a single core cpu (no worker available) M_ASSERT(numWorker >= 0); size_t numWorker_st = (size_t) numWorker; + M_GLOBAL_POOL(); g->worker = M_MEMORY_REALLOC(m_work3r_thread_ct, NULL, 0, numWorker_st); if (M_UNLIKELY_NOMEM (g->worker == NULL)) { M_MEMORY_FULL(m_work3r_thread_ct, numWorker_st); } - m_work3r_queue_init(g->queue_g, numWorker_st + extraQueue); + m_work3r_queue_init M_R(g->queue_g, numWorker_st + extraQueue); g->numWorker_g = (unsigned int) numWorker_st; g->resetFunc_g = resetFunc; g->clearFunc_g = clearFunc; @@ -440,13 +444,14 @@ M_INLINE void m_worker_clear(m_worker_t g) { M_ASSERT (m_work3r_queue_empty_p (g->queue_g)); + M_GLOBAL_POOL(); // Push the terminate order on the queue for(unsigned int i = 0; i < g->numWorker_g; i++) { m_work3r_order_ct w = M_WORK3R_EMPTY_ORDER; // 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 (g->queue_g, w, true); + m_work3r_queue_push_blocking M_R(g->queue_g, w, true); } // Wait for thread termination for(unsigned int i = 0; i < g->numWorker_g; i++) { @@ -456,7 +461,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(g->queue_g); + m_work3r_queue_clear M_R(g->queue_g); } /* Start a new collaboration between workers of pool 'g' @@ -478,8 +483,9 @@ M_INLINE void 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 (block->worker->queue_g, w) == true) { + && m_work3r_queue_push M_R(block->worker->queue_g, w) == 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; @@ -496,8 +502,9 @@ M_INLINE void 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 (block->worker->queue_g, w) == true) { + && m_work3r_queue_push M_R(block->worker->queue_g, w) == 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; @@ -515,8 +522,9 @@ M_INLINE void m_work3r_spawn_function(m_worker_sync_t block, std::function func, 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 (block->worker->queue_g, w) == true) { + && m_work3r_queue_push M_R(block->worker->queue_g, w) == 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; @@ -557,7 +565,8 @@ M_INLINE void m_worker_flush(m_worker_t g) { m_work3r_order_ct w; - while (m_work3r_queue_pop_blocking (&w, g->queue_g, false) == true) { + M_GLOBAL_POOL(); + while (m_work3r_queue_pop_blocking M_R(&w, g->queue_g, false) == true) { m_work3r_exec(&w); m_work3r_queue_pop_release(g->queue_g); }