deque_modeling_policy_data_aware.c 42 KB

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