diff --git a/bench/nucleotide/nucleotide.c b/bench/nucleotide/nucleotide.c index 94b6d599..f1028a08 100644 --- a/bench/nucleotide/nucleotide.c +++ b/bench/nucleotide/nucleotide.c @@ -139,21 +139,21 @@ static void compute_hash(dict_oligonucleotide_t hash_Table, const polynucleotide } worker_sync_t block; // Define a synchronization block - worker_start(block); + worker_start(block, worker); dict_oligonucleotide_t hash_Table3, *p_hash_Table3 = &hash_Table3; dict_oligonucleotide_t hash_Table2, *p_hash_Table2 = &hash_Table2; dict_oligonucleotide_t hash_Table1, *p_hash_Table1 = &hash_Table1; - WORKER_SPAWN(worker, block, (p_hash_Table1, size, tab, desiredLength), { + WORKER_SPAWN(block, (p_hash_Table1, size, tab, desiredLength), { init_hash(*p_hash_Table1, size, tab, desiredLength, 3, 4); }, (/*no output*/)); - WORKER_SPAWN(worker, block, (p_hash_Table2, size, tab, desiredLength), { + WORKER_SPAWN(block, (p_hash_Table2, size, tab, desiredLength), { init_hash(*p_hash_Table2, size, tab, desiredLength, 2, 4); }, (/*no output*/)); - WORKER_SPAWN(worker, block, (p_hash_Table3, size, tab, desiredLength), { + WORKER_SPAWN(block, (p_hash_Table3, size, tab, desiredLength), { init_hash(*p_hash_Table3, size, tab, desiredLength, 1, 4); }, (/*no output*/)); init_hash(hash_Table, size, tab, desiredLength, 0, 4); - worker_sync(worker, block); + worker_sync(block); // Merge all hashtables dict_oligonucleotide_splice(hash_Table, hash_Table1); @@ -237,24 +237,24 @@ int main() // Do the following functions in parallel. worker_sync_t block; // Define a synchronization block - worker_start(block); + worker_start(block, worker); - WORKER_SPAWN(worker, block, (ptr, output), { + WORKER_SPAWN(block, (ptr, output), { compute_count(*ptr, "GGTATTTTAATTTATAGT", output[6]); }, (/*no output*/)); - WORKER_SPAWN(worker, block, (ptr, output), { + WORKER_SPAWN(block, (ptr, output), { compute_count(*ptr, "GGTATTTTAATT", output[5]); }, (/*no output*/)); - WORKER_SPAWN(worker, block, (ptr, output), { + WORKER_SPAWN(block, (ptr, output), { compute_count(*ptr, "GGTATT", output[4]); }, (/*no output*/)); - WORKER_SPAWN(worker, block, (ptr, output), { + WORKER_SPAWN(block, (ptr, output), { compute_count(*ptr, "GGTA", output[3]); }, (/*no output*/)); compute_count(*ptr, "GGT", output[2]); compute_freq(*ptr, 2, output[1]); compute_freq(p, 1, output[0]); - worker_sync(worker, block); + worker_sync(block); // Output the results to stdout. for(int i=0; i set(*_get_at(a,i),x) - * DICT SET _set_at method has been renamed as _push + * Rename DICT SET _set_at method as _push Since there is no key, it is a push method, not a _set_at method. - * DICT_SET_DEF2 ==> DICT_SET_DEF + * Rename DICT_SET_DEF2 as DICT_SET_DEF since it only requires one argument. - * ALGO _contains ==> _contain_p + * Rename ALGO _contains as _contain_p since it is a predicate '_p' and it shall be in infinitive form. - * Rename algo _map into _for_each (Standard name). + * Rename algo _map as _for_each (Standard name). diff --git a/m-worker.h b/m-worker.h index bc9818b3..89f2d147 100644 --- a/m-worker.h +++ b/m-worker.h @@ -108,12 +108,6 @@ typedef struct worker_thread_s { m_thread_t id; } worker_thread_t; -/* This type defines a synchronization point for workers */ -typedef struct worker_sync_s { - atomic_int num_spawn; - atomic_int num_terminated_spawn; -} worker_sync_t[1]; - /* Let's define the queue which will manage all work orders */ BUFFER_DEF(worker_queue, worker_order_t, 0, BUFFER_QUEUE|BUFFER_UNBLOCKING_PUSH|BUFFER_BLOCKING_POP|BUFFER_THREAD_SAFE|BUFFER_DEFERRED_POP, M_POD_OPLIST) @@ -137,6 +131,13 @@ typedef struct worker_s { } worker_t[1]; +/* This type defines a synchronization point for workers */ +typedef struct worker_sync_s { + atomic_int num_spawn; + atomic_int num_terminated_spawn; + struct worker_s *worker; +} worker_sync_t[1]; + /* Return the number of CPU of the system */ static inline int workeri_get_cpu_count(void) @@ -266,20 +267,21 @@ worker_clear(worker_t g) /* Start a new collaboration between workers and define the synchronization point 'block' */ static inline void -worker_start(worker_sync_t block) +worker_start(worker_sync_t block, worker_t g) { atomic_init (&block->num_spawn, 0); atomic_init (&block->num_terminated_spawn, 0); + block->worker = g; } /* Spawn or not the given work order to workers, or do it ourself if no worker is available */ static inline void -worker_spawn(worker_t g, worker_sync_t block, void (*func)(void *data), void *data) +worker_spawn(worker_sync_t block, void (*func)(void *data), void *data) { const worker_order_t w = { block, data, func WORKER_EXTRA_ORDER }; - if (M_UNLIKELY (!worker_queue_full_p(g->queue_g)) - && worker_queue_push (g->queue_g, w) == true) { + if (M_UNLIKELY (!worker_queue_full_p(block->worker->queue_g)) + && worker_queue_push (block->worker->queue_g, w) == true) { //printf ("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; @@ -293,11 +295,11 @@ worker_spawn(worker_t g, worker_sync_t block, void (*func)(void *data), void *da /* Spawn or not the given work order to workers, or do it ourself if no worker is available */ static inline void -worker_spawn_block(worker_t g, worker_sync_t block, void (^func)(void *data), void *data) +worker_spawn_block(worker_sync_t block, void (^func)(void *data), void *data) { const worker_order_t w = { block, data, NULL, func }; - if (M_UNLIKELY (!worker_queue_full_p(g->queue_g)) - && worker_queue_push (g->queue_g, w) == true) { + if (M_UNLIKELY (!worker_queue_full_p(block->worker->queue_g)) + && worker_queue_push (block->worker->queue_g, w) == true) { //printf ("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; @@ -312,11 +314,11 @@ worker_spawn_block(worker_t g, worker_sync_t block, void (^func)(void *data), vo /* Spawn or not the given work order to workers, or do it ourself if no worker is available */ static inline void -worker_spawn_function(worker_t g, worker_sync_t block, std::function func, void *data) +worker_spawn_function(worker_sync_t block, std::function func, void *data) { const worker_order_t w = { block, data, NULL, func }; - if (M_UNLIKELY (!worker_queue_full_p(g->queue_g)) - && worker_queue_push (g->queue_g, w) == true) { + if (M_UNLIKELY (!worker_queue_full_p(block->worker->queue_g)) + && worker_queue_push (block->worker->queue_g, w) == true) { //printf ("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; @@ -339,15 +341,15 @@ worker_sync_p(worker_sync_t block) /* Wait for all work orders to be finished */ static inline void -worker_sync(worker_t g, worker_sync_t block) +worker_sync(worker_sync_t block) { //printf ("Waiting for thread terminasion.\n"); if (worker_sync_p(block)) return; - m_mutex_lock(g->lock); + m_mutex_lock(block->worker->lock); while (!worker_sync_p(block)) { - m_cond_wait(g->a_thread_ends, g->lock); + m_cond_wait(block->worker->a_thread_ends, block->worker->lock); } - m_mutex_unlock(g->lock); + m_mutex_unlock(block->worker->lock); } /* Flush any work order in the queue if some remains.*/ @@ -376,19 +378,19 @@ worker_count(worker_t g) 'output' is the list of output variables of the 'core' block within "( )" Output variables are only available after a synchronisation block. */ #if WORKER_USE_CLANG_BLOCK -#define WORKER_SPAWN(_worker, _block, _input, _core, _output) \ +#define WORKER_SPAWN(_block, _input, _core, _output) \ WORKER_DEF_DATA(_input, _output) \ WORKER_DEF_SUBBLOCK(_input, _output, _core) \ - worker_spawn_block ((_worker), (_block), WORKER_SPAWN_SUBFUNC_NAME, &WORKER_SPAWN_DATA_NAME) + worker_spawn_block ((_block), WORKER_SPAWN_SUBFUNC_NAME, &WORKER_SPAWN_DATA_NAME) #elif WORKER_USE_CPP_FUNCTION // TODO: Explicit pass all arguments by reference. -#define WORKER_SPAWN(_worker, _block, _input, _core, _output) \ - worker_spawn_function ((_worker), (_block), [&](void *param) {(void)param ; _core } , NULL) +#define WORKER_SPAWN(_block, _input, _core, _output) \ + worker_spawn_function ((_block), [&](void *param) {(void)param ; _core } , NULL) #else -#define WORKER_SPAWN(_worker, _block, _input, _core, _output) \ +#define WORKER_SPAWN(_block, _input, _core, _output) \ WORKER_DEF_DATA(_input, _output) \ WORKER_DEF_SUBFUNC(_input, _output, _core) \ - worker_spawn ((_worker), (_block), WORKER_SPAWN_SUBFUNC_NAME, &WORKER_SPAWN_DATA_NAME) + worker_spawn ((_block), WORKER_SPAWN_SUBFUNC_NAME, &WORKER_SPAWN_DATA_NAME) #endif #define WORKER_SPAWN_STRUCT_NAME M_C(worker_data_s_, __LINE__) @@ -467,11 +469,11 @@ typedef struct worker_s { #define worker_init(g, numWorker, extraQueue, resetFunc) do { (void) g; } while (0) #define worker_clear(g) do { (void) g; } while (0) -#define worker_start(b) do { (void) b; } while (0) -#define worker_spawn(w, b, f, d) do { f(d); } while (0) +#define worker_start(b, w) do { (void) b; } while (0) +#define worker_spawn(b, f, d) do { f(d); } while (0) #define worker_sync(b) do { (void) b; } while (0) #define worker_count(w) 1 -#define WORKER_SPAWN(w, b, i, c, o) do { c } while (0) +#define WORKER_SPAWN(b, i, c, o) do { c } while (0) #endif /* WORKER_DISABLE */ diff --git a/tests/test-mworker.c b/tests/test-mworker.c index b1078068..d9997b3e 100644 --- a/tests/test-mworker.c +++ b/tests/test-mworker.c @@ -48,11 +48,11 @@ static int fib(int n) struct fib2_s f; worker_sync_t b; - worker_start(b); + worker_start(b, w_g); f.n = n - 2; - worker_spawn (w_g, b, subfunc_1, &f); + worker_spawn (b, subfunc_1, &f); int y = fib (n-1); - worker_sync(w_g, b); + worker_sync(b); return f.x + y; } @@ -90,10 +90,10 @@ static int fib2(int n) worker_sync_t b; int y1, y2; - worker_start(b); - WORKER_SPAWN(w_g, b, (n), { y2 = fib2(n-2); }, (y2)); + worker_start(b, w_g); + WORKER_SPAWN(b, (n), { y2 = fib2(n-2); }, (y2)); y1 = fib (n-1); - worker_sync(w_g, b); + worker_sync(b); return y1 + y2; }