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

Port to new memory model (add global context 0)

Browse Source
This commit is contained in:
Patrick Pelissier committed 2025-05-07 21:02:24 +02:00
1 parent 8069fae3d5
commit 10d918bbc9
3 files changed
+40 -9
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

example/ex11-use-pool.c
+5
View File
@@ -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;
m-core.h
+17
View File
@@ -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
m-worker.h
+18 -9
View File
@@ -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<void(void *data)> 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);
}