diff --git a/m-worker.h b/m-worker.h new file mode 100644 index 00000000..5c200720 --- /dev/null +++ b/m-worker.h @@ -0,0 +1,325 @@ +/* + * M*LIB / WORKER - Extra worker interface + * + * Copyright (c) 2017, 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 __M_WORKER_H +#define __M_WORKER_H + +#include "m-atomic.h" +#include "m-buffer.h" +#include "m-mutex.h" + +#if defined(_WIN32) +# include +#elif (defined(__APPLE__) && defined(__MACH__)) \ + || defined(__DragonFly__) || defined(__FreeBSD__) \ + || defined(__NetBSD__) || defined(__OpenBSD__) +# include +# include +# define USE_SYSCTL +#else +# include +#endif + +/* Support for CLANG block since CLANG doesn't support nested function. + M-WORKER uses its 'blocks' extension instead, but it is not compatible + with function. + So you need to compile with "-fblocks" and link with "-lBlocksRuntime" + if you use clang & want to use the MACRO version. +*/ +#if defined(__has_extension) && !defined(WORKER_CLANG_BLOCK) +# if __has_extension(blocks) +# define WORKER_CLANG_BLOCK 1 +# endif +#endif +#ifndef WORKER_CLANG_BLOCK +# define WORKER_CLANG_BLOCK 0 +#endif + +/* This type defines a work order */ +typedef struct { + struct worker_block_s *block; + void * data; + void (*func) (void *data); +#if WORKER_CLANG_BLOCK + void (^blockFunc)(void *data); +#endif +} worker_order_t; + +/* This type defines a worker */ +typedef struct { + m_thread_t id; +} worker_thread_t; + +/* This type defines a synchronization point for workers */ +typedef struct worker_block_s { + atomic_int num_spawn; + atomic_int num_terminated_spawn; +} worker_block_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) + +/* This type defines the internal state of the workers */ +typedef struct worker_internal_s { + /* The work order queue */ + worker_queue_t queue_g; + + /* The table of workers */ + worker_thread_t *worker; + unsigned int numWorker_g; + + /* The global reset function */ + void (*resetFunc_g)(void); +} worker_t; + +/* The global state. Shared for all */ +worker_t worker_internal_g; + +/* Return the number of CPU of the system */ +static inline int +workeri_get_cpu_count(void) +{ +#if defined(_WIN32) + SYSTEM_INFO sysinfo; + GetSystemInfo(&sysinfo); + return sysinfo.dwNumberOfProcessors; +#elif defined(USE_SYSCTL) + int nm[2]; + int count = 0; + size_t len = sizeof (count); + nm[0] = CTL_HW; + nm[1] = HW_NCPU; + sysctl(nm, 2, &count, &len, NULL, 0); + return MAX(1, count); +#elif defined (_SC_NPROCESSORS_ONLN) + return sysconf(_SC_NPROCESSORS_ONLN); +#else + return 1; +#endif +} + +/* The worker thread */ +static inline void +workeri_thread(void *arg) +{ + (void) arg; + while (true) { + worker_order_t w; + if (worker_internal_g.resetFunc_g != NULL) + worker_internal_g.resetFunc_g(); + //printf ("Waiting for data (queue: %lu / %lu)\n", worker_queue_size(worker_internal_g.queue_g), worker_queue_capacity(worker_internal_g.queue_g)); + worker_queue_pop(&w, worker_internal_g.queue_g); + if (w.block == NULL) return; + //printf ("Starting thread with data %p\n", w.data); +#if WORKER_CLANG_BLOCK + //printf ("Running %s f=%p b=%p\n", (w.func == NULL) ? "Blocks" : "Function", w.func, w.blockFunc); + if (w.func == NULL) + w.blockFunc(w.data); + else +#endif + w.func(w.data); + worker_queue_pop_release(worker_internal_g.queue_g); + atomic_fetch_add (&w.block->num_terminated_spawn, 1); + } +} + +/* Initialization of the worker module + Input: + @numWorker: number of worker to create + @extraQueue: number of extra work order we can get if all workers are full + @resetFunc: function to reset the state of a worker between work orders (or NULL if none) +*/ +static inline void +worker_init(unsigned int numWorker, unsigned int extraQueue, void (*resetFunc)(void)) +{ + if (numWorker == 0) + numWorker = workeri_get_cpu_count()-1; + //printf ("Starting queue with: %d\n", numWorker + extraQueue); + worker_queue_init(worker_internal_g.queue_g, numWorker + extraQueue); + size_t numWorker_st = numWorker; + worker_internal_g.worker = M_MEMORY_REALLOC(worker_thread_t, NULL, numWorker_st); + M_ASSERT_INIT (worker_internal_g.worker != NULL); + for(size_t i = 0; i < numWorker; i++) { + m_thread_create(worker_internal_g.worker[i].id, workeri_thread, NULL); + } + worker_internal_g.numWorker_g = numWorker; + worker_internal_g.resetFunc_g = resetFunc; +} + +/* Clear of the worker module */ +static inline void +worker_clear(void) +{ + for(unsigned int i = 0; i < worker_internal_g.numWorker_g; i++) { + worker_order_t w = { NULL, NULL, NULL}; + worker_queue_push (worker_internal_g.queue_g, w); + } + + for(unsigned int i = 0; i < worker_internal_g.numWorker_g; i++) { + m_thread_join(worker_internal_g.worker[i].id); + } + M_MEMORY_FREE(worker_internal_g.worker); + + worker_queue_clear(worker_internal_g.queue_g); +} + +/* Start a new collaboration between workers + and define the synchronization point 'block' */ +static inline void +worker_start(worker_block_t block) +{ + atomic_init (&block->num_spawn, 0); + atomic_init (&block->num_terminated_spawn, 0); +} + +/* Spawn or not the given work order to workers, + or do it ourself if no worker is available */ +static inline void +worker_spawn(worker_block_t block, void (*func)(void *data), void *data) +{ + const worker_order_t w = { block, data, func }; + if (M_UNLIKELY (!worker_queue_full_p(worker_internal_g.queue_g)) + && worker_queue_push (worker_internal_g.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; + } + //printf ("Running data ourself: %p\n", data); + /* No thread available. Call the function ourself */ + (*func) (data); +} + +#if WORKER_CLANG_BLOCK +/* 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_block_t block, void (^func)(void *data), void *data) +{ + const worker_order_t w = { block, data, NULL, func }; + if (M_UNLIKELY (!worker_queue_full_p(worker_internal_g.queue_g)) + && worker_queue_push (worker_internal_g.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; + } + //printf ("Running data ourself as block: %p\n", data); + /* No thread available. Call the function ourself */ + func (data); +} +#endif + +/* Wait for all work orders to be finished */ +static inline void +worker_sync(worker_block_t block) +{ + //printf ("Waiting for thread terminasion.\n"); + /* If the number of spawns is greated than the number + of terminated spawns, some spawns are still working. + So wait for terminaison */ + while (atomic_load(&block->num_spawn) != atomic_load (&block->num_terminated_spawn)); +} + +/* Return the number of workers */ +static inline size_t +worker_count(void) +{ + return worker_internal_g.numWorker_g + 1; +} + +/* Spawn the 'core' block computation into another thread if + a worker thread is available. Compute it in the current thread otherwise. + 'block' shall be the initialised synchronised block for all threads. + 'input' is the list of input variables of the 'core' block within "( )" + 'output' is the list of output variables of the 'core' block within "( )" + Output variables are only available after a synchronisation block. */ +#if WORKER_CLANG_BLOCK +#define WORKER_SPAWN(_block, _input, _core, _output) \ + WORKER_DEF_DATA(_input, _output) \ + WORKER_DEF_SUBBLOCK(_input, _output, _core) \ + worker_spawn_block ((_block), WORKER_SPAWN_SUBFUNC_NAME, &WORKER_SPAWN_DATA_NAME) +#else +#define WORKER_SPAWN(_block, _input, _core, _output) \ + WORKER_DEF_DATA(_input, _output) \ + WORKER_DEF_SUBFUNC(_input, _output, _core) \ + worker_spawn ((_block), WORKER_SPAWN_SUBFUNC_NAME, &WORKER_SPAWN_DATA_NAME) +#endif +#define WORKER_SPAWN_STRUCT_NAME M_C(worker_data_s_, __LINE__) +#define WORKER_SPAWN_DATA_NAME M_C(worker_data_, __LINE__) +#define WORKER_SPAWN_SUBFUNC_NAME M_C(worker_subfunc_, __LINE__) +#define WORKER_DEF_DATA(_input, _output) \ + struct WORKER_SPAWN_STRUCT_NAME { \ + WORKER_DEF_DATA_INPUT _input \ + M_IF_EMPTY _output ( , WORKER_DEF_DATA_OUTPUT _output) \ + } WORKER_SPAWN_DATA_NAME = { \ + WORKER_INIT_DATA_INPUT _input \ + M_IF_EMPTY _output (, WORKER_INIT_DATA_OUTPUT _output) \ + }; +#define WORKER_DEF_SINGLE_INPUT(var) __typeof__(var) var; +#define WORKER_DEF_DATA_INPUT(...) \ + M_MAP(WORKER_DEF_SINGLE_INPUT, __VA_ARGS__) +#define WORKER_DEF_SINGLE_OUTPUT(var) \ + __typeof__(var) *M_C(var, __ptr); +#define WORKER_DEF_DATA_OUTPUT(...) \ + M_MAP(WORKER_DEF_SINGLE_OUTPUT, __VA_ARGS__) +#define WORKER_INIT_SINGLE_INPUT(var) \ + .var = var, +#define WORKER_INIT_DATA_INPUT(...) \ + M_MAP(WORKER_INIT_SINGLE_INPUT, __VA_ARGS__) +#define WORKER_INIT_SINGLE_OUTPUT(var) \ + .M_C(var, __ptr) = &var, +#define WORKER_INIT_DATA_OUTPUT(...) \ + M_MAP(WORKER_INIT_SINGLE_OUTPUT, __VA_ARGS__) +#define WORKER_DEF_SUBFUNC(_input, _output, _core) \ + auto void WORKER_SPAWN_SUBFUNC_NAME(void *) ; \ + void WORKER_SPAWN_SUBFUNC_NAME(void *_data) \ + { \ + struct WORKER_SPAWN_STRUCT_NAME *_s_data = _data ; \ + WORKER_INIT_LOCAL_INPUT _input \ + M_IF_EMPTY _output ( , WORKER_INIT_LOCAL_OUTPUT _output) \ + do { _core } while (0); \ + M_IF_EMPTY _output ( , WORKER_PROPAGATE_LOCAL_OUTPUT _output) \ + }; +#define WORKER_DEF_SUBBLOCK(_input, _output, _core) \ + void (^WORKER_SPAWN_SUBFUNC_NAME) (void *) = ^ void (void * _data) \ + { \ + struct WORKER_SPAWN_STRUCT_NAME *_s_data = _data ; \ + WORKER_INIT_LOCAL_INPUT _input \ + M_IF_EMPTY _output ( , WORKER_INIT_LOCAL_OUTPUT _output) \ + do { _core } while (0); \ + M_IF_EMPTY _output ( , WORKER_PROPAGATE_LOCAL_OUTPUT _output) \ + }; +#define WORKER_INIT_SINGLE_LOCAL_INPUT(var) \ + __typeof__(var) var = _s_data->var; +#define WORKER_INIT_LOCAL_INPUT(...) \ + M_MAP(WORKER_INIT_SINGLE_LOCAL_INPUT, __VA_ARGS__) +#define WORKER_INIT_SINGLE_LOCAL_OUTPUT(var) \ + __typeof__(var) var; +#define WORKER_INIT_LOCAL_OUTPUT(...) \ + M_MAP(WORKER_INIT_SINGLE_LOCAL_OUTPUT, __VA_ARGS__) +#define WORKER_PROPAGATE_SINGLE_OUTPUT(var) \ + *(_s_data->M_C(var, __ptr)) = var; +#define WORKER_PROPAGATE_LOCAL_OUTPUT(...) \ + M_MAP(WORKER_PROPAGATE_SINGLE_OUTPUT, __VA_ARGS__) + +#endif