deque_modeling_policy_data_aware.c 42 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330
  1. /* StarPU --- Runtime system for heterogeneous multicore architectures.
  2. *
  3. * Copyright (C) 2011-2017 Inria
  4. * Copyright (C) 2009-2019 Université de Bordeaux
  5. * Copyright (C) 2013 Joris Pablo
  6. * Copyright (C) 2010-2018 CNRS
  7. * Copyright (C) 2013 Simon Archipoff
  8. * Copyright (C) 2013 Thibaut Lambert
  9. * Copyright (C) 2011 Télécom-SudParis
  10. * Copyright (C) 2016 Uppsala University
  11. *
  12. * StarPU is free software; you can redistribute it and/or modify
  13. * it under the terms of the GNU Lesser General Public License as published by
  14. * the Free Software Foundation; either version 2.1 of the License, or (at
  15. * your option) any later version.
  16. *
  17. * StarPU is distributed in the hope that it will be useful, but
  18. * WITHOUT ANY WARRANTY; without even the implied warranty of
  19. * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
  20. *
  21. * See the GNU Lesser General Public License in COPYING.LGPL for more details.
  22. */
  23. /* Distributed queues using performance modeling to assign tasks */
  24. #include <starpu_config.h>
  25. #include <starpu_scheduler.h>
  26. #include <common/fxt.h>
  27. #include <core/task.h>
  28. #include <core/workers.h>
  29. #include <core/sched_policy.h>
  30. #include <core/debug.h>
  31. #include <sched_policies/fifo_queues.h>
  32. #include <limits.h>
  33. #ifndef DBL_MIN
  34. #define DBL_MIN __DBL_MIN__
  35. #endif
  36. #ifndef DBL_MAX
  37. #define DBL_MAX __DBL_MAX__
  38. #endif
  39. //#define NOTIFY_READY_SOON
  40. struct _starpu_dmda_data
  41. {
  42. double alpha;
  43. double beta;
  44. double _gamma;
  45. double idle_power;
  46. struct _starpu_fifo_taskq **queue_array;
  47. long int total_task_cnt;
  48. long int ready_task_cnt;
  49. long int eager_task_cnt; /* number of tasks scheduled without model */
  50. int num_priorities;
  51. };
  52. /* The dmda scheduling policy uses
  53. *
  54. * alpha * T_computation + beta * T_communication + gamma * Consumption
  55. *
  56. * Here are the default values of alpha, beta, gamma
  57. */
  58. #define _STARPU_SCHED_ALPHA_DEFAULT 1.0
  59. #define _STARPU_SCHED_BETA_DEFAULT 1.0
  60. #define _STARPU_SCHED_GAMMA_DEFAULT 1000.0
  61. #ifdef STARPU_USE_TOP
  62. static double alpha = _STARPU_SCHED_ALPHA_DEFAULT;
  63. static double beta = _STARPU_SCHED_BETA_DEFAULT;
  64. static double _gamma = _STARPU_SCHED_GAMMA_DEFAULT;
  65. static double idle_power = 0.0;
  66. static const float alpha_minimum=0;
  67. static const float alpha_maximum=10.0;
  68. static const float beta_minimum=0;
  69. static const float beta_maximum=10.0;
  70. static const float gamma_minimum=0;
  71. static const float gamma_maximum=10000.0;
  72. static const float idle_power_minimum=0;
  73. static const float idle_power_maximum=10000.0;
  74. #endif /* !STARPU_USE_TOP */
  75. static int count_non_ready_buffers(struct starpu_task *task, unsigned worker)
  76. {
  77. int cnt = 0;
  78. unsigned nbuffers = STARPU_TASK_GET_NBUFFERS(task);
  79. unsigned index;
  80. for (index = 0; index < nbuffers; index++)
  81. {
  82. starpu_data_handle_t handle;
  83. unsigned buffer_node = _starpu_task_data_get_node_on_worker(task, index, worker);
  84. handle = STARPU_TASK_GET_HANDLE(task, index);
  85. int is_valid;
  86. starpu_data_query_status(handle, buffer_node, NULL, &is_valid, NULL);
  87. if (!is_valid)
  88. cnt++;
  89. }
  90. return cnt;
  91. }
  92. #ifdef STARPU_USE_TOP
  93. static void param_modified(struct starpu_top_param* d)
  94. {
  95. #ifdef STARPU_DEVEL
  96. #warning FIXME: get sched ctx to get alpha/beta/gamma/idle values
  97. #endif
  98. /* Just to show parameter modification. */
  99. _STARPU_MSG("%s has been modified : "
  100. "alpha=%f|beta=%f|gamma=%f|idle_power=%f !\n",
  101. d->name, alpha,beta,_gamma, idle_power);
  102. }
  103. #endif /* !STARPU_USE_TOP */
  104. static int _normalize_prio(int priority, int num_priorities, unsigned sched_ctx_id)
  105. {
  106. int min = starpu_sched_ctx_get_min_priority(sched_ctx_id);
  107. int max = starpu_sched_ctx_get_max_priority(sched_ctx_id);
  108. return ((num_priorities-1)/(max-min)) * (priority - min);
  109. }
  110. /* This is called when a transfer request is actually pushed to the worker */
  111. static void _starpu_fifo_task_transfer_started(struct _starpu_fifo_taskq *fifo, struct starpu_task *task, int num_priorities)
  112. {
  113. double transfer_model = task->predicted_transfer;
  114. if (isnan(transfer_model))
  115. return;
  116. /* We now start the transfer, move it from predicted to pipelined */
  117. fifo->exp_len -= transfer_model;
  118. fifo->pipeline_len += transfer_model;
  119. fifo->exp_start = starpu_timing_now() + fifo->pipeline_len;
  120. fifo->exp_end = fifo->exp_start + fifo->exp_len;
  121. if(num_priorities != -1)
  122. {
  123. int i;
  124. int task_prio = _normalize_prio(task->priority, num_priorities, task->sched_ctx);
  125. for(i = 0; i <= task_prio; i++)
  126. fifo->exp_len_per_priority[i] -= transfer_model;
  127. }
  128. }
  129. /* This is called when a task is actually pushed to the worker (i.e. the transfer finished */
  130. static void _starpu_fifo_task_started(struct _starpu_fifo_taskq *fifo, struct starpu_task *task, int num_priorities)
  131. {
  132. double model = task->predicted;
  133. double transfer_model = task->predicted_transfer;
  134. if(!isnan(transfer_model))
  135. /* The transfer is over, remove it from pipelined */
  136. fifo->pipeline_len -= transfer_model;
  137. if(!isnan(model))
  138. {
  139. /* We now start the computation, move it from predicted to pipelined */
  140. fifo->exp_len -= model;
  141. fifo->pipeline_len += model;
  142. fifo->exp_start = starpu_timing_now() + fifo->pipeline_len;
  143. fifo->exp_end= fifo->exp_start + fifo->exp_len;
  144. if(num_priorities != -1)
  145. {
  146. int i;
  147. int task_prio = _normalize_prio(task->priority, num_priorities, task->sched_ctx);
  148. for(i = 0; i <= task_prio; i++)
  149. fifo->exp_len_per_priority[i] -= model;
  150. }
  151. }
  152. }
  153. /* This is called when a task is actually finished */
  154. static void _starpu_fifo_task_finished(struct _starpu_fifo_taskq *fifo, struct starpu_task *task, int num_priorities STARPU_ATTRIBUTE_UNUSED)
  155. {
  156. if(!isnan(task->predicted))
  157. /* The execution is over, remove it from pipelined */
  158. fifo->pipeline_len -= task->predicted;
  159. fifo->exp_start = STARPU_MAX(starpu_timing_now() + fifo->pipeline_len, fifo->exp_start);
  160. fifo->exp_end = fifo->exp_start + fifo->exp_len;
  161. }
  162. static struct starpu_task *_starpu_fifo_pop_first_ready_task(struct _starpu_fifo_taskq *fifo_queue, unsigned workerid, int num_priorities)
  163. {
  164. struct starpu_task *task = NULL, *current;
  165. if (fifo_queue->ntasks == 0)
  166. return NULL;
  167. if (fifo_queue->ntasks > 0)
  168. {
  169. fifo_queue->ntasks--;
  170. task = starpu_task_list_front(&fifo_queue->taskq);
  171. if (STARPU_UNLIKELY(!task))
  172. return NULL;
  173. int first_task_priority = task->priority;
  174. current = task;
  175. int non_ready_best = INT_MAX;
  176. while (current)
  177. {
  178. int priority = current->priority;
  179. if (priority >= first_task_priority)
  180. {
  181. int non_ready = count_non_ready_buffers(current, workerid);
  182. if (non_ready < non_ready_best)
  183. {
  184. non_ready_best = non_ready;
  185. task = current;
  186. if (non_ready == 0)
  187. break;
  188. }
  189. }
  190. current = current->next;
  191. }
  192. if(num_priorities != -1)
  193. {
  194. int i;
  195. int task_prio = _normalize_prio(task->priority, num_priorities, task->sched_ctx);
  196. for(i = 0; i <= task_prio; i++)
  197. fifo_queue->ntasks_per_priority[i]--;
  198. }
  199. starpu_task_list_erase(&fifo_queue->taskq, task);
  200. }
  201. return task;
  202. }
  203. static struct starpu_task *dmda_pop_ready_task(unsigned sched_ctx_id)
  204. {
  205. struct _starpu_dmda_data *dt = (struct _starpu_dmda_data*)starpu_sched_ctx_get_policy_data(sched_ctx_id);
  206. struct starpu_task *task;
  207. unsigned workerid = starpu_worker_get_id_check();
  208. struct _starpu_fifo_taskq *fifo = dt->queue_array[workerid];
  209. /* Take the opportunity to update start time */
  210. fifo->exp_start = STARPU_MAX(starpu_timing_now(), fifo->exp_start);
  211. fifo->exp_end = fifo->exp_start + fifo->exp_len;
  212. task = _starpu_fifo_pop_first_ready_task(fifo, workerid, dt->num_priorities);
  213. if (task)
  214. {
  215. _starpu_fifo_task_transfer_started(fifo, task, dt->num_priorities);
  216. starpu_sched_ctx_list_task_counters_decrement(sched_ctx_id, workerid);
  217. #ifdef STARPU_VERBOSE
  218. if (task->cl)
  219. {
  220. int non_ready = count_non_ready_buffers(task, workerid);
  221. if (non_ready == 0)
  222. dt->ready_task_cnt++;
  223. }
  224. dt->total_task_cnt++;
  225. #endif
  226. }
  227. return task;
  228. }
  229. static struct starpu_task *dmda_pop_task(unsigned sched_ctx_id)
  230. {
  231. struct _starpu_dmda_data *dt = (struct _starpu_dmda_data*)starpu_sched_ctx_get_policy_data(sched_ctx_id);
  232. struct starpu_task *task;
  233. unsigned workerid = starpu_worker_get_id_check();
  234. struct _starpu_fifo_taskq *fifo = dt->queue_array[workerid];
  235. /* Take the opportunity to update start time */
  236. fifo->exp_start = STARPU_MAX(starpu_timing_now(), fifo->exp_start);
  237. fifo->exp_end = fifo->exp_start + fifo->exp_len;
  238. STARPU_ASSERT_MSG(fifo, "worker %u does not belong to ctx %u anymore.\n", workerid, sched_ctx_id);
  239. task = _starpu_fifo_pop_local_task(fifo);
  240. if (task)
  241. {
  242. _starpu_fifo_task_transfer_started(fifo, task, dt->num_priorities);
  243. starpu_sched_ctx_list_task_counters_decrement(sched_ctx_id, workerid);
  244. #ifdef STARPU_VERBOSE
  245. if (task->cl)
  246. {
  247. int non_ready = count_non_ready_buffers(task, workerid);
  248. if (non_ready == 0)
  249. dt->ready_task_cnt++;
  250. }
  251. dt->total_task_cnt++;
  252. #endif
  253. }
  254. return task;
  255. }
  256. static struct starpu_task *dmda_pop_every_task(unsigned sched_ctx_id)
  257. {
  258. struct _starpu_dmda_data *dt = (struct _starpu_dmda_data*)starpu_sched_ctx_get_policy_data(sched_ctx_id);
  259. struct starpu_task *new_list, *task;
  260. unsigned workerid = starpu_worker_get_id_check();
  261. struct _starpu_fifo_taskq *fifo = dt->queue_array[workerid];
  262. /* Take the opportunity to update start time */
  263. fifo->exp_start = STARPU_MAX(starpu_timing_now(), fifo->exp_start);
  264. fifo->exp_end = fifo->exp_start + fifo->exp_len;
  265. _starpu_worker_lock_self();
  266. new_list = _starpu_fifo_pop_every_task(fifo, workerid);
  267. _starpu_worker_unlock_self();
  268. starpu_sched_ctx_list_task_counters_reset(sched_ctx_id, workerid);
  269. for (task = new_list; task; task = task->next)
  270. _starpu_fifo_task_transfer_started(fifo, task, dt->num_priorities);
  271. return new_list;
  272. }
  273. static int push_task_on_best_worker(struct starpu_task *task, int best_workerid,
  274. double predicted, double predicted_transfer,
  275. int prio, unsigned sched_ctx_id)
  276. {
  277. struct _starpu_dmda_data *dt = (struct _starpu_dmda_data*)starpu_sched_ctx_get_policy_data(sched_ctx_id);
  278. /* make sure someone could execute that task ! */
  279. STARPU_ASSERT(best_workerid != -1);
  280. if (_starpu_get_nsched_ctxs() > 1)
  281. {
  282. _starpu_worker_relax_on();
  283. _starpu_sched_ctx_lock_write(sched_ctx_id);
  284. _starpu_worker_relax_off();
  285. if (_starpu_sched_ctx_worker_is_master_for_child_ctx(sched_ctx_id, best_workerid, task))
  286. task = NULL;
  287. _starpu_sched_ctx_unlock_write(sched_ctx_id);
  288. if (!task)
  289. return 0;
  290. }
  291. struct _starpu_fifo_taskq *fifo = dt->queue_array[best_workerid];
  292. double now = starpu_timing_now();
  293. #ifdef STARPU_USE_SC_HYPERVISOR
  294. starpu_sched_ctx_call_pushed_task_cb(best_workerid, sched_ctx_id);
  295. #endif //STARPU_USE_SC_HYPERVISOR
  296. _starpu_worker_lock(best_workerid);
  297. /* Sometimes workers didn't take the tasks as early as we expected */
  298. fifo->exp_start = isnan(fifo->exp_start) ? now + fifo->pipeline_len : STARPU_MAX(fifo->exp_start, now);
  299. fifo->exp_end = fifo->exp_start + fifo->exp_len;
  300. if ((now + predicted_transfer) < fifo->exp_end)
  301. {
  302. /* We may hope that the transfer will be finished by
  303. * the start of the task. */
  304. predicted_transfer = 0.0;
  305. }
  306. else
  307. {
  308. /* The transfer will not be finished by then, take the
  309. * remainder into account */
  310. predicted_transfer = (now + predicted_transfer) - fifo->exp_end;
  311. }
  312. if(!isnan(predicted_transfer))
  313. {
  314. fifo->exp_len += predicted_transfer;
  315. if(dt->num_priorities != -1)
  316. {
  317. int i;
  318. int task_prio = _normalize_prio(task->priority, dt->num_priorities, task->sched_ctx);
  319. for(i = 0; i <= task_prio; i++)
  320. fifo->exp_len_per_priority[i] += predicted_transfer;
  321. }
  322. }
  323. if(!isnan(predicted))
  324. {
  325. fifo->exp_len += predicted;
  326. if(dt->num_priorities != -1)
  327. {
  328. int i;
  329. int task_prio = _normalize_prio(task->priority, dt->num_priorities, task->sched_ctx);
  330. for(i = 0; i <= task_prio; i++)
  331. fifo->exp_len_per_priority[i] += predicted;
  332. }
  333. }
  334. fifo->exp_end = fifo->exp_start + fifo->exp_len;
  335. _starpu_worker_unlock(best_workerid);
  336. task->predicted = predicted;
  337. task->predicted_transfer = predicted_transfer;
  338. #ifdef STARPU_USE_TOP
  339. starpu_top_task_prevision(task, best_workerid,
  340. (unsigned long long)(fifo->exp_end-predicted)/1000,
  341. (unsigned long long)fifo->exp_end/1000);
  342. #endif /* !STARPU_USE_TOP */
  343. if (starpu_get_prefetch_flag())
  344. starpu_prefetch_task_input_for(task, best_workerid);
  345. STARPU_AYU_ADDTOTASKQUEUE(starpu_task_get_job_id(task), best_workerid);
  346. if (_starpu_get_nsched_ctxs() > 1)
  347. {
  348. unsigned stream_ctx_id = starpu_worker_get_sched_ctx_id_stream(best_workerid);
  349. if(stream_ctx_id != STARPU_NMAX_SCHED_CTXS)
  350. {
  351. _starpu_worker_relax_on();
  352. _starpu_sched_ctx_lock_write(sched_ctx_id);
  353. _starpu_worker_relax_off();
  354. starpu_sched_ctx_move_task_to_ctx_locked(task, stream_ctx_id, 0);
  355. starpu_sched_ctx_revert_task_counters_ctx_locked(sched_ctx_id, task->flops);
  356. _starpu_sched_ctx_unlock_write(sched_ctx_id);
  357. }
  358. }
  359. int ret = 0;
  360. if (prio)
  361. {
  362. _starpu_worker_lock(best_workerid);
  363. ret =_starpu_fifo_push_sorted_task(dt->queue_array[best_workerid], task);
  364. if(dt->num_priorities != -1)
  365. {
  366. int i;
  367. int task_prio = _normalize_prio(task->priority, dt->num_priorities, task->sched_ctx);
  368. for(i = 0; i <= task_prio; i++)
  369. dt->queue_array[best_workerid]->ntasks_per_priority[i]++;
  370. }
  371. #if !defined(STARPU_NON_BLOCKING_DRIVERS) || defined(STARPU_SIMGRID)
  372. starpu_wake_worker_locked(best_workerid);
  373. #endif
  374. starpu_push_task_end(task);
  375. _starpu_worker_unlock(best_workerid);
  376. }
  377. else
  378. {
  379. _starpu_worker_lock(best_workerid);
  380. starpu_task_list_push_back (&dt->queue_array[best_workerid]->taskq, task);
  381. dt->queue_array[best_workerid]->ntasks++;
  382. dt->queue_array[best_workerid]->nprocessed++;
  383. #if !defined(STARPU_NON_BLOCKING_DRIVERS) || defined(STARPU_SIMGRID)
  384. starpu_wake_worker_locked(best_workerid);
  385. #endif
  386. starpu_push_task_end(task);
  387. _starpu_worker_unlock(best_workerid);
  388. }
  389. starpu_sched_ctx_list_task_counters_increment(sched_ctx_id, best_workerid);
  390. return ret;
  391. }
  392. /* TODO: factorize with dmda!! */
  393. static int _dm_push_task(struct starpu_task *task, unsigned prio, unsigned sched_ctx_id)
  394. {
  395. struct _starpu_dmda_data *dt = (struct _starpu_dmda_data*)starpu_sched_ctx_get_policy_data(sched_ctx_id);
  396. int best = -1;
  397. double best_exp_end = 0.0;
  398. double model_best = 0.0;
  399. double transfer_model_best = 0.0;
  400. int ntasks_best = -1;
  401. double ntasks_best_end = 0.0;
  402. int calibrating = 0;
  403. /* A priori, we know all estimations */
  404. int unknown = 0;
  405. unsigned best_impl = 0;
  406. struct starpu_worker_collection *workers = starpu_sched_ctx_get_worker_collection(sched_ctx_id);
  407. struct starpu_sched_ctx_iterator it;
  408. double now = starpu_timing_now();
  409. workers->init_iterator_for_parallel_tasks(workers, &it, task);
  410. while(workers->has_next(workers, &it))
  411. {
  412. unsigned nimpl;
  413. unsigned impl_mask;
  414. unsigned worker = workers->get_next(workers, &it);
  415. struct _starpu_fifo_taskq *fifo = dt->queue_array[worker];
  416. struct starpu_perfmodel_arch* perf_arch = starpu_worker_get_perf_archtype(worker, sched_ctx_id);
  417. /* Sometimes workers didn't take the tasks as early as we expected */
  418. double exp_start = isnan(fifo->exp_start) ? now + fifo->pipeline_len : STARPU_MAX(fifo->exp_start, now);
  419. if (!starpu_worker_can_execute_task_impl(worker, task, &impl_mask))
  420. continue;
  421. for (nimpl = 0; nimpl < STARPU_MAXIMPLEMENTATIONS; nimpl++)
  422. {
  423. if (!(impl_mask & (1U << nimpl)))
  424. {
  425. /* no one on that queue may execute this task */
  426. continue;
  427. }
  428. double exp_end;
  429. double local_length = starpu_task_expected_length(task, perf_arch, nimpl);
  430. double local_penalty = starpu_task_expected_data_transfer_time_for(task, worker);
  431. double ntasks_end = fifo->ntasks / starpu_worker_get_relative_speedup(perf_arch);
  432. //_STARPU_DEBUG("Scheduler dm: task length (%lf) worker (%u) kernel (%u) \n", local_length,worker,nimpl);
  433. /*
  434. * This implements a default greedy scheduler for the
  435. * case of tasks which have no performance model, or
  436. * whose performance model is not calibrated yet.
  437. *
  438. * It simply uses the number of tasks already pushed to
  439. * the workers, divided by the relative performance of
  440. * a CPU and of a GPU.
  441. *
  442. * This is always computed, but the ntasks_best
  443. * selection is only really used if the task indeed has
  444. * no performance model, or is not calibrated yet.
  445. */
  446. if (ntasks_best == -1
  447. /* Always compute the greedy decision, at least for
  448. * the tasks with no performance model. */
  449. || (!calibrating && ntasks_end < ntasks_best_end)
  450. /* The performance model of this task is not
  451. * calibrated on this worker, try to run it there
  452. * to calibrate it there. */
  453. || (!calibrating && isnan(local_length))
  454. /* the performance model of this task is not
  455. * calibrated on this worker either, rather run it
  456. * there if this one is low on scheduled tasks. */
  457. || (calibrating && isnan(local_length) && ntasks_end < ntasks_best_end)
  458. )
  459. {
  460. ntasks_best_end = ntasks_end;
  461. ntasks_best = worker;
  462. best_impl = nimpl;
  463. }
  464. if (isnan(local_length))
  465. {
  466. /* we are calibrating, we want to speed-up calibration time
  467. * so we privilege non-calibrated tasks (but still
  468. * greedily distribute them to avoid dumb schedules) */
  469. static int warned;
  470. if (!warned)
  471. {
  472. warned = 1;
  473. _STARPU_DISP("Warning: performance model for %s not finished calibrating on worker %u, using a dumb scheduling heuristic for now\n", starpu_task_get_name(task), worker);
  474. }
  475. calibrating = 1;
  476. }
  477. if (isnan(local_length) || _STARPU_IS_ZERO(local_length))
  478. /* there is no prediction available for that task
  479. * with that arch yet, so switch to a greedy strategy */
  480. unknown = 1;
  481. if (unknown)
  482. continue;
  483. exp_end = exp_start + fifo->exp_len + local_length;
  484. if (best == -1 || exp_end < best_exp_end)
  485. {
  486. /* a better solution was found */
  487. best_exp_end = exp_end;
  488. best = worker;
  489. model_best = local_length;
  490. transfer_model_best = local_penalty;
  491. best_impl = nimpl;
  492. }
  493. }
  494. }
  495. if (unknown)
  496. {
  497. best = ntasks_best;
  498. model_best = 0.0;
  499. transfer_model_best = 0.0;
  500. #ifdef STARPU_VERBOSE
  501. dt->eager_task_cnt++;
  502. #endif
  503. }
  504. //_STARPU_DEBUG("Scheduler dm: kernel (%u)\n", best_impl);
  505. starpu_task_set_implementation(task, best_impl);
  506. starpu_sched_task_break(task);
  507. /* we should now have the best worker in variable "best" */
  508. return push_task_on_best_worker(task, best,
  509. model_best, transfer_model_best, prio, sched_ctx_id);
  510. }
  511. /* TODO: factorise CPU computations, expensive with a lot of cores */
  512. static void compute_all_performance_predictions(struct starpu_task *task,
  513. unsigned nworkers,
  514. double local_task_length[nworkers][STARPU_MAXIMPLEMENTATIONS],
  515. double exp_end[nworkers][STARPU_MAXIMPLEMENTATIONS],
  516. double *max_exp_endp,
  517. double *best_exp_endp,
  518. double local_data_penalty[nworkers][STARPU_MAXIMPLEMENTATIONS],
  519. double local_energy[nworkers][STARPU_MAXIMPLEMENTATIONS],
  520. int *forced_worker, int *forced_impl, unsigned sched_ctx_id, unsigned sorted_decision)
  521. {
  522. int calibrating = 0;
  523. double max_exp_end = DBL_MIN;
  524. double best_exp_end = DBL_MAX;
  525. int ntasks_best = -1;
  526. int nimpl_best = 0;
  527. double ntasks_best_end = 0.0;
  528. /* A priori, we know all estimations */
  529. int unknown = 0;
  530. unsigned worker_ctx = 0;
  531. int task_prio = 0;
  532. starpu_task_bundle_t bundle = task->bundle;
  533. struct _starpu_dmda_data *dt = (struct _starpu_dmda_data*)starpu_sched_ctx_get_policy_data(sched_ctx_id);
  534. if(sorted_decision && dt->num_priorities != -1)
  535. task_prio = _normalize_prio(task->priority, dt->num_priorities, sched_ctx_id);
  536. struct starpu_worker_collection *workers = starpu_sched_ctx_get_worker_collection(sched_ctx_id);
  537. double now = starpu_timing_now();
  538. struct starpu_sched_ctx_iterator it;
  539. workers->init_iterator_for_parallel_tasks(workers, &it, task);
  540. while(worker_ctx<nworkers && workers->has_next(workers, &it))
  541. {
  542. unsigned nimpl;
  543. unsigned impl_mask;
  544. unsigned workerid = workers->get_next(workers, &it);
  545. struct _starpu_fifo_taskq *fifo = dt->queue_array[workerid];
  546. struct starpu_perfmodel_arch* perf_arch = starpu_worker_get_perf_archtype(workerid, sched_ctx_id);
  547. unsigned memory_node = starpu_worker_get_memory_node(workerid);
  548. STARPU_ASSERT_MSG(fifo != NULL, "workerid %u ctx %u\n", workerid, sched_ctx_id);
  549. /* Sometimes workers didn't take the tasks as early as we expected */
  550. double exp_start = isnan(fifo->exp_start) ? now + fifo->pipeline_len : STARPU_MAX(fifo->exp_start, now);
  551. if (!starpu_worker_can_execute_task_impl(workerid, task, &impl_mask))
  552. continue;
  553. for (nimpl = 0; nimpl < STARPU_MAXIMPLEMENTATIONS; nimpl++)
  554. {
  555. if (!(impl_mask & (1U << nimpl)))
  556. {
  557. /* no one on that queue may execute this task */
  558. continue;
  559. }
  560. int fifo_ntasks = fifo->ntasks;
  561. double prev_exp_len = fifo->exp_len;
  562. /* consider the priority of the task when deciding on which workerid to schedule,
  563. compute the expected_end of the task if it is inserted before other tasks already scheduled */
  564. if(sorted_decision)
  565. {
  566. if(dt->num_priorities != -1)
  567. {
  568. prev_exp_len = fifo->exp_len_per_priority[task_prio];
  569. fifo_ntasks = fifo->ntasks_per_priority[task_prio];
  570. }
  571. else
  572. {
  573. _starpu_worker_lock(workerid);
  574. prev_exp_len = _starpu_fifo_get_exp_len_prev_task_list(fifo, task, workerid, nimpl, &fifo_ntasks);
  575. _starpu_worker_unlock(workerid);
  576. }
  577. }
  578. exp_end[worker_ctx][nimpl] = exp_start + prev_exp_len;
  579. if (exp_end[worker_ctx][nimpl] > max_exp_end)
  580. max_exp_end = exp_end[worker_ctx][nimpl];
  581. //_STARPU_DEBUG("Scheduler dmda: task length (%lf) workerid (%u) kernel (%u) \n", local_task_length[workerid][nimpl],workerid,nimpl);
  582. if (bundle)
  583. {
  584. /* TODO : conversion time */
  585. local_task_length[worker_ctx][nimpl] = starpu_task_bundle_expected_length(bundle, perf_arch, nimpl);
  586. local_data_penalty[worker_ctx][nimpl] = starpu_task_bundle_expected_data_transfer_time(bundle, memory_node);
  587. local_energy[worker_ctx][nimpl] = starpu_task_bundle_expected_energy(bundle, perf_arch,nimpl);
  588. }
  589. else
  590. {
  591. local_task_length[worker_ctx][nimpl] = starpu_task_expected_length(task, perf_arch, nimpl);
  592. local_data_penalty[worker_ctx][nimpl] = starpu_task_expected_data_transfer_time_for(task, workerid);
  593. local_energy[worker_ctx][nimpl] = starpu_task_expected_energy(task, perf_arch,nimpl);
  594. double conversion_time = starpu_task_expected_conversion_time(task, perf_arch, nimpl);
  595. if (conversion_time > 0.0)
  596. local_task_length[worker_ctx][nimpl] += conversion_time;
  597. }
  598. double ntasks_end = fifo_ntasks / starpu_worker_get_relative_speedup(perf_arch);
  599. /*
  600. * This implements a default greedy scheduler for the
  601. * case of tasks which have no performance model, or
  602. * whose performance model is not calibrated yet.
  603. *
  604. * It simply uses the number of tasks already pushed to
  605. * the workers, divided by the relative performance of
  606. * a CPU and of a GPU.
  607. *
  608. * This is always computed, but the ntasks_best
  609. * selection is only really used if the task indeed has
  610. * no performance model, or is not calibrated yet.
  611. */
  612. if (ntasks_best == -1
  613. /* Always compute the greedy decision, at least for
  614. * the tasks with no performance model. */
  615. || (!calibrating && ntasks_end < ntasks_best_end)
  616. /* The performance model of this task is not
  617. * calibrated on this workerid, try to run it there
  618. * to calibrate it there. */
  619. || (!calibrating && isnan(local_task_length[worker_ctx][nimpl]))
  620. /* the performance model of this task is not
  621. * calibrated on this workerid either, rather run it
  622. * there if this one is low on scheduled tasks. */
  623. || (calibrating && isnan(local_task_length[worker_ctx][nimpl]) && ntasks_end < ntasks_best_end)
  624. )
  625. {
  626. ntasks_best_end = ntasks_end;
  627. ntasks_best = workerid;
  628. nimpl_best = nimpl;
  629. }
  630. if (isnan(local_task_length[worker_ctx][nimpl]))
  631. /* we are calibrating, we want to speed-up calibration time
  632. * so we privilege non-calibrated tasks (but still
  633. * greedily distribute them to avoid dumb schedules) */
  634. calibrating = 1;
  635. if (isnan(local_task_length[worker_ctx][nimpl])
  636. || _STARPU_IS_ZERO(local_task_length[worker_ctx][nimpl]))
  637. /* there is no prediction available for that task
  638. * with that arch (yet or at all), so switch to a greedy strategy */
  639. unknown = 1;
  640. if (unknown)
  641. continue;
  642. double task_starting_time = STARPU_MAX(exp_start + prev_exp_len, now + local_data_penalty[worker_ctx][nimpl]);
  643. exp_end[worker_ctx][nimpl] = task_starting_time + local_task_length[worker_ctx][nimpl];
  644. if (exp_end[worker_ctx][nimpl] < best_exp_end)
  645. {
  646. /* a better solution was found */
  647. best_exp_end = exp_end[worker_ctx][nimpl];
  648. nimpl_best = nimpl;
  649. }
  650. if (isnan(local_energy[worker_ctx][nimpl]))
  651. local_energy[worker_ctx][nimpl] = 0.;
  652. }
  653. worker_ctx++;
  654. }
  655. *forced_worker = unknown?ntasks_best:-1;
  656. *forced_impl = unknown?nimpl_best:-1;
  657. #ifdef STARPU_VERBOSE
  658. if (unknown)
  659. {
  660. dt->eager_task_cnt++;
  661. }
  662. #endif
  663. *best_exp_endp = best_exp_end;
  664. *max_exp_endp = max_exp_end;
  665. }
  666. static double _dmda_push_task(struct starpu_task *task, unsigned prio, unsigned sched_ctx_id, unsigned simulate, unsigned sorted_decision)
  667. {
  668. /* find the queue */
  669. int best = -1, best_in_ctx = -1;
  670. int selected_impl = 0;
  671. double model_best = 0.0;
  672. double transfer_model_best = 0.0;
  673. /* this flag is set if the corresponding worker is selected because
  674. there is no performance prediction available yet */
  675. int forced_best = -1;
  676. int forced_impl = -1;
  677. struct _starpu_dmda_data *dt = (struct _starpu_dmda_data*)starpu_sched_ctx_get_policy_data(sched_ctx_id);
  678. struct starpu_worker_collection *workers = starpu_sched_ctx_get_worker_collection(sched_ctx_id);
  679. unsigned nworkers_ctx = workers->nworkers;
  680. double local_task_length[nworkers_ctx][STARPU_MAXIMPLEMENTATIONS];
  681. double local_data_penalty[nworkers_ctx][STARPU_MAXIMPLEMENTATIONS];
  682. double local_energy[nworkers_ctx][STARPU_MAXIMPLEMENTATIONS];
  683. /* Expected end of this task on the workers */
  684. double exp_end[nworkers_ctx][STARPU_MAXIMPLEMENTATIONS];
  685. /* This is the minimum among the exp_end[] matrix */
  686. double best_exp_end;
  687. /* This is the maximum termination time of already-scheduled tasks over all workers */
  688. double max_exp_end = 0.0;
  689. double fitness[nworkers_ctx][STARPU_MAXIMPLEMENTATIONS];
  690. compute_all_performance_predictions(task,
  691. nworkers_ctx,
  692. local_task_length,
  693. exp_end,
  694. &max_exp_end,
  695. &best_exp_end,
  696. local_data_penalty,
  697. local_energy,
  698. &forced_best,
  699. &forced_impl, sched_ctx_id, sorted_decision);
  700. if (forced_best == -1)
  701. {
  702. double best_fitness = -1;
  703. unsigned worker_ctx = 0;
  704. struct starpu_sched_ctx_iterator it;
  705. workers->init_iterator_for_parallel_tasks(workers, &it, task);
  706. while(worker_ctx < nworkers_ctx && workers->has_next(workers, &it))
  707. {
  708. unsigned worker = workers->get_next(workers, &it);
  709. unsigned nimpl;
  710. unsigned impl_mask;
  711. if (!starpu_worker_can_execute_task_impl(worker, task, &impl_mask))
  712. continue;
  713. for (nimpl = 0; nimpl < STARPU_MAXIMPLEMENTATIONS; nimpl++)
  714. {
  715. if (!(impl_mask & (1U << nimpl)))
  716. {
  717. /* no one on that queue may execute this task */
  718. continue;
  719. }
  720. fitness[worker_ctx][nimpl] = dt->alpha*(exp_end[worker_ctx][nimpl] - best_exp_end)
  721. + dt->beta*(local_data_penalty[worker_ctx][nimpl])
  722. + dt->_gamma*(local_energy[worker_ctx][nimpl]);
  723. if (exp_end[worker_ctx][nimpl] > max_exp_end)
  724. {
  725. /* This placement will make the computation
  726. * longer, take into account the idle
  727. * consumption of other cpus */
  728. fitness[worker_ctx][nimpl] += dt->_gamma * dt->idle_power * (exp_end[worker_ctx][nimpl] - max_exp_end) / 1000000.0;
  729. }
  730. if (best == -1 || fitness[worker_ctx][nimpl] < best_fitness)
  731. {
  732. /* we found a better solution */
  733. best_fitness = fitness[worker_ctx][nimpl];
  734. best = worker;
  735. best_in_ctx = worker_ctx;
  736. selected_impl = nimpl;
  737. //_STARPU_DEBUG("best fitness (worker %d) %e = alpha*(%e) + beta(%e) +gamma(%e)\n", worker, best_fitness, exp_end[worker][nimpl] - best_exp_end, local_data_penalty[worker][nimpl], local_energy[worker][nimpl]);
  738. }
  739. }
  740. worker_ctx++;
  741. }
  742. }
  743. STARPU_ASSERT(forced_best != -1 || best != -1);
  744. if (forced_best != -1)
  745. {
  746. /* there is no prediction available for that task
  747. * with that arch we want to speed-up calibration time
  748. * so we force this measurement */
  749. best = forced_best;
  750. selected_impl = forced_impl;
  751. model_best = 0.0;
  752. transfer_model_best = 0.0;
  753. }
  754. else if (task->bundle)
  755. {
  756. struct starpu_perfmodel_arch* perf_arch = starpu_worker_get_perf_archtype(best_in_ctx, sched_ctx_id);
  757. unsigned memory_node = starpu_worker_get_memory_node(best);
  758. model_best = starpu_task_expected_length(task, perf_arch, selected_impl);
  759. transfer_model_best = starpu_task_expected_data_transfer_time(memory_node, task);
  760. }
  761. else
  762. {
  763. model_best = local_task_length[best_in_ctx][selected_impl];
  764. transfer_model_best = local_data_penalty[best_in_ctx][selected_impl];
  765. }
  766. //_STARPU_DEBUG("Scheduler dmda: kernel (%u)\n", best_impl);
  767. starpu_task_set_implementation(task, selected_impl);
  768. starpu_sched_task_break(task);
  769. if(!simulate)
  770. {
  771. /* we should now have the best worker in variable "best" */
  772. return push_task_on_best_worker(task, best, model_best, transfer_model_best, prio, sched_ctx_id);
  773. }
  774. else
  775. {
  776. return exp_end[best_in_ctx][selected_impl] ;
  777. }
  778. }
  779. static int dmda_push_sorted_decision_task(struct starpu_task *task)
  780. {
  781. return _dmda_push_task(task, 1, task->sched_ctx, 0, 1);
  782. }
  783. static int dmda_push_sorted_task(struct starpu_task *task)
  784. {
  785. #ifdef STARPU_DEVEL
  786. #warning TODO: after defining a scheduling window, use that instead of empty_ctx_tasks
  787. #endif
  788. return _dmda_push_task(task, 1, task->sched_ctx, 0, 0);
  789. }
  790. static int dm_push_task(struct starpu_task *task)
  791. {
  792. return _dm_push_task(task, 0, task->sched_ctx);
  793. }
  794. static int dmda_push_task(struct starpu_task *task)
  795. {
  796. STARPU_ASSERT(task);
  797. return _dmda_push_task(task, 0, task->sched_ctx, 0, 0);
  798. }
  799. static double dmda_simulate_push_task(struct starpu_task *task)
  800. {
  801. STARPU_ASSERT(task);
  802. return _dmda_push_task(task, 0, task->sched_ctx, 1, 0);
  803. }
  804. static double dmda_simulate_push_sorted_task(struct starpu_task *task)
  805. {
  806. STARPU_ASSERT(task);
  807. return _dmda_push_task(task, 1, task->sched_ctx, 1, 0);
  808. }
  809. static double dmda_simulate_push_sorted_decision_task(struct starpu_task *task)
  810. {
  811. STARPU_ASSERT(task);
  812. return _dmda_push_task(task, 1, task->sched_ctx, 1, 1);
  813. }
  814. #ifdef NOTIFY_READY_SOON
  815. static void dmda_notify_ready_soon(void *data STARPU_ATTRIBUTE_UNUSED, struct starpu_task *task, double delay)
  816. {
  817. if (!task->cl)
  818. return;
  819. /* fprintf(stderr, "task %lu %p %p %s %s will be ready within %f\n", starpu_task_get_job_id(task), task, task->cl, task->cl->name, task->cl->model?task->cl->model->symbol : NULL, delay); */
  820. /* TODO: do something with it */
  821. }
  822. #endif
  823. static void dmda_add_workers(unsigned sched_ctx_id, int *workerids, unsigned nworkers)
  824. {
  825. struct _starpu_dmda_data *dt = (struct _starpu_dmda_data*)starpu_sched_ctx_get_policy_data(sched_ctx_id);
  826. unsigned i;
  827. for (i = 0; i < nworkers; i++)
  828. {
  829. struct _starpu_fifo_taskq *q;
  830. int workerid = workerids[i];
  831. /* if the worker has alreadry belonged to this context
  832. the queue and the synchronization variables have been already initialized */
  833. q = dt->queue_array[workerid];
  834. if(q == NULL)
  835. {
  836. q = dt->queue_array[workerid] = _starpu_create_fifo();
  837. /* These are only stats, they can be read with races */
  838. STARPU_HG_DISABLE_CHECKING(q->exp_start);
  839. STARPU_HG_DISABLE_CHECKING(q->exp_len);
  840. STARPU_HG_DISABLE_CHECKING(q->exp_end);
  841. }
  842. if(dt->num_priorities != -1)
  843. {
  844. _STARPU_MALLOC(q->exp_len_per_priority, dt->num_priorities*sizeof(double));
  845. _STARPU_MALLOC(q->ntasks_per_priority, dt->num_priorities*sizeof(unsigned));
  846. int j;
  847. for(j = 0; j < dt->num_priorities; j++)
  848. {
  849. q->exp_len_per_priority[j] = 0.0;
  850. q->ntasks_per_priority[j] = 0;
  851. }
  852. }
  853. }
  854. }
  855. static void dmda_remove_workers(unsigned sched_ctx_id, int *workerids, unsigned nworkers)
  856. {
  857. struct _starpu_dmda_data *dt = (struct _starpu_dmda_data*)starpu_sched_ctx_get_policy_data(sched_ctx_id);
  858. unsigned i;
  859. for (i = 0; i < nworkers; i++)
  860. {
  861. int workerid = workerids[i];
  862. if(dt->queue_array[workerid] != NULL)
  863. {
  864. if(dt->num_priorities != -1)
  865. {
  866. free(dt->queue_array[workerid]->exp_len_per_priority);
  867. free(dt->queue_array[workerid]->ntasks_per_priority);
  868. }
  869. _starpu_destroy_fifo(dt->queue_array[workerid]);
  870. dt->queue_array[workerid] = NULL;
  871. }
  872. }
  873. }
  874. static void initialize_dmda_policy(unsigned sched_ctx_id)
  875. {
  876. struct _starpu_dmda_data *dt;
  877. _STARPU_CALLOC(dt, 1, sizeof(struct _starpu_dmda_data));
  878. starpu_sched_ctx_set_policy_data(sched_ctx_id, (void*)dt);
  879. _STARPU_MALLOC(dt->queue_array, STARPU_NMAXWORKERS*sizeof(struct _starpu_fifo_taskq*));
  880. int i;
  881. for(i = 0; i < STARPU_NMAXWORKERS; i++)
  882. dt->queue_array[i] = NULL;
  883. dt->alpha = starpu_get_env_float_default("STARPU_SCHED_ALPHA", _STARPU_SCHED_ALPHA_DEFAULT);
  884. dt->beta = starpu_get_env_float_default("STARPU_SCHED_BETA", _STARPU_SCHED_BETA_DEFAULT);
  885. dt->_gamma = starpu_get_env_float_default("STARPU_SCHED_GAMMA", _STARPU_SCHED_GAMMA_DEFAULT);
  886. dt->idle_power = starpu_get_env_float_default("STARPU_IDLE_POWER", 0.0);
  887. if(starpu_sched_ctx_min_priority_is_set(sched_ctx_id) != 0 && starpu_sched_ctx_max_priority_is_set(sched_ctx_id) != 0)
  888. dt->num_priorities = starpu_sched_ctx_get_max_priority(sched_ctx_id) - starpu_sched_ctx_get_min_priority(sched_ctx_id) + 1;
  889. else
  890. dt->num_priorities = -1;
  891. #ifdef STARPU_USE_TOP
  892. /* FIXME: broken, needs to access context variable */
  893. starpu_top_register_parameter_float("DMDA_ALPHA", &alpha,
  894. alpha_minimum, alpha_maximum, param_modified);
  895. starpu_top_register_parameter_float("DMDA_BETA", &beta,
  896. beta_minimum, beta_maximum, param_modified);
  897. starpu_top_register_parameter_float("DMDA_GAMMA", &_gamma,
  898. gamma_minimum, gamma_maximum, param_modified);
  899. starpu_top_register_parameter_float("DMDA_IDLE_POWER", &idle_power,
  900. idle_power_minimum, idle_power_maximum, param_modified);
  901. #endif /* !STARPU_USE_TOP */
  902. #ifdef NOTIFY_READY_SOON
  903. starpu_task_notify_ready_soon_register(dmda_notify_ready_soon, dt);
  904. #endif
  905. }
  906. static void initialize_dmda_sorted_policy(unsigned sched_ctx_id)
  907. {
  908. initialize_dmda_policy(sched_ctx_id);
  909. /* The application may use any integer */
  910. if (starpu_sched_ctx_min_priority_is_set(sched_ctx_id) == 0)
  911. starpu_sched_ctx_set_min_priority(sched_ctx_id, INT_MIN);
  912. if (starpu_sched_ctx_max_priority_is_set(sched_ctx_id) == 0)
  913. starpu_sched_ctx_set_max_priority(sched_ctx_id, INT_MAX);
  914. }
  915. static void deinitialize_dmda_policy(unsigned sched_ctx_id)
  916. {
  917. struct _starpu_dmda_data *dt = (struct _starpu_dmda_data*)starpu_sched_ctx_get_policy_data(sched_ctx_id);
  918. #ifdef STARPU_VERBOSE
  919. {
  920. struct _starpu_sched_ctx *sched_ctx = _starpu_get_sched_ctx_struct(sched_ctx_id);
  921. long int modelled_task_cnt = dt->total_task_cnt - dt->eager_task_cnt;
  922. _STARPU_DEBUG("%s sched policy (sched_ctx %u): total_task_cnt %ld ready_task_cnt %ld (%.1f%%), modelled_task_cnt = %ld (%.1f%%)%s\n",
  923. sched_ctx->sched_policy?sched_ctx->sched_policy->policy_name:"<none>",
  924. sched_ctx_id,
  925. dt->total_task_cnt,
  926. dt->ready_task_cnt,
  927. (100.0f*dt->ready_task_cnt)/dt->total_task_cnt,
  928. modelled_task_cnt,
  929. (100.0f*modelled_task_cnt)/dt->total_task_cnt,
  930. modelled_task_cnt==0?" *** Check if performance models are enabled and converging on a per-codelet basis, or use an non-modeling scheduling policy. ***":"");
  931. }
  932. #endif
  933. free(dt->queue_array);
  934. free(dt);
  935. }
  936. /* dmda_pre_exec_hook is called right after the data transfer is done and right
  937. * before the computation to begin, it is useful to update more precisely the
  938. * value of the expected start, end, length, etc... */
  939. static void dmda_pre_exec_hook(struct starpu_task *task, unsigned sched_ctx_id)
  940. {
  941. unsigned workerid = starpu_worker_get_id_check();
  942. struct _starpu_dmda_data *dt = (struct _starpu_dmda_data*)starpu_sched_ctx_get_policy_data(sched_ctx_id);
  943. struct _starpu_fifo_taskq *fifo = dt->queue_array[workerid];
  944. const double now = starpu_timing_now();
  945. /* Once the task is executing, we can update the predicted amount
  946. * of work. */
  947. _starpu_worker_lock_self();
  948. _starpu_fifo_task_started(fifo, task, dt->num_priorities);
  949. /* Take the opportunity to update start time */
  950. fifo->exp_start = STARPU_MAX(now + fifo->pipeline_len, fifo->exp_start);
  951. fifo->exp_end = fifo->exp_start + fifo->exp_len;
  952. _starpu_worker_unlock_self();
  953. }
  954. static void dmda_push_task_notify(struct starpu_task *task, int workerid, int perf_workerid, unsigned sched_ctx_id)
  955. {
  956. struct _starpu_dmda_data *dt = (struct _starpu_dmda_data*)starpu_sched_ctx_get_policy_data(sched_ctx_id);
  957. struct _starpu_fifo_taskq *fifo = dt->queue_array[workerid];
  958. /* Compute the expected penality */
  959. struct starpu_perfmodel_arch *perf_arch = starpu_worker_get_perf_archtype(perf_workerid, sched_ctx_id);
  960. double predicted = starpu_task_expected_length(task, perf_arch,
  961. starpu_task_get_implementation(task));
  962. double predicted_transfer = starpu_task_expected_data_transfer_time_for(task, workerid);
  963. double now = starpu_timing_now();
  964. /* Update the predictions */
  965. _starpu_worker_lock(workerid);
  966. /* Sometimes workers didn't take the tasks as early as we expected */
  967. fifo->exp_start = isnan(fifo->exp_start) ? now + fifo->pipeline_len : STARPU_MAX(fifo->exp_start, now);
  968. fifo->exp_end = fifo->exp_start + fifo->exp_len;
  969. /* If there is no prediction available, we consider the task has a null length */
  970. if (!isnan(predicted_transfer))
  971. {
  972. if (now + predicted_transfer < fifo->exp_end)
  973. {
  974. /* We may hope that the transfer will be finished by
  975. * the start of the task. */
  976. predicted_transfer = 0;
  977. }
  978. else
  979. {
  980. /* The transfer will not be finished by then, take the
  981. * remainder into account */
  982. predicted_transfer = (now + predicted_transfer) - fifo->exp_end;
  983. }
  984. task->predicted_transfer = predicted_transfer;
  985. fifo->exp_end += predicted_transfer;
  986. fifo->exp_len += predicted_transfer;
  987. if(dt->num_priorities != -1)
  988. {
  989. int i;
  990. int task_prio = _normalize_prio(task->priority, dt->num_priorities, task->sched_ctx);
  991. for(i = 0; i <= task_prio; i++)
  992. fifo->exp_len_per_priority[i] += predicted_transfer;
  993. }
  994. }
  995. /* If there is no prediction available, we consider the task has a null length */
  996. if (!isnan(predicted))
  997. {
  998. task->predicted = predicted;
  999. fifo->exp_end += predicted;
  1000. fifo->exp_len += predicted;
  1001. if(dt->num_priorities != -1)
  1002. {
  1003. int i;
  1004. int task_prio = _normalize_prio(task->priority, dt->num_priorities, task->sched_ctx);
  1005. for(i = 0; i <= task_prio; i++)
  1006. fifo->exp_len_per_priority[i] += predicted;
  1007. }
  1008. }
  1009. if(dt->num_priorities != -1)
  1010. {
  1011. int i;
  1012. int task_prio = _normalize_prio(task->priority, dt->num_priorities, task->sched_ctx);
  1013. for(i = 0; i <= task_prio; i++)
  1014. fifo->ntasks_per_priority[i]++;
  1015. }
  1016. fifo->ntasks++;
  1017. _starpu_worker_unlock(workerid);
  1018. }
  1019. static void dmda_post_exec_hook(struct starpu_task * task, unsigned sched_ctx_id)
  1020. {
  1021. struct _starpu_dmda_data *dt = (struct _starpu_dmda_data*)starpu_sched_ctx_get_policy_data(sched_ctx_id);
  1022. unsigned workerid = starpu_worker_get_id_check();
  1023. struct _starpu_fifo_taskq *fifo = dt->queue_array[workerid];
  1024. _starpu_worker_lock_self();
  1025. _starpu_fifo_task_finished(fifo, task, dt->num_priorities);
  1026. _starpu_worker_unlock_self();
  1027. }
  1028. struct starpu_sched_policy _starpu_sched_dm_policy =
  1029. {
  1030. .init_sched = initialize_dmda_policy,
  1031. .deinit_sched = deinitialize_dmda_policy,
  1032. .add_workers = dmda_add_workers ,
  1033. .remove_workers = dmda_remove_workers,
  1034. .push_task = dm_push_task,
  1035. .simulate_push_task = NULL,
  1036. .pop_task = dmda_pop_task,
  1037. .pre_exec_hook = dmda_pre_exec_hook,
  1038. .post_exec_hook = dmda_post_exec_hook,
  1039. .pop_every_task = dmda_pop_every_task,
  1040. .policy_name = "dm",
  1041. .policy_description = "performance model",
  1042. .worker_type = STARPU_WORKER_LIST,
  1043. };
  1044. struct starpu_sched_policy _starpu_sched_dmda_policy =
  1045. {
  1046. .init_sched = initialize_dmda_policy,
  1047. .deinit_sched = deinitialize_dmda_policy,
  1048. .add_workers = dmda_add_workers ,
  1049. .remove_workers = dmda_remove_workers,
  1050. .push_task = dmda_push_task,
  1051. .simulate_push_task = dmda_simulate_push_task,
  1052. .push_task_notify = dmda_push_task_notify,
  1053. .pop_task = dmda_pop_task,
  1054. .pre_exec_hook = dmda_pre_exec_hook,
  1055. .post_exec_hook = dmda_post_exec_hook,
  1056. .pop_every_task = dmda_pop_every_task,
  1057. .policy_name = "dmda",
  1058. .policy_description = "data-aware performance model",
  1059. .worker_type = STARPU_WORKER_LIST,
  1060. };
  1061. struct starpu_sched_policy _starpu_sched_dmda_sorted_policy =
  1062. {
  1063. .init_sched = initialize_dmda_sorted_policy,
  1064. .deinit_sched = deinitialize_dmda_policy,
  1065. .add_workers = dmda_add_workers ,
  1066. .remove_workers = dmda_remove_workers,
  1067. .push_task = dmda_push_sorted_task,
  1068. .simulate_push_task = dmda_simulate_push_sorted_task,
  1069. .push_task_notify = dmda_push_task_notify,
  1070. .pop_task = dmda_pop_ready_task,
  1071. .pre_exec_hook = dmda_pre_exec_hook,
  1072. .post_exec_hook = dmda_post_exec_hook,
  1073. .pop_every_task = dmda_pop_every_task,
  1074. .policy_name = "dmdas",
  1075. .policy_description = "data-aware performance model (sorted)",
  1076. .worker_type = STARPU_WORKER_LIST,
  1077. };
  1078. struct starpu_sched_policy _starpu_sched_dmda_sorted_decision_policy =
  1079. {
  1080. .init_sched = initialize_dmda_sorted_policy,
  1081. .deinit_sched = deinitialize_dmda_policy,
  1082. .add_workers = dmda_add_workers ,
  1083. .remove_workers = dmda_remove_workers,
  1084. .push_task = dmda_push_sorted_decision_task,
  1085. .simulate_push_task = dmda_simulate_push_sorted_decision_task,
  1086. .push_task_notify = dmda_push_task_notify,
  1087. .pop_task = dmda_pop_ready_task,
  1088. .pre_exec_hook = dmda_pre_exec_hook,
  1089. .post_exec_hook = dmda_post_exec_hook,
  1090. .pop_every_task = dmda_pop_every_task,
  1091. .policy_name = "dmdasd",
  1092. .policy_description = "data-aware performance model (sorted decision)",
  1093. .worker_type = STARPU_WORKER_LIST,
  1094. };
  1095. struct starpu_sched_policy _starpu_sched_dmda_ready_policy =
  1096. {
  1097. .init_sched = initialize_dmda_policy,
  1098. .deinit_sched = deinitialize_dmda_policy,
  1099. .add_workers = dmda_add_workers ,
  1100. .remove_workers = dmda_remove_workers,
  1101. .push_task = dmda_push_task,
  1102. .simulate_push_task = dmda_simulate_push_task,
  1103. .push_task_notify = dmda_push_task_notify,
  1104. .pop_task = dmda_pop_ready_task,
  1105. .pre_exec_hook = dmda_pre_exec_hook,
  1106. .post_exec_hook = dmda_post_exec_hook,
  1107. .pop_every_task = dmda_pop_every_task,
  1108. .policy_name = "dmdar",
  1109. .policy_description = "data-aware performance model (ready)",
  1110. .worker_type = STARPU_WORKER_LIST,
  1111. };