task.c 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817
  1. /* StarPU --- Runtime system for heterogeneous multicore architectures.
  2. *
  3. * Copyright (C) 2009-2012 Université de Bordeaux 1
  4. * Copyright (C) 2010, 2011, 2012 Centre National de la Recherche Scientifique
  5. * Copyright (C) 2011 Télécom-SudParis
  6. * Copyright (C) 2011 INRIA
  7. *
  8. * StarPU is free software; you can redistribute it and/or modify
  9. * it under the terms of the GNU Lesser General Public License as published by
  10. * the Free Software Foundation; either version 2.1 of the License, or (at
  11. * your option) any later version.
  12. *
  13. * StarPU is distributed in the hope that it will be useful, but
  14. * WITHOUT ANY WARRANTY; without even the implied warranty of
  15. * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
  16. *
  17. * See the GNU Lesser General Public License in COPYING.LGPL for more details.
  18. */
  19. #include <starpu.h>
  20. #include <starpu_profiling.h>
  21. #include <core/workers.h>
  22. #include <core/sched_ctx.h>
  23. #include <core/jobs.h>
  24. #include <core/task.h>
  25. #include <core/task_bundle.h>
  26. #include <common/config.h>
  27. #include <common/utils.h>
  28. #include <profiling/profiling.h>
  29. #include <profiling/bound.h>
  30. #include <math.h>
  31. #include <string.h>
  32. /* XXX this should be reinitialized when StarPU is shutdown (or we should make
  33. * sure that no task remains !) */
  34. /* TODO we could make this hierarchical to avoid contention ? */
  35. static pthread_cond_t submitted_cond = PTHREAD_COND_INITIALIZER;
  36. static pthread_mutex_t submitted_mutex = PTHREAD_MUTEX_INITIALIZER;
  37. static long int nsubmitted = 0, nready = 0;
  38. static void _starpu_increment_nsubmitted_tasks(void);
  39. /* This key stores the task currently handled by the thread, note that we
  40. * cannot use the worker structure to store that information because it is
  41. * possible that we have a task with a NULL codelet, which means its callback
  42. * could be executed by a user thread as well. */
  43. static pthread_key_t current_task_key;
  44. void starpu_task_init(struct starpu_task *task)
  45. {
  46. STARPU_ASSERT(task);
  47. /* As most of the fields must be initialised at NULL, let's put 0
  48. * everywhere */
  49. memset(task, 0, sizeof(struct starpu_task));
  50. /* Now we can initialise fields which recquire custom value */
  51. task->priority = STARPU_DEFAULT_PRIO;
  52. task->detach = 1;
  53. task->status = STARPU_TASK_INVALID;
  54. task->predicted = NAN;
  55. task->predicted_transfer = NAN;
  56. task->magic = 42;
  57. task->sched_ctx = _starpu_get_initial_sched_ctx()->id;
  58. task->control_task = 0;
  59. task->hypervisor_tag = 0;
  60. task->flops = 0.0;
  61. task->already_pushed = 0;
  62. }
  63. /* Free all the ressources allocated for a task, without deallocating the task
  64. * structure itself (this is required for statically allocated tasks). */
  65. void starpu_task_deinit(struct starpu_task *task)
  66. {
  67. STARPU_ASSERT(task);
  68. /* If a buffer was allocated to store the profiling info, we free it. */
  69. if (task->profiling_info)
  70. {
  71. free(task->profiling_info);
  72. task->profiling_info = NULL;
  73. }
  74. /* If case the task is (still) part of a bundle */
  75. starpu_task_bundle_t bundle = task->bundle;
  76. if (bundle)
  77. starpu_task_bundle_remove(bundle, task);
  78. struct _starpu_job *j = (struct _starpu_job *)task->starpu_private;
  79. if (j) {
  80. _starpu_job_destroy(j);
  81. task->starpu_private = NULL;
  82. }
  83. }
  84. struct starpu_task * __attribute__((malloc)) starpu_task_create(void)
  85. {
  86. struct starpu_task *task;
  87. task = (struct starpu_task *) malloc(sizeof(struct starpu_task));
  88. STARPU_ASSERT(task);
  89. starpu_task_init(task);
  90. /* Dynamically allocated tasks are destroyed by default */
  91. task->destroy = 1;
  92. return task;
  93. }
  94. /* Free the ressource allocated during starpu_task_create. This function can be
  95. * called automatically after the execution of a task by setting the "destroy"
  96. * flag of the starpu_task structure (default behaviour). Calling this function
  97. * on a statically allocated task results in an undefined behaviour. */
  98. void _starpu_task_destroy(struct starpu_task *task)
  99. {
  100. /* If starpu_task_destroy is called in a callback, we just set the destroy
  101. flag. The task will be destroyed after the callback returns */
  102. if (task == starpu_task_get_current()
  103. && _starpu_get_local_worker_status() == STATUS_CALLBACK)
  104. {
  105. task->destroy = 1;
  106. }
  107. else
  108. {
  109. starpu_task_deinit(task);
  110. /* TODO handle the case of task with detach = 1 and destroy = 1 */
  111. /* TODO handle the case of non terminated tasks -> return -EINVAL */
  112. free(task);
  113. }
  114. }
  115. void starpu_task_destroy(struct starpu_task *task)
  116. {
  117. STARPU_ASSERT(task);
  118. STARPU_ASSERT_MSG(!task->destroy || !task->detach, "starpu_task_destroy must not be called for task with destroy = 1 and detach = 1");
  119. _starpu_task_destroy(task);
  120. }
  121. int starpu_task_wait(struct starpu_task *task)
  122. {
  123. _STARPU_LOG_IN();
  124. STARPU_ASSERT(task);
  125. STARPU_ASSERT_MSG(!task->detach, "starpu_task_wait can only be called on tasks with detach = 0");
  126. if (task->detach || task->synchronous)
  127. {
  128. _STARPU_DEBUG("Task is detached or asynchronous. Waiting returns immediately\n");
  129. _STARPU_LOG_OUT_TAG("einval");
  130. return -EINVAL;
  131. }
  132. if (STARPU_UNLIKELY(!_starpu_worker_may_perform_blocking_calls()))
  133. {
  134. _STARPU_LOG_OUT_TAG("edeadlk");
  135. return -EDEADLK;
  136. }
  137. struct _starpu_job *j = (struct _starpu_job *)task->starpu_private;
  138. _starpu_wait_job(j);
  139. /* as this is a synchronous task, the liberation of the job
  140. structure was deferred */
  141. if (task->destroy)
  142. _starpu_task_destroy(task);
  143. _STARPU_LOG_OUT();
  144. return 0;
  145. }
  146. struct _starpu_job *_starpu_get_job_associated_to_task(struct starpu_task *task)
  147. {
  148. STARPU_ASSERT(task);
  149. if (!task->starpu_private)
  150. {
  151. struct _starpu_job *j = _starpu_job_create(task);
  152. task->starpu_private = j;
  153. }
  154. return (struct _starpu_job *)task->starpu_private;
  155. }
  156. /* NB in case we have a regenerable task, it is possible that the job was
  157. * already counted. */
  158. int _starpu_submit_job(struct _starpu_job *j)
  159. {
  160. struct starpu_task *task = j->task;
  161. _STARPU_LOG_IN();
  162. /* notify bound computation of a new task */
  163. _starpu_bound_record(j);
  164. _starpu_increment_nsubmitted_tasks();
  165. _starpu_increment_nsubmitted_tasks_of_sched_ctx(j->task->sched_ctx);
  166. #ifdef STARPU_USE_SCHED_CTX_HYPERVISOR
  167. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(j->task->sched_ctx);
  168. if(sched_ctx != NULL && j->task->sched_ctx != 0 && j->task->sched_ctx != STARPU_NMAX_SCHED_CTXS
  169. && sched_ctx->perf_counters != NULL)
  170. {
  171. _starpu_compute_buffers_footprint(j->task->cl->model, STARPU_CPU_DEFAULT, 0, j);
  172. sched_ctx->perf_counters->notify_submitted_job(j->task, j->footprint);
  173. }
  174. #endif
  175. /* We retain handle reference count */
  176. if (task->cl) {
  177. unsigned i;
  178. for (i=0; i<task->cl->nbuffers; i++) {
  179. starpu_data_handle_t handle = task->handles[i];
  180. _starpu_spin_lock(&handle->header_lock);
  181. handle->busy_count++;
  182. _starpu_spin_unlock(&handle->header_lock);
  183. }
  184. }
  185. _STARPU_PTHREAD_MUTEX_LOCK(&j->sync_mutex);
  186. /* Need to atomically set submitted to 1 and check dependencies, since
  187. * this is concucrent with _starpu_notify_cg */
  188. j->terminated = 0;
  189. if (!j->submitted)
  190. j->submitted = 1;
  191. else
  192. j->submitted = 2;
  193. int ret = _starpu_enforce_deps_and_schedule(j);
  194. _STARPU_LOG_OUT();
  195. return ret;
  196. }
  197. void _starpu_codelet_check_deprecated_fields(struct starpu_codelet *cl)
  198. {
  199. if (!cl)
  200. return;
  201. int is_where_unset = cl->where == 0;
  202. /* Check deprecated and unset fields (where, <device>_func,
  203. * <device>_funcs) */
  204. /* CPU */
  205. if (cl->cpu_func && cl->cpu_func != STARPU_MULTIPLE_CPU_IMPLEMENTATIONS && cl->cpu_funcs[0])
  206. {
  207. fprintf(stderr, "[warning] [struct starpu_codelet] both cpu_func and cpu_funcs are set. Ignoring cpu_func.\n");
  208. cl->cpu_func = STARPU_MULTIPLE_CPU_IMPLEMENTATIONS;
  209. }
  210. if (cl->cpu_func && cl->cpu_func != STARPU_MULTIPLE_CPU_IMPLEMENTATIONS)
  211. {
  212. cl->cpu_funcs[0] = cl->cpu_func;
  213. cl->cpu_func = STARPU_MULTIPLE_CPU_IMPLEMENTATIONS;
  214. }
  215. if (cl->cpu_funcs[0] && cl->cpu_func == 0)
  216. {
  217. cl->cpu_func = STARPU_MULTIPLE_CPU_IMPLEMENTATIONS;
  218. }
  219. if (cl->cpu_funcs[0] && is_where_unset)
  220. {
  221. cl->where |= STARPU_CPU;
  222. }
  223. /* CUDA */
  224. if (cl->cuda_func && cl->cuda_func != STARPU_MULTIPLE_CUDA_IMPLEMENTATIONS && cl->cuda_funcs[0])
  225. {
  226. fprintf(stderr, "[warning] [struct starpu_codelet] both cuda_func and cuda_funcs are set. Ignoring cuda_func.\n");
  227. cl->cuda_func = STARPU_MULTIPLE_CUDA_IMPLEMENTATIONS;
  228. }
  229. if (cl->cuda_func && cl->cuda_func != STARPU_MULTIPLE_CUDA_IMPLEMENTATIONS)
  230. {
  231. cl->cuda_funcs[0] = cl->cuda_func;
  232. cl->cuda_func = STARPU_MULTIPLE_CUDA_IMPLEMENTATIONS;
  233. }
  234. if (cl->cuda_funcs[0] && cl->cuda_func == 0)
  235. {
  236. cl->cuda_func = STARPU_MULTIPLE_CUDA_IMPLEMENTATIONS;
  237. }
  238. if (cl->cuda_funcs[0] && is_where_unset)
  239. {
  240. cl->where |= STARPU_CUDA;
  241. }
  242. /* OpenCL */
  243. if (cl->opencl_func && cl->opencl_func != STARPU_MULTIPLE_OPENCL_IMPLEMENTATIONS && cl->opencl_funcs[0])
  244. {
  245. fprintf(stderr, "[warning] [struct starpu_codelet] both opencl_func and opencl_funcs are set. Ignoring opencl_func.\n");
  246. cl->opencl_func = STARPU_MULTIPLE_OPENCL_IMPLEMENTATIONS;
  247. }
  248. if (cl->opencl_func && cl->opencl_func != STARPU_MULTIPLE_OPENCL_IMPLEMENTATIONS)
  249. {
  250. cl->opencl_funcs[0] = cl->opencl_func;
  251. cl->opencl_func = STARPU_MULTIPLE_OPENCL_IMPLEMENTATIONS;
  252. }
  253. if (cl->opencl_funcs[0] && cl->opencl_func == 0)
  254. {
  255. cl->opencl_func = STARPU_MULTIPLE_OPENCL_IMPLEMENTATIONS;
  256. }
  257. if (cl->opencl_funcs[0] && is_where_unset)
  258. {
  259. cl->where |= STARPU_OPENCL;
  260. }
  261. /* Gordon */
  262. if (cl->gordon_func && cl->gordon_func != STARPU_MULTIPLE_GORDON_IMPLEMENTATIONS)
  263. {
  264. cl->gordon_funcs[0] = cl->gordon_func;
  265. cl->gordon_func = STARPU_MULTIPLE_GORDON_IMPLEMENTATIONS;
  266. }
  267. if (cl->gordon_funcs[0] && cl->gordon_func == 0)
  268. {
  269. cl->gordon_func = STARPU_MULTIPLE_GORDON_IMPLEMENTATIONS;
  270. }
  271. if (cl->gordon_funcs[0] && is_where_unset)
  272. {
  273. cl->where = STARPU_GORDON;
  274. }
  275. }
  276. void _starpu_task_check_deprecated_fields(struct starpu_task *task)
  277. {
  278. if (task->cl)
  279. {
  280. unsigned i;
  281. for(i=0; i<task->cl->nbuffers ; i++)
  282. {
  283. if (task->buffers[i].handle && task->handles[i])
  284. {
  285. fprintf(stderr, "[warning][struct starpu_task] task->buffers[%u] and task->handles[%u] both set. Ignoring task->buffers[%u] ?\n", i, i, i);
  286. STARPU_ASSERT(task->buffers[i].mode == task->cl->modes[i]);
  287. STARPU_ABORT();
  288. }
  289. if (task->buffers[i].handle)
  290. {
  291. task->handles[i] = task->buffers[i].handle;
  292. task->cl->modes[i] = task->buffers[i].mode;
  293. }
  294. }
  295. }
  296. }
  297. /* application should submit new tasks to StarPU through this function */
  298. int starpu_task_submit(struct starpu_task *task)
  299. {
  300. STARPU_ASSERT(task);
  301. STARPU_ASSERT(task->magic == 42);
  302. unsigned nsched_ctxs = _starpu_get_nsched_ctxs();
  303. if(task->sched_ctx == 0 && nsched_ctxs != 1 && !task->control_task)
  304. task->sched_ctx = starpu_get_sched_ctx();
  305. // task->sched_ctx = (nsched_ctxs == 1 || task->control_task) ?
  306. // 0 : starpu_get_sched_ctx());
  307. int ret;
  308. unsigned is_sync = task->synchronous;
  309. _STARPU_LOG_IN();
  310. if (is_sync)
  311. {
  312. /* Perhaps it is not possible to submit a synchronous
  313. * (blocking) task */
  314. if (STARPU_UNLIKELY(!_starpu_worker_may_perform_blocking_calls()))
  315. {
  316. _STARPU_LOG_OUT_TAG("EDEADLK");
  317. return -EDEADLK;
  318. }
  319. task->detach = 0;
  320. }
  321. _starpu_task_check_deprecated_fields(task);
  322. _starpu_codelet_check_deprecated_fields(task->cl);
  323. if (task->cl)
  324. {
  325. unsigned i;
  326. /* Check the type of worker(s) required by the task exist */
  327. if (!_starpu_worker_exists(task))
  328. {
  329. _STARPU_LOG_OUT_TAG("ENODEV");
  330. return -ENODEV;
  331. }
  332. /* Check buffers */
  333. STARPU_ASSERT(task->cl->nbuffers <= STARPU_NMAXBUFS);
  334. for (i = 0; i < task->cl->nbuffers; i++)
  335. {
  336. /* Make sure handles are not partitioned */
  337. STARPU_ASSERT(task->handles[i]->nchildren == 0);
  338. }
  339. /* In case we require that a task should be explicitely
  340. * executed on a specific worker, we make sure that the worker
  341. * is able to execute this task. */
  342. if (task->execute_on_a_specific_worker && !starpu_combined_worker_can_execute_task(task->workerid, task, 0))
  343. {
  344. _STARPU_LOG_OUT_TAG("ENODEV");
  345. return -ENODEV;
  346. }
  347. _starpu_detect_implicit_data_deps(task);
  348. if (task->cl->model && task->cl->model->symbol)
  349. _starpu_load_perfmodel(task->cl->model);
  350. if (task->cl->power_model && task->cl->power_model->symbol)
  351. _starpu_load_perfmodel(task->cl->power_model);
  352. }
  353. /* If profiling is activated, we allocate a structure to store the
  354. * appropriate info. */
  355. struct starpu_task_profiling_info *info;
  356. int profiling = starpu_profiling_status_get();
  357. info = _starpu_allocate_profiling_info_if_needed(task);
  358. task->profiling_info = info;
  359. /* The task is considered as block until we are sure there remains not
  360. * dependency. */
  361. task->status = STARPU_TASK_BLOCKED;
  362. if (profiling)
  363. _starpu_clock_gettime(&info->submit_time);
  364. /* internally, StarPU manipulates a struct _starpu_job * which is a wrapper around a
  365. * task structure, it is possible that this job structure was already
  366. * allocated, for instance to enforce task depenencies. */
  367. struct _starpu_job *j = _starpu_get_job_associated_to_task(task);
  368. ret = _starpu_submit_job(j);
  369. if (is_sync)
  370. {
  371. _starpu_wait_job(j);
  372. if (task->destroy)
  373. _starpu_task_destroy(task);
  374. }
  375. _STARPU_LOG_OUT();
  376. return ret;
  377. }
  378. int _starpu_task_submit_internally(struct starpu_task *task)
  379. {
  380. task->control_task = 1;
  381. return starpu_task_submit(task);
  382. }
  383. /* application should submit new tasks to StarPU through this function */
  384. int starpu_task_submit_to_ctx(struct starpu_task *task, unsigned sched_ctx_id)
  385. {
  386. task->sched_ctx = sched_ctx_id;
  387. starpu_task_submit(task);
  388. }
  389. /* The StarPU core can submit tasks directly to the scheduler or a worker,
  390. * skipping dependencies completely (when it knows what it is doing). */
  391. int _starpu_task_submit_nodeps(struct starpu_task *task)
  392. {
  393. _starpu_task_check_deprecated_fields(task);
  394. _starpu_codelet_check_deprecated_fields(task->cl);
  395. if (task->cl)
  396. {
  397. if (task->cl->model)
  398. _starpu_load_perfmodel(task->cl->model);
  399. if (task->cl->power_model)
  400. _starpu_load_perfmodel(task->cl->power_model);
  401. }
  402. struct _starpu_job *j = _starpu_get_job_associated_to_task(task);
  403. _starpu_increment_nsubmitted_tasks();
  404. _starpu_increment_nsubmitted_tasks_of_sched_ctx(j->task->sched_ctx);
  405. _STARPU_PTHREAD_MUTEX_LOCK(&j->sync_mutex);
  406. j->submitted = 1;
  407. if (task->cl)
  408. {
  409. /* This would be done by data dependencies checking */
  410. unsigned i;
  411. for (i=0 ; i<task->cl->nbuffers ; i++)
  412. {
  413. j->ordered_buffers[i].handle = j->task->handles[i];
  414. j->ordered_buffers[i].mode = j->task->cl->modes[i];
  415. }
  416. }
  417. _STARPU_PTHREAD_MUTEX_UNLOCK(&j->sync_mutex);
  418. return _starpu_push_task(j);
  419. }
  420. /*
  421. * worker->sched_mutex must be locked when calling this function.
  422. */
  423. int _starpu_task_submit_conversion_task(struct starpu_task *task,
  424. unsigned int workerid)
  425. {
  426. STARPU_ASSERT(task->cl);
  427. STARPU_ASSERT(task->execute_on_a_specific_worker);
  428. _starpu_task_check_deprecated_fields(task);
  429. _starpu_codelet_check_deprecated_fields(task->cl);
  430. /* We should factorize that */
  431. if (task->cl->model)
  432. _starpu_load_perfmodel(task->cl->model);
  433. if (task->cl->power_model)
  434. _starpu_load_perfmodel(task->cl->power_model);
  435. /* We retain handle reference count */
  436. unsigned i;
  437. for (i=0; i<task->cl->nbuffers; i++) {
  438. starpu_data_handle_t handle = task->handles[i];
  439. _starpu_spin_lock(&handle->header_lock);
  440. handle->busy_count++;
  441. _starpu_spin_unlock(&handle->header_lock);
  442. }
  443. struct _starpu_job *j = _starpu_get_job_associated_to_task(task);
  444. _starpu_increment_nsubmitted_tasks();
  445. _starpu_increment_nsubmitted_tasks_of_sched_ctx(j->task->sched_ctx);
  446. _STARPU_PTHREAD_MUTEX_LOCK(&j->sync_mutex);
  447. j->submitted = 1;
  448. _starpu_increment_nready_tasks();
  449. for (i=0 ; i<task->cl->nbuffers ; i++)
  450. {
  451. j->ordered_buffers[i].handle = j->task->handles[i];
  452. j->ordered_buffers[i].mode = j->task->cl->modes[i];
  453. }
  454. _STARPU_LOG_IN();
  455. task->status = STARPU_TASK_READY;
  456. _starpu_profiling_set_task_push_start_time(task);
  457. unsigned node = starpu_worker_get_memory_node(workerid);
  458. if (starpu_get_prefetch_flag())
  459. starpu_prefetch_task_input_on_node(task, node);
  460. struct _starpu_worker *worker;
  461. worker = _starpu_get_worker_struct(workerid);
  462. starpu_task_list_push_front(&worker->local_tasks, task);
  463. _starpu_profiling_set_task_push_end_time(task);
  464. _STARPU_LOG_OUT();
  465. _STARPU_PTHREAD_MUTEX_UNLOCK(&j->sync_mutex);
  466. return 0;
  467. }
  468. void starpu_codelet_init(struct starpu_codelet *cl)
  469. {
  470. memset(cl, 0, sizeof(struct starpu_codelet));
  471. }
  472. void starpu_display_codelet_stats(struct starpu_codelet *cl)
  473. {
  474. unsigned worker;
  475. unsigned nworkers = starpu_worker_get_count();
  476. if (cl->name)
  477. fprintf(stderr, "Statistics for codelet %s\n", cl->name);
  478. else if (cl->model && cl->model->symbol)
  479. fprintf(stderr, "Statistics for codelet %s\n", cl->model->symbol);
  480. unsigned long total = 0;
  481. for (worker = 0; worker < nworkers; worker++)
  482. total += cl->per_worker_stats[worker];
  483. for (worker = 0; worker < nworkers; worker++)
  484. {
  485. char name[32];
  486. starpu_worker_get_name(worker, name, 32);
  487. fprintf(stderr, "\t%s -> %lu / %lu (%2.2f %%)\n", name, cl->per_worker_stats[worker], total, (100.0f*cl->per_worker_stats[worker])/total);
  488. }
  489. }
  490. /*
  491. * We wait for all the tasks that have already been submitted. Note that a
  492. * regenerable is not considered finished until it was explicitely set as
  493. * non-regenerale anymore (eg. from a callback).
  494. */
  495. int starpu_task_wait_for_all(void)
  496. {
  497. unsigned nsched_ctxs = _starpu_get_nsched_ctxs();
  498. unsigned sched_ctx = nsched_ctxs == 1 ? 0 : starpu_get_sched_ctx();
  499. _starpu_wait_for_all_tasks_of_sched_ctx(sched_ctx);
  500. return 0;
  501. }
  502. int starpu_task_wait_for_all_in_ctx(unsigned sched_ctx)
  503. {
  504. _starpu_wait_for_all_tasks_of_sched_ctx(sched_ctx);
  505. return 0;
  506. }
  507. /*
  508. * We wait until there is no ready task any more (i.e. StarPU will not be able
  509. * to progress any more).
  510. */
  511. int starpu_task_wait_for_no_ready(void)
  512. {
  513. if (STARPU_UNLIKELY(!_starpu_worker_may_perform_blocking_calls()))
  514. return -EDEADLK;
  515. _STARPU_PTHREAD_MUTEX_LOCK(&submitted_mutex);
  516. _STARPU_TRACE_TASK_WAIT_FOR_ALL;
  517. while (nready > 0)
  518. _STARPU_PTHREAD_COND_WAIT(&submitted_cond, &submitted_mutex);
  519. _STARPU_PTHREAD_MUTEX_UNLOCK(&submitted_mutex);
  520. return 0;
  521. }
  522. void _starpu_decrement_nsubmitted_tasks(void)
  523. {
  524. struct _starpu_machine_config *config = _starpu_get_machine_config();
  525. _STARPU_PTHREAD_MUTEX_LOCK(&submitted_mutex);
  526. if (--nsubmitted == 0) {
  527. if (!config->submitting)
  528. config->running = 0;
  529. _STARPU_PTHREAD_COND_BROADCAST(&submitted_cond);
  530. }
  531. _STARPU_TRACE_UPDATE_TASK_CNT(nsubmitted);
  532. _STARPU_PTHREAD_MUTEX_UNLOCK(&submitted_mutex);
  533. }
  534. void
  535. starpu_drivers_request_termination(void)
  536. {
  537. struct _starpu_machine_config *config = _starpu_get_machine_config();
  538. _STARPU_PTHREAD_MUTEX_LOCK(&submitted_mutex);
  539. config->submitting = 0;
  540. if (nsubmitted == 0) {
  541. config->running = 0;
  542. _STARPU_PTHREAD_COND_BROADCAST(&submitted_cond);
  543. }
  544. _STARPU_PTHREAD_MUTEX_UNLOCK(&submitted_mutex);
  545. }
  546. static void _starpu_increment_nsubmitted_tasks(void)
  547. {
  548. _STARPU_PTHREAD_MUTEX_LOCK(&submitted_mutex);
  549. nsubmitted++;
  550. _STARPU_TRACE_UPDATE_TASK_CNT(nsubmitted);
  551. _STARPU_PTHREAD_MUTEX_UNLOCK(&submitted_mutex);
  552. }
  553. void _starpu_increment_nready_tasks(void)
  554. {
  555. _STARPU_PTHREAD_MUTEX_LOCK(&submitted_mutex);
  556. nready++;
  557. _STARPU_PTHREAD_MUTEX_UNLOCK(&submitted_mutex);
  558. }
  559. void _starpu_decrement_nready_tasks(void)
  560. {
  561. _STARPU_PTHREAD_MUTEX_LOCK(&submitted_mutex);
  562. if (--nready == 0)
  563. _STARPU_PTHREAD_COND_BROADCAST(&submitted_cond);
  564. _STARPU_PTHREAD_MUTEX_UNLOCK(&submitted_mutex);
  565. }
  566. void _starpu_initialize_current_task_key(void)
  567. {
  568. pthread_key_create(&current_task_key, NULL);
  569. }
  570. /* Return the task currently executed by the worker, or NULL if this is called
  571. * either from a thread that is not a task or simply because there is no task
  572. * being executed at the moment. */
  573. struct starpu_task *starpu_task_get_current(void)
  574. {
  575. return (struct starpu_task *) pthread_getspecific(current_task_key);
  576. }
  577. void _starpu_set_current_task(struct starpu_task *task)
  578. {
  579. pthread_setspecific(current_task_key, task);
  580. }
  581. /*
  582. * Returns 0 if tasks does not use any multiformat handle, 1 otherwise.
  583. */
  584. int
  585. _starpu_task_uses_multiformat_handles(struct starpu_task *task)
  586. {
  587. unsigned i;
  588. for (i = 0; i < task->cl->nbuffers; i++)
  589. {
  590. if (_starpu_data_is_multiformat_handle(task->handles[i]))
  591. return 1;
  592. }
  593. return 0;
  594. }
  595. /*
  596. * Checks whether the given handle needs to be converted in order to be used on
  597. * the node given as the second argument.
  598. */
  599. int
  600. _starpu_handle_needs_conversion_task(starpu_data_handle_t handle,
  601. unsigned int node)
  602. {
  603. enum starpu_node_kind node_kind;
  604. node_kind = starpu_node_get_kind(node);
  605. /*
  606. * Here, we assume that CUDA devices and OpenCL devices use the
  607. * same data structure. A conversion is only needed when moving
  608. * data from a CPU to a GPU, or the other way around.
  609. */
  610. switch (node_kind)
  611. {
  612. case STARPU_CPU_RAM:
  613. switch(starpu_node_get_kind(handle->mf_node))
  614. {
  615. case STARPU_CPU_RAM:
  616. return 0;
  617. case STARPU_CUDA_RAM: /* Fall through */
  618. case STARPU_OPENCL_RAM:
  619. return 1;
  620. case STARPU_SPU_LS: /* Not supported */
  621. default:
  622. STARPU_ASSERT(0);
  623. }
  624. break;
  625. case STARPU_CUDA_RAM: /* Fall through */
  626. case STARPU_OPENCL_RAM:
  627. switch(starpu_node_get_kind(handle->mf_node))
  628. {
  629. case STARPU_CPU_RAM:
  630. return 1;
  631. case STARPU_CUDA_RAM:
  632. case STARPU_OPENCL_RAM:
  633. return 0;
  634. case STARPU_SPU_LS: /* Not supported */
  635. default:
  636. STARPU_ASSERT(0);
  637. }
  638. break;
  639. case STARPU_SPU_LS: /* Not supported */
  640. default:
  641. STARPU_ASSERT(0);
  642. }
  643. /* that instruction should never be reached */
  644. return -EINVAL;
  645. }
  646. starpu_cpu_func_t _starpu_task_get_cpu_nth_implementation(struct starpu_codelet *cl, unsigned nimpl)
  647. {
  648. STARPU_ASSERT(cl->cpu_func == STARPU_MULTIPLE_CPU_IMPLEMENTATIONS);
  649. return cl->cpu_funcs[nimpl];
  650. }
  651. starpu_cuda_func_t _starpu_task_get_cuda_nth_implementation(struct starpu_codelet *cl, unsigned nimpl)
  652. {
  653. STARPU_ASSERT(cl->cuda_func == STARPU_MULTIPLE_CUDA_IMPLEMENTATIONS);
  654. return cl->cuda_funcs[nimpl];
  655. }
  656. starpu_opencl_func_t _starpu_task_get_opencl_nth_implementation(struct starpu_codelet *cl, unsigned nimpl)
  657. {
  658. STARPU_ASSERT(cl->opencl_func == STARPU_MULTIPLE_OPENCL_IMPLEMENTATIONS);
  659. return cl->opencl_funcs[nimpl];
  660. }
  661. starpu_gordon_func_t _starpu_task_get_gordon_nth_implementation(struct starpu_codelet *cl, unsigned nimpl)
  662. {
  663. STARPU_ASSERT(cl->gordon_func == STARPU_MULTIPLE_GORDON_IMPLEMENTATIONS);
  664. return cl->gordon_funcs[nimpl];
  665. }