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

Update API so that the worker_t parameter is given on creating the sync bloc.

Browse Source 
(API breakage), more RAII like.
This commit is contained in:
Patrick Pelissier committed 2018-07-28 19:09:07 +02:00
1 parent 7678625071
commit 7b5ae3c15b
4 files changed
+57 -54
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

bench/nucleotide/nucleotide.c
+11 -11
View File
@@ -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<MAXMUM_NUMBER_OUTPUT; i++)
doc/API.txt
+9 -8
View File
@@ -1,10 +1,11 @@
Evolution of the API
--------------------
API incompatibility
-------------------
V0.1.*: * First formal version of API
V0.2.*: * worker_sync takes a 'worker_t' as an additional parameter
(This allows to perform a wait-on-event instead of a busy waiting).
V0.2.*: * worker_start takes a 'worker_t' as an additional parameter
(This allows to perform a wait-on-event instead of a busy waiting).
* worker_spawn doesn't take into a 'worker_t' anymore.
* worker_init takes an int as numWorker instead of an unsingned int.
* RBTREE pop method has been renamed into _pop_at
* SNAPSHOT_DEF has been removed. use SNAPSHOT_SPSC_DEF
@@ -12,12 +13,12 @@ V0.2.*: * worker_sync takes a 'worker_t' as an additional parameter
The interface is similar to dict _get_at, making the API cleaner
and more generic.
_set_at2(a,i,x) ==> 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).
m-worker.h
+31 -29
View File
@@ -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<void(void *data)> func, void *data)
worker_spawn_function(worker_sync_t block, std::function<void(void *data)> 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 */
tests/test-mworker.c
+6 -6
View File
@@ -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;
}