workers.h
/* StarPU --- Runtime system for heterogeneous multicore architectures.
*
* Copyright (C) 2008-2021 Université de Bordeaux, CNRS (LaBRI UMR 5800), Inria
* Copyright (C) 2013 Thibaut Lambert
* Copyright (C) 2016 Uppsala University
*
* StarPU is free software; you can redistribute it and/or modify
* it under the terms of the GNU Lesser General Public License as published by
* the Free Software Foundation; either version 2.1 of the License, or (at
* your option) any later version.
*
* StarPU is distributed in the hope that it will be useful, but
* WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
*
* See the GNU Lesser General Public License in COPYING.LGPL for more details.
*/
#ifndef __WORKERS_H__
#define __WORKERS_H__
/* @{ */
#include <limits.h>
#include <starpu.h>
#include <common/config.h>
#include <common/timing.h>
#include <common/fxt.h>
#include <common/thread.h>
#include <common/utils.h>
#include <core/jobs.h>
#include <core/perfmodel/perfmodel.h>
#include <core/sched_policy.h>
#include <core/topology.h>
#include <core/errorcheck.h>
#include <core/sched_ctx.h>
#include <core/sched_ctx_list.h>
#include <core/simgrid.h>
#ifdef STARPU_HAVE_HWLOC
#include <hwloc.h>
#endif
#include <core/drivers.h>
#include <drivers/cuda/driver_cuda.h>
#include <drivers/opencl/driver_opencl.h>
#ifdef STARPU_USE_MIC
#include <drivers/mic/driver_mic_source.h>
#endif /* STARPU_USE_MIC */
#ifdef STARPU_USE_MPI_MASTER_SLAVE
#include <drivers/mpi/driver_mpi_source.h>
#endif
#include <drivers/cpu/driver_cpu.h>
#include <datawizard/datawizard.h>
#include <starpu_parameters.h>
#define STARPU_MAX_PIPELINE 4
enum initialization { UNINITIALIZED = 0, CHANGING, INITIALIZED };
struct _starpu_ctx_change_list;
LIST_TYPE(_starpu_worker,
struct _starpu_machine_config *config;
starpu_pthread_mutex_t mutex;
enum starpu_worker_archtype arch;
uint32_t worker_mask;
struct starpu_perfmodel_arch perf_arch;
starpu_pthread_t worker_thread;
unsigned devid;
unsigned subworkerid;
int bindid;
int workerid;
int combined_workerid;
int current_rank;
int worker_size;
starpu_pthread_cond_t started_cond;
starpu_pthread_cond_t ready_cond;
unsigned memory_node;
unsigned numa_memory_node;
starpu_pthread_cond_t sched_cond;
starpu_pthread_mutex_t sched_mutex;
unsigned state_relax_refcnt;
#ifdef STARPU_SPINLOCK_CHECK
const char *relax_on_file;
int relax_on_line;
const char *relax_on_func;
const char *relax_off_file;
int relax_off_line;
const char *relax_off_func;
#endif
unsigned state_sched_op_pending;
unsigned state_changing_ctx_waiting;
unsigned state_changing_ctx_notice;
unsigned state_blocked_in_parallel;
unsigned state_blocked_in_parallel_observed;
unsigned state_block_in_parallel_req;
unsigned state_block_in_parallel_ack;
unsigned state_unblock_in_parallel_req;
unsigned state_unblock_in_parallel_ack;
unsigned block_in_parallel_ref_count;
starpu_pthread_t thread_changing_ctx;
struct _starpu_ctx_change_list ctx_change_list;
struct starpu_task_list local_tasks;
struct starpu_task **local_ordered_tasks;
unsigned local_ordered_tasks_size;
unsigned current_ordered_task;
unsigned current_ordered_task_order;
struct starpu_task *current_task;
struct starpu_task *current_tasks[STARPU_MAX_PIPELINE];
#ifdef STARPU_SIMGRID
starpu_pthread_wait_t wait;
#endif
struct timespec cl_start;
struct timespec cl_expend;
struct timespec cl_end;
unsigned char first_task;
unsigned char ntasks;
unsigned char pipeline_length;
unsigned char pipeline_stuck;
struct _starpu_worker_set *set;
unsigned worker_is_running;
unsigned worker_is_initialized;
enum _starpu_worker_status status;
unsigned state_keep_awake;
char name[128];
char short_name[32];
unsigned run_by_starpu;
struct _starpu_driver_ops *driver_ops;
struct _starpu_sched_ctx_list *sched_ctx_list;
int tmp_sched_ctx;
unsigned nsched_ctxs;
struct _starpu_barrier_counter tasks_barrier;
unsigned has_prev_init;
unsigned removed_from_ctx[STARPU_NMAX_SCHED_CTXS+1];
unsigned spinning_backoff ;
unsigned nb_buffers_transferred;
unsigned nb_buffers_totransfer;
struct starpu_task *task_transferring;
unsigned shares_tasks_lists[STARPU_NMAX_SCHED_CTXS+1];
unsigned poped_in_ctx[STARPU_NMAX_SCHED_CTXS+1];
unsigned reverse_phase[2];
unsigned pop_ctx_priority;
unsigned is_slave_somewhere;
struct _starpu_sched_ctx *stream_ctx;
#ifdef __GLIBC__
cpu_set_t cpu_set;
#endif /* __GLIBC__ */
#ifdef STARPU_HAVE_HWLOC
hwloc_bitmap_t hwloc_cpu_set;
hwloc_obj_t hwloc_obj;
#endif
char padding[STARPU_CACHELINE_SIZE];
);
struct _starpu_combined_worker
{
struct starpu_perfmodel_arch perf_arch;
uint32_t worker_mask;
int worker_size;
unsigned memory_node;
int combined_workerid[STARPU_NMAXWORKERS];
#ifdef STARPU_USE_MP
int count;
starpu_pthread_mutex_t count_mutex;
#endif
#ifdef __GLIBC__
cpu_set_t cpu_set;
#endif /* __GLIBC__ */
#ifdef STARPU_HAVE_HWLOC
hwloc_bitmap_t hwloc_cpu_set;
#endif
char padding[STARPU_CACHELINE_SIZE];
};
struct _starpu_worker_set
{
starpu_pthread_mutex_t mutex;
starpu_pthread_t worker_thread;
unsigned nworkers;
unsigned started;
void *retval;
struct _starpu_worker *workers;
starpu_pthread_cond_t ready_cond;
unsigned set_is_initialized;
};
#ifdef STARPU_USE_MPI_MASTER_SLAVE
extern struct _starpu_worker_set mpi_worker_set[STARPU_MAXMPIDEVS];
#endif
struct _starpu_machine_topology
{
unsigned nworkers;
unsigned ncombinedworkers;
unsigned nsched_ctxs;
#ifdef STARPU_HAVE_HWLOC
hwloc_topology_t hwtopology;
#endif
struct starpu_tree *tree;
unsigned nhwcpus;
unsigned nhwpus;
unsigned nhwcudagpus;
unsigned nhwopenclgpus;
unsigned nhwmpi;
unsigned ncpus;
unsigned ncudagpus;
unsigned nworkerpercuda;
int cuda_th_per_stream;
int cuda_th_per_dev;
unsigned nopenclgpus;
unsigned nmpidevices;
unsigned nhwmpidevices;
unsigned nhwmpicores[STARPU_MAXMPIDEVS];
unsigned nmpicores[STARPU_MAXMPIDEVS];
unsigned nhwmicdevices;
unsigned nmicdevices;
unsigned nhwmiccores[STARPU_MAXMICDEVS];
unsigned nmiccores[STARPU_MAXMICDEVS];
unsigned workers_bindid[STARPU_NMAXWORKERS];
unsigned workers_nbindid;
unsigned workers_cuda_gpuid[STARPU_NMAXWORKERS];
unsigned workers_opencl_gpuid[STARPU_NMAXWORKERS];
/*** Indicates the successive MIC devices that should be used
* by the MIC driver. It is either filled according to the
* user's explicit parameters (from starpu_conf) or according
* to the STARPU_WORKERS_MICID env. variable. Otherwise, they
* are taken in ID order. */
unsigned workers_mpi_ms_deviceid[STARPU_NMAXWORKERS];
};
struct _starpu_machine_config
{
struct _starpu_machine_topology topology;
#ifdef STARPU_HAVE_HWLOC
int cpu_depth;
int pu_depth;
#endif
int current_bindid;
char currently_bound[STARPU_NMAXWORKERS];
char currently_shared[STARPU_NMAXWORKERS];
int current_cuda_gpuid;
int current_opencl_gpuid;
int current_mic_deviceid;
int current_mpi_deviceid;
int cpus_nodeid;
int cuda_nodeid;
int opencl_nodeid;
int mic_nodeid;
int mpi_nodeid;
char padding1[STARPU_CACHELINE_SIZE];
struct _starpu_worker workers[STARPU_NMAXWORKERS];
struct _starpu_combined_worker combined_workers[STARPU_NMAX_COMBINEDWORKERS];
starpu_pthread_mutex_t submitted_mutex;
char padding2[STARPU_CACHELINE_SIZE];
struct
{
int *workerids;
unsigned nworkers;
} *bindid_workers;
unsigned nbindid;
uint32_t worker_mask;
struct starpu_conf conf;
unsigned running;
int disable_kernels;
int pause_depth;
struct _starpu_sched_ctx sched_ctxs[STARPU_NMAX_SCHED_CTXS+1];
unsigned submitting;
int watchdog_ok;
};
extern int _starpu_worker_parallel_blocks;
extern struct _starpu_machine_config _starpu_config STARPU_ATTRIBUTE_INTERNAL;
extern int _starpu_keys_initialized STARPU_ATTRIBUTE_INTERNAL;
extern starpu_pthread_key_t _starpu_worker_key STARPU_ATTRIBUTE_INTERNAL;
extern starpu_pthread_key_t _starpu_worker_set_key STARPU_ATTRIBUTE_INTERNAL;
void _starpu_set_argc_argv(int *argc, char ***argv);
int *_starpu_get_argc();
char ***_starpu_get_argv();
void _starpu_conf_check_environment(struct starpu_conf *conf);
void _starpu_may_pause(void);
static inline unsigned _starpu_machine_is_running(void)
{
unsigned ret;
/* running is just protected by a memory barrier */
STARPU_RMB();
ANNOTATE_HAPPENS_AFTER(&_starpu_config.running);
ret = _starpu_config.running;
ANNOTATE_HAPPENS_BEFORE(&_starpu_config.running);
return ret;
}
void _starpu_worker_init(struct _starpu_worker *workerarg, struct _starpu_machine_config *pconfig);
uint32_t _starpu_worker_exists(struct starpu_task *);
uint32_t _starpu_can_submit_cuda_task(void);
uint32_t _starpu_can_submit_cpu_task(void);
uint32_t _starpu_can_submit_opencl_task(void);
unsigned _starpu_worker_can_block(unsigned memnode, struct _starpu_worker *worker);
void _starpu_block_worker(int workerid, starpu_pthread_cond_t *cond, starpu_pthread_mutex_t *mutex);
void _starpu_driver_start(struct _starpu_worker *worker, unsigned fut_key, unsigned sync);
void _starpu_worker_start(struct _starpu_worker *worker, unsigned fut_key, unsigned sync);
static inline unsigned _starpu_worker_get_count(void)
{
return _starpu_config.topology.nworkers;
}
#define starpu_worker_get_count _starpu_worker_get_count
static inline void _starpu_set_local_worker_key(struct _starpu_worker *worker)
{
STARPU_ASSERT(_starpu_keys_initialized);
STARPU_PTHREAD_SETSPECIFIC(_starpu_worker_key, worker);
}
static inline struct _starpu_worker *_starpu_get_local_worker_key(void)
{
if (!_starpu_keys_initialized)
return NULL;
return (struct _starpu_worker *) STARPU_PTHREAD_GETSPECIFIC(_starpu_worker_key);
}
static inline void _starpu_set_local_worker_set_key(struct _starpu_worker_set *worker)
{
STARPU_ASSERT(_starpu_keys_initialized);
STARPU_PTHREAD_SETSPECIFIC(_starpu_worker_set_key, worker);
}
static inline struct _starpu_worker_set *_starpu_get_local_worker_set_key(void)
{
if (!_starpu_keys_initialized)
return NULL;
return (struct _starpu_worker_set *) STARPU_PTHREAD_GETSPECIFIC(_starpu_worker_set_key);
}
static inline struct _starpu_worker *_starpu_get_worker_struct(unsigned id)
{
STARPU_ASSERT(id < starpu_worker_get_count());
return &_starpu_config.workers[id];
}
static inline struct _starpu_sched_ctx *_starpu_get_sched_ctx_struct(unsigned id)
{
return (id > STARPU_NMAX_SCHED_CTXS) ? NULL : &_starpu_config.sched_ctxs[id];
}
struct _starpu_combined_worker *_starpu_get_combined_worker_struct(unsigned id);
static inline struct _starpu_machine_config *_starpu_get_machine_config(void)
{
return &_starpu_config;
}
static inline int _starpu_get_disable_kernels(void)
{
return _starpu_config.disable_kernels;
}
static inline enum _starpu_worker_status _starpu_worker_get_status(int workerid)
{
return _starpu_config.workers[workerid].status;
}
static inline void _starpu_worker_set_status(int workerid, enum _starpu_worker_status status)
{
_starpu_config.workers[workerid].status = status;
}
static inline struct _starpu_sched_ctx* _starpu_get_initial_sched_ctx(void)
{
return &_starpu_config.sched_ctxs[STARPU_GLOBAL_SCHED_CTX];
}
int starpu_worker_get_nids_by_type(enum starpu_worker_archtype type, int *workerids, int maxsize);
int starpu_worker_get_nids_ctx_free_by_type(enum starpu_worker_archtype type, int *workerids, int maxsize);
static inline unsigned _starpu_worker_mutex_is_sched_mutex(int workerid, starpu_pthread_mutex_t *mutex)
{
struct _starpu_worker *w = _starpu_get_worker_struct(workerid);
return &w->sched_mutex == mutex;
}
static inline int _starpu_worker_get_nsched_ctxs(int workerid)
{
return _starpu_config.workers[workerid].nsched_ctxs;
}
static inline unsigned _starpu_get_nsched_ctxs(void)
{
/* topology.nsched_ctxs may be increased asynchronously in sched_ctx_create */
STARPU_RMB();
return _starpu_config.topology.nsched_ctxs;
}
static inline int _starpu_worker_get_id(void)
{
struct _starpu_worker * worker;
worker = _starpu_get_local_worker_key();
if (worker)
{
return worker->workerid;
}
else
{
/* there is no worker associated to that thread, perhaps it is
* a thread from the application or this is some SPU worker */
return -1;
}
}
#define starpu_worker_get_id _starpu_worker_get_id
static inline unsigned __starpu_worker_get_id_check(const char *f, int l)
{
(void) l;
(void) f;
int id = starpu_worker_get_id();
STARPU_ASSERT_MSG(id>=0, "%s:%d Cannot be called from outside a worker\n", f, l);
return id;
}
#define _starpu_worker_get_id_check(f,l) __starpu_worker_get_id_check(f,l)
enum starpu_node_kind _starpu_worker_get_node_kind(enum starpu_worker_archtype type);
void _starpu_worker_set_stream_ctx(unsigned workerid, struct _starpu_sched_ctx *sched_ctx);
struct _starpu_sched_ctx* _starpu_worker_get_ctx_stream(unsigned stream_workerid);
static inline void _starpu_worker_request_blocking_in_parallel(struct _starpu_worker * const worker)
{
_starpu_worker_parallel_blocks = 1;
/* flush pending requests to start on a fresh transaction epoch */
while (worker->state_unblock_in_parallel_req)
STARPU_PTHREAD_COND_WAIT(&worker->sched_cond, &worker->sched_mutex);
/* announce blocking intent */
STARPU_ASSERT(worker->block_in_parallel_ref_count < UINT_MAX);
worker->block_in_parallel_ref_count++;
if (worker->block_in_parallel_ref_count == 1)
{
/* only the transition from 0 to 1 triggers the block_in_parallel_req */
STARPU_ASSERT(!worker->state_blocked_in_parallel);
STARPU_ASSERT(!worker->state_block_in_parallel_req);
STARPU_ASSERT(!worker->state_block_in_parallel_ack);
STARPU_ASSERT(!worker->state_unblock_in_parallel_req);
STARPU_ASSERT(!worker->state_unblock_in_parallel_ack);
/* trigger the block_in_parallel_req */
worker->state_block_in_parallel_req = 1;
STARPU_PTHREAD_COND_BROADCAST(&worker->sched_cond);
#ifdef STARPU_SIMGRID
starpu_pthread_queue_broadcast(&_starpu_simgrid_task_queue[worker->workerid]);
#endif
/* wait for block_in_parallel_req to be processed */
while (!worker->state_block_in_parallel_ack)
STARPU_PTHREAD_COND_WAIT(&worker->sched_cond, &worker->sched_mutex);
STARPU_ASSERT(worker->block_in_parallel_ref_count >= 1);
STARPU_ASSERT(worker->state_block_in_parallel_req);
STARPU_ASSERT(worker->state_blocked_in_parallel);
/* reset block_in_parallel_req state flags */
worker->state_block_in_parallel_req = 0;
worker->state_block_in_parallel_ack = 0;
/* broadcast block_in_parallel_req state flags reset */
STARPU_PTHREAD_COND_BROADCAST(&worker->sched_cond);
}
}
static inline void _starpu_worker_request_unblocking_in_parallel(struct _starpu_worker * const worker)
{
/* flush pending requests to start on a fresh transaction epoch */
while (worker->state_block_in_parallel_req)
STARPU_PTHREAD_COND_WAIT(&worker->sched_cond, &worker->sched_mutex);
/* unblocking may be requested unconditionnally
* thus, check is unblocking is really needed */
if (worker->state_blocked_in_parallel)
{
if (worker->block_in_parallel_ref_count == 1)
{
/* only the transition from 1 to 0 triggers the unblock_in_parallel_req */
STARPU_ASSERT(!worker->state_block_in_parallel_req);
STARPU_ASSERT(!worker->state_block_in_parallel_ack);
STARPU_ASSERT(!worker->state_unblock_in_parallel_req);
STARPU_ASSERT(!worker->state_unblock_in_parallel_ack);
/* trigger the unblock_in_parallel_req */
worker->state_unblock_in_parallel_req = 1;
STARPU_PTHREAD_COND_BROADCAST(&worker->sched_cond);
/* wait for the unblock_in_parallel_req to be processed */
while (!worker->state_unblock_in_parallel_ack)
STARPU_PTHREAD_COND_WAIT(&worker->sched_cond, &worker->sched_mutex);
STARPU_ASSERT(worker->state_unblock_in_parallel_req);
STARPU_ASSERT(!worker->state_blocked_in_parallel);
/* reset unblock_in_parallel_req state flags */
worker->state_unblock_in_parallel_req = 0;
worker->state_unblock_in_parallel_ack = 0;
/* broadcast unblock_in_parallel_req state flags reset */
STARPU_PTHREAD_COND_BROADCAST(&worker->sched_cond);
}
/* announce unblocking complete */
STARPU_ASSERT(worker->block_in_parallel_ref_count > 0);
worker->block_in_parallel_ref_count--;
}
}
static inline void _starpu_worker_process_block_in_parallel_requests(struct _starpu_worker * const worker)
{
while (worker->state_block_in_parallel_req)
{
STARPU_ASSERT(!worker->state_blocked_in_parallel);
STARPU_ASSERT(!worker->state_block_in_parallel_ack);
STARPU_ASSERT(!worker->state_unblock_in_parallel_req);
STARPU_ASSERT(!worker->state_unblock_in_parallel_ack);
STARPU_ASSERT(worker->block_in_parallel_ref_count > 0);
/* enter effective blocked state */
worker->state_blocked_in_parallel = 1;
/* notify block_in_parallel_req processing */
worker->state_block_in_parallel_ack = 1;
STARPU_PTHREAD_COND_BROADCAST(&worker->sched_cond);
/* block */
while (!worker->state_unblock_in_parallel_req)
STARPU_PTHREAD_COND_WAIT(&worker->sched_cond, &worker->sched_mutex);
STARPU_ASSERT(worker->state_blocked_in_parallel);
STARPU_ASSERT(!worker->state_block_in_parallel_req);
STARPU_ASSERT(!worker->state_block_in_parallel_ack);
STARPU_ASSERT(!worker->state_unblock_in_parallel_ack);
STARPU_ASSERT(worker->block_in_parallel_ref_count > 0);
/* leave effective blocked state */
worker->state_blocked_in_parallel = 0;
/* notify unblock_in_parallel_req processing */
worker->state_unblock_in_parallel_ack = 1;
STARPU_PTHREAD_COND_BROADCAST(&worker->sched_cond);
}
}
#ifdef STARPU_SPINLOCK_CHECK
static inline void __starpu_worker_enter_sched_op(struct _starpu_worker * const worker, const char*file, int line, const char* func)
#else
static inline void _starpu_worker_enter_sched_op(struct _starpu_worker * const worker)
#endif
{
STARPU_ASSERT(!worker->state_sched_op_pending);
if (!worker->state_blocked_in_parallel_observed)
{
/* process pending block requests before entering a sched_op region */
_starpu_worker_process_block_in_parallel_requests(worker);
while (worker->state_changing_ctx_notice)
{
STARPU_PTHREAD_COND_WAIT(&worker->sched_cond, &worker->sched_mutex);
/* new block requests may have been triggered during the wait,
* need to check again */
_starpu_worker_process_block_in_parallel_requests(worker);
}
}
else
{
/* if someone observed the worker state since the last call, postpone block request
* processing for one sched_op turn more, because the observer will not have seen
* new block requests between its observation and now.
*
* however, the worker still has to wait for context change operations to complete
* before entering sched_op again*/
while (worker->state_changing_ctx_notice)
{
STARPU_PTHREAD_COND_WAIT(&worker->sched_cond, &worker->sched_mutex);
}
}
/* no block request and no ctx change ahead,
* enter sched_op */
worker->state_sched_op_pending = 1;
worker->state_blocked_in_parallel_observed = 0;
worker->state_relax_refcnt = 0;
#ifdef STARPU_SPINLOCK_CHECK
worker->relax_on_file = file;
worker->relax_on_line = line;
worker->relax_on_func = func;
#endif
}
#ifdef STARPU_SPINLOCK_CHECK
#define _starpu_worker_enter_sched_op(worker) __starpu_worker_enter_sched_op((worker), __FILE__, __LINE__, __starpu_func__)
#endif
void _starpu_worker_apply_deferred_ctx_changes(void);
#ifdef STARPU_SPINLOCK_CHECK
static inline void __starpu_worker_leave_sched_op(struct _starpu_worker * const worker, const char*file, int line, const char* func)
#else
static inline void _starpu_worker_leave_sched_op(struct _starpu_worker * const worker)
#endif
{
STARPU_ASSERT(worker->state_sched_op_pending);
worker->state_relax_refcnt = 1;
#ifdef STARPU_SPINLOCK_CHECK
worker->relax_off_file = file;
worker->relax_off_line = line;
worker->relax_off_func = func;
#endif
worker->state_sched_op_pending = 0;
STARPU_PTHREAD_COND_BROADCAST(&worker->sched_cond);
_starpu_worker_apply_deferred_ctx_changes();
}
#ifdef STARPU_SPINLOCK_CHECK
#define _starpu_worker_leave_sched_op(worker) __starpu_worker_leave_sched_op((worker), __FILE__, __LINE__, __starpu_func__)
#endif
static inline int _starpu_worker_sched_op_pending(void)
{
int workerid = starpu_worker_get_id();
if (workerid == -1)
return 0;
struct _starpu_worker *worker = _starpu_get_worker_struct(workerid);
STARPU_ASSERT(worker != NULL);
return worker->state_sched_op_pending;
}
static inline void _starpu_worker_enter_changing_ctx_op(struct _starpu_worker * const worker)
{
STARPU_ASSERT(!starpu_pthread_equal(worker->thread_changing_ctx, starpu_pthread_self()));
/* flush pending requests to start on a fresh transaction epoch */
while (worker->state_changing_ctx_notice)
STARPU_PTHREAD_COND_WAIT(&worker->sched_cond, &worker->sched_mutex);
/* announce changing_ctx intent
*
* - an already started sched_op is allowed to complete
* - no new sched_op may be started
*/
worker->state_changing_ctx_notice = 1;
worker->thread_changing_ctx = starpu_pthread_self();
/* allow for an already started sched_op to complete */
if (worker->state_sched_op_pending)
{
/* request sched_op to broadcast when way is cleared */
worker->state_changing_ctx_waiting = 1;
/* wait for sched_op completion */
STARPU_PTHREAD_COND_BROADCAST(&worker->sched_cond);
#ifdef STARPU_SIMGRID
starpu_pthread_queue_broadcast(&_starpu_simgrid_task_queue[worker->workerid]);
#endif
do
{
STARPU_PTHREAD_COND_WAIT(&worker->sched_cond, &worker->sched_mutex);
}
while (worker->state_sched_op_pending);
/* reset flag so other sched_ops wont have to broadcast state */
worker->state_changing_ctx_waiting = 0;
}
}
static inline void _starpu_worker_leave_changing_ctx_op(struct _starpu_worker * const worker)
{
worker->thread_changing_ctx = (starpu_pthread_t)0;
worker->state_changing_ctx_notice = 0;
STARPU_PTHREAD_COND_BROADCAST(&worker->sched_cond);
}
#ifdef STARPU_SPINLOCK_CHECK
static inline void __starpu_worker_relax_on(const char*file, int line, const char* func)
#else
static inline void _starpu_worker_relax_on(void)
#endif
{
struct _starpu_worker *worker = _starpu_get_local_worker_key();
if (worker == NULL)
return;
if (!worker->state_sched_op_pending)
return;
STARPU_PTHREAD_MUTEX_LOCK_SCHED(&worker->sched_mutex);
#ifdef STARPU_SPINLOCK_CHECK
STARPU_ASSERT_MSG(worker->state_relax_refcnt<UINT_MAX, "relax last turn on in %s (%s:%d)\n", worker->relax_on_func, worker->relax_on_file, worker->relax_on_line);
#else
STARPU_ASSERT(worker->state_relax_refcnt<UINT_MAX);
#endif
worker->state_relax_refcnt++;
#ifdef STARPU_SPINLOCK_CHECK
worker->relax_on_file = file;
worker->relax_on_line = line;
worker->relax_on_func = func;
#endif
STARPU_PTHREAD_COND_BROADCAST(&worker->sched_cond);
STARPU_PTHREAD_MUTEX_UNLOCK_SCHED(&worker->sched_mutex);
}
#ifdef STARPU_SPINLOCK_CHECK
#define _starpu_worker_relax_on() __starpu_worker_relax_on(__FILE__, __LINE__, __starpu_func__)
#endif
#define starpu_worker_relax_on _starpu_worker_relax_on
#ifdef STARPU_SPINLOCK_CHECK
static inline void __starpu_worker_relax_on_locked(struct _starpu_worker *worker, const char*file, int line, const char* func)
#else
static inline void _starpu_worker_relax_on_locked(struct _starpu_worker *worker)
#endif
{
if (!worker->state_sched_op_pending)
return;
#ifdef STARPU_SPINLOCK_CHECK
STARPU_ASSERT_MSG(worker->state_relax_refcnt<UINT_MAX, "relax last turn on in %s (%s:%d)\n", worker->relax_on_func, worker->relax_on_file, worker->relax_on_line);
#else
STARPU_ASSERT(worker->state_relax_refcnt<UINT_MAX);
#endif
worker->state_relax_refcnt++;
#ifdef STARPU_SPINLOCK_CHECK
worker->relax_on_file = file;
worker->relax_on_line = line;
worker->relax_on_func = func;
#endif
STARPU_PTHREAD_COND_BROADCAST(&worker->sched_cond);
}
#ifdef STARPU_SPINLOCK_CHECK
#define _starpu_worker_relax_on_locked(worker) __starpu_worker_relax_on_locked(worker,__FILE__, __LINE__, __starpu_func__)
#endif
#ifdef STARPU_SPINLOCK_CHECK
static inline void __starpu_worker_relax_off(const char*file, int line, const char* func)
#else
static inline void _starpu_worker_relax_off(void)
#endif
{
int workerid = starpu_worker_get_id();
if (workerid == -1)
return;
struct _starpu_worker *worker = _starpu_get_worker_struct(workerid);
STARPU_ASSERT(worker != NULL);
if (!worker->state_sched_op_pending)
return;
STARPU_PTHREAD_MUTEX_LOCK_SCHED(&worker->sched_mutex);
#ifdef STARPU_SPINLOCK_CHECK
STARPU_ASSERT_MSG(worker->state_relax_refcnt>0, "relax last turn off in %s (%s:%d)\n", worker->relax_on_func, worker->relax_on_file, worker->relax_on_line);
#else
STARPU_ASSERT(worker->state_relax_refcnt>0);
#endif
worker->state_relax_refcnt--;
#ifdef STARPU_SPINLOCK_CHECK
worker->relax_off_file = file;
worker->relax_off_line = line;
worker->relax_off_func = func;
#endif
STARPU_PTHREAD_MUTEX_UNLOCK_SCHED(&worker->sched_mutex);
}
#ifdef STARPU_SPINLOCK_CHECK
#define _starpu_worker_relax_off() __starpu_worker_relax_off(__FILE__, __LINE__, __starpu_func__)
#endif
#define starpu_worker_relax_off _starpu_worker_relax_off
#ifdef STARPU_SPINLOCK_CHECK
static inline void __starpu_worker_relax_off_locked(const char*file, int line, const char* func)
#else
static inline void _starpu_worker_relax_off_locked(void)
#endif
{
int workerid = starpu_worker_get_id();
if (workerid == -1)
return;
struct _starpu_worker *worker = _starpu_get_worker_struct(workerid);
STARPU_ASSERT(worker != NULL);
if (!worker->state_sched_op_pending)
return;
#ifdef STARPU_SPINLOCK_CHECK
STARPU_ASSERT_MSG(worker->state_relax_refcnt>0, "relax last turn off in %s (%s:%d)\n", worker->relax_on_func, worker->relax_on_file, worker->relax_on_line);
#else
STARPU_ASSERT(worker->state_relax_refcnt>0);
#endif
worker->state_relax_refcnt--;
#ifdef STARPU_SPINLOCK_CHECK
worker->relax_off_file = file;
worker->relax_off_line = line;
worker->relax_off_func = func;
#endif
}
#ifdef STARPU_SPINLOCK_CHECK
#define _starpu_worker_relax_off_locked() __starpu_worker_relax_off_locked(__FILE__, __LINE__, __starpu_func__)
#endif
static inline int _starpu_worker_get_relax_state(void)
{
int workerid = starpu_worker_get_id();
if (workerid < 0)
return 1;
struct _starpu_worker *worker = _starpu_get_worker_struct(workerid);
STARPU_ASSERT(worker != NULL);
return worker->state_relax_refcnt != 0;
}
#define starpu_worker_get_relax_state _starpu_worker_get_relax_state
static inline void _starpu_worker_lock(int workerid)
{
struct _starpu_worker *worker = _starpu_get_worker_struct(workerid);
STARPU_ASSERT(worker != NULL);
int cur_workerid = starpu_worker_get_id();
if (workerid != cur_workerid)
{
starpu_worker_relax_on();
STARPU_PTHREAD_MUTEX_LOCK_SCHED(&worker->sched_mutex);
while (!worker->state_relax_refcnt)
{
STARPU_PTHREAD_COND_WAIT(&worker->sched_cond, &worker->sched_mutex);
}
}
else
{
STARPU_PTHREAD_MUTEX_LOCK_SCHED(&worker->sched_mutex);
}
}
static inline int _starpu_worker_trylock(int workerid)
{
struct _starpu_worker *cur_worker = _starpu_get_local_worker_key();
int cur_workerid = cur_worker->workerid;
struct _starpu_worker *worker = _starpu_get_worker_struct(workerid);
STARPU_ASSERT(worker != NULL);
/* Start with ourself */
int ret = STARPU_PTHREAD_MUTEX_TRYLOCK_SCHED(&cur_worker->sched_mutex);
if (ret)
return ret;
if (workerid == cur_workerid)
/* We only needed to lock ourself */
return 0;
/* Now try to lock the other worker */
ret = STARPU_PTHREAD_MUTEX_TRYLOCK_SCHED(&worker->sched_mutex);
if (!ret)
{
/* Good, check that it is relaxed */
ret = !worker->state_relax_refcnt;
if (ret)
STARPU_PTHREAD_MUTEX_UNLOCK_SCHED(&worker->sched_mutex);
}
if (!ret)
_starpu_worker_relax_on_locked(cur_worker);
STARPU_PTHREAD_MUTEX_UNLOCK_SCHED(&cur_worker->sched_mutex);
return ret;
}
static inline void _starpu_worker_unlock(int workerid)
{
struct _starpu_worker *worker = _starpu_get_worker_struct(workerid);
STARPU_ASSERT(worker != NULL);
STARPU_PTHREAD_MUTEX_UNLOCK_SCHED(&worker->sched_mutex);
int cur_workerid = starpu_worker_get_id();
if (workerid != cur_workerid)
{
starpu_worker_relax_off();
}
}
static inline void _starpu_worker_lock_self(void)
{
int workerid = starpu_worker_get_id_check();
struct _starpu_worker *worker = _starpu_get_worker_struct(workerid);
STARPU_ASSERT(worker != NULL);
STARPU_PTHREAD_MUTEX_LOCK_SCHED(&worker->sched_mutex);
}
static inline void _starpu_worker_unlock_self(void)
{
int workerid = starpu_worker_get_id_check();
struct _starpu_worker *worker = _starpu_get_worker_struct(workerid);
STARPU_ASSERT(worker != NULL);
STARPU_PTHREAD_MUTEX_UNLOCK_SCHED(&worker->sched_mutex);
}
static inline int _starpu_wake_worker_relax(int workerid)
{
_starpu_worker_lock(workerid);
int ret = starpu_wake_worker_locked(workerid);
_starpu_worker_unlock(workerid);
return ret;
}
int starpu_wake_worker_relax_light(int workerid);
void _starpu_worker_refuse_task(struct _starpu_worker *worker, struct starpu_task *task);
/* @}*/
#endif // __WORKERS_H__
_starpu_barrier_counter
Definition: barrier_counter.h:27
datawizard.h
driver_cpu.h
driver_cuda.h
driver_mic_source.h
driver_mpi_source.h
driver_opencl.h
drivers.h
errorcheck.h
_starpu_worker_status
_starpu_worker_status
Definition: errorcheck.h:26
fxt.h
jobs.h
perfmodel.h
sched_ctx.h
_starpu_worker_apply_deferred_ctx_changes
void _starpu_worker_apply_deferred_ctx_changes(void)
sched_ctx_list.h
_starpu_sched_ctx_list
Definition: sched_ctx_list.h:25
sched_policy.h
simgrid.h
starpu_parameters.h
_starpu_combined_worker
Definition: workers.h:204
_starpu_combined_worker::padding
char padding[STARPU_CACHELINE_SIZE]
Definition: workers.h:224
_starpu_combined_worker::memory_node
unsigned memory_node
Definition: workers.h:208
_starpu_combined_worker::perf_arch
struct starpu_perfmodel_arch perf_arch
Definition: workers.h:205
_starpu_combined_worker::worker_mask
uint32_t worker_mask
Definition: workers.h:206
_starpu_driver_ops
Definition: drivers.h:24
_starpu_machine_config
Definition: workers.h:359
_starpu_machine_config::mpi_nodeid
int mpi_nodeid
Definition: workers.h:393
_starpu_machine_config::combined_workers
struct _starpu_combined_worker combined_workers[STARPU_NMAX_COMBINEDWORKERS]
Definition: workers.h:404
_starpu_machine_config::opencl_nodeid
int opencl_nodeid
Definition: workers.h:389
_starpu_machine_config::running
unsigned running
Definition: workers.h:428
_starpu_machine_config::current_cuda_gpuid
int current_cuda_gpuid
Definition: workers.h:373
_starpu_machine_config::nbindid
unsigned nbindid
Definition: workers.h:417
_starpu_machine_config::pause_depth
int pause_depth
Definition: workers.h:434
_starpu_machine_config::current_opencl_gpuid
int current_opencl_gpuid
Definition: workers.h:376
_starpu_machine_config::cpus_nodeid
int cpus_nodeid
Definition: workers.h:385
_starpu_machine_config::mic_nodeid
int mic_nodeid
Definition: workers.h:391
_starpu_machine_config::bindid_workers
struct _starpu_machine_config::@4 * bindid_workers
_starpu_machine_config::sched_ctxs
struct _starpu_sched_ctx sched_ctxs[STARPU_NMAX_SCHED_CTXS+1]
Definition: workers.h:437
_starpu_machine_config::current_bindid
int current_bindid
Definition: workers.h:368
_starpu_machine_config::workers
struct _starpu_worker workers[STARPU_NMAXWORKERS]
Definition: workers.h:400
_starpu_machine_config::worker_mask
uint32_t worker_mask
Definition: workers.h:422
_starpu_machine_config::current_mic_deviceid
int current_mic_deviceid
Definition: workers.h:379
_starpu_machine_config::cuda_nodeid
int cuda_nodeid
Definition: workers.h:387
_starpu_machine_config::padding2
char padding2[STARPU_CACHELINE_SIZE]
Definition: workers.h:409
_starpu_machine_config::submitting
unsigned submitting
Definition: workers.h:440
_starpu_machine_config::padding1
char padding1[STARPU_CACHELINE_SIZE]
Definition: workers.h:396
_starpu_machine_config::current_mpi_deviceid
int current_mpi_deviceid
Definition: workers.h:382
_starpu_machine_config::conf
struct starpu_conf conf
Definition: workers.h:425
_starpu_machine_topology
Definition: workers.h:248
_starpu_machine_topology::tree
struct starpu_tree * tree
Definition: workers.h:262
_starpu_machine_topology::nmpidevices
unsigned nmpidevices
Definition: workers.h:302
_starpu_machine_topology::nhwmicdevices
unsigned nhwmicdevices
Definition: workers.h:310
_starpu_machine_topology::hwtopology
hwloc_topology_t hwtopology
Definition: workers.h:259
_starpu_machine_topology::workers_mpi_ms_deviceid
unsigned workers_mpi_ms_deviceid[STARPU_NMAXWORKERS]
Definition: workers.h:355
_starpu_machine_topology::workers_cuda_gpuid
unsigned workers_cuda_gpuid[STARPU_NMAXWORKERS]
Definition: workers.h:337
_starpu_machine_topology::nhwopenclgpus
unsigned nhwopenclgpus
Definition: workers.h:282
_starpu_machine_topology::workers_opencl_gpuid
unsigned workers_opencl_gpuid[STARPU_NMAXWORKERS]
Definition: workers.h:345
_starpu_machine_topology::nworkers
unsigned nworkers
Definition: workers.h:250
_starpu_machine_topology::ncudagpus
unsigned ncudagpus
Definition: workers.h:293
_starpu_machine_topology::nhwmpi
unsigned nhwmpi
Definition: workers.h:287
_starpu_machine_topology::ncpus
unsigned ncpus
Definition: workers.h:290
_starpu_machine_topology::ncombinedworkers
unsigned ncombinedworkers
Definition: workers.h:253
_starpu_machine_topology::nhwcudagpus
unsigned nhwcudagpus
Definition: workers.h:277
_starpu_machine_topology::nopenclgpus
unsigned nopenclgpus
Definition: workers.h:299
_starpu_machine_topology::workers_nbindid
unsigned workers_nbindid
Definition: workers.h:329
_starpu_machine_topology::nhwmpicores
unsigned nhwmpicores[STARPU_MAXMPIDEVS]
Definition: workers.h:305
_starpu_machine_topology::nhwpus
unsigned nhwpus
Definition: workers.h:272
_starpu_machine_topology::workers_bindid
unsigned workers_bindid[STARPU_NMAXWORKERS]
Definition: workers.h:323
_starpu_machine_topology::nhwmiccores
unsigned nhwmiccores[STARPU_MAXMICDEVS]
Definition: workers.h:313
_starpu_machine_topology::nhwcpus
unsigned nhwcpus
Definition: workers.h:267
_starpu_sched_ctx
Definition: sched_ctx.h:46
_starpu_sched_ctx::id
unsigned id
Definition: sched_ctx.h:48
_starpu_worker_set
Definition: workers.h:232
_starpu_worker_set::started
unsigned started
Definition: workers.h:236
_starpu_worker_set::worker_thread
starpu_pthread_t worker_thread
Definition: workers.h:234
_starpu_worker_set::ready_cond
starpu_pthread_cond_t ready_cond
Definition: workers.h:239
_starpu_worker
Definition: workers.h:70
_starpu_worker::state_blocked_in_parallel_observed
unsigned state_blocked_in_parallel_observed
Definition: workers.h:107
_starpu_worker::current_task
struct starpu_task * current_task
Definition: workers.h:134
_starpu_worker::state_block_in_parallel_ack
unsigned state_block_in_parallel_ack
Definition: workers.h:109
_starpu_worker::state_relax_refcnt
unsigned state_relax_refcnt
Definition: workers.h:94
_starpu_worker::task_transferring
struct starpu_task * task_transferring
Definition: workers.h:170
_starpu_worker::state_changing_ctx_waiting
unsigned state_changing_ctx_waiting
Definition: workers.h:104
_starpu_worker::subworkerid
unsigned subworkerid
Definition: workers.h:78
_starpu_worker::devid
unsigned devid
Definition: workers.h:77
_starpu_worker::thread_changing_ctx
starpu_pthread_t thread_changing_ctx
Definition: workers.h:120
_starpu_worker::current_rank
int current_rank
Definition: workers.h:82
_starpu_worker::state_unblock_in_parallel_req
unsigned state_unblock_in_parallel_req
Definition: workers.h:110
_starpu_worker::nb_buffers_transferred
unsigned nb_buffers_transferred
Definition: workers.h:168
_starpu_worker::worker_thread
starpu_pthread_t worker_thread
Definition: workers.h:76
_starpu_worker::ready_cond
starpu_pthread_cond_t ready_cond
Definition: workers.h:85
_starpu_worker::memory_node
unsigned memory_node
Definition: workers.h:86
_starpu_worker::workerid
int workerid
Definition: workers.h:80
_starpu_worker::nb_buffers_totransfer
unsigned nb_buffers_totransfer
Definition: workers.h:169
_starpu_worker::state_keep_awake
unsigned state_keep_awake
Definition: workers.h:151
_starpu_worker::set
struct _starpu_worker_set * set
Definition: workers.h:147
_starpu_worker::worker_size
int worker_size
Definition: workers.h:83
_starpu_worker::pop_ctx_priority
unsigned pop_ctx_priority
Definition: workers.h:186
_starpu_worker::state_unblock_in_parallel_ack
unsigned state_unblock_in_parallel_ack
Definition: workers.h:111
_starpu_worker::numa_memory_node
unsigned numa_memory_node
Definition: workers.h:87
_starpu_worker::pipeline_stuck
unsigned char pipeline_stuck
Definition: workers.h:146
_starpu_worker::pipeline_length
unsigned char pipeline_length
Definition: workers.h:145
_starpu_worker::nsched_ctxs
unsigned nsched_ctxs
Definition: workers.h:159
_starpu_worker::spinning_backoff
unsigned spinning_backoff
Definition: workers.h:166
_starpu_worker::state_blocked_in_parallel
unsigned state_blocked_in_parallel
Definition: workers.h:106
_starpu_worker::current_ordered_task_order
unsigned current_ordered_task_order
Definition: workers.h:133
_starpu_worker::started_cond
starpu_pthread_cond_t started_cond
Definition: workers.h:84
_starpu_worker::sched_mutex
starpu_pthread_mutex_t sched_mutex
Definition: workers.h:93
_starpu_worker::arch
enum starpu_worker_archtype arch
Definition: workers.h:73
_starpu_worker::run_by_starpu
unsigned run_by_starpu
Definition: workers.h:154
_starpu_worker::local_ordered_tasks_size
unsigned local_ordered_tasks_size
Definition: workers.h:131
_starpu_worker::local_ordered_tasks
struct starpu_task ** local_ordered_tasks
Definition: workers.h:130
_starpu_worker::is_slave_somewhere
unsigned is_slave_somewhere
Definition: workers.h:187
_starpu_worker::sched_cond
starpu_pthread_cond_t sched_cond
Definition: workers.h:92
_starpu_worker::has_prev_init
unsigned has_prev_init
Definition: workers.h:162
_starpu_worker::ntasks
unsigned char ntasks
Definition: workers.h:144
_starpu_worker::bindid
int bindid
Definition: workers.h:79
_starpu_worker::state_block_in_parallel_req
unsigned state_block_in_parallel_req
Definition: workers.h:108
_starpu_worker::state_changing_ctx_notice
unsigned state_changing_ctx_notice
Definition: workers.h:105
_starpu_worker::state_sched_op_pending
unsigned state_sched_op_pending
Definition: workers.h:103
_starpu_worker::combined_workerid
int combined_workerid
Definition: workers.h:81
_starpu_worker::first_task
unsigned char first_task
Definition: workers.h:143
_starpu_worker::current_ordered_task
unsigned current_ordered_task
Definition: workers.h:132
_starpu_worker::block_in_parallel_ref_count
unsigned block_in_parallel_ref_count
Definition: workers.h:119
_starpu_worker::worker_mask
uint32_t worker_mask
Definition: workers.h:74
_starpu_worker::status
enum _starpu_worker_status status
Definition: workers.h:150
thread.h
timing.h
topology.h
utils.h