heft.c 21 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634
  1. /* StarPU --- Runtime system for heterogeneous multicore architectures.
  2. *
  3. * Copyright (C) 2010, 2011-2012 Université de Bordeaux 1
  4. * Copyright (C) 2010, 2011, 2012 Centre National de la Recherche Scientifique
  5. * Copyright (C) 2011 Télécom-SudParis
  6. * Copyright (C) 2011 INRIA
  7. *
  8. * StarPU is free software; you can redistribute it and/or modify
  9. * it under the terms of the GNU Lesser General Public License as published by
  10. * the Free Software Foundation; either version 2.1 of the License, or (at
  11. * your option) any later version.
  12. *
  13. * StarPU is distributed in the hope that it will be useful, but
  14. * WITHOUT ANY WARRANTY; without even the implied warranty of
  15. * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
  16. *
  17. * See the GNU Lesser General Public License in COPYING.LGPL for more details.
  18. */
  19. /* Distributed queues using performance modeling to assign tasks */
  20. #include <float.h>
  21. #include <core/workers.h>
  22. #include <core/perfmodel/perfmodel.h>
  23. #include <core/task_bundle.h>
  24. #include <core/workers.h>
  25. #include <starpu_parameters.h>
  26. #include <starpu_task_bundle.h>
  27. #include <starpu_top.h>
  28. #ifndef DBL_MIN
  29. #define DBL_MIN __DBL_MIN__
  30. #endif
  31. #ifndef DBL_MAX
  32. #define DBL_MAX __DBL_MAX__
  33. #endif
  34. static double exp_start[STARPU_NMAXWORKERS]; /* of the first queued task */
  35. static double exp_end[STARPU_NMAXWORKERS]; /* of the set of queued tasks */
  36. static double exp_len[STARPU_NMAXWORKERS]; /* of the last queued task */
  37. static double ntasks[STARPU_NMAXWORKERS];
  38. typedef struct {
  39. double alpha;
  40. double beta;
  41. double _gamma;
  42. double idle_power;
  43. } heft_data;
  44. const float alpha_minimum=0;
  45. const float alpha_maximum=10.0;
  46. const float beta_minimum=0;
  47. const float beta_maximum=10.0;
  48. const float gamma_minimum=0;
  49. const float gamma_maximum=10000.0;
  50. const float idle_power_minimum=0;
  51. const float idle_power_maximum=10000.0;
  52. static void param_modified(struct starpu_top_param* d)
  53. {
  54. //just to show parameter modification
  55. fprintf(stderr,"%s has been modified : %f !\n", d->name, d->value);
  56. }
  57. static void heft_add_workers(unsigned sched_ctx_id, int *workerids, unsigned nworkers)
  58. {
  59. int workerid;
  60. unsigned i;
  61. for (i = 0; i < nworkers; i++)
  62. {
  63. workerid = workerids[i];
  64. struct _starpu_worker *workerarg = _starpu_get_worker_struct(workerid);
  65. /* init these structures only once for each worker */
  66. if(!workerarg->has_prev_init)
  67. {
  68. exp_start[workerid] = starpu_timing_now();
  69. exp_len[workerid] = 0.0;
  70. exp_end[workerid] = exp_start[workerid];
  71. ntasks[workerid] = 0;
  72. workerarg->has_prev_init = 1;
  73. }
  74. /* we push the tasks on the local lists of the workers
  75. therefore the synchronisations mechanisms of the strategy
  76. are the global ones */
  77. starpu_worker_set_sched_condition(sched_ctx_id, workerid, &workerarg->sched_mutex, &workerarg->sched_cond);
  78. }
  79. }
  80. static void heft_remove_workers(unsigned sched_ctx_id, int *workerids, unsigned nworkers)
  81. {
  82. int workerid;
  83. unsigned i;
  84. for (i = 0; i < nworkers; i++)
  85. {
  86. workerid = workerids[i];
  87. starpu_worker_set_sched_condition(sched_ctx_id, workerid, NULL, NULL);
  88. }
  89. }
  90. static void heft_init(unsigned sched_ctx_id)
  91. {
  92. starpu_create_worker_collection_for_sched_ctx(sched_ctx_id, WORKER_LIST);
  93. heft_data *hd = (heft_data*)malloc(sizeof(heft_data));
  94. hd->alpha = _STARPU_DEFAULT_ALPHA;
  95. hd->beta = _STARPU_DEFAULT_BETA;
  96. hd->_gamma = _STARPU_DEFAULT_GAMMA;
  97. hd->idle_power = 0.0;
  98. starpu_set_sched_ctx_policy_data(sched_ctx_id, (void*)hd);
  99. const char *strval_alpha = getenv("STARPU_SCHED_ALPHA");
  100. if (strval_alpha)
  101. hd->alpha = atof(strval_alpha);
  102. const char *strval_beta = getenv("STARPU_SCHED_BETA");
  103. if (strval_beta)
  104. hd->beta = atof(strval_beta);
  105. const char *strval_gamma = getenv("STARPU_SCHED_GAMMA");
  106. if (strval_gamma)
  107. hd->_gamma = atof(strval_gamma);
  108. const char *strval_idle_power = getenv("STARPU_IDLE_POWER");
  109. if (strval_idle_power)
  110. hd->idle_power = atof(strval_idle_power);
  111. starpu_top_register_parameter_float("HEFT_ALPHA", &hd->alpha, alpha_minimum,alpha_maximum,param_modified);
  112. starpu_top_register_parameter_float("HEFT_BETA", &hd->beta, beta_minimum,beta_maximum,param_modified);
  113. starpu_top_register_parameter_float("HEFT_GAMMA", &hd->_gamma, gamma_minimum,gamma_maximum,param_modified);
  114. starpu_top_register_parameter_float("HEFT_IDLE_POWER", &hd->idle_power, idle_power_minimum,idle_power_maximum,param_modified);
  115. }
  116. /* heft_pre_exec_hook is called right after the data transfer is done and right before
  117. * the computation to begin, it is useful to update more precisely the value
  118. * of the expected start, end, length, etc... */
  119. static void heft_pre_exec_hook(struct starpu_task *task)
  120. {
  121. unsigned sched_ctx_id = task->sched_ctx;
  122. int workerid = starpu_worker_get_id();
  123. double model = task->predicted;
  124. double transfer_model = task->predicted_transfer;
  125. pthread_mutex_t *sched_mutex;
  126. pthread_cond_t *sched_cond;
  127. starpu_worker_get_sched_condition(sched_ctx_id, workerid, &sched_mutex, &sched_cond);
  128. /* Once the task is executing, we can update the predicted amount
  129. * of work. */
  130. _STARPU_PTHREAD_MUTEX_LOCK(sched_mutex);
  131. exp_len[workerid] -= model + transfer_model;
  132. exp_start[workerid] = starpu_timing_now() + model;
  133. exp_end[workerid] = exp_start[workerid] + exp_len[workerid];
  134. ntasks[workerid]--;
  135. _STARPU_PTHREAD_MUTEX_UNLOCK(sched_mutex);
  136. }
  137. static void heft_push_task_notify(struct starpu_task *task, int workerid)
  138. {
  139. unsigned sched_ctx_id = task->sched_ctx;
  140. /* Compute the expected penality */
  141. enum starpu_perf_archtype perf_arch = starpu_worker_get_perf_archtype(workerid);
  142. unsigned memory_node = starpu_worker_get_memory_node(workerid);
  143. double predicted = starpu_task_expected_length(task, perf_arch,
  144. _starpu_get_job_associated_to_task(task)->nimpl);
  145. double predicted_transfer = starpu_task_expected_data_transfer_time(memory_node, task);
  146. pthread_mutex_t *sched_mutex;
  147. pthread_cond_t *sched_cond;
  148. starpu_worker_get_sched_condition(sched_ctx_id, workerid, &sched_mutex, &sched_cond);
  149. /* Update the predictions */
  150. _STARPU_PTHREAD_MUTEX_LOCK(sched_mutex);
  151. /* Sometimes workers didn't take the tasks as early as we expected */
  152. exp_start[workerid] = STARPU_MAX(exp_start[workerid], starpu_timing_now());
  153. exp_end[workerid] = exp_start[workerid] + exp_len[workerid];
  154. /* If there is no prediction available, we consider the task has a null length */
  155. if (!isnan(predicted))
  156. {
  157. task->predicted = predicted;
  158. exp_end[workerid] += predicted;
  159. exp_len[workerid] += predicted;
  160. }
  161. /* If there is no prediction available, we consider the task has a null length */
  162. if (!isnan(predicted_transfer))
  163. {
  164. if (starpu_timing_now() + predicted_transfer < exp_end[workerid])
  165. {
  166. /* We may hope that the transfer will be finished by
  167. * the start of the task. */
  168. predicted_transfer = 0;
  169. }
  170. else
  171. {
  172. /* The transfer will not be finished by then, take the
  173. * remainder into account */
  174. predicted_transfer = (starpu_timing_now() + predicted_transfer) - exp_end[workerid];
  175. }
  176. task->predicted_transfer = predicted_transfer;
  177. exp_end[workerid] += predicted_transfer;
  178. exp_len[workerid] += predicted_transfer;
  179. }
  180. ntasks[workerid]++;
  181. _STARPU_PTHREAD_MUTEX_UNLOCK(sched_mutex);
  182. }
  183. static int push_task_on_best_worker(struct starpu_task *task, int best_workerid, double predicted, double predicted_transfer, int prio, unsigned sched_ctx_id)
  184. {
  185. /* make sure someone coule execute that task ! */
  186. STARPU_ASSERT(best_workerid != -1);
  187. pthread_mutex_t *sched_mutex;
  188. pthread_cond_t *sched_cond;
  189. starpu_worker_get_sched_condition(sched_ctx_id, best_workerid, &sched_mutex, &sched_cond);
  190. #ifdef STARPU_USE_SCHED_CTX_HYPERVISOR
  191. starpu_call_pushed_task_cb(best_workerid, sched_ctx_id);
  192. #endif //STARPU_USE_SCHED_CTX_HYPERVISOR
  193. _STARPU_PTHREAD_MUTEX_LOCK(sched_mutex);
  194. /* Sometimes workers didn't take the tasks as early as we expected */
  195. exp_start[best_workerid] = STARPU_MAX(exp_start[best_workerid], starpu_timing_now());
  196. exp_end[best_workerid] = exp_start[best_workerid] + exp_len[best_workerid];
  197. exp_end[best_workerid] += predicted;
  198. exp_len[best_workerid] += predicted;
  199. if (starpu_timing_now() + predicted_transfer < exp_end[best_workerid])
  200. {
  201. /* We may hope that the transfer will be finished by
  202. * the start of the task. */
  203. predicted_transfer = 0;
  204. }
  205. else
  206. {
  207. /* The transfer will not be finished by then, take the
  208. * remainder into account */
  209. predicted_transfer = (starpu_timing_now() + predicted_transfer) - exp_end[best_workerid];
  210. }
  211. exp_end[best_workerid] += predicted_transfer;
  212. exp_len[best_workerid] += predicted_transfer;
  213. ntasks[best_workerid]++;
  214. _STARPU_PTHREAD_MUTEX_UNLOCK(sched_mutex);
  215. task->predicted = predicted;
  216. task->predicted_transfer = predicted_transfer;
  217. if (_starpu_top_status_get())
  218. _starpu_top_task_prevision(task, best_workerid,
  219. (unsigned long long)(exp_end[best_workerid]-predicted)/1000,
  220. (unsigned long long)exp_end[best_workerid]/1000);
  221. if (starpu_get_prefetch_flag())
  222. {
  223. unsigned memory_node = starpu_worker_get_memory_node(best_workerid);
  224. starpu_prefetch_task_input_on_node(task, memory_node);
  225. }
  226. //_STARPU_DEBUG("Heft : pushing local task\n");
  227. return starpu_push_local_task(best_workerid, task, prio);
  228. }
  229. /* TODO: factorize with dmda!! */
  230. static void compute_all_performance_predictions(struct starpu_task *task,
  231. double (*local_task_length)[STARPU_MAXIMPLEMENTATIONS],
  232. double (*exp_end)[STARPU_MAXIMPLEMENTATIONS],
  233. double *max_exp_endp, double *best_exp_endp,
  234. double (*local_data_penalty)[STARPU_MAXIMPLEMENTATIONS],
  235. double (*local_power)[STARPU_MAXIMPLEMENTATIONS],
  236. int *forced_worker, int *forced_impl,
  237. starpu_task_bundle_t bundle,
  238. unsigned sched_ctx_id)
  239. {
  240. int calibrating = 0;
  241. double max_exp_end = DBL_MIN;
  242. double best_exp_end = DBL_MAX;
  243. int ntasks_best = -1;
  244. int nimpl_best = 0;
  245. double ntasks_best_end = 0.0;
  246. /* A priori, we know all estimations */
  247. int unknown = 0;
  248. int worker, worker_ctx = 0;
  249. unsigned nimpl;
  250. struct worker_collection *workers = starpu_get_worker_collection_of_sched_ctx(sched_ctx_id);
  251. while(workers->has_next(workers))
  252. {
  253. worker = workers->get_next(workers);
  254. for (nimpl = 0; nimpl <STARPU_MAXIMPLEMENTATIONS; nimpl++)
  255. {
  256. /* Sometimes workers didn't take the tasks as early as we expected */
  257. pthread_mutex_t *sched_mutex;
  258. pthread_cond_t *sched_cond;
  259. starpu_worker_get_sched_condition(sched_ctx_id, worker, &sched_mutex, &sched_cond);
  260. _STARPU_PTHREAD_MUTEX_LOCK(sched_mutex);
  261. exp_start[worker] = STARPU_MAX(exp_start[worker], starpu_timing_now());
  262. exp_end[worker_ctx][nimpl] = exp_start[worker] + exp_len[worker];
  263. if (exp_end[worker_ctx][nimpl] > max_exp_end)
  264. max_exp_end = exp_end[worker_ctx][nimpl];
  265. _STARPU_PTHREAD_MUTEX_UNLOCK(sched_mutex);
  266. if (!starpu_worker_can_execute_task(worker, task, nimpl))
  267. {
  268. /* no one on that queue may execute this task */
  269. // worker_ctx++;
  270. continue;
  271. }
  272. enum starpu_perf_archtype perf_arch = starpu_worker_get_perf_archtype(worker);
  273. unsigned memory_node = starpu_worker_get_memory_node(worker);
  274. if (bundle)
  275. {
  276. /* TODO : conversion time */
  277. local_task_length[worker_ctx][nimpl] = starpu_task_bundle_expected_length(bundle, perf_arch, nimpl);
  278. local_data_penalty[worker_ctx][nimpl] = starpu_task_bundle_expected_data_transfer_time(bundle, memory_node);
  279. local_power[worker_ctx][nimpl] = starpu_task_bundle_expected_power(bundle, perf_arch, nimpl);
  280. //_STARPU_DEBUG("Scheduler heft bundle: task length (%lf) local power (%lf) worker (%u) kernel (%u) \n", local_task_length[worker_ctx],local_power[worker_ctx],worker,nimpl);
  281. }
  282. else
  283. {
  284. local_task_length[worker_ctx][nimpl] = starpu_task_expected_length(task, perf_arch, nimpl);
  285. local_data_penalty[worker_ctx][nimpl] = starpu_task_expected_data_transfer_time(memory_node, task);
  286. local_power[worker_ctx][nimpl] = starpu_task_expected_power(task, perf_arch, nimpl);
  287. double conversion_time = starpu_task_expected_conversion_time(task, perf_arch, nimpl);
  288. if (conversion_time > 0.0)
  289. local_task_length[worker_ctx][nimpl] += conversion_time;
  290. //_STARPU_DEBUG("Scheduler heft bundle: task length (%lf) local power (%lf) worker (%u) kernel (%u) \n", local_task_length[worker_ctx],local_power[worker_ctx],worker,nimpl);
  291. }
  292. double ntasks_end = ntasks[worker] / starpu_worker_get_relative_speedup(perf_arch);
  293. if (ntasks_best == -1
  294. || (!calibrating && ntasks_end < ntasks_best_end) /* Not calibrating, take better worker */
  295. || (!calibrating && isnan(local_task_length[worker_ctx][nimpl])) /* Not calibrating but this worker is being calibrated */
  296. || (calibrating && isnan(local_task_length[worker_ctx][nimpl]) && ntasks_end < ntasks_best_end) /* Calibrating, compete this worker with other non-calibrated */
  297. )
  298. {
  299. ntasks_best_end = ntasks_end;
  300. ntasks_best = worker;
  301. nimpl_best = nimpl;
  302. }
  303. if (isnan(local_task_length[worker_ctx][nimpl]))
  304. /* we are calibrating, we want to speed-up calibration time
  305. * so we privilege non-calibrated tasks (but still
  306. * greedily distribute them to avoid dumb schedules) */
  307. calibrating = 1;
  308. if (isnan(local_task_length[worker_ctx][nimpl])
  309. || _STARPU_IS_ZERO(local_task_length[worker_ctx][nimpl]))
  310. /* there is no prediction available for that task
  311. * with that arch (yet or at all), so switch to a greedy strategy */
  312. unknown = 1;
  313. if (unknown)
  314. continue;
  315. exp_end[worker_ctx][nimpl] = exp_start[worker] + exp_len[worker] + local_task_length[worker_ctx][nimpl];
  316. if (exp_end[worker_ctx][nimpl] < best_exp_end)
  317. {
  318. /* a better solution was found */
  319. best_exp_end = exp_end[worker_ctx][nimpl];
  320. nimpl_best = nimpl;
  321. }
  322. if (isnan(local_power[worker_ctx][nimpl]))
  323. local_power[worker_ctx][nimpl] = 0.;
  324. }
  325. worker_ctx++;
  326. }
  327. *forced_worker = unknown?ntasks_best:-1;
  328. *forced_impl = unknown?nimpl_best:-1;
  329. *best_exp_endp = best_exp_end;
  330. *max_exp_endp = max_exp_end;
  331. }
  332. static int push_conversion_tasks(struct starpu_task *task, unsigned int workerid)
  333. {
  334. unsigned i;
  335. int ret;
  336. unsigned int node = starpu_worker_get_memory_node(workerid);
  337. unsigned sched_ctx_id = task->sched_ctx;
  338. pthread_mutex_t *sched_mutex;
  339. pthread_cond_t *sched_cond;
  340. starpu_worker_get_sched_condition(sched_ctx_id, workerid, &sched_mutex, &sched_cond);
  341. _STARPU_PTHREAD_MUTEX_LOCK(&sched_mutex[workerid]);
  342. for (i = 0; i < task->cl->nbuffers; i++)
  343. {
  344. struct starpu_task *conversion_task;
  345. starpu_data_handle_t handle;
  346. handle = task->handles[i];
  347. if (!_starpu_handle_needs_conversion_task(handle, node))
  348. continue;
  349. conversion_task = _starpu_create_conversion_task(handle, node);
  350. conversion_task->execute_on_a_specific_worker = 1;
  351. conversion_task->workerid = workerid;
  352. conversion_task->mf_skip = 1;
  353. handle->mf_node = node;
  354. ret = _starpu_task_submit_conversion_task(conversion_task, workerid);
  355. STARPU_ASSERT(ret == 0);
  356. }
  357. task->execute_on_a_specific_worker = 1;
  358. task->workerid = workerid;
  359. task->mf_skip= 1;
  360. _STARPU_PTHREAD_MUTEX_UNLOCK(&sched_mutex[workerid]);
  361. return 0;
  362. }
  363. /* TODO: factorize with dmda */
  364. static int _heft_push_task(struct starpu_task *task, unsigned prio, unsigned sched_ctx_id)
  365. {
  366. heft_data *hd = (heft_data*)starpu_get_sched_ctx_policy_data(sched_ctx_id);
  367. int worker, worker_ctx = 0;
  368. unsigned nimpl;
  369. int best = -1, best_in_ctx = -1;
  370. int selected_impl= -1;
  371. /* this flag is set if the corresponding worker is selected because
  372. there is no performance prediction available yet */
  373. int forced_worker;
  374. int forced_impl;
  375. struct worker_collection *workers = starpu_get_worker_collection_of_sched_ctx(sched_ctx_id);
  376. unsigned nworkers_ctx = workers->nworkers;
  377. double local_task_length[nworkers_ctx][STARPU_MAXIMPLEMENTATIONS];
  378. double local_data_penalty[nworkers_ctx][STARPU_MAXIMPLEMENTATIONS];
  379. double local_power[nworkers_ctx][STARPU_MAXIMPLEMENTATIONS];
  380. double exp_end[nworkers_ctx][STARPU_MAXIMPLEMENTATIONS];
  381. double max_exp_end = 0.0;
  382. double best_exp_end;
  383. /*
  384. * Compute the expected end of the task on the various workers,
  385. * and detect if there is some calibration that needs to be done.
  386. */
  387. starpu_task_bundle_t bundle = NULL; //task->bundle;
  388. if(workers->init_cursor)
  389. workers->init_cursor(workers);
  390. compute_all_performance_predictions(task, local_task_length, exp_end,
  391. &max_exp_end, &best_exp_end,
  392. local_data_penalty,
  393. local_power, &forced_worker, &forced_impl,
  394. bundle, sched_ctx_id);
  395. /* If there is no prediction available for that task with that arch we
  396. * want to speed-up calibration time so we force this measurement */
  397. if (forced_worker != -1)
  398. {
  399. _starpu_get_job_associated_to_task(task)->nimpl = forced_impl;
  400. if (_starpu_task_uses_multiformat_handles(task) && !task->mf_skip)
  401. {
  402. /*
  403. * Our task uses multiformat handles, which may need to be converted.
  404. */
  405. push_conversion_tasks(task, forced_worker);
  406. prio = 0;
  407. }
  408. return push_task_on_best_worker(task, forced_worker, 0.0, 0.0, prio, sched_ctx_id);
  409. }
  410. /*
  411. * Determine which worker optimizes the fitness metric which is a
  412. * trade-off between load-balacing, data locality, and energy
  413. * consumption.
  414. */
  415. double fitness[nworkers_ctx][STARPU_MAXIMPLEMENTATIONS];
  416. double best_fitness = -1;
  417. while(workers->has_next(workers))
  418. {
  419. worker = workers->get_next(workers);
  420. for (nimpl = 0; nimpl < STARPU_MAXIMPLEMENTATIONS; nimpl++)
  421. {
  422. if (!starpu_worker_can_execute_task(worker, task, nimpl))
  423. {
  424. /* no one on that queue may execute this task */
  425. //worker_ctx++;
  426. continue;
  427. }
  428. fitness[worker_ctx][nimpl] = hd->alpha*(exp_end[worker_ctx][nimpl] - best_exp_end)
  429. + hd->beta*(local_data_penalty[worker_ctx][nimpl])
  430. + hd->_gamma*(local_power[worker_ctx][nimpl]);
  431. if (exp_end[worker_ctx][nimpl] > max_exp_end)
  432. /* This placement will make the computation
  433. * longer, take into account the idle
  434. * consumption of other cpus */
  435. fitness[worker_ctx][nimpl] += hd->_gamma * hd->idle_power * (exp_end[worker_ctx][nimpl] - max_exp_end) / 1000000.0;
  436. if (best == -1 || fitness[worker_ctx][nimpl] < best_fitness)
  437. {
  438. /* we found a better solution */
  439. best_fitness = fitness[worker_ctx][nimpl];
  440. best = worker;
  441. best_in_ctx = worker_ctx;
  442. selected_impl = nimpl;
  443. }
  444. }
  445. worker_ctx++;
  446. }
  447. /* By now, we must have found a solution */
  448. STARPU_ASSERT(best != -1);
  449. /* we should now have the best worker in variable "best" */
  450. double model_best, transfer_model_best;
  451. if (bundle)
  452. {
  453. /* If we have a task bundle, we have computed the expected
  454. * length for the entire bundle, but not for the task alone. */
  455. enum starpu_perf_archtype perf_arch = starpu_worker_get_perf_archtype(best);
  456. unsigned memory_node = starpu_worker_get_memory_node(best);
  457. model_best = starpu_task_expected_length(task, perf_arch, selected_impl);
  458. transfer_model_best = starpu_task_expected_data_transfer_time(memory_node, task);
  459. /* Remove the task from the bundle since we have made a
  460. * decision for it, and that other tasks should not consider it
  461. * anymore. */
  462. starpu_task_bundle_remove(bundle, task);
  463. }
  464. else
  465. {
  466. model_best = local_task_length[best_in_ctx][selected_impl];
  467. transfer_model_best = local_data_penalty[best_in_ctx][selected_impl];
  468. }
  469. if(workers->init_cursor)
  470. workers->deinit_cursor(workers);
  471. _starpu_get_job_associated_to_task(task)->nimpl = selected_impl;
  472. if (_starpu_task_uses_multiformat_handles(task) && !task->mf_skip)
  473. {
  474. /*
  475. * Our task uses multiformat handles, which may need to be converted.
  476. */
  477. push_conversion_tasks(task, forced_worker);
  478. prio = 0;
  479. }
  480. return push_task_on_best_worker(task, best, model_best, transfer_model_best, prio, sched_ctx_id);
  481. }
  482. static int heft_push_task(struct starpu_task *task)
  483. {
  484. unsigned sched_ctx_id = task->sched_ctx;
  485. pthread_mutex_t *changing_ctx_mutex = starpu_get_changing_ctx_mutex(sched_ctx_id);
  486. unsigned nworkers;
  487. int ret_val = -1;
  488. if (task->priority > 0)
  489. {
  490. _STARPU_PTHREAD_MUTEX_LOCK(changing_ctx_mutex);
  491. nworkers = starpu_get_nworkers_of_sched_ctx(sched_ctx_id);
  492. if(nworkers == 0)
  493. {
  494. _STARPU_PTHREAD_MUTEX_UNLOCK(changing_ctx_mutex);
  495. return ret_val;
  496. }
  497. ret_val = _heft_push_task(task, 1, sched_ctx_id);
  498. _STARPU_PTHREAD_MUTEX_UNLOCK(changing_ctx_mutex);
  499. return ret_val;
  500. }
  501. _STARPU_PTHREAD_MUTEX_LOCK(changing_ctx_mutex);
  502. nworkers = starpu_get_nworkers_of_sched_ctx(sched_ctx_id);
  503. if(nworkers == 0)
  504. {
  505. _STARPU_PTHREAD_MUTEX_UNLOCK(changing_ctx_mutex);
  506. return ret_val;
  507. }
  508. ret_val = _heft_push_task(task, 0, sched_ctx_id);
  509. _STARPU_PTHREAD_MUTEX_UNLOCK(changing_ctx_mutex);
  510. return ret_val;
  511. }
  512. static void heft_deinit(unsigned sched_ctx_id)
  513. {
  514. heft_data *ht = (heft_data*)starpu_get_sched_ctx_policy_data(sched_ctx_id);
  515. free(ht);
  516. starpu_delete_worker_collection_for_sched_ctx(sched_ctx_id);
  517. }
  518. struct starpu_sched_policy heft_policy =
  519. {
  520. .init_sched = heft_init,
  521. .deinit_sched = heft_deinit,
  522. .push_task = heft_push_task,
  523. .push_task_notify = heft_push_task_notify,
  524. .pop_task = NULL,
  525. .pop_every_task = NULL,
  526. .pre_exec_hook = heft_pre_exec_hook,
  527. .post_exec_hook = NULL,
  528. .add_workers = heft_add_workers ,
  529. .remove_workers = heft_remove_workers,
  530. .policy_name = "heft",
  531. .policy_description = "Heterogeneous Earliest Finish Task"
  532. };